add miner inactivity handling to local state machine, lots of test assertions

Author: kantai

libsigner/src/events.rs

@@ -32,6 +32,7 @@ use blockstack_lib::net::api::postblock_proposal::{
 use blockstack_lib::net::stackerdb::MINER_SLOT_COUNT;
 use blockstack_lib::util_lib::boot::boot_code_id;
 use blockstack_lib::version_string;
+use clarity::types::chainstate::StacksBlockId;
 use clarity::vm::types::serialization::SerializationError;
 use clarity::vm::types::QualifiedContractIdentifier;
 use serde::{Deserialize, Serialize};
@@ -202,13 +203,20 @@ pub enum SignerEvent<T: SignerEventTrait> {
         burn_height: u64,
         /// the burn hash for the newly processed burn block
         burn_header_hash: BurnchainHeaderHash,
+        /// the consensus hash for the newly processed burn block
+        consensus_hash: ConsensusHash,
         /// the time at which this event was received by the signer's event processor
         received_time: SystemTime,
     },
     /// A new processed Stacks block was received from the node with the given block hash
     NewBlock {
-        /// The block header hash for the newly processed stacks block
-        block_hash: Sha512Trunc256Sum,
+        /// The stacks block ID (or index block hash) of the new block
+        block_id: StacksBlockId,
+        /// The consensus hash of the block (either the tenure it was produced during for Stacks 3.0
+        ///   or the burn block that won the sortition in Stacks 2.0)
+        consensus_hash: ConsensusHash,
+        /// The signer sighash for the newly processed stacks block
+        signer_sighash: Sha512Trunc256Sum,
         /// The block height for the newly processed stacks block
         block_height: u64,
     },
@@ -556,6 +564,7 @@ struct BurnBlockEvent {
     reward_recipients: Vec<serde_json::Value>,
     reward_slot_holders: Vec<String>,
     burn_amount: u64,
+    consensus_hash: String,
 }
 
 impl<T: SignerEventTrait> TryFrom<BurnBlockEvent> for SignerEvent<T> {
@@ -571,34 +580,65 @@ impl<T: SignerEventTrait> TryFrom<BurnBlockEvent> for SignerEvent<T> {
                     .map_err(|e| EventError::Deserialize(format!("Invalid hex string: {e}")))
             })?;
 
+        let consensus_hash = burn_block_event
+            .consensus_hash
+            .get(2..)
+            .ok_or_else(|| EventError::Deserialize("Hex string should be 0x prefixed".into()))
+            .and_then(|hex| {
+                ConsensusHash::from_hex(hex)
+                    .map_err(|e| EventError::Deserialize(format!("Invalid hex string: {e}")))
+            })?;
+
         Ok(SignerEvent::NewBurnBlock {
             burn_height: burn_block_event.burn_block_height,
             received_time: SystemTime::now(),
             burn_header_hash,
+            consensus_hash,
         })
     }
 }
 
 #[derive(Debug, Deserialize)]
 struct BlockEvent {
-    block_hash: String,
+    index_block_hash: String,
+    signer_signature_hash: String,
+    consensus_hash: String,
     block_height: u64,
 }
 
 impl<T: SignerEventTrait> TryFrom<BlockEvent> for SignerEvent<T> {
     type Error = EventError;
 
     fn try_from(block_event: BlockEvent) -> Result<Self, Self::Error> {
-        let block_hash: Sha512Trunc256Sum = block_event
-            .block_hash
+        let signer_sighash = block_event
+            .signer_signature_hash
             .get(2..)
             .ok_or_else(|| EventError::Deserialize("Hex string should be 0x prefixed".into()))
             .and_then(|hex| {
                 Sha512Trunc256Sum::from_hex(hex)
                     .map_err(|e| EventError::Deserialize(format!("Invalid hex string: {e}")))
             })?;
+        let consensus_hash = block_event
+            .consensus_hash
+            .get(2..)
+            .ok_or_else(|| EventError::Deserialize("Hex string should be 0x prefixed".into()))
+            .and_then(|hex| {
+                ConsensusHash::from_hex(hex)
+                    .map_err(|e| EventError::Deserialize(format!("Invalid hex string: {e}")))
+            })?;
+        let block_id = block_event
+            .index_block_hash
+            .get(2..)
+            .ok_or_else(|| EventError::Deserialize("Hex string should be 0x prefixed".into()))
+            .and_then(|hex| {
+                StacksBlockId::from_hex(hex)
+                    .map_err(|e| EventError::Deserialize(format!("Invalid hex string: {e}")))
+            })?;
+
         Ok(SignerEvent::NewBlock {
-            block_hash,
+            block_id,
+            signer_sighash,
+            consensus_hash,
             block_height: block_event.block_height,
         })
     }

stacks-signer/src/runloop.rs

@@ -500,37 +500,6 @@ impl<Signer: SignerTrait<T>, T: StacksMessageCodec + Clone + Send + Debug>
             "Running one pass for the signer. state={:?}, event={event:?}",
             self.state
         );
