fix: protocol timeout (#462)

* Added timeouts for presignature and triple generation protocol

* Handle signature generation timeout

node/src/protocol/cryptography.rs

@@ -334,12 +334,22 @@ impl CryptographicProtocol for RunningState {
         sign_queue.organize(&self, ctx.me().await);
         let my_requests = sign_queue.my_requests(ctx.me().await);
         while presignature_manager.my_len() > 0 {
+            if signature_manager.failed_len() > 0 {
+                let Some(presignature) = presignature_manager.take_mine() else {
+                    break;
+                };
+                signature_manager.retry_failed_generation(presignature);
+                break;
+            }
+
             let Some((receipt_id, _)) = my_requests.iter().next() else {
                 break;
             };
+
             let Some(presignature) = presignature_manager.take_mine() else {
                 break;
             };
+
             let receipt_id = *receipt_id;
             let my_request = my_requests.remove(&receipt_id).unwrap();
             signature_manager.generate(
@@ -353,7 +363,7 @@ impl CryptographicProtocol for RunningState {
         }
         drop(sign_queue);
         drop(presignature_manager);
-        for (p, msg) in signature_manager.poke()? {
+        for (p, msg) in signature_manager.poke() {
             let info = self.participants.get(&p).unwrap();
             messages.push(info.clone(), MpcMessage::Signature(msg));
         }

node/src/protocol/presignature.rs

@@ -7,6 +7,7 @@ use cait_sith::{KeygenOutput, PresignArguments, PresignOutput};
 use k256::Secp256k1;
 use std::collections::hash_map::Entry;
 use std::collections::{HashMap, VecDeque};
+use std::time::Instant;
 
 /// Unique number used to identify a specific ongoing presignature generation protocol.
 /// Without `PresignatureId` it would be unclear where to route incoming cait-sith presignature
@@ -25,6 +26,40 @@ pub struct PresignatureGenerator {
     pub triple0: TripleId,
     pub triple1: TripleId,
     pub mine: bool,
+    pub timestamp: Instant,
+}
+
+impl PresignatureGenerator {
+    pub fn new(
+        protocol: PresignatureProtocol,
+        triple0: TripleId,
+        triple1: TripleId,
+        mine: bool,
+    ) -> Self {
+        Self {
+            protocol,
+            triple0,
+            triple1,
+            mine,
+            timestamp: Instant::now(),
+        }
+    }
+
+    pub fn poke(&mut self) -> Result<Action<PresignOutput<Secp256k1>>, ProtocolError> {
+        if self.timestamp.elapsed() > crate::types::PROTOCOL_TIMEOUT {
+            tracing::info!(
+                self.triple0,
+                self.triple1,
+                self.mine,
+                "presignature protocol timed out"
+            );
+            return Err(ProtocolError::Other(
+                anyhow::anyhow!("presignature protocol timed out").into(),
+            ));
+        }
+
+        self.protocol.poke()
+    }
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -111,12 +146,9 @@ impl PresignatureManager {
                 threshold,
             },
         )?);
-        Ok(PresignatureGenerator {
-            protocol,
-            triple0: triple0.id,
-            triple1: triple1.id,
-            mine,
-        })
+        Ok(PresignatureGenerator::new(
+            protocol, triple0.id, triple1.id, mine,
+        ))
     }
 
     /// Starts a new presignature generation protocol.
@@ -213,8 +245,7 @@ impl PresignatureManager {
         let mut result = Ok(());
         self.generators.retain(|id, generator| {
             loop {
-                let protocol = &mut generator.protocol;
-                let action = match protocol.poke() {
+                let action = match generator.poke() {
                     Ok(action) => action,
                     Err(e) => {
                         result = Err(e);

node/src/protocol/signature.rs

@@ -16,7 +16,8 @@ use rand::rngs::StdRng;
 use rand::seq::{IteratorRandom, SliceRandom};
 use rand::SeedableRng;
 use std::collections::hash_map::Entry;
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
+use std::time::Instant;
 
 pub struct SignRequest {
     pub receipt_id: CryptoHash,
@@ -95,11 +96,56 @@ pub struct SignatureGenerator {
     pub msg_hash: [u8; 32],
     pub epsilon: Scalar,
     pub delta: Scalar,
+    pub timestamp: Instant,
+}
+
+impl SignatureGenerator {
+    pub fn new(
+        protocol: SignatureProtocol,
+        proposer: Participant,
+        presignature_id: PresignatureId,
+        msg_hash: [u8; 32],
+        epsilon: Scalar,
+        delta: Scalar,
+    ) -> Self {
+        Self {
+            protocol,
+            proposer,
+            presignature_id,
+            msg_hash,
+            epsilon,
+            delta,
+            timestamp: Instant::now(),
+        }
+    }
+
+    pub fn poke(&mut self) -> Result<Action<FullSignature<Secp256k1>>, ProtocolError> {
+        if self.timestamp.elapsed() > crate::types::PROTOCOL_SIGNATURE_TIMEOUT {
+            tracing::info!(self.presignature_id, "signature protocol timed out");
+            return Err(ProtocolError::Other(
+                anyhow::anyhow!("signature protocol timed out").into(),
+            ));
+        }
+
+        self.protocol.poke()
+    }
+}
+
+/// Generator for signature thas has failed. Only retains essential information
+/// for starting up this failed signature once again.
+pub struct FailedGenerator {
+    pub proposer: Participant,
+    pub msg_hash: [u8; 32],
+    pub epsilon: Scalar,
+    pub delta: Scalar,
+    pub timestamp: Instant,
 }
 
 pub struct SignatureManager {
     /// Ongoing signature generation protocols.
     generators: HashMap<CryptoHash, SignatureGenerator>,
+    /// Failed signatures awaiting to be retried.
+    failed_generators: VecDeque<(CryptoHash, FailedGenerator)>,
     /// Generated signatures assigned to the current node that are yet to be published.
     signatures: Vec<(CryptoHash, [u8; 32], FullSignature<Secp256k1>)>,
 
@@ -118,6 +164,7 @@ impl SignatureManager {
     ) -> Self {
         Self {
             generators: HashMap::new(),
+            failed_generators: VecDeque::new(),
             signatures: Vec::new(),
             participants,
             me,
@@ -126,6 +173,10 @@ impl SignatureManager {
         }
     }
 
+    pub fn failed_len(&self) -> usize {
+        self.failed_generators.len()
+    }
+
     #[allow(clippy::too_many_arguments)]
     fn generate_internal(
         participants: &[Participant],
@@ -151,14 +202,31 @@ impl SignatureManager {
             output,
             Scalar::from_bytes(&msg_hash),
         )?);
-        Ok(SignatureGenerator {
+        Ok(SignatureGenerator::new(
             protocol,
             proposer,
-            presignature_id: presignature.id,
+            presignature.id,
             msg_hash,
             epsilon,
             delta,
-        })
+        ))
+    }
+
+    pub fn retry_failed_generation(&mut self, presignature: Presignature) -> Option<()> {
+        let (hash, failed_generator) = self.failed_generators.pop_front()?;
+        let generator = Self::generate_internal(
+            &self.participants,
+            self.me,
+            self.public_key,
+            failed_generator.proposer,
+            presignature,
+            failed_generator.msg_hash,
+            failed_generator.epsilon,
+            failed_generator.delta,
+        )
+        .unwrap();
+        self.generators.insert(hash, generator);
+        Some(())
     }
 
     /// Starts a new presignature generation protocol.
@@ -231,16 +299,24 @@ impl SignatureManager {
     /// messages to be sent to the respective participant.
     ///
     /// An empty vector means we cannot progress until we receive a new message.
-    pub fn poke(&mut self) -> Result<Vec<(Participant, SignatureMessage)>, ProtocolError> {
+    pub fn poke(&mut self) -> Vec<(Participant, SignatureMessage)> {
         let mut messages = Vec::new();
-        let mut result = Ok(());
         self.generators.retain(|receipt_id, generator| {
             loop {
-                let protocol = &mut generator.protocol;
-                let action = match protocol.poke() {
+                let action = match generator.poke() {
                     Ok(action) => action,
-                    Err(e) => {
-                        result = Err(e);
+                    Err(err) => {
+                        tracing::warn!(?err, "signature failed to be produced; pushing request back into failed queue");
+                        self.failed_generators.push_back((
+                            *receipt_id,
+                            FailedGenerator {
+                                proposer: generator.proposer,
+                                msg_hash: generator.msg_hash,
+                                epsilon: generator.epsilon,
+                                delta: generator.delta,
+                                timestamp: generator.timestamp,
+                            },
+                        ));
                         break false;
                     }
                 };
@@ -299,7 +375,7 @@ impl SignatureManager {
                 }
             }
         });
-        result.map(|_| messages)
+        messages
     }
 
     pub async fn publish<T: Signer + ExposeAccountId>(

node/src/protocol/triple.rs

@@ -4,13 +4,14 @@ use crate::storage::triple_storage::{LockTripleNodeStorageBox, TripleData};
 use crate::types::TripleProtocol;
 use crate::util::AffinePointExt;
 use cait_sith::protocol::{Action, InitializationError, Participant, ProtocolError};
-use cait_sith::triples::{TriplePub, TripleShare};
+use cait_sith::triples::{TripleGenerationOutput, TriplePub, TripleShare};
 use highway::{HighwayHash, HighwayHasher};
 use k256::elliptic_curve::group::GroupEncoding;
 use k256::Secp256k1;
 use serde::{Deserialize, Serialize};
 use std::collections::hash_map::Entry;
 use std::collections::{HashMap, VecDeque};
+use std::time::Instant;
 
 /// Unique number used to identify a specific ongoing triple generation protocol.
 /// Without `TripleId` it would be unclear where to route incoming cait-sith triple generation
@@ -25,13 +26,40 @@ pub struct Triple {
     pub public: TriplePub<Secp256k1>,
 }
 
+pub struct TripleGenerator {
+    pub id: TripleId,
+    pub protocol: TripleProtocol,
+    pub timestamp: Instant,
+}
+
+impl TripleGenerator {
+    pub fn new(id: TripleId, protocol: TripleProtocol) -> Self {
+        Self {
+            id,
+            protocol,
+            timestamp: Instant::now(),
+        }
+    }
+
+    pub fn poke(&mut self) -> Result<Action<TripleGenerationOutput<Secp256k1>>, ProtocolError> {
+        if self.timestamp.elapsed() > crate::types::PROTOCOL_TIMEOUT {
+            tracing::info!(id = self.id, "triple protocol timed out");
+            return Err(ProtocolError::Other(
+                anyhow::anyhow!("triple protocol timed out").into(),
+            ));
+        }
+
+        self.protocol.poke()
+    }
+}
+
 /// Abstracts how triples are generated by providing a way to request a new triple that will be
 /// complete some time in the future and a way to take an already generated triple.
 pub struct TripleManager {
     /// Completed unspent triples
     pub triples: HashMap<TripleId, Triple>,
     /// Ongoing triple generation protocols
-    pub generators: HashMap<TripleId, TripleProtocol>,
+    pub generators: HashMap<TripleId, TripleGenerator>,
     /// List of triple ids generation of which was initiated by the current node.
     pub mine: VecDeque<TripleId>,
     pub participants: Vec<Participant>,
@@ -100,7 +128,8 @@ impl TripleManager {
             self.me,
             self.threshold,
         )?);
-        self.generators.insert(id, protocol);
+        self.generators
+            .insert(id, TripleGenerator::new(id, protocol));
         Ok(())
     }
 
@@ -181,10 +210,10 @@ impl TripleManager {
                         self.me,
                         self.threshold,
                     )?);
-                    let generator = e.insert(protocol);
-                    Ok(Some(generator))
+                    let generator = e.insert(TripleGenerator::new(id, protocol));
+                    Ok(Some(&mut generator.protocol))
                 }
-                Entry::Occupied(e) => Ok(Some(e.into_mut())),
+                Entry::Occupied(e) => Ok(Some(&mut e.into_mut().protocol)),
             }
         }
     }
@@ -197,9 +226,9 @@ impl TripleManager {
         let mut messages = Vec::new();
         let mut result = Ok(());
         let mut triples_to_insert = Vec::new();
-        self.generators.retain(|id, protocol| {
+        self.generators.retain(|id, generator| {
             loop {
-                let action = match protocol.poke() {
+                let action = match generator.poke() {
                     Ok(action) => action,
                     Err(e) => {
                         result = Err(e);

node/src/types.rs

@@ -1,4 +1,5 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use cait_sith::protocol::{InitializationError, Participant};
 use cait_sith::triples::TripleGenerationOutput;
@@ -9,6 +10,12 @@ use tokio::sync::{RwLock, RwLockWriteGuard};
 
 use crate::protocol::contract::ResharingContractState;
 
+/// Default timeout for triple/presig generation protocols. Times out after 5 minutes of being alive.
+pub const PROTOCOL_TIMEOUT: Duration = Duration::from_secs(5 * 60);
+
+/// Default timeout for signature generation protocol. Times out after 10 minutes of being alive.
+pub const PROTOCOL_SIGNATURE_TIMEOUT: Duration = Duration::from_secs(10 * 60);
+
 pub type SecretKeyShare = <Secp256k1 as CurveArithmetic>::Scalar;
 pub type PublicKey = <Secp256k1 as CurveArithmetic>::AffinePoint;
 pub type TripleProtocol =