path: root/src/rpc.cpp

// Copyright (c) 2009 Satoshi Nakamoto / Copyright (c) 2026 Bitflash developers
// Distributed under the MIT/X11 software license.
//
// Bitflash pool server -- .btf Stratum pool, operator side.
//
// Architecture:
//   PoolEventLoop  -- single thread, owns all miner fds, all job/share/payout
//                     state. No locks needed on any of that data.
//   AcceptOneFn    -- one thread per relay. Registers, waits to be paired,
//                     wraps the tunnel, pushes finished socket to gReadyQueue.
//   AcceptLoopFn   -- spawns AcceptOneFn threads, re-polls relay list every
//                     60s so dynamically discovered relays get picked up.
//
// GUI-visible state is snapshotted under gStatsMutex every 2s.
// Accept queue handoff uses gReadyMutex (held for microseconds).

#pragma push_macro("snprintf")
#undef snprintf
#include <nlohmann/json.hpp>
#pragma pop_macro("snprintf")

#include "headers.h"
#ifdef snprintf
#undef snprintf
#endif
#include "btftunnel.h"

#ifndef _WIN32
#include <sys/socket.h>
#include <sys/select.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#endif

#include <atomic>
#include <mutex>
#include <map>
#include <set>
#include <deque>
#include <vector>
#include <algorithm>
#include <sstream>
#include <iomanip>

using json = nlohmann::json;

#ifdef _WIN32
#define sock_close(s)  BtfCloseSocket(s)
#define SEND_FLAGS     0
#else
#define sock_close(s)  BtfCloseSocket(s)
#define SEND_FLAGS     MSG_NOSIGNAL
#endif

volatile bool gPoolRunning = false;

// ---------------------------------------------------------------------------
// Hex helpers
// ---------------------------------------------------------------------------
static std::string ToHex(const void* p, size_t n)
{
    const unsigned char* b = (const unsigned char*)p;
    std::string s; s.reserve(n * 2);
    static const char* H = "0123456789abcdef";
    for (size_t i = 0; i < n; i++) { s += H[b[i] >> 4]; s += H[b[i] & 0xf]; }
    return s;
}

static std::vector<unsigned char> FromHex(const std::string& s)
{
    std::vector<unsigned char> v;
    for (size_t i = 0; i + 1 < s.size(); i += 2) {
        auto h = [](char c) -> int {
            if (c >= '0' && c <= '9') return c - '0';
            if (c >= 'a' && c <= 'f') return c - 'a' + 10;
            if (c >= 'A' && c <= 'F') return c - 'A' + 10;
            return 0;
        };
        v.push_back((unsigned char)((h(s[i]) << 4) | h(s[i + 1])));
    }
    return v;
}

// ---------------------------------------------------------------------------
// Job -- written and read exclusively by the event loop thread
// ---------------------------------------------------------------------------
struct StratumJob {
    std::string jobId;
    CBlock      block;
    uint256     target;  // full block target; share targets are per-miner via vardiff
    int         height;
};
static StratumJob gCurrentJob;
static bool       gHaveJob = false;

// Compute a share target from a Stratum difficulty value.
// Stratum difficulty 1 must correspond to THIS chain's easiest allowed target
// (bnProofOfWorkLimit = ~uint256(0) >> 12, compact 0x1f0fffff), not Bitcoin's
// SHA256 diff-1 target (0x1d00ffff). Those differ by about 2^20 (~1e6x):
// Bitcoin's diff-1 needs ~2^32 hashes/share on average, while this chain's PoW
// floor needs only ~2^12. Using the Bitcoin constant made every miner's
// "difficulty 1" about a million times harder than the chain's own floor, so a
// worker could hash for minutes without ever finding a share -- and vardiff
// never gets the samples it needs to correct course, because no share ever
// comes in for it to measure. Base off the actual chain limit instead so
// difficulty 1 is genuinely easy and vardiff has real data to retarget from.
// Higher difficulty = smaller (harder) target.
static uint256 ShareTargetFromDifficulty(double difficulty)
{
    if (difficulty <= 0.0) difficulty = 1.0;
    static const uint256 diff1 = CBigNum(bnProofOfWorkLimit).getuint256();
    // shareTarget = diff1 / difficulty
    CBigNum bn; bn.setuint256(diff1);
    // Multiply by 2^16 to keep precision, divide, shift back
    bn <<= 16;
    bn /= (uint64)(difficulty * 65536.0);
    uint256 t = bn.getuint256();
    // Cap at all-ones (maximum, easiest possible)
    if (t < diff1 && difficulty < 1.0) t = ~uint256(0);
    return t;
}

// Expected number of hashes to find one share at a given Stratum difficulty,
// on this chain's scale (difficulty 1 == bnProofOfWorkLimit, which needs on
// average 2^12 = 4096 hashes). Used to turn "shares per second" into H/s.
static const double SHARE_HASHES_AT_DIFF_1 = 4096.0;
static double ExpectedHashesForDifficulty(double difficulty)
{
    if (difficulty <= 0.0) difficulty = 1.0;
    return difficulty * SHARE_HASHES_AT_DIFF_1;
}

// Rebuild the current job from the best chain tip.
// Must be called from the event loop thread (or startup, before the loop).
// Returns true on success. Logs the specific failure reason on false.
static bool RebuildJob()
{
    CRITICAL_BLOCK(cs_main)
    {
        if (!pindexBest) {
            LogPrint("pool", "RebuildJob: no chain tip (pindexBest is null)\n");
            return false;
        }

        // Don't issue jobs while we are behind the network: mining against a
        // stale tip produces blocks the network will orphan, and payout
        // entries that can never be paid (their coinbases never exist on the
        // main chain). Require at least one peer, and require that no peer
        // claims a longer chain than ours.
        if (vNodes.empty()) {
            LogPrint("pool", "RebuildJob: no peers yet -- not issuing jobs\n");
            return false;
        }
        // This used to compare the tip's timestamp against 30 minutes, which
        // was wrong in both directions and cost real money.
        //
        // It cannot tell "we are not synced" from "the network is slow". A
        // pool mining its own fork keeps its tip timestamp fresh forever, so
        // the check passed for the whole of heights 3966-4085 -- 120
        // consecutive blocks, every one of them orphaned, 5890.50 BTF of
        // payouts voided. Then, once the pool stopped finding blocks, the
        // same check refused to issue jobs against a tip that was perfectly
        // valid and was the network's own best, and the pool -- roughly 90%
        // of the hashrate -- could not restart the chain it was the majority
        // of. Stale tip, so no jobs; no jobs, so the tip stays stale.
        //
        // The height peers announce in their version message answers the
        // question the age check was only guessing at. Median, so one peer
        // claiming an absurd height cannot stall us, and one block of slack
        // to match WarnIfBehind, because somebody is always mid-relay. A
        // peer that predates the announcement reports -1 and is ignored; if
        // nobody has told us anything we mine our best chain, which is what
        // the node would do without a pool attached.
        int nPeerHeight = GetPeerMedianHeight();
        if (nPeerHeight >= 0 && nBestHeight < nPeerHeight - 1) {
            LogPrint("pool", "RebuildJob: %d blocks behind the network "
                             "(height %d, peers report %d) -- not issuing jobs\n",
                     nPeerHeight - nBestHeight, nBestHeight, nPeerHeight);
            return false;
        }

        // The comparison above is blind in the one case that matters most.
        //
        // When the node goes deaf -- peers still in vNodes, nothing arriving --
        // no peer announces anything, GetPeerMedianHeight() returns -1, and the
        // fallback ("nobody has told us anything, so mine our best chain") let
        // the pool build a private fork for as long as the deafness lasted.
        // That fallback is right for a node that has simply not been told yet
        // and wrong for a node that has stopped being able to hear, and a
        // height nobody sent cannot tell those apart. Time can.
        //
        // Deliberately not behind a debug category, and deliberately not a
        // silent refusal: an operator whose pool has stopped issuing jobs needs
        // the reason in the log. /poolmineblind is the escape hatch for a node
        // that really is alone -- without one, a genuinely isolated pool that is
        // the majority of the hashrate could never restart its own chain, which
        // is the deadlock the tip-age check used to cause.
        if (nPeerHeight < 0 && !fPoolMineBlind) {
            int64 nAge = GetPeerHeightAgeSecs();
            if (nAge >= BTF_NO_PEER_HEIGHT_MAX_SECS) {
                static int64 nLastSaid = 0;
                if (GetTime() - nLastSaid > 60) {
                    nLastSaid = GetTime();
                    printf("RebuildJob: no peer has reported a height for %d minutes "
                           "-- not issuing jobs (our height %d, %d peers connected). "
                           "This node is deaf; mining now would fork it. "
                           "Pass /poolmineblind to override.\n",
                           (int)(nAge / 60), nBestHeight, (int)vNodes.size());
                }
                return false;
            }
        }

        CBlockIndex* pindexPrev = pindexBest;
        unsigned int nBits = GetNextWorkRequired(pindexPrev);

        // From the pool, so the template's payout key is in wallet.dat before
        // any miner is handed work against it.
        vector<unsigned char> vchPubKey = GetKeyFromPool();
        static std::atomic<uint32_t> sExtra{0};
        uint32_t extraNonce = ++sExtra;

        CTransaction txNew;
        txNew.vin.resize(1);
        txNew.vin[0].prevout.SetNull();
        // Height first, for the reason given at the matching line in
        // BitcoinMiner: it is what keeps two coinbases at different heights
        // from ever being byte-identical. This builder was missed when that
        // went in, and it produces real blocks -- a pool operator's template
        // is mined and submitted like any other. #58.
        txNew.vin[0].scriptSig << (pindexPrev ? pindexPrev->nHeight + 1 : 0) << nBits << (CBigNum)extraNonce;
        txNew.vout.resize(1);
        txNew.vout[0].scriptPubKey << vchPubKey << OP_CHECKSIG;

        CBlock block;
        block.vtx.push_back(txNew);

        {
            CTxDB txdb("r");
            CRITICAL_BLOCK(cs_mapTransactions)
            {
                map<uint256, CTxIndex> pool;
                unsigned int sz = 0;
                for (auto& kv : mapTransactions) {
                    CTransaction& tx = kv.second;
                    if (tx.IsCoinBase() || !tx.IsFinal()) continue;
                    int64 nFees = 0;
                    map<uint256, CTxIndex> tmp(pool);
                    if (!tx.ConnectInputs(txdb, tmp, CDiskTxPos(1,1,1), 0, nFees,
                                          false, true, tx.GetMinFee(block.vtx.size() < 100)))
                        continue;
                    pool = tmp;
                    block.vtx.push_back(tx);
                    sz += ::GetSerializeSize(tx, SER_NETWORK);
                    if (sz > MAX_SIZE / 2) break;
                }
            }
        }

        block.vtx[0].vout[0].nValue = block.GetBlockValue(pindexPrev->nHeight + 1, 0);
        block.hashPrevBlock  = pindexPrev->GetBlockHash();
        block.hashMerkleRoot = block.BuildMerkleTree();
        block.nTime = max((unsigned int)(pindexPrev->GetMedianTimePast() + 1),
                          (unsigned int)GetAdjustedTime());
        block.nBits  = nBits;
        block.nNonce = 1;
        // No AddKey here: the pool wrote this key to wallet.dat before it was
        // handed out.

        StratumJob job;
        job.jobId       = ToHex(&extraNonce, 4);
        job.block       = block;
        job.target      = CBigNum().SetCompact(nBits).getuint256();
        job.height      = nBestHeight + 1;

        gCurrentJob = job;
        gHaveJob    = true;
        return true;
    }
    return false;
}

