From 65130c31208af2eeed6cdff1314449d370daa770 Mon Sep 17 00:00:00 2001 From: DanS Date: Tue, 25 Aug 2026 19:18:30 +0200 Subject: [PATCH] async/wallet: close the loose ends around automated operations Six small defects, all found by an audit of the automated-operation path and all verified present before changing anything. AsyncRPCQueue::addOperation returned void and silently dropped the operation when the queue was closed or finishing. Every caller assumed success: the three schedulers left their running flag set with nothing in flight (which then blocks every later round), and z_sendmany / z_shieldcoinbase / z_mergetoaddress returned an opid for work that would never run -- z_shieldcoinbase and z_mergetoaddress having already locked their selected coins in the constructor. It now returns bool; the schedulers release the flag and log, and the three RPCs raise an error instead of handing back an opid. Coin locks are memory-only, so a refusal at shutdown reclaims them with the process; the lie about success was the defect. Nothing ever removed finished automated operations from the queue's map. popOperationForId is reached only from z_getoperationresult, so on a node running autoshield every 25 blocks the map grew by one entry per round forever. The schedulers now pop the operation they just cancelled. The worker already handles a missing id ("cannot find operation in map, may have been removed", asyncrpcqueue.cpp), and it releases lock_ before calling main(), so popping under cs_wallet introduces no lock cycle. The autoshield operation built its transaction against targetHeight_, the enqueue-time height, while SetExpiryHeight and the network-upgrade straddle guard both used tipHeight. Since the builder's height selects the consensus branch id, the guard was checking a height the transaction was not signed against -- it could not prevent the failure it exists to prevent. Now tipHeight throughout. cancel() in the sweep, consolidation and autoshield operations set CANCELLED unconditionally, dropping the base class's guard entirely. The schedulers cancel the previous operation when they enqueue the next, so a round that had already SUCCEEDED got its result relabelled as cancelled. Restored a narrower guard: still cancellable while READY or EXECUTING (the base class refuses the latter, which would defeat cancellation here), but a terminal state is left alone. CommitAutomatedTx dumped the whole transaction to stderr on every commit, duplicating the LogPrintf that CommitTransaction does one call later. ToString() emits a line per input, so with the 400-input autoshield cap that was tens of KB of stderr per round. Removed. Also corrected a comment that credited the immature-coinbase exclusion to fOnlySpendable (the argument is fOnlyConfirmed; the exclusion is unconditional in AvailableCoins), and noted that AUTOSHIELD_CTXIN_P2SH_SIZE is a byte size that merely happens to share the value 400 with the input cap. Verified on an isolated regtest chain, 8/8: nine consecutive autoshield rounds succeed after the builder-height change; the operation map stays at 1 entry across all nine (it grew one per round before); stderr totals 1608 bytes for the whole run with zero CommitAutomatedTx dumps, while debug.log still records all nine commits via CommitTransaction; no round is relabelled cancelled; funds shield correctly and no coin locks leak. Co-Authored-By: Claude Opus 5 (1M context) --- src/asyncrpcqueue.cpp | 9 +++-- src/asyncrpcqueue.h | 7 +++- .../asyncrpcoperation_autoshieldcoinbase.cpp | 23 +++++++++-- ...asyncrpcoperation_saplingconsolidation.cpp | 7 ++++ src/wallet/asyncrpcoperation_sweep.cpp | 7 ++++ src/wallet/rpcwallet.cpp | 18 +++++++-- src/wallet/wallet.cpp | 40 +++++++++++++++++-- 7 files changed, 96 insertions(+), 15 deletions(-) diff --git a/src/asyncrpcqueue.cpp b/src/asyncrpcqueue.cpp index 543951e18..ce356c146 100644 --- a/src/asyncrpcqueue.cpp +++ b/src/asyncrpcqueue.cpp @@ -96,18 +96,21 @@ void AsyncRPCQueue::run(size_t workerId) { * * Don't use std::make_shared(). */ -void AsyncRPCQueue::addOperation(const std::shared_ptr &ptrOperation) { +bool AsyncRPCQueue::addOperation(const std::shared_ptr &ptrOperation) { std::lock_guard guard(lock_); - // Don't add if queue is closed or finishing + // Don't add if queue is closed or finishing. Report it: silently dropping the + // operation made callers announce work that would never run. + // (isClosed/isFinishing read atomics, so calling them under the guard is safe.) if (isClosed() || isFinishing()) { - return; + return false; } AsyncRPCOperationId id = ptrOperation->getId(); operation_map_.emplace(id, ptrOperation); operation_id_queue_.push(id); this->condition_.notify_one(); + return true; } /** diff --git a/src/asyncrpcqueue.h b/src/asyncrpcqueue.h index 1ebc6c1ee..099b31706 100644 --- a/src/asyncrpcqueue.h +++ b/src/asyncrpcqueue.h @@ -63,7 +63,12 @@ public: size_t getOperationCount() const; std::shared_ptr getOperationForId(AsyncRPCOperationId) const; std::shared_ptr popOperationForId(AsyncRPCOperationId); - void addOperation(const std::shared_ptr &ptrOperation); + // Returns false if the queue is closed or finishing, in which case the + // operation was NOT queued and will never run. Callers must react: a caller + // that ignores this both reports success for work that will not happen and + // leaves any state it set for the operation (running flags, coin locks) + // stranded for the life of the process. + bool addOperation(const std::shared_ptr &ptrOperation); std::vector getAllOperationIds() const; private: diff --git a/src/wallet/asyncrpcoperation_autoshieldcoinbase.cpp b/src/wallet/asyncrpcoperation_autoshieldcoinbase.cpp index d50273364..4469adf92 100644 --- a/src/wallet/asyncrpcoperation_autoshieldcoinbase.cpp +++ b/src/wallet/asyncrpcoperation_autoshieldcoinbase.cpp @@ -34,6 +34,8 @@ static const size_t AUTOSHIELD_TX_OVERHEAD = (3 * AUTOSHIELD_SAPLING_OUTPUT_SIZE // conservative, so an under-estimate translates directly into an oversize tx. // The remainder is simply shielded on the next round. static const size_t AUTOSHIELD_MAX_INPUTS = 400; +// Unrelated to the cap above despite sharing the value: this is a SIZE IN BYTES for +// one spent P2SH input, mirroring CTXIN_SPEND_P2SH_SIZE in rpcwallet.cpp. static const size_t AUTOSHIELD_CTXIN_P2SH_SIZE = 400; // Expire unmined autoshield txs after this many blocks, so a tx cannot straddle // a network-upgrade activation. @@ -356,9 +358,10 @@ bool AsyncRPCOperation_autoshieldcoinbase::main_impl() { } // Gather matured, spendable coinbase UTXOs, byte-capped to a single tx. - // AvailableCoins with fOnlySpendable already excludes immature coinbase - // (< COINBASE_MATURITY) and outputs we don't own, so external - // -mineraddress / pool coinbase naturally yields zero inputs. + // AvailableCoins excludes immature coinbase unconditionally (wallet.cpp, + // `IsCoinBase() && GetBlocksToMaturity() > 0`) and only ever returns outputs + // we own, so external -mineraddress / pool coinbase yields zero inputs. The + // second argument here is fOnlyConfirmed, not fOnlySpendable. size_t estimatedTxSize = AUTOSHIELD_TX_OVERHEAD; std::vector vecOutputs; pwalletMain->AvailableCoins(vecOutputs, true, NULL, false, true); @@ -421,7 +424,12 @@ bool AsyncRPCOperation_autoshieldcoinbase::main_impl() { // Build the t->z shield tx. Proof generation happens in Build() WITHOUT // holding cs_wallet (mirrors the sweep op) so we don't stall wallet RPCs. - auto builder = TransactionBuilder(consensusParams, targetHeight_, pwalletMain); + // tipHeight, not targetHeight_: the builder's height selects the consensus + // branch id (transaction_builder.cpp CurrentEpochBranchId), and the NU-straddle + // guard above plus SetExpiryHeight below are both keyed off tipHeight. Using the + // stale enqueue-time height here meant the guard was checking a height the + // transaction was not actually signed against. + auto builder = TransactionBuilder(consensusParams, tipHeight, pwalletMain); builder.SetExpiryHeight(tipHeight + AUTOSHIELD_EXPIRY_DELTA); builder.SetFee(fee); @@ -485,6 +493,13 @@ void AsyncRPCOperation_autoshieldcoinbase::setResult() { } void AsyncRPCOperation_autoshieldcoinbase::cancel() { + // Cancelling is how the scheduler stops an in-flight round, so unlike the base + // class this must be able to move an EXECUTING operation to CANCELLED. What it + // must not do is overwrite a state that is already terminal: the scheduler + // cancels the previous operation when it enqueues the next one, and that one may + // have already SUCCEEDED, whose result would otherwise be relabelled as cancelled. + if (isSuccess() || isFailed() || isCancelled()) + return; set_state(OperationStatus::CANCELLED); } diff --git a/src/wallet/asyncrpcoperation_saplingconsolidation.cpp b/src/wallet/asyncrpcoperation_saplingconsolidation.cpp index 1a6dfec7c..2f360dfed 100644 --- a/src/wallet/asyncrpcoperation_saplingconsolidation.cpp +++ b/src/wallet/asyncrpcoperation_saplingconsolidation.cpp @@ -305,6 +305,13 @@ void AsyncRPCOperation_saplingconsolidation::setConsolidationResult(int numTxCre } void AsyncRPCOperation_saplingconsolidation::cancel() { + // Cancelling is how the scheduler stops an in-flight round, so unlike the base + // class this must be able to move an EXECUTING operation to CANCELLED. What it + // must not do is overwrite a state that is already terminal: the scheduler + // cancels the previous operation when it enqueues the next one, and that one may + // have already SUCCEEDED, whose result would otherwise be relabelled as cancelled. + if (isSuccess() || isFailed() || isCancelled()) + return; set_state(OperationStatus::CANCELLED); } diff --git a/src/wallet/asyncrpcoperation_sweep.cpp b/src/wallet/asyncrpcoperation_sweep.cpp index c9b7afaf6..a67052414 100644 --- a/src/wallet/asyncrpcoperation_sweep.cpp +++ b/src/wallet/asyncrpcoperation_sweep.cpp @@ -364,6 +364,13 @@ void AsyncRPCOperation_sweep::setSweepResult(int numTxCreated, const CAmount& am } void AsyncRPCOperation_sweep::cancel() { + // Cancelling is how the scheduler stops an in-flight round, so unlike the base + // class this must be able to move an EXECUTING operation to CANCELLED. What it + // must not do is overwrite a state that is already terminal: the scheduler + // cancels the previous operation when it enqueues the next one, and that one may + // have already SUCCEEDED, whose result would otherwise be relabelled as cancelled. + if (isSuccess() || isFailed() || isCancelled()) + return; set_state(OperationStatus::CANCELLED); } diff --git a/src/wallet/rpcwallet.cpp b/src/wallet/rpcwallet.cpp index 98fb82c29..b969994b0 100644 --- a/src/wallet/rpcwallet.cpp +++ b/src/wallet/rpcwallet.cpp @@ -5543,7 +5543,10 @@ UniValue z_sendmany(const UniValue& params, bool fHelp, const CPubKey& mypk) // Create operation and add to global queue std::shared_ptr q = getAsyncRPCQueue(); std::shared_ptr operation( new AsyncRPCOperation_sendmany(builder, contextualTx, fromaddress, taddrRecipients, zaddrRecipients, saplingNoteInputs, nMinDepth, nFee, contextInfo, opret) ); - q->addOperation(operation); + if (!q->addOperation(operation)) { + throw JSONRPCError(RPC_INTERNAL_ERROR, + "Async RPC queue is shutting down; the operation was not queued"); + } if(fZdebug) LogPrintf("%s: Submitted to async queue\n", __FUNCTION__); @@ -5771,7 +5774,13 @@ UniValue z_shieldcoinbase(const UniValue& params, bool fHelp, const CPubKey& myp // Create operation and add to global queue std::shared_ptr q = getAsyncRPCQueue(); std::shared_ptr operation( new AsyncRPCOperation_shieldcoinbase(builder, contextualTx, inputs, destaddress, nFee, donation, contextInfo) ); - q->addOperation(operation); + // The constructor has already locked the selected coins. Coin locks are + // memory-only, so a refused queue at shutdown reclaims them with the process; + // what must not happen is returning an opid for work that will never run. + if (!q->addOperation(operation)) { + throw JSONRPCError(RPC_INTERNAL_ERROR, + "Async RPC queue is shutting down; the operation was not queued"); + } AsyncRPCOperationId operationId = operation->getId(); // Return continuation information @@ -6125,7 +6134,10 @@ UniValue z_mergetoaddress(const UniValue& params, bool fHelp, const CPubKey& myp std::shared_ptr q = getAsyncRPCQueue(); std::shared_ptr operation( new AsyncRPCOperation_mergetoaddress(builder, contextualTx, utxoInputs, saplingNoteInputs, recipient, nFee, contextInfo) ); - q->addOperation(operation); + if (!q->addOperation(operation)) { + throw JSONRPCError(RPC_INTERNAL_ERROR, + "Async RPC queue is shutting down; the operation was not queued"); + } AsyncRPCOperationId operationId = operation->getId(); // Return continuation information diff --git a/src/wallet/wallet.cpp b/src/wallet/wallet.cpp index 7a78591f9..ff80f062f 100644 --- a/src/wallet/wallet.cpp +++ b/src/wallet/wallet.cpp @@ -651,11 +651,21 @@ void CWallet::RunSaplingSweep(int blockHeight) { std::shared_ptr lastOperation = q->getOperationForId(saplingSweepOperationId); if (lastOperation != nullptr) { lastOperation->cancel(); + // Drop it from the queue's map as well. Nothing else ever removes these: + // popOperationForId is only reached from z_getoperationresult, so on a node + // running this every interval the map grew without bound. + q->popOperationForId(saplingSweepOperationId); } pendingSaplingSweepTxs.clear(); std::shared_ptr operation(new AsyncRPCOperation_sweep(blockHeight + 5)); saplingSweepOperationId = operation->getId(); - q->addOperation(operation); + if (!q->addOperation(operation)) { + // Queue is closing (shutdown). Release the flag we just set, or it stays + // set with no operation in flight and blocks every later round. + LogPrintf("%s: async queue is not accepting operations; skipping this round\n", __func__); + fSweepRunning = false; + return; + } } void CWallet::RunSaplingConsolidation(int blockHeight) { @@ -699,11 +709,21 @@ void CWallet::RunSaplingConsolidation(int blockHeight) { std::shared_ptr lastOperation = q->getOperationForId(saplingConsolidationOperationId); if (lastOperation != nullptr) { lastOperation->cancel(); + // Drop it from the queue's map as well. Nothing else ever removes these: + // popOperationForId is only reached from z_getoperationresult, so on a node + // running this every interval the map grew without bound. + q->popOperationForId(saplingConsolidationOperationId); } pendingSaplingConsolidationTxs.clear(); std::shared_ptr operation(new AsyncRPCOperation_saplingconsolidation(blockHeight + 5)); saplingConsolidationOperationId = operation->getId(); - q->addOperation(operation); + if (!q->addOperation(operation)) { + // Queue is closing (shutdown). Release the flag we just set, or it stays + // set with no operation in flight and blocks every later round. + LogPrintf("%s: async queue is not accepting operations; skipping this round\n", __func__); + fConsolidationRunning = false; + return; + } } // Periodically drain matured transparent coinbase into a wallet-owned Sapling @@ -754,16 +774,28 @@ void CWallet::RunAutoShieldCoinbase(int blockHeight) { std::shared_ptr lastOperation = q->getOperationForId(saplingAutoShieldOperationId); if (lastOperation != nullptr) { lastOperation->cancel(); + // Drop it from the queue's map as well. Nothing else ever removes these: + // popOperationForId is only reached from z_getoperationresult, so on a node + // running this every interval the map grew without bound. + q->popOperationForId(saplingAutoShieldOperationId); } std::shared_ptr operation(new AsyncRPCOperation_autoshieldcoinbase(blockHeight + 5)); saplingAutoShieldOperationId = operation->getId(); - q->addOperation(operation); + if (!q->addOperation(operation)) { + // Queue is closing (shutdown). Release the flag we just set, or it stays + // set with no operation in flight and blocks every later round. + LogPrintf("%s: async queue is not accepting operations; skipping this round\n", __func__); + fAutoShieldRunning = false; + return; + } } bool CWallet::CommitAutomatedTx(const CTransaction& tx) { CWalletTx wtx(this, tx); CReserveKey reservekey(pwalletMain); - fprintf(stderr,"%s: %s\n",__func__,tx.ToString().c_str()); + // No tx dump here: CommitTransaction already LogPrintf's the same wtx.ToString(), + // and ToString() emits a line per vin, so with the 400-input autoshield cap this + // printed tens of KB to stderr on every automated round. return CommitTransaction(wtx, reservekey); }