Report unstable peers from TSS

CHANGELOG.md

@@ -46,6 +46,7 @@ runtime
 - Add TDX test network chainspec ([#1204](https://github.com/entropyxyz/entropy-core/pull/1204))
 - Test CLI command to retrieve quote and change endpoint / TSS account in one command ([#1198](https://github.com/entropyxyz/entropy-core/pull/1198))
 - On-chain unresponsiveness reporting [(#1215)](https://github.com/entropyxyz/entropy-core/pull/1215)
+- Report unstable peers from TSS [(#1228)](https://github.com/entropyxyz/entropy-core/pull/1228)
 - Add cli options for adding validator [(#1242)](https://github.com/entropyxyz/entropy-core/pull/1242)
 
 ### Changed

crates/protocol/src/listener.rs

@@ -29,7 +29,7 @@ use tokio::sync::{broadcast, mpsc, oneshot};
 pub type ListenerResult = Result<Broadcaster, ListenerErr>;
 
 /// Tracks which validators we are connected to for a particular protocol execution
-/// and sets up channels for exchaning protocol messages
+/// and sets up channels for exchanging protocol messages
 #[derive(Debug)]
 pub struct Listener {
     /// Endpoint to create subscriptions

crates/protocol/tests/helpers/mod.rs

@@ -255,6 +255,7 @@ async fn open_protocol_connections(
             // Check the response as to whether they accepted our SubscribeMessage
             let response_message = encrypted_connection.recv().await?;
             let subscribe_response: Result<(), String> = bincode::deserialize(&response_message)?;
+
             if let Err(error_message) = subscribe_response {
                 return Err(anyhow!(error_message));
             }

crates/testing-utils/src/constants.rs

@@ -17,12 +17,18 @@ use hex_literal::hex;
 use subxt::utils::AccountId32;
 
 lazy_static! {
-    pub static ref ALICE_STASH_ADDRESS: AccountId32 = hex!["be5ddb1579b72e84524fc29e78609e3caf42e85aa118ebfe0b0ad404b5bdd25f"].into();
+    pub static ref ALICE_STASH_ADDRESS: AccountId32 =
+        hex!["be5ddb1579b72e84524fc29e78609e3caf42e85aa118ebfe0b0ad404b5bdd25f"].into(); // subkey inspect //Alice//stash
+    pub static ref BOB_STASH_ADDRESS: AccountId32 =
+        hex!["fe65717dad0447d715f660a0a58411de509b42e6efb8375f562f58a554d5860e"].into(); // subkey inspect //Bob//stash
+    pub static ref CHARLIE_STASH_ADDRESS: AccountId32 =
+        hex!["1e07379407fecc4b89eb7dbd287c2c781cfb1907a96947a3eb18e4f8e7198625"].into(); // subkey inspect //Charlie//stash
+    pub static ref DAVE_STASH_ADDRESS: AccountId32 =
+        hex!["e860f1b1c7227f7c22602f53f15af80747814dffd839719731ee3bba6edc126c"].into(); // subkey inspect //Dave//stash
+
     pub static ref RANDOM_ACCOUNT: AccountId32 = hex!["8676839ca1e196624106d17c56b1efbb90508a86d8053f7d4fcd21127a9f7565"].into();
     pub static ref VALIDATOR_1_STASH_ID: AccountId32 =
         hex!["be5ddb1579b72e84524fc29e78609e3caf42e85aa118ebfe0b0ad404b5bdd25f"].into(); // alice stash;
-    pub static ref BOB_STASH_ADDRESS: AccountId32 =
-        hex!["fe65717dad0447d715f660a0a58411de509b42e6efb8375f562f58a554d5860e"].into(); // subkey inspect //Bob//stash
 
     // See entropy_tss::helpers::validator::get_signer_and_x25519_secret for how these are derived
     pub static ref TSS_ACCOUNTS: Vec<AccountId32> = vec![

crates/threshold-signature-server/src/helpers/signing.rs

@@ -92,9 +92,16 @@ pub async fn do_signing(
     .await?;
 
     let channels = {
-        let ready = timeout(Duration::from_secs(SETUP_TIMEOUT_SECONDS), rx_ready).await?;
-        let broadcast_out = ready??;
-        Channels(broadcast_out, rx_from_others)
+        match timeout(Duration::from_secs(SETUP_TIMEOUT_SECONDS), rx_ready).await {
+            Ok(ready) => {
+                let broadcast_out = ready??;
+                Channels(broadcast_out, rx_from_others)
+            },
+            Err(e) => {
+                let unsubscribed_peers = app_state.unsubscribed_peers(&session_id)?;
+                return Err(ProtocolErr::Timeout { source: e, inactive_peers: unsubscribed_peers });
+            },
+        }
     };
 
     let result = signing_service

crates/threshold-signature-server/src/helpers/user.rs

@@ -87,6 +87,7 @@ pub async fn do_dkg(
         x25519_secret_key,
     )
     .await?;
+
     let channels = {
         let ready = timeout(Duration::from_secs(SETUP_TIMEOUT_SECONDS), rx_ready).await?;
         let broadcast_out = ready??;

crates/threshold-signature-server/src/lib.rs

@@ -208,6 +208,19 @@ impl AppState {
     pub fn new(configuration: Configuration, kv_store: KvManager) -> Self {
         Self { listener_state: ListenerState::default(), configuration, kv_store }
     }
+
+    /// Gets the list of peers who haven't yet subscribed to us for this particular session.
+    pub fn unsubscribed_peers(
+        &self,
+        session_id: &entropy_protocol::SessionId,
+    ) -> Result<Vec<subxt::utils::AccountId32>, crate::signing_client::ProtocolErr> {
+        self.listener_state.unsubscribed_peers(session_id).map_err(|_| {
+            crate::signing_client::ProtocolErr::SessionError(format!(
+                "Unable to get unsubscribed peers for `SessionId` {:?}",
+                session_id,
+            ))
+        })
+    }
 }
 
 pub fn app(app_state: AppState) -> Router {

crates/threshold-signature-server/src/signing_client/api.rs

@@ -138,7 +138,6 @@ pub async fn ws_handler(
 async fn handle_socket_result(socket: WebSocket, app_state: AppState) {
     if let Err(err) = handle_socket(socket, app_state).await {
         tracing::warn!("Websocket connection closed unexpectedly {:?}", err);
-        // TODO here we should inform the chain that signing failed
     };
 }
 
@@ -293,7 +292,19 @@ pub async fn get_channels(
     )
     .await?;
 
-    let ready = timeout(Duration::from_secs(SETUP_TIMEOUT_SECONDS), rx_ready).await?;
-    let broadcast_out = ready??;
-    Ok(Channels(broadcast_out, rx_from_others))
+    match timeout(Duration::from_secs(SETUP_TIMEOUT_SECONDS), rx_ready).await {
+        Ok(ready) => {
+            let broadcast_out = ready??;
+            Ok(Channels(broadcast_out, rx_from_others))
+        },
+        Err(e) => {
+            let unsubscribed_peers = state.unsubscribed_peers(session_id).map_err(|_| {
+                crate::signing_client::ProtocolErr::SessionError(format!(
+                    "Unable to get unsubscribed peers for `SessionId` {:?}",
+                    session_id,
+                ))
+            })?;
+            Err(ProtocolErr::Timeout { source: e, inactive_peers: unsubscribed_peers })
+        },
+    }
 }

crates/threshold-signature-server/src/signing_client/errors.rs

@@ -22,6 +22,7 @@ use axum::{
 };
 use entropy_kvdb::kv_manager::error::InnerKvError;
 use entropy_protocol::errors::ProtocolExecutionErr;
+use subxt::utils::AccountId32;
 use thiserror::Error;
 use tokio::sync::oneshot::error::RecvError;
 
@@ -46,8 +47,8 @@ pub enum ProtocolErr {
     Deserialization(String),
     #[error("Oneshot timeout error: {0}")]
     OneshotTimeout(#[from] RecvError),
-    #[error("Subscribe API error: {0}")]
-    Subscribe(#[from] SubscribeErr),
+    #[error("Subscribe API error: {source} by TSS Account `{account_id:?}`")]
+    Subscribe { source: SubscribeErr, account_id: AccountId32 },
     #[error("reqwest error: {0}")]
     Reqwest(#[from] reqwest::Error),
     #[error("Utf8Error: {0:?}")]
@@ -70,18 +71,21 @@ pub enum ProtocolErr {
     UserError(String),
     #[error("Validation Error: {0}")]
     ValidationErr(#[from] crate::validation::errors::ValidationErr),
-    #[error("Subscribe message rejected: {0}")]
-    BadSubscribeMessage(String),
+    #[error("Subscribe message rejected: {message} by TSS Account `{account_id}`")]
+    BadSubscribeMessage { message: String, account_id: AccountId32 },
     #[error("From Hex Error: {0}")]
     FromHex(#[from] hex::FromHexError),
     #[error("Conversion Error: {0}")]
     Conversion(&'static str),
-    #[error("Could not open ws connection: {0}")]
-    ConnectionError(#[from] tokio_tungstenite::tungstenite::Error),
-    #[error("Timed out waiting for remote party")]
-    Timeout(#[from] tokio::time::error::Elapsed),
-    #[error("Encrypted connection error {0}")]
-    EncryptedConnection(String),
+    #[error("Could not open ws connection: {source} with the TSS Account `{account_id:?}`")]
+    ConnectionError { source: tokio_tungstenite::tungstenite::Error, account_id: AccountId32 },
+    #[error("Timed out while waiting for peer(s): {:?}", inactive_peers)]
+    Timeout { source: tokio::time::error::Elapsed, inactive_peers: Vec<AccountId32> },
+    #[error("Encrypted connection error {source:?} with the TSS Account `{account_id:?}`")]
+    EncryptedConnection {
+        source: entropy_protocol::protocol_transport::errors::EncryptedConnectionErr,
+        account_id: AccountId32,
+    },
     #[error("Program error: {0}")]
     ProgramError(#[from] crate::user::errors::ProgramError),
     #[error("Invalid length for converting address")]
@@ -143,6 +147,8 @@ pub enum SubscribeErr {
     VersionMismatch(
         #[from] entropy_protocol::protocol_transport::errors::ProtocolVersionMismatchError,
     ),
+    #[error("Unable to find `Listener` with `SessionId`: {0:?}")]
+    NoSessionId(entropy_protocol::SessionId),
 }
 
 impl IntoResponse for SubscribeErr {

crates/threshold-signature-server/src/signing_client/mod.rs

@@ -48,4 +48,20 @@ impl ListenerState {
             .map_err(|e| SubscribeErr::LockError(e.to_string()))?
             .contains_key(session_id))
     }
+
+    /// Gets the list of peers who haven't yet subscribed to us for this particular session.
+    pub fn unsubscribed_peers(
+        &self,
+        session_id: &SessionId,
+    ) -> Result<Vec<subxt::utils::AccountId32>, SubscribeErr> {
+        let listeners =
+            self.listeners.lock().map_err(|e| SubscribeErr::LockError(e.to_string()))?;
+        let listener =
+            listeners.get(session_id).ok_or(SubscribeErr::NoSessionId(session_id.clone()))?;
+
+        let unsubscribed_peers =
+            listener.validators.keys().map(|id| subxt::utils::AccountId32(*id)).collect();
+
+        Ok(unsubscribed_peers)
+    }
 }

crates/threshold-signature-server/src/signing_client/protocol_transport.rs

@@ -55,14 +55,31 @@ pub async fn open_protocol_connections(
     let connect_to_validators = validators_info
         .iter()
         .filter(|validators_info| {
-            // Decide whether to initiate a connection by comparing accound ids
-            // otherwise, we wait for them to connect to us
-            signer.public().0 > validators_info.tss_account.0
+            // Decide whether to initiate a connection by comparing account IDs, otherwise we wait
+            // for them to connect to us
+            let initiate_connection = signer.public().0 > validators_info.tss_account.0;
+            if !initiate_connection {
+                tracing::debug!(
+                    "Waiting for {:?} to open a connect with us.",
+                    validators_info.tss_account.0
+                );
+            }
+
+            initiate_connection
         })
         .map(|validator_info| async move {
+            tracing::debug!(
+                "Attempting to open protocol connections with {:?}",
+                validator_info.tss_account.0.clone()
+            );
+
             // Open a ws connection
             let ws_endpoint = format!("ws://{}/ws", validator_info.ip_address);
-            let (ws_stream, _response) = connect_async(ws_endpoint).await?;
+            let (ws_stream, _response) =
+                connect_async(ws_endpoint).await.map_err(|e| ProtocolErr::ConnectionError {
+                    source: e,
+                    account_id: validator_info.tss_account.clone(),
+                })?;
 
             // Send a SubscribeMessage in the payload of the final handshake message
             let subscribe_message_vec =
@@ -75,32 +92,50 @@ pub async fn open_protocol_connections(
                 subscribe_message_vec,
             )
             .await
-            .map_err(|e| ProtocolErr::EncryptedConnection(e.to_string()))?;
+            .map_err(|e| ProtocolErr::EncryptedConnection {
+                source: e,
+                account_id: validator_info.tss_account.clone(),
+            })?;
 
             // Check the response as to whether they accepted our SubscribeMessage
-            let response_message = encrypted_connection
-                .recv()
-                .await
-                .map_err(|e| ProtocolErr::EncryptedConnection(e.to_string()))?;
+            let response_message = encrypted_connection.recv().await.map_err(|e| {
+                ProtocolErr::EncryptedConnection {
+                    source: e,
+                    account_id: validator_info.tss_account.clone(),
+                }
+            })?;
+
             let subscribe_response: Result<(), String> = bincode::deserialize(&response_message)?;
             if let Err(error_message) = subscribe_response {
                 // In future versions, we can check here if the error is VersionTooNew(version)
                 // and if possible the downgrade protocol messages used to be backward compatible
-                return Err(ProtocolErr::BadSubscribeMessage(error_message));
+                return Err(ProtocolErr::BadSubscribeMessage {
+                    message: error_message,
+                    account_id: validator_info.tss_account.clone(),
+                });
             }
 
             // Setup channels
-            let ws_channels = get_ws_channels(state, session_id, &validator_info.tss_account)?;
+            let ws_channels = get_ws_channels(state, session_id, &validator_info.tss_account)
+                .map_err(|e| ProtocolErr::Subscribe {
+                    source: e,
+                    account_id: validator_info.tss_account.clone(),
+                })?;
 
             let remote_party_id = PartyId::new(validator_info.tss_account.clone());
+            let account_id = validator_info.tss_account.clone();
 
             // Handle protocol messages
             tokio::spawn(async move {
-                if let Err(err) =
-                    ws_to_channels(encrypted_connection, ws_channels, remote_party_id).await
-                {
-                    tracing::warn!("{:?}", err);
-                };
+                ws_to_channels(encrypted_connection, ws_channels, remote_party_id).await.map_err(
+                    |err| {
+                        tracing::warn!("{:?}", err);
+                        Err::<(), ProtocolErr>(ProtocolErr::EncryptedConnection {
+                            source: err.into(),
+                            account_id,
+                        })
+                    },
+                )
             });
 
             Ok::<_, ProtocolErr>(())

crates/threshold-signature-server/src/user/api.rs

@@ -41,6 +41,7 @@ use x25519_dalek::StaticSecret;
 
 use super::UserErr;
 use crate::chain_api::entropy::runtime_types::pallet_registry::pallet::RegisteredInfo;
+use crate::signing_client::ProtocolErr;
 use crate::{
     chain_api::{entropy, get_api, get_rpc, EntropyConfig},
     helpers::{
@@ -351,14 +352,15 @@ pub async fn sign_tx(
             request_limit,
             derivation_path,
         )
-        .await
-        .map(|signature| {
-            (
+        .await;
+
+        let signing_protocol_output = match signing_protocol_output {
+            Ok(signature) => Ok((
                 BASE64_STANDARD.encode(signature.to_rsv_bytes()),
                 signer.signer().sign(&signature.to_rsv_bytes()),
-            )
-        })
-        .map_err(|error| error.to_string());
+            )),
+            Err(e) => Err(handle_protocol_errors(&api, &rpc, &signer, e).await.unwrap_err()),
+        };
 
         // This response chunk is sent later with the result of the signing protocol
         if response_tx.try_send(serde_json::to_string(&signing_protocol_output)).is_err() {
@@ -370,6 +372,56 @@ pub async fn sign_tx(
     Ok((StatusCode::OK, Body::from_stream(response_rx)))
 }
 
+/// Helper for handling different protocol errors.
+///
+/// If the error is of the reportable type (e.g an offence from another peer) it will be reported
+/// on-chain by this helper.
+async fn handle_protocol_errors(
+    api: &OnlineClient<EntropyConfig>,
+    rpc: &LegacyRpcMethods<EntropyConfig>,
+    signer: &PairSigner<EntropyConfig, sr25519::Pair>,
+    error: ProtocolErr,
+) -> Result<(), String> {
+    let peers_to_report: Vec<SubxtAccountId32> = match &error {
+        ProtocolErr::ConnectionError { account_id, .. }
+        | ProtocolErr::EncryptedConnection { account_id, .. }
+        | ProtocolErr::BadSubscribeMessage { account_id, .. }
+        | ProtocolErr::Subscribe { account_id, .. } => vec![account_id.clone()],
+
+        ProtocolErr::Timeout { inactive_peers, .. } => inactive_peers.clone(),
+        _ => vec![],
+    };
+
+    // This is a non-reportable error, so we don't do any further processing with the error
+    if peers_to_report.is_empty() {
+        return Err(error.to_string());
+    }
+
+    tracing::debug!("Reporting `{:?}` for `{}`", peers_to_report.clone(), error.to_string());
+
+    let mut failed_reports = Vec::new();
+    for peer in peers_to_report {
+        let report_unstable_peer_tx =
+            entropy::tx().staking_extension().report_unstable_peer(peer.clone());
+
+        if let Err(tx_error) =
+            submit_transaction(api, rpc, signer, &report_unstable_peer_tx, None).await
+        {
+            failed_reports.push(format!("{}", tx_error));
+        }
+    }
+
+    if failed_reports.is_empty() {
+        Err(error.to_string())
+    } else {
+        Err(format!(
+            "Failed to report peers for `{}` due to `{}`)",
+            error,
+            failed_reports.join(", ")
+        ))
+    }
+}
+
 /// HTTP POST endpoint called by the off-chain worker (Propagation pallet) during the network
 /// jumpstart.
 ///