// ---------------------------------------------------------------------------
// Payouts -- event loop thread only
// ---------------------------------------------------------------------------
struct PendingPayout {
    // The block that funds this payout, identified by hash and not by height
    // alone. A reorg replaces the block at a height without changing the
    // height, so a height-keyed obligation outlives the coinbase that was
    // supposed to pay it: a 280-block reorg once left ~14.4k BTF of debt
    // owed against blocks that no longer existed, which then drained the
    // wallet paying fork-era miners out of real-chain coin.
    int                          blockHeight = 0;
    // Must be explicitly zeroed: uint256's default constructor is empty
    // (uint256.h:514), so leaving this unassigned gives every ledger entry
    // loaded from an older file a garbage hash, which then matches no block
    // and gets voided as "orphaned". That mistake voided a live payout.
    uint256                      blockHash = 0;  // zero = entry predates hash keying
    int                          matureAtHeight;
    std::map<std::string, int64> amounts; // address -> satoshis
    int                          retries = 0;    // exec-thread only: SendMoney attempts so far
    int                          deferrals = 0;  // exec-thread only: waits for spendable coin
    int64                        retryAfter = 0; // exec-thread only: unix time, 0 = ready now
};
static std::vector<PendingPayout> gPendingPayouts;

static std::string PayoutFilePath()
{
    return GetAppDir() + "/pending_payouts.json";
}

static void SavePendingPayouts()
{
    nlohmann::json arr = nlohmann::json::array();
    for (const auto& pp : gPendingPayouts) {
        nlohmann::json obj;
        obj["blockHeight"]    = pp.blockHeight;
        obj["blockHash"]      = pp.blockHash.ToString();
        obj["matureAtHeight"] = pp.matureAtHeight;
        nlohmann::json amounts = nlohmann::json::object();
        for (const auto& kv : pp.amounts)
            amounts[kv.first] = kv.second;
        obj["amounts"] = amounts;
        arr.push_back(obj);
    }
    std::string path = PayoutFilePath();
    FILE* f = fopen(path.c_str(), "w");
    if (f) {
        std::string s = arr.dump(2);
        fwrite(s.c_str(), 1, s.size(), f);
        fclose(f);
    } else {
        LogPrint("payout", "[payout] WARNING: could not write %s\n", path.c_str());
    }
}

static void LoadPendingPayouts()
{
    std::string path = PayoutFilePath();
    FILE* f = fopen(path.c_str(), "r");
    if (!f) return; // no file yet -- first run, nothing to load
    fseek(f, 0, SEEK_END);
    long sz = ftell(f);
    fseek(f, 0, SEEK_SET);
    std::string s(sz, '\0');
    fread(&s[0], 1, sz, f);
    fclose(f);
    try {
        nlohmann::json arr = nlohmann::json::parse(s);
        for (const auto& obj : arr) {
            PendingPayout pp;
            pp.matureAtHeight = obj["matureAtHeight"].get<int>();
            // Entries written before payouts were keyed by hash carry neither
            // field. Leave blockHash zero and they are grandfathered past the
            // orphan check rather than voided on a guess.
            if (obj.contains("blockHash") && obj["blockHash"].is_string())
                pp.blockHash.SetHex(obj["blockHash"].get<std::string>());
            if (obj.contains("blockHeight") && obj["blockHeight"].is_number())
                pp.blockHeight = obj["blockHeight"].get<int>();
            for (auto it = obj["amounts"].begin(); it != obj["amounts"].end(); ++it)
                pp.amounts[it.key()] = it.value().get<int64>();
            gPendingPayouts.push_back(pp);
        }
        LogPrint("payout", "[payout] loaded %zu pending payout(s) from disk\n",
                 gPendingPayouts.size());
    } catch (const std::exception& e) {
        LogPrint("payout", "[payout] WARNING: failed to parse %s: %s\n",
                 path.c_str(), e.what());
    }
}

// A payment in flight: one record per (payout entry, address) pair, which is
// the granularity SendMoney works at and therefore the granularity at which a
// payment can be half-done when the process dies.
//
// This list is persisted. It used to be a memory-only queue, and matured
// entries were erased from pending_payouts.json *before* being handed to it,
// so any restart in between lost those payments with no record that anyone
// was owed anything: a container restart on 2026-07-30 silently dropped
// ~1,156 BTF that way. Writing the records before removing the ledger entry,
// and reconciling against the wallet on startup, is what makes a restart
// survivable.
struct PayingRecord {
    int         blockHeight    = 0;
    uint256     blockHash      = 0; // zero = predates hash keying, see PendingPayout
    int         matureAtHeight = 0;
    std::string address;
    int64       amount     = 0;
    int         retries    = 0;
    int         deferrals  = 0;
    int64       retryAfter = 0;
};
static std::vector<PayingRecord> gPaying; // guarded by gPayoutExecMutex
static std::mutex                gPayoutExecMutex;
static std::atomic<bool>         gPayoutThreadRunning{false};

static std::string PayingFilePath()
{
    return GetAppDir() + "/paying_payouts.json";
}

// Caller must hold gPayoutExecMutex.
static void SavePayingLocked()
{
    nlohmann::json arr = nlohmann::json::array();
    for (const auto& r : gPaying) {
        nlohmann::json obj;
        obj["blockHeight"]    = r.blockHeight;
        obj["blockHash"]      = r.blockHash.ToString();
        obj["matureAtHeight"] = r.matureAtHeight;
        obj["address"]        = r.address;
        obj["amount"]         = r.amount;
        obj["retries"]        = r.retries;
        obj["deferrals"]      = r.deferrals;
        arr.push_back(obj);
    }
    std::string path = PayingFilePath();
    // Write to a temporary file and rename, so a crash mid-write cannot leave
    // a truncated list of who is still owed money.
    std::string tmp = path + ".tmp";
    FILE* f = fopen(tmp.c_str(), "w");
    if (!f) {
        LogPrint("payout", "[payout] WARNING: could not write %s\n", tmp.c_str());
        return;
    }
    std::string s = arr.dump(2);
    fwrite(s.c_str(), 1, s.size(), f);
    fflush(f);
    fclose(f);
    if (rename(tmp.c_str(), path.c_str()) != 0)
        LogPrint("payout", "[payout] WARNING: could not rename %s -> %s\n",
                 tmp.c_str(), path.c_str());
}

static void LoadPaying()
{
    std::string path = PayingFilePath();
    FILE* f = fopen(path.c_str(), "r");
    if (!f) return;
    fseek(f, 0, SEEK_END);
    long sz = ftell(f);
    fseek(f, 0, SEEK_SET);
    std::string s(sz, '\0');
    fread(&s[0], 1, sz, f);
    fclose(f);
    try {
        nlohmann::json arr = nlohmann::json::parse(s);
        for (const auto& obj : arr) {
            PayingRecord r;
            r.address        = obj["address"].get<std::string>();
            r.amount         = obj["amount"].get<int64>();
            r.matureAtHeight = obj["matureAtHeight"].get<int>();
            if (obj.contains("blockHash") && obj["blockHash"].is_string())
                r.blockHash.SetHex(obj["blockHash"].get<std::string>());
            if (obj.contains("blockHeight") && obj["blockHeight"].is_number())
                r.blockHeight = obj["blockHeight"].get<int>();
            if (obj.contains("retries") && obj["retries"].is_number())
                r.retries = obj["retries"].get<int>();
            if (obj.contains("deferrals") && obj["deferrals"].is_number())
                r.deferrals = obj["deferrals"].get<int>();
            gPaying.push_back(r);
        }
        LogPrint("payout", "[payout] loaded %zu in-flight payment(s) from disk\n",
                 gPaying.size());
    } catch (const std::exception& e) {
        LogPrint("payout", "[payout] WARNING: failed to parse %s: %s\n",
                 path.c_str(), e.what());
    }
}

