Merge branch 'develop' into feat/clarity-wasm-develop

Author: obycode

.github/workflows/bitcoin-tests.yml

@@ -93,6 +93,7 @@ jobs:
           - tests::signer::v0::bitcoind_forking_test
           - tests::signer::v0::multiple_miners
           - tests::signer::v0::mock_sign_epoch_25
+          - tests::signer::v0::signer_set_rollover
           - tests::signer::v0::miner_forking
           - tests::nakamoto_integrations::stack_stx_burn_op_integration_test
           - tests::nakamoto_integrations::check_block_heights
@@ -101,6 +102,7 @@ jobs:
           - tests::nakamoto_integrations::check_block_info
           - tests::nakamoto_integrations::check_block_info_rewards
           - tests::nakamoto_integrations::continue_tenure_extend
+          - tests::nakamoto_integrations::mock_mining
           - tests::nakamoto_integrations::multiple_miners
           # Do not run this one until we figure out why it fails in CI
           # - tests::neon_integrations::bitcoin_reorg_flap

.vscode/extensions.json

@@ -2,7 +2,7 @@
     "recommendations": [
         "rust-lang.rust-analyzer",
         "vadimcn.vscode-lldb",
-        "serayuzgur.crates",
+        "fill-labs.dependi",
         "editorconfig.editorconfig",
     ]
 }

libsigner/src/runloop.rs

@@ -262,7 +262,7 @@ impl<
 
         // start receiving events and doing stuff with them
         let runloop_thread = thread::Builder::new()
