Do not pause block broadcast until certain all signers have processed the last block pushed message

Signed-off-by: Jacinta Ferrant <jacinta.ferrant@gmail.com>

Author: jferrant

stacks-signer/src/lib.rs

@@ -53,6 +53,7 @@ use chainstate::SortitionsView;
 use config::GlobalConfig;
 use libsigner::{SignerEvent, SignerEventReceiver, SignerEventTrait, VERSION_STRING};
 use runloop::SignerResult;
+use signerdb::BlockInfo;
 use stacks_common::{info, warn};
 use v0::signer_state::LocalStateMachine;
 
@@ -81,6 +82,8 @@ pub trait Signer<T: SignerEventTrait>: Debug + Display {
     fn get_local_state_machine(&self) -> &LocalStateMachine;
     /// Get the number of pending block proposals
     fn get_pending_proposals_count(&self) -> u64;
+    /// Get canonical block according to this signer's db
+    fn get_canonical_tip(&self) -> Option<BlockInfo>;
 }
 
 /// A wrapper around the running signer type for the signer

stacks-signer/src/runloop.rs

@@ -25,6 +25,7 @@ use stacks_common::{debug, error, info, warn};
 use crate::chainstate::SortitionsView;
 use crate::client::{retry_with_exponential_backoff, ClientError, StacksClient};
 use crate::config::{GlobalConfig, SignerConfig, SignerConfigMode};
+use crate::signerdb::BlockInfo;
 use crate::v0::signer_state::LocalStateMachine;
 #[cfg(any(test, feature = "testing"))]
 use crate::v0::tests::TEST_SKIP_SIGNER_CLEANUP;
@@ -59,6 +60,9 @@ pub struct StateInfo {
     pub signer_state_machines: Vec<(u64, Option<LocalStateMachine>)>,
     /// The number of pending block proposals for this signer
     pub pending_proposals_count: u64,
+    /// The canonical tip block info according to the running signers
+    /// as a pair of (reward-cycle, block-info)
+    pub signer_canonical_tips: Vec<(u64, Option<BlockInfo>)>,
 }
 
 /// The signer result that can be sent across threads
@@ -544,6 +548,16 @@ impl<Signer: SignerTrait<T>, T: StacksMessageCodec + Clone + Send + Debug>
                         }
                     })
                     .unwrap_or(0),
+                signer_canonical_tips: self
+                    .stacks_signers
+                    .iter()
+                    .map(|(reward_cycle, signer)| {
+                        let ConfiguredSigner::RegisteredSigner(ref signer) = signer else {
+                            return (*reward_cycle, None);
+                        };
+                        (*reward_cycle, signer.get_canonical_tip())
+                    })
+                    .collect(),
             };
             info!("Signer status check requested: {state_info:?}");
 

stacks-signer/src/signerdb.rs

@@ -98,7 +98,7 @@ impl FromRow<BurnBlockInfo> for BurnBlockInfo {
     }
 }
 
-#[derive(Serialize, Deserialize, Debug, PartialEq, Default)]
+#[derive(Serialize, Deserialize, Debug, PartialEq, Default, Clone)]
 /// Store extra version-specific info in `BlockInfo`
 pub enum ExtraBlockInfo {
     #[default]
@@ -167,7 +167,7 @@ impl TryFrom<&str> for BlockState {
 }
 
 /// Additional Info about a proposed block
-#[derive(Serialize, Deserialize, Debug, PartialEq)]
+#[derive(Serialize, Deserialize, Debug, PartialEq, Clone)]
 pub struct BlockInfo {
     /// The block we are considering
     pub block: NakamotoBlock,

stacks-signer/src/v0/signer.rs

@@ -336,6 +336,14 @@ impl SignerTrait<SignerMessage> for Signer {
             .map(|results| u64::try_from(results.len()).unwrap())
             .unwrap_or(0)
     }
+
+    fn get_canonical_tip(&self) -> Option<BlockInfo> {
+        self.signer_db
+            .get_canonical_tip()
+            .inspect_err(|e| error!("{self}: Failed to check for canonical tip: {e:?}"))
+            .ok()
+            .flatten()
+    }
 }
 
 impl Signer {

testnet/stacks-node/src/tests/signer/v0.rs

@@ -7344,6 +7344,36 @@ fn miner_recovers_when_broadcast_block_delay_across_tenures_occurs() {
     // Clear the stackerdb chunks
     info!("Forcing miner to ignore block responses for block N+1");
     TEST_IGNORE_SIGNERS.set(true);
+
+    info!("------------------------- Wait for All Signers to Update their Canonical Tip to Block N -------------------------");
+    let current_rc = signer_test.get_current_reward_cycle();
+    let expected_tip = block_n.header.signer_signature_hash();
+    wait_for(30, || {
+        let states = signer_test.get_all_states();
+        let canonical_tips = states
+            .iter()
+            .filter(|state| {
+                state
+                    .signer_canonical_tips
+                    .iter()
+                    .find_map(|(rc, block_info_opt)| {
+                        if current_rc % 2 == *rc {
+                            block_info_opt.as_ref()
+                        } else {
+                            None
+                        }
+                    })
+                    .map(|block_info| {
+                        block_info.block.header.signer_signature_hash() == expected_tip
+                    })
+                    .unwrap_or(false)
+            })
+            .count();
+
+        Ok(canonical_tips == num_signers)
+    })
+    .expect("Timed out waiting for all signers to update their global state to Block N");
+
     info!("Delaying signer block N+1 broadcasting to the miner");
     TEST_PAUSE_BLOCK_BROADCAST.set(true);
     test_observer::clear();