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); }