feat: construct replay set after fork

Author: hstove

stacks-signer/src/client/stacks_client.rs

@@ -452,6 +452,22 @@ impl StacksClient {
         })
     }
 
+    /// Get the sortition info for a given consensus hash
+    pub fn get_sortition_by_consensus_hash(
+        &self,
+        consensus_hash: &ConsensusHash,
+    ) -> Result<SortitionInfo, ClientError> {
+        let path = self.sortition_by_consensus_hash_path(consensus_hash);
+        let response = self.stacks_node_client.get(&path).send()?;
+        if !response.status().is_success() {
+            return Err(ClientError::RequestFailure(response.status()));
+        }
+        let sortition_info = response.json::<Vec<SortitionInfo>>()?;
+        sortition_info.get(0).cloned().ok_or_else(|| {
+            ClientError::InvalidResponse("No sortition info found for given consensus hash".into())
+        })
+    }
+
     /// Get the current peer info data from the stacks node
     pub fn get_peer_info(&self) -> Result<PeerInfo, ClientError> {
         debug!("StacksClient: Getting peer info");
@@ -725,6 +741,14 @@ impl StacksClient {
         format!("{}{RPC_SORTITION_INFO_PATH}", self.http_origin)
     }
 
+    fn sortition_by_consensus_hash_path(&self, consensus_hash: &ConsensusHash) -> String {
+        format!(
+            "{}{RPC_SORTITION_INFO_PATH}/consensus/{}",
+            self.http_origin,
+            consensus_hash.to_hex()
+        )
+    }
+
     fn tenure_forking_info_path(&self, start: &ConsensusHash, stop: &ConsensusHash) -> String {
         format!(
             "{}{RPC_TENURE_FORKING_INFO_PATH}/{}/{}",

stacks-signer/src/tests/signer_state.rs

@@ -220,6 +220,7 @@ fn determine_global_states() {
         current_miner: (&current_miner).into(),
         active_signer_protocol_version: local_supported_signer_protocol_version, // a majority of signers are saying they support version the same local_supported_signer_protocol_version, so update it here...
         tx_replay_state: false,
+        tx_replay_set: None,
     };
 
     assert_eq!(
@@ -265,6 +266,7 @@ fn determine_global_states() {
         current_miner: (&new_miner).into(),
         active_signer_protocol_version: local_supported_signer_protocol_version, // a majority of signers are saying they support version the same local_supported_signer_protocol_version, so update it here...
         tx_replay_state: false,
+        tx_replay_set: None,
     };
 
     // Let's tip the scales over to a different miner

stacks-signer/src/v0/signer.rs

@@ -369,7 +369,7 @@ impl SignerTrait<SignerMessage> for Signer {
                     });
                 self.local_state_machine
                     .bitcoin_block_arrival(&self.signer_db, stacks_client, &self.proposal_config, Some(NewBurnBlock {
-                        height: *burn_height,
+                        burn_block_height: *burn_height,
                         consensus_hash: *consensus_hash,
                     }))
                     .unwrap_or_else(|e| error!("{self}: failed to update local state machine for latest bitcoin block arrival"; "err" => ?e));

stacks-signer/src/v0/signer_state.rs

@@ -18,6 +18,7 @@ use std::time::{Duration, UNIX_EPOCH};
 
 use blockstack_lib::chainstate::burn::ConsensusHashExtensions;
 use blockstack_lib::chainstate::nakamoto::{NakamotoBlock, NakamotoBlockHeader};
+use blockstack_lib::chainstate::stacks::{StacksTransaction, TransactionPayload};
 use clarity::types::chainstate::StacksAddress;
 use libsigner::v0::messages::{
     MessageSlotID, SignerMessage, StateMachineUpdate as StateMachineUpdateMessage,
@@ -151,6 +152,7 @@ impl GlobalStateEvaluator {
                 current_miner: current_miner.into(),
                 active_signer_protocol_version,
                 tx_replay_state: false,
+                tx_replay_set: None,
             };
             let entry = state_views
                 .entry(state_machine.clone())
@@ -250,6 +252,8 @@ pub struct SignerStateMachine {
     /// Whether or not we're in a tx replay state
     /// TODO: just a placeholder for now
     pub tx_replay_state: bool,
+    /// Transaction replay set
+    pub tx_replay_set: Option<Vec<StacksTransaction>>,
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Eq, Hash)]
@@ -324,7 +328,7 @@ pub enum StateMachineUpdate {
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct NewBurnBlock {
     /// The height of the new burn block
-    pub height: u64,
+    pub burn_block_height: u64,
     /// The hash of the new burn block
     pub consensus_hash: ConsensusHash,
 }
@@ -389,6 +393,7 @@ impl LocalStateMachine {
             current_miner: MinerState::NoValidMiner,
             active_signer_protocol_version: SUPPORTED_SIGNER_PROTOCOL_VERSION,
             tx_replay_state: false,
+            tx_replay_set: None,
         }
     }
 
@@ -687,10 +692,10 @@ impl LocalStateMachine {
         // TODO: test only, remove
         match expected_burn_block.clone() {
             Some(expected_burn_block) => {
-                if expected_burn_block.height > 230 {
+                if expected_burn_block.burn_block_height > 230 {
                     info!(
                         "---- bitcoin_block_arrival {} {} ----",
-                        expected_burn_block.height, expected_burn_block.consensus_hash
+                        expected_burn_block.burn_block_height, expected_burn_block.consensus_hash
                     );
                 }
             }
@@ -712,7 +717,8 @@ impl LocalStateMachine {
                         match expected_burn_block {
                             None => expected_burn_block = Some(pending_burn_block),
                             Some(ref expected) => {
-                                if pending_burn_block.height > expected.height {
+                                if pending_burn_block.burn_block_height > expected.burn_block_height
+                                {
                                     expected_burn_block = Some(pending_burn_block);
                                 }
                             }
@@ -728,36 +734,80 @@ impl LocalStateMachine {
         let next_burn_block_height = peer_info.burn_block_height;
         let next_burn_block_hash = peer_info.pox_consensus;
         let mut fork_detected = prior_state_machine.tx_replay_state;
+        let mut tx_replay_set = prior_state_machine.tx_replay_set.clone();
 
         if let Some(expected_burn_block) = expected_burn_block {
             // If the next height is less than the expected height, we need to wait.
             // OR if the next height is the same, but with a different hash, we need to wait.
-            if next_burn_block_height < expected_burn_block.height || {
-                next_burn_block_height == expected_burn_block.height
+            if next_burn_block_height < expected_burn_block.burn_block_height || {
+                next_burn_block_height == expected_burn_block.burn_block_height
                     && next_burn_block_hash != expected_burn_block.consensus_hash
             } {
                 let err_msg = format!(
                     "Node has not processed the next burn block ({}) yet",
-                    expected_burn_block.height
+                    expected_burn_block.burn_block_height
                 );
                 *self = Self::Pending {
                     update: StateMachineUpdate::BurnBlock(expected_burn_block),
                     prior: prior_state_machine,
                 };
                 return Err(ClientError::InvalidResponse(err_msg).into());
             }
-            if expected_burn_block.height <= prior_state_machine.burn_block_height
+            if expected_burn_block.burn_block_height <= prior_state_machine.burn_block_height
                 && expected_burn_block.consensus_hash != prior_state_machine.burn_block
+                // TODO: handle fork while still in replay
+                && tx_replay_set.is_none()
             {
                 fork_detected = true;
                 info!("---- Signer State: Possible fork! ----";
-                    "expected_burn_block.height" => expected_burn_block.height,
+                    "expected_burn_block.height" => expected_burn_block.burn_block_height,
                     "expected_burn_block.hash" => %expected_burn_block.consensus_hash,
                     "next_burn_block_height" => next_burn_block_height,
                     "next_burn_block_hash" => %next_burn_block_hash,
                     "prior_state_machine.burn_block_height" => prior_state_machine.burn_block_height,
                     "prior_state_machine.burn_block" => %prior_state_machine.burn_block,
                 );
+                // Determine the tenures that were forked
+                let mut sortition_info =
+                    client.get_sortition_by_consensus_hash(&prior_state_machine.burn_block)?;
+                let last_forked_tenure = prior_state_machine.burn_block;
+                let mut first_forked_tenure = prior_state_machine.burn_block;
+                let mut forked_tenures = vec![(
+                    prior_state_machine.burn_block,
+                    prior_state_machine.burn_block_height,
+                )];
+                while sortition_info.burn_block_height > expected_burn_block.burn_block_height {
+                    let Some(stacks_parent_ch) = sortition_info.stacks_parent_ch else {
+                        info!("No stacks parent ch found for sortition info";
+                            "sortition_info" => ?sortition_info,
+                        );
+                        break;
+                    };
+                    sortition_info = client.get_sortition_by_consensus_hash(&stacks_parent_ch)?;
+                    first_forked_tenure = sortition_info.consensus_hash;
+                    forked_tenures.push((stacks_parent_ch, sortition_info.burn_block_height));
+                }
+                let fork_info =
+                    client.get_tenure_forking_info(&first_forked_tenure, &last_forked_tenure)?;
+                let forked_txs = fork_info
+                    .iter()
+                    .flat_map(|fork_info| {
+                        fork_info
+                            .nakamoto_blocks
+                            .iter()
+                            .flat_map(|blocks| blocks.iter())
+                            .flat_map(|block| block.txs.iter())
+                    })
+                    .cloned()
+                    .filter(|tx| match tx.payload {
+                        // Don't include Coinbase, TenureChange, or PoisonMicroblock transactions
+                        TransactionPayload::TenureChange(..)
+                        | TransactionPayload::Coinbase(..)
+                        | TransactionPayload::PoisonMicroblock(..) => false,
+                        _ => true,
+                    })
+                    .collect::<Vec<_>>();
+                tx_replay_set = Some(forked_txs);
             }
         }
 
@@ -802,6 +852,7 @@ impl LocalStateMachine {
             current_miner: miner_state,
             active_signer_protocol_version: prior_state_machine.active_signer_protocol_version,
             tx_replay_state: fork_detected,
+            tx_replay_set,
         });
 
         if prior_state != *self {
@@ -847,6 +898,7 @@ impl LocalStateMachine {
                 current_miner: current_miner.into(),
                 active_signer_protocol_version,
                 tx_replay_state: false,
+                tx_replay_set: None,
             });
             // Because we updated our active signer protocol version, update local_update so its included in the subsequent evaluations
             let update: Result<StateMachineUpdateMessage, _> = (&*self).try_into();
@@ -880,6 +932,7 @@ impl LocalStateMachine {
                 current_miner: (&new_miner).into(),
                 active_signer_protocol_version,
                 tx_replay_state: false,
+                tx_replay_set: None,
             });
         }
     }

stackslib/src/chainstate/stacks/transaction.rs

@@ -14,6 +14,7 @@
 // You should have received a copy of the GNU General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+use std::hash::{Hash, Hasher};
 use std::io;
 use std::io::prelude::*;
 use std::io::{Read, Write};
@@ -683,6 +684,14 @@ impl StacksTransaction {
     }
 }
 
+impl Hash for StacksTransaction {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.txid().hash(state);
+    }
+}
+
+impl Eq for StacksTransaction {}
+
 impl StacksMessageCodec for StacksTransaction {
     fn consensus_serialize<W: Write>(&self, fd: &mut W) -> Result<(), codec_error> {
         write_next(fd, &(self.version as u8))?;

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

@@ -2666,6 +2666,15 @@ fn bitcoind_forking_test() {
 #[ignore]
 /// Trigger a Bitcoin fork and ensure that the signer
 /// both detects the fork and moves into a tx replay state
+///
+/// The test flow is:
+///
+/// - Mine 10 tenures after epoch 3
+/// - Include a STX transfer in the 10th tenure
+/// - Trigger a Bitcoin fork (3 blocks)
+/// - Verify that the signer moves into tx replay state
+/// - Verify that the signer correctly includes the stx transfer
+///   in the tx replay set
 fn tx_replay_forking_test() {
     if env::var("BITCOIND_TEST") != Ok("1".into()) {
         return;
@@ -2731,7 +2740,7 @@ fn tx_replay_forking_test() {
 
     let burn_blocks = test_observer::get_burn_blocks();
     let forked_blocks = burn_blocks.iter().rev().take(2).collect::<Vec<_>>();
-    let start_tenure: ConsensusHash = hex_bytes(
+    let last_forked_tenure: ConsensusHash = hex_bytes(
         &forked_blocks[0]
             .get("consensus_hash")
             .unwrap()
@@ -2741,7 +2750,7 @@ fn tx_replay_forking_test() {
     .unwrap()
     .as_slice()
     .into();
-    let end_tenure: ConsensusHash = hex_bytes(
+    let first_forked_tenure: ConsensusHash = hex_bytes(
         &forked_blocks[1]
             .get("consensus_hash")
             .unwrap()
@@ -2754,7 +2763,7 @@ fn tx_replay_forking_test() {
 
     let tip = get_chain_info(&signer_test.running_nodes.conf);
     // Make a transfer tx (this will get forked)
-    signer_test
+    let (txid, _) = signer_test
         .submit_transfer_tx(&sender_sk, send_fee, send_amt)
         .unwrap();
 
@@ -2797,8 +2806,7 @@ fn tx_replay_forking_test() {
 
     let fork_info = signer_test
         .stacks_client
-        // .get_tenure_forking_info(&start_tenure, &end_tenure)
-        .get_tenure_forking_info(&end_tenure, &start_tenure)
+        .get_tenure_forking_info(&first_forked_tenure, &last_forked_tenure)
         .unwrap();
 
     info!("---- Fork info: {fork_info:?} ----");
@@ -2846,7 +2854,6 @@ fn tx_replay_forking_test() {
             },
         )
         .unwrap();
-        // signer_test.check_signer_states_normal_missed_sortition();
     }
 
     let post_fork_1_nonce = get_account(&http_origin, &sender_addr).nonce;
@@ -2867,6 +2874,14 @@ fn tx_replay_forking_test() {
         match state {
             LocalStateMachine::Initialized(signer_state_machine) => {
                 assert!(signer_state_machine.tx_replay_state);
+                let Some(tx_replay_set) = signer_state_machine.tx_replay_set else {
+                    panic!(
+                        "Signer state machine is in tx replay state, but tx replay set is not set"
+                    );
+                };
+                info!("---- Tx replay set: {:?} ----", tx_replay_set);
+                assert_eq!(tx_replay_set.len(), 1);
+                assert_eq!(tx_replay_set[0].txid().to_hex(), txid);
             }
             _ => {
                 panic!("Signer state is not in the initialized state");