Feat/block rejections timeout heuristic

.github/workflows/bitcoin-tests.yml

@@ -134,6 +134,7 @@ jobs:
           - tests::signer::v0::block_commit_delay
           - tests::signer::v0::continue_after_fast_block_no_sortition
           - tests::signer::v0::block_validation_response_timeout
+          - tests::signer::v0::block_validation_check_rejection_timeout_heuristic
           - tests::signer::v0::block_validation_pending_table
           - tests::signer::v0::new_tenure_while_validating_previous_scenario
           - tests::signer::v0::tenure_extend_after_bad_commit

CHANGELOG.md

@@ -17,6 +17,7 @@ and this project adheres to the versioning scheme outlined in the [README.md](RE
 ### Changed
 
 - Miner will include other transactions in blocks with tenure extend transactions (#5760)
+- Add `block_rejection_timeout_steps` to miner configuration for defining rejections-based timeouts while waiting for signers response (#5705)
 
 ### Fixed
 

stackslib/src/config/mod.rs

@@ -16,7 +16,7 @@
 
 pub mod chain_data;
 
-use std::collections::{HashMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::net::{Ipv4Addr, SocketAddr, ToSocketAddrs};
 use std::path::PathBuf;
 use std::str::FromStr;
@@ -2155,6 +2155,8 @@ pub struct MinerConfig {
     pub tenure_extend_poll_secs: Duration,
     /// Duration to wait before attempting to issue a tenure extend
     pub tenure_timeout: Duration,
+    /// Define the timeout to apply while waiting for signers responses, based on the amount of rejections
+    pub block_rejection_timeout_steps: HashMap<u32, Duration>,
 }
 
 impl Default for MinerConfig {
@@ -2193,6 +2195,14 @@ impl Default for MinerConfig {
             ),
             tenure_extend_poll_secs: Duration::from_secs(DEFAULT_TENURE_EXTEND_POLL_SECS),
             tenure_timeout: Duration::from_secs(DEFAULT_TENURE_TIMEOUT_SECS),
+            block_rejection_timeout_steps: {
+                let mut rejections_timeouts_default_map = HashMap::<u32, Duration>::new();
+                rejections_timeouts_default_map.insert(0, Duration::from_secs(600));
+                rejections_timeouts_default_map.insert(10, Duration::from_secs(300));
+                rejections_timeouts_default_map.insert(20, Duration::from_secs(150));
+                rejections_timeouts_default_map.insert(30, Duration::from_secs(0));
+                rejections_timeouts_default_map
+            },
         }
     }
 }
@@ -2589,6 +2599,7 @@ pub struct MinerConfigFile {
     pub tenure_cost_limit_per_block_percentage: Option<u8>,
     pub tenure_extend_poll_secs: Option<u64>,
     pub tenure_timeout_secs: Option<u64>,
+    pub block_rejection_timeout_steps: Option<HashMap<String, u64>>,
 }
 
 impl MinerConfigFile {
@@ -2731,6 +2742,23 @@ impl MinerConfigFile {
             tenure_cost_limit_per_block_percentage,
             tenure_extend_poll_secs: self.tenure_extend_poll_secs.map(Duration::from_secs).unwrap_or(miner_default_config.tenure_extend_poll_secs),
             tenure_timeout: self.tenure_timeout_secs.map(Duration::from_secs).unwrap_or(miner_default_config.tenure_timeout),
+            block_rejection_timeout_steps: {
+                if let Some(block_rejection_timeout_items) = self.block_rejection_timeout_steps {
+                    let mut rejection_timeout_durations = HashMap::<u32, Duration>::new();
+                    for (slice, seconds) in block_rejection_timeout_items.iter() {
+                        match slice.parse::<u32>() {
+                            Ok(slice_slot) => rejection_timeout_durations.insert(slice_slot, Duration::from_secs(*seconds)),
+                            Err(e) => panic!("block_rejection_timeout_steps keys must be unsigned integers: {}", e)
+                        };
+                    }
+                    if !rejection_timeout_durations.contains_key(&0) {
+                        panic!("block_rejection_timeout_steps requires a definition for the '0' key/step");
+                    }
+                    rejection_timeout_durations
+                } else{
+                    miner_default_config.block_rejection_timeout_steps
+                }
+            }
         })
     }
 }

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

@@ -13,9 +13,12 @@
 // You should have received a copy of the GNU General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+use std::collections::BTreeMap;
+use std::ops::Bound::Included;
 use std::sync::atomic::AtomicBool;
 use std::sync::{Arc, Mutex};
 use std::thread::JoinHandle;
+use std::time::{Duration, Instant};
 
 use libsigner::v0::messages::{MinerSlotID, SignerMessage as SignerMessageV0};
 use libsigner::{BlockProposal, SignerSession, StackerDBSession};
