wallet: fix consolidation-scheduler wedge (ran every block; dead mutual-exclusion)

The Sapling auto-consolidation scheduler never advanced nextConsolidation after
init and never set fConsolidationRunning, so once the tip passed the init
threshold `-consolidation` dispatched a consolidation op every block instead of
once per -consolidationinterval, and every guard that reads fConsolidationRunning
(in RunSaplingSweep, and in the new autoshield driver) was dead.

Mirror the intended scheduler model:
- RunSaplingConsolidation sets fConsolidationRunning=true before dispatch and
  self-guards with `if (fConsolidationRunning) return;`.
- The consolidation op advances nextConsolidation = consolidationInterval +
  tipHeight and clears fConsolidationRunning on every terminal state
  (success/failure/exception/cancel), guarded by op id so only the current op
  mutates scheduler state.
- saplingConsolidationOperationId moved to public so the op can read it.

Restores the documented once-per-interval cadence and makes the
sweep/consolidation/autoshield mutual-exclusion guards effective.

Note: the sweep op has the same latent wedge (fSweepRunning/nextSweep are
cleared only on the sweepComplete success path, so a cancelled or throwing
sweep leaves fSweepRunning stuck true) - left for a follow-up; this commit is
the template to port.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/wallet/asyncrpcoperation_saplingconsolidation.cpp

@@ -28,8 +28,17 @@ AsyncRPCOperation_saplingconsolidation::AsyncRPCOperation_saplingconsolidation(i
 AsyncRPCOperation_saplingconsolidation::~AsyncRPCOperation_saplingconsolidation() {}
 
 void AsyncRPCOperation_saplingconsolidation::main() {
-    if (isCancelled())
+    if (isCancelled()) {
+        // Only the current op owns the scheduler flag; a stale/cancelled op must
+        // not clear it out from under a freshly-enqueued successor.
+        if (pwalletMain) {
+            LOCK(pwalletMain->cs_wallet);
+            if (getId() == pwalletMain->saplingConsolidationOperationId) {
+                pwalletMain->fConsolidationRunning = false;
+            }
+        }
         return;
+    }
 
     set_state(OperationStatus::EXECUTING);
     start_execution_clock();
@@ -76,6 +85,21 @@ void AsyncRPCOperation_saplingconsolidation::main() {
     LogPrintf("%s", s);
     unlock_notes(); // clean up
     LogPrint("zrpc", "%s: consolidation input notes unlocked\n", getId());
+
+    // Advance the interval and clear the running flag on EVERY terminal state
+    // (success, failure, exception) so consolidation runs once per interval
+    // instead of every block, and a failed round still lets the next one fire.
+    // Only the CURRENT op does this bookkeeping. This fixes the pre-existing
+    // wedge where nextConsolidation never advanced and fConsolidationRunning
+    // was never set/reset.
+    if (pwalletMain) {
+        LOCK2(cs_main, pwalletMain->cs_wallet);
+        if (getId() == pwalletMain->saplingConsolidationOperationId) {
+            int tipHeight = (chainActive.Tip() != NULL) ? chainActive.Tip()->GetHeight() : targetHeight_;
+            pwalletMain->nextConsolidation     = pwalletMain->consolidationInterval + tipHeight;
+            pwalletMain->fConsolidationRunning = false;
+        }
+    }
 }
 
 bool AsyncRPCOperation_saplingconsolidation::main_impl() {

src/wallet/wallet.cpp

@@ -644,6 +644,12 @@ void CWallet::RunSaplingConsolidation(int blockHeight) {
         return;
     }
 
+    // Self-guard: an op is already in flight (nextConsolidation only advances
+    // when it completes). Don't cancel + re-enqueue a fresh op every block.
+    if (fConsolidationRunning) {
+        return;
+    }
+
     LogPrintf("%s: consolidation enabled at blockHeight=%d fSweepRunning=%d\n", __func__,  blockHeight, fSweepRunning );
 
     if (fSweepRunning) {
@@ -657,6 +663,7 @@ void CWallet::RunSaplingConsolidation(int blockHeight) {
     }
 
     LogPrintf("%s:  creating consolidation operation at blockHeight=%d\n", __func__,  blockHeight);
+    fConsolidationRunning = true;
     std::shared_ptr<AsyncRPCQueue> q = getAsyncRPCQueue();
     std::shared_ptr<AsyncRPCOperation> lastOperation = q->getOperationForId(saplingConsolidationOperationId);
     if (lastOperation != nullptr) {
@@ -697,9 +704,6 @@ void CWallet::RunAutoShieldCoinbase(int blockHeight) {
 
     // Mutual exclusion: sweep/consolidation share the single async worker and
     // cs_wallet; don't queue an autoshield in the same connected block.
-    // NOTE: fConsolidationRunning is currently never set true (a pre-existing
-    // consolidation-scheduler bug); this half of the guard only becomes
-    // effective once that is fixed on the consolidation side.
     if (fSweepRunning || fConsolidationRunning) {
         LogPrintf("%s: not autoshielding since sweep/consolidation is running at height=%d\n", __func__, blockHeight);
         return;

src/wallet/wallet.h

@@ -784,7 +784,6 @@ private:
     TxNullifiers mapTxSaplingNullifiers;
     
     std::vector<CTransaction> pendingSaplingConsolidationTxs;
-    AsyncRPCOperationId saplingConsolidationOperationId;
 
     std::vector<CTransaction> pendingSaplingSweepTxs;
     AsyncRPCOperationId saplingSweepOperationId;
@@ -802,6 +801,9 @@ public:
     int64_t nWitnessCacheSize;
     bool needsRescan = false;
     int nextConsolidation = 0;
+    // Id of the in-flight consolidation op; read by the op to confirm it is
+    // still the current one before mutating scheduler state.
+    AsyncRPCOperationId saplingConsolidationOperationId;
 
     bool fSaplingConsolidationEnabled = false;
     bool fConsolidationRunning        = false;