diff --git a/src/app.cpp b/src/app.cpp index f9e9516..0dc9ce9 100644 --- a/src/app.cpp +++ b/src/app.cpp @@ -499,6 +499,7 @@ bool App::init() // Initialize RPC client rpc_ = std::make_unique(); + rpc_->setBusyGate(rpc_busy_gate_); // Initialize background RPC worker thread worker_ = std::make_unique(); @@ -940,6 +941,10 @@ void App::update() // enable the wallet UI), but there is no RPC daemon — so RPC polls (mining/balance/peers/txs) // must key off rpc_ being genuinely connected or they'd fire and fail ("Not connected"). const bool rpcConnected = rpc_ && rpc_->isConnected(); + // The node is too busy to answer: send it nothing that needs cs_main (it would only queue behind the + // rest and hold one of its few RPC slots) until the canary gets through. pumpRpcBusyCanary() runs it. + pumpRpcBusyCanary(); + const bool rpcBusy = rpc_busy_gate_->busy(); // Fast refresh (mining stats + daemon memory) every second // Skip when wallet is locked — no need to poll, and queued tasks @@ -958,7 +963,7 @@ void App::update() // Skip the mining poll while the daemon is in warmup (e.g. during -rescan). Otherwise // getmininginfo is rejected with -28 ("Rescanning...") every second and floods the log. // The rescan progress poll below still runs — that's how we track the warmup/rescan. - if (rpcConnected && !state_.warming_up && !runtime_rescan_active_) { + if (rpcConnected && !state_.warming_up && !runtime_rescan_active_ && !rpcBusy) { refreshMiningInfo(); } @@ -1210,7 +1215,7 @@ void App::update() rpc::RPCWorker* opidWorker = (fast_worker_ && fast_worker_->isRunning()) ? fast_worker_.get() : worker_.get(); if (network_refresh_.isDue(RefreshTimer::Opid) && !pending_opids_.empty() - && rpcConnected && opidWorker && !opid_poll_in_progress_) { + && rpcConnected && !rpcBusy && opidWorker && !opid_poll_in_progress_) { network_refresh_.reset(RefreshTimer::Opid); auto opids = pending_opids_; // copy for worker thread opid_poll_in_progress_ = true; @@ -1319,8 +1324,10 @@ void App::update() } // Per-category refresh with tab-aware intervals - // Skip when wallet is locked — same reason as above. - if (rpcConnected && !state_.isLocked()) { + // Skip when wallet is locked — same reason as above. Hold everything while the node is busy. + if (rpcConnected && !state_.isLocked() && rpcBusy) { + // nothing: the canary brings polling back (markImmediateRefresh) once the node answers + } else if (rpcConnected && !state_.isLocked()) { const bool walletDataPage = currentPageNeedsWalletDataRefresh(); if (!runtime_rescan_active_ && network_refresh_.consumeDue(RefreshTimer::Core)) { refreshCoreData(); @@ -1369,7 +1376,7 @@ void App::update() if (network_refresh_.consumeDue(RefreshTimer::Peers)) { refreshPeerInfo(); } - } else if (network_refresh_.consumeDue(RefreshTimer::Core)) { + } else if (!rpcBusy && network_refresh_.consumeDue(RefreshTimer::Core)) { if (!connection_in_progress_ && wizard_phase_ == WizardPhase::None && !bootstrap_downloading_) { @@ -5968,6 +5975,7 @@ void App::beginShutdown() // the render loop exits, so the UI stays responsive. if (rpc_) rpc_->requestAbort(); if (fast_rpc_) fast_rpc_->requestAbort(); + abortRpcBusyCanary(); if (worker_) { worker_->requestStop(); } @@ -6874,6 +6882,8 @@ void App::shutdown() // Ensure the mining-control thread is stopped + joined (idempotent; beginShutdown already did it on the // normal quit path, but shutdown() can also run without it). Must precede xmrig_manager_ teardown. stopMiningControlThread(); + abortRpcBusyCanary(); + if (canary_thread_.joinable()) canary_thread_.join(); // Wipe any copied secret from the OS clipboard before we exit — the 45s auto-clear timer // never fires if the user quits sooner, which would otherwise leave a key/seed resident. diff --git a/src/app.h b/src/app.h index f6c7198..a25b834 100644 --- a/src/app.h +++ b/src/app.h @@ -23,6 +23,7 @@ #include "rpc/connection.h" #include "daemon/stop_delivery.h" #include "daemon/daemon_busy_detect.h" +#include "rpc/rpc_busy_gate.h" #include "services/network_refresh_service.h" #include "services/wallet_security_controller.h" #include "services/wallet_security_workflow.h" @@ -1170,6 +1171,19 @@ private: std::atomic daemon_busy_reset_pending_{false}; // set from any thread; applied by feedDaemonBusy() std::chrono::steady_clock::time_point rpc_busy_since_{}; // epoch = not busy std::function raise_window_; + // S6: once the node is too busy to answer (a reply timeout or a full queue), the polling clients + // (rpc_, fast_rpc_) hold back everything that needs cs_main, and one canary request on its own + // client waits for the node to answer again. rpc_busy_gate_ is shared with those clients. + std::shared_ptr rpc_busy_gate_ = std::make_shared(); + rpc::ConnectRetryBackoff connect_backoff_; // S8: per-failure-kind delay before the next connect attempt + std::thread canary_thread_; + std::atomic canary_state_{0}; // CanaryState + std::mutex canary_mutex_; + rpc::RPCClient* canary_client_ = nullptr; // aborted on shutdown + rpc::ConnectRetryBackoff canary_backoff_; // a canary refused as queue-full waits before the next + std::chrono::steady_clock::time_point canary_retry_at_{}; + void pumpRpcBusyCanary(); + void abortRpcBusyCanary(); void feedDaemonBusy(const std::string& chunk); void resetDaemonBusy(); diff --git a/src/app_network.cpp b/src/app_network.cpp index 250b49f..ad87fbd 100644 --- a/src/app_network.cpp +++ b/src/app_network.cpp @@ -264,6 +264,79 @@ void App::detectWalletDegraded() VERBOSE_LOGF("[recovery] Daemon opened wallet in DEGRADED mode — new-key derivation disabled\n"); } +// One canary per busy period: a getinfo on its own client with NO client timeout, so it is the one request +// we let wait in the node's queue instead of re-sending (each re-send would take another slot). Its answer +// means the node serves cs_main requests again. +enum CanaryState { kCanaryIdle = 0, kCanaryRunning, kCanaryAnswered, kCanaryRefused, kCanaryGone }; + +void App::abortRpcBusyCanary() +{ + std::lock_guard lk(canary_mutex_); + if (canary_client_) canary_client_->requestAbort(); +} + +void App::pumpRpcBusyCanary() +{ + const int st = canary_state_.load(); + if (st >= kCanaryAnswered) { + if (canary_thread_.joinable()) canary_thread_.join(); + canary_state_ = kCanaryIdle; + if (st == kCanaryAnswered) { + DEBUG_LOGF("[busy] the node answered again after %.0fs; resuming normal polling\n", + rpc_busy_gate_->busySeconds()); + rpc_busy_gate_->clear(); + canary_backoff_.reset(); + connect_backoff_.reset(); + network_refresh_.markImmediateRefresh(); // also makes a pending connect attempt due at once + } else if (st == kCanaryRefused) { + const float wait = canary_backoff_.next(rpc::RpcFailureKind::QueueFull); + canary_retry_at_ = std::chrono::steady_clock::now() + + std::chrono::milliseconds(static_cast(wait * 1000.0f)); + DEBUG_LOGF("[busy] canary refused (queue full); next one in %.0fs\n", wait); + } else { + // Nothing listening any more (or our own abort): there's no busy node to wait for. Normal + // connect / lost-connection handling takes over. + DEBUG_LOGF("[busy] canary got no answer; leaving busy mode\n"); + rpc_busy_gate_->clear(); + canary_backoff_.reset(); + } + } + if (!rpc_busy_gate_->busy() || canary_state_.load() != kCanaryIdle || shutting_down_) return; + if (std::chrono::steady_clock::now() < canary_retry_at_) return; + + const rpc::ConnectionConfig cfg = (!saved_config_.rpcuser.empty() || !saved_config_.rpcpassword.empty()) + ? saved_config_ : rpc::Connection::autoDetectConfig(); + DEBUG_LOGF("[busy] the node is too busy to answer (%d request(s) given up on): holding back polls; " + "one canary getinfo waits for it\n", rpc_busy_gate_->abandoned()); + canary_state_ = kCanaryRunning; + canary_thread_ = std::thread([this, cfg]() { + rpc::RPCClient client; // deliberately ungated: this is the one request we allow + { + std::lock_guard lk(canary_mutex_); + canary_client_ = &client; + } + int result = kCanaryGone; + if (!shutting_down_ && client.open(cfg.host, cfg.port, cfg.rpcuser, cfg.rpcpassword, cfg.use_tls)) { + try { + rpc::RPCClient::TraceScope trace("Busy node / Canary"); + client.call("getinfo", json::array(), 0L); // 0 = no client timeout + result = kCanaryAnswered; + } catch (const std::exception& e) { + const rpc::RpcFailureKind kind = rpc::rpcFailureKindOf(e); + if (kind == rpc::RpcFailureKind::AppError || kind == rpc::RpcFailureKind::Warmup) + result = kCanaryAnswered; // a worker ran it: cs_main is free again + else if (kind == rpc::RpcFailureKind::QueueFull) + result = kCanaryRefused; + } + } + { + std::lock_guard lk(canary_mutex_); + canary_client_ = nullptr; + } + canary_state_ = result; + }); +} + void App::tryConnect() { // Lite builds have no full node / RPC daemon, so never run the RPC connection state machine @@ -324,8 +397,7 @@ void App::tryConnect() if (++daemon_wait_attempts_ == kDaemonWaitWarnAttempts) { ui::Notifications::instance().warning(TR("daemon_port_busy_warn"), 20.0f); } - network_refresh_.setTimer(services::NetworkRefreshService::Timer::Core, - services::RefreshScheduler::kCoreDefault - 1.0f); + network_refresh_.deferFor(services::NetworkRefreshService::Timer::Core, 4.0f); return; } daemon_wait_attempts_ = 0; // port is free — clear the bounded-wait counter @@ -338,8 +410,7 @@ void App::tryConnect() if (startEmbeddedDaemon()) { // Will retry connection after daemon starts VERBOSE_LOGF("[connect #%d] Embedded daemon starting, will retry connection...\n", connect_attempt); - network_refresh_.setTimer(services::NetworkRefreshService::Timer::Core, - services::RefreshScheduler::kCoreDefault - 1.0f); + network_refresh_.deferFor(services::NetworkRefreshService::Timer::Core, 1.0f); } else { // The daemon couldn't be started (binary not found, Sapling params missing, spawn // failure, …). Surface the actual reason instead of leaving the status stuck on @@ -356,8 +427,7 @@ void App::tryConnect() detail.empty() ? std::string(TR("sb_daemon_start_failed")) : detail, 30.0f); } // Keep retrying: a missing binary/params can be fixed without a restart. - network_refresh_.setTimer(services::NetworkRefreshService::Timer::Core, - services::RefreshScheduler::kCoreDefault - 1.0f); + network_refresh_.deferFor(services::NetworkRefreshService::Timer::Core, 5.0f); } } else if (!isUsingEmbeddedDaemon()) { VERBOSE_LOGF("[connect #%d] Embedded daemon disabled (using external). No config found at %s\n", @@ -451,6 +521,7 @@ void App::tryConnect() return [this, config, connected, warmingUp, warmupStatus, daemonStarting, externalDetected, attempt, connectErr, connectKind]() { noteDaemonRpcResult(connected, connectKind); // a busy node (timeouts / queue full) feeds the quit guard + if (connected) connect_backoff_.reset(); if (connected) { VERBOSE_LOGF("[connect #%d] Connected successfully%s\n", attempt, warmingUp ? " (daemon warming up)" : ""); @@ -525,10 +596,10 @@ void App::tryConnect() // daemon's own console output) instead of a bare technical error. const bool reachableButBusy = rpc::isRpcTimeout(connectKind); applyDaemonInitStatus(reachableButBusy); - VERBOSE_LOGF("[connect #%d] RPC connection failed (%s) — daemon still starting, will retry...\n", - attempt, connectErr.c_str()); - network_refresh_.setTimer(services::NetworkRefreshService::Timer::Core, - services::RefreshScheduler::kCoreDefault - 1.0f); + const float retryIn = connect_backoff_.next(connectKind); + VERBOSE_LOGF("[connect #%d] RPC connection failed (%s) — daemon still starting, retry in %.0fs\n", + attempt, connectErr.c_str(), retryIn); + network_refresh_.deferFor(services::NetworkRefreshService::Timer::Core, retryIn); } else if (externalDetected) { state_.connected = false; // An external daemon is on the RPC port but not answering. A timeout means it's @@ -542,10 +613,10 @@ void App::tryConnect() } else { connection_status_ = TR("sb_connecting_external"); } - VERBOSE_LOGF("[connect #%d] External daemon detected but RPC failed (%s), will retry...\n", - attempt, connectErr.c_str()); - network_refresh_.setTimer(services::NetworkRefreshService::Timer::Core, - services::RefreshScheduler::kCoreDefault - 1.0f); + const float retryIn = connect_backoff_.next(connectKind); + VERBOSE_LOGF("[connect #%d] External daemon detected but RPC failed (%s), retry in %.0fs\n", + attempt, connectErr.c_str(), retryIn); + network_refresh_.deferFor(services::NetworkRefreshService::Timer::Core, retryIn); } else { onDisconnected("Connection failed"); VERBOSE_LOGF("[connect #%d] RPC connection failed — no daemon starting, no external detected\n", attempt); @@ -692,6 +763,7 @@ void App::onConnected() // behind the main refresh batch. if (!fast_rpc_) { fast_rpc_ = std::make_unique(); + fast_rpc_->setBusyGate(rpc_busy_gate_); } if (!fast_worker_) { fast_worker_ = std::make_unique(); @@ -1800,7 +1872,11 @@ void App::refreshCoreData() constexpr int kCoreFailuresBeforeDisconnect = 3; if (!state_.warming_up) { if (!result.balanceOk && !result.blockchainOk) { - if (++consecutive_core_failures_ >= kCoreFailuresBeforeDisconnect && + // Only failures that put the connection itself in doubt count. A node that is merely + // busy (reply timeouts, a full queue, calls held back) is still there: tearing down + // then wiped the wallet state and restarted the connect probes that fill its queue. + if (rpc::rpcKindMeansLost(result.blockchainErrKind) && + ++consecutive_core_failures_ >= kCoreFailuresBeforeDisconnect && state_.connected) { consecutive_core_failures_ = 0; handleLostConnection("Lost connection to daemon"); diff --git a/src/rpc/rpc_busy_gate.h b/src/rpc/rpc_busy_gate.h new file mode 100644 index 0000000..c935096 --- /dev/null +++ b/src/rpc/rpc_busy_gate.h @@ -0,0 +1,168 @@ +#pragma once + +// Keeping the GUI from flooding a busy node (S6). dragonxd's HTTP server has 8 workers and a 16-deep +// queue. While a long operation holds cs_main (a witness rebuild can take an hour), every request that +// needs cs_main waits, and a client that gives up does NOT free its slot. The GUI's polls used to time +// out and be re-sent until all 24 slots were its own abandoned requests, after which even "stop" was +// refused. The gate stops that: after the first sign the node is busy (a reply timeout, or a queue-full +// refusal), the GUI's clients send nothing that needs cs_main until a single "canary" request (which +// waits as long as it takes) gets through. Pure and header-only; shared by the GUI's RPC clients. + +#include "rpc_failure.h" + +#include +#include +#include + +namespace dragonx { +namespace rpc { + +// Kinds that mean "reached a node that is too busy to answer", as opposed to "no node there". +inline bool rpcKindMeansBusy(RpcFailureKind kind) +{ + return kind == RpcFailureKind::ReplyTimeout || kind == RpcFailureKind::QueueFull || + kind == RpcFailureKind::Deferred; +} + +// Kinds that mean the connection itself is in doubt (counted toward a reconnect). Busy kinds, a +// warmup reply, an application error and our own abort are not: the node is there. +inline bool rpcKindMeansLost(RpcFailureKind kind) +{ + switch (kind) { + case RpcFailureKind::None: + case RpcFailureKind::ReplyTimeout: + case RpcFailureKind::QueueFull: + case RpcFailureKind::Deferred: + case RpcFailureKind::Warmup: + case RpcFailureKind::AppError: + case RpcFailureKind::Aborted: + return false; + default: + return true; + } +} + +class RpcBusyGate { +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"; + } + + // May this request be sent now? Lock-free methods always; the rest only while the node isn't busy. + bool admit(const std::string& method) const + { + if (isLockFree(method)) return true; + std::lock_guard lk(mx_); + return !busy_; + } + + // How a sent request went. Any real answer (even an error) means the node is serving cs_main + // requests again; a reply timeout or a queue-full refusal means it isn't. Other failures (nothing + // listening, auth, our own abort) say nothing about busyness and leave the state alone. + void record(const std::string& method, RpcFailureKind kind) + { + if (isLockFree(method)) return; + std::lock_guard lk(mx_); + if (kind == RpcFailureKind::None || kind == RpcFailureKind::AppError || kind == RpcFailureKind::Warmup) { + clearLocked(); + } else if (kind == RpcFailureKind::ReplyTimeout || kind == RpcFailureKind::QueueFull) { + if (!busy_) { + busy_ = true; + since_ = std::chrono::steady_clock::now(); + ++busyPeriods_; + } + if (kind == RpcFailureKind::ReplyTimeout) ++abandoned_; // it still holds a daemon slot + } + } + + // The canary got through (or the node is gone): poll normally again. + void clear() + { + std::lock_guard lk(mx_); + clearLocked(); + } + + bool busy() const + { + std::lock_guard lk(mx_); + return busy_; + } + // Requests we gave up on that the node may still be holding, since the gate last cleared. + int abandoned() const + { + std::lock_guard lk(mx_); + return abandoned_; + } + // Increments each time the gate turns busy: lets the owner arm exactly one canary per busy period. + int busyPeriods() const + { + std::lock_guard lk(mx_); + return busyPeriods_; + } + double busySeconds() const + { + std::lock_guard lk(mx_); + if (!busy_) return 0.0; + return std::chrono::duration(std::chrono::steady_clock::now() - since_).count(); + } + +private: + void clearLocked() + { + busy_ = false; + abandoned_ = 0; + } + + mutable std::mutex mx_; + bool busy_ = false; + int abandoned_ = 0; + int busyPeriods_ = 0; + std::chrono::steady_clock::time_point since_{}; +}; + +/** + * How long to wait before the next connection attempt after a failure of this kind. Each repeat of the + * same kind waits longer, up to its cap; a success or a different kind starts over. + * nothing listening / connect timeout: 1, 2, 5s (the node is starting, or gone) + * queue full: 2, 4, 8, 16, 30s (the queue is someone's; hammering it only adds to it) + * busy (reply timeout / deferred): 30s; the canary brings the connection back as soon as it can + * anything else: 4s + */ +class ConnectRetryBackoff { +public: + float next(RpcFailureKind kind) + { + streak_ = (kind == last_) ? streak_ + 1 : 0; + last_ = kind; + switch (kind) { + case RpcFailureKind::NotListening: + case RpcFailureKind::ConnectTimeout: { + static const float steps[] = {1.0f, 2.0f, 5.0f}; + return steps[streak_ < 2 ? streak_ : 2]; + } + case RpcFailureKind::QueueFull: { + static const float steps[] = {2.0f, 4.0f, 8.0f, 16.0f, 30.0f}; + return steps[streak_ < 4 ? streak_ : 4]; + } + case RpcFailureKind::ReplyTimeout: + case RpcFailureKind::Deferred: + return 30.0f; + default: + return 4.0f; + } + } + void reset() + { + streak_ = 0; + last_ = RpcFailureKind::None; + } + +private: + RpcFailureKind last_ = RpcFailureKind::None; + int streak_ = 0; +}; + +} // namespace rpc +} // namespace dragonx diff --git a/src/rpc/rpc_client.cpp b/src/rpc/rpc_client.cpp index bbbcf40..27bd92f 100644 --- a/src/rpc/rpc_client.cpp +++ b/src/rpc/rpc_client.cpp @@ -360,6 +360,8 @@ json RPCClient::makePayload(const std::string& method, const json& params) std::string RPCClient::performCall(const std::string& method, const json& params, long& httpCodeOut) { + if (busy_gate_ && !busy_gate_->admit(method)) + throw RpcTransportError(RpcFailureKind::Deferred, "RPC deferred: the node is busy", CURLE_OK, 0, false); emitRpcTrace(method); json payload = makePayload(method, params); @@ -380,13 +382,20 @@ std::string RPCClient::performCall(const std::string& method, const json& params long requestSize = 0; curl_easy_getinfo(impl_->curl, CURLINFO_REQUEST_SIZE, &requestSize); const bool sent = requestSize > 0; - throw RpcTransportError(classifyRpcFailure(res, sent, 0, 0, std::string()), - "RPC request failed: " + std::string(curl_easy_strerror(res)), res, 0, sent); + const RpcFailureKind kind = classifyRpcFailure(res, sent, 0, 0, std::string()); + if (busy_gate_) busy_gate_->record(method, kind); + throw RpcTransportError(kind, "RPC request failed: " + std::string(curl_easy_strerror(res)), res, 0, sent); } // Check HTTP response code httpCodeOut = 0; curl_easy_getinfo(impl_->curl, CURLINFO_RESPONSE_CODE, &httpCodeOut); + if (busy_gate_) { + // Any HTTP answer means a daemon worker ran our request, except the queue-full refusal. + const bool queueFull = httpCodeOut == 500 && + response_data.find("Work queue depth exceeded") != std::string::npos; + busy_gate_->record(method, queueFull ? RpcFailureKind::QueueFull : RpcFailureKind::None); + } return response_data; } diff --git a/src/rpc/rpc_client.h b/src/rpc/rpc_client.h index f7552d3..e22fc38 100644 --- a/src/rpc/rpc_client.h +++ b/src/rpc/rpc_client.h @@ -6,6 +6,7 @@ #include "types.h" #include "rpc_failure.h" // RpcError, RpcTransportError, RpcFailureKind +#include "rpc_busy_gate.h" #include #include #include @@ -81,6 +82,14 @@ public: */ RpcFailureKind trySendStop(long timeoutSec = 2); + /** + * @brief Share a busy gate with the app's other polling clients (nullptr: none) + * + * While the gate is busy, calls that need the daemon's cs_main fail at once with a Deferred + * RpcTransportError instead of being sent; every sent call reports how it went to the gate. + */ + void setBusyGate(std::shared_ptr gate) { busy_gate_ = std::move(gate); } + /** * @brief Disconnect from dragonxd */ @@ -297,6 +306,7 @@ private: std::string warmup_status_; std::string last_connect_error_; RpcFailureKind last_connect_kind_ = RpcFailureKind::None; + std::shared_ptr busy_gate_; std::string logged_connect_error_; // last "Connection failed" text logged (rate limit) unsigned connect_error_repeats_ = 0; // identical failures since it was logged json last_connect_info_; diff --git a/src/rpc/rpc_failure.h b/src/rpc/rpc_failure.h index 42b550c..b46733c 100644 --- a/src/rpc/rpc_failure.h +++ b/src/rpc/rpc_failure.h @@ -28,6 +28,7 @@ enum class RpcFailureKind { Warmup, // JSON-RPC -28 (RPC_IN_WARMUP): the node is still loading AppError, // the daemon ran the method and answered with a JSON-RPC error Aborted, // our own requestAbort() cut the transfer + Deferred, // not sent: the node is busy (RpcBusyGate), so we held back rather than queue more Other, }; @@ -45,6 +46,7 @@ inline const char* rpcFailureKindName(RpcFailureKind kind) case RpcFailureKind::Warmup: return "Warmup"; case RpcFailureKind::AppError: return "AppError"; case RpcFailureKind::Aborted: return "Aborted"; + case RpcFailureKind::Deferred: return "Deferred"; case RpcFailureKind::Other: return "Other"; } return "Other"; diff --git a/src/services/network_refresh_service.cpp b/src/services/network_refresh_service.cpp index 0b17fe5..69f572b 100644 --- a/src/services/network_refresh_service.cpp +++ b/src/services/network_refresh_service.cpp @@ -1,4 +1,5 @@ #include "network_refresh_service.h" +#include "rpc/rpc_busy_gate.h" #include "../util/logger.h" #include @@ -324,7 +325,9 @@ NetworkRefreshService::CoreRefreshResult NetworkRefreshService::collectCoreRefre if (lc > 0 && b < lc - 2) behind = true; } - if (includeBalance && !behind) { + // Don't follow a chain-info call that found the node busy (a reply timeout or a full queue) with two + // balance scans: they need the same cs_main, and each would only take another of its few RPC slots. + if (includeBalance && !behind && !rpc::rpcKindMeansBusy(blockErrKind)) { // z_gettotalbalance is O(mapWallet) and holds the daemon's cs_main for its whole duration — // seconds on a large shielded wallet. Time it so the caller can throttle how often it polls // (balanceRefreshDue()), keeping balance scans from starving block connection. diff --git a/src/services/network_refresh_service.h b/src/services/network_refresh_service.h index 22c4b3e..e22b340 100644 --- a/src/services/network_refresh_service.h +++ b/src/services/network_refresh_service.h @@ -386,6 +386,7 @@ public: void reset(Timer timer) { scheduler_.reset(timer); } 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); } 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 4b0d685..eca39a1 100644 --- a/src/services/refresh_scheduler.cpp +++ b/src/services/refresh_scheduler.cpp @@ -47,10 +47,16 @@ void RefreshScheduler::tick(float deltaSeconds) timers_.fast += delta; timers_.txAge += delta; timers_.opid += delta; + for (float& h : holds_) { + if (h > 0.0f) h = std::max(0.0f, h - delta); + } } bool RefreshScheduler::isDue(Timer timer) const { + const float h = hold(timer); + if (h > 0.0f) return false; // held by deferFor() + if (h == 0.0f) return true; // the hold ran out: due now, whatever the interval float timerInterval = interval(timer); return timerInterval > 0.0f && timerRef(timer) >= timerInterval; } @@ -65,19 +71,27 @@ bool RefreshScheduler::consumeDue(Timer timer) void RefreshScheduler::reset(Timer timer) { timerRef(timer) = 0.0f; + holdRef(timer) = -1.0f; } void RefreshScheduler::markDue(Timer timer) { + holdRef(timer) = -1.0f; // an explicit "now" (user action) overrides a backoff float timerInterval = interval(timer); timerRef(timer) = timerInterval > 0.0f ? timerInterval : 0.0f; } void RefreshScheduler::setTimer(Timer timer, float seconds) { + holdRef(timer) = -1.0f; timerRef(timer) = std::max(0.0f, seconds); } +void RefreshScheduler::deferFor(Timer timer, float seconds) +{ + holdRef(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 c5adf1e..ddf5b45 100644 --- a/src/services/refresh_scheduler.h +++ b/src/services/refresh_scheduler.h @@ -2,6 +2,8 @@ #include "ui/sidebar.h" +#include + namespace dragonx { namespace services { @@ -55,6 +57,11 @@ public: void reset(Timer timer); void markDue(Timer timer); void setTimer(Timer timer, float seconds); + // Hold a timer for exactly `seconds`, whatever the page's interval (which only measures elapsed + // time): not due until then, due as soon as it runs out. consumeDue/reset/markDue/setTimer end it. + // 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); float timer(Timer timer) const; float interval(Timer timer) const; @@ -79,9 +86,14 @@ private: float& timerRef(Timer timer); const float& timerRef(Timer timer) const; + static constexpr int kTimerCount = 8; + float& holdRef(Timer timer) { return holds_[static_cast(timer)]; } + float hold(Timer timer) const { return holds_[static_cast(timer)]; } Intervals intervals_{kCoreDefault, kTransactionDefault, kAddressDefault, kPeerDefault}; Timers timers_; + // deferFor() holds: > 0 counting down (not due), 0 ran out (due), < 0 none. + std::array holds_{{-1.0f, -1.0f, -1.0f, -1.0f, -1.0f, -1.0f, -1.0f, -1.0f}}; }; } // namespace services diff --git a/tests/test_phase4.cpp b/tests/test_phase4.cpp index dedbcd8..d1c5710 100644 --- a/tests/test_phase4.cpp +++ b/tests/test_phase4.cpp @@ -24,6 +24,7 @@ #include "data/wallet_state.h" #include "rpc/connection.h" #include "rpc/rpc_failure.h" +#include "rpc/rpc_busy_gate.h" #include "resources/embedded_resources.h" #include "services/network_refresh_service.h" #include "services/refresh_scheduler.h" @@ -1077,6 +1078,58 @@ void testSpendableFiltering() EXPECT_EQ(includeZero.size(), static_cast(3)); } +// S8: deferFor() is an absolute hold, independent of the page interval (the retry-storm fix). +void testRefreshSchedulerDeferFor() +{ + using dragonx::services::RefreshScheduler; + using Timer = RefreshScheduler::Timer; + { // The old bug, pinned: "setTimer(kCoreDefault - 1)" meant 4s ELAPSED, and Overview's interval is 2s, + // so the retry was due at once (every frame). + RefreshScheduler s; + s.applyPage(dragonx::ui::NavPage::Overview); + s.setTimer(Timer::Core, RefreshScheduler::kCoreDefault - 1.0f); + EXPECT_TRUE(s.isDue(Timer::Core)); + } + { // Overview + deferFor(4): not due at 3.9, due at 4.0. + RefreshScheduler s; + s.applyPage(dragonx::ui::NavPage::Overview); + s.deferFor(Timer::Core, 4.0f); + s.tick(3.9f); + EXPECT_FALSE(s.isDue(Timer::Core)); + s.tick(0.1f); + EXPECT_TRUE(s.isDue(Timer::Core)); + } + { // A page change mid-hold doesn't move it. + RefreshScheduler s; + s.applyPage(dragonx::ui::NavPage::Overview); + s.deferFor(Timer::Core, 4.0f); + s.tick(1.0f); + s.applyPage(dragonx::ui::NavPage::Console); // 10s interval + s.tick(2.9f); + EXPECT_FALSE(s.isDue(Timer::Core)); + s.tick(0.1f); + EXPECT_TRUE(s.isDue(Timer::Core)); + } + { // Shorter than the interval: due at 1.0 on Console, not at 10. + RefreshScheduler s; + s.applyPage(dragonx::ui::NavPage::Console); + s.deferFor(Timer::Core, 1.0f); + s.tick(1.0f); + EXPECT_TRUE(s.consumeDue(Timer::Core)); + // consumeDue ended the hold: back to the 10s interval. + s.tick(5.0f); + EXPECT_FALSE(s.isDue(Timer::Core)); + } + { // A user action (markDue / markImmediateRefresh) overrides a backoff. + RefreshScheduler s; + s.applyPage(dragonx::ui::NavPage::Overview); + s.deferFor(Timer::Core, 30.0f); + s.tick(1.0f); + s.markImmediateRefresh(); + EXPECT_TRUE(s.isDue(Timer::Core)); + } +} + void testRefreshScheduler() { using dragonx::services::RefreshScheduler; @@ -1612,7 +1665,10 @@ void testNetworkRefreshRpcCollectors() EXPECT_FALSE(busyCore.blockchainOk); EXPECT_FALSE(busyCore.balanceOk); EXPECT_TRUE(busyCore.blockchainErrKind == RpcFailureKind::QueueFull); - EXPECT_TRUE(busyCore.balanceErrKind == RpcFailureKind::ReplyTimeout); + // S6: after the chain call found the node busy, no balance scan is sent at all (each would only + // queue behind the same cs_main and take another slot). + EXPECT_TRUE(busyRpc.methodNames() == std::vector({"getblockchaininfo"})); + EXPECT_TRUE(busyCore.balanceErrKind == RpcFailureKind::None); } MockRefreshRpc peerRpc; @@ -4167,6 +4223,104 @@ void testOutputSinceAcrossTrim() EXPECT_TRUE(got.empty()); } +// --------------------------------------------------------------------------------------------------- +// S6: the busy gate (don't flood a busy node's RPC queue) and the connect backoff. +// --------------------------------------------------------------------------------------------------- + +void testRpcBusyGate() +{ + using dragonx::rpc::RpcBusyGate; + using K = dragonx::rpc::RpcFailureKind; + RpcBusyGate g; + EXPECT_TRUE(g.admit("getinfo")); + EXPECT_FALSE(g.busy()); + + // The first reply timeout on a cs_main call flips to busy: nothing needing cs_main goes out. + g.record("getblockchaininfo", K::ReplyTimeout); + EXPECT_TRUE(g.busy()); + EXPECT_EQ(g.abandoned(), 1); + EXPECT_EQ(g.busyPeriods(), 1); + EXPECT_FALSE(g.admit("getinfo")); + EXPECT_FALSE(g.admit("z_gettotalbalance")); + EXPECT_TRUE(g.admit("stop")); // lock-free: always + EXPECT_TRUE(g.admit("help")); + EXPECT_TRUE(g.admit("getrescaninfo")); + + // Lock-free calls say nothing about cs_main; "nothing listening" says nothing about busyness. + g.record("help", K::ReplyTimeout); + g.record("getinfo", K::NotListening); + EXPECT_EQ(g.abandoned(), 1); + EXPECT_TRUE(g.busy()); + g.record("getinfo", K::ReplyTimeout); // a second timeout in the same busy period + EXPECT_EQ(g.busyPeriods(), 1); // ...is not a new period (one canary per period) + + // Any answer clears it, even an application error. + g.record("z_sendmany", K::AppError); + EXPECT_FALSE(g.busy()); + EXPECT_EQ(g.abandoned(), 0); + + // Queue full is busy too, but the refused request holds no slot. + g.record("getinfo", K::QueueFull); + EXPECT_TRUE(g.busy()); + EXPECT_EQ(g.abandoned(), 0); + EXPECT_EQ(g.busyPeriods(), 2); + g.clear(); + EXPECT_FALSE(g.busy()); + + // An hour of polling on two lanes against a node stuck on cs_main: only what was already in flight + // when it turned busy gets abandoned; everything after is held back. (Was ~24: every slot.) + RpcBusyGate hour; + int sent = 0; + for (int second = 0; second < 3600; ++second) { + for (int lane = 0; lane < 2; ++lane) { + if (!hour.admit("getblockchaininfo")) continue; + ++sent; + if (second > 0 || lane == 1) hour.record("getblockchaininfo", K::ReplyTimeout); + } + } + EXPECT_TRUE(sent <= 2); + EXPECT_TRUE(hour.abandoned() <= 2); +} + +void testRpcKindClasses() +{ + using namespace dragonx::rpc; + using K = RpcFailureKind; + for (K k : {K::ReplyTimeout, K::QueueFull, K::Deferred}) { + EXPECT_TRUE(rpcKindMeansBusy(k)); + EXPECT_FALSE(rpcKindMeansLost(k)); // a busy node is still there: never a reconnect + } + for (K k : {K::NotListening, K::ConnectTimeout, K::Auth, K::ShuttingDown, K::Unhandled, K::Other}) { + EXPECT_FALSE(rpcKindMeansBusy(k)); + EXPECT_TRUE(rpcKindMeansLost(k)); + } + for (K k : {K::None, K::Warmup, K::AppError, K::Aborted}) { + EXPECT_FALSE(rpcKindMeansBusy(k)); + EXPECT_FALSE(rpcKindMeansLost(k)); + } +} + +void testConnectRetryBackoff() +{ + using dragonx::rpc::ConnectRetryBackoff; + using K = dragonx::rpc::RpcFailureKind; + ConnectRetryBackoff b; + EXPECT_NEAR(b.next(K::NotListening), 1.0, 1e-6); + EXPECT_NEAR(b.next(K::NotListening), 2.0, 1e-6); + EXPECT_NEAR(b.next(K::ConnectTimeout), 1.0, 1e-6); // a different kind starts over + b.reset(); + const double qf[] = {2, 4, 8, 16, 30, 30}; + for (double want : qf) EXPECT_NEAR(b.next(K::QueueFull), want, 1e-6); + EXPECT_NEAR(b.next(K::ReplyTimeout), 30.0, 1e-6); // the canary owns recovery + EXPECT_NEAR(b.next(K::Other), 4.0, 1e-6); + // A minute of queue-full refusals: a handful of attempts, not thousands (was ~3,000/min). + b.reset(); + double t = 0; + int attempts = 0; + while (t < 60.0) { t += b.next(K::QueueFull); ++attempts; } + EXPECT_TRUE(attempts <= 6); +} + // S3: on Linux a node in its final shutdown phase reads comm "hush-shutoff"; its executable still says // dragonxd, so it must still count as running (it holds the datadir lock until it exits). void testDaemonProcessIdentity() @@ -8737,6 +8891,10 @@ int main() testAmountFormatting(); testSpendableFiltering(); testRefreshScheduler(); + testRefreshSchedulerDeferFor(); + testRpcBusyGate(); + testRpcKindClasses(); + testConnectRetryBackoff(); testNetworkRefreshService(); testNetworkRefreshSnapshotHelpers(); testParseDaemonRescanOutput();