-            .name("signer_runloop".to_string())
+            .name(format!("signer_runloop:{}", bind_addr.port()))
             .stack_size(THREAD_STACK_SIZE)
             .spawn(move || {
                 signer_loop.main_loop(event_recv, command_receiver, result_sender, stop_signaler)

stacks-signer/src/chainstate.rs

@@ -303,11 +303,12 @@ impl SortitionsView {
             let last_in_tenure = signer_db
                 .get_last_signed_block_in_tenure(&block.header.consensus_hash)
                 .map_err(|e| ClientError::InvalidResponse(e.to_string()))?;
-            if last_in_tenure.is_some() {
+            if let Some(last_in_tenure) = last_in_tenure {
                 warn!(
                     "Miner block proposal contains a tenure change, but we've already signed a block in this tenure. Considering proposal invalid.";
                     "proposed_block_consensus_hash" => %block.header.consensus_hash,
                     "proposed_block_signer_sighash" => %block.header.signer_signature_hash(),
+                    "last_in_tenure_signer_sighash" => %last_in_tenure.block.header.signer_signature_hash(),
                 );
                 return Ok(false);
             }

stacks-signer/src/lib.rs

@@ -100,6 +100,8 @@ pub struct SpawnedSigner<S: Signer<T> + Send, T: SignerEventTrait> {
     pub cmd_send: Sender<RunLoopCommand>,
     /// The result receiver for interacting with the running signer
     pub res_recv: Receiver<Vec<SignerResult>>,
+    /// The spawned signer's config
+    pub config: GlobalConfig,
     /// Phantom data for the signer type
     _phantom: std::marker::PhantomData<S>,
 }
@@ -136,7 +138,7 @@ impl<S: Signer<T> + Send + 'static, T: SignerEventTrait + 'static> SpawnedSigner
         {
             crate::monitoring::start_serving_monitoring_metrics(config.clone()).ok();
         }
-        let runloop = RunLoop::new(config);
+        let runloop = RunLoop::new(config.clone());
         let mut signer: RunLoopSigner<S, T> =
             libsigner::Signer::new(runloop, ev, cmd_recv, res_send);
         let running_signer = signer.spawn(endpoint).expect("Failed to spawn signer");
@@ -145,6 +147,7 @@ impl<S: Signer<T> + Send + 'static, T: SignerEventTrait + 'static> SpawnedSigner
             cmd_send,
             res_recv,
             _phantom: std::marker::PhantomData,
+            config,
         }
     }
 }

stackslib/src/chainstate/nakamoto/coordinator/mod.rs

@@ -546,7 +546,7 @@ pub fn load_nakamoto_reward_set<U: RewardSetProvider>(
            "burnchain_height" => %anchor_block_sn.block_height);
 
     let reward_set = provider.get_reward_set_nakamoto(
-        prepare_end_height.saturating_sub(1),
+        prepare_end_height,
         chain_state,
         burnchain,
         sort_db,

stackslib/src/chainstate/nakamoto/coordinator/tests.rs

@@ -44,6 +44,7 @@ use crate::chainstate::nakamoto::coordinator::load_nakamoto_reward_set;
 use crate::chainstate::nakamoto::miner::NakamotoBlockBuilder;
 use crate::chainstate::nakamoto::signer_set::NakamotoSigners;
 use crate::chainstate::nakamoto::test_signers::TestSigners;
+use crate::chainstate::nakamoto::test_stall::*;
 use crate::chainstate::nakamoto::tests::get_account;
 use crate::chainstate::nakamoto::tests::node::TestStacker;
 use crate::chainstate::nakamoto::{
@@ -2453,3 +2454,89 @@ pub fn simple_nakamoto_coordinator_10_extended_tenures_10_sortitions() -> TestPe
 fn test_nakamoto_coordinator_10_tenures_and_extensions_10_blocks() {
     simple_nakamoto_coordinator_10_extended_tenures_10_sortitions();
 }
+
+#[test]
+fn process_next_nakamoto_block_deadlock() {
+    let private_key = StacksPrivateKey::from_seed(&[2]);
+    let addr = StacksAddress::p2pkh(false, &StacksPublicKey::from_private(&private_key));
+
+    let num_stackers: u32 = 4;
+    let mut signing_key_seed = num_stackers.to_be_bytes().to_vec();
+    signing_key_seed.extend_from_slice(&[1, 1, 1, 1]);
+    let signing_key = StacksPrivateKey::from_seed(signing_key_seed.as_slice());
+    let test_stackers = (0..num_stackers)
+        .map(|index| TestStacker {
+            signer_private_key: signing_key.clone(),
+            stacker_private_key: StacksPrivateKey::from_seed(&index.to_be_bytes()),
+            amount: u64::MAX as u128 - 10000,
+            pox_addr: Some(PoxAddress::Standard(
+                StacksAddress::new(
+                    C32_ADDRESS_VERSION_TESTNET_SINGLESIG,
+                    Hash160::from_data(&index.to_be_bytes()),
+                ),
+                Some(AddressHashMode::SerializeP2PKH),
+            )),
+            max_amount: None,
+        })
+        .collect::<Vec<_>>();
+    let test_signers = TestSigners::new(vec![signing_key]);
+    let mut pox_constants = TestPeerConfig::default().burnchain.pox_constants;
+    pox_constants.reward_cycle_length = 10;
+    pox_constants.v2_unlock_height = 21;
+    pox_constants.pox_3_activation_height = 26;
+    pox_constants.v3_unlock_height = 27;
+    pox_constants.pox_4_activation_height = 28;
+
+    let mut boot_plan = NakamotoBootPlan::new(function_name!())
+        .with_test_stackers(test_stackers.clone())
+        .with_test_signers(test_signers.clone())
+        .with_private_key(private_key);
+    boot_plan.pox_constants = pox_constants;
+
+    info!("Creating peer");
+
+    let mut peer = boot_plan.boot_into_nakamoto_peer(vec![], None);
+    let mut sortition_db = peer.sortdb().reopen().unwrap();
+    let (chainstate, _) = &mut peer
+        .stacks_node
+        .as_mut()
+        .unwrap()
+        .chainstate
+        .reopen()
+        .unwrap();
+
+    enable_process_block_stall();
+
+    let miner_thread = std::thread::spawn(move || {
+        info!("  -------------------------------   MINING TENURE");
+        let (block, burn_height, ..) =
+            peer.single_block_tenure(&private_key, |_| {}, |_| {}, |_| true);
+        info!("  -------------------------------   TENURE MINED");
+    });
+
+    // Wait a bit, to ensure the miner has reached the stall
+    std::thread::sleep(std::time::Duration::from_secs(10));
+
+    // Lock the sortdb
+    info!("  -------------------------------   TRYING TO LOCK THE SORTDB");
+    let sort_tx = sortition_db.tx_begin().unwrap();
+    info!("  -------------------------------   SORTDB LOCKED");
+
+    // Un-stall the block processing
+    disable_process_block_stall();
+
+    // Wait a bit, to ensure the tenure will have grabbed any locks it needs
+    std::thread::sleep(std::time::Duration::from_secs(10));
+
+    // Lock the chainstate db
+    info!("  -------------------------------   TRYING TO LOCK THE CHAINSTATE");
+    let chainstate_tx = chainstate.chainstate_tx_begin().unwrap();
+
+    info!("  -------------------------------   SORTDB AND CHAINSTATE LOCKED");
+    drop(chainstate_tx);
+    drop(sort_tx);
+    info!("  -------------------------------   MAIN THREAD FINISHED");
+
+    // Wait for the blocker and miner threads to finish
+    miner_thread.join().unwrap();
+}

stackslib/src/chainstate/nakamoto/mod.rs

@@ -270,6 +270,31 @@ lazy_static! {
     ];
 }
 
+#[cfg(test)]
+mod test_stall {
+    pub static TEST_PROCESS_BLOCK_STALL: std::sync::Mutex<Option<bool>> =
+        std::sync::Mutex::new(None);
+
+    pub fn stall_block_processing() {
+        if *TEST_PROCESS_BLOCK_STALL.lock().unwrap() == Some(true) {
+            // Do an extra check just so we don't log EVERY time.
+            warn!("Block processing is stalled due to testing directive.");
+            while *TEST_PROCESS_BLOCK_STALL.lock().unwrap() == Some(true) {
+                std::thread::sleep(std::time::Duration::from_millis(10));
+            }
+            info!("Block processing is no longer stalled due to testing directive.");
+        }
+    }
+
+    pub fn enable_process_block_stall() {
+        TEST_PROCESS_BLOCK_STALL.lock().unwrap().replace(true);
+    }
+
+    pub fn disable_process_block_stall() {
+        TEST_PROCESS_BLOCK_STALL.lock().unwrap().replace(false);
+    }
+}
+
 /// Trait for common MARF getters between StacksDBConn and StacksDBTx
 pub trait StacksDBIndexed {
     fn get(&mut self, tip: &StacksBlockId, key: &str) -> Result<Option<String>, DBError>;
@@ -1722,6 +1747,9 @@ impl NakamotoChainState {
         canonical_sortition_tip: &SortitionId,
         dispatcher_opt: Option<&'a T>,
     ) -> Result<Option<StacksEpochReceipt>, ChainstateError> {
+        #[cfg(test)]
+        test_stall::stall_block_processing();
+
         let nakamoto_blocks_db = stacks_chain_state.nakamoto_blocks_db();
         let Some((next_ready_block, block_size)) =
             nakamoto_blocks_db.next_ready_nakamoto_block(stacks_chain_state.db())?
@@ -1992,22 +2020,14 @@ impl NakamotoChainState {
             next_ready_block.header.consensus_hash
         );
 
-        // set stacks block accepted
-        let mut sort_tx = sort_db.tx_handle_begin(canonical_sortition_tip)?;
-        sort_tx.set_stacks_block_accepted(
-            &next_ready_block.header.consensus_hash,
-            &next_ready_block.header.block_hash(),
-            next_ready_block.header.chain_length,
-        )?;
-
         // this will panic if the Clarity commit fails.
         clarity_commit.commit();
         chainstate_tx.commit()
-            .unwrap_or_else(|e| {
-                error!("Failed to commit chainstate transaction after committing Clarity block. The chainstate database is now corrupted.";
-                       "error" => ?e);
-                panic!()
-            });
+        .unwrap_or_else(|e| {
+            error!("Failed to commit chainstate transaction after committing Clarity block. The chainstate database is now corrupted.";
+            "error" => ?e);
+            panic!()
+        });
 
         // as a separate transaction, mark this block as processed.
         // This is done separately so that the staging blocks DB, which receives writes
@@ -2019,6 +2039,22 @@ impl NakamotoChainState {
 
         let signer_bitvec = (&next_ready_block).header.pox_treatment.clone();
 
+        // set stacks block accepted
+        let mut sort_tx = sort_db.tx_handle_begin(canonical_sortition_tip)?;
+        sort_tx.set_stacks_block_accepted(
+            &next_ready_block.header.consensus_hash,
+            &next_ready_block.header.block_hash(),
+            next_ready_block.header.chain_length,
+        )?;
+
+        sort_tx
+            .commit()
+            .unwrap_or_else(|e| {
+                error!("Failed to commit sortition db transaction after committing chainstate and clarity block. The chainstate database is now corrupted.";
+                       "error" => ?e);
+                panic!()
+            });
+
         // announce the block, if we're connected to an event dispatcher
         if let Some(dispatcher) = dispatcher_opt {
             let block_event = (
@@ -2045,14 +2081,6 @@ impl NakamotoChainState {
             );
         }
 
-        sort_tx
-            .commit()
-            .unwrap_or_else(|e| {
-                error!("Failed to commit sortition db transaction after committing chainstate and clarity block. The chainstate database is now corrupted.";
-                       "error" => ?e);
-                panic!()
-            });
-
         Ok(Some(receipt))
     }
 

testnet/stacks-node/src/event_dispatcher.rs

@@ -6,7 +6,9 @@ use std::thread::sleep;
 use std::time::Duration;
 
 use async_h1::client;
+use async_std::future::timeout;
 use async_std::net::TcpStream;
+use async_std::task;
 use clarity::vm::analysis::contract_interface_builder::build_contract_interface;
 use clarity::vm::costs::ExecutionCost;
 use clarity::vm::events::{FTEventType, NFTEventType, STXEventType};
@@ -138,6 +140,7 @@ pub struct MinedNakamotoBlockEvent {
     pub signer_signature_hash: Sha512Trunc256Sum,
     pub tx_events: Vec<TransactionEvent>,
     pub signer_bitvec: String,
+    pub signer_signature: Vec<MessageSignature>,
 }
 
 impl InnerStackerDBChannel {
@@ -318,27 +321,33 @@ impl EventObserver {
         };
 
         let backoff = Duration::from_millis((1.0 * 1_000.0) as u64);
+        let connection_timeout = Duration::from_secs(5);
 
         loop {
             let body = body.clone();
             let mut req = Request::new(Method::Post, url.clone());
             req.append_header("Content-Type", "application/json");
             req.set_body(body);
 
-            let response = async_std::task::block_on(async {
-                let stream = match TcpStream::connect(self.endpoint.clone()).await {
-                    Ok(stream) => stream,
-                    Err(err) => {
-                        warn!("Event dispatcher: connection failed  - {:?}", err);
-                        return None;
-                    }
-                };
+            let response = task::block_on(async {
+                let stream =
+                    match timeout(connection_timeout, TcpStream::connect(&self.endpoint)).await {
+                        Ok(Ok(stream)) => stream,
+                        Ok(Err(err)) => {
+                            warn!("Event dispatcher: connection failed  - {:?}", err);
+                            return None;
+                        }
+                        Err(_) => {
+                            error!("Event dispatcher: connection attempt timed out");
+                            return None;
+                        }
+                    };
 
                 match client::connect(stream, req).await {
                     Ok(response) => Some(response),
                     Err(err) => {
                         warn!("Event dispatcher: rpc invocation failed  - {:?}", err);
-                        return None;
+                        None
                     }
                 }
             });
@@ -1261,6 +1270,7 @@ impl EventDispatcher {
             tx_events,
             miner_signature: block.header.miner_signature.clone(),
             signer_signature_hash: block.header.signer_signature_hash(),
+            signer_signature: block.header.signer_signature.clone(),
             signer_bitvec,
         })
         .unwrap();