diff --git a/qa/rpc-tests/spark_batching.py b/qa/rpc-tests/spark_batching.py
index 8367b200ba..0e3dce68cc 100755
--- a/qa/rpc-tests/spark_batching.py
+++ b/qa/rpc-tests/spark_batching.py
@@ -1,10 +1,9 @@
 #!/usr/bin/env python3
-"""Test deferred Spark batch proof verification (-batching) across reindex.
+"""Test historical and recent Spark batch proof verification across reindex.
 
-All blocks are mined with timestamps more than a day in the past, so a
-later reindex takes the old-block batching path: Spark spend proofs are
-collected into the batch container and must batch-verify before the node
-persists validation state and clears the reindex flag.
+Old blocks share an accumulated batch; spends in a recent block share a
+separate per-block batch. Reindex must verify both before completing and
+reach the same chain and wallet balance with batching disabled.
 """
 import os
 import time
@@ -92,10 +91,27 @@ def run_test(self):
         assert BATCH_SUCCESS_LOG in log, \
             "batched reindex did not batch verify Spark proofs"
         assert "Spark batch verification failed." not in log
+        assert_equal(log.count(BATCH_SUCCESS_LOG), 1)
+        self.wait_spark_balance(spark_balance)
+
+        # Put two spends in one recent block. Reindex clears the mempool proof
+        # cache, so these must share one per-block batch, separate from the
+        # accumulated historical batch above.
+        set_node_times(self.nodes, int(time.time()))
+        for amount in (1, 2):
+            self.nodes[0].spendspark({self.nodes[0].getnewaddress(): {
+                "amount": amount, "subtractFee": False}})
+        self.nodes[0].generate(6)
+        spark_balance = self.nodes[0].getsparkbalance()
+        self.reindex(batching=True)
+        log = self.read_debug_log()
+        assert_equal(log.count(BATCH_SUCCESS_LOG), 2)
+        assert "Spark batch verification failed." not in log
         self.wait_spark_balance(spark_balance)
 
         # Control: block-by-block verification reaches the same chain.
         self.reindex(batching=False)
+        assert BATCH_SUCCESS_LOG not in self.read_debug_log()
         self.wait_spark_balance(spark_balance)
 
         print("Success")
diff --git a/src/batchproof_container.cpp b/src/batchproof_container.cpp
index 79e68ef06f..a469b67109 100644
--- a/src/batchproof_container.cpp
+++ b/src/batchproof_container.cpp
@@ -2,12 +2,119 @@
 #include "ui_interface.h"
 #include "spark/state.h"
 #include "util.h"
+#include "validation.h"
 
 #include <boost/filesystem.hpp>
+#include <unordered_map>
 
-extern bool fReindex;
+namespace {
 
-std::unique_ptr<BatchProofContainer> BatchProofContainer::instance;
+using CoverSets = std::unordered_map<int32_t, std::vector<spark::Coin>>;
+
+// Load each deployed state group once, shared by current and historical proofs.
+// The snapshot owns the coins, so verification does not access chain state.
+CoverSets LoadCoverSets(
+    std::vector<spark::SpendTransaction>& transactions,
+    std::vector<spark::SpendTransaction>& historicalTransactions)
+{
+    AssertLockHeld(cs_main);
+    CoverSets coverSets;
+    for (auto* batch : {&transactions, &historicalTransactions}) {
+        for (auto& tx : *batch) {
+            for (uint64_t id : tx.getCoinGroupIds()) {
+                const int32_t stateId = static_cast<int32_t>(id);
+                auto entry = coverSets.try_emplace(stateId);
+                if (entry.second) {
+                    uint256 blockHash;
+                    std::vector<unsigned char> setHash;
+                    // Consensus can reference newer coins than the wallet's
+                    // confirmation window. Snapshot the whole active group.
+                    spark::CSparkState::GetState()->GetCoinSetForSpend(
+                        &chainActive, chainActive.Height(), stateId,
+                        blockHash, entry.first->second, setHash);
+                }
+            }
+        }
+    }
+    return coverSets;
+}
+
+bool VerifySparkBatch(
+    const std::vector<spark::SpendTransaction>& sparkTransactions,
+    const std::vector<uint256>& sparkTxIds,
+    const std::vector<spark::SpendTransaction>& historicalSparkTransactions,
+    const std::vector<uint256>& historicalSparkTxIds,
+    const CoverSets& coverSets)
+{
+    if (sparkTransactions.empty() && historicalSparkTransactions.empty())
+        return true;
+
+    LogPrintf("Spark batch verification started.\n");
+    uiInterface.UpdateProgressBarLabel("Batch verifying Spark Proofs...");
+
+    const spark::SpendTransaction::CoverSetProvider coverSetProvider =
+        [&coverSets](uint64_t id) -> const std::vector<spark::Coin>& {
+            return coverSets.at(static_cast<int32_t>(id));
+        };
+    auto* params = spark::Params::get_default();
+
+    bool passed = true;
+    try {
+        if (!sparkTransactions.empty()) {
+            passed = spark::SpendTransaction::verify(
+                params, sparkTransactions, coverSetProvider);
+        }
+        if (passed && !historicalSparkTransactions.empty()) {
+            passed = spark::SpendTransaction::verifyHistorical(
+                params, historicalSparkTransactions, coverSetProvider);
+        }
+    } catch (const std::bad_alloc&) {
+        throw;
+    } catch (const std::exception &) {
+        passed = false;
+    }
+
+    if (!passed) {
+        // Re-verify the retained proofs individually so the operator can see
+        // exactly which spends are invalid without a diagnostic reindex.
+        for (std::size_t i = 0; i < sparkTransactions.size(); ++i) {
+            bool fProofValid;
+            try {
+                fProofValid = spark::SpendTransaction::verify(
+                    params, {sparkTransactions[i]}, coverSetProvider);
+            } catch (const std::bad_alloc&) {
+                throw;
+            } catch (const std::exception &) {
+                fProofValid = false;
+            }
+            if (!fProofValid) {
+                LogPrintf("Spark batch verification failed for spend transaction %s.\n", sparkTxIds[i].ToString());
+            }
+        }
+        for (std::size_t i = 0; i < historicalSparkTransactions.size(); ++i) {
+            bool fProofValid;
+            try {
+                fProofValid = spark::SpendTransaction::verifyHistorical(
+                    params, {historicalSparkTransactions[i]}, coverSetProvider);
+            } catch (const std::bad_alloc&) {
+                throw;
+            } catch (const std::exception &) {
+                fProofValid = false;
+            }
+            if (!fProofValid) {
+                LogPrintf("Spark batch verification failed for spend transaction %s.\n", historicalSparkTxIds[i].ToString());
+            }
+        }
+        LogPrintf("Spark batch verification failed.\n");
+        return false;
+    }
+
+    LogPrintf("Spark batch verification finished successfully.\n");
+    return true;
+}
+
+
+} // namespace
 
 static boost::filesystem::path RecoveryMarkerPath()
 {
@@ -36,70 +143,157 @@ void BatchProofContainer::RemoveRecoveryMarker()
     boost::filesystem::remove(RecoveryMarkerPath());
 }
 
-BatchProofContainer* BatchProofContainer::get_instance() {
-    if (instance) {
-        return instance.get();
-    } else {
-        instance.reset(new BatchProofContainer());
-        return instance.get();
-    }
+
+BatchProofContainer* BatchProofContainer::get_instance()
+{
+    static BatchProofContainer instance;
+    return &instance;
 }
 
-void BatchProofContainer::init() {
+void BatchProofContainer::init(Mode nextMode)
+{
+    LOCK(cs_main);
     tempSparkTransactions.clear();
     tempSparkTxIds.clear();
     tempHistoricalSparkTransactions.clear();
     tempHistoricalSparkTxIds.clear();
-    if (fCollectProofs)
+    mode = nextMode;
+    if (mode == Mode::Deferred)
         WriteRecoveryMarker();
 }
 
-void BatchProofContainer::finalize() {
-    if (fCollectProofs) {
-        sparkTransactions.insert(sparkTransactions.end(), tempSparkTransactions.begin(), tempSparkTransactions.end());
-        sparkTxIds.insert(sparkTxIds.end(), tempSparkTxIds.begin(), tempSparkTxIds.end());
-        historicalSparkTransactions.insert(
-            historicalSparkTransactions.end(),
-            tempHistoricalSparkTransactions.begin(),
-            tempHistoricalSparkTransactions.end());
-        historicalSparkTxIds.insert(
-            historicalSparkTxIds.end(),
-            tempHistoricalSparkTxIds.begin(),
-            tempHistoricalSparkTxIds.end());
+void BatchProofContainer::finalize()
+{
+    LOCK(cs_main);
+    assert(mode != Mode::Block);
+    if (mode == Mode::Deferred) {
+        const auto size = sparkTransactions.size();
+        const auto historicalSize = historicalSparkTransactions.size();
+        try {
+            sparkTransactions.insert(sparkTransactions.end(), tempSparkTransactions.begin(), tempSparkTransactions.end());
+            sparkTxIds.insert(sparkTxIds.end(), tempSparkTxIds.begin(), tempSparkTxIds.end());
+            historicalSparkTransactions.insert(historicalSparkTransactions.end(), tempHistoricalSparkTransactions.begin(), tempHistoricalSparkTransactions.end());
+            historicalSparkTxIds.insert(historicalSparkTxIds.end(), tempHistoricalSparkTxIds.begin(), tempHistoricalSparkTxIds.end());
+        } catch (...) {
+            // Keep proof/txid pairs aligned; temps still own the whole block.
+            sparkTransactions.erase(sparkTransactions.begin() + size, sparkTransactions.end());
+            sparkTxIds.erase(sparkTxIds.begin() + size, sparkTxIds.end());
+            historicalSparkTransactions.erase(historicalSparkTransactions.begin() + historicalSize, historicalSparkTransactions.end());
+            historicalSparkTxIds.erase(historicalSparkTxIds.begin() + historicalSize, historicalSparkTxIds.end());
+            throw;
+        }
+        ++generation;
     }
-    tempSparkTransactions.clear();
-    tempSparkTxIds.clear();
-    tempHistoricalSparkTransactions.clear();
-    tempHistoricalSparkTxIds.clear();
-    fCollectProofs = false;
+    init();
 }
 
-bool BatchProofContainer::verify_pending() {
-    bool passed = true;
-    if (!fCollectProofs) {
-        init();
-        passed = batch_spark();
-        if (!passed)
+bool BatchProofContainer::is_deferred() const
+{
+    LOCK(cs_main);
+    return mode == Mode::Deferred;
+}
+
+bool BatchProofContainer::verify_block_batch()
+{
+    AssertLockHeld(cs_main);
+    if (mode != Mode::Block)
+        return true;
+
+    const auto coverSets = LoadCoverSets(tempSparkTransactions, tempHistoricalSparkTransactions);
+    const bool passed = VerifySparkBatch(
+        tempSparkTransactions, tempSparkTxIds,
+        tempHistoricalSparkTransactions, tempHistoricalSparkTxIds, coverSets);
+    init();
+    return passed;
+}
+
+bool BatchProofContainer::verify_pending()
+{
+    AssertLockNotHeld(cs_main);
+    LOCK(cs_verify);
+    for (;;) {
+        std::vector<spark::SpendTransaction> transactions, historicalTransactions;
+        std::vector<uint256> txIds, historicalTxIds;
+        CoverSets coverSets;
+        uint64_t snapshotGeneration;
+        const CBlockIndex* snapshotTip;
+        {
+            LOCK(cs_main);
+            if (mode != Mode::Disabled)
+                return true;
+            if (fBatchFailed)
+                return false;
+            if (sparkTransactions.empty() && historicalSparkTransactions.empty()) {
+                if (!fReindex)
+                    RemoveRecoveryMarker();
+                return true;
+            }
+            // Retain the canonical proofs. Disconnects can remove them, and an
+            // exception cannot destroy the only copy of an unchecked batch.
+            transactions = sparkTransactions;
+            txIds = sparkTxIds;
+            historicalTransactions = historicalSparkTransactions;
+            historicalTxIds = historicalSparkTxIds;
+            coverSets = LoadCoverSets(transactions, historicalTransactions);
+            snapshotGeneration = generation;
+            snapshotTip = chainActive.Tip();
+        }
+
+        const bool passed = VerifySparkBatch(
+            transactions, txIds, historicalTransactions, historicalTxIds, coverSets);
+
+        LOCK(cs_main);
+        if (generation != snapshotGeneration || chainActive.Tip() != snapshotTip)
+            continue;
+        if (!passed) {
+            fBatchFailed = true;
             WriteRecoveryMarker();
-        else if (!fReindex)
+            return false;
+        }
+        sparkTransactions.clear();
+        sparkTxIds.clear();
+        historicalSparkTransactions.clear();
+        historicalSparkTxIds.clear();
+        ++generation;
+        if (!fReindex)
             RemoveRecoveryMarker();
+        return true;
     }
-    fCollectProofs = false;
-    return passed;
 }
 
-void BatchProofContainer::add(const spark::SpendTransaction& tx, const uint256& txHash) {
-    tempSparkTransactions.push_back(tx);
+bool BatchProofContainer::add(const spark::SpendTransaction& tx, const uint256& txHash)
+{
+    LOCK(cs_main);
+    if (mode == Mode::Disabled)
+        return false;
     tempSparkTxIds.push_back(txHash);
+    try {
+        tempSparkTransactions.push_back(tx);
+    } catch (...) {
+        tempSparkTxIds.pop_back();
+        throw;
+    }
+    return true;
 }
 
-void BatchProofContainer::addHistorical(
-    const spark::SpendTransaction& tx, const uint256& txHash) {
-    tempHistoricalSparkTransactions.push_back(tx);
+bool BatchProofContainer::addHistorical(const spark::SpendTransaction& tx, const uint256& txHash)
+{
+    LOCK(cs_main);
+    if (mode == Mode::Disabled)
+        return false;
     tempHistoricalSparkTxIds.push_back(txHash);
+    try {
+        tempHistoricalSparkTransactions.push_back(tx);
+    } catch (...) {
+        tempHistoricalSparkTxIds.pop_back();
+        throw;
+    }
+    return true;
 }
 
-void BatchProofContainer::remove(const spark::SpendTransaction& tx) {
+void BatchProofContainer::remove(const spark::SpendTransaction& tx)
+{
+    LOCK(cs_main);
     bool fBatchChanged = false;
     for (std::size_t i = sparkTransactions.size(); i-- > 0;) {
         if (sparkTransactions[i].getUsedLTags() == tx.getUsedLTags()) {
@@ -116,81 +310,7 @@ void BatchProofContainer::remove(const spark::SpendTransaction& tx) {
         }
     }
     if (fBatchChanged) {
-        // the pending batch changed, so a previous failure verdict no longer applies
+        ++generation;
         fBatchFailed = false;
     }
 }
-
-bool BatchProofContainer::batch_spark() {
-    if (sparkTransactions.empty() && historicalSparkTransactions.empty())
-        return true;
-    if (fBatchFailed)
-        return false;
-
-    LogPrintf("Spark batch verification started.\n");
-    uiInterface.UpdateProgressBarLabel("Batch verifying Spark Proofs...");
-
-    spark::CSparkState* sparkState = spark::CSparkState::GetState();
-    std::vector<spark::Coin> loadedCoverSet;
-    const spark::SpendTransaction::CoverSetProvider coverSetProvider =
-        [sparkState, &loadedCoverSet](uint64_t id)
-            -> const std::vector<spark::Coin>& {
-        loadedCoverSet.clear();
-        sparkState->GetCoinSet(static_cast<int32_t>(id), loadedCoverSet);
-        return loadedCoverSet;
-    };
-    auto* params = spark::Params::get_default();
-
-    bool passed = true;
-    try {
-        if (!sparkTransactions.empty()) {
-            passed = spark::SpendTransaction::verify(
-                params, sparkTransactions, coverSetProvider);
-        }
-        if (passed && !historicalSparkTransactions.empty()) {
-            passed = spark::SpendTransaction::verifyHistorical(
-                params, historicalSparkTransactions, coverSetProvider);
-        }
-    } catch (const std::exception &) {
-        passed = false;
-    }
-
-    if (!passed) {
-        // Re-verify the retained proofs individually so the operator can see
-        // exactly which spends are invalid without a diagnostic reindex.
-        for (std::size_t i = 0; i < sparkTransactions.size(); ++i) {
-            bool fProofValid;
-            try {
-                fProofValid = spark::SpendTransaction::verify(
-                    params, {sparkTransactions[i]}, coverSetProvider);
-            } catch (const std::exception &) {
-                fProofValid = false;
-            }
-            if (!fProofValid) {
-                LogPrintf("Spark batch verification failed for spend transaction %s.\n", sparkTxIds[i].ToString());
-            }
-        }
-        for (std::size_t i = 0; i < historicalSparkTransactions.size(); ++i) {
-            bool fProofValid;
-            try {
-                fProofValid = spark::SpendTransaction::verifyHistorical(
-                    params, {historicalSparkTransactions[i]}, coverSetProvider);
-            } catch (const std::exception &) {
-                fProofValid = false;
-            }
-            if (!fProofValid) {
-                LogPrintf("Spark batch verification failed for spend transaction %s.\n", historicalSparkTxIds[i].ToString());
-            }
-        }
-        LogPrintf("Spark batch verification failed.\n");
-        fBatchFailed = true;
-        return false;
-    }
-
-    LogPrintf("Spark batch verification finished successfully.\n");
-    sparkTransactions.clear();
-    sparkTxIds.clear();
-    historicalSparkTransactions.clear();
-    historicalSparkTxIds.clear();
-    return true;
-}
diff --git a/src/batchproof_container.h b/src/batchproof_container.h
index 8a8deecbba..4fdc8ceddf 100644
--- a/src/batchproof_container.h
+++ b/src/batchproof_container.h
@@ -4,21 +4,29 @@
 #include <memory>
 #include "chain.h"
 #include "libspark/spend_transaction.h"
+#include "sync.h"
 
 extern CChain chainActive;
 
 class BatchProofContainer {
 public:
+    enum class Mode { Disabled, Deferred, Block };
+
     static BatchProofContainer* get_instance();
 
-    void init();
+    void init(Mode mode = Mode::Disabled);
 
     void finalize();
 
+    bool is_deferred() const;
+
+    /** Verify this block's temps under cs_main, without touching pending proofs. */
+    bool verify_block_batch();
+
     /**
-     * Verify the finalized pending Spark batch when proofs are not being
-     * collected. Matches master's verify() gate: a no-op while fCollectProofs
-     * is set, so IBD keeps accumulating until a recent tip.
+     * Verify a retained snapshot, retrying if the pending batch or active tip
+     * changes. Concurrent callers wait for the current verifier. Call without
+     * cs_main: only snapshot preparation and verdict publication hold it.
      *
      * @return true if collecting, if no batch is pending, or if the batch
      *         verifies; false on verification failure (pending proofs kept).
@@ -28,16 +36,16 @@ class BatchProofContainer {
     static bool HasRecoveryMarker();
     static void RemoveRecoveryMarker();
 
-    void add(const spark::SpendTransaction& tx, const uint256& txHash);
-    void addHistorical(const spark::SpendTransaction& tx, const uint256& txHash);
+    bool add(const spark::SpendTransaction& tx, const uint256& txHash);
+    bool addHistorical(const spark::SpendTransaction& tx, const uint256& txHash);
     void remove(const spark::SpendTransaction& tx);
-public:
-    bool fCollectProofs = 0;
 
 private:
-    bool batch_spark();
-
-    static std::unique_ptr<BatchProofContainer> instance;
+    // Lock order: cs_verify -> cs_main. Collection never takes cs_verify.
+    CCriticalSection cs_verify;
+    // All remaining mutable state is protected by cs_main.
+    Mode mode = Mode::Disabled;
+    uint64_t generation = 0;
     // a pending batch failed verification; fail fast until the batch changes
     bool fBatchFailed = false;
     // temp spark transaction proofs and the txids they came from
diff --git a/src/init.cpp b/src/init.cpp
index 2c1e4b4f9a..dfbd52e89a 100644
--- a/src/init.cpp
+++ b/src/init.cpp
@@ -268,13 +268,6 @@ void Shutdown()
     StopHTTPServer();
     llmq::StopLLMQSystem();
 
-    {
-        LOCK(cs_main);
-        BatchProofContainer::get_instance()->finalize();
-        CValidationState state;
-        VerifyPendingSparkBatch(state, "shutdown");
-    }
-
 #ifdef ENABLE_WALLET
     if (pwalletMain)
         pwalletMain->Flush(false);
@@ -302,6 +295,10 @@ void Shutdown()
     // cleanup; the embedded Tor itself is torn down by process exit.
     g_connman.reset();
     UnregisterNodeSignals(GetNodeSignals());
+    BatchProofContainer::get_instance()->finalize();
+    CValidationState batchState;
+    VerifyPendingSparkBatch(batchState, "shutdown");
+
     if (fDumpMempoolLater)
         DumpMempool();
 
@@ -776,14 +773,11 @@ void ThreadImport(std::vector <boost::filesystem::path> vImportFiles) {
             LoadExternalBlockFile(chainparams, file, &pos);
             nFile++;
         }
-        {
-            LOCK(cs_main);
-            BatchProofContainer::get_instance()->finalize();
-            CValidationState state;
-            if (!VerifyPendingSparkBatch(state, "clearing reindex flag")) {
-                LogPrintf("Reindexing stopped before clearing reindex flag: %s\n", FormatStateMessage(state));
-                return;
-            }
+        BatchProofContainer::get_instance()->finalize();
+        CValidationState state;
+        if (!VerifyPendingSparkBatch(state, "clearing reindex flag")) {
+            LogPrintf("Reindexing stopped before clearing reindex flag: %s\n", FormatStateMessage(state));
+            return;
         }
         pblocktree->WriteReindexing(false);
         fReindex = false;
diff --git a/src/net_processing.cpp b/src/net_processing.cpp
index 6eda344a82..d8177232ac 100644
--- a/src/net_processing.cpp
+++ b/src/net_processing.cpp
@@ -1020,6 +1020,35 @@ static void RelayAddress(const CAddress& addr, bool fReachable, CConnman& connma
 
 void static ProcessGetData(CNode* pfrom, const Consensus::Params& consensusParams, CConnman& connman, const std::atomic<bool>& interruptMsgProc)
 {
+    if (pfrom->fPauseSend || interruptMsgProc)
+        return;
+
+    // At most one block is served below. Activate it before taking the serving
+    // lock: activation may wait for a verifier that needs cs_main to finish.
+    for (const CInv& inv : pfrom->vRecvGetData) {
+        if (inv.type != MSG_BLOCK && inv.type != MSG_FILTERED_BLOCK &&
+            inv.type != MSG_CMPCT_BLOCK && inv.type != MSG_WITNESS_BLOCK)
+            continue;
+        bool activate;
+        {
+            LOCK(cs_main);
+            auto mi = mapBlockIndex.find(inv.hash);
+            activate = mi != mapBlockIndex.end() && mi->second->nChainTx &&
+                !mi->second->IsValid(BLOCK_VALID_SCRIPTS) && mi->second->IsValid(BLOCK_VALID_TREE);
+        }
+        if (activate) {
+            std::shared_ptr<const CBlock> recentBlock;
+            {
+                LOCK(cs_most_recent_block);
+                recentBlock = most_recent_block;
+            }
+            CValidationState state;
+            if (!ActivateBestChain(state, Params(), recentBlock))
+                return;
+        }
+        break;
+    }
+
     std::deque<CInv>::iterator it = pfrom->vRecvGetData.begin();
     std::vector<CInv> vNotFound;
     const CNetMsgMaker msgMaker(pfrom->GetSendVersion());
@@ -1043,21 +1072,6 @@ void static ProcessGetData(CNode* pfrom, const Consensus::Params& consensusParam
                 BlockMap::iterator mi = mapBlockIndex.find(inv.hash);
                 if (mi != mapBlockIndex.end())
                 {
-                    if (mi->second->nChainTx && !mi->second->IsValid(BLOCK_VALID_SCRIPTS) &&
-                            mi->second->IsValid(BLOCK_VALID_TREE)) {
-                        // If we have the block and all of its parents, but have not yet validated it,
-                        // we might be in the middle of connecting it (ie in the unlock of cs_main
-                        // before ActivateBestChain but after AcceptBlock).
-                        // In this case, we need to run ActivateBestChain prior to checking the relay
-                        // conditions below.
-                        std::shared_ptr<const CBlock> a_recent_block;
-                        {
-                            LOCK(cs_most_recent_block);
-                            a_recent_block = most_recent_block;
-                        }
-                        CValidationState dummy;
-                        ActivateBestChain(dummy, Params(), a_recent_block);
-                    }
                     if (chainActive.Contains(mi->second)) {
                         send = true;
                     } else {
@@ -2082,7 +2096,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr
             inv.type = State(pfrom->GetId())->fWantsCmpctWitness ? MSG_WITNESS_BLOCK : MSG_BLOCK;
             inv.hash = req.blockhash;
             pfrom->vRecvGetData.push_back(inv);
-            ProcessGetData(pfrom, chainparams.GetConsensus(), connman, interruptMsgProc);
+            // ProcessMessages serves this queued request without cs_main.
             return true;
         }
 
diff --git a/src/spark/state.cpp b/src/spark/state.cpp
index 45988c3c3e..132037cbf4 100644
--- a/src/spark/state.cpp
+++ b/src/spark/state.cpp
@@ -1066,7 +1066,7 @@ bool CheckSparkSpendTransaction(
     std::unordered_map<uint64_t, CoverSetData> cover_set_data;
 
     BatchProofContainer* batchProofContainer = BatchProofContainer::get_instance();
-    bool useBatching = batchProofContainer->fCollectProofs && !isVerifyDB && !isCheckWallet && sparkTxInfo && !sparkTxInfo->fInfoIsComplete;
+    const bool fCanBatch = !isVerifyDB && !isCheckWallet && sparkTxInfo && !sparkTxInfo->fInfoIsComplete;
 
     for (const auto& idAndHash : idAndBlockHashes) {
         const uint64_t wireGroupId = idAndHash.first;
@@ -1212,29 +1212,21 @@ bool CheckSparkSpendTransaction(
         return loadedCoverSet;
     };
     
-    // if we are collecting proofs, skip verification and collect proofs
-    // add proofs into container
-    if (useBatching) {
+    // Recent blocks retain the mempool cache fast path. Historical batches
+    // still collect every proof, and VerifyDB always re-verifies its own view.
+    bool haveCachedSuccess = false;
+    if (!isVerifyDB && (!fCanBatch || !batchProofContainer->is_deferred())) {
+        LOCK(cs_checkedSparkSpendTransactions);
+        haveCachedSuccess = gCheckedSparkSpendTransactions.exists(hashTx);
+    }
+    const bool fAddedToBatch = !haveCachedSuccess && fCanBatch && ((isChaumV2 || requireChaumV1SingleInput)
+        ? batchProofContainer->add(*spend, hashTx)
+        : batchProofContainer->addHistorical(*spend, hashTx));
+    if (haveCachedSuccess || fAddedToBatch) {
         passVerify = true;
-        if (isChaumV2 || requireChaumV1SingleInput) {
-            batchProofContainer->add(*spend, hashTx);
-        } else {
-            batchProofContainer->addHistorical(*spend, hashTx);
-        }
     } else {
         try {
-            bool haveCachedSuccess = false;
-            // The cache is txid-only. VerifyDB reconstructs cover sets at the
-            // historical height, so a mempool success must not skip re-verify.
-            if (!isVerifyDB) {
-                LOCK(cs_checkedSparkSpendTransactions);
-                haveCachedSuccess = gCheckedSparkSpendTransactions.exists(hashTx);
-            }
-            if (haveCachedSuccess) {
-                LogPrintf("CheckSparkSpendTransaction: already checked tx %s\n", hashTx.ToString());
-                passVerify = true;
-            }
-            else if (isMempoolAcceptance) {
+            if (isMempoolAcceptance) {
                 passVerify = spark::SpendTransaction::verify(
                     spark::Params::get_default(),
                     {*spend},
diff --git a/src/sync.cpp b/src/sync.cpp
index a7450a013b..ed5baedd7c 100644
--- a/src/sync.cpp
+++ b/src/sync.cpp
@@ -170,6 +170,9 @@ void AssertLockHeldInternal(const char* pszName, const char* pszFile, int nLine,
 
 void AssertLockNotHeldInternal(const char* pszName, const char* pszFile, int nLine, void* cs)
 {
+    // A thread that has never acquired a tracked lock has no lock stack yet.
+    if (!lockstack.get())
+        return;
     for (const std::pair<void*, CLockLocation>& i : *lockstack) {
         if (i.first == cs) {
             fprintf(stderr, "Assertion failed: lock %s held in %s:%i; locks held:\n%s", pszName, pszFile, nLine, LocksHeld().c_str());
diff --git a/src/test/evo_deterministicmns_tests.cpp b/src/test/evo_deterministicmns_tests.cpp
index 3037e8c4de..3ecaf377a1 100644
--- a/src/test/evo_deterministicmns_tests.cpp
+++ b/src/test/evo_deterministicmns_tests.cpp
@@ -355,12 +355,12 @@ BOOST_FIXTURE_TEST_CASE(dip3_activation, TestChainDIP3BeforeActivationSetup)
 
     // This block should activate DIP3
     CreateAndProcessBlock({}, coinbaseKey);
-    LOCK(cs_main);
     BOOST_ASSERT(chainActive.Height() == nHeight + 1);
 
     // Mining a block with a DIP3 transaction should succeed now
     block = std::make_shared<CBlock>(CreateBlock(txns, coinbaseKey));
     ProcessNewBlock(Params(), block, true, nullptr);
+    LOCK(cs_main);
     deterministicMNManager->UpdatedBlockTip(chainActive.Tip());
 
     BOOST_ASSERT(chainActive.Height() == nHeight + 2);
diff --git a/src/test/mtp_trans_tests.cpp b/src/test/mtp_trans_tests.cpp
index d10d397d52..e7c593d8da 100644
--- a/src/test/mtp_trans_tests.cpp
+++ b/src/test/mtp_trans_tests.cpp
@@ -130,9 +130,8 @@ BOOST_AUTO_TEST_CASE(mtp_transition)
     b = CreateAndProcessBlock(scriptPubKeyMtp, mtp);
     BOOST_CHECK_MESSAGE(previousHeight == chainActive.Height() - 1, "Block not connected");
     coinbaseTxns.push_back(*b.vtx[0]);
-    LOCK(cs_main);
     {
-        LOCK(pwalletMain->cs_wallet);
+        LOCK2(cs_main, pwalletMain->cs_wallet);
         pwalletMain->AddToWalletIfInvolvingMe(*b.vtx[0], chainActive.Tip(), 0, true);
     }
 
@@ -140,6 +139,7 @@ BOOST_AUTO_TEST_CASE(mtp_transition)
     //Disconnect MTP block
     BOOST_CHECK_MESSAGE(DisconnectBlocks(1), "Block disconnect failed");
     {
+        LOCK(cs_main);
         CValidationState state;
         const CChainParams& chainparams = Params();
         InvalidateBlock(state, chainparams, mapBlockIndex[b.GetHash()]);
@@ -157,7 +157,7 @@ BOOST_AUTO_TEST_CASE(mtp_transition)
     b = CreateAndProcessBlock(scriptPubKeyMtp, mtp);
     coinbaseTxns.push_back(*b.vtx[0]);
     {
-        LOCK(pwalletMain->cs_wallet);
+        LOCK2(cs_main, pwalletMain->cs_wallet);
         pwalletMain->AddToWalletIfInvolvingMe(*b.vtx[0], chainActive.Tip(), 0, true);
     }
 
@@ -171,7 +171,7 @@ BOOST_AUTO_TEST_CASE(mtp_transition)
     b = CreateAndProcessBlock(scriptPubKeyMtp, mtp);
     coinbaseTxns.push_back(*b.vtx[0]);
     {
-        LOCK(pwalletMain->cs_wallet);
+        LOCK2(cs_main, pwalletMain->cs_wallet);
         pwalletMain->AddToWalletIfInvolvingMe(*b.vtx[0], chainActive.Tip(), 0, true);
     }
 
diff --git a/src/test/spark_batch_test.cpp b/src/test/spark_batch_test.cpp
index 71976dc777..8bbd74e867 100644
--- a/src/test/spark_batch_test.cpp
+++ b/src/test/spark_batch_test.cpp
@@ -4,8 +4,12 @@
 #include "../wallet/wallet.h"
 #include "fixtures.h"
 #include "test_bitcoin.h"
+#include "../ui_interface.h"
 
 #include <boost/test/unit_test.hpp>
+#include <atomic>
+#include <chrono>
+#include <future>
 
 BOOST_FIXTURE_TEST_SUITE(spark_batch_tests, SparkTestingSetup)
 
@@ -41,8 +45,7 @@ BOOST_AUTO_TEST_CASE(spark_batch_fail_closed)
         LOCK(cs_main);
         CValidationState state;
         spark::CSparkTxInfo info;
-        container->fCollectProofs = true;
-        container->init();
+        container->init(BatchProofContainer::Mode::Deferred);
         BOOST_CHECK(spark::CheckSparkTransaction(
             spendTx, state, spendTx.GetHash(), false, chainActive.Height(), false, true, &info));
         container->finalize();
@@ -62,8 +65,7 @@ BOOST_AUTO_TEST_CASE(spark_batch_fail_closed)
     BOOST_REQUIRE(invalidSpend.getUsedLTags() != spark::ParseSparkSpend(spendTx).getUsedLTags());
 
     collectSpend();
-    container->fCollectProofs = true;
-    container->init();
+    container->init(BatchProofContainer::Mode::Deferred);
     container->add(invalidSpend, spendTxB.GetHash());
     container->finalize();
 
@@ -92,4 +94,82 @@ BOOST_AUTO_TEST_CASE(spark_batch_fail_closed)
     BOOST_CHECK(container->verify_pending());
 }
 
+BOOST_AUTO_TEST_CASE(spark_batch_concurrent_verification)
+{
+    GenerateBlocks(501);
+    std::vector<CMutableTransaction> mintTxs;
+    GenerateMints({10 * COIN, 20 * COIN}, mintTxs);
+    GenerateBlock(mintTxs);
+    GenerateBlocks(6);
+
+    CAmount fee;
+    const CTransaction firstSpend(*pwalletMain->CreateSparkSpendTransaction(
+        {{GetScriptForDestination(GenerateAddress().GetID()), COIN, false}}, {}, fee, nullptr).tx);
+    const CTransaction secondSpend(*pwalletMain->CreateSparkSpendTransaction(
+        {{GetScriptForDestination(GenerateAddress().GetID()), 15 * COIN, false}}, {}, fee, nullptr).tx);
+    auto* container = BatchProofContainer::get_instance();
+    auto collect = [&](const CTransaction& tx) {
+        LOCK(cs_main);
+        CValidationState state;
+        spark::CSparkTxInfo info;
+        container->init(BatchProofContainer::Mode::Deferred);
+        const bool collected = spark::CheckSparkTransaction(
+            tx, state, tx.GetHash(), false, chainActive.Height(), false, true, &info);
+        container->finalize();
+        return collected;
+    };
+    BOOST_REQUIRE(collect(firstSpend));
+
+    std::promise<void> started, release, secondStarted;
+    auto ready = started.get_future();
+    auto released = release.get_future();
+    auto secondReady = secondStarted.get_future();
+    std::atomic<int> snapshots{0};
+    std::atomic<bool> timedOut{false};
+    boost::signals2::scoped_connection pause = uiInterface.UpdateProgressBarLabel.connect(
+        [&](const std::string&) {
+            if (snapshots.fetch_add(1) == 0) {
+                started.set_value();
+                timedOut = released.wait_for(std::chrono::seconds(10)) != std::future_status::ready;
+            }
+        });
+    auto first = std::async(std::launch::async, [&] { return container->verify_pending(); });
+    const bool paused = ready.wait_for(std::chrono::seconds(5)) == std::future_status::ready;
+    BOOST_CHECK(paused);
+    if (paused) {
+        // Verification releases cs_main, and a disconnect can still remove the
+        // retained proof. Replace it with a different, equally-sized batch.
+        container->remove(spark::ParseSparkSpend(firstSpend));
+        BOOST_CHECK(collect(secondSpend));
+    }
+    auto second = std::async(std::launch::async, [&] {
+        secondStarted.set_value();
+        return container->verify_pending();
+    });
+    secondReady.wait();
+    BOOST_CHECK(second.wait_for(std::chrono::milliseconds(100)) == std::future_status::timeout);
+    BOOST_CHECK(BatchProofContainer::HasRecoveryMarker());
+    release.set_value();
+    BOOST_CHECK(first.get());
+    BOOST_CHECK(second.get());
+    BOOST_CHECK(!timedOut);
+    BOOST_CHECK_EQUAL(snapshots.load(), 2);
+    pause.disconnect();
+    BOOST_CHECK(!BatchProofContainer::HasRecoveryMarker());
+
+    // An allocation failure must leave the pending batch available to retry.
+    BOOST_REQUIRE(collect(secondSpend));
+    snapshots = 0;
+    boost::signals2::scoped_connection failOnce = uiInterface.UpdateProgressBarLabel.connect(
+        [&](const std::string&) {
+            if (snapshots.fetch_add(1) == 0)
+                throw std::bad_alloc();
+        });
+    BOOST_CHECK_THROW(container->verify_pending(), std::bad_alloc);
+    BOOST_CHECK(BatchProofContainer::HasRecoveryMarker());
+    BOOST_CHECK(container->verify_pending());
+    BOOST_CHECK_EQUAL(snapshots.load(), 2);
+    BOOST_CHECK(!BatchProofContainer::HasRecoveryMarker());
+}
+
 BOOST_AUTO_TEST_SUITE_END()
diff --git a/src/test/spark_tests.cpp b/src/test/spark_tests.cpp
index bb1e9366fb..1cd720625c 100644
--- a/src/test/spark_tests.cpp
+++ b/src/test/spark_tests.cpp
@@ -13,6 +13,7 @@
 #include "../miner.h"
 #include "../policy/policy.h"
 #include "../hash.h"
+#include "../ui_interface.h"
 
 #include "test_bitcoin.h"
 #include "fixtures.h"
@@ -471,8 +472,6 @@ BOOST_AUTO_TEST_CASE(connect_and_disconnect_block)
 {
     // util function
     auto reconnect = [](CBlock const &block) {
-        LOCK(cs_main);
-
         std::shared_ptr<CBlock const> sharedBlock =
                 std::make_shared<CBlock const>(block);
 
@@ -1528,7 +1527,6 @@ BOOST_AUTO_TEST_CASE(spark_v2_activation_and_wallet_selection)
         }
         ~ResetActivationHeights()
         {
-            BatchProofContainer::get_instance()->fCollectProofs = false;
             BatchProofContainer::get_instance()->init();
             consensus.nSparkNamesStartBlock = sparkNamesStartBlock;
         }
@@ -1709,8 +1707,7 @@ BOOST_AUTO_TEST_CASE(spark_v2_activation_and_wallet_selection)
         &activeV2Info));
 
     BatchProofContainer* batch = BatchProofContainer::get_instance();
-    batch->init();
-    batch->fCollectProofs = true;
+    batch->init(BatchProofContainer::Mode::Deferred);
     CValidationState batchedV2State;
     CSparkTxInfo batchedV2Info;
     BOOST_REQUIRE(CheckSparkTransaction(
@@ -1976,7 +1973,6 @@ BOOST_AUTO_TEST_CASE(spark_v2_activation_and_wallet_selection)
     }
     BOOST_CHECK(!mempool.exists(v2MultiAtFork.GetHash()));
     {
-        LOCK(cs_main);
         CValidationState reconnectState;
         BOOST_REQUIRE(ActivateBestChain(
             reconnectState,
@@ -2105,13 +2101,11 @@ BOOST_AUTO_TEST_CASE(unbound_cover_set_is_rejected_after_chaum_v2)
     referencedBlock->sparkSetHash.erase(groupId);
 
     BatchProofContainer* batch = BatchProofContainer::get_instance();
-    batch->init();
-    batch->fCollectProofs = true;
+    batch->init(BatchProofContainer::Mode::Deferred);
     struct RestoreBatchCollection {
         BatchProofContainer* batch;
         ~RestoreBatchCollection()
         {
-            batch->fCollectProofs = false;
             batch->init();
         }
     } restoreBatch{batch};
@@ -2970,7 +2964,6 @@ BOOST_AUTO_TEST_CASE(spark_single_input_historical_batch_verification)
         }
         ~ResetBatchAndActivation()
         {
-            batch->fCollectProofs = false;
             batch->init();
         }
     } reset{batch};
@@ -3008,8 +3001,7 @@ BOOST_AUTO_TEST_CASE(spark_single_input_historical_batch_verification)
         parsedAlias.getCoinGroupIds().front() >
         static_cast<uint64_t>(std::numeric_limits<int32_t>::max()));
 
-    batch->init();
-    batch->fCollectProofs = true;
+    batch->init(BatchProofContainer::Mode::Deferred);
     CValidationState historicalState;
     CSparkTxInfo historicalInfo;
     BOOST_REQUIRE(CheckSparkTransaction(
@@ -3040,8 +3032,7 @@ BOOST_AUTO_TEST_CASE(spark_single_input_historical_batch_verification)
         true,
         &legacyAliasInfo));
 
-    batch->init();
-    batch->fCollectProofs = true;
+    batch->init(BatchProofContainer::Mode::Deferred);
     CValidationState legacyBatchState;
     CSparkTxInfo legacyBatchInfo;
     BOOST_REQUIRE(CheckSparkTransaction(
@@ -3060,8 +3051,7 @@ BOOST_AUTO_TEST_CASE(spark_single_input_historical_batch_verification)
     // Exercise the post-single-input batch as well; it has a separate
     // collection and verification path.
     UpdateRegtestSparkSingleInputHeight(chainActive.Height());
-    batch->init();
-    batch->fCollectProofs = true;
+    batch->init(BatchProofContainer::Mode::Deferred);
     CValidationState currentBatchState;
     CSparkTxInfo currentBatchInfo;
     BOOST_REQUIRE(CheckSparkTransaction(
@@ -3108,8 +3098,7 @@ BOOST_AUTO_TEST_CASE(spark_single_input_historical_batch_verification)
     BOOST_REQUIRE(activeAliasState.IsInvalid(activeAliasDoS));
     BOOST_CHECK_EQUAL(activeAliasDoS, 100);
 
-    batch->init();
-    batch->fCollectProofs = true;
+    batch->init(BatchProofContainer::Mode::Deferred);
     CValidationState activeBatchAliasState;
     CSparkTxInfo activeBatchAliasInfo;
     BOOST_CHECK(!CheckSparkTransaction(
@@ -3124,8 +3113,7 @@ BOOST_AUTO_TEST_CASE(spark_single_input_historical_batch_verification)
     batch->finalize();
     BOOST_CHECK(batch->verify_pending());
 
-    batch->init();
-    batch->fCollectProofs = true;
+    batch->init(BatchProofContainer::Mode::Deferred);
     CValidationState activeState;
     CSparkTxInfo activeInfo;
     BOOST_CHECK(!CheckSparkTransaction(
@@ -3142,6 +3130,90 @@ BOOST_AUTO_TEST_CASE(spark_single_input_historical_batch_verification)
     BOOST_CHECK_EQUAL(activeDoS, 100);
 }
 
+BOOST_AUTO_TEST_CASE(recent_spark_blocks_preserve_batch_and_cache_semantics)
+{
+    RestoreSparkActivationHeights resetActivationHeights;
+    auto* batch = BatchProofContainer::get_instance();
+    struct ResetBatch {
+        BatchProofContainer* batch;
+        ~ResetBatch() { batch->init(); }
+    } reset{batch};
+    GenerateBlocks(500);
+
+    for (auto version : {SpendTransactionVersion::V1, SpendTransactionVersion::V2}) {
+        if (version == SpendTransactionVersion::V2)
+            UpdateRegtestSparkChaumV2Height(chainActive.Height() + 1);
+
+        std::vector<CMutableTransaction> mintTransactions;
+        const auto mints = GenerateMints({5 * COIN, 10 * COIN}, mintTransactions);
+        mempool.clear();
+        BOOST_REQUIRE(GenerateBlock(mintTransactions));
+        BOOST_REQUIRE_EQUAL(mints.size(), 2U);
+
+        // These valid proofs reference the latest block, outside the wallet's
+        // confirmation window. Batching must use the same consensus cover set.
+        const CTransaction first = GenerateCustomSparkSpend(
+            {pwalletMain->sparkWallet->getMintMeta(mints[0].k)}, COIN,
+            0, version, 0, chainActive.Height());
+        const CTransaction second = GenerateCustomSparkSpend(
+            {pwalletMain->sparkWallet->getMintMeta(mints[1].k)}, 2 * COIN,
+            0, version, 0, chainActive.Height());
+        ClearSparkSpendProofCache();
+        int verifications = 0;
+        boost::signals2::scoped_connection observe = uiInterface.UpdateProgressBarLabel.connect(
+            [&](const std::string& label) {
+                if (label == "Batch verifying Spark Proofs...")
+                    ++verifications;
+            });
+        auto check = [&](const CTransaction& tx, CSparkTxInfo* info) {
+            CValidationState state;
+            return CheckSparkTransaction(tx, state, tx.GetHash(), false,
+                info ? chainActive.Height() + 1 : INT_MAX, false, true, info);
+        };
+
+        {
+            LOCK(cs_main);
+            batch->init(BatchProofContainer::Mode::Deferred);
+            CSparkTxInfo historicalInfo;
+            BOOST_REQUIRE(check(first, &historicalInfo));
+            batch->finalize();
+            BOOST_CHECK_EQUAL(verifications, 0);
+
+            // Verifying a recent block must neither consume nor clear the
+            // recovery marker for a separate historical pending batch.
+            batch->init(BatchProofContainer::Mode::Block);
+            CSparkTxInfo blockInfo;
+            BOOST_REQUIRE(check(second, &blockInfo));
+            BOOST_CHECK(batch->verify_block_batch());
+            BOOST_CHECK_EQUAL(verifications, 1);
+            BOOST_CHECK(BatchProofContainer::HasRecoveryMarker());
+        }
+        BOOST_CHECK(batch->verify_pending());
+        BOOST_CHECK_EQUAL(verifications, 2);
+        BOOST_CHECK(!BatchProofContainer::HasRecoveryMarker());
+
+        {
+            LOCK(cs_main);
+            // Populate the normal mempool proof cache, then check that a
+            // cached spend requires no additional per-block batch work.
+            BOOST_REQUIRE(check(first, nullptr));
+            batch->init(BatchProofContainer::Mode::Block);
+            CSparkTxInfo cachedInfo;
+            BOOST_REQUIRE(check(first, &cachedInfo));
+            BOOST_CHECK(batch->verify_block_batch());
+            BOOST_CHECK_EQUAL(verifications, 2);
+        }
+        ClearSparkSpendProofCache();
+
+        // Exercise the real ConnectBlock path with two uncached spends.
+        BOOST_REQUIRE(GenerateBlock({CMutableTransaction(first), CMutableTransaction(second)}));
+        BOOST_CHECK_EQUAL(verifications, 3);
+        BOOST_CHECK(batch->verify_pending());
+        BOOST_CHECK_EQUAL(verifications, 3);
+        BOOST_CHECK(!BatchProofContainer::HasRecoveryMarker());
+    }
+}
+
 BOOST_AUTO_TEST_CASE(batched_spark_proofs_are_verified_inside_connect_block)
 {
     BatchProofContainer* batch = BatchProofContainer::get_instance();
@@ -3149,7 +3221,6 @@ BOOST_AUTO_TEST_CASE(batched_spark_proofs_are_verified_inside_connect_block)
         BatchProofContainer* batch;
         ~ResetBatch()
         {
-            batch->fCollectProofs = false;
             batch->init();
         }
     } reset{batch};
@@ -3173,8 +3244,7 @@ BOOST_AUTO_TEST_CASE(batched_spark_proofs_are_verified_inside_connect_block)
     candidateIndex.phashBlock = &candidateHash;
     candidateIndex.pprev = chainActive.Tip();
     candidateIndex.nHeight = chainActive.Height() + 1;
-    // Use a recent block time so ConnectBlock verifies proofs inline instead
-    // of deferring them (master IBD batching path).
+    // Use a recent block time so ConnectBlock verifies this block's batch.
     candidateIndex.nTime = GetSystemTimeInSeconds();
 
     CValidationState state;
@@ -3182,7 +3252,16 @@ BOOST_AUTO_TEST_CASE(batched_spark_proofs_are_verified_inside_connect_block)
     {
         LOCK(cs_main);
         BOOST_CHECK(!ConnectBlock(
-            candidate, state, &candidateIndex, view, ::Params(), true));
+            candidate, state, &candidateIndex, view, ::Params(), false));
+    }
+    BOOST_CHECK_EQUAL(state.GetRejectReason(), "bad-spark-batch-proof");
+
+    CValidationState checkOnlyState;
+    CCoinsViewCache checkOnlyView(pcoinsTip);
+    {
+        LOCK(cs_main);
+        BOOST_CHECK(!ConnectBlock(
+            candidate, checkOnlyState, &candidateIndex, checkOnlyView, ::Params(), true));
     }
 
     // VerifyDB must avoid tip-state mutation without skipping the proof.
@@ -3211,7 +3290,6 @@ BOOST_AUTO_TEST_CASE(abandoned_connect_block_clears_batched_spark_proofs)
         BatchProofContainer* batch;
         ~ResetBatch()
         {
-            batch->fCollectProofs = false;
             batch->init();
         }
     } reset{batch};
@@ -3248,13 +3326,11 @@ BOOST_AUTO_TEST_CASE(abandoned_connect_block_clears_batched_spark_proofs)
     {
         LOCK(cs_main);
         BOOST_CHECK(!ConnectBlock(
-            candidate, state, &candidateIndex, view, ::Params(), true));
+            candidate, state, &candidateIndex, view, ::Params(), false));
     }
 
-    // Master deferred batching does not abort() on ConnectBlock failure; the
-    // next ConnectBlock init() (or an explicit init here) drops temps.
-    batch->init();
-    batch->fCollectProofs = false;
+    // Even a later finalize cannot enqueue proofs from an abandoned block.
+    batch->finalize();
     BOOST_CHECK(batch->verify_pending());
 
     mempool.clear();
@@ -3263,7 +3339,6 @@ BOOST_AUTO_TEST_CASE(abandoned_connect_block_clears_batched_spark_proofs)
 
 BOOST_AUTO_TEST_CASE(verifydb_level_four_reconnects_spark_spend_and_mints)
 {
-    BatchProofContainer::get_instance()->fCollectProofs = false;
     BatchProofContainer::get_instance()->init();
 
     GenerateBlocks(500);
@@ -3286,7 +3361,6 @@ BOOST_AUTO_TEST_CASE(verifydb_level_four_reconnects_spark_spend_and_mints)
 
 BOOST_AUTO_TEST_CASE(verifydb_rejects_invalid_standalone_spark_mint)
 {
-    BatchProofContainer::get_instance()->fCollectProofs = false;
     BatchProofContainer::get_instance()->init();
 
     GenerateBlocks(500);
@@ -3323,7 +3397,6 @@ BOOST_AUTO_TEST_CASE(verifydb_rejects_invalid_standalone_spark_mint)
 
 BOOST_AUTO_TEST_CASE(verifydb_rejects_same_block_spark_double_spend)
 {
-    BatchProofContainer::get_instance()->fCollectProofs = false;
     BatchProofContainer::get_instance()->init();
 
     GenerateBlocks(500);
@@ -3392,7 +3465,6 @@ BOOST_AUTO_TEST_CASE(verifydb_rejects_same_block_spark_double_spend)
 
 BOOST_AUTO_TEST_CASE(verifydb_rejects_cross_block_spark_double_spend)
 {
-    BatchProofContainer::get_instance()->fCollectProofs = false;
     BatchProofContainer::get_instance()->init();
 
     GenerateBlocks(500);
@@ -3548,7 +3620,6 @@ BOOST_AUTO_TEST_CASE(spark_single_input_block_boundary_and_reorg)
     BOOST_REQUIRE_EQUAL(chainActive.Height(), baseHeight);
 
     const auto reconnect = [](const CBlock& block) {
-        LOCK(cs_main);
         CValidationState state;
         const auto shared = std::make_shared<const CBlock>(block);
         BOOST_REQUIRE(ActivateBestChain(state, ::Params(), shared));
@@ -3874,7 +3945,7 @@ BOOST_AUTO_TEST_CASE(spark_unknown_cover_set_reference_is_not_mempool_admissible
     UpdateRegtestSparkChaumV2Height(exactReferencesHeight);
 
     BatchProofContainer* batch = BatchProofContainer::get_instance();
-    batch->fCollectProofs = false;
+    batch->init();
     CValidationState legacyExtraReferenceState;
     CSparkTxInfo legacyExtraReferenceInfo;
     BOOST_REQUIRE(CheckSparkTransaction(
@@ -3887,8 +3958,7 @@ BOOST_AUTO_TEST_CASE(spark_unknown_cover_set_reference_is_not_mempool_admissible
         true,
         &legacyExtraReferenceInfo));
 
-    batch->init();
-    batch->fCollectProofs = true;
+    batch->init(BatchProofContainer::Mode::Deferred);
     CValidationState legacyBatchedExtraReferenceState;
     CSparkTxInfo legacyBatchedExtraReferenceInfo;
     BOOST_REQUIRE(CheckSparkTransaction(
@@ -4009,9 +4079,6 @@ BOOST_AUTO_TEST_CASE(coingroup)
 
     // util function
     auto reconnect = [](CBlock const &block) {
-        LOCK2(cs_main, pwalletMain->cs_wallet);
-        LOCK(mempool.cs);
-
         std::shared_ptr<CBlock const> sharedBlock =
                 std::make_shared<CBlock const>(block);
 
diff --git a/src/validation.cpp b/src/validation.cpp
index ed69613b9a..aa3fdfa495 100644
--- a/src/validation.cpp
+++ b/src/validation.cpp
@@ -2315,7 +2315,13 @@ static bool ShouldBatchSparkProofs(const CBlockIndex* pindex)
 
 bool VerifyPendingSparkBatch(CValidationState& state, const std::string& reason)
 {
-    if (!BatchProofContainer::get_instance()->verify_pending()) {
+    bool passed;
+    try {
+        passed = BatchProofContainer::get_instance()->verify_pending();
+    } catch (const std::exception& e) {
+        return AbortNode(state, strprintf("Unable to verify Spark batch before %s: %s", reason, e.what()));
+    }
+    if (!passed) {
         return AbortNode(state,
                          strprintf("Spark batch verification failed before %s", reason),
                          _("Spark batch verification failed. The invalid spend transactions are listed in debug.log. Restart the node: batching is disabled and a reindex is started automatically so chainstate is rebuilt and Spark proofs are checked block by block."));
@@ -2797,10 +2803,20 @@ bool ConnectBlock(const CBlock& block, CValidationState& state, CBlockIndex* pin
 
     std::set<uint256> txIds;
     bool isMainNet = chainparams.GetConsensus().IsMain();
-    // batch verify Lelantus/Sigma if block is older than a day, that means we are syncing or reindexing
     BatchProofContainer* batchProofContainer = BatchProofContainer::get_instance();
-    batchProofContainer->fCollectProofs = ShouldBatchSparkProofs(pindex);
-    batchProofContainer->init();
+    // Keep accumulated historical batches, but verify recent blocks before
+    // publishing state. Check-only paths must verify proofs directly.
+    auto batchMode = BatchProofContainer::Mode::Disabled;
+    if (!fJustCheck && GetBoolArg("-batching", true)) {
+        batchMode = ShouldBatchSparkProofs(pindex)
+            ? BatchProofContainer::Mode::Deferred : BatchProofContainer::Mode::Block;
+    }
+    batchProofContainer->init(batchMode);
+    struct ResetSparkBatch
+    {
+        BatchProofContainer* container;
+        ~ResetSparkBatch() { container->init(); }
+    } resetSparkBatch{batchProofContainer};
     std::size_t nSigma = 0;
     std::size_t nLelantus = 0;
 
@@ -2991,6 +3007,15 @@ bool ConnectBlock(const CBlock& block, CValidationState& state, CBlockIndex* pin
         }
     }
 
+    // Special transaction processing can publish notifications and cache
+    // changes. A recent block's proofs must pass before any of those effects.
+    try {
+        if (!batchProofContainer->verify_block_batch())
+            return state.DoS(100, false, REJECT_INVALID, "bad-spark-batch-proof");
+    } catch (const std::bad_alloc&) {
+        return state.Error("ConnectBlock(): memory allocation failed while verifying Spark batch");
+    }
+
     if (!ProcessSpecialTxsInBlock(block, pindex, state, isVerifyDB ? false : fJustCheck, fScriptChecks, !isVerifyDB)) {
         return error("ConnectBlock(): ProcessSpecialTxsInBlock for block %s at height %i failed with %s",
                     pindex->GetBlockHash().ToString(), pindex->nHeight, FormatStateMessage(state));
@@ -3115,7 +3140,7 @@ bool ConnectBlock(const CBlock& block, CValidationState& state, CBlockIndex* pin
     // add this block to the view's block chain
     view.SetBestBlock(pindex->GetBlockHash());
 
-    // do batch verification if remains a day or collect proofs
+    // Only historical blocks contribute to the deferred batch.
     batchProofContainer->finalize();
 
     int64_t nTime5 = GetTimeMicros(); nTimeIndex += nTime5 - nTime4;
@@ -3691,24 +3716,26 @@ bool DisconnectBlocks(int blocks) {
 }
 
 void ReprocessBlocks(int nBlocks) {
-    LOCK(cs_main);
+    {
+        LOCK(cs_main);
 
-    std::map<uint256, int64_t>::iterator it = mapRejectedBlocks.begin();
-    while (it != mapRejectedBlocks.end()) {
-        //use a window twice as large as is usual for the nBlocks we want to reset
-        if ((*it).second > GetTime() - (nBlocks * 60 * 5)) {
-            BlockMap::iterator mi = mapBlockIndex.find((*it).first);
-            if (mi != mapBlockIndex.end() && (*mi).second) {
+        std::map<uint256, int64_t>::iterator it = mapRejectedBlocks.begin();
+        while (it != mapRejectedBlocks.end()) {
+            //use a window twice as large as is usual for the nBlocks we want to reset
+            if ((*it).second > GetTime() - (nBlocks * 60 * 5)) {
+                BlockMap::iterator mi = mapBlockIndex.find((*it).first);
+                if (mi != mapBlockIndex.end() && (*mi).second) {
 
-                CBlockIndex *pindex = (*mi).second;
-                LogPrintf("ReprocessBlocks -- %s\n", (*it).first.ToString());
+                    CBlockIndex *pindex = (*mi).second;
+                    LogPrintf("ReprocessBlocks -- %s\n", (*it).first.ToString());
 
-                ResetBlockFailureFlags(pindex);            }
+                    ResetBlockFailureFlags(pindex);            }
+            }
+            ++it;
         }
-        ++it;
-    }
 
-    DisconnectBlocks(nBlocks);
+        DisconnectBlocks(nBlocks);
+    }
 
     CValidationState state;
     ActivateBestChain(state, Params());
@@ -3988,12 +4015,12 @@ bool ActivateBestChain(CValidationState &state, const CChainParams& chainparams,
                 for (unsigned int i = 0; i < block.vtx.size(); i++)
                     GetMainSignals().SyncTransaction(*block.vtx[i], pair.first, i);
             }
-            BatchProofContainer* batchProofContainer = BatchProofContainer::get_instance();
-            batchProofContainer->fCollectProofs = ShouldBatchSparkProofs(pindexNewTip);
-            if (!VerifyPendingSparkBatch(state, "connecting new tip"))
-                return false;
         }
 
+        if (!ShouldBatchSparkProofs(pindexNewTip) &&
+            !VerifyPendingSparkBatch(state, "connecting new tip"))
+            return false;
+
         // When we reach this point, we switched to a new tip (stored in pindexNewTip).
 
         // Notifications/callbacks that can run without cs_main
diff --git a/src/validation.h b/src/validation.h
index a0ffae67db..255d3ac03c 100644
--- a/src/validation.h
+++ b/src/validation.h
@@ -301,10 +301,11 @@ void ThreadScriptCheck();
 bool IsInitialBlockDownload();
 /** Retrieve a transaction (from memory pool, or from disk, if possible) */
 bool GetTransaction(const uint256 &hash, CTransactionRef &tx, const Consensus::Params& params, uint256 &hashBlock, bool fAllowSlow = false);
-/** Find the best known block, and make it the tip of the block chain */
+/** Find and activate the best known block. Call without cs_main. */
 bool ActivateBestChain(CValidationState& state, const CChainParams& chainparams, std::shared_ptr<const CBlock> pblock = std::shared_ptr<const CBlock>());
 /**
  * Verify the pending Spark batch when proofs are not being collected.
+ * Call without cs_main so concurrent verification can wait without deadlocking.
  * On failure the node is aborted, a datadir marker is written so the next
  * start disables batching and reindexes, and false is returned (no throw).
  */
diff --git a/src/wallet/test/spark_wallet_tests.cpp b/src/wallet/test/spark_wallet_tests.cpp
index 43d8f07246..f986926b8e 100644
--- a/src/wallet/test/spark_wallet_tests.cpp
+++ b/src/wallet/test/spark_wallet_tests.cpp
@@ -403,11 +403,13 @@ BOOST_AUTO_TEST_CASE(mintspark_and_mint_all)
     }
 
     auto generateBlocksPerScripts = [&](size_t blocks, size_t blocksPerScript) -> std::vector<CScript> {
-        LOCK2(cs_main, pwalletMain->cs_wallet);
         std::vector<CScript> scripts;
         while (blocks != 0) {
             CPubKey key;
-            key = pwalletMain->GenerateNewKey();
+            {
+                LOCK(pwalletMain->cs_wallet);
+                key = pwalletMain->GenerateNewKey();
+            }
             scripts.push_back(GetScriptForDestination(key.GetID()));
             auto blockCount = std::min(blocksPerScript, blocks);
             GenerateBlocks(blockCount, &scripts.back());
diff --git a/src/wallet/test/wallet_tests.cpp b/src/wallet/test/wallet_tests.cpp
index 7f7671f73e..33c2d2ef8b 100644
--- a/src/wallet/test/wallet_tests.cpp
+++ b/src/wallet/test/wallet_tests.cpp
@@ -365,12 +365,15 @@ BOOST_AUTO_TEST_CASE(ApproximateBestSubset)
 
 BOOST_FIXTURE_TEST_CASE(rescan, TestChain100Setup)
 {
-    LOCK(cs_main);
-
     // Cap last block file size, and mine new block in a new block file.
-    CBlockIndex* oldTip = chainActive.Tip();
-    GetBlockFileInfo(oldTip->GetBlockPos().nFile)->nSize = MAX_BLOCKFILE_SIZE;
+    CBlockIndex* oldTip;
+    {
+        LOCK(cs_main);
+        oldTip = chainActive.Tip();
+        GetBlockFileInfo(oldTip->GetBlockPos().nFile)->nSize = MAX_BLOCKFILE_SIZE;
+    }
     CreateAndProcessBlock({}, GetScriptForRawPubKey(coinbaseKey.GetPubKey()));
+    LOCK(cs_main);
     CBlockIndex* newTip = chainActive.Tip();
 
     // Verify ScanForWalletTransactions picks up transactions in both the old
@@ -437,7 +440,6 @@ BOOST_FIXTURE_TEST_CASE(rescan, TestChain100Setup)
 BOOST_FIXTURE_TEST_CASE(importwallet_rescan, TestChain100Setup)
 {
     CWallet *pwalletMainBackup = ::pwalletMain;
-    LOCK(cs_main);
 
     // Create two blocks with same timestamp to verify that importwallet rescan
     // will pick up both blocks, not just the first.
@@ -452,6 +454,7 @@ BOOST_FIXTURE_TEST_CASE(importwallet_rescan, TestChain100Setup)
     SetMockTime(KEY_TIME);
     coinbaseTxns.emplace_back(*CreateAndProcessBlock({}, GetScriptForRawPubKey(coinbaseKey.GetPubKey())).vtx[0]);
 
+    LOCK(cs_main);
     // Import key into wallet and call dumpwallet to create backup file.
     {
         CWallet wallet;