node.cpp

// SPDX-License-Identifier: AGPL-3.0-or-later
#include "node.hpp"

#include <core/common.hpp>
#include <core/hash.hpp>
#include <core/random.hpp>
#include <core/target_utils.hpp>
#include <sharechain/prepared_list.hpp>
#include <impl/dgb/get_shares_walk.hpp>
#include <impl/dgb/download_stops.hpp>
#include <impl/dgb/pool_efficiency.hpp>
#include <impl/dgb/expected_time_to_block.hpp>
#include <impl/dgb/coin/binomial_conf_interval.hpp>
#include <impl/dgb/coin/share_tx_relay_refs.hpp>  // SSOT: per-share new-tx-ref probe (#905)

#include <algorithm>
#include <filesystem>
#include <fstream>
#include <iomanip>
#include <random>

// Static members for DensePPLNSWindow precomputed decay table
std::vector<uint64_t> 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 "???";
    // Oracle-faithful Wilson score interval via the dgb::coin SSOT
    // (coin/binomial_conf_interval.hpp; p2pool util/math.py:133). This delegates
    // the prior inline z=1.96 + plain-[0,1]-clip approximation onto the SSOT,
    // which uses z = sqrt(2)*ierf(conf) and add_to_range bracketing. Output is
    // oracle-faithful and NOT byte-identical to the old literal-z form; diag
    // display only (heartbeat log), zero consensus surface.
    const std::array<double, 2> ci = dgb::coin::binomial_conf_interval(
        static_cast<double>(x), static_cast<double>(n), conf);
    const double p = static_cast<double>(x) / n;
    std::ostringstream os;
    os << "~" << std::fixed << std::setprecision(1) << (100.0 * p) << "% ("
       << static_cast<int>(std::floor(100.0 * ci[0])) << "-"
       << static_cast<int>(std::ceil(100.0 * ci[1])) << "%)";
    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 "???";
    // Stale-rate CI via the oracle-faithful SSOT, then mapped to efficiency.
    // See format_binomial_conf note: oracle-faithful, NOT byte-identical to the
    // prior inline z=1.96 form; diag display only.
    const std::array<double, 2> ci = dgb::coin::binomial_conf_interval(
        static_cast<double>(stale), static_cast<double>(n), 0.95);
    const double p = static_cast<double>(stale) / n;
    const double lo_stale = ci[0];
    const double hi_stale = ci[1];
    const double denom = (stale_prop < 0.999) ? (1.0 - stale_prop) : 1.0;
    const 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<int>(std::floor(100.0 * eff_lo)) << "-"
       << static_cast<int>(std::ceil(100.0 * eff_hi)) << "%)";
    return os.str();
}

