Merge remote-tracking branch 'origin/fix/burn-view' into feat/burn-view-fix-burn-checks

Author: kantai

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

@@ -13,10 +13,11 @@
 //
 // 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::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
 #[cfg(test)]
 use std::sync::LazyLock;
 use std::thread;
-use std::thread::JoinHandle;
 use std::time::{Duration, Instant};
 
 use clarity::boot_util::boot_code_id;
@@ -52,7 +53,7 @@ use stacks_common::types::{PrivateKey, StacksEpochId};
 use stacks_common::util::tests::TestFlag;
 use stacks_common::util::vrf::VRFProof;
 
-use super::relayer::RelayerThread;
+use super::relayer::{MinerStopHandle, RelayerThread};
 use super::{Config, Error as NakamotoNodeError, EventDispatcher, Keychain};
 use crate::nakamoto_node::signer_coordinator::SignerCoordinator;
 use crate::nakamoto_node::VRF_MOCK_MINER_KEY;
@@ -203,6 +204,8 @@ pub struct BlockMinerThread {
     /// Rather, this burn block is used to determine whether or not a new
     ///  burn block has arrived since this thread started.
     burn_tip_at_start: ConsensusHash,
+    /// flag to indicate an abort driven from the relayer
+    abort_flag: Arc<AtomicBool>,
 }
 
 impl BlockMinerThread {
@@ -233,9 +236,14 @@ impl BlockMinerThread {
             signer_set_cache: None,
             burn_tip_at_start: burn_tip_at_start.clone(),
             tenure_change_time: Instant::now(),
+            abort_flag: Arc::new(AtomicBool::new(false)),
         }
     }
 
