diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 944ca371e..aff400c59 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -110,7 +110,7 @@ jobs: cmake --build build_ci \ --target test_hardening test_share_messages test_coin_broadcaster \ test_redistribute_address test_redistribute test_auto_ratchet \ - test_stratum_extensions \ + test_stratum_extensions test_stratum_hotel_interim \ core_test sharechain_test share_test btc_share_test \ test_threading test_weights \ test_header_chain test_mempool test_template_builder \ @@ -267,7 +267,7 @@ jobs: boost_sentinel \ test_hardening test_share_messages \ test_redistribute_address test_redistribute test_auto_ratchet \ - test_stratum_extensions \ + test_stratum_extensions test_stratum_hotel_interim \ core_test sharechain_test share_test btc_share_test \ test_threading test_weights \ test_header_chain test_mempool test_template_builder \ diff --git a/src/c2pool/main_dash.cpp b/src/c2pool/main_dash.cpp index b26980396..49b5f8939 100644 --- a/src/c2pool/main_dash.cpp +++ b/src/c2pool/main_dash.cpp @@ -46,6 +46,7 @@ #include #include +#include // raise_nofile_limit (hotel interim fix #4) #include #include // NetService (dashd RPC endpoint) @@ -539,6 +540,18 @@ int run_mine_block(bool testnet, const std::string& rpc_endpoint, int main(int argc, char** argv) { + // Mining-hotel interim fix #4: raise RLIMIT_NOFILE to 65536 at startup + // (one fd per stratum/miner session + RPC + sharechain P2P; distro-default + // 1024 starves the accept loop). Report the effective soft limit. + { + const uint64_t nofile = core::raise_nofile_limit(65536); + if (nofile == 0) + std::cout << "[init] RLIMIT_NOFILE: unsupported on this platform (or query failed)\n"; + else + std::cout << "[init] RLIMIT_NOFILE soft limit: " << nofile + << (nofile < 65536 ? " (< 65536; hard limit too low)" : "") << "\n"; + } + bool want_help = false; bool want_run = false; bool want_mine = false; diff --git a/src/c2pool/main_ltc.cpp b/src/c2pool/main_ltc.cpp index 4e8654e30..9a3f7b8b8 100644 --- a/src/c2pool/main_ltc.cpp +++ b/src/c2pool/main_ltc.cpp @@ -22,6 +22,7 @@ // Core includes #include #include +#include #include #include #include @@ -427,7 +428,8 @@ void print_help() { std::cout << " --stratum-max-diff N Maximum per-connection difficulty (default: 65536)\n"; std::cout << " --stratum-target-time N Target seconds per pseudoshare (default: 3)\n"; std::cout << " --no-vardiff Disable automatic difficulty adjustment\n"; - std::cout << " --max-coinbase-outputs N Max coinbase outputs per block (default: 4000, matches p2pool)\n\n"; + std::cout << " --max-coinbase-outputs N Max coinbase outputs per block (default: 4000, matches p2pool)\n"; + std::cout << " --max-stratum-connections N Max concurrent miners on this node (default: 100, 0 = unlimited)\n\n"; std::cout << "EMBEDDED NODE OPTIONS:\n"; std::cout << " --embedded-ltc Use embedded LTC SPV node (no daemon needed)\n"; @@ -536,7 +538,22 @@ int main(int argc, char* argv[]) { // Initialize logging core::log::Logger::init(); - + + // Mining-hotel interim fix #4: raise RLIMIT_NOFILE to 65536 at startup. + // One fd per stratum session + HTTP + RPC + P2P + LevelDB — distro-default + // 1024 starves the accept loop under miner churn. Log the effective limit. + { + const uint64_t nofile = core::raise_nofile_limit(65536); + if (nofile == 0) + LOG_WARNING << "RLIMIT_NOFILE: unsupported on this platform (or query failed)"; + else if (nofile < 65536) + LOG_WARNING << "RLIMIT_NOFILE soft limit is " << nofile + << " (< 65536; hard limit too low — raise with ulimit -Hn / limits.conf)"; + else + LOG_INFO << "RLIMIT_NOFILE soft limit: " << nofile; + } + + // Banner intentionally deferred — only the bordered log-file banner is // emitted (see post-parse section below). Printing an unbordered copy // here too cluttered journalctl output with duplicate headers. @@ -1042,6 +1059,10 @@ int main(int argc, char* argv[]) { stratum_config.max_coinbase_outputs = static_cast(std::stoul(argv[++i])); cli_explicit.insert("max_coinbase_outputs"); } + else if (arg == "--max-stratum-connections" && i + 1 < argc) { + stratum_config.max_stratum_connections = static_cast(std::stoul(argv[++i])); + cli_explicit.insert("max_stratum_connections"); + } // Operational tuning else if (arg == "--log-file" && i + 1 < argc) { log_file = argv[++i]; @@ -1252,6 +1273,8 @@ int main(int argc, char* argv[]) { stratum_config.vardiff_enabled = cfg["vardiff_enabled"].as(); if (!cli_explicit.count("max_coinbase_outputs") && cfg["max_coinbase_outputs"]) stratum_config.max_coinbase_outputs = cfg["max_coinbase_outputs"].as(); + if (!cli_explicit.count("max_stratum_connections") && cfg["max_stratum_connections"]) + stratum_config.max_stratum_connections = cfg["max_stratum_connections"].as(); // Operational tuning if (!cli_explicit.count("log_file") && cfg["log_file"]) @@ -1890,7 +1913,8 @@ int main(int argc, char* argv[]) { << " max_diff=" << stratum_config.max_difficulty << " target_time=" << stratum_config.target_time << "s" << " vardiff=" << (stratum_config.vardiff_enabled ? "on" : "off") - << " max_cb_outputs=" << stratum_config.max_coinbase_outputs; + << " max_cb_outputs=" << stratum_config.max_coinbase_outputs + << " max_stratum_connections=" << stratum_config.max_stratum_connections; LOG_INFO << "Operational config: p2p_max_peers=" << p2p_max_peers << " ban_duration=" << p2p_ban_duration << "s" << " rss_limit=" << rss_limit_mb << "MB" diff --git a/src/core/core_util.cpp b/src/core/core_util.cpp index 704aa4fb0..0cf3bba32 100644 --- a/src/core/core_util.cpp +++ b/src/core/core_util.cpp @@ -2,6 +2,10 @@ #include +#ifndef _WIN32 +#include +#endif + namespace core { @@ -10,4 +14,36 @@ uint32_t timestamp() return std::time(nullptr); } +uint64_t raise_nofile_limit(uint64_t target) +{ +#ifndef _WIN32 + struct rlimit rl{}; + if (getrlimit(RLIMIT_NOFILE, &rl) != 0) + return 0; + + if (rl.rlim_cur != RLIM_INFINITY && static_cast(rl.rlim_cur) < target) { + struct rlimit want = rl; + // Clamp to the hard limit — unprivileged processes cannot exceed it. + if (rl.rlim_max == RLIM_INFINITY + || static_cast(rl.rlim_max) >= target) + want.rlim_cur = static_cast(target); + else + want.rlim_cur = rl.rlim_max; + // Best effort: on failure the original soft limit stays in effect. + (void)setrlimit(RLIMIT_NOFILE, &want); + } + + if (getrlimit(RLIMIT_NOFILE, &rl) != 0) + return 0; + if (rl.rlim_cur == RLIM_INFINITY) + return UINT64_MAX; + return static_cast(rl.rlim_cur); +#else + // Windows: no setrlimit; socket handles are not fd-table bound the same + // way. Report "unsupported" and let the caller log accordingly. + (void)target; + return 0; +#endif +} + } // namespace core diff --git a/src/core/core_util.hpp b/src/core/core_util.hpp index afa7d284e..e55ca5eef 100644 --- a/src/core/core_util.hpp +++ b/src/core/core_util.hpp @@ -13,4 +13,14 @@ namespace core // core <-> {pool, ltc, c2pool} static-link cycle (V36 SCC gate). uint32_t timestamp(); +// Raise the process RLIMIT_NOFILE soft limit to `target` (default 65536), +// clamped to the hard limit. Mining-hotel interim fix #4: a capped node +// still holds one fd per stratum session + HTTP + RPC + P2P + LevelDB — +// distro defaults (1024) starve the accept loop under miner churn. +// +// Returns the soft limit actually in effect after the call (even if raising +// failed), or 0 where unsupported (non-POSIX). Deliberately log-free (this +// leaf is dependency-free) — callers log the returned value at startup. +uint64_t raise_nofile_limit(uint64_t target = 65536); + } // namespace core diff --git a/src/core/stratum_server.cpp b/src/core/stratum_server.cpp index f72da1a7e..f72d750bf 100644 --- a/src/core/stratum_server.cpp +++ b/src/core/stratum_server.cpp @@ -153,10 +153,46 @@ void StratumServer::accept_connections() void StratumServer::handle_accept(boost::system::error_code ec, tcp::socket socket) { if (!ec) { - // Create, register, and start Stratum session - auto session = std::make_shared(std::move(socket), mining_interface_, this); - register_session(session); - session->start(); + // ── STRICT per-node miner cap (hotel interim fix #5) ── + // Admission control BEFORE register_session: if the node already holds + // max_stratum_connections live sessions, close the new socket cleanly, + // WARN, bump the refused counter, and keep accepting. 0 = unlimited. + // Dead sessions are normally pruned lazily in notify_all(); prune here + // too when at cap so lingering closed sockets never starve real miners. + const size_t cap = mining_interface_ + ? mining_interface_->get_stratum_config().max_stratum_connections + : 0; + size_t live = 0; + if (cap > 0) { + std::lock_guard lock(sessions_mutex_); + if (sessions_.size() >= cap) { + for (auto it = sessions_.begin(); it != sessions_.end(); ) { + if (!(*it)->is_connected()) + it = sessions_.erase(it); + else + ++it; + } + } + live = sessions_.size(); + } + if (cap > 0 && live >= cap) { + refused_connections_.fetch_add(1); + boost::system::error_code ep_ec; + const auto ep = socket.remote_endpoint(ep_ec); + LOG_WARNING << "Stratum connection refused: node at miner cap " + << live << "/" << cap << " (max_stratum_connections)" + << (ep_ec ? std::string{} : " peer=" + ep.address().to_string() + + ":" + std::to_string(ep.port())) + << " refused_total=" << refused_connections_.load(); + boost::system::error_code ignore; + socket.shutdown(tcp::socket::shutdown_both, ignore); + socket.close(ignore); + } else { + // Create, register, and start Stratum session + auto session = std::make_shared(std::move(socket), mining_interface_, this); + register_session(session); + session->start(); + } } else { LOG_ERROR << "Stratum accept error: " << ec.message(); } @@ -179,6 +215,29 @@ void StratumServer::unregister_session(std::shared_ptr s) sessions_.erase(s); } +size_t StratumServer::get_session_count() const +{ + std::lock_guard lock(sessions_mutex_); + return sessions_.size(); +} + +std::pair StratumServer::get_job_payload_stats() const +{ + // Snapshot sessions under the mutex, count outside it (session job maps + // are io-thread state — see header comment: call quiesced). + std::vector> snapshot; + { + std::lock_guard lock(sessions_mutex_); + snapshot.assign(sessions_.begin(), sessions_.end()); + } + size_t distinct = 0, total = 0; + for (const auto& s : snapshot) { + distinct += s->distinct_job_payloads(); + total += s->active_job_count(); + } + return {distinct, total}; +} + double StratumServer::get_total_hashrate() const { // Sum all hashrate from the addr rate monitor (consistent with get_local_addr_rates) @@ -947,6 +1006,9 @@ nlohmann::json StratumSession::handle_submit(const nlohmann::json& params, const // CRITICAL: copy the job snapshot — do NOT hold a reference. // handle_submit() may trigger notify_all() → send_notify_work() → new job // added to active_jobs_ → map rebalance → reference invalidated → SIGSEGV. + // (The heavyweight template data is behind job.payload, a shared_ptr, so + // this copy is cheap and the payload stays alive even if the entry is + // evicted mid-processing.) const auto job = job_it->second; // ASICBoost: apply version rolling bits to the job version @@ -993,9 +1055,9 @@ nlohmann::json StratumSession::handle_submit(const nlohmann::json& params, const // every BTC stratum submission gets a garbage scrypt-based diff and // rejects at the vardiff gate. double share_difficulty = mining_interface_->compute_share_difficulty( - job.coinb1, job.coinb2, + job.payload->coinb1, job.payload->coinb2, extranonce1_, extranonce2, ntime, nonce, - effective_version, job.gbt_prevhash, job.nbits, job.merkle_branches); + effective_version, job.gbt_prevhash, job.nbits, job.payload->merkle_branches); // VARDIFF: acceptance threshold is the per-connection adaptive difficulty. // p2pool accepts ALL pseudoshares meeting effective_target (VARDIFF level), // not just those meeting pool share_bits. This gives smooth hashrate data. @@ -1027,13 +1089,13 @@ nlohmann::json StratumSession::handle_submit(const nlohmann::json& params, const } MiningInterface::JobSnapshot snapshot; - snapshot.coinb1 = job.coinb1; - snapshot.coinb2 = job.coinb2; + snapshot.coinb1 = job.payload->coinb1; + snapshot.coinb2 = job.payload->coinb2; snapshot.gbt_prevhash = job.gbt_prevhash; snapshot.nbits = job.nbits; // share target bits (for header construction) snapshot.block_nbits = job.gbt_block_nbits; // original GBT block bits (for block target check) snapshot.version = effective_version; // use rolled version for block construction - snapshot.merkle_branches = job.merkle_branches; + snapshot.merkle_branches = job.payload->merkle_branches; snapshot.tx_data = job.tx_data; snapshot.mweb = job.mweb; snapshot.segwit_active = job.segwit_active; @@ -1051,9 +1113,9 @@ nlohmann::json StratumSession::handle_submit(const nlohmann::json& params, const snapshot.frozen_ref.bits = job.frozen_bits; snapshot.frozen_ref.timestamp = job.frozen_timestamp; snapshot.frozen_ref.merged_payout_hash = job.frozen_merged_payout_hash; - snapshot.frozen_ref.frozen_merkle_branches = job.frozen_merkle_branches; + snapshot.frozen_ref.frozen_merkle_branches = job.payload->frozen_merkle_branches; snapshot.frozen_ref.frozen_witness_root = job.frozen_witness_root; - snapshot.frozen_ref.frozen_merged_coinbase_info = job.frozen_merged_coinbase_info; + snapshot.frozen_ref.frozen_merged_coinbase_info = job.payload->frozen_merged_coinbase_info; snapshot.frozen_ref.share_version = job.share_version; snapshot.frozen_ref.desired_version = job.desired_version; // NOTE: stale_info is NOT propagated here. It must match what ref_hash_fn @@ -1399,9 +1461,61 @@ void StratumSession::send_notify_work(bool force_clean, const uint256* frozen_be bool clean_jobs = force_clean || (prevhash != last_prevhash_); last_prevhash_ = prevhash; - // Track this job — evict oldest if at capacity (keep MAX_ACTIVE_JOBS for late shares) - while (active_jobs_.size() >= MAX_ACTIVE_JOBS) { - active_jobs_.erase(active_jobs_.begin()); + // ── Shared per-generation payload (hotel interim fix #2) ── + // coinb1/coinb2 (coinb2 carries the PPLNS outputs) + the merkle branch + // vectors are the dominant per-job allocations. Within one work generation + // a session rebuilds byte-identical copies on every notify (VARDIFF pushes, + // safety timer, best-share-unchanged refreshes) — fold them into ONE + // refcounted block shared by all jobs of that generation. Reuse is gated + // on a full byte-compare, so any change (new template, new tip, rolled + // extranonce, changed payout set) allocates a fresh payload: the bytes a + // miner receives are IDENTICAL to the pre-fix per-job copies. + const uint64_t work_generation = mining_interface_->get_work_generation(); + std::shared_ptr payload; + if (payload_cache_ && payload_cache_generation_ == work_generation + && payload_cache_->coinb1 == coinb1 + && payload_cache_->coinb2 == coinb2 + && payload_cache_->merkle_branches == merkle_branches_vec + && payload_cache_->frozen_merkle_branches == cbr.snapshot.frozen_ref.frozen_merkle_branches + && payload_cache_->frozen_merged_coinbase_info == cbr.snapshot.frozen_ref.frozen_merged_coinbase_info) { + payload = payload_cache_; // byte-identical → share the block + } else { + auto p = std::make_shared(); + p->coinb1 = coinb1; + p->coinb2 = coinb2; + p->merkle_branches = std::move(merkle_branches_vec); + p->frozen_merkle_branches = cbr.snapshot.frozen_ref.frozen_merkle_branches; + p->frozen_merged_coinbase_info = cbr.snapshot.frozen_ref.frozen_merged_coinbase_info; + payload = std::move(p); + payload_cache_ = payload; + payload_cache_generation_ = work_generation; + } + + // ── Job tracking: FIFO + 300 s TTL eviction (hotel interim fix #1) ── + // active_jobs_ is an unordered_map: erase(begin()) removed an ARBITRARY + // entry — possibly the job the miner is hashing right now — producing + // nondeterministic stale-rejects under notify storms. job_order_ records + // insertion order so we evict the genuinely-OLDEST job first, and a TTL + // sweep expires entries older than JOB_TTL (300 s), mirroring p2pool's + // ExpiringDict semantics. The most recently issued job (deque back) is + // never evicted — it is the job the miner is currently on. + const auto now_steady = std::chrono::steady_clock::now(); + // TTL sweep (oldest-first; stop at the first young entry). + while (job_order_.size() > 1) { + auto oldest = active_jobs_.find(job_order_.front()); + if (oldest == active_jobs_.end()) { // defensive: desynced id + job_order_.pop_front(); + continue; + } + if (now_steady - oldest->second.created_at < JOB_TTL) + break; + active_jobs_.erase(oldest); + job_order_.pop_front(); + } + // FIFO capacity eviction (keep MAX_ACTIVE_JOBS for late shares). + while (active_jobs_.size() >= MAX_ACTIVE_JOBS && job_order_.size() > 1) { + active_jobs_.erase(job_order_.front()); + job_order_.pop_front(); } { JobEntry je; @@ -1409,12 +1523,12 @@ void StratumSession::send_notify_work(bool force_clean, const uint256* frozen_be je.gbt_prevhash = gbt_prevhash; je.nbits = nbits; je.ntime = curtime; - je.coinb1 = coinb1; - je.coinb2 = coinb2; + je.payload = payload; je.version = version_u32; - je.merkle_branches = std::move(merkle_branches_vec); je.gbt_block_nbits = gbt_block_nbits; + je.created_at = now_steady; active_jobs_[job_id] = std::move(je); + job_order_.push_back(job_id); } // Store the SAME frozen prev_share_hash that was used for ref_hash computation @@ -1441,9 +1555,9 @@ void StratumSession::send_notify_work(bool force_clean, const uint256* frozen_be je.frozen_bits = cbr.snapshot.frozen_ref.bits; je.frozen_timestamp = cbr.snapshot.frozen_ref.timestamp; je.frozen_merged_payout_hash = cbr.snapshot.frozen_ref.merged_payout_hash; - je.frozen_merkle_branches = cbr.snapshot.frozen_ref.frozen_merkle_branches; + // frozen_merkle_branches + frozen_merged_coinbase_info live in the + // shared payload (je.payload) — see fix #2 above. je.frozen_witness_root = cbr.snapshot.frozen_ref.frozen_witness_root; - je.frozen_merged_coinbase_info = cbr.snapshot.frozen_ref.frozen_merged_coinbase_info; je.share_version = cbr.snapshot.frozen_ref.share_version; je.desired_version = cbr.snapshot.frozen_ref.desired_version; je.has_frozen = true; @@ -1506,6 +1620,20 @@ void StratumSession::schedule_work_push_timer() }); } +size_t StratumSession::distinct_job_payloads() const +{ + // Pointer-identity census of the shared per-generation payloads (fix #2). + // Bounded-memory evidence for the flat-RSS KAT: with N active jobs of one + // work generation this returns 1, not N. + std::set ptrs; + for (const auto& [id, je] : active_jobs_) { + (void)id; + if (je.payload) + ptrs.insert(static_cast(je.payload.get())); + } + return ptrs.size(); +} + void StratumSession::cancel_timers() { // Cancel pending timer — callback fires with ec=operation_aborted → returns. diff --git a/src/core/stratum_server.hpp b/src/core/stratum_server.hpp index 7d6e68993..979b25b90 100644 --- a/src/core/stratum_server.hpp +++ b/src/core/stratum_server.hpp @@ -113,14 +113,26 @@ class StratumSession : public std::enable_shared_from_this // Active jobs for stale detection (job_id → prevhash at time of issue) struct JobEntry { + // Heavyweight per-generation template data, refcounted and SHARED + // across all jobs a session issues for one work generation (hotel + // interim fix #2 — de-dup of the per-job memory bomb). coinb2 carries + // the PPLNS payout outputs, the dominant per-job allocation. Contents + // are immutable after construction; a new payload is only allocated + // when any byte differs from the previous job's payload (byte-identical + // reuse ⇒ provably no wire change). + struct SharedJobPayload { + std::string coinb1; + std::string coinb2; // includes PPLNS outputs + std::vector merkle_branches; + std::vector frozen_merkle_branches; // segwit txid_merkle_link at template time + std::vector frozen_merged_coinbase_info; // pre-serialized + }; std::string prevhash; // Stratum-format (swapped) for stale detection std::string gbt_prevhash; // Raw GBT previousblockhash (BE display hex) for header reconstruction std::string nbits; // share target bits (used in header the miner hashes) uint32_t ntime{}; - std::string coinb1; - std::string coinb2; + std::shared_ptr payload; // coinb1/coinb2/branches (shared per work generation) uint32_t version{}; - std::vector merkle_branches; std::shared_ptr> tx_data; // raw tx hex from GBT template (a1: shared/lazy) std::string mweb; // MWEB extension data bool segwit_active{false}; @@ -137,18 +149,33 @@ class StratumSession : public std::enable_shared_from_this uint32_t frozen_bits{0}; uint32_t frozen_timestamp{0}; uint256 frozen_merged_payout_hash; - std::vector frozen_merkle_branches; // segwit txid_merkle_link at template time uint256 frozen_witness_root; // wtxid_merkle_root at template time - std::vector frozen_merged_coinbase_info; // pre-serialized bool has_frozen{false}; int64_t share_version{36}; // AutoRatchet-determined share version at template time uint64_t desired_version{36}; // Version vote int stale_info{0}; // 0=none, 253=orphan (block template changed), 254=doa + // Creation stamp for TTL eviction (steady clock — immune to wall-clock + // jumps; the wire ntime field above is unchanged and stays wall-clock). + std::chrono::steady_clock::time_point created_at{}; }; std::unordered_map active_jobs_; + // Insertion-order companion to active_jobs_ (hotel interim fix #1). + // unordered_map::begin() is ARBITRARY, so the old + // `erase(active_jobs_.begin())` capacity eviction could drop the job the + // miner is CURRENTLY hashing — the nondeterministic stale-reject bug. + // This deque records job issue order so eviction is genuinely oldest-first + // (FIFO), and a 300 s TTL sweep on insert mirrors p2pool's ExpiringDict. + // The most recently issued job (back of the deque) is never evicted. + std::deque job_order_; + // Per-generation payload cache: reuse the previous job's SharedJobPayload + // when the freshly built bytes are identical (same work generation, same + // coinbase, same branches). Byte-compare gate ⇒ wire-identical by design. + std::shared_ptr payload_cache_; + uint64_t payload_cache_generation_ = 0; std::string last_prevhash_; // track prevhash for clean_jobs detection uint64_t last_pushed_generation_ = 0; // work generation at last push (for safety timer) static constexpr size_t MAX_ACTIVE_JOBS = 256; // p2pool keeps all jobs from current block + static constexpr std::chrono::seconds JOB_TTL{300}; // p2pool ExpiringDict expiry for jobs // Per-worker statistics uint64_t accepted_shares_ = 0; @@ -184,6 +211,10 @@ class StratumSession : public std::enable_shared_from_this bool is_connected() const { return socket_.is_open(); } bool is_subscribed() const { return subscribed_; } + // 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(); } + size_t distinct_job_payloads() const; // distinct SharedJobPayload blocks retained double get_hashrate() const { return hashrate_tracker_.get_current_hashrate(); } const std::array& get_pubkey_hash() const { return pubkey_hash_; } const std::map& get_merged_addresses() const { return merged_addresses_; } @@ -241,6 +272,10 @@ class StratumServer mutable std::mutex sessions_mutex_; std::set> sessions_; + // STRICT per-node miner cap (hotel interim fix #5): connections refused + // because sessions_.size() >= StratumConfig::max_stratum_connections. + std::atomic refused_connections_{0}; + // p2pool RateMonitor pair (work.py:223-226): // local_rate_monitor: per-user hashrate (for get_local_rates) // local_addr_rate_monitor: per-address hashrate (for get_local_addr_rates) @@ -297,6 +332,16 @@ class StratumServer void register_session(std::shared_ptr s); void unregister_session(std::shared_ptr s); + /// Connections refused by the strict per-node miner cap (fix #5). + uint64_t get_refused_connections() const { return refused_connections_.load(); } + /// Live session count (dead sessions pruned lazily — see notify_all()). + size_t get_session_count() const; + /// Diagnostics: {distinct SharedJobPayload blocks, total active jobs} + /// summed across all sessions. Bounded-memory evidence: distinct payloads + /// track retained work GENERATIONS, not sessions × jobs. Test/diag hook — + /// call quiesced (sessions mutate their job maps on the io thread). + std::pair get_job_payload_stats() const; + std::string get_bind_address() const { return bind_address_; } uint16_t get_port() const { return port_; } bool is_running() const { return running_; } diff --git a/src/core/stratum_types.hpp b/src/core/stratum_types.hpp index 987d1be06..07d3d4edc 100644 --- a/src/core/stratum_types.hpp +++ b/src/core/stratum_types.hpp @@ -44,6 +44,14 @@ struct StratumConfig { // 2^16 (65536) for scrypt nets (LTC/DOGE), 1 for SHA256d nets (bitcoin). // Default preserves the scrypt convention; SHA256d work sources override to 1.0. double set_difficulty_multiplier = 65536.0; + // STRICT per-node miner cap (mining-hotel interim fix): maximum number of + // concurrent stratum TCP sessions this node accepts. When the cap is hit + // the excess socket is closed cleanly, a WARN is logged, the + // refused_connections counter increments, and the accept loop keeps + // running. 0 = unlimited (legacy behavior). Wire-through: + // max_stratum_connections (YAML) / --max-stratum-connections (CLI). + // Admission control only — zero wire-byte change for admitted sessions. + size_t max_stratum_connections = 100; }; /// Frozen share-construction fields returned by ref_hash_fn. These diff --git a/src/core/web_server.cpp b/src/core/web_server.cpp index 434dacead..69b4b1030 100644 --- a/src/core/web_server.cpp +++ b/src/core/web_server.cpp @@ -8488,7 +8488,7 @@ void WebServer::trigger_work_refresh() stratum_server_->notify_all(); } -void WebServer::trigger_work_refresh_debounced() +void WebServer::execute_debounced_work_refresh() { // Capture previous tip hash BEFORE refresh updates it — needed for delta precompute. // Absent tip (fresh bootstrap, no shares yet) → empty prev_tip_hash, precompute_delta @@ -8501,7 +8501,6 @@ void WebServer::trigger_work_refresh_debounced() if (auto* mm = mining_interface_->get_mm_manager()) mm->mark_pplns_dirty(); - // No debounce — call refresh immediately. trigger_work_refresh(); // Invalidate sharechain window cache (new share changed the tip) @@ -8519,6 +8518,61 @@ void WebServer::trigger_work_refresh_debounced() if (mining_interface_->sse_subscriber_count() > 0) { mining_interface_->sse_push(to_json(mining_interface_->get_sharechain_tip()).dump()); } + + // Record execution for the debounce window + trailing-edge event gate. + m_last_work_refresh = std::chrono::steady_clock::now(); + auto refreshed_tip = mining_interface_->get_sharechain_tip(); + m_last_refresh_tip_hash = refreshed_tip ? refreshed_tip->hash : std::string{}; +} + +void WebServer::trigger_work_refresh_debounced() +{ + // ── Notify debounce (hotel interim fix #3) ── + // Share-arrival storms used to fan out into one full refresh_work() + + // notify_all() per share (the old body here was a no-op stub that called + // refresh immediately). Semantics now: + // * Leading edge: a call outside the debounce window refreshes + // IMMEDIATELY (first share after quiet keeps p2pool-grade latency). + // * Trailing coalesce: calls inside the window collapse into ONE + // deferred refresh ~300 ms after the leading edge, event-gated on a + // REAL work change (sharechain tip differs from the last executed + // refresh) so a redundant burst costs nothing. + // The new-block path is untouched: trigger_work_refresh() stays immediate. + // Timing/coalescing only — the bytes of any notify that IS sent are + // unchanged. Single-threaded: callers + timer run on the main ioc_. + using namespace std::chrono; + static constexpr auto DEBOUNCE_WINDOW = milliseconds(300); + + const auto now = steady_clock::now(); + const bool in_window = (m_last_work_refresh.time_since_epoch().count() != 0) + && (now - m_last_work_refresh < DEBOUNCE_WINDOW); + + if (!in_window && !m_work_refresh_pending) { + execute_debounced_work_refresh(); // leading edge — immediate + return; + } + + if (m_work_refresh_pending) + return; // trailing refresh already scheduled — this call coalesces + + m_work_refresh_pending = true; + if (!m_work_refresh_timer) + m_work_refresh_timer = std::make_shared(ioc_); + // Fire at the end of the window that opened with the leading-edge refresh. + const auto remaining = DEBOUNCE_WINDOW - duration_cast(now - m_last_work_refresh); + m_work_refresh_timer->expires_after(remaining > milliseconds(1) ? remaining : milliseconds(1)); + m_work_refresh_timer->async_wait([this](beast::error_code ec) { + m_work_refresh_pending = false; + if (ec || !running_) return; + // Event gate: only refresh on a real work change. If the sharechain + // tip is still what the last executed refresh saw, the coalesced + // calls were redundant (already covered by the leading edge). + auto tip = mining_interface_->get_sharechain_tip(); + std::string tip_hash = tip ? tip->hash : std::string{}; + if (tip_hash == m_last_refresh_tip_hash) + return; + execute_debounced_work_refresh(); + }); } void WebServer::set_coin_node(core::coin::ICoinNode* node) diff --git a/src/core/web_server.hpp b/src/core/web_server.hpp index af9e5ad62..2869cf374 100644 --- a/src/core/web_server.hpp +++ b/src/core/web_server.hpp @@ -1498,9 +1498,15 @@ class WebServer bool solo_mode_; std::string solo_address_; - // Debounce timer for trigger_work_refresh_debounced() + // Debounce state for trigger_work_refresh_debounced() (hotel interim fix + // #3). Leading-edge-immediate + ~300 ms trailing-coalesce; the trailing + // refresh is event-gated on a REAL work change (sharechain tip moved since + // the last executed refresh). All state is touched only from the main + // ioc_ thread (callers + timer handler run there) — no extra locking. std::shared_ptr m_work_refresh_timer; bool m_work_refresh_pending = false; + std::chrono::steady_clock::time_point m_last_work_refresh{}; // last executed refresh + std::string m_last_refresh_tip_hash; // sharechain tip at last executed refresh // Payout system integration c2pool::payout::PayoutManager* payout_manager_ptr_ = nullptr; @@ -1574,6 +1580,9 @@ class WebServer private: void accept_connections(); void handle_accept(beast::error_code ec, tcp::socket socket); + // Body of a debounced work refresh (capture prev tip → refresh → caches → + // SSE). Runs on the leading edge and on the trailing coalesced timer. + void execute_debounced_work_refresh(); }; // StratumSession and StratumServer — see stratum_server.hpp diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 64e58428a..ca8961c3c 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -938,6 +938,24 @@ if (BUILD_TESTING AND GTest_FOUND) target_link_libraries(test_stratum_extensions PRIVATE c2pool_payout ltc_coin c2pool_storage c2pool_merged_mining pool) # OBJECT-lib SCC direct-naming (#22/#39) gtest_add_tests(test_stratum_extensions "" AUTO) + # Mining-hotel interim fix KATs: strict per-node miner cap (refused + # connections + counter), FIFO+TTL job eviction under a 600-notify stale + # storm, and the flat-RSS shared-payload pointer-identity scaffold. + # Loopback-only (synthetic IWorkSource, ephemeral test port) — no daemon / + # RPC / sharechain. DASH byte-parity note: the notify/coinbase/share BYTE + # checks live in the dash lane against the frstrtr/p2pool-dash oracle + # (share version 16→36), not here. + add_executable(test_stratum_hotel_interim test_stratum_hotel_interim.cpp) + target_link_libraries(test_stratum_hotel_interim PRIVATE + GTest::gtest_main GTest::gtest + c2pool_hashrate + core ltc + nlohmann_json::nlohmann_json + ${Boost_LIBRARIES} + ) + target_link_libraries(test_stratum_hotel_interim PRIVATE c2pool_payout ltc_coin c2pool_storage c2pool_merged_mining pool) # OBJECT-lib SCC direct-naming (#22/#39) + gtest_add_tests(test_stratum_hotel_interim "" AUTO) + add_executable(test_multiaddress_pplns test_multiaddress_pplns.cpp) target_link_libraries(test_multiaddress_pplns PRIVATE GTest::gtest_main GTest::gtest diff --git a/test/test_stratum_hotel_interim.cpp b/test/test_stratum_hotel_interim.cpp new file mode 100644 index 000000000..63b2cac43 --- /dev/null +++ b/test/test_stratum_hotel_interim.cpp @@ -0,0 +1,495 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later +// +// Mining-hotel interim fix KATs — stratum admission cap + job eviction + +// shared-payload memory bound (core::StratumServer / core::StratumSession). +// +// Covers the three behavior changes of the "minimal stratum hotel-interim +// fix + strict per-node miner cap" slice: +// +// 1. CAP ENFORCEMENT — with max_stratum_connections=2 the 3rd (and 4th) +// TCP connection is refused cleanly (socket closed, counter increments, +// accept loop keeps running: a freed slot admits a new miner). +// +// 2. STALE-STORM / FIFO+TTL EVICTION — issue a job, then a 600-notify +// storm. Pre-fix, capacity eviction was `erase(active_jobs_.begin())` +// on an unordered_map: an ARBITRARY entry — possibly the job the miner +// is currently hashing — was dropped, producing nondeterministic +// "Stale share" rejects. Post-fix eviction is genuinely-oldest-first +// (FIFO, cap 256) + 300 s TTL (p2pool ExpiringDict semantics). The rig +// asserts the CURRENT job (and a recent one inside the FIFO window) +// submits ACCEPTED after the storm — this FAILS on the pre-fix +// arbitrary-evict with overwhelming probability (345 arbitrary +// evictions over 601 string-keyed entries) — and that the genuinely +// oldest job IS stale (cap still enforced, oldest-first). +// NOTE on the spec's "submit against N after N+600 → ACCEPTED": with +// MAX_ACTIVE_JOBS=256 (p2pool parity) a job 600 generations back is +// GENUINELY oldest and correctly evicted; the bug being fixed is +// arbitrary eviction of RECENT jobs. The assertion set here encodes +// exactly that: newest-256 always survive, oldest are the ones dropped. +// +// 3. FLAT-RSS SCAFFOLD — N sessions × job churn: total active jobs stay +// bounded by MAX_ACTIVE_JOBS per session, and the heavyweight template +// payload (coinb1/coinb2 incl. PPLNS outputs + merkle branches) is +// POINTER-IDENTICAL across all jobs of one work generation (one +// refcounted block per generation per session, not jobs × copies). +// +// WIRE SAFETY / BYTE-PARITY: everything here is admission/eviction/memory +// only — no notify param, coinbase (coinb1‖en1‖en2‖coinb2), or share byte +// changes. The notify/coinbase/share BYTE-PARITY KAT is NOT this file: for +// DASH it must run against the frstrtr/p2pool-dash oracle (share version +// 16→36 transition line), NOT the v36-uniform baseline. See the dash lane +// parity tests (test_dash_coinbase_parity / test_dash_stratum_binding) for +// the oracle-pinned byte checks; this file deliberately reuses a synthetic +// work source and never asserts wire bytes. +// +// Self-contained: synthetic IWorkSource, loopback TCP on an ephemeral test +// port, no coin daemon / RPC / sharechain. + +#include + +#include +#include + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace asio = boost::asio; +using tcp = asio::ip::tcp; +using json = nlohmann::json; +using namespace std::chrono_literals; + +namespace { + +// ── Synthetic work source ─────────────────────────────────────────────────── +// Minimal IWorkSource: constant template + constant per-connection coinbase. +// Constant bytes within one work generation are exactly what makes the +// shared-payload dedup observable (pointer identity across jobs). +class FakeWorkSource : public core::stratum::IWorkSource { +public: + core::stratum::StratumConfig cfg; + std::atomic generation{1}; + + FakeWorkSource() + { + tmpl_["previousblockhash"] = + "00000000000000000000000000000000000000000000000000000000000000aa"; + tmpl_["version"] = 0x20000000; + tmpl_["bits"] = "1e0fffff"; + tmpl_["curtime"] = 1700000000ULL; + tmpl_["height"] = 1; + tx_data_ = std::make_shared>(); + } + + const core::stratum::StratumConfig& get_stratum_config() const override { return cfg; } + std::function get_best_share_hash_fn() const override { return {}; } + std::string get_current_gbt_prevhash() const override + { + return tmpl_["previousblockhash"].get(); + } + uint64_t get_work_generation() const override { return generation.load(); } + bool has_merged_chain(uint32_t) const override { return false; } + + void register_stratum_worker(const std::string&, + const core::stratum::WorkerInfo&) override {} + void unregister_stratum_worker(const std::string&) override {} + void update_stratum_worker(const std::string&, double, double, double, + uint64_t, uint64_t, uint64_t) override {} + + nlohmann::json get_current_work_template() const override { return tmpl_; } + std::vector get_stratum_merkle_branches() const override { return {}; } + std::pair get_coinbase_parts() const override { return {"", ""}; } + + core::stratum::CoinbaseResult build_connection_coinbase( + const uint256&, const std::string&, + const std::vector&, + const std::vector>>&) const override + { + core::stratum::CoinbaseResult r; + r.coinb1 = "01000000010000000000000000000000000000000000000000000000000000" + "000000000000ffffffff"; + r.coinb2 = "0000000000f2052a010000001976a914000000000000000000000000000000" + "000000000088ac00000000"; + r.snapshot.tx_data = tx_data_; + return r; + } + + nlohmann::json mining_submit(const std::string&, const std::string&, + const std::string&, const std::string&, + const std::string&, const std::string&, + const std::string&, + const std::map>&, + const core::stratum::JobSnapshot*) override + { + json r; + r["result"] = true; + r["error"] = nullptr; + return r; + } + + uint32_t get_share_bits() const override { return 0; } + uint32_t get_share_max_bits() const override { return 0; } + + double compute_share_difficulty(const std::string&, const std::string&, + const std::string&, const std::string&, + const std::string&, const std::string&, + uint32_t, const std::string&, const std::string&, + const std::vector&) const override + { + return 1.0e12; // always above vardiff → submission accepted + } + +private: + nlohmann::json tmpl_; + std::shared_ptr> tx_data_; +}; + +// ── Server harness: real StratumServer on loopback, io thread in the test ── +// (The PRODUCT stays single-threaded; the extra thread here only drives the +// test's io_context the way each coin main's ioc.run() does.) +class ServerHarness { +public: + asio::io_context ioc; + std::shared_ptr ws = std::make_shared(); + std::unique_ptr server; + uint16_t port = 0; + + bool start() + { + // Ephemeral-ish port with retry (StratumServer::start binds a fixed port). + for (uint16_t p = 39411; p < 39511; ++p) { + auto s = std::make_unique(ioc, "127.0.0.1", p, ws); + if (s->start()) { + server = std::move(s); + port = p; + break; + } + } + if (!server) return false; + guard_.emplace(asio::make_work_guard(ioc)); + th_ = std::thread([this] { ioc.run(); }); + return true; + } + + // Run fn on the io thread and wait for it (quiesce barrier: asio handlers + // execute FIFO on the single io thread, so fn sees all prior work done). + template + auto on_io(F&& fn) -> decltype(fn()) + { + std::promise prom; + auto fut = prom.get_future(); + asio::post(ioc, [&] { prom.set_value(fn()); }); + return fut.get(); + } + + void notify_storm(int n) + { + std::promise done; + asio::post(ioc, [&] { + for (int i = 0; i < n; ++i) + server->notify_all(); + done.set_value(); + }); + done.get_future().wait(); + } + + ~ServerHarness() + { + if (server) { + std::promise done; + asio::post(ioc, [&] { server->stop(); done.set_value(); }); + done.get_future().wait(); + } + guard_.reset(); + ioc.stop(); + if (th_.joinable()) th_.join(); + } + +private: + std::optional> guard_; + std::thread th_; +}; + +// ── Minimal blocking stratum client with timeouts ─────────────────────────── +class Client { +public: + explicit Client() : sock_(io_) {} + + bool connect(uint16_t port) + { + boost::system::error_code ec; + sock_.connect(tcp::endpoint(asio::ip::make_address("127.0.0.1"), port), ec); + return !ec; + } + + void send_json(const json& j) + { + auto line = std::make_shared(j.dump() + "\n"); + boost::system::error_code ec; + asio::write(sock_, asio::buffer(*line), ec); + } + + // Read one \n-terminated line; nullopt on timeout or close. + std::optional read_line(std::chrono::milliseconds timeout = 3000ms) + { + std::optional out; + bool finished = false; + asio::async_read_until(sock_, buf_, '\n', + [&](boost::system::error_code ec, std::size_t) { + finished = true; + if (!ec) { + std::istream is(&buf_); + std::string line; + std::getline(is, line); + if (!line.empty() && line.back() == '\r') line.pop_back(); + out = line; + } else { + closed_ = true; + } + }); + io_.restart(); + io_.run_for(timeout); + if (!finished) { + boost::system::error_code ec; + sock_.cancel(ec); + io_.restart(); + io_.run(); // drain the aborted handler + closed_ = false; // timeout, not close + } + return out; + } + + bool was_closed_by_peer() const { return closed_; } + + // Drive subscribe (+ optional authorize); records notify job ids in order. + bool subscribe(int id = 1) + { + send_json({{"id", id}, {"method", "mining.subscribe"}, {"params", json::array()}}); + return wait_response(id).has_value(); + } + bool authorize(const std::string& user, int id = 2) + { + send_json({{"id", id}, {"method", "mining.authorize"}, + {"params", json::array({user, "x"})}}); + auto resp = wait_response(id); + return resp && (*resp)["result"].is_boolean() && (*resp)["result"].get(); + } + + // Read until the response with the given id arrives; notifies encountered + // along the way are recorded into jobs/last_ntime. + std::optional wait_response(int id, std::chrono::milliseconds timeout = 5000ms) + { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + auto line = read_line(500ms); + if (!line) { + if (was_closed_by_peer()) return std::nullopt; + continue; + } + json j = json::parse(*line, nullptr, false); + if (j.is_discarded()) continue; + record_notify(j); + if (j.contains("id") && !j["id"].is_null() && j["id"] == id) + return j; + } + return std::nullopt; + } + + // Drain notifies until `count` job ids collected (or timeout). + bool collect_notifies(size_t count, std::chrono::milliseconds timeout = 20000ms) + { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (jobs.size() < count && std::chrono::steady_clock::now() < deadline) { + auto line = read_line(500ms); + if (!line) { + if (was_closed_by_peer()) return false; + continue; + } + json j = json::parse(*line, nullptr, false); + if (j.is_discarded()) continue; + record_notify(j); + } + return jobs.size() >= count; + } + + // Submit against a job id; returns the response json (or nullopt). + std::optional submit(const std::string& user, const std::string& job_id, int id) + { + send_json({{"id", id}, {"method", "mining.submit"}, + {"params", json::array({user, job_id, "00000000", last_ntime, "12345678"})}}); + return wait_response(id); + } + + std::vector jobs; // job ids, in notify order + std::string last_ntime = "65a812c0"; // overwritten by record_notify + +private: + void record_notify(const json& j) + { + if (j.contains("method") && j["method"] == "mining.notify" + && j.contains("params") && j["params"].is_array() + && j["params"].size() >= 9) { + jobs.push_back(j["params"][0].get()); + last_ntime = j["params"][7].get(); + } + } + + asio::io_context io_; + tcp::socket sock_; + asio::streambuf buf_; + bool closed_ = false; +}; + +} // namespace + +// ════════════════════════════════════════════════════════════════════════════ +// KAT 1 — STRICT per-node miner cap: 3rd connection refused, counter bumps, +// accept loop survives and admits again once a slot frees. +// ════════════════════════════════════════════════════════════════════════════ +TEST(StratumHotelInterim, CapEnforcementThirdConnectionRefused) +{ + ServerHarness h; + h.ws->cfg.max_stratum_connections = 2; + ASSERT_TRUE(h.start()); + + auto c1 = std::make_unique(); + auto c2 = std::make_unique(); + ASSERT_TRUE(c1->connect(h.port)); + ASSERT_TRUE(c1->subscribe()); + ASSERT_TRUE(c2->connect(h.port)); + ASSERT_TRUE(c2->subscribe()); + EXPECT_EQ(h.on_io([&] { return h.server->get_session_count(); }), 2u); + + // 3rd connection: TCP-accepted then cleanly closed by the cap gate. + Client c3; + ASSERT_TRUE(c3.connect(h.port)); + auto line = c3.read_line(3000ms); + EXPECT_FALSE(line.has_value()); + EXPECT_TRUE(c3.was_closed_by_peer()) << "3rd connection must be closed by the cap"; + EXPECT_EQ(h.on_io([&] { return h.server->get_refused_connections(); }), 1u); + + // 4th connection also refused; counter increments again. + Client c4; + ASSERT_TRUE(c4.connect(h.port)); + (void)c4.read_line(3000ms); + EXPECT_TRUE(c4.was_closed_by_peer()); + EXPECT_EQ(h.on_io([&] { return h.server->get_refused_connections(); }), 2u); + + // Existing miners are untouched: c1 still gets work on demand. + h.notify_storm(1); + EXPECT_TRUE(c1->collect_notifies(2, 5000ms)); // subscribe notify + storm notify + + // Free a slot → the accept loop must still be alive and admit a new miner + // (dead-session prune runs at-cap inside handle_accept). + c1.reset(); + std::this_thread::sleep_for(300ms); // let the server observe the close + Client c5; + ASSERT_TRUE(c5.connect(h.port)); + EXPECT_TRUE(c5.subscribe()) << "freed slot must admit a new miner"; +} + +// ════════════════════════════════════════════════════════════════════════════ +// KAT 2 — Stale-storm: 600-notify storm, then submit. +// * current job (newest) → ACCEPTED (pre-fix arbitrary-evict rig: +// this is the assertion that fails on `erase(active_jobs_.begin())`) +// * recent job (inside FIFO window of 256) → ACCEPTED +// * genuinely-oldest pre-storm job → STALE (cap enforced, oldest-first) +// ════════════════════════════════════════════════════════════════════════════ +TEST(StratumHotelInterim, StaleStormFifoEviction) +{ + ServerHarness h; + ASSERT_TRUE(h.start()); + + Client c; + ASSERT_TRUE(c.connect(h.port)); + ASSERT_TRUE(c.subscribe()); + ASSERT_TRUE(c.authorize("hoteltestworker1")); + ASSERT_TRUE(c.collect_notifies(1, 5000ms)); + const size_t base = c.jobs.size(); // jobs issued by subscribe/authorize + + // Notify storm: 600 further jobs on this session. + h.notify_storm(600); + ASSERT_TRUE(c.collect_notifies(base + 600)) << "storm notifies not received"; + + // Session retains at most MAX_ACTIVE_JOBS (256) — FIFO cap held. + const auto stats = h.on_io([&] { return h.server->get_job_payload_stats(); }); + EXPECT_LE(stats.second, 256u); + EXPECT_GE(stats.second, 200u); // storm actually filled the window + + // (a) CURRENT job — the one the miner is hashing right now — must never + // have been evicted. This submit FAILS pre-fix (arbitrary eviction). + { + auto resp = c.submit("hoteltestworker1", c.jobs.back(), 100); + ASSERT_TRUE(resp.has_value()); + EXPECT_TRUE((*resp)["result"].is_boolean() && (*resp)["result"].get()) + << "current job rejected: " << resp->dump(); + } + // (b) A recent job inside the newest-256 window survives too. + { + const std::string& recent = c.jobs[c.jobs.size() - 100]; + auto resp = c.submit("hoteltestworker1", recent, 101); + ASSERT_TRUE(resp.has_value()); + EXPECT_TRUE((*resp)["result"].is_boolean() && (*resp)["result"].get()) + << "recent (in-window) job rejected: " << resp->dump(); + } + // (c) The genuinely-oldest pre-storm job is beyond the 256-job FIFO window + // → correctly stale (error 21). Guards against the cap silently vanishing. + { + auto resp = c.submit("hoteltestworker1", c.jobs.front(), 102); + ASSERT_TRUE(resp.has_value()); + ASSERT_TRUE(resp->contains("error") && (*resp)["error"].is_array()); + EXPECT_EQ((*resp)["error"][0].get(), 21) << resp->dump(); + } +} + +// ════════════════════════════════════════════════════════════════════════════ +// KAT 3 — Flat-RSS scaffold: N sessions × job churn keeps retained job memory +// bounded; the heavyweight template payload is POINTER-IDENTICAL across all +// jobs of one work generation (1 shared block per generation per session). +// ════════════════════════════════════════════════════════════════════════════ +TEST(StratumHotelInterim, FlatRssSharedPayloadIdentity) +{ + ServerHarness h; + ASSERT_TRUE(h.start()); + + constexpr size_t kSessions = 3; + constexpr int kChurn = 50; + + std::vector> clients; + for (size_t i = 0; i < kSessions; ++i) { + auto c = std::make_unique(); + ASSERT_TRUE(c->connect(h.port)); + ASSERT_TRUE(c->subscribe()); + clients.push_back(std::move(c)); + } + + h.notify_storm(kChurn); + for (auto& c : clients) + ASSERT_TRUE(c->collect_notifies(1 + kChurn)); + + // Quiesced read on the io thread: {distinct payload blocks, total jobs}. + const auto [distinct, total] = + h.on_io([&] { return h.server->get_job_payload_stats(); }); + + // Job churn happened… + EXPECT_GE(total, kSessions * kChurn); + // …and per-session job count is bounded by the FIFO cap. + EXPECT_LE(total, kSessions * 256u); + // Pointer identity: ONE shared payload per (session × work generation) — + // the fake source never bumps the generation and emits byte-identical + // coinbase parts, so each session must hold exactly one payload block, not + // one copy per job. This is the de-duped "memory bomb" assertion. + EXPECT_EQ(distinct, kSessions) + << "expected 1 shared payload per session (jobs=" << total << ")"; +}