Merge pull request #5409 from stacks-network/feat/signer-track-validation-submission-with-config-timeout

Feat/signer track validation submission with config timeout

Author: jferrant

.github/workflows/bitcoin-tests.yml

@@ -125,6 +125,7 @@ jobs:
           - tests::signer::v0::multiple_miners_with_custom_chain_id
           - tests::signer::v0::block_commit_delay
           - tests::signer::v0::continue_after_fast_block_no_sortition
+          - tests::signer::v0::block_validation_response_timeout
           - tests::nakamoto_integrations::burn_ops_integration_test
           - tests::nakamoto_integrations::check_block_heights
           - tests::nakamoto_integrations::clarity_burn_state

stacks-signer/src/client/mod.rs

@@ -412,6 +412,7 @@ pub(crate) mod tests {
             first_proposal_burn_block_timing: config.first_proposal_burn_block_timing,
             block_proposal_timeout: config.block_proposal_timeout,
             tenure_last_block_proposal_timeout: config.tenure_last_block_proposal_timeout,
+            block_proposal_validation_timeout: config.block_proposal_validation_timeout,
         }
     }
 

stacks-signer/src/config.rs

@@ -35,6 +35,7 @@ use crate::client::SignerSlotID;
 
 const EVENT_TIMEOUT_MS: u64 = 5000;
 const BLOCK_PROPOSAL_TIMEOUT_MS: u64 = 600_000;
+const BLOCK_PROPOSAL_VALIDATION_TIMEOUT_MS: u64 = 120_000;
 const DEFAULT_FIRST_PROPOSAL_BURN_BLOCK_TIMING_SECS: u64 = 60;
 const DEFAULT_TENURE_LAST_BLOCK_PROPOSAL_TIMEOUT_SECS: u64 = 30;
 
@@ -132,6 +133,8 @@ pub struct SignerConfig {
     /// Time to wait for the last block of a tenure to be globally accepted or rejected
     /// before considering a new miner's block at the same height as potentially valid.
     pub tenure_last_block_proposal_timeout: Duration,
+    /// How much time to wait for a block proposal validation response before marking the block invalid
+    pub block_proposal_validation_timeout: Duration,
 }
 
 /// The parsed configuration for the signer
@@ -165,6 +168,9 @@ pub struct GlobalConfig {
     /// Time to wait for the last block of a tenure to be globally accepted or rejected
     /// before considering a new miner's block at the same height as potentially valid.
     pub tenure_last_block_proposal_timeout: Duration,
+    /// How long to wait for a response from a block proposal validation response from the node
+    /// before marking that block as invalid and rejecting it
+    pub block_proposal_validation_timeout: Duration,
 }
 
 /// Internal struct for loading up the config file
@@ -187,16 +193,19 @@ struct RawConfigFile {
     pub db_path: String,
     /// Metrics endpoint
     pub metrics_endpoint: Option<String>,
-    /// How much time must pass in seconds between the first block proposal in a tenure and the next bitcoin block
-    ///  before a subsequent miner isn't allowed to reorg the tenure
+    /// How much time (in secs) must pass between the first block proposal in a tenure and the next bitcoin block
+    /// before a subsequent miner isn't allowed to reorg the tenure
     pub first_proposal_burn_block_timing_secs: Option<u64>,
-    /// How much time to wait for a miner to propose a block following a sortition in milliseconds
+    /// How much time (in millisecs) to wait for a miner to propose a block following a sortition
     pub block_proposal_timeout_ms: Option<u64>,
     /// An optional custom Chain ID
     pub chain_id: Option<u32>,
     /// Time in seconds to wait for the last block of a tenure to be globally accepted or rejected
     /// before considering a new miner's block at the same height as potentially valid.
     pub tenure_last_block_proposal_timeout_secs: Option<u64>,
+    /// How long to wait (in millisecs) for a response from a block proposal validation response from the node
+    /// before marking that block as invalid and rejecting it
+    pub block_proposal_validation_timeout_ms: Option<u64>,
 }
 
 impl RawConfigFile {
@@ -282,6 +291,12 @@ impl TryFrom<RawConfigFile> for GlobalConfig {
                 .unwrap_or(DEFAULT_TENURE_LAST_BLOCK_PROPOSAL_TIMEOUT_SECS),
         );
 
+        let block_proposal_validation_timeout = Duration::from_millis(
+            raw_data
+                .block_proposal_validation_timeout_ms
+                .unwrap_or(BLOCK_PROPOSAL_VALIDATION_TIMEOUT_MS),
+        );
+
         Ok(Self {
             node_host: raw_data.node_host,
             endpoint,
@@ -296,6 +311,7 @@ impl TryFrom<RawConfigFile> for GlobalConfig {
             block_proposal_timeout,
             chain_id: raw_data.chain_id,
             tenure_last_block_proposal_timeout,
+            block_proposal_validation_timeout,
         })
     }
 }

