diff --git a/src/impl/dgb/CMakeLists.txt b/src/impl/dgb/CMakeLists.txt index a8a92ecc2..fc4f568ed 100644 --- a/src/impl/dgb/CMakeLists.txt +++ b/src/impl/dgb/CMakeLists.txt @@ -17,8 +17,23 @@ # slice under daemon/. Top-level add_subdirectory(src/impl/dgb) registration # follows the OBJECT-lib convention PR; held ~2-3 heartbeats, not raced here. if(COIN_DGB) - message(STATUS "c2pool: DGB Scrypt-only coin module enabled (skeleton)") + message(STATUS "c2pool: DGB Scrypt-only coin module enabled") add_subdirectory(coin) # add_subdirectory(daemon) # add_subdirectory(test) + + # dgb -- DGB Scrypt pool-layer OBJECT lib. Mirrors the ltc OBJECT lib + # (src/impl/ltc/CMakeLists.txt): compiles the real NodeImpl translation unit + # (node.cpp + protocol handlers) so share_tracker.hpp / share_check.hpp are + # pulled into a compiled TU. Links the shared core / pool / sharechain base + # plus dgb_coin (embedded P2P+mempool) and c2pool_storage (LevelDB sharechain). + add_library(dgb OBJECT + config_coin.hpp config_coin.cpp config_pool.hpp config_pool.cpp + node.hpp node.cpp + peer.hpp + protocol_actual.cpp protocol_legacy.cpp + messages.hpp + share_types.hpp share.hpp + ) + target_link_libraries(dgb core pool sharechain dgb_coin btclibs c2pool_storage ${SECP256K1_LIBRARIES}) endif() diff --git a/src/impl/dgb/config_coin.cpp b/src/impl/dgb/config_coin.cpp new file mode 100644 index 000000000..47b506d59 --- /dev/null +++ b/src/impl/dgb/config_coin.cpp @@ -0,0 +1,35 @@ +#include "config_coin.hpp" + +#include + +namespace dgb +{ + +std::ofstream& CoinConfig::get_default(std::ofstream& file) +{ + YAML::Node out; + + out["symbol"] = "defaultNet"; + out["p2p"] = config::P2PData(); + out["rpc"] = config::RPCData(); + out["share_period"] = 0; + out["testnet"] = false; + + file << out; + return file; +} + +void CoinConfig::load() +{ + YAML::Node node = YAML::LoadFile(m_filepath.string()); + + PARSE_CONFIG(node, symbol, std::string); + + m_p2p = node["p2p"].as(); + m_rpc = node["rpc"].as(); + + PARSE_CONFIG(node, share_period, int); + PARSE_CONFIG(node, testnet, bool); +} + +} // namespace dgb diff --git a/src/impl/dgb/config_coin.hpp b/src/impl/dgb/config_coin.hpp index f4340b74f..5547b80e2 100644 --- a/src/impl/dgb/config_coin.hpp +++ b/src/impl/dgb/config_coin.hpp @@ -4,6 +4,9 @@ #include #include +#include +#include + #include #include @@ -25,6 +28,49 @@ namespace config }; } // config +} // namespace dgb + +namespace YAML +{ +template<> struct convert +{ + static Node encode(const dgb::config::P2PData& rhs) + { + Node node; + node["prefix"] = HexStr(rhs.prefix); + node["address"] = rhs.address; + return node; + } + + static bool decode(const Node& node, dgb::config::P2PData& rhs) + { + rhs.prefix = ParseHexBytes(node["prefix"].as()); + rhs.address = node["address"].as(); + return true; + } +}; + +template<> struct convert +{ + static Node encode(const dgb::config::RPCData& rhs) + { + Node node; + node["address"] = rhs.address; + node["userpass"] = rhs.userpass; + return node; + } + + static bool decode(const Node& node, dgb::config::RPCData& rhs) + { + rhs.address = node["address"].as(); + rhs.userpass = node["userpass"].as(); + return true; + } +}; +} + +namespace dgb +{ /// DigiByte Scrypt coin parameters. /// Source of truth: p2pool-dgb-scrypt oracle bitcoin/networks/digibyte.py diff --git a/src/impl/dgb/config_pool.cpp b/src/impl/dgb/config_pool.cpp new file mode 100644 index 000000000..728e2a632 --- /dev/null +++ b/src/impl/dgb/config_pool.cpp @@ -0,0 +1,48 @@ +#include "config_pool.hpp" + +#include +#include + +namespace dgb +{ + +std::ofstream& PoolConfig::get_default(std::ofstream& file) +{ + YAML::Node out; + + // Use the canonical LTC p2pool mainnet prefix when no prefix is set yet + out["prefix"] = m_prefix.empty() ? DEFAULT_PREFIX_HEX : HexStr(m_prefix); + out["worker"] = m_worker; + + YAML::Node addrs_node; + for (const auto& host : DEFAULT_BOOTSTRAP_HOSTS) + addrs_node.push_back(host + ":" + std::to_string(P2P_PORT)); + out["bootstrap_addrs"] = addrs_node; + + file << out; + return file; +} + +void PoolConfig::load() +{ + YAML::Node node = YAML::LoadFile(m_filepath.string()); + + // prefix + m_prefix = ParseHexBytes(node["prefix"].as()); + + PARSE_CONFIG(node, worker, std::string); + + // Bootstrap addresses: load from YAML if present, otherwise use hardcoded defaults + if (node["bootstrap_addrs"] && node["bootstrap_addrs"].IsSequence()) + { + for (const auto& item : node["bootstrap_addrs"]) + m_bootstrap_addrs.emplace_back(item.as()); + } + else + { + for (const auto& host : DEFAULT_BOOTSTRAP_HOSTS) + m_bootstrap_addrs.emplace_back(host + ":" + std::to_string(P2P_PORT)); + } +} + +} // namespace dgb diff --git a/src/impl/dgb/node.cpp b/src/impl/dgb/node.cpp index 8ea40e7b5..645bef45b 100644 --- a/src/impl/dgb/node.cpp +++ b/src/impl/dgb/node.cpp @@ -1,41 +1,2322 @@ #include "node.hpp" -#include +#include +#include +#include +#include +#include -// c2pool-dgb node skeleton (slice #4, Option B). -// -// COMPILING SKELETON ONLY. The real embedded-daemon + pool/sharechain body -// (DensePPLNSWindow / PPLNS, p2p run-loop, template builder, broadcaster) is -// M3 / Phase B per the embedded-daemon roadmap: DGB is PORT-not-activation — -// Phase A p2p + mempool already landed under impl/dgb/coin/, and Phase B -// brings the pool pillars. When the pool-pillars node.cpp lands it REPLACES -// this file via a clean dgb-only re-cut off master; see -// c2pool-dgb-embedded-impl-plan.md (frstrtr/the docs/v36). -// -// Deliberately free of pool logic so c2pool-dgb LINKS today without dragging -// Phase B forward, instantiating the Fileconfig-derived config classes (whose -// load()/get_default() bodies are Phase B), or touching the shared -// bitcoin_family / DOGE-aux surface. +#include +#include +#include +#include +#include + +// Static members for DensePPLNSWindow precomputed decay table +std::vector dgb::DensePPLNSWindow::s_decay_table; +uint64_t dgb::DensePPLNSWindow::s_decay_per = 0; +bool dgb::DensePPLNSWindow::s_table_initialized = false; + +// Helper: read current RSS from /proc/self/status (Linux only) +static long get_rss_mb() { + std::ifstream f("/proc/self/status"); + std::string line; + while (std::getline(f, line)) { + if (line.rfind("VmRSS:", 0) == 0) { + long kb = 0; + sscanf(line.c_str(), "VmRSS: %ld", &kb); + return kb / 1024; + } + } + return 0; +} + +static long g_rss_limit_mb = 4000; // abort if RSS exceeds this (configurable) + +// p2pool-style hashrate formatting: auto-scale to H/s, kH/s, MH/s, GH/s, TH/s +static std::string format_hashrate(double hs) { + std::ostringstream os; + os << std::fixed; + if (hs >= 1e12) os << std::setprecision(2) << hs / 1e12 << "TH/s"; + else if (hs >= 1e9) os << std::setprecision(2) << hs / 1e9 << "GH/s"; + else if (hs >= 1e6) os << std::setprecision(2) << hs / 1e6 << "MH/s"; + else if (hs >= 1e3) os << std::setprecision(1) << hs / 1e3 << "kH/s"; + else os << std::setprecision(0) << hs << "H/s"; + return os.str(); +} + +// p2pool-style duration formatting: auto-scale to seconds, hours, days, years +static std::string format_duration(double secs) { + if (secs <= 0 || !std::isfinite(secs)) return "???"; + std::ostringstream os; + os << std::fixed; + if (secs >= 86400.0 * 365.25) + os << std::setprecision(1) << secs / (86400.0 * 365.25) << " years"; + else if (secs >= 86400.0) + os << std::setprecision(1) << secs / 86400.0 << " days"; + else if (secs >= 3600.0) + os << std::setprecision(1) << secs / 3600.0 << " hours"; + else if (secs >= 60.0) + os << std::setprecision(1) << secs / 60.0 << " minutes"; + else + os << std::setprecision(1) << secs << " seconds"; + return os.str(); +} + +// p2pool-style Wilson score confidence interval (util/math.py:133-152) +// Returns "~X.Y% (lo-hi%)" string for binomial proportion x/n at 95% confidence. +static std::string format_binomial_conf(int x, int n, double conf = 0.95) { + if (n == 0) return "???"; + // z for 95% ≈ 1.96 (inverse error function approximation) + double z = 1.96; + double p = static_cast(x) / n; + double topa = p + z * z / (2.0 * n); + double topb = z * std::sqrt(p * (1.0 - p) / n + z * z / (4.0 * n * n)); + double bottom = 1.0 + z * z / n; + double lo = std::max(0.0, (topa - topb) / bottom); + double hi = std::min(1.0, (topa + topb) / bottom); + std::ostringstream os; + os << "~" << std::fixed << std::setprecision(1) << (100.0 * p) << "% (" + << static_cast(std::floor(100.0 * lo)) << "-" + << static_cast(std::ceil(100.0 * hi)) << "%)"; + return os.str(); +} + +// Wilson score confidence interval for efficiency: 1 - stale_rate, scaled +static std::string format_binomial_conf_efficiency(int stale, int n, double stale_prop) { + if (n == 0) return "???"; + double z = 1.96; + double p = static_cast(stale) / n; + double topa = p + z * z / (2.0 * n); + double topb = z * std::sqrt(p * (1.0 - p) / n + z * z / (4.0 * n * n)); + double bottom = 1.0 + z * z / n; + double lo_stale = std::max(0.0, (topa - topb) / bottom); + double hi_stale = std::min(1.0, (topa + topb) / bottom); + // Efficiency = (1 - stale_rate) / (1 - stale_prop) + double denom = (stale_prop < 0.999) ? (1.0 - stale_prop) : 1.0; + double eff = (1.0 - p) / denom; + double eff_lo = (1.0 - hi_stale) / denom; + double eff_hi = (1.0 - lo_stale) / denom; + eff_lo = std::max(0.0, eff_lo); + eff_hi = std::min(1.0, eff_hi); + std::ostringstream os; + os << "~" << std::fixed << std::setprecision(1) << (100.0 * eff) << "% (" + << static_cast(std::floor(100.0 * eff_lo)) << "-" + << static_cast(std::ceil(100.0 * eff_hi)) << "%)"; + return os.str(); +} namespace dgb { -// One-line summary of the DGB-Scrypt network this binary targets. Reads the -// compile-time constants from config_{coin,pool}.hpp (header-only) — no -// Fileconfig instantiation. -std::string network_summary() +static uint64_t make_random_nonce() { - return std::string("DigiByte Scrypt-only (V36) — ") - + "pool_p2p_port=" + std::to_string(PoolConfig::P2P_PORT) - + " coin_p2p_port=" + std::to_string(CoinParams::MAINNET_P2P_PORT) - + " block_period=" + std::to_string(CoinParams::BLOCK_PERIOD) + "s" - + " gbt_algo=" + std::string(CoinParams::GBT_ALGO); + std::mt19937_64 rng(std::random_device{}()); + return rng(); } -// Skeleton entry point. Phase B replaces this with the real node run-loop. -int run_skeleton() +void NodeImpl::send_ping(peer_ptr peer) { - return 0; + auto rmsg = dgb::message_ping::make_raw(); + peer->write(std::move(rmsg)); +}; + +void NodeImpl::connected(std::shared_ptr socket) +{ + auto addr = socket->get_addr(); + bool is_outbound = m_pending_outbound.erase(addr) > 0; + + // Reject banned peers + if (is_banned(addr)) + { + LOG_INFO << "[Pool] Rejecting connection from banned peer " << addr.to_string(); + socket->close(); + return; + } + + // Let BaseNode create the peer and set up the timeout timer + base_t::connected(socket); + + if (is_outbound) + m_outbound_addrs.insert(addr); + + auto peer = m_connections[addr]; + send_version(peer); +} + +void NodeImpl::error(const message_error_type& err, const NetService& service, const std::source_location where) +{ + // If peer disconnected within 10s of us sending shares, those shares + // were likely rejected (e.g. PoW-invalid). Mark them so we don't + // keep re-broadcasting the same bad share on every reconnection. + { + auto it = m_last_broadcast_to.find(service); + if (it != m_last_broadcast_to.end()) { + auto elapsed = std::chrono::steady_clock::now() - it->second.when; + if (elapsed < std::chrono::seconds(10)) { + for (const auto& h : it->second.hashes) { + if (m_rejected_share_hashes.insert(h).second) { + LOG_WARNING << "[Pool] Marking share " << h.GetHex().substr(0, 16) + << " as rejected (peer " << service.to_string() + << " disconnected " << std::chrono::duration_cast(elapsed).count() + << "ms after broadcast)"; + } + } + } + m_last_broadcast_to.erase(it); + } + } + + // Drop stale nonce->peer entries for this endpoint before base cleanup. + // Without this, reconnects can be rejected as false duplicates because + // m_peers still contains the old nonce mapping. + for (auto it = m_peers.begin(); it != m_peers.end(); ) + { + if (it->second && it->second->addr() == service) + it = m_peers.erase(it); + else + ++it; + } + + // Clean outbound tracking before base removes the peer + m_pending_outbound.erase(service); + m_outbound_addrs.erase(service); + + // p2pool p2p.py:595: self.get_shares.respond_all(reason) + // Cancel pending share requests for this peer — invokes callbacks with + // empty response so m_downloading_shares entries are cleaned up immediately + // instead of waiting for the 5s ReplyMatcher timeout. + { + std::vector to_cancel; + for (auto& [req_id, peer_addr] : m_pending_share_reqs) { + if (peer_addr == service) + to_cancel.push_back(req_id); + } + for (auto& req_id : to_cancel) { + m_pending_share_reqs.erase(req_id); + m_share_getter.cancel(req_id); + } + } + + base_t::error(err, service, where); +} + +void NodeImpl::close_connection(const NetService& service) +{ + // Same rejection tracking as error() — close_connection is another + // path for peer disconnection. + { + auto it = m_last_broadcast_to.find(service); + if (it != m_last_broadcast_to.end()) { + auto elapsed = std::chrono::steady_clock::now() - it->second.when; + if (elapsed < std::chrono::seconds(10)) { + for (const auto& h : it->second.hashes) + m_rejected_share_hashes.insert(h); + } + m_last_broadcast_to.erase(it); + } + } + + m_pending_outbound.erase(service); + m_outbound_addrs.erase(service); + + // Cancel pending share requests for this peer (same as in error()) + { + std::vector to_cancel; + for (auto& [req_id, peer_addr] : m_pending_share_reqs) { + if (peer_addr == service) + to_cancel.push_back(req_id); + } + for (auto& req_id : to_cancel) { + m_pending_share_reqs.erase(req_id); + m_share_getter.cancel(req_id); + } + } + + base_t::close_connection(service); +} + +NodeImpl::TrackerSnapshot NodeImpl::get_tracker_snapshot() const { + std::lock_guard lock(m_snapshot_mutex); + return m_snapshot; +} + +int NodeImpl::get_chain_count() const { return get_tracker_snapshot().chain_count; } +int NodeImpl::get_verified_count() const { return get_tracker_snapshot().verified_count; } + +void NodeImpl::send_version(peer_ptr peer) +{ + auto rmsg = dgb::message_version::make_raw( + m_tracker.m_params->minimum_protocol_version, + 1, // services + addr_t{1, peer->addr()}, // addr_to (the remote) + addr_t{1, NetService{"0.0.0.0", m_tracker.m_params->p2p_port}}, // addr_from (us) + m_nonce, + m_software_version, + 1, // mode (always 1 for legacy compat) + advertised_best_share() // verified head, or raw head pre-sync (ROOT-2) + ); + peer->write(std::move(rmsg)); +} + +std::optional NodeImpl::handle_version(std::unique_ptr rmsg, peer_ptr peer) +{ + LOG_DEBUG_POOL << "handle message_version"; + std::unique_ptr msg; + msg = dgb::message_version::make(rmsg->m_data); + + LOG_INFO << "[Pool] Peer " + << msg->m_addr_from.m_endpoint.to_string() + << " says protocol version is " + << msg->m_version + << ", client version " + << msg->m_subversion; + + if (peer->m_other_version.has_value()) + { + LOG_DEBUG_POOL << "more than one version message"; + throw std::runtime_error("more than one version message"); + } + + peer->m_other_version = msg->m_version; + peer->m_other_subversion = msg->m_subversion; + peer->m_other_services = msg->m_services; + + if (m_nonce == msg->m_nonce) + { + LOG_WARNING << "[Pool] was connected to self"; + return std::nullopt; + } + + if (m_peers.contains(msg->m_nonce)) + { + LOG_DEBUG_POOL << "Detected duplicate connection, disconnecting from " << peer->addr().to_string(); + return std::nullopt; + } + + peer->m_nonce = msg->m_nonce; + m_peers[peer->m_nonce] = peer; + + // Request peers from the newly established connection + { + auto getaddrs_msg = dgb::message_getaddrs::make_raw(8); + peer->write(std::move(getaddrs_msg)); + } + + // Reject peers running too-old protocol + if (msg->m_version < m_tracker.m_params->minimum_protocol_version) + { + LOG_WARNING << "Peer " << msg->m_addr_from.m_endpoint.to_string() + << " protocol " << msg->m_version + << " < minimum " << m_tracker.m_params->minimum_protocol_version + << ", disconnecting"; + throw std::runtime_error("peer protocol too old"); + } + + if (!msg->m_best_share.IsNull()) + { + LOG_INFO << "Best share hash for " << msg->m_addr_from.m_endpoint.to_string() + << " = " << msg->m_best_share.ToString(); + + if (!m_chain->contains(msg->m_best_share)) { + // Start downloading shares we don't have + download_shares(peer, msg->m_best_share); + } else { + // p2pool: handle_share_hashes → handle_shares → set_best_share() + // Even when the share is known, re-run think() to re-evaluate + // best chain with the peer's perspective. Critical after restart: + // shares loaded from LevelDB may have stale best_share selection. + run_think(); + } + } + + // Advertise ourselves to the peer (matching Python p2pool sendAdvertisement) + { + auto port = core::Server::listen_port(); + auto addrme_msg = dgb::message_addrme::make_raw(port); + peer->write(std::move(addrme_msg)); + } + + return pool::PeerConnectionType::legacy; +} + +void NodeImpl::processing_shares(HandleSharesData& data_ref, NetService addr) +{ + // Take ownership immediately so the caller can return/free its local. + auto data = std::make_shared(std::move(data_ref)); + size_t n = data->m_items.size(); + if (n == 0) return; + + // Phase 1 (thread pool, parallel): run share_init_verify() for each share. + // share_init_verify() does scrypt-1024 (~20ms each) — must NOT block io_context. + // Each share's hash computation is independent, so we can fully parallelize. + auto remaining = std::make_shared>(static_cast(n)); + for (size_t i = 0; i < n; i++) + { + boost::asio::post(m_verify_pool, + [i, data, remaining, this, addr]() + { + auto& share = data->m_items[i]; + if (share.hash().IsNull()) + { + try + { + share.ACTION({ + obj->m_hash = share_init_verify(*obj, *m_tracker.m_params, true); + }); + } + catch (const std::exception&) + { + // leave hash null — phase 2 will skip this share + } + } + // When all verifications are done, schedule phase 2 on io_context + if (--(*remaining) == 0) + { + boost::asio::post(*m_context, + [data, this, addr]() + { + processing_shares_phase2(*data, addr); + }); + } + }); + } +} + +void NodeImpl::processing_shares_phase2(HandleSharesData& data, NetService addr) +{ + // Phase 2 (io_context thread): topological sort + chain insertion + LevelDB store. + // All shared state (m_tracker, m_chain, m_raw_share_cache, m_storage) touched here. + // + // Non-blocking mutex: if think() holds the exclusive lock on the compute + // thread, queue this batch for processing after think() releases. The IO + // thread never blocks — keepalive timers and network I/O continue. + { + std::unique_lock lock(m_tracker_mutex, std::try_to_lock); + if (!lock.owns_lock()) { + LOG_INFO << "[ASYNC-DEFER] processing_shares_phase2: mutex busy, queuing " + << data.m_items.size() << " shares from " << addr.to_string() + << " (pending=" << m_pending_adds.size() + 1 << ")"; + m_pending_adds.push_back(PendingShareBatch{ + std::make_unique(std::move(data)), addr}); + return; + } + } + // Lock released — proceed with normal processing. + // No lock needed for the rest: we only enter here when think() is NOT + // running (m_tracker_mutex was available), and ASIO single-thread + // guarantees no other IO handler overlaps. + + // Step 1: collect verified shares (skip any that failed verification, hash still null) + std::vector valid_shares; + valid_shares.reserve(data.m_items.size()); + for (size_t idx = 0; idx < data.m_items.size(); ++idx) + { + auto& share = data.m_items[idx]; + if (share.hash().IsNull()) + continue; // verification failed in phase 1 + + // Cache original raw bytes for relay (keyed by computed hash) + if (idx < data.m_raw_items.size() && !data.m_raw_items[idx].contents.m_data.empty()) + m_raw_share_cache[share.hash()] = std::move(data.m_raw_items[idx]); + valid_shares.push_back(share); + } + + // Step 2: Topologically sort valid shares by hash/prev_hash linkage + chain::PreparedList prepare_shares(valid_shares); + std::vector shares = prepare_shares.build_list(); + + // Step 3: Process sorted shares + int32_t new_count = 0; + int32_t dup_count = 0; + std::map all_new_txs; + std::vector db_batch; + for (int i = 0; i < (int)shares.size(); ++i) + { + auto& share = shares[i]; + + // Safety: abort if RSS exceeds limit + if (i % 100 == 0) { + long rss_now = get_rss_mb(); + if (rss_now > g_rss_limit_mb) { + LOG_ERROR << "RSS LIMIT EXCEEDED (" << rss_now << "MB > " << g_rss_limit_mb << "MB) — aborting!"; + std::abort(); + } + } + + auto& new_txs = data.m_txs[share.hash()]; + if (!new_txs.empty()) + { + for (auto& new_tx : new_txs) + { + PackStream packed_tx = pack(coin::TX_WITH_WITNESS(new_tx)); + all_new_txs[Hash(packed_tx.get_span())] = new_tx; + } + } + + if (m_chain->contains(share.hash())) + { + ++dup_count; + continue; + } + + ++new_count; + + // Log received share — p2pool format: "Received share diff=X hash=Y miner=Z" + share.invoke([](auto* obj) { + auto target = chain::bits_to_target(obj->m_bits); + double diff = chain::target_to_difficulty(target); + // Extract miner identity (pubkey_hash for v17/v33/v36, address script for v34/v35) + std::string miner_hex; + if constexpr (requires { obj->m_pubkey_hash; }) + miner_hex = obj->m_pubkey_hash.GetHex().substr(0, 16); + else if constexpr (requires { obj->m_address; }) + miner_hex = "script"; + LOG_INFO << "Received share: diff=" << std::scientific << std::setprecision(2) << diff + << " hash=" << obj->m_hash.GetHex().substr(0, 16) + << " miner=" << miner_hex; + }); + + m_tracker.add(share); + + // Log fork detection: if this share's prev_hash has other children, it forks + { + uint256 prev; + bool is_local = false; + share.invoke([&](auto* obj) { + prev = obj->m_prev_hash; + is_local = (obj->peer_addr == NetService{"0.0.0.0", 0} || + obj->peer_addr == NetService{}); + }); + if (is_local && !prev.IsNull()) { + auto& rev = m_tracker.chain.get_reverse(); + auto it = rev.find(prev); + if (it != rev.end() && it->second.size() > 1) { + static int fork_log = 0; + if (fork_log++ < 50) + LOG_WARNING << "[FORK] Local share forks! prev=" << prev.GetHex().substr(0,16) + << " siblings=" << it->second.size() + << " verified_best=" << m_best_share_hash.GetHex().substr(0,16); + } + } + } + + // Verification is deferred to think() Phase 1 (called after this batch). + // p2pool: handle_shares() only adds, set_best_share()→think() verifies. + // Inline verification was redundant and caused double-verify CPU waste. + + // NOTE: Do NOT trim inside the processing loop. The trim in run_think() + // handles pruning between batches. Trimming here is unsafe because + // shares added at the tail can be freed while the loop still holds + // dangling raw pointers to them (use-after-free). + + // Collect for batch LevelDB persist (committed atomically after loop) + if (m_storage && m_storage->is_available()) + { + std::vector bytes; + auto raw_it = m_raw_share_cache.find(share.hash()); + if (raw_it != m_raw_share_cache.end() && + raw_it->second.type == share.version() && + !raw_it->second.contents.m_data.empty()) + { + bytes.assign(raw_it->second.contents.m_data.begin(), + raw_it->second.contents.m_data.end()); + } + else + { + PackStream ps = pack(share); + auto span = ps.get_span(); + bytes.assign(reinterpret_cast(span.data()), + reinterpret_cast(span.data()) + span.size()); + } + uint64_t ver = share.version(); + std::vector versioned; + versioned.resize(8 + bytes.size()); + std::memcpy(versioned.data(), &ver, 8); + std::memcpy(versioned.data() + 8, bytes.data(), bytes.size()); + + share.ACTION({ + uint256 target = chain::bits_to_target(obj->m_bits); + uint256 abswork_256; + std::copy(obj->m_abswork.begin(), obj->m_abswork.end(), abswork_256.begin()); + c2pool::storage::SharechainStorage::ShareBatchEntry entry; + entry.hash = obj->m_hash; + entry.serialized_data = std::move(versioned); + entry.prev_hash = obj->m_prev_hash; + entry.height = obj->m_absheight; + entry.timestamp = obj->m_timestamp; + entry.work = abswork_256; + entry.target = target; + db_batch.push_back(std::move(entry)); + }); + } + } + + // Commit all shares to LevelDB atomically (one WriteBatch for entire batch). + // Crash-safe: either ALL shares persisted or NONE. + if (!db_batch.empty() && m_storage && m_storage->is_available()) { + m_storage->store_shares_batch(db_batch); + } + + if (new_count > 0) { + auto as2 = addr.to_string(); + std::string source = (addr.port() == 0) ? as2.substr(0, as2.rfind(':')) : as2; + LOG_INFO << "Processing " << new_count << " shares from " + << source << "... (dup=" << dup_count + << " chain=" << m_tracker.chain.size() << ")"; + } + + // Trigger think() after every share batch (p2pool: set_best_share after handle_shares). + // p2pool calls set_best_share() after EVERY batch with new_count > 0 — no size gate. + // think() scores heads and updates best_share + desired set for download_shares. + if (new_count > 0) { + run_think(); + } +} + +std::vector NodeImpl::handle_get_share(std::vector hashes, uint64_t parents, std::vector stops, NetService peer_addr) +{ + // try_to_lock per the architectural rule (node.hpp:67) — IO thread MUST + // never block on m_tracker_mutex. A blocking shared_lock here was the + // root cause of the periodic event-loop freeze + SIGABRT cycle on + // contabo (2026-04-12, -16, -19, -21, -25): when the compute thread + // held the exclusive lock for a long think+clean cycle on a wedged + // chain (~30+s), an incoming SHAREREQ on the IO thread would block + // here, the watchdog would fire after 30s of io_context unresponsive, + // and systemd would restart. + // + // Empty reply does NOT cause peer disconnect — p2pool's downloader + // (node.py:120) picks a random peer per request and retries; an empty + // hit just shifts to a different peer next iteration. + std::shared_lock lock(m_tracker_mutex, std::try_to_lock); + if (!lock.owns_lock()) + { + static int defer_log = 0; + if (defer_log++ % 50 == 0) + LOG_INFO << "[handle_get_share] tracker busy — returning empty to " + << peer_addr.to_string() + << " (peer will retry against another peer)"; + return {}; + } + + parents = std::min(parents, (uint64_t)1000/hashes.size()); + std::vector shares; + for (const auto& handle_hash : hashes) + { + if (!m_chain->contains(handle_hash)) + { + static int miss_log = 0; + if (miss_log++ < 5) + LOG_WARNING << "[handle_get_share] hash NOT in chain: " + << handle_hash.ToString().substr(0, 16) + << " chain_size=" << m_chain->size() + << " tracker_chain_size=" << m_tracker.chain.size(); + continue; + } + uint64_t n = std::min(parents+1, (uint64_t) m_chain->get_height(handle_hash)); + for (auto [hash, data] : m_chain->get_chain(handle_hash, n)) + { + if (std::find(stops.begin(), stops.end(), hash) != stops.end()) + break; + if (m_rejected_share_hashes.count(hash)) + continue; + shares.push_back(data.share); + } + } + + if (!shares.empty()) + { + LOG_INFO << "[Pool] Sending " << shares.size() << " shares to " << peer_addr.to_string(); + } + return shares; +} + +void NodeImpl::send_shares(peer_ptr peer, const std::vector& share_hashes) +{ + // try_to_lock per the architectural rule (node.hpp:67) — see freeze + // analysis in handle_get_share above. If we can't acquire NOW, skip + // this batch. The shares are still in our chain; the next broadcast + // cycle (or the next think() result) picks them up. + std::shared_lock lock(m_tracker_mutex, std::try_to_lock); + if (!lock.owns_lock()) + { + static int defer_log = 0; + if (defer_log++ % 50 == 0) + LOG_INFO << "[send_shares] tracker busy — skipping send to " + << peer->addr().to_string() << " (will retry next cycle)"; + return; + } + + // Collect shares that exist in our chain (skip rejected) + std::vector shares; + for (const auto& hash : share_hashes) + { + if (!m_chain->contains(hash)) + continue; + if (m_rejected_share_hashes.count(hash)) + continue; + // Retrieve the share via get_chain(hash, 1) — first element is the share itself + for (auto [h, data] : m_chain->get_chain(hash, 1)) + { + shares.push_back(data.share); + break; + } + } + + if (shares.empty()) + return; + + // Collect transactions that the peer doesn't know about + std::set needed_txs; + for (auto& share : shares) + { + share.invoke([&](auto* obj) { + if constexpr (requires { obj->m_new_transaction_hashes; }) + { + for (const auto& th : obj->m_new_transaction_hashes) + { + if (!peer->m_remote_txs.count(th) && + !peer->m_remembered_txs.count(th)) + needed_txs.insert(th); + } + } + }); + } + + // Send remember_tx for txs the peer needs + if (!needed_txs.empty()) + { + std::vector known_hashes; // hashes in peer's remote set + std::vector full_txs; // full txs otherwise + + for (const auto& th : needed_txs) + { + if (peer->m_remote_txs.count(th)) + { + known_hashes.push_back(th); + } + else + { + auto it = m_known_txs.find(th); + if (it != m_known_txs.end()) + full_txs.emplace_back(it->second); + } + } + + if (!known_hashes.empty() || !full_txs.empty()) + { + auto rtx_msg = message_remember_tx::make_raw(known_hashes, full_txs); + peer->write(std::move(rtx_msg)); + } + } + + // Pack and send shares — use cached original bytes when available + std::vector rshares; + rshares.reserve(shares.size()); + for (size_t i = 0; i < shares.size(); ++i) + { + auto it = m_raw_share_cache.find(shares[i].hash()); + if (it != m_raw_share_cache.end()) + { + rshares.push_back(it->second); + } + else + { + rshares.emplace_back(shares[i].version(), pack(shares[i])); + } + } + + auto shares_msg = message_shares::make_raw(rshares); + peer->write(std::move(shares_msg)); + + // Send forget_tx so peer can free the remembered txs + if (!needed_txs.empty()) + { + std::vector forget_vec(needed_txs.begin(), needed_txs.end()); + auto ftx_msg = message_forget_tx::make_raw(forget_vec); + peer->write(std::move(ftx_msg)); + } + + LOG_INFO << "[Pool] Sent " << shares.size() << " shares (+" << needed_txs.size() + << " txs) to " << peer->addr().to_string(); +} + +void NodeImpl::broadcast_share(const uint256& share_hash) +{ + // try_to_lock per the architectural rule (node.hpp:67) — see freeze + // analysis in handle_get_share above. If think+clean holds the + // exclusive lock right now, defer this broadcast: the share is still + // in our chain; the next local share creation, the next think cycle, + // or the next peer-driven SHAREREQ will pick it up. Blocking here + // was a contributing freeze trigger (called from local-share creation + // and from the share-add hot path). + std::shared_lock lock(m_tracker_mutex, std::try_to_lock); + if (!lock.owns_lock()) + { + static int defer_log = 0; + if (defer_log++ % 50 == 0) + LOG_INFO << "[broadcast_share] tracker busy — deferring broadcast of " + << share_hash.GetHex().substr(0, 16) + << " (next cycle will pick it up)"; + return; + } + + // Walk the chain back from share_hash, collecting un-broadcast shares + std::vector to_send; + int32_t height = m_chain->get_height(share_hash); + int32_t walk = std::min(height, 5); + + for (auto [hash, data] : m_chain->get_chain(share_hash, walk)) + { + if (m_shared_share_hashes.count(hash)) + break; + if (m_rejected_share_hashes.count(hash)) + continue; // skip shares previously rejected by peers + m_shared_share_hashes.insert(hash); + to_send.push_back(hash); + } + + if (to_send.empty()) + return; + + auto now = std::chrono::steady_clock::now(); + for (auto& [nonce, peer] : m_peers) { + send_shares(peer, to_send); + m_last_broadcast_to[peer->addr()] = {to_send, now}; + } +} + +void NodeImpl::notify_local_share(const uint256& share_hash) +{ + // p2pool: set_best_share() → think() synchronously on the reactor thread. + // Use think() for ALL best_share decisions, matching p2pool exactly. + if (share_hash.IsNull() || !m_tracker.chain.contains(share_hash)) + return; + + // Try inline verify — if think() holds the mutex, defer to next think() cycle. + // The share is already in the chain; think() will verify + score it. + { + std::unique_lock lock(m_tracker_mutex, std::try_to_lock); + if (lock.owns_lock()) + m_tracker.attempt_verify(share_hash); + } + + // Trigger async think() — will pick up the local share through scoring. + run_think(); +} + +uint256 NodeImpl::best_share_hash() +{ + // p2pool's think() returns the best VERIFIED head — not the raw chain tip. + // This ensures shares are built on a chain that all peers agree on. + // Use think().best ONLY if it's on the VERIFIED chain. + // p2pool's think() only considers verified shares for the best head. + // If think().best is on an unverified fork (e.g., our own shares that + // p2pool hasn't verified yet), fall back to the best verified head. + // This prevents building a self-reinforcing fork that never joins + // the main chain. + if (!m_best_share_hash.IsNull() && m_tracker.verified.contains(m_best_share_hash)) { + static int log_count = 0; + if (log_count++ % 60 == 0) { + auto h = m_tracker.verified.get_height(m_best_share_hash); + LOG_INFO << "[best_share] using think() result height=" << h + << " verified=" << m_tracker.verified.size() + << " raw=" << (m_chain ? m_chain->size() : 0); + } + return m_best_share_hash; + } + // Fallback: if think() hasn't run yet, pick best verified head by work + auto& verified = m_tracker.verified; + if (verified.size() > 0) { + uint256 best; + uint288 best_work; + bool first = true; + for (const auto& [head_hash, tail_hash] : verified.get_heads()) { + auto* idx = verified.get_index(head_hash); + if (idx && (first || idx->work > best_work)) { + best = head_hash; + best_work = idx->work; + first = false; + } + } + if (!best.IsNull()) { + static int log_count2 = 0; + auto best_height = verified.get_height(best); + if (log_count2++ % 60 == 0) + LOG_INFO << "[best_share] fallback VERIFIED head height=" << best_height + << " work=" << best_work.GetHex().substr(0, 16) + << " verified=" << verified.size() + << " raw=" << (m_chain ? m_chain->size() : 0); + return best; + } + } + + // No verified chain yet — return null to prevent creating shares with + // MAX_TARGET max_bits. p2pool does the same: best_share_var.value = None + // until the verified chain exists, and generate_transaction refuses to + // create work when best_share is None (with peers connected). + // + // On a genesis node (no peers, fresh chain), this returns ZERO which + // triggers genesis share creation with 100% donation payout. + if (!m_peers.empty()) { + static int wait_log = 0; + if (wait_log++ % 12 == 0) + LOG_INFO << "[best_share] waiting for verified chain (peers=" + << m_peers.size() << " raw=" + << (m_chain ? m_chain->size() : 0) << ")"; + return uint256::ZERO; + } + + // True genesis: no peers at all — use raw chain to bootstrap + if (!m_chain || m_chain->size() == 0) + return uint256::ZERO; + + uint256 best; + int32_t best_height = -1; + for (const auto& [head_hash, tail_hash] : m_chain->get_heads()) { + auto h = m_chain->get_height(head_hash); + if (h > best_height) { + best = head_hash; + best_height = h; + } + } + static int raw_log = 0; + if (raw_log++ % 60 == 0) + LOG_INFO << "[best_share] using RAW head (genesis, no peers) height=" << best_height; + return best; +} + +uint256 NodeImpl::advertised_best_share() +{ + // Peer-facing advertisement ONLY (version handshake + timer re-announce). + // NEVER used for work/share creation — best_share_hash() owns that and + // deliberately returns a VERIFIED head (or NULL) so we never build local + // work on an unagreed chain. For ADVERTISEMENT the opposite is right: a + // peer must learn our tallest head to call download_shares() and pull our + // chain. ROOT-2 (Option-A net-id soak): a --genesis node whose verified + // chain is still empty at handshake advertises NULL via best_share_hash(); + // the peer never downloads and broadcast can't backfill (head shares are + // already de-dup-marked). Advertising the raw head breaks that deadlock + // without touching work creation. + uint256 v = best_share_hash(); + if (!v.IsNull()) + return v; + + // Verified chain still empty — advertise our tallest RAW head instead. + if (m_chain && m_chain->size() > 0) { + uint256 best; + int32_t best_height = -1; + for (const auto& [head_hash, tail_hash] : m_chain->get_heads()) { + auto h = m_chain->get_height(head_hash); + if (h > best_height) { + best = head_hash; + best_height = h; + } + } + return best; + } + return uint256::ZERO; +} + +void NodeImpl::readvertise_best_share() +{ + // ROOT-2 re-advertisement. A peer that finished its version handshake + // while our verified chain was empty got a NULL best_share and never + // called download_shares(); broadcast_share() can't wake it either because + // our head shares are already in m_shared_share_hashes (de-dup-marked at + // creation) so its walk breaks immediately. Here we re-push the tip walk + // to every peer WITHOUT consulting the de-dup set. Advertise-only: never + // affects local work creation. + uint256 head = advertised_best_share(); + if (head.IsNull()) + return; + + // Same try_to_lock discipline as broadcast_share (node.hpp:67): never block + // the IO thread on the tracker mutex. If think() holds it now, the next + // trigger (best-change or the timer) retries. + std::shared_lock lock(m_tracker_mutex, std::try_to_lock); + if (!lock.owns_lock()) + return; + + if (!m_chain->contains(head)) + return; + + std::vector to_send; + int32_t height = m_chain->get_height(head); + int32_t walk = std::min(height, 5); + for (auto [hash, data] : m_chain->get_chain(head, walk)) { + if (m_rejected_share_hashes.count(hash)) + continue; // never re-broadcast peer-rejected shares + to_send.push_back(hash); + } + if (to_send.empty()) + return; + + auto now = std::chrono::steady_clock::now(); + for (auto& [nonce, peer] : m_peers) { + send_shares(peer, to_send); + m_last_broadcast_to[peer->addr()] = {to_send, now}; + } + LOG_INFO << "[readvertise] re-pushed " << to_send.size() + << " head share(s) to " << m_peers.size() << " peer(s) (ROOT-2)"; +} + +void NodeImpl::download_shares(peer_ptr /*unused_peer*/, const uint256& target_hash) +{ + // p2pool node.py:108-141 download_shares() — exact translation. + // + // Key differences from old c2pool implementation: + // 1. RANDOM peer selection (not the reporting peer) + // 2. RANDOM parent count 0-499 (not fixed 500) + // 3. STOPS list: known heads + their 10th parents (not empty) + // 4. Log format: "Requesting parent share XXXX from IP:PORT" + + // Already downloading this hash — avoid duplicate requests + if (m_downloading_shares.count(target_hash)) + return; + + // Stop requesting hashes that have returned empty too many times — + // the share is pruned from the network. p2pool uses reactive desired_var + // + sleep(1) backoff; we use explicit failure counting. + if (m_download_fail_count[target_hash] >= MAX_EMPTY_RETRIES) { + static int fail_log = 0; + if (fail_log++ % 20 == 0) + LOG_INFO << "[Pool] Skipping permanently failed hash " + << target_hash.ToString().substr(0,16) + << " (failed " << m_download_fail_count[target_hash] << " times)"; + return; + } + + m_downloading_shares.insert(target_hash); + + // p2pool: if len(self.peers) == 0: sleep(1); continue + if (m_peers.empty()) { + m_downloading_shares.erase(target_hash); + return; + } + + // p2pool: peer = random.choice(self.peers.values()) + auto peer_it = m_peers.begin(); + if (m_peers.size() > 1) { + std::advance(peer_it, core::random::random_uint256().GetLow64() % m_peers.size()); + } + auto& peer = peer_it->second; + + // p2pool: parents=random.randrange(500) + uint64_t parents = core::random::random_uint256().GetLow64() % 500; + + // p2pool: stops=list(set(tracker.heads) | set( + // tracker.get_nth_parent_hash(head, min(max(0, height-1), 10)) + // for head in tracker.heads))[:100] + // + // Always include stops to bound per-request reply size to the actual + // info-gap between our chain and the peer's. Without stops, peer dumps + // up to `parents` shares in one single-branch burst — and with bootstrap + // parents=random(500), the chain grows along one lineage until it + // crosses 2*CHAIN_LENGTH+10, at which point clean_tracker drop-tails + // collapses the whole branch and verified resets to 0. + std::vector stops; + { + std::set stop_set; + for (auto& [head_hash, tail_hash] : m_tracker.chain.get_heads()) { + stop_set.insert(head_hash); + auto h = m_tracker.chain.get_acc_height(head_hash); + auto nth = std::min(std::max(0, h - 1), 10); + if (nth > 0) { + auto parent = m_tracker.chain.get_nth_parent_via_skip(head_hash, nth); + if (!parent.IsNull()) + stop_set.insert(parent); + } + } + int count = 0; + for (auto& s : stop_set) { + if (count++ >= 100) break; + stops.push_back(s); + } + } + + auto req_id = core::random::random_uint256(); + std::vector hashes = { target_hash }; + + // Track req_id → peer for selective cancellation on disconnect + m_pending_share_reqs[req_id] = peer->addr(); + + // p2pool: print 'Requesting parent share %s from %s' + LOG_INFO << "[Pool] Requesting parent share " + << target_hash.ToString().substr(0,16) + << " from " << peer->addr().to_string() + << " (parents=" << parents << " stops=" << stops.size() << ")"; + + // weak_ptr prevents use-after-free if peer disconnects before reply + std::weak_ptr> weak_peer = peer; + auto peer_addr_for_log = peer->addr(); + + request_shares(req_id, peer, hashes, parents, stops, + [this, weak_peer, target_hash, peer_addr_for_log, req_id](dgb::ShareReplyData reply) + { + m_downloading_shares.erase(target_hash); + m_pending_share_reqs.erase(req_id); + + if (reply.m_items.empty()) + { + // Empty reply = timeout or peer had no matching shares. + // Track failures — stop requesting after MAX_EMPTY_RETRIES. + auto& fail_cnt = m_download_fail_count[target_hash]; + ++fail_cnt; + LOG_INFO << "[Pool] Share request empty for " + << target_hash.ToString().substr(0,16) + << " from " << peer_addr_for_log.to_string() + << " (fail " << fail_cnt << "/" << MAX_EMPTY_RETRIES << ")"; + return; + } + + // Success — clear failure count for this hash + m_download_fail_count.erase(target_hash); + + LOG_INFO << "[Pool] Received " << reply.m_items.size() << " shares for download request"; + + // Feed into processing pipeline + HandleSharesData data; + for (size_t idx = 0; idx < reply.m_items.size(); ++idx) + { + if (idx < reply.m_raw_items.size()) + data.add(reply.m_items[idx], {}, reply.m_raw_items[idx]); + else + data.add(reply.m_items[idx], {}); + } + processing_shares(data, peer_addr_for_log); + + // Find the oldest share's parent — if unknown, keep fetching + uint256 oldest_parent; + reply.m_items.back().invoke([&](auto* obj) { oldest_parent = obj->m_prev_hash; }); + + if (!oldest_parent.IsNull() && !m_chain->contains(oldest_parent)) + { + auto locked = weak_peer.lock(); + if (locked) + download_shares(locked, oldest_parent); + } + } + ); +} + +void NodeImpl::load_persisted_shares() +{ + if (!m_storage || !m_storage->is_available()) + return; + + // load_sharechain iterates the height index and loads each share. + // Limit to keep_per_head shares to avoid loading unbounded history + // into memory (LevelDB is never pruned, so it accumulates forever). + LOG_INFO << "[Pool] Scanning LevelDB height index for persisted shares..."; + auto t0 = std::chrono::steady_clock::now(); + auto all_hashes = m_storage->get_shares_by_height_range(0, UINT64_MAX); + auto scan_ms = std::chrono::duration_cast( + std::chrono::steady_clock::now() - t0).count(); + { + std::set unique(all_hashes.begin(), all_hashes.end()); + LOG_INFO << "[Pool] Height index scan: " << all_hashes.size() << " entries (" + << unique.size() << " unique) in " << scan_ms << "ms"; + } + if (all_hashes.empty()) + { + LOG_INFO << "No persisted shares found in LevelDB"; + return; + } + + const size_t keep_per_head = m_tracker.m_params->chain_length * 2 + 10; + const size_t total_in_db = all_hashes.size(); + + // Only load the most recent shares (highest height = end of vector) + size_t skip = 0; + if (total_in_db > keep_per_head) + { + skip = total_in_db - keep_per_head; + LOG_INFO << "[Pool] LevelDB has " << total_in_db << " shares, loading only newest " + << keep_per_head << " (skipping " << skip << " old shares)"; + } + + const size_t to_load = total_in_db - skip; + LOG_INFO << "[Pool] Loading shares from LevelDB: " << to_load << " shares to process..."; + int loaded = 0, skipped_contains = 0, skipped_load = 0, skipped_small = 0; + std::vector verified_hashes; // shares to pre-populate into verified + for (size_t i = skip; i < total_in_db; ++i) + { + const auto& hash = all_hashes[i]; + std::vector data; + core::ShareMetadata meta; + if (!m_storage->load_share(hash, data, meta)) { + ++skipped_load; + continue; + } + if (data.size() < 8) { + ++skipped_small; + continue; + } + + try + { + // First 8 bytes = version (uint64_t LE), rest = packed share + uint64_t ver; + std::memcpy(&ver, data.data(), 8); + + std::vector share_bytes(data.begin() + 8, data.end()); + PackStream ps(share_bytes); + + auto share = dgb::load_share(static_cast(ver), ps, NetService{"database", 0}); + // m_hash is not part of the serialized format — it's computed + // during share_check validation. Restore it from the LevelDB key. + share.ACTION({ obj->m_hash = hash; }); + if (!m_chain->contains(share.hash())) + { + m_tracker.add(share); + loaded++; + + // Restore or compute pow_hash on the index. + // p2pool recomputes pow_hash on every load (data.py:1362). + // We cache it in LevelDB to avoid scrypt, but recompute if missing. + { + auto* idx = m_tracker.chain.get_index(share.hash()); + if (idx) { + if (!meta.pow_hash.IsNull()) { + idx->pow_hash = meta.pow_hash; + } else { + // Recompute scrypt pow_hash (matching p2pool's __init__ behavior) + try { + g_last_pow_hash = uint256(); + g_last_init_is_block = false; + uint256 computed_hash; + share.ACTION({ computed_hash = share_init_verify(*obj, *m_tracker.m_params, true); }); + if (!g_last_pow_hash.IsNull()) + idx->pow_hash = g_last_pow_hash; + if (!computed_hash.IsNull() && computed_hash != hash) { + static int hash_mismatch = 0; + if (++hash_mismatch <= 3) + LOG_WARNING << "pow_hash recompute: hash mismatch stored=" + << hash.GetHex().substr(0,16) + << " computed=" << computed_hash.GetHex().substr(0,16); + } + } catch (...) {} + } + } + } + + // p2pool known_verified: pre-populate verified tracker without re-verification + if (meta.is_verified) + verified_hashes.push_back(share.hash()); + + if (loaded % 1000 == 0) { + size_t progress = i - skip + 1; + LOG_INFO << "[Pool] Loading shares: " << progress << "/" << to_load + << " (" << (100 * progress / to_load) << "%)"; + } + } else { + ++skipped_contains; + } + } + catch (const std::exception& e) + { + LOG_WARNING << "Failed to load share " << hash.ToString() + << " from LevelDB: " << e.what(); + } + } + + // Pre-populate verified chain from persisted known_verified set (p2pool node.py:190-192). + // Shares are loaded by ascending height, so parent-before-child order is guaranteed. + int pre_verified = 0; + for (const auto& vh : verified_hashes) { + if (m_tracker.chain.contains(vh) && !m_tracker.verified.contains(vh)) { + try { + auto& share_var = m_tracker.chain.get_share(vh); + m_tracker.verified.add(share_var); + ++pre_verified; + } catch (...) {} + } + } + if (pre_verified > 0) + LOG_INFO << "[Pool] Pre-populated " << pre_verified << " verified shares from LevelDB"; + + LOG_INFO << "[Pool] Loaded " << loaded << " shares from LevelDB storage" + << " (DB total: " << total_in_db << ", limit: " << keep_per_head + << ", skipped: load=" << skipped_load << " small=" << skipped_small + << " dup=" << skipped_contains << ")"; + + // Prune old shares from LevelDB that we skipped + if (skip > 0) + { + size_t pruned = 0; + for (size_t i = 0; i < skip; ++i) + { + if (m_storage->remove_share(all_hashes[i])) + ++pruned; + } + LOG_INFO << "[Pool] Pruned " << pruned << " old shares from LevelDB"; + } +} + +void NodeImpl::flush_verified_to_leveldb() +{ + if (m_verified_flush_buf.empty() || !m_storage || !m_storage->is_available()) + return; + // Persist pow_hash from tracker index alongside verified status + std::vector> hash_pow_pairs; + for (const auto& hash : m_verified_flush_buf) { + uint256 pow; + auto* idx = m_tracker.chain.get_index(hash); + if (idx) pow = idx->pow_hash; + hash_pow_pairs.emplace_back(hash, pow); + } + m_storage->mark_shares_verified_with_pow(hash_pow_pairs); + m_verified_flush_buf.clear(); +} + +void NodeImpl::shutdown() +{ + LOG_INFO << "[Pool] NodeImpl shutdown: flushing pending LevelDB buffers..."; + flush_verified_to_leveldb(); + if (!m_removal_flush_buf.empty() && m_storage && m_storage->is_available()) + { + m_storage->remove_shares_batch(m_removal_flush_buf); + m_removal_flush_buf.clear(); + } + LOG_INFO << "[Pool] NodeImpl shutdown complete"; +} + +void NodeImpl::start_outbound_connections() +{ + if (m_target_outbound_peers == 0) + { + LOG_INFO << "Outbound peer dialing disabled (target=0)"; + return; + } + + // Try to connect to peers right away + auto try_connect_peers = [this]() { + size_t outbound = m_outbound_addrs.size(); + if (outbound >= m_target_outbound_peers || m_connections.size() >= m_max_peers) + return; + + size_t needed = m_target_outbound_peers - outbound; + auto good_peers = get_good_peers(needed + 4); // ask for a few extra in case some are already connected + + for (auto& ap : good_peers) + { + if (needed == 0) + break; + // Skip if already connected, already dialing, or banned + if (m_connections.contains(ap.addr) || m_pending_outbound.contains(ap.addr) || is_banned(ap.addr)) + continue; + + LOG_INFO << "[Pool] Dialing outbound peer " << ap.addr.to_string(); + m_pending_outbound.insert(ap.addr); + core::Client::connect(ap.addr); + --needed; + } + }; + + // Initial burst + try_connect_peers(); + + // Periodic maintenance — every 30 seconds, check if we need more outbound peers + m_connect_timer = std::make_unique(m_context, true); + m_connect_timer->start(30, try_connect_peers); +} + +void NodeImpl::prune_shares(const uint256& /*best_share*/) +{ + // Matches p2pool node.py:381-398 tail-dropping exactly: + // - Check each tail: if min(height(head) for heads) < 2*CL+10 → skip + // - Remove ONE child of qualifying tail per iteration + // - Loop up to 1000 times (gradual, not bulk) + // - Also cascade removal to verified + const auto CL = static_cast(m_tracker.m_params->chain_length); + const int32_t min_depth = 2 * CL + 10; + + for (int iter = 0; iter < 1000; ++iter) + { + // p2pool node.py:382-398: find ONE qualifying tail child to remove + uint256 to_remove; + bool found = false; + auto tails_copy = m_tracker.chain.get_tails(); + for (auto& [tail_hash, head_hashes] : tails_copy) + { + // Check min height across ALL heads for this tail + int32_t min_height = std::numeric_limits::max(); + for (auto& hh : head_hashes) { + if (!m_tracker.chain.contains(hh)) continue; + try { + min_height = std::min(min_height, m_tracker.chain.get_height(hh)); + } catch (...) { continue; } + } + if (min_height < min_depth) continue; + + // Find ONE child of this tail (p2pool removes one at a time) + auto& rev = m_tracker.chain.get_reverse(); + auto rev_it = rev.find(tail_hash); + if (rev_it != rev.end() && !rev_it->second.empty()) { + to_remove = *rev_it->second.begin(); + found = true; + break; // one per iteration + } + } + + if (!found) break; + + if (!m_tracker.chain.contains(to_remove)) continue; + // p2pool node.py:393 — safety check: parent must be a tail + auto* idx = m_tracker.chain.get_index(to_remove); + if (!idx) continue; + bool parent_is_tail = m_tracker.chain.get_tails().contains(idx->tail); + if (!parent_is_tail) { + LOG_DEBUG_POOL << "prune skip: parent " << idx->tail.ToString().substr(0,16) + << " not a tail"; + continue; + } + if (m_tracker.verified.contains(to_remove)) + m_tracker.verified.remove(to_remove, /*owns_data=*/false); + m_tracker.chain.remove(to_remove); + } + + // Cache cleanup (kept from original) + if (m_shared_share_hashes.size() > m_max_shared_hashes) + m_shared_share_hashes.clear(); + if (m_known_txs.size() > m_max_known_txs) + m_known_txs.clear(); + if (m_raw_share_cache.size() > m_max_raw_shares) + m_raw_share_cache.clear(); +} + +// (old phases 5-7 removed — replaced by p2pool-style pruning above) + +void NodeImpl::run_think() +{ + // Skip if a think() is already running on the compute thread. + // The atomic flag serializes: only one think() in flight at a time. + if (m_think_running.exchange(true)) { + static int skip_log = 0; + if (skip_log++ % 20 == 0) + LOG_INFO << "[ASYNC-THINK] skipped — compute thread busy (skip_count=" << skip_log << ")"; + return; + } + LOG_INFO << "[ASYNC-THINK] dispatching to compute thread" + << " pending_adds=" << m_pending_adds.size() + << " peers=" << m_peers.size(); + + // Capture block_rel_height fn by value for thread safety + auto block_rel_height = m_block_rel_height_fn + ? m_block_rel_height_fn + : std::function([](uint256) -> int32_t { return 0; }); + + // ── Post think() to the dedicated compute thread ────────────────────── + // The compute thread acquires m_tracker_mutex exclusively, guaranteeing + // no concurrent IO-thread modifications to the tracker. The IO thread + // uses try_to_lock for all tracker access — if the mutex is held, the + // operation is deferred (shares queued, clean_tracker skipped). Network + // I/O, keepalive timers, and stratum continue unimpeded. + // + // Previous attempt at threaded think() caused SIGSEGV because there was + // NO synchronization — prune_shares freed shares while think traversed + // them. The shared_mutex + try_to_lock pattern eliminates this race: + // no mutations occur while the compute thread holds the exclusive lock. + boost::asio::post(m_think_pool, [this, block_rel_height]() { + m_compute_thread_id.store(std::this_thread::get_id(), std::memory_order_relaxed); + + // ── Compute thread: exclusive tracker access ────────────────────── + // Collect results under the lock, then release before posting to IO. + TrackerThinkResult result; + bool best_changed = false; + bool needs_continue = false; + int64_t think_ms = 0; + + try { + auto lock_t0 = std::chrono::steady_clock::now(); + std::unique_lock lock(m_tracker_mutex); // exclusive — IO defers + auto lock_ms = std::chrono::duration_cast( + std::chrono::steady_clock::now() - lock_t0).count(); + if (lock_ms > 10) + LOG_INFO << "[ASYNC-THINK] mutex acquired after " << lock_ms << "ms wait"; + + uint256 prev_block; + uint32_t bits = 0; + + // Bootstrap mode: no verified chain yet → verify everything in one + // pass with unlimited budget. During bootstrap, stratum isn't serving + // real work so no IO callbacks need the tracker lock. Matches p2pool + // where think() runs synchronously on the reactor during initial sync. + bool bootstrap = m_tracker.verified.size() == 0; + LOG_INFO << "[ASYNC-THINK] compute: chain=" << m_tracker.chain.size() + << " verified=" << m_tracker.verified.size() + << " heads=" << m_tracker.chain.get_heads().size() + << " v_heads=" << m_tracker.verified.get_heads().size() + << (bootstrap ? " BOOTSTRAP" : ""); + auto think_t0 = std::chrono::steady_clock::now(); + result = m_tracker.think(block_rel_height, prev_block, bits, bootstrap); + think_ms = std::chrono::duration_cast( + std::chrono::steady_clock::now() - think_t0).count(); + + // Apply tracker-mutating results while still holding the lock: + // - Ban list and download set are NodeImpl members (not tracker), + // but safe here since IO thread doesn't touch them during think. + // - best_share update and top5_heads are consumed by IO-thread + // callbacks, so just record the values for the IO-thread phase. + m_last_top5_heads = std::move(result.top5_heads); + + if (!result.best.IsNull()) { + best_changed = (m_best_share_hash != result.best); + m_best_share_hash = result.best; + } + + // Publish lock-free snapshot for all IO-thread consumers + publish_snapshot(); + + needs_continue = m_tracker.m_think_needs_continue; + + // Flush verified hashes to LevelDB while lock is held + flush_verified_to_leveldb(); + + // Work refresh runs on the IO thread AFTER the lock is released. + // It takes 1-5s (PPLNS + MM coinbase) — must NOT hold exclusive lock + // or it blocks all shared_lock readers (handle_get_share, send_shares) + // and kills peer connections. + + } catch (const std::exception& e) { + LOG_ERROR << "run_think() failed on compute thread: " << e.what(); + } catch (...) { + LOG_ERROR << "run_think() failed on compute thread: unknown error"; + } + // ── Lock released here ──────────────────────────────────────────── + + LOG_INFO << "[ASYNC-THINK] compute done in " << think_ms << "ms" + << " best_changed=" << best_changed + << " needs_continue=" << needs_continue + << " bads=" << result.bad_peer_addresses.size() + << " desired=" << result.desired.size(); + + // ── Post IO-thread phase: work refresh + drain pending ──────────── + // These operations need the IO thread (send to peers, stratum notify). + // The mutex is NOT held — IO thread can acquire it for drain_pending. + boost::asio::post(*m_context, [this, result = std::move(result), + best_changed, needs_continue]() { + try { + // Ban peers that provided invalid/unverifiable shares + auto now = std::chrono::steady_clock::now(); + for (const auto& bad_addr : result.bad_peer_addresses) { + if (bad_addr.port() == 0) + continue; + LOG_WARNING << "run_think: banning peer " << bad_addr.to_string() + << " for unverifiable shares"; + m_ban_list[bad_addr] = now + m_ban_duration; + } + + // Expire old bans + for (auto it = m_ban_list.begin(); it != m_ban_list.end(); ) { + if (it->second <= now) it = m_ban_list.erase(it); + else ++it; + } + + // Clear stale download set (p2pool node.py:108-141) + m_downloading_shares.clear(); + + // Reset fail counts each think() cycle — matches p2pool which + // re-adds desired hashes from scratch every cycle with sleep(1) + // backoff. Permanent blacklisting caused bootstrap stalls: + // the tail parent hash would get 3 empty replies and never be + // requested again, leaving the chain stuck at 1) + std::advance(peer_it, core::random::random_uint256().GetLow64() % m_peers.size()); + download_shares(peer_it->second, hash); + } + } + + // Work refresh on IO thread — NO lock held. Takes 1-5s but doesn't + // block shared_lock readers (handle_get_share, send_shares). + // The tracker is not being modified during this window (think finished, + // pending adds not yet drained), so reading it is safe. + if (best_changed) { + LOG_INFO << "[ASYNC-THINK] IO-phase: best=" << m_best_share_hash.GetHex() + << " work refresh starting"; + auto wr_t0 = std::chrono::steady_clock::now(); + if (m_on_best_share_changed) m_on_best_share_changed(); + auto wr_ms = std::chrono::duration_cast( + std::chrono::steady_clock::now() - wr_t0).count(); + LOG_INFO << "[ASYNC-THINK] IO-phase: work refresh done in " << wr_ms << "ms"; + } else if (result.best.IsNull()) { + LOG_WARNING << "[ASYNC-THINK] IO-phase: result.best is NULL — verified_tails=" + << m_tracker.verified.get_tails().size() + << " verified_heads=" << m_tracker.verified.get_heads().size(); + } + + // ROOT-2: re-advertise our head so a peer that handshook while our + // verified chain was empty can finally download it. On best-change + // push immediately; additionally fire ONE delayed re-advert when the + // verified chain first becomes non-empty (covers peers that connected + // during the empty window and never see a best-change event). + if (best_changed && !m_peers.empty()) { + readvertise_best_share(); + } + if (m_verified_was_empty && m_tracker.verified.size() > 0) { + m_verified_was_empty = false; + if (!m_readvert_timer) + m_readvert_timer = std::make_unique(m_context, false); + m_readvert_timer->start(10, [this]() { readvertise_best_share(); }); + LOG_INFO << "[readvertise] verified chain populated — scheduled " + "10s re-advert (ROOT-2)"; + } + + // Drain share batches queued while think() held the mutex + LOG_INFO << "[ASYNC-THINK] IO-phase: draining " << m_pending_adds.size() + << " pending batches, peers=" << m_peers.size(); + drain_pending_adds(); + + // Clear flag AFTER drain — prevents new think() from starting + // while deferred shares are still being added to the tracker. + m_think_running.store(false); + LOG_INFO << "[ASYNC-THINK] cycle complete — IO thread free"; + + // If verification budget was exhausted, schedule continuation. + // The IO thread processes network I/O between verification batches. + if (needs_continue) { + boost::asio::post(*m_context, [this]() { run_think(); }); + } + } catch (const std::exception& e) { + LOG_ERROR << "run_think() IO phase failed: " << e.what(); + m_think_running.store(false); + } catch (...) { + LOG_ERROR << "run_think() IO phase failed: unknown error"; + m_think_running.store(false); + } + }); + }); +} + +// ── Drain pending share batches ───────────────────────────────────────── +// Called on the IO thread after think() releases m_tracker_mutex. +// Processes share batches that arrived while think was running. +void NodeImpl::drain_pending_adds() +{ + if (m_pending_adds.empty()) + return; + + // Take ownership of pending queue + auto pending = std::move(m_pending_adds); + m_pending_adds.clear(); + + LOG_INFO << "[drain] Processing " << pending.size() + << " deferred share batches"; + + for (auto& batch : pending) { + processing_shares_phase2(*batch.data, batch.addr); + } +} + +// ── Heartbeat logging ─────────────────────────────────────────────────── +// p2pool-style status lines: chain height, local/pool hashrate, orphan/DOA. +// Runs on a separate 30s timer — diagnostic only, not consensus-critical. +// Previously ran inside run_think() blocking the IO thread on every share. +void NodeImpl::heartbeat_log() +{ + // Try shared lock — if think is running, skip this heartbeat + std::shared_lock lock(m_tracker_mutex, std::try_to_lock); + if (!lock.owns_lock()) + return; + + auto chain_sz = m_tracker.chain.size(); + auto verified = m_tracker.verified.size(); + auto peers = m_peers.size(); + + int incoming_peers = 0; + for (auto& [nonce, peer] : m_peers) { + if (peer && !m_outbound_addrs.contains(peer->addr())) + ++incoming_peers; + } + + int height = 0; + if (!m_best_share_hash.IsNull()) { + if (m_tracker.chain.contains(m_best_share_hash)) + height = m_tracker.chain.get_height(m_best_share_hash); + } + if (height == 0 && verified > 0) + height = static_cast(verified); + + // L1: chain status + LOG_INFO << "c2pool: " << height << " shares in chain (" + << verified << " verified/" << chain_sz << " total)" + << " Peers: " << peers << " (" << incoming_peers << " incoming)"; + + // L2: local hashrate + uint32_t share_bits = 0; + if (!m_best_share_hash.IsNull() && m_tracker.chain.contains(m_best_share_hash)) { + m_tracker.chain.get(m_best_share_hash).share.invoke([&](auto* s) { + share_bits = s->m_bits; + }); + } + if (m_local_rate_stats_fn) { + auto stats = m_local_rate_stats_fn(); + std::ostringstream local_line; + local_line << " Local: " << format_hashrate(stats.hashrate); + if (stats.effective_dt > 0) + local_line << " in last " << format_duration(stats.effective_dt); + local_line << " Local dead on arrival: " + << format_binomial_conf(stats.dead_datums, stats.total_datums); + if (stats.hashrate > 0 && share_bits != 0) { + auto share_target = chain::bits_to_target(share_bits); + auto share_aps = chain::target_to_average_attempts(share_target); + double ets = static_cast(share_aps.GetLow64()) / stats.hashrate; + local_line << " Expected time to share: " << format_duration(ets); + } + LOG_INFO << local_line.str(); + } else if (m_local_hashrate_fn) { + LOG_INFO << " Local: " << format_hashrate(m_local_hashrate_fn()); + } + + // L3: orphan/DOA/stale stats (chain walk — diagnostic only) + int orphan_count = 0, doa_count = 0, total_recent = 0; + uint256 walk_start = m_best_share_hash; + if (walk_start.IsNull() || !m_tracker.chain.contains(walk_start)) { + auto& vheads = m_tracker.verified.get_heads(); + if (!vheads.empty()) + walk_start = vheads.begin()->first; + } + if (!walk_start.IsNull() && m_tracker.chain.contains(walk_start)) { + int window = std::min(height, static_cast( + std::min(size_t(3600) / m_tracker.m_params->share_period, size_t(height)))); + if (window > 0) { + auto walkable = m_tracker.chain.get_height(walk_start); + auto walk_n = std::min(window, walkable); + if (walk_n > 0) { + try { + auto view = m_tracker.chain.get_chain(walk_start, walk_n); + for (auto [hash, data] : view) { + data.share.invoke([&](auto* s) { + if (s->m_stale_info == dgb::StaleInfo::orphan) ++orphan_count; + else if (s->m_stale_info == dgb::StaleInfo::doa) ++doa_count; + }); + ++total_recent; + } + } catch (...) {} + } + } + } + double stale_prop = total_recent > 0 + ? static_cast(orphan_count + doa_count) / total_recent : 0.0; + + { + std::ostringstream shares_line; + shares_line << " Shares: " << chain_sz + << " (" << orphan_count << " orphan, " << doa_count << " dead)" + << " Stale rate: " << format_binomial_conf(orphan_count + doa_count, total_recent) + << " Efficiency: " << format_binomial_conf_efficiency(orphan_count + doa_count, total_recent, stale_prop); + + if (m_current_pplns_fn) { + auto outputs = m_current_pplns_fn(); + uint64_t my_payout = 0; + std::set local_scripts; + if (!m_node_payout_script_hex.empty()) + local_scripts.insert(m_node_payout_script_hex); + if (m_local_miner_scripts_fn) { + for (const auto& s : m_local_miner_scripts_fn()) + local_scripts.insert(s); + } + for (const auto& [script, amount] : outputs) { + if (local_scripts.count(script)) + my_payout += amount; + } + if (my_payout > 0) { + double coins = static_cast(my_payout) / 1e8; + shares_line << " Current payout: (" << std::fixed << std::setprecision(4) + << coins << ")=" << coins << " tDGB"; + } + } + + shares_line << " [heads=" << m_tracker.chain.get_heads().size() + << " v_heads=" << m_tracker.verified.get_heads().size() + << " rss=" << get_rss_mb() << "MB]"; + LOG_INFO << shares_line.str(); + } + + // L4: pool hashrate + ETB + if (height > 2) { + try { + auto aps = m_tracker.get_pool_attempts_per_second( + m_best_share_hash, + std::min(height - 1, static_cast(m_tracker.m_params->target_lookbehind)), + /*min_work=*/false); + double pool_hs = static_cast(aps.GetLow64()); + double real_pool_hs = (stale_prop < 0.999 && pool_hs > 0) + ? pool_hs / (1.0 - stale_prop) : pool_hs; + double etb_secs = 0; + uint32_t block_bits = 0; + if (!m_best_share_hash.IsNull() && m_tracker.chain.contains(m_best_share_hash)) { + m_tracker.chain.get(m_best_share_hash).share.invoke([&](auto* s) { + block_bits = s->m_min_header.m_bits; + }); + } + if (real_pool_hs > 0 && block_bits != 0) { + auto block_target = chain::bits_to_target(block_bits); + auto block_aps = chain::target_to_average_attempts(block_target); + etb_secs = static_cast(block_aps.GetLow64()) / real_pool_hs; + if (block_aps.IsNull() && !block_target.IsNull()) + etb_secs = 1e18; + } + LOG_INFO << " Pool: " << format_hashrate(real_pool_hs) + << " Stale rate: " << std::fixed << std::setprecision(1) + << (100.0 * stale_prop) << "% Expected time to block: " + << format_duration(etb_secs); + } catch (...) {} + } +} + +// Periodic maintenance: eat stale heads, drop tails. +// Direct translation of p2pool node.py:355-402 clean_tracker(). +// +// Runs on the compute thread under exclusive lock — matching p2pool where +// clean_tracker + think() execute on the same reactor thread. Chain +// modifications (remove shares) MUST NOT happen concurrently with think(). +void NodeImpl::clean_tracker() +{ + // Prevent concurrent clean_tracker (timer re-entry safety). + if (m_clean_running.exchange(true)) + return; + + // Skip if think() is already in flight — the 5s timer will retry. + if (m_think_running.load()) { + m_clean_running.store(false); + return; + } + + // Post the entire clean_tracker body to the compute thread. + // This guarantees chain modifications happen under exclusive lock, + // never concurrent with think() or IO-thread reads. + m_think_running.store(true); // block think() re-entry during clean + + boost::asio::post(m_think_pool, [this]() { + m_compute_thread_id.store(std::this_thread::get_id(), std::memory_order_relaxed); + + bool clean_best_changed = false; + bool bootstrap = false; + try { + std::unique_lock lock(m_tracker_mutex); // exclusive + + // Step 1: Run think() inline (already holds the lock) + { + auto block_rel_height = m_block_rel_height_fn + ? m_block_rel_height_fn + : std::function([](uint256) -> int32_t { return 0; }); + + uint256 prev_block; + uint32_t bits = 0; + + bootstrap = m_tracker.verified.size() == 0; + LOG_INFO << "[CLEAN] think+clean starting on compute thread: chain=" + << m_tracker.chain.size() << " verified=" << m_tracker.verified.size() + << (bootstrap ? " BOOTSTRAP" : ""); + auto think_t0 = std::chrono::steady_clock::now(); + auto result = m_tracker.think(block_rel_height, prev_block, bits, bootstrap); + auto think_ms = std::chrono::duration_cast( + std::chrono::steady_clock::now() - think_t0).count(); + + m_last_top5_heads = std::move(result.top5_heads); + + if (!result.best.IsNull()) { + m_best_share_hash = result.best; + } + + flush_verified_to_leveldb(); + // Work refresh deferred to IO thread (after lock release) + } + + // Steps 2-3: Prune (still holding exclusive lock) + auto now_sec = static_cast(std::time(nullptr)); + auto CL = static_cast(m_tracker.m_params->chain_length); + + // Step 2: Eat stale heads (p2pool node.py:358-378) + // Three guards protect useful heads: + // 1. Top-5 scored heads (decorated_heads[-5:]) + // 2. Heads seen < 300s ago + // 3. Unverified heads whose tail has recent child activity (< 120s) + if (!m_last_top5_heads.empty()) // p2pool node.py:359: if decorated_heads: + { + // Build top-5 set for O(1) lookup + std::set top5_set(m_last_top5_heads.begin(), m_last_top5_heads.end()); + + for (int iter = 0; iter < 1000; ++iter) + { + std::vector to_remove; + auto heads_copy = m_tracker.chain.get_heads(); + for (auto& [head_hash, tail_hash] : heads_copy) + { + if (!m_tracker.chain.contains(head_hash)) continue; + + // Guard 1: keep top-5 scored heads (p2pool node.py:363) + if (top5_set.count(head_hash)) continue; + + // Guard 2: keep heads seen < 300s ago (p2pool node.py:366) + auto* idx = m_tracker.chain.get_index(head_hash); + if (!idx || idx->time_seen > now_sec - 300) continue; + + // Guard 3: keep unverified heads with recent tail activity (p2pool node.py:369) + if (!m_tracker.verified.contains(head_hash)) + { + auto& rev = m_tracker.chain.get_reverse(); + auto rev_it = rev.find(tail_hash); + if (rev_it != rev.end() && !rev_it->second.empty()) + { + int64_t max_child_ts = 0; + for (const auto& child : rev_it->second) + { + auto* cidx = m_tracker.chain.get_index(child); + if (cidx && cidx->time_seen > max_child_ts) + max_child_ts = cidx->time_seen; + } + if (max_child_ts > now_sec - 120) continue; + } + } + + to_remove.push_back(head_hash); + } + + if (to_remove.empty()) break; + + size_t _dh_idx = 0; + for (const auto& h : to_remove) + { + try { + // DIAG: per-remove marker so a freeze inside chain.contains() + // or verified.remove() leaves the last touched hash in the log. + std::fprintf(stderr, + "[drop-heads-PRE] drop_idx=%zu/%zu hash=%.16s " + "chain_size=%zu\n", + _dh_idx, to_remove.size(), h.GetHex().c_str(), + m_tracker.chain.size()); + std::fflush(stderr); + ++_dh_idx; + + if (m_tracker.verified.contains(h)) + m_tracker.verified.remove(h, /*owns_data=*/false); + if (m_tracker.chain.contains(h)) + m_tracker.chain.remove(h); + } catch (...) {} + } + } + } + + // Step 3: Drop tails — remove ALL children of qualifying tails. + // Exact translation of p2pool node.py:382-398. + // p2pool has NO best-chain protection — the 2*CHAIN_LENGTH+10 threshold + // ensures only shares far beyond the PPLNS window are removed. + { + int total_dropped = 0; + for (int iter = 0; iter < 1000; ++iter) + { + std::vector to_remove; + auto tails_copy = m_tracker.chain.get_tails(); + for (auto& [tail_hash, head_hashes] : tails_copy) + { + int32_t min_height = 0; // default 0 → skip if no valid heads + for (auto& hh : head_hashes) { + if (!m_tracker.chain.contains(hh)) continue; + auto h = m_tracker.chain.get_height(hh); + if (min_height == 0 || h < min_height) + min_height = h; + } + if (min_height < 2 * CL + 10) continue; + + if (iter == 0) { + LOG_WARNING << "[drop-tails-QUALIFY] tail=" << tail_hash.GetHex().substr(0,16) + << " min_height=" << min_height << " threshold=" << (2*CL+10) + << " n_heads=" << head_hashes.size() + << " chain_size=" << m_tracker.chain.size(); + } + + // p2pool node.py:386: to_remove.update(tracker.reverse.get(tail, set())) + auto& rev = m_tracker.chain.get_reverse(); + auto rev_it = rev.find(tail_hash); + if (rev_it != rev.end()) { + for (const auto& child : rev_it->second) + to_remove.push_back(child); + } + } + + if (to_remove.empty()) break; + + // p2pool node.py:392-398 + size_t _drop_idx = 0; + for (const auto& aftertail : to_remove) + { + try { + // DIAG: contabo 2026-04-21 froze 40s inside + // unordered_map::contains() on an aftertail at bucket 2332. + // If we spin here again this is the last line flushed — + // gives us the iter index, which aftertail triggered it, + // and the queue size to correlate with the freeze. + std::fprintf(stderr, + "[drop-tails-PRE] iter=%d drop_idx=%zu/%zu aftertail=%.16s " + "chain_size=%zu\n", + iter, _drop_idx, to_remove.size(), + aftertail.GetHex().c_str(), + m_tracker.chain.size()); + std::fflush(stderr); + ++_drop_idx; + + if (!m_tracker.chain.contains(aftertail)) continue; + // p2pool node.py:393: if items[aftertail].previous_hash not in tails: continue + auto* idx = m_tracker.chain.get_index(aftertail); + if (!idx) continue; + if (!m_tracker.chain.get_tails().count(idx->tail)) { + continue; + } + if (m_tracker.verified.contains(aftertail)) + m_tracker.verified.remove(aftertail, /*owns_data=*/false); + m_tracker.chain.remove(aftertail); + ++total_dropped; + } catch (...) {} + } + } + if (total_dropped > 0) + LOG_INFO << "[clean-drop-tails] dropped " << total_dropped << " shares" + << " chain_size=" << m_tracker.chain.size() + << " heads=" << m_tracker.chain.get_heads().size(); + } + + // Step 4: re-score after pruning (inline, already holding lock) + { + auto block_rel_height = m_block_rel_height_fn + ? m_block_rel_height_fn + : std::function([](uint256) -> int32_t { return 0; }); + uint256 prev_block; + uint32_t bits = 0; + auto result = m_tracker.think(block_rel_height, prev_block, bits, bootstrap); + m_last_top5_heads = std::move(result.top5_heads); + if (!result.best.IsNull()) { + clean_best_changed = (m_best_share_hash != result.best); + m_best_share_hash = result.best; + } + publish_snapshot(); + flush_verified_to_leveldb(); + // Work refresh deferred to IO thread (after lock release) + } + + // Step 5: Flush pruned shares from LevelDB (p2pool main.py:269-270) + if (!m_removal_flush_buf.empty() && m_storage && m_storage->is_available()) + { + auto count = m_removal_flush_buf.size(); + if (m_storage->remove_shares_batch(m_removal_flush_buf)) + LOG_INFO << "[clean-leveldb] removed " << count << " pruned shares from LevelDB"; + else + LOG_WARNING << "[clean-leveldb] batch remove failed, count=" << count; + m_removal_flush_buf.clear(); + } + + // Orphan/fork diagnostics — understand chain topology + { + auto& chain = m_tracker.chain; + auto& verified = m_tracker.verified; + size_t total_heads = chain.get_heads().size(); + size_t verified_heads = verified.get_heads().size(); + size_t total_tails = chain.get_tails().size(); + + // Count c2pool's own shares in verified vs total + int local_in_chain = 0, local_in_verified = 0; + int total_in_chain = 0; + for (auto& [head_hash, tail_hash] : chain.get_heads()) { + auto height = chain.get_height(head_hash); + total_in_chain += height; + // Check if this head is in verified + if (verified.contains(head_hash)) { + // Walk backward and count local shares + // (expensive — only sample first 50) + } + } + + static int diag_count = 0; + if (diag_count++ % 6 == 0) { // every 30s (5s * 6) + LOG_INFO << "[FORK-DIAG] heads=" << total_heads + << " verified_heads=" << verified_heads + << " tails=" << total_tails + << " chain=" << chain.size() + << " verified=" << verified.size() + << " gap=" << (chain.size() - verified.size()) + << " best_h=" << (m_best_share_hash.IsNull() ? 0 + : (verified.contains(m_best_share_hash) + ? verified.get_height(m_best_share_hash) : -1)) + << " chain_h=" << (m_best_share_hash.IsNull() ? 0 + : (chain.contains(m_best_share_hash) + ? chain.get_height(m_best_share_hash) : -1)) + << " best=" << (m_best_share_hash.IsNull() + ? std::string("null") + : m_best_share_hash.GetHex().substr(0, 16)); + } + } + + } catch (const std::exception& e) { + LOG_ERROR << "[CLEAN] failed on compute thread: " << e.what(); + } catch (...) { + LOG_ERROR << "[CLEAN] failed on compute thread: unknown error"; + } + // Lock released — post work refresh + drain to IO thread. + // Work refresh (1-5s) runs WITHOUT any lock so shared_lock readers + // (handle_get_share, send_shares) are never blocked. + boost::asio::post(*m_context, [this, clean_best_changed]() { + if (clean_best_changed && m_on_best_share_changed) { + LOG_INFO << "[CLEAN] IO-phase: work refresh (best changed)"; + m_on_best_share_changed(); + } + drain_pending_adds(); + m_think_running.store(false); + m_clean_running.store(false); + }); + }); // end of m_think_pool lambda +} + +bool NodeImpl::is_whitelisted(const NetService& addr) const +{ + const std::string ip = addr.address(); + if (m_whitelist_ips.contains(ip)) return true; + if (m_whitelist_hosts.contains(addr)) return true; + return false; +} + +bool NodeImpl::is_banned(const NetService& addr) const +{ + // Whitelist bypass: permanent dial targets are immune to bans. + if (is_whitelisted(addr)) return false; + + auto now = std::chrono::steady_clock::now(); + auto it = m_ban_list.find(addr); + if (it != m_ban_list.end() && it->second > now) return true; + + auto ip_it = m_ip_ban_list.find(addr.address()); + if (ip_it != m_ip_ban_list.end() && ip_it->second > now) return true; + + return false; +} + +// ── Admin API implementation ────────────────────────────────────────── + +void NodeImpl::set_whitelist_path(const std::string& path) +{ + m_whitelist_path = path; + if (!path.empty()) load_whitelist_from_disk(); +} + +void NodeImpl::load_whitelist_from_disk() +{ + if (m_whitelist_path.empty()) return; + std::ifstream f(m_whitelist_path); + if (!f) return; + try { + nlohmann::json j; + f >> j; + if (!j.contains("entries") || !j["entries"].is_array()) return; + for (const auto& e : j["entries"]) { + if (!e.contains("host") || !e.contains("port")) continue; + std::string host = e["host"].get(); + uint16_t port = e["port"].get(); + m_whitelist_ips.insert(host); + m_whitelist_hosts.insert(NetService(host, port)); + } + LOG_INFO << "[Pool] Loaded " << m_whitelist_hosts.size() + << " whitelist entries from " << m_whitelist_path; + } catch (const std::exception& e) { + LOG_WARNING << "[Pool] Failed to parse whitelist " << m_whitelist_path + << ": " << e.what(); + } +} + +void NodeImpl::save_whitelist_to_disk() const +{ + if (m_whitelist_path.empty()) return; + try { + nlohmann::json j; + j["version"] = 1; + auto arr = nlohmann::json::array(); + auto now_unix = std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()).count(); + for (const auto& host : m_whitelist_hosts) { + arr.push_back({ + {"host", host.address()}, + {"port", host.port()}, + {"added_unix", now_unix} + }); + } + j["entries"] = arr; + std::string tmp = m_whitelist_path + ".new"; + { + std::ofstream f(tmp); + f << j.dump(2); + } + std::filesystem::rename(tmp, m_whitelist_path); + } catch (const std::exception& e) { + LOG_WARNING << "[Pool] Failed to persist whitelist: " << e.what(); + } +} + +static nlohmann::json build_bans_json( + const std::map& peer_bans, + const std::map& ip_bans) +{ + auto now = std::chrono::steady_clock::now(); + auto arr = nlohmann::json::array(); + for (const auto& [addr, expiry] : peer_bans) { + if (expiry <= now) continue; + auto secs = std::chrono::duration_cast(expiry - now).count(); + arr.push_back({{"host", addr.address()}, {"port", addr.port()}, + {"expires_in_sec", secs}, {"source", "auto"}}); + } + for (const auto& [ip, expiry] : ip_bans) { + if (expiry <= now) continue; + auto secs = std::chrono::duration_cast(expiry - now).count(); + arr.push_back({{"host", ip}, {"port", 0}, + {"expires_in_sec", secs}, {"source", "admin"}}); + } + return arr; +} + +static nlohmann::json build_whitelist_json(const std::set& hosts) +{ + auto arr = nlohmann::json::array(); + for (const auto& h : hosts) + arr.push_back({{"host", h.address()}, {"port", h.port()}}); + return arr; +} + +nlohmann::json NodeImpl::admin_list_bans() const +{ + return {{"ok", true}, {"bans", build_bans_json(m_ban_list, m_ip_ban_list)}}; +} + +nlohmann::json NodeImpl::admin_ban_ip(const std::string& ip, int duration_sec) +{ + if (ip.empty()) + return {{"ok", false}, {"error", "ip required"}}; + int dur = duration_sec > 0 ? duration_sec : static_cast(m_ban_duration.count()); + auto expiry = std::chrono::steady_clock::now() + std::chrono::seconds(dur); + m_ip_ban_list[ip] = expiry; + LOG_INFO << "[Pool] Admin ban: " << ip << " for " << dur << "s"; + return {{"ok", true}, {"action", "ban"}, {"target", ip}, + {"duration_sec", dur}, + {"bans", build_bans_json(m_ban_list, m_ip_ban_list)}}; +} + +nlohmann::json NodeImpl::admin_unban_ip(const std::string& ip) +{ + if (ip.empty()) + return {{"ok", false}, {"error", "ip required"}}; + size_t removed = m_ip_ban_list.erase(ip); + for (auto it = m_ban_list.begin(); it != m_ban_list.end(); ) { + if (it->first.address() == ip) { it = m_ban_list.erase(it); ++removed; } + else ++it; + } + LOG_INFO << "[Pool] Admin unban: " << ip << " (" << removed << " entries removed)"; + return {{"ok", true}, {"action", "unban"}, {"target", ip}, + {"removed", removed}, + {"bans", build_bans_json(m_ban_list, m_ip_ban_list)}}; +} + +nlohmann::json NodeImpl::admin_list_whitelist() const +{ + return {{"ok", true}, {"whitelist", build_whitelist_json(m_whitelist_hosts)}}; +} + +nlohmann::json NodeImpl::admin_whitelist_add(const std::string& host, uint16_t port) +{ + if (host.empty() || port == 0) + return {{"ok", false}, {"error", "host and port required"}}; + NetService addr(host, port); + m_whitelist_ips.insert(host); + m_whitelist_hosts.insert(addr); + // Remove any existing ban for this host — whitelisting is an explicit override. + m_ip_ban_list.erase(host); + for (auto it = m_ban_list.begin(); it != m_ban_list.end(); ) { + if (it->first.address() == host) it = m_ban_list.erase(it); + else ++it; + } + save_whitelist_to_disk(); + LOG_INFO << "[Pool] Whitelisted " << addr.to_string(); + + // Proactive dial: if not already connected, try immediately. + if (!m_connections.contains(addr) && !m_pending_outbound.contains(addr)) { + LOG_INFO << "[Pool] Dialing whitelisted peer " << addr.to_string(); + m_pending_outbound.insert(addr); + core::Client::connect(addr); + } + return {{"ok", true}, {"action", "whitelist_add"}, + {"target", addr.to_string()}, + {"whitelist", build_whitelist_json(m_whitelist_hosts)}}; +} + +nlohmann::json NodeImpl::admin_whitelist_remove(const std::string& host, uint16_t port) +{ + if (host.empty() || port == 0) + return {{"ok", false}, {"error", "host and port required"}}; + NetService addr(host, port); + size_t removed_h = m_whitelist_hosts.erase(addr); + // Only remove the IP from whitelist if no other host:port remains for it. + bool other_on_same_ip = std::any_of( + m_whitelist_hosts.begin(), m_whitelist_hosts.end(), + [&](const NetService& n) { return n.address() == host; }); + if (!other_on_same_ip) m_whitelist_ips.erase(host); + if (removed_h) save_whitelist_to_disk(); + LOG_INFO << "[Pool] De-whitelisted " << addr.to_string(); + return {{"ok", true}, {"action", "whitelist_remove"}, + {"target", addr.to_string()}, {"removed", removed_h}, + {"whitelist", build_whitelist_json(m_whitelist_hosts)}}; +} + +nlohmann::json NodeImpl::admin_list_peers() const +{ + auto arr = nlohmann::json::array(); + for (const auto& [addr, peer] : m_connections) { + bool outbound = m_outbound_addrs.contains(addr); + arr.push_back({ + {"host", addr.address()}, + {"port", addr.port()}, + {"outbound", outbound}, + {"whitelisted", is_whitelisted(addr)} + }); + } + return {{"ok", true}, {"peers", arr}}; +} + +nlohmann::json NodeImpl::admin_drop_peer(const std::string& ip) +{ + if (ip.empty()) + return {{"ok", false}, {"error", "ip required"}}; + std::vector targets; + for (const auto& [addr, peer] : m_connections) + if (addr.address() == ip) targets.push_back(addr); + for (const auto& addr : targets) + close_connection(addr); + LOG_INFO << "[Pool] Admin drop " << ip << " (" << targets.size() << " connections)"; + return {{"ok", true}, {"action", "drop"}, {"target", ip}, + {"dropped", targets.size()}}; +} + +nlohmann::json NodeImpl::admin_dial_peer(const std::string& host, uint16_t port) +{ + if (host.empty() || port == 0) + return {{"ok", false}, {"error", "host and port required"}}; + NetService addr(host, port); + if (m_connections.contains(addr)) + return {{"ok", true}, {"action", "dial"}, {"target", addr.to_string()}, + {"note", "already connected"}}; + if (m_pending_outbound.contains(addr)) + return {{"ok", true}, {"action", "dial"}, {"target", addr.to_string()}, + {"note", "dial already pending"}}; + m_pending_outbound.insert(addr); + core::Client::connect(addr); + return {{"ok", true}, {"action", "dial"}, {"target", addr.to_string()}}; +} + +void NodeImpl::set_rss_limit_mb(long mb) +{ + g_rss_limit_mb = mb; } } // namespace dgb diff --git a/src/impl/dgb/node.hpp b/src/impl/dgb/node.hpp index d8f777502..f1fa1eeb3 100644 --- a/src/impl/dgb/node.hpp +++ b/src/impl/dgb/node.hpp @@ -1,45 +1,639 @@ #pragma once -/// DigiByte Scrypt P2Pool Node -/// -/// DGB Scrypt uses identical share format and PoW to Litecoin Scrypt. -/// The node implementation reuses the LTC share validation, tracker, -/// and protocol handlers — only the pool/coin configuration differs. -/// -/// When c2pool is started with --net digibyte: -/// - Pool network params from dgb::PoolConfig (port 5024, share period 25s) -/// - Coin params from dgb::CoinParams (port 12024, 15s blocks, D-prefix addresses) -/// - GBT requests include rules=["scrypt"] for DGB multi-algo filtering -/// - AuxPoW merged mining works identically (DOGE, PEP, BELLS, etc.) - #include "config_pool.hpp" #include "config_coin.hpp" -#include +#include +#include "params.hpp" +#include "share.hpp" +#include "share_tracker.hpp" +#include "peer.hpp" +#include "messages.hpp" + +#include +#include +#include +#include +#include +#include +#include -// DGB Scrypt shares use the same format as LTC shares. -// The share type, validation logic, tracker, and protocol are shared. -// Only config_pool and config_coin differ between LTC and DGB networks. +#include +#include +#include +#include +#include +#include namespace dgb { -// Type aliases — DGB uses the same share format as LTC -// These will be used when instantiating the node template -using PoolConfigType = dgb::PoolConfig; -using CoinParamsType = dgb::CoinParams; +// DGB config aggregator. LTC defines this in config.hpp; the DGB tree splits +// config into config_pool.hpp + config_coin.hpp with no aggregator header, so +// the alias lives here (the sole consumer is the node template below). +using Config = core::Config; + +struct HandleSharesData; +struct ShareReplyData +{ + std::vector m_items; + std::vector m_raw_items; +}; + +class NodeImpl : public pool::BaseNode +{ + // Async share downloader: + // ID = uint256 (matches sharereq id to sharereply id) + // RESPONSE = parsed shares plus their original raw payloads + // REQUEST args: req_id, peer, hashes, parents, stops + using share_getter_t = ReplyMatcher::ID + ::RESPONSE + ::REQUEST, uint64_t, std::vector>; + +protected: + core::CoinParams m_coin_params; + dgb::Handler m_handler; + share_getter_t m_share_getter; + ShareTracker m_tracker; + std::unique_ptr m_storage; + + // Global pool of known transactions, populated by remember_tx and coin daemon. + // Protocol handlers look up tx hashes here when processing shares. + std::map m_known_txs; + + // Thread pool for parallel share_init_verify (scrypt CPU work). + // Keeps expensive crypto off the io_context thread. + boost::asio::thread_pool m_verify_pool{4}; + + // ── Async compute pipeline ────────────────────────────────────────── + // think() runs on m_think_pool (1 thread) holding m_tracker_mutex exclusively. + // The IO thread NEVER calls lock() — only try_to_lock(). If the mutex is + // held by the compute thread, the IO thread defers the operation and continues + // processing network I/O, keepalive timers, and stratum. This eliminates the + // event loop freeze that previously caused all peers to timeout simultaneously. + // + // Synchronization contract: + // Compute thread: unique_lock(m_tracker_mutex) — exclusive, blocking + // IO thread reads: shared_lock(try_to_lock) — non-blocking, skip if busy + // IO thread writes: unique_lock(try_to_lock) — non-blocking, queue if busy + boost::asio::thread_pool m_think_pool{1}; + std::atomic m_think_running{false}; + std::atomic m_clean_running{false}; + mutable std::shared_mutex m_tracker_mutex; + + // ── Lock-free stats snapshot ───────────────────────────────────────── + // Published by think() on the compute thread under m_tracker_mutex. + // Read by ALL consumers (sync_status, loading page, global_stats, + // graph data, etc.) WITHOUT needing the tracker lock. + // + // This is the c2pool equivalent of p2pool's reactor-thread stats + // variables: updated atomically each think() cycle, always available. + // Eliminates the systemic "0/empty" data problem where 20+ callbacks + // returned stale data whenever think() held the exclusive lock. + struct TrackerSnapshot { + int chain_count{0}; + int verified_count{0}; + int head_count{0}; + int orphan_shares{0}; + int dead_shares{0}; + int fork_count{0}; + double pool_hashrate{0}; + }; + void publish_snapshot() { + TrackerSnapshot s; + s.chain_count = static_cast(m_tracker.chain.size()); + s.verified_count = static_cast(m_tracker.verified.size()); + s.head_count = static_cast(m_tracker.chain.get_heads().size()); + s.fork_count = s.head_count; + std::lock_guard lock(m_snapshot_mutex); + m_snapshot = s; + } + mutable std::mutex m_snapshot_mutex; + TrackerSnapshot m_snapshot; + + // Identity of the compute thread (m_think_pool's single thread). + // Used by TrackerReadGuard to skip shared_lock when the caller is + // already on the compute thread (which holds exclusive lock). + std::atomic m_compute_thread_id{}; + + // Pending share batches queued while think() holds the mutex. + // Drained on the IO thread after think() releases the lock. + // Uses unique_ptr because HandleSharesData is forward-declared here. + struct PendingShareBatch { + std::unique_ptr data; + NetService addr; + }; + std::vector m_pending_adds; + + // Top-5 scored heads from last think() — used by clean_tracker() + // to protect the best chains from head pruning (p2pool node.py:363). + std::vector m_last_top5_heads; + + // Buffer of newly verified share hashes, flushed to LevelDB periodically + std::vector m_verified_flush_buf; + + // Buffer of pruned share hashes, batch-deleted from LevelDB after clean_tracker() + std::vector m_removal_flush_buf; + +public: + NodeImpl() + : m_coin_params(dgb::make_coin_params(false)), + m_share_getter(nullptr, + [](uint256, peer_ptr, std::vector, uint64_t, std::vector){}) + { + m_tracker.m_params = &m_coin_params; + } + + NodeImpl(boost::asio::io_context* ctx, config_t* config) + : m_coin_params(dgb::make_coin_params(config->m_testnet)), + base_t(ctx, config), + m_share_getter(ctx, + [](uint256 req_id, peer_ptr to_peer, + std::vector hashes, uint64_t parents, + std::vector stops) + { + auto rmsg = dgb::message_sharereq::make_raw(req_id, hashes, parents, stops); + to_peer->write(std::move(rmsg)); + }, + 15) // p2pool p2p.py:80: timeout=15 for share requests + { + m_tracker.m_params = &m_coin_params; + + // Seed addr store with hardcoded bootstrap peers + m_addrs.load(config->pool()->m_bootstrap_addrs); + // Randomise our nonce so we detect self-connections + std::mt19937_64 rng(std::random_device{}()); + m_nonce = rng(); + // Route m_chain (used by BaseNode) to the tracker's main chain + m_chain = &m_tracker.chain; + + // Open LevelDB storage and load any persisted shares + std::string net_name = config->m_testnet ? "litecoin_testnet" : "litecoin"; + m_storage = std::make_unique(net_name); + load_persisted_shares(); + + // Wire up verified-hash persistence callback (p2pool known_verified pattern) + m_tracker.m_on_share_verified = [this](const uint256& hash) { + m_verified_flush_buf.push_back(hash); + if (m_verified_flush_buf.size() >= 50) + flush_verified_to_leveldb(); + }; + + // Wire up share removal → LevelDB cleanup (p2pool main.py:269-270) + // Buffer removals; clean_tracker() flushes after drop-tails. + // Safe on crash: unflushed shares get pruned at next startup by load_persisted_shares(). + m_tracker.chain.on_removed([this](const uint256& hash) { + m_removal_flush_buf.push_back(hash); + }); + } + + // INetwork: Pool node does not initiate disconnect — peer connections + // manage their own lifecycle via close_connection()/error() below. + void disconnect() override { } + void connected(std::shared_ptr socket) override; + + // ICommunicator (override BaseNode to track outbound lifecycle): + void error(const message_error_type& err, const NetService& service, const std::source_location where = std::source_location::current()) override; + void close_connection(const NetService& service) override; + + // BaseNode: + void send_ping(peer_ptr peer) override; + std::optional handle_version(std::unique_ptr rmsg, peer_ptr peer) override; + + // ltc + void send_version(peer_ptr peer); + void processing_shares(HandleSharesData& data, NetService addr); + void processing_shares_phase2(HandleSharesData& data, NetService addr); + /// Direct tracker access — compute-thread-only (already holds exclusive lock) + /// or startup code (before compute thread exists). + /// IO-thread code MUST use read_tracker() instead. + ShareTracker& tracker() { return m_tracker; } + const core::CoinParams& coin_params() const { return m_coin_params; } + + /// RAII guard for IO-thread tracker reads. + /// - IO thread: acquires shared_lock(try_to_lock). Returns falsy if busy. + /// - Compute thread: skips locking (exclusive already held). Always truthy. + /// Guard lifetime = lock lifetime. No way to hold a dangling reference. + class TrackerReadGuard { + std::shared_lock lock_; + ShareTracker& tracker_; + bool ok_; + public: + TrackerReadGuard(std::shared_mutex& mtx, ShareTracker& t, bool on_compute) + : lock_(mtx, std::defer_lock), tracker_(t) + { + if (on_compute) { + ok_ = true; // exclusive lock already held by caller + } else { + ok_ = lock_.try_lock(); + } + } + TrackerReadGuard(TrackerReadGuard&&) = default; + TrackerReadGuard(const TrackerReadGuard&) = delete; + TrackerReadGuard& operator=(const TrackerReadGuard&) = delete; + TrackerReadGuard& operator=(TrackerReadGuard&&) = default; + + explicit operator bool() const { return ok_; } + ShareTracker& operator*() { return tracker_; } + ShareTracker* operator->() { return &tracker_; } + }; + + /// True if the calling thread is the compute thread (m_think_pool). + /// Compute thread already holds exclusive lock — shared_lock must be skipped. + bool is_compute_thread() const { + return std::this_thread::get_id() == m_compute_thread_id.load(std::memory_order_relaxed); + } + + /// Preferred tracker accessor for IO-thread callbacks. + /// Returns a guard that: + /// - On IO thread: acquires shared_lock(try_to_lock). Check `if (!guard)` before use. + /// - On compute thread: skips locking (exclusive lock already held). Always valid. + TrackerReadGuard read_tracker() { + return TrackerReadGuard(m_tracker_mutex, m_tracker, is_compute_thread()); + } + + /// Acquire shared (reader) lock on the tracker mutex — BLOCKING. + /// Only for consensus-critical paths (share creation) where skipping is not acceptable. + /// IO-thread callers: prefer read_tracker() (non-blocking) unless you MUST have the lock. + std::shared_lock tracker_shared_lock() { + return std::shared_lock(m_tracker_mutex); + } + + // Async share download — response delivered to callback when sharereply arrives + void request_shares(uint256 id, peer_ptr peer, + std::vector hashes, uint64_t parents, + std::vector stops, + std::function callback) + { + m_share_getter.request(id, callback, id, peer, hashes, parents, stops); + } + + // Called from HANDLER(sharereply) to complete a pending async request + void got_share_reply(uint256 id, dgb::ShareReplyData shares) + { + try { m_share_getter.got_response(id, shares); } + catch (const std::invalid_argument&) { /* request already timed out */ } + } + + std::vector handle_get_share(std::vector hashes, uint64_t parents, std::vector stops, NetService peer_addr); + + /// Broadcast a new best-block notification to all connected P2P peers. + void broadcast_bestblock(const coin::BlockHeaderType& header) { + for (auto& [nonce, peer] : m_peers) + peer->write(message_bestblock::make_raw(header)); + } + + /// Return a JSON array of connected peer info for the /peer_list endpoint. + nlohmann::json get_peer_info_json() const { + nlohmann::json arr = nlohmann::json::array(); + for (const auto& [nonce, peer] : m_peers) { + auto addr = peer->addr(); + bool incoming = (m_outbound_addrs.find(addr) == m_outbound_addrs.end()); + auto uptime_sec = std::chrono::duration_cast( + std::chrono::steady_clock::now() - peer->m_connected_at).count(); + arr.push_back({ + {"address", addr.to_string()}, + {"version", peer->m_other_subversion}, + {"incoming", incoming}, + {"uptime", uptime_sec}, + {"downtime", 0}, + {"web_port", 0} + }); + } + return arr; + } + + /// Register a callback invoked whenever a bestblock message is received + /// from any peer (after relaying). Use this to trigger work refresh. + void set_on_bestblock(std::function fn) { m_on_bestblock = std::move(fn); } + + /// Set the software version string announced to peers in P2P version messages. + void set_software_version(std::string ver) { m_software_version = std::move(ver); } + + /// Send a set of shares (with any needed txs) to a single peer. + /// Skips shares that originated from that peer. + void send_shares(peer_ptr peer, const std::vector& share_hashes); + + /// Broadcast a locally-generated (or newly-received) share to all peers. + void broadcast_share(const uint256& share_hash); + + /// Start downloading shares from a peer, beginning at `target_hash`. + /// Recursively fetches parents until the chain is connected or CHAIN_LENGTH reached. + void download_shares(peer_ptr peer, const uint256& target_hash); + + /// Return the hash of our tallest chain head, or uint256::ZERO if empty. + uint256 best_share_hash(); + + /// Peer-facing advertisement of our head (version handshake + timer + /// re-announce ONLY — never work creation). Returns the verified head + /// when we have one, otherwise the tallest RAW head so a fresh peer can + /// begin download_shares() before our verified chain exists. ROOT-2: on a + /// --genesis node whose verified chain is still empty at handshake, + /// best_share_hash() advertises NULL and the peer never downloads; + /// advertising the raw head breaks that deadlock. + uint256 advertised_best_share(); + + /// Re-push our advertised head to every connected peer, bypassing the + /// broadcast_share() de-dup set. Fired on best-change and once on a timer + /// after the verified chain first becomes non-empty, so a peer that + /// handshook during the empty window can still ingest our chain. + void readvertise_best_share(); + + /// Load persisted shares from LevelDB storage into the tracker. + void load_persisted_shares(); + void flush_verified_to_leveldb(); + + /// Graceful shutdown: flush pending verified/removal buffers to LevelDB. + void shutdown(); + + /// Start dialing outbound peers from AddrStore / bootstrap list. + /// Maintains target outbound peer count active outbound connections. + void start_outbound_connections(); + + /// Set desired number of outbound peers maintained by connection loop. + /// A value of 0 disables outbound dialing. + void set_target_outbound_peers(size_t count) { m_target_outbound_peers = count; } + + /// Set max total P2P peers (inbound + outbound). + void set_max_peers(size_t count) { m_max_peers = count; } + + /// Set P2P ban duration (seconds). + void set_ban_duration(int seconds) { m_ban_duration = std::chrono::seconds(seconds); } + + /// Set cache size limits for memory control. + void set_cache_limits(size_t max_shared, size_t max_known_txs, size_t max_raw) { + m_max_shared_hashes = max_shared; + m_max_known_txs = max_known_txs; + m_max_raw_shares = max_raw; + } + + /// Set RSS memory limit in MB (abort if exceeded). Static because checked in process_shares. + static void set_rss_limit_mb(long mb); + + /// Expose tracker mutex for IO-thread callbacks that access the tracker. + /// Callers MUST use shared_lock(try_to_lock) — NEVER blocking lock(). + std::shared_mutex& tracker_mutex() { return m_tracker_mutex; } + + /// Unified share retention: single-pass prune of chain + verified + LevelDB. + /// Replaces multi-pass trim with work-based dead head detection and + /// deferred destruction for verified cascade safety. + /// Called from run_think() on the ioc thread. + void prune_shares(const uint256& best_share); + + /// Run the share tracker think() cycle: verifies chains, scores heads, + /// identifies bad peers, and requests needed shares. + /// Should be called periodically (e.g. after processing_shares or on a timer). + void run_think(); + + /// Fast-path: update best share after creating a local share. + /// Bypasses run_think() scoring — just sets the new tip and triggers + /// work refresh so miners immediately build on the new share. + /// p2pool equivalent: set_best_share() → work_event.happened(). + void notify_local_share(const uint256& share_hash); + + /// Periodic maintenance: eat stale heads, drop tails, then run_think(). + /// Matches p2pool's clean_tracker() (node.py:355-402). + void clean_tracker(); + + /// Drain share batches queued while think() held the tracker mutex. + /// Called on the IO thread after think() releases the lock. + void drain_pending_adds(); + + /// p2pool-style heartbeat: chain height, local/pool hashrate, orphan/DOA stats. + /// Runs on a separate 30s timer — diagnostic only, not consensus-critical. + void heartbeat_log(); + + /// Set the block_rel_height function used by run_think() for chain scoring. + /// fn(block_hash) should return confirmations from the coin daemon: + /// >= 0 : block is in main chain (0 = tip, higher = deeper) + /// < 0 : block is NOT in main chain (orphaned/stale) + using block_rel_height_fn_t = std::function; + void set_block_rel_height_fn(block_rel_height_fn_t fn) { m_block_rel_height_fn = std::move(fn); } + + /// Lock-free stats snapshot — published by think(), never fails, never needs tracker lock. + /// Inline definition must precede callers in the same header. + TrackerSnapshot get_tracker_snapshot() const; + int get_chain_count() const; + int get_verified_count() const; + + /// Called when best_share changes (p2pool: new_work_event) + /// Triggers immediate work update for all stratum miners. + void set_on_best_share_changed(std::function fn) { m_on_best_share_changed = std::move(fn); } + + /// Callback to get local hashrate from stratum server (H/s) + void set_local_hashrate_fn(std::function fn) { m_local_hashrate_fn = std::move(fn); } + + /// Local mining stats from RateMonitor (for p2pool-style status lines) + struct LocalRateStats { + double hashrate = 0; // H/s (total local) + double effective_dt = 0; // seconds of data in window + int total_datums = 0; // pseudoshares in window + int dead_datums = 0; // dead (DOA) pseudoshares in window + }; + void set_local_rate_stats_fn(std::function fn) { m_local_rate_stats_fn = std::move(fn); } + + /// Current PPLNS outputs {script_hex, satoshis} for payout display + void set_current_pplns_fn(std::function>()> fn) { + m_current_pplns_fn = std::move(fn); + } + + /// Node operator's payout script hex (for matching in PPLNS outputs) + void set_node_payout_script_hex(const std::string& hex) { m_node_payout_script_hex = hex; } + const std::string& get_node_payout_script_hex() const { return m_node_payout_script_hex; } + + /// Local miner scripts (from stratum sessions' pubkey_hashes → all script forms) + void set_local_miner_scripts_fn(std::function()> fn) { + m_local_miner_scripts_fn = std::move(fn); + } + + /// Check whether a peer address is currently banned. + bool is_banned(const NetService& addr) const; + + /// ── Runtime admin API (pool peer bans + whitelist) ───────────────── + /// All methods assumed to run on the io_context thread — callers + /// (web_server HTTP handlers) dispatch via thread_safe_wrap(). + /// + /// Returned JSON uses the shape: + /// {"ok": true|false, "error"?: "...", "bans": [...], "whitelist": [...]} + nlohmann::json admin_list_bans() const; + nlohmann::json admin_ban_ip(const std::string& ip, int duration_sec); + nlohmann::json admin_unban_ip(const std::string& ip); + nlohmann::json admin_list_whitelist() const; + nlohmann::json admin_whitelist_add(const std::string& host, uint16_t port); + nlohmann::json admin_whitelist_remove(const std::string& host, uint16_t port); + nlohmann::json admin_list_peers() const; + nlohmann::json admin_drop_peer(const std::string& ip); + nlohmann::json admin_dial_peer(const std::string& host, uint16_t port); + + /// Path to persisted whitelist file (~/.c2pool/pool_whitelist.json). + /// Set by c2pool_refactored.cpp before start(); empty = no persistence. + void set_whitelist_path(const std::string& path); + + /// True if addr's IP matches a whitelist entry (IP or host:port). + bool is_whitelisted(const NetService& addr) const; + +protected: + std::string m_software_version = "/c2pool:0.1/"; // overridden by set_software_version() + std::function m_on_bestblock; + std::function m_on_best_share_changed; + std::function m_local_hashrate_fn; + std::function m_local_rate_stats_fn; + std::function>()> m_current_pplns_fn; + std::function()> m_local_miner_scripts_fn; + std::string m_node_payout_script_hex; + std::set m_shared_share_hashes; // de-dup set for broadcast_share + std::set m_rejected_share_hashes; // shares rejected by peers — never re-broadcast + std::set m_downloading_shares; // hashes currently being fetched + + // Track share hashes that peers couldn't provide (empty reply). + // After MAX_EMPTY_RETRIES failures, stop requesting — the share is + // pruned from the network. p2pool avoids this via reactive desired_var + // + sleep(1) backoff. We use explicit failure counting. + static constexpr int MAX_EMPTY_RETRIES = 3; + std::unordered_map m_download_fail_count; + + // Track req_id → peer addr for selective cancellation on disconnect. + // p2pool has per-peer get_shares (GenericDeferrer), so connectionLost calls + // respond_all() on just that peer's deferrer. c2pool has a shared m_share_getter, + // so we track which req_ids belong to which peer for cancel-on-disconnect. + std::map m_pending_share_reqs; // req_id → peer addr + + // Track recently-broadcast share hashes + timestamp so we can detect + // rapid disconnections (peer rejected our share → PoW invalid loop). + struct BroadcastRecord { + std::vector hashes; + std::chrono::steady_clock::time_point when; + }; + std::map m_last_broadcast_to; // per-peer + + // Connection maintenance + static constexpr size_t DEFAULT_TARGET_OUTBOUND_PEERS = 8; + size_t m_max_peers = 30; + size_t m_target_outbound_peers = DEFAULT_TARGET_OUTBOUND_PEERS; + std::unique_ptr m_connect_timer; + std::unique_ptr m_readvert_timer; // one-shot ROOT-2 re-advert + std::set m_pending_outbound; // addresses currently being dialed + std::set m_outbound_addrs; // successfully connected outbound peers + + // Peer banning: maps address → ban expiry time + std::map m_ban_list; + std::chrono::seconds m_ban_duration{300}; // 5 minutes (configurable) + + // IP-only manual bans (admin endpoint). Keyed by IP string so the + // operator can ban/unban without knowing the peer's source port. + std::map m_ip_ban_list; + + // Whitelist: IPs that bypass is_banned() and host:port entries kept as + // permanent dial targets. Persists across restart via m_whitelist_path. + std::set m_whitelist_ips; + std::set m_whitelist_hosts; + std::string m_whitelist_path; + + void load_whitelist_from_disk(); + void save_whitelist_to_disk() const; + + // Rate-limiting no longer needed: think() runs on a dedicated compute + // thread (m_think_pool), serialized by m_think_running atomic flag. + // The IO thread never blocks. p2pool's reactor.callLater equivalent. + + // Cache limits (configurable) + size_t m_max_shared_hashes = 50000; + size_t m_max_known_txs = 10000; + size_t m_max_raw_shares = 50000; + + // Block depth function for chain scoring (set via set_block_rel_height_fn) + block_rel_height_fn_t m_block_rel_height_fn; + + // Cached best share hash from the most recent think() cycle + uint256 m_best_share_hash; + + // ROOT-2: fire exactly one delayed re-advert when the verified chain first + // transitions from empty to non-empty (peers that handshook during the + // empty window otherwise never see a best-change event). + bool m_verified_was_empty = true; + + // Cache of original raw serialized bytes keyed by share hash. + // Used for relay so we send the exact bytes we received, avoiding + // any round-trip serialization differences. + std::unordered_map m_raw_share_cache; +}; + +struct HandleSharesData +{ + std::vector m_items; + std::vector m_raw_items; // original raw bytes, parallel with m_items + std::map> m_txs; + + void add(const ShareType& share, std::vector txs) + { + m_items.push_back(share); + m_raw_items.emplace_back(); // no cached raw bytes + m_txs[share.hash()] = std::move(txs); + } + + void add(const ShareType& share, std::vector txs, + const chain::RawShare& raw) + { + m_items.push_back(share); + m_raw_items.push_back(raw); + m_txs[share.hash()] = std::move(txs); + } +}; + + +/* + void handle_message_addrs(std::shared_ptr msg, PoolProtocol* protocol); + void handle_message_addrme(std::shared_ptr msg, PoolProtocol* protocol); + void handle_message_ping(std::shared_ptr msg, PoolProtocol* protocol); + void handle_message_getaddrs(std::shared_ptr msg, PoolProtocol* protocol); + void handle_message_shares(std::shared_ptr msg, PoolProtocol* protocol); + void handle_message_sharereq(std::shared_ptr msg, PoolProtocol* protocol); + void handle_message_sharereply(std::shared_ptr msg, PoolProtocol* protocol); + void handle_message_bestblock(std::shared_ptr msg, PoolProtocol* protocol); + void handle_message_have_tx(std::shared_ptr msg, PoolProtocol* protocol); + void handle_message_losing_tx(std::shared_ptr msg, PoolProtocol* protocol); + void handle_message_remember_tx(std::shared_ptr msg, PoolProtocol* protocol); + void handle_message_forget_tx(std::shared_ptr msg, PoolProtocol* protocol); +*/ + +class Legacy : public pool::Protocol +{ +public: + void handle_message(std::unique_ptr rmsg, NodeImpl::peer_ptr peer) override; + + ADD_HANDLER(addrs, dgb::message_addrs); + ADD_HANDLER(addrme, dgb::message_addrme); + ADD_HANDLER(ping, dgb::message_ping); + ADD_HANDLER(getaddrs, dgb::message_getaddrs); + ADD_HANDLER(shares, dgb::message_shares); + ADD_HANDLER(sharereq, dgb::message_sharereq); + ADD_HANDLER(sharereply, dgb::message_sharereply); + ADD_HANDLER(bestblock, dgb::message_bestblock); + ADD_HANDLER(have_tx, dgb::message_have_tx); + ADD_HANDLER(losing_tx, dgb::message_losing_tx); + ADD_HANDLER(remember_tx, dgb::message_remember_tx); + ADD_HANDLER(forget_tx, dgb::message_forget_tx); +}; -// --------------------------------------------------------------------------- -// Skeleton entry points (slice #4, Option B). Defined in node.cpp. -// COMPILING SKELETON ONLY — the real node run-loop is M3 / Phase B. -// --------------------------------------------------------------------------- +class Actual : public pool::Protocol +{ +public: + void handle_message(std::unique_ptr rmsg, NodeImpl::peer_ptr peer) override; -// One-line summary of the DGB-Scrypt network from the header-only config -// constants (no Fileconfig instantiation). -std::string network_summary(); + ADD_HANDLER(addrs, dgb::message_addrs); + ADD_HANDLER(addrme, dgb::message_addrme); + ADD_HANDLER(ping, dgb::message_ping); + ADD_HANDLER(getaddrs, dgb::message_getaddrs); + ADD_HANDLER(shares, dgb::message_shares); + ADD_HANDLER(sharereq, dgb::message_sharereq); + ADD_HANDLER(sharereply, dgb::message_sharereply); + ADD_HANDLER(bestblock, dgb::message_bestblock); + ADD_HANDLER(have_tx, dgb::message_have_tx); + ADD_HANDLER(losing_tx, dgb::message_losing_tx); + ADD_HANDLER(remember_tx, dgb::message_remember_tx); + ADD_HANDLER(forget_tx, dgb::message_forget_tx); +}; -// Skeleton run entry. Returns a process exit code. Phase B replaces this with -// the embedded-daemon + pool/sharechain run-loop. -int run_skeleton(); +using Node = pool::NodeBridge; } // namespace dgb diff --git a/src/impl/dgb/protocol_actual.cpp b/src/impl/dgb/protocol_actual.cpp new file mode 100644 index 000000000..45fab872b --- /dev/null +++ b/src/impl/dgb/protocol_actual.cpp @@ -0,0 +1,274 @@ +#include "node.hpp" +#include "share.hpp" + +#include +#include +#include + +#include + +namespace dgb +{ + +void Actual::handle_message(std::unique_ptr rmsg, peer_ptr peer) +{ + dgb::Handler::result_t result; + try + { + result = m_handler.parse(rmsg); + } catch (const std::exception& ec) + { + LOG_WARNING << "Failed to parse message '" << rmsg->m_command << "' from " + << peer->addr().to_string() << ": " << ec.what(); + return; + } + + try + { + std::visit([&](auto& msg){ handle(std::move(msg), peer); }, result); + } + catch (const std::exception& ec) + { + LOG_WARNING << "Handler error for '" << rmsg->m_command << "' from " + << peer->addr().to_string() << ": " << ec.what(); + return; + } +} + +void Actual::HANDLER(addrs) +{ + for (auto& addr : msg->m_addrs) + { + addr.m_timestamp = std::min((uint64_t) core::timestamp(), addr.m_timestamp); + got_addr(addr.m_endpoint, addr.m_services, addr.m_timestamp); + + if ((core::random::random_float(0, 1) < 0.8) && (!m_connections.empty())) + { + auto wpeer = core::random::random_choice(m_connections); + auto rmsg = message_addrs::make_raw({addr}); + wpeer->write(std::move(rmsg)); + } + } +} + +void Actual::HANDLER(addrme) +{ + if (peer->addr().address() == "127.0.0.1") + { + if (!m_peers.empty() && (core::random::random_float(0, 1) < 0.8)) + { + auto random_peer = core::random::random_choice(m_peers); + auto rmsg = dgb::message_addrme::make_raw(msg->m_port); + random_peer->write(std::move(rmsg)); + } + } else + { + auto endpoint = NetService{peer->addr().address(), msg->m_port}; + got_addr(endpoint, peer->m_other_services, core::timestamp()); + if (!m_peers.empty() && (core::random::random_float(0, 1) < 0.8)) + { + auto random_peer = core::random::random_choice(m_peers); + auto rmsg = dgb::message_addrs::make_raw({addr_record_t{peer->m_other_services, endpoint, core::timestamp()} }); + random_peer->write(std::move(rmsg)); + } + } +} + +void Actual::HANDLER(ping) +{ +} + +void Actual::HANDLER(getaddrs) +{ + if (msg->m_count > 100) + msg->m_count = 100; + + std::vector addrs; + for (const auto& pair : get_good_peers(msg->m_count)) + { + addrs.push_back({pair.value.m_service, pair.addr, pair.value.m_last_seen}); + } + + auto rmsg = message_addrs::make_raw({addrs}); + peer->write(std::move(rmsg)); +} + +void Actual::HANDLER(shares) +{ + dgb::HandleSharesData result; + + for (auto wrappedshare : msg->m_shares) + { + // Save a copy of the original raw bytes before deserialization consumes them + chain::RawShare raw_copy(wrappedshare.type, + BaseScript(wrappedshare.contents.m_data)); + + dgb::ShareType share; + try + { + share = dgb::load_share(wrappedshare, peer->addr()); + } + catch(const std::exception& e) + { + LOG_WARNING << "Failed to load share (type=" << wrappedshare.type + << ") from " << peer->addr().to_string() << ": " << e.what(); + continue; + } + + std::vector txs; + share.ACTION + ({ + if constexpr (share_t::version >= 13 && share_t::version < 34) + for (auto tx_hash : obj->m_tx_info.m_new_transaction_hashes) + { + auto it = m_known_txs.find(tx_hash); + if (it != m_known_txs.end()) + { + txs.emplace_back(it->second); + } + else + { + LOG_WARNING << "Peer referenced unknown transaction " << tx_hash.ToString(); + } + } + }); + + result.add(share, txs, raw_copy); + } + + processing_shares(result, peer->addr()); +} + +void Actual::HANDLER(sharereq) +{ + auto shares = handle_get_share(msg->m_hashes, msg->m_parents, msg->m_stops, peer->addr()); + + std::vector rshares; + + try + { + for (auto& share : shares) + { + rshares.emplace_back(share.version(), pack(share)); + } + auto reply_msg = message_sharereply::make_raw(msg->m_id, dgb::ShareReplyResult::good, rshares); + peer->write(std::move(reply_msg)); + } + catch (const std::invalid_argument &e) + { + auto reply_msg = message_sharereply::make_raw(msg->m_id, dgb::ShareReplyResult::too_long, {}); + peer->write(std::move(reply_msg)); + } +} + +void Actual::HANDLER(sharereply) +{ + dgb::ShareReplyData result; + if (msg->m_result == ShareReplyResult::good) + { + result.m_items.reserve(msg->m_shares.size()); + result.m_raw_items.reserve(msg->m_shares.size()); + for (auto& rshare : msg->m_shares) + { + try + { + auto share = dgb::load_share(rshare, peer->addr()); + result.m_items.push_back(share); + result.m_raw_items.push_back(rshare); + } + catch(const std::exception& e) + { + LOG_WARNING << "Failed to deserialize share (type=" << rshare.type + << ") from " << peer->addr().to_string() << ": " << e.what(); + continue; + } + } + } + got_share_reply(msg->m_id, result); +} + +void Actual::HANDLER(bestblock) +{ + auto header_hash = Hash(pack(msg->m_header).get_span()); + LOG_INFO << "[Pool] New best block from peer " << peer->addr().to_string() + << ": " << header_hash.ToString(); + + // p2pool does NOT relay bestblock — each node broadcasts only from its own + // block source (bitcoind_work.changed → best_block_header.changed). + // Relaying creates an amplification loop: A→B→C→A with alternating hashes. + if (m_on_bestblock) m_on_bestblock(header_hash); +} + +void Actual::HANDLER(have_tx) +{ + peer->m_remote_txs.insert(msg->m_tx_hashes.begin(), msg->m_tx_hashes.end()); + if (peer->m_remote_txs.size() > 10000) + { + peer->m_remote_txs.erase(peer->m_remote_txs.begin(), std::next(peer->m_remote_txs.begin(), peer->m_remote_txs.size() - 10000)); + } +} + +void Actual::HANDLER(losing_tx) +{ + std::set losing_txs; + losing_txs.insert(msg->m_tx_hashes.begin(), msg->m_tx_hashes.end()); + + std::set diff_txs; + std::set_difference(peer->m_remote_txs.begin(), peer->m_remote_txs.end(), + losing_txs.begin(), losing_txs.end(), + std::inserter(diff_txs, diff_txs.begin())); + + peer->m_remote_txs = diff_txs; +} + +void Actual::HANDLER(remember_tx) +{ + // Phase 1: tx_hashes — peer tells us to remember these by hash (must be in known_txs) + for (auto tx_hash : msg->m_tx_hashes) + { + if (peer->m_remembered_txs.contains(tx_hash)) + { + LOG_WARNING << "Peer referenced transaction twice: " << tx_hash.ToString(); + continue; + } + + auto it = m_known_txs.find(tx_hash); + if (it != m_known_txs.end()) + { + peer->m_remembered_txs.insert_or_assign(tx_hash, it->second); + } + else + { + LOG_WARNING << "Peer referenced unknown transaction " << tx_hash.ToString(); + } + } + + // Phase 2: txs — peer sends full transactions (compute hash, store) + for (auto& tx : msg->m_txs) + { + auto packed = pack(coin::TX_WITH_WITNESS(tx)); + auto tx_hash = Hash(packed.get_span()); + + if (peer->m_remembered_txs.contains(tx_hash)) + { + LOG_WARNING << "Peer sent duplicate transaction: " << tx_hash.ToString(); + continue; + } + + coin::Transaction full_tx(tx); + peer->m_remembered_txs.insert_or_assign(tx_hash, full_tx); + + if (!m_known_txs.contains(tx_hash)) + m_known_txs.emplace(tx_hash, std::move(full_tx)); + } +} + +void Actual::HANDLER(forget_tx) +{ + for (auto tx_hash : msg->m_tx_hashes) + { + peer->m_remembered_txs.erase(tx_hash); + } +} + +} // namespace dgb diff --git a/src/impl/dgb/protocol_legacy.cpp b/src/impl/dgb/protocol_legacy.cpp new file mode 100644 index 000000000..92ba46749 --- /dev/null +++ b/src/impl/dgb/protocol_legacy.cpp @@ -0,0 +1,310 @@ +#include "node.hpp" +#include "share.hpp" + +#include +#include +#include + +namespace dgb +{ + +void Legacy::handle_message(std::unique_ptr rmsg, peer_ptr peer) +{ + dgb::Handler::result_t result; + try + { + result = m_handler.parse(rmsg); + } catch (const std::exception& ec) + { + LOG_WARNING << "Failed to parse message '" << rmsg->m_command << "' from " + << peer->addr().to_string() << ": " << ec.what(); + return; + } + + try + { + std::visit([&](auto& msg){ handle(std::move(msg), peer); }, result); + } + catch (const std::exception& ec) + { + LOG_WARNING << "Handler error for '" << rmsg->m_command << "' from " + << peer->addr().to_string() << ": " << ec.what(); + return; + } +} + +void Legacy::HANDLER(addrs) +{ + for (auto& addr : msg->m_addrs) + { + addr.m_timestamp = std::min((uint64_t) core::timestamp(), addr.m_timestamp); + got_addr(addr.m_endpoint, addr.m_services, addr.m_timestamp); + + if ((core::random::random_float(0, 1) < 0.8) && (!m_connections.empty())) + { + auto wpeer = core::random::random_choice(m_connections); + auto rmsg = message_addrs::make_raw({addr}); + wpeer->write(std::move(rmsg)); + } + } +} + +void Legacy::HANDLER(addrme) +{ + if (peer->addr().address() == "127.0.0.0") + { + if (m_peers.empty() && (core::random::random_float(0, 1) < 0.8)) + { + auto random_peer = core::random::random_choice(m_peers); + auto rmsg = dgb::message_addrme::make_raw(msg->m_port); + random_peer->write(std::move(rmsg)); + } + } else + { + auto endpoint = NetService{peer->addr().address(), msg->m_port}; + got_addr(endpoint, peer->m_other_services, core::timestamp()); + if (m_peers.empty() && (core::random::random_float(0, 1) < 0.8)) + { + auto random_peer = core::random::random_choice(m_peers); + auto rmsg = dgb::message_addrs::make_raw({addr_record_t{peer->m_other_services, endpoint, core::timestamp()} }); + random_peer->write(std::move(rmsg)); + } + } +} + +void Legacy::HANDLER(ping) +{ +} + +void Legacy::HANDLER(getaddrs) +{ + if (msg->m_count > 100) + msg->m_count = 100; + + std::vector addrs; + for (const auto& pair : get_good_peers(msg->m_count)) + { + addrs.push_back({pair.value.m_service, pair.addr, pair.value.m_last_seen}); + } + + auto rmsg = message_addrs::make_raw({addrs}); + peer->write(std::move(rmsg)); +} + +void Legacy::HANDLER(shares) +{ + try { + dgb::HandleSharesData result; //share, txs + + for (auto wrappedshare : msg->m_shares) + { + dgb::ShareType share; + try + { + share = dgb::load_share(wrappedshare, peer->addr()); + } + catch(const std::exception& e) + { + LOG_WARNING << "Failed to load share (type=" << wrappedshare.type + << ") from " << peer->addr().to_string() << ": " << e.what(); + continue; + } + + std::vector txs; + share.ACTION + ({ + if constexpr (share_t::version >= 13 && share_t::version < 34) + for (auto tx_hash : obj->m_tx_info.m_new_transaction_hashes) + { + auto it = m_known_txs.find(tx_hash); + if (it != m_known_txs.end()) + { + txs.emplace_back(it->second); + } + else + { + LOG_WARNING << "Peer referenced unknown transaction " << tx_hash.ToString(); + } + } + }); + + result.add(share, txs); + } + + processing_shares(result, peer->addr()); + } catch (const std::exception& e) { + LOG_ERROR << "[Pool] shares handler exception: " << e.what(); + } +} + +void Legacy::HANDLER(sharereq) +{ + // Debug: log what's being requested + { + static int dbg_req = 0; + if (dbg_req < 20) + { + ++dbg_req; + std::string req_hashes; + for (auto& h : msg->m_hashes) req_hashes += h.ToString().substr(0, 16) + " "; + LOG_WARNING << "SHAREREQ #" << dbg_req << " from " << peer->addr().to_string() + << " hashes=[" << req_hashes << "] parents=" << msg->m_parents + << " stops=" << msg->m_stops.size(); + } + } + + auto shares = handle_get_share(msg->m_hashes, msg->m_parents, msg->m_stops, peer->addr()); + + // Debug: log what we're returning + { + static int dbg_rep = 0; + if (dbg_rep < 20) + { + ++dbg_rep; + LOG_WARNING << "SHAREREPLY #" << dbg_rep << " returning " << shares.size() << " shares"; + for (size_t i = 0; i < std::min(shares.size(), (size_t)3); ++i) + { + LOG_WARNING << " share[" << i << "] hash=" << shares[i].hash().ToString().substr(0, 16); + } + } + } + + std::vector rshares; + + try + { + for (auto& share : shares) + { + rshares.emplace_back(share.version(), pack(share)); + } + auto reply_msg = message_sharereply::make_raw(msg->m_id, dgb::ShareReplyResult::good, rshares); + peer->write(std::move(reply_msg)); + } + catch (const std::invalid_argument &e) + { + // Serialization overflow: the packed shares exceeded the P2P message + // size limit (32 MB). Reply with too_long so the peer requests a + // smaller batch. This is the correct behavior per Python p2pool. + LOG_WARNING << "Share reply too large, sending too_long: " << e.what(); + auto reply_msg = message_sharereply::make_raw(msg->m_id, dgb::ShareReplyResult::too_long, {}); + peer->write(std::move(reply_msg)); + } +} + +void Legacy::HANDLER(sharereply) +{ + dgb::ShareReplyData result; + if (msg->m_result == ShareReplyResult::good) + { + result.m_items.reserve(msg->m_shares.size()); + result.m_raw_items.reserve(msg->m_shares.size()); + for (auto& rshare : msg->m_shares) + { + try + { + auto share = dgb::load_share(rshare, peer->addr()); + result.m_items.push_back(share); + result.m_raw_items.push_back(rshare); + } + catch(const std::exception& e) + { + LOG_WARNING << "Failed to deserialize share (type=" << rshare.type + << ") from " << peer->addr().to_string() << ": " << e.what(); + continue; + } + } + } + // Resolve the async request that originally sent the sharereq + got_share_reply(msg->m_id, result); +} + +void Legacy::HANDLER(bestblock) +{ + try { + auto header_hash = Hash(pack(msg->m_header).get_span()); + LOG_INFO << "[Pool] New best block from peer " << peer->addr().to_string() + << ": " << header_hash.ToString(); + + // p2pool does NOT relay bestblock — each node broadcasts only from its own + // block source (bitcoind_work.changed → best_block_header.changed). + // Relaying creates an amplification loop: A→B→C→A with alternating hashes. + if (m_on_bestblock) m_on_bestblock(header_hash); + } catch (const std::exception& e) { + LOG_WARNING << "[Pool] bestblock handler exception: " << e.what(); + } +} + +void Legacy::HANDLER(have_tx) +{ + peer->m_remote_txs.insert(msg->m_tx_hashes.begin(), msg->m_tx_hashes.end()); + if (peer->m_remote_txs.size() > 10000) + { + peer->m_remote_txs.erase(peer->m_remote_txs.begin(), std::next(peer->m_remote_txs.begin(), peer->m_remote_txs.size() - 10000)); + } +} + +void Legacy::HANDLER(losing_tx) +{ + //remove all msg->txs hashes from remote_tx_hashes + std::set losing_txs; + losing_txs.insert(msg->m_tx_hashes.begin(), msg->m_tx_hashes.end()); + + std::set diff_txs; + std::set_difference(peer->m_remote_txs.begin(), peer->m_remote_txs.end(), + losing_txs.begin(), losing_txs.end(), + std::inserter(diff_txs, diff_txs.begin())); + + peer->m_remote_txs = diff_txs; +} + +void Legacy::HANDLER(remember_tx) +{ + // Phase 1: tx_hashes — peer tells us to remember these by hash (must be in known_txs) + for (auto tx_hash : msg->m_tx_hashes) + { + if (peer->m_remembered_txs.contains(tx_hash)) + { + LOG_WARNING << "Peer referenced transaction twice: " << tx_hash.ToString(); + continue; + } + + auto it = m_known_txs.find(tx_hash); + if (it != m_known_txs.end()) + { + peer->m_remembered_txs.insert_or_assign(tx_hash, it->second); + } + else + { + LOG_WARNING << "Peer referenced unknown transaction " << tx_hash.ToString(); + } + } + + // Phase 2: txs — peer sends full transactions (compute hash, store) + for (auto& tx : msg->m_txs) + { + auto packed = pack(coin::TX_WITH_WITNESS(tx)); + auto tx_hash = Hash(packed.get_span()); + + if (peer->m_remembered_txs.contains(tx_hash)) + { + LOG_WARNING << "Peer sent duplicate transaction: " << tx_hash.ToString(); + continue; + } + + coin::Transaction full_tx(tx); + peer->m_remembered_txs.insert_or_assign(tx_hash, full_tx); + + if (!m_known_txs.contains(tx_hash)) + m_known_txs.emplace(tx_hash, std::move(full_tx)); + } +} + +void Legacy::HANDLER(forget_tx) +{ + for (auto tx_hash : msg->m_tx_hashes) + { + peer->m_remembered_txs.erase(tx_hash); + } +} + +} // namespace dgb \ No newline at end of file