-        // This is the only event that we respond to from the outer signer runloop
-        if let Some(SignerEvent::StatusCheck) = event {
-            info!("Signer status check requested: {:?}.", self.state);
-            if let Err(e) = res.send(vec![StateInfo {
-                runloop_state: self.state,
-                reward_cycle_info: self.current_reward_cycle_info,
-                running_signers: self
-                    .stacks_signers
-                    .values()
-                    .map(|s| s.reward_cycle())
-                    .collect(),
-                signer_state_machines: self
-                    .stacks_signers
-                    .iter()
-                    .map(|(reward_cycle, signer)| {
-                        let ConfiguredSigner::RegisteredSigner(ref signer) = signer else {
-                            return (*reward_cycle, None);
-                        };
-                        (
-                            *reward_cycle,
-                            Some(signer.get_local_state_machine().clone()),
-                        )
-                    })
-                    .collect(),
-            }
-            .into()])
-            {
-                error!("Failed to send status check result: {e}.");
-            }
-        }
-
         if self.state == State::Uninitialized {
             if let Err(e) = self.initialize_runloop() {
                 error!("Failed to initialize signer runloop: {e}.");
@@ -564,6 +533,38 @@ impl<Signer: SignerTrait<T>, T: StacksMessageCodec + Clone + Send + Debug>
                 current_reward_cycle,
             );
         }
