Add miner_recovers_when_broadcast_block_delay_across_tenures_occurs

Signed-off-by: Jacinta Ferrant <jacinta@trustmachines.co>

.github/workflows/bitcoin-tests.yml

@@ -103,6 +103,7 @@ jobs:
           - tests::signer::v0::locally_accepted_blocks_overriden_by_global_rejection
           - tests::signer::v0::locally_rejected_blocks_overriden_by_global_acceptance
           - tests::signer::v0::reorg_locally_accepted_blocks_across_tenures_succeeds
+          - tests::signer::v0::miner_recovers_when_broadcast_block_delay_across_tenures_occurs
           - tests::nakamoto_integrations::stack_stx_burn_op_integration_test
           - tests::nakamoto_integrations::check_block_heights
           - tests::nakamoto_integrations::clarity_burn_state

stacks-signer/src/client/mod.rs

@@ -571,7 +571,6 @@ pub(crate) mod tests {
             db_path: config.db_path.clone(),
             first_proposal_burn_block_timing: config.first_proposal_burn_block_timing,
             block_proposal_timeout: config.block_proposal_timeout,
-            broadcast_signed_blocks: true,
         }
     }
 

stacks-signer/src/client/stacks_client.rs

@@ -706,17 +706,23 @@ impl StacksClient {
     /// Returns `true` if the block was accepted or `false` if the block
     ///   was rejected.
     pub fn post_block(&self, block: &NakamotoBlock) -> Result<bool, ClientError> {
-        let response = self
-            .stacks_node_client
-            .post(format!(
-                "{}{}?broadcast=1",
-                self.http_origin,
-                postblock_v3::PATH
-            ))
-            .header("Content-Type", "application/octet-stream")
-            .header(AUTHORIZATION, self.auth_password.clone())
-            .body(block.serialize_to_vec())
-            .send()?;
+        let send_request = || {
+            self.stacks_node_client
+                .post(format!(
+                    "{}{}?broadcast=1",
+                    self.http_origin,
+                    postblock_v3::PATH
+                ))
+                .header("Content-Type", "application/octet-stream")
+                .header(AUTHORIZATION, self.auth_password.clone())
+                .body(block.serialize_to_vec())
+                .send()
+                .map_err(|e| {
+                    debug!("Failed to submit block to the Stacks node: {e:?}");
+                    backoff::Error::transient(e)
+                })
+        };
+        let response = retry_with_exponential_backoff(send_request)?;
         if !response.status().is_success() {
             return Err(ClientError::RequestFailure(response.status()));
         }

stacks-signer/src/config.rs

@@ -157,8 +157,6 @@ pub struct SignerConfig {
     pub first_proposal_burn_block_timing: Duration,
     /// How much time to wait for a miner to propose a block following a sortition
     pub block_proposal_timeout: Duration,
-    /// Broadcast a block to the node if we gather enough signatures from other signers
-    pub broadcast_signed_blocks: bool,
 }
 
 /// The parsed configuration for the signer
@@ -203,8 +201,6 @@ pub struct GlobalConfig {
     pub first_proposal_burn_block_timing: Duration,
     /// How much time to wait for a miner to propose a block following a sortition
     pub block_proposal_timeout: Duration,
-    /// Broadcast a block to the node if we gather enough signatures from other signers
-    pub broadcast_signed_blocks: bool,
 }
 
 /// Internal struct for loading up the config file
@@ -361,7 +357,6 @@ impl TryFrom<RawConfigFile> for GlobalConfig {
             metrics_endpoint,
             first_proposal_burn_block_timing,
             block_proposal_timeout,
-            broadcast_signed_blocks: true,
         })
     }
 }

stacks-signer/src/runloop.rs

@@ -335,7 +335,6 @@ impl<Signer: SignerTrait<T>, T: StacksMessageCodec + Clone + Send + Debug> RunLo
             max_tx_fee_ustx: self.config.max_tx_fee_ustx,
             db_path: self.config.db_path.clone(),
             block_proposal_timeout: self.config.block_proposal_timeout,
-            broadcast_signed_blocks: self.config.broadcast_signed_blocks,
         }))
     }
 

stacks-signer/src/signerdb.rs

@@ -225,16 +225,20 @@ impl BlockInfo {
         block_info
     }
 
-    /// Mark this block as locally accepted, valid, signed over, and records a timestamp in the block info if it wasn't
+    /// Mark this block as locally accepted, valid, signed over, and records either the self or group signed timestamp in the block info if it wasn't
     ///  already set.
