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 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
+15
-5
@@ -499,6 +499,7 @@ bool App::init()
|
||||
|
||||
// Initialize RPC client
|
||||
rpc_ = std::make_unique<rpc::RPCClient>();
|
||||
rpc_->setBusyGate(rpc_busy_gate_);
|
||||
|
||||
// Initialize background RPC worker thread
|
||||
worker_ = std::make_unique<rpc::RPCWorker>();
|
||||
@@ -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.
|
||||
|
||||
@@ -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<bool> 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<void()> 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::RpcBusyGate> rpc_busy_gate_ = std::make_shared<rpc::RpcBusyGate>();
|
||||
rpc::ConnectRetryBackoff connect_backoff_; // S8: per-failure-kind delay before the next connect attempt
|
||||
std::thread canary_thread_;
|
||||
std::atomic<int> 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();
|
||||
|
||||
|
||||
+91
-15
@@ -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<std::mutex> 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<int>(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<std::mutex> 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<std::mutex> 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<rpc::RPCClient>();
|
||||
fast_rpc_->setBusyGate(rpc_busy_gate_);
|
||||
}
|
||||
if (!fast_worker_) {
|
||||
fast_worker_ = std::make_unique<rpc::RPCWorker>();
|
||||
@@ -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");
|
||||
|
||||
@@ -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 <chrono>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
|
||||
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<std::mutex> 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<std::mutex> 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<std::mutex> lk(mx_);
|
||||
clearLocked();
|
||||
}
|
||||
|
||||
bool busy() const
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<std::mutex> 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<std::mutex> lk(mx_);
|
||||
return busyPeriods_;
|
||||
}
|
||||
double busySeconds() const
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(mx_);
|
||||
if (!busy_) return 0.0;
|
||||
return std::chrono::duration<double>(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
|
||||
+11
-2
@@ -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;
|
||||
}
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
|
||||
#include "types.h"
|
||||
#include "rpc_failure.h" // RpcError, RpcTransportError, RpcFailureKind
|
||||
#include "rpc_busy_gate.h"
|
||||
#include <atomic>
|
||||
#include <string>
|
||||
#include <functional>
|
||||
@@ -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<RpcBusyGate> 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<RpcBusyGate> 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_;
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
#include "network_refresh_service.h"
|
||||
#include "rpc/rpc_busy_gate.h"
|
||||
#include "../util/logger.h"
|
||||
|
||||
#include <algorithm>
|
||||
@@ -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.
|
||||
|
||||
@@ -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(); }
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
|
||||
#include "ui/sidebar.h"
|
||||
|
||||
#include <array>
|
||||
|
||||
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<int>(timer)]; }
|
||||
float hold(Timer timer) const { return holds_[static_cast<int>(timer)]; }
|
||||
|
||||
Intervals intervals_{kCoreDefault, kTransactionDefault, kAddressDefault, kPeerDefault};
|
||||
Timers timers_;
|
||||
// deferFor() holds: > 0 counting down (not due), 0 ran out (due), < 0 none.
|
||||
std::array<float, kTimerCount> holds_{{-1.0f, -1.0f, -1.0f, -1.0f, -1.0f, -1.0f, -1.0f, -1.0f}};
|
||||
};
|
||||
|
||||
} // namespace services
|
||||
|
||||
Reference in New Issue
Block a user