+
+        // This is the only event that we respond to from the outer signer runloop
+        if let Some(SignerEvent::StatusCheck) = event {
+            info!("Signer status check requested: {:?}.", self.state);
+            if let Err(e) = res.send(vec![StateInfo {
+                runloop_state: self.state,
+                reward_cycle_info: self.current_reward_cycle_info,
+                running_signers: self
+                    .stacks_signers
+                    .values()
+                    .map(|s| s.reward_cycle())
+                    .collect(),
+                signer_state_machines: self
+                    .stacks_signers
+                    .iter()
+                    .map(|(reward_cycle, signer)| {
+                        let ConfiguredSigner::RegisteredSigner(ref signer) = signer else {
+                            return (*reward_cycle, None);
+                        };
+                        (
+                            *reward_cycle,
+                            Some(signer.get_local_state_machine().clone()),
+                        )
+                    })
+                    .collect(),
+            }
+            .into()])
+            {
+                error!("Failed to send status check result: {e}.");
+            }
+        }
+
         if self.state == State::NoRegisteredSigners && event.is_some() {
             let next_reward_cycle = current_reward_cycle.saturating_add(1);
             info!("Signer is not registered for the current reward cycle ({current_reward_cycle}). Reward set is not yet determined or signer is not registered for the upcoming reward cycle ({next_reward_cycle}).");

stacks-signer/src/signerdb.rs

@@ -510,6 +510,15 @@ ALTER TABLE block_rejection_signer_addrs
     ADD COLUMN reject_code INTEGER;
 "#;
 
+static ADD_CONSENSUS_HASH: &str = r#"
+ALTER TABLE burn_blocks
+    ADD COLUMN consensus_hash TEXT;
+"#;
+
+static ADD_CONSENSUS_HASH_INDEX: &str = r#"
+CREATE INDEX IF NOT EXISTS burn_blocks_ch on burn_blocks (consensus_hash);
+"#;
+
 static SCHEMA_1: &[&str] = &[
     DROP_SCHEMA_0,
     CREATE_DB_CONFIG,
@@ -579,9 +588,15 @@ static SCHEMA_9: &[&str] = &[
     "INSERT INTO db_config (version) VALUES (9);",
 ];
 
+static SCHEMA_10: &[&str] = &[
+    ADD_CONSENSUS_HASH,
+    ADD_CONSENSUS_HASH_INDEX,
+    "INSERT INTO db_config (version) VALUES (10);",
+];
+
 impl SignerDb {
     /// The current schema version used in this build of the signer binary.
-    pub const SCHEMA_VERSION: u32 = 9;
+    pub const SCHEMA_VERSION: u32 = 10;
 
     /// Create a new `SignerState` instance.
     /// This will create a new SQLite database at the given path
@@ -723,7 +738,7 @@ impl SignerDb {
         Ok(())
     }
 
-    /// Migrate from schema 9 to schema 9
+    /// Migrate from schema 8 to schema 9
     fn schema_9_migration(tx: &Transaction) -> Result<(), DBError> {
         if Self::get_schema_version(tx)? >= 9 {
             // no migration necessary
@@ -737,6 +752,20 @@ impl SignerDb {
         Ok(())
     }
 
+    /// Migrate from schema 9 to schema 10
+    fn schema_10_migration(tx: &Transaction) -> Result<(), DBError> {
+        if Self::get_schema_version(tx)? >= 10 {
+            // no migration necessary
+            return Ok(());
+        }
+
+        for statement in SCHEMA_10.iter() {
+            tx.execute_batch(statement)?;
+        }
+
+        Ok(())
+    }
+
     /// Register custom scalar functions used by the database
     fn register_scalar_functions(&self) -> Result<(), DBError> {
         // Register helper function for determining if a block is a tenure change transaction
@@ -779,7 +808,8 @@ impl SignerDb {
                 6 => Self::schema_7_migration(&sql_tx)?,
                 7 => Self::schema_8_migration(&sql_tx)?,
                 8 => Self::schema_9_migration(&sql_tx)?,
-                9 => break,
+                9 => Self::schema_10_migration(&sql_tx)?,
+                10 => break,
                 x => return Err(DBError::Other(format!(
                     "Database schema is newer than supported by this binary. Expected version = {}, Database version = {x}",
                     Self::SCHEMA_VERSION,
@@ -911,18 +941,20 @@ impl SignerDb {
     pub fn insert_burn_block(
         &mut self,
         burn_hash: &BurnchainHeaderHash,
+        consensus_hash: &ConsensusHash,
         burn_height: u64,
         received_time: &SystemTime,
     ) -> Result<(), DBError> {
         let received_ts = received_time
             .duration_since(std::time::UNIX_EPOCH)
             .map_err(|e| DBError::Other(format!("Bad system time: {e}")))?
             .as_secs();
-        debug!("Inserting burn block info"; "burn_block_height" => burn_height, "burn_hash" => %burn_hash, "received" => received_ts);
+        debug!("Inserting burn block info"; "burn_block_height" => burn_height, "burn_hash" => %burn_hash, "received" => received_ts, "ch" => %consensus_hash);
         self.db.execute(
-            "INSERT OR REPLACE INTO burn_blocks (block_hash, block_height, received_time) VALUES (?1, ?2, ?3)",
+            "INSERT OR REPLACE INTO burn_blocks (block_hash, consensus_hash, block_height, received_time) VALUES (?1, ?2, ?3, ?4)",
             params![
                 burn_hash,
+                consensus_hash,
                 u64_to_sql(burn_height)?,
                 u64_to_sql(received_ts)?,
             ],
@@ -947,6 +979,23 @@ impl SignerDb {
         Ok(Some(receive_time))
     }
 
+    /// Get timestamp (epoch seconds) at which a burn block was received over the event dispatcheer by this signer
+    /// if that burn block has been received.
+    pub fn get_burn_block_receive_time_ch(
+        &self,
+        ch: &ConsensusHash,
+    ) -> Result<Option<u64>, DBError> {
+        let query = "SELECT received_time FROM burn_blocks WHERE consensus_hash = ? LIMIT 1";
+        let Some(receive_time_i64) = query_row::<i64, _>(&self.db, query, &[ch])? else {
+            return Ok(None);
+        };
+        let receive_time = u64::try_from(receive_time_i64).map_err(|e| {
+            error!("Failed to parse db received_time as u64: {e}");
+            DBError::Corruption
+        })?;
+        Ok(Some(receive_time))
+    }
+
     /// Insert or replace a block into the database.
     /// Preserves the `broadcast` column if replacing an existing block.
     pub fn insert_block(&mut self, block_info: &BlockInfo) -> Result<(), DBError> {
@@ -1497,12 +1546,14 @@ mod tests {
         let db_path = tmp_db_path();
         let mut db = SignerDb::new(db_path).expect("Failed to create signer db");
         let test_burn_hash = BurnchainHeaderHash([10; 32]);
+        let test_consensus_hash = ConsensusHash([13; 20]);
         let stime = SystemTime::now();
         let time_to_epoch = stime
             .duration_since(SystemTime::UNIX_EPOCH)
             .unwrap()
             .as_secs();
-        db.insert_burn_block(&test_burn_hash, 10, &stime).unwrap();
+        db.insert_burn_block(&test_burn_hash, &test_consensus_hash, 10, &stime)
+            .unwrap();
 
         let stored_time = db
             .get_burn_block_receive_time(&test_burn_hash)

stacks-signer/src/tests/chainstate.rs

@@ -251,7 +251,12 @@ fn reorg_timing_testing(
     let sortition_time = SystemTime::UNIX_EPOCH
         + Duration::from_secs(block_info_1.proposed_time + sortition_timing_secs);
     signer_db
-        .insert_burn_block(&view.cur_sortition.burn_block_hash, 3, &sortition_time)
+        .insert_burn_block(
+            &view.cur_sortition.burn_block_hash,
+            &view.cur_sortition.consensus_hash,
+            3,
+            &sortition_time,
+        )
         .unwrap();
 
     let MockServerClient {
@@ -385,10 +390,11 @@ fn check_block_proposal_timeout() {
 
     // Ensure we have a burn height to compare against
     let burn_hash = view.cur_sortition.burn_block_hash;
+    let consensus_hash = view.cur_sortition.consensus_hash;
     let burn_height = 1;
     let received_time = SystemTime::now();
     signer_db
-        .insert_burn_block(&burn_hash, burn_height, &received_time)
+        .insert_burn_block(&burn_hash, &consensus_hash, burn_height, &received_time)
         .unwrap();
 
     view.check_proposal(
@@ -456,10 +462,11 @@ fn check_sortition_timeout() {
     };
     // Ensure we have a burn height to compare against
     let burn_hash = sortition.burn_block_hash;
+    let consensus_hash = sortition.consensus_hash;
     let burn_height = 1;
     let received_time = SystemTime::now();
     signer_db
-        .insert_burn_block(&burn_hash, burn_height, &received_time)
+        .insert_burn_block(&burn_hash, &consensus_hash, burn_height, &received_time)
         .unwrap();
 
     std::thread::sleep(Duration::from_secs(1));