DragonX/ObsidianDragon
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity

feat(merge): run per-address consolidation as a paced background job

Browse Source 
"Keep in each address" posted one task that called z_mergetoaddress for
every batch of every address back to back; each call is a whole-wallet
scan under cs_main, and the daemon runs the ops one at a time anyway.
It is now an App-owned services::ConsolidationRun:

- one RPC at a time; the next merge only after the previous batch left the
  queue and a gap as long as the last call took (2-120 s);
- pauses while the node syncs, the wallet is locked, a send of ours is in
  progress, or another wallet scan is running;
- failures classified: nothing that may have queued is ever resent; a
  breaker stops after 3 dead addresses or 3 failed ops in a row;
- each merge is charged to the scan budget; one re-read and one forced
  balance at the end;
- it keeps running when the dialog closes: a status-bar chip, a run view
  with progress, the wait reason and a Stop button.

The "Into one" merge can no longer be confirmed twice while its call is
in flight. fake_dragonxd gains FAKE_SCAN_SECS, FAKE_MERGE_SERIAL and
FAKE_MERGE_FAIL.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DanS
2026-10-09 11:06:34 -05:00
co-authored by Claude Opus 5.5
parent b29a09eaa0
commit cee8b9d7f6
19 changed files with 1875 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app_network.cpp
+188
View File
@@ -690,6 +690,8 @@ void App::tryConnect()
void App::onConnected()
{
maint_abort_.store(false); // maintenance calls queued before this connect have already failed (FIFO)
++rpc_session_;
if (consolidation_.abort()) ui::Notifications::instance().warning(TR("merge_run_aborted_toast"));
state_.connected = true;
ui::ShieldDialog::invalidateScope(); // inputs read before this connect may be another wallet's
const bool completedSwitch = wallet_switch_pending_confirm_.exchange(false); // this connect confirms a switch
@@ -873,6 +875,8 @@ void App::onDisconnected(const std::string& reason)
first_balance_read_tried_ = false;
address_owed_next_try_ = {};
mempool_known_ = false;
++rpc_session_;
if (consolidation_.abort()) ui::Notifications::instance().warning(TR("merge_run_aborted_toast"));
ui::ShieldDialog::invalidateScope();
address_validation_cache_dirty_ = true;
resetTransactionHistoryCacheSession();
@@ -1352,6 +1356,17 @@ void App::markWalletMutated(bool forceBalance)
network_refresh_.markWalletMutationRefresh(); // Core / Transactions / Addresses due now
}
services::ScanTicket App::beginAdhocWalletScan()
{
return scan_governor_.beginExplicit(services::ScanStream::Adhoc, currentScanSignals(), scanClockNow(),
/*servesRoutine=*/false);
}
void App::endAdhocWalletScan(const services::ScanTicket& ticket, double costMs, services::ScanOutcome outcome)
{
scan_governor_.complete(ticket, costMs, outcome, scanClockNow());
}
bool App::invokeSendResultCallback(const std::string& opid, bool ok,
const std::string& result)
{
@@ -1511,6 +1526,10 @@ void App::switchToWallet(const std::string& walletFile, bool stopDaemonConfirmed
ui::Notifications::instance().warning("Finish or cancel the seed migration before switching wallets.");
return;
}
if (consolidation_.active()) {
ui::Notifications::instance().warning(TR("merge_run_busy"));
return;
}
if (!isUsingEmbeddedDaemon()) {
ui::Notifications::instance().warning("Switching wallets needs the embedded daemon — stop any external dragonxd first.");
return;
@@ -5468,6 +5487,167 @@ void App::maybeRemindSeedBackup()
// shrinks in place and shielded-note witness data accumulates, so a mining/shielded wallet can grow
// unbounded. Fires ONCE (persisted flag) a warning toast + a clickable "Consolidate notes…" entry in the
// bell/alert panel that opens Merge to Address; re-arms if the file later drops back under the threshold.
bool App::startConsolidation(const std::vector<services::NoteGroup>& groups, double fee, int batchLimit)
{
auto& n = ui::Notifications::instance();
if (isLiteBuild() || capture_mode_ || shutting_down_.load()) return false;
if (consolidation_.active()) {
n.warning(TR("merge_run_busy"));
return false;
}
if (show_seed_migration_ || (settings_ && settings_->getSeedMigrationPending())) {
n.warning(TR("merge_run_migration_pending"));
return false;
}
if (!rpc_ || !rpc_->isConnected() || !worker_ || !state_.connected || isWalletSwitchInProgress()) {
n.error(TR("merge_run_start_failed"));
return false;
}
services::RunConfig cfg;
cfg.fee = fee;
cfg.limit = batchLimit;
if (!consolidation_.start(groups, cfg, services::steadySeconds())) {
n.warning(TR("merge_keep_each_select"));
return false;
}
n.success(TR("merge_started"));
return true; // syncing, the lock, a busy node and pending sends are the run's own pauses
}
void App::pumpConsolidation()
{
if (!consolidation_.active()) return;
const double now = services::steadySeconds();
const services::ScanContext ctx = scanContext();
const bool unlockExpired = state_.encrypted && !state_.locked && state_.unlocked_until > 0 &&
std::time(nullptr) >= static_cast<std::time_t>(state_.unlocked_until);
services::RunEnv env;
// Not walletScanReady(): the run is the user's job (no background hold, no connect gate), and it tells its
// pauses apart. Not rpc_->isWarmingUp(): that client flag sticks for the session.
env.nodeUsable = !isLiteBuild() && !shutting_down_.load() && !daemon_restarting_.load() && worker_ &&
ctx.connected && !ctx.warmingUp && !ctx.runtimeRescan && !state_.sync.rescanning;
env.busy = ctx.rpcBusy;
env.callsAllowed = !ctx.syncing;
env.locked = ctx.locked || unlockExpired;
env.userSendBusy = hasTransactionSendProgress(); // a send / shield / auto-shield / chat op of ours goes first
env.scanBusy = scan_governor_.anyInFlight(); // never stack a merge scan on another GUI scan (both lanes)
if ((consolidation_.consumeLockResyncRequest() || unlockExpired) && now >= run_lock_resync_at_) {
run_lock_resync_at_ = now + 30.0;
refreshEncryptionState();
}
const services::RunStep step = consolidation_.next(env, now);
if (step.action == services::RunAction::Call) postConsolidationCall(step.addrIdx);
else if (step.action == services::RunAction::Poll) postConsolidationPoll();
else if (step.action == services::RunAction::Finish) finishConsolidation();
}
void App::postConsolidationCall(int addrIdx)
{
if (!worker_ || !rpc_) return; // before markCallPosted: asked again next update
const auto& addrs = consolidation_.addresses();
if (addrIdx < 0 || addrIdx >= static_cast<int>(addrs.size())) return;
const std::string address = addrs[addrIdx].group.address;
const std::uint64_t gen = consolidation_.gen();
const std::uint64_t session = rpc_session_;
const double fee = consolidation_.config().fee;
const int limit = consolidation_.config().limit;
const services::ScanTicket ticket = beginAdhocWalletScan(); // GetFilteredNotes: a whole-wallet scan
consolidation_.markCallPosted(addrIdx, services::steadySeconds());
worker_->post([this, rpc = rpc_.get(), address, addrIdx, gen, fee, limit, session, ticket]() -> rpc::RPCWorker::MainCb {
services::MergeCallResult r;
rpc::RpcFailureKind kind = rpc::RpcFailureKind::None;
const auto t0 = std::chrono::steady_clock::now();
try {
MaintenanceCallScope maint(*this); // a disconnect aborts it, or refuses it unsent
rpc::RPCClient::TraceScope trace("Consolidate / Merge batch");
// fromaddresses = [addr], toaddress = addr: the address's notes merge back into it.
const nlohmann::json res = rpc->call("z_mergetoaddress",
{nlohmann::json::array({address}), address, fee, limit, limit}, services::kRunCallTimeoutSec);
services::readMergeCallResult(res, r);
} catch (const std::exception& e) {
r = services::MergeCallResult{};
r.error = e.what();
if (r.error.empty()) r.error = "error";
kind = services::mergeCallFailureKind(e); // sent but cut off: uncertain, never resent
} catch (...) {
r = services::MergeCallResult{};
r.error = "unknown error";
kind = rpc::RpcFailureKind::Other;
}
const double secs = std::chrono::duration<double>(std::chrono::steady_clock::now() - t0).count();
return [this, gen, addrIdx, r, kind, secs, session, ticket]() {
endAdhocWalletScan(ticket, secs * 1000.0, services::chatScanOutcome(r.error.empty(), kind));
const std::string orphan = consolidation_.onCallResult(gen, addrIdx, r, kind, secs, services::steadySeconds());
if (!orphan.empty() && session == rpc_session_) trackOperation(orphan); // never lose an accepted merge
};
});
}
void App::postConsolidationPoll()
{
if (!worker_ || !rpc_) return;
const std::vector<std::string> ids = consolidation_.outstandingOpids();
const auto filter = NetworkRefreshService::buildOperationFilterParams(ids);
if (!filter) return;
const std::set<std::string> wanted(ids.begin(), ids.end());
const std::uint64_t gen = consolidation_.gen();
consolidation_.markPollPosted(services::steadySeconds());
// Not charged: z_getoperationstatus reads only the async ops map. It needs the filter fallback itself: the
// shared opid poller only learns that the filter is unsupported when it has ops of its own.
worker_->post([this, rpc = rpc_.get(), params = *filter, wanted, gen]() -> rpc::RPCWorker::MainCb {
bool ok = false;
std::vector<services::OpReport> reports;
try {
rpc::RPCClient::TraceScope trace("Consolidate / Batch status");
nlohmann::json result;
if (!opid_filter_unsupported_.load(std::memory_order_relaxed)) {
try {
result = rpc->call("z_getoperationstatus", params);
} catch (const rpc::RpcError& e) {
if (e.kind() == rpc::RpcFailureKind::Warmup) throw;
result = rpc->call("z_getoperationstatus", nlohmann::json::array());
opid_filter_unsupported_.store(true, std::memory_order_relaxed);
}
} else {
result = rpc->call("z_getoperationstatus", nlohmann::json::array());
}
ok = result.is_array();
reports = services::parseOpReports(result, wanted);
} catch (...) {
ok = false;
reports.clear();
}
return [this, gen, ok, reports = std::move(reports)]() {
consolidation_.onPollResult(gen, ok, reports, services::steadySeconds());
};
});
}
void App::finishConsolidation()
{
const services::RunSummary s = consolidation_.summary();
// 2E: a batch of user ops re-reads the wallet and forces one balance, at its end.
if (s.batchesSucceeded > 0) markWalletMutated(/*forceBalance=*/true);
else if (s.batchesSubmitted > 0) noteWalletMutation();
std::string err = s.firstError.empty() ? std::string(TR("shield_unknown_error")) : s.firstError;
if (err.size() > 160) err = err.substr(0, 160) + "...";
auto& n = ui::Notifications::instance();
switch (services::runVerdict(s)) {
case services::RunVerdict::Clean: n.success(TR("shield_merge_done")); break;
case services::RunVerdict::NothingSubmitted: n.error(TR("merge_keep_each_nothing_submitted")); break;
case services::RunVerdict::SomeFailed: n.error(std::string(TR("shield_op_failed")) + err); break;
case services::RunVerdict::Unconfirmed: n.warning(TR("send_status_unconfirmed")); break;
case services::RunVerdict::Stopped: n.info(TR("merge_run_stopped_toast")); break;
case services::RunVerdict::Halted: {
char b[512];
std::snprintf(b, sizeof(b), TR("merge_run_halted_fmt"), err.c_str());
n.error(b);
break;
}
default: break; // Aborted is announced where it happens; it never reaches Finish
}
}
void App::maybeWarnLargeWallet()
{
if (capture_mode_ || lite_wallet_) return; // no live nags during a UI sweep; lite has no wallet.dat
@@ -5660,6 +5840,10 @@ void App::beginCreateSeedWallet()
void App::showSeedMigrationDialog()
{
if (consolidation_.active()) {
ui::Notifications::instance().warning(TR("merge_run_busy"));
return;
}
show_seed_migration_ = true;
// Resume a pending migration. If a sweep was already submitted (txid persisted), resume at the
// confirm/adopt stage — re-derived from the chain — rather than sweeping again; otherwise start
@@ -5765,6 +5949,10 @@ void App::refreshSeedMigrationBalance()
// huge-input wallet may leave a small remainder in the (preserved) legacy backup.
void App::beginSweepToSeedWallet()
{
if (consolidation_.active()) {
ui::Notifications::instance().warning(TR("merge_run_busy"));
return;
}
if (!rpc_ || !worker_ || !state_.connected) {
seed_migration_status_ = "Not connected to the daemon.";
seed_migration_step_ = SeedMigrationStep::Error; return;