CRC: Fix use of weight vs entry in capitulate_miner_view and various cleanups

Signed-off-by: Jacinta Ferrant <[email protected]>

Author: jferrant

libsigner/src/events.rs

@@ -192,9 +192,14 @@ pub enum SignerEvent<T: SignerEventTrait> {
     /// The `Vec<T>` will contain any signer messages made by the miner.
     MinerMessages(Vec<T>),
     /// The signer messages for other signers and miners to observe
-    /// - The `u32` represents the signer set to which the message belongs (either 0 or 1).
-    /// - Each message of type `T` is paired with the `StacksPublicKey` of the slot from which it was retrieved.
-    SignerMessages(u32, Vec<(StacksPublicKey, T)>),
+    SignerMessages {
+        /// The signer set to which the message belongs (either 0 or 1)
+        signer_set: u32,
+        /// Each message of type `T` is paired with the `StacksPublicKey` of the slot from which it was retreived
+        messages: Vec<(StacksPublicKey, T)>,
+        /// the time at which this event was received by the signer's event processor
+        received_time: SystemTime,
+    },
     /// A new block proposal validation response from the node
     BlockValidationResponse(BlockValidateResponse),
     /// Status endpoint request
@@ -519,6 +524,7 @@ impl<T: SignerEventTrait> TryFrom<StackerDBChunksEvent> for SignerEvent<T> {
     type Error = EventError;
 
     fn try_from(event: StackerDBChunksEvent) -> Result<Self, Self::Error> {
+        let received_time = SystemTime::now();
         let signer_event = if event.contract_id.name.as_str() == MINERS_NAME
             && event.contract_id.is_boot()
         {
@@ -547,7 +553,11 @@ impl<T: SignerEventTrait> TryFrom<StackerDBChunksEvent> for SignerEvent<T> {
                     ))
                 })
                 .collect();
-            SignerEvent::SignerMessages(signer_set, messages)
+            SignerEvent::SignerMessages {
+                signer_set,
+                messages,
+                received_time,
+            }
         } else {
             return Err(EventError::UnrecognizedStackerDBContract(event.contract_id));
         };

libsigner/src/tests/mod.rs

@@ -20,7 +20,7 @@ use std::fmt::Debug;
 use std::io::{Read, Write};
 use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
 use std::sync::mpsc::{channel, Receiver, Sender};
-use std::time::Duration;
+use std::time::{Duration, SystemTime};
 use std::{mem, thread};
 
 use blockstack_lib::chainstate::nakamoto::signer_set::NakamotoSigners;
@@ -192,7 +192,11 @@ fn test_simple_signer() {
                 .recover_pk()
                 .expect("Faield to recover public key of slot");
             let signer_message = read_next::<SignerMessage, _>(&mut &msg[..]).unwrap();
-            SignerEvent::SignerMessages(0, vec![(pubkey, signer_message)])
+            SignerEvent::SignerMessages {
+                signer_set: 0,
+                messages: vec![(pubkey, signer_message)],
+                received_time: SystemTime::now(),
+            }
         })
         .collect();
 

libsigner/src/v0/messages.rs

@@ -574,7 +574,7 @@ pub enum StateMachineUpdateContent {
 }
 
 /// Message for update the Signer State infos