// Is the block that funds this payout still on the main chain?
//
// A zero hash means the entry was written before payouts recorded one, so
// there is nothing to check and we assume it is good -- voiding those would
// mean refusing to pay miners on the strength of a missing field.
static bool PayoutBlockStillOnChain(const uint256& blockHash)
{
    if (blockHash == 0)
        return true;
    CRITICAL_BLOCK(cs_main)
    {
        std::map<uint256, CBlockIndex*>::iterator mi = mapBlockIndex.find(blockHash);
        if (mi == mapBlockIndex.end() || mi->second == NULL)
            return false;
        return mi->second->IsInMainChain();
    }
    return false;
}

static std::string PayoutComment(int blockHeight)
{
    return strprintf("Pool payout height %d", blockHeight);
}

// Did we already send this payment before dying? Looks for a wallet
// transaction of ours carrying this payout's comment and paying exactly this
// amount to this address.
//
// Biased towards reporting "already paid" on any doubt: skipping a payment a
// miner is owed is visible and recoverable from the log, while paying twice
// is neither.
static bool WalletAlreadyPaid(const PayingRecord& r)
{
    uint160 h160;
    if (!AddressToHash160(r.address, h160))
        return false;
    CScript want;
    want << OP_DUP << OP_HASH160 << h160 << OP_EQUALVERIFY << OP_CHECKSIG;
    const std::string comment = PayoutComment(r.blockHeight);

    CRITICAL_BLOCK(cs_mapWallet)
    {
        for (std::map<uint256, CWalletTx>::iterator it = mapWallet.begin();
             it != mapWallet.end(); ++it) {
            CWalletTx& wtx = it->second;
            // The comment is written only by the payout path, so its presence
            // already identifies the transaction as one of ours.
            std::map<std::string, std::string>::const_iterator ci =
                wtx.mapValue.find("comment");
            if (ci == wtx.mapValue.end() || ci->second != comment)
                continue;
            foreach (const CTxOut& txout, wtx.vout)
                if (txout.nValue == r.amount && txout.scriptPubKey == want)
                    return true;
        }
    }
    return false;
}

// Startup reconciliation. Drops in-flight records the wallet shows were
// already sent, and drops ledger entries that were already exploded into
// records (which happens if we died between writing the records and rewriting
// the ledger -- both files then describe the same debt).
static void ReconcilePayouts()
{
    std::lock_guard<std::mutex> lk(gPayoutExecMutex);

    size_t settled = 0;
    int64  settledAmount = 0;
    for (auto it = gPaying.begin(); it != gPaying.end(); ) {
        if (WalletAlreadyPaid(*it)) {
            settledAmount += it->amount;
            settled++;
            it = gPaying.erase(it);
        } else {
            ++it;
        }
    }
    if (settled) {
        LogPrint("payout", "[payout] %zu in-flight payment(s) totalling %s were"
                 " already sent before the last shutdown -- not resending\n",
                 settled, FormatMoney(settledAmount).c_str());
        SavePayingLocked();
    }

    size_t dropped = 0;
    for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ) {
        bool exploded = false;
        for (const auto& r : gPaying)
            if (r.matureAtHeight == it->matureAtHeight && r.blockHash == it->blockHash) {
                exploded = true;
                break;
            }
        if (exploded) {
            LogPrint("payout", "[payout] ledger entry for block %d is already"
                     " in flight -- dropping the duplicate\n", it->blockHeight);
            it = gPendingPayouts.erase(it);
            dropped++;
        } else {
            ++it;
        }
    }
    if (dropped) SavePendingPayouts();

    if (!gPaying.empty()) {
        int64 owed = 0;
        for (const auto& r : gPaying) owed += r.amount;
        LogPrint("payout", "[payout] resuming %zu in-flight payment(s) totalling %s\n",
                 gPaying.size(), FormatMoney(owed).c_str());
    }
}

static void QueuePayouts(int blockHeight,
                         const uint256& blockHash,
                         const std::map<std::string, uint64>& shareCount,
                         uint64 total)
{
    if (total == 0) {
        LogPrint("payout", "[payout] block %d: zero shares recorded -- nothing to pay out\n",
                 blockHeight);
        return;
    }

    double feeFrac = dPoolFeePercent / 100.0;
    if (feeFrac < 0.0) feeFrac = 0.0;
    if (feeFrac > 1.0) feeFrac = 1.0;

    // Compute the block subsidy at the height that was just found.
    // We do not include tx fees in miner payout -- those stay in the operator
    // wallet as part of the coinbase, which is the operator's cut on top of feeFrac.
    int64 nSubsidy = 50 * COIN;
    nSubsidy >>= (blockHeight / 210000); // halving schedule
    int64 minerReward = (int64)((double)nSubsidy * (1.0 - feeFrac));

    PendingPayout pp;
    pp.blockHeight = blockHeight;
    pp.blockHash   = blockHash;
    // COINBASE_SPENDABLE_DEPTH, not COINBASE_MATURITY. Paying at depth 100
    // means the block's own coinbase is still 20 blocks short of spendable, so
    // the payment has to come out of some *other* mature coin -- the pool runs
    // a permanent ~19-block float and every payout fails until unrelated coin
    // matures. The earlier "96% of 934 payments went through at depth 100"
    // measurement that justified the shorter wait was taken while this node was
    // alone on a private fork paying itself, so it says nothing about a pool
    // competing on the real chain. Waiting the extra 20 blocks makes each
    // payout funded by construction, and puts the maturity gate past the
    // reorg depth that orphaned 280 blocks of obligations.
    pp.matureAtHeight = blockHeight + COINBASE_SPENDABLE_DEPTH;

    int skipped = 0;
    int64 burned = 0; // amount owed to miners with bad/empty addresses
    for (auto& kv : shareCount) {
        int64 amount = (int64)(((double)kv.second / (double)total) * (double)minerReward);
        if (amount < CENT) {
            skipped++;
            continue;
        }
        uint160 addrCheck;
        if (kv.first.empty() || !AddressToHash160(kv.first, addrCheck)) {
            // Address was invalid at authorize time (warned then). Their share
            // of the reward cannot be paid -- log the amount so it's visible.
            burned += amount;
            LogPrint("payout", "[payout] WARNING: %s owed to miner with invalid"
                     " address '%s' -- cannot pay out, amount forfeited\n",
                     FormatMoney(amount).c_str(), kv.first.c_str());
            continue;
        }
        pp.amounts[kv.first] = amount;
    }

    gPendingPayouts.push_back(pp);
    SavePendingPayouts();

    int64 operatorCut = nSubsidy - minerReward;
    LogPrint("payout",
             "[payout] block %d: subsidy %s BTF, miners get %s"
             " (fee %.2f%% = %s to operator),"
             " %zu recipients, matures height %d,"
             " %d below dust%s\n",
             blockHeight,
             FormatMoney(nSubsidy).c_str(),
             FormatMoney(minerReward).c_str(),
             dPoolFeePercent,
             FormatMoney(operatorCut).c_str(),
             pp.amounts.size(), pp.matureAtHeight, skipped,
             burned > 0 ? (", " + FormatMoney(burned) + " forfeited (bad address)").c_str()
                        : "");
}

// Called from event loop: move matured entries into the in-flight list (fast).
//
// Ordering matters and is deliberate: the in-flight records are written and
// fsync'd to their own file *before* the ledger entry is erased, so the debt
// is always described by at least one of the two files. Dying in the window
// where both describe it is handled at startup by ReconcilePayouts, which
// drops the ledger entry it can see is already in flight.
static void FlushMaturePayouts()
{
    // Void obligations whose funding block lost a reorg. The coin that was
    // meant to pay them does not exist, so paying anyway spends real-chain
    // coin the pool owes to somebody else.
    bool voided = false;
    for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ) {
        if (PayoutBlockStillOnChain(it->blockHash)) { ++it; continue; }
        int64 total = 0;
        for (auto& kv : it->amounts) total += kv.second;
        LogPrint("payout", "[payout] block %d (%s) was orphaned -- voiding %zu"
                 " payment(s) totalling %s\n",
                 it->blockHeight, it->blockHash.ToString().substr(0, 14).c_str(),
                 it->amounts.size(), FormatMoney(total).c_str());
        it = gPendingPayouts.erase(it);
        voided = true;
    }
    if (voided) SavePendingPayouts();

    std::vector<PendingPayout> ready;
    for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ) {
        if (nBestHeight < it->matureAtHeight) { ++it; continue; }
        LogPrint("payout", "[payout] block height %d matured -- queuing %zu payments\n",
                 it->matureAtHeight, it->amounts.size());
        ready.push_back(*it);   // copy first; erase after save
        ++it;
    }
    if (!ready.empty()) {
        // Record the individual payments first, and only then drop the ledger
        // entries that produced them.
        {
            std::lock_guard<std::mutex> lk(gPayoutExecMutex);
            for (const auto& pp : ready) {
                for (const auto& kv : pp.amounts) {
                    PayingRecord r;
                    r.blockHeight    = pp.blockHeight;
                    r.blockHash      = pp.blockHash;
                    r.matureAtHeight = pp.matureAtHeight;
                    r.address        = kv.first;
                    r.amount         = kv.second;
                    gPaying.push_back(r);
                }
            }
            SavePayingLocked();
        }
        for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ) {
            if (nBestHeight >= it->matureAtHeight)
                it = gPendingPayouts.erase(it);
            else
                ++it;
        }
        SavePendingPayouts();
    }
}

// Identity of a payment, for finding its record again after SendMoney.
// amounts is a map keyed by address, so (block, address) is already unique.
static bool SamePayment(const PayingRecord& a, const PayingRecord& b)
{
    return a.matureAtHeight == b.matureAtHeight
        && a.blockHash      == b.blockHash
        && a.address        == b.address
        && a.amount         == b.amount;
}

// Settle a payment: it is either done or abandoned, either way it stops being
// owed. Removing the record and rewriting the file is what makes the payment
// not happen twice after a restart.
static void ForgetPayment(const PayingRecord& r)
{
    std::lock_guard<std::mutex> lk(gPayoutExecMutex);
    for (auto it = gPaying.begin(); it != gPaying.end(); ++it)
        if (SamePayment(*it, r)) { gPaying.erase(it); break; }
    SavePayingLocked();
}