@@ -37,7 +40,9 @@ use stacks::util_lib::boot::boot_code_id;
 use super::stackerdb_listener::StackerDBListenerComms;
 use super::Error as NakamotoNodeError;
 use crate::event_dispatcher::StackerDBChannel;
-use crate::nakamoto_node::stackerdb_listener::{StackerDBListener, EVENT_RECEIVER_POLL};
+use crate::nakamoto_node::stackerdb_listener::{
+    BlockStatus, StackerDBListener, EVENT_RECEIVER_POLL,
+};
 use crate::neon::Counters;
 use crate::Config;
 
@@ -66,6 +71,8 @@ pub struct SignerCoordinator {
     /// Rather, this burn block is used to determine whether or not a new
     ///  burn block has arrived since this thread started.
     burn_tip_at_start: ConsensusHash,
+    /// The timeout configuration based on the percentage of rejections
+    block_rejection_timeout_steps: BTreeMap<u64, Duration>,
 }
 
 impl SignerCoordinator {
@@ -101,6 +108,14 @@ impl SignerCoordinator {
         let miners_contract_id = boot_code_id(MINERS_NAME, is_mainnet);
         let miners_session = StackerDBSession::new(&rpc_socket.to_string(), miners_contract_id);
 
+        // build a BTreeMap of the various timeout steps
+        let mut block_rejection_timeout_steps = BTreeMap::<u64, Duration>::new();
+        for (percentage, duration) in config.miner.block_rejection_timeout_steps.iter() {
+            let rejections_amount =
+                ((f64::from(listener.total_weight) / 100.0) * f64::from(*percentage)) as u64;
+            block_rejection_timeout_steps.insert(rejections_amount, *duration);
+        }
+
         let mut sc = Self {
             message_key,
             is_mainnet,
@@ -111,6 +126,7 @@ impl SignerCoordinator {
             keep_running,
             listener_thread: None,
             burn_tip_at_start: burn_tip_at_start.clone(),
+            block_rejection_timeout_steps,
         };
 
         // Spawn the signer DB listener thread
@@ -293,19 +309,44 @@ impl SignerCoordinator {
         sortdb: &SortitionDB,
         counters: &Counters,
     ) -> Result<Vec<MessageSignature>, NakamotoNodeError> {
+        // the amount of current rejections (used to eventually modify the timeout)
+        let mut rejections: u64 = 0;
+        // default timeout (the 0 entry must be always present)
+        let mut rejections_timeout = self
+            .block_rejection_timeout_steps
+            .get(&rejections)
+            .ok_or_else(|| {
+                NakamotoNodeError::SigningCoordinatorFailure(
+                    "Invalid rejection timeout step function definition".into(),
+                )
+            })?;
+        // this is used for comparing block_status to identify if it has been changed from the previous event
+        let mut block_status_tracker = BlockStatus::default();
+
+        // this is used to track the start of the waiting cycle
+        let rejections_timer = Instant::now();
         loop {
+            // At every iteration wait for the block_status.
+            // Exit when the amount of confirmations/rejections reaches the threshold (or until timeout)
+            // Based on the amount of rejections, eventually modify the timeout.
             let block_status = match self.stackerdb_comms.wait_for_block_status(
                 block_signer_sighash,
                 EVENT_RECEIVER_POLL,
                 |status| {
-                    status.total_weight_signed < self.weight_threshold
-                        && status
-                            .total_reject_weight
-                            .saturating_add(self.weight_threshold)
-                            <= self.total_weight
+                    if rejections_timer.elapsed() > *rejections_timeout {
+                        return false;
+                    }
+                    if *status != block_status_tracker {
+                        return false;
+                    }
+                    return true;
                 },
             )? {
-                Some(status) => status,
+                Some(status) => {
+                    // keep track of the last status
+                    block_status_tracker = status.clone();
+                    status
+                }
                 None => {
                     // If we just received a timeout, we should check if the burnchain
                     // tip has changed or if we received this signed block already in
@@ -336,10 +377,46 @@ impl SignerCoordinator {
                         return Err(NakamotoNodeError::BurnchainTipChanged);
                     }
 
+                    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)
+                        );
+                        return Err(NakamotoNodeError::SigningCoordinatorFailure(
+                            "Gave up while trying reaching the threshold".into(),
+                        ));
+                    }
+
                     continue;
                 }
             };
 
+            if rejections != block_status.total_reject_weight as u64 {
+                rejections = block_status.total_reject_weight as u64;
+                let rejections_timeout_tuple = self
+                    .block_rejection_timeout_steps
+                    .range((Included(0), Included(rejections)))
+                    .last()
+                    .ok_or_else(|| {
+                        NakamotoNodeError::SigningCoordinatorFailure(
+                            "Invalid rejection timeout step function definition".into(),
+                        )
+                    })?;
+                rejections_timeout = rejections_timeout_tuple.1;
+                info!("Number of received rejections updated, resetting timeout";
+                                    "rejections" => rejections,
+                                    "rejections_timeout" => rejections_timeout.as_secs(),
+                                    "rejections_threshold" => self.total_weight.saturating_sub(self.weight_threshold));
+
+                Counters::set(
+                    &counters.naka_miner_current_rejections_timeout_secs,
+                    rejections_timeout.as_secs(),
+                );
+                Counters::set(&counters.naka_miner_current_rejections, rejections);
+            }
+
             if block_status
                 .total_reject_weight
                 .saturating_add(self.weight_threshold)
@@ -357,10 +434,18 @@ impl SignerCoordinator {
                     "block_signer_sighash" => %block_signer_sighash,
                 );
                 return Ok(block_status.gathered_signatures.values().cloned().collect());
-            } else {
+            } 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)
+                );
                 return Err(NakamotoNodeError::SigningCoordinatorFailure(
-                    "Unblocked without reaching the threshold".into(),
+                    "Gave up while trying reaching the threshold".into(),
                 ));
