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};
@@ -66,6 +69,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 +106,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 +124,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,16 +307,38 @@ 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 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
+                    // rejections-based timeout expired?
+                    if rejections_timer.elapsed() > *rejections_timeout {
+                        return false;
+                    }
+                    // number or rejections changed?
+                    if status.total_reject_weight as u64 != rejections {
+                        return false;
+                    }
+                    // enough signatures?
+                    return status.total_weight_signed < self.weight_threshold;
                 },
             )? {
                 Some(status) => status,
@@ -336,10 +372,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 +429,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/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

@@ -937,6 +937,7 @@ fn setup_stx_btc_node<G: FnMut(&mut NeonConfig)>(
     let counters = run_loop.counters();
 
     let coord_channel = run_loop.coordinator_channels();
+
     let run_loop_thread = thread::spawn(move || run_loop.start(None, 0));
 
     // Give the run loop some time to start up!