// Persist an updated attempt count / backoff for a payment still owed.
static void ReschedulePayment(const PayingRecord& r)
{
    std::lock_guard<std::mutex> lk(gPayoutExecMutex);
    for (auto& e : gPaying)
        if (SamePayment(e, r)) {
            e.retries    = r.retries;
            e.deferrals  = r.deferrals;
            e.retryAfter = r.retryAfter;
            break;
        }
    SavePayingLocked();
}

// Payout execution thread: works through gPaying, calls SendMoney.
// Runs independently of the event loop so slow wallet operations
// (SendMoney takes cs_main) never stall miner I/O.
static void PayoutThreadFn(void*)
{
    gPayoutThreadRunning = true;
    LogPrint("payout", "[payout] payout thread started\n");

    while (!fShutdown && gPoolRunning) {
        std::vector<PayingRecord> batch;
        {
            std::lock_guard<std::mutex> lk(gPayoutExecMutex);
            int64 now = GetTime();
            for (const auto& r : gPaying)
                if (r.retryAfter <= now)
                    batch.push_back(r);
        }

        for (PayingRecord r : batch) {
            // Re-check here and not only at maturity: a payment can sit in this
            // list for hours across retries, and the reorg that kills its
            // funding block can land at any point in that window.
            if (!PayoutBlockStillOnChain(r.blockHash)) {
                LogPrint("payout", "[payout] block %d (%s) orphaned while queued"
                         " -- dropping %s to %s\n",
                         r.blockHeight, r.blockHash.ToString().substr(0, 14).c_str(),
                         FormatMoney(r.amount).c_str(), r.address.c_str());
                ForgetPayment(r);
                continue;
            }

            uint160 h160;
            if (!AddressToHash160(r.address, h160)) {
                // Should not happen -- bad addresses are filtered in QueuePayouts
                LogPrint("payout", "[payout] SKIPPED bad address '%s'\n",
                         r.address.c_str());
                ForgetPayment(r);
                continue;
            }
            CScript sc;
            sc << OP_DUP << OP_HASH160 << h160 << OP_EQUALVERIFY << OP_CHECKSIG;

            // Waiting for coin to mature is not a failure, so it must not
            // consume the retry budget. Burning a retry every 60s means a
            // payment the pool fully intends to honour can hit GIVING UP
            // 24h later purely because the subsidy funding it had not
            // matured yet -- the miner is then silently never paid.
            int64 nSpendable = GetBalance();
            if (nSpendable < r.amount) {
                r.deferrals++;
                r.retryAfter = GetTime() + 60;
                // Hourly, not every minute: a dry pool would otherwise
                // write one line per payment per minute forever.
                if (r.deferrals == 1 || r.deferrals % 60 == 0)
                    LogPrint("payout", "[payout] deferring %s to %s -- only %s"
                             " spendable (waiting %d min so far)\n",
                             FormatMoney(r.amount).c_str(), r.address.c_str(),
                             FormatMoney(nSpendable).c_str(), r.deferrals);
                ReschedulePayment(r);
                continue;
            }

            CWalletTx wtx;
            wtx.mapValue["comment"] = PayoutComment(r.blockHeight);

            if (SendMoney(sc, r.amount, wtx)) {
                // The wallet holds the transaction before this returns, so a
                // crash here is recoverable: ReconcilePayouts finds it by
                // comment+address+amount and does not resend.
                ForgetPayment(r);
                LogPrint("payout", "[payout] paid %s to %s (tx %s)\n",
                         FormatMoney(r.amount).c_str(), r.address.c_str(),
                         wtx.GetHash().ToString().substr(0, 12).c_str());
            } else if (r.retries < 1440) {
                // Transient failures (reorg, locked wallet) must not silently
                // burn the payment -- retry with backoff. Insufficient funds
                // never lands here; it defers above.
                r.retries++;
                r.retryAfter = GetTime() + 60;
                ReschedulePayment(r);
                LogPrint("payout", "[payout] FAILED to pay %s to %s"
                         " (%s spendable at the time) -- retry %d/1440 in 60s\n",
                         FormatMoney(r.amount).c_str(), r.address.c_str(),
                         FormatMoney(nSpendable).c_str(), r.retries);
            } else {
                // Reaching here means 1440 attempts that were NOT
                // insufficient-funds waits (those defer instead) against a
                // block still on the main chain -- i.e. a real wallet fault,
                // not a maturity or reorg question.
                LogPrint("payout", "[payout] GIVING UP on %s to %s after %d"
                         " retries -- block %d is still on-chain and %s was"
                         " spendable, so this is a wallet fault, not maturity"
                         " -- payment forfeited\n",
                         FormatMoney(r.amount).c_str(), r.address.c_str(),
                         r.retries, r.blockHeight,
                         FormatMoney(nSpendable).c_str());
                ForgetPayment(r);
            }
        }

        Sleep(500); // check for new work every 500ms
    }

    gPayoutThreadRunning = false;
    LogPrint("payout", "[payout] payout thread stopped\n");
}

// ---------------------------------------------------------------------------
// GUI stats snapshot -- event loop writes, GUI threads read
// ---------------------------------------------------------------------------
struct StatsSnapshot {
    int    authorizedMiners = 0;
    int    blocksFound      = 0;
    uint64 roundShares      = 0;
    double totalHashRate    = 0.0;
};
static StatsSnapshot                   gStats;
static std::vector<PoolWorkerStatView> gWorkerStats;
static std::vector<PendingPayoutView>  gPayoutViews;
static std::mutex                      gStatsMutex;

void GetPoolOperatorStats(int& authorizedMiners, int& blocksFound, uint64& roundShares, double& totalHashRate)
{
    std::lock_guard<std::mutex> lk(gStatsMutex);
    authorizedMiners = gStats.authorizedMiners;
    blocksFound      = gStats.blocksFound;
    roundShares      = gStats.roundShares;
    totalHashRate    = gStats.totalHashRate;
}

void GetPoolWorkerStats(std::vector<PoolWorkerStatView>& out)
{
    std::lock_guard<std::mutex> lk(gStatsMutex);
    out = gWorkerStats;
}

void GetPendingPayouts(std::vector<PendingPayoutView>& out)
{
    std::lock_guard<std::mutex> lk(gStatsMutex);
    out = gPayoutViews;
}

// ---------------------------------------------------------------------------
// Miner state -- event loop thread only
// ---------------------------------------------------------------------------
static std::atomic<uint32_t> gNextExtranonce2{1};

// Vardiff target: aim for one share every VARDIFF_TARGET_SECS seconds per miner.
// After VARDIFF_SHARES_PER_CHECK shares, recalculate and adjust if rate is off
// by more than 50%. Shares are the stable unit; difficulty floats to match.
static const int64  VARDIFF_TARGET_SECS       = 15;  // one share every 15s
static const uint64 VARDIFF_SHARES_PER_CHECK  = 5;   // re-evaluate every 5 shares
static const double VARDIFF_RETARGET_RATIO    = 1.5; // adjust if >50% off target
static const double VARDIFF_MIN               = 0.001;
static const double VARDIFF_MAX               = 1e12;
// If a miner has gone this long since its vardiff window opened with ZERO
// shares, the normal vardiff retarget (which only fires after
// VARDIFF_SHARES_PER_CHECK shares) can never trigger -- it has no samples to
// act on. This is the residual gap the ShareTargetFromDifficulty fix above
// doesn't cover: even with a correctly-easy diff-1 target, a miner stuck in
// RandomX light mode (or otherwise much slower than expected) can go
// indefinitely without a single share, and difficulty would never drop to
// meet it. Force a difficulty cut on a timeout instead of waiting for shares
// that will never come.
static const int64  VARDIFF_STALL_SECS        = 45;  // 3x share target -- reacts faster without false-triggering on normal variance
static const double VARDIFF_STALL_CUT         = 4.0; // divide difficulty by this much

struct Miner {
    btf_socket_t fd;
    bool         authorised    = false;
    std::string  address;
    std::string  worker;
    std::string  readBuf;
    int64        connectTime;
    uint64       sessionShares  = 0;
    uint64       roundShares    = 0;
    uint64       totalShares    = 0;
    int64        lastSeen       = 0;
    uint32_t     extranonce2;

    // Vardiff state
    double       difficulty     = 1.0;   // current share difficulty sent to this miner
    uint256      shareTarget;            // derived from difficulty; what we check against
    int64        vardiffWindowStart = 0; // when the current vardiff window opened
    uint64       vardiffShares  = 0;     // shares found in current window

    // Hashrate estimate -- EWMA of (expected hashes per share at current
    // difficulty) / (seconds since the previous share). We had all the data
    // to compute this (difficulty + share timing) but never actually did,
    // so pool/GUI hashrate always showed as zero/missing.
    int64        lastShareTime  = 0;     // GetTime() of the previous accepted share
    double       hashRate       = 0.0;   // smoothed H/s estimate

    explicit Miner(btf_socket_t f)
        : fd(f), connectTime(GetTime()),
          extranonce2(gNextExtranonce2.fetch_add(1)) {}
};

static std::vector<Miner*> gMiners;

struct FoundBlock { int height; int64 ts; std::string addr; };
static std::deque<FoundBlock> gFoundBlocks;

// ---------------------------------------------------------------------------
// Accept queue -- accept threads push; event loop pops at top of each tick
// ---------------------------------------------------------------------------
static std::vector<btf_socket_t> gReadyQueue;
static std::mutex                gReadyMutex;

static void PushReady(btf_socket_t s)
{
    std::lock_guard<std::mutex> lk(gReadyMutex);
    gReadyQueue.push_back(s);
}