-#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, Eq)]
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, Eq, Hash)]
 pub enum StateMachineUpdateMinerState {
     /// There is an active miner
     ActiveMiner {
@@ -595,29 +595,6 @@ pub enum StateMachineUpdateMinerState {
     NoValidMiner,
 }
 
-impl Hash for StateMachineUpdateMinerState {
-    fn hash<H: Hasher>(&self, state: &mut H) {
-        match self {
-            StateMachineUpdateMinerState::ActiveMiner {
-                current_miner_pkh,
-                tenure_id,
-                parent_tenure_id,
-                parent_tenure_last_block,
-                parent_tenure_last_block_height,
-            } => {
-                current_miner_pkh.hash(state);
-                tenure_id.hash(state);
-                parent_tenure_id.hash(state);
-                parent_tenure_last_block.hash(state);
-                parent_tenure_last_block_height.hash(state);
-            }
-            StateMachineUpdateMinerState::NoValidMiner => {
-                0.hash(state);
-            }
-        }
-    }
-}
-
 impl StateMachineUpdate {
     /// Construct a StateMachineUpdate message, checking to ensure that the
     /// supplied content is supported by the supplied protocol versions.

stacks-signer/src/signerdb.rs

@@ -367,6 +367,10 @@ CREATE INDEX IF NOT EXISTS blocks_consensus_hash_status_height ON blocks (consen
 CREATE INDEX IF NOT EXISTS blocks_reward_cycle_state on blocks (reward_cycle, state);
 "#;
 
+static CREATE_INDEXES_11: &str = r#"
+CREATE INDEX IF NOT EXISTS signer_state_machine_updates_reward_cycle_received_time ON signer_state_machine_updates (reward_cycle, received_time ASC);
+"#;
+
 static CREATE_SIGNER_STATE_TABLE: &str = "
 CREATE TABLE IF NOT EXISTS signer_states (
     reward_cycle INTEGER PRIMARY KEY,
@@ -524,6 +528,7 @@ CREATE TABLE IF NOT EXISTS signer_state_machine_updates (
     signer_addr TEXT NOT NULL,
     reward_cycle INTEGER NOT NULL,
     state_update TEXT NOT NULL,
+    received_time TEXT NOT NULL,
     PRIMARY KEY (signer_addr, reward_cycle)
 ) STRICT;"#;
 
@@ -604,6 +609,7 @@ static SCHEMA_10: &[&str] = &[
 
 static SCHEMA_11: &[&str] = &[
     CREATE_SIGNER_STATE_MACHINE_UPDATES_TABLE,
+    CREATE_INDEXES_11,
     "INSERT INTO db_config (version) VALUES (11);",
 ];
 
@@ -1355,7 +1361,12 @@ impl SignerDb {
         reward_cycle: u64,
         address: &StacksAddress,
         update: &StateMachineUpdate,
+        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();
         let update_str =
             serde_json::to_string(&update).expect("Unable to serialize state machine update");
         debug!("Inserting update.";
@@ -1364,10 +1375,11 @@ impl SignerDb {
             "active_signer_protocol_version" => update.active_signer_protocol_version,
             "local_supported_signer_protocol_version" => update.local_supported_signer_protocol_version
         );
-        self.db.execute("INSERT OR REPLACE INTO signer_state_machine_updates (signer_addr, reward_cycle, state_update) VALUES (?1, ?2, ?3)", params![
+        self.db.execute("INSERT OR REPLACE INTO signer_state_machine_updates (signer_addr, reward_cycle, state_update, received_time) VALUES (?1, ?2, ?3, ?4)", params![
             address.to_string(),
             u64_to_sql(reward_cycle)?,
-            update_str
+            update_str,
+            u64_to_sql(received_ts)?
         ])?;
         Ok(())
     }
@@ -2450,12 +2462,17 @@ mod tests {
             "The database should be empty for reward_cycle {reward_cycle_1}"
         );
 
-        db.insert_state_machine_update(reward_cycle_1, &address_1, &update_1)
+        db.insert_state_machine_update(reward_cycle_1, &address_1, &update_1, &SystemTime::now())
             .expect("Unable to insert block into db");
-        db.insert_state_machine_update(reward_cycle_1, &address_2, &update_2)
-            .expect("Unable to insert block into db");
-        db.insert_state_machine_update(reward_cycle_1 + 1, &address_3, &update_3)
+        db.insert_state_machine_update(reward_cycle_1, &address_2, &update_2, &SystemTime::now())
             .expect("Unable to insert block into db");
+        db.insert_state_machine_update(
+            reward_cycle_1 + 1,
+            &address_3,
+            &update_3,
+            &SystemTime::now(),
+        )
+        .expect("Unable to insert block into db");
 
         let updates = db.get_signer_state_machine_updates(reward_cycle_1).unwrap();
         assert_eq!(updates.len(), 2);
@@ -2464,7 +2481,7 @@ mod tests {
         assert_eq!(updates.get(&address_2), Some(&update_2));
         assert_eq!(updates.get(&address_3), None);
 
-        db.insert_state_machine_update(reward_cycle_1, &address_2, &update_3)
+        db.insert_state_machine_update(reward_cycle_1, &address_2, &update_3, &SystemTime::now())
             .expect("Unable to insert block into db");
         let updates = db.get_signer_state_machine_updates(reward_cycle_1).unwrap();
         assert_eq!(updates.len(), 2);

stacks-signer/src/v0/signer.rs

@@ -17,7 +17,7 @@ use std::fmt::Debug;
 use std::sync::mpsc::Sender;
 #[cfg(any(test, feature = "testing"))]
 use std::sync::LazyLock;
-use std::time::{Duration, Instant};
+use std::time::{Duration, Instant, SystemTime};
 
 use blockstack_lib::chainstate::nakamoto::{NakamotoBlock, NakamotoBlockHeader};
 use blockstack_lib::net::api::postblock_proposal::{
@@ -206,7 +206,9 @@ impl SignerTrait<SignerMessage> for Signer {
             | Some(SignerEvent::NewBlock { .. })
             | Some(SignerEvent::StatusCheck)
             | None => None,
-            Some(SignerEvent::SignerMessages(msg_parity, ..)) => Some(u64::from(*msg_parity) % 2),
+            Some(SignerEvent::SignerMessages { signer_set, .. }) => {
+                Some(u64::from(*signer_set) % 2)
+            }
         };
         let other_signer_parity = (self.reward_cycle + 1) % 2;
         if event_parity == Some(other_signer_parity) {
@@ -246,7 +248,11 @@ impl SignerTrait<SignerMessage> for Signer {
                     sortition_state,
                 )
             }
-            SignerEvent::SignerMessages(_, messages) => {
+            SignerEvent::SignerMessages {
+                received_time,
+                messages,
+                ..
+            } => {
                 debug!(
                     "{self}: Received {} messages from the other signers",
                     messages.len()
@@ -260,7 +266,11 @@ impl SignerTrait<SignerMessage> for Signer {
                             sortition_state,
                         ),
                         SignerMessage::StateMachineUpdate(update) => {
-                            self.handle_state_machine_update(signer_public_key, update);
+                            self.handle_state_machine_update(
+                                signer_public_key,
+                                update,
+                                received_time,
+                            );
                         }
                         _ => {}
                     }
@@ -589,13 +599,16 @@ impl Signer {
         &mut self,
         signer_public_key: &Secp256k1PublicKey,
         update: &StateMachineUpdate,
+        received_time: &SystemTime,
     ) {
         let address = StacksAddress::p2pkh(self.mainnet, signer_public_key);
         // Store the state machine update so we can reload it if we crash
-        if let Err(e) =
-            self.signer_db
-                .insert_state_machine_update(self.reward_cycle, &address, update)
-        {
+        if let Err(e) = self.signer_db.insert_state_machine_update(
+            self.reward_cycle,
+            &address,
+            update,
+            received_time,
+        ) {
             warn!("{self}: Failed to update global state in signerdb: {e}");
         }
         self.global_state_evaluator

stacks-signer/src/v0/signer_state.rs

@@ -104,7 +104,7 @@ impl GlobalStateEvaluator {
 
             let entry = burn_blocks.entry(burn_block).or_insert_with(|| 0);
             *entry += weight;
-            if *entry >= self.total_weight * 7 / 10 {
+            if self.reached_agreement(*entry) {
                 return Some((burn_block, burn_block_height));
             }
         }
@@ -116,8 +116,7 @@ impl GlobalStateEvaluator {
     /// NOTE: do not call this function unless the evaluator has already been updated with the current local state
     pub fn determine_global_state(&self) -> Option<SignerStateMachine> {
         let active_signer_protocol_version = self.determine_active_signer_protocol_version()?;
-        let (global_burn_block, _) = self.determine_global_burn_view()?;
-        let mut miner_views = HashMap::new();
+        let mut state_views = HashMap::new();
         for (address, update) in &self.address_updates {
             let Some(weight) = self.address_weights.get(address) else {
                 continue;
@@ -128,20 +127,18 @@ impl GlobalStateEvaluator {
                 current_miner,
                 ..
             } = &update.content;
-
-            if *burn_block != global_burn_block {
-                continue;
-            }
-
-            let entry = miner_views.entry(current_miner).or_insert_with(|| 0);
+            let state_machine = SignerStateMachine {
+                burn_block: *burn_block,
+                burn_block_height: *burn_block_height,
+                current_miner: current_miner.into(),
+                active_signer_protocol_version,
+            };
+            let entry = state_views
+                .entry(state_machine.clone())
+                .or_insert_with(|| 0);
             *entry += weight;
-            if *entry >= self.total_weight * 7 / 10 {
-                return Some(SignerStateMachine {
-                    burn_block: *burn_block,
-                    burn_block_height: *burn_block_height,
-                    current_miner: current_miner.into(),
-                    active_signer_protocol_version,
-                });
+            if self.reached_agreement(*entry) {
+                return Some(state_machine);
             }
         }
         None
@@ -169,7 +166,7 @@ impl GlobalStateEvaluator {
 
             let entry = miner_views.entry(current_miner).or_insert_with(|| 0);
             *entry += weight;
-            if *entry >= self.total_weight * 7 / 10 {
+            if self.reached_agreement(*entry) {
                 return Some(current_miner.clone());
             }
         }
@@ -217,7 +214,7 @@ impl GlobalStateEvaluator {
                 let nmb_blocks = signerdb
                     .get_globally_accepted_block_count_in_tenure(&tenure_id)
                     .unwrap_or(0);
-                if nmb_blocks > 0 || *weight >= self.total_weight * 7 / 10 {
+                if nmb_blocks > 0 || self.reached_agreement(*entry) {
                     return Some(current_miner.clone());
                 }
             }
@@ -237,12 +234,18 @@ impl GlobalStateEvaluator {
         self.address_updates.insert(address, update);
         true
     }
+
+    /// Check if the supplied vote weight crosses the global agreement threshold.
+    /// Returns true if it has, false otherwise.
+    fn reached_agreement(&self, vote_weight: u32) -> bool {
+        vote_weight >= self.total_weight * 7 / 10
+    }
 }
 
 /// A signer state machine view. This struct can
 ///  be used to encode the local signer's view or
 ///  the global view.
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Eq, Hash)]
 pub struct SignerStateMachine {
     /// The tip burn block (i.e., the latest bitcoin block) seen by this signer
     pub burn_block: ConsensusHash,
@@ -254,7 +257,7 @@ pub struct SignerStateMachine {
     pub active_signer_protocol_version: u64,
 }
 
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Eq, Hash)]
 /// Enum for capturing the signer state machine's view of who
 ///  should be the active miner and what their tenure should be
 ///  built on top of.

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

@@ -235,7 +235,12 @@ impl StackerDBListener {
             }) else {
                 continue;
             };
-            let SignerEvent::SignerMessages(signer_set, messages) = signer_event else {
+            let SignerEvent::SignerMessages {
+                signer_set,
+                messages,
+                ..
+            } = signer_event
+            else {
                 debug!("StackerDBListener: Received signer event other than a signer message. Ignoring.");
                 continue;
             };