component test: test_presig_stall_from_slow_redis

chain-signatures/node/src/protocol/presignature.rs

@@ -29,6 +29,19 @@ use tokio::sync::{mpsc, watch};
 use tokio::task::JoinHandle;
 use tokio::time;
 
+pub const PRESIGNATURE_POSIT_TIMEOUT: Duration =
+    Duration::from_millis(if cfg!(feature = "test-feature") {
+        3_000
+    } else {
+        10_000
+    });
+pub const PRESIGNATURE_POSIT_EXTRA_DELAY: Duration =
+    Duration::from_millis(if cfg!(feature = "test-feature") {
+        500
+    } else {
+        2_000
+    });
+
 /// Unique number used to identify a specific ongoing presignature generation protocol.
 /// Without `PresignatureId` it would be unclear where to route incoming cait-sith presignature
 /// generation messages.
@@ -711,7 +724,7 @@ impl PresignatureSpawner {
         loop {
             tokio::select! {
                 _ = expiration_interval.tick() => {
-                    for (id, action) in self.posits.expire_and_start(self.threshold, Duration::from_secs(10), Duration::from_secs(2)) {
+                    for (id, action) in self.posits.expire_and_start(self.threshold, PRESIGNATURE_POSIT_TIMEOUT, PRESIGNATURE_POSIT_EXTRA_DELAY) {
                         let PositInternalAction::StartProtocol(participants, positor) = action else {
                             tracing::warn!(
                                 ?id,

chain-signatures/node/src/storage/protocol_storage.rs

@@ -3,8 +3,11 @@ use deadpool_redis::{Connection, Pool};
 use near_sdk::AccountId;
 use redis::{AsyncCommands, FromRedisValue, ToRedisArgs};
 use std::collections::{HashMap, HashSet};
+use std::fmt;
 use std::sync::{Arc, OnceLock};
-use std::{fmt, time::Instant};
+#[cfg(feature = "test-feature")]
+use std::time::Duration;
+use std::time::Instant;
 use tokio::sync::RwLock;
 use tracing;
 
@@ -207,6 +210,9 @@ impl<Id: Eq + std::hash::Hash> ReservedState<Id> {
     }
 }
 
+#[cfg(feature = "test-feature")]
+type OperationDelays = Arc<std::sync::Mutex<HashMap<&'static str, Duration>>>;
+
 #[derive(Debug)]
 pub struct ProtocolStorage<A: ProtocolArtifact> {
     redis_pool: Pool,
@@ -215,6 +221,8 @@ pub struct ProtocolStorage<A: ProtocolArtifact> {
     owner_keys: String,
     account_id: AccountId,
     me: Arc<OnceLock<Participant>>,
+    #[cfg(feature = "test-feature")]
+    operation_delays: OperationDelays,
     _phantom: std::marker::PhantomData<A>,
 }
 
@@ -227,6 +235,8 @@ impl<A: ProtocolArtifact> Clone for ProtocolStorage<A> {
             owner_keys: self.owner_keys.clone(),
             account_id: self.account_id.clone(),
             me: self.me.clone(),
+            #[cfg(feature = "test-feature")]
+            operation_delays: self.operation_delays.clone(),
             _phantom: std::marker::PhantomData,
         }
     }
@@ -245,9 +255,21 @@ impl<A: ProtocolArtifact> ProtocolStorage<A> {
             owner_keys,
             account_id: account_id.clone(),
             me: Arc::new(OnceLock::new()),
+            #[cfg(feature = "test-feature")]
+            operation_delays: Arc::new(std::sync::Mutex::new(HashMap::new())),
             _phantom: std::marker::PhantomData,
         }
     }
+
+    /// Inject artificial latency for a specific redis operation (e.g. "take_mine",
+    /// "insert", "contains"). Shared across clones via `Arc`.
+    #[cfg(feature = "test-feature")]
+    pub fn set_operation_delay(&self, operation: &'static str, delay: Duration) {
+        self.operation_delays
+            .lock()
+            .unwrap()
+            .insert(operation, delay);
+    }
 }
 
 impl<A: ProtocolArtifact> ProtocolStorage<A> {
@@ -310,6 +332,19 @@ impl<A: ProtocolArtifact> ProtocolStorage<A> {
             .ok()
     }
 
+    #[cfg(feature = "test-feature")]
+    async fn apply_operation_delay(&self, operation: &str) {
+        let delay = self
+            .operation_delays
+            .lock()
+            .unwrap()
+            .get(operation)
+            .copied();
+        if let Some(d) = delay {
+            tokio::time::sleep(d).await;
+        }
+    }
+
     pub async fn fetch_owned(&self) -> Result<Vec<A::Id>, StorageError> {
         self.fetch_owned_by(self.me()?).await
     }
@@ -506,6 +541,8 @@ impl<A: ProtocolArtifact> ProtocolStorage<A> {
     /// persisted as a dedicated Redis set for later holder-tracking.
     /// Private: callers must use `create_slot()` + `ArtifactSlot::insert()`.
     async fn insert(&self, artifact: A, owner: Participant) -> bool {
+        #[cfg(feature = "test-feature")]
+        self.apply_operation_delay("insert").await;
         const SCRIPT: &str = r#"
             local artifact_key = KEYS[1]
             local owner_keys = KEYS[2]
@@ -566,6 +603,8 @@ impl<A: ProtocolArtifact> ProtocolStorage<A> {
     }
 
     pub async fn contains(&self, id: A::Id) -> bool {
+        #[cfg(feature = "test-feature")]
+        self.apply_operation_delay("contains").await;
         let Some(mut conn) = self.connect().await else {
             return false;
         };
@@ -755,6 +794,8 @@ impl<A: ProtocolArtifact> ProtocolStorage<A> {
     /// It is very important to NOT reuse the same artifact twice for two different
     /// protocols.
     pub async fn take_mine(&self) -> Option<ArtifactTaken<A>> {
+        #[cfg(feature = "test-feature")]
+        self.apply_operation_delay("take_mine").await;
         const SCRIPT: &str = r#"
             local artifact_key = KEYS[1]
             local mine_key = KEYS[2]