diff --git a/CMakeLists.txt b/CMakeLists.txt index c59a53e..4a41225 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -470,6 +470,7 @@ set(APP_SOURCES src/services/network_refresh_service.cpp src/services/note_consolidation.cpp src/services/refresh_scheduler.cpp + src/services/wallet_scan_governor.cpp src/services/wallet_security_controller.cpp src/services/wallet_security_workflow.cpp src/services/wallet_security_workflow_executor.cpp @@ -617,6 +618,7 @@ set(APP_HEADERS src/services/history_scan_policy.h src/services/network_refresh_service.h src/services/refresh_scheduler.h + src/services/wallet_scan_governor.h src/services/wallet_security_controller.h src/services/wallet_security_workflow.h src/services/wallet_security_workflow_executor.h @@ -1154,6 +1156,7 @@ if(BUILD_TESTING) src/services/network_refresh_service.cpp src/services/note_consolidation.cpp src/services/refresh_scheduler.cpp + src/services/wallet_scan_governor.cpp src/services/wallet_security_controller.cpp src/services/wallet_security_workflow.cpp src/services/wallet_security_workflow_executor.cpp diff --git a/src/app.cpp b/src/app.cpp index 0689dff..428d58b 100644 --- a/src/app.cpp +++ b/src/app.cpp @@ -821,6 +821,20 @@ void App::update() if (fast_worker_) { fast_worker_->drainResults(); } + // A scan ticket whose job is gone without a result (its task threw before returning a callback) would + // hold the scan budget's one slot: complete it as timed out. Safe right after the drains: a ticket is + // issued in the same call as its job, and a job's flag clears just before its result is applied. + { + using Job = services::NetworkRefreshService::Job; + using services::ScanStream; + const double scanNow = scanClockNow(); + if (!network_refresh_.jobInProgress(Job::Core)) scan_governor_.dropOrphans(ScanStream::Balance, scanNow); + if (!network_refresh_.jobInProgress(Job::Addresses)) scan_governor_.dropOrphans(ScanStream::Addresses, scanNow); + if (!network_refresh_.jobInProgress(Job::Transactions)) { + scan_governor_.dropOrphans(ScanStream::History, scanNow); + scan_governor_.dropOrphans(ScanStream::RecentTx, scanNow); + } + } // Apply any lite-wallet refresh the controller's background worker produced (main thread). if (lite_wallet_) { @@ -1356,6 +1370,9 @@ void App::update() std::remove(pending_opids_.begin(), pending_opids_.end(), id), pending_opids_.end()); } + // An op finished (or vanished): whatever it did to the wallet, the scans gated on change + // signals must look again. + if (!terminalOpids.empty()) noteWalletMutation(); // Stale opids (no longer reported by the daemon): let any waiting send UI // know the outcome couldn't be confirmed rather than spinning forever. @@ -1418,10 +1435,11 @@ void App::update() // runtime rescan, when the daemon holds cs_main and these calls would block. if (!state_.warming_up && !runtime_rescan_active_) { // History: refreshTransactionData() decides what this cycle may scan (services::decideTxCycle): - // wallet-wide scans only as their measured cost allows, except explicit work (the first load, an + // wallet-wide scans only as the scan budget allows, except explicit work (the first load, an // invalidation, a new address); sends and other mutations get their rows through Light cycles. + // Held back by the budget, it asks again soon (a waiting stream must keep asking). if (network_refresh_.consumeDue(RefreshTimer::Transactions) && shouldRunWalletTransactionRefresh()) { - refreshTransactionData(); + if (refreshTransactionData()) network_refresh_.retrySoon(RefreshTimer::Transactions, kScanRetrySec); } // 0-conf chat: a DEDICATED ~2.5s mempool scan of the chat address, independent of the page's // (slower, 10–15s) Transactions cadence — so incoming messages surface in a few seconds. The @@ -1433,16 +1451,32 @@ void App::update() chat_fast_scan_accum_ = 0.0f; fastScanChatMemos(); } - if (network_refresh_.consumeDue(RefreshTimer::Addresses)) { + if (network_refresh_.isDue(RefreshTimer::Addresses)) { // Explicit need (a changed address set, or an in-flight send tracking its change output) - // must refresh now; the routine periodic poll on a wallet page is throttled by the last - // address scan's measured cost (addressRefreshDue) — this is the z_listunspent hammering - // that was starving cs_main while synced on a large wallet. + // refreshes now. The routine poll on a wallet page runs only when something it depends on + // changed (tip, mempool, a wallet mutation) or its safety sweep is due, and only as the scan + // budget allows; held back, it asks again soon. + bool retry = false; if (addresses_dirty_ || hasTransactionSendProgress()) { refreshAddressData(); - } else if (walletDataPage && addressRefreshDue()) { - refreshAddressData(); + } else if (walletDataPage && !effectivelySyncing()) { + using services::ScanStream; + const double scanNow = scanClockNow(); + const auto scanSig = currentScanSignals(); + if (scan_governor_.needed(ScanStream::Addresses, scanSig, scanNow)) { + if (network_refresh_.jobInProgress(services::NetworkRefreshService::Job::Addresses)) { + retry = true; + } else if (auto ticket = scan_governor_.tryAdmit(ScanStream::Addresses, + currentScanVisibility().addresses, + scanSig, scanNow)) { + refreshAddressData(ticket); + } else { + retry = true; + } + } } + if (retry) network_refresh_.retrySoon(RefreshTimer::Addresses, kScanRetrySec); + else network_refresh_.reset(RefreshTimer::Addresses); } } if (network_refresh_.consumeDue(RefreshTimer::Peers)) { @@ -5253,6 +5287,8 @@ void App::refreshNow() { // Trigger immediate refresh on all categories network_refresh_.markImmediateRefresh(); + noteWalletMutation(); // the user asked: the change-gated scans run, the balance too + requestForcedBalanceRefresh(); markTransactionsDirty(); // Force transaction list update addresses_dirty_ = true; // Force address/balance update last_tx_block_height_ = -1; // Reset tx cache diff --git a/src/app.h b/src/app.h index 69ed6bc..ac0c6b0 100644 --- a/src/app.h +++ b/src/app.h @@ -27,6 +27,7 @@ #include "daemon/daemon_busy_detect.h" #include "rpc/rpc_busy_gate.h" #include "services/history_scan_policy.h" +#include "services/wallet_scan_governor.h" #include "services/network_refresh_service.h" #include "services/wallet_security_controller.h" #include "services/wallet_security_workflow.h" @@ -439,6 +440,17 @@ public: // form (every retained op) for the rest of this connection. bool opidFilterUnsupported() const { return opid_filter_unsupported_.load(std::memory_order_relaxed); } + // Something changed the wallet that the scans' change signals (tip, mempool) don't show: a send or other + // operation finished, an import, a Console command. The change-gated scans run again. + void noteWalletMutation() { ++wallet_mutation_epoch_; } + // Minimized, hidden or occluded: no page is visible, so no scan stream gets the visible head start. + void setWindowBackground(bool background) { window_background_ = background; } + // A balance scan is in flight (the staleness badge waits for it). + bool isBalanceRefreshInFlight() const { return scan_governor_.inFlight(services::ScanStream::Balance); } + // The balance is waiting for the scan budget and still within what it promises (its floor plus one scan): + // an old value then is the budget at work, not a failing node. + bool isBalanceHeldByScanBudget() const; + // RAII around one long ("maintenance") rpc_ call made on a worker thread with a timeout above the // default 30 s, e.g. a merge or a whole-wallet scan given 120 s. rpc_ holds its curl lock for the // whole transfer and disconnect() takes the same lock, so these are counted for @@ -1348,7 +1360,6 @@ private: // off in proportion to its own measured cost. See effectivelySyncing() / balanceRefreshDue(). bool was_core_syncing_ = false; // previous Core-refresh sync state, to detect the caught-up edge std::time_t sync_settle_until_ = 0; // hold the sync-throttle until this wall-clock time (0 = not settling) - double last_balance_scan_ms_ = 0.0; // measured cost of the last z_gettotalbalance scan bool force_balance_refresh_ = false; // a wallet mutation forces the next balance poll through the throttle // Bumped with each forced-refresh request, so the poll that carried the forced balance clears the // flag only if no newer request came in meanwhile (see requestForcedBalanceRefresh()). @@ -1360,17 +1371,21 @@ private: // that actually sent the forced balance call, not at dispatch, so a dispatch the job queue refuses // (a Core job already in flight) or one that skips the balance (node behind or busy) keeps it. void requestForcedBalanceRefresh() { force_balance_refresh_ = true; ++force_balance_seq_; } - // Same adaptive back-off applied to the other two O(mapWallet) scans that hold the daemon's cs_main: - // the address scan (z_listunspent) and the history scan (z_listreceivedbyaddress). Without this, a - // large shielded wallet re-scans them every tab cadence (~seconds each), saturating cs_main and - // starving block connection near the tip (where effectivelySyncing() reads false). See - // addressRefreshDue() / txRefreshDue(); only the routine periodic poll is throttled — explicit - // refreshes (tab switch, dirty set, in-flight send) call the refresh directly and bypass this. - double last_address_scan_ms_ = 0.0; // measured cost of the last address scan - double last_tx_scan_ms_ = 0.0; // measured cost of the last wallet-wide history scan - // When that scan ran (wall-clock): the history throttle's clock. Not state_.last_tx_update, which - // every pending-send row edit and probe-less poll also stamps (that pushed the next scan out). - std::int64_t last_tx_scan_at_ = 0; + // One budget for every wallet-wide scan the GUI sends on its own (balance, addresses, history, the + // between-blocks probe): one in flight at a time across both lanes, ~15% of wall-clock in total. They + // used to be throttled one by one at 10% each, which added up to 30-40% with all of them due. + // Explicit needs (dirty sets, the user's own send, a never-loaded list) bypass it but are charged. + services::WalletScanGovernor scan_governor_; + // The change signals the routine scans are gated on, besides the tip hash (state_.sync.best_blockhash). + std::uint64_t wallet_mutation_epoch_ = 0; // bumped by noteWalletMutation() + bool mempool_known_ = false; // the last Core poll's getmempoolinfo + std::int64_t mempool_txs_ = 0; + std::int64_t mempool_bytes_ = 0; + bool window_background_ = false; // setWindowBackground() + static constexpr float kScanRetrySec = 2.0f; // a routine scan the governor held back asks again this soon + services::ScanSignals currentScanSignals() const; + services::ScanVisibility currentScanVisibility() const; + static double scanClockNow(); // the governor's clock: steady-clock seconds // Auto-clear for secrets copied to the clipboard. Only a hash of the copied secret is kept. std::uint64_t clipboard_secret_hash_ = 0; double clipboard_clear_deadline_ = 0.0; @@ -1520,6 +1535,8 @@ private: // mined (a mempool hint doesn't use that up, so the row also catches its confirmation). std::unordered_map history_hint_floors_; std::unordered_set history_hinted_txids_; + // The between-blocks probe's progress through the z-addresses since what it watches last changed. + services::ProbeRotation recent_probe_rotation_; // The z-addresses of the last successful address enumeration this session: a never-scanned address new // to it (an imported key) resets the sent sweep's watermark, even if the list in between came back blank. std::unordered_set history_listed_z_; @@ -1761,12 +1778,14 @@ private: // Private methods - data refresh void refreshData(); // Orchestrator: dispatches per-category refreshes void refreshCoreData(bool allowBalance = true); // Balance + blockchain info (can use fast_worker_) - void refreshAddressData(); // Address lists + balances + // Address lists + balances. With no ticket it is an explicit refresh (charged to the scan budget). + void refreshAddressData(services::ScanTicket ticket = {}); // Transaction history: decides the cycle (services::decideTxCycle) and starts it. Safe to call any // time: it refuses when the node can't be scanned (not connected, locked, warming up, rescanning, busy). - void refreshTransactionData(); - void startTransactionCycle(bool full, bool routineAllowed); // Full or Light cycle - void startRecentTransactionPoll(bool probe); // between-blocks poll + // Returns true when a routine scan was wanted but the scan budget held it back (ask again soon). + bool refreshTransactionData(); + void startTransactionCycle(bool full, bool routineAllowed, services::ScanTicket ticket); // Full or Light + void startRecentTransactionPoll(bool probe, services::ScanTicket ticket); // between blocks void markTransactionsDirty() { transactions_dirty_ = true; ++tx_dirty_generation_; } std::vector shieldedAddressList() const; services::ShieldedHistoryPass::Work historyScanWork() const; @@ -1780,12 +1799,6 @@ private: void refreshWalletEncryptionState(); void applyRefreshPolicy(ui::NavPage page); bool effectivelySyncing() const; // syncing, or within the post-sync settle window (hysteresis) - bool balanceRefreshDue() const; // adaptive: enough time elapsed given the last balance-scan cost? - bool addressRefreshDue() const; // same adaptive back-off for the address scan (z_listunspent) - bool txRefreshDue() const; // same adaptive back-off for wallet-wide history scans - // Shared duty-cycle rule: a scan may resume only once (lastScanMs / kScanDutyCycle) has elapsed since - // lastUpdate, so any single O(mapWallet) scan can occupy at most ~kScanDutyCycle of wall-clock. - bool scanRefreshDue(std::int64_t lastUpdate, double lastScanMs) const; bool currentPageNeedsWalletDataRefresh() const; bool shouldRunWalletTransactionRefresh() const; bool shouldRefreshRecentTransactions() const; diff --git a/src/app_network.cpp b/src/app_network.cpp index 0f8074f..a281c87 100644 --- a/src/app_network.cpp +++ b/src/app_network.cpp @@ -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(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(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::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(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(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(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"); diff --git a/src/data/wallet_state.h b/src/data/wallet_state.h index 058677a..a838e25 100644 --- a/src/data/wallet_state.h +++ b/src/data/wallet_state.h @@ -318,6 +318,12 @@ struct WalletState { // Timestamps for refresh logic int64_t last_balance_update = 0; + // A Core poll that skipped the balance confirmed it is still current (nothing it depends on changed): + // the staleness badge counts from the later of the two. + int64_t balance_verified_at = 0; + // The last balance call that was sent failed (or timed out): the staleness badge then shows at its usual + // thresholds even while the scan budget is holding the next one. + bool balance_last_attempt_failed = false; int64_t last_address_update = 0; // set when an address-list refresh applies; 0 = never loaded yet int64_t last_tx_update = 0; int64_t last_peer_update = 0; @@ -362,6 +368,8 @@ struct WalletState { // badge (and any "updated X ago" reader) briefly reports it as current until the first refresh // re-stamps it. All readers treat 0 as "never" (formatTimeAgoShort/timeAgo return ""). last_balance_update = last_address_update = last_tx_update = last_peer_update = last_mining_update = 0; + balance_verified_at = 0; + balance_last_attempt_failed = false; } // Rebuild combined addresses list from z/t lists diff --git a/src/rpc/rpc_busy_gate.h b/src/rpc/rpc_busy_gate.h index c935096..616cfd2 100644 --- a/src/rpc/rpc_busy_gate.h +++ b/src/rpc/rpc_busy_gate.h @@ -47,7 +47,8 @@ public: // Methods dragonxd runs without cs_main: always sent, and their results say nothing about cs_main. static bool isLockFree(const std::string& method) { - return method == "stop" || method == "help" || method == "getrescaninfo"; + // getmempoolinfo takes only mempool.cs (rpc/blockchain.cpp), and the dispatcher no global lock. + return method == "stop" || method == "help" || method == "getrescaninfo" || method == "getmempoolinfo"; } // May this request be sent now? Lock-free methods always; the rest only while the node isn't busy. diff --git a/src/services/history_scan_policy.cpp b/src/services/history_scan_policy.cpp index 15cb926..78a0511 100644 --- a/src/services/history_scan_policy.cpp +++ b/src/services/history_scan_policy.cpp @@ -81,6 +81,23 @@ std::size_t oldestScannedAddressIndex(const std::vector& addresses, return best; } +std::string ProbeRotation::next(const std::vector& addresses, + const std::unordered_map& heights) +{ + std::unordered_set listed(addresses.begin(), addresses.end()); + for (auto it = left_.begin(); it != left_.end();) { + if (listed.count(*it)) ++it; + else it = left_.erase(it); + } + if (left_.empty()) left_ = std::move(listed); + std::vector left; + for (const auto& address : addresses) { + if (left_.count(address)) left.push_back(address); + } + if (left.empty()) return {}; + return left[oldestScannedAddressIndex(left, heights)]; +} + int ShieldedHistoryPass::floorAt(int tip) const { return tip > 0 ? std::min(startHeight_, tip) : startHeight_; diff --git a/src/services/history_scan_policy.h b/src/services/history_scan_policy.h index 5789df1..b0b5766 100644 --- a/src/services/history_scan_policy.h +++ b/src/services/history_scan_policy.h @@ -17,6 +17,7 @@ #include #include #include +#include #include namespace dragonx { @@ -61,6 +62,22 @@ bool fullSentSweepFailureCounts(rpc::RpcFailureKind kind); std::size_t oldestScannedAddressIndex(const std::vector& addresses, const std::unordered_map& heights); +// The between-blocks probe scans one address per run. After what it watches changed (the mempool, a wallet +// mutation) it isn't done until it has probed every z-address once: the change may be any of them. Tracks +// which are left, by address (heights can't tell: a rotation at one tip leaves them all equal). +class ProbeRotation { +public: + // The next address to probe: the oldest-scanned of those left. Drops addresses no longer listed, and + // starts a new rotation when none are left. Empty when there are no addresses. + std::string next(const std::vector& addresses, const std::unordered_map& heights); + void probed(const std::string& address) { left_.erase(address); } + bool complete() const { return left_.empty(); } + void reset() { left_.clear(); } + +private: + std::unordered_set left_; +}; + // One pass of receive scans over every z-address. An address is done once it was scanned at or above the // pass's start height (its floor), so a pass ends after ceil(A / budget) cycles however the tip moves. An // explicit pass (the first load of a wallet, or after an invalidation) may run back to back; a routine diff --git a/src/services/network_refresh_service.cpp b/src/services/network_refresh_service.cpp index 7d7c5df..4517439 100644 --- a/src/services/network_refresh_service.cpp +++ b/src/services/network_refresh_service.cpp @@ -343,6 +343,25 @@ NetworkRefreshService::CoreRefreshResult NetworkRefreshService::collectCoreRefre DEBUG_LOGF("BlockchainInfo error: %s\n", e.what()); } + // The mempool's size: with the tip hash and the wallet's mutation epoch, what tells the caller whether a + // wallet scan could see anything new. getmempoolinfo takes no cs_main. Failing leaves it unknown, which + // counts as changed. + bool mempoolFetched = false, mempoolOk = false; + std::int64_t mempoolTxs = 0, mempoolBytes = 0; + if (plan.fetchMempool && blockOk) { + mempoolFetched = true; + try { + json pool = rpc.call("getmempoolinfo", json::array()); + if (pool.is_object()) { + mempoolOk = true; + if (pool.contains("size") && pool["size"].is_number_integer()) mempoolTxs = pool["size"].get(); + if (pool.contains("bytes") && pool["bytes"].is_number_integer()) mempoolBytes = pool["bytes"].get(); + } + } catch (const std::exception& e) { + DEBUG_LOGF("getmempoolinfo error: %s\n", e.what()); + } + } + // If getblockchaininfo shows we're behind (same 2-block tolerance as applyCoreRefreshResult), skip // the balance scan this cycle regardless of includeBalance: the balance is incomplete mid-sync // anyway, and skipping it lets the sync-state update — and hence kSyncProfile — take effect at the @@ -400,6 +419,10 @@ NetworkRefreshService::CoreRefreshResult NetworkRefreshService::collectCoreRefre result.balanceErrKind = balanceErrKind; result.balanceAttempted = balanceAttempted; result.spendableQueried = spendableQueried; + result.mempoolFetched = mempoolFetched; + result.mempoolOk = mempoolOk; + result.mempoolTxs = mempoolTxs; + result.mempoolBytes = mempoolBytes; return result; } @@ -773,6 +796,7 @@ NetworkRefreshService::AddressRefreshResult NetworkRefreshService::collectAddres } catch (const std::exception& e) { DEBUG_LOGF("z_listaddresses error: %s\n", e.what()); result.addressListOk = false; // enumeration failed → the shielded list may be falsely short + result.replyTimedOut |= rpc::rpcFailureKindOf(e) == rpc::RpcFailureKind::ReplyTimeout; } try { @@ -799,11 +823,12 @@ NetworkRefreshService::AddressRefreshResult NetworkRefreshService::collectAddres } } catch (const std::exception& e) { DEBUG_LOGF("z_listunspent unavailable (%s), falling back to z_getbalance\n", e.what()); + result.replyTimedOut |= rpc::rpcFailureKindOf(e) == rpc::RpcFailureKind::ReplyTimeout; for (auto& info : result.shieldedAddresses) { try { // display total (minconf=0, includes pending change) json total = rpc.call("z_getbalance", json::array({info.address, 0})); if (!total.is_null()) info.balance = total.get(); - } catch (...) {} + } catch (...) { result.balancesOk = false; } try { // spendable (minconf=1); degrade to the display value on old daemons json conf = rpc.call("z_getbalance", json::array({info.address, 1})); info.spendableBalance = (!conf.is_null()) ? conf.get() : info.balance; @@ -817,6 +842,7 @@ NetworkRefreshService::AddressRefreshResult NetworkRefreshService::collectAddres } catch (const std::exception& e) { DEBUG_LOGF("getaddressesbyaccount error: %s\n", e.what()); result.addressListOk = false; // enumeration failed → the transparent list may be falsely short + result.replyTimedOut |= rpc::rpcFailureKindOf(e) == rpc::RpcFailureKind::ReplyTimeout; } try { @@ -824,6 +850,8 @@ NetworkRefreshService::AddressRefreshResult NetworkRefreshService::collectAddres applyTransparentBalancesFromUnspent(result.transparentAddresses, unspent); } catch (const std::exception& e) { DEBUG_LOGF("listunspent error: %s\n", e.what()); + result.balancesOk = false; + result.replyTimedOut |= rpc::rpcFailureKindOf(e) == rpc::RpcFailureKind::ReplyTimeout; } result.scanMs = std::chrono::duration( @@ -1114,6 +1142,7 @@ NetworkRefreshService::TransactionRefreshResult NetworkRefreshService::collectTr result.transparentChainOk = true; } catch (const std::exception& e) { transactionRpcError = true; + result.replyTimedOut |= rpc::rpcFailureKindOf(e) == rpc::RpcFailureKind::ReplyTimeout; DEBUG_LOGF("listtransactions error: %s\n", e.what()); } } @@ -1155,6 +1184,7 @@ NetworkRefreshService::TransactionRefreshResult NetworkRefreshService::collectTr if (currentBlockHeight >= 0) result.shieldedScanHeights[address] = currentBlockHeight; } catch (const std::exception& e) { transactionRpcError = true; + result.replyTimedOut |= rpc::rpcFailureKindOf(e) == rpc::RpcFailureKind::ReplyTimeout; DEBUG_LOGF("z_listreceivedbyaddress error: %s\n", e.what()); } }; @@ -1243,6 +1273,7 @@ NetworkRefreshService::TransactionRefreshResult NetworkRefreshService::collectTr } catch (const std::exception& e) { transactionRpcError = true; result.sentSweepErrKind = rpc::rpcFailureKindOf(e); + result.replyTimedOut |= result.sentSweepErrKind == rpc::RpcFailureKind::ReplyTimeout; DEBUG_LOGF("z_listsentbyaddress error: %s\n", e.what()); } result.sentSweepMs = msSince(sweepStart); @@ -1447,6 +1478,7 @@ NetworkRefreshService::TransactionRefreshResult NetworkRefreshService::collectTr sortTransactionsNewestFirst(result.transactions); result.scanMs = msSince(scanStart); + result.rpcError = transactionRpcError; return result; } @@ -1499,6 +1531,8 @@ NetworkRefreshService::TransactionRefreshResult NetworkRefreshService::collectRe } } } catch (const std::exception& e) { + result.rpcError = true; + result.replyTimedOut |= rpc::rpcFailureKindOf(e) == rpc::RpcFailureKind::ReplyTimeout; DEBUG_LOGF("recent listtransactions error: %s\n", e.what()); } @@ -1540,6 +1574,8 @@ NetworkRefreshService::TransactionRefreshResult NetworkRefreshService::collectRe std::chrono::steady_clock::now() - callStart).count(); } catch (const std::exception& e) { ++result.shieldedAddressesScanned; + result.rpcError = true; + result.replyTimedOut |= rpc::rpcFailureKindOf(e) == rpc::RpcFailureKind::ReplyTimeout; DEBUG_LOGF("recent z_listreceivedbyaddress error: %s\n", e.what()); } } diff --git a/src/services/network_refresh_service.h b/src/services/network_refresh_service.h index 68445cf..20925f7 100644 --- a/src/services/network_refresh_service.h +++ b/src/services/network_refresh_service.h @@ -132,6 +132,11 @@ public: bool balanceAttempted = false; bool spendableQueried = false; // the minconf=1 call was made std::optional totalBalanceZat; // the minconf=0 "total" in zatoshis, exact + // getmempoolinfo (CoreRefreshPlan::fetchMempool): a change signal that takes no cs_main. + bool mempoolFetched = false; + bool mempoolOk = false; + std::int64_t mempoolTxs = 0; + std::int64_t mempoolBytes = 0; }; // What one Core refresh fetches. The minconf=1 (spendable) z_gettotalbalance is a second whole-wallet @@ -141,6 +146,7 @@ public: bool includeBalance = true; bool needSpendable = true; std::optional prevTotalZat; + bool fetchMempool = false; // add getmempoolinfo after a successful getblockchaininfo }; struct MiningRefreshResult { @@ -200,6 +206,8 @@ public: // The wallet's unspent notes from this same z_listunspent scan, so a consumer (e.g. the chat // note-budget) can be fed for free instead of running its own duplicate z_listunspent. std::vector unspentNotes; + bool balancesOk = true; // false: a balance call failed, the per-address balances may be stale + bool replyTimedOut = false; // a call hit the reply timeout (the daemon kept running it) }; struct AddressRefreshSnapshot { @@ -283,6 +291,8 @@ public: // Txids found but not enriched (over the per-cycle cap, or the view failed transiently): the caller // keeps them and feeds them back as enrichBacklogTxids. std::unordered_set unenrichedTxids; + bool rpcError = false; // a wallet-wide call failed + bool replyTimedOut = false; // ...with a reply timeout (the daemon kept running it) }; struct OperationStatusPollResult { @@ -465,6 +475,7 @@ public: void markDue(Timer timer) { scheduler_.markDue(timer); } void setTimer(Timer timer, float seconds) { scheduler_.setTimer(timer, seconds); } void deferFor(Timer timer, float seconds) { scheduler_.deferFor(timer, seconds); } + void retrySoon(Timer timer, float seconds) { scheduler_.retrySoon(timer, seconds); } float timer(Timer timer) const { return scheduler_.timer(timer); } float interval(Timer timer) const { return scheduler_.interval(timer); } void markImmediateRefresh() { scheduler_.markImmediateRefresh(); } diff --git a/src/services/refresh_scheduler.cpp b/src/services/refresh_scheduler.cpp index eca39a1..367c595 100644 --- a/src/services/refresh_scheduler.cpp +++ b/src/services/refresh_scheduler.cpp @@ -92,6 +92,12 @@ void RefreshScheduler::deferFor(Timer timer, float seconds) holdRef(timer) = std::max(0.0f, seconds); } +void RefreshScheduler::retrySoon(Timer timer, float seconds) +{ + holdRef(timer) = -1.0f; + timerRef(timer) = std::max(0.0f, interval(timer) - std::max(0.0f, seconds)); +} + float RefreshScheduler::timer(Timer timer) const { return timerRef(timer); diff --git a/src/services/refresh_scheduler.h b/src/services/refresh_scheduler.h index ddf5b45..c7b376e 100644 --- a/src/services/refresh_scheduler.h +++ b/src/services/refresh_scheduler.h @@ -62,6 +62,9 @@ public: // For retry backoff: the old "setTimer(interval - delay)" collapsed to an immediate retry wherever // the page interval was shorter than the delay (every frame on Overview/Send). void deferFor(Timer timer, float seconds); + // Due again `seconds` from now (a scan the governor held back asks again soon). Unlike deferFor, never + // due on a 0-interval timer (the sync profile's), so a held-back scan can't fire mid-sync. + void retrySoon(Timer timer, float seconds); float timer(Timer timer) const; float interval(Timer timer) const; diff --git a/src/services/wallet_scan_governor.cpp b/src/services/wallet_scan_governor.cpp new file mode 100644 index 0000000..b9320a1 --- /dev/null +++ b/src/services/wallet_scan_governor.cpp @@ -0,0 +1,312 @@ +// DragonX Wallet - ImGui Edition +// Copyright 2024-2026 The Hush Developers +// Released under the GPLv3 + +#include "wallet_scan_governor.h" + +#include +#include + +namespace dragonx { +namespace services { + +ScanVisibility scanVisibilityForPage(ui::NavPage page, bool windowBackground) +{ + ScanVisibility v; + if (windowBackground) return v; + using NP = ui::NavPage; + switch (page) { + case NP::Overview: + v.balance = v.addresses = v.history = true; // totals, per-address balances, recent activity + break; + case NP::Send: + case NP::Receive: + v.addresses = v.history = true; // per-address balances, recent sends / receives + break; + case NP::History: + v.history = true; + break; + case NP::Market: + v.balance = true; // portfolio value + break; + case NP::Chat: + v.chat = true; + break; + default: + break; + } + return v; +} + +double hiddenBalanceIntervalSec(double emaCostSec, bool autoShieldArmed) +{ + const double floorSec = autoShieldArmed ? 120.0 : 600.0; + return std::max(floorSec, 50.0 * std::max(0.0, emaCostSec)); +} + +BalancePollDecision decideBalancePoll(const BalancePollInputs& in) +{ + if (!in.allowBalance || in.syncing) return BalancePollDecision::Skip; + if (in.forced || in.neverFetched) return BalancePollDecision::Explicit; + if (!in.changed) return BalancePollDecision::Skip; + if (in.visible) { + // Just came into view with an old value: fetch it now rather than queue behind the other streams. + if (!in.waiting && in.shownAgeSec >= in.visibleFloorSec && in.sinceLastScanSec >= in.visibleFloorSec) + return BalancePollDecision::Explicit; + return BalancePollDecision::AskGovernor; + } + if (in.sinceLastScanSec < hiddenBalanceIntervalSec(in.emaCostSec, in.autoShieldArmed)) + return BalancePollDecision::Skip; + return BalancePollDecision::AskGovernor; +} + +double WalletScanGovernor::budgetForTipLag(double base, int blocksBehind) +{ + if (blocksBehind <= 0) return base; + if (blocksBehind == 1) return base * 2.0 / 3.0; + return base / 3.0; +} + +bool WalletScanGovernor::signalsMatch(ScanStream stream, const ScanSignals& current) const +{ + const Stream& st = at(stream); + if (!st.basisValid) return false; + const unsigned mask = cfg_.streams[static_cast(stream)].signalMask; + if (mask & kScanSignalTip) { + if (current.tipHash.empty() || st.basis.tipHash.empty() || current.tipHash != st.basis.tipHash) return false; + } + if (mask & kScanSignalMempool) { + if (!current.mempoolKnown || !st.basis.mempoolKnown) return false; + if (current.mempoolTxs != st.basis.mempoolTxs || current.mempoolBytes != st.basis.mempoolBytes) return false; + } + if (mask & kScanSignalEpoch) { + if (current.walletEpoch != st.basis.walletEpoch) return false; + } + return true; +} + +bool WalletScanGovernor::needed(ScanStream stream, const ScanSignals& current, double now) const +{ + const Stream& st = at(stream); + if (!st.basisValid) return true; + if (!signalsMatch(stream, current)) return true; + const double maxAge = cfg_.streams[static_cast(stream)].maxAgeSec; + return maxAge > 0.0 && now - st.basisAt >= maxAge; +} + +double WalletScanGovernor::floorSec(const Stream& stream) const +{ + return std::max(stream.askVisible ? cfg_.visibleFloorSec : cfg_.hiddenFloorSec, + cfg_.floorCostMultiple * stream.emaCostMs / 1000.0); +} + +ScanStream WalletScanGovernor::bestWaiter(double now) const +{ + const bool quiet = now < quietUntil_; + ScanStream best = ScanStream::Count; + double bestKey = std::numeric_limits::infinity(); + for (std::size_t i = 0; i < streams_.size(); ++i) { + const Stream& st = streams_[i]; + if (st.waitingSince < 0.0) continue; + if (now - st.lastAskAt > cfg_.waiterTtlSec) continue; // stopped asking + if (quiet && now - st.waitingSince < floorSec(st)) continue; // during quiet, only via its floor + const double key = st.waitingSince - (st.askVisible ? cfg_.visibleHeadStartSec : 0.0); + if (key < bestKey) { + bestKey = key; + best = static_cast(i); + } + } + return best; +} + +ScanTicket WalletScanGovernor::issue(ScanStream stream, ScanKind kind, const ScanSignals& current, double now, + bool servesRoutine) +{ + ScanTicket t; + t.stream = stream; + t.kind = kind; + t.id = ++nextId_; + t.generation = generation_; + t.dispatchedAt = now; + t.servesRoutine = servesRoutine; + t.basis = current; + live_.push_back(Live{t.id, stream, kind, now, servesRoutine}); + return t; +} + +ScanTicket WalletScanGovernor::ticketFor(const Live& live) const +{ + ScanTicket t; + t.stream = live.stream; + t.kind = live.kind; + t.id = live.id; + t.generation = generation_; + t.dispatchedAt = live.dispatchedAt; + t.servesRoutine = live.servesRoutine; + return t; +} + +ScanTicket WalletScanGovernor::tryAdmit(ScanStream stream, bool visible, const ScanSignals& current, double now) +{ + Stream& st = at(stream); + if (st.waitingSince < 0.0 || now - st.lastAskAt > cfg_.waitLapseSec) st.waitingSince = now; // a new wait + st.lastAskAt = now; + st.askVisible = visible; + pruneLeases(now); + if (!live_.empty()) return {}; // one wallet scan in the daemon at a time + if (bestWaiter(now) != stream) return {}; + ScanTicket t = issue(stream, ScanKind::Routine, current, now, true); + t.waitedSince = st.waitingSince; + st.waitingSince = -1.0; + return t; +} + +ScanTicket WalletScanGovernor::beginExplicit(ScanStream stream, const ScanSignals& current, double now, + bool servesRoutine) +{ + ScanTicket t = issue(stream, ScanKind::Explicit, current, now, servesRoutine); + if (servesRoutine) { + Stream& st = at(stream); + t.waitedSince = st.waitingSince; + st.waitingSince = -1.0; // served + } + return t; +} + +void WalletScanGovernor::eraseLive(std::uint64_t id) +{ + live_.erase(std::remove_if(live_.begin(), live_.end(), [id](const Live& l) { return l.id == id; }), live_.end()); +} + +namespace { +// A ticket that sent nothing gives back the wait it ended (keeping a wait begun since, if older). +void restoreWait(double& waitingSince, double waitedSince) +{ + if (waitedSince < 0.0) return; + waitingSince = waitingSince < 0.0 ? waitedSince : std::min(waitingSince, waitedSince); +} +} // namespace + +void WalletScanGovernor::cancel(const ScanTicket& ticket) +{ + if (!ticket || ticket.generation != generation_) return; + const bool wasLive = std::any_of(live_.begin(), live_.end(), [&](const Live& l) { return l.id == ticket.id; }); + if (!wasLive) return; + eraseLive(ticket.id); + restoreWait(at(ticket.stream).waitingSince, ticket.waitedSince); +} + +void WalletScanGovernor::complete(const ScanTicket& ticket, double costMs, ScanOutcome outcome, double now) +{ + if (!ticket || ticket.generation != generation_) return; // from before a reset() + const bool wasLive = std::any_of(live_.begin(), live_.end(), [&](const Live& l) { return l.id == ticket.id; }); + if (!wasLive) return; // already completed (or dropped as an orphan) + eraseLive(ticket.id); + Stream& st = at(ticket.stream); + if (outcome == ScanOutcome::Skipped) { // nothing was sent: no charge, and it keeps its place in line + restoreWait(st.waitingSince, ticket.waitedSince); + return; + } + st.lastScanAt = std::max(st.lastScanAt, ticket.dispatchedAt); + double ms = std::max(0.0, costMs); + if (outcome == ScanOutcome::TimedOut && ticket.servesRoutine) ms = std::max(ms, st.emaCostMs); + if (ms > 0.0) { + // Quiet long enough that this scan was `budget` of the time since it started. + const double debt = ms / 1000.0 * (1.0 / budgetForTipLag(cfg_.budget, tipLag_) - 1.0); + const double cap = now + (ticket.kind == ScanKind::Explicit ? cfg_.explicitDebtCapSec : cfg_.maxQuietSec); + quietUntil_ = std::max(quietUntil_, std::min(std::max(quietUntil_, now) + debt, cap)); + } + if (!ticket.servesRoutine) return; // charged; the routine scans' cost and basis are theirs + if (outcome == ScanOutcome::Ok || outcome == ScanOutcome::Partial || outcome == ScanOutcome::TimedOut) { + // Rises at once, decays slowly: a scan that got expensive is spaced out right away. + st.emaCostMs = (st.emaCostMs <= 0.0 || ms > st.emaCostMs) ? ms : 0.7 * st.emaCostMs + 0.3 * ms; + } + if (outcome == ScanOutcome::Failed || outcome == ScanOutcome::TimedOut) { + if (st.pendingValid) st.pendingPoisoned = true; + } else if (outcome == ScanOutcome::Partial) { + if (!st.pendingValid) { + st.pending = ticket.basis; + st.pendingAt = ticket.dispatchedAt; + st.pendingValid = true; + st.pendingPoisoned = false; + } + } else if (outcome == ScanOutcome::Ok) { + // A pass that started at the pending basis is current for that basis (the earliest), unless one of + // its dispatches failed. + if (!(st.pendingValid && st.pendingPoisoned)) { + st.basis = st.pendingValid ? st.pending : ticket.basis; + st.basisAt = st.pendingValid ? st.pendingAt : ticket.dispatchedAt; + st.basisValid = true; + } + st.pendingValid = false; + st.pendingPoisoned = false; + } +} + +void WalletScanGovernor::dropOrphans(ScanStream stream, double now) +{ + std::vector orphans; + for (const auto& l : live_) { + if (l.stream == stream) orphans.push_back(l); + } + for (const auto& l : orphans) complete(ticketFor(l), 0.0, ScanOutcome::TimedOut, now); +} + +void WalletScanGovernor::pruneLeases(double now) +{ + std::vector expired; + for (const auto& l : live_) { + const double lease = std::max(cfg_.minLeaseSec, 4.0 * at(l.stream).emaCostMs / 1000.0); + if (now - l.dispatchedAt > lease) expired.push_back(l); + } + for (const auto& l : expired) complete(ticketFor(l), 0.0, ScanOutcome::TimedOut, now); +} + +void WalletScanGovernor::invalidate(ScanStream stream) +{ + Stream& st = at(stream); + st.basisValid = false; + st.pendingValid = false; + st.pendingPoisoned = false; +} + +void WalletScanGovernor::reset() +{ + const std::uint64_t nextGeneration = generation_ + 1; + const std::uint64_t nextId = nextId_; + *this = WalletScanGovernor(cfg_); + generation_ = nextGeneration; + nextId_ = nextId; +} + +bool WalletScanGovernor::inFlight(ScanStream stream) const +{ + return std::any_of(live_.begin(), live_.end(), [stream](const Live& l) { return l.stream == stream; }); +} + +double WalletScanGovernor::secondsSinceScan(ScanStream stream, double now) const +{ + const double at_ = at(stream).lastScanAt; + return at_ < 0.0 ? 1e9 : now - at_; +} + +bool WalletScanGovernor::waiting(ScanStream stream, double now) const +{ + const Stream& st = at(stream); + return st.waitingSince >= 0.0 && now - st.lastAskAt <= cfg_.waitLapseSec; +} + +bool WalletScanGovernor::heldWithinFloor(ScanStream stream, double now) const +{ + if (!waiting(stream, now)) return false; + const Stream& st = at(stream); + return now - st.waitingSince < floorSec(st) + st.emaCostMs / 1000.0; +} + +double WalletScanGovernor::visibleFloorSecFor(ScanStream stream) const +{ + return std::max(cfg_.visibleFloorSec, cfg_.floorCostMultiple * at(stream).emaCostMs / 1000.0); +} + +} // namespace services +} // namespace dragonx diff --git a/src/services/wallet_scan_governor.h b/src/services/wallet_scan_governor.h new file mode 100644 index 0000000..e31354a --- /dev/null +++ b/src/services/wallet_scan_governor.h @@ -0,0 +1,214 @@ +// DragonX Wallet - ImGui Edition +// Copyright 2024-2026 The Hush Developers +// Released under the GPLv3 + +#pragma once + +// One budget for every wallet-wide scan the GUI sends on its own: the balance (z_gettotalbalance), the +// address scan (z_listunspent + listunspent), the history (z_listreceivedbyaddress, z_listsentbyaddress) and +// the between-blocks probe. Each holds dragonxd's cs_main for its whole run, seconds per call on a large +// wallet, which starves block connection. They used to be throttled one by one at 10% each, which adds up +// to 30-40% with all of them due. Here: one scan in flight at a time across both RPC lanes, and one shared +// quiet clock that keeps the total near `budget` of wall-clock. A stream that keeps asking is served FIFO +// (visible ones a little first), and a floor scaled to its own cost keeps any stream from starving. +// Explicit needs (the user's own send, a dirty set, a never-loaded list) are never refused, but are charged. +// +// Each stream also keeps the signals it last scanned at (tip hash, mempool size, a wallet-mutation epoch): +// a routine scan is only needed when they changed, or after the stream's safety sweep. +// +// An explicit ticket that doesn't do the stream's routine work (a history cycle that only lists rows) is +// charged to the quiet clock but leaves the stream's wait, cost estimate and basis alone (servesRoutine). +// +// Pure and main-thread only: every call takes `now` in seconds of a steady clock. + +#include "ui/sidebar.h" + +#include +#include +#include +#include +#include + +namespace dragonx { +namespace services { + +enum class ScanStream : std::uint8_t { Balance, Addresses, History, RecentTx, ChatNotes, ChatFast, Adhoc, Count }; +enum class ScanKind : std::uint8_t { Routine, Explicit }; +enum class ScanOutcome : std::uint8_t { + Ok, // applied: the stream's signals basis advances + Partial, // applied, more dispatches needed before the basis may advance (a multi-cycle pass) + Failed, // charged; the basis stays (and a pending pass is poisoned) + TimedOut, // a reply timeout: charged at least the stream's usual cost (the daemon still runs it) + Skipped, // admitted, but no scan was sent (node behind, busy, held back): not charged +}; +enum : unsigned { kScanSignalTip = 1, kScanSignalMempool = 2, kScanSignalEpoch = 4, kScanSignalAll = 7 }; + +// What a scan's result depends on, read without cs_main: the tip (bestblockhash, fetched each Core cycle), +// the mempool (getmempoolinfo takes only mempool.cs) and the App's wallet-mutation epoch. +struct ScanSignals { + std::string tipHash; + bool mempoolKnown = false; + std::int64_t mempoolTxs = 0; + std::int64_t mempoolBytes = 0; + std::uint64_t walletEpoch = 0; +}; + +struct ScanTicket { + ScanStream stream = ScanStream::Balance; + ScanKind kind = ScanKind::Routine; + std::uint64_t id = 0; + std::uint64_t generation = 0; + double dispatchedAt = 0.0; + double waitedSince = -1.0; // when its wait began, if it ended one (cancel() / Skipped restore it) + bool servesRoutine = true; // does the stream's routine work: ends its wait, feeds its cost and basis + ScanSignals basis; // the signals at dispatch: what an Ok result is current for + explicit operator bool() const { return id != 0; } +}; + +// Which streams the current page shows (visible streams get a head start and a shorter starvation floor). +struct ScanVisibility { + bool balance = false; + bool addresses = false; + bool history = false; + bool chat = false; +}; +ScanVisibility scanVisibilityForPage(ui::NavPage page, bool windowBackground); + +// The balance part of a Core poll. +enum class BalancePollDecision { Skip, AskGovernor, Explicit }; +struct BalancePollInputs { + bool allowBalance = true; // false for the locked wallet's chain-state poll + bool syncing = false; // the node is catching up (or settling after it) + bool forced = false; // the user's own send + bool neverFetched = false; // no balance yet this session + bool changed = true; // the governor's change gate + bool visible = true; // the page shows the aggregate balance + bool autoShieldArmed = false; // client-side auto-shield runs from balance results + double sinceLastScanSec = 1e9; // since the last balance scan that was sent + double emaCostSec = 0.0; + // Coming into view with an old balance (back to Overview, the window restored): one explicit poll + // instead of a fresh wait behind the other streams. Only when the stream isn't already waiting and + // both the value on screen and the last scan are older than the stream's visible floor. + bool waiting = false; // the Balance stream is waiting for admission + double shownAgeSec = 0.0; // age of the value on screen (last fetch, or last poll that confirmed it) + double visibleFloorSec = 30.0; // WalletScanGovernor::visibleFloorSecFor(Balance) +}; +// How often a page that doesn't show the balance still polls it: max(10 min, 50x its cost), or every 2 min +// while client-side auto-shield needs it. +double hiddenBalanceIntervalSec(double emaCostSec, bool autoShieldArmed); +BalancePollDecision decideBalancePoll(const BalancePollInputs& in); + +class WalletScanGovernor { +public: + struct StreamConfig { + unsigned signalMask = kScanSignalAll; // which signals the change gate compares + double maxAgeSec = 0.0; // safety sweep: needed this long after its basis (0 = never) + }; + struct Config { + double budget = 0.15; // share of wall-clock for all routine scans together + double explicitDebtCapSec = 120; // an explicit scan never pushes the quiet clock past now + this + double maxQuietSec = 900; // no scan pushes it past now + this (no runaway debt) + double visibleFloorSec = 30; // a waiting visible stream is admitted after this, quiet or not... + double hiddenFloorSec = 900; // ...a hidden one after this... + double floorCostMultiple = 15; // ...or after 15x its own cost if longer (<= ~6% each from floors) + double visibleHeadStartSec = 60; // visible waiters rank as if they had waited this much longer + double waiterTtlSec = 15; // a stream that hasn't asked for this long isn't waiting + double waitLapseSec = 60; // asking again after this long starts a new wait + double minLeaseSec = 600; // a ticket older than max(this, 4x its cost) is presumed lost + std::array(ScanStream::Count)> streams{{ + {kScanSignalAll, 300}, // Balance + {kScanSignalAll, 300}, // Addresses + {kScanSignalTip | kScanSignalEpoch, 0}, // History + {kScanSignalMempool | kScanSignalEpoch, 0}, // RecentTx (between blocks: the tip is History's) + {kScanSignalAll, 60}, // ChatNotes + {kScanSignalAll, 60}, // ChatFast + {0, 0}, // Adhoc + }}; + }; + + WalletScanGovernor() = default; + explicit WalletScanGovernor(const Config& config) : cfg_(config) {} + + // The budget while the node is behind: full at the tip, 2/3 one block behind, 1/3 two or more. + static double budgetForTipLag(double base, int blocksBehind); + + // A routine scan of this stream is needed: no basis yet, the signals changed, or the sweep is due. + bool needed(ScanStream stream, const ScanSignals& current, double now) const; + // The stream's basis matches these signals (masked). An unknown tip or mempool never matches. + bool signalsMatch(ScanStream stream, const ScanSignals& current) const; + + // Ask to run a routine scan now: a ticket when admitted, else an empty one (the stream keeps waiting + // while it keeps asking). Admitted only when no scan is in flight and this stream is the best waiter. + ScanTicket tryAdmit(ScanStream stream, bool visible, const ScanSignals& current, double now); + // An explicit need: never refused, charged (with capped debt). servesRoutine = false for one that + // doesn't do the stream's routine work: it is charged, but the stream keeps waiting for that work and + // its cost estimate and basis stay those of the routine scans. + ScanTicket beginExplicit(ScanStream stream, const ScanSignals& current, double now, bool servesRoutine = true); + // The scan wasn't sent after all (the job queue refused it): drop the ticket with no charge. + void cancel(const ScanTicket& ticket); + // The scan's result is in. `basis` may differ from the ticket's (e.g. what the worker saw). + void complete(const ScanTicket& ticket, double costMs, ScanOutcome outcome, double now); + // Tickets of this stream whose job is gone without a result: completed as timed out. + void dropOrphans(ScanStream stream, double now); + void setTipLag(int blocksBehind) { tipLag_ = blocksBehind > 0 ? blocksBehind : 0; } + void invalidate(ScanStream stream); // forget its basis: the next routine scan is needed + void reset(); // a new connection: forget everything; old tickets complete as no-ops + + bool inFlight(ScanStream stream) const; + bool anyInFlight() const { return !live_.empty(); } + double quietUntil() const { return quietUntil_; } + double emaCostMs(ScanStream stream) const { return at(stream).emaCostMs; } + // Since the dispatch of the last scan that was sent (cancelled and Skipped tickets don't count); 1e9 if none. + double secondsSinceScan(ScanStream stream, double now) const; + double currentBudget() const { return budgetForTipLag(cfg_.budget, tipLag_); } + // The stream asked within the lapse window and hasn't been admitted yet. + bool waiting(ScanStream stream, double now) const; + // Waiting, and still within what the budget promises it: its floor plus one scan of its own. Past that, + // or not asking, a stale value means something else is wrong. + bool heldWithinFloor(ScanStream stream, double now) const; + // The floor it would get as a visible waiter: max(visibleFloorSec, floorCostMultiple x its cost). + double visibleFloorSecFor(ScanStream stream) const; + +private: + struct Stream { + double waitingSince = -1.0; + double lastAskAt = -1e18; + bool askVisible = false; + double emaCostMs = 0.0; + double lastScanAt = -1.0; // dispatch time of the last scan that was sent + bool basisValid = false; + ScanSignals basis; + double basisAt = 0.0; + bool pendingValid = false; + bool pendingPoisoned = false; + ScanSignals pending; + double pendingAt = 0.0; + }; + struct Live { + std::uint64_t id = 0; + ScanStream stream = ScanStream::Balance; + ScanKind kind = ScanKind::Routine; + double dispatchedAt = 0.0; + bool servesRoutine = true; + }; + + Stream& at(ScanStream stream) { return streams_[static_cast(stream)]; } + const Stream& at(ScanStream stream) const { return streams_[static_cast(stream)]; } + double floorSec(const Stream& stream) const; + ScanStream bestWaiter(double now) const; + ScanTicket issue(ScanStream stream, ScanKind kind, const ScanSignals& current, double now, bool servesRoutine); + ScanTicket ticketFor(const Live& live) const; + void eraseLive(std::uint64_t id); + void pruneLeases(double now); + + Config cfg_; + std::array(ScanStream::Count)> streams_{}; + std::vector live_; + double quietUntil_ = 0.0; + int tipLag_ = 0; + std::uint64_t nextId_ = 0; + std::uint64_t generation_ = 1; +}; + +} // namespace services +} // namespace dragonx diff --git a/src/ui/windows/balance_tab.cpp b/src/ui/windows/balance_tab.cpp index 2058ea4..0014cca 100644 --- a/src/ui/windows/balance_tab.cpp +++ b/src/ui/windows/balance_tab.cpp @@ -423,15 +423,22 @@ static void RenderBalanceClassic(App* app) // hasn't refreshed in a while (a busy daemon can fail z_gettotalbalance without // dropping the whole connection). The node banner only covers full disconnects, so // this pill is the sole signal that the shown number may be out of date. - StalenessBadge badge = evaluateStalenessBadge( - state.last_balance_update, std::time(nullptr), state.connected); + // Current as of the later of: the last balance fetch, and the last Core poll that found + // nothing it depends on changed (the scan budget skips the scan then). Not while a balance + // scan is in flight (the new value is on its way), nor while the scan budget holds the next + // one within what it promises, unless the last attempt failed. + const std::int64_t balanceAsOf = std::max(state.last_balance_update, state.balance_verified_at); + const bool budgetHeld = !state.balance_last_attempt_failed && app->isBalanceHeldByScanBudget(); + StalenessBadge badge = (app->isBalanceRefreshInFlight() || budgetHeld) + ? StalenessBadge{} + : evaluateStalenessBadge(balanceAsOf, std::time(nullptr), state.connected); if (badge.show) { const bool err = (badge.severity == StalenessSeverity::Error); ImU32 fg = err ? Error() : Warning(); ImU32 bg = WithAlpha(fg, 38); ImU32 bd = WithAlpha(fg, 90); snprintf(buf, sizeof(buf), "%s %s", - TR("data_stale_prefix"), timeAgo(state.last_balance_update).c_str()); + TR("data_stale_prefix"), timeAgo(balanceAsOf).c_str()); ImVec2 pillSz = DrawPill(dl, ImVec2(cx, cy), buf, capFont, fg, bg, bd, ImVec2(4.0f * dp, 2.0f * dp)); // Hover → explain what stale means and how old the data actually is. if (material::IsRectHovered(ImVec2(cx, cy), ImVec2(cx + pillSz.x, cy + pillSz.y))) diff --git a/src/ui/windows/console_command_executor.cpp b/src/ui/windows/console_command_executor.cpp index 8f357ae..2264ccd 100644 --- a/src/ui/windows/console_command_executor.cpp +++ b/src/ui/windows/console_command_executor.cpp @@ -79,6 +79,9 @@ void FullNodeConsoleExecutor::submit(const std::string& cmd) } return [this, result_str, is_error]() { in_flight_.fetch_sub(1); + // A Console command may have changed the wallet (a send, an import, a new address): the + // scans gated on change signals must look again. (Main thread: this is a drained callback.) + if (app_) app_->noteWalletMutation(); std::lock_guard lk(results_mutex_); results_.push_back({result_str, is_error}); }; @@ -93,6 +96,7 @@ void FullNodeConsoleExecutor::submit(const std::string& cmd) result_str = e.what(); is_error = true; } + if (app_) app_->noteWalletMutation(); std::lock_guard lk(results_mutex_); results_.push_back({result_str, is_error}); } diff --git a/tests/test_phase4.cpp b/tests/test_phase4.cpp index d851958..9d41916 100644 --- a/tests/test_phase4.cpp +++ b/tests/test_phase4.cpp @@ -28,6 +28,7 @@ #include "rpc/rpc_busy_gate.h" #include "resources/embedded_resources.h" #include "services/history_scan_policy.h" +#include "services/wallet_scan_governor.h" #include "services/network_refresh_service.h" #include "services/refresh_scheduler.h" #include "services/note_consolidation.h" @@ -10476,6 +10477,526 @@ void testHistoryCollectorSweepAndEnrichment() } } +// C37: the scan governor's admission, floors, charging, change gate and lifecycle. +void testWalletScanGovernor() +{ + using namespace dragonx::services; + using G = WalletScanGovernor; + ScanSignals sig; + sig.tipHash = "tip-a"; + sig.mempoolKnown = true; + sig.mempoolTxs = 1; + sig.mempoolBytes = 100; + + { // One scan at a time; after it, quiet until it was `budget` (15%) of the time since it started. + G gov; + auto t = gov.tryAdmit(ScanStream::Balance, true, sig, 0.0); + EXPECT_TRUE(static_cast(t)); + EXPECT_FALSE(static_cast(gov.tryAdmit(ScanStream::Addresses, true, sig, 0.5))); // one at a time + gov.complete(t, 1000.0, ScanOutcome::Ok, 1.0); + EXPECT_NEAR(gov.quietUntil(), 1.0 + 1.0 * (1.0 / 0.15 - 1.0), 1e-6); // ~6.67 + EXPECT_FALSE(static_cast(gov.tryAdmit(ScanStream::Balance, true, sig, 6.5))); // quiet + EXPECT_FALSE(static_cast(gov.tryAdmit(ScanStream::Balance, true, sig, 6.7))); // Addresses waited longer + EXPECT_TRUE(static_cast(gov.tryAdmit(ScanStream::Addresses, true, sig, 6.8))); + } + { // FIFO among waiters; a visible one gets a 60 s head start; a stream that stopped asking isn't waiting. + G gov; + auto busy = gov.beginExplicit(ScanStream::Adhoc, sig, 0.0); // keeps the slot taken while they queue + gov.tryAdmit(ScanStream::Addresses, false, sig, 1.0); + gov.tryAdmit(ScanStream::History, false, sig, 2.0); + gov.complete(busy, 0.0, ScanOutcome::Skipped, 3.0); + gov.tryAdmit(ScanStream::History, false, sig, 3.0); // still waiting, but asked later + EXPECT_FALSE(static_cast(gov.tryAdmit(ScanStream::History, false, sig, 3.5))); + EXPECT_TRUE(static_cast(gov.tryAdmit(ScanStream::Addresses, false, sig, 3.5))); // first in + G head; + auto slot = head.beginExplicit(ScanStream::Adhoc, sig, 0.0); + head.tryAdmit(ScanStream::Addresses, false, sig, 1.0); // hidden, waiting since 1 + head.tryAdmit(ScanStream::Balance, true, sig, 30.0); // visible, waiting since 30 (ranks as -30) + head.complete(slot, 0.0, ScanOutcome::Skipped, 31.0); + head.tryAdmit(ScanStream::Addresses, false, sig, 31.0); + EXPECT_TRUE(static_cast(head.tryAdmit(ScanStream::Balance, true, sig, 31.5))); + G ttl; + auto held = ttl.beginExplicit(ScanStream::Adhoc, sig, 0.0); + ttl.tryAdmit(ScanStream::Addresses, false, sig, 1.0); // then stops asking + ttl.complete(held, 0.0, ScanOutcome::Skipped, 20.0); + EXPECT_TRUE(static_cast(ttl.tryAdmit(ScanStream::History, false, sig, 20.0))); // Addresses lapsed + } + { // During quiet, a waiter gets in only via its floor: max(30 s visible / 900 s hidden, 15x its cost). + G gov; + auto big = gov.tryAdmit(ScanStream::History, true, sig, 0.0); + gov.complete(big, 200000.0, ScanOutcome::Ok, 200.0); // quiet capped at now + 900 + EXPECT_NEAR(gov.quietUntil(), 1100.0, 1e-6); + double t = 200.0; + bool admitted = false; + for (; t < 1100.0 && !admitted; t += 2.0) admitted = static_cast(gov.tryAdmit(ScanStream::Balance, true, sig, t)); + EXPECT_TRUE(admitted); + EXPECT_NEAR(t - 2.0, 230.0, 2.1); // waited 30 s (EMA 0) + G costly; + auto first = costly.tryAdmit(ScanStream::Addresses, true, sig, 0.0); + costly.complete(first, 40000.0, ScanOutcome::Ok, 40.0); // EMA 40 s: floor 600 s + auto again = costly.tryAdmit(ScanStream::History, true, sig, 300.0); // after that quiet + EXPECT_TRUE(static_cast(again)); + costly.complete(again, 200000.0, ScanOutcome::Ok, 500.0); // quiet to 500 + 900 + admitted = false; + for (t = 500.0; t < 1400.0 && !admitted; t += 2.0) admitted = static_cast(costly.tryAdmit(ScanStream::Addresses, true, sig, t)); + EXPECT_TRUE(admitted); + EXPECT_NEAR(t - 2.0, 1100.0, 2.1); // 600 s + G hidden; + auto h = hidden.tryAdmit(ScanStream::History, true, sig, 0.0); + hidden.complete(h, 200000.0, ScanOutcome::Ok, 200.0); // quiet to 1100 + auto blocker = hidden.beginExplicit(ScanStream::Adhoc, sig, 1050.0); + hidden.complete(blocker, 200000.0, ScanOutcome::Ok, 1060.0); // explicit: quiet only to +120 = 1180 + admitted = false; + for (t = 200.0; t < 2000.0 && !admitted; t += 2.0) admitted = static_cast(hidden.tryAdmit(ScanStream::Balance, false, sig, t)); + EXPECT_TRUE(admitted); + EXPECT_TRUE(t - 2.0 >= 1100.0 - 0.1); // hidden: not before its 900 s floor + } + { // Charging: routine debt accumulates but quiet never runs past now + 900; explicit debt only to + // now + 120 and never shortens quiet; TimedOut charges at least the usual cost; Skipped nothing. + G gov; + double now = 0.0; + for (int i = 0; i < 5; ++i) { + auto t = gov.beginExplicit(ScanStream::History, sig, now); + gov.complete(t, 0.0, ScanOutcome::Skipped, now); + auto r = gov.tryAdmit(ScanStream::Balance, true, sig, gov.quietUntil() + 1.0); + now = gov.quietUntil() + 1.0; + if (r) gov.complete(r, 200000.0, ScanOutcome::Ok, now); + EXPECT_TRUE(gov.quietUntil() - now <= 900.0 + 1e-6); + } + G ex; + auto e = ex.beginExplicit(ScanStream::Balance, sig, 0.0); + ex.complete(e, 100000.0, ScanOutcome::Ok, 10.0); + EXPECT_NEAR(ex.quietUntil(), 130.0, 1e-6); + auto r = ex.beginExplicit(ScanStream::Addresses, sig, 10.0); + ex.complete(r, 1.0, ScanOutcome::Ok, 11.0); + EXPECT_TRUE(ex.quietUntil() >= 130.0 && ex.quietUntil() <= 131.0); // adds its own bit, never shortens + G timed; + auto a = timed.beginExplicit(ScanStream::Balance, sig, 0.0); + timed.complete(a, 10000.0, ScanOutcome::Ok, 10.0); + const double q1 = timed.quietUntil(); + auto b = timed.beginExplicit(ScanStream::Balance, sig, 200.0); + timed.complete(b, 1000.0, ScanOutcome::TimedOut, 201.0); // charged the 10 s EMA, not 1 s + EXPECT_TRUE(timed.quietUntil() >= 201.0 + 10.0 * (1.0 / 0.15 - 1.0) - 1e-6 || timed.quietUntil() >= 201.0 + 120.0 - 1e-6); + EXPECT_TRUE(q1 > 0.0); + auto c = timed.beginExplicit(ScanStream::Addresses, sig, 1000.0); + const double before = timed.quietUntil(); + timed.complete(c, 50000.0, ScanOutcome::Skipped, 1001.0); + EXPECT_NEAR(timed.quietUntil(), before, 1e-9); // Skipped: nothing sent, nothing charged + G ema; + auto e1 = ema.beginExplicit(ScanStream::Balance, sig, 0.0); + ema.complete(e1, 10000.0, ScanOutcome::Ok, 1.0); + auto e2 = ema.beginExplicit(ScanStream::Balance, sig, 2.0); + ema.complete(e2, 30000.0, ScanOutcome::Ok, 3.0); + EXPECT_NEAR(ema.emaCostMs(ScanStream::Balance), 30000.0, 1e-6); // rises at once + auto e3 = ema.beginExplicit(ScanStream::Balance, sig, 4.0); + ema.complete(e3, 10000.0, ScanOutcome::Ok, 5.0); + EXPECT_NEAR(ema.emaCostMs(ScanStream::Balance), 24000.0, 1e-6); // decays slowly (0.7 / 0.3) + G lag; + EXPECT_NEAR(lag.currentBudget(), 0.15, 1e-12); + lag.setTipLag(1); + EXPECT_NEAR(lag.currentBudget(), 0.10, 1e-12); + lag.setTipLag(5); + EXPECT_NEAR(lag.currentBudget(), 0.05, 1e-12); + } + { // The change gate. + G gov; + EXPECT_TRUE(gov.needed(ScanStream::Balance, sig, 0.0)); // no basis yet + auto t = gov.beginExplicit(ScanStream::Balance, sig, 0.0); + gov.complete(t, 100.0, ScanOutcome::Ok, 1.0); + EXPECT_FALSE(gov.needed(ScanStream::Balance, sig, 10.0)); + ScanSignals pool = sig; + pool.mempoolTxs = 2; + EXPECT_TRUE(gov.needed(ScanStream::Balance, pool, 10.0)); + ScanSignals epoch = sig; + epoch.walletEpoch = 1; + EXPECT_TRUE(gov.needed(ScanStream::Balance, epoch, 10.0)); + EXPECT_TRUE(gov.needed(ScanStream::Balance, sig, 300.0)); // the safety sweep + ScanSignals unknown = sig; + unknown.mempoolKnown = false; + EXPECT_TRUE(gov.needed(ScanStream::Balance, unknown, 10.0)); // unknown never matches + ScanSignals noTip = sig; + noTip.tipHash.clear(); + EXPECT_TRUE(gov.needed(ScanStream::Balance, noTip, 10.0)); + // History compares the tip and the epoch only. + auto h = gov.beginExplicit(ScanStream::History, sig, 20.0); + gov.complete(h, 100.0, ScanOutcome::Ok, 21.0); + EXPECT_FALSE(gov.needed(ScanStream::History, pool, 30.0)); + ScanSignals tip = sig; + tip.tipHash = "tip-b"; + EXPECT_TRUE(gov.needed(ScanStream::History, tip, 30.0)); + // A failure keeps the basis. + auto f = gov.beginExplicit(ScanStream::Balance, tip, 40.0); + gov.complete(f, 100.0, ScanOutcome::Failed, 41.0); + EXPECT_TRUE(gov.needed(ScanStream::Balance, tip, 50.0)); + EXPECT_FALSE(gov.needed(ScanStream::Balance, sig, 50.0)); + // A change that lands while the scan runs is still needed afterwards (the basis is the dispatch's). + auto d = gov.beginExplicit(ScanStream::Balance, pool, 60.0); + ScanSignals later = pool; + later.mempoolTxs = 3; + gov.complete(d, 100.0, ScanOutcome::Ok, 61.0); + EXPECT_FALSE(gov.needed(ScanStream::Balance, pool, 62.0)); + EXPECT_TRUE(gov.needed(ScanStream::Balance, later, 62.0)); + // A pass: Partial then Ok adopts the first dispatch's basis; a failure inside poisons it. + G pass; + ScanSignals p1 = sig, p2 = sig; + p1.tipHash = "p1"; + p2.tipHash = "p2"; + auto pa = pass.beginExplicit(ScanStream::Addresses, p1, 0.0); + pass.complete(pa, 10.0, ScanOutcome::Partial, 1.0); + auto pb = pass.beginExplicit(ScanStream::Addresses, p2, 2.0); + pass.complete(pb, 10.0, ScanOutcome::Ok, 3.0); + EXPECT_FALSE(pass.needed(ScanStream::Addresses, p1, 4.0)); + EXPECT_TRUE(pass.needed(ScanStream::Addresses, p2, 4.0)); + G poisoned; + auto qa = poisoned.beginExplicit(ScanStream::Addresses, p1, 0.0); + poisoned.complete(qa, 10.0, ScanOutcome::Partial, 1.0); + auto qb = poisoned.beginExplicit(ScanStream::Addresses, p1, 2.0); + poisoned.complete(qb, 10.0, ScanOutcome::Failed, 3.0); + auto qc = poisoned.beginExplicit(ScanStream::Addresses, p1, 4.0); + poisoned.complete(qc, 10.0, ScanOutcome::Ok, 5.0); + EXPECT_TRUE(poisoned.needed(ScanStream::Addresses, p1, 6.0)); + gov.invalidate(ScanStream::Balance); + EXPECT_TRUE(gov.needed(ScanStream::Balance, sig, 70.0)); + } + { // Lifecycle. + G gov; + auto t = gov.tryAdmit(ScanStream::Balance, true, sig, 0.0); + gov.cancel(t); // not sent: no charge, wait restored + EXPECT_FALSE(gov.anyInFlight()); + EXPECT_NEAR(gov.quietUntil(), 0.0, 1e-12); + EXPECT_FALSE(static_cast(gov.tryAdmit(ScanStream::Addresses, true, sig, 1.0))); // Balance is ahead + EXPECT_TRUE(static_cast(gov.tryAdmit(ScanStream::Balance, true, sig, 2.0))); + G reset; + auto old = reset.beginExplicit(ScanStream::Balance, sig, 0.0); + reset.reset(); + reset.complete(old, 1e6, ScanOutcome::Ok, 1.0); // from the old connection: no-op + EXPECT_NEAR(reset.quietUntil(), 0.0, 1e-12); + EXPECT_TRUE(reset.needed(ScanStream::Balance, sig, 2.0)); + G orphan; + auto o1 = orphan.beginExplicit(ScanStream::Addresses, sig, 0.0); + orphan.complete(o1, 10000.0, ScanOutcome::Ok, 10.0); + const double q = orphan.quietUntil(); + orphan.beginExplicit(ScanStream::Addresses, sig, 200.0); + EXPECT_TRUE(orphan.inFlight(ScanStream::Addresses)); + orphan.dropOrphans(ScanStream::Addresses, 300.0); // its job is gone: freed, charged + EXPECT_FALSE(orphan.inFlight(ScanStream::Addresses)); + EXPECT_TRUE(orphan.quietUntil() > q); + G lease; + auto stuck = lease.tryAdmit(ScanStream::History, true, sig, 0.0); + EXPECT_TRUE(static_cast(stuck)); + EXPECT_FALSE(static_cast(lease.tryAdmit(ScanStream::Balance, true, sig, 599.0))); + EXPECT_TRUE(static_cast(lease.tryAdmit(ScanStream::Balance, true, sig, 601.0))); // presumed lost + lease.complete(stuck, 1e6, ScanOutcome::Ok, 602.0); // a late result: ignored + EXPECT_NEAR(lease.quietUntil(), 0.0, 1e-12); + G adhoc; + auto a1 = adhoc.beginExplicit(ScanStream::Adhoc, sig, 0.0); + auto a2 = adhoc.beginExplicit(ScanStream::Adhoc, sig, 0.0); + adhoc.complete(a1, 0.0, ScanOutcome::Skipped, 1.0); + EXPECT_TRUE(adhoc.inFlight(ScanStream::Adhoc)); + adhoc.complete(a2, 0.0, ScanOutcome::Skipped, 1.0); + EXPECT_FALSE(adhoc.inFlight(ScanStream::Adhoc)); + } + { // A stream that stopped asking for over 60 s starts a new wait: no floor bypass from the old one. + G lapse; + auto big = lapse.tryAdmit(ScanStream::History, true, sig, 0.0); + lapse.complete(big, 200000.0, ScanOutcome::Ok, 200.0); // quiet to 1100 + EXPECT_FALSE(static_cast(lapse.tryAdmit(ScanStream::Balance, true, sig, 200.0))); + double t = 261.0; // silent for 61 s + bool admitted = false; + for (; t < 1100.0 && !admitted; t += 2.0) admitted = static_cast(lapse.tryAdmit(ScanStream::Balance, true, sig, t)); + EXPECT_TRUE(admitted); + EXPECT_NEAR(t - 2.0, 291.0, 2.1); // 30 s from 261 (without the rule: 261) + } + { // An explicit charge never shortens a routine quiet. + G keep; + auto r = keep.tryAdmit(ScanStream::History, true, sig, 0.0); + keep.complete(r, 200000.0, ScanOutcome::Ok, 200.0); // quiet to 1100 + auto e = keep.beginExplicit(ScanStream::Balance, sig, 300.0); + keep.complete(e, 50000.0, ScanOutcome::Ok, 310.0); // its own cap is 430 + EXPECT_NEAR(keep.quietUntil(), 1100.0, 1e-6); + } + { // An explicit ticket that doesn't do the routine work (a Light history cycle): charged, but the stream + // keeps waiting for that work, and its cost estimate stays the routine scans'. + G gov; + auto h = gov.tryAdmit(ScanStream::History, true, sig, 0.0); + gov.complete(h, 120000.0, ScanOutcome::Ok, 0.0); // EMA 120 s, quiet to 680 + auto slot = gov.beginExplicit(ScanStream::Adhoc, sig, 700.0); + gov.tryAdmit(ScanStream::History, true, sig, 701.0); + gov.tryAdmit(ScanStream::Addresses, true, sig, 702.0); + auto light = gov.beginExplicit(ScanStream::History, sig, 703.0, false); + gov.complete(light, 1000.0, ScanOutcome::Ok, 704.0); // quiet to ~709.7 + EXPECT_NEAR(gov.emaCostMs(ScanStream::History), 120000.0, 1e-6); + EXPECT_TRUE(gov.quietUntil() > 709.0); + gov.complete(slot, 0.0, ScanOutcome::Skipped, 705.0); + EXPECT_FALSE(static_cast(gov.tryAdmit(ScanStream::Addresses, true, sig, 712.0))); // History is older + EXPECT_TRUE(static_cast(gov.tryAdmit(ScanStream::History, true, sig, 712.5))); + } + { // Only scans that were sent count: a cancelled or Skipped ticket doesn't restart the hidden interval, + // and a Skipped routine ticket keeps its place in line. + G gov; + auto c = gov.tryAdmit(ScanStream::Balance, false, sig, 0.0); + gov.cancel(c); + EXPECT_TRUE(gov.secondsSinceScan(ScanStream::Balance, 10.0) > 1e8); + auto s1 = gov.tryAdmit(ScanStream::Balance, false, sig, 10.0); + EXPECT_TRUE(static_cast(s1)); + gov.complete(s1, 0.0, ScanOutcome::Skipped, 11.0); + EXPECT_TRUE(gov.secondsSinceScan(ScanStream::Balance, 20.0) > 1e8); + auto o = gov.tryAdmit(ScanStream::Balance, false, sig, 20.0); + EXPECT_TRUE(static_cast(o)); + gov.complete(o, 1000.0, ScanOutcome::Ok, 21.0); + EXPECT_NEAR(gov.secondsSinceScan(ScanStream::Balance, 30.0), 10.0, 1e-9); + auto f = gov.beginExplicit(ScanStream::Balance, sig, 40.0); + gov.complete(f, 1000.0, ScanOutcome::Failed, 41.0); // sent, so it counts + EXPECT_NEAR(gov.secondsSinceScan(ScanStream::Balance, 50.0), 10.0, 1e-9); + G fifo; + auto busy = fifo.beginExplicit(ScanStream::Adhoc, sig, 0.0); + fifo.tryAdmit(ScanStream::Balance, true, sig, 1.0); + fifo.tryAdmit(ScanStream::Addresses, true, sig, 5.0); + fifo.complete(busy, 0.0, ScanOutcome::Skipped, 9.0); + auto b = fifo.tryAdmit(ScanStream::Balance, true, sig, 10.0); + EXPECT_TRUE(static_cast(b)); + fifo.complete(b, 0.0, ScanOutcome::Skipped, 11.0); // the node was behind: nothing sent + EXPECT_FALSE(static_cast(fifo.tryAdmit(ScanStream::Addresses, true, sig, 12.0))); // Balance kept its place + EXPECT_TRUE(static_cast(fifo.tryAdmit(ScanStream::Balance, true, sig, 12.5))); + } + { // waiting / heldWithinFloor / visibleFloorSecFor. + G gov; + EXPECT_NEAR(gov.visibleFloorSecFor(ScanStream::Balance), 30.0, 1e-9); + auto first = gov.beginExplicit(ScanStream::Balance, sig, 0.0); + gov.complete(first, 21000.0, ScanOutcome::Ok, 21.0); // EMA 21 s: floor 315 s + EXPECT_NEAR(gov.visibleFloorSecFor(ScanStream::Balance), 315.0, 1e-9); + auto slot = gov.beginExplicit(ScanStream::Adhoc, sig, 30.0); + EXPECT_FALSE(gov.waiting(ScanStream::Balance, 30.0)); + EXPECT_FALSE(gov.heldWithinFloor(ScanStream::Balance, 30.0)); + EXPECT_FALSE(static_cast(gov.tryAdmit(ScanStream::Balance, true, sig, 31.0))); // denied: waits + EXPECT_TRUE(gov.waiting(ScanStream::Balance, 32.0)); + EXPECT_TRUE(gov.heldWithinFloor(ScanStream::Balance, 32.0)); + EXPECT_FALSE(gov.heldWithinFloor(ScanStream::Balance, 92.0)); // stopped asking (> 60 s) + double t = 31.0; + for (; t < 360.0; t += 2.0) gov.tryAdmit(ScanStream::Balance, true, sig, t); // the slot stays taken + EXPECT_TRUE(gov.heldWithinFloor(ScanStream::Balance, 360.0)); // 329 s < 315 + 21 + for (; t < 370.0; t += 2.0) gov.tryAdmit(ScanStream::Balance, true, sig, t); + EXPECT_FALSE(gov.heldWithinFloor(ScanStream::Balance, 370.0)); // past floor + one scan: something's wrong + gov.complete(slot, 0.0, ScanOutcome::Skipped, 371.0); + EXPECT_TRUE(static_cast(gov.tryAdmit(ScanStream::Balance, true, sig, 372.0))); + EXPECT_FALSE(gov.waiting(ScanStream::Balance, 373.0)); // admitted: no longer waiting + } +} + +// The between-blocks probe's rotation: every z-address once (oldest-scanned first) before it is done. +void testProbeRotation() +{ + using dragonx::services::ProbeRotation; + const std::vector addresses{"za", "zb", "zc"}; + std::unordered_map heights{{"za", 100}, {"zb", 90}, {"zc", 100}}; + ProbeRotation rotation; + EXPECT_TRUE(rotation.complete()); + EXPECT_EQ(rotation.next(addresses, heights), std::string("zb")); // oldest first + rotation.probed("zb"); + heights["zb"] = 101; + EXPECT_FALSE(rotation.complete()); + EXPECT_EQ(rotation.next(addresses, heights), std::string("za")); // tie: list order + rotation.probed("za"); + heights["za"] = 101; + EXPECT_EQ(rotation.next(addresses, heights), std::string("zc")); + rotation.probed("zc"); + EXPECT_TRUE(rotation.complete()); + // A new rotation after a change: all heights equal now, still every address once. + heights["zc"] = 101; + EXPECT_EQ(rotation.next(addresses, heights), std::string("za")); + rotation.probed("za"); + EXPECT_EQ(rotation.next(addresses, heights), std::string("zb")); + // An address that left the list is dropped; one never scanned goes first. + const std::vector changed{"zb", "zd"}; + EXPECT_EQ(rotation.next(changed, heights), std::string("zb")); // za/zc gone, zd new but not in this rotation + rotation.probed("zb"); + EXPECT_TRUE(rotation.complete()); + EXPECT_EQ(rotation.next(changed, heights), std::string("zd")); // new rotation: never scanned first + rotation.reset(); + EXPECT_TRUE(rotation.complete()); + EXPECT_EQ(rotation.next({}, heights), std::string()); +} + +// C37 under overload: three visible streams that always want to scan (21 s, 25 s and 120 s per scan, a +// large wallet) for 2 h. Together they stay near the budget (the old per-stream 10% rules added up to +// 30-40%), none starves, and quiet never runs away; hidden, each still gets in within its floor. +void testWalletScanGovernorOverload() +{ + using namespace dragonx::services; + WalletScanGovernor gov; + const ScanStream streams[3] = {ScanStream::Balance, ScanStream::Addresses, ScanStream::History}; + const double cost[3] = {21.0, 25.0, 120.0}; + double lastAdmit[3] = {0.0, 0.0, 0.0}; + double maxGapVisible[3] = {0.0, 0.0, 0.0}; + double maxGapHidden[3] = {0.0, 0.0, 0.0}; + ScanTicket live; + int liveIndex = -1; + double liveEnd = 0.0, scanSec = 0.0, maxQuietAhead = 0.0; + const ScanSignals sig; + const double visibleEnd = 7200.0, end = 10800.0; + for (double now = 0.0; now < end; now += 2.0) { + const bool visible = now < visibleEnd; + if (live && now >= liveEnd) { + gov.complete(live, cost[liveIndex] * 1000.0, ScanOutcome::Ok, now); + live = ScanTicket{}; + } + for (int i = 0; i < 3; ++i) { + auto t = gov.tryAdmit(streams[i], visible, sig, now); + if (!t) continue; + live = t; + liveIndex = i; + liveEnd = now + cost[i]; + if (now < visibleEnd) scanSec += cost[i]; + const double gap = now - lastAdmit[i]; + if (now < visibleEnd) maxGapVisible[i] = std::max(maxGapVisible[i], gap); + else if (lastAdmit[i] >= visibleEnd) maxGapHidden[i] = std::max(maxGapHidden[i], gap); + lastAdmit[i] = now; + } + maxQuietAhead = std::max(maxQuietAhead, gov.quietUntil() - now); + } + EXPECT_TRUE(scanSec / visibleEnd <= 0.22); + EXPECT_TRUE(scanSec / visibleEnd >= 0.10); + EXPECT_TRUE(maxQuietAhead <= 900.0 + 1e-6); + // A stream's longest wait: its floor, plus the other two scans if both are queued ahead of it. + const double others[3] = {25.0 + 120.0, 21.0 + 120.0, 21.0 + 25.0}; + const double visibleFloor[3] = {315.0, 375.0, 1800.0}; // max(300 s, 15x cost) + for (int i = 0; i < 3; ++i) EXPECT_TRUE(maxGapVisible[i] <= visibleFloor[i] + others[i] + 4.0); + const double hiddenFloor[3] = {900.0, 900.0, 1800.0}; // max(900 s, 15x cost) + for (int i = 0; i < 3; ++i) EXPECT_TRUE(maxGapHidden[i] <= hiddenFloor[i] + others[i] + 4.0); +} + +void testBalancePollDecisionAndVisibility() +{ + using namespace dragonx::services; + using D = BalancePollDecision; + EXPECT_NEAR(hiddenBalanceIntervalSec(0.05, false), 600.0, 1e-9); + EXPECT_NEAR(hiddenBalanceIntervalSec(20.0, false), 1000.0, 1e-9); + EXPECT_NEAR(hiddenBalanceIntervalSec(0.05, true), 120.0, 1e-9); + BalancePollInputs in; + EXPECT_TRUE(decideBalancePoll(in) == D::AskGovernor); + { auto x = in; x.allowBalance = false; x.forced = true; EXPECT_TRUE(decideBalancePoll(x) == D::Skip); } + { auto x = in; x.syncing = true; x.forced = true; EXPECT_TRUE(decideBalancePoll(x) == D::Skip); } + { auto x = in; x.forced = true; x.changed = false; EXPECT_TRUE(decideBalancePoll(x) == D::Explicit); } + { auto x = in; x.neverFetched = true; x.changed = false; EXPECT_TRUE(decideBalancePoll(x) == D::Explicit); } + { auto x = in; x.changed = false; EXPECT_TRUE(decideBalancePoll(x) == D::Skip); } + { auto x = in; x.visible = false; x.sinceLastScanSec = 100.0; EXPECT_TRUE(decideBalancePoll(x) == D::Skip); } + { auto x = in; x.visible = false; x.sinceLastScanSec = 700.0; EXPECT_TRUE(decideBalancePoll(x) == D::AskGovernor); } + { auto x = in; x.visible = false; x.autoShieldArmed = true; x.sinceLastScanSec = 130.0; + EXPECT_TRUE(decideBalancePoll(x) == D::AskGovernor); } + // Coming into view with an old value: one explicit poll. Each guard alone flips it back. + BalancePollInputs back = in; + back.shownAgeSec = 400.0; + back.visibleFloorSec = 315.0; + back.sinceLastScanSec = 400.0; + EXPECT_TRUE(decideBalancePoll(back) == D::Explicit); + { auto x = back; x.waiting = true; EXPECT_TRUE(decideBalancePoll(x) == D::AskGovernor); } // already queued + { auto x = back; x.shownAgeSec = 100.0; EXPECT_TRUE(decideBalancePoll(x) == D::AskGovernor); } + { auto x = back; x.sinceLastScanSec = 100.0; EXPECT_TRUE(decideBalancePoll(x) == D::AskGovernor); } // just tried + { auto x = back; x.changed = false; EXPECT_TRUE(decideBalancePoll(x) == D::Skip); } + { auto x = back; x.visible = false; EXPECT_TRUE(decideBalancePoll(x) == D::Skip); } // hidden: its own interval + + using NP = dragonx::ui::NavPage; + auto overview = scanVisibilityForPage(NP::Overview, false); + EXPECT_TRUE(overview.balance && overview.addresses && overview.history && !overview.chat); + for (auto page : {NP::Send, NP::Receive}) { + auto v = scanVisibilityForPage(page, false); + EXPECT_TRUE(!v.balance && v.addresses && v.history); + } + auto history = scanVisibilityForPage(NP::History, false); + EXPECT_TRUE(!history.balance && !history.addresses && history.history); + auto market = scanVisibilityForPage(NP::Market, false); + EXPECT_TRUE(market.balance && !market.addresses && !market.history); + auto chat = scanVisibilityForPage(NP::Chat, false); + EXPECT_TRUE(chat.chat && !chat.balance); + auto mining = scanVisibilityForPage(NP::Mining, false); + EXPECT_TRUE(!mining.balance && !mining.addresses && !mining.history && !mining.chat); + auto background = scanVisibilityForPage(NP::Overview, true); + EXPECT_TRUE(!background.balance && !background.addresses && !background.history && !background.chat); +} + +// C36: the Core poll reads the mempool's size (no cs_main) right after the chain, for the change gate. +void testCoreRefreshMempool() +{ + using Refresh = dragonx::services::NetworkRefreshService; + using json = nlohmann::json; + using dragonx::rpc::RpcFailureKind; + using dragonx::rpc::RpcTransportError; + const json chain = json{{"blocks", 150}, {"headers", 150}, {"longestchain", 150}, {"bestblockhash", "tip-150"}}; + const json total = json{{"private", "3.00000000"}, {"transparent", "1.25000000"}, {"total", "4.25000000"}}; + Refresh::CoreRefreshPlan plan; + plan.fetchMempool = true; + plan.needSpendable = false; + plan.prevTotalZat = 425000000; + { + MockRefreshRpc rpc; + rpc.addResponse("getblockchaininfo", chain); + rpc.addResponse("getmempoolinfo", json{{"size", 3}, {"bytes", 4500}, {"usage", 9000}}); + rpc.addResponse("z_gettotalbalance", total); + auto result = Refresh::collectCoreRefreshResult(rpc, plan); + EXPECT_TRUE(rpc.methodNames() == std::vector({"getblockchaininfo", "getmempoolinfo", "z_gettotalbalance"})); + EXPECT_EQ(rpc.calls[1].params, json::array()); + EXPECT_TRUE(result.mempoolFetched && result.mempoolOk); + EXPECT_EQ(result.mempoolTxs, static_cast(3)); + EXPECT_EQ(result.mempoolBytes, static_cast(4500)); + } + { // A failed read leaves the mempool unknown (counts as changed); the balance still runs. + MockRefreshRpc rpc; + rpc.addResponse("getblockchaininfo", chain); + rpc.addFailure("getmempoolinfo", "boom"); + rpc.addResponse("z_gettotalbalance", total); + auto result = Refresh::collectCoreRefreshResult(rpc, plan); + EXPECT_TRUE(result.mempoolFetched && !result.mempoolOk); + EXPECT_TRUE(result.balanceOk); + } + { // Behind: chain + mempool only. + MockRefreshRpc rpc; + rpc.addResponse("getblockchaininfo", json{{"blocks", 100}, {"headers", 150}, {"longestchain", 150}}); + rpc.addResponse("getmempoolinfo", json{{"size", 0}, {"bytes", 0}}); + Refresh::collectCoreRefreshResult(rpc, plan); + EXPECT_TRUE(rpc.methodNames() == std::vector({"getblockchaininfo", "getmempoolinfo"})); + } + { // The chain read found the node busy: nothing else. + MockRefreshRpc rpc; + rpc.addException("getblockchaininfo", std::make_exception_ptr(RpcTransportError( + RpcFailureKind::QueueFull, "RPC error: HTTP 500: Work queue depth exceeded", CURLE_OK, 500, true))); + auto result = Refresh::collectCoreRefreshResult(rpc, plan); + EXPECT_TRUE(rpc.methodNames() == std::vector({"getblockchaininfo"})); + EXPECT_FALSE(result.mempoolFetched); + } +} + +void testRefreshSchedulerRetrySoon() +{ + using dragonx::services::RefreshScheduler; + using Timer = RefreshScheduler::Timer; + RefreshScheduler early; + early.setIntervals({5.0f, 10.0f, 15.0f, 20.0f}); + early.reset(Timer::Transactions); + early.retrySoon(Timer::Transactions, 2.0f); + EXPECT_FALSE(early.isDue(Timer::Transactions)); + early.tick(1.9f); + EXPECT_FALSE(early.isDue(Timer::Transactions)); + RefreshScheduler due; + due.setIntervals({5.0f, 10.0f, 15.0f, 20.0f}); + due.reset(Timer::Transactions); + due.retrySoon(Timer::Transactions, 2.0f); + due.tick(2.0f); + EXPECT_TRUE(due.isDue(Timer::Transactions)); + RefreshScheduler syncing; // the sync profile disables the timer: a retry never fires mid-sync + syncing.setIntervals(RefreshScheduler::kSyncProfile); + syncing.retrySoon(Timer::Transactions, 2.0f); + syncing.tick(60.0f); + EXPECT_FALSE(syncing.isDue(Timer::Transactions)); + + // getmempoolinfo takes no cs_main: always sent, and its replies say nothing about busyness. + dragonx::rpc::RpcBusyGate gate; + gate.record("z_gettotalbalance", dragonx::rpc::RpcFailureKind::ReplyTimeout); + EXPECT_TRUE(gate.busy()); + EXPECT_TRUE(gate.admit("getmempoolinfo")); + gate.record("getmempoolinfo", dragonx::rpc::RpcFailureKind::None); + EXPECT_TRUE(gate.busy()); +} + int main() { testConnectionConfig(); @@ -10504,6 +11025,12 @@ int main() testHistoryPassWithCollector(); testHistoryCollectorSweepAndEnrichment(); testHistoryDutyModel(); + testWalletScanGovernor(); + testWalletScanGovernorOverload(); + testProbeRotation(); + testBalancePollDecisionAndVisibility(); + testCoreRefreshMempool(); + testRefreshSchedulerRetrySoon(); testNetworkRefreshResultModels(); testTrackedSendRetirement(); testBalanceDisplayLerp(); diff --git a/tools/fake_dragonxd/README.md b/tools/fake_dragonxd/README.md index fa96a38..90593b2 100644 --- a/tools/fake_dragonxd/README.md +++ b/tools/fake_dragonxd/README.md @@ -122,6 +122,7 @@ inherits the GUI's environment. | `FAKE_STUCK_SHUTDOWN` | 0 | the Sep 29 orphan: hold the datadir lock, no listener, ignore SIGTERM, never exit | | `FAKE_WALLET_NOTES` | (off) | a note wallet: `addr:count:amount,...` shielded notes; enables `z_mergetoaddress` + its operations (dragonxd rules: confirmed unlocked notes, a batch locked on return, merged note next block) | | `FAKE_MERGE_SECS` | 6 | how long a `z_mergetoaddress` operation runs before it succeeds | +| `FAKE_METHODS_ALL` | 0 | `1` = one `call m= p=` event per request (the GUI's RPC cadence; the `methods` lines keep only the top 10 counts) | | `FAKE_HEIGHT`, `FAKE_REBUILD_TOTAL`, `FAKE_NOTES`, `FAKE_VERSION` | 3327145, 62583, 13632, 1030050 | cosmetic numbers | ## Test cases diff --git a/tools/fake_dragonxd/fake_dragonxd.cpp b/tools/fake_dragonxd/fake_dragonxd.cpp index 5f2c056..812da70 100644 --- a/tools/fake_dragonxd/fake_dragonxd.cpp +++ b/tools/fake_dragonxd/fake_dragonxd.cpp @@ -749,6 +749,13 @@ json execOne(const json& req, int& status) json params = req.contains("params") ? req["params"] : json::array(); if (!params.is_array()) params = json::array(); status = 200; + // FAKE_METHODS_ALL=1: one "call" event per request (method + params), to see the GUI's RPC cadence. + static const bool logAll = [] { const char* v = std::getenv("FAKE_METHODS_ALL"); return v && std::string(v) == "1"; }(); + if (logAll) { + std::string p = params.dump(); + if (p.size() > 160) p = p.substr(0, 160) + "..."; + event("call", "m=" + m + " p=" + p); + } // Like dragonxd: stop and help take no lock, nor does getrescaninfo (it only reads wallet fields). const bool lockFree = (m == "stop" || m == "help" || m == "getrescaninfo");