fix: missing signers' votes (#1243)

* move processing_delay to request_decider. Retry sending decisions

* revert back bitcoin_processing_delay

* remove unused import

* address tests comments

* update new tests

* add concatenate_blocks to TestData Params. fix test

* revert to old behaviour for concatenate_blocks

* remove refuse

* don't bail on retry fail

* improve tests

* fix get_deposit_signer_decisions in_memory

* refactor fetching_deposit_signer_decisions test

* improve test comments

* use inspect_err

signer/src/api/new_block.rs

@@ -667,6 +667,7 @@ mod tests {
             num_deposit_requests_per_block: 1,
             num_withdraw_requests_per_block: 1,
             num_signers_per_request: 0,
+            consecutive_blocks: false,
         };
         let db = ctx.inner_storage();
         let test_data = TestData::generate(&mut rng, &[], &test_params);
@@ -739,6 +740,7 @@ mod tests {
             num_deposit_requests_per_block: 1,
             num_withdraw_requests_per_block: 1,
             num_signers_per_request: 0,
+            consecutive_blocks: false,
         };
         let db = ctx.inner_storage();
         let test_data = TestData::generate(&mut rng, &[], &test_params);
@@ -787,6 +789,7 @@ mod tests {
             num_deposit_requests_per_block: 2,
             num_withdraw_requests_per_block: 2,
             num_signers_per_request: 0,
+            consecutive_blocks: false,
         };
 
         let db = ctx.inner_storage();
@@ -859,6 +862,7 @@ mod tests {
             num_deposit_requests_per_block: 2,
             num_withdraw_requests_per_block: 2,
             num_signers_per_request: 0,
+            consecutive_blocks: false,
         };
 
         let db = ctx.inner_storage();
@@ -920,6 +924,7 @@ mod tests {
             num_deposit_requests_per_block: 2,
             num_withdraw_requests_per_block: 2,
             num_signers_per_request: 0,
+            consecutive_blocks: false,
         };
 
         let test_data = TestData::generate(&mut rng, &[], &test_params);

signer/src/config/mod.rs

