Merge pull request #6229 from kantai/fix/bitcoin-reorg-extend

Fix: miner coordinator behavior in reorg/extend

Author: kantai

stacks-signer/src/v0/signer.rs

@@ -482,7 +482,7 @@ impl Signer {
                             );
                             #[cfg(any(test, feature = "testing"))]
                             if self.test_skip_block_broadcast(b) {
-                                return;
+                                continue;
                             }
                             self.handle_post_block(stacks_client, b);
                         }
@@ -491,7 +491,7 @@ impl Signer {
                                 Ok(epoch) => epoch,
                                 Err(e) => {
                                     warn!("{self}: Failed to determine node epoch. Cannot mock sign: {e}");
-                                    return;
+                                    continue;
                                 }
                             };
                             info!("{self}: received a mock block proposal.";

stackslib/src/chainstate/nakamoto/mod.rs

@@ -2851,29 +2851,6 @@ impl NakamotoChainState {
         Self::get_block_header_nakamoto(chainstate_conn.sqlite(), &block_id)
     }
 
-    /// DO NOT USE IN CONSENSUS CODE.  Different nodes can have different blocks for the same
-    /// tenure.
-    ///
-    /// Get the highest block in a given tenure (identified by its consensus hash).
-    /// Ties will be broken by timestamp.
-    ///
-    /// Used to verify that a signer-submitted block proposal builds atop the highest known block
-    /// in the given tenure, regardless of which fork it's on.
-    pub fn get_highest_known_block_header_in_tenure(
-        db: &Connection,
-        consensus_hash: &ConsensusHash,
-    ) -> Result<Option<StacksHeaderInfo>, ChainstateError> {
-        // see if we have a nakamoto block in this tenure
-        let qry = "SELECT * FROM nakamoto_block_headers WHERE consensus_hash = ?1 ORDER BY block_height DESC, timestamp DESC LIMIT 1";
-        let args = params![consensus_hash];
-        if let Some(header) = query_row(db, qry, args)? {
-            return Ok(Some(header));
-        }
-
-        // see if this is an epoch2 header. If it exists, then there will only be one.
-        Ok(StacksChainState::get_stacks_block_header_info_by_consensus_hash(db, consensus_hash)?)
-    }
-
     /// DO NOT USE IN CONSENSUS CODE.  Different nodes can have different blocks for the same
     /// tenure.
     ///

stackslib/src/net/relay.rs

@@ -2497,6 +2497,12 @@ impl Relayer {
                 let tx = self.stacker_dbs.tx_begin(config.clone())?;
                 for sync_result in sync_results.into_iter() {
                     for chunk in sync_result.chunks_to_store.into_iter() {
+                        if let Some(event_list) = all_events.get_mut(&sync_result.contract_id) {
+                            event_list.push(chunk.clone());
+                        } else {
+                            all_events.insert(sync_result.contract_id.clone(), vec![chunk.clone()]);
+                        }
+
                         let md = chunk.get_slot_metadata();
                         if let Err(e) = tx.try_replace_chunk(&sc, &md, &chunk.data) {
                             if matches!(e, Error::StaleChunk { .. }) {
@@ -2525,11 +2531,6 @@ impl Relayer {
                             debug!("Stored chunk"; "stackerdb_contract_id" => %sync_result.contract_id, "slot_id" => md.slot_id, "slot_version" => md.slot_version);
                         }
 
-                        if let Some(event_list) = all_events.get_mut(&sync_result.contract_id) {
-                            event_list.push(chunk.clone());
-                        } else {
-                            all_events.insert(sync_result.contract_id.clone(), vec![chunk.clone()]);
-                        }
                         let msg = StacksMessageType::StackerDBPushChunk(StackerDBPushChunkData {
                             contract_id: sc.clone(),
                             rc_consensus_hash: rc_consensus_hash.clone(),

stackslib/src/net/stackerdb/db.rs

@@ -25,7 +25,6 @@ use stacks_common::util::get_epoch_time_secs;
 use stacks_common::util::hash::Sha512Trunc256Sum;
 use stacks_common::util::secp256k1::MessageSignature;
 
-use super::StackerDBEventDispatcher;
 use crate::net::stackerdb::{StackerDBConfig, StackerDBTx, StackerDBs, STACKERDB_INV_MAX};
 use crate::net::{Error as net_error, StackerDBChunkData};
 use crate::util_lib::db::{
@@ -396,20 +395,6 @@ impl StackerDBTx<'_> {
         Ok(())
     }
 
-    /// Try to upload a chunk to the StackerDB instance, notifying
-    ///  and subscribed listeners via the `dispatcher`
-    pub fn put_chunk<ED: StackerDBEventDispatcher>(
-        self,
-        contract: &QualifiedContractIdentifier,
-        chunk: StackerDBChunkData,
-        dispatcher: &ED,
-    ) -> Result<(), net_error> {
-        self.try_replace_chunk(contract, &chunk.get_slot_metadata(), &chunk.data)?;
-        self.commit()?;
-        dispatcher.new_stackerdb_chunks(contract.clone(), vec![chunk]);
-        Ok(())
-    }
-
     /// Add or replace a chunk for a given reward cycle, if it is valid
     /// Otherwise, this errors out with Error::StaleChunk
     pub fn try_replace_chunk(

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

@@ -420,8 +420,9 @@ impl SignerCoordinator {
 
                     // Check if a new Stacks block has arrived in the parent tenure
                     let highest_in_tenure =
-                        NakamotoChainState::get_highest_known_block_header_in_tenure(
-                            &mut chain_state.index_conn(),
+                        NakamotoChainState::find_highest_known_block_header_in_tenure(
+                            &chain_state,
+                            &sortdb,
                             &parent_tenure_header.consensus_hash,
                         )?
                         .ok_or(NakamotoNodeError::UnexpectedChainState)?;
@@ -436,7 +437,10 @@ impl SignerCoordinator {
                         };
                         return Ok(stored_block.signer_signature);
                     } else if highest_stacks_block_id != parent_block_id {
-                        info!("SignCoordinator: Exiting due to new stacks tip");
+                        info!("SignCoordinator: Exiting due to new stacks tip";
+                              "new_block_hash" => %highest_in_tenure.anchored_header.block_hash(),
+                              "new_block_height" => %highest_in_tenure.anchored_header.height(),
+                        );
                         return Err(NakamotoNodeError::StacksTipChanged);
                     }
 

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

@@ -1388,18 +1388,18 @@ impl<S: Signer<T> + Send + 'static, T: SignerEventTrait + 'static> SignerTest<Sp
             SignerSlotID(0), // We are just reading so again, don't care about index.
             SignerDb::new(":memory:").unwrap(),
         );
-        let latest_msgs = StackerDB::get_messages(
+        let mut latest_msgs = StackerDB::get_messages(
             stackerdb
                 .get_session_mut(&MessageSlotID::BlockResponse)
                 .expect("Failed to get BlockResponse stackerdb session"),
             &[slot_id],
         )
         .expect("Failed to get message from stackerdb");
-        let latest_msg = latest_msgs.last().unwrap();
+        let latest_msg = latest_msgs.pop().unwrap();
         let SignerMessage::BlockResponse(block_response) = latest_msg else {
             panic!("Latest message from slot #{slot_id} isn't a block acceptance");
         };
-        block_response.clone()
+        block_response
     }
 
     /// Get the latest block acceptance from the given slot

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

@@ -17349,6 +17349,13 @@ fn bitcoin_reorg_extended_tenure() {
     info!("Bitcoin fork triggered"; "ch" => %before_fork, "btc_height" => burn_block_height);
     info!("Chain info before fork: {:?}", get_chain_info(&conf_1));
 
+    // Make sure signers don't perform block broadcast for the next bits:
+    //  we want to ensure that the *miner* is the one broadcast blocks,
+    //  because when we stall p2p broadcast, we don't want to accidentally
+    //  stall the miner in the situation where they produce block A, signers broadcast it,
+    //  we initiate the stall, and then the miner attempts to broadcast A.
+    stacks_signer::v0::tests::TEST_SKIP_BLOCK_BROADCAST.set(true);
+
     let mut after_fork = get_chain_info(&conf_1).pox_consensus;
     wait_for(60, || {
         after_fork = get_chain_info(&conf_1).pox_consensus;
@@ -17371,7 +17378,8 @@ fn bitcoin_reorg_extended_tenure() {
     //  so that we can ensure all the signers approve the proposal
     //  before it gets accepted by stacks-nodes
     TEST_P2P_BROADCAST_STALL.set(true);
-    stacks_signer::v0::tests::TEST_SKIP_BLOCK_BROADCAST.set(true);
+
+    info!("Stalled broadcast, submitting a contract call!");
 
     // the signer signature hash is the same as the block header hash.
     // we use the latest_signer_sighash to make sure we're getting block responses for the
@@ -17392,11 +17400,18 @@ fn bitcoin_reorg_extended_tenure() {
     let rc = miners.signer_test.get_current_reward_cycle();
     let slot_ids = miners.signer_test.get_signer_indices(rc);
     let mut block_responses: Vec<_> = vec![];
+
     wait_for(60, || {
         block_responses = slot_ids
             .iter()
             .filter_map(|slot_id| {
                 let latest_br = miners.signer_test.get_latest_block_response(slot_id.0);
+                info!(
+                    "[{}] Checking response for {}. accepted = {}",
+                    slot_id.0,
+                    latest_br.get_signer_signature_hash(),
+                    latest_br.as_block_accepted().is_some()
+                );
                 if latest_br.get_signer_signature_hash() != latest_signer_sighash {
                     Some(latest_br)
                 } else {