+            } else {
+                continue;
             }
         }
     }

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

@@ -50,7 +50,7 @@ pub static TEST_IGNORE_SIGNERS: LazyLock<TestFlag<bool>> = LazyLock::new(TestFla
 /// waking up to check timeouts?
 pub static EVENT_RECEIVER_POLL: Duration = Duration::from_millis(500);
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Default, PartialEq)]
 pub struct BlockStatus {
     pub responded_signers: HashSet<StacksPublicKey>,
     pub gathered_signatures: BTreeMap<u32, MessageSignature>,
@@ -337,10 +337,8 @@ impl StackerDBListener {
                         block.gathered_signatures.insert(slot_id, signature);
                         block.responded_signers.insert(signer_pubkey);
 
-                        if block.total_weight_signed >= self.weight_threshold {
-                            // Signal to anyone waiting on this block that we have enough signatures
-                            cvar.notify_all();
-                        }
+                        // Signal to anyone waiting on this block that we have a new status
+                        cvar.notify_all();
 
                         // Update the idle timestamp for this signer
                         self.update_idle_timestamp(
@@ -396,14 +394,8 @@ impl StackerDBListener {
                             "server_version" => rejected_data.metadata.server_version,
                         );
 
-                        if block
-                            .total_reject_weight
-                            .saturating_add(self.weight_threshold)
-                            > self.total_weight
-                        {
-                            // Signal to anyone waiting on this block that we have enough rejections
-                            cvar.notify_all();
-                        }
+                        // Signal to anyone waiting on this block that we have a new status
+                        cvar.notify_all();
 
                         // Update the idle timestamp for this signer
                         self.update_idle_timestamp(

testnet/stacks-node/src/run_loop/neon.rs

@@ -118,6 +118,9 @@ pub struct Counters {
     pub naka_miner_directives: RunLoopCounter,
     pub naka_submitted_commit_last_stacks_tip: RunLoopCounter,
 
+    pub naka_miner_current_rejections: RunLoopCounter,
+    pub naka_miner_current_rejections_timeout_secs: RunLoopCounter,
+
     #[cfg(test)]
     pub naka_skip_commit_op: TestFlag<bool>,
 }
@@ -136,12 +139,12 @@ impl Counters {
     fn inc(_ctr: &RunLoopCounter) {}
 
     #[cfg(test)]
-    fn set(ctr: &RunLoopCounter, value: u64) {
+    pub fn set(ctr: &RunLoopCounter, value: u64) {
         ctr.0.store(value, Ordering::SeqCst);
     }
 
     #[cfg(not(test))]
-    fn set(_ctr: &RunLoopCounter, _value: u64) {}
+    pub fn set(_ctr: &RunLoopCounter, _value: u64) {}
 
     pub fn bump_blocks_processed(&self) {
         Counters::inc(&self.blocks_processed);

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

@@ -86,6 +86,7 @@
     pub counters: Counters,
     pub coord_channel: Arc<Mutex<CoordinatorChannels>>,
     pub conf: NeonConfig,
+    pub counters: Counters,
 }
 
 /// A test harness for running a v0 or v1 signer integration test
@@ -937,6 +938,9 @@
     let counters = run_loop.counters();
 
     let coord_channel = run_loop.coordinator_channels();
+
+    let run_loop_counters = run_loop.counters();
+
     let run_loop_thread = thread::spawn(move || run_loop.start(None, 0));
 
     // Give the run loop some time to start up!
@@ -974,5 +978,6 @@
         coord_channel,
         counters,
         conf: naka_conf,
+        counters: run_loop_counters,
     }
 }