Do not terminate tenure early and process block proposals that are fr…

…om prior cycles. Adds test.

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

stacks-signer/src/lib.rs

@@ -81,6 +81,8 @@ pub trait Signer<T: SignerEventTrait>: Debug + Display {
         current_reward_cycle: u64,
         command: Option<RunLoopCommand>,
     );
+    /// Check if the signer is stale, i.e. its tenure is complete and it has no pending blocks to process
+    fn has_pending_blocks(&self) -> bool;
 }
 
 /// A wrapper around the running signer type for the signer

stacks-signer/src/runloop.rs

@@ -375,13 +375,21 @@ impl<Signer: SignerTrait<T>, T: StacksMessageCodec + Clone + Send + Debug> RunLo
     fn cleanup_stale_signers(&mut self, current_reward_cycle: u64) {
         let mut to_delete = Vec::new();
         for (idx, signer) in &mut self.stacks_signers {
-            if signer.reward_cycle() < current_reward_cycle {
+            let reward_cycle = signer.reward_cycle();
+            let next_reward_cycle = reward_cycle.wrapping_add(1);
+            let stale = match next_reward_cycle.cmp(&current_reward_cycle) {
+                std::cmp::Ordering::Less => true, // We are more than one reward cycle behind, so we are stale
+                std::cmp::Ordering::Equal => !signer.has_pending_blocks(), // We are the next reward cycle, so check if we have any pending blocks to process
+                std::cmp::Ordering::Greater => false, // We are the current reward cycle, so we are not stale
+            };
+            if stale {
                 debug!("{signer}: Signer's tenure has completed.");
                 to_delete.push(*idx);
                 continue;
             }
         }
         for idx in to_delete {
+            println!("DELETING");
             self.stacks_signers.remove(&idx);
         }
     }

stacks-signer/src/signerdb.rs

@@ -100,6 +100,7 @@ CREATE TABLE IF NOT EXISTS blocks (
 const CREATE_INDEXES: &str = "
 CREATE INDEX IF NOT EXISTS blocks_signed_over ON blocks (signed_over);
 CREATE INDEX IF NOT EXISTS blocks_consensus_hash ON blocks (consensus_hash);
+CREATE INDEX IF NOT EXISTS blocks_valid ON blocks ((json_extract(block_info, '$.valid')));
 ";
 
 const CREATE_SIGNER_STATE_TABLE: &str = "
@@ -230,6 +231,15 @@ impl SignerDb {
 
         Ok(())
     }
+
+    /// Determine if there are any pending blocks that have not yet been processed by checking the block_info.valid field
+    pub fn has_pending_blocks(&self, reward_cycle: u64) -> Result<bool, DBError> {
+        let query = "SELECT block_info FROM blocks WHERE reward_cycle = ? AND json_extract(block_info, '$.valid') IS NULL LIMIT 1";
+        let result: Option<String> =
+            query_row(&self.db, query, params!(&u64_to_sql(reward_cycle)?))?;
+
+        Ok(result.is_some())
+    }
 }
 
 fn try_deserialize<T>(s: Option<String>) -> Result<Option<T>, DBError>
@@ -420,6 +430,34 @@ mod tests {
             .is_none());
     }
 
