Fixes discovered while bringing up the pre-launch testnet

libsigner/src/events.rs

@@ -50,11 +50,22 @@ use wsts::state_machine::signer;
 use crate::http::{decode_http_body, decode_http_request};
 use crate::{EventError, SignerMessage};
 
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+/// BlockProposal sent to signers
+pub struct BlockProposalSigners {
+    /// The block itself
+    pub block: NakamotoBlock,
+    /// The burn height the block is mined during
+    pub burn_height: u64,
+    /// The reward cycle the block is mined during
+    pub reward_cycle: u64,
+}
+
 /// Event enum for newly-arrived signer subscribed events
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub enum SignerEvent {
     /// The miner proposed blocks for signers to observe and sign
-    ProposedBlocks(Vec<NakamotoBlock>),
+    ProposedBlocks(Vec<BlockProposalSigners>),
     /// 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<SignerMessage>),
@@ -64,6 +75,26 @@ pub enum SignerEvent {
     StatusCheck,
 }
 
+impl StacksMessageCodec for BlockProposalSigners {
+    fn consensus_serialize<W: Write>(&self, fd: &mut W) -> Result<(), CodecError> {
+        self.block.consensus_serialize(fd)?;
+        self.burn_height.consensus_serialize(fd)?;
+        self.reward_cycle.consensus_serialize(fd)?;
+        Ok(())
+    }
+
+    fn consensus_deserialize<R: Read>(fd: &mut R) -> Result<Self, CodecError> {
+        let block = NakamotoBlock::consensus_deserialize(fd)?;
+        let burn_height = u64::consensus_deserialize(fd)?;
+        let reward_cycle = u64::consensus_deserialize(fd)?;
+        Ok(BlockProposalSigners {
+            block,
+            burn_height,
+            reward_cycle,
+        })
+    }
+}
+
 /// Trait to implement a stop-signaler for the event receiver thread.
 /// The caller calls `send()` and the event receiver loop (which lives in a separate thread) will
 /// terminate.
@@ -195,11 +226,15 @@ impl EventStopSignaler for SignerStopSignaler {
             // We need to send actual data to trigger the event receiver
             let body = "Yo. Shut this shit down!".to_string();
             let req = format!(
-                "POST /shutdown HTTP/1.0\r\nContent-Length: {}\r\n\r\n{}",
-                &body.len(),
+                "POST /shutdown HTTP/1.1\r\nHost: {}\r\nConnection: close\r\nContent-Length: {}\r\nContent-Type: text/plain\r\n\r\n{}",
+                self.local_addr,
+                body.len(),
                 body
             );
-            stream.write_all(req.as_bytes()).unwrap();
+            match stream.write_all(req.as_bytes()) {
+                Err(e) => error!("Failed to send shutdown request: {}", e),
+                _ => (),
+            };
         }
     }
 }
