feat(wallet): defer + retry a merge that hits a transient mempool conflict
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 <noreply@anthropic.com>
This commit is contained in:
@@ -192,6 +192,22 @@ MergeFailure classifyMergeFailure(rpc::RpcFailureKind kind, const std::string& m
|
|||||||
return MergeFailure::Fatal;
|
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)
|
rpc::RpcFailureKind mergeCallFailureKind(const std::exception& e)
|
||||||
{
|
{
|
||||||
const rpc::RpcFailureKind kind = rpc::rpcFailureKindOf(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);
|
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;
|
b.state = st;
|
||||||
if (b.addrIdx < 0 || b.addrIdx >= static_cast<int>(addrs_.size())) return;
|
if (b.addrIdx < 0 || b.addrIdx >= static_cast<int>(addrs_.size())) return;
|
||||||
@@ -307,10 +323,20 @@ void ConsolidationRun::applyOpState(RunBatch& b, OpState st, const OpReport* rep
|
|||||||
case OpState::Success:
|
case OpState::Success:
|
||||||
if (rep) b.txid = rep->txid;
|
if (rep) b.txid = rep->txid;
|
||||||
++a.opsDone;
|
++a.opsDone;
|
||||||
|
a.defers = 0; // progress made: this address's consecutive-defer budget resets
|
||||||
opFailStreak_ = 0;
|
opFailStreak_ = 0;
|
||||||
break;
|
break;
|
||||||
case OpState::Failed:
|
case OpState::Failed:
|
||||||
if (rep) b.error = rep->error;
|
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;
|
if (a.failure.empty()) a.failure = b.error;
|
||||||
a.done = true;
|
a.done = true;
|
||||||
worsen(a, AddrOutcome::Failed);
|
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.locked || now < unlockHoldUntil_) return wait(RunWait::Locked);
|
||||||
if (env.userSendBusy) return wait(RunWait::UserSend);
|
if (env.userSendBusy) return wait(RunWait::UserSend);
|
||||||
if (now < nextCallAt_) return wait(RunWait::Batch);
|
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))
|
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
|
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
|
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;
|
b.misses = 0;
|
||||||
const OpState st = opStateFromStatus(it->second->status);
|
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);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -113,6 +113,10 @@ enum class MergeFailure : std::uint8_t {
|
|||||||
Fatal, // anything else: this address is done
|
Fatal, // anything else: this address is done
|
||||||
};
|
};
|
||||||
MergeFailure classifyMergeFailure(rpc::RpcFailureKind kind, const std::string& message);
|
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
|
// 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
|
// (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.
|
// 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 kRunMaxUnlockRefusals = 2;
|
||||||
inline constexpr int kRunMaxFailStreak = 3;
|
inline constexpr int kRunMaxFailStreak = 3;
|
||||||
inline constexpr int kRunMaxPollMisses = 2;
|
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 {
|
struct RunConfig {
|
||||||
double fee = 0.0001;
|
double fee = 0.0001;
|
||||||
@@ -169,6 +177,8 @@ struct RunAddress {
|
|||||||
int opsDone = 0;
|
int opsDone = 0;
|
||||||
int transient = 0;
|
int transient = 0;
|
||||||
int unlockRefusals = 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;
|
std::string failure;
|
||||||
};
|
};
|
||||||
struct RunBatch {
|
struct RunBatch {
|
||||||
@@ -226,7 +236,7 @@ private:
|
|||||||
enum class Phase : std::uint8_t { Idle, Running, Finished };
|
enum class Phase : std::uint8_t { Idle, Running, Finished };
|
||||||
static void worsen(RunAddress& a, AddrOutcome o) { if (o > a.outcome) a.outcome = o; }
|
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 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; }
|
void halt(const std::string& msg) { halted_ = true; if (haltError_.empty()) haltError_ = msg; }
|
||||||
|
|
||||||
Phase phase_ = Phase::Idle;
|
Phase phase_ = Phase::Idle;
|
||||||
|
|||||||
@@ -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<NoteGroup> 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<OpReport> 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()
|
void testDatadirLockGate()
|
||||||
{
|
{
|
||||||
using dragonx::daemon::EmbeddedDaemon;
|
using dragonx::daemon::EmbeddedDaemon;
|
||||||
@@ -20298,6 +20344,7 @@ int main()
|
|||||||
testOwnershipRecord();
|
testOwnershipRecord();
|
||||||
testDebugLogTailRules();
|
testDebugLogTailRules();
|
||||||
testNoteConsolidation();
|
testNoteConsolidation();
|
||||||
|
testConsolidationDefer();
|
||||||
testScopeScanAndLockedNotes();
|
testScopeScanAndLockedNotes();
|
||||||
testMergeCallResult();
|
testMergeCallResult();
|
||||||
testOpReports();
|
testOpReports();
|
||||||
|
|||||||
Reference in New Issue
Block a user