// Loopback stratum listener: lets a miner running on the SAME machine reach
// this pool over plain TCP (127.0.0.1:8334), skipping the .btf rendezvous
// entirely. The rendezvous relay only allows one service registration per
// pubkey, so the pool service and the node's own peer-accept fight over the
// node identity at the relay -- loopback avoids that fight for local miners.
static void LocalAcceptFn(void*)
{
    SOCKET ls = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (ls == INVALID_SOCKET)
        return;
    int yes = 1;
    setsockopt(ls, SOL_SOCKET, SO_REUSEADDR, (char*)&yes, sizeof(yes));
    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
    addr.sin_port = htons(8334);
    if (bind(ls, (struct sockaddr*)&addr, sizeof(addr)) != 0 || listen(ls, 16) != 0)
    {
        LogPrint("pool", "[pool] loopback stratum bind/listen failed on 127.0.0.1:8334\n");
        closesocket(ls);
        return;
    }
    LogPrint("pool", "[pool] loopback stratum listening on 127.0.0.1:8334\n");
    while (!fShutdown && gPoolRunning)
    {
        SOCKET c = accept(ls, NULL, NULL);
        if (c == INVALID_SOCKET)
        {
            if (fShutdown || !gPoolRunning)
                break;
            continue;
        }
        LogPrint("pool", "[pool] miner arrived via loopback -- pushing to event loop\n");
        PushReady((btf_socket_t)c);
    }
    closesocket(ls);
}

// ---------------------------------------------------------------------------
// I/O helpers -- event loop thread only
// ---------------------------------------------------------------------------

// Blocking send of a complete JSON line. Called only from the event loop so
// no other thread is writing to the same fd concurrently.
static bool SendLine(btf_socket_t fd, const json& j)
{
    std::string s = j.dump() + "\n";
    int off = 0, total = (int)s.size();
    while (off < total) {
        int r = send(fd, s.c_str() + off, total - off, SEND_FLAGS);
        if (r <= 0) return false;
        off += r;
    }
    return true;
}

// Ceilings on what an unauthenticated peer can make the pool operator hold.
// Stratum lines are a few hundred bytes and a real pool this size will not see
// anything near a thousand miners.
static const size_t MAX_STRATUM_LINE = 64 * 1024;
static const size_t MAX_POOL_MINERS  = 1000;

// Non-blocking recv: pull whatever bytes are available now into buf, then
// return one complete line if one is ready. Three outcomes:
//   true  -- line is filled; buf may still contain more lines
//   false, errno EAGAIN/EWOULDBLOCK (or FIONREAD==0) -- no data right now
//   false, r==0 -- clean disconnect; caller must treat fd as dead
// Caller distinguishes disconnect from "not yet" by checking whether
// line.empty() after a false return (line is "" on both paths, so we
// use a separate out-param for the disconnect signal).
static bool RecvLine(btf_socket_t fd, std::string& buf,
                     std::string& line, bool& disconnected)
{
    disconnected = false;
    char tmp[4096];

#ifdef _WIN32
    u_long avail = 0;
    ioctlsocket(fd, FIONREAD, &avail);
    if (avail > 0) {
        int r = recv(fd, tmp, (int)std::min((u_long)(sizeof(tmp)-1), avail), 0);
        if (r > 0)      { tmp[r] = 0; buf += tmp; }
        else if (r == 0){ disconnected = true; return false; }
        // r < 0: WSAEWOULDBLOCK -- fall through with what we have
    }
#else
    {
        int r = recv(fd, tmp, sizeof(tmp) - 1, MSG_DONTWAIT);
        if (r > 0)       { tmp[r] = 0; buf += tmp; }
        else if (r == 0) { disconnected = true; return false; }
        // r < 0: EAGAIN/EWOULDBLOCK -- fall through with what we have
    }
#endif

    // A peer that never sends a newline would otherwise grow this buffer for
    // as long as it keeps writing, and it does not have to authorize first --
    // the read happens before any of that. Stratum lines are a few hundred
    // bytes; anything past the cap is not a slow line, it is a flood.
    if (buf.size() > MAX_STRATUM_LINE) {
        disconnected = true;
        buf.clear();
        return false;
    }

    size_t pos = buf.find('\n');
    if (pos == std::string::npos) return false; // no complete line yet

    line = buf.substr(0, pos);
    if (!line.empty() && line.back() == '\r') line.pop_back();
    buf = buf.substr(pos + 1);
    return true;
}

// ---------------------------------------------------------------------------
// Job helpers
// ---------------------------------------------------------------------------
static CBlock BuildMinerBlock(const StratumJob& job, uint32_t extranonce2)
{
    CBlock blk = job.block;
    blk.vtx[0].vin[0].scriptSig << (CBigNum)extranonce2;
    blk.hashMerkleRoot = blk.BuildMerkleTree();
    return blk;
}

// shareTarget is per-miner (set by vardiff) and passed explicitly so the
// value in p[3] of mining.notify always matches what HandleLine checks on submit.
// This is the single source of truth for share difficulty.
static json MakeNotifyParams(const StratumJob& job, const CBlock& b,
                              const uint256& shareTarget, bool clean)
{
    unsigned char hdr[80];
    memcpy(hdr,    &b.nVersion,       4);
    memcpy(hdr+4,  &b.hashPrevBlock,  32);
    memcpy(hdr+36, &b.hashMerkleRoot, 32);
    memcpy(hdr+68, &b.nTime,          4);
    memcpy(hdr+72, &b.nBits,          4);
    memcpy(hdr+76, &b.nNonce,         4);
    unsigned char tgt[32]; memcpy(tgt, &shareTarget, 32);
    return json::array({job.jobId, ToHex(hdr, 80),
                        b.hashPrevBlock.GetHex(), ToHex(tgt, 32), clean});
}

static void SendJob(Miner* m, bool clean)
{
    if (!gHaveJob) return;
    CBlock mb = BuildMinerBlock(gCurrentJob, m->extranonce2);
    // Use this miner's own shareTarget (set by vardiff) in the notify params.
    // The participant miner reads p[3] as its share target, and HandleLine
    // checks m->shareTarget on submit -- both sides now use the same value.
    json n = {{"id",nullptr},{"method","mining.notify"},
              {"params", MakeNotifyParams(gCurrentJob, mb, m->shareTarget, clean)}};
    if (!SendLine(m->fd, n))
        LogPrint("worker", "[pool->worker] job send FAILED for %s -- miner will be dropped\n",
                 m->address.empty() ? "(not yet authorized)" : m->address.c_str());
}

static void BroadcastJob(bool clean)
{
    if (!gHaveJob) return;
    int sent = 0;
    for (Miner* m : gMiners) {
        if (!m->authorised) continue;
        SendJob(m, clean);
        sent++;
    }
    LogPrint("pool", "[pool] job %s broadcast to %d miner(s)\n",
             gCurrentJob.jobId.c_str(), sent);
}