-    pub fn mark_locally_accepted(&mut self) -> Result<(), String> {
+    pub fn mark_locally_accepted(&mut self, group_signed: bool) -> Result<(), String> {
         self.valid = Some(true);
         self.signed_over = true;
-        self.signed_self.get_or_insert(get_epoch_time_secs());
+        if group_signed {
+            self.signed_group.get_or_insert(get_epoch_time_secs());
+        } else {
+            self.signed_self.get_or_insert(get_epoch_time_secs());
+        }
         self.move_to(BlockState::LocallyAccepted)
     }
 
-    /// Mark this block as globally accepted, valid, signed over, and records a timestamp in the block info if it wasn't
+    /// Mark this block as valid, signed over, and records a group timestamp in the block info if it wasn't
     ///  already set.
     pub fn mark_globally_accepted(&mut self) -> Result<(), String> {
         self.valid = Some(true);
@@ -785,15 +789,20 @@ impl SignerDb {
         query_rows(&self.db, qry, args)
     }
 
-    /// Mark a block as having been broadcasted
+    /// Mark a block as having been broadcasted and therefore GloballyAccepted
     pub fn set_block_broadcasted(
         &self,
         reward_cycle: u64,
         block_sighash: &Sha512Trunc256Sum,
         ts: u64,
     ) -> Result<(), DBError> {
-        let qry = "UPDATE blocks SET broadcasted = ?1 WHERE reward_cycle = ?2 AND signer_signature_hash = ?3";
-        let args = params![u64_to_sql(ts)?, u64_to_sql(reward_cycle)?, block_sighash];
+        let qry = "UPDATE blocks SET broadcasted = ?1, block_info = json_set(block_info, '$.state', ?2) WHERE reward_cycle = ?3 AND signer_signature_hash = ?4";
+        let args = params![
+            u64_to_sql(ts)?,
+            BlockState::GloballyAccepted.to_string(),
+            u64_to_sql(reward_cycle)?,
+            block_sighash
+        ];
 
         debug!("Marking block {} as broadcasted at {}", block_sighash, ts);
         self.db.execute(qry, args)?;
@@ -1015,7 +1024,7 @@ mod tests {
             .is_none());
 
         block_info
-            .mark_locally_accepted()
+            .mark_locally_accepted(false)
             .expect("Failed to mark block as locally accepted");
         db.insert_block(&block_info).unwrap();
 
@@ -1175,12 +1184,32 @@ mod tests {
             )
             .unwrap()
             .is_none());
+        assert_eq!(
+            db.block_lookup(
+                block_info_1.reward_cycle,
+                &block_info_1.signer_signature_hash()
+            )
+            .expect("Unable to get block from db")
+            .expect("Unable to get block from db")
+            .state,
+            BlockState::Unprocessed
+        );
         db.set_block_broadcasted(
             block_info_1.reward_cycle,
             &block_info_1.signer_signature_hash(),
             12345,
         )
         .unwrap();
+        assert_eq!(
+            db.block_lookup(
+                block_info_1.reward_cycle,
+                &block_info_1.signer_signature_hash()
+            )
+            .expect("Unable to get block from db")
+            .expect("Unable to get block from db")
+            .state,
+            BlockState::GloballyAccepted
+        );
         db.insert_block(&block_info_1)
             .expect("Unable to insert block into db a second time");
 

stacks-signer/src/tests/chainstate.rs

@@ -243,7 +243,7 @@ fn reorg_timing_testing(
     };
     let mut header_clone = block_proposal_1.block.header.clone();
     let mut block_info_1 = BlockInfo::from(block_proposal_1);
-    block_info_1.mark_locally_accepted().unwrap();
+    block_info_1.mark_locally_accepted(false).unwrap();
     signer_db.insert_block(&block_info_1).unwrap();
 
     let sortition_time = SystemTime::UNIX_EPOCH

stacks-signer/src/v0/signer.rs

@@ -55,6 +55,14 @@ pub static TEST_IGNORE_ALL_BLOCK_PROPOSALS: std::sync::Mutex<
     Option<Vec<stacks_common::types::chainstate::StacksPublicKey>>,
 > = std::sync::Mutex::new(None);
 
+#[cfg(any(test, feature = "testing"))]
+/// Pause the block broadcast
+pub static TEST_PAUSE_BLOCK_BROADCAST: std::sync::Mutex<Option<bool>> = std::sync::Mutex::new(None);
+
+#[cfg(any(test, feature = "testing"))]
+/// Skip broadcasting the block to the network
+pub static TEST_SKIP_BLOCK_BROADCAST: std::sync::Mutex<Option<bool>> = std::sync::Mutex::new(None);
+
 /// The stacks signer registered for the reward cycle
 #[derive(Debug)]
 pub struct Signer {
@@ -78,8 +86,6 @@ pub struct Signer {
     pub signer_db: SignerDb,
     /// Configuration for proposal evaluation
     pub proposal_config: ProposalEvalConfig,
-    /// Whether or not to broadcast signed blocks if we gather all signatures
-    pub broadcast_signed_blocks: bool,
 }
 
 impl std::fmt::Display for Signer {
@@ -179,13 +185,23 @@ impl SignerTrait<SignerMessage> for Signer {
                             );
                         }
                         SignerMessage::BlockPushed(b) => {
-                            let block_push_result = stacks_client.post_block(b);
+                            // This will infinitely loop until the block is acknowledged by the node
                             info!(
                                 "{self}: Got block pushed message";
                                 "block_id" => %b.block_id(),
                                 "signer_sighash" => %b.header.signer_signature_hash(),
-                                "push_result" => ?block_push_result,
                             );
+                            loop {
+                                match stacks_client.post_block(b) {
+                                    Ok(block_push_result) => {
+                                        debug!("{self}: Block pushed to stacks node: {block_push_result:?}");
+                                        break;
+                                    }
+                                    Err(e) => {
+                                        warn!("{self}: Failed to push block to stacks node: {e}. Retrying...");
+                                    }
+                                };
+                            }
                         }
                         SignerMessage::MockProposal(mock_proposal) => {
                             let epoch = match stacks_client.get_node_epoch() {
@@ -306,7 +322,6 @@ impl From<SignerConfig> for Signer {
             reward_cycle: signer_config.reward_cycle,
             signer_db,
             proposal_config,
-            broadcast_signed_blocks: signer_config.broadcast_signed_blocks,
         }
     }
 }
@@ -555,7 +570,7 @@ impl Signer {
                 return None;
             }
         };
-        if let Err(e) = block_info.mark_locally_accepted() {
+        if let Err(e) = block_info.mark_locally_accepted(false) {
             warn!("{self}: Failed to mark block as locally accepted: {e:?}",);
             return None;
         }
@@ -876,10 +891,11 @@ impl Signer {
             warn!("{self}: No such block {block_hash}");
             return;
         };
-        // move block to globally accepted state. If this is not possible, we have a bug in our block handling logic.
-        if let Err(e) = block_info.mark_globally_accepted() {
+        // move block to LOCALLY accepted state.
+        // We only mark this GLOBALLY accepted if we manage to broadcast it...
+        if let Err(e) = block_info.mark_locally_accepted(true) {
             // Do not abort as we should still try to store the block signature threshold
-            warn!("{self}: Failed to mark block as globally accepted: {e:?}");
+            warn!("{self}: Failed to mark block as locally accepted: {e:?}");
         }
         let _ = self.signer_db.insert_block(&block_info).map_err(|e| {
             warn!(
@@ -888,17 +904,24 @@ impl Signer {
             );
             e
         });
-
-        if self.broadcast_signed_blocks {
-            self.broadcast_signed_block(stacks_client, block_info.block, &addrs_to_sigs);
-        } else {
-            debug!(
-                "{self}: Not broadcasting signed block {block_hash} since broadcast_signed_blocks is false";
-                "stacks_block_id" => %block_info.block.block_id(),
-                "parent_block_id" => %block_info.block.header.parent_block_id,
-                "burnchain_consensus_hash" => %block_info.block.header.consensus_hash
-            );
+        #[cfg(any(test, feature = "testing"))]
+        {
+            if *TEST_PAUSE_BLOCK_BROADCAST.lock().unwrap() == Some(true) {
+                // Do an extra check just so we don't log EVERY time.
+                warn!("Block broadcast is stalled due to testing directive.";
+                    "block_id" => %block_info.block.block_id(),
+                    "height" => block_info.block.header.chain_length,
+                );
+                while *TEST_PAUSE_BLOCK_BROADCAST.lock().unwrap() == Some(true) {
+                    std::thread::sleep(std::time::Duration::from_millis(10));
+                }
+                info!("Block validation is no longer stalled due to testing directive.";
+                    "block_id" => %block_info.block.block_id(),
+                    "height" => block_info.block.header.chain_length,
+                );
+            }
         }
+        self.broadcast_signed_block(stacks_client, block_info.block, &addrs_to_sigs);
     }
 
     fn broadcast_signed_block(
@@ -918,11 +941,30 @@ impl Signer {
         block.header.signer_signature_hash();
         block.header.signer_signature = signatures;
 
+        #[cfg(any(test, feature = "testing"))]
+        {
+            if *TEST_SKIP_BLOCK_BROADCAST.lock().unwrap() == Some(true) {
+                warn!(
+                    "{self}: Skipping block broadcast due to testing directive";
+                    "block_id" => %block.block_id(),
+                    "height" => block.header.chain_length,
+                    "consensus_hash" => %block.header.consensus_hash
+                );
+
+                if let Err(e) = self.signer_db.set_block_broadcasted(
+                    self.reward_cycle,
+                    &block_hash,
+                    get_epoch_time_secs(),
+                ) {
+                    warn!("{self}: Failed to set block broadcasted for {block_hash}: {e:?}");
+                }
+                return;
+            }
+        }
         debug!(
             "{self}: Broadcasting Stacks block {} to node",
             &block.block_id()
         );
-
         if let Err(e) = stacks_client.post_block(&block) {
             warn!(
                 "{self}: Failed to post block {block_hash}: {e:?}";