Repository: stacks-network/stacks-core (language: python)

Issue:
Feat/shadow block tooling
This adds tooling to `stacks-inspect` to generate and insert shadow blocks into a production chainstate.

It'll probably get more commits as we test it out, but it's opened here for review so we can start making images off of it and testing it out on Nakamoto testnet.


Discussion / hints:

Fix patch:
diff --git a/.github/actions/dockerfiles/Dockerfile.debian-source b/.github/actions/dockerfiles/Dockerfile.debian-source
index b8da585fe2..80c434e8d5 100644
--- a/.github/actions/dockerfiles/Dockerfile.debian-source
+++ b/.github/actions/dockerfiles/Dockerfile.debian-source
@@ -24,5 +24,5 @@ RUN --mount=type=tmpfs,target=${BUILD_DIR} cp -R /src/. ${BUILD_DIR}/ \
     && cp -R ${BUILD_DIR}/target/${TARGET}/release/. /out
 
 FROM --platform=${TARGETPLATFORM} debian:bookworm
-COPY --from=build /out/stacks-node /out/stacks-signer /bin/
+COPY --from=build /out/stacks-node /out/stacks-signer /out/stacks-inspect /bin/
 CMD ["stacks-node", "mainnet"]
diff --git a/stacks-common/src/types/mod.rs b/stacks-common/src/types/mod.rs
index 5e13a6d330..8089c6c0a1 100644
--- a/stacks-common/src/types/mod.rs
+++ b/stacks-common/src/types/mod.rs
@@ -170,6 +170,21 @@ impl StacksEpochId {
         }
     }
 
+    /// Whether or not this epoch supports shadow blocks
+    pub fn supports_shadow_blocks(&self) -> bool {
+        match self {
+            StacksEpochId::Epoch10
+            | StacksEpochId::Epoch20
+            | StacksEpochId::Epoch2_05
+            | StacksEpochId::Epoch21
+            | StacksEpochId::Epoch22
+            | StacksEpochId::Epoch23
+            | StacksEpochId::Epoch24
+            | StacksEpochId::Epoch25 => false,
+            StacksEpochId::Epoch30 => true,
+        }
+    }
+
     /// Does this epoch support unlocking PoX contributors that miss a slot?
     ///
     /// Epoch 2.0 - 2.05 didn't support this feature, but they weren't epoch-guarded on it. Instead,
diff --git a/stackslib/src/burnchains/mod.rs b/stackslib/src/burnchains/mod.rs
index 0bc68897cb..3e153df53b 100644
--- a/stackslib/src/burnchains/mod.rs
+++ b/stackslib/src/burnchains/mod.rs
@@ -450,6 +450,7 @@ impl PoxConstants {
         )
     }
 