// ---------------------------------------------------------------------------
// Stratum message handler -- called for each complete line from a miner
// Returns false if the miner should be disconnected immediately.
// ---------------------------------------------------------------------------
static bool HandleLine(Miner* m, const std::string& rawLine,
                       std::map<std::string, uint64>& roundShareCount,
                       uint64& roundShareTotal,
                       int& blocksFoundThisSession)
{
    json req;
    try {
        // Fast pre-parse depth check to prevent stack overflow in json::parse.
        // Legitimate Stratum requests have a depth of at most 4.
        int depth = 0;
        bool in_string = false;
        bool escape = false;
        for (char c : rawLine) {
            if (in_string) {
                if (escape) escape = false;
                else if (c == '\\') escape = true;
                else if (c == '"') in_string = false;
            } else {
                if (c == '"') in_string = true;
                else if (c == '{' || c == '[') {
                    if (++depth > 20) {
                        LogPrint("worker", "[worker] JSON too deeply nested from %s\n",
                                 m->address.empty() ? "(unauth)" : m->address.c_str());
                        return true;
                    }
                } else if (c == '}' || c == ']') {
                    depth--;
                }
            }
        }
        req = json::parse(rawLine);
    }
    catch (...) {
        LogPrint("worker", "[worker] JSON parse error from %s: %.80s\n",
                 m->address.empty() ? "(unauth)" : m->address.c_str(),
                 rawLine.c_str());
        return true; // don't disconnect on bad JSON, just ignore the line
    }

    // value() throws when the document is not an object, and again when a key
    // is present with the wrong type -- "[1,2,3]" and {"method":1} are both
    // valid JSON that parse cleanly and then blow up on the next two lines.
    if (!req.is_object() || (req.contains("method") && !req["method"].is_string())) {
        LogPrint("worker", "[worker] malformed request from %s: %.80s\n",
                 m->address.empty() ? "(unauth)" : m->address.c_str(),
                 rawLine.c_str());
        return true;
    }

    json        id     = req.value("id", json(nullptr));
    std::string method = req.value("method", "");

    // Reply helper: always sends result+error pair as Stratum requires
    auto reply = [&](json result, std::string error = "") {
        json err = error.empty() ? json(nullptr) : json(error);
        if (!SendLine(m->fd, json{{"id",id},{"result",result},{"error",err}}))
            LogPrint("worker", "[pool->worker] reply send FAILED for %s (method=%s)\n",
                     m->address.empty() ? "(unauth)" : m->address.c_str(),
                     method.c_str());
    };

    // ------------------------------------------------------------------
    if (method == "mining.subscribe") {
        std::string sid = ToHex(&m->fd, 4);
        LogPrint("worker", "[worker->pool] subscribe from fd=%d sid=%s\n",
                 (int)m->fd, sid.c_str());
        reply(json::array({json::array({json::array({"mining.notify",sid})}),sid,4}));
        return true;
    }

    // ------------------------------------------------------------------
    if (method == "mining.authorize") {
        // is_array() matters as much as the element checks: indexing a json
        // object with a number throws, and size() on an object returns its key
        // count, so "params":{"a":1} would pass a bare size() test.
        auto& p = req["params"];
        bool fArr = p.is_array();
        m->address = (fArr && p.size() > 0 && p[0].is_string()) ? p[0].get<std::string>() : "";
        m->worker  = (fArr && p.size() > 1 && p[1].is_string()) ? p[1].get<std::string>() : "worker";
        m->lastSeen = GetTime();

        // Validate address before authorising
        uint160 h160; bool validAddr = AddressToHash160(m->address, h160);

        m->authorised = true; // authorise regardless -- shares just won't pay out
        LogPrint("worker", "[worker->pool] authorize address=%s worker=%s addr_valid=%s\n",
                 m->address.c_str(), m->worker.c_str(), validAddr ? "yes" : "NO");

        if (!validAddr)
            LogPrint("worker", "[pool] WARNING: address '%s' is invalid -- "
                     "miner will hash but shares cannot be paid out\n",
                     m->address.c_str());

        reply(true);

        // Initialize vardiff state and send initial difficulty
        m->difficulty       = 1.0;
        m->shareTarget      = ShareTargetFromDifficulty(m->difficulty);
        m->vardiffWindowStart = GetTime();
        m->vardiffShares    = 0;
        LogPrint("worker", "[pool->worker] initial difficulty=%.4f for %s\n",
                 m->difficulty, m->address.c_str());
        if (!SendLine(m->fd, json{{"id",nullptr},{"method","mining.set_difficulty"},
                                  {"params",json::array({m->difficulty})}}))
            LogPrint("worker", "[pool->worker] set_difficulty send FAILED for %s\n",
                     m->address.c_str());

        if (gHaveJob) {
            LogPrint("worker", "[pool->worker] sending job %s to %s\n",
                     gCurrentJob.jobId.c_str(), m->address.c_str());
            SendJob(m, true);
        } else {
            LogPrint("worker", "[pool->worker] no job available yet for %s"
                     " -- will be sent one as soon as the pool builds it\n",
                     m->address.c_str());
        }
        return true;
    }

    // ------------------------------------------------------------------
    if (method == "mining.extranonce.subscribe") { reply(true); return true; }
    if (method == "mining.get_transactions")     { reply(json::array()); return true; }

    // ------------------------------------------------------------------
    if (method == "mining.submit") {
        if (!m->authorised) {
            LogPrint("worker", "[worker->pool] submit from unauthorized miner (fd=%d)\n",
                     (int)m->fd);
            reply(false, "not authorised");
            return true;
        }

        auto& p = req["params"];
        // Check the shape as well as the length: a submit carrying
        // "params":[1,2,3] would otherwise reach get<std::string>() and throw
        // a json::type_error that nothing above catches, taking the pool down
        // with it. Authorising costs an attacker nothing but an address.
        if (!p.is_array() || p.size() < 3 || !p[1].is_string() || !p[2].is_string()) {
            LogPrint("worker", "[worker->pool] submit bad params from %s\n",
                     m->address.c_str());
            reply(false, "bad params");
            return true;
        }

        std::string submitJobId = p[1].get<std::string>();
        std::string nonceHex    = p[2].get<std::string>();

        if (!gHaveJob) {
            LogPrint("worker", "[worker->pool] submit from %s but pool has no current job\n",
                     m->address.c_str());
            reply(false, "stale job");
            return true;
        }
        if (gCurrentJob.jobId != submitJobId) {
            LogPrint("worker", "[worker->pool] submit stale job from %s"
                     " (submitted=%s current=%s)\n",
                     m->address.c_str(), submitJobId.c_str(), gCurrentJob.jobId.c_str());
            reply(false, "stale job");
            return true;
        }

        auto nb = FromHex(nonceHex);
        if (nb.size() < 4) {
            LogPrint("worker", "[worker->pool] submit bad nonce from %s: '%s'\n",
                     m->address.c_str(), nonceHex.c_str());
            reply(false, "bad nonce");
            return true;
        }
        unsigned int nNonce; memcpy(&nNonce, nb.data(), 4);

        // Rebuild this miner's exact block (unique merkle via extranonce2)
        CBlock b = BuildMinerBlock(gCurrentJob, m->extranonce2);
        b.nNonce = nNonce;
        unsigned char hdr[80];
        memcpy(hdr,    &b.nVersion,       4);
        memcpy(hdr+4,  &b.hashPrevBlock,  32);
        memcpy(hdr+36, &b.hashMerkleRoot, 32);
        memcpy(hdr+68, &b.nTime,          4);
        memcpy(hdr+72, &b.nBits,          4);
        memcpy(hdr+76, &b.nNonce,         4);

        uint256 powHash = RandomXPoWHash(hdr, 80);

        // Check against this miner's personal share target (set by vardiff)
        if (powHash > m->shareTarget) {
            LogPrint("worker", "[worker->pool] high-hash from %s"
                     " nonce=%u (difficulty=%.4f)\n",
                     m->address.c_str(), nNonce, m->difficulty);
            reply(false, "high-hash");
            return true;
        }

        // Valid share -- record it
        m->sessionShares++;
        m->roundShares++;
        m->totalShares++;
        m->vardiffShares++;
        int64 shareTime = GetTime();
        m->lastSeen = shareTime;
        roundShareCount[m->address]++;
        roundShareTotal++;

        // Update hashrate estimate: expected hashes for this share divided by
        // the time since the last one, smoothed with an EWMA so a single fast
        // or slow share doesn't make the number jump around.
        if (m->lastShareTime > 0) {
            int64 dt = shareTime - m->lastShareTime;
            if (dt <= 0) dt = 1;
            double inst = ExpectedHashesForDifficulty(m->difficulty) / (double)dt;
            const double alpha = 0.25;
            m->hashRate = (m->hashRate <= 0.0) ? inst
                                                : alpha * inst + (1.0 - alpha) * m->hashRate;
        }
        m->lastShareTime = shareTime;

        LogPrint("worker", "[worker->pool] valid share from %s"
                 " nonce=%u job=%s difficulty=%.4f (session: %llu)\n",
                 m->address.c_str(), nNonce,
                 gCurrentJob.jobId.c_str(), m->difficulty,
                 (unsigned long long)m->sessionShares);
        reply(true);

        // Vardiff: after every VARDIFF_SHARES_PER_CHECK shares, check rate
        if (m->vardiffShares >= VARDIFF_SHARES_PER_CHECK) {
            int64 elapsed = GetTime() - m->vardiffWindowStart;
            if (elapsed <= 0) elapsed = 1;
            double actualSecsPerShare = (double)elapsed / (double)m->vardiffShares;
            double ratio = actualSecsPerShare / (double)VARDIFF_TARGET_SECS;
            // ratio < 1 means shares arriving faster than target (too easy)
            // ratio > 1 means shares arriving slower than target (too hard)
            if (ratio < (1.0 / VARDIFF_RETARGET_RATIO) ||
                ratio > VARDIFF_RETARGET_RATIO) {
                // New difficulty scales inversely with actual share rate
                double newDiff = m->difficulty / ratio;
                if (newDiff < VARDIFF_MIN) newDiff = VARDIFF_MIN;
                if (newDiff > VARDIFF_MAX) newDiff = VARDIFF_MAX;
                // Only apply if the change is meaningful (>10%)
                if (newDiff / m->difficulty > 1.1 || m->difficulty / newDiff > 1.1) {
                    m->difficulty    = newDiff;
                    m->shareTarget   = ShareTargetFromDifficulty(m->difficulty);
                    LogPrint("worker", "[pool] vardiff: %s new difficulty=%.4f"
                             " (%.1fs/share actual vs %ds target)\n",
                             m->address.c_str(), m->difficulty,
                             actualSecsPerShare, (int)VARDIFF_TARGET_SECS);
                    SendLine(m->fd, json{{"id",nullptr},
                                        {"method","mining.set_difficulty"},
                                        {"params",json::array({m->difficulty})}});
                    // set_difficulty alone doesn't change what the worker
                    // actually hashes against -- it only reads its share
                    // target from mining.notify params[3]. Without this, the
                    // new difficulty wouldn't take effect until the next
                    // natural job broadcast (new block, or the 60s trickle
                    // refresh), so push a fresh notify for this miner now.
                    SendJob(m, false);
                }
            }
            // Reset window
            m->vardiffWindowStart = GetTime();
            m->vardiffShares      = 0;
        }

        if (powHash > gCurrentJob.target) return true; // valid share, not a block

        // Full difficulty met -- submit block to the chain
        LogPrint("pool", "[pool] BLOCK CANDIDATE from %s height=%d nNonce=%u -- submitting\n",
                 m->address.c_str(), gCurrentJob.height, nNonce);

        CBlock* pblock = new CBlock(b);
        bool accepted = false;
        CRITICAL_BLOCK(cs_main)
        {
            int heightBefore = nBestHeight;
            if (ProcessBlock(NULL, pblock)) {
                // ProcessBlock returns true for orphans too -- verify the chain
                // actually advanced. If nBestHeight didn't increase, the block
                // was shelved as an orphan (pool was behind the chain tip) and
                // no coinbase was added to the wallet. Do not pay out.
                if (nBestHeight > heightBefore) {
                    accepted = true;
                    LogPrint("pool", "[pool] BLOCK ACCEPTED height=%d by %s\n",
                             gCurrentJob.height, m->address.c_str());
                    gFoundBlocks.push_front({gCurrentJob.height, GetTime(), m->address});
                    if (gFoundBlocks.size() > 50) gFoundBlocks.pop_back();
                    blocksFoundThisSession++;
                    gHaveJob = false; // invalidate job immediately
                } else {
                    LogPrint("pool", "[pool] BLOCK ORPHANED height=%d by %s"
                             " -- pool was behind chain tip, no payout\n",
                             gCurrentJob.height, m->address.c_str());
                    gHaveJob = false; // job is stale regardless
                }
            } else {
                LogPrint("pool", "[pool] BLOCK REJECTED height=%d (ProcessBlock returned false)\n",
                         gCurrentJob.height);
                delete pblock;
            }
        }

        if (accepted) {
            // Snapshot height before RebuildJob() (which increments it)
            int foundHeight = gCurrentJob.height;
            // b is the block we submitted, untouched by ProcessBlock (which
            // took a copy), so its hash is what the payout must be keyed on.
            QueuePayouts(foundHeight, b.GetHash(), roundShareCount, roundShareTotal);
            // Reset round shares
            roundShareCount.clear();
            roundShareTotal = 0;
            for (Miner* mi : gMiners) mi->roundShares = 0;
            // Build and broadcast new job immediately
            if (RebuildJob()) {
                LogPrint("pool", "[pool] new job %s built after block %d\n",
                         gCurrentJob.jobId.c_str(), foundHeight);
                BroadcastJob(true);
            } else {
                LogPrint("pool", "[pool] WARNING: RebuildJob() failed after block %d"
                         " -- miners will wait\n", foundHeight);
            }
        }
        return true;
    }

    // Unknown method -- log it (can reveal miner software quirks)
    LogPrint("worker", "[worker->pool] unknown method '%s' from %s\n",
             method.c_str(),
             m->address.empty() ? "(unauth)" : m->address.c_str());
    reply(nullptr, "unknown method");
    return true;
}