@@ -337,10 +372,10 @@ fn process_stackerdb_event(
         .map_err(|e| EventError::Deserialize(format!("Could not decode body to JSON: {:?}", &e)))?;
 
     let signer_event = if event.contract_id == boot_code_id(MINERS_NAME, is_mainnet) {
-        let blocks: Vec<NakamotoBlock> = event
+        let blocks: Vec<BlockProposalSigners> = event
             .modified_slots
             .iter()
-            .filter_map(|chunk| read_next::<NakamotoBlock, _>(&mut &chunk.data[..]).ok())
+            .filter_map(|chunk| read_next::<BlockProposalSigners, _>(&mut &chunk.data[..]).ok())
             .collect();
         SignerEvent::ProposedBlocks(blocks)
     } else if event.contract_id.name.to_string().starts_with(SIGNERS_NAME)

libsigner/src/http.rs

@@ -238,12 +238,12 @@ pub fn run_http_request<S: Read + Write>(
 
     let req_txt = if let Some(content_type) = content_type {
         format!(
-            "{} {} HTTP/1.0\r\nHost: {}\r\nConnection: close\r\nContent-Type: {}\r\n{}User-Agent: libsigner/0.1\r\nAccept: */*\r\n\r\n",
+            "{} {} HTTP/1.1\r\nHost: {}\r\nConnection: close\r\nContent-Type: {}\r\n{}User-Agent: libsigner/0.1\r\nAccept: */*\r\n\r\n",
             verb, path, host, content_type, content_length_hdr
         )
     } else {
         format!(
-            "{} {} HTTP/1.0\r\nHost: {}\r\nConnection: close\r\n{}User-Agent: libsigner/0.1\r\nAccept: */*\r\n\r\n",
+            "{} {} HTTP/1.1\r\nHost: {}\r\nConnection: close\r\n{}User-Agent: libsigner/0.1\r\nAccept: */*\r\n\r\n",
             verb, path, host, content_length_hdr
         )
     };

libsigner/src/libsigner.rs

@@ -45,7 +45,8 @@ mod session;
 
 pub use crate::error::{EventError, RPCError};
 pub use crate::events::{
-    EventReceiver, EventStopSignaler, SignerEvent, SignerEventReceiver, SignerStopSignaler,
+    BlockProposalSigners, EventReceiver, EventStopSignaler, SignerEvent, SignerEventReceiver,
+    SignerStopSignaler,
 };
 pub use crate::messages::{
     BlockRejection, BlockResponse, RejectCode, SignerMessage, BLOCK_MSG_ID, TRANSACTIONS_MSG_ID,

libsigner/src/messages.rs

@@ -823,6 +823,27 @@ pub enum BlockResponse {
     Rejected(BlockRejection),
 }
 
+impl std::fmt::Display for BlockResponse {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            BlockResponse::Accepted(a) => {
+                write!(
+                    f,
+                    "BlockAccepted: signer_sighash = {}, signature = {}",
+                    a.0, a.1
+                )
+            }
+            BlockResponse::Rejected(r) => {
+                write!(
+                    f,
+                    "BlockRejected: signer_sighash = {}, code = {}, reason = {}",
+                    r.reason_code, r.reason, r.signer_signature_hash
+                )
+            }
+        }
+    }
+}
+
 impl BlockResponse {
     /// Create a new accepted BlockResponse for the provided block signer signature hash and signature
     pub fn accepted(hash: Sha512Trunc256Sum, sig: Signature) -> Self {

libsigner/src/tests/mod.rs

@@ -135,7 +135,12 @@ fn test_simple_signer() {
 
             let ev = &thread_chunks[num_sent];
             let body = serde_json::to_string(ev).unwrap();
-            let req = format!("POST /stackerdb_chunks HTTP/1.0\r\nConnection: close\r\nContent-Length: {}\r\n\r\n{}", &body.len(), body);
+            let req = format!(
+                "POST /stackerdb_chunks HTTP/1.1\r\nHost: {}\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{}",
+                endpoint,
+                &body.len(),
+                body
+            );
             debug!("Send:\n{}", &req);
 
             sock.write_all(req.as_bytes()).unwrap();
@@ -188,13 +193,16 @@ fn test_status_endpoint() {
                 return;
             }
         };
-        let req = "GET /status HTTP/1.0\r\nConnection: close\r\n\r\n";
+        let req = format!(
+            "GET /status HTTP/1.1\r\nHost: {}\r\nConnection: close\r\n\r\n",
+            endpoint
+        );
 
         sock.write_all(req.as_bytes()).unwrap();
         let mut buf = [0; 128];
         let _ = sock.read(&mut buf).unwrap();
         let res_str = std::str::from_utf8(&buf).unwrap();
-        let expected_status_res = "HTTP/1.0 200 OK\r\n";
+        let expected_status_res = "HTTP/1.1 200 OK\r\n";
         assert_eq!(expected_status_res, &res_str[..expected_status_res.len()]);
         sock.flush().unwrap();
     });

stacks-signer/src/coordinator.rs

@@ -69,6 +69,9 @@ impl From<PublicKeys> for CoordinatorSelector {
     }
 }
 
