diff --git a/src/c2pool/main_dash.cpp b/src/c2pool/main_dash.cpp index d151caaea..b9390a67b 100644 --- a/src/c2pool/main_dash.cpp +++ b/src/c2pool/main_dash.cpp @@ -92,6 +92,8 @@ #include #include +#include // io-thread-decouple: background RPC pool +#include #include #include // std::getenv @@ -1495,49 +1497,94 @@ int run_node(bool testnet, const std::string& rpc_endpoint, // trivial RPC; failures are swallowed so a daemon hiccup never crashes the // run-loop (retry next tick). Reuses the existing NodeRPC client — no new // dependency, no dashd config change, no new notify mechanism. + // io-thread-decouple: dedicated single-thread pool for the fallback arm's + // BLOCKING dashd RPC (getbestblockhash tip probe + the background template + // re-source). Mirrors main_ltc.cpp hdr_pool ("keeps scrypt off io_context"): + // synchronous beast I/O runs HERE, never on the stratum io_context, so 60+ + // sessions never starve while dashd is queried (or wedged -- the NodeRPC + // socket timeout + m_rpc_mutex bound this thread). Declared here (after rpc / + // work_source / stratum_server) so its explicit stop()+join() after the run + // loop -- and its destructor -- happen BEFORE those objects unwind: no + // background probe is ever mid-flight against freed state. Only created on + // the fallback arm; null on the embedded arm (legacy inline path unchanged). + std::shared_ptr rpc_pool; if (!coin_p2p && rpc && stratum_server) { + rpc_pool = std::make_shared(1); + + // Non-blocking template re-source: cached_work() hands the blocking + // select_work()/GBT to rpc_pool as a single-flight background job + // instead of blocking the io thread on every stale/generation miss (the + // per-share ~15-30 s GBT block). The io thread serves the cached template + // immediately; the pool updates it and the next notify picks it up. + work_source->set_refresh_executor( + [rpc_pool](std::function job) { + boost::asio::post(*rpc_pool, std::move(job)); + }); + auto tip_timer = std::make_shared(ioc); auto last_tip = std::make_shared(); auto tip_tick = std::make_shared< std::function>(); + // tip_tick runs ON ioc when the 3 s timer fires. It does NOT call the + // blocking RPC itself: it hands getbestblockhash to rpc_pool and posts + // the tip-change follow-up (invalidate + bump + notify) + the timer + // re-arm BACK onto ioc (ws/ss/tip_timer are io-thread-confined), exactly + // like main_ltc.cpp's post-to-pool -> post-back-to-ioc pattern. The + // timer is re-armed only AFTER the RPC completes, so a slow dashd cannot + // pile up overlapping polls. Lost-block-prevention is preserved: a real + // tip change still fires invalidate + notify -- only WHERE the probe runs + // has moved off the stratum io thread. *tip_tick = [rpc = rpc.get(), ws = work_source.get(), - ss = stratum_server.get(), tip_timer, last_tip, tip_tick]( - const boost::system::error_code& ec) { + ss = stratum_server.get(), tip_timer, last_tip, tip_tick, + rpc_pool, &ioc](const boost::system::error_code& ec) { if (ec) return; // cancelled at shutdown - try { - const std::string tip = rpc->getbestblockhash(); - if (!tip.empty() && *last_tip != tip) { - const bool first_seen = last_tip->empty(); - *last_tip = tip; - // Skip the notify on the very first observation (baseline); - // only a genuine tip CHANGE forces a refresh. - if (!first_seen) { - ws->invalidate_template_cache( - "tip-poll: dashd best-block changed"); - ws->bump_work_generation(); - ss->notify_all(); - LOG_INFO << "[Stratum] tip-poll: new tip " - << tip.substr(0, 16) - << ", forcing template refresh + notify"; - std::cout << "[Stratum] tip-poll: new tip " - << tip.substr(0, 16) - << ", forcing template refresh + notify\n"; + boost::asio::post(*rpc_pool, + [rpc, ws, ss, tip_timer, last_tip, tip_tick, &ioc]() { + std::string tip; + bool ok = false; + try { + tip = rpc->getbestblockhash(); // BLOCKING -- BACKGROUND THREAD + ok = true; + } catch (const std::exception& e) { + LOG_WARNING << "[Stratum] tip-poll getbestblockhash failed " + "(non-fatal, retry next tick): " << e.what(); + } catch (...) { + // swallow — never crash on a tip probe } - } - } catch (const std::exception& e) { - LOG_WARNING << "[Stratum] tip-poll getbestblockhash failed " - "(non-fatal, retry next tick): " << e.what(); - } catch (...) { - // swallow — never crash the run-loop on a tip probe - } - tip_timer->expires_after(std::chrono::seconds(3)); - tip_timer->async_wait(*tip_tick); + // Follow-up + timer re-arm run BACK ON ioc (io-thread-confined + // state). If ioc is already stopped (shutdown) this handler + // simply never runs -> the poll stops cleanly. + boost::asio::post(ioc, + [ok, tip = std::move(tip), ws, ss, tip_timer, last_tip, tip_tick]() { + if (ok && !tip.empty() && *last_tip != tip) { + const bool first_seen = last_tip->empty(); + *last_tip = tip; + // Skip the notify on the very first observation + // (baseline); only a genuine tip CHANGE refreshes. + if (!first_seen) { + ws->invalidate_template_cache( + "tip-poll: dashd best-block changed"); + ws->bump_work_generation(); + ss->notify_all(); + LOG_INFO << "[Stratum] tip-poll: new tip " + << tip.substr(0, 16) + << ", forcing template refresh + notify"; + std::cout << "[Stratum] tip-poll: new tip " + << tip.substr(0, 16) + << ", forcing template refresh + notify\n"; + } + } + tip_timer->expires_after(std::chrono::seconds(3)); + tip_timer->async_wait(*tip_tick); + }); + }); }; tip_timer->expires_after(std::chrono::seconds(3)); tip_timer->async_wait(*tip_tick); std::cout << "[run] fallback-arm tip-poll ARMED (dashd getbestblockhash " - "every 3 s -> event-driven template refresh + clean_jobs " - "notify on tip change)\n"; + "every 3 s on a dedicated RPC thread -> event-driven template " + "refresh + clean_jobs notify on tip change; io thread never " + "blocks on dashd)\n"; } std::cout << "[run] run-loop up (Ctrl-C to stop); won blocks relay DUAL-PATH:\n" @@ -1567,6 +1614,17 @@ int run_node(bool testnet, const std::string& rpc_endpoint, } } + // io-thread-decouple: join the background RPC pool FIRST, before any of the + // objects it dereferences (rpc / work_source / stratum_server) unwind. run() + // has returned (ioc stopped), so no NEW work is posted; stop()+join() waits + // out any in-flight getbestblockhash/GBT re-source (bounded by the NodeRPC + // socket timeout) so no pool thread ever touches freed state. Any post-back + // to the stopped ioc simply never executes. + if (rpc_pool) { + rpc_pool->stop(); + rpc_pool->join(); + } + // Tear the acceptor + sessions down while the work source and node_coin_state // it references are still alive -- explicit reset keeps destruction order safe // (stratum_server was declared before them, so it would otherwise outlive them). diff --git a/src/core/stratum_server.cpp b/src/core/stratum_server.cpp index bd0d45e95..2cdea2042 100644 --- a/src/core/stratum_server.cpp +++ b/src/core/stratum_server.cpp @@ -92,6 +92,7 @@ StratumServer::StratumServer(net::io_context& ioc, const std::string& address, u , bind_address_(address) , port_(port) , running_(false) + , idle_reap_timer_(ioc) { } @@ -113,7 +114,8 @@ bool StratumServer::start() running_ = true; accept_connections(); - + start_idle_reaper(); // no-op unless session_idle_timeout_sec > 0 + LOG_INFO << "StratumServer started on " << bind_address_ << ":" << port_; return true; @@ -133,6 +135,7 @@ void StratumServer::stop() boost::system::error_code ec; acceptor_.cancel(ec); acceptor_.close(ec); + idle_reap_timer_.cancel(); // stop the zombie-session reaper running_ = false; // Snapshot + clear the live-sessions set under the mutex, then close @@ -383,6 +386,58 @@ void StratumServer::notify_all() } } +void StratumServer::start_idle_reaper() +{ + // Zombie-session reaper (belt-and-suspenders alongside OS TCP keepalive): + // periodically close sessions that have sent no inbound line past the idle + // timeout. A live miner submits shares far inside the window; a half-open + // NAT-dead peer sends nothing. No-op (never re-armed) unless the DASH-set + // StratumConfig::session_idle_timeout_sec > 0 -> other coins unchanged. + const auto& cfg0 = mining_interface_ ? mining_interface_->get_stratum_config() + : core::stratum::StratumConfig{}; + const uint32_t timeout = cfg0.session_idle_timeout_sec; + const bool keepalive_on = cfg0.tcp_keepalive_enabled; + if (timeout == 0 || !running_) return; + + // Sweep at half the timeout, bounded to [15 s, 60 s], so a dead session is + // reclaimed within ~1.5x the timeout. + const uint32_t period = std::max(15, std::min(60, timeout / 2)); + idle_reap_timer_.expires_after(std::chrono::seconds(period)); + idle_reap_timer_.async_wait([this, timeout, keepalive_on](const boost::system::error_code& ec) { + if (ec) return; // cancelled at shutdown + if (!running_) return; + std::vector> snapshot; + { + std::lock_guard lk(sessions_mutex_); + snapshot.assign(sessions_.begin(), sessions_.end()); + } + size_t reaped = 0; + for (auto& s : snapshot) { + if (!s->is_connected()) + continue; // socket already dead -> notify_all prunes it + if (s->seconds_since_activity() <= timeout) + continue; // recently active + // KEEPALIVE-AWARE: when OS TCP keepalive is enabled it is the + // liveness authority -- a still-open socket means keepalive has NOT + // declared the peer dead (it force-closes dead NAT paths, which the + // is_connected() guard above then catches). Do NOT reap an idle-but- + // keepalive-validated AUTHORIZED rig: a high fixed-diff suffix + // (ADDR+N) can legitimately exceed the idle window between submits, + // and reaping it would drop a LIVE, paying miner. Idle-time reaping + // is the FALLBACK only where keepalive is unavailable; unauthorized + // sessions (never completed handshake) are always fair game. + if (keepalive_on && s->is_authorized()) + continue; + s->drop_stale("idle timeout (no inbound share/line)"); + ++reaped; + } + if (reaped) + LOG_INFO << "[Stratum] idle-reaper closed " << reaped + << " stale session(s) (> " << timeout << "s idle)"; + start_idle_reaper(); // re-arm + }); +} + /// StratumSession Implementation StratumSession::StratumSession(tcp::socket socket, std::shared_ptr mining_interface, StratumServer* server) @@ -390,7 +445,9 @@ StratumSession::StratumSession(tcp::socket socket, std::shared_ptr , mining_interface_(mining_interface) , server_(server) , connected_at_(std::chrono::steady_clock::now()) + , last_activity_(connected_at_) , work_push_timer_(socket_.get_executor()) + , handshake_timer_(socket_.get_executor()) { subscription_id_ = generate_subscription_id(); extranonce1_ = generate_extranonce1(); @@ -411,6 +468,8 @@ void StratumSession::start() auto ep = socket_.remote_endpoint(ec); if (ec) return; // socket closed before start() was dispatched LOG_INFO << "StratumSession started for client: " << ep; + apply_socket_keepalive(); // OS keepalive (no-op unless enabled) -- reaps NAT-dead peers + arm_handshake_timer(); // authorize deadline (no-op unless configured) read_message(); } @@ -445,10 +504,13 @@ void StratumSession::read_message() void StratumSession::process_message(std::size_t bytes_read) { try { + // Zombie-session reaper: a received line is proof of a live peer. + last_activity_ = std::chrono::steady_clock::now(); + std::istream is(&buffer_); std::string line; std::getline(is, line); - + if (!line.empty() && line.back() == '\r') { line.pop_back(); // Remove \r if present } @@ -531,6 +593,8 @@ nlohmann::json StratumSession::handle_authorize(const nlohmann::json& params, co if (params.size() >= 1 && params[0].is_string()) { username_ = params[0]; authorized_ = true; + // Handshake completed -> disarm the authorize deadline (zombie-session fix). + handshake_timer_.cancel(); // ─── Step 1: Strip fixed difficulty suffix (+N) before any parsing ─── // Format: "ADDRESS+1024" or "ADDR,ADDR+512" → suggested_difficulty_=N @@ -1289,6 +1353,20 @@ nlohmann::json StratumSession::handle_submit(const nlohmann::json& params, const void StratumSession::send_response(const nlohmann::json& response) { try { + // Write-queue backlog cap (zombie-session fix): a dead peer whose kernel + // send buffer has filled never drains the async_write, so the queue + // grows unbounded. Past the configured depth, drop the session rather + // than accumulate. 0 = unlimited (legacy; other coins unchanged). + const size_t cap = mining_interface_ + ? mining_interface_->get_stratum_config().max_write_queue_depth + : 0; + if (cap > 0 && write_queue_.size() >= cap) { + LOG_WARNING << "[Stratum] session " << session_id_ + << " write backlog " << write_queue_.size() + << " >= cap " << cap << " -- dropping stuck session"; + drop_stale("write-queue backlog exceeded"); + return; + } std::string message = response.dump() + "\n"; write_queue_.push_back(std::move(message)); if (!writing_) @@ -1790,12 +1868,82 @@ void StratumSession::cancel_timers() // Cancel pending timer — callback fires with ec=operation_aborted → returns. // Timer is a member, so it outlives all its callbacks (no use-after-free). work_push_timer_.cancel(); + handshake_timer_.cancel(); // Close socket so is_connected() returns false for any already-dequeued // callbacks that fire with ec=success after cancel(). boost::system::error_code ec; socket_.close(ec); } +// ── Zombie-session fixes (transport/liveness only; consensus-neutral) ──────── + +void StratumSession::apply_socket_keepalive() +{ + // Enable OS TCP keepalive so the kernel actively probes an idle peer and + // errors the pending async_read when a NAT path has gone dead (the peer + // never sent FIN/RST). This is THE root fix for immortal subscribed + // sessions -- socket_.is_open() alone never falsifies for a half-open NAT + // drop. No-op unless the (DASH-set) StratumConfig knob is enabled, so + // LTC/BTC/DGB behaviour is unchanged. + const auto& cfg = mining_interface_->get_stratum_config(); + if (!cfg.tcp_keepalive_enabled) return; + + boost::system::error_code ec; + socket_.set_option(boost::asio::socket_base::keep_alive(true), ec); // portable (asio) + if (ec) { + LOG_WARNING << "[Stratum] SO_KEEPALIVE set failed: " << ec.message(); + return; + } + // Per-probe tuning (idle/interval/count) is Linux-specific -- Windows tunes + // keepalive via WSAIoctl(SIO_KEEPALIVE_VALS) and macOS via TCP_KEEPALIVE, so + // guard exactly like sample_tcp_rtt_ms() above (#ifdef __linux__). Off-Linux + // the portable keep_alive(true) above still arms keepalive with OS defaults; + // the DASH deploy target is Linux, where the 60/10/3 tuning (~90 s detect) + // applies. This keeps core building on the Windows/macOS CI lanes. +#ifdef __linux__ + const int fd = socket_.native_handle(); + int idle = static_cast(cfg.tcp_keepalive_idle_sec); + int intvl = static_cast(cfg.tcp_keepalive_interval_sec); + int cnt = static_cast(cfg.tcp_keepalive_count); + ::setsockopt(fd, IPPROTO_TCP, TCP_KEEPIDLE, &idle, sizeof(idle)); + ::setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, &intvl, sizeof(intvl)); + ::setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, &cnt, sizeof(cnt)); + LOG_TRACE << "[Stratum] TCP keepalive armed (idle=" << idle + << "s interval=" << intvl << "s count=" << cnt << ")"; +#else + LOG_TRACE << "[Stratum] TCP keepalive armed (OS-default probe timing; " + "per-probe tuning is Linux-only)"; +#endif +} + +void StratumSession::arm_handshake_timer() +{ + // Drop a session that never completes mining.authorize within the deadline + // (a live rig authorizes in << 1 s; a half-open probe never does). Disarmed + // in handle_authorize(). No-op unless configured. Member timer + weak self + // keepalive so a fired-after-close callback is a safe no-op. + const auto& cfg = mining_interface_->get_stratum_config(); + if (cfg.handshake_timeout_sec == 0) return; + handshake_timer_.expires_after(std::chrono::seconds(cfg.handshake_timeout_sec)); + std::weak_ptr weak = shared_from_this(); + handshake_timer_.async_wait([weak](const boost::system::error_code& ec) { + if (ec) return; // cancelled (authorized, or shutdown) + auto self = weak.lock(); + if (!self) return; + if (!self->authorized_ && self->is_connected()) + self->drop_stale("handshake (authorize) deadline"); + }); +} + +void StratumSession::drop_stale(const char* reason) +{ + LOG_INFO << "[Stratum] dropping stale session " << session_id_ + << " (" << reason << ")"; + // cancel_timers() closes the socket; the pending async_read then fails and + // runs the standard disconnect path (unregister worker, log). Idempotent. + cancel_timers(); +} + // Parse multi-chain addresses from username string. // Tries multiple separator formats for maximum miner compatibility: // 1. Slash+colon: "LTC_ADDR/98:DOGE_ADDR" (explicit chain ID) diff --git a/src/core/stratum_server.hpp b/src/core/stratum_server.hpp index 5cbc8bac7..3ff20c581 100644 --- a/src/core/stratum_server.hpp +++ b/src/core/stratum_server.hpp @@ -190,6 +190,10 @@ class StratumSession : public std::enable_shared_from_this uint64_t stale_shares_ = 0; uint64_t doa_shares_ = 0; // block-template changed at submit (live-vs-frozen prevhash mismatch); accounting only, share still broadcasts std::chrono::steady_clock::time_point connected_at_; + // Last inbound-line timestamp (zombie-session reaper, StratumConfig:: + // session_idle_timeout_sec). Updated on every received stratum line; a + // half-open NAT-dead peer never advances it. Init to connected_at_. + std::chrono::steady_clock::time_point last_activity_; std::string session_id_; // Reject-reason diagnostics throttle: emit the low-diff (P5) and @@ -218,6 +222,11 @@ class StratumSession : public std::enable_shared_from_this // where the transport/protocol object owns its timers. boost::asio::steady_timer work_push_timer_; + // Handshake deadline (zombie-session fix, StratumConfig::handshake_timeout_ + // sec): armed at start(), disarmed once authorize completes. A session that + // never authorizes (half-open / probe) is dropped when it fires. + boost::asio::steady_timer handshake_timer_; + public: explicit StratumSession(tcp::socket socket, std::shared_ptr mining_interface, StratumServer* server = nullptr); @@ -225,6 +234,15 @@ class StratumSession : public std::enable_shared_from_this bool is_connected() const { return socket_.is_open(); } bool is_subscribed() const { return subscribed_; } + bool is_authorized() const { return authorized_; } + // Seconds since the last inbound stratum line (zombie-session reaper). + double seconds_since_activity() const { + return std::chrono::duration( + std::chrono::steady_clock::now() - last_activity_).count(); + } + // Reap a stale/zombie session: cancel timers + close the socket. The pending + // async_read then fails and runs the normal disconnect path. Idempotent. + void drop_stale(const char* reason); // Diagnostics / test introspection (read-only; call quiesced or from the // session's io_context thread — active_jobs_ mutates on notify/submit). size_t active_job_count() const { return active_jobs_.size(); } @@ -267,6 +285,10 @@ class StratumSession : public std::enable_shared_from_this void start_periodic_work_push(); void schedule_work_push_timer(); void cancel_timers(); + // Zombie-session fixes (all no-op unless the DASH-set StratumConfig knobs + // enable them): apply OS TCP keepalive, arm the authorize deadline. + void apply_socket_keepalive(); + void arm_handshake_timer(); std::string generate_extranonce1(); void parse_address_separators(std::string& username, std::string& merged_addr_raw); @@ -303,6 +325,13 @@ class StratumServer mutable double addr_rates_cache_ts_ = 0.0; mutable std::mutex cache_mutex_; + // Zombie-session idle reaper (StratumConfig::session_idle_timeout_sec). + // Periodically closes sessions with no inbound activity past the timeout so + // a NAT-dead peer that OS keepalive somehow missed is still reclaimed. + // Disarmed (never scheduled) when the timeout is 0 -> legacy no-op. + boost::asio::steady_timer idle_reap_timer_; + void start_idle_reaper(); + public: StratumServer(net::io_context& ioc, const std::string& address, uint16_t port, std::shared_ptr mining_interface); ~StratumServer(); diff --git a/src/core/stratum_types.hpp b/src/core/stratum_types.hpp index 4764f1b55..3cbf47424 100644 --- a/src/core/stratum_types.hpp +++ b/src/core/stratum_types.hpp @@ -67,6 +67,36 @@ struct StratumConfig { // fetches, and the legacy stub-coinbase fallback is unreachable. // Default false: LTC/BTC/DGB behavior is byte-unchanged. bool require_job_snapshot{false}; + + // ── Live-session hygiene (mining-hotel ZOMBIE-SESSION LEAK fix) ────────── + // A NAT-dropped miner's TCP connection is frequently never FIN/RST'd, so + // StratumSession::is_connected() (socket_.is_open()) stays true FOREVER and + // the session is never reaped -- every failed rig retry then mints an + // immortal subscribed session that keeps drawing full per-notify job builds + // (observed: 66 sockets for ~23 rigs). At the default cap these zombies also + // lock real rigs out (admission control). These knobs bound that leak. ALL + // default OFF/neutral so LTC/BTC/DGB behaviour is BYTE-UNCHANGED; DASH opts + // in (impl/dash/stratum/work_source.cpp ctor). Transport/liveness only -- + // zero wire-byte change and consensus-neutral. + // + // (a) OS TCP keepalive: the kernel probes an idle peer and errors the + // pending async_read on a dead NAT path, which runs the normal + // disconnect+prune. THE root fix for the immortal-session class. + bool tcp_keepalive_enabled = false; + uint32_t tcp_keepalive_idle_sec = 60; // begin probing after 60 s idle + uint32_t tcp_keepalive_interval_sec = 10; // 10 s between probes + uint32_t tcp_keepalive_count = 3; // drop after 3 failed probes + // (b) Handshake deadline: drop a session that has not completed + // mining.authorize within N s (a live rig authorizes in << 1 s). 0=off. + uint32_t handshake_timeout_sec = 0; + // (c) Application idle reaper (belt-and-suspenders for keepalive): reap a + // session that has sent NO inbound line for N s. A live miner submits + // shares far inside this window; a half-open zombie sends nothing. 0=off. + uint32_t session_idle_timeout_sec = 0; + // (d) Write-queue backlog cap: drop a session whose un-acked write backlog + // exceeds this many frames (a dead peer whose kernel send buffer filled + // accumulates unbounded). 0=unlimited (legacy). + size_t max_write_queue_depth = 0; }; /// Frozen share-construction fields returned by ref_hash_fn. These diff --git a/src/impl/dash/coin/rpc.cpp b/src/impl/dash/coin/rpc.cpp index aa42adf06..04a415f58 100644 --- a/src/impl/dash/coin/rpc.cpp +++ b/src/impl/dash/coin/rpc.cpp @@ -9,6 +9,11 @@ #include // ParseHex / HexStr +#ifndef _WIN32 +#include // setsockopt SO_SND/RCVTIMEO (Send() deadline) +#include // struct timeval +#endif + namespace dash { @@ -75,6 +80,10 @@ void NodeRPC::connect(NetService address, std::string userpass) { try { + // Bound the synchronous Send() I/O BEFORE check() + // runs its first RPC, so a wedged dashd cannot hang + // the caller (io thread or background rpc_pool). + apply_socket_timeouts(); if (check()) { m_connected = true; @@ -147,11 +156,59 @@ void NodeRPC::sync_reconnect() LOG_WARNING << "CoindRPC sync_reconnect connect failed: " << ec.message(); return; } + apply_socket_timeouts(); // re-arm the Send() deadline on the fresh socket LOG_INFO << "CoindRPC reconnected (sync)"; } +void NodeRPC::close_stream() +{ + // Same teardown idiom as sync_reconnect()/the destructor. m_connected is + // cleared so the next Send() write-fails into a clean sync_reconnect(). + beast::error_code ec; + m_stream.socket().shutdown(io::ip::tcp::socket::shutdown_both, ec); + m_stream.close(); + m_connected = false; +} + +void NodeRPC::apply_socket_timeouts() +{ + // Force the socket back to BLOCKING mode, then set kernel send/receive + // timeouts. async_connect (connect()) leaves asio's non_blocking flag set; + // under it a synchronous recv returns EAGAIN immediately and SO_RCVTIMEO is + // a no-op. In blocking mode the sync http::write/http::read inside Send() + // do true blocking send()/recv() which the kernel bounds by SO_SND/RCVTIMEO, + // returning an error after RPC_IO_TIMEOUT_SECONDS instead of hanging on a + // wedged dashd. (beast tcp_stream::expires_after cannot be used: it only + // governs ASYNC ops -- the sync path calls socket.read_some() directly.) + // Per-operation inactivity timeout: a large-but-steadily-arriving GBT does + // NOT trip it; only a genuine stall does. POSIX (SO_*TIMEO takes a timeval); + // guarded off Windows (no /timeval, and Winsock SO_RCVTIMEO takes + // a DWORD) -- the DASH deploy target is Linux, and macOS keeps the real + // deadline. On Windows this is a no-op (pre-PR behaviour: no deadline). +#ifndef _WIN32 + if (!m_stream.socket().is_open()) + return; + boost::system::error_code ec; + m_stream.socket().non_blocking(false, ec); // guarantee SO_*TIMEO applies + if (ec) + LOG_WARNING << "CoindRPC: could not set blocking mode: " << ec.message(); + struct timeval tv; + tv.tv_sec = RPC_IO_TIMEOUT_SECONDS; + tv.tv_usec = 0; + const int fd = m_stream.socket().native_handle(); + ::setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); + ::setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); +#endif +} + std::string NodeRPC::Send(const std::string &request) { + // io-thread-decouple / LTC parity (impl/ltc/coin/rpc.cpp:136): serialize the + // whole write+read+reconnect. getwork()/submit_block_hex (io thread) and the + // tip-poll's getbestblockhash (background rpc_pool thread) both drive this + // single shared beast client; m_http_request/m_stream are not thread-safe. + // The socket deadline (apply_socket_timeouts) bounds how long the lock is held. + std::lock_guard _rpc_lock(m_rpc_mutex); // Retry once after synchronous reconnect on write/read failure for (int attempt = 0; attempt < 2; ++attempt) { @@ -169,6 +226,10 @@ std::string NodeRPC::Send(const std::string &request) sync_reconnect(); continue; } + // Deadline-desync guard (see the read-fail path below): tear the + // socket down before giving up so a partially-written request can + // never be answered into a LATER Send() on a reused connection. + close_stream(); return {}; } @@ -187,6 +248,17 @@ std::string NodeRPC::Send(const std::string &request) sync_reconnect(); continue; } + // Deadline-desync guard (SO_RCVTIMEO deadline hazard): this final + // attempt has already WRITTEN the request into dashd but its response + // is unread (read timed out). jsonrpccxx reuses the constant id + // "curltest" and never validates response ids, so reusing this open + // connection would make the NEXT Send() read THIS request's late + // response -> permanent off-by-one desync (a GBT would parse a + // getbestblockhash string -> zeroed work -> a stuck "honest set-gap" + // outage until the connection churns). Tear the socket down so the + // next Send() write-fails into a clean sync_reconnect(). Already + // under m_rpc_mutex. + close_stream(); return {}; } diff --git a/src/impl/dash/coin/rpc.hpp b/src/impl/dash/coin/rpc.hpp index 4bb797115..f1d2742f1 100644 --- a/src/impl/dash/coin/rpc.hpp +++ b/src/impl/dash/coin/rpc.hpp @@ -42,6 +42,7 @@ #include #include +#include #include #include @@ -76,6 +77,17 @@ class NodeRPC : public jsonrpccxx::IClientConnector beast::tcp_stream m_stream; boost::asio::ip::tcp::resolver m_resolver; http::request m_http_request; + // Serializes Send() (verbatim LTC parity, impl/ltc/coin/rpc.hpp:43-47). + // io-thread-decouple drives this ONE shared client from TWO threads: the + // stratum io_context thread (getwork() during a template re-source, and ARM + // B submit_block_hex on a won block) AND the background rpc_pool thread (the + // tip poll's getbestblockhash + the single-flight GBT refresh). m_http_request + // + m_stream are NOT thread-safe: without this lock one thread's + // prepare_payload() frees the Content-Length field element mid-write on the + // other -> UAF / interleaved HTTP frames / cross-wired responses (a garbled + // submitblock = lost block on the paying node). The lock also serializes the + // sync_reconnect() retry path, which is only ever entered from inside Send(). + std::mutex m_rpc_mutex; std::unique_ptr m_auth; jsonrpccxx::JsonRpcClient m_client; @@ -96,6 +108,28 @@ class NodeRPC : public jsonrpccxx::IClientConnector std::string Send(const std::string &request) override; nlohmann::json CallAPIMethod(const std::string& method, const jsonrpccxx::positional_parameter& params = {}); + // io-thread-decouple: a REAL deadline on the synchronous Send() I/O so a + // wedged-but-connected dashd (socket open, no bytes) cannot hang the caller + // forever -- which on the background rpc_pool thread would wedge + // template_refresh_inflight_ true (permanent set-gap after the next tip + // change), stop the tip-poll re-arming, AND block rpc_pool->join() at + // shutdown (SIGKILL needed). beast's SYNCHRONOUS read_some/write_some call + // socket.read_some() directly and do NOT honour tcp_stream::expires_after + // (that timer only fires for ASYNC ops -- basic_stream.hpp), so the robust + // bound is a kernel SO_RCV/SNDTIMEO on the socket forced back to BLOCKING + // mode (async_connect leaves asio's non_blocking flag set, under which + // SO_*TIMEO is a no-op). apply_socket_timeouts() does both; it is called + // after every successful (re)connect. Linux release target. + static constexpr int RPC_IO_TIMEOUT_SECONDS = 12; + void apply_socket_timeouts(); + // Tear down m_stream (sync_reconnect idiom). Called on a FINAL-attempt Send() + // failure so a half-written / unread request can never be answered into a + // later Send() on a reused connection -- the deadline-desync guard (the + // constant "curltest" id + jsonrpccxx not validating response ids makes any + // reused-connection off-by-one a permanent set-gap outage). Caller holds + // m_rpc_mutex. + void close_stream(); + public: NodeRPC(io::io_context* context, dash::interfaces::Node* coin, bool testnet); ~NodeRPC(); diff --git a/src/impl/dash/stratum/work_source.cpp b/src/impl/dash/stratum/work_source.cpp index 21af1548c..ff951fa56 100644 --- a/src/impl/dash/stratum/work_source.cpp +++ b/src/impl/dash/stratum/work_source.cpp @@ -142,6 +142,25 @@ DASHWorkSource::DASHWorkSource(const coin::NodeCoinState& coin_state, // Derived directly from a smoothed hashrate, so it cannot oscillate. DASH-only; // other coins keep the legacy ratio path (use_hashrate_vardiff=false). config_.use_hashrate_vardiff = true; + // ── Zombie-session leak fix (mining-hotel): OPT DASH IN to the live-session + // hygiene knobs (default-off in StratumConfig so other coins are unchanged). + // A NAT-dropped rig's TCP session is never FIN/RST'd, so socket_.is_open() + // stays true forever and the session is never reaped -- every failed retry + // minted an immortal subscribed session drawing full per-notify job builds + // (66 sockets for ~23 rigs). Transport/liveness only; consensus-neutral. + config_.tcp_keepalive_enabled = true; // kernel probes dead NAT paths (root fix) + config_.tcp_keepalive_idle_sec = 60; + config_.tcp_keepalive_interval_sec = 10; + config_.tcp_keepalive_count = 3; // ~90 s to detect a dead peer + config_.handshake_timeout_sec = 30; // drop never-authorize probes + // Idle reaper is a BACKSTOP to TCP keepalive (the real liveness authority), + // not the primary zombie killer. 1800 s + the keepalive-aware skip in + // start_idle_reaper() means an authorized rig on a high fixed-diff suffix + // (e.g. ADDR+4096 at modest X11 hashrate -> multi-minute share intervals) is + // NEVER reaped for idleness while its socket is keepalive-validated; only + // genuinely dead / never-authorized sessions are reclaimed. + config_.session_idle_timeout_sec = 1800; + config_.max_write_queue_depth = 256; // drop a stuck-write dead peer LOG_INFO << "[DASH-STRATUM] DASHWorkSource constructed" << " (min_diff=" << config_.min_difficulty << " max_diff=" << config_.max_difficulty @@ -266,24 +285,14 @@ std::shared_ptr DASHWorkSource::peek_template() const // IWorkSource: work generation -- Stage 4c (the template trio). // ───────────────────────────────────────────────────────────────────────────── -std::shared_ptr DASHWorkSource::cached_work() const +void DASHWorkSource::resource_template_now() const { - const uint64_t gen = work_generation_.load(std::memory_order_relaxed); - const auto now = std::chrono::steady_clock::now(); - { - std::lock_guard lk(template_mutex_); - if (template_cache_ && template_cache_gen_ == gen - && now - template_cache_at_ < kStaleAfter) - return template_cache_; - // Negative cache: a recent failed sourcing attempt -> don't re-poll yet. - if (!template_cache_ && template_last_fail_at_.time_since_epoch().count() != 0 - && now - template_last_fail_at_ < kRetryAfter) - return nullptr; - } - - // Re-source OUTSIDE the lock (the fallback arm is a blocking dashd RPC). - // Embedded arm when the node-held bundle is populated, retained dashd GBT - // fallback otherwise (the never-removed [GBT-XCHECK] safety path). + // The blocking select_work()/dashd-GBT re-source + cache update. Runs either + // inline on the caller (legacy blocking path) OR on the background rpc_pool + // thread (io-thread-decouple, via refresh_executor_). select_work() is done + // OUTSIDE the lock (the fallback arm is a blocking dashd RPC); the cache is + // then updated under template_mutex_. Byte-for-byte the SAME template the + // old inline path produced -- transport/threading change only. coin::DashWorkData work; if (coin_state_.populated() || dashd_fallback_) { try { @@ -308,6 +317,7 @@ std::shared_ptr DASHWorkSource::cached_work() const } } + const auto now = std::chrono::steady_clock::now(); std::lock_guard lk(template_mutex_); if (work.m_bits == 0 || work.m_previous_block.IsNull()) { // Set-gap (unarmed fallback / empty GBT) OR a zero-prev template (the @@ -316,7 +326,7 @@ std::shared_ptr DASHWorkSource::cached_work() const // serving a stale tip is worse than waiting. template_cache_.reset(); template_last_fail_at_ = now; - return nullptr; + return; } // Tip moved since the last snapshot? Bump work_generation_ so sessions // detect stale work between their timer firings and re-push. @@ -326,6 +336,58 @@ std::shared_ptr DASHWorkSource::cached_work() const template_cache_gen_ = work_generation_.load(std::memory_order_relaxed); template_cache_at_ = now; template_last_fail_at_ = {}; +} + +std::shared_ptr DASHWorkSource::cached_work() const +{ + const uint64_t gen = work_generation_.load(std::memory_order_relaxed); + const auto now = std::chrono::steady_clock::now(); + { + std::lock_guard lk(template_mutex_); + // FRESH cache (same generation, inside the TTL) -> serve it. The + // generation key is load-bearing: bump_work_generation() is the "the tip + // may have rotated, re-source" signal (double-fetch-race fix), so it must + // still trigger a refresh -- but io-thread-decouple moves WHERE that + // refresh runs off the stratum io thread (below). + if (template_cache_ && template_cache_gen_ == gen + && now - template_cache_at_ < kStaleAfter) + return template_cache_; + + if (refresh_executor_) { + // NON-BLOCKING scale path (the fix for 60+ sessions freezing on each + // re-source): the io thread NEVER runs the blocking select_work()/ + // dashd-GBT. Hand it to the background rpc_pool as a SINGLE-FLIGHT + // job and serve the CURRENT cache (possibly stale, or null on a + // set-gap) immediately. A minted share bumps the generation ~every + // 15-30 s; before, that blocked the io thread on a full GBT -- now it + // costs one background fetch, coalesced. A stale serve is bounded by + // the bg refresh latency AND the 3 s tip-poll invalidation (which + // RESETS the cache on a real tip change, so a rotated tip is served + // as a set-gap until the fresh template lands, never as stale work). + const bool recently_failed = + !template_cache_ + && template_last_fail_at_.time_since_epoch().count() != 0 + && now - template_last_fail_at_ < kRetryAfter; + if (!recently_failed && !template_refresh_inflight_.exchange(true)) { + refresh_executor_([this]() { + resource_template_now(); + template_refresh_inflight_.store(false); + }); + } + return template_cache_; // stale-or-null; null == honest set-gap + } + + // Legacy inline-blocking path (executor not wired: embedded/tests) -- + // BYTE-IDENTICAL to the pre-decouple behaviour. Negative cache: a recent + // failed sourcing attempt -> don't re-poll yet. + if (!template_cache_ && template_last_fail_at_.time_since_epoch().count() != 0 + && now - template_last_fail_at_ < kRetryAfter) + return nullptr; + } + + // Legacy path only: re-source inline (blocks the caller) then serve. + resource_template_now(); + std::lock_guard lk(template_mutex_); return template_cache_; } diff --git a/src/impl/dash/stratum/work_source.hpp b/src/impl/dash/stratum/work_source.hpp index eac5e4f09..62b7a80c2 100644 --- a/src/impl/dash/stratum/work_source.hpp +++ b/src/impl/dash/stratum/work_source.hpp @@ -305,6 +305,23 @@ class DASHWorkSource : public core::stratum::IWorkSource /// Bumps work_generation_ when a refresh observes a moved coin tip so /// stratum sessions re-push work on their next heartbeat. std::shared_ptr cached_work() const; + // io-thread-decouple: the blocking select_work()/dashd-GBT re-source, factored + // out so it runs either inline (legacy blocking path, no executor wired) OR + // on the background rpc_pool thread (via refresh_executor_). Updates the + // template cache under template_mutex_. + void resource_template_now() const; + +public: + // io-thread-decouple: wire a background executor (main_dash.cpp posts onto + // the dedicated rpc_pool). When set, cached_work() NEVER blocks the caller + // on a dashd GBT: it serves the (possibly stale) cache immediately and hands + // the blocking re-source to this executor as a SINGLE-FLIGHT background job. + // Set once at startup before the io loop runs. Only wired on the dashd- + // fallback arm; unset -> the legacy inline-blocking path (embedded/tests). + void set_refresh_executor(std::function)> fn) + { refresh_executor_ = std::move(fn); } + +private: // External dependencies (non-owning references) -- see Lifetime note. const coin::NodeCoinState& coin_state_; ///< embedded work arm (populated -> Embedded) std::function dashd_fallback_; ///< always-reachable dashd GBT RPC arm (never removed) @@ -353,6 +370,13 @@ class DASHWorkSource : public core::stratum::IWorkSource mutable uint64_t template_cache_gen_{0}; mutable std::chrono::steady_clock::time_point template_cache_at_{}; mutable std::chrono::steady_clock::time_point template_last_fail_at_{}; + + // io-thread-decouple: background single-flight template refresh. + // refresh_executor_ posts the blocking re-source onto the rpc_pool thread + // (set_refresh_executor); template_refresh_inflight_ collapses concurrent + // refreshes to one. Empty executor -> legacy inline-blocking cached_work(). + std::function)> refresh_executor_; + mutable std::atomic template_refresh_inflight_{false}; }; } // namespace dash::stratum diff --git a/test/test_dash_stratum_work_source.cpp b/test/test_dash_stratum_work_source.cpp index 7f4f36b2c..b30906fb2 100644 --- a/test/test_dash_stratum_work_source.cpp +++ b/test/test_dash_stratum_work_source.cpp @@ -32,10 +32,14 @@ #include +#include #include +#include #include +#include #include #include +#include #include namespace { @@ -791,6 +795,87 @@ TEST(DashStratumWorkSource, FallbackArmTipChangeRefreshesTemplateAndBumpsGenerat EXPECT_EQ(tmpl_after.value("height", 0u), 424243u); } +// ── io-thread-decouple KAT (mining-hotel stratum-stall fix, v0.2.3.8) ──────── +// The stall fix: the stratum io_context thread must NEVER block on the dashd +// fallback getblocktemplate. cached_work() re-sources through a background +// executor (the dedicated rpc_pool in main_dash.cpp) as a SINGLE-FLIGHT job and +// serves the cached template immediately, so a generation bump (a minted share, +// ~every 15-30 s) no longer freezes 60+ sessions on a blocking GBT. Pins: +// (a) with an executor wired, a cache-miss / gen-bump does NOT call the (slow) +// fallback on the calling (io) thread -- it schedules ONE background job +// and serves the current cache immediately; +// (b) the re-source runs on the EXECUTOR's thread, not the caller's; +// (c) single-flight: concurrent misses schedule at most one refresh; +// (d) the gen-key freshness contract (double-fetch-race fix) is preserved -- +// a bump still triggers a refresh, only OFF the io thread. +TEST(DashStratumWorkSource, IoThreadDecoupleServesCachedAndRefreshesOffThread) +{ + dash::coin::NodeCoinState cs; // unpopulated -> dashd-fallback arm + std::atomic fallback_calls{0}; + std::atomic fallback_on_caller{false}; + const auto caller_tid = std::this_thread::get_id(); + + auto ws = std::make_unique( + cs, + [&]() -> dash::coin::DashWorkData { + fallback_calls.fetch_add(1); + if (std::this_thread::get_id() == caller_tid) + fallback_on_caller.store(true); + return rich_work(); + }); + + // Controllable executor: capture jobs so the test drives WHEN the background + // re-source runs. Single-flight is enforced inside the work source. + std::mutex jm; + std::vector> jobs; + ws->set_refresh_executor([&](std::function job) { + std::lock_guard l(jm); + jobs.push_back(std::move(job)); + }); + + // (1) First request: cache empty. The io thread does NOT call the fallback -- + // it schedules ONE background job and serves an empty set-gap template. + auto t0 = ws->get_current_work_template(); + EXPECT_TRUE(t0.empty()); + EXPECT_EQ(fallback_calls.load(), 0); // io thread never blocked on dashd + { std::lock_guard l(jm); ASSERT_EQ(jobs.size(), 1u); } + + // (1b) Single-flight: another request while a refresh is in flight schedules + // no additional job. + (void)ws->get_current_work_template(); + { std::lock_guard l(jm); EXPECT_EQ(jobs.size(), 1u); } + + // (2) Run the background job on a DIFFERENT thread (the rpc_pool stand-in). + // The blocking fallback runs THERE; the cache is populated. + std::function job0; + { std::lock_guard l(jm); job0 = jobs[0]; jobs.clear(); } + std::thread(job0).join(); + EXPECT_EQ(fallback_calls.load(), 1); + EXPECT_FALSE(fallback_on_caller.load()); // re-source ran OFF the caller thread + + // (3) Now cached + fresh: served immediately, no new fallback call, no job. + auto t1 = ws->get_current_work_template(); + EXPECT_FALSE(t1.empty()); + EXPECT_EQ(t1.value("previousblockhash", ""), std::string(kPrevHashHex)); + EXPECT_EQ(fallback_calls.load(), 1); + { std::lock_guard l(jm); EXPECT_TRUE(jobs.empty()); } + + // (4) A generation bump (a minted best-share) breaks freshness, but the io + // thread STILL does not block: it serves the cached template and + // schedules a single background refresh. + ws->bump_work_generation(); + auto t2 = ws->get_current_work_template(); + EXPECT_FALSE(t2.empty()); // served cached template -- no stall + EXPECT_EQ(fallback_calls.load(), 1); // fallback NOT called on the io thread + { std::lock_guard l(jm); ASSERT_EQ(jobs.size(), 1u); } + + // Draining the scheduled refresh runs the fallback off-thread again. + std::function job1; + { std::lock_guard l(jm); job1 = jobs[0]; jobs.clear(); } + std::thread(job1).join(); + EXPECT_EQ(fallback_calls.load(), 2); +} + // Fix 3 pin (work-source side of the zero-hash pre-auth job_0 defect): a // template with a zeroed prev is NOT mineable work — honest set-gap, never a // zero-prev job.