// ---------------------------------------------------------------------------
// Stats refresh -- called every 2s from event loop
// ---------------------------------------------------------------------------
static void RefreshStats(int blocksFoundThisSession, uint64 roundShareTotal)
{
    std::lock_guard<std::mutex> lk(gStatsMutex);

    gStats.authorizedMiners = 0;
    gStats.blocksFound      = blocksFoundThisSession;
    gStats.roundShares      = roundShareTotal;
    gStats.totalHashRate    = 0.0;
    for (Miner* m : gMiners)
        if (m->authorised) gStats.authorizedMiners++;

    gWorkerStats.clear();
    for (Miner* m : gMiners) {
        if (!m->authorised || m->lastSeen <= 0) continue;
        PoolWorkerStatView v;
        v.address     = m->address;
        v.worker      = m->worker;
        v.roundShares = m->roundShares;
        v.totalShares = m->totalShares;
        v.lastSeen    = m->lastSeen;
        v.hashRate    = m->hashRate;
        gWorkerStats.push_back(v);
        gStats.totalHashRate += m->hashRate;
    }
    std::sort(gWorkerStats.begin(), gWorkerStats.end(),
        [](const PoolWorkerStatView& a, const PoolWorkerStatView& b) {
            return a.roundShares != b.roundShares ? a.roundShares > b.roundShares
                                                  : a.lastSeen > b.lastSeen;
        });

    // gPendingPayouts is event-loop-only data (no lock needed here since
    // RefreshStats is always called from the event loop thread).
    gPayoutViews.clear();
    for (const PendingPayout& pp : gPendingPayouts) {
        PendingPayoutView pv;
        pv.matureAtHeight = pp.matureAtHeight;
        pv.recipients     = (int)pp.amounts.size();
        pv.totalAmount    = 0;
        for (auto& kv : pp.amounts) pv.totalAmount += kv.second;
        gPayoutViews.push_back(pv);
    }
}

// ---------------------------------------------------------------------------
// Event loop
// ---------------------------------------------------------------------------
static void PoolEventLoop()
{
    int   lastHeight            = -1;
    int64 lastRefresh           = GetTime();
    int64 lastStatsRefresh      = 0;
    int   blocksFoundThisSession = 0;
    int64 noJobSince            = 0;   // 0 = have a job
    bool  warnedNoJob           = false;
    int64 lastNoJobRetry        = 0;   // throttle RebuildJob calls when no tip

    std::map<std::string, uint64> roundShareCount;
    uint64 roundShareTotal = 0;

    static const int64 JOB_REFRESH_SECS  = 60;  // re-stamp nTime periodically
    static const int64 NO_JOB_WARN_SECS  = 15;  // warn after this long with no job
    static const int64 NO_JOB_RETRY_SECS = 5;   // how often to retry RebuildJob

    while (!fShutdown && gPoolRunning) {

        // 1. Drain accept queue
        {
            std::lock_guard<std::mutex> lk(gReadyMutex);
            for (btf_socket_t fd : gReadyQueue) {
                // Connecting costs nothing and every accepted socket carries a
                // read buffer, so an unbounded miner list is an unbounded
                // allocation for whoever runs the pool.
                if (gMiners.size() >= MAX_POOL_MINERS) {
                    LogPrint("worker", "[pool] refusing miner fd=%d --"
                             " already at %zu connections\n",
                             (int)fd, gMiners.size());
                    sock_close(fd);
                    continue;
                }
                gMiners.push_back(new Miner(fd));
                LogPrint("worker", "[pool] miner fd=%d entered event loop"
                         " -- awaiting subscribe/authorize\n", (int)fd);
            }
            gReadyQueue.clear();
        }

        // 2. Detect new block.
        // When our own miner found the block, HandleLine already called
        // RebuildJob() + BroadcastJob(true) and set gHaveJob=true.
        // We still flush mature payouts and update bookkeeping, but skip
        // rebuilding if a fresh job already exists for this height.
        {
            bool newBlock = false;
            CRITICAL_BLOCK(cs_main)
            {
                if (pindexBest && nBestHeight != lastHeight) {
                    lastHeight = nBestHeight;
                    newBlock   = true;
                }
            }
            if (newBlock) {
                FlushMaturePayouts();
                if (!gHaveJob) {
                    // External block (not found by our miners), or HandleLine's
                    // RebuildJob failed. Either way this round is over with no
                    // payout from us, so clear round shares before starting the
                    // next one -- otherwise they'd keep accumulating across
                    // rounds that our own miners didn't win.
                    roundShareCount.clear();
                    roundShareTotal = 0;
                    for (Miner* mi : gMiners) mi->roundShares = 0;

                    // Build and broadcast now.
                    if (RebuildJob()) {
                        LogPrint("pool", "[pool] new block height=%d -- job %s ready,"
                                 " %zu miner(s) connected\n",
                                 lastHeight, gCurrentJob.jobId.c_str(), gMiners.size());
                        BroadcastJob(true);
                    } else {
                        LogPrint("pool", "[pool] new block height=%d but RebuildJob() failed"
                                 " -- miners will wait\n", lastHeight);
                    }
                } else {
                    // HandleLine already built and broadcast for this height.
                    LogPrint("pool", "[pool] new block height=%d -- job already"
                             " broadcast by submit handler\n", lastHeight);
                }
                lastRefresh  = GetTime();
                noJobSince   = 0;
                warnedNoJob  = false;
            }
        }

        // 3. No-job watchdog (throttled: only retry every NO_JOB_RETRY_SECS)
        if (!gHaveJob) {
            int64 now = GetTime();
            if (noJobSince == 0) noJobSince = now;

            if (now - lastNoJobRetry >= NO_JOB_RETRY_SECS) {
                lastNoJobRetry = now;
                if (RebuildJob()) {
                    LogPrint("pool", "[pool] job built after %llds without one\n",
                             (long long)(now - noJobSince));
                    BroadcastJob(false);
                    noJobSince  = 0;
                    warnedNoJob = false;
                    lastRefresh = now;
                } else if (!warnedNoJob && now - noJobSince >= NO_JOB_WARN_SECS) {
                    LogPrint("pool", "[pool] STUCK: no job for %llds -- %s."
                             " Miners are waiting. Will keep retrying every %llds.\n",
                             (long long)(now - noJobSince),
                             pindexBest ? "RebuildJob() keeps failing (check above for why)"
                                        : "no chain tip yet (node still syncing)",
                             (long long)NO_JOB_RETRY_SECS);
                    warnedNoJob = true;
                }
            }
        } else if (GetTime() - lastRefresh >= JOB_REFRESH_SECS) {
            // Periodic nTime refresh -- stale timestamps cause valid hashes to be rejected
            if (RebuildJob()) BroadcastJob(false);
            lastRefresh = GetTime();
        }

        // 3.5. Vardiff stall watchdog -- runs independently of share count.
        // The normal vardiff check in HandleLine only fires once
        // VARDIFF_SHARES_PER_CHECK shares have arrived; a miner that never
        // finds a single share never reaches that check, so it can be stuck
        // at its initial (or last) difficulty forever. Catch that here.
        for (Miner* m : gMiners) {
            if (!m->authorised || m->vardiffShares != 0 || m->vardiffWindowStart <= 0)
                continue;
            int64 stalled = GetTime() - m->vardiffWindowStart;
            if (stalled < VARDIFF_STALL_SECS) continue;

            double newDiff = m->difficulty / VARDIFF_STALL_CUT;
            if (newDiff < VARDIFF_MIN) newDiff = VARDIFF_MIN;

            if (newDiff < m->difficulty) {
                m->difficulty         = newDiff;
                m->shareTarget        = ShareTargetFromDifficulty(m->difficulty);
                m->vardiffWindowStart = GetTime();
                LogPrint("worker", "[pool] vardiff STALL: %s sent zero shares in"
                         " %llds -- forcing difficulty down to %.6f (no share data"
                         " to retarget from normally)\n",
                         m->address.c_str(), (long long)stalled, m->difficulty);
                SendLine(m->fd, json{{"id",nullptr},
                                    {"method","mining.set_difficulty"},
                                    {"params",json::array({m->difficulty})}});
                // Same reasoning as the normal vardiff retarget above: the
                // worker only reads its share target from mining.notify, so
                // push a fresh job now instead of waiting on the next
                // natural broadcast.
                SendJob(m, false);
            } else {
                // Already at the difficulty floor -- this miner's problem isn't
                // difficulty, it's something else (slow hashing, bad connection,
                // stuck RandomX light mode, etc). Log once per stall window so
                // it's visible without spamming.
                LogPrint("worker", "[pool] vardiff STALL: %s still at zero shares"
                         " after %llds at floor difficulty %.6f -- likely a"
                         " hashrate/connectivity problem, not a difficulty problem\n",
                         m->address.c_str(), (long long)stalled, m->difficulty);
                m->vardiffWindowStart = GetTime(); // avoid re-logging every tick
            }
        }

        // 4. Select over miner sockets
        if (!gMiners.empty()) {
            fd_set fds; FD_ZERO(&fds);
            int maxfd = 0;
            for (Miner* m : gMiners) {
                FD_SET(m->fd, &fds);
                if ((int)m->fd > maxfd) maxfd = (int)m->fd;
            }
            struct timeval tv = {0, 50000}; // 50ms
            if (select(maxfd + 1, &fds, NULL, NULL, &tv) > 0) {
                std::vector<Miner*> dead;
                for (Miner* m : gMiners) {
                    if (!FD_ISSET(m->fd, &fds)) continue;
                    std::string line;
                    bool disc = false;
                    // Drain all complete lines buffered for this miner.
                    // HandleLine always returns true; disc signals clean disconnect.
                    // ThreadRPCServer has no handler of its own, so anything
                    // thrown here would reach the top of the thread and
                    // terminate the whole node -- not just the pool. Every
                    // line arrives from an unauthenticated socket.
                    while (RecvLine(m->fd, m->readBuf, line, disc)) {
                        try {
                            HandleLine(m, line, roundShareCount,
                                       roundShareTotal, blocksFoundThisSession);
                        } catch (const std::exception& e) {
                            LogPrint("worker", "[pool] dropping miner %s --"
                                     " exception handling line: %s\n",
                                     m->address.empty() ? "(unauth)" : m->address.c_str(),
                                     e.what());
                            disc = true;
                            break;
                        } catch (...) {
                            LogPrint("worker", "[pool] dropping miner %s --"
                                     " unknown exception handling line\n",
                                     m->address.empty() ? "(unauth)" : m->address.c_str());
                            disc = true;
                            break;
                        }
                    }
                    if (disc) dead.push_back(m);
                }
                for (Miner* m : dead) {
                    LogPrint("worker", "[pool] miner %s disconnected"
                             " (session shares: %llu, round shares: %llu)\n",
                             m->worker.empty() ? "(never authorized)" : m->worker.c_str(),
                             (unsigned long long)m->sessionShares,
                             (unsigned long long)m->roundShares);
                    sock_close(m->fd);
                    gMiners.erase(std::find(gMiners.begin(), gMiners.end(), m));
                    delete m;
                }
            }
        } else {
            Sleep(50);
        }

        // 5. Refresh GUI stats every 2s
        if (GetTime() - lastStatsRefresh >= 2) {
            RefreshStats(blocksFoundThisSession, roundShareTotal);
            lastStatsRefresh = GetTime();
        }
    }

    for (Miner* m : gMiners) { sock_close(m->fd); delete m; }
    gMiners.clear();
}