stacks-signer/src/runloop.rs

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

stacks-signer/src/v0/signer.rs

@@ -15,6 +15,7 @@
 use std::collections::HashMap;
 use std::fmt::Debug;
 use std::sync::mpsc::Sender;
+use std::time::{Duration, Instant};
 
 use blockstack_lib::chainstate::nakamoto::{NakamotoBlock, NakamotoBlockHeader};
 use blockstack_lib::net::api::postblock_proposal::{
@@ -85,6 +86,11 @@ pub struct Signer {
     pub signer_db: SignerDb,
     /// Configuration for proposal evaluation
     pub proposal_config: ProposalEvalConfig,
+    /// How long to wait for a block proposal validation response to arrive before
+    /// marking a submitted block as invalid
+    pub block_proposal_validation_timeout: Duration,
+    /// The current submitted block proposal and its submission time
+    pub submitted_block_proposal: Option<(BlockProposal, Instant)>,
 }
 
 impl std::fmt::Display for Signer {
@@ -127,6 +133,7 @@ impl SignerTrait<SignerMessage> for Signer {
         if event_parity == Some(other_signer_parity) {
             return;
         }
+        self.check_submitted_block_proposal();
         debug!("{self}: Processing event: {event:?}");
         let Some(event) = event else {
             // No event. Do nothing.
@@ -274,6 +281,8 @@ impl From<SignerConfig> for Signer {
             reward_cycle: signer_config.reward_cycle,
             signer_db,
             proposal_config,
+            submitted_block_proposal: None,
+            block_proposal_validation_timeout: signer_config.block_proposal_validation_timeout,
         }
     }
 }
@@ -355,7 +364,7 @@ impl Signer {
         }
 
         info!(
-            "{self}: received a block proposal for a new block. Submit block for validation. ";
+            "{self}: received a block proposal for a new block.";
             "signer_sighash" => %signer_signature_hash,
             "block_id" => %block_proposal.block.block_id(),
             "block_height" => block_proposal.block.header.chain_length,
@@ -456,14 +465,35 @@ impl Signer {
                 Ok(_) => debug!("{self}: Block rejection accepted by stacker-db"),
             }
         } else {
-            // We don't know if proposal is valid, submit to stacks-node for further checks and store it locally.
-            // Do not store invalid blocks as this could DOS the signer. We only store blocks that are valid or unknown.
-            stacks_client
-                .submit_block_for_validation(block_info.block.clone())
-                .unwrap_or_else(|e| {
-                    warn!("{self}: Failed to submit block for validation: {e:?}");
-                });
+            // Just in case check if the last block validation submission timed out.
+            self.check_submitted_block_proposal();
+            if self.submitted_block_proposal.is_none() {
+                // We don't know if proposal is valid, submit to stacks-node for further checks and store it locally.
+                info!(
+                    "{self}: submitting block proposal for validation";
+                    "signer_sighash" => %signer_signature_hash,
+                    "block_id" => %block_proposal.block.block_id(),
+                    "block_height" => block_proposal.block.header.chain_length,
+                    "burn_height" => block_proposal.burn_height,
+                );
+                match stacks_client.submit_block_for_validation(block_info.block.clone()) {
+                    Ok(_) => {
+                        self.submitted_block_proposal =
+                            Some((block_proposal.clone(), Instant::now()));
+                    }
+                    Err(e) => {
+                        warn!("{self}: Failed to submit block for validation: {e:?}");
+                    }
+                };
+            } else {
+                // Still store the block but log we can't submit it for validation. We may receive enough signatures/rejections
+                // from other signers to push the proposed block into a global rejection/acceptance regardless of our participation.
+                // However, we will not be able to participate beyond this until our block submission times out or we receive a response
+                // from our node.
+                warn!("{self}: cannot submit block proposal for validation as we are already waiting for a response for a prior submission")
+            }
 
+            // Do not store KNOWN invalid blocks as this could DOS the signer. We only store blocks that are valid or unknown.
             self.signer_db
                 .insert_block(&block_info)
                 .unwrap_or_else(|_| panic!("{self}: Failed to insert block in DB"));
@@ -493,6 +523,16 @@ impl Signer {
     ) -> Option<BlockResponse> {
         crate::monitoring::increment_block_validation_responses(true);
         let signer_signature_hash = block_validate_ok.signer_signature_hash;
+        if self
+            .submitted_block_proposal
+            .as_ref()
+            .map(|(proposal, _)| {
+                proposal.block.header.signer_signature_hash() == signer_signature_hash
+            })
+            .unwrap_or(false)
+        {
+            self.submitted_block_proposal = None;
+        }
         // For mutability reasons, we need to take the block_info out of the map and add it back after processing
         let mut block_info = match self
             .signer_db
@@ -542,6 +582,16 @@ impl Signer {
     ) -> Option<BlockResponse> {
         crate::monitoring::increment_block_validation_responses(false);
         let signer_signature_hash = block_validate_reject.signer_signature_hash;
+        if self
+            .submitted_block_proposal
+            .as_ref()
+            .map(|(proposal, _)| {
+                proposal.block.header.signer_signature_hash() == signer_signature_hash
+            })
+            .unwrap_or(false)
+        {
+            self.submitted_block_proposal = None;
+        }
         let mut block_info = match self
             .signer_db
             .block_lookup(self.reward_cycle, &signer_signature_hash)
@@ -617,6 +667,81 @@ impl Signer {
         }
     }
 
+    /// Check the current tracked submitted block proposal to see if it has timed out.
+    /// Broadcasts a rejection and marks the block locally rejected if it has.
+    fn check_submitted_block_proposal(&mut self) {
+        let Some((block_proposal, block_submission)) = self.submitted_block_proposal.take() else {
+            // Nothing to check.
+            return;
+        };
+        if block_submission.elapsed() < self.block_proposal_validation_timeout {
+            // Not expired yet. Put it back!
+            self.submitted_block_proposal = Some((block_proposal, block_submission));
+            return;
+        }
+        let signature_sighash = block_proposal.block.header.signer_signature_hash();
+        // For mutability reasons, we need to take the block_info out of the map and add it back after processing
+        let mut block_info = match self
+            .signer_db
+            .block_lookup(self.reward_cycle, &signature_sighash)
+        {
+            Ok(Some(block_info)) => {
+                if block_info.state == BlockState::GloballyRejected
+                    || block_info.state == BlockState::GloballyAccepted
+                {
+                    // The block has already reached consensus.
+                    return;
+                }
+                block_info
+            }
+            Ok(None) => {
+                // This is weird. If this is reached, its probably an error in code logic or the db was flushed.
+                // Why are we tracking a block submission for a block we have never seen / stored before.
+                error!("{self}: tracking an unknown block validation submission.";
+                    "signer_sighash" => %signature_sighash,
+                    "block_id" => %block_proposal.block.block_id(),
+                );
+                return;
+            }
+            Err(e) => {
+                error!("{self}: Failed to lookup block in signer db: {e:?}",);
+                return;
+            }
+        };
+        // We cannot determine the validity of the block, but we have not reached consensus on it yet.
+        // Reject it so we aren't holding up the network because of our inaction.
+        warn!(
+            "{self}: Failed to receive block validation response within {} ms. Rejecting block.", self.block_proposal_validation_timeout.as_millis();
+            "signer_sighash" => %signature_sighash,
+            "block_id" => %block_proposal.block.block_id(),
+        );
+        let rejection = BlockResponse::rejected(
+            block_proposal.block.header.signer_signature_hash(),
+            RejectCode::ConnectivityIssues,
+            &self.private_key,
+            self.mainnet,
+        );
+        if let Err(e) = block_info.mark_locally_rejected() {
+            warn!("{self}: Failed to mark block as locally rejected: {e:?}",);
+        };
+        debug!("{self}: Broadcasting a block response to stacks node: {rejection:?}");
+        let res = self
+            .stackerdb
+            .send_message_with_retry::<SignerMessage>(rejection.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.signer_db
+            .insert_block(&block_info)
+            .unwrap_or_else(|_| panic!("{self}: Failed to insert block in DB"));
+    }
+
     /// Compute the signing weight, given a list of signatures
     fn compute_signature_signing_weight<'a>(
         &self,
@@ -723,6 +848,15 @@ impl Signer {
             error!("{self}: Failed to update block state: {e:?}",);
             panic!("{self} Failed to update block state: {e}");
         }
+        if self
+            .submitted_block_proposal
+            .as_ref()
+            .map(|(proposal, _)| &proposal.block.header.signer_signature_hash() == block_hash)
+            .unwrap_or(false)
+        {
+            // Consensus reached! No longer bother tracking its validation submission to the node as we are too late to participate in the decision anyway.
+            self.submitted_block_proposal = None;
+        }
     }
 
     /// Handle an observed signature from another signer
@@ -865,6 +999,15 @@ impl Signer {
             }
         }
         self.broadcast_signed_block(stacks_client, block_info.block, &addrs_to_sigs);
+        if self
+            .submitted_block_proposal
+            .as_ref()
+            .map(|(proposal, _)| &proposal.block.header.signer_signature_hash() == block_hash)
+            .unwrap_or(false)
+        {
+            // Consensus reached! No longer bother tracking its validation submission to the node as we are too late to participate in the decision anyway.
+            self.submitted_block_proposal = None;
+        }
     }
 
     fn broadcast_signed_block(

stackslib/src/net/api/postblock_proposal.rs

@@ -343,6 +343,17 @@ impl NakamotoBlockProposal {
         sortdb: &SortitionDB,
         chainstate: &mut StacksChainState, // not directly used; used as a handle to open other chainstates
     ) -> Result<BlockValidateOk, BlockValidateRejectReason> {
+        #[cfg(any(test, feature = "testing"))]
+        {
+            if *TEST_VALIDATE_STALL.lock().unwrap() == Some(true) {
+                // Do an extra check just so we don't log EVERY time.
+                warn!("Block validation is stalled due to testing directive.");
+                while *TEST_VALIDATE_STALL.lock().unwrap() == Some(true) {
+                    std::thread::sleep(std::time::Duration::from_millis(10));
+                }
+                info!("Block validation is no longer stalled due to testing directive.");
+            }
+        }
         let ts_start = get_epoch_time_ms();
         // Measure time from start of function
         let time_elapsed = || get_epoch_time_ms().saturating_sub(ts_start);
@@ -533,24 +544,6 @@ impl NakamotoBlockProposal {
             });
         }
 
-        #[cfg(any(test, feature = "testing"))]
-        {
-            if *TEST_VALIDATE_STALL.lock().unwrap() == Some(true) {
-                // Do an extra check just so we don't log EVERY time.
-                warn!("Block validation is stalled due to testing directive.";
-                    "block_id" => %block.block_id(),
-                    "height" => block.header.chain_length,
-                );
-                while *TEST_VALIDATE_STALL.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.block_id(),
-                    "height" => block.header.chain_length,
-                );
-            }
-        }
-
         info!(
             "Participant: validated anchored block";
             "block_header_hash" => %computed_block_header_hash,