perf(net): one scan budget for all wallet scans, gated on what changed
The balance, address scan, history and probe were throttled one by one at 10% each, which added up to 30-40% of dragonxd's cs_main with all of them due. services::WalletScanGovernor runs one wallet scan at a time across both RPC lanes and keeps one shared quiet clock at 15% of wall-clock (cut when the node is behind). Waiters are served first in, first out, with a head start for what the page shows and a starvation floor of max(30 s visible / 900 s hidden, 15x the stream's cost). Explicit work (a dirty set, the user's send, a first load) is never refused but is charged. Routine scans run only when the tip, the mempool size (getmempoolinfo: no cs_main) or a wallet-mutation epoch changed since their last good scan, or a safety sweep is due. Pages that don't show the balance poll it at most every max(10 min, 50x its cost) (2 min with client-side auto-shield); coming back into view with an old value fetches it at once. The staleness badge counts from the last poll that confirmed nothing changed, and hides while the budget holds the balance within its guarantee. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
+212
-86
@@ -842,7 +842,9 @@ void App::onDisconnected(const std::string& reason)
|
||||
last_total_balance_zat_.reset();
|
||||
history_transparent_rows_ = 0; // the history's scan state (the pass, epochs and watermark are
|
||||
history_transparent_tip_ = -1; // reset with the cache session below); cost estimates are kept
|
||||
last_tx_scan_at_ = 0;
|
||||
scan_governor_.reset(); // tickets of this connection's scans complete as no-ops
|
||||
recent_probe_rotation_.reset();
|
||||
mempool_known_ = false;
|
||||
ui::ShieldDialog::invalidateScope();
|
||||
address_validation_cache_dirty_ = true;
|
||||
resetTransactionHistoryCacheSession();
|
||||
@@ -1015,39 +1017,30 @@ bool App::walletScanReady() const
|
||||
return !rpc_busy_gate_->busy();
|
||||
}
|
||||
|
||||
// Adaptive throttle: the next balance poll must wait at least (lastScanCost / kBalanceDutyCycle) since
|
||||
// the last one, so balance scanning can never occupy more than ~kBalanceDutyCycle of wall-clock. A
|
||||
// cheap wallet (sub-cadence cost) is unaffected — the tab's Core timer stays the real cadence; a ~20s
|
||||
// scan on a large wallet backs off to roughly every ~200s instead of every 2s, freeing cs_main for
|
||||
// block connection. A wallet mutation bypasses this via force_balance_refresh_.
|
||||
bool App::scanRefreshDue(std::int64_t lastUpdate, double lastScanMs) const
|
||||
services::ScanSignals App::currentScanSignals() const
|
||||
{
|
||||
constexpr double kScanDutyCycle = 0.10;
|
||||
if (lastUpdate == 0) return true; // never fetched
|
||||
if (lastScanMs <= 0.0) return true; // no cost measured yet
|
||||
const double minInterval = (lastScanMs / 1000.0) / kScanDutyCycle;
|
||||
return std::difftime(std::time(nullptr), static_cast<std::time_t>(lastUpdate)) >= minInterval;
|
||||
services::ScanSignals sig;
|
||||
sig.tipHash = state_.sync.best_blockhash;
|
||||
sig.mempoolKnown = mempool_known_;
|
||||
sig.mempoolTxs = mempool_txs_;
|
||||
sig.mempoolBytes = mempool_bytes_;
|
||||
sig.walletEpoch = wallet_mutation_epoch_;
|
||||
return sig;
|
||||
}
|
||||
|
||||
bool App::balanceRefreshDue() const
|
||||
bool App::isBalanceHeldByScanBudget() const
|
||||
{
|
||||
return scanRefreshDue(state_.last_balance_update, last_balance_scan_ms_);
|
||||
return scan_governor_.heldWithinFloor(services::ScanStream::Balance, scanClockNow());
|
||||
}
|
||||
|
||||
// The address scan (z_listunspent) and history scan (z_listreceivedbyaddress) are just as O(mapWallet)
|
||||
// and cs_main-bound as the balance scan, so they get the same adaptive back-off. This is what keeps a
|
||||
// large synced wallet from re-scanning them every tab cadence — the near-tip starvation the balance-only
|
||||
// throttle missed (only the periodic poll is gated; explicit refreshes call the refresh directly).
|
||||
bool App::addressRefreshDue() const
|
||||
services::ScanVisibility App::currentScanVisibility() const
|
||||
{
|
||||
return scanRefreshDue(state_.last_address_update, last_address_scan_ms_);
|
||||
return services::scanVisibilityForPage(current_page_, window_background_);
|
||||
}
|
||||
|
||||
bool App::txRefreshDue() const
|
||||
double App::scanClockNow()
|
||||
{
|
||||
// Keyed on the last wallet-wide scan, not state_.last_tx_update (pending-send row edits and probe-less
|
||||
// polls stamp that too, which kept pushing the next scan out).
|
||||
return scanRefreshDue(last_tx_scan_at_, last_tx_scan_ms_);
|
||||
return std::chrono::duration<double>(std::chrono::steady_clock::now().time_since_epoch()).count();
|
||||
}
|
||||
|
||||
bool App::currentPageNeedsWalletDataRefresh() const
|
||||
@@ -1672,6 +1665,7 @@ void App::invalidateShieldedHistoryScanProgress(bool persistCache, bool keepSent
|
||||
++history_scan_epoch_; // results of scans already in flight no longer apply
|
||||
history_hint_floors_.clear();
|
||||
history_hinted_txids_.clear();
|
||||
recent_probe_rotation_.reset();
|
||||
if (!keepSentScan) {
|
||||
history_sent_scan_height_ = -1; // the next sweep covers the whole wallet again
|
||||
history_enrich_backlog_.clear();
|
||||
@@ -1908,6 +1902,7 @@ void App::refreshCoreData(bool allowBalance)
|
||||
// by the getrescaninfo / daemon-log pollers) before declaring it done.
|
||||
if (state_.sync.rescanning && rescan_confirmed_active_) {
|
||||
resetWitnessRescanProgress();
|
||||
noteWalletMutation();
|
||||
state_.sync.rescan_progress = 1.0f;
|
||||
// Notes/witnesses were rebuilt — force a fresh history + balance pull.
|
||||
markTransactionsDirty();
|
||||
@@ -1953,32 +1948,82 @@ void App::refreshCoreData(bool allowBalance)
|
||||
? fast_rpc_.get() : rpc_.get();
|
||||
if (!w || !rpc) return;
|
||||
ui::NavPage tracePage = current_page_;
|
||||
// Decide whether to include the balance call (z_gettotalbalance — O(mapWallet), holds cs_main).
|
||||
// Suppress it (a) while syncing or within the post-sync settle window, so it can't starve block
|
||||
// connection, and (b) unless enough time has elapsed given the LAST scan's measured cost, so a
|
||||
// large shielded wallet backs off automatically instead of re-scanning every couple of seconds.
|
||||
// A wallet mutation (send/shield) forces the next poll through so the user's own action updates the
|
||||
// balance immediately. Captured on the main thread to avoid reading state_ off the worker thread.
|
||||
// allowBalance=false (the locked wallet's chain-state poll) skips it and leaves a forced refresh
|
||||
// pending for after the unlock.
|
||||
const bool includeBalance = allowBalance && !effectivelySyncing() &&
|
||||
(force_balance_refresh_ || balanceRefreshDue());
|
||||
// A Core job already in flight: the enqueue would refuse this one anyway (and it may hold a ticket).
|
||||
if (network_refresh_.jobInProgress(services::NetworkRefreshService::Job::Core)) return;
|
||||
// The balance call (z_gettotalbalance — O(mapWallet), holds cs_main) is one of the wallet-wide scans
|
||||
// the governor budgets. Skipped while syncing (or settling after it) and for the locked wallet's
|
||||
// chain-state poll; only when something it depends on changed (tip, mempool, wallet mutation) or its
|
||||
// safety sweep is due; rarely on pages that don't show it. Forced (the user's own send) and first
|
||||
// fetches are explicit: never held back, but charged.
|
||||
const double scanNow = scanClockNow();
|
||||
const auto scanSig = currentScanSignals();
|
||||
services::BalancePollInputs balanceIn;
|
||||
balanceIn.allowBalance = allowBalance;
|
||||
balanceIn.syncing = effectivelySyncing();
|
||||
balanceIn.forced = force_balance_refresh_;
|
||||
balanceIn.neverFetched = state_.last_balance_update == 0;
|
||||
balanceIn.changed = scan_governor_.needed(services::ScanStream::Balance, scanSig, scanNow);
|
||||
balanceIn.visible = currentScanVisibility().balance;
|
||||
balanceIn.autoShieldArmed = settings_ && settings_->getAutoShield() && !daemon_autoshield_active_;
|
||||
balanceIn.sinceLastScanSec = scan_governor_.secondsSinceScan(services::ScanStream::Balance, scanNow);
|
||||
balanceIn.emaCostSec = scan_governor_.emaCostMs(services::ScanStream::Balance) / 1000.0;
|
||||
balanceIn.waiting = scan_governor_.waiting(services::ScanStream::Balance, scanNow);
|
||||
balanceIn.shownAgeSec = std::difftime(std::time(nullptr), static_cast<std::time_t>(
|
||||
std::max(state_.last_balance_update, state_.balance_verified_at)));
|
||||
balanceIn.visibleFloorSec = scan_governor_.visibleFloorSecFor(services::ScanStream::Balance);
|
||||
services::ScanTicket balanceTicket;
|
||||
switch (services::decideBalancePoll(balanceIn)) {
|
||||
case services::BalancePollDecision::Explicit:
|
||||
balanceTicket = scan_governor_.beginExplicit(services::ScanStream::Balance, scanSig, scanNow);
|
||||
break;
|
||||
case services::BalancePollDecision::AskGovernor:
|
||||
balanceTicket = scan_governor_.tryAdmit(services::ScanStream::Balance, balanceIn.visible, scanSig, scanNow);
|
||||
break;
|
||||
case services::BalancePollDecision::Skip:
|
||||
break;
|
||||
}
|
||||
const bool includeBalance = static_cast<bool>(balanceTicket);
|
||||
// The forced request this poll carries. The result callback clears the flag only if the balance
|
||||
// call was really sent and no newer request came in; clearing it here lost it whenever the enqueue
|
||||
// was refused (a Core job already in flight) or the worker skipped the balance (behind, busy).
|
||||
const std::uint32_t forcedSeq = (includeBalance && force_balance_refresh_) ? force_balance_seq_ : 0;
|
||||
NetworkRefreshService::CoreRefreshPlan plan;
|
||||
plan.includeBalance = includeBalance;
|
||||
plan.fetchMempool = !balanceIn.syncing; // a change signal for the scans; takes no cs_main
|
||||
plan.needSpendable = NetworkRefreshService::spendableBalanceNeeded(
|
||||
state_, !pending_send_info_.empty() || !pending_opids_.empty() || !send_txids_.empty(),
|
||||
force_balance_refresh_);
|
||||
plan.prevTotalZat = last_total_balance_zat_;
|
||||
|
||||
auto enqueued = network_refresh_.enqueue(services::NetworkRefreshService::Job::Core, *w, [this, rpc, tracePage, plan, forcedSeq]() -> rpc::RPCWorker::MainCb {
|
||||
auto enqueued = network_refresh_.enqueue(services::NetworkRefreshService::Job::Core, *w, [this, rpc, tracePage, plan, forcedSeq, balanceTicket]() -> rpc::RPCWorker::MainCb {
|
||||
AppRefreshRpcGateway refreshRpc(*rpc, traceSource(tracePage, "Core refresh"));
|
||||
auto result = NetworkRefreshService::collectCoreRefreshResult(refreshRpc, plan);
|
||||
return [this, result, forcedSeq]() {
|
||||
return [this, result, forcedSeq, balanceTicket]() {
|
||||
try {
|
||||
if (result.blockchainOk) {
|
||||
// The scan budget shrinks while the node is behind (every chain read, not only those
|
||||
// that carried a balance).
|
||||
int lag = 0;
|
||||
if (result.blocks && result.longestChain && *result.longestChain > 0)
|
||||
lag = std::max(0, *result.longestChain - *result.blocks);
|
||||
scan_governor_.setTipLag(lag);
|
||||
}
|
||||
if (balanceTicket) {
|
||||
// The balance is current for what the worker saw just before it: its tip and mempool.
|
||||
services::ScanTicket t = balanceTicket;
|
||||
if (result.bestBlockHash) t.basis.tipHash = *result.bestBlockHash;
|
||||
if (result.mempoolFetched) {
|
||||
t.basis.mempoolKnown = result.mempoolOk;
|
||||
t.basis.mempoolTxs = result.mempoolTxs;
|
||||
t.basis.mempoolBytes = result.mempoolBytes;
|
||||
}
|
||||
const auto outcome = !result.balanceAttempted ? services::ScanOutcome::Skipped
|
||||
: result.balanceOk ? services::ScanOutcome::Ok
|
||||
: result.balanceErrKind == rpc::RpcFailureKind::ReplyTimeout ? services::ScanOutcome::TimedOut
|
||||
: services::ScanOutcome::Failed;
|
||||
scan_governor_.complete(t, result.balanceScanMs, outcome, scanClockNow());
|
||||
}
|
||||
if (result.balanceAttempted) state_.balance_last_attempt_failed = !result.balanceOk;
|
||||
NetworkRefreshService::applyCoreRefreshResult(state_, result, std::time(nullptr));
|
||||
if (result.balanceOk) {
|
||||
// Fresh aggregates: re-apply the pending sends' debits to them. The per-address
|
||||
@@ -1988,11 +2033,17 @@ void App::refreshCoreData(bool allowBalance)
|
||||
last_total_balance_zat_ = NetworkRefreshService::nextTotalBaseline(result, last_total_balance_zat_);
|
||||
if (forcedSeq != 0 && result.balanceAttempted && forcedSeq == force_balance_seq_)
|
||||
force_balance_refresh_ = false;
|
||||
if (result.mempoolFetched) {
|
||||
mempool_known_ = result.mempoolOk;
|
||||
mempool_txs_ = result.mempoolTxs;
|
||||
mempool_bytes_ = result.mempoolBytes;
|
||||
}
|
||||
// No balance call, but nothing the balance depends on changed since the last one: what's
|
||||
// shown is current (the staleness badge counts from here).
|
||||
if (!result.balanceAttempted && result.blockchainOk && state_.last_balance_update != 0 &&
|
||||
scan_governor_.signalsMatch(services::ScanStream::Balance, currentScanSignals()))
|
||||
state_.balance_verified_at = std::time(nullptr);
|
||||
|
||||
// Feed the adaptive balance throttle + sync-settle hysteresis. Record the last scan's
|
||||
// cost (0 when balance was skipped), and arm the settle window only on the
|
||||
// syncing→caught-up edge so a wallet synced from the start is never throttled at connect.
|
||||
if (result.balanceScanMs > 0.0) last_balance_scan_ms_ = result.balanceScanMs;
|
||||
// Sticky-behind hysteresis: hold the low-impact profile for kSyncSettleSeconds after the
|
||||
// LAST time we read "behind" — not only on the caught-up edge. While actually syncing we
|
||||
// re-arm every Core tick, so the profile stays engaged continuously; once we truly reach
|
||||
@@ -2118,16 +2169,23 @@ void App::refreshCoreData(bool allowBalance)
|
||||
}
|
||||
};
|
||||
}, 3);
|
||||
if (!enqueued.enqueued) return;
|
||||
if (!enqueued.enqueued && balanceTicket) scan_governor_.cancel(balanceTicket);
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Address Data: z/t address lists + per-address balances
|
||||
// ============================================================================
|
||||
|
||||
void App::refreshAddressData()
|
||||
void App::refreshAddressData(services::ScanTicket ticket)
|
||||
{
|
||||
if (!worker_ || !rpc_ || !state_.connected) return;
|
||||
if (!worker_ || !rpc_ || !state_.connected ||
|
||||
network_refresh_.jobInProgress(services::NetworkRefreshService::Job::Addresses)) {
|
||||
if (ticket) scan_governor_.cancel(ticket); // not sent after all
|
||||
return;
|
||||
}
|
||||
// No ticket: an explicit refresh (a dirty set, a send in progress, tab entry). Never held back, but
|
||||
// charged to the scan budget like the routine ones.
|
||||
if (!ticket) ticket = scan_governor_.beginExplicit(services::ScanStream::Addresses, currentScanSignals(), scanClockNow());
|
||||
const std::size_t previousAddressCount = state_.z_addresses.size() + state_.t_addresses.size();
|
||||
const std::string previousWalletIdentity = transactionHistoryCacheWalletIdentity();
|
||||
const auto zList = shieldedAddressList();
|
||||
@@ -2136,14 +2194,16 @@ void App::refreshAddressData()
|
||||
? NetworkRefreshService::AddressRefreshSnapshot{}
|
||||
: NetworkRefreshService::buildAddressRefreshSnapshot(state_);
|
||||
ui::NavPage tracePage = current_page_;
|
||||
auto enqueued = network_refresh_.enqueue(services::NetworkRefreshService::Job::Addresses, *worker_, [this, previousAddressCount, previousWalletIdentity, previousZ = std::move(previousZ), addressSnapshot = std::move(addressSnapshot), tracePage]() -> rpc::RPCWorker::MainCb {
|
||||
auto enqueued = network_refresh_.enqueue(services::NetworkRefreshService::Job::Addresses, *worker_, [this, previousAddressCount, previousWalletIdentity, previousZ = std::move(previousZ), addressSnapshot = std::move(addressSnapshot), tracePage, ticket]() -> rpc::RPCWorker::MainCb {
|
||||
AppRefreshRpcGateway refreshRpc(*rpc_, traceSource(tracePage, "Address refresh"));
|
||||
auto result = NetworkRefreshService::collectAddressRefreshResult(refreshRpc, addressSnapshot);
|
||||
|
||||
return [this, previousAddressCount, previousWalletIdentity, previousZ, result = std::move(result)]() mutable {
|
||||
return [this, previousAddressCount, previousWalletIdentity, previousZ, ticket, result = std::move(result)]() mutable {
|
||||
const bool addrListOk = result.addressListOk; // capture before the move
|
||||
const double scanMs = result.scanMs; // feed the adaptive throttle (addressRefreshDue)
|
||||
if (scanMs > 0.0) last_address_scan_ms_ = scanMs;
|
||||
scan_governor_.complete(ticket, result.scanMs,
|
||||
(addrListOk && result.balancesOk) ? services::ScanOutcome::Ok
|
||||
: result.replyTimedOut ? services::ScanOutcome::TimedOut : services::ScanOutcome::Failed,
|
||||
scanClockNow());
|
||||
// #3: this scan already listed the wallet's unspent notes — feed the chat send-budget from it
|
||||
// (dedups the dedicated chat z_listunspent). Grab it before `result` is moved below.
|
||||
auto unspentNotes = std::move(result.unspentNotes);
|
||||
@@ -2186,6 +2246,7 @@ void App::refreshAddressData()
|
||||
}
|
||||
history_listed_z_ = std::move(listedZ);
|
||||
}
|
||||
if (addressSetChanged) scan_governor_.invalidate(services::ScanStream::Balance); // the totals moved
|
||||
if (state_.transactions.empty() || addressSetChanged) {
|
||||
markTransactionsDirty();
|
||||
last_tx_block_height_ = -1;
|
||||
@@ -2204,20 +2265,24 @@ void App::refreshAddressData()
|
||||
updateWalletIndexForActiveWallet(/*markOpened=*/false);
|
||||
};
|
||||
}, 3);
|
||||
if (!enqueued.enqueued) return;
|
||||
if (!enqueued.enqueued) scan_governor_.cancel(ticket);
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Transaction Data: transparent + shielded receives + z_viewtransaction enrichment
|
||||
// ============================================================================
|
||||
|
||||
void App::refreshTransactionData()
|
||||
bool App::refreshTransactionData()
|
||||
{
|
||||
using services::ScanStream;
|
||||
using services::TxCycle;
|
||||
// The one entry point for history refreshes (the periodic timer, connect, tab entry, the address
|
||||
// callback's re-route), so every caller gets the same rules. Wallet-wide scans hold the daemon's
|
||||
// cs_main: never while it can't take them.
|
||||
if (!worker_ || !rpc_ || !rpc_->isConnected() || !state_.connected) return;
|
||||
if (state_.isLocked() || state_.warming_up || runtime_rescan_active_ || rpc_busy_gate_->busy()) return;
|
||||
if (!worker_ || !rpc_ || !rpc_->isConnected() || !state_.connected) return false;
|
||||
if (state_.isLocked() || state_.warming_up || runtime_rescan_active_ || rpc_busy_gate_->busy()) return false;
|
||||
// A cycle in flight: the enqueue would refuse this one.
|
||||
if (network_refresh_.jobInProgress(services::NetworkRefreshService::Job::Transactions)) return false;
|
||||
|
||||
const int tip = state_.sync.blocks;
|
||||
const auto work = historyScanWork();
|
||||
@@ -2228,32 +2293,67 @@ void App::refreshTransactionData()
|
||||
in.sendTracking = !send_txids_.empty();
|
||||
in.explicitScanWork = work.explicitWork;
|
||||
in.routineScanWork = work.routineWork;
|
||||
in.gateOpen = txRefreshDue();
|
||||
in.syncing = effectivelySyncing();
|
||||
in.heightKnown = tip > 0;
|
||||
in.recentDue = currentPageNeedsWalletDataRefresh() && shouldRefreshRecentTransactions();
|
||||
const auto cycle = services::decideTxCycle(in);
|
||||
// What this cycle is without and with a routine wallet-wide scan: the scan budget (the governor)
|
||||
// decides between the two.
|
||||
in.gateOpen = false;
|
||||
const TxCycle held = services::decideTxCycle(in);
|
||||
in.gateOpen = true;
|
||||
const TxCycle open = services::decideTxCycle(in);
|
||||
|
||||
const double now = scanClockNow();
|
||||
const auto sig = currentScanSignals();
|
||||
TxCycle cycle = held;
|
||||
bool routineAllowed = false;
|
||||
bool heldBack = false;
|
||||
services::ScanTicket ticket;
|
||||
// The probe only when something changed since its last run (tip, mempool, a wallet mutation).
|
||||
const bool wantsProbe = open == TxCycle::RecentWithProbe && scan_governor_.needed(ScanStream::RecentTx, sig, now);
|
||||
if (open == TxCycle::Full || wantsProbe) {
|
||||
// Explicit work runs anyway (held is Full then); admission lets routine work ride along in it.
|
||||
ticket = scan_governor_.tryAdmit(open == TxCycle::Full ? ScanStream::History : ScanStream::RecentTx,
|
||||
currentScanVisibility().history, sig, now);
|
||||
if (ticket) {
|
||||
cycle = open;
|
||||
routineAllowed = true;
|
||||
} else {
|
||||
heldBack = held != open;
|
||||
}
|
||||
}
|
||||
switch (cycle) {
|
||||
case services::TxCycle::None:
|
||||
return;
|
||||
case services::TxCycle::Recent:
|
||||
case services::TxCycle::RecentWithProbe:
|
||||
startRecentTransactionPoll(cycle == services::TxCycle::RecentWithProbe);
|
||||
return;
|
||||
case services::TxCycle::Full:
|
||||
case services::TxCycle::Light:
|
||||
case TxCycle::None:
|
||||
return heldBack;
|
||||
case TxCycle::Recent:
|
||||
startRecentTransactionPoll(false, {});
|
||||
return heldBack;
|
||||
case TxCycle::RecentWithProbe:
|
||||
startRecentTransactionPoll(true, ticket);
|
||||
return false;
|
||||
case TxCycle::Full:
|
||||
case TxCycle::Light:
|
||||
if (addresses_dirty_) {
|
||||
if (ticket) scan_governor_.cancel(ticket);
|
||||
// The history needs the current address list; the address callback marks it due again.
|
||||
refreshAddressData();
|
||||
network_refresh_.markDue(services::NetworkRefreshService::Timer::Transactions);
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
startTransactionCycle(cycle == services::TxCycle::Full, in.gateOpen);
|
||||
return;
|
||||
// Explicit work, a mutation's Light cycle, the first load: never held back, but charged. Only the first
|
||||
// load's (or an invalidation's) pass does the routine work too: the others leave History waiting for it,
|
||||
// and their cost out of its estimate.
|
||||
if (!ticket) {
|
||||
ticket = scan_governor_.beginExplicit(ScanStream::History, sig, now,
|
||||
cycle == TxCycle::Full && !initial_history_scan_complete_);
|
||||
}
|
||||
startTransactionCycle(cycle == TxCycle::Full, routineAllowed, ticket);
|
||||
return heldBack;
|
||||
}
|
||||
return heldBack;
|
||||
}
|
||||
|
||||
void App::startTransactionCycle(bool full, bool routineAllowed)
|
||||
void App::startTransactionCycle(bool full, bool routineAllowed, services::ScanTicket ticket)
|
||||
{
|
||||
const int currentBlocks = state_.sync.blocks;
|
||||
auto transactionSnapshot = NetworkRefreshService::buildTransactionRefreshSnapshot(
|
||||
@@ -2322,13 +2422,13 @@ void App::startTransactionCycle(bool full, bool routineAllowed)
|
||||
|
||||
auto enqueued = network_refresh_.enqueue(services::NetworkRefreshService::Job::Transactions, *worker_, [this, currentBlocks, full,
|
||||
transactionSnapshot = std::move(transactionSnapshot), tracePage, scanEpoch, sentEpoch, dirtyGen,
|
||||
snapshotHadZ]() -> rpc::RPCWorker::MainCb {
|
||||
snapshotHadZ, ticket]() -> rpc::RPCWorker::MainCb {
|
||||
AppRefreshRpcGateway refreshRpc(*rpc_, traceSource(tracePage, full ? "Transaction refresh" : "Transaction refresh (light)"),
|
||||
this);
|
||||
auto result = NetworkRefreshService::collectTransactionRefreshResult(
|
||||
refreshRpc, transactionSnapshot, currentBlocks, MAX_VIEWTX_PER_CYCLE);
|
||||
|
||||
return [this, full, scanEpoch, sentEpoch, dirtyGen, snapshotHadZ, result = std::move(result)]() mutable {
|
||||
return [this, full, scanEpoch, sentEpoch, dirtyGen, snapshotHadZ, ticket, result = std::move(result)]() mutable {
|
||||
const std::time_t now = std::time(nullptr);
|
||||
const int resultTip = result.blockHeight;
|
||||
const std::size_t attempted = result.shieldedAddressesScanned;
|
||||
@@ -2337,14 +2437,16 @@ void App::startTransactionCycle(bool full, bool routineAllowed)
|
||||
// Only history_sent_epoch_ invalidates the sent side (watermark, backoff, backlog): a plain
|
||||
// refresh keeps the sent scan, so it doesn't throw away an in-flight whole-wallet sweep.
|
||||
const bool sameSentEpoch = sentEpoch == history_sent_epoch_;
|
||||
// Feed the throttle (txRefreshDue): Full cycles are the wallet-wide scans it meters. Not the
|
||||
// whole-wallet sweep, which has its own cadence (once per connect or full invalidation, with a
|
||||
// backoff): charging up to 120 s of it held routine scans back for 10x that after every connect.
|
||||
if (full) {
|
||||
// Charge the scan budget with this cycle's cost, except a whole-wallet sweep, which has its own
|
||||
// cadence (once per connect or full invalidation, with a backoff): charging up to 120 s of it held
|
||||
// routine scans back long after every connect.
|
||||
{
|
||||
double charged = result.scanMs;
|
||||
if (result.sentSweepRan && result.sentSweepFull) charged = std::max(1.0, result.scanMs - result.sentSweepMs);
|
||||
if (charged > 0.0) last_tx_scan_ms_ = charged;
|
||||
last_tx_scan_at_ = now;
|
||||
scan_governor_.complete(ticket, charged,
|
||||
!result.rpcError ? services::ScanOutcome::Ok
|
||||
: result.replyTimedOut ? services::ScanOutcome::TimedOut : services::ScanOutcome::Failed,
|
||||
scanClockNow());
|
||||
}
|
||||
if (sameScanEpoch) {
|
||||
if (succeeded > 0) {
|
||||
@@ -2434,7 +2536,10 @@ void App::startTransactionCycle(bool full, bool routineAllowed)
|
||||
maybeFinishTransactionSendProgress();
|
||||
};
|
||||
}, 3);
|
||||
if (!enqueued.enqueued) return;
|
||||
if (!enqueued.enqueued) {
|
||||
scan_governor_.cancel(ticket);
|
||||
return;
|
||||
}
|
||||
|
||||
network_refresh_.resetTxAge();
|
||||
}
|
||||
@@ -2463,7 +2568,7 @@ void App::pruneTrackedSendTxids()
|
||||
}
|
||||
}
|
||||
|
||||
void App::startRecentTransactionPoll(bool probe)
|
||||
void App::startRecentTransactionPoll(bool probe, services::ScanTicket ticket)
|
||||
{
|
||||
const int currentBlocks = state_.sync.blocks;
|
||||
auto transactionSnapshot = NetworkRefreshService::buildTransactionRefreshSnapshot(
|
||||
@@ -2471,31 +2576,46 @@ void App::startRecentTransactionPoll(bool probe)
|
||||
transactionSnapshot.pendingOpids.insert(pending_opids_.begin(), pending_opids_.end());
|
||||
if (settings_) transactionSnapshot.miningAddresses = settings_->getMiningAddresses();
|
||||
transactionSnapshot.shieldedScanHeights = shielded_history_scan_heights_;
|
||||
// Between blocks: recent listtransactions rows, plus (when the scan throttle allows) one receive scan
|
||||
// of the address scanned longest ago. It no longer advances the pass cursor.
|
||||
// Between blocks: recent listtransactions rows, plus (when the scan budget allows) one receive scan: the
|
||||
// oldest-scanned address this rotation hasn't probed yet. It doesn't advance the pass cursor.
|
||||
transactionSnapshot.skipShieldedScans = !probe;
|
||||
transactionSnapshot.maxShieldedReceiveScans = 1;
|
||||
transactionSnapshot.shieldedScanStartIndex = probe
|
||||
? services::oldestScannedAddressIndex(transactionSnapshot.shieldedAddresses, shielded_history_scan_heights_)
|
||||
: 0;
|
||||
transactionSnapshot.shieldedScanStartIndex = 0;
|
||||
std::string probeTarget;
|
||||
if (probe) {
|
||||
const auto& addresses = transactionSnapshot.shieldedAddresses;
|
||||
probeTarget = recent_probe_rotation_.next(addresses, shielded_history_scan_heights_);
|
||||
const auto it = std::find(addresses.begin(), addresses.end(), probeTarget);
|
||||
if (it != addresses.end()) transactionSnapshot.shieldedScanStartIndex = static_cast<std::size_t>(it - addresses.begin());
|
||||
}
|
||||
const std::uint64_t scanEpoch = history_scan_epoch_;
|
||||
ui::NavPage tracePage = current_page_;
|
||||
|
||||
auto enqueued = network_refresh_.enqueue(services::NetworkRefreshService::Job::Transactions, *worker_, [this, currentBlocks,
|
||||
transactionSnapshot = std::move(transactionSnapshot), tracePage, probe, scanEpoch]() -> rpc::RPCWorker::MainCb {
|
||||
transactionSnapshot = std::move(transactionSnapshot), tracePage, probe, scanEpoch, ticket,
|
||||
probeTarget]() -> rpc::RPCWorker::MainCb {
|
||||
AppRefreshRpcGateway refreshRpc(*rpc_, traceSource(tracePage, "Recent transaction poll"));
|
||||
auto result = NetworkRefreshService::collectRecentTransactionRefreshResult(
|
||||
refreshRpc, transactionSnapshot, currentBlocks);
|
||||
|
||||
return [this, probe, scanEpoch, result = std::move(result)]() mutable {
|
||||
return [this, probe, scanEpoch, ticket, probeTarget, result = std::move(result)]() mutable {
|
||||
// The poll doesn't consume the tip: newTip stays set until a Full or Light cycle applies at it,
|
||||
// so the Full cycle still runs once the throttle allows. (Read here, not at dispatch, so a reset
|
||||
// to -1 meanwhile is kept.)
|
||||
int pollTip = last_tx_block_height_;
|
||||
if (ticket) {
|
||||
// The probe is a wallet-wide scan: charged to the scan budget. Partial until the rotation has
|
||||
// probed every z-address since what it watches changed (the change may be at any of them).
|
||||
auto outcome = !result.rpcError ? services::ScanOutcome::Ok
|
||||
: result.replyTimedOut ? services::ScanOutcome::TimedOut : services::ScanOutcome::Failed;
|
||||
if (outcome == services::ScanOutcome::Ok && probe) {
|
||||
if (scanEpoch == history_scan_epoch_ && result.shieldedScansSucceeded > 0 && !probeTarget.empty())
|
||||
recent_probe_rotation_.probed(probeTarget);
|
||||
if (!recent_probe_rotation_.complete()) outcome = services::ScanOutcome::Partial;
|
||||
}
|
||||
scan_governor_.complete(ticket, result.scanMs, outcome, scanClockNow());
|
||||
}
|
||||
if (probe && result.shieldedAddressesScanned > 0) {
|
||||
// The probe is a wallet-wide scan: it counts for the throttle.
|
||||
if (result.scanMs > 0.0) last_tx_scan_ms_ = result.scanMs;
|
||||
last_tx_scan_at_ = std::time(nullptr);
|
||||
if (scanEpoch == history_scan_epoch_ && result.shieldedScansSucceeded > 0) {
|
||||
history_addr_scan_ms_ = services::updateScanCostEstimateMs(
|
||||
history_addr_scan_ms_, result.receiveScanOkMs / static_cast<double>(result.shieldedScansSucceeded));
|
||||
@@ -2540,7 +2660,10 @@ void App::startRecentTransactionPoll(bool probe)
|
||||
maybeFinishTransactionSendProgress();
|
||||
};
|
||||
}, 3);
|
||||
if (!enqueued.enqueued) return;
|
||||
if (!enqueued.enqueued) {
|
||||
if (ticket) scan_governor_.cancel(ticket);
|
||||
return;
|
||||
}
|
||||
|
||||
network_refresh_.resetTxAge();
|
||||
}
|
||||
@@ -3454,6 +3577,7 @@ void App::invalidateAddressValidationCache()
|
||||
{
|
||||
address_validation_cache_dirty_ = true;
|
||||
addresses_dirty_ = true;
|
||||
noteWalletMutation(); // an import or sweep: the balance changes without the tip or mempool showing it
|
||||
invalidateShieldedHistoryScanProgress(true);
|
||||
}
|
||||
|
||||
@@ -6085,6 +6209,7 @@ void App::submitZSendMany(const std::string& from, const std::string& to, double
|
||||
markTransactionsDirty();
|
||||
last_tx_block_height_ = -1;
|
||||
requestForcedBalanceRefresh(); // the user's own send must update the balance now, past the throttle
|
||||
noteWalletMutation();
|
||||
network_refresh_.markWalletMutationRefresh();
|
||||
// z_sendmany only returned an opid: the transaction is built/signed/
|
||||
// broadcast asynchronously by the daemon. Defer the user-facing
|
||||
@@ -6280,6 +6405,7 @@ void App::runtimeRescan(int startHeight)
|
||||
if (ok) {
|
||||
state_.sync.rescan_progress = 1.0f;
|
||||
markTransactionsDirty();
|
||||
noteWalletMutation();
|
||||
last_tx_block_height_ = -1;
|
||||
invalidateShieldedHistoryScanProgress(true);
|
||||
ui::Notifications::instance().success("Blockchain rescan complete");
|
||||
|
||||
Reference in New Issue
Block a user