@@ -289,6 +289,9 @@ pub struct SignerConfig {
     /// How many bitcoin blocks back from the chain tip the signer will
     /// look for requests.
     pub context_window: u16,
+    /// How many bitcoin blocks back from the chain tip the signer will
+    /// look for deposit decisions to retry to propagate.
+    pub deposit_decisions_retry_window: u16,
     /// The maximum duration of a signing round before the coordinator will
     /// time out and return an error.
     #[serde(deserialize_with = "duration_seconds_deserializer")]
@@ -468,6 +471,7 @@ impl Settings {
         // after https://github.com/stacks-network/sbtc/issues/1004 gets
         // done.
         cfg_builder = cfg_builder.set_default("signer.context_window", 1000)?;
+        cfg_builder = cfg_builder.set_default("signer.deposit_decisions_retry_window", 3)?;
         cfg_builder = cfg_builder.set_default("signer.dkg_max_duration", 120)?;
         cfg_builder = cfg_builder.set_default("signer.bitcoin_presign_request_max_duration", 30)?;
         cfg_builder = cfg_builder.set_default("signer.signer_round_max_duration", 30)?;
@@ -596,6 +600,7 @@ mod tests {
         assert_eq!(settings.signer.sbtc_bitcoin_start_height, Some(101));
         assert_eq!(settings.signer.bootstrap_signatures_required, 2);
         assert_eq!(settings.signer.context_window, 1000);
+        assert_eq!(settings.signer.deposit_decisions_retry_window, 3);
         assert!(settings.signer.prometheus_exporter_endpoint.is_none());
         assert_eq!(
             settings.signer.bitcoin_presign_request_max_duration,
@@ -914,6 +919,7 @@ mod tests {
                 .remove(parameter);
         };
         remove_parameter("context_window");
+        remove_parameter("deposit_decisions_retry_window");
         remove_parameter("signer_round_max_duration");
         remove_parameter("bitcoin_presign_request_max_duration");
         remove_parameter("dkg_max_duration");
@@ -926,6 +932,7 @@ mod tests {
         let settings = Settings::new(Some(&new_config.path())).unwrap();
 
         assert_eq!(settings.signer.context_window, 1000);
+        assert_eq!(settings.signer.deposit_decisions_retry_window, 3);
         assert_eq!(
             settings.signer.bitcoin_presign_request_max_duration,
             Duration::from_secs(30)

signer/src/main.rs

@@ -370,6 +370,7 @@ async fn run_request_decider(ctx: impl Context) -> Result<(), Error> {
         network,
         context: ctx.clone(),
         context_window: config.signer.context_window,
+        deposit_decisions_retry_window: config.signer.deposit_decisions_retry_window,
         blocklist_checker: BlocklistClient::new(&ctx),
         signer_private_key: config.signer.private_key,
     };

signer/src/message.rs

@@ -7,6 +7,7 @@ use crate::bitcoin::validation::TxRequestIds;
 use crate::keys::PublicKey;
 use crate::stacks::contracts::ContractCall;
 use crate::stacks::contracts::StacksTx;
+use crate::storage::model;
 use crate::storage::model::BitcoinBlockHash;
 use crate::storage::model::StacksTxId;
 
@@ -139,6 +140,17 @@ pub struct SignerDepositDecision {
     pub can_sign: bool,
 }
 
+impl From<model::DepositSigner> for SignerDepositDecision {
+    fn from(signer: model::DepositSigner) -> Self {
+        Self {
+            txid: signer.txid.into(),
+            output_index: signer.output_index,
+            can_accept: signer.can_accept,
+            can_sign: signer.can_sign,
+        }
+    }
+}
+
 /// Represents a decision related to signer withdrawal.
 #[derive(Debug, Clone, PartialEq)]
 #[cfg_attr(feature = "testing", derive(fake::Dummy))]

signer/src/request_decider.rs

@@ -50,6 +50,9 @@ pub struct RequestDeciderEventLoop<C, N, B> {
     pub signer_private_key: PrivateKey,
     /// How many bitcoin blocks back from the chain tip the signer will look for requests.
     pub context_window: u16,
+    /// How many bitcoin blocks back from the chain tip the signer will look for deposit
+    /// decisions to retry to propagate.
+    pub deposit_decisions_retry_window: u16,
 }
 
 /// This function defines which messages this event loop is interested
@@ -137,6 +140,23 @@ where
         let span = tracing::Span::current();
         span.record("chain_tip", tracing::field::display(chain_tip));
 
+        // We retry the deposit decisions because some signers' bitcoin nodes might have
+        // been running behind and ignored the previous messages.
+        let deposit_decisions_to_retry = db
+            .get_deposit_signer_decisions(
+                &chain_tip,
+                self.deposit_decisions_retry_window,
+                &signer_public_key,
+            )
+            .await?;
+
+        let _ = self
+            .handle_deposit_decisions_to_retry(deposit_decisions_to_retry, &chain_tip)
+            .await
+            .inspect_err(
+                |error| tracing::warn!(%error, "error handling deposit decisions to retry"),
+            );
+
         let deposit_requests = db
             .get_pending_deposit_requests(&chain_tip, self.context_window, &signer_public_key)
             .await?;
@@ -256,6 +276,24 @@ where
         Ok(())
     }
 
+    /// Send the given deposit decisions to the other signers for redundancy.
+    #[tracing::instrument(skip_all)]
+    pub async fn handle_deposit_decisions_to_retry(
+        &mut self,
+        decisions: Vec<model::DepositSigner>,
+        chain_tip: &BitcoinBlockHash,
+    ) -> Result<(), Error> {
+        for decision in decisions.into_iter().map(SignerDepositDecision::from) {
+            let _ = self
+                .send_message(decision, chain_tip)
+                .await
+                .inspect_err(|error| {
+                    tracing::warn!(%error, "error sending deposit decision to retry, skipping");
+                });
+        }
+        Ok(())
+    }
+
     #[tracing::instrument(skip_all)]
     async fn handle_pending_withdrawal_request(
         &mut self,
@@ -478,6 +516,7 @@ mod tests {
             num_deposit_requests_per_block: 5,
             num_withdraw_requests_per_block: 5,
             num_signers_per_request: 0,
+            consecutive_blocks: false,
         };
 
         let context = TestContext::builder()
@@ -488,6 +527,7 @@ mod tests {
         testing::request_decider::TestEnvironment {
             context,
             context_window: 6,
+            deposit_decisions_retry_window: 1,
             num_signers: 7,
             signing_threshold: 5,
             test_model_parameters,

signer/src/stacks/wallet.rs

@@ -677,6 +677,7 @@ mod tests {
             num_deposit_requests_per_block: 0,
             num_withdraw_requests_per_block: 0,
             num_signers_per_request: 0,
+            consecutive_blocks: false,
         };
         let test_data = TestData::generate(&mut rng, &[], &test_params);
         test_data.write_to(&db).await;

signer/src/storage/in_memory.rs

@@ -908,6 +908,43 @@ impl super::DbRead for SharedStore {
             .get(sighash)
             .map(|s| (s.will_sign, s.aggregate_key)))
     }
+
+    async fn get_deposit_signer_decisions(
+        &self,
+        chain_tip: &model::BitcoinBlockHash,
+        context_window: u16,
+        signer_public_key: &PublicKey,
+    ) -> Result<Vec<model::DepositSigner>, Error> {
+        let store = self.lock().await;
+        let deposit_requests = store.get_deposit_requests(chain_tip, context_window);
+        let voted: HashSet<(model::BitcoinTxId, u32)> = store
+            .signer_to_deposit_request
+            .get(signer_public_key)
+            .cloned()
+            .unwrap_or(Vec::new())
+            .into_iter()
+            .collect();
+
+        let result = deposit_requests
+            .into_iter()
+            .filter_map(|request| {
+                if !voted.contains(&(request.txid, request.output_index)) {
+                    return None;
+                }
+                store
+                    .deposit_request_to_signers
+                    .get(&(request.txid, request.output_index))
+                    .and_then(|signers| {
+                        signers
+                            .iter()
+                            .find(|signer| signer.signer_pub_key == *signer_public_key)
+                            .cloned()
+                    })
+            })
+            .collect();
+
+        Ok(result)
+    }
 }
 
 impl super::DbWrite for SharedStore {

signer/src/storage/mod.rs

@@ -124,6 +124,15 @@ pub trait DbRead {
         output_index: u32,
     ) -> impl Future<Output = Result<Vec<model::DepositSigner>, Error>> + Send;
 
+    /// Get all the deposit decisions for the given signer in the given window
+    /// of blocks.
+    fn get_deposit_signer_decisions(
+        &self,
+        chain_tip: &model::BitcoinBlockHash,
+        context_window: u16,
+        signer_public_key: &PublicKey,
+    ) -> impl Future<Output = Result<Vec<model::DepositSigner>, Error>> + Send;
+
     /// Returns whether the given `signer_public_key` can provide signature
     /// shares for the deposit transaction.
     ///

signer/src/storage/postgres.rs

@@ -488,7 +488,7 @@ impl PgStore {
                 JOIN sbtc_signer.bitcoin_transactions AS bt USING (txid)
                 WHERE prevout_type = 'signers_input'
             )
-            SELECT 
+            SELECT
                 bo.txid
               , bb.block_height
             FROM sbtc_signer.bitcoin_tx_outputs AS bo
@@ -1796,7 +1796,7 @@ impl super::DbRead for PgStore {
         sqlx::query_as::<_, model::SweptDepositRequest>(
             "
             WITH RECURSIVE bitcoin_blockchain AS (
-                SELECT 
+                SELECT
                     block_hash
                   , block_height
                 FROM bitcoin_blockchain_of($1, $2)
@@ -1810,9 +1810,9 @@ impl super::DbRead for PgStore {
                 JOIN bitcoin_blockchain as bb
                     ON bb.block_hash = stacks_blocks.bitcoin_anchor
                 WHERE stacks_blocks.block_hash = $3
-        
+
                 UNION ALL
-        
+
                 SELECT
                     parent.block_hash
                   , parent.block_height
@@ -1841,7 +1841,7 @@ impl super::DbRead for PgStore {
             LEFT JOIN completed_deposit_events AS cde
               ON cde.bitcoin_txid = deposit_req.txid
              AND cde.output_index = deposit_req.output_index
-            LEFT JOIN stacks_blockchain AS sb 
+            LEFT JOIN stacks_blockchain AS sb
               ON sb.block_hash = cde.block_hash
             GROUP BY
                 bc_trx.txid
@@ -1921,6 +1921,40 @@ impl super::DbRead for PgStore {
         .await
         .map_err(Error::SqlxQuery)
     }
+
+    async fn get_deposit_signer_decisions(
+        &self,
+        chain_tip: &model::BitcoinBlockHash,
+        context_window: u16,
+        signer_public_key: &PublicKey,
+    ) -> Result<Vec<model::DepositSigner>, Error> {
+        sqlx::query_as::<_, model::DepositSigner>(
+            r#"
+            WITH target_block AS (
+                SELECT blocks.block_hash, blocks.created_at
+                FROM sbtc_signer.bitcoin_blockchain_of($1, $2) chain
+                JOIN sbtc_signer.bitcoin_blocks blocks USING (block_hash)
+                ORDER BY chain.block_height ASC
+                LIMIT 1
+            )
+            SELECT
+                ds.txid,
+                ds.output_index,
+                ds.signer_pub_key,
+                ds.can_sign,
+                ds.can_accept
+            FROM sbtc_signer.deposit_signers ds
+            WHERE ds.signer_pub_key = $3
+              AND ds.created_at >= (SELECT created_at FROM target_block)
+            "#,
+        )
+        .bind(chain_tip)
+        .bind(i32::from(context_window))
+        .bind(signer_public_key)
+        .fetch_all(&self.0)
+        .await
+        .map_err(Error::SqlxQuery)
+    }
 }
 
 impl super::DbWrite for PgStore {

signer/src/testing/request_decider.rs

@@ -42,6 +42,7 @@ impl<C: Context + 'static> RequestDeciderEventLoopHarness<C> {
         context: C,
         network: SignerNetwork,
         context_window: u16,
+        deposit_decisions_retry_window: u16,
         signer_private_key: PrivateKey,
     ) -> Self {
         Self {
@@ -51,6 +52,7 @@ impl<C: Context + 'static> RequestDeciderEventLoopHarness<C> {
                 blocklist_checker: Some(()),
                 signer_private_key,
                 context_window,
+                deposit_decisions_retry_window,
             },
             context,
         }
@@ -122,6 +124,8 @@ pub struct TestEnvironment<C> {
     pub context: C,
     /// Bitcoin context window
     pub context_window: u16,
+    /// Deposit decisions retry window
+    pub deposit_decisions_retry_window: u16,
     /// Num signers
     pub num_signers: usize,
     /// Signing threshold
@@ -156,6 +160,7 @@ where
             self.context.clone(),
             signer_network,
             self.context_window,
+            self.deposit_decisions_retry_window,
             coordinator_signer_info.signer_private_key,
         );
 
@@ -238,6 +243,7 @@ where
             self.context.clone(),
             signer_network,
             self.context_window,
+            self.deposit_decisions_retry_window,
             coordinator_signer_info.signer_private_key,
         );
 
@@ -315,6 +321,7 @@ where
                     ctx,
                     net,
                     self.context_window,
+                    self.deposit_decisions_retry_window,
                     signer_info.signer_private_key,
                 );