+    #[test]
+    fn test_has_pending_blocks() {
+        let db_path = tmp_db_path();
+        let mut db = SignerDb::new(db_path).expect("Failed to create signer db");
+        let (mut block_info_1, _block_proposal) = create_block();
+        let (block_info_2, _block_proposal) = create_block();
+        let (block_info_3, _block_proposal) = create_block();
+        let (block_info_4, _block_proposal) = create_block();
+
+        db.insert_block(&block_info_1)
+            .expect("Unable to insert block into db");
+        db.insert_block(&block_info_2)
+            .expect("Unable to insert block into db");
+        db.insert_block(&block_info_3)
+            .expect("Unable to insert block into db");
+        db.insert_block(&block_info_4)
+            .expect("Unable to insert block into db");
+
+        assert!(db.has_pending_blocks(block_info_1.reward_cycle).unwrap());
+
+        block_info_1.valid = Some(true);
+
+        db.insert_block(&block_info_1)
+            .expect("Unable to update block in db");
+
+        assert!(!db.has_pending_blocks(block_info_1.reward_cycle).unwrap());
+    }
+
     #[test]
     fn test_sqlite_version() {
         let db_path = tmp_db_path();

stacks-signer/src/v0/signer.rs

@@ -82,13 +82,13 @@ impl SignerTrait<SignerMessage> for Signer {
         sortition_state: &mut Option<SortitionsView>,
         event: Option<&SignerEvent<SignerMessage>>,
         _res: Sender<Vec<SignerResult>>,
-        current_reward_cycle: u64,
+        _current_reward_cycle: u64,
     ) {
         let event_parity = match event {
-            Some(SignerEvent::BlockValidationResponse(_)) => Some(current_reward_cycle % 2),
             // Block proposal events do have reward cycles, but each proposal has its own cycle,
             //  and the vec could be heterogenous, so, don't differentiate.
-            Some(SignerEvent::MinerMessages(..))
+            Some(SignerEvent::BlockValidationResponse(_))
+            | Some(SignerEvent::MinerMessages(..))
             | Some(SignerEvent::NewBurnBlock(_))
             | Some(SignerEvent::StatusCheck)
             | None => None,
@@ -163,6 +163,16 @@ impl SignerTrait<SignerMessage> for Signer {
             warn!("{self}: Received a command: {command:?}. V0 Signers do not support commands. Ignoring...")
         }
     }
+
+    fn has_pending_blocks(&self) -> bool {
+        self.signer_db
+            .has_pending_blocks(self.reward_cycle)
+            .unwrap_or_else(|e| {
+                error!("{self}: Failed to check for pending blocks: {e:?}",);
+                // Assume we have pending blocks to prevent premature cleanup
+                true
+            })
+    }
 }
 
 impl From<SignerConfig> for Signer {
@@ -381,7 +391,6 @@ impl Signer {
                     }
                 };
                 block_info.valid = Some(true);
-                // TODO: do not sign the block if it fails signer state checks (forks, etc.)
                 let signature = self
                     .private_key
                     .sign(&signer_signature_hash.0)

stacks-signer/src/v1/signer.rs

@@ -267,6 +267,16 @@ impl SignerTrait<SignerMessage> for Signer {
         }
         self.process_next_command(stacks_client, current_reward_cycle);
     }