// ---------------------------------------------------------------------------
// Accept threads
// ---------------------------------------------------------------------------
struct AcceptCtx { unsigned char pk[32]; unsigned char sk[32]; std::string relay; };

// One persistent thread per relay. Loops: register -> wrap -> push -> repeat.
static void AcceptOneFn(void* arg)
{
    AcceptCtx* ctx = (AcceptCtx*)arg;
    unsigned char pk[32], sk[32];
    memcpy(pk, ctx->pk, 32); memcpy(sk, ctx->sk, 32);
    std::string relay = ctx->relay;
    delete ctx;

    size_t colon = relay.rfind(':');
    if (colon == std::string::npos) {
        LogPrint("pool", "[pool] AcceptOneFn: bad relay address '%s'\n", relay.c_str());
        return;
    }
    std::string host = relay.substr(0, colon);
    int         port = atoi(relay.substr(colon + 1).c_str());

    LogPrint("pool", "[pool] accept thread started for relay %s\n", relay.c_str());

    while (!fShutdown && gPoolRunning) {
        // Advertise this relay as the pool's meeting node BEFORE registering,
        // so that miners which resolve our descriptor find a relay where the
        // pool (not the node's peer-accept) holds the registration.
        BtfSetActiveRelay(relay);
        // Returns as soon as the relay lists us -- registered, not yet paired.
        btf::RvSocket rv = btf::RvServiceRegister(host.c_str(), (unsigned short)port, pk);

        if (fShutdown || !gPoolRunning) {
            if (rv != btf::RV_INVALID) btf::RvClose(rv);
            break;
        }
        if (rv == btf::RV_INVALID) {
            LogPrint("pool", "[pool] relay %s: RvServiceRegister failed"
                     " -- backing off 30s before retry\n", relay.c_str());
            for (int i = 0; i < 30 && !fShutdown && gPoolRunning; i++) Sleep(1000);
            continue;
        }

        // Block for a miner to actually arrive. RvServiceRegister used to do
        // this itself and returned with the relay's 0x01 pairing byte already
        // consumed; it no longer does. Without waiting here, BtfServiceWrap
        // reads that 0x01 as the first byte of the miner's 32-byte ephemeral
        // key -- every subsequent byte is shifted by one, so the X25519
        // handshake completes against a garbage shared secret (it has no
        // confirmation step to catch it), the tunnel is handed to the event
        // loop, and the miner's mining.subscribe never decrypts into a valid
        // line. The miner sees the pool close on it; the pool sees a peer that
        // never authorized.
        if (!btf::RvServiceWaitPaired(rv)) {
            btf::RvClose(rv);
            LogPrint("pool", "[pool] relay %s: dropped us before any miner dialed"
                     " -- re-registering\n", relay.c_str());
            continue;
        }

        // Echan handshake -- blocking but fast (local + relay round-trip)
        btf_socket_t fd = btf::BtfServiceWrap(rv, sk);
        if (fd == INVALID_SOCKET) {
            LogPrint("pool", "[pool] relay %s: BtfServiceWrap failed (echan handshake)"
                     " -- miner dropped, will accept next\n", relay.c_str());
            continue;
        }

        LogPrint("worker", "[pool] miner arrived via relay %s -- pushing to event loop\n",
                 relay.c_str());
        PushReady(fd);
        // Loop immediately: register again so the next miner can connect
    }

    LogPrint("pool", "[pool] accept thread stopped for relay %s\n", relay.c_str());
}

// Spawns one AcceptOneFn per relay. Re-polls relay list every 60s to pick up
// newly discovered relays. Uses a set to avoid spawning duplicate threads.
static void AcceptLoopFn(void*)
{
    LogPrint("pool", "[pool] AcceptLoopFn started -- miners can connect via %s\n",
             BtfLocalAddress().c_str());

    // Wait for our identity to be ready
    unsigned char pk[32], sk[32];
    int waitSecs = 0;
    while (!BtfGetIdentity(pk, sk)) {
        if (fShutdown || !gPoolRunning) return;
        if (waitSecs == 0)
            LogPrint("pool", "[pool] waiting for .btf identity...\n");
        waitSecs++;
        Sleep(2000);
    }
    LogPrint("pool", "[pool] .btf identity ready after %ds\n", waitSecs * 2);

    std::set<std::string> spawned; // relays we've already started a thread for

    while (!fShutdown && gPoolRunning) {
        vector<string> relays = BtfAllRelays();
        if (relays.empty()) {
            LogPrint("pool", "[pool] no relays available -- miners cannot connect yet\n");
        } else {
            for (const string& relay : relays) {
                if (spawned.count(relay)) continue; // already running
                AcceptCtx* ctx = new AcceptCtx();
                memcpy(ctx->pk, pk, 32); memcpy(ctx->sk, sk, 32);
                ctx->relay = relay;
                uintptr_t tid = _beginthread(AcceptOneFn, 0, ctx);
                if (tid == (uintptr_t)-1) {
                    delete ctx;
                    LogPrint("pool", "[pool] failed to spawn accept thread for %s\n",
                             relay.c_str());
                } else {
                    spawned.insert(relay);
                    LogPrint("pool", "[pool] accept thread spawned for relay %s"
                             " (%zu total)\n", relay.c_str(), spawned.size());
                }
            }
        }
        // Re-poll every 60s -- newly discovered relays get picked up
        for (int i = 0; i < 60 && !fShutdown && gPoolRunning; i++) Sleep(1000);
    }
}

// ---------------------------------------------------------------------------
// Entry point
// ---------------------------------------------------------------------------
void ThreadRPCServer(void*)
{
    gPoolRunning = true;

    LogPrint("pool", "[pool] ============================================\n");
    LogPrint("pool", "[pool] Pool server starting\n");
    LogPrint("pool", "[pool] .btf address: %s\n", BtfLocalAddress().c_str());
    LogPrint("pool", "[pool] fee: %.2f%%\n", dPoolFeePercent);

    if (RebuildJob()) {
        LogPrint("pool", "[pool] initial job %s seeded at height %d\n",
                 gCurrentJob.jobId.c_str(), nBestHeight + 1);
    } else {
        LogPrint("pool", "[pool] no chain tip yet -- job will build on first block\n");
    }

    LoadPendingPayouts();
    LoadPaying();
    // Must run before the payout thread starts, so nothing is resent that the
    // wallet already shows as sent, and nothing is owed twice because we died
    // between recording the payments and rewriting the ledger.
    ReconcilePayouts();

    _beginthread(AcceptLoopFn, 0, NULL);
    _beginthread(LocalAcceptFn, 0, NULL);
    _beginthread(ThreadBtfPoolAnnouncer, 0, NULL);
    _beginthread(PayoutThreadFn, 0, NULL);

    PoolEventLoop(); // blocks until gPoolRunning = false or fShutdown

    LogPrint("pool", "[pool] pool server stopped\n");
    LogPrint("pool", "[pool] ============================================\n");
    gPoolRunning = false;
}