From 0ebd1ac5d5567e4e48e5eafe736b870dc96ef446 Mon Sep 17 00:00:00 2001 From: DanS Date: Mon, 26 Oct 2026 18:26:28 -0500 Subject: [PATCH] feat(wallet): defer + retry a merge that hits a transient mempool conflict MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A consolidation batch can be rejected by the daemon with bad-txns-shielded-requirements-not-met (REJECT_DUPLICATE) when it clashes with an earlier, still-unconfirmed batch — the batch outran its predecessors into the mempool. That isn't a real failure: once the earlier batches confirm, the same merge succeeds. Previously it failed the whole address. Now a merge OP failure whose message is a mempool conflict (shielded- / joinsplit-requirements-not-met, txn-mempool-conflict, inputs-spent — new pure helper isDeferrableMergeOpError) HOLDS the address for kRunMempoolClearSec (2 min) instead of failing it, then retries from where it left off; up to kRunMaxDeferRetries consecutive defers (reset on any success), after which it is a genuine failure. Deferrals don't feed the halt breaker, and since the address outcome stays non-failed the run verdict reads Clean when the retry succeeds. applyOpState takes `now` to stamp the hold; next() waits on a deferred address until its window passes. Unit-tested (testConsolidationDefer): the classifier; and a driven run where an op reports the conflict → the address is deferred (not failed), held before the window and retried after it. Build + tests + hygiene green. Co-Authored-By: Claude Opus 4.8 --- src/services/note_consolidation.cpp | 31 +++++++++++++++++-- src/services/note_consolidation.h | 12 +++++++- tests/test_phase4.cpp | 47 +++++++++++++++++++++++++++++ 3 files changed, 87 insertions(+), 3 deletions(-) diff --git a/src/services/note_consolidation.cpp b/src/services/note_consolidation.cpp index da593da..667a8a4 100644 --- a/src/services/note_consolidation.cpp +++ b/src/services/note_consolidation.cpp @@ -192,6 +192,22 @@ MergeFailure classifyMergeFailure(rpc::RpcFailureKind kind, const std::string& m return MergeFailure::Fatal; } +bool isDeferrableMergeOpError(const std::string& message) +{ + // A batch rejected from the mempool because it clashed with an earlier, still-unconfirmed batch (its + // Sapling nullifier/anchor, or an input, already in flight). Re-trying after those confirm succeeds. + // Substring match — the daemon prefixes a reject code (e.g. "18: bad-txns-shielded-requirements-not-met"). + static const char* const kNeedles[] = { + "shielded-requirements-not-met", // Sapling nullifier/anchor clash (REJECT_DUPLICATE) + "joinsplit-requirements-not-met", // the Sprout analogue + "txn-mempool-conflict", // spends an output another mempool tx already spends + "bad-txns-inputs-spent", // an input already spent by an unconfirmed sibling + }; + for (const char* n : kNeedles) + if (message.find(n) != std::string::npos) return true; + return false; +} + rpc::RpcFailureKind mergeCallFailureKind(const std::exception& e) { const rpc::RpcFailureKind kind = rpc::rpcFailureKindOf(e); @@ -298,7 +314,7 @@ void ConsolidationRun::endAddress(RunAddress& a, AddrOutcome o, const std::strin if (a.run.opids.empty() && ++deadStreak_ >= kRunMaxFailStreak) halt(msg); } -void ConsolidationRun::applyOpState(RunBatch& b, OpState st, const OpReport* rep) +void ConsolidationRun::applyOpState(RunBatch& b, OpState st, const OpReport* rep, double now) { b.state = st; if (b.addrIdx < 0 || b.addrIdx >= static_cast(addrs_.size())) return; @@ -307,10 +323,20 @@ void ConsolidationRun::applyOpState(RunBatch& b, OpState st, const OpReport* rep case OpState::Success: if (rep) b.txid = rep->txid; ++a.opsDone; + a.defers = 0; // progress made: this address's consecutive-defer budget resets opFailStreak_ = 0; break; case OpState::Failed: if (rep) b.error = rep->error; + // A transient mempool conflict (clashed with an earlier, still-unconfirmed batch): don't fail the + // address — hold it and retry once the earlier batches confirm. Bounded, and it doesn't feed the + // halt breaker. The batch itself stays terminal (Failed) so it isn't re-polled; the address stays + // not-done and gets a fresh batch after the hold (see next()). + if (isDeferrableMergeOpError(b.error) && a.defers < kRunMaxDeferRetries) { + ++a.defers; + a.deferUntil = now + kRunMempoolClearSec; + break; + } if (a.failure.empty()) a.failure = b.error; a.done = true; worsen(a, AddrOutcome::Failed); @@ -375,6 +401,7 @@ RunStep ConsolidationRun::next(const RunEnv& env, double now) if (env.locked || now < unlockHoldUntil_) return wait(RunWait::Locked); if (env.userSendBusy) return wait(RunWait::UserSend); if (now < nextCallAt_) return wait(RunWait::Batch); + if (now < addrs_[cur_].deferUntil) return wait(RunWait::Batch); // held after a mempool conflict; retry once it clears if (!batches_.empty() && (batches_.back().state == OpState::Unknown || batches_.back().state == OpState::Queued)) return wait(RunWait::Batch); // at most 2 unfinished ops, at most 2 x limit notes locked if (env.scanBusy) return wait(RunWait::Scan); // another GUI wallet scan is in the daemon @@ -484,7 +511,7 @@ void ConsolidationRun::onPollResult(std::uint64_t gen, bool ok, const std::vecto } b.misses = 0; const OpState st = opStateFromStatus(it->second->status); - if (st != OpState::Unknown) applyOpState(b, st, it->second); + if (st != OpState::Unknown) applyOpState(b, st, it->second, now); } } diff --git a/src/services/note_consolidation.h b/src/services/note_consolidation.h index bd55ab8..c6ecd49 100644 --- a/src/services/note_consolidation.h +++ b/src/services/note_consolidation.h @@ -113,6 +113,10 @@ enum class MergeFailure : std::uint8_t { Fatal, // anything else: this address is done }; MergeFailure classifyMergeFailure(rpc::RpcFailureKind kind, const std::string& message); +// True when an async op FAILURE message (not an RPC-call failure) is a transient mempool conflict — a batch +// that clashed with an earlier, still-unconfirmed batch (bad-txns-shielded-requirements-not-met and kin). +// The run holds the address and retries once the mempool clears instead of failing it. Pure; substring match. +bool isDeferrableMergeOpError(const std::string& message); // The failure kind of a z_mergetoaddress call that threw. Sent, then cut off before any HTTP status // (connection reset, an empty or partial reply): the daemon may have queued the merge, exactly as after a // reply timeout, so it reads as one. Any other failure keeps its kind. @@ -138,6 +142,10 @@ inline constexpr int kRunMaxTransientRetries = 5; inline constexpr int kRunMaxUnlockRefusals = 2; inline constexpr int kRunMaxFailStreak = 3; inline constexpr int kRunMaxPollMisses = 2; +// An op that failed with a transient mempool conflict (it clashed with an earlier, still-unconfirmed batch) +// holds its address for this long so the earlier batches can confirm, then retries — up to this many times. +inline constexpr double kRunMempoolClearSec = 120.0; +inline constexpr int kRunMaxDeferRetries = 5; struct RunConfig { double fee = 0.0001; @@ -169,6 +177,8 @@ struct RunAddress { int opsDone = 0; int transient = 0; int unlockRefusals = 0; + double deferUntil = 0.0; // held until this time after a mempool-conflict op failure, then retried + int defers = 0; // consecutive mempool-conflict defers (reset on a success; capped at kRunMaxDeferRetries) std::string failure; }; struct RunBatch { @@ -226,7 +236,7 @@ private: enum class Phase : std::uint8_t { Idle, Running, Finished }; static void worsen(RunAddress& a, AddrOutcome o) { if (o > a.outcome) a.outcome = o; } void endAddress(RunAddress& a, AddrOutcome o, const std::string& msg); - void applyOpState(RunBatch& b, OpState st, const OpReport* rep); + void applyOpState(RunBatch& b, OpState st, const OpReport* rep, double now = 0.0); void halt(const std::string& msg) { halted_ = true; if (haltError_.empty()) haltError_ = msg; } Phase phase_ = Phase::Idle; diff --git a/tests/test_phase4.cpp b/tests/test_phase4.cpp index e40a21c..6168e02 100644 --- a/tests/test_phase4.cpp +++ b/tests/test_phase4.cpp @@ -7395,6 +7395,52 @@ void testNoteConsolidation() } } +// A merge op that fails with a transient mempool conflict defers its address and retries once the window +// clears, instead of failing it — so a batch that outran an earlier, still-unconfirmed batch recovers. +void testConsolidationDefer() +{ + using namespace dragonx::services; + + // The classifier: mempool-conflict reject codes defer; a genuine "no funds" failure does not. + EXPECT_TRUE(isDeferrableMergeOpError("18: bad-txns-shielded-requirements-not-met")); + EXPECT_TRUE(isDeferrableMergeOpError("txn-mempool-conflict")); + EXPECT_TRUE(isDeferrableMergeOpError("bad-txns-inputs-spent")); + EXPECT_FALSE(isDeferrableMergeOpError("could not find any funds")); + EXPECT_FALSE(isDeferrableMergeOpError("")); + + ConsolidationRun run; + RunConfig cfg; cfg.limit = 10; cfg.maxBatches = 3; cfg.fee = 0.0001; + std::vector groups{ {"zA", 100, 50.0} }; + EXPECT_TRUE(run.start(groups, cfg, 0.0)); + RunEnv env; env.nodeUsable = true; // idle + callable (callsAllowed defaults true; the rest false) + + // A batch is submitted for address 0. + RunStep s = run.next(env, 1.0); + EXPECT_TRUE(s.action == RunAction::Call && s.addrIdx == 0); + run.markCallPosted(0, 1.0); + MergeCallResult cr; cr.opid = "op1"; cr.merging = 10; cr.remaining = 90; + run.onCallResult(run.gen(), 0, cr, dragonx::rpc::RpcFailureKind::None, 0.5, 2.0); + EXPECT_FALSE(run.addresses()[0].done); + + // Its op is polled and reports a mempool conflict. + EXPECT_TRUE(run.next(env, 3.0).action == RunAction::Poll); + run.markPollPosted(3.0); + std::vector reps{ {"op1", "failed", "18: bad-txns-shielded-requirements-not-met", ""} }; + run.onPollResult(run.gen(), true, reps, 4.0); + + // Deferred, not failed: still live, held, and counted as a defer (not an address failure). + EXPECT_TRUE(run.active()); + EXPECT_FALSE(run.addresses()[0].done); + EXPECT_TRUE(run.addresses()[0].outcome != AddrOutcome::Failed); + EXPECT_EQ(run.addresses()[0].defers, 1); + EXPECT_EQ(run.summary().addressesFailed, 0); + + // Held before the window, retried after it. + EXPECT_TRUE(run.next(env, 5.0).action != RunAction::Call); + RunStep retry = run.next(env, 4.0 + kRunMempoolClearSec + 1.0); + EXPECT_TRUE(retry.action == RunAction::Call && retry.addrIdx == 0); +} + void testDatadirLockGate() { using dragonx::daemon::EmbeddedDaemon; @@ -20298,6 +20344,7 @@ int main() testOwnershipRecord(); testDebugLogTailRules(); testNoteConsolidation(); + testConsolidationDefer(); testScopeScanAndLockedNotes(); testMergeCallResult(); testOpReports();