namespace dgb
{

static uint64_t make_random_nonce()
{
    std::mt19937_64 rng(std::random_device{}());
    return rng();
}

void NodeImpl::send_ping(peer_ptr peer)
{
        auto rmsg = dgb::message_ping::make_raw();
        peer->write(std::move(rmsg));
};

void NodeImpl::connected(std::shared_ptr<core::Socket> 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<std::chrono::milliseconds>(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);
    publish_peer_info_snapshot();   // IO thread: refresh after peer removal

    // 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<uint256> 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<uint256> 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);
    publish_peer_info_snapshot();   // IO thread: refresh after close
}

NodeImpl::TrackerSnapshot NodeImpl::get_tracker_snapshot() const {
    std::lock_guard<std::mutex> 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->advertised_protocol_version,  // advertise our capability (oracle p2p.py VERSION 3501), NOT the accept-floor (handle_version keeps minimum_protocol_version as the reject floor)
        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<pool::PeerConnectionType> NodeImpl::handle_version(std::unique_ptr<RawMessage> rmsg, peer_ptr peer)
{
    LOG_DEBUG_POOL << "handle message_version";
        std::unique_ptr<dgb::message_version> 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;
        publish_peer_info_snapshot();   // IO thread: refresh the display snapshot

        // 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. The floor is the RUNTIME
        // ratcheted value (cold 1400, lifted to 3500 once >=95% of window work
        // desires the best share's version -- apply_min_protocol_ratchet), NOT the
        // immutable config floor. Read lock-free; compute thread publishes via store.
        const uint32_t min_proto =
            m_runtime_min_protocol_version.load(std::memory_order_relaxed);
        if (msg->m_version < min_proto)
        {
            LOG_WARNING << "Peer " << msg->m_addr_from.m_endpoint.to_string()
                        << " protocol " << msg->m_version
                        << " < minimum " << min_proto
                        << ", 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));
        }

        // Canonical p2p.py:276 — one FULL have_tx as soon as the handshake
        // completes, so the peer's view of our tx pool starts correct (and its
        // dashboard stops showing us as txpool = 0). Subsequent sweeps send
        // deltas only. Advert-only; no consensus/mint state touched.
        //
        // ORDERING IS LOAD-BEARING: this is the LARGEST write the handshake
        // issues (up to TX_ADVERT_MAX_HASHES_PER_MESSAGE hashes, ~32 KB) and it
        // goes LAST, after the tiny getaddrs/addrme (and any sharereq) above.
        // core::Socket::write starts a composed async_write with no outbound
        // queue (core/socket.cpp:110-137), so initiating another write while one
        // is still draining interleaves bytes mid-message and gets us dropped on
        // a bad checksum. Going last means NOTHING follows it in this handler;
        // the ~32 KB advert may well take several write_some rounds, which is
        // harmless precisely because nothing else is queued behind it, and the
        // per-peer min-emit interval keeps the next timer sweep from landing on
        // top of it. See core/tx_advertiser.hpp.
        advertise_known_txs(peer);

        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<HandleSharesData>(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<std::atomic<int>>(static_cast<int>(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<HandleSharesData>(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<ShareType> 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<uint256, ShareType> prepare_shares(valid_shares);
    std::vector<ShareType> shares = prepare_shares.build_list();

    // Step 3: Process sorted shares
    int32_t new_count = 0;
    int32_t dup_count = 0;
    std::map<uint256, coin::MutableTransaction> all_new_txs;
    std::vector<c2pool::storage::SharechainStorage::ShareBatchEntry> 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<uint8_t> 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<const uint8_t*>(span.data()),
                             reinterpret_cast<const uint8_t*>(span.data()) + span.size());
            }
            uint64_t ver = share.version();
            std::vector<uint8_t> 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<dgb::ShareType> NodeImpl::handle_get_share(std::vector<uint256> hashes, uint64_t parents, std::vector<uint256> 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<std::shared_mutex> 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 {};
    }

    // Delegate the walk to the SSOT (get_shares_walk.hpp) so the parents cap,
    // the per-hash walk bound, the stop-hash break, and the missing/rejected
    // skips stay byte-identical to the p2pool node.py oracle and are pinned by
    // the get_shares_walk conformance KAT. The try_to_lock guard above and the
    // node-local logging stay here; only the pure walk moves to the SSOT.
    auto shares = dgb::collect_get_shares<dgb::ShareType>(
        *m_chain,
        hashes,
        parents,
        stops,
        [this](const uint256& h) { return m_rejected_share_hashes.count(h) != 0; },
        [this](const uint256& 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();
        });

	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<uint256>& 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<std::shared_mutex> 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<ShareType> 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. A share's
    // referenced new-tx hashes live INSIDE m_tx_info (dgb::ShareTxInfo) on
    // v17/v33 and are absent on v34/v35/v36. The pre-fix probe named a
    // top-level m_new_transaction_hashes NO dgb share type declares, so it was
    // ALWAYS false and this relay block was silently dead for every version.
    // append_share_tx_refs is the SSOT that guards on m_tx_info (#905).
    std::set<uint256> needed_txs;
    for (auto& share : shares)
    {
        share.invoke([&](auto* obj) {
            std::vector<uint256> refs;
            dgb::append_share_tx_refs(obj, refs);
            for (const auto& th : refs)
            {
                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<uint256> known_hashes;   // hashes in peer's remote set
        std::vector<coin::MutableTransaction> 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<chain::RawShare> 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<uint256> 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<std::shared_mutex> 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<uint256> 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, as p2pool does.
    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();
}

void NodeImpl::apply_min_protocol_ratchet()
{
    // Runtime wiring of the accept-floor ratchet #602 landed as a pure function.
    // Mirrors oracle main.py:213-216 (run on each best-share advance) + data.py:857
    // update_min_protocol_version: over the window [CHAIN_LENGTH*9/10, CHAIN_LENGTH]
    // behind the best share's PARENT, lift the inbound P2P accept-floor 1400 -> 3500
    // once the best share's VERSION holds >=95% of the work-weighted desired-version
    // tally. Runtime sibling of the 60% version-switch gate (share_check.hpp step 2),
    // reading the SAME window. MUST be called under exclusive m_tracker_mutex.
    const uint32_t target  = PoolConfig::SHARE_MINIMUM_PROTOCOL_VERSION;  // 3500
    const uint32_t current =
        m_runtime_min_protocol_version.load(std::memory_order_relaxed);
    if (current >= target)                    // already ratcheted -> latched, no-op
        return;
    if (m_best_share_hash.IsNull() || !m_tracker.chain.contains(m_best_share_hash))
        return;

    // best share's VERSION (share TYPE version) + its parent hash
    // (oracle: previous_share = shares[best_share.previous_share_hash]).
    int64_t best_version = 0;
    uint256 prev_hash;
    m_tracker.chain.get_share(m_best_share_hash).invoke([&](auto* obj) {
        best_version = std::remove_pointer_t<decltype(obj)>::version;
        prev_hash    = obj->m_prev_hash;
    });
    if (prev_hash.IsNull() || !m_tracker.chain.contains(prev_hash))
        return;

    const int32_t CL            = static_cast<int32_t>(m_tracker.m_params->chain_length);
    const int32_t parent_height = m_tracker.chain.get_height(prev_hash);

    // Sample the SAME window the 60% switch gate reads (share_check.hpp:1610):
    // anchor = nth_parent(parent, CHAIN_LENGTH*9/10), size = CHAIN_LENGTH/10.
    const uint32_t window_start = (static_cast<uint32_t>(CL) * 9) / 10;
    const uint32_t window_size  =  static_cast<uint32_t>(CL) / 10;
    auto anchor  = m_tracker.chain.get_nth_parent_key(prev_hash, window_start);
    auto weights = m_tracker.get_desired_version_weights(
        anchor, static_cast<int32_t>(window_size));

    // apply_min_protocol_ratchet_decision applies the main.py:212 full-window
    // guard (parent_height >= CHAIN_LENGTH) the pure ratchet omits, then delegates.
    const uint32_t lifted = dgb::apply_min_protocol_ratchet_decision(
        parent_height, CL, weights, best_version, current, target);
    if (lifted != current) {
        m_runtime_min_protocol_version.store(lifted, std::memory_order_relaxed);
        LOG_INFO << "[min-proto-ratchet] MINIMUM_PROTOCOL_VERSION " << current
                 << " -> " << lifted << " (>=95% window work desires v"
                 << best_version << ", best="
                 << m_best_share_hash.GetHex().substr(0, 16) << ")";
    }
}

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<std::shared_mutex> lock(m_tracker_mutex, std::try_to_lock);
    if (!lock.owns_lock())
        return;

    if (!m_chain->contains(head))
        return;

    std::vector<uint256> 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)
{
    // download_shares(): C++ implementation of the p2pool share-download loop.
    //
    // 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.
    //
    // Delegate the stops construction to the SSOT (download_stops.hpp) so the
    // per-head nth-parent depth min(max(0,height-1),10), the head-inclusion,
    // the null-parent skip, and the 100-cap stay byte-identical to the p2pool
    // node.py download_shares oracle and are pinned by the download_stops
    // conformance KAT (#348). Only the pure stops computation moves; the
    // duplicate-request / fail-count / random-peer / random-parents node-local
    // logic above stays here. The std::set<uint256> ordering inside the SSOT
    // reproduces this region's prior inline std::set insert + ordered-iteration
    // (and 100-cap survivor selection) exactly.
    std::vector<uint256> heads;
    for (auto& [head_hash, tail_hash] : m_tracker.chain.get_heads())
        heads.push_back(head_hash);
    std::vector<uint256> stops = dgb::compute_download_stops(
        heads,
        [this](const uint256& h) { return m_tracker.chain.get_acc_height(h); },
        [this](const uint256& h, int nth) {
            return m_tracker.chain.get_nth_parent_via_skip(h, nth);
        });

    auto req_id = core::random::random_uint256();
    std::vector<uint256> 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<pool::Peer<dgb::Peer>> 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::milliseconds>(
        std::chrono::steady_clock::now() - t0).count();
    {
        std::set<uint256> 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<uint256> 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<uint8_t> 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<unsigned char> share_bytes(data.begin() + 8, data.end());
            PackStream ps(share_bytes);

            auto share = dgb::load_share(static_cast<int64_t>(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<std::pair<uint256, uint256>> 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()
{
    // ── have_tx / losing_tx delta sweep ──────────────────────────────────
    // Started BEFORE the outbound-dialing early-return below: a node with
    // outbound dialing disabled still accepts inbound peers and must still
    // advertise its tx pool to them. Advert-only, no consensus state.
    if (m_context)
    {
        m_tx_advert_timer = std::make_unique<core::Timer>(m_context, true);
        m_tx_advert_timer->start(core::TX_ADVERT_INTERVAL_SECONDS,
                                 [this]() { advertise_known_txs(); });
    }

    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<core::Timer>(m_context, true);
    m_connect_timer->start(30, try_connect_peers);
}

void NodeImpl::prune_shares(const uint256& /*best_share*/)
{
    // Tail-dropping (as in p2pool's clean_tracker):
    // - 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<int32_t>(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<int32_t>::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();
    // Recency-preserving bounded eviction (was: wholesale clear()). Dropping the
    // whole cache at once destroyed every forwardable tx byte, so a share relay
    // that referenced a just-dropped tx could no longer remember_tx-forward it
    // and the canonical peer disconnected with "referenced unknown transaction".
    // Evict oldest-first down to the cap, keeping the most-recently-learned txs.
    if (m_known_txs.size() > m_max_known_txs)
        core::evict_known_txs_to_cap(m_known_txs, m_known_txs_order, m_max_known_txs);

    // /p2p_stats gauges (observe-only): the SEND-side truth behind a peer
    // dashboard reporting TXPOOL=0 for us. Published here because prune_shares
    // is the periodic pass that already touches both containers. Relaxed
    // stores, no lock, no consensus/mint/payout state read or written.
    core::obs::p2p_stats().known_txs_size.store(
        m_known_txs.size(), std::memory_order_relaxed);
    core::obs::p2p_stats().known_txs_order_size.store(
        static_cast<std::int64_t>(m_known_txs_order.size()), std::memory_order_relaxed);
    core::obs::p2p_stats().known_txs_updated_at.store(
        static_cast<std::int64_t>(core::timestamp()), std::memory_order_relaxed);
    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<int32_t(uint256)>([](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::milliseconds>(
            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::milliseconds>(
            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;
            apply_min_protocol_ratchet();  // oracle main.py:216 (per best-share advance)
        }

        // Phase 1c: drive the whale-departure detector with the fresh best
        // share while we still hold the tracker lock exclusively (race-free).
        // detect() emits [WHALE-DEPARTURE]/[WHALE-RECOVERY] log lines itself;
        // we publish its verdict for the local work path (local_desired_target).
        m_whale_departure_active.store(
            m_whale_departure.detect(m_tracker, m_best_share_hash),
            std::memory_order_relaxed);

        // Phase 3L: log-based pool monitor on a ~30s cadence (same exclusive
        // tracker lock). Consensus-neutral — emits [MONITOR-*] lines only.
        {
            auto mon_now = static_cast<uint32_t>(std::time(nullptr));
            if (mon_now - m_last_monitor_ts >= MONITOR_INTERVAL_S) {
                m_last_monitor_ts = mon_now;
                m_pool_monitor.run_cycle(m_tracker, m_best_share_hash);
            }
        }

        // 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 <CHAIN_LENGTH.
            m_download_fail_count.clear();

            // Request desired shares from random peers
            if (!result.desired.empty() && !m_peers.empty()) {
                for (auto& [peer_addr, hash] : result.desired) {
                    auto peer_it = m_peers.begin();
                    if (m_peers.size() > 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::milliseconds>(
                    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<core::Timer>(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();
            publish_peer_info_snapshot();   // IO thread: keep uptime fresh

            // 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<int>(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<double>(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<int>(
            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 = dgb::compute_stale_prop(orphan_count, doa_count, total_recent);

    {
        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<std::string> 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<double>(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<int>(m_tracker.m_params->target_lookbehind)),
                /*min_work=*/false);
            double pool_hs = static_cast<double>(aps.GetLow64());
            double real_pool_hs = dgb::compute_real_pool_hashrate(pool_hs, stale_prop);
            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 = dgb::compute_expected_time_to_block(
                    static_cast<double>(block_aps.GetLow64()), real_pool_hs,
                    /*average_attempts_overflowed=*/block_aps.IsNull(),
                    /*block_target_nonzero=*/!block_target.IsNull());
            }
            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.
// C++ implementation of the p2pool clean_tracker() design (stale-head eating + tail dropping).
//
// 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<int32_t(uint256)>([](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::milliseconds>(
                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;
                apply_min_protocol_ratchet();  // oracle main.py:216 (per best-share advance)
            }

            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<int64_t>(std::time(nullptr));
        auto CL = static_cast<int32_t>(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<uint256> top5_set(m_last_top5_heads.begin(), m_last_top5_heads.end());

        for (int iter = 0; iter < 1000; ++iter)
        {
            std::vector<uint256> 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.
    // C++ implementation of the p2pool tail-dropping step.
    // 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<uint256> 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<int32_t(uint256)>([](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;
            apply_min_protocol_ratchet();  // oracle main.py:216 (per best-share advance)
        }
        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
}



// ── Admin API implementation ──────────────────────────────────────────










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_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<NetService> 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