Merge pull request #5962 from jferrant/feat/signer-state-conflict-resolution-strategies

Feat/signer state conflict resolution strategies

Author: wileyj

libsigner/src/events.rs

@@ -192,8 +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 is the signer set to which the message belongs (either 0 or 1)
-    SignerMessages(u32, Vec<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
@@ -518,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()
         {
@@ -536,12 +543,21 @@ impl<T: SignerEventTrait> TryFrom<StackerDBChunksEvent> for SignerEvent<T> {
                 return Err(EventError::UnrecognizedStackerDBContract(event.contract_id));
             };
             // signer-XXX-YYY boot contract
-            let signer_messages: Vec<T> = event
+            let messages: Vec<(StacksPublicKey, T)> = event
                 .modified_slots
                 .iter()
-                .filter_map(|chunk| read_next::<T, _>(&mut &chunk.data[..]).ok())
+                .filter_map(|chunk| {
+                    Some((
+                        chunk.recover_pk().ok()?,
+                        read_next::<T, _>(&mut &chunk.data[..]).ok()?,
+                    ))
+                })
                 .collect();
-            SignerEvent::SignerMessages(signer_set, signer_messages)
+            SignerEvent::SignerMessages {
+                signer_set,
+                messages,
+                received_time,
+            }
         } else {
             return Err(EventError::UnrecognizedStackerDBContract(event.contract_id));
         };

libsigner/src/tests/mod.rs

@@ -20,15 +20,15 @@ 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;
 use blockstack_lib::chainstate::nakamoto::{NakamotoBlock, NakamotoBlockHeader};
 use blockstack_lib::chainstate::stacks::boot::SIGNERS_NAME;
 use blockstack_lib::chainstate::stacks::events::StackerDBChunksEvent;
 use blockstack_lib::util_lib::boot::boot_code_id;
-use clarity::types::chainstate::{ConsensusHash, StacksBlockId, TrieHash};
+use clarity::types::chainstate::{ConsensusHash, StacksBlockId, StacksPublicKey, TrieHash};
 use clarity::util::hash::Sha512Trunc256Sum;
 use clarity::util::secp256k1::MessageSignature;
 use clarity::vm::types::QualifiedContractIdentifier;
@@ -142,6 +142,13 @@ fn test_simple_signer() {
         chunks.push(chunk_event);
     }
 
+    chunks.sort_by(|ev1, ev2| {
+        ev1.modified_slots[0]
+            .slot_id
+            .partial_cmp(&ev2.modified_slots[0].slot_id)
+            .unwrap()
+    });
+
     let thread_chunks = chunks.clone();
 
     // simulate a node that's trying to push data
@@ -177,23 +184,44 @@ fn test_simple_signer() {
     sleep_ms(5000);
     let accepted_events = running_signer.stop().unwrap();
 
-    chunks.sort_by(|ev1, ev2| {
-        ev1.modified_slots[0]
-            .slot_id
-            .partial_cmp(&ev2.modified_slots[0].slot_id)
-            .unwrap()
-    });
-
     let sent_events: Vec<SignerEvent<SignerMessage>> = chunks
         .iter()
         .map(|chunk| {
             let msg = chunk.modified_slots[0].data.clone();
+            let pubkey = chunk.modified_slots[0]
+                .recover_pk()
+                .expect("Faield to recover public key of slot");
             let signer_message = read_next::<SignerMessage, _>(&mut &msg[..]).unwrap();
-            SignerEvent::SignerMessages(0, vec![signer_message])
+            SignerEvent::SignerMessages {
+                signer_set: 0,
+                messages: vec![(pubkey, signer_message)],
+                received_time: SystemTime::now(),
+            }
         })
         .collect();
 
-    assert_eq!(sent_events, accepted_events);
+    for (sent_event, accepted_event) in sent_events.iter().zip(accepted_events.iter()) {
+        let SignerEvent::SignerMessages {
+            signer_set,
+            messages,
+            received_time,
+        } = sent_event
+        else {
+            panic!("BUG: should not have sent anything but a signer message");
+        };
+        let SignerEvent::SignerMessages {
+            signer_set: accepted_signer_set,
+            messages: accepted_messages,
+            received_time: accepted_time,
+        } = accepted_event
+        else {
+            panic!("BUG: should not have accepted anything but a signer message");
+        };
+
+        assert_eq!(signer_set, accepted_signer_set);
+        assert_eq!(messages, accepted_messages);
+        assert_ne!(received_time, accepted_time);
+    }
     mock_stacks_node.join().unwrap();
 }
 

libsigner/src/v0/messages.rs

@@ -24,6 +24,7 @@
 //! and the `SignerMessage` enum.
 
 use std::fmt::{Debug, Display};
+use std::hash::{Hash, Hasher};
 use std::io::{Read, Write};
 use std::marker::PhantomData;
 use std::net::{SocketAddr, TcpListener, TcpStream};
@@ -573,7 +574,7 @@ pub enum StateMachineUpdateContent {
 }
 
 /// Message for update the Signer State infos
-#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, Eq, Hash)]
 pub enum StateMachineUpdateMinerState {
     /// There is an active miner
     ActiveMiner {