+    pub fn get_abort_flag(&self) -> Arc<AtomicBool> {
+        self.abort_flag.clone()
+    }
+
     #[cfg(test)]
     fn fault_injection_block_broadcast_stall(new_block: &NakamotoBlock) {
         if TEST_BROADCAST_STALL.get() {
@@ -298,29 +306,6 @@ impl BlockMinerThread {
     }
 
     /// Stop a miner tenure by blocking the miner and then joining the tenure thread
-    pub fn stop_miner(
-        globals: &Globals,
-        prior_miner: JoinHandle<Result<(), NakamotoNodeError>>,
-    ) -> Result<(), NakamotoNodeError> {
-        debug!(
-            "Stopping prior miner thread ID {:?}",
-            prior_miner.thread().id()
-        );
-        globals.block_miner();
-        let prior_miner_result = prior_miner
-            .join()
-            .map_err(|_| ChainstateError::MinerAborted)?;
-        if let Err(e) = prior_miner_result {
-            // it's okay if the prior miner thread exited with an error.
-            // in many cases this is expected (i.e., a burnchain block occurred)
-            // if some error condition should be handled though, this is the place
-            //  to do that handling.
-            debug!("Prior mining thread exited with: {e:?}");
-        }
-        globals.unblock_miner();
-        Ok(())
-    }
-
     #[cfg(test)]
     fn fault_injection_stall_miner() {
         if TEST_MINE_STALL.get() {
@@ -338,7 +323,7 @@ impl BlockMinerThread {
 
     pub fn run_miner(
         mut self,
-        prior_miner: Option<JoinHandle<Result<(), NakamotoNodeError>>>,
+        prior_miner: Option<MinerStopHandle>,
     ) -> Result<(), NakamotoNodeError> {
         // when starting a new tenure, block the mining thread if its currently running.
         // the new mining thread will join it (so that the new mining thread stalls, not the relayer)
@@ -352,7 +337,12 @@ impl BlockMinerThread {
             "reason" => %self.reason,
         );
         if let Some(prior_miner) = prior_miner {
-            Self::stop_miner(&self.globals, prior_miner)?;
+            debug!(
+                "Miner thread {:?}: will try and stop prior miner {:?}",
+                thread::current().id(),
+                prior_miner.inner_thread().id()
+            );
+            prior_miner.stop(&self.globals)?;
         }
         let mut stackerdbs = StackerDBs::connect(&self.config.get_stacker_db_file_path(), true)?;
         let mut last_block_rejected = false;
@@ -490,6 +480,13 @@ impl BlockMinerThread {
                     break Some(x);
                 }
                 Err(NakamotoNodeError::MiningFailure(ChainstateError::MinerAborted)) => {
+                    if self.abort_flag.load(Ordering::SeqCst) {
+                        info!("Miner interrupted while mining in order to shut down");
+                        self.globals
+                            .raise_initiative(format!("MiningFailure: aborted by node"));
+                        return Err(ChainstateError::MinerAborted.into());
+                    }
+
                     info!("Miner interrupted while mining, will try again");
                     // sleep, and try again. if the miner was interrupted because the burnchain
                     // view changed, the next `mine_block()` invocation will error

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

@@ -15,13 +15,15 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 use core::fmt;
 use std::collections::HashSet;
-use std::fs;
 use std::io::Read;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::mpsc::{Receiver, RecvTimeoutError};
+use std::sync::Arc;
 #[cfg(test)]
 use std::sync::LazyLock;
 use std::thread::JoinHandle;
 use std::time::{Duration, Instant};
+use std::{fs, thread};
 
 use rand::{thread_rng, Rng};
 use stacks::burnchains::{Burnchain, Txid};
@@ -40,6 +42,7 @@ use stacks::chainstate::stacks::db::StacksChainState;
 use stacks::chainstate::stacks::miner::{
     get_mining_spend_amount, signal_mining_blocked, signal_mining_ready,
 };
+use stacks::chainstate::stacks::Error as ChainstateError;
 use stacks::core::mempool::MemPoolDB;
 use stacks::core::STACKS_EPOCH_3_1_MARKER;
 use stacks::monitoring::increment_stx_blocks_mined_counter;
@@ -186,6 +189,101 @@ impl LastCommit {
     }
 }
 
+pub type MinerThreadJoinHandle = JoinHandle<Result<(), NakamotoNodeError>>;
+
+/// Miner thread join handle.
+/// This can be a "bare" miner thread, or a "tenure-stop" miner thread which itself stops a "bare"
+/// miner thread.
+pub enum MinerStopHandle {
+    Miner(MinerThreadJoinHandle, Arc<AtomicBool>),
+    TenureStop(MinerThreadJoinHandle, Arc<AtomicBool>),
+}
+
+impl MinerStopHandle {
+    pub fn new_miner(jh: MinerThreadJoinHandle, abort_flag: Arc<AtomicBool>) -> Self {
+        Self::Miner(jh, abort_flag)
+    }
+
+    pub fn new_tenure_stop(jh: MinerThreadJoinHandle, abort_flag: Arc<AtomicBool>) -> Self {
+        Self::TenureStop(jh, abort_flag)
+    }
+
+    pub fn inner_thread(&self) -> &std::thread::Thread {
+        match self {
+            Self::Miner(jh, ..) => jh.thread(),
+            Self::TenureStop(jh, ..) => jh.thread(),
+        }
+    }
+
+    pub fn into_inner(self) -> MinerThreadJoinHandle {
+        match self {
+            Self::Miner(jh, ..) => jh,
+            Self::TenureStop(jh, ..) => jh,
+        }
+    }
+
+    pub fn is_tenure_stop(&self) -> bool {
+        match self {
+            Self::TenureStop(..) => true,
+            _ => false,
+        }
+    }
+
+    pub fn is_miner(&self) -> bool {
+        match self {
+            Self::Miner(..) => true,
+            _ => false,
+        }
+    }
+
+    pub fn set_abort_flag(&self) {
+        match self {
+            Self::Miner(_, abort_flag) => {
+                (*abort_flag).store(true, Ordering::SeqCst);
+            }
+            Self::TenureStop(_, abort_flag) => {
+                (*abort_flag).store(true, Ordering::SeqCst);
+            }
+        }
+    }
+
+    pub fn get_abort_flag(&self) -> Arc<AtomicBool> {
+        match self {
+            Self::Miner(_, abort_flag) => abort_flag.clone(),
+            Self::TenureStop(_, abort_flag) => abort_flag.clone(),
+        }
+    }
+
+    pub fn stop(self, globals: &Globals) -> Result<(), NakamotoNodeError> {
+        let my_id = thread::current().id();
+        let prior_thread_id = self.inner_thread().id();
+        debug!(
+            "[Thread {:?}]: Stopping prior miner thread ID {:?}",
+            &my_id, &prior_thread_id
+        );
+
+        self.set_abort_flag();
+        globals.block_miner();
+
+        let prior_miner = self.into_inner();
+        let prior_miner_result = prior_miner.join().map_err(|_| {
+            error!("Miner: failed to join prior miner");
+            ChainstateError::MinerAborted
+        })?;
+        debug!("Stopped prior miner thread ID {:?}", &prior_thread_id);
+        if let Err(e) = prior_miner_result {
+            // it's okay if the prior miner thread exited with an error.
+            // in many cases this is expected (i.e., a burnchain block occurred)
+            // if some error condition should be handled though, this is the place
+            //  to do that handling.
+            debug!("Prior mining thread exited with: {e:?}");
+        }
+
+        globals.unblock_miner();
+        Ok(())
+    }
+}
+
 /// Relayer thread
 /// * accepts network results and stores blocks and microblocks
 /// * forwards new blocks, microblocks, and transactions to the p2p thread
@@ -242,7 +340,7 @@ pub struct RelayerThread {
     relayer: Relayer,
 
     /// handle to the subordinate miner thread
-    miner_thread: Option<JoinHandle<Result<(), NakamotoNodeError>>>,
+    miner_thread: Option<MinerStopHandle>,
     /// miner thread's burn view
     miner_thread_burn_view: Option<BlockSnapshot>,
 
@@ -1053,6 +1151,7 @@ impl RelayerThread {
             reason,
             burn_tip_at_start,
         )?;
+        let miner_abort_flag = new_miner_state.get_abort_flag();
 
         debug!("Relayer: starting new tenure thread");
 
@@ -1062,6 +1161,10 @@ impl RelayerThread {
             .name(format!("miner.{parent_tenure_start}.{rand_id}",))
             .stack_size(BLOCK_PROCESSOR_STACK_SIZE)
             .spawn(move || {
+                debug!(
+                    "New block miner thread ID is {:?}",
+                    std::thread::current().id()
+                );
                 Self::fault_injection_stall_miner_thread_startup();
                 if let Err(e) = new_miner_state.run_miner(prior_tenure_thread) {
                     info!("Miner thread failed: {e:?}");
@@ -1078,7 +1181,10 @@ impl RelayerThread {
             "Relayer: started tenure thread ID {:?}",
             new_miner_handle.thread().id()
         );
-        self.miner_thread.replace(new_miner_handle);
+        self.miner_thread.replace(MinerStopHandle::new_miner(
+            new_miner_handle,
+            miner_abort_flag,
+        ));
         self.miner_thread_burn_view.replace(burn_tip);
         Ok(())
     }
@@ -1092,18 +1198,23 @@ impl RelayerThread {
         };
         self.miner_thread_burn_view = None;
 
-        let id = prior_tenure_thread.thread().id();
+        let id = prior_tenure_thread.inner_thread().id();
+        let abort_flag = prior_tenure_thread.get_abort_flag();
         let globals = self.globals.clone();
 
         let stop_handle = std::thread::Builder::new()
-            .name(format!("tenure-stop-{}", self.local_peer.data_url))
-            .spawn(move || BlockMinerThread::stop_miner(&globals, prior_tenure_thread))
+            .name(format!(
+                "tenure-stop({:?})-{}",
+                id, self.local_peer.data_url
+            ))
+            .spawn(move || prior_tenure_thread.stop(&globals))
             .map_err(|e| {
                 error!("Relayer: Failed to spawn a stop-tenure thread: {e:?}");
                 NakamotoNodeError::SpawnError(e)
             })?;
 
-        self.miner_thread.replace(stop_handle);
+        self.miner_thread
+            .replace(MinerStopHandle::new_tenure_stop(stop_handle, abort_flag));
         debug!("Relayer: stopped tenure thread ID {id:?}");
         Ok(())
     }

testnet/stacks-node/src/tests/nakamoto_integrations.rs

@@ -196,9 +196,7 @@ lazy_static! {
 pub static TEST_SIGNING: Mutex<Option<TestSigningChannel>> = Mutex::new(None);
 
 pub struct TestSigningChannel {
-    // pub recv: Option<Receiver<ThresholdSignature>>,
     pub recv: Option<Receiver<Vec<MessageSignature>>>,
-    // pub send: Sender<ThresholdSignature>,
     pub send: Sender<Vec<MessageSignature>>,
 }
 
@@ -208,8 +206,6 @@ impl TestSigningChannel {
     /// Returns None if the singleton isn't instantiated and the miner should coordinate
     ///  a real signer set signature.
     /// Panics if the blind-signer times out.
-    ///
-    /// TODO: update to use signatures vec
     pub fn get_signature() -> Option<Vec<MessageSignature>> {
         let mut signer = TEST_SIGNING.lock().unwrap();
         let sign_channels = signer.as_mut()?;