Merge pull request #5877 from stacks-network/feat/resend-proposal

feat: on timeout, re-propose the same block

Author: obycode

CHANGELOG.md

@@ -7,13 +7,17 @@ and this project adheres to the versioning scheme outlined in the [README.md](RE
 
 ## [Unreleased]
 
+### Changed
+
+- When a miner times out waiting for signatures, it will re-propose the same block instead of building a new block ([#5877](https://github.com/stacks-network/stacks-core/pull/5877))
+
 ## [3.1.0.0.7]
 
-## Added
+### Added
 
 - Add `disable_retries` mode for events_observer disabling automatic retry on error
 
-## Changed
+### Changed
 
 - Implement faster cost tracker for default cost functions in Clarity
 - By default, miners will wait for a new tenure to start for a configurable amount of time after receiving a burn block before
@@ -23,7 +27,7 @@ and this project adheres to the versioning scheme outlined in the [README.md](RE
 
 ## [3.1.0.0.6]
 
-## Added
+### Added
 
 - The `BlockProposal` StackerDB message serialization struct now includes a `server_version` string, which represents the version of the node that the miner is using. ([#5803](https://github.com/stacks-network/stacks-core/pull/5803))
 - Add `vrf_seed` to the `/v3/sortitions` rpc endpoint

stacks-signer/src/chainstate.rs

@@ -402,7 +402,7 @@ impl SortitionsView {
                 false,
             );
             let epoch_time = get_epoch_time_secs();
-            let enough_time_passed = epoch_time > extend_timestamp;
+            let enough_time_passed = epoch_time >= extend_timestamp;
             if !changed_burn_view && !enough_time_passed {
                 warn!(
                     "Miner block proposal contains a tenure extend, but the burnchain view has not changed and enough time has not passed to refresh the block limit. Considering proposal invalid.";

stacks-signer/src/v0/signer.rs

@@ -15,6 +15,8 @@
 use std::collections::HashMap;
 use std::fmt::Debug;
 use std::sync::mpsc::Sender;
+#[cfg(any(test, feature = "testing"))]
+use std::sync::LazyLock;
 use std::time::{Duration, Instant};
 
 use blockstack_lib::chainstate::nakamoto::{NakamotoBlock, NakamotoBlockHeader};
@@ -23,9 +25,15 @@ use blockstack_lib::net::api::postblock_proposal::{
 };
 use blockstack_lib::util_lib::db::Error as DBError;
 use clarity::types::chainstate::StacksPrivateKey;
+#[cfg(any(test, feature = "testing"))]
+use clarity::types::chainstate::StacksPublicKey;
 use clarity::types::{PrivateKey, StacksEpochId};
 use clarity::util::hash::{MerkleHashFunc, Sha512Trunc256Sum};
 use clarity::util::secp256k1::Secp256k1PublicKey;
+#[cfg(any(test, feature = "testing"))]
+use clarity::util::sleep_ms;
+#[cfg(any(test, feature = "testing"))]
+use clarity::util::tests::TestFlag;
 use libsigner::v0::messages::{
     BlockAccepted, BlockRejection, BlockResponse, MessageSlotID, MockProposal, MockSignature,
     RejectReason, RejectReasonPrefix, SignerMessage,
@@ -44,6 +52,12 @@ use crate::runloop::SignerResult;
 use crate::signerdb::{BlockInfo, BlockState, SignerDb};
 use crate::Signer as SignerTrait;
 
+/// A global variable that can be used to make signers repeat their proposal
+/// response if their public key is in the provided list
+#[cfg(any(test, feature = "testing"))]
+pub static TEST_REPEAT_PROPOSAL_RESPONSE: LazyLock<TestFlag<Vec<StacksPublicKey>>> =
+    LazyLock::new(TestFlag::default);
+
 /// Signer running mode (whether dry-run or real)
 #[derive(Debug)]
 pub enum SignerMode {
@@ -464,6 +478,45 @@ impl Signer {
         }
     }
 
+    /// The actual `send_block_response` implementation. Declared so that we do
+    /// not need to duplicate in testing.
+    fn impl_send_block_response(&mut self, block_response: BlockResponse) {
+        let res = self
+            .stackerdb
+            .send_message_with_retry::<SignerMessage>(block_response.clone().into());
+        match res {
+            Err(e) => warn!("{self}: Failed to send block rejection to stacker-db: {e:?}"),
+            Ok(ack) if !ack.accepted => warn!(
+                "{self}: Block rejection not accepted by stacker-db: {:?}",
+                ack.reason
+            ),
+            Ok(_) => debug!("{self}: Block rejection accepted by stacker-db"),
+        }
+    }
+
+    #[cfg(any(test, feature = "testing"))]
+    fn send_block_response(&mut self, block_response: BlockResponse) {
+        const NUM_REPEATS: usize = 1;
+        let mut count = 0;
+        let public_keys = TEST_REPEAT_PROPOSAL_RESPONSE.get();
+        if !public_keys.contains(
+            &stacks_common::types::chainstate::StacksPublicKey::from_private(&self.private_key),
+        ) {
+            count = NUM_REPEATS;
+        }
+        while count <= NUM_REPEATS {
+            self.impl_send_block_response(block_response.clone());
+
+            count += 1;
+            sleep_ms(1000);
+        }
+    }
+
+    #[cfg(not(any(test, feature = "testing")))]
+    fn send_block_response(&mut self, block_response: BlockResponse) {
+        self.impl_send_block_response(block_response)
+    }
+
     /// Handle block proposal messages submitted to signers stackerdb
     fn handle_block_proposal(
         &mut self,
@@ -575,18 +628,7 @@ impl Signer {
         if let Some(block_response) = block_response {
             // We know proposal is invalid. Send rejection message, do not do further validation and do not store it.
             debug!("{self}: Broadcasting a block response to stacks node: {block_response:?}");
-            let res = self
-                .stackerdb
-                .send_message_with_retry::<SignerMessage>(block_response.into());
-
-            match res {
-                Err(e) => warn!("{self}: Failed to send block rejection to stacker-db: {e:?}"),
-                Ok(ack) if !ack.accepted => warn!(
-                    "{self}: Block rejection not accepted by stacker-db: {:?}",
-                    ack.reason
-                ),
-                Ok(_) => debug!("{self}: Block rejection accepted by stacker-db"),
-            }
+            self.send_block_response(block_response);
         } else {
             // Just in case check if the last block validation submission timed out.
             self.check_submitted_block_proposal();

testnet/stacks-node/src/nakamoto_node.rs

@@ -143,6 +143,8 @@ pub enum Error {
     /// NetError wrapper
     #[error("NetError: {0}")]
     NetError(#[from] NetError),
+    #[error("Timed out waiting for signatures")]
+    SignatureTimeout,
 }
 
 impl StacksNode {

testnet/stacks-node/src/nakamoto_node/signer_coordinator.rs

@@ -254,46 +254,54 @@ impl SignerCoordinator {
         };
 
         let block_proposal_message = SignerMessageV0::BlockProposal(block_proposal);
-        debug!("Sending block proposal message to signers";
-            "signer_signature_hash" => %block.header.signer_signature_hash(),
-        );
-        Self::send_miners_message::<SignerMessageV0>(
-            &self.message_key,
-            sortdb,
-            election_sortition,
-            stackerdbs,
-            block_proposal_message,
-            MinerSlotID::BlockProposal,
-            self.is_mainnet,
-            &mut self.miners_session,
-            &election_sortition.consensus_hash,
-        )?;
-        counters.bump_naka_proposed_blocks();
 
-        #[cfg(test)]
-        {
-            info!(
-                "SignerCoordinator: sent block proposal to .miners, waiting for test signing channel"
+        loop {
+            debug!("Sending block proposal message to signers";
+                "signer_signature_hash" => %block.header.signer_signature_hash(),
             );
-            // In test mode, short-circuit waiting for the signers if the TEST_SIGNING
-            //  channel has been created. This allows integration tests for the stacks-node
-            //  independent of the stacks-signer.
-            if let Some(signatures) =
-                crate::tests::nakamoto_integrations::TestSigningChannel::get_signature()
+            Self::send_miners_message::<SignerMessageV0>(
+                &self.message_key,
+                sortdb,
+                election_sortition,
+                stackerdbs,
+                block_proposal_message.clone(),
+                MinerSlotID::BlockProposal,
+                self.is_mainnet,
+                &mut self.miners_session,
+                &election_sortition.consensus_hash,
+            )?;
+            counters.bump_naka_proposed_blocks();
+
+            #[cfg(test)]
             {
-                debug!("Short-circuiting waiting for signers, using test signature");
-                return Ok(signatures);
+                info!(
+                "SignerCoordinator: sent block proposal to .miners, waiting for test signing channel"
+            );
+                // In test mode, short-circuit waiting for the signers if the TEST_SIGNING
+                //  channel has been created. This allows integration tests for the stacks-node
+                //  independent of the stacks-signer.
+                if let Some(signatures) =
+                    crate::tests::nakamoto_integrations::TestSigningChannel::get_signature()
+                {
+                    debug!("Short-circuiting waiting for signers, using test signature");
+                    return Ok(signatures);
+                }
             }
-        }
 
-        self.get_block_status(
-            &block.header.signer_signature_hash(),
-            &block.block_id(),
-            block.header.parent_block_id,
-            chain_state,
-            sortdb,
-            counters,
-        )
+            let res = self.get_block_status(
+                &block.header.signer_signature_hash(),
+                &block.block_id(),
+                block.header.parent_block_id,
+                chain_state,
+                sortdb,
+                counters,
+            );
+
+            match res {
+                Err(NakamotoNodeError::SignatureTimeout) => continue,
+                _ => return res,
+            }
+        }
     }
 
     /// Get the block status for a given block hash.
@@ -340,7 +348,7 @@ impl SignerCoordinator {
                     if rejections_timer.elapsed() > *rejections_timeout {
                         return false;
                     }
-                    // number or rejections changed?
+                    // number of rejections changed?
                     if status.total_weight_rejected != rejections {
                         return false;
                     }
@@ -353,7 +361,7 @@ impl SignerCoordinator {
                     // If we just received a timeout, we should check if the burnchain
                     // tip has changed or if we received this signed block already in
                     // the staging db.
-                    debug!("SignerCoordinator: Timeout waiting for block signatures");
+                    debug!("SignerCoordinator: Intermediate timeout waiting for block status");
 
                     // Look in the nakamoto staging db -- a block can only get stored there
                     // if it has enough signing weight to clear the threshold.
@@ -380,15 +388,17 @@ impl SignerCoordinator {
                     }
 
                     if rejections_timer.elapsed() > *rejections_timeout {
-                        warn!("Timed out while waiting for responses from signers";
-                                  "elapsed" => rejections_timer.elapsed().as_secs(),
-                                  "rejections_timeout" => rejections_timeout.as_secs(),
-                                  "rejections" => rejections,
-                                  "rejections_threshold" => self.total_weight.saturating_sub(self.weight_threshold)
+                        warn!("Timed out while waiting for responses from signers, resending proposal";
+                            "elapsed" => rejections_timer.elapsed().as_secs(),
+                            "rejections_timeout" => rejections_timeout.as_secs(),
+                            "rejections" => rejections,
+                            "rejections_threshold" => self.total_weight.saturating_sub(self.weight_threshold)
                         );
-                        return Err(NakamotoNodeError::SigningCoordinatorFailure(
-                            "Timed out while waiting for signatures".into(),
-                        ));
+
+                        // Reset the rejections in the stackerdb listener
+                        self.stackerdb_comms.reset_rejections(block_signer_sighash);
+
+                        return Err(NakamotoNodeError::SignatureTimeout);
                     }
 
                     // Check if a new Stacks block has arrived in the parent tenure
@@ -399,7 +409,7 @@ impl SignerCoordinator {
                         )?
                         .ok_or(NakamotoNodeError::UnexpectedChainState)?;
                     if highest_in_tenure.index_block_hash() != parent_block_id {
-                        debug!("SignCoordinator: Exiting due to new stacks tip");
+                        info!("SignCoordinator: Exiting due to new stacks tip");
                         return Err(NakamotoNodeError::StacksTipChanged);
                     }
 
@@ -448,14 +458,16 @@ impl SignerCoordinator {
                 return Ok(block_status.gathered_signatures.values().cloned().collect());
             } else if rejections_timer.elapsed() > *rejections_timeout {
                 warn!("Timed out while waiting for responses from signers";
-                          "elapsed" => rejections_timer.elapsed().as_secs(),
-                          "rejections_timeout" => rejections_timeout.as_secs(),
-                          "rejections" => rejections,
-                          "rejections_threshold" => self.total_weight.saturating_sub(self.weight_threshold)
+                    "elapsed" => rejections_timer.elapsed().as_secs(),
+                    "rejections_timeout" => rejections_timeout.as_secs(),
+                    "rejections" => rejections,
+                    "rejections_threshold" => self.total_weight.saturating_sub(self.weight_threshold)
                 );
-                return Err(NakamotoNodeError::SigningCoordinatorFailure(
-                    "Timed out while waiting for signatures".into(),
-                ));
+
+                // Reset the rejections in the stackerdb listener
+                self.stackerdb_comms.reset_rejections(block_signer_sighash);
+
+                return Err(NakamotoNodeError::SignatureTimeout);
             } else {
                 continue;
             }

testnet/stacks-node/src/nakamoto_node/stackerdb_listener.rs

@@ -52,9 +52,13 @@ pub static EVENT_RECEIVER_POLL: Duration = Duration::from_millis(500);
 
 #[derive(Debug, Clone)]
 pub struct BlockStatus {
-    pub responded_signers: HashSet<StacksPublicKey>,
+    /// Set of the slot ids of signers who have responded
+    pub responded_signers: HashSet<u32>,
+    /// Map of the slot id of signers who have signed the block and their signature
     pub gathered_signatures: BTreeMap<u32, MessageSignature>,
+    /// Total weight of signers who have signed the block
     pub total_weight_approved: u32,
+    /// Total weight of signers who have rejected the block
     pub total_weight_rejected: u32,
 }
 
@@ -342,7 +346,7 @@ impl StackerDBListener {
                             "server_version" => metadata.server_version,
                         );
                         block.gathered_signatures.insert(slot_id, signature);
-                        block.responded_signers.insert(signer_pubkey);
+                        block.responded_signers.insert(slot_id);
 
                         if block.total_weight_approved >= self.weight_threshold {
                             // Signal to anyone waiting on this block that we have enough signatures
@@ -384,11 +388,13 @@ impl StackerDBListener {
                                 continue;
                             }
                         };
-                        block.responded_signers.insert(rejected_pubkey);
-                        block.total_weight_rejected = block
-                            .total_weight_rejected
-                            .checked_add(signer_entry.weight)
-                            .expect("FATAL: total weight rejected exceeds u32::MAX");
+
+                        if block.responded_signers.insert(slot_id) {
+                            block.total_weight_rejected = block
+                                .total_weight_rejected
+                                .checked_add(signer_entry.weight)
+                                .expect("FATAL: total weight rejected exceeds u32::MAX");
+                        }
 
                         info!("StackerDBListener: Signer rejected block";
                             "block_signer_sighash" => %rejected_data.signer_signature_hash,
@@ -496,6 +502,25 @@ impl StackerDBListenerComms {
         blocks.insert(block.signer_signature_hash(), block_status);
     }
 
+    /// Reset rejections for a block proposal.
+    /// This is used when a block proposal times out and we need to retry it by
+    /// clearing the block's rejections. Block approvals cannot be cleared
+    /// because an old approval could always be used to make a block reach
+    /// the approval threshold.
+    pub fn reset_rejections(&self, signer_sighash: &Sha512Trunc256Sum) {
+        let (lock, _cvar) = &*self.blocks;
+        let mut blocks = lock.lock().expect("FATAL: failed to lock block status");
+        if let Some(block) = blocks.get_mut(signer_sighash) {
+            block.responded_signers.clear();
+            block.total_weight_rejected = 0;
+
+            // Add approving signers back to the responded signers set
+            for (slot_id, _) in block.gathered_signatures.iter() {
+                block.responded_signers.insert(*slot_id);
+            }
+        }
+    }
+
     /// Get the status for `block` from the Stacker DB listener.
     /// If the block is not found in the map, return an error.
     /// If the block is found, call `condition` to check if the block status