[DRAFT] [miner] Reduce fork and orphan rate with an interruptable miner

src/chainstate/burn/db/sortdb.rs

@@ -89,7 +89,7 @@ use crate::chainstate::stacks::index::{ClarityMarfTrieId, MARFValue};
 use stacks_common::types::chainstate::StacksAddress;
 use stacks_common::types::chainstate::TrieHash;
 use stacks_common::types::chainstate::{
-    BlockHeaderHash, BurnchainHeaderHash, PoxId, SortitionId, VRFSeed,
+    BlockHeaderHash, BurnchainHeaderHash, PoxId, SortitionId, StacksBlockId, VRFSeed,
 };
 
 const BLOCK_HEIGHT_MAX: u64 = ((1 as u64) << 63) - 1;
@@ -1434,7 +1434,7 @@ impl<'a> SortitionHandleTx<'a> {
             )?;
         } else {
             // see if this block builds off of a Stacks block mined on this burnchain fork
-            let height_opt = match SortitionDB::get_accepted_stacks_block_pointer(
+            let parent_height_opt = match SortitionDB::get_accepted_stacks_block_pointer(
                 self,
                 &burn_tip.consensus_hash,
                 parent_stacks_block_hash,
@@ -1452,18 +1452,19 @@ impl<'a> SortitionHandleTx<'a> {
                     }
                 }
             };
-            match height_opt {
-                Some(height) => {
+            match parent_height_opt {
+                Some(parent_height) => {
                     if stacks_block_height > burn_tip.canonical_stacks_tip_height {
-                        assert!(stacks_block_height > height, "BUG: DB corruption -- block height {} <= {} means we accepted a block out-of-order", stacks_block_height, height);
+                        assert!(stacks_block_height > parent_height, "BUG: DB corruption -- block height {} <= {} means we accepted a block out-of-order", stacks_block_height, parent_height);
+
                         // This block builds off of a parent that is _concurrent_ with the memoized canonical stacks chain pointer.
                         // i.e. this block will reorg the Stacks chain on the canonical burnchain fork.
                         // Memoize this new stacks chain tip to the canonical burn chain snapshot.
                         // Note that we don't have to check continuity of accepted blocks -- we already
                         // are guaranteed by the Stacks chain state code that Stacks blocks in a given
                         // Stacks fork will be marked as accepted in sequential order (i.e. at height h, h+1,
                         // h+2, etc., without any gaps).
-                        debug!("Accepted Stacks block {}/{} builds on a previous canonical Stacks tip on this burnchain fork ({})", consensus_hash, stacks_block_hash, &burn_tip.burn_header_hash);
+                        debug!("Accepted Stacks block {}/{} ({}) builds on a previous canonical Stacks tip on this burnchain fork ({})", consensus_hash, stacks_block_hash, stacks_block_height, &burn_tip.burn_header_hash);
                         let args: &[&dyn ToSql] = &[
                             consensus_hash,
                             stacks_block_hash,
@@ -1477,7 +1478,7 @@ impl<'a> SortitionHandleTx<'a> {
                         // This block was mined on this fork, but it's acceptance doesn't overtake
                         // the current stacks chain tip.  Remember it so that we can process its children,
                         // which might do so later.
-                        debug!("Accepted Stacks block {}/{} builds on a non-canonical Stacks tip in this burnchain fork ({})", consensus_hash, stacks_block_hash, &burn_tip.burn_header_hash);
+                        debug!("Accepted Stacks block {}/{} ({}) builds on a non-canonical Stacks tip in this burnchain fork ({} height {})", consensus_hash, stacks_block_hash, stacks_block_height, &burn_tip.burn_header_hash, burn_tip.canonical_stacks_tip_height);
                     }
                     SortitionDB::insert_accepted_stacks_block_pointer(
                         self,
@@ -2475,8 +2476,8 @@ impl SortitionDB {
     pub fn is_db_version_supported_in_epoch(epoch: StacksEpochId, version: &str) -> bool {
         match epoch {
             StacksEpochId::Epoch10 => false,
-            StacksEpochId::Epoch20 => (version == "1" || version == "2" || version == "3"),
-            StacksEpochId::Epoch2_05 => (version == "2" || version == "3" || version == "4"),
+            StacksEpochId::Epoch20 => version == "1" || version == "2" || version == "3",
+            StacksEpochId::Epoch2_05 => version == "2" || version == "3" || version == "4",
         }
     }
 

src/chainstate/coordinator/mod.rs

@@ -19,6 +19,8 @@ use std::convert::{TryFrom, TryInto};
 use std::fs;
 use std::path::PathBuf;
 use std::sync::mpsc::SyncSender;
+use std::sync::Arc;
+use std::sync::Mutex;
 use std::time::Duration;
 
 use crate::burnchains::{
@@ -39,6 +41,7 @@ use crate::chainstate::stacks::{
         StacksHeaderInfo,
     },
     events::{StacksTransactionEvent, StacksTransactionReceipt, TransactionOrigin},
+    miner::{signal_mining_blocked, signal_mining_ready, MinerStatus},
     Error as ChainstateError, StacksBlock, TransactionPayload,
 };
 use crate::core::StacksEpoch;
@@ -272,6 +275,7 @@ impl<'a, T: BlockEventDispatcher, CE: CostEstimator + ?Sized, FE: FeeEstimator +
         atlas_config: AtlasConfig,
         cost_estimator: Option<&mut CE>,
         fee_estimator: Option<&mut FE>,
+        miner_status: Arc<Mutex<MinerStatus>>,
     ) where
         T: BlockEventDispatcher,
     {
@@ -311,18 +315,23 @@ impl<'a, T: BlockEventDispatcher, CE: CostEstimator + ?Sized, FE: FeeEstimator +
             // timeout so that we handle Ctrl-C a little gracefully
             match comms.wait_on() {
                 CoordinatorEvents::NEW_STACKS_BLOCK => {
+                    signal_mining_blocked(miner_status.clone());
                     debug!("Received new stacks block notice");
                     if let Err(e) = inst.handle_new_stacks_block() {
                         warn!("Error processing new stacks block: {:?}", e);
                     }
+                    signal_mining_ready(miner_status.clone());
                 }
                 CoordinatorEvents::NEW_BURN_BLOCK => {
+                    signal_mining_blocked(miner_status.clone());
                     debug!("Received new burn block notice");
                     if let Err(e) = inst.handle_new_burnchain_block() {
                         warn!("Error processing new burn block: {:?}", e);
                     }
+                    signal_mining_ready(miner_status.clone());
                 }
                 CoordinatorEvents::STOP => {
+                    signal_mining_blocked(miner_status.clone());
                     debug!("Received stop notice");
                     return;
                 }

src/chainstate/stacks/db/blocks.rs

@@ -1501,7 +1501,7 @@ impl StacksChainState {
                 }
             }
 
-            test_debug!(
+            debug!(
                 "Loaded microblock {}/{}-{} (parent={}, expect_seq={})",
                 &parent_consensus_hash,
                 &parent_anchored_block_hash,
@@ -1533,6 +1533,16 @@ impl StacksChainState {
             }
         }
         ret.reverse();
+
+        if ret.len() > 0 {
+            // should start with 0
+            if ret[0].header.sequence != 0 {
+                warn!("Invalid microblock stream from {}/{} to {}: sequence does not start with 0, but with {}",
+                      parent_consensus_hash, parent_anchored_block_hash, tip_microblock_hash, ret[0].header.sequence);
+
+                return Ok(None);
+            }
+        }
         Ok(Some(ret))
     }
 
@@ -1617,10 +1627,11 @@ impl StacksChainState {
             return Ok(None);
         }
 
-        let mut ret = vec![];
+        let mut ret: Vec<StacksMicroblock> = vec![];
         let mut tip: Option<StacksMicroblock> = None;
         let mut fork_poison = None;
         let mut expected_sequence = start_seq;
+        let mut parents: HashMap<BlockHeaderHash, usize> = HashMap::new();
 
         // load associated staging microblock data, but best-effort.
         // Stop loading once we find a fork juncture.
@@ -1657,6 +1668,21 @@ impl StacksChainState {
                 break;
             }
 
+            if let Some(idx) = parents.get(&mblock.header.prev_block) {
+                let conflict = ret[*idx].clone();
+                warn!(
+                    "Microblock fork found: microblocks {} and {} share parent {}",
+                    mblock.block_hash(),
+                    conflict.block_hash(),
+                    &mblock.header.prev_block
+                );
+                fork_poison = Some(TransactionPayload::PoisonMicroblock(
+                    mblock.header,
+                    conflict.header,
+                ));
+                break;
+            }
+
             // expect forks, so expected_sequence may not always increase
             expected_sequence =
                 cmp::min(mblock.header.sequence, expected_sequence).saturating_add(1);
@@ -1677,6 +1703,10 @@ impl StacksChainState {
             }
 
             tip = Some(mblock.clone());
+
+            let prev_block = mblock.header.prev_block.clone();
+            parents.insert(prev_block, ret.len());
+
             ret.push(mblock);
         }
         if fork_poison.is_none() && ret.len() == 0 {
@@ -3453,6 +3483,20 @@ impl StacksChainState {
         Ok(count - to_write)
     }
 
+    /// Check whether or not there exists a Stacks block at or higher than a given height that is
+    /// unprocessed.  This is used by miners to determine whether or not the block-commit they're
+    /// about to send is about to be invalidated
+    pub fn has_higher_unprocessed_blocks(conn: &DBConn, height: u64) -> Result<bool, Error> {
+        let sql =
+            "SELECT 1 FROM staging_blocks WHERE orphaned = 0 AND processed = 0 AND height >= ?1";
+        let args: &[&dyn ToSql] = &[&u64_to_sql(height)?];
+        let res = conn
+            .query_row(sql, args, |_r| Ok(()))
+            .optional()
+            .map(|x| x.is_some())?;
+        Ok(res)
+    }
+
     fn extract_signed_microblocks(
         parent_anchored_block_header: &StacksBlockHeader,
         microblocks: &Vec<StacksMicroblock>,
@@ -3793,6 +3837,49 @@ impl StacksChainState {
         Ok(Some((block_commit.burn_fee, sortition_burns)))
     }
 
+    /// Do we already have an anchored block?
+    pub fn has_anchored_block(
+        conn: &DBConn,
+        blocks_path: &str,
+        consensus_hash: &ConsensusHash,
+        block: &StacksBlock,
+    ) -> Result<bool, Error> {
+        let index_block_hash =
+            StacksBlockHeader::make_index_block_hash(consensus_hash, &block.block_hash());
+        if StacksChainState::has_stored_block(
+            &conn,
+            blocks_path,
+            consensus_hash,
+            &block.block_hash(),
+        )? {
+            debug!(
+                "Block already stored and processed: {}/{} ({})",
+                consensus_hash,
+                &block.block_hash(),
+                &index_block_hash
+            );
+            return Ok(true);
+        } else if StacksChainState::has_staging_block(conn, consensus_hash, &block.block_hash())? {
+            debug!(
+                "Block already stored (but not processed): {}/{} ({})",
+                consensus_hash,
+                &block.block_hash(),
+                &index_block_hash
+            );
+            return Ok(true);
+        } else if StacksChainState::has_block_indexed(&blocks_path, &index_block_hash)? {
+            debug!(
+                "Block already stored to chunk store: {}/{} ({})",
+                consensus_hash,
+                &block.block_hash(),
+                &index_block_hash
+            );
+            return Ok(true);
+        }
+
+        Ok(false)
+    }
+
     /// Pre-process and store an anchored block to staging, queuing it up for
     /// subsequent processing once all of its ancestors have been processed.
     ///
@@ -3828,43 +3915,21 @@ impl StacksChainState {
         let mainnet = self.mainnet;
         let chain_id = self.chain_id;
         let blocks_path = self.blocks_path.clone();
-        let mut block_tx = self.db_tx_begin()?;
 
-        // already in queue or already processed?
-        let index_block_hash =
-            StacksBlockHeader::make_index_block_hash(consensus_hash, &block.block_hash());
-        if StacksChainState::has_stored_block(
-            &block_tx,
-            &blocks_path,
+        // optimistic check (before opening a tx): already in queue or already processed?
+        if StacksChainState::has_anchored_block(
+            self.db(),
+            &self.blocks_path,
             consensus_hash,
-            &block.block_hash(),
-        )? {
-            debug!(
-                "Block already stored and processed: {}/{} ({})",
-                consensus_hash,
-                &block.block_hash(),
-                &index_block_hash
-            );
-            return Ok(false);
-        } else if StacksChainState::has_staging_block(
-            &block_tx,
-            consensus_hash,
-            &block.block_hash(),
+            block,
         )? {
-            debug!(
-                "Block already stored (but not processed): {}/{} ({})",
-                consensus_hash,
-                &block.block_hash(),
-                &index_block_hash
-            );
             return Ok(false);
-        } else if StacksChainState::has_block_indexed(&blocks_path, &index_block_hash)? {
-            debug!(
-                "Block already stored to chunk store: {}/{} ({})",
-                consensus_hash,
-                &block.block_hash(),
-                &index_block_hash
-            );
+        }
+
+        let mut block_tx = self.db_tx_begin()?;
+
+        // already in queue or already processed (within the tx; things might have changed)
+        if StacksChainState::has_anchored_block(&block_tx, &blocks_path, consensus_hash, block)? {
             return Ok(false);
         }
 

src/chainstate/stacks/db/mod.rs

@@ -192,7 +192,7 @@ impl DBConfig {
     pub fn supports_epoch(&self, epoch_id: StacksEpochId) -> bool {
         match epoch_id {
             StacksEpochId::Epoch10 => false,
-            StacksEpochId::Epoch20 => (self.version == "1" || self.version == "2"),
+            StacksEpochId::Epoch20 => self.version == "1" || self.version == "2",
             StacksEpochId::Epoch2_05 => self.version == "2",
         }
     }

src/chainstate/stacks/db/unconfirmed.rs

@@ -24,6 +24,7 @@ use crate::chainstate::stacks::db::accounts::*;
 use crate::chainstate::stacks::db::blocks::*;
 use crate::chainstate::stacks::db::*;
 use crate::chainstate::stacks::events::*;
+use crate::chainstate::stacks::index::marf::MARFOpenOpts;
 use crate::chainstate::stacks::Error;
 use crate::chainstate::stacks::*;
 use crate::clarity_vm::clarity::{ClarityInstance, Error as clarity_error};
@@ -84,6 +85,10 @@ pub struct UnconfirmedState {
     num_mblocks_added: u64,
     have_state: bool,
 
+    mainnet: bool,
+    clarity_state_index_root: String,
+    marf_opts: Option<MARFOpenOpts>,
+
     // fault injection for testing
     pub disable_cost_check: bool,
     pub disable_bytes_check: bool,
@@ -120,11 +125,51 @@ impl UnconfirmedState {
             num_mblocks_added: 0,
             have_state: false,
 
+            mainnet: chainstate.mainnet,
+            clarity_state_index_root: chainstate.clarity_state_index_root.clone(),
+            marf_opts: chainstate.marf_opts.clone(),
+
             disable_cost_check: check_fault_injection(FAULT_DISABLE_MICROBLOCKS_COST_CHECK),
             disable_bytes_check: check_fault_injection(FAULT_DISABLE_MICROBLOCKS_BYTES_CHECK),
         })
     }
 
+    /// Make a read-only copy of this unconfirmed state.  The resulting unconfiremd state cannot
+    /// be refreshed, but it will represent a snapshot of the existing unconfirmed state.
+    pub fn make_readonly_owned(&self) -> Result<UnconfirmedState, Error> {
+        let marf = MarfedKV::open_unconfirmed(
+            &self.clarity_state_index_root,
+            None,
+            self.marf_opts.clone(),
+        )?;
+
+        let clarity_instance = ClarityInstance::new(self.mainnet, marf);
+
+        Ok(UnconfirmedState {
+            confirmed_chain_tip: self.confirmed_chain_tip.clone(),
+            unconfirmed_chain_tip: self.unconfirmed_chain_tip.clone(),
+            clarity_inst: clarity_instance,
+            mined_txs: self.mined_txs.clone(),
+            cost_so_far: self.cost_so_far.clone(),
+            bytes_so_far: self.bytes_so_far,
+
+            last_mblock: self.last_mblock.clone(),
+            last_mblock_seq: self.last_mblock_seq,
+
+            readonly: true,
+            dirty: false,
+            num_mblocks_added: self.num_mblocks_added,
+            have_state: self.have_state,
+
+            mainnet: self.mainnet,
+            clarity_state_index_root: self.clarity_state_index_root.clone(),
+            marf_opts: self.marf_opts.clone(),
+
+            disable_cost_check: self.disable_cost_check,
+            disable_bytes_check: self.disable_bytes_check,
+        })
+    }
+
     /// Make a new unconfirmed state, but don't do anything with it yet, and deny refreshes.
     fn new_readonly(
         chainstate: &StacksChainState,
@@ -157,6 +202,10 @@ impl UnconfirmedState {
             num_mblocks_added: 0,
             have_state: false,
 
+            mainnet: chainstate.mainnet,
+            clarity_state_index_root: chainstate.clarity_state_index_root.clone(),
+            marf_opts: chainstate.marf_opts.clone(),
+
             disable_cost_check: check_fault_injection(FAULT_DISABLE_MICROBLOCKS_COST_CHECK),
             disable_bytes_check: check_fault_injection(FAULT_DISABLE_MICROBLOCKS_BYTES_CHECK),
         })