+/// Whether or not to rotate to new coordinators in `update_coordinator`
+const ROTATE_COORDINATORS: bool = false;
+
 impl CoordinatorSelector {
     /// Update the coordinator id
     fn update_coordinator(&mut self, new_coordinator_ids: Vec<u32>) {
@@ -81,20 +84,24 @@ impl CoordinatorSelector {
                 .coordinator_ids
                 .first()
                 .expect("FATAL: No registered signers");
-            if new_coordinator_id == self.coordinator_id {
+            if ROTATE_COORDINATORS && new_coordinator_id == self.coordinator_id {
                 // If the newly selected coordinator is the same as the current and we have more than one available, advance immediately to the next
                 if self.coordinator_ids.len() > 1 {
                     new_index = new_index.saturating_add(1);
                 }
             }
             new_index
         } else {
-            let mut new_index = self.coordinator_index.saturating_add(1);
-            if new_index == self.coordinator_ids.len() {
-                // We have exhausted all potential coordinators. Go back to the start
-                new_index = 0;
+            if ROTATE_COORDINATORS {
+                let mut new_index = self.coordinator_index.saturating_add(1);
+                if new_index == self.coordinator_ids.len() {
+                    // We have exhausted all potential coordinators. Go back to the start
+                    new_index = 0;
+                }
+                new_index
+            } else {
+                self.coordinator_index
             }
-            new_index
         };
         self.coordinator_id = *self
             .coordinator_ids
@@ -136,7 +143,7 @@ impl CoordinatorSelector {
         )
     }
 
-    /// Calculate the ordered list of coordinator ids by comparing the provided public keys against the pox consensus hash
+    /// Calculate the ordered list of coordinator ids by comparing the provided public keys
     pub fn calculate_coordinator_ids(
         public_keys: &PublicKeys,
         pox_consensus_hash: &ConsensusHash,

stacks-signer/src/runloop.rs

@@ -229,7 +229,7 @@ impl RunLoop {
     }
 
     /// Refresh signer configuration for a specific reward cycle
-    fn refresh_signer_config(&mut self, reward_cycle: u64) {
+    fn refresh_signer_config(&mut self, reward_cycle: u64, current: bool) {
         let reward_index = reward_cycle % 2;
         let mut needs_refresh = false;
         if let Some(signer) = self.stacks_signers.get_mut(&reward_index) {
@@ -266,7 +266,12 @@ impl RunLoop {
                     .insert(reward_index, Signer::from(new_signer_config));
                 debug!("Reward cycle #{reward_cycle} Signer #{signer_id} initialized.");
             } else {
-                warn!("Signer is not registered for reward cycle {reward_cycle}. Waiting for confirmed registration...");
+                // TODO: Update `current` here once the signer binary is tracking its own latest burnchain/stacks views.
+                if current {
+                    warn!("Signer is not registered for the current reward cycle ({reward_cycle}). Waiting for confirmed registration...");
+                } else {
+                    debug!("Signer is not registered for reward cycle {reward_cycle}. Waiting for confirmed registration...");
+                }
             }
         }
     }
@@ -275,11 +280,19 @@ impl RunLoop {
     /// Note: this will trigger DKG if required
     fn refresh_signers(&mut self, current_reward_cycle: u64) -> Result<(), ClientError> {
         let next_reward_cycle = current_reward_cycle.saturating_add(1);
-        self.refresh_signer_config(current_reward_cycle);
-        self.refresh_signer_config(next_reward_cycle);
+        self.refresh_signer_config(current_reward_cycle, true);
+        self.refresh_signer_config(next_reward_cycle, false);
         // TODO: do not use an empty consensus hash
         let pox_consensus_hash = ConsensusHash::empty();
-        for signer in self.stacks_signers.values_mut() {
+        let mut to_delete = Vec::new();
+        for (idx, signer) in &mut self.stacks_signers {
+            if signer.reward_cycle < current_reward_cycle {
+                debug!("{signer}: Signer's tenure has completed.");
+                // We don't really need this state, but it's useful for debugging
+                signer.state = SignerState::TenureCompleted;
+                to_delete.push(*idx);
+                continue;
+            }
             let old_coordinator_id = signer.coordinator_selector.get_coordinator().0;
             let updated_coordinator_id = signer
                 .coordinator_selector
@@ -302,6 +315,11 @@ impl RunLoop {
                 })?;
             }
         }
+        for i in to_delete.into_iter() {
+            if let Some(signer) = self.stacks_signers.remove(&i) {
+                info!("{signer}: Tenure has completed. Removing signer from runloop.",);
+            }
+        }
         if self.stacks_signers.is_empty() {
             info!("Signer is not registered for the current reward cycle ({current_reward_cycle}) or next reward cycle ({next_reward_cycle}). Waiting for confirmed registration...");
             self.state = State::Uninitialized;
@@ -357,6 +375,26 @@ impl SignerRunLoop<Vec<OperationResult>, RunLoopCommand> for RunLoop {
             error!("Failed to refresh signers: {e}. Signer may have an outdated view of the network. Attempting to process event anyway.");
         }
         for signer in self.stacks_signers.values_mut() {
+            if signer.state == SignerState::TenureCompleted {
+                warn!("{signer}: Signer's tenure has completed. This signer should have been cleaned up during refresh.");
+                continue;
+            }
+            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::ProposedBlocks(_)) => None,
+                Some(SignerEvent::SignerMessages(msg_parity, ..)) => {
+                    Some(u64::from(msg_parity) % 2)
+                }
+                Some(SignerEvent::StatusCheck) => None,
+                None => None,
+            };
+            let other_signer_parity = (signer.reward_cycle + 1) % 2;
+            if event_parity == Some(other_signer_parity) {
+                continue;
+            }
+
             if let Err(e) = signer.process_event(
                 &self.stacks_client,
                 event.as_ref(),