+    // NOTE: this is the *old* pre-Nakamoto testnet
     pub fn testnet_default() -> PoxConstants {
         PoxConstants::new(
             POX_REWARD_CYCLE_LENGTH / 2,   // 1050
@@ -468,6 +469,10 @@ impl PoxConstants {
         ) // total liquid supply is 40000000000000000 µSTX
     }
 
+    pub fn nakamoto_testnet_default() -> PoxConstants {
+        PoxConstants::new(900, 100, 51, 100, 0, u64::MAX, u64::MAX, 242, 243, 246, 244)
+    }
+
     // TODO: add tests from mutation testing results #4838
     #[cfg_attr(test, mutants::skip)]
     pub fn regtest_default() -> PoxConstants {
diff --git a/stackslib/src/chainstate/burn/operations/leader_block_commit.rs b/stackslib/src/chainstate/burn/operations/leader_block_commit.rs
index 910315f082..c3a378ddf6 100644
--- a/stackslib/src/chainstate/burn/operations/leader_block_commit.rs
+++ b/stackslib/src/chainstate/burn/operations/leader_block_commit.rs
@@ -136,7 +136,8 @@ impl LeaderBlockCommitOp {
         block_header_hash: &BlockHeaderHash,
         block_height: u64,
         new_seed: &VRFSeed,
-        parent: &LeaderBlockCommitOp,
+        parent_block_height: u32,
+        parent_vtxindex: u16,
         key_block_ptr: u32,
         key_vtxindex: u16,
         burn_fee: u64,
@@ -148,8 +149,8 @@ impl LeaderBlockCommitOp {
             new_seed: new_seed.clone(),
             key_block_ptr: key_block_ptr,
             key_vtxindex: key_vtxindex,
-            parent_block_ptr: parent.block_height as u32,
-            parent_vtxindex: parent.vtxindex as u16,
+            parent_block_ptr: parent_block_height,
+            parent_vtxindex: parent_vtxindex,
             memo: vec![],
             burn_fee: burn_fee,
             input: input.clone(),
@@ -696,8 +697,19 @@ impl LeaderBlockCommitOp {
         //  is descendant
         let directly_descended_from_anchor = epoch_id.block_commits_to_parent()
             && self.block_header_hash == reward_set_info.anchor_block;
-        let descended_from_anchor = directly_descended_from_anchor || tx
-            .descended_from(parent_block_height, &reward_set_info.anchor_block)
+
+        // second, if we're in a nakamoto epoch, and the parent block has vtxindex 0 (i.e. the
+        // coinbase of the burnchain block), then assume that this block descends from the anchor
+        // block for the purposes of validating its PoX payouts.  The block validation logic will
+        // check that the parent block is indeed a shadow block, and that `self.parent_block_ptr`
+        // points to the shadow block's tenure's burnchain block.
+        let maybe_shadow_parent = epoch_id.supports_shadow_blocks()
+            && self.parent_block_ptr != 0
+            && self.parent_vtxindex == 0;
+
+        let descended_from_anchor = directly_descended_from_anchor
+            || maybe_shadow_parent
+            || tx.descended_from(parent_block_height, &reward_set_info.anchor_block)
             .map_err(|e| {
                 error!("Failed to check whether parent (height={}) is descendent of anchor block={}: {}",
                        parent_block_height, &reward_set_info.anchor_block, e);
@@ -1031,10 +1043,12 @@ impl LeaderBlockCommitOp {
             return Err(op_error::BlockCommitNoParent);
         } else if self.parent_block_ptr != 0 || self.parent_vtxindex != 0 {
             // not building off of genesis, so the parent block must exist
+            // unless the parent is a shadow block
             let has_parent = tx
                 .get_block_commit_parent(parent_block_height, self.parent_vtxindex.into(), &tx_tip)?
                 .is_some();
-            if !has_parent {
+            let maybe_shadow_block = self.parent_vtxindex == 0 && epoch_id.supports_shadow_blocks();
+            if !has_parent && !maybe_shadow_block {
                 warn!("Invalid block commit: no parent block in this fork";
                       "apparent_sender" => %apparent_sender_repr
                 );
diff --git a/stackslib/src/chainstate/nakamoto/coordinator/mod.rs b/stackslib/src/chainstate/nakamoto/coordinator/mod.rs
index cb1966d806..fc7c8ba504 100644
--- a/stackslib/src/chainstate/nakamoto/coordinator/mod.rs
+++ b/stackslib/src/chainstate/nakamoto/coordinator/mod.rs
@@ -58,6 +58,9 @@ use crate::monitoring::increment_stx_blocks_processed_counter;
 use crate::net::Error as NetError;
 use crate::util_lib::db::Error as DBError;
 
+#[cfg(any(test, feature = "testing"))]
+pub static TEST_COORDINATOR_STALL: std::sync::Mutex<Option<bool>> = std::sync::Mutex::new(None);
+
 #[cfg(test)]
 pub mod tests;
 
@@ -484,7 +487,14 @@ pub fn load_nakamoto_reward_set<U: RewardSetProvider>(
     let Some(anchor_block_header) = prepare_phase_sortitions
         .into_iter()
         .find_map(|sn| {
-            if !sn.sortition {
+            let shadow_tenure = match chain_state.nakamoto_blocks_db().is_shadow_tenure(&sn.consensus_hash) {
+                Ok(x) => x,
+                Err(e) => {
+                    return Some(Err(e));
+                }
+            };
+
+            if !sn.sortition && !shadow_tenure {
                 return None
             }
 
@@ -757,6 +767,21 @@ impl<
         true
     }
 
+    #[cfg(any(test, feature = "testing"))]
+    fn fault_injection_pause_nakamoto_block_processing() {
+        if *TEST_COORDINATOR_STALL.lock().unwrap() == Some(true) {
+            // Do an extra check just so we don't log EVERY time.
+            warn!("Coordinator is stalled due to testing directive");
+            while *TEST_COORDINATOR_STALL.lock().unwrap() == Some(true) {
+                std::thread::sleep(std::time::Duration::from_millis(10));
+            }
+            warn!("Coordinator is no longer stalled due to testing directive. Continuing...");
+        }
+    }
+
+    #[cfg(not(any(test, feature = "testing")))]
+    fn fault_injection_pause_nakamoto_block_processing() {}
+
     /// Handle one or more new Nakamoto Stacks blocks.
     /// If we process a PoX anchor block, then return its block hash.  This unblocks processing the
     /// next reward cycle's burnchain blocks.  Subsequent calls to this function will terminate
@@ -769,6 +794,8 @@ impl<
         );
 
         loop {
+            Self::fault_injection_pause_nakamoto_block_processing();
+
             // process at most one block per loop pass
             let mut processed_block_receipt = match NakamotoChainState::process_next_nakamoto_block(
                 &mut self.chain_state_db,
diff --git a/stackslib/src/chainstate/nakamoto/miner.rs b/stackslib/src/chainstate/nakamoto/miner.rs
index 0291b1dad2..74ecd19bc1 100644
--- a/stackslib/src/chainstate/nakamoto/miner.rs
+++ b/stackslib/src/chainstate/nakamoto/miner.rs
@@ -28,7 +28,10 @@ use clarity::vm::clarity::TransactionConnection;
 use clarity::vm::costs::{ExecutionCost, LimitedCostTracker, TrackerData};
 use clarity::vm::database::BurnStateDB;
 use clarity::vm::errors::Error as InterpreterError;
-use clarity::vm::types::{QualifiedContractIdentifier, TypeSignature};
+use clarity::vm::types::{
+    QualifiedContractIdentifier, StacksAddressExtensions as ClarityStacksAddressExtensions,
+    TypeSignature,
+};
 use libstackerdb::StackerDBChunkData;
 use serde::Deserialize;
 use stacks_common::codec::{read_next, write_next, Error as CodecError, StacksMessageCodec};
@@ -37,8 +40,9 @@ use stacks_common::types::chainstate::{
 };
 use stacks_common::types::StacksPublicKeyBuffer;
 use stacks_common::util::get_epoch_time_ms;
-use stacks_common::util::hash::{Hash160, MerkleTree, Sha512Trunc256Sum};
+use stacks_common::util::hash::{hex_bytes, Hash160, MerkleTree, Sha512Trunc256Sum};
 use stacks_common::util::secp256k1::{MessageSignature, Secp256k1PrivateKey};
+use stacks_common::util::vrf::VRFProof;
 
 use crate::burnchains::{PrivateKey, PublicKey};
 use crate::chainstate::burn::db::sortdb::{
@@ -58,8 +62,8 @@ use crate::chainstate::stacks::db::transactions::{
     handle_clarity_runtime_error, ClarityRuntimeTxError,
 };
 use crate::chainstate::stacks::db::{
-    ChainstateTx, ClarityTx, MinerRewardInfo, StacksBlockHeaderTypes, StacksChainState,
-    StacksHeaderInfo, MINER_REWARD_MATURITY,
+    ChainstateTx, ClarityTx, MinerRewardInfo, StacksAccount, StacksBlockHeaderTypes,
+    StacksChainState, StacksHeaderInfo, MINER_REWARD_MATURITY,
 };
 use crate::chainstate::stacks::events::{StacksTransactionEvent, StacksTransactionReceipt};
 use crate::chainstate::stacks::miner::{
@@ -117,7 +121,7 @@ pub struct NakamotoBlockBuilder {
     /// Total burn this block represents
     total_burn: u64,
     /// Matured miner rewards to process, if any.
-    matured_miner_rewards_opt: Option<MaturedMinerRewards>,
+    pub(crate) matured_miner_rewards_opt: Option<MaturedMinerRewards>,
     /// bytes of space consumed so far
     pub bytes_so_far: u64,
     /// transactions selected
@@ -143,7 +147,7 @@ pub struct MinerTenureInfo<'a> {
     pub coinbase_height: u64,
     pub cause: Option<TenureChangeCause>,
     pub active_reward_set: boot::RewardSet,
-    pub tenure_block_commit: LeaderBlockCommitOp,
+    pub tenure_block_commit_opt: Option<LeaderBlockCommitOp>,
 }
 
 impl NakamotoBlockBuilder {
@@ -244,7 +248,21 @@ impl NakamotoBlockBuilder {
         burn_dbconn: &'a SortitionHandleConn,
         cause: Option<TenureChangeCause>,
     ) -> Result<MinerTenureInfo<'a>, Error> {
-        debug!("Nakamoto miner tenure begin");
+        self.inner_load_tenure_info(chainstate, burn_dbconn, cause, false)
+    }
+
+    /// This function should be called before `tenure_begin`.
+    /// It creates a MinerTenureInfo struct which owns connections to the chainstate and sortition
+    /// DBs, so that block-processing is guaranteed to terminate before the lives of these handles
+    /// expire.
+    pub(crate) fn inner_load_tenure_info<'a>(
+        &self,
+        chainstate: &'a mut StacksChainState,
+        burn_dbconn: &'a SortitionHandleConn,
+        cause: Option<TenureChangeCause>,
+        shadow_block: bool,
+    ) -> Result<MinerTenureInfo<'a>, Error> {
+        debug!("Nakamoto miner tenure begin"; "shadow" => shadow_block, "tenure_change" => ?cause);
 
         let Some(tenure_election_sn) =
             SortitionDB::get_block_snapshot_consensus(&burn_dbconn, &self.header.consensus_hash)?
@@ -256,19 +274,25 @@ impl NakamotoBlockBuilder {
             );
             return Err(Error::NoSuchBlockError);
         };
-        let Some(tenure_block_commit) = SortitionDB::get_block_commit(
-            &burn_dbconn,
-            &tenure_election_sn.winning_block_txid,
-            &tenure_election_sn.sortition_id,
-        )?
-        else {
-            warn!("Could not find winning block commit for burn block that elected the miner";
-                "consensus_hash" => %self.header.consensus_hash,
-                "stacks_block_hash" => %self.header.block_hash(),
-                "stacks_block_id" => %self.header.block_id(),
-                "winning_txid" => %tenure_election_sn.winning_block_txid
-            );
-            return Err(Error::NoSuchBlockError);
+
+        let tenure_block_commit_opt = if shadow_block {
+            None
+        } else {
+            let Some(tenure_block_commit) = SortitionDB::get_block_commit(
+                &burn_dbconn,
+                &tenure_election_sn.winning_block_txid,
+                &tenure_election_sn.sortition_id,
+            )?
+            else {
+                warn!("Could not find winning block commit for burn block that elected the miner";
+                    "consensus_hash" => %self.header.consensus_hash,
+                    "stacks_block_hash" => %self.header.block_hash(),
+                    "stacks_block_id" => %self.header.block_id(),
+                    "winning_txid" => %tenure_election_sn.winning_block_txid
+                );
+                return Err(Error::NoSuchBlockError);
+            };
+            Some(tenure_block_commit)
         };
 
         let elected_height = tenure_election_sn.block_height;
@@ -372,11 +396,11 @@ impl NakamotoBlockBuilder {
             cause,
             coinbase_height,
             active_reward_set,
-            tenure_block_commit,
+            tenure_block_commit_opt,
         })
     }
 
-    /// Begin/resume mining a tenure's transactions.
+    /// Begin/resume mining a (normal) tenure's transactions.
     /// Returns an open ClarityTx for mining the block.
     /// NOTE: even though we don't yet know the block hash, the Clarity VM ensures that a
     /// transaction can't query information about the _current_ block (i.e. information that is not
@@ -386,6 +410,12 @@ impl NakamotoBlockBuilder {
         burn_dbconn: &'a SortitionHandleConn,
         info: &'b mut MinerTenureInfo<'a>,
     ) -> Result<ClarityTx<'b, 'b>, Error> {
+        let Some(block_commit) = info.tenure_block_commit_opt.as_ref() else {
+            return Err(Error::InvalidStacksBlock(
+                "Block-commit is required; cannot mine a shadow block".into(),
+            ));
+        };
+
         let SetupBlockResult {
             clarity_tx,
             matured_miner_rewards_opt,
@@ -398,7 +428,6 @@ impl NakamotoBlockBuilder {
             &burn_dbconn.context.pox_constants,
             info.parent_consensus_hash,
             info.parent_header_hash,
-            info.parent_stacks_block_height,
             info.parent_burn_block_height,
             info.burn_tip,
             info.burn_tip_height,
@@ -406,7 +435,7 @@ impl NakamotoBlockBuilder {
             info.coinbase_height,
             info.cause == Some(TenureChangeCause::Extended),
             &self.header.pox_treatment,
-            &info.tenure_block_commit,
+            block_commit,
             &info.active_reward_set,
         )?;
         self.matured_miner_rewards_opt = matured_miner_rewards_opt;
diff --git a/stackslib/src/chainstate/nakamoto/mod.rs b/stackslib/src/chainstate/nakamoto/mod.rs
index e5ce7b0637..ca37e30121 100644
--- a/stackslib/src/chainstate/nakamoto/mod.rs
+++ b/stackslib/src/chainstate/nakamoto/mod.rs
@@ -119,6 +119,7 @@ use crate::{chainstate, monitoring};
 pub mod coordinator;
 pub mod keys;
 pub mod miner;
+pub mod shadow;
 pub mod signer_set;
 pub mod staging_blocks;
 pub mod tenure;
@@ -840,6 +841,12 @@ impl NakamotoBlockHeader {
             ));
         };
 
+        // if this is a shadow block, then its signing weight is as if every signer signed it, even
+        // though the signature vector is undefined.
+        if self.is_shadow_block() {
+            return Ok(self.get_shadow_signer_weight(reward_set)?);
+        }
+
         let mut total_weight_signed: u32 = 0;
         // `last_index` is used to prevent out-of-order signatures
         let mut last_index = None;
@@ -1400,6 +1407,7 @@ impl NakamotoBlock {
                   "consensus_hash" => %self.header.consensus_hash,
                   "stacks_block_hash" => %self.header.block_hash(),
                   "stacks_block_id" => %self.block_id(),
+                  "parent_block_id" => %self.header.parent_block_id,
                   "commit_seed" => %block_commit.new_seed,
                   "proof_seed" => %VRFSeed::from_proof(&parent_vrf_proof),
                   "parent_vrf_proof" => %parent_vrf_proof.to_hex(),
@@ -1433,10 +1441,15 @@ impl NakamotoBlock {
     }
 
     /// Verify the miner signature over this block.
+    /// If this is a shadow block, then this is always Ok(())
     pub(crate) fn check_miner_signature(
         &self,
         miner_pubkey_hash160: &Hash160,
     ) -> Result<(), ChainstateError> {
+        if self.is_shadow_block() {
+            return Ok(());
+        }
+
         let recovered_miner_hash160 = self.recover_miner_pubkh()?;
         if &recovered_miner_hash160 != miner_pubkey_hash160 {
             warn!(
@@ -1501,11 +1514,13 @@ impl NakamotoBlock {
 
     /// Verify that if this block has a coinbase, that its VRF proof is consistent with the leader
     /// public key's VRF key. If there is no coinbase tx, then this is a no-op.
-    pub(crate) fn check_coinbase_tx(
+    fn check_normal_coinbase_tx(
         &self,
         leader_vrf_key: &VRFPublicKey,
         sortition_hash: &SortitionHash,
     ) -> Result<(), ChainstateError> {
+        assert!(!self.is_shadow_block());
+
         // If this block has a coinbase, then verify that its VRF proof was generated by this
         // block's miner.  We'll verify that the seed of this block-commit was generated from the
         // parnet tenure's VRF proof via the `validate_vrf_seed()` method, which requires that we
@@ -1514,11 +1529,12 @@ impl NakamotoBlock {
             let (_, _, vrf_proof_opt) = coinbase_tx
                 .try_as_coinbase()
                 .expect("FATAL: `get_coinbase_tx()` did not return a coinbase");
+
             let vrf_proof = vrf_proof_opt.ok_or(ChainstateError::InvalidStacksBlock(
                 "Nakamoto coinbase must have a VRF proof".into(),
             ))?;
 
-            // this block's VRF proof must have ben generated from the last sortition's sortition
+            // this block's VRF proof must have been generated from the last sortition's sortition
             // hash (which includes the last commit's VRF seed)
             let valid = match VRF::verify(leader_vrf_key, vrf_proof, sortition_hash.as_bytes()) {
                 Ok(v) => v,
@@ -1548,27 +1564,15 @@ impl NakamotoBlock {
         Ok(())
     }
 
-    /// Validate this Nakamoto block header against burnchain state.
-    /// Used to determine whether or not we'll keep a block around (even if we don't yet have its parent).
-    ///
-    /// Arguments
-    /// -- `tenure_burn_chain_tip` is the BlockSnapshot containing the block-commit for this block's
-    /// tenure.  It is not always the tip of the burnchain.
-    /// -- `expected_burn` is the total number of burnchain tokens spent, if known.
-    /// -- `leader_key` is the miner's leader key registration transaction
+    /// Verify properties of blocks against the burnchain that are common to both normal and shadow
+    /// blocks.
     ///
-    /// Verifies the following:
     /// -- (self.header.consensus_hash) that this block falls into this block-commit's tenure
     /// -- (self.header.burn_spent) that this block's burn total matches `burn_tip`'s total burn
-    /// -- (self.header.miner_signature) that this miner signed this block
-    /// -- if this block has a tenure change, then it's consistent with the miner's public key and
-    /// self.header.consensus_hash
-    /// -- if this block has a coinbase, then that it's VRF proof was generated by this miner
-    pub fn validate_against_burnchain(
+    fn common_validate_against_burnchain(
         &self,
         tenure_burn_chain_tip: &BlockSnapshot,
         expected_burn: Option<u64>,
-        leader_key: &LeaderKeyRegisterOp,
     ) -> Result<(), ChainstateError> {
         // this block's consensus hash must match the sortition that selected it
         if tenure_burn_chain_tip.consensus_hash != self.header.consensus_hash {
@@ -1599,24 +1603,37 @@ impl NakamotoBlock {
             }
         }
 
-        // miner must have signed this block
-        let miner_pubkey_hash160 = leader_key
-            .interpret_nakamoto_signing_key()
-            .ok_or(ChainstateError::NoSuchBlockError)
-            .map_err(|e| {
-                warn!(
-                    "Leader key did not contain a hash160 of the miner signing public key";
-                    "leader_key" => ?leader_key,
-                );
-                e
-            })?;
+        Ok(())
+    }
 
-        self.check_miner_signature(&miner_pubkey_hash160)?;
+    /// Validate this Nakamoto block header against burnchain state.
+    /// Used to determine whether or not we'll keep a block around (even if we don't yet have its parent).
+    ///
+    /// Arguments
+    /// -- `mainnet`: whether or not the chain is mainnet
+    /// -- `tenure_burn_chain_tip` is the BlockSnapshot containing the block-commit for this block's
+    /// tenure.  It is not always the tip of the burnchain.
+    /// -- `expected_burn` is the total number of burnchain tokens spent, if known.
+    /// -- `leader_key` is the miner's leader key registration transaction
+    ///
+    /// Verifies the following:
+    /// -- (self.header.consensus_hash) that this block falls into this block-commit's tenure
+    /// -- (self.header.burn_spent) that this block's burn total matches `burn_tip`'s total burn
+    /// -- (self.header.miner_signature) that this miner signed this block
+    /// -- if this block has a tenure change, then it's consistent with the miner's public key and
+    /// self.header.consensus_hash
+    /// -- if this block has a coinbase, then that it's VRF proof was generated by this miner
+    fn validate_normal_against_burnchain(
+        &self,
+        tenure_burn_chain_tip: &BlockSnapshot,
+        expected_burn: Option<u64>,
+        miner_pubkey_hash160: &Hash160,
+        vrf_public_key: &VRFPublicKey,
+    ) -> Result<(), ChainstateError> {
+        self.common_validate_against_burnchain(tenure_burn_chain_tip, expected_burn)?;
+        self.check_miner_signature(miner_pubkey_hash160)?;
         self.check_tenure_tx()?;
-        self.check_coinbase_tx(
-            &leader_key.public_key,
-            &tenure_burn_chain_tip.sortition_hash,
-        )?;
+        self.check_normal_coinbase_tx(vrf_public_key, &tenure_burn_chain_tip.sortition_hash)?;
 
         // not verified by this method:
         // * chain_length       (need parent block header)
@@ -1801,23 +1818,36 @@ impl NakamotoChainState {
         let block_id = next_ready_block.block_id();
 
         // find corresponding snapshot
-        let next_ready_block_snapshot = SortitionDB::get_block_snapshot_consensus(
+        let Some(next_ready_block_snapshot) = SortitionDB::get_block_snapshot_consensus(
             sort_db.conn(),
             &next_ready_block.header.consensus_hash,
         )?
-        .unwrap_or_else(|| {
+        else {
+            // might not have snapshot yet, even if the block is burn-attachable, because it could
+            // be a shadow block
+            if next_ready_block.is_shadow_block() {
+                test_debug!(
+                    "Stop processing Nakamoto blocks at shadow block {}",
+                    &next_ready_block.block_id()
+                );
+                return Ok(None);
+            }
+
+            // but this isn't allowed for non-shadow blocks, which must be marked burn-attachable
+            // separately
             panic!(
                 "CORRUPTION: staging Nakamoto block {}/{} does not correspond to a burn block",
                 &next_ready_block.header.consensus_hash,
                 &next_ready_block.header.block_hash()
-            )
-        });
+            );
+        };
 
         debug!("Process staging Nakamoto block";
                "consensus_hash" => %next_ready_block.header.consensus_hash,
                "stacks_block_hash" => %next_ready_block.header.block_hash(),
                "stacks_block_id" => %next_ready_block.header.block_id(),
-               "burn_block_hash" => %next_ready_block_snapshot.burn_header_hash
+               "burn_block_hash" => %next_ready_block_snapshot.burn_header_hash,
+               "parent_block_id" => %next_ready_block.header.parent_block_id,
         );
 
         let elected_height = sort_db
@@ -1972,7 +2002,7 @@ impl NakamotoChainState {
             ));
         };
 
-        let (commit_burn, sortition_burn) = if new_tenure {
+        let (commit_burn, sortition_burn) = if new_tenure && !next_ready_block.is_shadow_block() {
             // find block-commit to get commit-burn
             let block_commit = SortitionDB::get_block_commit(
                 sort_db.conn(),
@@ -1985,6 +2015,7 @@ impl NakamotoChainState {
                 SortitionDB::get_block_burn_amount(sort_db.conn(), &next_ready_block_snapshot)?;
             (block_commit.burn_fee, sort_burn)
         } else {
+            // non-tenure-change blocks and shadow blocks both have zero additional spends
             (0, 0)
         };
 
@@ -2171,21 +2202,17 @@ impl NakamotoChainState {
         Ok(Some(burn_view_sn.total_burn))
     }
 
-    /// Validate that a Nakamoto block attaches to the burn chain state.
-    /// Called before inserting the block into the staging DB.
-    /// Wraps `NakamotoBlock::validate_against_burnchain()`, and
-    /// verifies that all transactions in the block are allowed in this epoch.
-    pub fn validate_nakamoto_block_burnchain(
+    /// Verify that the given Nakamoto block attaches to the canonical burnchain fork.
+    /// Return Ok(snapshot) on success, where `snapshot` is the sortition corresponding to this
+    /// block's tenure.
+    /// Return Err(..) otherwise
+    fn validate_nakamoto_tenure_snapshot(
         db_handle: &SortitionHandleConn,
-        expected_burn: Option<u64>,
         block: &NakamotoBlock,
-        mainnet: bool,
-        chain_id: u32,
-    ) -> Result<(), ChainstateError> {
+    ) -> Result<BlockSnapshot, ChainstateError> {
         // find the sortition-winning block commit for this block, as well as the block snapshot
         // containing the parent block-commit.  This is the snapshot that corresponds to when the
         // miner begain its tenure; it may not be the burnchain tip.
-        let block_hash = block.header.block_hash();
         let consensus_hash = &block.header.consensus_hash;
 
         let sort_tip = SortitionDB::get_canonical_burn_chain_tip(db_handle)?;
@@ -2194,7 +2221,7 @@ impl NakamotoChainState {
         let Some(tenure_burn_chain_tip) =
             SortitionDB::get_block_snapshot_consensus(db_handle, consensus_hash)?
         else {
-            warn!("No sortition for {}", &consensus_hash);
+            warn!("No sortition for {}", consensus_hash);
             return Err(ChainstateError::InvalidStacksBlock(
                 "No sortition for block's consensus hash".into(),
             ));
@@ -2221,7 +2248,58 @@ impl NakamotoChainState {
             ));
         };
 
-        // the block-commit itself
+        Ok(tenure_burn_chain_tip)
+    }
+
+    /// Statically validate the block's transactions against the burnchain epoch.
+    /// Return Ok(()) if they pass all static checks
+    /// Return Err(..) if not.
+    fn validate_nakamoto_block_transactions_static(
+        mainnet: bool,
+        chain_id: u32,
+        sortdb_conn: &Connection,
+        block: &NakamotoBlock,
+        block_tenure_burn_height: u64,
+    ) -> Result<(), ChainstateError> {
+        // check the _next_ block's tenure, since when Nakamoto's miner activates, the current chain tip
+        // will be in epoch 2.5 (the next block will be epoch 3.0)
+        let cur_epoch = SortitionDB::get_stacks_epoch(sortdb_conn, block_tenure_burn_height + 1)?
+            .expect("FATAL: no epoch defined for current Stacks block");
+
+        // static checks on transactions all pass
+        let valid = block.validate_transactions_static(mainnet, chain_id, cur_epoch.epoch_id);
+        if !valid {
+            warn!(
+                "Invalid Nakamoto block, transactions failed static checks: {}/{} (epoch {})",
+                &block.header.consensus_hash,
+                &block.header.block_hash(),
+                cur_epoch.epoch_id
+            );
+            return Err(ChainstateError::InvalidStacksBlock(
+                "Invalid Nakamoto block: failed static transaction checks".into(),
+            ));
+        }
+
+        Ok(())
+    }
+
+    /// Validate that a normal Nakamoto block attaches to the burn chain state.
+    /// Called before inserting the block into the staging DB.
+    /// Wraps `NakamotoBlock::validate_against_burnchain()`, and
+    /// verifies that all transactions in the block are allowed in this epoch.
+    pub(crate) fn validate_normal_nakamoto_block_burnchain(
+        staging_db: NakamotoStagingBlocksConnRef,
+        db_handle: &SortitionHandleConn,
+        expected_burn: Option<u64>,
+        block: &NakamotoBlock,
+        mainnet: bool,
+        chain_id: u32,
+    ) -> Result<(), ChainstateError> {
+        assert!(!block.is_shadow_block());
+
+        let tenure_burn_chain_tip = Self::validate_nakamoto_tenure_snapshot(db_handle, block)?;
+
+        // block-commit of this sortition
         let Some(block_commit) = db_handle.get_block_commit_by_txid(
             &tenure_burn_chain_tip.sortition_id,
             &tenure_burn_chain_tip.winning_block_txid,
@@ -2229,13 +2307,20 @@ impl NakamotoChainState {
         else {
             warn!(
                 "No block commit for {} in sortition for {}",
-                &tenure_burn_chain_tip.winning_block_txid, &consensus_hash
+                &tenure_burn_chain_tip.winning_block_txid, &block.header.consensus_hash
             );
             return Err(ChainstateError::InvalidStacksBlock(
                 "No block-commit in sortition for block's consensus hash".into(),
             ));
         };
 
+        // if the *parent* of this block is a shadow block, then the block-commit's
+        // parent_vtxindex *MUST* be 0 and the parent_block_ptr *MUST* be the tenure of the
+        // shadow block.
+        //
+        // if the parent is not a shadow block, then this is a no-op.
+        Self::validate_shadow_parent_burnchain(staging_db, db_handle, block, &block_commit)?;
+
         // key register of the winning miner
         let leader_key = db_handle
             .get_leader_key_at(
@@ -2244,40 +2329,42 @@ impl NakamotoChainState {
             )?
             .expect("FATAL: have block commit but no leader key");
 
+        // miner key hash160.
+        let miner_pubkey_hash160 = leader_key
+            .interpret_nakamoto_signing_key()
+            .ok_or(ChainstateError::NoSuchBlockError)
+            .map_err(|e| {
+                warn!(
+                    "Leader key did not contain a hash160 of the miner signing public key";
+                    "leader_key" => ?leader_key,
+                );
+                e
+            })?;
+
         // attaches to burn chain
-        if let Err(e) =
-            block.validate_against_burnchain(&tenure_burn_chain_tip, expected_burn, &leader_key)
-        {
+        if let Err(e) = block.validate_normal_against_burnchain(
+            &tenure_burn_chain_tip,
+            expected_burn,
+            &miner_pubkey_hash160,
+            &leader_key.public_key,
+        ) {
             warn!(
                 "Invalid Nakamoto block, could not validate on burnchain";
-                "consensus_hash" => %consensus_hash,
-                "stacks_block_hash" => %block_hash,
+                "consensus_hash" => %block.header.consensus_hash,
+                "stacks_block_hash" => %block.header.block_hash(),
                 "error" => ?e
             );
 
             return Err(e);
         }
 
-        // check the _next_ block's tenure, since when Nakamoto's miner activates, the current chain tip
-        // will be in epoch 2.5 (the next block will be epoch 3.0)
-        let cur_epoch = SortitionDB::get_stacks_epoch(
+        Self::validate_nakamoto_block_transactions_static(
+            mainnet,
+            chain_id,
             db_handle.deref(),
-            tenure_burn_chain_tip.block_height + 1,
-        )?
-        .expect("FATAL: no epoch defined for current Stacks block");
-
-        // static checks on transactions all pass
-        let valid = block.validate_transactions_static(mainnet, chain_id, cur_epoch.epoch_id);
-        if !valid {
-            warn!(
-                "Invalid Nakamoto block, transactions failed static checks: {}/{} (epoch {})",
-                consensus_hash, block_hash, cur_epoch.epoch_id
-            );
-            return Err(ChainstateError::InvalidStacksBlock(
-                "Invalid Nakamoto block: failed static transaction checks".into(),
-            ));
-        }
-
+            block,
+            tenure_burn_chain_tip.block_height,
+        )?;
         Ok(())
     }
 
@@ -2397,9 +2484,31 @@ impl NakamotoChainState {
         // checked on `::append_block()`
         let expected_burn_opt = Self::get_expected_burns(db_handle, headers_conn, block)?;
 
+        if block.is_shadow_block() {
+            // this block is already present in the staging DB, so just perform some prefunctory
+            // validation (since they're constructed a priori to be valid)
+            if let Err(e) = Self::validate_shadow_nakamoto_block_burnchain(
+                staging_db_tx.conn(),
+                db_handle,
+                expected_burn_opt,
+                block,
+                config.mainnet,
+                config.chain_id,
+            ) {
+                error!("Unacceptable shadow Nakamoto block";
+                      "stacks_block_id" => %block.block_id(),
+                      "error" => ?e
+                );
+                panic!("Unacceptable shadow Nakamoto block");
+            }
+
+            return Ok(false);
+        }
+
         // this block must be consistent with its miner's leader-key and block-commit, and must
         // contain only transactions that are valid in this epoch.
-        if let Err(e) = Self::validate_nakamoto_block_burnchain(
+        if let Err(e) = Self::validate_normal_nakamoto_block_burnchain(
+            staging_db_tx.conn(),
             db_handle,
             expected_burn_opt,
             block,
@@ -2511,6 +2620,24 @@ impl NakamotoChainState {
         Ok(None)
     }
 
+    /// Load the block version of a Nakamoto blocok
+    pub fn get_nakamoto_block_version(
+        chainstate_conn: &Connection,
+        index_block_hash: &StacksBlockId,
+    ) -> Result<Option<u8>, ChainstateError> {
+        let sql = "SELECT version FROM nakamoto_block_headers WHERE index_block_hash = ?1";
+        let args = rusqlite::params![index_block_hash];
+        let mut stmt = chainstate_conn.prepare(sql)?;
+        let result = stmt
+            .query_row(args, |row| {
+                let version: u8 = row.get(0)?;
+                Ok(version)
+            })
+            .optional()?;
+
+        Ok(result)
+    }
+
     /// Load the parent block ID of a Nakamoto block
     pub fn get_nakamoto_parent_block_id(
         chainstate_conn: &Connection,
@@ -2782,6 +2909,12 @@ impl NakamotoChainState {
         consensus_hash: &ConsensusHash,
         block_commit_txid: &Txid,
     ) -> Result<VRFProof, ChainstateError> {
+        // is the tip a shadow block (and necessarily a Nakamoto block)?
+        if let Some(shadow_vrf_proof) = Self::get_shadow_vrf_proof(chainstate_conn, tip_block_id)? {
+            return Ok(shadow_vrf_proof);
+        }
+
+        // parent tenure is a normal tenure
         let sn = SortitionDB::get_block_snapshot_consensus(sortdb_conn, consensus_hash)?.ok_or(
             ChainstateError::InvalidStacksBlock("No sortition for consensus hash".into()),
         )?;
@@ -2803,7 +2936,10 @@ impl NakamotoChainState {
 
         let parent_vrf_proof =
             Self::get_block_vrf_proof(chainstate_conn, tip_block_id, &parent_sn.consensus_hash)?
-                .ok_or(ChainstateError::NoSuchBlockError)
+                .ok_or_else(|| {
+                    warn!("No VRF proof for {}", &parent_sn.consensus_hash);
+                    ChainstateError::NoSuchBlockError
+                })
                 .map_err(|e| {
                     warn!("Could not find parent VRF proof";
                       "tip_block_id" => %tip_block_id,
@@ -2923,6 +3059,11 @@ impl NakamotoChainState {
         sortdb_conn: &Connection,
         block: &NakamotoBlock,
     ) -> Result<(), ChainstateError> {
+        if block.is_shadow_block() {
+            // no-op
+            return Ok(());
+        }
+
         // get the block-commit for this block
         let sn =
             SortitionDB::get_block_snapshot_consensus(sortdb_conn, &block.header.consensus_hash)?
@@ -3526,6 +3667,143 @@ impl NakamotoChainState {
         ))
     }
 
+    /// Begin block-processing for a normal block and return all of the pre-processed state within a
+    /// `SetupBlockResult`.  Used by the Nakamoto miner, and called by Self::setup_normal_block()
+    pub fn setup_block<'a, 'b>(
+        chainstate_tx: &'b mut ChainstateTx,
+        clarity_instance: &'a mut ClarityInstance,
+        sortition_dbconn: &'b dyn SortitionDBRef,
+        first_block_height: u64,
+        pox_constants: &PoxConstants,
+        parent_consensus_hash: ConsensusHash,
+        parent_header_hash: BlockHeaderHash,
+        parent_burn_height: u32,
+        burn_header_hash: BurnchainHeaderHash,
+        burn_header_height: u32,
+        new_tenure: bool,
+        coinbase_height: u64,
+        tenure_extend: bool,
+        block_bitvec: &BitVec<4000>,
+        tenure_block_commit: &LeaderBlockCommitOp,
+        active_reward_set: &RewardSet,
+    ) -> Result<SetupBlockResult<'a, 'b>, ChainstateError> {
+        // this block's bitvec header must match the miner's block commit punishments
+        Self::check_pox_bitvector(block_bitvec, tenure_block_commit, active_reward_set)?;
+        Self::inner_setup_block(
+            chainstate_tx,
+            clarity_instance,
+            sortition_dbconn,
+            first_block_height,
+            pox_constants,
+            parent_consensus_hash,
+            parent_header_hash,
+            parent_burn_height,
+            burn_header_hash,
+            burn_header_height,
+            new_tenure,
+            coinbase_height,
+            tenure_extend,
+        )
+    }
+
+    /// Begin block-processing for a normal block and return all of the pre-processed state within a
+    /// `SetupBlockResult`.
+    ///
+    /// Called as part of block processing
+    fn setup_normal_block_processing<'a, 'b>(
+        chainstate_tx: &'b mut ChainstateTx,
+        clarity_instance: &'a mut ClarityInstance,
+        sortition_dbconn: &'b dyn SortitionDBRef,
+        first_block_height: u64,
+        pox_constants: &PoxConstants,
+        parent_chain_tip: &StacksHeaderInfo,
+        parent_consensus_hash: ConsensusHash,
+        parent_header_hash: BlockHeaderHash,
+        parent_burn_height: u32,
+        tenure_block_snapshot: BlockSnapshot,
+        block: &NakamotoBlock,
+        new_tenure: bool,
+        coinbase_height: u64,
+        tenure_extend: bool,
+        block_bitvec: &BitVec<4000>,
+        active_reward_set: &RewardSet,
+    ) -> Result<SetupBlockResult<'a, 'b>, ChainstateError> {
+        let burn_header_hash = tenure_block_snapshot.burn_header_hash.clone();
+        let burn_header_height =
+            u32::try_from(tenure_block_snapshot.block_height).map_err(|_| {
+                ChainstateError::InvalidStacksBlock(
+                    "Could not downcast burn block height to u32".into(),
+                )
+            })?;
+        let tenure_block_commit = SortitionDB::get_block_commit(
+            sortition_dbconn.sqlite_conn(),
+            &tenure_block_snapshot.winning_block_txid,
+            &tenure_block_snapshot.sortition_id,
+        )?
+        .ok_or_else(|| {
+            warn!("Invalid Nakamoto block: has no block-commit in its sortition";
+                "consensus_hash" => %block.header.consensus_hash,
+                "stacks_block_hash" => %block.header.block_hash(),
+                "stacks_block_id" => %block.header.block_id(),
+                "sortition_id" => %tenure_block_snapshot.sortition_id,
+                "block_commit_txid" => %tenure_block_snapshot.winning_block_txid
+            );
+            ChainstateError::NoSuchBlockError
+        })?;
+
+        // this block's tenure's block-commit contains the hash of the parent tenure's tenure-start
+        // block.
+        // (note that we can't check this earlier, since we need the parent tenure to have been
+        // processed)
+        if new_tenure && parent_chain_tip.is_nakamoto_block() && !block.is_first_mined() {
+            let parent_block_id = StacksBlockId::new(&parent_consensus_hash, &parent_header_hash);
+            let parent_tenure_start_header = Self::get_nakamoto_tenure_start_block_header(
+                chainstate_tx.as_tx(),
+                &parent_block_id,
+                &parent_consensus_hash,
+            )?
+            .ok_or_else(|| {
+                warn!("Invalid Nakamoto block: no start-tenure block for parent";
+                      "parent_consensus_hash" => %parent_consensus_hash,
+                      "consensus_hash" => %block.header.consensus_hash,
+                      "stacks_block_hash" => %block.header.block_hash(),
+                      "stacks_block_id" => %block.header.block_id());
+                ChainstateError::NoSuchBlockError
+            })?;
+
+            if parent_tenure_start_header.index_block_hash() != tenure_block_commit.last_tenure_id()
+            {
+                warn!("Invalid Nakamoto block: its tenure's block-commit's block ID hash does not match its parent tenure's start block";
+                    "parent_consensus_hash" => %parent_consensus_hash,
+                    "parent_tenure_start_block_id" => %parent_tenure_start_header.index_block_hash(),
+                    "block_commit.last_tenure_id" => %tenure_block_commit.last_tenure_id(),
+                    "parent_tip" => %parent_block_id,
+                );
+                test_debug!("Faulty commit: {:?}", &tenure_block_commit);
+
+                return Err(ChainstateError::NoSuchBlockError);
+            }
+        }
+        Self::setup_block(
+            chainstate_tx,
+            clarity_instance,
+            sortition_dbconn,
+            first_block_height,
+            pox_constants,
+            parent_consensus_hash,
+            parent_header_hash,
+            parent_burn_height,
+            burn_header_hash,
+            burn_header_height,
+            new_tenure,
+            coinbase_height,
+            tenure_extend,
+            block_bitvec,
+            &tenure_block_commit,
+            active_reward_set,
+        )
+    }
+
     /// Begin block-processing and return all of the pre-processed state within a
     /// `SetupBlockResult`.
     ///
@@ -3550,15 +3828,12 @@ impl NakamotoChainState {
     /// * coinbase_height: the number of tenures that this block confirms (including epoch2 blocks)
     ///   (this is equivalent to the number of coinbases)
     /// * tenure_extend: whether or not to reset the tenure's ongoing execution cost
-    /// * block_bitvec: the bitvec that will control PoX reward handling for this block
-    /// * tenure_block_commit: the block commit that elected this miner
-    /// * active_reward_set: the reward and signer set active during `tenure_block_commit`
     ///
     /// Returns clarity_tx, list of receipts, microblock execution cost,
     /// microblock fees, microblock burns, list of microblock tx receipts,
     /// miner rewards tuples, the stacks epoch id, and a boolean that
     /// represents whether the epoch transition has been applied.
-    pub fn setup_block<'a, 'b>(
+    fn inner_setup_block<'a, 'b>(
         chainstate_tx: &'b mut ChainstateTx,
         clarity_instance: &'a mut ClarityInstance,
         sortition_dbconn: &'b dyn SortitionDBRef,
@@ -3566,19 +3841,13 @@ impl NakamotoChainState {
         pox_constants: &PoxConstants,
         parent_consensus_hash: ConsensusHash,
         parent_header_hash: BlockHeaderHash,
-        _parent_stacks_height: u64,
         parent_burn_height: u32,
         burn_header_hash: BurnchainHeaderHash,
         burn_header_height: u32,
         new_tenure: bool,
         coinbase_height: u64,
         tenure_extend: bool,
-        block_bitvec: &BitVec<4000>,
-        tenure_block_commit: &LeaderBlockCommitOp,
-        active_reward_set: &RewardSet,
     ) -> Result<SetupBlockResult<'a, 'b>, ChainstateError> {
-        Self::check_pox_bitvector(block_bitvec, tenure_block_commit, active_reward_set)?;
-
         let parent_index_hash = StacksBlockId::new(&parent_consensus_hash, &parent_header_hash);
         let parent_sortition_id = sortition_dbconn
             .get_sortition_id_from_consensus_hash(&parent_consensus_hash)
@@ -3814,79 +4083,84 @@ impl NakamotoChainState {
         Ok(lockup_events)
     }
 
+    /// Verify that the PoX bitvector from the block header is consistent with the block-commit's
+    /// PoX outputs, as determined by the active reward set and whether or not the 0's in the
+    /// bitvector correspond to signers' PoX outputs.
     fn check_pox_bitvector(
         block_bitvec: &BitVec<4000>,
         tenure_block_commit: &LeaderBlockCommitOp,
         active_reward_set: &RewardSet,
     ) -> Result<(), ChainstateError> {
-        if !tenure_block_commit.treatment.is_empty() {
-            let address_to_indeces: HashMap<_, Vec<_>> = active_reward_set
-                .rewarded_addresses
+        if tenure_block_commit.treatment.is_empty() {
+            return Ok(());
+        }
+
+        let address_to_indeces: HashMap<_, Vec<_>> = active_reward_set
+            .rewarded_addresses
+            .iter()
+            .enumerate()
+            .fold(HashMap::new(), |mut map, (ix, addr)| {
+                map.entry(addr).or_insert_with(Vec::new).push(ix);
+                map
+            });
+
+        // our block commit issued a punishment, check the reward set and bitvector
+        //  to ensure that this was valid.
+        for treated_addr in tenure_block_commit.treatment.iter() {
+            if treated_addr.is_burn() {
+                // Don't need to assert anything about burn addresses.
+                // If they were in the reward set, "punishing" them is meaningless.
+                continue;
+            }
+            // otherwise, we need to find the indices in the rewarded_addresses
+            //  corresponding to this address.
+            let empty_vec = vec![];
+            let address_indices = address_to_indeces
+                .get(treated_addr.deref())
+                .unwrap_or(&empty_vec);
+
+            // if any of them are 0, punishment is okay.
+            // if all of them are 1, punishment is not okay.
+            // if all of them are 0, *must* have punished
+            let bitvec_values: Result<Vec<_>, ChainstateError> = address_indices
                 .iter()
-                .enumerate()
-                .fold(HashMap::new(), |mut map, (ix, addr)| {
-                    map.entry(addr).or_insert_with(Vec::new).push(ix);
-                    map
-                });
-
-            // our block commit issued a punishment, check the reward set and bitvector
-            //  to ensure that this was valid.
-            for treated_addr in tenure_block_commit.treatment.iter() {
-                if treated_addr.is_burn() {
-                    // Don't need to assert anything about burn addresses.
-                    // If they were in the reward set, "punishing" them is meaningless.
-                    continue;
-                }
-                // otherwise, we need to find the indices in the rewarded_addresses
-                //  corresponding to this address.
-                let empty_vec = vec![];
-                let address_indices = address_to_indeces
-                    .get(treated_addr.deref())
-                    .unwrap_or(&empty_vec);
-
-                // if any of them are 0, punishment is okay.
-                // if all of them are 1, punishment is not okay.
-                // if all of them are 0, *must* have punished
-                let bitvec_values: Result<Vec<_>, ChainstateError> = address_indices
-                    .iter()
-                    .map(
-                        |ix| {
-                            let ix = u16::try_from(*ix)
-                                .map_err(|_| ChainstateError::InvalidStacksBlock("Reward set index outside of u16".into()))?;
-                            let bitvec_value = block_bitvec.get(ix)
-                                .unwrap_or_else(|| {
-                                    warn!("Block header's bitvec is smaller than the reward set, defaulting higher indexes to 1");
-                                    true
-                                });
-                            Ok(bitvec_value)
-                        }
-                    )
-                    .collect();
-                let bitvec_values = bitvec_values?;
-                let all_1 = bitvec_values.iter().all(|x| *x);
-                let all_0 = bitvec_values.iter().all(|x| !x);
-                if all_1 {
-                    if treated_addr.is_punish() {
-                        warn!(
-                            "Invalid Nakamoto block: punished PoX address when bitvec contained 1s for the address";
-                            "reward_address" => %treated_addr.deref(),
-                            "bitvec_values" => ?bitvec_values,
-                        );
-                        return Err(ChainstateError::InvalidStacksBlock(
-                            "Bitvec does not match the block commit's PoX handling".into(),
-                        ));
-                    }
-                } else if all_0 {
-                    if treated_addr.is_reward() {
-                        warn!(
-                            "Invalid Nakamoto block: rewarded PoX address when bitvec contained 0s for the address";
-                            "reward_address" => %treated_addr.deref(),
-                            "bitvec_values" => ?bitvec_values,
-                        );
-                        return Err(ChainstateError::InvalidStacksBlock(
-                            "Bitvec does not match the block commit's PoX handling".into(),
-                        ));
+                .map(
+                    |ix| {
+                        let ix = u16::try_from(*ix)
+                            .map_err(|_| ChainstateError::InvalidStacksBlock("Reward set index outside of u16".into()))?;
+                        let bitvec_value = block_bitvec.get(ix)
+                            .unwrap_or_else(|| {
+                                warn!("Block header's bitvec is smaller than the reward set, defaulting higher indexes to 1");
+                                true
+                            });
+                        Ok(bitvec_value)
                     }
+                )
+                .collect();
+            let bitvec_values = bitvec_values?;
+            let all_1 = bitvec_values.iter().all(|x| *x);
+            let all_0 = bitvec_values.iter().all(|x| !x);
+            if all_1 {
+                if treated_addr.is_punish() {
+                    warn!(
+                        "Invalid Nakamoto block: punished PoX address when bitvec contained 1s for the address";
+                        "reward_address" => %treated_addr.deref(),
+                        "bitvec_values" => ?bitvec_values,
+                    );
+                    return Err(ChainstateError::InvalidStacksBlock(
+                        "Bitvec does not match the block commit's PoX handling".into(),
+                    ));
+                }
+            } else if all_0 {
+                if treated_addr.is_reward() {
+                    warn!(
+                        "Invalid Nakamoto block: rewarded PoX address when bitvec contained 0s for the address";
+                        "reward_address" => %treated_addr.deref(),
+                        "bitvec_values" => ?bitvec_values,
+                    );
+                    return Err(ChainstateError::InvalidStacksBlock(
+                        "Bitvec does not match the block commit's PoX handling".into(),
+                    ));
                 }
             }
         }
@@ -4013,8 +4287,6 @@ impl NakamotoChainState {
         // It must exist in the same Bitcoin fork as our `burn_dbconn`.
         let tenure_block_snapshot =
             Self::check_sortition_exists(burn_dbconn, &block.header.consensus_hash)?;
-        let burn_header_hash = tenure_block_snapshot.burn_header_hash.clone();
-        let burn_header_height = tenure_block_snapshot.block_height;
         let block_hash = block.header.block_hash();
 
         let new_tenure = block.is_wellformed_tenure_start_block().map_err(|_| {
@@ -4091,54 +4363,6 @@ impl NakamotoChainState {
             ));
         }
 
-        // this block's bitvec header must match the miner's block commit punishments
-        let tenure_block_commit = SortitionDB::get_block_commit(
-            burn_dbconn.conn(),
-            &tenure_block_snapshot.winning_block_txid,
-            &tenure_block_snapshot.sortition_id,
-        )?
-        .ok_or_else(|| {
-            warn!("Invalid Nakamoto block: has no block-commit in its sortition";
-                "consensus_hash" => %block.header.consensus_hash,
-                "stacks_block_hash" => %block.header.block_hash(),
-                "stacks_block_id" => %block.header.block_id(),
-                "sortition_id" => %tenure_block_snapshot.sortition_id,
-                "block_commit_txid" => %tenure_block_snapshot.winning_block_txid
-            );
-            ChainstateError::NoSuchBlockError
-        })?;
-
-        // this block's tenure's block-commit contains the hash of the parent tenure's tenure-start
-        // block.
-        // (note that we can't check this earlier, since we need the parent tenure to have been
-        // processed)
-        if new_tenure && parent_chain_tip.is_nakamoto_block() && !block.is_first_mined() {
-            let parent_tenure_start_header = Self::get_nakamoto_tenure_start_block_header(
-                chainstate_tx.as_tx(),
-                &parent_block_id,
-                &parent_ch,
-            )?
-            .ok_or_else(|| {
-                warn!("Invalid Nakamoto block: no start-tenure block for parent";
-                      "parent_consensus_hash" => %parent_ch,
-                      "consensus_hash" => %block.header.consensus_hash,
-                      "stacks_block_hash" => %block.header.block_hash(),
-                      "stacks_block_id" => %block.header.block_id());
-                ChainstateError::NoSuchBlockError
-            })?;
-
-            if parent_tenure_start_header.index_block_hash() != tenure_block_commit.last_tenure_id()
-            {
-                warn!("Invalid Nakamoto block: its tenure's block-commit's block ID hash does not match its parent tenure's start block";
-                    "parent_consensus_hash" => %parent_ch,
-                    "parent_tenure_start_block_id" => %parent_tenure_start_header.index_block_hash(),
-                    "block_commit.last_tenure_id" => %tenure_block_commit.last_tenure_id()
-                );
-
-                return Err(ChainstateError::NoSuchBlockError);
-            }
-        }
-
         // verify VRF proof, if present
         // only need to do this once per tenure
         // get the resulting vrf proof bytes
@@ -4198,27 +4422,43 @@ impl NakamotoChainState {
             mut auto_unlock_events,
             signer_set_calc,
             burn_vote_for_aggregate_key_ops,
-        } = Self::setup_block(
-            chainstate_tx,
-            clarity_instance,
-            burn_dbconn,
-            first_block_height,
-            pox_constants,
-            parent_ch,
-            parent_block_hash,
-            parent_chain_tip.stacks_block_height,
-            parent_chain_tip.burn_header_height,
-            burn_header_hash,
-            burn_header_height.try_into().map_err(|_| {
-                ChainstateError::InvalidStacksBlock("Burn block height exceeded u32".into())
-            })?,
-            new_tenure,
-            coinbase_height,
-            tenure_extend,
-            &block.header.pox_treatment,
-            &tenure_block_commit,
-            active_reward_set,
-        )?;
+        } = if block.is_shadow_block() {
+            // shadow block
+            Self::setup_shadow_block_processing(
+                chainstate_tx,
+                clarity_instance,
+                burn_dbconn,
+                first_block_height,
+                pox_constants,
+                parent_ch,
+                parent_block_hash,
+                parent_chain_tip.burn_header_height,
+                tenure_block_snapshot,
+                new_tenure,
+                coinbase_height,
+                tenure_extend,
+            )?
+        } else {
+            // normal block
+            Self::setup_normal_block_processing(
+                chainstate_tx,
+                clarity_instance,
+                burn_dbconn,
+                first_block_height,
+                pox_constants,
+                &parent_chain_tip,
+                parent_ch,
+                parent_block_hash,
+                parent_chain_tip.burn_header_height,
+                tenure_block_snapshot,
+                block,
+                new_tenure,
+                coinbase_height,
+                tenure_extend,
+                &block.header.pox_treatment,
+                active_reward_set,
+            )?
+        };
 
         let starting_cost = clarity_tx.cost_so_far();
 
diff --git a/stackslib/src/chainstate/nakamoto/shadow.rs b/stackslib/src/chainstate/nakamoto/shadow.rs
new file mode 100644
index 0000000000..cdc099e120
--- /dev/null
+++ b/stackslib/src/chainstate/nakamoto/shadow.rs
@@ -0,0 +1,1008 @@
+// Copyright (C) 2013-2020 Blockstack PBC, a public benefit corporation
+// Copyright (C) 2020-2024 Stacks Open Internet Foundation
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// 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 clarity::vm::costs::ExecutionCost;
+use rusqlite::params;
+/// Shadow blocks
+///
+/// In the event of an emergency chain halt, a SIP will be written to declare that a chain halt has
+/// happened, and what transactions and blocks (if any) need to be mined at which burnchain block
+/// heights to recover the chain.
+///
+/// If this remedy is necessary, these blocks will be mined into one or more _shadow_ blocks and
+/// _shadow_ tenures.
+///
+/// Shadow blocks are blocks that are inserted directly into the staging blocks DB as part of a
+/// schema update. They are neither mined nor relayed.  Instead, they are synthesized as part of an
+/// emergency node upgrade in order to ensure that the conditions which lead to the chain stall
+/// never occur.
+///
+/// For example, if a prepare phase is mined without a single block-commit hitting the Bitcoin
+/// chain, a pair of shadow block tenures will be synthesized to create a PoX anchor block and
+/// restore the chain's liveness.  As another example, if insufficiently many STX are locked in PoX
+/// to get a healthy set of signers, a shadow block can be synthesized with extra `stack-stx`
+/// transactions submitted from healthy stackers in order to create a suitable PoX reward set.
+///
+/// This module contains shadow block-specific logic for the Nakamoto block header, Nakamoto block,
+/// Nakamoto chainstate, and Nakamoto miner structures.
+use rusqlite::Connection;
+use stacks_common::codec::StacksMessageCodec;
+use stacks_common::types::chainstate::{
+    BlockHeaderHash, ConsensusHash, StacksAddress, StacksBlockId, StacksPrivateKey, StacksPublicKey,
+};
+use stacks_common::util::hash::Hash160;
+use stacks_common::util::vrf::VRFProof;
+
+use crate::burnchains::PoxConstants;
+use crate::chainstate::nakamoto::miner::{MinerTenureInfo, NakamotoBlockBuilder};
+use crate::chainstate::nakamoto::{
+    BlockSnapshot, ChainstateError, LeaderBlockCommitOp, NakamotoBlock, NakamotoBlockHeader,
+    NakamotoBlockObtainMethod, NakamotoChainState, NakamotoStagingBlocksConn,
+    NakamotoStagingBlocksConnRef, NakamotoStagingBlocksTx, SetupBlockResult, SortitionDB,
+    SortitionHandleConn, StacksDBIndexed,
+};
+use crate::chainstate::stacks::boot::RewardSet;
+use crate::chainstate::stacks::db::blocks::DummyEventDispatcher;
+use crate::chainstate::stacks::db::{
+    ChainstateTx, ClarityTx, StacksAccount, StacksChainState, StacksHeaderInfo,
+};
+use crate::chainstate::stacks::miner::{
+    BlockBuilder, BlockLimitFunction, TransactionError, TransactionProblematic, TransactionResult,
+    TransactionSkipped,
+};
+use crate::chainstate::stacks::{
+    CoinbasePayload, Error, StacksTransaction, StacksTransactionSigner, TenureChangeCause,
+    TenureChangePayload, TransactionAnchorMode, TransactionAuth, TransactionPayload,
+    TransactionVersion,
+};
+use crate::clarity::vm::types::StacksAddressExtensions;
+use crate::clarity_vm::clarity::ClarityInstance;
+use crate::clarity_vm::database::SortitionDBRef;
+use crate::net::Error as NetError;
+use crate::util_lib::db::{query_row, u64_to_sql, Error as DBError};
+
+impl NakamotoBlockHeader {
+    /// Is this a shadow block?
+    ///
+    /// This is a special kind of block that is directly inserted into the chainstate by means of a
+    /// consensus rule.  It won't be downloaded or broadcasted, but every node will have it.  They
+    /// get created as a result of a consensus-level SIP in order to restore the chain to working
+    /// order.
+    ///
+    /// Shadow blocks have the high bit of their version field set.
+    pub fn is_shadow_block(&self) -> bool {
+        Self::is_shadow_block_version(self.version)
+    }
+
+    /// Is a block version a shadow block version?
+    pub fn is_shadow_block_version(version: u8) -> bool {
+        version & 0x80 != 0
+    }
+
+    /// Get the signing weight of a shadow block
+    pub fn get_shadow_signer_weight(&self, reward_set: &RewardSet) -> Result<u32, Error> {
+        let Some(signers) = &reward_set.signers else {
+            return Err(ChainstateError::InvalidStacksBlock(
+                "No signers in the reward set".to_string(),
+            ));
+        };
+        let shadow_weight = signers
+            .iter()
+            .fold(0u32, |acc, signer| acc.saturating_add(signer.weight));
+
+        Ok(shadow_weight)
+    }
+}
+
+impl NakamotoBlock {
+    /// Is this block a shadow block?
+    /// Check the header
+    pub fn is_shadow_block(&self) -> bool {
+        self.header.is_shadow_block()
+    }
+
+    /// Verify that if this shadow block has a coinbase, that its VRF proof is consistent with the leader
+    /// public key's VRF key. If there is no coinbase tx, then this is a no-op.
+    pub(crate) fn check_shadow_coinbase_tx(&self, mainnet: bool) -> Result<(), ChainstateError> {
+        if !self.is_shadow_block() {
+            error!(
+                "FATAL: tried to validate non-shadow block in a shadow-block-specific validator"
+            );
+            panic!();
+        }
+
+        // If this shadow block has a coinbase, then verify that it has a VRF proof (which will be
+        // verified later) and that its recipient is the burn address.  Shadow blocks do not award
+        // STX.
+        if let Some(coinbase_tx) = self.get_coinbase_tx() {
+            let (_, recipient_opt, vrf_proof_opt) = coinbase_tx
+                .try_as_coinbase()
+                .expect("FATAL: `get_coinbase_tx()` did not return a coinbase");
+
+            if vrf_proof_opt.is_none() {
+                return Err(ChainstateError::InvalidStacksBlock(
+                    "Shadow Nakamoto coinbase must have a VRF proof".into(),
+                ));
+            }
+
+            let Some(recipient) = recipient_opt else {
+                warn!("Invalid shadow block: no recipient");
+                return Err(ChainstateError::InvalidStacksBlock(
+                    "Shadow block did not pay to burn address".into(),
+                ));
+            };
+
+            // must be the standard burn address for this network
+            let burn_addr = StacksAddress::burn_address(mainnet).to_account_principal();
+            if burn_addr != *recipient {
+                warn!("Invalid shadow block: recipient does not burn");
+                return Err(ChainstateError::InvalidStacksBlock(
+                    "Shadow block did not pay to burn address".into(),
+                ));
+            }
+
+            // can't check the VRF proof because the creator of the shadow block (e.g. the SIP
+            // process) isn't a miner, so it could be anything.
+        }
+        Ok(())
+    }
+
+    /// Validate this Nakamoto shadow block header against burnchain state.
+    ///
+    /// Arguments
+    /// -- `mainnet`: whether or not the chain is mainnet
+    /// -- `tenure_burn_chain_tip` is the BlockSnapshot containing the block-commit for this block's
+    /// tenure.  It is not always the tip of the burnchain.
+    /// -- `expected_burn` is the total number of burnchain tokens spent, if known.
+    ///
+    /// Verifies the following:
+    /// -- (self.header.consensus_hash) that this block falls into this block-commit's tenure
+    /// -- (self.header.burn_spent) that this block's burn total matches `burn_tip`'s total burn
+    /// -- if this block has a tenure change, then it's consistent with the miner's public key and
+    /// self.header.consensus_hash
+    ///
+    /// NOTE: unlike normal blocks, we do not need to verify the VRF proof or miner signature
+    pub(crate) fn validate_shadow_against_burnchain(
+        &self,
+        mainnet: bool,
+        tenure_burn_chain_tip: &BlockSnapshot,
+        expected_burn: Option<u64>,
+    ) -> Result<(), ChainstateError> {
+        if !self.is_shadow_block() {
+            error!(
+                "FATAL: tried to validate non-shadow block in a shadow-block-specific validator"
+            );
+            panic!();
+        }
+        self.common_validate_against_burnchain(tenure_burn_chain_tip, expected_burn)?;
+        self.check_tenure_tx()?;
+        self.check_shadow_coinbase_tx(mainnet)?;
+
+        // not verified by this method:
+        // * chain_length       (need parent block header)
+        // * parent_block_id    (need parent block header)
+        // * block-commit seed  (need parent block)
+        // * tx_merkle_root     (already verified; validated on deserialization)
+        // * state_index_root   (validated on process_block())
+        // * stacker signature  (validated on accept_block())
+        Ok(())
+    }
+}
+
+impl NakamotoChainState {
+    /// Verify that the shadow parent of a normal block is consistent with the normal block's
+    /// tenure's block-commit.
+    ///
+    /// * the block-commit vtxindex must be 0 (i.e. burnchain coinbase)
+    /// * the block-commit block ptr must be the shadow parent tenure's sortition
+    ///
+    /// Returns Ok(()) if the parent is _not_ a shadow block
+    /// Returns Ok(()) if the parent is a shadow block, and the above criteria are met
+    /// Returns Err(ChainstateError::InvalidStacksBlock(..)) if the parent is a shadow block, and
+    /// some of the criteria above are false
+    /// Returns Err(..) on other (DB-related) errors
+    pub(crate) fn validate_shadow_parent_burnchain(
+        staging_db: NakamotoStagingBlocksConnRef,
+        db_handle: &SortitionHandleConn,
+        block: &NakamotoBlock,
+        block_commit: &LeaderBlockCommitOp,
+    ) -> Result<(), ChainstateError> {
+        // only applies if the parent is a nakamoto block (since all shadow blocks are nakamoto
+        // blocks)
+        let Some(parent_header) =
+            staging_db.get_nakamoto_block_header(&block.header.parent_block_id)?
+        else {
+            return Ok(());
+        };
+
+        if !parent_header.is_shadow_block() {
+            return Ok(());
+        }
+
+        if block_commit.parent_vtxindex != 0 {
+            warn!("Invalid Nakamoto block: parent {} of {} is a shadow block but block-commit vtxindex is {}", &parent_header.block_id(), &block.block_id(), block_commit.parent_vtxindex);
+            return Err(ChainstateError::InvalidStacksBlock("Invalid Nakamoto block: invalid block-commit parent vtxindex for parent shadow block".into()));
+        }
+        let Some(parent_sn) =
+            SortitionDB::get_block_snapshot_consensus(db_handle, &parent_header.consensus_hash)?
+        else {
+            warn!(
+                "Invalid Nakamoto block: No sortition for parent shadow block {}",
+                &block.header.parent_block_id
+            );
+            return Err(ChainstateError::InvalidStacksBlock(
+                "Invalid Nakamoto block: parent shadow block has no sortition".into(),
+            ));
+        };
+        if u64::from(block_commit.parent_block_ptr) != parent_sn.block_height {
+            warn!("Invalid Nakamoto block: parent {} of {} is a shadow block but block-commit parent ptr is {}", &parent_header.block_id(), &block.block_id(), block_commit.parent_block_ptr);
+            return Err(ChainstateError::InvalidStacksBlock("Invalid Nakamoto block: invalid block-commit parent block ptr for parent shadow block".into()));
+        }
+
+        Ok(())
+    }
+
+    /// Validate a shadow Nakamoto block against burnchain state.
+    /// Wraps `NakamotoBlock::validate_shadow_against_burnchain()`, and
+    /// verifies that all transactions in the block are allowed in this epoch.
+    pub(crate) fn validate_shadow_nakamoto_block_burnchain(
+        staging_db: NakamotoStagingBlocksConnRef,
+        db_handle: &SortitionHandleConn,
+        expected_burn: Option<u64>,
+        block: &NakamotoBlock,
+        mainnet: bool,
+        chain_id: u32,
+    ) -> Result<(), ChainstateError> {
+        if !block.is_shadow_block() {
+            error!(
+                "FATAL: tried to validate non-shadow block in a shadow-block-specific validator"
+            );
+            panic!();
+        }
+
+        // this block must already be stored
+        if !staging_db.has_shadow_nakamoto_block_with_index_hash(&block.block_id())? {
+            warn!("Invalid shadow Nakamoto block, must already be stored";
+                "consensus_hash" => %block.header.consensus_hash,
+                "stacks_block_hash" => %block.header.block_hash(),
+                "block_id" => %block.header.block_id()
+            );
+
+            return Err(ChainstateError::InvalidStacksBlock(
+                "Shadow block must already be stored".into(),
+            ));
+        }
+
+        let tenure_burn_chain_tip = Self::validate_nakamoto_tenure_snapshot(db_handle, block)?;
+        if let Err(e) =
+            block.validate_shadow_against_burnchain(mainnet, &tenure_burn_chain_tip, expected_burn)
+        {
+            warn!(
+                "Invalid shadow Nakamoto block, could not validate on burnchain";
+                "consensus_hash" => %block.header.consensus_hash,
+                "stacks_block_hash" => %block.header.block_hash(),
+                "block_id" => %block.header.block_id(),
+                "error" => ?e
+            );
+
+            return Err(e);
+        }
+        Self::validate_nakamoto_block_transactions_static(
+            mainnet,
+            chain_id,
+            db_handle.conn(),
+            block,
+            tenure_burn_chain_tip.block_height,
+        )?;
+        Ok(())
+    }
+
+    /// Load the stored VRF proof for the given shadow block's tenure.
+    ///
+    /// Returns Ok(Some(vrf proof)) on success
+    /// Returns Ok(None) if the parent tenure isn't a shadow tenure
+    pub(crate) fn get_shadow_vrf_proof<SDBI: StacksDBIndexed>(
+        chainstate_conn: &mut SDBI,
+        tip_block_id: &StacksBlockId,
+    ) -> Result<Option<VRFProof>, ChainstateError> {
+        // is the tip a shadow block (and necessarily a Nakamoto block)?
+        let Some(parent_version) =
+            NakamotoChainState::get_nakamoto_block_version(chainstate_conn.sqlite(), tip_block_id)?
+        else {
+            return Ok(None);
+        };
+
+        if !NakamotoBlockHeader::is_shadow_block_version(parent_version) {
+            return Ok(None);
+        }
+
+        // this is a shadow block
+        let tenure_consensus_hash = NakamotoChainState::get_block_header_nakamoto_tenure_id(
+            chainstate_conn.sqlite(),
+            tip_block_id,
+        )?
+        .ok_or_else(|| {
+            warn!("No tenure consensus hash for block {}", tip_block_id);
+            ChainstateError::NoSuchBlockError
+        })?;
+
+        // the shadow tenure won't have a block-commit, but we just found its tenure ID anyway
+        debug!(
+            "Load VRF proof for shadow tenure {}",
+            &tenure_consensus_hash
+        );
+        let vrf_proof =
+            Self::get_block_vrf_proof(chainstate_conn, tip_block_id, &tenure_consensus_hash)?
+                .ok_or_else(|| {
+                    warn!("No VRF proof for {}", &tenure_consensus_hash);
+                    ChainstateError::NoSuchBlockError
+                })
+                .map_err(|e| {
+                    warn!("Could not find shadow tenure VRF proof";
+                      "tip_block_id" => %tip_block_id,
+                      "shadow consensus_hash" => %tenure_consensus_hash);
+                    e
+                })?;
+
+        return Ok(Some(vrf_proof));
+    }
+
+    /// Begin block-processing for a shadow block and return all of the pre-processed state within a
+    /// `SetupBlockResult`.
+    ///
+    /// Called to begin processing a shadow block
+    pub(crate) fn setup_shadow_block_processing<'a, 'b>(
+        chainstate_tx: &'b mut ChainstateTx,
+        clarity_instance: &'a mut ClarityInstance,
+        sortition_dbconn: &'b dyn SortitionDBRef,
+        first_block_height: u64,
+        pox_constants: &PoxConstants,
+        parent_consensus_hash: ConsensusHash,
+        parent_header_hash: BlockHeaderHash,
+        parent_burn_height: u32,
+        tenure_block_snapshot: BlockSnapshot,
+        new_tenure: bool,
+        coinbase_height: u64,
+        tenure_extend: bool,
+    ) -> Result<SetupBlockResult<'a, 'b>, ChainstateError> {
+        let burn_header_hash = &tenure_block_snapshot.burn_header_hash;
+        let burn_header_height =
+            u32::try_from(tenure_block_snapshot.block_height).map_err(|_| {
+                ChainstateError::InvalidStacksBlock(
+                    "Failed to downcast burn block height to u32".into(),
+                )
+            })?;
+        let block_consensus_hash = &tenure_block_snapshot.consensus_hash;
+
+        let parent_block_id = StacksBlockId::new(&parent_consensus_hash, &parent_header_hash);
+
+        // tenure start header must exist and be processed
+        let _ = Self::get_nakamoto_tenure_start_block_header(
+            chainstate_tx.as_tx(),
+            &parent_block_id,
+            &parent_consensus_hash,
+        )?
+        .ok_or_else(|| {
+            warn!("Invalid shadow Nakamoto block: no start-tenure block for parent";
+                  "parent_consensus_hash" => %parent_consensus_hash,
+                  "consensus_hash" => %block_consensus_hash
+            );
+            ChainstateError::NoSuchBlockError
+        })?;
+
+        Self::inner_setup_block(
+            chainstate_tx,
+            clarity_instance,
+            sortition_dbconn,
+            first_block_height,
+            pox_constants,
+            parent_consensus_hash,
+            parent_header_hash,
+            parent_burn_height,
+            burn_header_hash.clone(),
+            burn_header_height,
+            new_tenure,
+            coinbase_height,
+            tenure_extend,
+        )
+    }
+}
+
+impl NakamotoBlockBuilder {
+    /// This function should be called before `tenure_begin`.
+    /// It creates a MinerTenureInfo struct which owns connections to the chainstate and sortition
+    /// DBs, so that block-processing is guaranteed to terminate before the lives of these handles
+    /// expire.
+    ///
+    /// It's used to create shadow blocks.
+    pub(crate) fn shadow_load_tenure_info<'a>(
+        &self,
+        chainstate: &'a mut StacksChainState,
+        burn_dbconn: &'a SortitionHandleConn,
+        cause: Option<TenureChangeCause>,
+    ) -> Result<MinerTenureInfo<'a>, Error> {
+        self.inner_load_tenure_info(chainstate, burn_dbconn, cause, true)
+    }
+
+    /// Begin/resume mining a shadow tenure's transactions.
+    /// Returns an open ClarityTx for mining the block.
+    /// NOTE: even though we don't yet know the block hash, the Clarity VM ensures that a
+    /// transaction can't query information about the _current_ block (i.e. information that is not
+    /// yet known).
+    pub fn shadow_tenure_begin<'a, 'b>(
+        &mut self,
+        burn_dbconn: &'a SortitionHandleConn,
+        info: &'b mut MinerTenureInfo<'a>,
+        tenure_id_consensus_hash: &ConsensusHash,
+    ) -> Result<ClarityTx<'b, 'b>, Error> {
+        let tenure_snapshot = SortitionDB::get_block_snapshot_consensus(
+            burn_dbconn.conn(),
+            tenure_id_consensus_hash,
+        )?
+        .ok_or_else(|| Error::NoSuchBlockError)?;
+
+        let SetupBlockResult {
+            clarity_tx,
+            matured_miner_rewards_opt,
+            ..
+        } = NakamotoChainState::setup_shadow_block_processing(
+            &mut info.chainstate_tx,
+            info.clarity_instance,
+            burn_dbconn,
+            burn_dbconn.context.first_block_height,
+            &burn_dbconn.context.pox_constants,
+            info.parent_consensus_hash,
+            info.parent_header_hash,
+            info.parent_burn_block_height,
+            tenure_snapshot,
+            info.cause == Some(TenureChangeCause::BlockFound),
+            info.coinbase_height,
+            info.cause == Some(TenureChangeCause::Extended),
+        )?;
+        self.matured_miner_rewards_opt = matured_miner_rewards_opt;
+        Ok(clarity_tx)
+    }
+
+    /// Get an address's account
+    pub fn get_account(
+        chainstate: &mut StacksChainState,
+        sortdb: &SortitionDB,
+        addr: &StacksAddress,
+        tip: &StacksHeaderInfo,
+    ) -> Result<StacksAccount, Error> {
+        let snapshot =
+            SortitionDB::get_block_snapshot_consensus(&sortdb.conn(), &tip.consensus_hash)?
+                .ok_or_else(|| Error::NoSuchBlockError)?;
+
+        let account = chainstate
+            .with_read_only_clarity_tx(
+                &sortdb.index_handle(&snapshot.sortition_id),
+                &tip.index_block_hash(),
+                |clarity_conn| {
+                    StacksChainState::get_account(clarity_conn, &addr.to_account_principal())
+                },
+            )
+            .ok_or_else(|| Error::NoSuchBlockError)?;
+
+        Ok(account)
+    }
+
+    /// Make a shadow block from transactions
+    pub fn make_shadow_block_from_txs(
+        mut builder: NakamotoBlockBuilder,
+        chainstate_handle: &StacksChainState,
+        burn_dbconn: &SortitionHandleConn,
+        tenure_id_consensus_hash: &ConsensusHash,
+        mut txs: Vec<StacksTransaction>,
+    ) -> Result<(NakamotoBlock, u64, ExecutionCost), Error> {
+        use clarity::vm::ast::ASTRules;
+
+        debug!(
+            "Build shadow Nakamoto block from {} transactions",
+            txs.len()
+        );
+        let (mut chainstate, _) = chainstate_handle.reopen()?;
+
+        let mut tenure_cause = None;
+        for tx in txs.iter() {
+            let TransactionPayload::TenureChange(payload) = &tx.payload else {
+                continue;
+            };
+            tenure_cause = Some(payload.cause);
+            break;
+        }
+
+        let mut miner_tenure_info =
+            builder.shadow_load_tenure_info(&mut chainstate, burn_dbconn, tenure_cause)?;
+        let mut tenure_tx = builder.shadow_tenure_begin(
+            burn_dbconn,
+            &mut miner_tenure_info,
+            tenure_id_consensus_hash,
+        )?;
+        for tx in txs.drain(..) {
+            let tx_len = tx.tx_len();
+            match builder.try_mine_tx_with_len(
+                &mut tenure_tx,
+                &tx,
+                tx_len,
+                &BlockLimitFunction::NO_LIMIT_HIT,
+                ASTRules::PrecheckSize,
+            ) {
+                TransactionResult::Success(..) => {
+                    debug!("Included {}", &tx.txid());
+                }
+                TransactionResult::Skipped(TransactionSkipped { error, .. })
+                | TransactionResult::ProcessingError(TransactionError { error, .. }) => {
+                    match error {
+                        Error::BlockTooBigError => {
+                            // done mining -- our execution budget is exceeded.
+                            // Make the block from the transactions we did manage to get
+                            debug!("Block budget exceeded on tx {}", &tx.txid());
+                        }
+                        Error::InvalidStacksTransaction(_emsg, true) => {
+                            // if we have an invalid transaction that was quietly ignored, don't warn here either
+                            test_debug!(
+                                "Failed to apply tx {}: InvalidStacksTransaction '{:?}'",
+                                &tx.txid(),
+                                &_emsg
+                            );
+                            continue;
+                        }
+                        Error::ProblematicTransaction(txid) => {
+                            test_debug!("Encountered problematic transaction. Aborting");
+                            return Err(Error::ProblematicTransaction(txid));
+                        }
+                        e => {
+                            warn!("Failed to apply tx {}: {:?}", &tx.txid(), &e);
+                            continue;
+                        }
+                    }
+                }
+                TransactionResult::Problematic(TransactionProblematic { tx, .. }) => {
+                    // drop from the mempool
+                    debug!("Encountered problematic transaction {}", &tx.txid());
+                    return Err(Error::ProblematicTransaction(tx.txid()));
+                }
+            }
+        }
+        let block = builder.mine_nakamoto_block(&mut tenure_tx);
+        let size = builder.bytes_so_far;
+        let cost = builder.tenure_finish(tenure_tx)?;
+        Ok((block, size, cost))
+    }
+
+    /// Produce a single-block shadow tenure.
+    /// Used by tooling to synthesize shadow blocks in case of an emergency.
+    /// The details and circumstances will be recorded in an accompanying SIP.
+    ///
+    /// `naka_tip_id` is the Stacks chain tip on top of which the shadow block will be built.
+    /// `tenure_id_consensus_hash` is the sortition in which the shadow block will be built.
+    /// `txs` are transactions to include, beyond a coinbase and tenure-change
+    pub fn make_shadow_tenure(
+        chainstate: &mut StacksChainState,
+        sortdb: &SortitionDB,
+        naka_tip_id: StacksBlockId,
+        tenure_id_consensus_hash: ConsensusHash,
+        mut txs: Vec<StacksTransaction>,
+    ) -> Result<NakamotoBlock, Error> {
+        let mainnet = chainstate.config().mainnet;
+        let chain_id = chainstate.config().chain_id;
+
+        let recipient = StacksAddress::burn_address(mainnet).to_account_principal();
+        let vrf_proof_bytes = vec![
+            0x92, 0x75, 0xdf, 0x67, 0xa6, 0x8c, 0x87, 0x45, 0xc0, 0xff, 0x97, 0xb4, 0x82, 0x01,
+            0xee, 0x6d, 0xb4, 0x47, 0xf7, 0xc9, 0x3b, 0x23, 0xae, 0x24, 0xcd, 0xc2, 0x40, 0x0f,
+            0x52, 0xfd, 0xb0, 0x8a, 0x1a, 0x6a, 0xc7, 0xec, 0x71, 0xbf, 0x9c, 0x9c, 0x76, 0xe9,
+            0x6e, 0xe4, 0x67, 0x5e, 0xbf, 0xf6, 0x06, 0x25, 0xaf, 0x28, 0x71, 0x85, 0x01, 0x04,
+            0x7b, 0xfd, 0x87, 0xb8, 0x10, 0xc2, 0xd2, 0x13, 0x9b, 0x73, 0xc2, 0x3b, 0xd6, 0x9d,
+            0xe6, 0x63, 0x60, 0x95, 0x3a, 0x64, 0x2c, 0x2a, 0x33, 0x0a,
+        ];
+
+        // safety -- we know it's a good proof
+        let vrf_proof = VRFProof::from_bytes(vrf_proof_bytes.as_slice()).unwrap();
+
+        let naka_tip_header = NakamotoChainState::get_block_header(chainstate.db(), &naka_tip_id)?
+            .ok_or_else(|| {
+                warn!("No such Nakamoto tip: {:?}", &naka_tip_id);
+                Error::NoSuchBlockError
+            })?;
+
+        let naka_tip_tenure_start_header = NakamotoChainState::get_tenure_start_block_header(
+            &mut chainstate.index_conn(),
+            &naka_tip_id,
+            &naka_tip_header.consensus_hash,
+        )?
+        .ok_or_else(|| {
+            Error::InvalidStacksBlock(format!(
+                "No tenure-start block header for tenure {}",
+                &naka_tip_header.consensus_hash
+            ))
+        })?;
+
+        if naka_tip_header.anchored_header.height() + 1
+            <= naka_tip_tenure_start_header.anchored_header.height()
+        {
+            return Err(Error::InvalidStacksBlock(
+                "Nakamoto tip is lower than its tenure-start block".into(),
+            ));
+        }
+
+        let coinbase_payload = CoinbasePayload(naka_tip_tenure_start_header.index_block_hash().0);
+
+        // the miner key is irrelevant
+        let miner_key = StacksPrivateKey::new();
+        let miner_addr = StacksAddress::p2pkh(mainnet, &StacksPublicKey::from_private(&miner_key));
+        let miner_tx_auth = TransactionAuth::from_p2pkh(&miner_key).ok_or_else(|| {
+            Error::InvalidStacksBlock(
+                "Unable to construct transaction auth from transient private key".into(),
+            )
+        })?;
+
+        let tx_version = if mainnet {
+            TransactionVersion::Mainnet
+        } else {
+            TransactionVersion::Testnet
+        };
+        let miner_account = Self::get_account(chainstate, sortdb, &miner_addr, &naka_tip_header)?;
+
+        // tenure change payload (BlockFound)
+        let tenure_change_payload = TenureChangePayload {
+            tenure_consensus_hash: tenure_id_consensus_hash.clone(),
+            prev_tenure_consensus_hash: naka_tip_header.consensus_hash,
+            burn_view_consensus_hash: tenure_id_consensus_hash.clone(),
+            previous_tenure_end: naka_tip_id,
+            previous_tenure_blocks: (naka_tip_header.anchored_header.height() + 1
+                - naka_tip_tenure_start_header.anchored_header.height())
+                as u32,
+            cause: TenureChangeCause::BlockFound,
+            pubkey_hash: Hash160::from_node_public_key(&StacksPublicKey::from_private(&miner_key)),
+        };
+
+        // tenure-change tx
+        let tenure_change_tx = {
+            let mut tx_tenure_change = StacksTransaction::new(
+                tx_version.clone(),
+                miner_tx_auth.clone(),
+                TransactionPayload::TenureChange(tenure_change_payload),
+            );
+            tx_tenure_change.chain_id = chain_id;
+            tx_tenure_change.anchor_mode = TransactionAnchorMode::OnChainOnly;
+            tx_tenure_change.auth.set_origin_nonce(miner_account.nonce);
+
+            let mut tx_signer = StacksTransactionSigner::new(&tx_tenure_change);
+            tx_signer.sign_origin(&miner_key)?;
+            let tx_tenure_change_signed = tx_signer
+                .get_tx()
+                .ok_or_else(|| Error::InvalidStacksBlock("Failed to sign tenure change".into()))?;
+            tx_tenure_change_signed
+        };
+
+        // coinbase tx
+        let coinbase_tx = {
+            let mut tx_coinbase = StacksTransaction::new(
+                tx_version.clone(),
+                miner_tx_auth.clone(),
+                TransactionPayload::Coinbase(coinbase_payload, Some(recipient), Some(vrf_proof)),
+            );
+            tx_coinbase.chain_id = chain_id;
+            tx_coinbase.anchor_mode = TransactionAnchorMode::OnChainOnly;
+            tx_coinbase.auth.set_origin_nonce(miner_account.nonce + 1);
+
+            let mut tx_signer = StacksTransactionSigner::new(&tx_coinbase);
+            tx_signer.sign_origin(&miner_key)?;
+            let tx_coinbase_signed = tx_signer
+                .get_tx()
+                .ok_or_else(|| Error::InvalidStacksBlock("Failed to sign coinbase".into()))?;
+            tx_coinbase_signed
+        };
+
+        // `burn_tip` corresponds to the burn view consensus hash of the tenure.
+        let burn_tip =
+            SortitionDB::get_block_snapshot_consensus(sortdb.conn(), &tenure_id_consensus_hash)?
+                .ok_or_else(|| Error::InvalidStacksBlock("No such tenure ID".into()))?;
+
+        debug!(
+            "Build Nakamoto shadow block in tenure {} sortition {} parent_tip {}",
+            &tenure_id_consensus_hash, &burn_tip.consensus_hash, &naka_tip_id
+        );
+
+        // make a block
+        let builder = NakamotoBlockBuilder::new(
+            &naka_tip_header,
+            &tenure_id_consensus_hash,
+            burn_tip.total_burn,
+            Some(&tenure_change_tx),
+            Some(&coinbase_tx),
+            1,
+            None,
+        )?;
+
+        let mut block_txs = vec![tenure_change_tx, coinbase_tx];
+        block_txs.append(&mut txs);
+        let (mut shadow_block, _size, _cost) = Self::make_shadow_block_from_txs(
+            builder,
+            &chainstate,
+            &sortdb.index_handle(&burn_tip.sortition_id),
+            &tenure_id_consensus_hash,
+            block_txs,
+        )?;
+
+        shadow_block.header.version |= 0x80;
+
+        // no need to sign with the signer set; just the miner is sufficient
+        // (and it can be any miner)
+        shadow_block.header.sign_miner(&miner_key)?;
+
+        Ok(shadow_block)
+    }
+}
+
+impl<'a> NakamotoStagingBlocksConnRef<'a> {
+    /// Determine if we have a particular block with the given index hash.
+    /// Returns Ok(true) if so
+    /// Returns Ok(false) if not
+    /// Returns Err(..) on DB error
+    pub fn has_shadow_nakamoto_block_with_index_hash(
+        &self,
+        index_block_hash: &StacksBlockId,
+    ) -> Result<bool, ChainstateError> {
+        let qry = "SELECT 1 FROM nakamoto_staging_blocks WHERE index_block_hash = ?1 AND obtain_method = ?2";
+        let args = params![
+            index_block_hash,
+            &NakamotoBlockObtainMethod::Shadow.to_string()
+        ];
+        let res: Option<i64> = query_row(self, qry, args)?;
+        Ok(res.is_some())
+    }
+
+    /// Is this a shadow tenure?
+    /// If any block is a shadow block in the tenure, they must all be.
+    ///
+    /// Returns true if the tenure has at least one shadow block.
+    pub fn is_shadow_tenure(
+        &self,
+        consensus_hash: &ConsensusHash,
+    ) -> Result<bool, ChainstateError> {
+        let qry = "SELECT 1 FROM nakamoto_staging_blocks WHERE consensus_hash = ?1 AND obtain_method = ?2";
+        let args = rusqlite::params![
+            consensus_hash,
+            NakamotoBlockObtainMethod::Shadow.to_string()
+        ];
+        let present: Option<u32> = query_row(self, qry, args)?;
+        Ok(present.is_some())
+    }
+
+    /// Shadow blocks, unlike Stacks blocks, have a unique place in the chain history.
+    /// They are inserted post-hoc, so they and their underlying burnchain blocks don't get
+    /// invalidated via a fork.  A consensus hash can identify (1) no tenures, (2) a single
+    /// shadow tenure, or (3) one or more non-shadow tenures.
+    ///
+    /// This is important when downloading a tenure that is ended by a shadow block, since it won't
+    /// be processed beforehand and its hash isn't learned from the burnchain (so we must be able
+    /// to infer that if this is a shadow tenure, none of the blocks in it have siblings).
+    pub fn get_shadow_tenure_start_block(
+        &self,
+        ch: &ConsensusHash,
+    ) -> Result<Option<NakamotoBlock>, ChainstateError> {
+        let qry = "SELECT data FROM nakamoto_staging_blocks WHERE consensus_hash = ?1 AND obtain_method = ?2 ORDER BY height DESC LIMIT 1";
+        let args = params![ch, &NakamotoBlockObtainMethod::Shadow.to_string()];
+        let res: Option<Vec<u8>> = query_row(self, qry, args)?;
+        let Some(block_bytes) = res else {
+            return Ok(None);
+        };
+        let block = NakamotoBlock::consensus_deserialize(&mut block_bytes.as_slice())?;
+        if !block.is_shadow_block() {
+            error!("Staging DB corruption: expected shadow block from {}", ch);
+            return Err(DBError::Corruption.into());
+        }
+        Ok(Some(block))
+    }
+}
+
+impl<'a> NakamotoStagingBlocksTx<'a> {
+    /// Add a shadow block.
+    /// Fails if there are any non-shadow blocks present in the tenure.
+    pub fn add_shadow_block(&self, shadow_block: &NakamotoBlock) -> Result<(), ChainstateError> {
+        if !shadow_block.is_shadow_block() {
+            return Err(ChainstateError::InvalidStacksBlock(
+                "Not a shadow block".into(),
+            ));
+        }
+        let block_id = shadow_block.block_id();
+
+        // is this block stored already?
+        let qry = "SELECT 1 FROM nakamoto_staging_blocks WHERE index_block_hash = ?1";
+        let args = params![block_id];
+        let present: Option<i64> = query_row(self, qry, args)?;
+        if present.is_some() {
+            return Ok(());
+        }
+
+        // this tenure must be empty, or it must be a shadow tenure
+        let qry = "SELECT 1 FROM nakamoto_staging_blocks WHERE consensus_hash = ?1";
+        let args = rusqlite::params![&shadow_block.header.consensus_hash];
+        let present: Option<u32> = query_row(self, qry, args)?;
+        if present.is_some()
+            && !self
+                .conn()
+                .is_shadow_tenure(&shadow_block.header.consensus_hash)?
+        {
+            return Err(ChainstateError::InvalidStacksBlock(
+                "Shadow block cannot be inserted into non-empty non-shadow tenure".into(),
+            ));
+        }
+
+        // there must not be a block at this height in this tenure
+        let qry = "SELECT 1 FROM nakamoto_staging_blocks WHERE consensus_hash = ?1 AND height = ?2";
+        let args = rusqlite::params![
+            &shadow_block.header.consensus_hash,
+            u64_to_sql(shadow_block.header.chain_length)?
+        ];
+        let present: Option<u32> = query_row(self, qry, args)?;
+        if present.is_some() {
+            return Err(ChainstateError::InvalidStacksBlock(format!(
+                "Conflicting block at height {} in tenure {}",
+                shadow_block.header.chain_length, &shadow_block.header.consensus_hash
+            )));
+        }
+
+        // the shadow block is crafted post-hoc, so we know the consensus hash exists.
+        // thus, it's always burn-attachable
+        let burn_attachable = true;
+
+        // shadow blocks cannot be replaced
+        let signing_weight = u32::MAX;
+
+        self.store_block(
+            shadow_block,
+            burn_attachable,
+            signing_weight,
+            NakamotoBlockObtainMethod::Shadow,
+        )?;
+        Ok(())
+    }
+}
+
+/// DO NOT RUN ON A RUNNING NODE (unless you're testing).
+///
+/// Insert and process a shadow block into the Stacks chainstate.
+pub fn process_shadow_block(
+    chain_state: &mut StacksChainState,
+    sort_db: &mut SortitionDB,
+    shadow_block: NakamotoBlock,
+) -> Result<(), ChainstateError> {
+    let tx = chain_state.staging_db_tx_begin()?;
+    tx.add_shadow_block(&shadow_block)?;
+    tx.commit()?;
+
+    let no_dispatch: Option<DummyEventDispatcher> = None;
+    loop {
+        let sort_tip = SortitionDB::get_canonical_burn_chain_tip(sort_db.conn())?;
+
+        // process at most one block per loop pass
+        let processed_block_receipt = match NakamotoChainState::process_next_nakamoto_block(
+            chain_state,
+            sort_db,
+            &sort_tip.sortition_id,
+            no_dispatch.as_ref(),
+        ) {
+            Ok(receipt_opt) => receipt_opt,
+            Err(ChainstateError::InvalidStacksBlock(msg)) => {
+                warn!("Encountered invalid block: {}", &msg);
+                continue;
+            }
+            Err(ChainstateError::NetError(NetError::DeserializeError(msg))) => {
+                // happens if we load a zero-sized block (i.e. an invalid block)
+                warn!("Encountered invalid block (codec error): {}", &msg);
+                continue;
+            }
+            Err(e) => {
+                // something else happened
+                return Err(e.into());
+            }
+        };
+
+        if processed_block_receipt.is_none() {
+            // out of blocks
+            info!("No more blocks to process (no receipts)");
+            break;
+        };
+
+        let Some((_, processed, orphaned, _)) = chain_state
+            .nakamoto_blocks_db()
+            .get_block_processed_and_signed_weight(
+                &shadow_block.header.consensus_hash,
+                &shadow_block.header.block_hash(),
+            )?
+        else {
+            return Err(ChainstateError::InvalidStacksBlock(format!(
+                "Shadow block {} for tenure {} not store",
+                &shadow_block.block_id(),
+                &shadow_block.header.consensus_hash
+            )));
+        };
+
+        if orphaned {
+            return Err(ChainstateError::InvalidStacksBlock(format!(
+                "Shadow block {} for tenure {} was orphaned",
+                &shadow_block.block_id(),
+                &shadow_block.header.consensus_hash
+            )));
+        }
+
+        if processed {
+            break;
+        }
+    }
+    Ok(())
+}
+
+/// DO NOT RUN ON A RUNNING NODE (unless you're testing).
+///
+/// Automatically repair a node that has been stalled due to an empty prepare phase.
+/// Works by synthesizing, inserting, and processing shadow tenures in-between the last sortition
+/// with a winner and the burnchain tip.
+///
+/// This is meant to be accessed by the tooling. Once the blocks are synthesized, they would be
+/// added into other broken nodes' chainstates by the same tooling.  Ultimately, a patched node
+/// would be released with these shadow blocks added in as part of the chainstate schema.
+///
+/// Returns the syntheisized shadow blocks on success.
+/// Returns error on failure.
+pub fn shadow_chainstate_repair(
+    chain_state: &mut StacksChainState,
+    sort_db: &mut SortitionDB,
+) -> Result<Vec<NakamotoBlock>, ChainstateError> {
+    let sort_tip = SortitionDB::get_canonical_burn_chain_tip(sort_db.conn())?;
+
+    let header = NakamotoChainState::get_canonical_block_header(chain_state.db(), &sort_db)?
+        .ok_or_else(|| ChainstateError::NoSuchBlockError)?;
+
+    let header_sn =
+        SortitionDB::get_block_snapshot_consensus(sort_db.conn(), &header.consensus_hash)?
+            .ok_or_else(|| {
+                ChainstateError::InvalidStacksBlock(
+                    "Canonical stacks header does not have a sortition".into(),
+                )
+            })?;
+
+    let mut shadow_blocks = vec![];
+    for burn_height in (header_sn.block_height + 1)..sort_tip.block_height {
+        let sort_tip = SortitionDB::get_canonical_burn_chain_tip(sort_db.conn())?;
+        let sort_handle = sort_db.index_handle(&sort_tip.sortition_id);
+        let sn = sort_handle
+            .get_block_snapshot_by_height(burn_height)?
+            .ok_or_else(|| ChainstateError::InvalidStacksBlock("No sortition at height".into()))?;
+
+        let header = NakamotoChainState::get_canonical_block_header(chain_state.db(), &sort_db)?
+            .ok_or_else(|| ChainstateError::NoSuchBlockError)?;
+
+        let chain_tip = header.index_block_hash();
+        let shadow_block = NakamotoBlockBuilder::make_shadow_tenure(
+            chain_state,
+            sort_db,
+            chain_tip.clone(),
+            sn.consensus_hash,
+            vec![],
+        )?;
+
+        shadow_blocks.push(shadow_block.clone());
+
+        process_shadow_block(chain_state, sort_db, shadow_block)?;
+    }
+
+    Ok(shadow_blocks)
+}
diff --git a/stackslib/src/chainstate/nakamoto/staging_blocks.rs b/stackslib/src/chainstate/nakamoto/staging_blocks.rs
index 382c708850..c3e8432878 100644
--- a/stackslib/src/chainstate/nakamoto/staging_blocks.rs
+++ b/stackslib/src/chainstate/nakamoto/staging_blocks.rs
@@ -28,7 +28,7 @@ use stacks_common::util::{get_epoch_time_secs, sleep_ms};
 
 use crate::chainstate::burn::db::sortdb::{SortitionDB, SortitionHandle};
 use crate::chainstate::burn::BlockSnapshot;
-use crate::chainstate::nakamoto::{NakamotoBlock, NakamotoChainState};
+use crate::chainstate::nakamoto::{NakamotoBlock, NakamotoBlockHeader, NakamotoChainState};
 use crate::chainstate::stacks::db::StacksChainState;
 use crate::chainstate::stacks::index::marf::MarfConnection;
 use crate::chainstate::stacks::{Error as ChainstateError, StacksBlock, StacksBlockHeader};
@@ -41,10 +41,16 @@ use crate::util_lib::db::{
 /// The means by which a block is obtained.
 #[derive(Debug, PartialEq, Clone, Copy)]
 pub enum NakamotoBlockObtainMethod {
+    /// The block was fetched by te block downloader
     Downloaded,
+    /// The block was uploaded to us via p2p
     Pushed,
+    /// This node mined the block
     Mined,
+    /// The block was uploaded to us via HTTP
     Uploaded,
+    /// This is a shadow block -- it was created by a SIP to fix a consensus bug
+    Shadow,
 }
 
 impl fmt::Display for NakamotoBlockObtainMethod {
@@ -149,7 +155,12 @@ pub const NAKAMOTO_STAGING_DB_SCHEMA_2: &'static [&'static str] = &[
     r#"INSERT INTO db_version (version) VALUES (2)"#,
 ];
 
-pub const NAKAMOTO_STAGING_DB_SCHEMA_LATEST: u32 = 2;
+pub const NAKAMOTO_STAGING_DB_SCHEMA_3: &'static [&'static str] = &[
+    r#"CREATE INDEX nakamoto_staging_blocks_by_obtain_method ON nakamoto_staging_blocks(consensus_hash,obtain_method);"#,
+    r#"UPDATE db_version SET version = 3"#,
+];
+
+pub const NAKAMOTO_STAGING_DB_SCHEMA_LATEST: u32 = 3;
 
 pub struct NakamotoStagingBlocksConn(rusqlite::Connection);
 
@@ -211,6 +222,21 @@ impl<'a> DerefMut for NakamotoStagingBlocksTx<'a> {
         &mut self.0
     }
 }
+/// Open a Blob handle to a Nakamoto block
+fn inner_open_nakamoto_block<'a>(
+    conn: &'a Connection,
+    rowid: i64,
+    readwrite: bool,
+) -> Result<Blob<'a>, ChainstateError> {
+    let blob = conn.blob_open(
+        rusqlite::DatabaseName::Main,
+        "nakamoto_staging_blocks",
+        "data",
+        rowid,
+        !readwrite,
+    )?;
+    Ok(blob)
+}
 
 impl NakamotoStagingBlocksConn {
     /// Open a Blob handle to a Nakamoto block
@@ -219,18 +245,20 @@ impl NakamotoStagingBlocksConn {
         rowid: i64,
         readwrite: bool,
     ) -> Result<Blob<'a>, ChainstateError> {
-        let blob = self.blob_open(
-            rusqlite::DatabaseName::Main,
-            "nakamoto_staging_blocks",
-            "data",
-            rowid,
-            !readwrite,
-        )?;
-        Ok(blob)
+        inner_open_nakamoto_block(self.deref(), rowid, readwrite)
     }
 }
 
 impl<'a> NakamotoStagingBlocksConnRef<'a> {
+    /// Open a Blob handle to a Nakamoto block
+    pub fn open_nakamoto_block(
+        &'a self,
+        rowid: i64,
+        readwrite: bool,
+    ) -> Result<Blob<'a>, ChainstateError> {
+        inner_open_nakamoto_block(self.deref(), rowid, readwrite)
+    }
+
     /// Determine if we have a particular block with the given index hash.
     /// Returns Ok(true) if so
     /// Returns Ok(false) if not
@@ -250,7 +278,7 @@ impl<'a> NakamotoStagingBlocksConnRef<'a> {
     /// There will be at most one such block.
     ///
     /// NOTE: for Nakamoto blocks, the sighash is the same as the block hash.
-    pub(crate) fn get_block_processed_and_signed_weight(
+    pub fn get_block_processed_and_signed_weight(
         &self,
         consensus_hash: &ConsensusHash,
         block_hash: &BlockHeaderHash,
@@ -332,6 +360,32 @@ impl<'a> NakamotoStagingBlocksConnRef<'a> {
         )))
     }
 
+    /// Get a Nakamoto block header by index block hash.
+    /// Verifies its integrity
+    /// Returns Ok(Some(header)) if the block was present
+    /// Returns Ok(None) if there was no such block
+    /// Returns Err(..) on DB error, including corruption
+    pub fn get_nakamoto_block_header(
+        &self,
+        index_block_hash: &StacksBlockId,
+    ) -> Result<Option<NakamotoBlockHeader>, ChainstateError> {
+        let Some(rowid) = self.get_nakamoto_block_rowid(index_block_hash)? else {
+            return Ok(None);
+        };
+
+        let mut fd = self.open_nakamoto_block(rowid, false)?;
+        let block_header = NakamotoBlockHeader::consensus_deserialize(&mut fd)?;
+        if &block_header.block_id() != index_block_hash {
+            error!(
+                "Staging DB corruption: expected {}, got {}",
+                index_block_hash,
+                &block_header.block_id()
+            );
+            return Err(DBError::Corruption.into());
+        }
+        Ok(Some(block_header))
+    }
+
     /// Get the size of a Nakamoto block, given its index block hash
     /// Returns Ok(Some(size)) if the block was present
     /// Returns Ok(None) if there was no such block
@@ -443,14 +497,6 @@ impl<'a> NakamotoStagingBlocksConnRef<'a> {
             })
     }
 
-    /// Given a block ID, determine if it has children that have been processed and accepted
-    pub fn has_children(&self, index_block_hash: &StacksBlockId) -> Result<bool, ChainstateError> {
-        let qry = "SELECT 1 FROM nakamoto_staging_blocks WHERE parent_block_id = ?1 AND processed = 1 AND orphaned = 0 LIMIT 1";
-        let args = rusqlite::params![index_block_hash];
-        let children_flags: Option<u32> = query_row(self, qry, args)?;
-        Ok(children_flags.is_some())
-    }
-
     /// Given a consensus hash, determine if the burn block has been processed.
     /// Because this is stored in a denormalized way, we'll want to do this whenever we store a
     /// block (so we can set `burn_attachable` accordingly)
@@ -534,6 +580,19 @@ impl<'a> NakamotoStagingBlocksTx<'a> {
                 .is_burn_block_processed(&block.header.consensus_hash)?
         };
 
+        let obtain_method = if block.is_shadow_block() {
+            // override
+            NakamotoBlockObtainMethod::Shadow
+        } else {
+            obtain_method
+        };
+
+        if self.conn().is_shadow_tenure(&block.header.consensus_hash)? && !block.is_shadow_block() {
+            return Err(ChainstateError::InvalidStacksBlock(
+                "Tried to insert a non-shadow block into a shadow tenure".into(),
+            ));
+        }
+
         self.execute(
             "INSERT INTO nakamoto_staging_blocks (
                      block_hash,
@@ -715,15 +774,37 @@ impl StacksChainState {
 
     /// Perform migrations
     pub fn migrate_nakamoto_staging_blocks(conn: &Connection) -> Result<(), ChainstateError> {
-        let mut version = Self::get_nakamoto_staging_blocks_db_version(conn)?;
-        if version < 2 {
-            debug!("Migrate Nakamoto staging blocks DB to schema 2");
-            for cmd in NAKAMOTO_STAGING_DB_SCHEMA_2.iter() {
-                conn.execute(cmd, NO_PARAMS)?;
+        loop {
+            let version = Self::get_nakamoto_staging_blocks_db_version(conn)?;
+            if version == NAKAMOTO_STAGING_DB_SCHEMA_LATEST {
+                return Ok(());
+            }
+            match version {
+                1 => {
+                    debug!("Migrate Nakamoto staging blocks DB to schema 2");
+                    for cmd in NAKAMOTO_STAGING_DB_SCHEMA_2.iter() {
+                        conn.execute(cmd, NO_PARAMS)?;
+                    }
+                    let version = Self::get_nakamoto_staging_blocks_db_version(conn)?;
+                    assert_eq!(version, 2, "Nakamoto staging DB migration failure");
+                    debug!("Migrated Nakamoto staging blocks DB to schema 2");
+                }
+                2 => {
+                    debug!("Migrate Nakamoto staging blocks DB to schema 3");
+                    for cmd in NAKAMOTO_STAGING_DB_SCHEMA_3.iter() {
+                        conn.execute(cmd, NO_PARAMS)?;
+                    }
+                    let version = Self::get_nakamoto_staging_blocks_db_version(conn)?;
+                    assert_eq!(version, 3, "Nakamoto staging DB migration failure");
+                    debug!("Migrated Nakamoto staging blocks DB to schema 3");
+                }
+                NAKAMOTO_STAGING_DB_SCHEMA_LATEST => {
+                    break;
+                }
+                _ => {
+                    panic!("Unusable staging DB: Unknown schema version {}", version);
+                }
             }
-            version = Self::get_nakamoto_staging_blocks_db_version(conn)?;
-            assert_eq!(version, 2, "Nakamoto staging DB migration failure");
-            debug!("Migrated Nakamoto staging blocks DB to schema 2");
         }
         Ok(())
     }
diff --git a/stackslib/src/chainstate/nakamoto/tenure.rs b/stackslib/src/chainstate/nakamoto/tenure.rs
index 5c729d845d..9852733311 100644
--- a/stackslib/src/chainstate/nakamoto/tenure.rs
+++ b/stackslib/src/chainstate/nakamoto/tenure.rs
@@ -749,8 +749,18 @@ impl NakamotoChainState {
                 warn!("Invalid tenure-change: parent snapshot comes after current tip"; "burn_view_consensus_hash" => %tenure_payload.burn_view_consensus_hash, "prev_tenure_consensus_hash" => %tenure_payload.prev_tenure_consensus_hash);
                 return Ok(None);
             }
-            if !prev_sn.sortition {
-                // parent wasn't a sortition-induced tenure change
+
+            // is the parent a shadow block?
+            // Only possible if the parent is also a nakamoto block
+            let is_parent_shadow_block = NakamotoChainState::get_nakamoto_block_version(
+                headers_conn.sqlite(),
+                &block_header.parent_block_id,
+            )?
+            .map(|parent_version| NakamotoBlockHeader::is_shadow_block_version(parent_version))
+            .unwrap_or(false);
+
+            if !is_parent_shadow_block && !prev_sn.sortition {
+                // parent wasn't a shadow block (we expect a sortition), but this wasn't a sortition-induced tenure change
                 warn!("Invalid tenure-change: no block found";
                       "prev_tenure_consensus_hash" => %tenure_payload.prev_tenure_consensus_hash
                 );
@@ -758,8 +768,8 @@ impl NakamotoChainState {
             }
         }
 
-        // the tenure must correspond to sortitions
-        if !tenure_sn.sortition {
+        // if this isn't a shadow block, then the tenure must correspond to sortitions
+        if !block_header.is_shadow_block() && !tenure_sn.sortition {
             warn!("Invalid tenure-change: no block found";
                   "tenure_consensus_hash" => %tenure_payload.tenure_consensus_hash
             );
@@ -1056,6 +1066,15 @@ impl NakamotoChainState {
                 ChainstateError::NoSuchBlockError
             })?;
 
+        if snapshot.consensus_hash != *block_consensus_hash {
+            // should be unreachable, but check defensively
+            warn!(
+                "Snapshot for {} is not the same as the one for {}",
+                &burn_header_hash, block_consensus_hash
+            );
+            return Err(ChainstateError::NoSuchBlockError);
+        }
+
         Ok(snapshot)
     }
 }
diff --git a/stackslib/src/chainstate/stacks/db/accounts.rs b/stackslib/src/chainstate/stacks/db/accounts.rs
index 7c81410e87..b05365d5ac 100644
--- a/stackslib/src/chainstate/stacks/db/accounts.rs
+++ b/stackslib/src/chainstate/stacks/db/accounts.rs
@@ -856,6 +856,14 @@ impl StacksChainState {
             burn_total
         );
 
+        // in the case of shadow blocks, there will be zero burns.
+        // the coinbase is still generated, but it's rendered unspendable
+        let (this_burn_total, burn_total) = if burn_total == 0 {
+            (1, 1)
+        } else {
+            (this_burn_total, burn_total)
+        };
+
         // each participant gets a share of the coinbase proportional to the fraction it burned out
         // of all participants' burns.
         let coinbase_reward = participant
diff --git a/stackslib/src/main.rs b/stackslib/src/main.rs
index bcb7dfc964..16fbd7c2d2 100644
--- a/stackslib/src/main.rs
+++ b/stackslib/src/main.rs
@@ -49,6 +49,10 @@ use blockstack_lib::chainstate::burn::db::sortdb::{
 use blockstack_lib::chainstate::burn::operations::BlockstackOperationType;
 use blockstack_lib::chainstate::burn::{BlockSnapshot, ConsensusHash};
 use blockstack_lib::chainstate::coordinator::{get_reward_cycle_info, OnChainRewardSetProvider};
+use blockstack_lib::chainstate::nakamoto::miner::NakamotoBlockBuilder;
+use blockstack_lib::chainstate::nakamoto::shadow::{
+    process_shadow_block, shadow_chainstate_repair,
+};
 use blockstack_lib::chainstate::nakamoto::{NakamotoBlock, NakamotoChainState};
 use blockstack_lib::chainstate::stacks::db::blocks::{DummyEventDispatcher, StagingBlock};
 use blockstack_lib::chainstate::stacks::db::{
@@ -247,6 +251,56 @@ impl P2PSession {
     }
 }
 
+fn open_nakamoto_chainstate_dbs(
+    chainstate_dir: &str,
+    network: &str,
+) -> (SortitionDB, StacksChainState) {
+    let (mainnet, chain_id, pox_constants, dirname) = match network {
+        "mainnet" => (
+            true,
+            CHAIN_ID_MAINNET,
+            PoxConstants::mainnet_default(),
+            network,
+        ),
+        "krypton" => (
+            false,
+            0x80000100,
+            PoxConstants::nakamoto_testnet_default(),
+            network,
+        ),
+        "naka3" => (
+            false,
+            0x80000000,
+            PoxConstants::new(20, 5, 3, 100, 0, u64::MAX, u64::MAX, 104, 105, 106, 107),
+            "nakamoto-neon",
+        ),
+        _ => {
+            panic!("Unrecognized network name '{}'", network);
+        }
+    };
+
+    let chain_state_path = format!("{}/{}/chainstate/", chainstate_dir, dirname);
+    let sort_db_path = format!("{}/{}/burnchain/sortition/", chainstate_dir, dirname);
+
+    let sort_db = SortitionDB::open(&sort_db_path, true, pox_constants)
+        .unwrap_or_else(|_| panic!("Failed to open {sort_db_path}"));
+
+    let (chain_state, _) = StacksChainState::open(mainnet, chain_id, &chain_state_path, None)
+        .expect("Failed to open stacks chain state");
+
+    (sort_db, chain_state)
+}
+
+fn check_shadow_network(network: &str) {
+    if network != "mainnet" && network != "krypton" && network != "naka3" {
+        eprintln!(
+            "Unknown network '{}': only support 'mainnet', 'krypton', or 'naka3'",
+            &network
+        );
+        process::exit(1);
+    }
+}
+
 #[cfg_attr(test, mutants::skip)]
 fn main() {
     let mut argv: Vec<String> = env::args().collect();
@@ -1166,6 +1220,204 @@ simulating a miner.
         println!("{:?}", inv);
     }
 
+    if argv[1] == "get-nakamoto-tip" {
+        if argv.len() < 4 {
+            eprintln!(
+                "Usage: {} get-nakamoto-tip CHAINSTATE_DIR NETWORK",
+                &argv[0]
+            );
+            process::exit(1);
+        }
+
+        let chainstate_dir = argv[2].as_str();
+        let network = argv[3].as_str();
+
+        check_shadow_network(network);
+        let (sort_db, chain_state) = open_nakamoto_chainstate_dbs(chainstate_dir, network);
+
+        let header = NakamotoChainState::get_canonical_block_header(chain_state.db(), &sort_db)
+            .unwrap()
+            .unwrap();
+        println!("{}", &header.index_block_hash());
+        process::exit(0);
+    }
+
+    if argv[1] == "get-account" {
+        if argv.len() < 5 {
+            eprintln!(
+                "Usage: {} get-account CHAINSTATE_DIR mainnet|krypton ADDRESS [CHAIN_TIP]",
+                &argv[0]
+            );
+            process::exit(1);
+        }
+
+        let chainstate_dir = argv[2].as_str();
+        let network = argv[3].as_str();
+        let addr = StacksAddress::from_string(&argv[4]).unwrap();
+        let chain_tip: Option<StacksBlockId> =
+            argv.get(5).map(|tip| StacksBlockId::from_hex(tip).unwrap());
+
+        check_shadow_network(network);
+        let (sort_db, mut chain_state) = open_nakamoto_chainstate_dbs(chainstate_dir, network);
+
+        let chain_tip_header = chain_tip
+            .map(|tip| {
+                let header = NakamotoChainState::get_block_header_nakamoto(chain_state.db(), &tip)
+                    .unwrap()
+                    .unwrap();
+                header
+            })
+            .unwrap_or_else(|| {
+                let header =
+                    NakamotoChainState::get_canonical_block_header(chain_state.db(), &sort_db)
+                        .unwrap()
+                        .unwrap();
+                header
+            });
+
+        let account =
+            NakamotoBlockBuilder::get_account(&mut chain_state, &sort_db, &addr, &chain_tip_header)
+                .unwrap();
+        println!("{:#?}", &account);
+        process::exit(0);
+    }
+
+    if argv[1] == "make-shadow-block" {
+        if argv.len() < 5 {
+            eprintln!(
+                "Usage: {} make-shadow-block CHAINSTATE_DIR NETWORK CHAIN_TIP_HASH [TX...]",
+                &argv[0]
+            );
+            process::exit(1);
+        }
+        let chainstate_dir = argv[2].as_str();
+        let network = argv[3].as_str();
+        let chain_tip = StacksBlockId::from_hex(argv[4].as_str()).unwrap();
+        let txs = argv[5..]
+            .iter()
+            .map(|tx_str| {
+                let tx_bytes = hex_bytes(&tx_str).unwrap();
+                let tx = StacksTransaction::consensus_deserialize(&mut &tx_bytes[..]).unwrap();
+                tx
+            })
+            .collect();
+
+        check_shadow_network(network);
+        let (sort_db, mut chain_state) = open_nakamoto_chainstate_dbs(chainstate_dir, network);
+        let header = NakamotoChainState::get_block_header(chain_state.db(), &chain_tip)
+            .unwrap()
+            .unwrap();
+
+        let shadow_block = NakamotoBlockBuilder::make_shadow_tenure(
+            &mut chain_state,
+            &sort_db,
+            chain_tip,
+            header.consensus_hash,
+            txs,
+        )
+        .unwrap();
+
+        println!("{}", to_hex(&shadow_block.serialize_to_vec()));
+        process::exit(0);
+    }
+
+    // Generates the shadow blocks needed to restore this node to working order.
+    // Automatically inserts and processes them as well.
+    // Prints out the generated shadow blocks (as JSON)
+    if argv[1] == "shadow-chainstate-repair" {
+        if argv.len() < 4 {
+            eprintln!(
+                "Usage: {} shadow-chainstate-repair CHAINSTATE_DIR NETWORK",
+                &argv[0]
+            );
+            process::exit(1);
+        }
+
+        let chainstate_dir = argv[2].as_str();
+        let network = argv[3].as_str();
+
+        check_shadow_network(network);
+
+        let (mut sort_db, mut chain_state) = open_nakamoto_chainstate_dbs(chainstate_dir, network);
+        let shadow_blocks = shadow_chainstate_repair(&mut chain_state, &mut sort_db).unwrap();
+
+        let shadow_blocks_hex: Vec<_> = shadow_blocks
+            .into_iter()
+            .map(|blk| to_hex(&blk.serialize_to_vec()))
+            .collect();
+
+        println!("{}", serde_json::to_string(&shadow_blocks_hex).unwrap());
+        process::exit(0);
+    }
+
+    // Inserts and processes shadow blocks generated from `shadow-chainstate-repair`
+    if argv[1] == "shadow-chainstate-patch" {
+        if argv.len() < 5 {
+            eprintln!(
+                "Usage: {} shadow-chainstate-patch CHAINSTATE_DIR NETWORK SHADOW_BLOCKS_PATH.JSON",
+                &argv[0]
+            );
+            process::exit(1);
+        }
+
+        let chainstate_dir = argv[2].as_str();
+        let network = argv[3].as_str();
+        let shadow_blocks_json_path = argv[4].as_str();
+
+        let shadow_blocks_hex = {
+            let mut blocks_json_file =
+                File::open(shadow_blocks_json_path).expect("Unable to open file");
+            let mut buffer = vec![];
+            blocks_json_file.read_to_end(&mut buffer).unwrap();
+            let shadow_blocks_hex: Vec<String> = serde_json::from_slice(&buffer).unwrap();
+            shadow_blocks_hex
+        };
+
+        let shadow_blocks: Vec<_> = shadow_blocks_hex
+            .into_iter()
+            .map(|blk_hex| {
+                NakamotoBlock::consensus_deserialize(&mut hex_bytes(&blk_hex).unwrap().as_slice())
+                    .unwrap()
+            })
+            .collect();
+
+        check_shadow_network(network);
+
+        let (mut sort_db, mut chain_state) = open_nakamoto_chainstate_dbs(chainstate_dir, network);
+        for shadow_block in shadow_blocks.into_iter() {
+            process_shadow_block(&mut chain_state, &mut sort_db, shadow_block).unwrap();
+        }
+
+        process::exit(0);
+    }
+
+    if argv[1] == "add-shadow-block" {
+        if argv.len() < 5 {
+            eprintln!(
+                "Usage: {} add-shadow-block CHAINSTATE_DIR NETWORK SHADOW_BLOCK_HEX",
+                &argv[0]
+            );
+            process::exit(1);
+        }
+        let chainstate_dir = argv[2].as_str();
+        let network = argv[3].as_str();
+        let block_hex = argv[4].as_str();
+        let shadow_block =
+            NakamotoBlock::consensus_deserialize(&mut hex_bytes(block_hex).unwrap().as_slice())
+                .unwrap();
+
+        assert!(shadow_block.is_shadow_block());
+
+        check_shadow_network(network);
+        let (_, mut chain_state) = open_nakamoto_chainstate_dbs(chainstate_dir, network);
+
+        let tx = chain_state.staging_db_tx_begin().unwrap();
+        tx.add_shadow_block(&shadow_block).unwrap();
+        tx.commit().unwrap();
+
+        process::exit(0);
+    }
+
     if argv[1] == "replay-chainstate" {
         if argv.len() < 7 {
             eprintln!("Usage: {} OLD_CHAINSTATE_PATH OLD_SORTITION_DB_PATH OLD_BURNCHAIN_DB_PATH NEW_CHAINSTATE_PATH NEW_BURNCHAIN_DB_PATH", &argv[0]);
diff --git a/stackslib/src/net/api/get_tenures_fork_info.rs b/stackslib/src/net/api/get_tenures_fork_info.rs
index 8bcf32ce1d..2cb2847290 100644
--- a/stackslib/src/net/api/get_tenures_fork_info.rs
+++ b/stackslib/src/net/api/get_tenures_fork_info.rs
@@ -231,21 +231,31 @@ impl RPCRequestHandler for GetTenuresForkInfo {
                 chainstate,
                 &network.stacks_tip.block_id(),
             )?);
-            let handle = sortdb.index_handle(&cursor.sortition_id);
             let mut depth = 0;
             while depth < DEPTH_LIMIT && cursor.consensus_hash != recurse_end {
-                depth += 1;
                 if height_bound >= cursor.block_height {
                     return Err(ChainError::NotInSameFork);
                 }
-                cursor = handle
-                    .get_last_snapshot_with_sortition(cursor.block_height.saturating_sub(1))?;
-                results.push(TenureForkingInfo::from_snapshot(
-                    &cursor,
-                    sortdb,
-                    chainstate,
-                    &network.stacks_tip.block_id(),
-                )?);
+                cursor =
+                    SortitionDB::get_block_snapshot(sortdb.conn(), &cursor.parent_sortition_id)?
+                        .ok_or_else(|| ChainError::NoSuchBlockError)?;
+                if cursor.sortition
+                    || chainstate
+                        .nakamoto_blocks_db()
+                        .is_shadow_tenure(&cursor.consensus_hash)?
+                {
+                    results.push(TenureForkingInfo::from_snapshot(
+                        &cursor,
+                        sortdb,
+                        chainstate,
+                        &network.stacks_tip.block_id(),
+                    )?);
+                }
+                if cursor.sortition {
+                    // don't count shadow blocks towards the depth, since there can be a large
+                    // swath of them.
+                    depth += 1;
+                }
             }
 
             Ok(results)
diff --git a/stackslib/src/net/api/getsortition.rs b/stackslib/src/net/api/getsortition.rs
index 9b22d8b82f..b41e516cbf 100644
--- a/stackslib/src/net/api/getsortition.rs
+++ b/stackslib/src/net/api/getsortition.rs
@@ -29,7 +29,9 @@ use {serde, serde_json};
 
 use crate::chainstate::burn::db::sortdb::SortitionDB;
 use crate::chainstate::burn::BlockSnapshot;
-use crate::chainstate::nakamoto::{NakamotoBlock, NakamotoChainState, NakamotoStagingBlocksConn};
+use crate::chainstate::nakamoto::{
+    NakamotoBlock, NakamotoChainState, NakamotoStagingBlocksConn, StacksDBIndexed,
+};
 use crate::chainstate::stacks::db::StacksChainState;
 use crate::chainstate::stacks::Error as ChainError;
 use crate::net::api::getblock_v3::NakamotoBlockStream;
@@ -85,6 +87,11 @@ pub struct SortitionInfo {
     pub consensus_hash: ConsensusHash,
     /// Boolean indicating whether or not there was a succesful sortition (i.e. a winning
     ///  block or miner was chosen).
+    ///
+    /// This will *also* be true if this sortition corresponds to a shadow block.  This is because
+    /// the signer does not distinguish between shadow blocks and blocks with sortitions, so until
+    /// we can update the signer and this interface, we'll have to report the presence of a shadow
+    /// block tenure in a way that the signer currently understands.
     pub was_sortition: bool,
     /// If sortition occurred, and the miner's VRF key registration
     ///  associated a nakamoto mining pubkey with their commit, this
@@ -150,13 +157,41 @@ impl GetSortitionHandler {
     fn get_sortition_info(
         sortition_sn: BlockSnapshot,
         sortdb: &SortitionDB,
+        chainstate: &mut StacksChainState,
+        tip: &StacksBlockId,
     ) -> Result<SortitionInfo, ChainError> {
+        let is_shadow = chainstate
+            .nakamoto_blocks_db()
+            .is_shadow_tenure(&sortition_sn.consensus_hash)?;
         let (miner_pk_hash160, stacks_parent_ch, committed_block_hash, last_sortition_ch) =
-            if !sortition_sn.sortition {
+            if !sortition_sn.sortition && !is_shadow {
                 let handle = sortdb.index_handle(&sortition_sn.sortition_id);
                 let last_sortition =
                     handle.get_last_snapshot_with_sortition(sortition_sn.block_height)?;
                 (None, None, None, Some(last_sortition.consensus_hash))
+            } else if !sortition_sn.sortition && is_shadow {
+                // this is a shadow tenure.
+                let parent_tenure_ch = chainstate
+                    .index_conn()
+                    .get_parent_tenure_consensus_hash(tip, &sortition_sn.consensus_hash)?
+                    .ok_or_else(|| DBError::NotFoundError)?;
+
+                let parent_tenure_start_header =
+                    NakamotoChainState::get_nakamoto_tenure_start_block_header(
+                        &mut chainstate.index_conn(),
+                        tip,
+                        &parent_tenure_ch,
+                    )?
+                    .ok_or_else(|| DBError::NotFoundError)?;
+
+                (
+                    Some(Hash160([0x00; 20])),
+                    Some(parent_tenure_ch.clone()),
+                    Some(BlockHeaderHash(
+                        parent_tenure_start_header.index_block_hash().0,
+                    )),
+                    Some(parent_tenure_ch),
+                )
             } else {
                 let block_commit = SortitionDB::get_block_commit(sortdb.conn(), &sortition_sn.winning_block_txid, &sortition_sn.sortition_id)?
                         .ok_or_else(|| {
@@ -211,7 +246,7 @@ impl GetSortitionHandler {
             sortition_id: sortition_sn.sortition_id,
             parent_sortition_id: sortition_sn.parent_sortition_id,
             consensus_hash: sortition_sn.consensus_hash,
-            was_sortition: sortition_sn.sortition,
+            was_sortition: sortition_sn.sortition || is_shadow,
             miner_pk_hash160,
             stacks_parent_ch,
             last_sortition_ch,
@@ -277,7 +312,7 @@ impl RPCRequestHandler for GetSortitionHandler {
         _contents: HttpRequestContents,
         node: &mut StacksNodeState,
     ) -> Result<(HttpResponsePreamble, HttpResponseContents), NetError> {
-        let result = node.with_node_state(|network, sortdb, _chainstate, _mempool, _rpc_args| {
+        let result = node.with_node_state(|network, sortdb, chainstate, _mempool, _rpc_args| {
             let query_result = match self.query {
                 QuerySpecifier::Latest => Ok(Some(network.burnchain_tip.clone())),
                 QuerySpecifier::ConsensusHash(ref consensus_hash) => {
@@ -306,7 +341,12 @@ impl RPCRequestHandler for GetSortitionHandler {
                 }
             };
             let sortition_sn = query_result?.ok_or_else(|| ChainError::NoSuchBlockError)?;
-            Self::get_sortition_info(sortition_sn, sortdb)
+            Self::get_sortition_info(
+                sortition_sn,
+                sortdb,
+                chainstate,
+                &network.stacks_tip.block_id(),
+            )
         });
 
         let block = match result {
@@ -334,13 +374,18 @@ impl RPCRequestHandler for GetSortitionHandler {
         if self.query == QuerySpecifier::LatestAndLast {
             // if latest **and** last are requested, lookup the sortition info for last_sortition_ch
             if let Some(last_sortition_ch) = last_sortition_ch {
-                let result = node.with_node_state(|_, sortdb, _, _, _| {
+                let result = node.with_node_state(|network, sortdb, chainstate, _, _| {
                     let last_sortition_sn = SortitionDB::get_block_snapshot_consensus(
                         sortdb.conn(),
                         &last_sortition_ch,
                     )?
                     .ok_or_else(|| ChainError::NoSuchBlockError)?;
-                    Self::get_sortition_info(last_sortition_sn, sortdb)
+                    Self::get_sortition_info(
+                        last_sortition_sn,
+                        sortdb,
+                        chainstate,
+                        &network.stacks_tip.block_id(),
+                    )
                 });
                 let last_block = match result {
                     Ok(block) => block,
diff --git a/stackslib/src/net/api/postblock_proposal.rs b/stackslib/src/net/api/postblock_proposal.rs
index daa8aaae3b..b6f91c59b8 100644
--- a/stackslib/src/net/api/postblock_proposal.rs
+++ b/stackslib/src/net/api/postblock_proposal.rs
@@ -403,7 +403,8 @@ impl NakamotoBlockProposal {
         };
 
         // Static validation checks
-        NakamotoChainState::validate_nakamoto_block_burnchain(
+        NakamotoChainState::validate_normal_nakamoto_block_burnchain(
+            chainstate.nakamoto_blocks_db(),
             &db_handle,
             expected_burn_opt,
             &self.block,
@@ -647,6 +648,12 @@ impl HttpRequest for RPCBlockProposalRequestHandler {
             }
         };
 
+        if block_proposal.block.is_shadow_block() {
+            return Err(Error::DecodeError(
+                "Shadow blocks cannot be submitted for validation".to_string(),
+            ));
+        }
+
         self.block_proposal = Some(block_proposal);
         Ok(HttpRequestContents::new().query_string(query))
     }
diff --git a/stackslib/src/net/download/nakamoto/download_state_machine.rs b/stackslib/src/net/download/nakamoto/download_state_machine.rs
index 42d228aca1..4c509ed5c1 100644
--- a/stackslib/src/net/download/nakamoto/download_state_machine.rs
+++ b/stackslib/src/net/download/nakamoto/download_state_machine.rs
@@ -1184,6 +1184,16 @@ impl NakamotoDownloadStateMachine {
                 continue;
             }
 
+            let _ = downloader
+                .try_advance_from_chainstate(chainstate)
+                .map_err(|e| {
+                    warn!(
+                        "Failed to advance downloader in state {} for {}: {:?}",
+                        &downloader.state, &downloader.naddr, &e
+                    );
+                    e
+                });
+
             debug!(
                 "Send request to {} for tenure {:?} (state {})",
                 &naddr,
@@ -1301,13 +1311,16 @@ impl NakamotoDownloadStateMachine {
     fn download_confirmed_tenures(
         &mut self,
         network: &mut PeerNetwork,
+        chainstate: &mut StacksChainState,
         max_count: usize,
     ) -> HashMap<ConsensusHash, Vec<NakamotoBlock>> {
         // queue up more downloaders
         self.update_tenure_downloaders(max_count, &network.current_reward_sets);
 
         // run all downloaders
-        let new_blocks = self.tenure_downloads.run(network, &mut self.neighbor_rpc);
+        let new_blocks = self
+            .tenure_downloads
+            .run(network, &mut self.neighbor_rpc, chainstate);
 
         new_blocks
     }
@@ -1318,7 +1331,7 @@ impl NakamotoDownloadStateMachine {
         &mut self,
         network: &mut PeerNetwork,
         sortdb: &SortitionDB,
-        chainstate: &StacksChainState,
+        chainstate: &mut StacksChainState,
         highest_processed_block_id: Option<StacksBlockId>,
     ) -> HashMap<ConsensusHash, Vec<NakamotoBlock>> {
         // queue up more downloaders
@@ -1340,7 +1353,7 @@ impl NakamotoDownloadStateMachine {
         // already downloaded all confirmed tenures), so there's no risk of clobberring any other
         // in-flight requests.
         let new_confirmed_blocks = if self.tenure_downloads.inflight() > 0 {
-            self.download_confirmed_tenures(network, 0)
+            self.download_confirmed_tenures(network, chainstate, 0)
         } else {
             HashMap::new()
         };
@@ -1415,7 +1428,7 @@ impl NakamotoDownloadStateMachine {
         burnchain_height: u64,
         network: &mut PeerNetwork,
         sortdb: &SortitionDB,
-        chainstate: &StacksChainState,
+        chainstate: &mut StacksChainState,
         ibd: bool,
     ) -> HashMap<ConsensusHash, Vec<NakamotoBlock>> {
         debug!(
@@ -1462,6 +1475,7 @@ impl NakamotoDownloadStateMachine {
             NakamotoDownloadState::Confirmed => {
                 let new_blocks = self.download_confirmed_tenures(
                     network,
+                    chainstate,
                     usize::try_from(network.get_connection_opts().max_inflight_blocks)
                         .expect("FATAL: max_inflight_blocks exceeds usize::MAX"),
                 );
diff --git a/stackslib/src/net/download/nakamoto/tenure.rs b/stackslib/src/net/download/nakamoto/tenure.rs
index ba1ac81033..0f4e3d53cb 100644
--- a/stackslib/src/net/download/nakamoto/tenure.rs
+++ b/stackslib/src/net/download/nakamoto/tenure.rs
@@ -98,6 +98,10 @@ impl WantedTenure {
 pub struct TenureStartEnd {
     /// Consensus hash that identifies the start of the tenure
     pub tenure_id_consensus_hash: ConsensusHash,
+    /// Consensus hash that identifies the snapshot with the start block ID
+    pub start_block_snapshot_consensus_hash: ConsensusHash,
+    /// Consensus hash that identifies the snapshot with the end block ID
+    pub end_block_snapshot_consensus_hash: ConsensusHash,
     /// Burnchain block height of tenure ID consensus hash
     pub tenure_id_burn_block_height: u64,
     /// Tenure-start block ID
@@ -122,7 +126,9 @@ impl TenureStartEnd {
     pub fn new(
         tenure_id_consensus_hash: ConsensusHash,
         tenure_id_burn_block_height: u64,
+        start_block_snapshot_consensus_hash: ConsensusHash,
         start_block_id: StacksBlockId,
+        end_block_snapshot_consensus_hash: ConsensusHash,
         end_block_id: StacksBlockId,
         start_reward_cycle: u64,
         end_reward_cycle: u64,
@@ -131,7 +137,9 @@ impl TenureStartEnd {
         Self {
             tenure_id_consensus_hash,
             tenure_id_burn_block_height,
+            start_block_snapshot_consensus_hash,
             start_block_id,
+            end_block_snapshot_consensus_hash,
             end_block_id,
             start_reward_cycle,
             end_reward_cycle,
@@ -219,7 +227,9 @@ impl TenureStartEnd {
             let tenure_start_end = TenureStartEnd::new(
                 wt.tenure_id_consensus_hash.clone(),
                 wt.burn_height,
+                wt_start.tenure_id_consensus_hash.clone(),
                 wt_start.winning_block_id.clone(),
+                wt_end.tenure_id_consensus_hash.clone(),
                 wt_end.winning_block_id.clone(),
                 rc,
                 rc,
@@ -328,7 +338,9 @@ impl TenureStartEnd {
             let mut tenure_start_end = TenureStartEnd::new(
                 wt.tenure_id_consensus_hash.clone(),
                 wt.burn_height,
+                wt_start.tenure_id_consensus_hash.clone(),
                 wt_start.winning_block_id.clone(),
+                wt_end.tenure_id_consensus_hash.clone(),
                 wt_end.winning_block_id.clone(),
                 rc,
                 pox_constants
diff --git a/stackslib/src/net/download/nakamoto/tenure_downloader.rs b/stackslib/src/net/download/nakamoto/tenure_downloader.rs
index 4c5efaccdd..e2716e8252 100644
--- a/stackslib/src/net/download/nakamoto/tenure_downloader.rs
+++ b/stackslib/src/net/download/nakamoto/tenure_downloader.rs
@@ -43,7 +43,7 @@ use crate::chainstate::nakamoto::{
 use crate::chainstate::stacks::boot::RewardSet;
 use crate::chainstate::stacks::db::StacksChainState;
 use crate::chainstate::stacks::{
-    Error as chainstate_error, StacksBlockHeader, TenureChangePayload,
+    Error as chainstate_error, StacksBlockHeader, TenureChangePayload, TransactionPayload,
 };
 use crate::core::{
     EMPTY_MICROBLOCK_PARENT_HASH, FIRST_BURNCHAIN_CONSENSUS_HASH, FIRST_STACKS_BLOCK_HASH,
@@ -119,9 +119,13 @@ impl fmt::Display for NakamotoTenureDownloadState {
 pub struct NakamotoTenureDownloader {
     /// Consensus hash that identifies this tenure
     pub tenure_id_consensus_hash: ConsensusHash,
+    /// Consensus hash that identifies the snapshot from whence we obtained tenure_start_block_id
+    pub start_block_snapshot_consensus_hash: ConsensusHash,
     /// Stacks block ID of the tenure-start block.  Learned from the inventory state machine and
     /// sortition DB.
     pub tenure_start_block_id: StacksBlockId,
+    /// Consensus hash that identifies the snapshot from whence we obtained tenure_end_block_id
+    pub end_block_snapshot_consensus_hash: ConsensusHash,
     /// Stacks block ID of the last block in this tenure (this will be the tenure-start block ID
     /// for some other tenure).  Learned from the inventory state machine and sortition DB.
     pub tenure_end_block_id: StacksBlockId,
@@ -150,19 +154,27 @@ pub struct NakamotoTenureDownloader {
 impl NakamotoTenureDownloader {
     pub fn new(
         tenure_id_consensus_hash: ConsensusHash,
+        start_block_snapshot_consensus_hash: ConsensusHash,
         tenure_start_block_id: StacksBlockId,
+        end_block_snapshot_consensus_hash: ConsensusHash,
         tenure_end_block_id: StacksBlockId,
         naddr: NeighborAddress,
         start_signer_keys: RewardSet,
         end_signer_keys: RewardSet,
     ) -> Self {
         debug!(
-            "Instantiate downloader to {} for tenure {}: {}-{}",
-            &naddr, &tenure_id_consensus_hash, &tenure_start_block_id, &tenure_end_block_id,
+            "Instantiate downloader to {}-{} for tenure {}: {}-{}",
+            &naddr,
+            &tenure_id_consensus_hash,
+            &start_block_snapshot_consensus_hash,
+            &tenure_start_block_id,
+            &tenure_end_block_id,
         );
         Self {
             tenure_id_consensus_hash,
+            start_block_snapshot_consensus_hash,
             tenure_start_block_id,
+            end_block_snapshot_consensus_hash,
             tenure_end_block_id,
             naddr,
             start_signer_keys,
@@ -270,7 +282,9 @@ impl NakamotoTenureDownloader {
             return Err(NetError::InvalidState);
         };
 
-        if self.tenure_end_block_id != tenure_end_block.header.block_id() {
+        if self.tenure_end_block_id != tenure_end_block.header.block_id()
+            && self.tenure_end_block_id != StacksBlockId([0x00; 32])
+        {
             // not the block we asked for
             warn!("Invalid tenure-end block: unexpected";
                   "tenure_id" => %self.tenure_id_consensus_hash,
@@ -541,6 +555,177 @@ impl NakamotoTenureDownloader {
         Ok(Some(request))
     }
 
+    /// Advance the state of the downloader from chainstate, if possible.
+    /// For example, a tenure-start or tenure-end block may have been pushed to us already (or they
+    /// may be shadow blocks)
+    pub fn try_advance_from_chainstate(
+        &mut self,
+        chainstate: &mut StacksChainState,
+    ) -> Result<(), NetError> {
+        loop {
+            match self.state {
+                NakamotoTenureDownloadState::GetTenureStartBlock(
+                    start_block_id,
+                    start_request_time,
+                ) => {
+                    if chainstate
+                        .nakamoto_blocks_db()
+                        .is_shadow_tenure(&self.start_block_snapshot_consensus_hash)?
+                    {
+                        debug!(
+                            "Tenure {} start-block confirmed by shadow tenure {}",
+                            &self.tenure_id_consensus_hash,
+                            &self.start_block_snapshot_consensus_hash
+                        );
+                        let Some(shadow_block) = chainstate
+                            .nakamoto_blocks_db()
+                            .get_shadow_tenure_start_block(
+                                &self.start_block_snapshot_consensus_hash,
+                            )?
+                        else {
+                            warn!(
+                                "No tenure-start block for shadow tenure {}",
+                                &self.start_block_snapshot_consensus_hash
+                            );
+                            break;
+                        };
+
+                        // the coinbase of a tenure-start block of a shadow tenure contains the
+                        // block-id of the parent tenure's start block (i.e. the information that
+                        // would have been gleaned from a block-commit, if there was one).
+                        let Some(shadow_coinbase) = shadow_block.get_coinbase_tx() else {
+                            warn!("Shadow block {} has no coinbase", &shadow_block.block_id());
+                            break;
+                        };
+
+                        let TransactionPayload::Coinbase(coinbase_payload, ..) =
+                            &shadow_coinbase.payload
+                        else {
+                            warn!(
+                                "Shadow block {} coinbase tx is not a Coinbase",
+                                &shadow_block.block_id()
+                            );
+                            break;
+                        };
+
+                        let tenure_start_block_id = StacksBlockId(coinbase_payload.0.clone());
+
+                        info!(
+                            "Tenure {} starts at shadow tenure-start {}, not {}",
+                            &self.tenure_id_consensus_hash, &tenure_start_block_id, &start_block_id
+                        );
+                        self.tenure_start_block_id = tenure_start_block_id.clone();
+                        self.state = NakamotoTenureDownloadState::GetTenureStartBlock(
+                            tenure_start_block_id,
+                            start_request_time,
+                        );
+                        if let Some((tenure_start_block, _sz)) = chainstate
+                            .nakamoto_blocks_db()
+                            .get_nakamoto_block(&self.tenure_start_block_id)?
+                        {
+                            // normal block on disk
+                            self.try_accept_tenure_start_block(tenure_start_block)?;
+                        }
+                    } else if let Some((tenure_start_block, _sz)) = chainstate
+                        .nakamoto_blocks_db()
+                        .get_nakamoto_block(&start_block_id)?
+                    {
+                        // we have downloaded this block already
+                        self.try_accept_tenure_start_block(tenure_start_block)?;
+                    } else {
+                        break;
+                    }
+                    if let NakamotoTenureDownloadState::GetTenureStartBlock(..) = &self.state {
+                        break;
+                    }
+                }
+                NakamotoTenureDownloadState::GetTenureEndBlock(
+                    end_block_id,
+                    start_request_time,
+                ) => {
+                    if chainstate
+                        .nakamoto_blocks_db()
+                        .is_shadow_tenure(&self.end_block_snapshot_consensus_hash)?
+                    {
+                        debug!(
+                            "Tenure {} end-block confirmed by shadow tenure {}",
+                            &self.tenure_id_consensus_hash, &self.end_block_snapshot_consensus_hash
+                        );
+                        let Some(shadow_block) = chainstate
+                            .nakamoto_blocks_db()
+                            .get_shadow_tenure_start_block(
+                                &self.end_block_snapshot_consensus_hash,
+                            )?
+                        else {
+                            warn!(
+                                "No tenure-start block for shadow tenure {}",
+                                &self.end_block_snapshot_consensus_hash
+                            );
+                            break;
+                        };
+
+                        // the coinbase of a tenure-start block of a shadow tenure contains the
+                        // block-id of the parent tenure's start block (i.e. the information that
+                        // would have been gleaned from a block-commit, if there was one).
+                        let Some(shadow_coinbase) = shadow_block.get_coinbase_tx() else {
+                            warn!("Shadow block {} has no coinbase", &shadow_block.block_id());
+                            break;
+                        };
+
+                        let TransactionPayload::Coinbase(coinbase_payload, ..) =
+                            &shadow_coinbase.payload
+                        else {
+                            warn!(
+                                "Shadow block {} coinbase tx is not a Coinbase",
+                                &shadow_block.block_id()
+                            );
+                            break;
+                        };
+
+                        let tenure_end_block_id = StacksBlockId(coinbase_payload.0.clone());
+
+                        info!(
+                            "Tenure {} ends at shadow tenure-start {}, not {}",
+                            &self.tenure_id_consensus_hash, &tenure_end_block_id, &end_block_id
+                        );
+                        self.tenure_end_block_id = tenure_end_block_id.clone();
+                        self.state = NakamotoTenureDownloadState::GetTenureEndBlock(
+                            tenure_end_block_id,
+                            start_request_time,
+                        );
+                        if let Some((tenure_end_block, _sz)) = chainstate
+                            .nakamoto_blocks_db()
+                            .get_nakamoto_block(&self.tenure_end_block_id)?
+                        {
+                            // normal block on disk
+                            self.try_accept_tenure_end_block(&tenure_end_block)?;
+                        }
+                    } else if let Some((tenure_end_block, _sz)) = chainstate
+                        .nakamoto_blocks_db()
+                        .get_nakamoto_block(&end_block_id)?
+                    {
+                        // normal block on disk
+                        self.try_accept_tenure_end_block(&tenure_end_block)?;
+                    } else {
+                        break;
+                    };
+                    if let NakamotoTenureDownloadState::GetTenureEndBlock(..) = &self.state {
+                        break;
+                    }
+                }
+                NakamotoTenureDownloadState::GetTenureBlocks(..) => {
+                    // TODO: look at the chainstate and find out what we don't have to download
+                    // TODO: skip shadow tenures
+                    break;
+                }
+                NakamotoTenureDownloadState::Done => {
+                    break;
+                }
+            }
+        }
+        Ok(())
+    }
+
     /// Begin the next download request for this state machine.  The request will be sent to the
     /// data URL corresponding to self.naddr.
     /// Returns Ok(true) if we sent the request, or there's already an in-flight request.  The
diff --git a/stackslib/src/net/download/nakamoto/tenure_downloader_set.rs b/stackslib/src/net/download/nakamoto/tenure_downloader_set.rs
index b5514558b8..08714f5cbf 100644
--- a/stackslib/src/net/download/nakamoto/tenure_downloader_set.rs
+++ b/stackslib/src/net/download/nakamoto/tenure_downloader_set.rs
@@ -513,7 +513,9 @@ impl NakamotoTenureDownloaderSet {
 
             let tenure_download = NakamotoTenureDownloader::new(
                 ch.clone(),
+                tenure_info.start_block_snapshot_consensus_hash.clone(),
                 tenure_info.start_block_id.clone(),
+                tenure_info.end_block_snapshot_consensus_hash.clone(),
                 tenure_info.end_block_id.clone(),
                 naddr.clone(),
                 start_reward_set.clone(),
@@ -540,6 +542,7 @@ impl NakamotoTenureDownloaderSet {
         &mut self,
         network: &mut PeerNetwork,
         neighbor_rpc: &mut NeighborRPC,
+        chainstate: &mut StacksChainState,
     ) -> HashMap<ConsensusHash, Vec<NakamotoBlock>> {
         let addrs: Vec<_> = self.peers.keys().cloned().collect();
         let mut finished = vec![];
@@ -565,6 +568,17 @@ impl NakamotoTenureDownloaderSet {
                 finished_tenures.push(CompletedTenure::from(downloader));
                 continue;
             }
+
+            let _ = downloader
+                .try_advance_from_chainstate(chainstate)
+                .map_err(|e| {
+                    warn!(
+                        "Failed to advance downloader in state {} for {}: {:?}",
+                        &downloader.state, &downloader.naddr, &e
+                    );
+                    e
+                });
+
             debug!(
                 "Send request to {naddr} for tenure {} (state {})",
                 &downloader.tenure_id_consensus_hash, &downloader.state
diff --git a/stackslib/src/net/download/nakamoto/tenure_downloader_unconfirmed.rs b/stackslib/src/net/download/nakamoto/tenure_downloader_unconfirmed.rs
index ddfd35fa97..9a9ee51b07 100644
--- a/stackslib/src/net/download/nakamoto/tenure_downloader_unconfirmed.rs
+++ b/stackslib/src/net/download/nakamoto/tenure_downloader_unconfirmed.rs
@@ -735,7 +735,9 @@ impl NakamotoUnconfirmedTenureDownloader {
         );
         let ntd = NakamotoTenureDownloader::new(
             tenure_tip.parent_consensus_hash.clone(),
+            tenure_tip.consensus_hash.clone(),
             tenure_tip.parent_tenure_start_block_id.clone(),
+            tenure_tip.consensus_hash.clone(),
             tenure_tip.tenure_start_block_id.clone(),
             self.naddr.clone(),
             confirmed_signer_keys.clone(),
@@ -777,6 +779,44 @@ impl NakamotoUnconfirmedTenureDownloader {
         }
     }
 
+    /// Advance the state of the downloader from chainstate, if possible.
+    /// For example, a tenure-start block may have been pushed to us already (or it
+    /// may be a shadow block)
+    pub fn try_advance_from_chainstate(
+        &mut self,
+        chainstate: &StacksChainState,
+    ) -> Result<(), NetError> {
+        loop {
+            match self.state {
+                NakamotoUnconfirmedDownloadState::GetTenureInfo => {
+                    // gotta send that request
+                    break;
+                }
+                NakamotoUnconfirmedDownloadState::GetTenureStartBlock(start_block_id) => {
+                    // if we have this, then load it up
+                    let Some((tenure_start_block, _sz)) = chainstate
+                        .nakamoto_blocks_db()
+                        .get_nakamoto_block(&start_block_id)?
+                    else {
+                        break;
+                    };
+                    self.try_accept_unconfirmed_tenure_start_block(tenure_start_block)?;
+                    if let NakamotoUnconfirmedDownloadState::GetTenureStartBlock(..) = &self.state {
+                        break;
+                    }
+                }
+                NakamotoUnconfirmedDownloadState::GetUnconfirmedTenureBlocks(..) => {
+                    // TODO: look at the chainstate and find out what we don't have to download
+                    break;
+                }
+                NakamotoUnconfirmedDownloadState::Done => {
+                    break;
+                }
+            }
+        }
+        Ok(())
+    }
+
     /// Begin the next download request for this state machine.
     /// Returns Ok(()) if we sent the request, or there's already an in-flight request.  The
     /// caller should try this again until it gets one of the other possible return values.  It's
diff --git a/stackslib/src/net/httpcore.rs b/stackslib/src/net/httpcore.rs
index c58355a6a9..9b2dd1e106 100644
--- a/stackslib/src/net/httpcore.rs
+++ b/stackslib/src/net/httpcore.rs
@@ -1197,8 +1197,9 @@ impl StacksHttp {
         let (response_preamble, response_contents) = match request_result {
             Ok((rp, rc)) => (rp, rc),
             Err(NetError::Http(e)) => {
+                debug!("RPC handler for {} failed: {:?}", decoded_path, &e);
                 return StacksHttpResponse::new_error(&request_preamble, &*e.into_http_error())
-                    .try_into_contents()
+                    .try_into_contents();
             }
             Err(e) => {
                 warn!("Irrecoverable error when handling request"; "path" => %request_preamble.path_and_query_str, "error" => %e);
diff --git a/stackslib/src/net/inv/nakamoto.rs b/stackslib/src/net/inv/nakamoto.rs
index d5b08f56d2..e832b70184 100644
--- a/stackslib/src/net/inv/nakamoto.rs
+++ b/stackslib/src/net/inv/nakamoto.rs
@@ -660,10 +660,12 @@ impl NakamotoTenureInv {
         match reply.payload {
             StacksMessageType::NakamotoInv(inv_data) => {
                 debug!(
-                    "{:?}: got NakamotoInv: {:?}",
+                    "{:?}: got NakamotoInv from {:?}: {:?}",
                     network.get_local_peer(),
+                    &self.neighbor_address,
                     &inv_data
                 );
+
                 let ret = self.merge_tenure_inv(inv_data.tenures, self.reward_cycle());
                 self.next_reward_cycle();
                 return Ok(ret);
diff --git a/stackslib/src/net/mod.rs b/stackslib/src/net/mod.rs
index 389d565af5..89e56fe29c 100644
--- a/stackslib/src/net/mod.rs
+++ b/stackslib/src/net/mod.rs
@@ -4061,6 +4061,22 @@ pub mod test {
             self.sortdb.as_ref().unwrap()
         }
 
+        pub fn with_dbs<F, R>(&mut self, f: F) -> R
+        where
+            F: FnOnce(&mut TestPeer, &mut SortitionDB, &mut TestStacksNode, &mut MemPoolDB) -> R,
+        {
+            let mut sortdb = self.sortdb.take().unwrap();
+            let mut stacks_node = self.stacks_node.take().unwrap();
+            let mut mempool = self.mempool.take().unwrap();
+
+            let res = f(self, &mut sortdb, &mut stacks_node, &mut mempool);
+
+            self.stacks_node = Some(stacks_node);
+            self.sortdb = Some(sortdb);
+            self.mempool = Some(mempool);
+            res
+        }
+
         pub fn with_db_state<F, R>(&mut self, f: F) -> Result<R, net_error>
         where
             F: FnOnce(
@@ -4726,6 +4742,9 @@ pub mod test {
             all_blocks: Vec<NakamotoBlock>,
             expected_siblings: usize,
         ) {
+            if !self.mine_malleablized_blocks {
+                return;
+            }
             for block in all_blocks.iter() {
                 let sighash = block.header.signer_signature_hash();
                 let siblings = self
diff --git a/stackslib/src/net/relay.rs b/stackslib/src/net/relay.rs
index b5fbf76cf4..cb7d310321 100644
--- a/stackslib/src/net/relay.rs
+++ b/stackslib/src/net/relay.rs
@@ -933,6 +933,11 @@ impl Relayer {
             &obtained_method;
             "block_id" => %block.header.block_id(),
         );
+        if block.is_shadow_block() {
+            // drop, since we can get these from ourselves when downloading a tenure that ends in
+            // a shadow block.
+            return Ok(BlockAcceptResponse::AlreadyStored);
+        }
 
         if fault_injection::ignore_block(block.header.chain_length, &burnchain.working_dir) {
             return Ok(BlockAcceptResponse::Rejected(


Test patch:
diff --git a/.github/workflows/bitcoin-tests.yml b/.github/workflows/bitcoin-tests.yml
index ebe9f433a9..f3b48d5dfa 100644
--- a/.github/workflows/bitcoin-tests.yml
+++ b/.github/workflows/bitcoin-tests.yml
@@ -139,6 +139,7 @@ jobs:
           - tests::nakamoto_integrations::utxo_check_on_startup_panic
           - tests::nakamoto_integrations::utxo_check_on_startup_recover
           - tests::nakamoto_integrations::v3_signer_api_endpoint
+          - tests::nakamoto_integrations::test_shadow_recovery
           - tests::nakamoto_integrations::signer_chainstate
           - tests::nakamoto_integrations::clarity_cost_spend_down
           - tests::nakamoto_integrations::v3_blockbyheight_api_endpoint
diff --git a/stackslib/src/burnchains/tests/mod.rs b/stackslib/src/burnchains/tests/mod.rs
index e7fa51a89c..887b56861b 100644
--- a/stackslib/src/burnchains/tests/mod.rs
+++ b/stackslib/src/burnchains/tests/mod.rs
@@ -351,10 +351,30 @@ impl TestMinerFactory {
 
 impl TestBurnchainBlock {
     pub fn new(parent_snapshot: &BlockSnapshot, fork_id: u64) -> TestBurnchainBlock {
+        let burn_header_hash = BurnchainHeaderHash::from_test_data(
+            parent_snapshot.block_height + 1,
+            &parent_snapshot.index_root,
+            fork_id,
+        );
         TestBurnchainBlock {
             parent_snapshot: parent_snapshot.clone(),
             block_height: parent_snapshot.block_height + 1,
-            txs: vec![],
+            txs: vec![
+                // make sure that no block-commit gets vtxindex == 0 unless explicitly structured.
+                // This prestx mocks a burnchain coinbase
+                BlockstackOperationType::PreStx(PreStxOp {
+                    output: StacksAddress::burn_address(false),
+                    txid: Txid::from_test_data(
+                        parent_snapshot.block_height + 1,
+                        0,
+                        &burn_header_hash,
+                        128,
+                    ),
+                    vtxindex: 0,
+                    block_height: parent_snapshot.block_height + 1,
+                    burn_header_hash,
+                }),
+            ],
             fork_id: fork_id,
             timestamp: get_epoch_time_secs(),
         }
@@ -397,6 +417,7 @@ impl TestBurnchainBlock {
         parent_block_snapshot: Option<&BlockSnapshot>,
         new_seed: Option<VRFSeed>,
         epoch_marker: u8,
+        parent_is_shadow: bool,
     ) -> LeaderBlockCommitOp {
         let pubks = miner
             .privks
@@ -435,6 +456,13 @@ impl TestBurnchainBlock {
         )
         .expect("FATAL: failed to read block commit");
 
+        if parent_is_shadow {
+            assert!(
+                get_commit_res.is_none(),
+                "FATAL: shadow parent should not have a block-commit"
+            );
+        }
+
         let input = SortitionDB::get_last_block_commit_by_sender(ic.conn(), &apparent_sender)
             .unwrap()
             .map(|commit| (commit.txid.clone(), 1 + (commit.commit_outs.len() as u32)))
@@ -454,7 +482,8 @@ impl TestBurnchainBlock {
                     block_hash,
                     self.block_height,
                     &new_seed,
-                    &parent,
+                    parent.block_height as u32,
+                    parent.vtxindex as u16,
                     leader_key.block_height as u32,
                     leader_key.vtxindex as u16,
                     burn_fee,
@@ -464,16 +493,42 @@ impl TestBurnchainBlock {
                 txop
             }
             None => {
-                // initial
-                let txop = LeaderBlockCommitOp::initial(
-                    block_hash,
-                    self.block_height,
-                    &new_seed,
-                    leader_key,
-                    burn_fee,
-                    &input,
-                    &apparent_sender,
-                );
+                let txop = if parent_is_shadow {
+                    test_debug!(
+                        "Block-commit for {} (burn height {}) builds on shadow sortition",
+                        block_hash,
+                        self.block_height
+                    );
+
+                    LeaderBlockCommitOp::new(
+                        block_hash,
+                        self.block_height,
+                        &new_seed,
+                        last_snapshot_with_sortition.block_height as u32,
+                        0,
+                        leader_key.block_height as u32,
+                        leader_key.vtxindex as u16,
+                        burn_fee,
+                        &input,
+                        &apparent_sender,
+                    )
+                } else {
+                    // initial
+                    test_debug!(
+                        "Block-commit for {} (burn height {}) builds on genesis",
+                        block_hash,
+                        self.block_height,
+                    );
+                    LeaderBlockCommitOp::initial(
+                        block_hash,
+                        self.block_height,
+                        &new_seed,
+                        leader_key,
+                        burn_fee,
+                        &input,
+                        &apparent_sender,
+                    )
+                };
                 txop
             }
         };
@@ -517,6 +572,7 @@ impl TestBurnchainBlock {
             parent_block_snapshot,
             None,
             STACKS_EPOCH_2_4_MARKER,
+            false,
         )
     }
 
diff --git a/stackslib/src/chainstate/nakamoto/coordinator/tests.rs b/stackslib/src/chainstate/nakamoto/coordinator/tests.rs
index 23bf3313e9..0525717981 100644
--- a/stackslib/src/chainstate/nakamoto/coordinator/tests.rs
+++ b/stackslib/src/chainstate/nakamoto/coordinator/tests.rs
@@ -576,7 +576,7 @@ impl<'a> TestPeer<'a> {
         coinbase_tx: &StacksTransaction,
         miner_setup: F,
         after_block: G,
-    ) -> NakamotoBlock
+    ) -> Result<NakamotoBlock, ChainstateError>
     where
         F: FnMut(&mut NakamotoBlockBuilder),
         G: FnMut(&mut NakamotoBlock) -> bool,
@@ -606,7 +606,7 @@ impl<'a> TestPeer<'a> {
         coinbase_tx: &StacksTransaction,
         miner_setup: F,
         after_block: G,
-    ) -> NakamotoBlock
+    ) -> Result<NakamotoBlock, ChainstateError>
     where
         F: FnMut(&mut NakamotoBlockBuilder),
         G: FnMut(&mut NakamotoBlock) -> bool,
@@ -631,7 +631,7 @@ impl<'a> TestPeer<'a> {
                         sortdb,
                         &sender_key,
                         sender_acct.nonce,
-                        100,
+                        200,
                         1,
                         &recipient_addr,
                     );
@@ -642,10 +642,10 @@ impl<'a> TestPeer<'a> {
                 }
             },
             after_block,
-        );
+        )?;
         assert_eq!(blocks_and_sizes.len(), 1);
         let block = blocks_and_sizes.pop().unwrap().0;
-        block
+        Ok(block)
     }
 
     pub fn mine_tenure<F>(&mut self, block_builder: F) -> Vec<(NakamotoBlock, u64, ExecutionCost)>
@@ -707,15 +707,41 @@ impl<'a> TestPeer<'a> {
             block_builder,
             |_| true,
         )
+        .unwrap()
     }
 
     pub fn single_block_tenure<S, F, G>(
         &mut self,
         sender_key: &StacksPrivateKey,
         miner_setup: S,
-        mut after_burn_ops: F,
+        after_burn_ops: F,
         after_block: G,
     ) -> (NakamotoBlock, u64, StacksTransaction, StacksTransaction)
+    where
+        S: FnMut(&mut NakamotoBlockBuilder),
+        F: FnMut(&mut Vec<BlockstackOperationType>),
+        G: FnMut(&mut NakamotoBlock) -> bool,
+    {
+        self.single_block_tenure_fallible(sender_key, miner_setup, after_burn_ops, after_block)
+            .unwrap()
+    }
+
+    /// Produce a single-block tenure, containing a stx-transfer sent from `sender_key`.
+    ///
+    /// * `after_burn_ops` is called right after `self.begin_nakamoto_tenure` to modify any burn ops
+    /// for this tenure
+    ///
+    /// * `miner_setup` is called right after the Nakamoto block builder is constructed, but before
+    /// any txs are mined
+    ///
+    /// * `after_block` is called right after the block is assembled, but before it is signed.
+    pub fn single_block_tenure_fallible<S, F, G>(
+        &mut self,
+        sender_key: &StacksPrivateKey,
+        miner_setup: S,
+        mut after_burn_ops: F,
+        after_block: G,
+    ) -> Result<(NakamotoBlock, u64, StacksTransaction, StacksTransaction), ChainstateError>
     where
         S: FnMut(&mut NakamotoBlockBuilder),
         F: FnMut(&mut Vec<BlockstackOperationType>),
@@ -770,9 +796,9 @@ impl<'a> TestPeer<'a> {
             &coinbase_tx,
             miner_setup,
             after_block,
-        );
+        )?;
 
-        (block, burn_height, tenure_change_tx, coinbase_tx)
+        Ok((block, burn_height, tenure_change_tx, coinbase_tx))
     }
 }
 
@@ -1422,24 +1448,27 @@ fn pox_treatment() {
 
     // set the bitvec to a heterogenous one: either punish or
     //  reward is acceptable, so this block should just process.
-    let block = peer.mine_single_block_tenure(
-        &private_key,
-        &tenure_change_tx,
-        &coinbase_tx,
-        |_| {},
-        |block| {
-            // each stacker has 3 entries in the bitvec.
-            // entries are ordered by PoxAddr, so this makes every entry a 1-of-3
-            block.header.pox_treatment = BitVec::try_from(
-                [
-                    false, false, true, false, false, true, false, false, true, false, false, true,
-                ]
-                .as_slice(),
-            )
-            .unwrap();
-            true
-        },
-    );
+    let block = peer
+        .mine_single_block_tenure(
+            &private_key,
+            &tenure_change_tx,
+            &coinbase_tx,
+            |_| {},
+            |block| {
+                // each stacker has 3 entries in the bitvec.
+                // entries are ordered by PoxAddr, so this makes every entry a 1-of-3
+                block.header.pox_treatment = BitVec::try_from(
+                    [
+                        false, false, true, false, false, true, false, false, true, false, false,
+                        true,
+                    ]
+                    .as_slice(),
+                )
+                .unwrap();
+                true
+            },
+        )
+        .unwrap();
     blocks.push(block);
 
     // now we need to test punishment!
@@ -1510,23 +1539,26 @@ fn pox_treatment() {
 
     // set the bitvec to a heterogenous one: either punish or
     //  reward is acceptable, so this block should just process.
-    let block = peer.mine_single_block_tenure(
-        &private_key,
-        &tenure_change_tx,
-        &coinbase_tx,
-        |miner| {
-            // each stacker has 3 entries in the bitvec.
-            // entries are ordered by PoxAddr, so this makes every entry a 1-of-3
-            miner.header.pox_treatment = BitVec::try_from(
-                [
-                    false, false, true, false, false, true, false, false, true, false, false, true,
-                ]
-                .as_slice(),
-            )
-            .unwrap();
-        },
-        |_block| true,
-    );
+    let block = peer
+        .mine_single_block_tenure(
+            &private_key,
+            &tenure_change_tx,
+            &coinbase_tx,
+            |miner| {
+                // each stacker has 3 entries in the bitvec.
+                // entries are ordered by PoxAddr, so this makes every entry a 1-of-3
+                miner.header.pox_treatment = BitVec::try_from(
+                    [
+                        false, false, true, false, false, true, false, false, true, false, false,
+                        true,
+                    ]
+                    .as_slice(),
+                )
+                .unwrap();
+            },
+            |_block| true,
+        )
+        .unwrap();
     blocks.push(block);
 
     let tip = {
@@ -3212,7 +3244,7 @@ fn test_stacks_on_burnchain_ops() {
 
             // mocked
             txid: Txid([i as u8; 32]),
-            vtxindex: 1,
+            vtxindex: 11,
             block_height: block_height + 1,
             burn_header_hash: BurnchainHeaderHash([0x00; 32]),
         }));
@@ -3232,7 +3264,7 @@ fn test_stacks_on_burnchain_ops() {
 
             // mocked
             txid: Txid([(i as u8) | 0x80; 32]),
-            vtxindex: 2,
+            vtxindex: 12,
             block_height: block_height + 1,
             burn_header_hash: BurnchainHeaderHash([0x00; 32]),
         }));
@@ -3244,7 +3276,7 @@ fn test_stacks_on_burnchain_ops() {
 
             // mocked
             txid: Txid([(i as u8) | 0x40; 32]),
-            vtxindex: 3,
+            vtxindex: 13,
             block_height: block_height + 1,
             burn_header_hash: BurnchainHeaderHash([0x00; 32]),
         }));
@@ -3263,7 +3295,7 @@ fn test_stacks_on_burnchain_ops() {
 
                 // mocked
                 txid: Txid([(i as u8) | 0xc0; 32]),
-                vtxindex: 4,
+                vtxindex: 14,
                 block_height: block_height + 1,
                 burn_header_hash: BurnchainHeaderHash([0x00; 32]),
             },
diff --git a/stackslib/src/chainstate/nakamoto/tests/mod.rs b/stackslib/src/chainstate/nakamoto/tests/mod.rs
index ea163730ec..94ef81c077 100644
--- a/stackslib/src/chainstate/nakamoto/tests/mod.rs
+++ b/stackslib/src/chainstate/nakamoto/tests/mod.rs
@@ -1663,7 +1663,9 @@ pub fn test_load_store_update_nakamoto_blocks() {
 /// Tests:
 /// * NakamotoBlockHeader::check_miner_signature
 /// * NakamotoBlockHeader::check_tenure_tx
-/// * NakamotoBlockHeader::check_coinbase_tx
+/// * NakamotoBlockHeader::is_shadow_block
+/// * NakamotoBlockHeader::check_normal_coinbase_tx
+/// * NakamotoBlockHeader::check_shadow_coinbase_tx
 #[test]
 fn test_nakamoto_block_static_verification() {
     let private_key = StacksPrivateKey::new();
@@ -1674,9 +1676,25 @@ fn test_nakamoto_block_static_verification() {
     let sortition_hash = SortitionHash([0x01; 32]);
     let vrf_proof = VRF::prove(&vrf_privkey, sortition_hash.as_bytes());
 
+    let burn_recipient = StacksAddress::burn_address(false).to_account_principal();
+    let alt_recipient = StacksAddress::p2pkh(false, &StacksPublicKey::from_private(&private_key_2))
+        .to_account_principal();
+
     let coinbase_payload =
         TransactionPayload::Coinbase(CoinbasePayload([0x12; 32]), None, Some(vrf_proof.clone()));
 
+    let coinbase_recipient_payload = TransactionPayload::Coinbase(
+        CoinbasePayload([0x12; 32]),
+        Some(alt_recipient),
+        Some(vrf_proof.clone()),
+    );
+
+    let coinbase_shadow_recipient_payload = TransactionPayload::Coinbase(
+        CoinbasePayload([0x12; 32]),
+        Some(burn_recipient),
+        Some(vrf_proof.clone()),
+    );
+
     let mut coinbase_tx = StacksTransaction::new(
         TransactionVersion::Testnet,
         TransactionAuth::from_p2pkh(&private_key).unwrap(),
@@ -1685,6 +1703,22 @@ fn test_nakamoto_block_static_verification() {
     coinbase_tx.chain_id = 0x80000000;
     coinbase_tx.anchor_mode = TransactionAnchorMode::OnChainOnly;
 
+    let mut coinbase_recipient_tx = StacksTransaction::new(
+        TransactionVersion::Testnet,
+        TransactionAuth::from_p2pkh(&private_key).unwrap(),
+        coinbase_recipient_payload.clone(),
+    );
+    coinbase_recipient_tx.chain_id = 0x80000000;
+    coinbase_recipient_tx.anchor_mode = TransactionAnchorMode::OnChainOnly;
+
+    let mut coinbase_shadow_recipient_tx = StacksTransaction::new(
+        TransactionVersion::Testnet,
+        TransactionAuth::from_p2pkh(&private_key).unwrap(),
+        coinbase_shadow_recipient_payload.clone(),
+    );
+    coinbase_shadow_recipient_tx.chain_id = 0x80000000;
+    coinbase_shadow_recipient_tx.anchor_mode = TransactionAnchorMode::OnChainOnly;
+
     let tenure_change_payload = TenureChangePayload {
         tenure_consensus_hash: ConsensusHash([0x04; 20]), // same as in nakamoto header
         prev_tenure_consensus_hash: ConsensusHash([0x01; 20]),
@@ -1754,6 +1788,29 @@ fn test_nakamoto_block_static_verification() {
         MerkleTree::<Sha512Trunc256Sum>::new(&txid_vecs).root()
     };
 
+    let nakamoto_recipient_txs = vec![tenure_change_tx.clone(), coinbase_recipient_tx.clone()];
+    let nakamoto_recipient_tx_merkle_root = {
+        let txid_vecs = nakamoto_recipient_txs
+            .iter()
+            .map(|tx| tx.txid().as_bytes().to_vec())
+            .collect();
+
+        MerkleTree::<Sha512Trunc256Sum>::new(&txid_vecs).root()
+    };
+
+    let nakamoto_shadow_recipient_txs = vec![
+        tenure_change_tx.clone(),
+        coinbase_shadow_recipient_tx.clone(),
+    ];
+    let nakamoto_shadow_recipient_tx_merkle_root = {
+        let txid_vecs = nakamoto_shadow_recipient_txs
+            .iter()
+            .map(|tx| tx.txid().as_bytes().to_vec())
+            .collect();
+
+        MerkleTree::<Sha512Trunc256Sum>::new(&txid_vecs).root()
+    };
+
     let nakamoto_txs_bad_ch = vec![tenure_change_tx_bad_ch.clone(), coinbase_tx.clone()];
     let nakamoto_tx_merkle_root_bad_ch = {
         let txid_vecs = nakamoto_txs_bad_ch
@@ -1837,6 +1894,48 @@ fn test_nakamoto_block_static_verification() {
         txs: nakamoto_txs_bad_miner_sig,
     };
 
+    let mut nakamoto_recipient_header = NakamotoBlockHeader {
+        version: 1,
+        chain_length: 457,
+        burn_spent: 126,
+        consensus_hash: tenure_change_payload.tenure_consensus_hash.clone(),
+        parent_block_id: StacksBlockId([0x03; 32]),
+        tx_merkle_root: nakamoto_recipient_tx_merkle_root,
+        state_index_root: TrieHash([0x07; 32]),
+        timestamp: 8,
+        miner_signature: MessageSignature::empty(),
+        signer_signature: vec![],
+        pox_treatment: BitVec::zeros(1).unwrap(),
+    };
+    nakamoto_recipient_header.sign_miner(&private_key).unwrap();
+
+    let nakamoto_recipient_block = NakamotoBlock {
+        header: nakamoto_recipient_header.clone(),
+        txs: nakamoto_recipient_txs,
+    };
+
+    let mut nakamoto_shadow_recipient_header = NakamotoBlockHeader {
+        version: 1,
+        chain_length: 457,
+        burn_spent: 126,
+        consensus_hash: tenure_change_payload.tenure_consensus_hash.clone(),
+        parent_block_id: StacksBlockId([0x03; 32]),
+        tx_merkle_root: nakamoto_shadow_recipient_tx_merkle_root,
+        state_index_root: TrieHash([0x07; 32]),
+        timestamp: 8,
+        miner_signature: MessageSignature::empty(),
+        signer_signature: vec![],
+        pox_treatment: BitVec::zeros(1).unwrap(),
+    };
+    nakamoto_shadow_recipient_header
+        .sign_miner(&private_key)
+        .unwrap();
+
+    let nakamoto_shadow_recipient_block = NakamotoBlock {
+        header: nakamoto_shadow_recipient_header.clone(),
+        txs: nakamoto_shadow_recipient_txs,
+    };
+
     assert_eq!(
         nakamoto_block.header.recover_miner_pk().unwrap(),
         StacksPublicKey::from_private(&private_key)
@@ -1863,13 +1962,78 @@ fn test_nakamoto_block_static_verification() {
     let vrf_alt_pubkey = VRFPublicKey::from_private(&vrf_alt_privkey);
 
     assert!(nakamoto_block
-        .check_coinbase_tx(&vrf_pubkey, &sortition_hash)
+        .check_normal_coinbase_tx(&vrf_pubkey, &sortition_hash)
         .is_ok());
     assert!(nakamoto_block
-        .check_coinbase_tx(&vrf_pubkey, &SortitionHash([0x02; 32]))
+        .check_normal_coinbase_tx(&vrf_pubkey, &SortitionHash([0x02; 32]))
         .is_err());
     assert!(nakamoto_block
-        .check_coinbase_tx(&vrf_alt_pubkey, &sortition_hash)
+        .check_normal_coinbase_tx(&vrf_alt_pubkey, &sortition_hash)
+        .is_err());
+
+    let mut shadow_block = nakamoto_shadow_recipient_block.clone();
+    shadow_block.header.version |= 0x80;
+
+    assert!(!nakamoto_shadow_recipient_block.is_shadow_block());
+    assert!(shadow_block.is_shadow_block());
+
+    // miner key not checked for shadow blocks
+    assert!(shadow_block
+        .check_miner_signature(&Hash160::from_node_public_key(
+            &StacksPublicKey::from_private(&private_key_2)
+        ))
+        .is_ok());
+
+    // shadow block VRF is not checked
+    assert!(shadow_block.check_shadow_coinbase_tx(false).is_ok());
+
+    // shadow blocks need burn recipeints for coinbases
+    let mut shadow_block_no_recipient = nakamoto_block.clone();
+    shadow_block_no_recipient.header.version |= 0x80;
+
+    assert!(shadow_block_no_recipient.is_shadow_block());
+    assert!(shadow_block_no_recipient
+        .check_shadow_coinbase_tx(false)
+        .is_err());
+
+    let mut shadow_block_alt_recipient = nakamoto_block.clone();
+    shadow_block_alt_recipient.header.version |= 0x80;
+
+    assert!(shadow_block_alt_recipient.is_shadow_block());
+    assert!(shadow_block_alt_recipient
+        .check_shadow_coinbase_tx(false)
+        .is_err());
+
+    // tenure tx requirements still hold for shadow blocks
+    let mut shadow_nakamoto_block = nakamoto_block.clone();
+    let mut shadow_nakamoto_block_bad_ch = nakamoto_block_bad_ch.clone();
+    let mut shadow_nakamoto_block_bad_miner_sig = nakamoto_block_bad_miner_sig.clone();
+
+    shadow_nakamoto_block.header.version |= 0x80;
+    shadow_nakamoto_block_bad_ch.header.version |= 0x80;
+    shadow_nakamoto_block_bad_miner_sig.header.version |= 0x80;
+
+    shadow_nakamoto_block
+        .header
+        .sign_miner(&private_key)
+        .unwrap();
+    shadow_nakamoto_block_bad_ch
+        .header
+        .sign_miner(&private_key)
+        .unwrap();
+    shadow_nakamoto_block_bad_miner_sig
+        .header
+        .sign_miner(&private_key)
+        .unwrap();
+
+    assert!(shadow_nakamoto_block.is_shadow_block());
+    assert!(shadow_nakamoto_block_bad_ch.is_shadow_block());
+    assert!(shadow_nakamoto_block_bad_miner_sig.is_shadow_block());
+
+    assert!(shadow_nakamoto_block.check_tenure_tx().is_ok());
+    assert!(shadow_nakamoto_block_bad_ch.check_tenure_tx().is_err());
+    assert!(shadow_nakamoto_block_bad_miner_sig
+        .check_tenure_tx()
         .is_err());
 }
 
diff --git a/stackslib/src/chainstate/nakamoto/tests/node.rs b/stackslib/src/chainstate/nakamoto/tests/node.rs
index 0645ecd15b..9a488d6a09 100644
--- a/stackslib/src/chainstate/nakamoto/tests/node.rs
+++ b/stackslib/src/chainstate/nakamoto/tests/node.rs
@@ -26,12 +26,13 @@ use hashbrown::HashMap;
 use rand::seq::SliceRandom;
 use rand::{CryptoRng, RngCore, SeedableRng};
 use rand_chacha::ChaCha20Rng;
+use rusqlite::params;
 use stacks_common::address::*;
 use stacks_common::consts::{FIRST_BURNCHAIN_CONSENSUS_HASH, FIRST_STACKS_BLOCK_HASH};
 use stacks_common::types::chainstate::{
     BlockHeaderHash, SortitionId, StacksAddress, StacksBlockId, VRFSeed,
 };
-use stacks_common::util::hash::Hash160;
+use stacks_common::util::hash::{hex_bytes, Hash160};
 use stacks_common::util::secp256k1::Secp256k1PrivateKey;
 use stacks_common::util::sleep_ms;
 use stacks_common::util::vrf::{VRFProof, VRFPublicKey};
@@ -51,11 +52,15 @@ use crate::chainstate::coordinator::{
 use crate::chainstate::nakamoto::coordinator::{
     get_nakamoto_next_recipients, load_nakamoto_reward_set,
 };
-use crate::chainstate::nakamoto::miner::NakamotoBlockBuilder;
-use crate::chainstate::nakamoto::staging_blocks::NakamotoBlockObtainMethod;
+use crate::chainstate::nakamoto::miner::{MinerTenureInfo, NakamotoBlockBuilder};
+use crate::chainstate::nakamoto::staging_blocks::{
+    NakamotoBlockObtainMethod, NakamotoStagingBlocksConnRef,
+};
 use crate::chainstate::nakamoto::test_signers::TestSigners;
 use crate::chainstate::nakamoto::tests::get_account;
-use crate::chainstate::nakamoto::{NakamotoBlock, NakamotoBlockHeader, NakamotoChainState};
+use crate::chainstate::nakamoto::{
+    NakamotoBlock, NakamotoBlockHeader, NakamotoChainState, StacksDBIndexed,
+};
 use crate::chainstate::stacks::address::PoxAddress;
 use crate::chainstate::stacks::db::blocks::test::store_staging_block;
 use crate::chainstate::stacks::db::test::*;
@@ -71,7 +76,7 @@ use crate::cost_estimates::UnitEstimator;
 use crate::net::relay::{BlockAcceptResponse, Relayer};
 use crate::net::test::{TestPeer, TestPeerConfig, *};
 use crate::util_lib::boot::boot_code_addr;
-use crate::util_lib::db::Error as db_error;
+use crate::util_lib::db::{query_row, Error as db_error};
 
 #[derive(Debug, Clone)]
 pub struct TestStacker {
@@ -182,6 +187,7 @@ impl TestBurnchainBlock {
         fork_snapshot: Option<&BlockSnapshot>,
         parent_block_snapshot: Option<&BlockSnapshot>,
         vrf_seed: VRFSeed,
+        parent_is_shadow_block: bool,
     ) -> LeaderBlockCommitOp {
         let tenure_id_as_block_hash = BlockHeaderHash(last_tenure_id.0.clone());
         self.inner_add_block_commit(
@@ -194,6 +200,7 @@ impl TestBurnchainBlock {
             parent_block_snapshot,
             Some(vrf_seed),
             STACKS_EPOCH_3_0_MARKER,
+            parent_is_shadow_block,
         )
     }
 }
@@ -221,15 +228,26 @@ impl TestMiner {
         recipient: Option<PrincipalData>,
         vrf_proof: VRFProof,
         nonce: u64,
+    ) -> StacksTransaction {
+        self.make_nakamoto_coinbase_with_nonce_and_payload(
+            recipient,
+            vrf_proof,
+            nonce,
+            CoinbasePayload([(self.nonce % 256) as u8; 32]),
+        )
+    }
+
+    pub fn make_nakamoto_coinbase_with_nonce_and_payload(
+        &mut self,
+        recipient: Option<PrincipalData>,
+        vrf_proof: VRFProof,
+        nonce: u64,
+        payload: CoinbasePayload,
     ) -> StacksTransaction {
         let mut tx_coinbase = StacksTransaction::new(
             TransactionVersion::Testnet,
             self.as_transaction_auth().unwrap(),
-            TransactionPayload::Coinbase(
-                CoinbasePayload([(self.nonce % 256) as u8; 32]),
-                recipient,
-                Some(vrf_proof),
-            ),
+            TransactionPayload::Coinbase(payload, recipient, Some(vrf_proof)),
         );
         tx_coinbase.chain_id = self.chain_id;
         tx_coinbase.anchor_mode = TransactionAnchorMode::OnChainOnly;
@@ -273,6 +291,15 @@ impl TestMiner {
     }
 }
 
+impl<'a> NakamotoStagingBlocksConnRef<'a> {
+    pub fn get_any_normal_tenure(&self) -> Result<Option<ConsensusHash>, ChainstateError> {
+        let qry = "SELECT consensus_hash FROM nakamoto_staging_blocks WHERE obtain_method != ?1 ORDER BY RANDOM() LIMIT 1";
+        let args = params![&NakamotoBlockObtainMethod::Shadow.to_string()];
+        let res: Option<ConsensusHash> = query_row(self, qry, args)?;
+        Ok(res)
+    }
+}
+
 impl TestStacksNode {
     pub fn add_nakamoto_tenure_commit(
         sortdb: &SortitionDB,
@@ -283,6 +310,7 @@ impl TestStacksNode {
         key_op: &LeaderKeyRegisterOp,
         parent_block_snapshot: Option<&BlockSnapshot>,
         vrf_seed: VRFSeed,
+        parent_is_shadow_block: bool,
     ) -> LeaderBlockCommitOp {
         let block_commit_op = {
             let ic = sortdb.index_conn();
@@ -296,6 +324,7 @@ impl TestStacksNode {
                 Some(&parent_snapshot),
                 parent_block_snapshot,
                 vrf_seed,
+                parent_is_shadow_block,
             )
         };
         block_commit_op
@@ -350,6 +379,7 @@ impl TestStacksNode {
         miner_key: &LeaderKeyRegisterOp,
         parent_block_snapshot_opt: Option<&BlockSnapshot>,
         expect_success: bool,
+        parent_is_shadow_block: bool,
     ) -> LeaderBlockCommitOp {
         info!(
             "Miner {}: Commit to Nakamoto tenure starting at {}",
@@ -385,6 +415,7 @@ impl TestStacksNode {
             miner_key,
             parent_block_snapshot_opt,
             vrf_seed,
+            parent_is_shadow_block,
         );
 
         test_debug!(
@@ -453,71 +484,125 @@ impl TestStacksNode {
     ) -> (LeaderBlockCommitOp, TenureChangePayload) {
         // this is the tenure that the block-commit confirms.
         // It's not the last-ever tenure; it's the one just before it.
-        let (last_tenure_id, parent_block_snapshot) =
-            if let Some(parent_blocks) = parent_nakamoto_tenure {
-                // parent is an epoch 3 nakamoto block
-                let first_parent = parent_blocks.first().unwrap();
-                let last_parent = parent_blocks.last().unwrap();
-                let parent_tenure_id = StacksBlockId::new(
-                    &first_parent.header.consensus_hash,
-                    &first_parent.header.block_hash(),
-                );
-                let parent_sortition = SortitionDB::get_block_snapshot_consensus(
-                    &sortdb.conn(),
-                    &first_parent.header.consensus_hash,
+        let (last_tenure_id, parent_block_snapshot, parent_is_shadow) = if let Some(parent_blocks) =
+            parent_nakamoto_tenure
+        {
+            // parent is an epoch 3 nakamoto block
+            let first_parent = parent_blocks.first().unwrap();
+            let last_parent = parent_blocks.last().unwrap();
+            let parent_tenure_id = StacksBlockId::new(
+                &first_parent.header.consensus_hash,
+                &first_parent.header.block_hash(),
+            );
+
+            let parent_sortition = if last_parent.is_shadow_block() {
+                // load up sortition that the shadow block replaces
+                SortitionDB::get_block_snapshot_consensus(
+                    sortdb.conn(),
+                    &last_parent.header.consensus_hash,
                 )
                 .unwrap()
-                .unwrap();
+                .unwrap()
+            } else {
+                // parent sortition must be the last sortition _with a winner_.
+                // This is not guaranteed with shadow blocks, so we have to search back if
+                // necessary.
+                let mut cursor = first_parent.header.consensus_hash;
+                let parent_sortition = loop {
+                    let parent_sortition =
+                        SortitionDB::get_block_snapshot_consensus(&sortdb.conn(), &cursor)
+                            .unwrap()
+                            .unwrap();
 
-                test_debug!(
-                    "Work in {} {} for Nakamoto parent: {},{}. Last tenure ID is {}",
-                    burn_block.block_height,
-                    burn_block.parent_snapshot.burn_header_hash,
-                    parent_sortition.total_burn,
-                    last_parent.header.chain_length + 1,
-                    &parent_tenure_id,
-                );
+                    if parent_sortition.sortition {
+                        break parent_sortition;
+                    }
 
-                (parent_tenure_id, parent_sortition)
-            } else if let Some(parent_stacks_block) = parent_stacks_block {
-                // building off an existing stacks block
-                let parent_stacks_block_snapshot = {
-                    let ic = sortdb.index_conn();
-                    let parent_stacks_block_snapshot =
-                        SortitionDB::get_block_snapshot_for_winning_stacks_block(
-                            &ic,
-                            &burn_block.parent_snapshot.sortition_id,
-                            &parent_stacks_block.block_hash(),
+                    // last tenure was a shadow tenure?
+                    let Ok(Some(tenure_start_header)) =
+                        NakamotoChainState::get_tenure_start_block_header(
+                            &mut self.chainstate.index_conn(),
+                            &parent_tenure_id,
+                            &cursor,
+                        )
+                    else {
+                        panic!("No tenure-start block header for tenure {}", &cursor);
+                    };
+
+                    let version = tenure_start_header
+                        .anchored_header
+                        .as_stacks_nakamoto()
+                        .unwrap()
+                        .version;
+
+                    assert!(NakamotoBlockHeader::is_shadow_block_version(version));
+                    cursor = self
+                        .chainstate
+                        .index_conn()
+                        .get_parent_tenure_consensus_hash(
+                            &tenure_start_header.index_block_hash(),
+                            &cursor,
                         )
                         .unwrap()
                         .unwrap();
-                    parent_stacks_block_snapshot
                 };
+                parent_sortition
+            };
 
-                let parent_chain_tip = StacksChainState::get_anchored_block_header_info(
-                    self.chainstate.db(),
-                    &parent_stacks_block_snapshot.consensus_hash,
-                    &parent_stacks_block.header.block_hash(),
-                )
-                .unwrap()
-                .unwrap();
-
-                let parent_tenure_id = parent_chain_tip.index_block_hash();
-
-                test_debug!(
-                    "Work in {} {} for Stacks 2.x parent: {},{}. Last tenure ID is {}",
+            test_debug!(
+                    "Work in {} {} for Nakamoto parent: {},{}. Last tenure ID is {}. Parent sortition is {}",
                     burn_block.block_height,
                     burn_block.parent_snapshot.burn_header_hash,
-                    parent_stacks_block_snapshot.total_burn,
-                    parent_chain_tip.anchored_header.height(),
+                    parent_sortition.total_burn,
+                    last_parent.header.chain_length + 1,
                     &parent_tenure_id,
+                    &parent_sortition.consensus_hash
                 );
 
-                (parent_tenure_id, parent_stacks_block_snapshot)
-            } else {
-                panic!("Neither Nakamoto nor epoch2 parent found");
+            (
+                parent_tenure_id,
+                parent_sortition,
+                last_parent.is_shadow_block(),
+            )
+        } else if let Some(parent_stacks_block) = parent_stacks_block {
+            // building off an existing stacks block
+            let parent_stacks_block_snapshot = {
+                let ic = sortdb.index_conn();
+                let parent_stacks_block_snapshot =
+                    SortitionDB::get_block_snapshot_for_winning_stacks_block(
+                        &ic,
+                        &burn_block.parent_snapshot.sortition_id,
+                        &parent_stacks_block.block_hash(),
+                    )
+                    .unwrap()
+                    .unwrap();
+                parent_stacks_block_snapshot
             };
 
+            let parent_chain_tip = StacksChainState::get_anchored_block_header_info(
+                self.chainstate.db(),
+                &parent_stacks_block_snapshot.consensus_hash,
+                &parent_stacks_block.header.block_hash(),
+            )
+            .unwrap()
+            .unwrap();
+
+            let parent_tenure_id = parent_chain_tip.index_block_hash();
+
+            test_debug!(
+                "Work in {} {} for Stacks 2.x parent: {},{}. Last tenure ID is {}",
+                burn_block.block_height,
+                burn_block.parent_snapshot.burn_header_hash,
+                parent_stacks_block_snapshot.total_burn,
+                parent_chain_tip.anchored_header.height(),
+                &parent_tenure_id,
+            );
+
+            (parent_tenure_id, parent_stacks_block_snapshot, false)
+        } else {
+            panic!("Neither Nakamoto nor epoch2 parent found");
+        };
+
         // the tenure-change contains a pointer to the end of the last tenure, which is currently
         // the canonical tip unless overridden
         let (previous_tenure_end, previous_tenure_consensus_hash, previous_tenure_blocks) =
@@ -551,7 +636,9 @@ impl TestStacksNode {
                     );
                     (hdr.index_block_hash(), hdr.consensus_hash, tenure_len)
                 } else {
-                    // building atop epoch2
+                    // building atop epoch2 (so the parent block can't be a shadow block, meaning
+                    // that parent_block_snapshot is _guaranteed_ to be the snapshot that chose
+                    // last_tenure_id).
                     debug!(
                         "Tenure length of epoch2 tenure {} is {}; tipped at {}",
                         &parent_block_snapshot.consensus_hash, 1, &last_tenure_id
@@ -585,6 +672,7 @@ impl TestStacksNode {
             miner_key,
             Some(&parent_block_snapshot),
             tenure_change_cause == TenureChangeCause::BlockFound,
+            parent_is_shadow,
         );
 
         (block_commit_op, tenure_change_payload)
@@ -599,6 +687,10 @@ impl TestStacksNode {
     /// The first block will contain a coinbase, if coinbase is Some(..)
     /// Process the blocks via the chains coordinator as we produce them.
     ///
+    /// If malleablize is true, then malleablized blocks will be created by varying the number of
+    /// signatures. Each malleablized block will be processed and stored if its signatures clear
+    /// the signing threshold.
+    ///
     /// Returns a list of
     ///     * the block
     ///     * its size
@@ -626,7 +718,7 @@ impl TestStacksNode {
         mut after_block: G,
         malleablize: bool,
         mined_canonical: bool,
-    ) -> Vec<(NakamotoBlock, u64, ExecutionCost, Vec<NakamotoBlock>)>
+    ) -> Result<Vec<(NakamotoBlock, u64, ExecutionCost, Vec<NakamotoBlock>)>, ChainstateError>
     where
         S: FnMut(&mut NakamotoBlockBuilder),
         F: FnMut(
@@ -665,7 +757,7 @@ impl TestStacksNode {
 
             let parent_tip_opt = if let Some(parent_id) = parent_id_opt {
                 if let Some(nakamoto_parent) =
-                    NakamotoChainState::get_block_header(chainstate.db(), &parent_id).unwrap()
+                    NakamotoChainState::get_block_header(chainstate.db(), &parent_id)?
                 {
                     debug!(
                         "Use parent tip identified by produced TenureChange ({})",
@@ -674,8 +766,7 @@ impl TestStacksNode {
                     Some(nakamoto_parent)
                 } else {
                     warn!("Produced Tenure change transaction does not point to a real block");
-                    NakamotoChainState::get_canonical_block_header(chainstate.db(), &sortdb)
-                        .unwrap()
+                    NakamotoChainState::get_canonical_block_header(chainstate.db(), &sortdb)?
                 }
             } else if let Some(tenure_change) = tenure_change.as_ref() {
                 // make sure parent tip is consistent with a tenure change
@@ -683,9 +774,7 @@ impl TestStacksNode {
                     if let Some(nakamoto_parent) = NakamotoChainState::get_block_header(
                         chainstate.db(),
                         &payload.previous_tenure_end,
-                    )
-                    .unwrap()
-                    {
+                    )? {
                         debug!(
                             "Use parent tip identified by given TenureChange ({})",
                             &payload.previous_tenure_end
@@ -693,17 +782,16 @@ impl TestStacksNode {
                         Some(nakamoto_parent)
                     } else {
                         debug!("Use parent tip identified by canonical tip pointer (no parent block {})", &payload.previous_tenure_end);
-                        NakamotoChainState::get_canonical_block_header(chainstate.db(), &sortdb)
-                            .unwrap()
+                        NakamotoChainState::get_canonical_block_header(chainstate.db(), &sortdb)?
                     }
                 } else {
                     panic!("Tenure change transaction does not have a TenureChange payload");
                 }
             } else {
-                NakamotoChainState::get_canonical_block_header(chainstate.db(), &sortdb).unwrap()
+                NakamotoChainState::get_canonical_block_header(chainstate.db(), &sortdb)?
             };
 
-            let burn_tip = SortitionDB::get_canonical_burn_chain_tip(sortdb.conn()).unwrap();
+            let burn_tip = SortitionDB::get_canonical_burn_chain_tip(sortdb.conn())?;
 
             debug!(
                 "Build Nakamoto block in tenure {} sortition {} parent_tip {:?}",
@@ -730,8 +818,7 @@ impl TestStacksNode {
                     },
                     1,
                     None,
-                )
-                .unwrap()
+                )?
             } else {
                 NakamotoBlockBuilder::new_first_block(
                     &tenure_change.clone().unwrap(),
@@ -748,22 +835,21 @@ impl TestStacksNode {
                 chainstate,
                 &sortdb.index_handle_at_tip(),
                 txs,
-            )
-            .unwrap();
+            )?;
             let try_to_process = after_block(&mut nakamoto_block);
             miner.sign_nakamoto_block(&mut nakamoto_block);
 
             let tenure_sn =
-                SortitionDB::get_block_snapshot_consensus(sortdb.conn(), tenure_id_consensus_hash)
-                    .unwrap()
-                    .unwrap();
+                SortitionDB::get_block_snapshot_consensus(sortdb.conn(), tenure_id_consensus_hash)?
+                    .ok_or_else(|| ChainstateError::NoSuchBlockError)?;
+
             let cycle = sortdb
                 .pox_constants
                 .block_height_to_reward_cycle(sortdb.first_block_height, tenure_sn.block_height)
                 .unwrap();
 
             // Get the reward set
-            let sort_tip_sn = SortitionDB::get_canonical_burn_chain_tip(sortdb.conn()).unwrap();
+            let sort_tip_sn = SortitionDB::get_canonical_burn_chain_tip(sortdb.conn())?;
             let reward_set = load_nakamoto_reward_set(
                 miner
                     .burnchain
@@ -804,9 +890,11 @@ impl TestStacksNode {
                 &block_id, &nakamoto_block.txs
             );
 
-            let sort_tip = SortitionDB::get_canonical_sortition_tip(sortdb.conn()).unwrap();
+            let sort_tip = SortitionDB::get_canonical_sortition_tip(sortdb.conn())?;
             let mut sort_handle = sortdb.index_handle(&sort_tip);
-            let stacks_tip = sort_handle.get_nakamoto_tip_block_id().unwrap().unwrap();
+            let stacks_tip = sort_handle
+                .get_nakamoto_tip_block_id()?
+                .ok_or_else(|| ChainstateError::NoSuchBlockError)?;
 
             let mut block_to_store = nakamoto_block.clone();
             let mut processed_blocks = vec![];
@@ -865,9 +953,8 @@ impl TestStacksNode {
                         let stacks_chain_tip = NakamotoChainState::get_canonical_block_header(
                             chainstate.db(),
                             &sortdb,
-                        )
-                        .unwrap()
-                        .unwrap();
+                        )?
+                        .ok_or_else(|| ChainstateError::NoSuchBlockError)?;
                         let nakamoto_chain_tip = stacks_chain_tip
                             .anchored_header
                             .as_stacks_nakamoto()
@@ -918,11 +1005,11 @@ impl TestStacksNode {
             all_malleablized_blocks.push(malleablized_blocks);
             block_count += 1;
         }
-        blocks
+        Ok(blocks
             .into_iter()
             .zip(all_malleablized_blocks.into_iter())
             .map(|((blk, sz, cost), mals)| (blk, sz, cost, mals))
-            .collect()
+            .collect())
     }
 
     pub fn make_nakamoto_block_from_txs(
@@ -1107,33 +1194,74 @@ impl<'a> TestPeer<'a> {
 
         // find the VRF leader key register tx to use.
         // it's the one pointed to by the parent tenure
-        let parent_consensus_hash_opt = if let Some(parent_tenure) = parent_tenure_opt.as_ref() {
-            let tenure_start_block = parent_tenure.first().unwrap();
-            Some(tenure_start_block.header.consensus_hash)
-        } else if let Some(parent_block) = parent_block_opt.as_ref() {
-            let parent_header_info =
-                StacksChainState::get_stacks_block_header_info_by_index_block_hash(
-                    stacks_node.chainstate.db(),
-                    &last_tenure_id,
+        let parent_consensus_hash_and_tenure_start_id_opt =
+            if let Some(parent_tenure) = parent_tenure_opt.as_ref() {
+                let tenure_start_block = parent_tenure.first().unwrap();
+                Some((
+                    tenure_start_block.header.consensus_hash,
+                    tenure_start_block.block_id(),
+                ))
+            } else if let Some(parent_block) = parent_block_opt.as_ref() {
+                let parent_header_info =
+                    StacksChainState::get_stacks_block_header_info_by_index_block_hash(
+                        stacks_node.chainstate.db(),
+                        &last_tenure_id,
+                    )
+                    .unwrap()
+                    .unwrap();
+                Some((
+                    parent_header_info.consensus_hash,
+                    parent_header_info.index_block_hash(),
+                ))
+            } else {
+                None
+            };
+
+        let last_key = if let Some((ch, parent_tenure_start_block_id)) =
+            parent_consensus_hash_and_tenure_start_id_opt.clone()
+        {
+            // it's possible that the parent was a shadow block.
+            // if so, find the highest non-shadow ancestor's block-commit, so we can
+            let mut cursor = ch;
+            let (tenure_sn, tenure_block_commit) = loop {
+                let tenure_sn = SortitionDB::get_block_snapshot_consensus(sortdb.conn(), &cursor)
+                    .unwrap()
+                    .unwrap();
+
+                let Some(tenure_block_commit) = get_block_commit_by_txid(
+                    sortdb.conn(),
+                    &tenure_sn.sortition_id,
+                    &tenure_sn.winning_block_txid,
                 )
-                .unwrap()
-                .unwrap();
-            Some(parent_header_info.consensus_hash)
-        } else {
-            None
-        };
+                .unwrap() else {
+                    // parent must be a shadow block
+                    let header = NakamotoChainState::get_block_header_nakamoto(
+                        stacks_node.chainstate.db(),
+                        &parent_tenure_start_block_id,
+                    )
+                    .unwrap()
+                    .unwrap()
+                    .anchored_header
+                    .as_stacks_nakamoto()
+                    .cloned()
+                    .unwrap();
+
+                    if !header.is_shadow_block() {
+                        panic!("Parent tenure start block ID {} has no block-commit and is not a shadow block", &parent_tenure_start_block_id);
+                    }
+
+                    cursor = stacks_node
+                        .chainstate
+                        .index_conn()
+                        .get_parent_tenure_consensus_hash(&parent_tenure_start_block_id, &cursor)
+                        .unwrap()
+                        .unwrap();
+
+                    continue;
+                };
+                break (tenure_sn, tenure_block_commit);
+            };
 
-        let last_key = if let Some(ch) = parent_consensus_hash_opt.clone() {
-            let tenure_sn = SortitionDB::get_block_snapshot_consensus(sortdb.conn(), &ch)
-                .unwrap()
-                .unwrap();
-            let tenure_block_commit = get_block_commit_by_txid(
-                sortdb.conn(),
-                &tenure_sn.sortition_id,
-                &tenure_sn.winning_block_txid,
-            )
-            .unwrap()
-            .unwrap();
             let tenure_leader_key = SortitionDB::get_leader_key_at(
                 &sortdb.index_conn(),
                 tenure_block_commit.key_block_ptr.into(),
@@ -1318,6 +1446,7 @@ impl<'a> TestPeer<'a> {
             block_builder,
             |_| true,
         )
+        .unwrap()
     }
 
     /// Produce and process a Nakamoto tenure, after processing the block-commit from
@@ -1333,7 +1462,7 @@ impl<'a> TestPeer<'a> {
         miner_setup: S,
         block_builder: F,
         after_block: G,
-    ) -> Vec<(NakamotoBlock, u64, ExecutionCost)>
+    ) -> Result<Vec<(NakamotoBlock, u64, ExecutionCost)>, ChainstateError>
     where
         S: FnMut(&mut NakamotoBlockBuilder),
         F: FnMut(
@@ -1345,59 +1474,55 @@ impl<'a> TestPeer<'a> {
         G: FnMut(&mut NakamotoBlock) -> bool,
     {
         let cycle = self.get_reward_cycle();
-        let mut stacks_node = self.stacks_node.take().unwrap();
-        let mut sortdb = self.sortdb.take().unwrap();
+        self.with_dbs(|peer, sortdb, stacks_node, mempool| {
+            // Ensure the signers are setup for the current cycle
+            signers.generate_aggregate_key(cycle);
 
-        // Ensure the signers are setup for the current cycle
-        signers.generate_aggregate_key(cycle);
-
-        let blocks = TestStacksNode::make_nakamoto_tenure_blocks(
-            &mut stacks_node.chainstate,
-            &mut sortdb,
-            &mut self.miner,
-            signers,
-            &tenure_change
-                .try_as_tenure_change()
-                .unwrap()
-                .tenure_consensus_hash
-                .clone(),
-            Some(tenure_change),
-            Some(coinbase),
-            &mut self.coord,
-            miner_setup,
-            block_builder,
-            after_block,
-            self.mine_malleablized_blocks,
-            self.nakamoto_parent_tenure_opt.is_none(),
-        );
-
-        let just_blocks = blocks
-            .clone()
-            .into_iter()
-            .map(|(block, _, _, _)| block)
-            .collect();
+            let blocks = TestStacksNode::make_nakamoto_tenure_blocks(
+                &mut stacks_node.chainstate,
+                sortdb,
+                &mut peer.miner,
+                signers,
+                &tenure_change
+                    .try_as_tenure_change()
+                    .unwrap()
+                    .tenure_consensus_hash
+                    .clone(),
+                Some(tenure_change),
+                Some(coinbase),
+                &mut peer.coord,
+                miner_setup,
+                block_builder,
+                after_block,
+                peer.mine_malleablized_blocks,
+                peer.nakamoto_parent_tenure_opt.is_none(),
+            )?;
+
+            let just_blocks = blocks
+                .clone()
+                .into_iter()
+                .map(|(block, _, _, _)| block)
+                .collect();
 
-        stacks_node.add_nakamoto_tenure_blocks(just_blocks);
+            stacks_node.add_nakamoto_tenure_blocks(just_blocks);
 
-        let mut malleablized_blocks: Vec<NakamotoBlock> = blocks
-            .clone()
-            .into_iter()
-            .map(|(_, _, _, malleablized)| malleablized)
-            .flatten()
-            .collect();
+            let mut malleablized_blocks: Vec<NakamotoBlock> = blocks
+                .clone()
+                .into_iter()
+                .map(|(_, _, _, malleablized)| malleablized)
+                .flatten()
+                .collect();
 
-        self.malleablized_blocks.append(&mut malleablized_blocks);
+            peer.malleablized_blocks.append(&mut malleablized_blocks);
 
-        let block_data = blocks
-            .clone()
-            .into_iter()
-            .map(|(blk, sz, cost, _)| (blk, sz, cost))
-            .collect();
-
-        self.stacks_node = Some(stacks_node);
-        self.sortdb = Some(sortdb);
+            let block_data = blocks
+                .clone()
+                .into_iter()
+                .map(|(blk, sz, cost, _)| (blk, sz, cost))
+                .collect();
 
-        block_data
+            Ok(block_data)
+        })
     }
 
     /// Produce and process a Nakamoto tenure extension.
@@ -1461,7 +1586,8 @@ impl<'a> TestPeer<'a> {
             |_| true,
             self.mine_malleablized_blocks,
             self.nakamoto_parent_tenure_opt.is_none(),
-        );
+        )
+        .unwrap();
 
         let just_blocks = blocks
             .clone()
@@ -1833,7 +1959,7 @@ impl<'a> TestPeer<'a> {
         );
         let parent_vrf_proof = NakamotoChainState::get_parent_vrf_proof(
             &mut chainstate.index_conn(),
-            &block.block_id(),
+            &block.header.parent_block_id,
             &sortdb.conn(),
             &block.header.consensus_hash,
             &tenure_block_commit.txid,
@@ -2197,5 +2323,287 @@ impl<'a> TestPeer<'a> {
             )
             .unwrap());
         }
+
+        // validate_shadow_parent_burnchain
+        // should always succeed
+        NakamotoChainState::validate_shadow_parent_burnchain(
+            chainstate.nakamoto_blocks_db(),
+            &sortdb.index_handle_at_tip(),
+            block,
+            &tenure_block_commit,
+        )
+        .unwrap();
+
+        if parent_block_header
+            .anchored_header
+            .as_stacks_nakamoto()
+            .map(|hdr| hdr.is_shadow_block())
+            .unwrap_or(false)
+        {
+            // test error cases
+            let mut bad_tenure_block_commit_vtxindex = tenure_block_commit.clone();
+            bad_tenure_block_commit_vtxindex.parent_vtxindex = 1;
+
+            let mut bad_tenure_block_commit_block_ptr = tenure_block_commit.clone();
+            bad_tenure_block_commit_block_ptr.parent_block_ptr += 1;
+
+            let mut bad_block_no_parent = block.clone();
+            bad_block_no_parent.header.parent_block_id = StacksBlockId([0x11; 32]);
+
+            // not a problem if there's no (nakamoto) parent, since the parent can be a
+            // (non-shadow) epoch2 block not present in the staging chainstate
+            NakamotoChainState::validate_shadow_parent_burnchain(
+                chainstate.nakamoto_blocks_db(),
+                &sortdb.index_handle_at_tip(),
+                &bad_block_no_parent,
+                &tenure_block_commit,
+            )
+            .unwrap();
+
+            // should fail because vtxindex must be 0
+            let ChainstateError::InvalidStacksBlock(_) =
+                NakamotoChainState::validate_shadow_parent_burnchain(
+                    chainstate.nakamoto_blocks_db(),
+                    &sortdb.index_handle_at_tip(),
+                    block,
+                    &bad_tenure_block_commit_vtxindex,
+                )
+                .unwrap_err()
+            else {
+                panic!("validate_shadow_parent_burnchain did not fail as expected");
+            };
+
+            // should fail because it doesn't point to shadow tenure
+            let ChainstateError::InvalidStacksBlock(_) =
+                NakamotoChainState::validate_shadow_parent_burnchain(
+                    chainstate.nakamoto_blocks_db(),
+                    &sortdb.index_handle_at_tip(),
+                    block,
+                    &bad_tenure_block_commit_block_ptr,
+                )
+                .unwrap_err()
+            else {
+                panic!("validate_shadow_parent_burnchain did not fail as expected");
+            };
+        }
+
+        if block.is_shadow_block() {
+            // block is stored
+            assert!(chainstate
+                .nakamoto_blocks_db()
+                .has_shadow_nakamoto_block_with_index_hash(&block.block_id())
+                .unwrap());
+
+            // block is in a shadow tenure
+            assert!(chainstate
+                .nakamoto_blocks_db()
+                .is_shadow_tenure(&block.header.consensus_hash)
+                .unwrap());
+
+            // shadow tenure has a start block
+            assert!(chainstate
+                .nakamoto_blocks_db()
+                .get_shadow_tenure_start_block(&block.header.consensus_hash)
+                .unwrap()
+                .is_some());
+
+            // succeeds without burn
+            NakamotoChainState::validate_shadow_nakamoto_block_burnchain(
+                chainstate.nakamoto_blocks_db(),
+                &sortdb.index_handle_at_tip(),
+                None,
+                &block,
+                false,
+                0x80000000,
+            )
+            .unwrap();
+
+            // succeeds with expected burn
+            NakamotoChainState::validate_shadow_nakamoto_block_burnchain(
+                chainstate.nakamoto_blocks_db(),
+                &sortdb.index_handle_at_tip(),
+                Some(block.header.burn_spent),
+                &block,
+                false,
+                0x80000000,
+            )
+            .unwrap();
+
+            // fails with invalid burn
+            let ChainstateError::InvalidStacksBlock(_) =
+                NakamotoChainState::validate_shadow_nakamoto_block_burnchain(
+                    chainstate.nakamoto_blocks_db(),
+                    &sortdb.index_handle_at_tip(),
+                    Some(block.header.burn_spent + 1),
+                    &block,
+                    false,
+                    0x80000000,
+                )
+                .unwrap_err()
+            else {
+                panic!("validate_shadow_nakamoto_block_burnchain succeeded when it shouldn't have");
+            };
+
+            // block must be stored alreay
+            let mut bad_block = block.clone();
+            bad_block.header.version += 1;
+
+            // fails because block_id() isn't present
+            let ChainstateError::InvalidStacksBlock(_) =
+                NakamotoChainState::validate_shadow_nakamoto_block_burnchain(
+                    chainstate.nakamoto_blocks_db(),
+                    &sortdb.index_handle_at_tip(),
+                    None,
+                    &bad_block,
+                    false,
+                    0x80000000,
+                )
+                .unwrap_err()
+            else {
+                panic!("validate_shadow_nakamoto_block_burnchain succeeded when it shouldn't have");
+            };
+
+            // VRF proof must be present
+            assert!(NakamotoChainState::get_shadow_vrf_proof(
+                &mut chainstate.index_conn(),
+                &block.block_id()
+            )
+            .unwrap()
+            .is_some());
+        } else {
+            // not a shadow block
+            assert!(!chainstate
+                .nakamoto_blocks_db()
+                .has_shadow_nakamoto_block_with_index_hash(&block.block_id())
+                .unwrap());
+            assert!(!chainstate
+                .nakamoto_blocks_db()
+                .is_shadow_tenure(&block.header.consensus_hash)
+                .unwrap());
+            assert!(chainstate
+                .nakamoto_blocks_db()
+                .get_shadow_tenure_start_block(&block.header.consensus_hash)
+                .unwrap()
+                .is_none());
+            assert!(NakamotoChainState::get_shadow_vrf_proof(
+                &mut chainstate.index_conn(),
+                &block.block_id()
+            )
+            .unwrap()
+            .is_none());
+        }
+    }
+
+    /// Add a shadow tenure on a given tip.
+    /// * Advance the burnchain and create an empty sortition (so we have a new consensus hash)
+    /// * Generate a shadow block for the empty sortition
+    /// * Store the shadow block to the staging DB
+    /// * Process it
+    ///
+    /// Tests:
+    /// * NakamotoBlockHeader::get_shadow_signer_weight()
+    pub fn make_shadow_tenure(&mut self, tip: Option<StacksBlockId>) -> NakamotoBlock {
+        let naka_tip_id = tip.unwrap_or(self.network.stacks_tip.block_id());
+        let (_, _, tenure_id_consensus_hash) = self.next_burnchain_block(vec![]);
+
+        test_debug!(
+            "\n\nMake shadow tenure for tenure {} off of tip {}\n\n",
+            &tenure_id_consensus_hash,
+            &naka_tip_id
+        );
+
+        let mut stacks_node = self.stacks_node.take().unwrap();
+        let sortdb = self.sortdb.take().unwrap();
+
+        let shadow_block = NakamotoBlockBuilder::make_shadow_tenure(
+            &mut stacks_node.chainstate,
+            &sortdb,
+            naka_tip_id,
+            tenure_id_consensus_hash,
+            vec![],
+        )
+        .unwrap();
+
+        // Get the reward set
+        let sort_tip_sn = SortitionDB::get_canonical_burn_chain_tip(sortdb.conn()).unwrap();
+        let reward_set = load_nakamoto_reward_set(
+            self.miner
+                .burnchain
+                .block_height_to_reward_cycle(sort_tip_sn.block_height)
+                .expect("FATAL: no reward cycle for sortition"),
+            &sort_tip_sn.sortition_id,
+            &self.miner.burnchain,
+            &mut stacks_node.chainstate,
+            &shadow_block.header.parent_block_id,
+            &sortdb,
+            &OnChainRewardSetProvider::new(),
+        )
+        .expect("Failed to load reward set")
+        .expect("Expected a reward set")
+        .0
+        .known_selected_anchor_block_owned()
+        .expect("Unknown reward set");
+
+        // check signer weight
+        let mut max_signing_weight = 0;
+        for signer in reward_set.signers.as_ref().unwrap().iter() {
+            max_signing_weight += signer.weight;
+        }
+
+        assert_eq!(
+            shadow_block
+                .header
+                .get_shadow_signer_weight(&reward_set)
+                .unwrap(),
+            max_signing_weight
+        );
+
+        // put it into Stacks staging DB
+        let tx = stacks_node.chainstate.staging_db_tx_begin().unwrap();
+        tx.add_shadow_block(&shadow_block).unwrap();
+
+        // inserts of the same block are idempotent
+        tx.add_shadow_block(&shadow_block).unwrap();
+
+        tx.commit().unwrap();
+
+        let rollback_tx = stacks_node.chainstate.staging_db_tx_begin().unwrap();
+
+        if let Some(normal_tenure) = rollback_tx.conn().get_any_normal_tenure().unwrap() {
+            // can't insert into a non-shadow tenure
+            let mut bad_shadow_block_tenure = shadow_block.clone();
+            bad_shadow_block_tenure.header.consensus_hash = normal_tenure;
+
+            let ChainstateError::InvalidStacksBlock(_) = rollback_tx
+                .add_shadow_block(&bad_shadow_block_tenure)
+                .unwrap_err()
+            else {
+                panic!("add_shadow_block succeeded when it should have failed");
+            };
+        }
+
+        // can't insert into the same height twice with different blocks
+        let mut bad_shadow_block_height = shadow_block.clone();
+        bad_shadow_block_height.header.version += 1;
+        let ChainstateError::InvalidStacksBlock(_) = rollback_tx
+            .add_shadow_block(&bad_shadow_block_height)
+            .unwrap_err()
+        else {
+            panic!("add_shadow_block succeeded when it should have failed");
+        };
+
+        drop(rollback_tx);
+
+        self.stacks_node = Some(stacks_node);
+        self.sortdb = Some(sortdb);
+
+        // process it
+        self.coord.handle_new_nakamoto_stacks_block().unwrap();
+
+        // verify that it processed
+        self.refresh_burnchain_view();
+        assert_eq!(self.network.stacks_tip.block_id(), shadow_block.block_id());
+
+        shadow_block
     }
 }
diff --git a/stackslib/src/net/api/tests/getsigner.rs b/stackslib/src/net/api/tests/getsigner.rs
index ffaa486f27..a3b112d0e3 100644
--- a/stackslib/src/net/api/tests/getsigner.rs
+++ b/stackslib/src/net/api/tests/getsigner.rs
@@ -139,7 +139,7 @@ fn test_try_make_response() {
     let response = responses.remove(0);
     info!("response: {:?}", &response);
     let signer_response = response.decode_signer().unwrap();
-    assert_eq!(signer_response.blocks_signed, 40);
+    assert_eq!(signer_response.blocks_signed, 20);
 
     // Signer doesn't exist so it should not have signed anything
     let response = responses.remove(0);
diff --git a/stackslib/src/net/api/tests/mod.rs b/stackslib/src/net/api/tests/mod.rs
index 40d329686d..0a6ad69762 100644
--- a/stackslib/src/net/api/tests/mod.rs
+++ b/stackslib/src/net/api/tests/mod.rs
@@ -47,7 +47,7 @@ use crate::net::httpcore::{StacksHttpRequest, StacksHttpResponse};
 use crate::net::relay::Relayer;
 use crate::net::rpc::ConversationHttp;
 use crate::net::test::{TestEventObserver, TestPeer, TestPeerConfig};
-use crate::net::tests::inv::nakamoto::make_nakamoto_peers_from_invs;
+use crate::net::tests::inv::nakamoto::make_nakamoto_peers_from_invs_ext;
 use crate::net::{
     Attachment, AttachmentInstance, MemPoolEventDispatcher, RPCHandlerArgs, StackerDBConfig,
     StacksNodeState, UrlString,
@@ -849,8 +849,18 @@ impl<'a> TestRPC<'a> {
             true, true, true, true, true, true, true, true, true, true,
         ]];
 
-        let (mut peer, mut other_peers) =
-            make_nakamoto_peers_from_invs(function_name!(), observer, 10, 3, bitvecs.clone(), 1);
+        let (mut peer, mut other_peers) = make_nakamoto_peers_from_invs_ext(
+            function_name!(),
+            observer,
+            bitvecs.clone(),
+            |boot_plan| {
+                boot_plan
+                    .with_pox_constants(10, 3)
+                    .with_extra_peers(1)
+                    .with_initial_balances(vec![])
+                    .with_malleablized_blocks(false)
+            },
+        );
         let mut other_peer = other_peers.pop().unwrap();
 
         let peer_1_indexer = BitcoinIndexer::new_unit_test(&peer.config.burnchain.working_dir);
diff --git a/stackslib/src/net/api/tests/postblock_proposal.rs b/stackslib/src/net/api/tests/postblock_proposal.rs
index a8087bf36a..c742bcf00b 100644
--- a/stackslib/src/net/api/tests/postblock_proposal.rs
+++ b/stackslib/src/net/api/tests/postblock_proposal.rs
@@ -378,12 +378,14 @@ fn test_try_make_response() {
     let observer = ProposalTestObserver::new();
     let proposal_observer = Arc::clone(&observer.proposal_observer);
 
+    info!("Run requests with observer");
     let mut responses = rpc_test.run_with_observer(requests, Some(&observer));
 
     let response = responses.remove(0);
 
     // Wait for the results to be non-empty
     loop {
+        info!("Wait for results to be non-empty");
         if proposal_observer
             .lock()
             .unwrap()
diff --git a/stackslib/src/net/tests/download/nakamoto.rs b/stackslib/src/net/tests/download/nakamoto.rs
index e2bea6fd50..a479dad07a 100644
--- a/stackslib/src/net/tests/download/nakamoto.rs
+++ b/stackslib/src/net/tests/download/nakamoto.rs
@@ -33,18 +33,23 @@ use crate::burnchains::PoxConstants;
 use crate::chainstate::burn::db::sortdb::SortitionHandle;
 use crate::chainstate::burn::BlockSnapshot;
 use crate::chainstate::nakamoto::test_signers::TestSigners;
-use crate::chainstate::nakamoto::{NakamotoBlock, NakamotoBlockHeader, NakamotoChainState};
+use crate::chainstate::nakamoto::{
+    NakamotoBlock, NakamotoBlockHeader, NakamotoChainState, NakamotoStagingBlocksConnRef,
+};
 use crate::chainstate::stacks::db::{StacksChainState, StacksHeaderInfo};
 use crate::chainstate::stacks::{
-    CoinbasePayload, StacksTransaction, TenureChangeCause, TenureChangePayload, TokenTransferMemo,
-    TransactionAnchorMode, TransactionAuth, TransactionPayload, TransactionVersion,
+    CoinbasePayload, Error as ChainstateError, StacksTransaction, TenureChangeCause,
+    TenureChangePayload, TokenTransferMemo, TransactionAnchorMode, TransactionAuth,
+    TransactionPayload, TransactionVersion,
 };
 use crate::clarity::vm::types::StacksAddressExtensions;
 use crate::net::api::gettenureinfo::RPCGetTenureInfo;
 use crate::net::download::nakamoto::{TenureStartEnd, WantedTenure, *};
 use crate::net::inv::nakamoto::NakamotoTenureInv;
-use crate::net::test::{dns_thread_start, TestEventObserver};
-use crate::net::tests::inv::nakamoto::{make_nakamoto_peer_from_invs, peer_get_nakamoto_invs};
+use crate::net::test::{dns_thread_start, to_addr, TestEventObserver};
+use crate::net::tests::inv::nakamoto::{
+    make_nakamoto_peer_from_invs, make_nakamoto_peers_from_invs_ext, peer_get_nakamoto_invs,
+};
 use crate::net::tests::{NakamotoBootPlan, TestPeer};
 use crate::net::{Error as NetError, Hash160, NeighborAddress, SortitionDB};
 use crate::stacks_common::types::Address;
@@ -97,6 +102,45 @@ impl NakamotoDownloadStateMachine {
     }
 }
 
+impl<'a> NakamotoStagingBlocksConnRef<'a> {
+    pub fn load_nakamoto_tenure(
+        &self,
+        tip: &StacksBlockId,
+    ) -> Result<Option<Vec<NakamotoBlock>>, ChainstateError> {
+        let Some((block, ..)) = self.get_nakamoto_block(tip)? else {
+            return Ok(None);
+        };
+        if block.is_wellformed_tenure_start_block().map_err(|_| {
+            ChainstateError::InvalidStacksBlock("Malformed tenure-start block".into())
+        })? {
+            // we're done
+            return Ok(Some(vec![block]));
+        }
+
+        // this is an intermediate block
+        let mut tenure = vec![];
+        let mut cursor = block.header.parent_block_id.clone();
+        tenure.push(block);
+        loop {
+            let Some((block, _)) = self.get_nakamoto_block(&cursor)? else {
+                return Ok(None);
+            };
+
+            let is_tenure_start = block.is_wellformed_tenure_start_block().map_err(|e| {
+                ChainstateError::InvalidStacksBlock("Malformed tenure-start block".into())
+            })?;
+            cursor = block.header.parent_block_id.clone();
+            tenure.push(block);
+
+            if is_tenure_start {
+                break;
+            }
+        }
+        tenure.reverse();
+        Ok(Some(tenure))
+    }
+}
+
 #[test]
 fn test_nakamoto_tenure_downloader() {
     let ch = ConsensusHash([0x11; 20]);
@@ -240,8 +284,10 @@ fn test_nakamoto_tenure_downloader() {
     };
 
     let mut td = NakamotoTenureDownloader::new(
+        tenure_start_block.header.consensus_hash.clone(),
         tenure_start_block.header.consensus_hash.clone(),
         tenure_start_block.header.block_id(),
+        next_tenure_start_block.header.consensus_hash.clone(),
         next_tenure_start_block.header.block_id(),
         naddr.clone(),
         reward_set.clone(),
@@ -361,6 +407,7 @@ fn test_nakamoto_unconfirmed_tenure_downloader() {
     );
     let (mut peer, reward_cycle_invs) =
         peer_get_nakamoto_invs(peer, &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]);
+    peer.mine_malleablized_blocks = false;
 
     let nakamoto_start =
         NakamotoBootPlan::nakamoto_first_tenure_height(&peer.config.burnchain.pox_constants);
@@ -2161,7 +2208,9 @@ fn test_nakamoto_download_run_2_peers() {
                     "Booting peer's stacks tip is now {:?}",
                     &boot_peer.network.stacks_tip
                 );
-                if stacks_tip_ch == canonical_stacks_tip_ch {
+                if stacks_tip_ch == canonical_stacks_tip_ch
+                    && stacks_tip_bhh == canonical_stacks_tip_bhh
+                {
                     break;
                 }
             }
@@ -2249,6 +2298,793 @@ fn test_nakamoto_unconfirmed_download_run_2_peers() {
 
     let (mut boot_dns_client, boot_dns_thread_handle) = dns_thread_start(100);
 
+    // start running that peer so we can boot off of it
+    let (term_sx, term_rx) = sync_channel(1);
+    thread::scope(|s| {
+        s.spawn(move || {
+            let (mut last_stacks_tip_ch, mut last_stacks_tip_bhh) =
+                SortitionDB::get_canonical_stacks_chain_tip_hash(boot_peer.sortdb().conn())
+                    .unwrap();
+            loop {
+                boot_peer
+                    .run_with_ibd(true, Some(&mut boot_dns_client))
+                    .unwrap();
+
+                let (stacks_tip_ch, stacks_tip_bhh) =
+                    SortitionDB::get_canonical_stacks_chain_tip_hash(boot_peer.sortdb().conn())
+                        .unwrap();
+
+                last_stacks_tip_ch = stacks_tip_ch;
+                last_stacks_tip_bhh = stacks_tip_bhh;
+
+                debug!(
+                    "Booting peer's stacks tip is now {:?}",
+                    &boot_peer.network.stacks_tip
+                );
+                if stacks_tip_ch == canonical_stacks_tip_ch
+                    && stacks_tip_bhh == canonical_stacks_tip_bhh
+                {
+                    break;
+                }
+            }
+
+            term_sx.send(()).unwrap();
+        });
+
+        loop {
+            if term_rx.try_recv().is_ok() {
+                break;
+            }
+            peer.step_with_ibd(false).unwrap();
+        }
+    });
+
+    boot_dns_thread_handle.join().unwrap();
+}
+
+/// Test the case where one or more blocks from tenure _T_ get orphend by a tenure-start block in
+/// tenure _T + 1_.  The unconfirmed downloader should be able to handle this case.
+#[test]
+fn test_nakamoto_microfork_download_run_2_peers() {
+    let sender_key = StacksPrivateKey::new();
+    let sender_addr = to_addr(&sender_key);
+    let initial_balances = vec![(sender_addr.to_account_principal(), 1000000000)];
+
+    let observer = TestEventObserver::new();
+    let bitvecs = vec![
+        // full rc
+        vec![true, true, true, true, true, true, true, true, true, true],
+    ];
+
+    let rc_len = 10u64;
+
+    let (mut peer, _) = make_nakamoto_peers_from_invs_ext(
+        function_name!(),
+        &observer,
+        bitvecs.clone(),
+        |boot_plan| {
+            boot_plan
+                .with_pox_constants(rc_len as u32, 5)
+                .with_extra_peers(0)
+                .with_initial_balances(initial_balances)
+                .with_malleablized_blocks(false)
+        },
+    );
+    peer.refresh_burnchain_view();
+
+    let nakamoto_start =
+        NakamotoBootPlan::nakamoto_first_tenure_height(&peer.config.burnchain.pox_constants);
+
+    // create a microfork
+    let naka_tip_ch = peer.network.stacks_tip.consensus_hash.clone();
+    let naka_tip_bh = peer.network.stacks_tip.block_hash.clone();
+    let naka_tip = peer.network.stacks_tip.block_id();
+
+    let sortdb = peer.sortdb_ref().reopen().unwrap();
+    let (chainstate, _) = peer.chainstate_ref().reopen().unwrap();
+
+    let naka_tip_header = NakamotoChainState::get_block_header_nakamoto(chainstate.db(), &naka_tip)
+        .unwrap()
+        .unwrap();
+
+    // load the full tenure for this tip
+    let mut naka_tip_tenure = chainstate
+        .nakamoto_blocks_db()
+        .load_nakamoto_tenure(&naka_tip)
+        .unwrap()
+        .unwrap();
+
+    assert!(naka_tip_tenure.len() > 1);
+
+    // make a microfork -- orphan naka_tip_tenure.last()
+    naka_tip_tenure.pop();
+
+    debug!("test: mine off of tenure");
+    debug!(
+        "test: first {}: {:?}",
+        &naka_tip_tenure.first().as_ref().unwrap().block_id(),
+        &naka_tip_tenure.first().as_ref().unwrap()
+    );
+    debug!(
+        "test: last {}: {:?}",
+        &naka_tip_tenure.last().as_ref().unwrap().block_id(),
+        &naka_tip_tenure.last().as_ref().unwrap()
+    );
+
+    peer.mine_nakamoto_on(naka_tip_tenure);
+    let (fork_naka_block, ..) = peer.single_block_tenure(&sender_key, |_| {}, |_| {}, |_| true);
+    debug!(
+        "test: produced fork {}: {:?}",
+        &fork_naka_block.block_id(),
+        &fork_naka_block
+    );
+
+    peer.refresh_burnchain_view();
+
+    peer.mine_nakamoto_on(vec![fork_naka_block.clone()]);
+    let (fork_naka_block_2, ..) = peer.single_block_tenure(&sender_key, |_| {}, |_| {}, |_| true);
+    debug!(
+        "test: confirmed fork with {}: {:?}",
+        &fork_naka_block_2.block_id(),
+        &fork_naka_block_2
+    );
+
+    peer.refresh_burnchain_view();
+
+    // get reward cyclce data
+    let (mut peer, reward_cycle_invs) =
+        peer_get_nakamoto_invs(peer, &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]);
+
+    // make a neighbor from this peer
+    let boot_observer = TestEventObserver::new();
+    let privk = StacksPrivateKey::from_seed(&[0, 1, 2, 3, 4]);
+    let mut boot_peer = peer.neighbor_with_observer(privk, Some(&boot_observer));
+
+    let (canonical_stacks_tip_ch, canonical_stacks_tip_bhh) =
+        SortitionDB::get_canonical_stacks_chain_tip_hash(peer.sortdb().conn()).unwrap();
+
+    let all_sortitions = peer.sortdb().get_all_snapshots().unwrap();
+    let tip = SortitionDB::get_canonical_burn_chain_tip(peer.sortdb().conn()).unwrap();
+    let nakamoto_tip = peer
+        .sortdb()
+        .index_handle(&tip.sortition_id)
+        .get_nakamoto_tip_block_id()
+        .unwrap()
+        .unwrap();
+
+    assert_eq!(tip.block_height, 53);
+
+    // boot up the boot peer's burnchain
+    for height in 25..tip.block_height {
+        let ops = peer
+            .get_burnchain_block_ops_at_height(height + 1)
+            .unwrap_or(vec![]);
+        let sn = {
+            let ih = peer.sortdb().index_handle(&tip.sortition_id);
+            let sn = ih.get_block_snapshot_by_height(height).unwrap().unwrap();
+            sn
+        };
+        test_debug!(
+            "boot_peer tip height={} hash={}",
+            sn.block_height,
+            &sn.burn_header_hash
+        );
+        test_debug!("ops = {:?}", &ops);
+        let block_header = TestPeer::make_next_burnchain_block(
+            &boot_peer.config.burnchain,
+            sn.block_height,
+            &sn.burn_header_hash,
+            ops.len() as u64,
+            false,
+        );
+        TestPeer::add_burnchain_block(&boot_peer.config.burnchain, &block_header, ops.clone());
+    }
+
+    let (mut boot_dns_client, boot_dns_thread_handle) = dns_thread_start(100);
+
+    // start running that peer so we can boot off of it
+    let (term_sx, term_rx) = sync_channel(1);
+    thread::scope(|s| {
+        s.spawn(move || {
+            let (mut last_stacks_tip_ch, mut last_stacks_tip_bhh) =
+                SortitionDB::get_canonical_stacks_chain_tip_hash(boot_peer.sortdb().conn())
+                    .unwrap();
+            loop {
+                boot_peer
+                    .run_with_ibd(true, Some(&mut boot_dns_client))
+                    .unwrap();
+
+                let (stacks_tip_ch, stacks_tip_bhh) =
+                    SortitionDB::get_canonical_stacks_chain_tip_hash(boot_peer.sortdb().conn())
+                        .unwrap();
+
+                last_stacks_tip_ch = stacks_tip_ch;
+                last_stacks_tip_bhh = stacks_tip_bhh;
+
+                debug!(
+                    "Booting peer's stacks tip is now {:?}",
+                    &boot_peer.network.stacks_tip
+                );
+                if stacks_tip_ch == canonical_stacks_tip_ch
+                    && stacks_tip_bhh == canonical_stacks_tip_bhh
+                {
+                    break;
+                }
+            }
+
+            term_sx.send(()).unwrap();
+        });
+
+        loop {
+            if term_rx.try_recv().is_ok() {
+                break;
+            }
+            peer.step_with_ibd(false).unwrap();
+        }
+    });
+
+    boot_dns_thread_handle.join().unwrap();
+}
+
+/// Test booting up a node where there is one shadow block in the prepare phase, as well as some
+/// blocks that mine atop it.
+#[test]
+fn test_nakamoto_download_run_2_peers_with_one_shadow_block() {
+    let observer = TestEventObserver::new();
+    let sender_key = StacksPrivateKey::new();
+    let sender_addr = to_addr(&sender_key);
+    let initial_balances = vec![(sender_addr.to_account_principal(), 1000000000)];
+    let bitvecs = vec![vec![true, true, false, false]];
+
+    let rc_len = 10u64;
+    let (mut peer, _) = make_nakamoto_peers_from_invs_ext(
+        function_name!(),
+        &observer,
+        bitvecs.clone(),
+        |boot_plan| {
+            boot_plan
+                .with_pox_constants(rc_len as u32, 5)
+                .with_extra_peers(0)
+                .with_initial_balances(initial_balances)
+                .with_malleablized_blocks(false)
+        },
+    );
+    peer.refresh_burnchain_view();
+    let (mut peer, reward_cycle_invs) =
+        peer_get_nakamoto_invs(peer, &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]);
+
+    let nakamoto_start =
+        NakamotoBootPlan::nakamoto_first_tenure_height(&peer.config.burnchain.pox_constants);
+
+    // create a shadow block
+    let naka_tip_ch = peer.network.stacks_tip.consensus_hash.clone();
+    let naka_tip_bh = peer.network.stacks_tip.block_hash.clone();
+    let naka_tip = peer.network.stacks_tip.block_id();
+
+    let sortdb = peer.sortdb_ref().reopen().unwrap();
+    let (chainstate, _) = peer.chainstate_ref().reopen().unwrap();
+
+    let naka_tip_header = NakamotoChainState::get_block_header_nakamoto(chainstate.db(), &naka_tip)
+        .unwrap()
+        .unwrap();
+
+    let naka_tip_tenure = chainstate
+        .nakamoto_blocks_db()
+        .load_nakamoto_tenure(&naka_tip)
+        .unwrap()
+        .unwrap();
+
+    assert!(naka_tip_tenure.len() > 1);
+
+    peer.mine_nakamoto_on(naka_tip_tenure);
+    let shadow_block = peer.make_shadow_tenure(None);
+    debug!(
+        "test: produced shadow block {}: {:?}",
+        &shadow_block.block_id(),
+        &shadow_block
+    );
+
+    peer.refresh_burnchain_view();
+
+    peer.mine_nakamoto_on(vec![shadow_block.clone()]);
+    let (next_block, ..) = peer.single_block_tenure(&sender_key, |_| {}, |_| {}, |_| true);
+    debug!(
+        "test: confirmed shadow block with {}: {:?}",
+        &next_block.block_id(),
+        &next_block
+    );
+
+    peer.refresh_burnchain_view();
+    peer.mine_nakamoto_on(vec![next_block.clone()]);
+
+    for _ in 0..9 {
+        let (next_block, ..) = peer.single_block_tenure(&sender_key, |_| {}, |_| {}, |_| true);
+        debug!(
+            "test: confirmed shadow block with {}: {:?}",
+            &next_block.block_id(),
+            &next_block
+        );
+
+        peer.refresh_burnchain_view();
+        peer.mine_nakamoto_on(vec![next_block.clone()]);
+    }
+
+    let all_sortitions = peer.sortdb().get_all_snapshots().unwrap();
+    let tip = SortitionDB::get_canonical_burn_chain_tip(peer.sortdb().conn()).unwrap();
+    let nakamoto_tip = peer
+        .sortdb()
+        .index_handle(&tip.sortition_id)
+        .get_nakamoto_tip_block_id()
+        .unwrap()
+        .unwrap();
+
+    /*
+    assert_eq!(
+        tip.block_height,
+        56
+    );
+    */
+
+    // make a neighbor from this peer
+    let boot_observer = TestEventObserver::new();
+    let privk = StacksPrivateKey::from_seed(&[0, 1, 2, 3, 4]);
+    let mut boot_peer = peer.neighbor_with_observer(privk, Some(&boot_observer));
+
+    let (canonical_stacks_tip_ch, canonical_stacks_tip_bhh) =
+        SortitionDB::get_canonical_stacks_chain_tip_hash(peer.sortdb().conn()).unwrap();
+
+    // boot up the boot peer's burnchain
+    for height in 25..tip.block_height {
+        let ops = peer
+            .get_burnchain_block_ops_at_height(height + 1)
+            .unwrap_or(vec![]);
+        let sn = {
+            let ih = peer.sortdb().index_handle(&tip.sortition_id);
+            let sn = ih.get_block_snapshot_by_height(height).unwrap().unwrap();
+            sn
+        };
+        test_debug!(
+            "boot_peer tip height={} hash={}",
+            sn.block_height,
+            &sn.burn_header_hash
+        );
+        test_debug!("ops = {:?}", &ops);
+        let block_header = TestPeer::make_next_burnchain_block(
+            &boot_peer.config.burnchain,
+            sn.block_height,
+            &sn.burn_header_hash,
+            ops.len() as u64,
+            false,
+        );
+        TestPeer::add_burnchain_block(&boot_peer.config.burnchain, &block_header, ops.clone());
+    }
+
+    {
+        let mut node = boot_peer.stacks_node.take().unwrap();
+        let tx = node.chainstate.staging_db_tx_begin().unwrap();
+        tx.add_shadow_block(&shadow_block).unwrap();
+        tx.commit().unwrap();
+        boot_peer.stacks_node = Some(node);
+    }
+
+    let (mut boot_dns_client, boot_dns_thread_handle) = dns_thread_start(100);
+
+    // start running that peer so we can boot off of it
+    let (term_sx, term_rx) = sync_channel(1);
+    thread::scope(|s| {
+        s.spawn(move || {
+            let (mut last_stacks_tip_ch, mut last_stacks_tip_bhh) =
+                SortitionDB::get_canonical_stacks_chain_tip_hash(boot_peer.sortdb().conn())
+                    .unwrap();
+            loop {
+                boot_peer
+                    .run_with_ibd(true, Some(&mut boot_dns_client))
+                    .unwrap();
+
+                let (stacks_tip_ch, stacks_tip_bhh) =
+                    SortitionDB::get_canonical_stacks_chain_tip_hash(boot_peer.sortdb().conn())
+                        .unwrap();
+
+                last_stacks_tip_ch = stacks_tip_ch;
+                last_stacks_tip_bhh = stacks_tip_bhh;
+
+                debug!(
+                    "Booting peer's stacks tip is now {:?}",
+                    &boot_peer.network.stacks_tip
+                );
+                if stacks_tip_ch == canonical_stacks_tip_ch {
+                    break;
+                }
+            }
+
+            term_sx.send(()).unwrap();
+        });
+
+        loop {
+            if term_rx.try_recv().is_ok() {
+                break;
+            }
+            peer.step_with_ibd(false).unwrap();
+        }
+    });
+
+    boot_dns_thread_handle.join().unwrap();
+}
+
+/// Test booting up a node where the whole prepare phase is shadow blocks
+#[test]
+fn test_nakamoto_download_run_2_peers_shadow_prepare_phase() {
+    let observer = TestEventObserver::new();
+    let sender_key = StacksPrivateKey::new();
+    let sender_addr = to_addr(&sender_key);
+    let initial_balances = vec![(sender_addr.to_account_principal(), 1000000000)];
+    let bitvecs = vec![vec![true, true]];
+
+    let rc_len = 10u64;
+    let (mut peer, _) = make_nakamoto_peers_from_invs_ext(
+        function_name!(),
+        &observer,
+        bitvecs.clone(),
+        |boot_plan| {
+            boot_plan
+                .with_pox_constants(rc_len as u32, 5)
+                .with_extra_peers(0)
+                .with_initial_balances(initial_balances)
+                .with_malleablized_blocks(false)
+        },
+    );
+    peer.refresh_burnchain_view();
+    let (mut peer, reward_cycle_invs) =
+        peer_get_nakamoto_invs(peer, &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]);
+
+    let nakamoto_start =
+        NakamotoBootPlan::nakamoto_first_tenure_height(&peer.config.burnchain.pox_constants);
+
+    // create a shadow block
+    let naka_tip_ch = peer.network.stacks_tip.consensus_hash.clone();
+    let naka_tip_bh = peer.network.stacks_tip.block_hash.clone();
+    let naka_tip = peer.network.stacks_tip.block_id();
+
+    let sortdb = peer.sortdb_ref().reopen().unwrap();
+    let (chainstate, _) = peer.chainstate_ref().reopen().unwrap();
+
+    let naka_tip_header = NakamotoChainState::get_block_header_nakamoto(chainstate.db(), &naka_tip)
+        .unwrap()
+        .unwrap();
+
+    let naka_tip_tenure = chainstate
+        .nakamoto_blocks_db()
+        .load_nakamoto_tenure(&naka_tip)
+        .unwrap()
+        .unwrap();
+
+    assert!(naka_tip_tenure.len() > 1);
+
+    peer.mine_nakamoto_on(naka_tip_tenure);
+
+    let mut shadow_blocks = vec![];
+    for _ in 0..10 {
+        let shadow_block = peer.make_shadow_tenure(None);
+        debug!(
+            "test: produced shadow block {}: {:?}",
+            &shadow_block.block_id(),
+            &shadow_block
+        );
+        shadow_blocks.push(shadow_block.clone());
+        peer.refresh_burnchain_view();
+
+        peer.mine_nakamoto_on(vec![shadow_block.clone()]);
+    }
+
+    match peer.single_block_tenure_fallible(&sender_key, |_| {}, |_| {}, |_| true) {
+        Ok((next_block, ..)) => {
+            debug!(
+                "test: confirmed shadow block with {}: {:?}",
+                &next_block.block_id(),
+                &next_block
+            );
+
+            peer.refresh_burnchain_view();
+            peer.mine_nakamoto_on(vec![next_block.clone()]);
+        }
+        Err(ChainstateError::NoSuchBlockError) => {
+            // tried to mine but our commit was invalid (e.g. because we haven't mined often
+            // enough)
+            peer.refresh_burnchain_view();
+        }
+        Err(e) => {
+            panic!("FATAL: {:?}", &e);
+        }
+    };
+
+    for _ in 0..10 {
+        let (next_block, ..) =
+            match peer.single_block_tenure_fallible(&sender_key, |_| {}, |_| {}, |_| true) {
+                Ok(x) => x,
+                Err(ChainstateError::NoSuchBlockError) => {
+                    // tried to mine but our commit was invalid (e.g. because we haven't mined often
+                    // enough)
+                    peer.refresh_burnchain_view();
+                    continue;
+                }
+                Err(e) => {
+                    panic!("FATAL: {:?}", &e);
+                }
+            };
+
+        debug!(
+            "test: confirmed shadow block with {}: {:?}",
+            &next_block.block_id(),
+            &next_block
+        );
+
+        peer.refresh_burnchain_view();
+        peer.mine_nakamoto_on(vec![next_block.clone()]);
+    }
+
+    let all_sortitions = peer.sortdb().get_all_snapshots().unwrap();
+    let tip = SortitionDB::get_canonical_burn_chain_tip(peer.sortdb().conn()).unwrap();
+    let nakamoto_tip = peer
+        .sortdb()
+        .index_handle(&tip.sortition_id)
+        .get_nakamoto_tip_block_id()
+        .unwrap()
+        .unwrap();
+
+    // make a neighbor from this peer
+    let boot_observer = TestEventObserver::new();
+    let privk = StacksPrivateKey::from_seed(&[0, 1, 2, 3, 4]);
+    let mut boot_peer = peer.neighbor_with_observer(privk, Some(&boot_observer));
+
+    let (canonical_stacks_tip_ch, canonical_stacks_tip_bhh) =
+        SortitionDB::get_canonical_stacks_chain_tip_hash(peer.sortdb().conn()).unwrap();
+
+    // boot up the boot peer's burnchain
+    for height in 25..tip.block_height {
+        let ops = peer
+            .get_burnchain_block_ops_at_height(height + 1)
+            .unwrap_or(vec![]);
+        let sn = {
+            let ih = peer.sortdb().index_handle(&tip.sortition_id);
+            let sn = ih.get_block_snapshot_by_height(height).unwrap().unwrap();
+            sn
+        };
+        test_debug!(
+            "boot_peer tip height={} hash={}",
+            sn.block_height,
+            &sn.burn_header_hash
+        );
+        test_debug!("ops = {:?}", &ops);
+        let block_header = TestPeer::make_next_burnchain_block(
+            &boot_peer.config.burnchain,
+            sn.block_height,
+            &sn.burn_header_hash,
+            ops.len() as u64,
+            false,
+        );
+        TestPeer::add_burnchain_block(&boot_peer.config.burnchain, &block_header, ops.clone());
+    }
+    {
+        let mut node = boot_peer.stacks_node.take().unwrap();
+        let tx = node.chainstate.staging_db_tx_begin().unwrap();
+        for shadow_block in shadow_blocks.into_iter() {
+            tx.add_shadow_block(&shadow_block).unwrap();
+        }
+        tx.commit().unwrap();
+        boot_peer.stacks_node = Some(node);
+    }
+
+    let (mut boot_dns_client, boot_dns_thread_handle) = dns_thread_start(100);
+
+    // start running that peer so we can boot off of it
+    let (term_sx, term_rx) = sync_channel(1);
+    thread::scope(|s| {
+        s.spawn(move || {
+            let (mut last_stacks_tip_ch, mut last_stacks_tip_bhh) =
+                SortitionDB::get_canonical_stacks_chain_tip_hash(boot_peer.sortdb().conn())
+                    .unwrap();
+            loop {
+                boot_peer
+                    .run_with_ibd(true, Some(&mut boot_dns_client))
+                    .unwrap();
+
+                let (stacks_tip_ch, stacks_tip_bhh) =
+                    SortitionDB::get_canonical_stacks_chain_tip_hash(boot_peer.sortdb().conn())
+                        .unwrap();
+
+                last_stacks_tip_ch = stacks_tip_ch;
+                last_stacks_tip_bhh = stacks_tip_bhh;
+
+                debug!(
+                    "Booting peer's stacks tip is now {:?}",
+                    &boot_peer.network.stacks_tip
+                );
+                if stacks_tip_ch == canonical_stacks_tip_ch {
+                    break;
+                }
+            }
+
+            term_sx.send(()).unwrap();
+        });
+
+        loop {
+            if term_rx.try_recv().is_ok() {
+                break;
+            }
+            peer.step_with_ibd(false).unwrap();
+        }
+    });
+
+    boot_dns_thread_handle.join().unwrap();
+}
+
+/// Test booting up a node where multiple reward cycles are shadow blocks
+#[test]
+fn test_nakamoto_download_run_2_peers_shadow_reward_cycles() {
+    let observer = TestEventObserver::new();
+    let sender_key = StacksPrivateKey::new();
+    let sender_addr = to_addr(&sender_key);
+    let initial_balances = vec![(sender_addr.to_account_principal(), 1000000000)];
+    let bitvecs = vec![vec![true, true]];
+
+    let rc_len = 10u64;
+    let (mut peer, _) = make_nakamoto_peers_from_invs_ext(
+        function_name!(),
+        &observer,
+        bitvecs.clone(),
+        |boot_plan| {
+            boot_plan
+                .with_pox_constants(rc_len as u32, 5)
+                .with_extra_peers(0)
+                .with_initial_balances(initial_balances)
+                .with_malleablized_blocks(false)
+        },
+    );
+    peer.refresh_burnchain_view();
+    let (mut peer, reward_cycle_invs) =
+        peer_get_nakamoto_invs(peer, &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]);
+
+    let nakamoto_start =
+        NakamotoBootPlan::nakamoto_first_tenure_height(&peer.config.burnchain.pox_constants);
+
+    // create a shadow block
+    let naka_tip_ch = peer.network.stacks_tip.consensus_hash.clone();
+    let naka_tip_bh = peer.network.stacks_tip.block_hash.clone();
+    let naka_tip = peer.network.stacks_tip.block_id();
+
+    let sortdb = peer.sortdb_ref().reopen().unwrap();
+    let (chainstate, _) = peer.chainstate_ref().reopen().unwrap();
+
+    let naka_tip_header = NakamotoChainState::get_block_header_nakamoto(chainstate.db(), &naka_tip)
+        .unwrap()
+        .unwrap();
+
+    let naka_tip_tenure = chainstate
+        .nakamoto_blocks_db()
+        .load_nakamoto_tenure(&naka_tip)
+        .unwrap()
+        .unwrap();
+
+    assert!(naka_tip_tenure.len() > 1);
+
+    peer.mine_nakamoto_on(naka_tip_tenure);
+
+    let mut shadow_blocks = vec![];
+    for _ in 0..30 {
+        let shadow_block = peer.make_shadow_tenure(None);
+        debug!(
+            "test: produced shadow block {}: {:?}",
+            &shadow_block.block_id(),
+            &shadow_block
+        );
+        shadow_blocks.push(shadow_block.clone());
+        peer.refresh_burnchain_view();
+
+        peer.mine_nakamoto_on(vec![shadow_block.clone()]);
+    }
+
+    match peer.single_block_tenure_fallible(&sender_key, |_| {}, |_| {}, |_| true) {
+        Ok((next_block, ..)) => {
+            debug!(
+                "test: confirmed shadow block with {}: {:?}",
+                &next_block.block_id(),
+                &next_block
+            );
+
+            peer.refresh_burnchain_view();
+            peer.mine_nakamoto_on(vec![next_block.clone()]);
+        }
+        Err(ChainstateError::NoSuchBlockError) => {
+            // tried to mine but our commit was invalid (e.g. because we haven't mined often
+            // enough)
+            peer.refresh_burnchain_view();
+        }
+        Err(e) => {
+            panic!("FATAL: {:?}", &e);
+        }
+    };
+
+    for _ in 0..10 {
+        let (next_block, ..) =
+            match peer.single_block_tenure_fallible(&sender_key, |_| {}, |_| {}, |_| true) {
+                Ok(x) => x,
+                Err(ChainstateError::NoSuchBlockError) => {
+                    // tried to mine but our commit was invalid (e.g. because we haven't mined often
+                    // enough)
+                    peer.refresh_burnchain_view();
+                    continue;
+                }
+                Err(e) => {
+                    panic!("FATAL: {:?}", &e);
+                }
+            };
+
+        debug!(
+            "test: confirmed shadow block with {}: {:?}",
+            &next_block.block_id(),
+            &next_block
+        );
+
+        peer.refresh_burnchain_view();
+        peer.mine_nakamoto_on(vec![next_block.clone()]);
+    }
+
+    let all_sortitions = peer.sortdb().get_all_snapshots().unwrap();
+    let tip = SortitionDB::get_canonical_burn_chain_tip(peer.sortdb().conn()).unwrap();
+    let nakamoto_tip = peer
+        .sortdb()
+        .index_handle(&tip.sortition_id)
+        .get_nakamoto_tip_block_id()
+        .unwrap()
+        .unwrap();
+
+    assert_eq!(tip.block_height, 84);
+
+    // make a neighbor from this peer
+    let boot_observer = TestEventObserver::new();
+    let privk = StacksPrivateKey::from_seed(&[0, 1, 2, 3, 4]);
+    let mut boot_peer = peer.neighbor_with_observer(privk, Some(&boot_observer));
+
+    let (canonical_stacks_tip_ch, canonical_stacks_tip_bhh) =
+        SortitionDB::get_canonical_stacks_chain_tip_hash(peer.sortdb().conn()).unwrap();
+
+    // boot up the boot peer's burnchain
+    for height in 25..tip.block_height {
+        let ops = peer
+            .get_burnchain_block_ops_at_height(height + 1)
+            .unwrap_or(vec![]);
+        let sn = {
+            let ih = peer.sortdb().index_handle(&tip.sortition_id);
+            let sn = ih.get_block_snapshot_by_height(height).unwrap().unwrap();
+            sn
+        };
+        test_debug!(
+            "boot_peer tip height={} hash={}",
+            sn.block_height,
+            &sn.burn_header_hash
+        );
+        test_debug!("ops = {:?}", &ops);
+        let block_header = TestPeer::make_next_burnchain_block(
+            &boot_peer.config.burnchain,
+            sn.block_height,
+            &sn.burn_header_hash,
+            ops.len() as u64,
+            false,
+        );
+        TestPeer::add_burnchain_block(&boot_peer.config.burnchain, &block_header, ops.clone());
+    }
+    {
+        let mut node = boot_peer.stacks_node.take().unwrap();
+        let tx = node.chainstate.staging_db_tx_begin().unwrap();
+        for shadow_block in shadow_blocks.into_iter() {
+            tx.add_shadow_block(&shadow_block).unwrap();
+        }
+        tx.commit().unwrap();
+        boot_peer.stacks_node = Some(node);
+    }
+
+    let (mut boot_dns_client, boot_dns_thread_handle) = dns_thread_start(100);
+
     // start running that peer so we can boot off of it
     let (term_sx, term_rx) = sync_channel(1);
     thread::scope(|s| {
diff --git a/stackslib/src/net/tests/inv/nakamoto.rs b/stackslib/src/net/tests/inv/nakamoto.rs
index fac9623d3f..5f889cde3e 100644
--- a/stackslib/src/net/tests/inv/nakamoto.rs
+++ b/stackslib/src/net/tests/inv/nakamoto.rs
@@ -404,15 +404,12 @@ pub fn make_nakamoto_peers_from_invs<'a>(
     bitvecs: Vec<Vec<bool>>,
     num_peers: usize,
 ) -> (TestPeer<'a>, Vec<TestPeer<'a>>) {
-    inner_make_nakamoto_peers_from_invs(
-        test_name,
-        observer,
-        rc_len,
-        prepare_len,
-        bitvecs,
-        num_peers,
-        vec![],
-    )
+    make_nakamoto_peers_from_invs_ext(test_name, observer, bitvecs, |boot_plan| {
+        boot_plan
+            .with_pox_constants(rc_len, prepare_len)
+            .with_extra_peers(num_peers)
+            .with_initial_balances(vec![])
+    })
 }
 
 /// NOTE: The second return value does _not_ need `<'a>`, since `observer` is never installed into
@@ -426,31 +423,26 @@ pub fn make_nakamoto_peers_from_invs_and_balances<'a>(
     num_peers: usize,
     initial_balances: Vec<(PrincipalData, u64)>,
 ) -> (TestPeer<'a>, Vec<TestPeer<'a>>) {
-    inner_make_nakamoto_peers_from_invs(
-        test_name,
-        observer,
-        rc_len,
-        prepare_len,
-        bitvecs,
-        num_peers,
-        initial_balances,
-    )
+    make_nakamoto_peers_from_invs_ext(test_name, observer, bitvecs, |boot_plan| {
+        boot_plan
+            .with_pox_constants(rc_len, prepare_len)
+            .with_extra_peers(num_peers)
+            .with_initial_balances(initial_balances)
+    })
 }
 
 /// Make peers from inventories and balances
-fn inner_make_nakamoto_peers_from_invs<'a>(
+/// NOTE: The second return value does _not_ need `<'a>`, since `observer` is never installed into
+/// the peers here.  However, it appears unavoidable to the borrow-checker.
+pub fn make_nakamoto_peers_from_invs_ext<'a, F>(
     test_name: &str,
     observer: &'a TestEventObserver,
-    rc_len: u32,
-    prepare_len: u32,
     bitvecs: Vec<Vec<bool>>,
-    num_peers: usize,
-    mut initial_balances: Vec<(PrincipalData, u64)>,
-) -> (TestPeer<'a>, Vec<TestPeer<'a>>) {
-    for bitvec in bitvecs.iter() {
-        assert_eq!(bitvec.len() as u32, rc_len);
-    }
-
+    boot_config: F,
+) -> (TestPeer<'a>, Vec<TestPeer<'a>>)
+where
+    F: FnOnce(NakamotoBootPlan) -> NakamotoBootPlan,
+{
     let private_key = StacksPrivateKey::from_seed(&[2]);
     let addr = StacksAddress::from_public_keys(
         C32_ADDRESS_VERSION_TESTNET_SINGLESIG,
@@ -461,6 +453,7 @@ fn inner_make_nakamoto_peers_from_invs<'a>(
     .unwrap();
     let recipient_addr =
         StacksAddress::from_string("ST2YM3J4KQK09V670TD6ZZ1XYNYCNGCWCVTASN5VM").unwrap();
+    let mut initial_balances = vec![(addr.to_account_principal(), 1_000_000)];
 
     let mut sender_nonce = 0;
 
@@ -525,14 +518,13 @@ fn inner_make_nakamoto_peers_from_invs<'a>(
         0, 0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2, 3, 3, 3, 3, 3, 3,
     ]);
 
-    initial_balances.push((addr.into(), 1_000_000));
-    let plan = NakamotoBootPlan::new(test_name)
-        .with_private_key(private_key)
-        .with_pox_constants(rc_len, prepare_len)
-        .with_initial_balances(initial_balances)
-        .with_extra_peers(num_peers)
-        .with_test_signers(test_signers)
-        .with_test_stackers(test_stackers);
+    let mut plan = boot_config(
+        NakamotoBootPlan::new(test_name)
+            .with_private_key(private_key)
+            .with_test_signers(test_signers)
+            .with_test_stackers(test_stackers),
+    );
+    plan.initial_balances.append(&mut initial_balances);
 
     let (peer, other_peers) = plan.boot_into_nakamoto_peers(boot_tenures, Some(observer));
     (peer, other_peers)
@@ -2382,3 +2374,87 @@ fn test_nakamoto_make_tenure_inv_from_old_tips() {
         assert_eq!(bits, expected_bits[0..bit_len]);
     }
 }
+
+#[test]
+fn test_nakamoto_invs_shadow_blocks() {
+    let observer = TestEventObserver::new();
+    let sender_key = StacksPrivateKey::new();
+    let sender_addr = to_addr(&sender_key);
+    let initial_balances = vec![(sender_addr.to_account_principal(), 1000000000)];
+    let mut bitvecs = vec![vec![
+        true, true, true, true, true, true, true, true, true, true,
+    ]];
+
+    let (mut peer, _) = make_nakamoto_peers_from_invs_and_balances(
+        function_name!(),
+        &observer,
+        10,
+        3,
+        bitvecs.clone(),
+        0,
+        initial_balances,
+    );
+    let nakamoto_start =
+        NakamotoBootPlan::nakamoto_first_tenure_height(&peer.config.burnchain.pox_constants);
+
+    let mut expected_ids = vec![];
+
+    // construct and add shadow blocks to this peer's chainstate
+    peer.refresh_burnchain_view();
+    let shadow_block = peer.make_shadow_tenure(None);
+    expected_ids.push(shadow_block.block_id());
+    peer.mine_nakamoto_on(vec![shadow_block]);
+
+    peer.refresh_burnchain_view();
+    let (naka_block, ..) = peer.single_block_tenure(&sender_key, |_| {}, |_| {}, |_| true);
+    expected_ids.push(naka_block.block_id());
+    peer.mine_nakamoto_on(vec![naka_block]);
+
+    peer.refresh_burnchain_view();
+    let shadow_block = peer.make_shadow_tenure(None);
+    expected_ids.push(shadow_block.block_id());
+    peer.mine_nakamoto_on(vec![shadow_block]);
+
+    peer.refresh_burnchain_view();
+    let (naka_block, ..) = peer.single_block_tenure(&sender_key, |_| {}, |_| {}, |_| true);
+    expected_ids.push(naka_block.block_id());
+    peer.mine_nakamoto_on(vec![naka_block]);
+
+    peer.refresh_burnchain_view();
+    let shadow_block = peer.make_shadow_tenure(None);
+    expected_ids.push(shadow_block.block_id());
+    peer.mine_nakamoto_on(vec![shadow_block]);
+
+    peer.refresh_burnchain_view();
+    let (naka_block, ..) = peer.single_block_tenure(&sender_key, |_| {}, |_| {}, |_| true);
+    expected_ids.push(naka_block.block_id());
+    peer.mine_nakamoto_on(vec![naka_block]);
+
+    let (mut peer, reward_cycle_invs) =
+        peer_get_nakamoto_invs(peer, &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]);
+
+    // the inv should show `true` for each shadow tenure
+    bitvecs.push(vec![true, true, true, true, true, true]);
+    check_inv_messages(bitvecs, 10, nakamoto_start, reward_cycle_invs);
+
+    // shadow blocks are part of the history
+    peer.refresh_burnchain_view();
+    let tip = peer.network.stacks_tip.block_id();
+
+    let mut stored_block_ids = vec![];
+    let mut cursor = tip;
+    for _ in 0..expected_ids.len() {
+        let block = peer
+            .chainstate()
+            .nakamoto_blocks_db()
+            .get_nakamoto_block(&cursor)
+            .unwrap()
+            .unwrap()
+            .0;
+        stored_block_ids.push(block.block_id());
+        cursor = block.header.parent_block_id;
+    }
+
+    stored_block_ids.reverse();
+    assert_eq!(stored_block_ids, expected_ids);
+}
diff --git a/stackslib/src/net/tests/mempool/mod.rs b/stackslib/src/net/tests/mempool/mod.rs
index 7a44a56788..d3f30aca19 100644
--- a/stackslib/src/net/tests/mempool/mod.rs
+++ b/stackslib/src/net/tests/mempool/mod.rs
@@ -973,63 +973,68 @@ pub fn test_mempool_storage_nakamoto() {
             StacksAddress::from_string("ST2YM3J4KQK09V670TD6ZZ1XYNYCNGCWCVTASN5VM").unwrap();
 
         let mempool_txs = RefCell::new(vec![]);
-        let blocks_and_sizes = peer.make_nakamoto_tenure_and(
-            tenure_change_tx,
-            coinbase_tx,
-            &mut test_signers,
-            |_| {},
-            |miner, chainstate, sortdb, blocks_so_far| {
-                let mut txs = vec![];
-                if blocks_so_far.len() < num_blocks {
-                    let account = get_account(chainstate, sortdb, &addr);
-
-                    let stx_transfer = make_token_transfer(
-                        chainstate,
-                        sortdb,
-                        &private_key,
-                        account.nonce,
-                        200,
-                        200,
-                        &recipient_addr,
-                    );
-                    txs.push(stx_transfer.clone());
-                    (*mempool_txs.borrow_mut()).push(stx_transfer.clone());
-                    all_txs.push(stx_transfer.clone());
-                }
-                txs
-            },
-            |_| {
-                let tip = NakamotoChainState::get_canonical_block_header(chainstate.db(), &sortdb)
+        let blocks_and_sizes = peer
+            .make_nakamoto_tenure_and(
+                tenure_change_tx,
+                coinbase_tx,
+                &mut test_signers,
+                |_| {},
+                |miner, chainstate, sortdb, blocks_so_far| {
+                    let mut txs = vec![];
+                    if blocks_so_far.len() < num_blocks {
+                        let account = get_account(chainstate, sortdb, &addr);
+
+                        let stx_transfer = make_token_transfer(
+                            chainstate,
+                            sortdb,
+                            &private_key,
+                            account.nonce,
+                            200,
+                            200,
+                            &recipient_addr,
+                        );
+                        txs.push(stx_transfer.clone());
+                        (*mempool_txs.borrow_mut()).push(stx_transfer.clone());
+                        all_txs.push(stx_transfer.clone());
+                    }
+                    txs
+                },
+                |_| {
+                    let tip =
+                        NakamotoChainState::get_canonical_block_header(chainstate.db(), &sortdb)
+                            .unwrap()
+                            .unwrap();
+                    let sort_tip = SortitionDB::get_block_snapshot_consensus(
+                        sortdb.conn(),
+                        &tip.consensus_hash,
+                    )
                     .unwrap()
                     .unwrap();
-                let sort_tip =
-                    SortitionDB::get_block_snapshot_consensus(sortdb.conn(), &tip.consensus_hash)
+                    let epoch = SortitionDB::get_stacks_epoch(sortdb.conn(), sort_tip.block_height)
                         .unwrap()
                         .unwrap();
-                let epoch = SortitionDB::get_stacks_epoch(sortdb.conn(), sort_tip.block_height)
-                    .unwrap()
-                    .unwrap();
-
-                // submit each transaction to the mempool
-                for mempool_tx in (*mempool_txs.borrow()).as_slice() {
-                    mempool
-                        .submit(
-                            &mut chainstate,
-                            &sortdb,
-                            &tip.consensus_hash,
-                            &tip.anchored_header.block_hash(),
-                            &mempool_tx,
-                            None,
-                            &epoch.block_limit,
-                            &epoch.epoch_id,
-                        )
-                        .unwrap();
-                }
 
-                (*mempool_txs.borrow_mut()).clear();
-                true
-            },
-        );
+                    // submit each transaction to the mempool
+                    for mempool_tx in (*mempool_txs.borrow()).as_slice() {
+                        mempool
+                            .submit(
+                                &mut chainstate,
+                                &sortdb,
+                                &tip.consensus_hash,
+                                &tip.anchored_header.block_hash(),
+                                &mempool_tx,
+                                None,
+                                &epoch.block_limit,
+                                &epoch.epoch_id,
+                            )
+                            .unwrap();
+                    }
+
+                    (*mempool_txs.borrow_mut()).clear();
+                    true
+                },
+            )
+            .unwrap();
 
         total_blocks += num_blocks;
     }
diff --git a/stackslib/src/net/tests/mod.rs b/stackslib/src/net/tests/mod.rs
index d9c7402bf8..6729dbc4a8 100644
--- a/stackslib/src/net/tests/mod.rs
+++ b/stackslib/src/net/tests/mod.rs
@@ -105,6 +105,8 @@ pub struct NakamotoBootPlan {
     pub num_peers: usize,
     /// Whether to add an initial balance for `private_key`'s account
     pub add_default_balance: bool,
+    /// Whether or not to produce malleablized blocks
+    pub malleablized_blocks: bool,
     pub network_id: u32,
 }
 
@@ -121,6 +123,7 @@ impl NakamotoBootPlan {
             observer: Some(TestEventObserver::new()),
             num_peers: 0,
             add_default_balance: true,
+            malleablized_blocks: true,
             network_id: TestPeerConfig::default().network_id,
         }
     }
@@ -177,6 +180,11 @@ impl NakamotoBootPlan {
         self
     }
 
+    pub fn with_malleablized_blocks(mut self, malleablized_blocks: bool) -> Self {
+        self.malleablized_blocks = malleablized_blocks;
+        self
+    }
+
     /// This is the first tenure in which nakamoto blocks will be built.
     /// However, it is also the last sortition for an epoch 2.x block.
     pub fn nakamoto_start_burn_height(pox_consts: &PoxConstants) -> u64 {
@@ -406,6 +414,8 @@ impl NakamotoBootPlan {
         peer_config.burnchain.pox_constants = self.pox_constants.clone();
         let mut peer = TestPeer::new_with_observer(peer_config.clone(), observer);
 
+        peer.mine_malleablized_blocks = self.malleablized_blocks;
+
         let mut other_peers = vec![];
         for i in 0..self.num_peers {
             let mut other_config = peer_config.clone();
@@ -416,7 +426,11 @@ impl NakamotoBootPlan {
             other_config.private_key = StacksPrivateKey::from_seed(&(i as u128).to_be_bytes());
 
             other_config.add_neighbor(&peer.to_neighbor());
-            other_peers.push(TestPeer::new_with_observer(other_config, None));
+
+            let mut other_peer = TestPeer::new_with_observer(other_config, None);
+            other_peer.mine_malleablized_blocks = self.malleablized_blocks;
+
+            other_peers.push(other_peer);
         }
 
         self.advance_to_nakamoto(&mut peer, &mut other_peers);
diff --git a/testnet/stacks-node/src/nakamoto_node/relayer.rs b/testnet/stacks-node/src/nakamoto_node/relayer.rs
index 7c8dc6f2c5..39ffaf0a55 100644
--- a/testnet/stacks-node/src/nakamoto_node/relayer.rs
+++ b/testnet/stacks-node/src/nakamoto_node/relayer.rs
@@ -31,7 +31,7 @@ use stacks::chainstate::burn::operations::{
 };
 use stacks::chainstate::burn::{BlockSnapshot, ConsensusHash};
 use stacks::chainstate::nakamoto::coordinator::get_nakamoto_next_recipients;
-use stacks::chainstate::nakamoto::NakamotoChainState;
+use stacks::chainstate::nakamoto::{NakamotoBlockHeader, NakamotoChainState};
 use stacks::chainstate::stacks::address::PoxAddress;
 use stacks::chainstate::stacks::db::StacksChainState;
 use stacks::chainstate::stacks::miner::{
@@ -557,6 +557,7 @@ impl RelayerThread {
         tip_block_ch: &ConsensusHash,
         tip_block_bh: &BlockHeaderHash,
     ) -> Result<LastCommit, NakamotoNodeError> {
+        let tip_block_id = StacksBlockId::new(&tip_block_ch, &tip_block_bh);
         let sort_tip = SortitionDB::get_canonical_burn_chain_tip(self.sortdb.conn())
             .map_err(|_| NakamotoNodeError::SnapshotNotFoundForChainTip)?;
 
@@ -630,18 +631,41 @@ impl RelayerThread {
             return Err(NakamotoNodeError::ParentNotFound);
         };
 
-        // find the parent block-commit of this commit
+        // find the parent block-commit of this commit, so we can find the parent vtxindex
+        // if the parent is a shadow block, then the vtxindex would be 0.
         let commit_parent_block_burn_height = tip_tenure_sortition.block_height;
-        let Ok(Some(parent_winning_tx)) = SortitionDB::get_block_commit(
-            self.sortdb.conn(),
-            &tip_tenure_sortition.winning_block_txid,
-            &tip_tenure_sortition.sortition_id,
-        ) else {
-            error!("Relayer: Failed to lookup the block commit of parent tenure ID"; "tenure_consensus_hash" => %tip_block_ch);
-            return Err(NakamotoNodeError::SnapshotNotFoundForChainTip);
-        };
+        let commit_parent_winning_vtxindex = if let Ok(Some(parent_winning_tx)) =
+            SortitionDB::get_block_commit(
+                self.sortdb.conn(),
+                &tip_tenure_sortition.winning_block_txid,
+                &tip_tenure_sortition.sortition_id,
+            ) {
+            parent_winning_tx.vtxindex
+        } else {
+            debug!(
+                "{}/{} ({}) must be a shadow block, since it has no block-commit",
+                &tip_block_bh, &tip_block_ch, &tip_block_id
+            );
+            let Ok(Some(parent_version)) =
+                NakamotoChainState::get_nakamoto_block_version(self.chainstate.db(), &tip_block_id)
+            else {
+                error!(
+                    "Relayer: Failed to lookup block version of {}",
+                    &tip_block_id
+                );
+                return Err(NakamotoNodeError::ParentNotFound);
+            };
+
+            if !NakamotoBlockHeader::is_shadow_block_version(parent_version) {
+                error!(
+                    "Relayer: parent block-commit of {} not found, and it is not a shadow block",
+                    &tip_block_id
+                );
+                return Err(NakamotoNodeError::ParentNotFound);
+            }
 
-        let commit_parent_winning_vtxindex = parent_winning_tx.vtxindex;
+            0
+        };
 
         // epoch in which this commit will be sent (affects how the burnchain client processes it)
         let Ok(Some(target_epoch)) =
diff --git a/testnet/stacks-node/src/tests/nakamoto_integrations.rs b/testnet/stacks-node/src/tests/nakamoto_integrations.rs
index 5d712ad550..ef6199d331 100644
--- a/testnet/stacks-node/src/tests/nakamoto_integrations.rs
+++ b/testnet/stacks-node/src/tests/nakamoto_integrations.rs
@@ -37,8 +37,9 @@ use stacks::chainstate::burn::operations::{
 };
 use stacks::chainstate::coordinator::comm::CoordinatorChannels;
 use stacks::chainstate::coordinator::OnChainRewardSetProvider;
-use stacks::chainstate::nakamoto::coordinator::load_nakamoto_reward_set;
+use stacks::chainstate::nakamoto::coordinator::{load_nakamoto_reward_set, TEST_COORDINATOR_STALL};
 use stacks::chainstate::nakamoto::miner::NakamotoBlockBuilder;
+use stacks::chainstate::nakamoto::shadow::shadow_chainstate_repair;
 use stacks::chainstate::nakamoto::test_signers::TestSigners;
 use stacks::chainstate::nakamoto::{NakamotoBlock, NakamotoBlockHeader, NakamotoChainState};
 use stacks::chainstate::stacks::address::{PoxAddress, StacksAddressExtensions};
@@ -90,6 +91,7 @@ use stacks_common::util::secp256k1::{MessageSignature, Secp256k1PrivateKey, Secp
 use stacks_common::util::{get_epoch_time_secs, sleep_ms};
 use stacks_signer::chainstate::{ProposalEvalConfig, SortitionsView};
 use stacks_signer::signerdb::{BlockInfo, BlockState, ExtraBlockInfo, SignerDb};
+use stacks_signer::v0::SpawnedSigner;
 
 use super::bitcoin_regtest::BitcoinCoreController;
 use crate::config::{EventKeyType, InitialBalance};
@@ -104,6 +106,7 @@ use crate::tests::neon_integrations::{
     get_neighbors, get_pox_info, next_block_and_wait, run_until_burnchain_height, submit_tx,
     submit_tx_fallible, test_observer, wait_for_runloop,
 };
+use crate::tests::signer::SignerTest;
 use crate::tests::{
     gen_random_port, get_chain_info, make_contract_call, make_contract_publish,
     make_contract_publish_versioned, make_stacks_transfer, to_addr,
@@ -3805,8 +3808,13 @@ fn follower_bootup_across_multiple_cycles() {
         .reward_cycle_length
         * 2
     {
+        let commits_before = commits_submitted.load(Ordering::SeqCst);
         next_block_and_process_new_stacks_block(&mut btc_regtest_controller, 60, &coord_channel)
             .unwrap();
+        wait_for(20, || {
+            Ok(commits_submitted.load(Ordering::SeqCst) > commits_before)
+        })
+        .unwrap();
     }
 
     info!("Nakamoto miner has advanced two reward cycles");
@@ -9625,6 +9633,183 @@ fn skip_mining_long_tx() {
     run_loop_thread.join().unwrap();
 }
 
+/// Verify that a node in which there is no prepare-phase block can be recovered by
+/// live-instantiating shadow tenures in the prepare phase
+#[test]
+#[ignore]
+fn test_shadow_recovery() {
+    if env::var("BITCOIND_TEST") != Ok("1".into()) {
+        return;
+    }
+
+    let mut signer_test: SignerTest<SpawnedSigner> = SignerTest::new(1, vec![]);
+    signer_test.boot_to_epoch_3();
+
+    let naka_conf = signer_test.running_nodes.conf.clone();
+    let btc_regtest_controller = &mut signer_test.running_nodes.btc_regtest_controller;
+    let coord_channel = signer_test.running_nodes.coord_channel.clone();
+    let commits_submitted = signer_test.running_nodes.commits_submitted.clone();
+
+    let burnchain = naka_conf.get_burnchain();
+
+    // make another tenure
+    next_block_and_mine_commit(
+        btc_regtest_controller,
+        60,
+        &coord_channel,
+        &commits_submitted,
+    )
+    .unwrap();
+
+    let block_height = btc_regtest_controller.get_headers_height();
+    let reward_cycle = btc_regtest_controller
+        .get_burnchain()
+        .block_height_to_reward_cycle(block_height)
+        .unwrap();
+    let prepare_phase_start = btc_regtest_controller
+        .get_burnchain()
+        .pox_constants
+        .prepare_phase_start(
+            btc_regtest_controller.get_burnchain().first_block_height,
+            reward_cycle,
+        );
+
+    let blocks_until_next_rc = prepare_phase_start + 1 - block_height
+        + (btc_regtest_controller
+            .get_burnchain()
+            .pox_constants
+            .prepare_length as u64)
+        + 1;
+
+    // kill the chain by blowing through a prepare phase
+    btc_regtest_controller.bootstrap_chain(blocks_until_next_rc);
+    let target_burn_height = btc_regtest_controller.get_headers_height();
+
+    let burnchain = naka_conf.get_burnchain();
+    let mut sortdb = burnchain.open_sortition_db(true).unwrap();
+    let (mut chainstate, _) = StacksChainState::open(
+        false,
+        CHAIN_ID_TESTNET,
+        &naka_conf.get_chainstate_path_str(),
+        None,
+    )
+    .unwrap();
+
+    wait_for(30, || {
+        let burn_height = get_chain_info(&naka_conf).burn_block_height;
+        if burn_height >= target_burn_height {
+            return Ok(true);
+        }
+        sleep_ms(500);
+        Ok(false)
+    })
+    .unwrap();
+
+    let stacks_height_before = get_chain_info(&naka_conf).stacks_tip_height;
+
+    // TODO: stall block processing; otherwise this test can flake
+    // stop block processing on the node
+    TEST_COORDINATOR_STALL.lock().unwrap().replace(true);
+
+    // fix node
+    let shadow_blocks = shadow_chainstate_repair(&mut chainstate, &mut sortdb).unwrap();
+    assert!(shadow_blocks.len() > 0);
+
+    wait_for(30, || {
+        let Some(info) = get_chain_info_opt(&naka_conf) else {
+            sleep_ms(500);
+            return Ok(false);
+        };
+        Ok(info.stacks_tip_height >= stacks_height_before)
+    })
+    .unwrap();
+
+    TEST_COORDINATOR_STALL.lock().unwrap().replace(false);
+    info!("Beginning post-shadow tenures");
+
+    // revive ATC-C by waiting for commits
+    for _i in 0..4 {
+        btc_regtest_controller.bootstrap_chain(1);
+        sleep_ms(30_000);
+    }
+
+    // make another tenure
+    next_block_and_mine_commit(
+        btc_regtest_controller,
+        60,
+        &coord_channel,
+        &commits_submitted,
+    )
+    .unwrap();
+
+    // all shadow blocks are present and processed
+    let mut shadow_ids = HashSet::new();
+    for sb in shadow_blocks {
+        let (_, processed, orphaned, _) = chainstate
+            .nakamoto_blocks_db()
+            .get_block_processed_and_signed_weight(
+                &sb.header.consensus_hash,
+                &sb.header.block_hash(),
+            )
+            .unwrap()
+            .unwrap();
+        assert!(processed);
+        assert!(!orphaned);
+        shadow_ids.insert(sb.block_id());
+    }
+
+    let tip = NakamotoChainState::get_canonical_block_header(chainstate.db(), &sortdb)
+        .unwrap()
+        .unwrap();
+    let mut cursor = tip.index_block_hash();
+
+    // the chainstate has four parts:
+    // * epoch 2
+    // * epoch 3 prior to failure
+    // * shadow blocks
+    // * epoch 3 after recovery
+    // Make sure they're all there
+
+    let mut has_epoch_3_recovery = false;
+    let mut has_shadow_blocks = false;
+    let mut has_epoch_3_failure = false;
+
+    loop {
+        let header = NakamotoChainState::get_block_header(chainstate.db(), &cursor)
+            .unwrap()
+            .unwrap();
+        if header.anchored_header.as_stacks_epoch2().is_some() {
+            break;
+        }
+
+        let header = header.anchored_header.as_stacks_nakamoto().clone().unwrap();
+
+        if header.is_shadow_block() {
+            assert!(shadow_ids.contains(&header.block_id()));
+        } else {
+            assert!(!shadow_ids.contains(&header.block_id()));
+        }
+
+        if !header.is_shadow_block() && !has_epoch_3_recovery {
+            has_epoch_3_recovery = true;
+        } else if header.is_shadow_block() && has_epoch_3_recovery && !has_shadow_blocks {
+            has_shadow_blocks = true;
+        } else if !header.is_shadow_block()
+            && has_epoch_3_recovery
+            && has_shadow_blocks
+            && !has_epoch_3_failure
+        {
+            has_epoch_3_failure = true;
+        }
+
+        cursor = header.parent_block_id;
+    }
+
+    assert!(has_epoch_3_recovery);
+    assert!(has_shadow_blocks);
+    assert!(has_epoch_3_failure);
+}
+
 #[test]
 #[ignore]
 /// This test is testing that the clarity cost spend down works as expected,
diff --git a/testnet/stacks-node/src/tests/signer/mod.rs b/testnet/stacks-node/src/tests/signer/mod.rs
index a9053d8c5d..946a566c13 100644
--- a/testnet/stacks-node/src/tests/signer/mod.rs
+++ b/testnet/stacks-node/src/tests/signer/mod.rs
@@ -111,7 +111,7 @@ pub struct SignerTest<S> {
 }
 
 impl<S: Signer<T> + Send + 'static, T: SignerEventTrait + 'static> SignerTest<SpawnedSigner<S, T>> {
-    fn new(num_signers: usize, initial_balances: Vec<(StacksAddress, u64)>) -> Self {
+    pub fn new(num_signers: usize, initial_balances: Vec<(StacksAddress, u64)>) -> Self {
         Self::new_with_config_modifications(
             num_signers,
             initial_balances,
diff --git a/testnet/stacks-node/src/tests/signer/v0.rs b/testnet/stacks-node/src/tests/signer/v0.rs
index b55b9bafe6..ced65c6a15 100644
--- a/testnet/stacks-node/src/tests/signer/v0.rs
+++ b/testnet/stacks-node/src/tests/signer/v0.rs
@@ -227,7 +227,7 @@ impl SignerTest<SpawnedSigner> {
     }
 
     /// Run the test until the epoch 3 boundary
-    fn boot_to_epoch_3(&mut self) {
+    pub fn boot_to_epoch_3(&mut self) {
         boot_to_epoch_3_reward_set(
             &self.running_nodes.conf,
             &self.running_nodes.blocks_processed,