+
+    fn has_pending_blocks(&self) -> bool {
+        self.signer_db
+            .has_pending_blocks(self.reward_cycle)
+            .unwrap_or_else(|e| {
+                error!("{self}: Failed to check if there are pending blocks: {e:?}");
+                // Assume there are pending blocks to prevent premature cleanup
+                true
+            })
+    }
 }
 
 impl Signer {

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

@@ -100,7 +100,11 @@ pub struct SignerTest<S> {
 }
 
 impl<S: Signer<T> + Send + 'static, T: SignerEventTrait + 'static> SignerTest<SpawnedSigner<S, T>> {
-    fn new(num_signers: usize, initial_balances: Vec<(StacksAddress, u64)>) -> Self {
+    fn new(
+        num_signers: usize,
+        initial_balances: Vec<(StacksAddress, u64)>,
+        wait_on_signers: Option<Duration>,
+    ) -> Self {
         // Generate Signer Data
         let signer_stacks_private_keys = (0..num_signers)
             .map(|_| StacksPrivateKey::new())
@@ -118,7 +122,11 @@ impl<S: Signer<T> + Send + 'static, T: SignerEventTrait + 'static> SignerTest<Sp
         // That's the kind of thing an idiot would have on his luggage!
         let password = "12345";
         naka_conf.connection_options.block_proposal_token = Some(password.to_string());
-        naka_conf.miner.wait_on_signers = Duration::from_secs(10);
+        if let Some(wait_on_signers) = wait_on_signers {
+            naka_conf.miner.wait_on_signers = wait_on_signers;
+        } else {
+            naka_conf.miner.wait_on_signers = Duration::from_secs(10);
+        }
 
         let run_stamp = rand::random();
 
@@ -547,6 +555,7 @@ fn setup_stx_btc_node(
             EventKeyType::StackerDBChunks,
             EventKeyType::BlockProposal,
             EventKeyType::MinedBlocks,
+            EventKeyType::BurnchainBlocks,
         ],
     });
 

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

@@ -41,7 +41,9 @@ use tracing_subscriber::{fmt, EnvFilter};
 
 use super::SignerTest;
 use crate::tests::nakamoto_integrations::{boot_to_epoch_3_reward_set, next_block_and};
-use crate::tests::neon_integrations::{get_chain_info, next_block_and_wait, submit_tx};
+use crate::tests::neon_integrations::{
+    get_chain_info, next_block_and_wait, submit_tx, test_observer,
+};
 use crate::tests::{self, make_stacks_transfer};
 use crate::{nakamoto_node, BurnchainController};
 
@@ -254,7 +256,7 @@ fn block_proposal_rejection() {
 
     info!("------------------------- Test Setup -------------------------");
     let num_signers = 5;
-    let mut signer_test: SignerTest<SpawnedSigner> = SignerTest::new(num_signers, vec![]);
+    let mut signer_test: SignerTest<SpawnedSigner> = SignerTest::new(num_signers, vec![], None);
     signer_test.boot_to_epoch_3();
     let short_timeout = Duration::from_secs(30);
 
@@ -360,7 +362,7 @@ fn miner_gather_signatures() {
 
     info!("------------------------- Test Setup -------------------------");
     let num_signers = 5;
-    let mut signer_test: SignerTest<SpawnedSigner> = SignerTest::new(num_signers, vec![]);
+    let mut signer_test: SignerTest<SpawnedSigner> = SignerTest::new(num_signers, vec![], None);
     signer_test.boot_to_epoch_3();
     let timeout = Duration::from_secs(30);
 
@@ -412,7 +414,7 @@ fn mine_2_nakamoto_reward_cycles() {
     info!("------------------------- Test Setup -------------------------");
     let nmb_reward_cycles = 2;
     let num_signers = 5;
-    let mut signer_test: SignerTest<SpawnedSigner> = SignerTest::new(num_signers, vec![]);
+    let mut signer_test: SignerTest<SpawnedSigner> = SignerTest::new(num_signers, vec![], None);
     let timeout = Duration::from_secs(200);
     signer_test.boot_to_epoch_3();
     let curr_reward_cycle = signer_test.get_current_reward_cycle();
@@ -468,13 +470,36 @@ fn end_of_tenure() {
     let mut signer_test: SignerTest<SpawnedSigner> = SignerTest::new(
         num_signers,
         vec![(sender_addr.clone(), send_amt + send_fee)],
+        Some(Duration::from_secs(500)),
     );
     let http_origin = format!("http://{}", &signer_test.running_nodes.conf.node.rpc_bind);
+    let long_timeout = Duration::from_secs(200);
+    let short_timeout = Duration::from_secs(20);
 
     signer_test.boot_to_epoch_3();
+    let curr_reward_cycle = signer_test.get_current_reward_cycle();
+    // Advance to one before the next reward cycle to ensure we are on the reward cycle boundary
+    let final_reward_cycle = curr_reward_cycle + 1;
+    let final_reward_cycle_height_boundary = signer_test
+        .running_nodes
+        .btc_regtest_controller
+        .get_burnchain()
+        .reward_cycle_to_block_height(final_reward_cycle)
+        - 2;
 
-    signer_test.mine_nakamoto_block(Duration::from_secs(30));
+    info!("------------------------- Test Mine to Next Reward Cycle Boundary  -------------------------");
+    signer_test.run_until_burnchain_height_nakamoto(
+        long_timeout,
+        final_reward_cycle_height_boundary,
+        num_signers,
+    );
+    println!("Advanced to nexct reward cycle boundary: {final_reward_cycle_height_boundary}");
+    assert_eq!(
+        signer_test.get_current_reward_cycle(),
+        final_reward_cycle - 1
+    );
 
+    info!("------------------------- Test Block Validation Stalled -------------------------");
     TEST_VALIDATE_STALL.lock().unwrap().replace(true);
 
     let proposals_before = signer_test
@@ -486,21 +511,26 @@ fn end_of_tenure() {
         .nakamoto_blocks_mined
         .load(Ordering::SeqCst);
 
+    let info = get_chain_info(&signer_test.running_nodes.conf);
+    let start_height = info.stacks_tip_height;
     // submit a tx so that the miner will mine an extra block
     let sender_nonce = 0;
     let transfer_tx =
         make_stacks_transfer(&sender_sk, sender_nonce, send_fee, &recipient, send_amt);
     submit_tx(&http_origin, &transfer_tx);
 
     info!("Submitted transfer tx and waiting for block proposal");
-    loop {
-        let blocks_proposed = signer_test
-            .running_nodes
-            .nakamoto_blocks_proposed
-            .load(Ordering::SeqCst);
-        if blocks_proposed > proposals_before {
-            break;
-        }
+    let start_time = Instant::now();
+    while signer_test
+        .running_nodes
+        .nakamoto_blocks_proposed
+        .load(Ordering::SeqCst)
+        <= proposals_before
+    {
+        assert!(
+            start_time.elapsed() <= short_timeout,
+            "Timed out waiting for block proposal"
+        );
         std::thread::sleep(Duration::from_millis(100));
     }
 
@@ -516,14 +546,13 @@ fn end_of_tenure() {
         blocks_before
     );
 
+    info!("Triggering a new block to be mined");
+
+    // Mine a couple blocks into the next reward cycle
     let commits_before = signer_test
         .running_nodes
         .commits_submitted
         .load(Ordering::SeqCst);
-
-    info!("Triggering a new block to be mined");
-
-    // Trigger the next block to be mined and commit submitted
     next_block_and(
         &mut signer_test.running_nodes.btc_regtest_controller,
         10,
@@ -536,23 +565,52 @@ fn end_of_tenure() {
         },
     )
     .unwrap();
+    for _ in 0..2 {
+        next_block_and(
+            &mut signer_test.running_nodes.btc_regtest_controller,
+            10,
+            || Ok(true),
+        )
+        .unwrap();
+    }
+    assert_eq!(signer_test.get_current_reward_cycle(), final_reward_cycle);
+
+    while test_observer::get_burn_blocks()
+        .last()
+        .unwrap()
+        .get("burn_block_height")
+        .unwrap()
+        .as_u64()
+        .unwrap()
+        >= final_reward_cycle_height_boundary + 3
+    {
+        std::thread::sleep(Duration::from_secs(1));
+        assert!(
+            start_time.elapsed() <= short_timeout,
+            "Timed out waiting for bun block events"
+        );
+    }
 
-    info!("Disabling the stall and waiting for the blocks to be processed");
+    std::thread::sleep(short_timeout);
+    info!("Unpausing block validation and waiting for block to be processed");
     // Disable the stall and wait for the block to be processed
     TEST_VALIDATE_STALL.lock().unwrap().replace(false);
-    loop {
-        let blocks_mined = signer_test
-            .running_nodes
-            .nakamoto_blocks_mined
-            .load(Ordering::SeqCst);
-        if blocks_mined > blocks_before + 1 {
-            break;
-        }
+    let start_time = Instant::now();
+    while signer_test
+        .running_nodes
+        .nakamoto_blocks_mined
+        .load(Ordering::SeqCst)
+        <= blocks_before
+    {
+        assert!(
+            start_time.elapsed() <= short_timeout,
+            "Timed out waiting for block to be mined"
+        );
         std::thread::sleep(Duration::from_millis(100));
     }
 
     let info = get_chain_info(&signer_test.running_nodes.conf);
-    assert_eq!(info.stacks_tip_height, 30);
+    assert_eq!(info.stacks_tip_height, start_height + 1);
 
     signer_test.shutdown();
 }
@@ -580,6 +638,7 @@ fn retry_on_timeout() {
     let mut signer_test: SignerTest<SpawnedSigner> = SignerTest::new(
         num_signers,
         vec![(sender_addr.clone(), send_amt + send_fee)],
+        Some(Duration::from_secs(5)),
     );
     let http_origin = format!("http://{}", &signer_test.running_nodes.conf.node.rpc_bind);
 
@@ -619,8 +678,8 @@ fn retry_on_timeout() {
 
     info!("Block proposed, verifying that it is not processed");
 
-    // Wait 20 seconds to be sure that the timeout has occurred
-    std::thread::sleep(Duration::from_secs(20));
+    // Wait 10 seconds to be sure that the timeout has occurred
+    std::thread::sleep(Duration::from_secs(10));
     assert_eq!(
         signer_test
             .running_nodes