ObsidianDragon/src/services/wallet_scan_governor.h

// 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 <array>
#include <cstddef>
#include <cstdint>
#include <string>
#include <vector>

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): only a first read runs
    bool background = false;       // the window is in the background: no first read while syncing
    bool firstReadTried = false;   // the first read while syncing was already sent this connection
    bool gateOpen = true;          // the connect gate (C44): closed right after a connect, at most ~20 s
    bool forced = false;           // the user's own send (waits while the window is in the background)
    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<StreamConfig, static_cast<std::size_t>(ScanStream::Count)> streams{{
            {kScanSignalAll, 300},   // Balance
            {kScanSignalAll, 300},   // Addresses
            {kScanSignalTip | kScanSignalEpoch, 0},   // History
            {kScanSignalMempool | kScanSignalEpoch, 0},   // RecentTx (between blocks: the tip is History's)
            {0, 0},   // ChatNotes: admission only; the need is services::shouldScanChatNotes (tip settle, chat ops)
            {0, 0},   // ChatFast: admission only; the need is getwalletinfo's txcount (chatInboxScanNeeded). Not
                      // the tip or mempool: they move before the wallet notifier stores the tx, so a scan fired
                      // on them can read the old wallet and consume the change.
            {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<std::size_t>(stream)]; }
    const Stream& at(ScanStream stream) const { return streams_[static_cast<std::size_t>(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<Stream, static_cast<std::size_t>(ScanStream::Count)> streams_{};
    std::vector<Live> live_;
    double quietUntil_ = 0.0;
    int tipLag_ = 0;
    std::uint64_t nextId_ = 0;
    std::uint64_t generation_ = 1;
};

}  // namespace services
}  // namespace dragonx