From 0f25c67600eb64279236c09c4b9ee4aac0fd32e8 Mon Sep 17 00:00:00 2001 From: DanS Date: Fri, 2 Oct 2026 09:49:55 -0500 Subject: [PATCH] feat(rpc): stop flooding a busy node's RPC queue; fix the reconnect storm dragonxd serves RPC with 8 workers and a 16-deep queue. While a witness rebuild holds cs_main, every request needing it waits, and a request the client gave up on keeps its slot. The GUI kept polling and re-probing until all 24 slots were its own abandoned requests (even stop was then refused), treated the busy node as lost (tearing down the wallet state), and retried connect at frame rate on some pages. Busy gate (src/rpc/rpc_busy_gate.h), enforced in RPCClient::performCall for the shared rpc_ / fast_rpc_: - the first reply timeout or queue-full refusal marks the node busy; from then on nothing needing cs_main is sent (calls fail at once with the new RpcFailureKind::Deferred). stop, help and getrescaninfo are lock-free in dragonxd and always go through - one canary getinfo with no client timeout waits on its own client and thread; its answer clears the gate and triggers an immediate refresh (queue-full: retried after 2..30 s; no answer: back to normal handling) - the timer dispatch skips Core/Tx/Addresses/Peers/chat/mining/opid and connect attempts while busy - the 3-strike teardown counts only failures that put the connection in doubt, never busy ones; no balance scans follow a busy chain-info call Reconnect storm: retries were armed with setTimer(Core, kCoreDefault - 1), i.e. "4 s elapsed", which against Overview's 2 s interval meant every frame. RefreshScheduler::deferFor() is an absolute hold, and ConnectRetryBackoff picks the delay by failure kind (not listening 1/2/5 s, queue full 2..30 s, busy 30 s, else 4 s). Measured against the fake daemon: abandoned requests during a rebuild went from ~24 (every slot) to at most 2, with no teardown and recovery when the rebuild ends; a saturated queue sees about one canary per back-off step instead of ~190 requests/s. Co-Authored-By: Claude Opus 5.5 --- src/app.cpp | 20 ++- src/app.h | 14 ++ src/app_network.cpp | 106 ++++++++++++-- src/rpc/rpc_busy_gate.h | 168 +++++++++++++++++++++++ src/rpc/rpc_client.cpp | 13 +- src/rpc/rpc_client.h | 10 ++ src/rpc/rpc_failure.h | 2 + src/services/network_refresh_service.cpp | 5 +- src/services/network_refresh_service.h | 1 + src/services/refresh_scheduler.cpp | 14 ++ src/services/refresh_scheduler.h | 12 ++ tests/test_phase4.cpp | 160 ++++++++++++++++++++- 12 files changed, 501 insertions(+), 24 deletions(-) create mode 100644 src/rpc/rpc_busy_gate.h 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();