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>
#include <fcntl.h>
#include <dirent.h>
#else
#include <io.h>
#endif

#include <atomic>
#include <mutex>
#include <thread>
#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();
    // Representability floor. shareTarget = diff1 / difficulty, and diff1 is
    // ~uint256(0) >> 12, i.e. a hair under 2^244 -- so the true target stops
    // fitting in a uint256 once difficulty reaches 2^-12 (= 1/4096), and the
    // fixed-point divide below breaks even earlier in a second way: the
    // (uint64)(difficulty * 65536.0) divisor truncates to zero for any
    // difficulty under 2^-16. Both thresholds sit far below VARDIFF_MIN
    // (0.001), so live operation never comes near this, but a caller that
    // bypasses vardiff (or a future VARDIFF_MIN) must not divide by zero or
    // hand back getuint256()'s low-256-bit truncation of an oversized
    // quotient -- a wrapped target is not merely wrong, it is arbitrarily
    // HARDER than difficulty 1. At these difficulties every hash is meant to
    // be a share, so the honest answer is the easiest representable target.
    // (This replaces a dead branch that read `if (t < diff1 && difficulty <
    // 1.0)`: t = diff1/difficulty, so difficulty < 1.0 implies t > diff1 and
    // the two conditions could never both hold.)
    if (difficulty < 1.0 / 4096.0) return ~uint256(0);
    // 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);
    return bn.getuint256();
}

// 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);

        // Fill the block the same way BitcoinMiner() does, because this is the
        // same job: a block that gets mined and submitted for real.
        //
        // It used to differ in two ways, and both were bugs.
        //
        // One pass over mapTransactions is not enough. That map is keyed by
        // hash, so it has no dependency order, and a transaction spending the
        // still-unconfirmed output of another (change from a recent send, most
        // commonly) fails ConnectInputs whenever its parent happens to sit
        // later in map order. A single pass then drops it -- and because map
        // order does not change between rebuilds, every later rebuild drops it
        // again at the same point. Deterministic exclusion, not bad luck: if
        // pool operators mine most of the blocks, a transaction of that shape
        // can stay unconfirmed for as long as its parent does.
        //
        // Diagnosis and fix both come from Vic-Nas, who found it in his fork
        // of this code and described it exactly. The fork has since been
        // deleted, so this carries it back rather than losing it.
        int64 nFees = 0;
        {
            CTxDB txdb("r");
            CRITICAL_BLOCK(cs_mapTransactions)
            {
                map<uint256, CTxIndex> pool;
                vector<char> vfAlreadyAdded(mapTransactions.size());
                bool fFoundSomething = true;
                unsigned int sz = 0;
                // The cap is MAX_BLOCK_SIZE/2, not MAX_SIZE/2. MAX_SIZE is the
                // 32 MB serialisation ceiling, so half of it is 16 MB -- this
                // builder would happily assemble a template that its own
                // CheckBlock() rejects at the 1 MB consensus cap. This is the
                // pool's job builder, so a block over the cap is not one
                // wasted solve, it is every miner on the pool working a
                // template the network will reject. The inner break bounds the
                // overshoot to a single transaction within a pass.
                while (fFoundSomething && sz < MAX_BLOCK_SIZE / 2)
                {
                    fFoundSomething = false;
                    unsigned int n = 0;
                    for (map<uint256, CTransaction>::iterator mi = mapTransactions.begin();
                         mi != mapTransactions.end(); ++mi, ++n)
                    {
                        if (vfAlreadyAdded[n])
                            continue;
                        CTransaction& tx = (*mi).second;
                        if (tx.IsCoinBase() || !tx.IsFinal()) continue;
                        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;
                        swap(pool, tmp);
                        block.vtx.push_back(tx);
                        sz += ::GetSerializeSize(tx, SER_NETWORK);
                        vfAlreadyAdded[n] = true;
                        fFoundSomething = true;
                        if (sz > MAX_BLOCK_SIZE / 2) break;
                    }
                }
            }
        }

        // And the fees actually go to the coinbase now. nFees was declared
        // inside the scan loop, so every transaction reset it and the total was
        // thrown away, while this line passed a hardcoded 0. A pool operator
        // collected fees into a block and paid none of them to itself. Solo
        // mining has always passed nFees here; only this builder did not.
        block.vtx[0].vout[0].nValue = block.GetBlockValue(pindexPrev->nHeight + 1, nFees);
        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.
    //
    // Upstream keys these on the string "block-<height>". That is the same
    // bug this fork already paid for, and height keying additionally lets
    // FindRecordedPoolPayout match the *previous* round's transaction after a
    // reorg and silently skip a miner, so their `id` is deliberately dropped.
    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 = 0;
    std::map<std::string, int64> amounts; // address -> satoshis
    // From upstream: which recipients already have a transaction. This is what
    // lets an entry stay on disk until every address in `amounts` is paid,
    // instead of being erased the moment the batch is handed off.
    std::map<std::string, std::string> paidTxids; // address -> txid
    // Retry and backoff state does NOT live here. It used to be declared here
    // as well as on PayingRecord, and the copy on this side was never written:
    // `retryAfter` was declared, serialised, loaded and tested at the flush
    // gate, and the only assignment to it anywhere was `= 0`, so the gate could
    // not fire. `retries` and `deferrals` were the same story. The real state
    // is per (round, recipient) -- which is the granularity a payment can fail
    // at -- and it lives on PayingRecord, which is now what persists it.
};
static std::vector<PendingPayout> gPendingPayouts;
static std::mutex                 gPayoutLedgerMutex;
// Bumped on every serialisation, under gPayoutLedgerMutex. See PublishVersioned.
static uint64                     gPayoutLedgerSeq = 0;
// Serialises writers of pending_payouts.json, and guards the sequence number
// of the image currently on disk. Never taken while a data mutex is held.
static std::mutex                 gPayoutLedgerFileMutex;
static uint64                     gPayoutLedgerPublishedSeq = 0;

struct PoolRoundProof {
    std::string                  id;
    int                          blockHeight;
    int                          matureAtHeight;
    int64                        createdAt;
    std::string                  blockHash;
    std::string                  coinbaseTxid;
    std::string                  foundBy;
    double                       feePercent;
    uint64                       totalShares;
    int64                        coinbaseValue;
    int64                        subsidy;
    int64                        minerReward;
    int64                        operatorReward;
    int                          skippedDust;
    int64                        forfeitedBadAddress;
    std::map<std::string, uint64> shares;
    std::map<std::string, int64>  payouts;
    std::map<std::string, std::string> paidTxids;
    // Recipients the ledger has written off, and why. Without this the public
    // round proof goes on advertising an address as owed with "txid": null
    // forever, while the ledger has already dropped the obligation -- the two
    // records of the same round permanently disagreeing, with the operator
    // holding the money.
    std::map<std::string, std::string> forfeited; // address -> reason

    PoolRoundProof()
        : blockHeight(0), matureAtHeight(0), createdAt(0), feePercent(0.0),
          totalShares(0), coinbaseValue(0), subsidy(0), minerReward(0), operatorReward(0),
          skippedDust(0), forfeitedBadAddress(0) {}
};
static std::vector<PoolRoundProof> gPoolRounds;
static std::mutex                  gPoolRoundsMutex;
static uint64                      gPoolRoundsSeq = 0; // guarded by gPoolRoundsMutex
static std::mutex                  gPoolRoundsFileMutex;
static uint64                      gPoolRoundsPublishedSeq = 0; // guarded by gPoolRoundsFileMutex
static const size_t                MAX_POOL_ROUNDS = 200;

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

static std::string PoolRoundsFilePath()
{
    if (!strPoolRoundsFile.empty())
        return strPoolRoundsFile;
    return GetAppDir() + "/pool_rounds.json";
}

// ---------------------------------------------------------------------------
// Atomic, ordered publication of the payout state files
// ---------------------------------------------------------------------------
//
// Every one of these files is a full dump of an in-memory structure, written
// tmp-then-rename so a crash cannot truncate it. Two things have to be true
// for that to actually be safe, and only the first was:
//
//   1. the rename must be atomic (it is), and
//   2. exactly one thread at a time may be writing the temp file, and a write
//      must never publish an OLDER image over a newer one.
//
// (2) was broken for pending_payouts.json. The event loop wrote it with
// gPayoutLedgerMutex held, the payout thread wrote it with the mutex
// released, and nothing serialised the two: both opened the same
// "<path>.tmp" with independent handles and both renamed it. The interleaved
// file that lands is either a mixture of two images (LoadPendingPayouts then
// throws mid-loop and comes up with a nearly empty ledger -- every obligation
// not yet exploded, gone) or a whole newly-queued round silently overwritten
// by an older snapshot that predates it.
//
// The fix is a per-file write mutex, held only across the I/O and never
// across a data-structure mutex, plus a version on the snapshot. Serialising
// happens under the data mutex and stamps a monotonically increasing sequence
// number; publishing happens with the data mutex released and refuses any
// blob whose sequence number is not newer than what is already on disk. So
// writers can queue in any order and finish in any order -- the file only
// ever moves forwards.
//
// The other half of the rule is that no thread waits for I/O while holding
// gPayoutLedgerMutex, gPayoutExecMutex or gPoolRoundsMutex, which is the
// contention the merge was right to remove from the event loop. That was
// claimed here before it was true: pool_status.json is written by the same
// WriteFileAtomic and RefreshStats held gPayoutLedgerMutex at function scope
// across the call, so once the writes gained fsync() the event loop was doing
// two of them every ten seconds with the payout thread's mutex held -- worse
// than the buffered write the merge replaced. The status file now goes through
// the same serialise-under-the-lock / publish-outside-it split as the ledgers,
// and the rule holds for all four files. It is checked, not asserted: see the
// "no payout mutex is held across a publish" section of the payout-ledger
// self-test.

// A serialised file image plus the version of the in-memory state it was
// taken from.
struct VersionedBlob {
    std::string blob;
    uint64      seq = 0;
};

// Push a file's contents out of the C library's buffer AND out of the kernel's
// page cache, so the bytes exist on the device before anything points at them.
static bool SyncFileHandle(FILE* f)
{
    if (fflush(f) != 0)
        return false;
#ifdef _WIN32
    return _commit(_fileno(f)) == 0;
#else
    return fsync(fileno(f)) == 0;
#endif
}

// Make the directory entry that rename() created durable too.
//
// A tmp+rename is only atomic with respect to a crash of the process. Against
// a power cut or a hard VM reset the rename can reach the platter while the
// data it points at has not, and the file that comes back is short or empty.
// For pending_payouts.json that is not a corrupt file, it is a silent one:
// LoadPendingPayouts catches the parse error, logs a single warning, and the
// node comes up owing nobody anything.
//
// Best effort -- a filesystem that does not allow the directory to be opened
// for this is not a reason to refuse to write the ledger.
static void SyncDirectoryOf(const std::string& path)
{
#ifndef _WIN32
    size_t slash = path.find_last_of('/');
    std::string dir = (slash == std::string::npos) ? std::string(".")
                                                   : path.substr(0, slash);
    if (dir.empty())
        dir = "/";
    int fd = open(dir.c_str(), O_RDONLY);
    if (fd < 0)
        return;
    fsync(fd);
    close(fd);
#endif
}

// Replace `path` with `data` atomically and durably. Caller must hold the
// write mutex for `path`: the temp name is derived from the target, so two
// threads in here at once for one path is precisely the race this exists to
// prevent.
//
// Every failure removes the temp file before returning. It used to be left
// where it fell, so a disk that filled up once left a short, valid-looking
// "pending_payouts.json.tmp" next to the good ledger for as long as the node
// ran. That file is a trap: it has the name of a half-finished save, and an
// operator cleaning up after an incident is exactly the person who will move
// it into place.
//
// Failures print unconditionally rather than through LogPrint(). These three
// files are written under two different debug categories ("payout" for the
// ledgers, "pool" for the status file), and a write failure is not a trace of
// normal operation -- it is the file the operator is about to read not being
// there. Gating it behind a category that had to be chosen before the failure
// happened is how "could not replace pool_status.json" went missing.
static bool WriteFileAtomic(const std::string& path, const std::string& data,
                            const char* what)
{
    std::string tmp = path + ".tmp";
    FILE* f = fopen(tmp.c_str(), "w");
    if (!f) {
        printf("[payout] WARNING: could not write %s (%s)\n",
               tmp.c_str(), what);
        return false;
    }
    if (!data.empty() && fwrite(data.c_str(), 1, data.size(), f) != data.size()) {
        printf("[payout] WARNING: short write to %s (%s)\n",
               tmp.c_str(), what);
        fclose(f);
        remove(tmp.c_str());
        return false;
    }
    // Before the rename, not after: publishing a name that points at nothing
    // is the failure this ordering exists to prevent. If the data cannot be
    // made durable, do not rename -- the previous file is still intact and
    // still correct, and the next write will try again.
    if (!SyncFileHandle(f)) {
        printf("[payout] WARNING: could not flush %s to disk (%s)"
               " -- keeping the previous file\n", tmp.c_str(), what);
        fclose(f);
        remove(tmp.c_str());
        return false;
    }
    // fclose() is where buffered data actually reaches the OS, so a failure
    // here means the .tmp is short -- renaming it over the real file would
    // destroy the good copy, which is exactly what these files exist to
    // prevent.
    if (fclose(f) != 0) {
        printf("[payout] WARNING: could not close %s (%s)\n",
               tmp.c_str(), what);
        remove(tmp.c_str());
        return false;
    }
#ifdef _WIN32
    // rename() on Windows will not replace an existing file.
    remove(path.c_str());
#endif
    if (rename(tmp.c_str(), path.c_str()) != 0) {
        printf("[payout] WARNING: could not replace %s (%s)\n",
               path.c_str(), what);
        remove(tmp.c_str());
        return false;
    }
    SyncDirectoryOf(path);
    return true;
}

// Paths whose loader threw part-way through, and which must therefore never be
// written again by this process.
//
// All three loaders push entries into the live structure INSIDE the try, so an
// exception half way down a file leaves a PARTIAL ledger in memory -- the
// entries before the bad one, and nothing after it. The first publish after
// that then writes the partial state over the file, and the only record of
// what was owed is gone. ReconcilePayouts publishes unconditionally at
// startup, so "the first publish" is a few milliseconds later, before an
// operator could possibly intervene. a8e2f58 named this failure mode in its
// own commit message and then only fixed the write side.
//
// The file is renamed out of the way instead of being left in place, so the
// evidence survives even if the operator restarts into the same fault, and so
// a node that comes up owing nobody anything is visibly missing its ledger
// rather than quietly appearing to have none.
static std::set<std::string> gRefusedFiles;
static std::mutex            gRefusedFilesMutex;

static bool FileIsRefused(const std::string& path)
{
    std::lock_guard<std::mutex> lk(gRefusedFilesMutex);
    return gRefusedFiles.count(path) > 0;
}

// Quarantine `path` and refuse to write it for the rest of the process.
//
// Prints unconditionally: an operator who has just lost a payout ledger must
// be told at whatever log settings they were already running with, and the
// name it was moved to is the only way back to the data.
static void RefuseFileAfterLoadFailure(const std::string& path, const char* what,
                                       const char* why)
{
    {
        std::lock_guard<std::mutex> lk(gRefusedFilesMutex);
        gRefusedFiles.insert(path);
    }
    std::string bad = strprintf("%s.bad.%lld", path.c_str(), (long long)GetTime());
    bool moved = (rename(path.c_str(), bad.c_str()) == 0);
    printf("[payout] ERROR: %s (%s) did not parse: %s\n"
           "[payout] ERROR: what loaded before the error is a PARTIAL picture,"
           " so this file will not be written again by this process --"
           " restart once the cause is fixed\n"
           "[payout] ERROR: the file %s\n",
           path.c_str(), what, why,
           moved ? strprintf("has been kept as %s", bad.c_str()).c_str()
                 : "could NOT be renamed out of the way -- copy it now");
}

// Publish a versioned image of some in-memory state to `path`.
//
// `fileMutex` serialises writers of `path` and guards `publishedSeq`. A blob
// older than what is already published is dropped: these files are whole-state
// dumps, so a higher sequence number is by construction a later state of the
// same structure and publishing it loses nothing. On a write failure
// `publishedSeq` is deliberately left alone, so the next (newer) write still
// gets a chance to land.
static void PublishVersioned(const std::string& path, const VersionedBlob& v,
                             std::mutex& fileMutex, uint64& publishedSeq,
                             const char* what)
{
    // A default-constructed blob is not a snapshot of anything; it is a
    // VersionedBlob nobody filled in. Today they are already dropped, because
    // every serialiser stamps ++gSeq and so never produces 0, and 0 fails the
    // seq <= publishedSeq test below -- but only for as long as publishedSeq
    // starts at 0 and nobody writes `seq = gSeq++`. One character, and the
    // ledger file becomes "[]". Say it out loud instead of relying on it.
    if (!v.seq)
        return;
    // A file whose loader threw is not a file this process knows the contents
    // of. Writing it would replace a full record of what was owed with
    // whatever happened to parse before the error.
    if (FileIsRefused(path))
        return;
    std::lock_guard<std::mutex> lk(fileMutex);
    if (v.seq <= publishedSeq)
        return;
    if (!WriteFileAtomic(path, v.blob, what))
        return;
    publishedSeq = v.seq;
}

// Identity of a payout round. Used as the pool-round proof id and, more
// importantly, as the `pool_payout_id` written into wallet transaction
// metadata and matched by FindRecordedPoolPayout().
//
// Derived from the block HASH, not from its height alone. Upstream uses
// "block-<height>", which names two different blocks identically across a
// reorg -- and because this id is what the crash-recovery probe matches on, a
// height-keyed id lets that probe find the *previous* round's transaction and
// conclude a miner has already been paid when they have not. The height stays
// in the string for readability only.
//
// A zero hash means the entry predates hash keying; those keep the old form so
// anything already recorded against them still matches.
static std::string PayoutIdForBlock(int blockHeight, const uint256& blockHash)
{
    if (blockHash == 0)
        return strprintf("block-%d", blockHeight);
    return strprintf("block-%d-%s", blockHeight,
                     blockHash.ToString().substr(0, 16).c_str());
}

// Do these two references name the same payout round?
//
// The block hash alone does not decide it, and every ledger lookup used to
// assume it did. Entries grandfathered in from a file written before payouts
// were keyed by hash all carry hash 0, so under a hash-only comparison the
// first zero-hash entry answers for every other one: RecordPayoutTxid writes
// a txid against the wrong round (which is then erased as "fully paid" while
// the real debt still stands), and ReconcilePayouts drops a live in-flight
// record because a *different* zero-hash round happened to pay that address.
//
// A non-zero hash is unique and decides on its own -- that is the whole point
// of keying on it. A zero hash falls back to the height, which
// LoadPendingPayouts supplies for any entry that can name its own maturity.
// Mixed hash-ness never matches: refusing to match is the safe direction for
// every caller here, because each of them treats a match as "this debt is
// already dealt with".
static bool SamePayoutRound(int aHeight, const uint256& aHash,
                            int bHeight, const uint256& bHash)
{
    if (aHash != bHash)
        return false;
    if (aHash != 0)
        return true;
    return aHeight == bHeight;
}

// An entry with neither a hash nor a height cannot be told apart from any
// other entry in the same state, so nothing may be recorded against it: a
// txid written to the wrong one reports a debt as settled that is still owed.
// Such an entry is never paid and is logged instead -- refusing to act is the
// safe failure, and no file this tree has ever written can produce one.
static bool PayoutRoundIdentifiable(int height, const uint256& hash)
{
    return hash != 0 || height != 0;
}

static bool PayoutFullyPaid(const PendingPayout& pp)
{
    for (const auto& kv : pp.amounts)
        if (pp.paidTxids.count(kv.first) == 0)
            return false;
    return true;
}

// Serialise the ledger and stamp it with the version it was taken at. Caller
// must hold gPayoutLedgerMutex.
//
// Split from the file write so no thread holds the ledger mutex across the
// I/O. Upstream writes the file inside the per-recipient loop with
// gPayoutLedgerMutex held, while the event loop's FlushMaturePayouts blocks on
// that same mutex -- one slow fsync per recipient, on the thread the miners'
// I/O shares. The sequence number is what lets the write happen later, and
// possibly out of order, without a stale image reaching the disk.
static VersionedBlob SerializePendingPayoutsLocked()
{
    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;
        // No retryAfter: it was never written with anything but zero, and the
        // retry clock that is actually used is on PayingRecord. Files that
        // still carry the key load fine (it is ignored), and a build that
        // expects it defaults an absent one to 0, which means "ready now" --
        // the same answer it always got.
        nlohmann::json amounts = nlohmann::json::object();
        for (const auto& kv : pp.amounts)
            amounts[kv.first] = kv.second;
        obj["amounts"] = amounts;
        nlohmann::json paid = nlohmann::json::object();
        for (const auto& kv : pp.paidTxids)
            paid[kv.first] = kv.second;
        obj["paidTxids"] = paid;
        arr.push_back(obj);
    }
    VersionedBlob v;
    v.blob = arr.dump(2);
    v.seq  = ++gPayoutLedgerSeq;
    return v;
}

// Publish a ledger snapshot. Holds no data mutex; safe -- and intended -- to
// call with gPayoutLedgerMutex released.
static void PublishPendingPayouts(const VersionedBlob& v)
{
    PublishVersioned(PayoutFilePath(), v, gPayoutLedgerFileMutex,
                     gPayoutLedgerPublishedSeq, "pending payout ledger");
}

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);
        std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
        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>();
            // Upstream's tolerance for an entry with no usable height, kept
            // because it costs nothing and an entry that cannot name its block
            // is worse than one derived from its maturity. COINBASE_MATURITY
            // and not COINBASE_SPENDABLE_DEPTH: this reverses whatever the
            // writer of *that* file used, and only files old enough to lack
            // blockHeight are affected -- they were written when maturity was
            // still COINBASE_MATURITY.
            if (pp.blockHeight == 0 && pp.matureAtHeight > COINBASE_MATURITY)
                pp.blockHeight = pp.matureAtHeight - COINBASE_MATURITY;
            // "retryAfter" is deliberately not read. Files written by the
            // merged build carry it, always as 0, and there is nowhere left to
            // put it -- the retry clock is PayingRecord's.
            for (auto it = obj["amounts"].begin(); it != obj["amounts"].end(); ++it)
                pp.amounts[it.key()] = it.value().get<int64>();
            if (obj.contains("paidTxids") && obj["paidTxids"].is_object())
                for (auto it = obj["paidTxids"].begin(); it != obj["paidTxids"].end(); ++it)
                    pp.paidTxids[it.key()] = it.value().get<std::string>();
            gPendingPayouts.push_back(pp);
        }
        LogPrint("payout", "[payout] loaded %zu pending payout(s) from disk\n",
                 gPendingPayouts.size());
    } catch (const std::exception& e) {
        RefuseFileAfterLoadFailure(path, "pending payout ledger", 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 {
    // Identity of this SLOT, distinct from the identity of the payment it
    // describes. ForgetPayment and ReschedulePayment used to find their record
    // by comparing every field, which is only an identity while no two records
    // can be equal -- and a restart used to produce byte-identical duplicates
    // (see FlushMaturePayouts), so the first match absorbed every update and
    // the second was never touched again.
    //
    // Process-local and deliberately not persisted: gPaying is rebuilt from
    // disk at every start, so an id only ever has to name a slot within one
    // run, and reading ids back would mean trusting a file to contain no
    // duplicates. Assigned under gPayoutExecMutex from gPayingNextId.
    uint64      id             = 0;
    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;
    // Ask the wallet whether this payment already went out, once, before the
    // first send attempt. Upstream asks on every attempt: FindRecordedPoolPayout
    // is a linear scan of mapWallet under cs_mapWallet, per recipient, so with
    // 2000 rotating payout addresses and a retry budget of 1440 that is a
    // scan per address per minute for a day. The question it answers -- "did a
    // previous process send this before it died?" -- can only change across a
    // restart, and every restart re-sets this flag (ReconcilePayouts), so
    // asking once per record is the whole of the answer.
    //
    // Not persisted: a record read back from disk is by definition a restart.
    bool        fProbeWallet = true;
};
static std::vector<PayingRecord> gPaying; // guarded by gPayoutExecMutex
static std::mutex                gPayoutExecMutex;
static uint64                    gPayingNextId = 0; // guarded by gPayoutExecMutex
static uint64                    gPayingSeq = 0; // guarded by gPayoutExecMutex
static std::mutex                gPayingFileMutex;
static uint64                    gPayingPublishedSeq = 0; // guarded by gPayingFileMutex
static std::atomic<bool>         gPayoutThreadRunning{false};

// Upstream's gPayoutExecQueue/gPayoutExecMutex/gPayoutThreadRunning are
// declared above with gPaying: the in-flight list IS the exec queue in this
// tree, at per-recipient rather than per-round granularity, so a second queue
// would be a second answer to "who is owed money". Only the dedupe set is new.
//
// Ids of rounds currently exploded into gPaying, so a flush cannot queue the
// same round twice. Guarded by gPayoutExecMutex. Seeded from the restored
// in-flight list by ReconcilePayouts -- it is memory-only, and leaving it
// empty at startup was half of why a restart duplicated everything in flight.
static std::set<std::string>      gQueuedPayouts;

// Is this exact payment already in flight? gPaying is keyed on
// (round, address): the ledger holds one amount per address per round, so two
// records for that pair are two standing orders to send the same money.
// Caller must hold gPayoutExecMutex.
static bool PaymentInFlightLocked(int blockHeight, const uint256& blockHash,
                                  const std::string& address)
{
    for (const PayingRecord& e : gPaying)
        if (e.address == address
            && SamePayoutRound(e.blockHeight, e.blockHash, blockHeight, blockHash))
            return true;
    return false;
}

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

// Serialise the in-flight list. Caller must hold gPayoutExecMutex.
//
// gPayoutExecMutex already serialises every writer of this file, so it never
// had the interleaving race pending_payouts.json had. It is still split the
// same way for a different reason: FlushMaturePayouts runs on the event loop
// and holds gPayoutExecMutex, so writing (and, since this series, fsyncing)
// the file under that mutex parks miner I/O behind the payout thread's disk.
// The version stamp keeps the file monotonic now that the write is outside
// the mutex that orders the writers.
static VersionedBlob SerializePayingLocked()
{
    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;
        // The live retry clock, which used not to be written at all: every
        // restart handed every backing-off payment a retryAfter of 0 and put
        // the whole set into the very next 500 ms batch, so a node restarted
        // in a loop retried continuously instead of once a minute.
        obj["retryAfter"]     = r.retryAfter;
        arr.push_back(obj);
    }
    VersionedBlob v;
    v.blob = arr.dump(2);
    v.seq  = ++gPayingSeq;
    return v;
}

// Publish an in-flight-list snapshot. Call with gPayoutExecMutex released.
static void PublishPaying(const VersionedBlob& v)
{
    PublishVersioned(PayingFilePath(), v, gPayingFileMutex,
                     gPayingPublishedSeq, "in-flight payment list");
}

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);
        // gPaying and gPayingNextId are both annotated as guarded by
        // gPayoutExecMutex and this was mutating them without it. Safe today
        // -- LoadPaying only runs at startup, before the payout thread is
        // spawned -- but "safe because of where it is called from" is a fact
        // about the caller, not about this function, and the annotation is
        // what every other reader will trust. The mutex is uncontended here by
        // construction, so taking it costs nothing.
        std::lock_guard<std::mutex> lk(gPayoutExecMutex);
        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>();
            // Absent in files written before this change; 0 means ready now,
            // which is exactly the behaviour those files used to get.
            if (obj.contains("retryAfter") && obj["retryAfter"].is_number())
                r.retryAfter = obj["retryAfter"].get<int64>();
            // Collapse duplicates as they are read. A build without the
            // ReconcilePayouts seeding below wrote a second (and after another
            // restart, a third) byte-identical record for every unpaid
            // recipient, so a file written by one of those has to be cleaned
            // up here or the pathology is inherited. The retry and deferral
            // counts are merged upwards so collapsing does not hand the
            // survivor a fresh budget.
            PayingRecord* dup = NULL;
            for (PayingRecord& e : gPaying)
                if (e.address == r.address
                    && SamePayoutRound(e.blockHeight, e.blockHash,
                                       r.blockHeight, r.blockHash)) {
                    dup = &e;
                    break;
                }
            if (dup) {
                dup->retries    = std::max(dup->retries, r.retries);
                dup->deferrals  = std::max(dup->deferrals, r.deferrals);
                dup->retryAfter = std::max(dup->retryAfter, r.retryAfter);
                LogPrint("payout", "[payout] collapsing duplicate in-flight"
                         " record for %s to %s -- one payment, one record\n",
                         FormatMoney(r.amount).c_str(), r.address.c_str());
                continue;
            }
            r.id = ++gPayingNextId;
            gPaying.push_back(r);
        }
        LogPrint("payout", "[payout] loaded %zu in-flight payment(s) from disk\n",
                 gPaying.size());
    } catch (const std::exception& e) {
        RefuseFileAfterLoadFailure(path, "in-flight payment list", e.what());
    }
}

// ---------------------------------------------------------------------------
// Pool round proofs (upstream cbdd06d). Disjoint from the in-flight ledger
// above -- this is the operator's published record of how a round was split,
// not an obligation to pay. Both sides added code at this offset; both are
// kept.
// ---------------------------------------------------------------------------
static nlohmann::json PoolRoundToJson(const PoolRoundProof& r)
{
    nlohmann::json j;
    j["id"] = r.id;
    j["blockHeight"] = r.blockHeight;
    j["matureAtHeight"] = r.matureAtHeight;
    j["createdAt"] = r.createdAt;
    j["blockHash"] = r.blockHash;
    j["coinbaseTxid"] = r.coinbaseTxid;
    j["foundBy"] = r.foundBy;
    j["feePercent"] = r.feePercent;
    j["totalShares"] = r.totalShares;
    j["coinbaseValueSatoshis"] = r.coinbaseValue;
    j["coinbaseValueBtf"] = FormatMoney(r.coinbaseValue);
    j["subsidySatoshis"] = r.subsidy;
    j["subsidyBtf"] = FormatMoney(r.subsidy);
    j["minerRewardSatoshis"] = r.minerReward;
    j["minerRewardBtf"] = FormatMoney(r.minerReward);
    j["operatorRewardSatoshis"] = r.operatorReward;
    j["operatorRewardBtf"] = FormatMoney(r.operatorReward);
    j["skippedDust"] = r.skippedDust;
    j["forfeitedBadAddressSatoshis"] = r.forfeitedBadAddress;
    j["forfeitedBadAddressBtf"] = FormatMoney(r.forfeitedBadAddress);

    j["shares"] = nlohmann::json::object();
    for (const auto& kv : r.shares)
        j["shares"][kv.first] = kv.second;

    j["payouts"] = nlohmann::json::object();
    for (const auto& kv : r.payouts) {
        nlohmann::json p;
        p["amountSatoshis"] = kv.second;
        p["amountBtf"] = FormatMoney(kv.second);
        std::map<std::string, std::string>::const_iterator tx = r.paidTxids.find(kv.first);
        if (tx != r.paidTxids.end())
            p["txid"] = tx->second;
        else
            p["txid"] = nullptr;
        // Only emitted when it happened, so the common shape of the file is
        // unchanged for anything already parsing it.
        std::map<std::string, std::string>::const_iterator fo = r.forfeited.find(kv.first);
        if (fo != r.forfeited.end()) {
            p["forfeited"] = true;
            p["forfeitReason"] = fo->second;
        }
        j["payouts"][kv.first] = p;
    }
    return j;
}

// Serialise the published round proofs. Caller must hold gPoolRoundsMutex.
//
// Like the in-flight list, gPoolRoundsMutex already serialises the writers
// (RecordPoolRound on the event loop, MarkPoolRoundPaid on the payout thread),
// so this file never had the interleaving race either -- but the event loop
// should not be waiting on the payout thread's fsync to publish a round, so
// the write moves outside the mutex on the same versioned terms.
static VersionedBlob SerializePoolRoundsLocked()
{
    nlohmann::json root;
    root["schema"] = "bitflash-pool-rounds-1";
    root["updatedAt"] = GetTime();
    root["rounds"] = nlohmann::json::array();
    for (const PoolRoundProof& r : gPoolRounds)
        root["rounds"].push_back(PoolRoundToJson(r));

    VersionedBlob v;
    v.blob = root.dump(2);
    v.seq  = ++gPoolRoundsSeq;
    return v;
}

// Publish a round-proof snapshot. Call with gPoolRoundsMutex released.
static void PublishPoolRounds(const VersionedBlob& v)
{
    PublishVersioned(PoolRoundsFilePath(), v, gPoolRoundsFileMutex,
                     gPoolRoundsPublishedSeq, "pool round proofs");
}

static void LoadPoolRounds()
{
    std::string path = PoolRoundsFilePath();
    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 root = nlohmann::json::parse(s);
        nlohmann::json arr = root.is_array() ? root : root["rounds"];
        std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
        for (const auto& obj : arr) {
            PoolRoundProof r;
            if (obj.contains("id") && obj["id"].is_string()) r.id = obj["id"].get<std::string>();
            if (obj.contains("blockHeight")) r.blockHeight = obj["blockHeight"].get<int>();
            if (obj.contains("matureAtHeight")) r.matureAtHeight = obj["matureAtHeight"].get<int>();
            if (obj.contains("createdAt")) r.createdAt = obj["createdAt"].get<int64>();
            if (obj.contains("blockHash") && obj["blockHash"].is_string()) r.blockHash = obj["blockHash"].get<std::string>();
            if (obj.contains("coinbaseTxid") && obj["coinbaseTxid"].is_string()) r.coinbaseTxid = obj["coinbaseTxid"].get<std::string>();
            if (obj.contains("foundBy") && obj["foundBy"].is_string()) r.foundBy = obj["foundBy"].get<std::string>();
            if (obj.contains("feePercent")) r.feePercent = obj["feePercent"].get<double>();
            if (obj.contains("totalShares")) r.totalShares = obj["totalShares"].get<uint64>();
            if (obj.contains("coinbaseValueSatoshis")) r.coinbaseValue = obj["coinbaseValueSatoshis"].get<int64>();
            if (obj.contains("subsidySatoshis")) r.subsidy = obj["subsidySatoshis"].get<int64>();
            if (obj.contains("minerRewardSatoshis")) r.minerReward = obj["minerRewardSatoshis"].get<int64>();
            if (obj.contains("operatorRewardSatoshis")) r.operatorReward = obj["operatorRewardSatoshis"].get<int64>();
            if (obj.contains("skippedDust")) r.skippedDust = obj["skippedDust"].get<int>();
            if (obj.contains("forfeitedBadAddressSatoshis")) r.forfeitedBadAddress = obj["forfeitedBadAddressSatoshis"].get<int64>();
            if (obj.contains("shares") && obj["shares"].is_object())
                for (auto it = obj["shares"].begin(); it != obj["shares"].end(); ++it)
                    r.shares[it.key()] = it.value().get<uint64>();
            if (obj.contains("payouts") && obj["payouts"].is_object())
                for (auto it = obj["payouts"].begin(); it != obj["payouts"].end(); ++it) {
                    if (it.value().is_object()) {
                        r.payouts[it.key()] = it.value().value("amountSatoshis", (int64)0);
                        if (it.value().contains("txid") && it.value()["txid"].is_string())
                            r.paidTxids[it.key()] = it.value()["txid"].get<std::string>();
                        if (it.value().value("forfeited", false))
                            r.forfeited[it.key()] =
                                it.value().value("forfeitReason", std::string("unspecified"));
                    } else if (it.value().is_number_integer()) {
                        r.payouts[it.key()] = it.value().get<int64>();
                    }
                }
            if (!r.id.empty())
                gPoolRounds.push_back(r);
        }
        while (gPoolRounds.size() > MAX_POOL_ROUNDS)
            gPoolRounds.erase(gPoolRounds.begin());
        LogPrint("payout", "[payout] loaded %zu pool round proof(s) from disk\n",
                 gPoolRounds.size());
    } catch (const std::exception& e) {
        RefuseFileAfterLoadFailure(path, "pool round proofs", 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);
}

// Startup reconciliation.
//
// This used to be the crash-recovery mechanism: it scanned the wallet with
// WalletAlreadyPaid() and then dropped ledger entries that had already been
// exploded into in-flight records. Neither job is right any more.
//
// The ledger is now authoritative and an entry survives until every recipient
// has a txid recorded against it (upstream cbdd06d), so dropping ledger
// entries here would reintroduce exactly the loss this merge is adopting that
// model to prevent. And the wallet scan is now FindRecordedPoolPayout(), done
// once per record on its first attempt, which is the same question asked in
// the one place that can act on the answer. Running both would be two
// authorities for the same debt.
//
// What is left is the useful part: the in-flight list restored from disk may
// contain records the ledger already shows as paid (we died between the send
// and the in-flight file being rewritten). The ledger wins.
//
// And the part that was missing: gQueuedPayouts is memory-only, so it came up
// empty while gPaying came up full. FlushMaturePayouts deduped against that
// empty set and against the ledger's paidTxids, and against nothing else --
// so a round that was exploded but not yet paid got a SECOND record per
// unpaid recipient on the first flush after every restart, deterministically.
// (Observed live: two byte-identical records for one (block, address).) The
// surviving in-flight list is exactly the statement of which rounds are
// already exploded, so it is what the set is rebuilt from.
static void ReconcilePayouts()
{
    VersionedBlob payingSnap;
    {
        std::lock_guard<std::mutex> exec(gPayoutExecMutex);
        std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);

        size_t settled = 0;
        int64  settledAmount = 0;
        for (auto it = gPaying.begin(); it != gPaying.end(); ) {
            bool paid = false;
            for (const PendingPayout& pp : gPendingPayouts)
                if (SamePayoutRound(pp.blockHeight, pp.blockHash,
                                    it->blockHeight, it->blockHash)
                    && pp.paidTxids.count(it->address)) {
                    paid = true;
                    break;
                }
            if (paid) {
                settledAmount += it->amount;
                settled++;
                it = gPaying.erase(it);
            } else {
                // Every restored record gets one wallet probe before it is
                // allowed to send: this is precisely the window where the
                // wallet may hold a transaction the ledger has not recorded.
                it->fProbeWallet = true;
                ++it;
            }
        }
        if (settled) {
            LogPrint("payout", "[payout] %zu in-flight payment(s) totalling %s"
                     " are already recorded paid in the ledger -- not"
                     " resending\n",
                     settled, FormatMoney(settledAmount).c_str());
        }

        // Rebuild the "already exploded" set from what survived. Without this
        // the first flush after a restart re-explodes every round in flight.
        gQueuedPayouts.clear();
        for (const PayingRecord& r : gPaying)
            gQueuedPayouts.insert(PayoutIdForBlock(r.blockHeight, r.blockHash));

        payingSnap = SerializePayingLocked();

        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());
        }
    }
    PublishPaying(payingSnap);
}

static void RecordPoolRound(const PoolRoundProof& round)
{
    VersionedBlob snap;
    {
        std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
        bool replaced = false;
        for (PoolRoundProof& existing : gPoolRounds) {
            if (existing.id == round.id) {
                existing = round;
                replaced = true;
                break;
            }
        }
        if (!replaced) {
            gPoolRounds.push_back(round);
            while (gPoolRounds.size() > MAX_POOL_ROUNDS)
                gPoolRounds.erase(gPoolRounds.begin());
        }
        snap = SerializePoolRoundsLocked();
    }
    PublishPoolRounds(snap);
}

static void MarkPoolRoundPaid(const std::string& id,
                              const std::string& address,
                              const std::string& txid)
{
    VersionedBlob snap;
    {
        std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
        bool found = false;
        for (PoolRoundProof& r : gPoolRounds) {
            if (r.id != id)
                continue;
            r.paidTxids[address] = txid;
            found = true;
            break;
        }
        if (!found)
            return;
        snap = SerializePoolRoundsLocked();
    }
    PublishPoolRounds(snap);
}

// Publish the fact that a recipient was written off, so the round proof and
// the ledger agree. The ledger erases the amount (it is not owed any more and
// must not be reported as sent); the proof keeps the amount visible and marks
// it forfeited with a reason, because a miner who was promised a share is
// entitled to see what happened to it.
static void MarkPoolRoundForfeited(const std::string& id,
                                   const std::string& address,
                                   const std::string& reason)
{
    VersionedBlob snap;
    {
        std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
        bool found = false;
        for (PoolRoundProof& r : gPoolRounds) {
            if (r.id != id)
                continue;
            r.forfeited[address] = reason;
            found = true;
            break;
        }
        if (!found)
            return;
        snap = SerializePoolRoundsLocked();
    }
    PublishPoolRounds(snap);
}

// Which wallet transaction, if any, already paid this?
//
// Two kinds of evidence, in order of quality.
//
// Structured (pool_payout_id / pool_payout_address / pool_payout_amount) is
// upstream's, and it is exact: the id is derived from the block hash, so it
// names one round and one round only. Every transaction this build sends
// carries it.
//
// But NO transaction this pool has ever sent carries it. Every payout in the
// existing wallet has only a "comment". So on the first boot after this
// upgrade -- precisely the boot where the binary is swapped after an unclean
// shutdown with payments in flight -- a structured-only probe finds nothing
// and every one of those recipients is paid a second time. The deleted
// WalletAlreadyPaid() matched on what the payment WAS, and that evidence is
// still there; it just has to be read.
//
// The legacy fallback is therefore restored, with three constraints that
// together stop it answering for a round that is not this one:
//
//   1. it is only consulted when the structured lookup found nothing;
//   2. it ignores any transaction carrying pool_payout_id. That key means this
//      build wrote the transaction, and this build's transactions are
//      adjudicated by the exact path -- so a NEW round's payment can never be
//      offered as evidence for an old one. Only pre-upgrade transactions,
//      which is exactly the population the fallback exists for, are eligible;
//   3. `excluded` holds every txid already credited to some OTHER
//      (round, address) pair -- from the ledger's paidTxids, which is durable,
//      and from what this process has already claimed. One legacy transaction
//      can therefore settle at most one debt, so N transactions can never
//      satisfy more than N obligations.
//
// What is left unresolved by construction is two pre-upgrade rounds at the
// SAME height paying the same address the same amount, since the comment
// carries only the height. That needs a reorg.
//
// This used to say the losing twin's ledger entry is voided by the orphan
// check before it can be paid. That is wrong: PayoutBlockStillOnChain()
// returns true for a zero hash, and EVERY pre-upgrade entry has a zero hash,
// so the orphan check is a no-op for exactly the population this fallback
// serves.
//
// The outcome is still safe, by a different mechanism. SamePayoutRound()
// treats two zero-hash entries at the same height as the SAME round, so the
// pair is indistinguishable everywhere, not only here: FlushMaturePayouts
// explodes one of them and skips the other as already queued,
// PaymentInFlightLocked refuses a second record for that (round, address), and
// RecordPayoutTxid settles against the first match. The pair therefore
// presents as ONE debt and can be paid at most once -- which is the direction
// that matters for a fallback whose failure mode is treating one payment as
// proof of two.
//
// The cost of that mechanism, stated plainly: only one of a genuine
// same-height pair would ever be paid. It takes a reorg AND both rounds
// predating hash keying, and no file this tree has written since can produce
// one, so it is a property of history rather than a live risk.
//
// Where the evidence genuinely runs out the fallback simply does not match,
// and the payment is made -- paying a debt we cannot prove was settled is the
// correct default for a pool.
enum PayoutProbeResult {
    PAYOUT_PROBE_NONE = 0,   // the wallet holds no record of this payment
    PAYOUT_PROBE_STRUCTURED, // matched this build's pool_payout_* metadata
    PAYOUT_PROBE_LEGACY      // matched a pre-upgrade comment and output
};

static PayoutProbeResult FindRecordedPoolPayout(const std::string& id,
                                                int blockHeight,
                                                const std::string& address,
                                                int64 amount,
                                                const std::set<std::string>& excluded,
                                                std::string& txidRet)
{
    // Built before cs_mapWallet is taken: an invalid address has no script to
    // compare against, and the caller has already rejected those.
    uint160 h160;
    bool fHaveScript = AddressToHash160(address, h160);
    CScript want;
    if (fHaveScript)
        want << OP_DUP << OP_HASH160 << h160 << OP_EQUALVERIFY << OP_CHECKSIG;
    const std::string comment = PayoutComment(blockHeight);

    // Candidate legacy matches, adjudicated after cs_mapWallet is released.
    //
    // The depth check cannot happen inside the scan: GetDepthInMainChain()
    // reads mapBlockIndex and pindexBest, which belong to cs_main, and cs_main
    // is taken BEFORE cs_mapWallet everywhere else in this tree
    // (CreateTransaction, ConnectBlock). Taking it the other way round here
    // would be a lock-order inversion in the payout thread. There is at most a
    // handful of these: a candidate has to carry this exact height's comment
    // and an output of exactly this amount to this address.
    std::vector<uint256> legacyCandidates;

    // One pass, not two: this scan runs under cs_mapWallet, which the whole
    // node contends on, and the merge is already paying for it once per record.
    CRITICAL_BLOCK(cs_mapWallet)
    {
        for (map<uint256, CWalletTx>::const_iterator it = mapWallet.begin();
             it != mapWallet.end(); ++it)
        {
            const CWalletTx& wtx = it->second;
            map<string, string>::const_iterator idIt = wtx.mapValue.find("pool_payout_id");
            if (idIt != wtx.mapValue.end()) {
                // Written by this build. Adjudicated exactly, and never
                // offered to the fallback below.
                if (idIt->second != id)
                    continue;
                map<string, string>::const_iterator addrIt = wtx.mapValue.find("pool_payout_address");
                if (addrIt == wtx.mapValue.end() || addrIt->second != address)
                    continue;
                map<string, string>::const_iterator amtIt = wtx.mapValue.find("pool_payout_amount");
                if (amtIt == wtx.mapValue.end() || amtIt->second != strprintf("%lld", amount))
                    continue;
                txidRet = wtx.GetHash().GetHex();
                return PAYOUT_PROBE_STRUCTURED;
            }

            if (!fHaveScript)
                continue;
            // The comment is written only by the payout path, so its presence
            // already identifies the transaction as one of ours.
            map<string, string>::const_iterator ci = wtx.mapValue.find("comment");
            if (ci == wtx.mapValue.end() || ci->second != comment)
                continue;
            if (excluded.count(wtx.GetHash().GetHex()))
                continue;   // already credited to a different debt
            foreach (const CTxOut& txout, wtx.vout)
                if (txout.nValue == amount && txout.scriptPubKey == want) {
                    legacyCandidates.push_back(it->first);
                    break;
                }
        }
    }

    // A legacy match must have CONFIRMED.
    //
    // The structured path does not need this and deliberately does not have
    // it: this process sent that transaction moments ago, and a send that
    // failed leaves nothing behind, because SendMoney calls
    // RollbackTransactionSpent, which calls EraseFromWallet.
    //
    // The legacy path is the opposite case in every respect. Those
    // transactions are days or weeks old, so whether they confirmed is a
    // settled fact rather than an open question, and a pre-upgrade payout that
    // never confirmed is evidence of NON-payment about to be read as payment.
    // This chain has had a 280-block reorg (see FlushMaturePayouts): a payout
    // invalidated by it is still sitting in mapWallet with its comment intact,
    // and the first boot after this upgrade is exactly when it gets matched.
    // The cost of being wrong here is a miner who is never paid and no record
    // that they were owed.
    CRITICAL_BLOCK(cs_main)
    CRITICAL_BLOCK(cs_mapWallet)
    {
        for (const uint256& hash : legacyCandidates) {
            map<uint256, CWalletTx>::const_iterator it = mapWallet.find(hash);
            if (it == mapWallet.end())
                continue;   // erased under us; it is not evidence of anything
            int depth = it->second.GetDepthInMainChain();
            if (depth < 1) {
                // Loud: this is the difference between a miner having been
                // paid and a miner about to be paid twice, decided on a fact
                // that is easy to check and impossible to recover afterwards.
                LogPrint("payout", "[payout] wallet tx %s carries the payout"
                         " comment for height %d and an output of exactly %s to"
                         " %s, but it is NOT in the main chain (depth %d) --"
                         " refusing to read it as payment; this recipient will"
                         " be paid\n",
                         hash.GetHex().c_str(), blockHeight,
                         FormatMoney(amount).c_str(), address.c_str(), depth);
                continue;
            }
            txidRet = hash.GetHex();
            return PAYOUT_PROBE_LEGACY;
        }
    }
    return PAYOUT_PROBE_NONE;
}

// txid -> "<round id>|<address>", for payments this run has already credited
// to a debt on legacy evidence. Stops one pre-upgrade transaction being read
// as proof that two different obligations were met. The ledger's paidTxids
// covers the same ground durably; this covers the window before the recovered
// txid reaches the ledger, and the case where the ledger entry has already
// been retired.
static std::map<std::string, std::string> gLegacyClaimedTxids;
static std::mutex                         gClaimedTxidMutex;

// Every txid already spoken for by some payment OTHER than this one. Collected
// before cs_mapWallet is taken: the ledger mutex must not be acquired under a
// wallet lock.
static void CollectForeignPayoutTxids(int blockHeight, const uint256& blockHash,
                                      const std::string& address,
                                      const std::string& claimKey,
                                      std::set<std::string>& out)
{
    {
        std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
        for (const PendingPayout& pp : gPendingPayouts) {
            bool sameRound = SamePayoutRound(pp.blockHeight, pp.blockHash,
                                             blockHeight, blockHash);
            for (const auto& kv : pp.paidTxids)
                if (!(sameRound && kv.first == address))
                    out.insert(kv.second);
        }
    }
    {
        std::lock_guard<std::mutex> lk(gClaimedTxidMutex);
        for (const auto& kv : gLegacyClaimedTxids)
            if (kv.second != claimKey)
                out.insert(kv.first);
    }
}

static void ClaimLegacyPayoutTxid(const std::string& txid, const std::string& claimKey)
{
    std::lock_guard<std::mutex> lk(gClaimedTxidMutex);
    gLegacyClaimedTxids[txid] = claimKey;
}

// blockHash stays a uint256 and not upstream's display string: it is the key
// this fork's payout ledger is built on, and a key that has been through
// ToString() is a key that can be compared wrong. It is converted to text only
// where the round proof needs text. The other three arguments are upstream's,
// carried for the round-proof data.
static void QueuePayouts(int blockHeight,
                         const uint256& blockHash,
                         const std::string& coinbaseTxid,
                         int64 coinbaseValue,
                         const std::string& foundBy,
                         const std::map<std::string, uint64>& shareCount,
                         uint64 total)
{
    PoolRoundProof round;
    round.id = PayoutIdForBlock(blockHeight, blockHash);
    round.blockHeight = blockHeight;
    round.matureAtHeight = blockHeight + COINBASE_SPENDABLE_DEPTH;
    round.createdAt = GetTime();
    round.blockHash = blockHash.ToString();
    round.coinbaseTxid = coinbaseTxid;
    round.coinbaseValue = coinbaseValue;
    round.foundBy = foundBy;
    round.feePercent = dPoolFeePercent;
    round.totalShares = total;
    round.shares = shareCount;

    if (total == 0) {
        RecordPoolRound(round);
        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));
    int64 operatorCut = nSubsidy - minerReward;
    round.subsidy = nSubsidy;
    round.minerReward = minerReward;
    round.operatorReward = operatorCut;

    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. Upstream uses
    // COINBASE_MATURITY here; this is a considered disagreement, not an
    // oversight -- see the merge commit message.
    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;
        round.payouts[kv.first] = amount;
    }
    round.skippedDust = skipped;
    round.forfeitedBadAddress = burned;

    {
        VersionedBlob snap;
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            gPendingPayouts.push_back(pp);
            snap = SerializePendingPayoutsLocked();
        }
        PublishPendingPayouts(snap);
    }
    RecordPoolRound(round);

    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 the event loop: explode matured ledger entries into per-recipient
// in-flight records (fast; no wallet or disk work beyond one small write).
//
// The ledger entry SURVIVES here. That is upstream's model and it is the
// crash-safe one: this fork used to erase the entry as soon as its payments
// were handed off, so a death in that window left the debt described by
// nothing at all -- a container restart on 2026-07-30 silently dropped
// ~1,156 BTF exactly that way. An entry is removed only once every recipient
// in it has a txid recorded (PayoutFullyPaid), which the payout thread does.
//
// gPaying is kept on top of that as the in-flight working set: it carries the
// per-recipient retry/deferral state that a per-round `nextAttempt` cannot
// express, and it is what the payout thread iterates. It is a cache of work
// derived from the ledger, never an independent record of debt -- if it were
// lost entirely the ledger would rebuild it on the next flush.
static void FlushMaturePayouts()
{
    // Which funding blocks are still on the main chain? Asked with no payout
    // lock held: PayoutBlockStillOnChain takes cs_main, and cs_main is held by
    // ProcessBlock and SendMoney for long stretches. Holding the ledger mutex
    // across it would let the event loop park the payout thread behind a chain
    // lock it has no interest in. (It is not a deadlock either way -- nothing
    // takes the ledger mutex while holding cs_main -- but it is exactly the
    // kind of stall this merge is removing from the ledger path.)
    std::vector<uint256> orphaned;
    {
        std::vector<uint256> hashes;
        {
            std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);
            for (const PendingPayout& pp : gPendingPayouts)
                hashes.push_back(pp.blockHash);
        }
        for (const uint256& h : hashes)
            if (!PayoutBlockStillOnChain(h))
                orphaned.push_back(h);
    }

    std::vector<PendingPayout> ready;
    VersionedBlob ledgerSnap;
    {
        std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);

        // 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 (std::find(orphaned.begin(), orphaned.end(), it->blockHash) == orphaned.end()) {
                ++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) ledgerSnap = SerializePendingPayoutsLocked();

        for (const PendingPayout& pp : gPendingPayouts) {
            if (nBestHeight < pp.matureAtHeight)
                continue;
            if (PayoutFullyPaid(pp))
                continue;
            // An entry with neither a hash nor a height cannot be told apart
            // from another in the same state, so a txid recorded against it
            // could settle the wrong debt. Paying it is the unsafe move;
            // leaving it standing and saying so is not.
            if (!PayoutRoundIdentifiable(pp.blockHeight, pp.blockHash)) {
                LogPrint("payout", "[payout] WARNING: ledger entry maturing at"
                         " height %d names neither a block hash nor a height --"
                         " refusing to pay %zu recipient(s) that cannot be told"
                         " apart from another round's\n",
                         pp.matureAtHeight, pp.amounts.size());
                continue;
            }
            ready.push_back(pp);
        }
    }
    // Outside the ledger mutex: publishing is I/O and the event loop is the
    // one thread that must never wait on the payout thread's disk.
    PublishPendingPayouts(ledgerSnap);

    if (ready.empty())
        return;

    VersionedBlob payingSnap;
    {
        std::lock_guard<std::mutex> queue(gPayoutExecMutex);
        size_t added = 0;
        for (const PendingPayout& pp : ready) {
            const std::string id = PayoutIdForBlock(pp.blockHeight, pp.blockHash);
            if (gQueuedPayouts.count(id))
                continue;   // already exploded into gPaying and still in flight
            size_t n = 0;
            for (const auto& kv : pp.amounts) {
                // Upstream's per-recipient dedupe: a recipient already carrying
                // a txid is done, even though the entry as a whole is not.
                if (pp.paidTxids.count(kv.first))
                    continue;
                // The invariant, checked against the thing it is an invariant
                // of: at most one in-flight record per (round, address). The
                // round-level set above is an optimisation that can be wrong --
                // it is memory only, it is rebuilt from gPaying at startup, and
                // a stale `ready` snapshot can race a round finishing --
                // whereas gPaying is the list being protected. A second record
                // here is a second standing order to send the same money.
                if (PaymentInFlightLocked(pp.blockHeight, pp.blockHash, kv.first))
                    continue;
                PayingRecord r;
                r.id             = ++gPayingNextId;
                r.blockHeight    = pp.blockHeight;
                r.blockHash      = pp.blockHash;
                r.matureAtHeight = pp.matureAtHeight;
                r.address        = kv.first;
                r.amount         = kv.second;
                gPaying.push_back(r);
                n++;
            }
            gQueuedPayouts.insert(id);
            added += n;
            LogPrint("payout", "[payout] block height %d matured -- queuing"
                     " payout %s (%zu of %zu recipients still owed)\n",
                     pp.matureAtHeight, id.c_str(), n, pp.amounts.size());
        }
        if (added) payingSnap = SerializePayingLocked();
    }
    PublishPaying(payingSnap);
}

// Settle a payment: it is either done or abandoned, either way it stops being
// in flight. Note this only retires the working record -- whether the miner is
// still OWED the money is the ledger's business, and the ledger is updated
// separately by RecordPayoutTxid().
//
// Erases by record id, not by comparing fields. Value-matching was only an
// identity for as long as no two records could be equal, and the restart bug
// above produced records that were equal in every field: the first match then
// absorbed every erase and every state update, and the second was never
// touched again. The id names the slot the caller took its copy from, so the
// update lands where it belongs whatever else is in the list.
static void ForgetPayment(const PayingRecord& r)
{
    VersionedBlob snap;
    {
        std::lock_guard<std::mutex> lk(gPayoutExecMutex);
        for (auto it = gPaying.begin(); it != gPaying.end(); ++it)
            if (it->id == r.id) { gPaying.erase(it); break; }

        // When the last record for a round is gone the round is no longer in
        // flight, so a later flush may explode it again if the ledger still
        // shows recipients owed. That is the intended behaviour and it is now
        // the only way a stalled payment ever leaves this list: a payment that
        // keeps failing is retried forever rather than given up on, so its
        // record stays here, and the flush that would re-explode it is stopped
        // by PaymentInFlightLocked.
        bool stillInFlight = false;
        for (const auto& e : gPaying)
            if (SamePayoutRound(e.blockHeight, e.blockHash,
                                r.blockHeight, r.blockHash)) {
                stillInFlight = true;
                break;
            }
        if (!stillInFlight)
            gQueuedPayouts.erase(PayoutIdForBlock(r.blockHeight, r.blockHash));

        snap = SerializePayingLocked();
    }
    PublishPaying(snap);
}

// Persist an updated attempt count / backoff for a payment still owed.
static void ReschedulePayment(const PayingRecord& r)
{
    VersionedBlob snap;
    {
        std::lock_guard<std::mutex> lk(gPayoutExecMutex);
        for (auto& e : gPaying)
            if (e.id == r.id) {
                e.retries      = r.retries;
                e.deferrals    = r.deferrals;
                e.retryAfter   = r.retryAfter;
                // Carry the cleared probe flag back, or every retry re-scans
                // the whole wallet -- the cost this merge is specifically
                // removing.
                e.fProbeWallet = r.fProbeWallet;
                break;
            }
        snap = SerializePayingLocked();
    }
    PublishPaying(snap);
}

// Record that a recipient has been paid, in the ledger, which is the only
// durable statement that the debt is settled.
//
// The file write happens with gPayoutLedgerMutex RELEASED. Upstream writes the
// file inside the per-recipient loop with the mutex held, so every recipient
// costs the event loop a full serialise+fsync it has to wait for in
// FlushMaturePayouts. Ordering against the event loop's own writes is the
// snapshot version's job, not the mutex's -- see PublishVersioned.
static void RecordPayoutTxid(const PayingRecord& r, const std::string& txid)
{
    VersionedBlob blob;
    bool complete = false;
    {
        std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
        for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ++it) {
            if (!SamePayoutRound(it->blockHeight, it->blockHash,
                                 r.blockHeight, r.blockHash))
                continue;
            it->paidTxids[r.address] = txid;
            if (PayoutFullyPaid(*it)) {
                LogPrint("payout", "[payout] payout for block %d complete --"
                         " removing from ledger\n", it->blockHeight);
                gPendingPayouts.erase(it);
                complete = true;
            }
            break;
        }
        blob = SerializePendingPayoutsLocked();
    }
    PublishPendingPayouts(blob);
    // The round proof records a txid per address, so this is per recipient and
    // not only on completion.
    MarkPoolRoundPaid(PayoutIdForBlock(r.blockHeight, r.blockHash), r.address, txid);
    if (complete)
        LogPrint("payout", "[payout] block %d fully settled\n", r.blockHeight);
}

// Write a recipient off.
//
// Only ever called for a state that is DECIDABLE -- currently one: the ledger
// names an address that is not a valid address, so no transaction paying it
// can be constructed by this or any other build. It is not called, and must
// not be, because a payment has failed some number of times. "The wallet said
// no N times" is not a decidable state, it is a description of how long we
// have been asking; a debt written off on the strength of it is a debt the
// operator keeps, and the arithmetic that decided WHY the wallet said no was
// wrong for a year (see PayoutThreadFn).
//
// The amount is removed from the entry rather than marked paid, so the money
// is not reported as sent. Once nothing is left owed the entry retires. The
// round proof is told separately, because dropping the obligation from the
// ledger while pool_rounds.json goes on publishing the address as owed with
// "txid": null leaves the pool's two records of the same round permanently
// disagreeing.
static void ForfeitPayoutRecipient(const PayingRecord& r, const std::string& reason)
{
    LogPrint("payout", "[payout] writing off %s owed to %s for block %d: %s\n",
             FormatMoney(r.amount).c_str(), r.address.c_str(), r.blockHeight,
             reason.c_str());
    VersionedBlob blob;
    {
        std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
        for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ++it) {
            if (!SamePayoutRound(it->blockHeight, it->blockHash,
                                 r.blockHeight, r.blockHash))
                continue;
            it->amounts.erase(r.address);
            // PayoutFullyPaid() returns true for an empty amounts map -- there
            // is no recipient in it without a txid -- so the
            // `amounts.empty()` disjunct that used to be here decided nothing.
            if (PayoutFullyPaid(*it))
                gPendingPayouts.erase(it);
            break;
        }
        blob = SerializePendingPayoutsLocked();
    }
    PublishPendingPayouts(blob);
    MarkPoolRoundForfeited(PayoutIdForBlock(r.blockHeight, r.blockHash),
                           r.address, reason);
}

// What a failed send means for the payment's retry budget.
//
// A deferral is "the wallet's condition is temporarily wrong": wait, and do
// not hold it against this payment. A fault is "the wallet is funded and
// unlocked and still would not do it", which is worth logging loudly and
// retrying with backoff.
//
// This used to be decided by re-reading GetBalance() after the failure and
// comparing it against `amount + nTransactionFee + CENT`. Every term of that
// was wrong:
//
//   * nTransactionFee is 0 in production. It is only ever assigned from the
//     GUI preferences dialog (gui.cpp), which a headless pool node does not
//     have, so the whole reserve was the CENT.
//
//   * The fee the reserve is for is GetMinFee(fDiscount=true), a STEP
//     function of the serialised size: 0 below 10,000 bytes, and
//     (1 + bytes/1000) * CENT at or above it -- so 11 * CENT at the step.
//     There is no required-fee value between 1 and 11 CENT. A one-CENT
//     reserve is therefore never the right number: below the step nothing
//     needs reserving, and at the step it is short by a factor of eleven.
//
//   * A pool wallet close to broke selects many small inputs BECAUSE it is
//     close to broke, and ~56 signed P2PKH inputs is 10 KB. The trigger and
//     the failure are positively correlated: the guess is wrong precisely
//     when the pool is nearly out of money.
//
// Concretely, with a balance of amount + 0.05: the pre-send gate passes,
// CreateTransaction builds a >10 KB transaction, the fee becomes 0.11,
// SelectCoins(amount + 0.11) fails, SendMoney returns false -- and the old
// arithmetic said (amount + 0.05) < (amount + 0.01) is false, so "funded",
// so a fault. The condition is stable, so it recurred every 60 s, and 1440 of
// them erased the debt with the coin still in the operator's wallet.
//
// So stop guessing. The wallet is asked to say what it did, and this only
// classifies the answer. WALLET_SEND_INSUFFICIENT_FUNDS is coin selection
// failing to cover amount plus whatever fee that size of transaction
// required, whatever that turned out to be -- no reserve to get wrong.
enum PayoutFailureClass {
    PAYOUT_FAILURE_DEFER = 0, // conditions are wrong; wait for them to change
    PAYOUT_FAILURE_FAULT      // funded and unlocked, and it still would not go
};

static PayoutFailureClass ClassifyPayoutSendFailure(WalletSendResult result,
                                                    bool fWalletLocked)
{
    // A locked wallet cannot sign. Nothing about this payment is wrong.
    if (fWalletLocked)
        return PAYOUT_FAILURE_DEFER;
    if (result == WALLET_SEND_INSUFFICIENT_FUNDS)
        return PAYOUT_FAILURE_DEFER;
    return PAYOUT_FAILURE_FAULT;
}

// 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) {
            const std::string id = PayoutIdForBlock(r.blockHeight, r.blockHash);

            // 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. Upstream has
            // no equivalent -- it decides once, when the entry matures.
            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;
            }

            // Upstream's per-recipient dedupe, against the ledger rather than
            // against the batch: if a txid is already recorded for this
            // address, the money went out and nothing more is owed.
            {
                bool alreadyPaid = false;
                {
                    std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
                    for (const PendingPayout& pp : gPendingPayouts)
                        if (SamePayoutRound(pp.blockHeight, pp.blockHash,
                                            r.blockHeight, r.blockHash)
                            && pp.paidTxids.count(r.address)) {
                            alreadyPaid = true;
                            break;
                        }
                }
                if (alreadyPaid) {
                    ForgetPayment(r);
                    continue;
                }
            }

            uint160 h160;
            if (!AddressToHash160(r.address, h160)) {
                // Should not happen -- bad addresses are filtered in
                // QueuePayouts. If one is here anyway it is decidable and
                // permanent: there is no transaction that pays a string which
                // is not an address, so retrying is not waiting for anything.
                //
                // This used to only ForgetPayment(), which drops the in-flight
                // record and leaves the ledger entry claiming the debt and
                // pool_rounds.json advertising the recipient as unpaid
                // forever, with the next flush re-exploding it. Write it off
                // properly and say why.
                const std::string reason =
                    strprintf("'%s' is not a valid address, so no transaction"
                              " can pay it", r.address.c_str());
                LogPrint("payout", "[payout] SKIPPED bad address '%s' --"
                         " %s owed for block %d cannot be paid\n",
                         r.address.c_str(), FormatMoney(r.amount).c_str(),
                         r.blockHeight);
                ForfeitPayoutRecipient(r, reason);
                ForgetPayment(r);
                continue;
            }
            CScript sc;
            sc << OP_DUP << OP_HASH160 << h160 << OP_EQUALVERIFY << OP_CHECKSIG;

            // Did a previous process send this and die before recording it?
            // Only asked once per record -- see PayingRecord::fProbeWallet.
            if (r.fProbeWallet) {
                std::string recovered;
                const std::string claimKey = id + "|" + r.address;
                std::set<std::string> excluded;
                CollectForeignPayoutTxids(r.blockHeight, r.blockHash, r.address,
                                          claimKey, excluded);
                r.fProbeWallet = false;
                PayoutProbeResult probe = FindRecordedPoolPayout(
                    id, r.blockHeight, r.address, r.amount, excluded, recovered);
                if (probe != PAYOUT_PROBE_NONE) {
                    if (probe == PAYOUT_PROBE_LEGACY) {
                        // Loud on purpose: this is the upgrade path, where the
                        // only evidence a payment went out is a comment written
                        // by a binary that is no longer running.
                        LogPrint("payout", "[payout] %s to %s was already paid by"
                                 " a pre-upgrade build (wallet tx %s carries the"
                                 " payout comment for height %d and an output of"
                                 " exactly this amount, and no structured payout"
                                 " metadata) -- not resending\n",
                                 FormatMoney(r.amount).c_str(), r.address.c_str(),
                                 recovered.c_str(), r.blockHeight);
                        ClaimLegacyPayoutTxid(recovered, claimKey);
                    } else {
                        LogPrint("payout", "[payout] recovered recorded payout %s to %s"
                                 " from wallet tx %s -- not resending\n",
                                 FormatMoney(r.amount).c_str(), r.address.c_str(),
                                 recovered.c_str());
                    }
                    RecordPayoutTxid(r, recovered);
                    ForgetPayment(r);
                    continue;
                }
            }

            // 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.
            //
            // The comparison includes the fee. It used to be against the
            // amount alone, so a balance anywhere between amount and
            // amount+fee sailed past this gate, failed in SendMoney, and was
            // counted as a send FAILURE rather than a wait -- 1440 of those is
            // 24 hours, and the recipient was erased with the money left in
            // the operator's wallet and nothing on disk saying it was owed.
            //
            // nTransactionFee is the part of the fee that can be known before
            // the transaction exists, and in production it is 0 -- it is only
            // ever assigned from the GUI preferences dialog, which a pool node
            // does not have -- so including it here changes nothing on this
            // pool and is kept only because a GUI operator can set it. The
            // rest is GetMinFee(), a function of the serialised size and
            // therefore of how many inputs SelectCoins ends up picking, which
            // cannot be predicted from here -- so this gate deliberately does
            // NOT try to reserve it. Refusing to attempt a payment the wallet
            // could actually have made would be its own way of not paying a
            // miner. That residual is handled after the fact, by asking the
            // wallet why it refused -- see ClassifyPayoutSendFailure.
            int64 nSpendable = GetBalance();
            if (nSpendable < r.amount + nTransactionFee) {
                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, need %s with fee (waiting %d min so far)\n",
                             FormatMoney(r.amount).c_str(), r.address.c_str(),
                             FormatMoney(nSpendable).c_str(),
                             FormatMoney(r.amount + nTransactionFee).c_str(),
                             r.deferrals);
                ReschedulePayment(r);
                continue;
            }

            // Likewise a locked wallet: it cannot sign, so every attempt fails
            // for a reason that has nothing to do with this payment being
            // payable. Left to the retry budget, an operator who locks the
            // wallet overnight comes back to erased debts.
            if (IsWalletLocked()) {
                r.deferrals++;
                r.retryAfter = GetTime() + 60;
                if (r.deferrals == 1 || r.deferrals % 60 == 0)
                    LogPrint("payout", "[payout] deferring %s to %s -- wallet is"
                             " locked (waiting %d min so far)\n",
                             FormatMoney(r.amount).c_str(), r.address.c_str(),
                             r.deferrals);
                ReschedulePayment(r);
                continue;
            }

            CWalletTx wtx;
            wtx.mapValue["comment"] = PayoutComment(r.blockHeight);
            // Upstream's structured metadata. This is what makes the wallet
            // probe above possible at all -- the old comment-only form could
            // not distinguish two rounds paying the same address the same
            // amount. Keyed by the hash-derived id, so a reorg cannot make one
            // round's transaction answer for another's.
            wtx.mapValue["pool_payout_id"]      = id;
            wtx.mapValue["pool_payout_address"] = r.address;
            wtx.mapValue["pool_payout_amount"]  = strprintf("%lld", r.amount);

            SendMoneyOutcome outcome;
            if (SendMoney(sc, r.amount, wtx, outcome)) {
                // The wallet holds the transaction before this returns, so a
                // crash between here and the ledger write is recoverable: the
                // probe above finds it by id+address+amount and does not
                // resend. Ledger first, in-flight record second -- the debt is
                // described by at least one of them at every instant.
                RecordPayoutTxid(r, wtx.GetHash().GetHex());
                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 {
                // Why did it fail? The answer used to decide whether the
                // payment eventually got erased, and it was decided by
                // guessing from a balance snapshot -- see
                // ClassifyPayoutSendFailure for the arithmetic and what it
                // cost. The wallet now says what it did, and nothing here
                // erases anything.
                //
                // The lock state is re-read rather than trusted from before
                // the attempt: SendMoney takes cs_main and can block for a
                // long time, so it may have moved underneath us. The funding
                // answer does not need re-reading -- it is a statement about
                // the attempt that was actually made.
                bool fLocked = IsWalletLocked();
                int64 nAfter = GetBalance();

                if (ClassifyPayoutSendFailure(outcome.result, fLocked)
                        == PAYOUT_FAILURE_DEFER) {
                    // Conditions are temporarily wrong. Wait.
                    r.deferrals++;
                    r.retryAfter = GetTime() + 60;
                    if (r.deferrals == 1 || r.deferrals % 60 == 0)
                        LogPrint("payout", "[payout] deferring %s to %s -- send"
                                 " failed with %s (%s spendable; the wallet"
                                 " needed a fee of %s for a transaction this"
                                 " size); waiting %d min so far\n",
                                 FormatMoney(r.amount).c_str(), r.address.c_str(),
                                 fLocked ? "the wallet locked" : "too little coin",
                                 FormatMoney(nAfter).c_str(),
                                 FormatMoney(outcome.nFeeRequired).c_str(),
                                 r.deferrals);
                    ReschedulePayment(r);
                } else {
                    // Funded, unlocked, block on the main chain, and the
                    // wallet still would not build, sign or accept the
                    // transaction. That is worth shouting about; it is not
                    // grounds to keep the money.
                    //
                    // There is deliberately no give-up branch. It used to
                    // forfeit the debt after 1440 of these, which is not a
                    // decidable state -- it is a count of how long we have
                    // been asking, and it was reached by misclassifying a
                    // funding shortfall as a fault. A debt that is never
                    // written off cannot be written off wrongly. The payment
                    // stays in flight, keeps its ledger entry, keeps being
                    // reported in pool_status.json as owed, and keeps
                    // retrying once a minute until the operator or the wallet
                    // makes it possible.
                    r.retries++;
                    r.retryAfter = GetTime() + 60;
                    ReschedulePayment(r);
                    // Every minute at first, then hourly: a fault that is not
                    // going away must not turn the log into one line a minute
                    // forever, and must not go quiet either.
                    if (r.retries <= 5 || r.retries % 60 == 0)
                        LogPrint("payout", "[payout] FAILED to pay %s to %s"
                                 " (%s spendable, wallet unlocked, block %d on"
                                 " the main chain) -- attempt %d, retrying in"
                                 " 60s; this debt stands until it is paid\n",
                                 FormatMoney(r.amount).c_str(), r.address.c_str(),
                                 FormatMoney(nAfter).c_str(), r.blockHeight,
                                 r.retries);
                }
            }
        }

        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;
static int64                           gLastPoolStatusWrite = 0;

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;
}

static std::string PoolStatusFilePath()
{
    if (!strPoolStatusFile.empty())
        return strPoolStatusFile;
    return GetAppDir() + "/pool_status.json";
}

// Serialise the operator status file. Caller must hold gStatsMutex (this reads
// gStats / gWorkerStats / gPayoutViews); gPoolRoundsMutex is taken and
// released inside.
//
// Split from the write for the same reason as the three ledgers, and for a
// worse instance of the same problem. This used to be called from RefreshStats
// with gPayoutLedgerMutex held at function scope, and once the ledger writes
// gained fsync() that call became two fsyncs -- one on the file, one on its
// directory -- performed by the event loop, the thread that owns every miner's
// socket, with the mutex the payout thread needs in its hand. Measured on the
// build host (ext4 on LVM on SSD): 3.78 ms per publish, every 10 seconds, and
// unbounded on a shared virtual disk under load.
//
// Nothing here needs the ledger mutex at all: the pending-payout rows were
// already copied into gPayoutViews under it, further up RefreshStats.
static std::string SerializePoolStatusLocked()
{
    nlohmann::json j;
    j["schema"] = "bitflash-pool-status-1";
    j["network"] = "bitflash";
    j["updatedAt"] = GetTime();

    j["operator"] = {
        {"name", strPoolName},
        {"btfAddress", BtfLocalAddress()},
        {"feePercent", dPoolFeePercent},
        {"dashboardUrl", strPoolDashboardUrl},
        {"roundsFile", "pool_rounds.json"}
    };

    j["node"] = {
        {"height", nBestHeight},
        {"poolRunning", (bool)gPoolRunning},
        {"mode", nMineMode}
    };

    j["pool"] = {
        {"authorizedMiners", gStats.authorizedMiners},
        {"blocksFoundSession", gStats.blocksFound},
        {"roundShares", gStats.roundShares},
        {"hashRate", gStats.totalHashRate}
    };

    j["workers"] = nlohmann::json::array();
    for (const PoolWorkerStatView& row : gWorkerStats) {
        j["workers"].push_back({
            {"address", row.address},
            {"worker", row.worker},
            {"totalShares", row.totalShares},
            {"roundShares", row.roundShares},
            {"lastSeen", row.lastSeen},
            {"hashRate", row.hashRate}
        });
    }

    j["pendingPayouts"] = nlohmann::json::array();
    for (const PendingPayoutView& row : gPayoutViews) {
        j["pendingPayouts"].push_back({
            {"matureAtHeight", row.matureAtHeight},
            {"recipients", row.recipients},
            {"totalAmountSatoshis", row.totalAmount},
            {"totalAmountBtf", FormatMoney(row.totalAmount)}
        });
    }

    j["recentRounds"] = nlohmann::json::array();
    {
        std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
        size_t start = gPoolRounds.size() > 10 ? gPoolRounds.size() - 10 : 0;
        for (size_t i = start; i < gPoolRounds.size(); i++) {
            const PoolRoundProof& r = gPoolRounds[i];
            // A forfeited recipient is not "unpaid and pending" -- the ledger
            // has already dropped the obligation. Counting it as unpaid is
            // what made the round proof and the ledger disagree for good.
            int unpaid = 0;
            int forfeited = 0;
            for (const auto& kv : r.payouts) {
                if (r.paidTxids.count(kv.first))
                    continue;
                if (r.forfeited.count(kv.first))
                    forfeited++;
                else
                    unpaid++;
            }
            j["recentRounds"].push_back({
                {"id", r.id},
                {"blockHeight", r.blockHeight},
                {"blockHash", r.blockHash},
                {"coinbaseTxid", r.coinbaseTxid},
                {"totalShares", r.totalShares},
                {"payouts", (int)r.payouts.size()},
                {"unpaidPayouts", unpaid},
                {"forfeitedPayouts", forfeited},
                {"coinbaseValueBtf", FormatMoney(r.coinbaseValue)},
                {"minerRewardBtf", FormatMoney(r.minerReward)}
            });
        }
    }

    return j.dump(2);
}

// Publish the operator status file. Call with every payout mutex released.
//
// Still no version guard and no write mutex, deliberately, and the split does
// not change the argument: RefreshStats remains the only caller of either
// half, in production it runs only on the event loop, and it now calls this
// after leaving the scope that serialised -- so the serialise and the write
// stay strictly ordered on one thread and pool_status.json still has a single
// writer by construction. It shares WriteFileAtomic so the durability rules
// stay in one place.
//
// "In production" is doing real work in that sentence. The payout-ledger
// self-test also calls RefreshStats, from one thread at a time and with the
// pool not running, which is why it is still a single writer there too. A
// second concurrent caller anywhere would invalidate this paragraph and want
// the same versioned publish the ledgers use.
static void PublishPoolStatus(const std::string& blob)
{
    WriteFileAtomic(PoolStatusFilePath(), blob, "pool status");
}

// ---------------------------------------------------------------------------
// 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

// #7: m->hashRate is an EWMA sampled only when a share arrives, so a miner
// that stops hashing keeps its last estimate forever and RefreshStats sums
// it into the announced pool total for as long as the connection sits open
// -- the published pool hashrate ran several times over ground truth
// because of exactly this. The decay pass in PoolEventLoop applies this
// policy: a miner gets a grace of HASHRATE_IDLE_MISSES expected share
// intervals (expected interval = expected hashes per share at its current
// difficulty / its last rate estimate) with no share, then its estimate
// decays with a half-life of one more such interval, computed from
// absolute time so the result does not depend on how often the loop ticks.
// The interval is clamped to [HASHRATE_MIN_INTERVAL, HASHRATE_MAX_INTERVAL]
// so a fast miner at the vardiff floor (expected interval of milliseconds)
// is not punished for ordinary share-delivery jitter, and a very slow
// miner's grace does not stretch past the point where its estimate means
// anything. At the vardiff target of one share per 15s, a stopped miner's
// announced rate is gone within minutes.
static const double HASHRATE_IDLE_MISSES  = 4.0;    // grace, in expected share intervals
static const double HASHRATE_MIN_INTERVAL = 1.0;    // seconds; floor for the expected interval
static const double HASHRATE_MAX_INTERVAL = 3600.0; // seconds; ceiling for the expected interval

struct Miner {
    btf_socket_t fd;
    bool         authorised    = false;
    bool         subscribed    = false;  // completed mining.subscribe (#19 census)
    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

    // Replay guard for mining.submit (#8): nonces already credited on this
    // connection for the job it is currently working. Keyed to the job id in
    // seenNoncesJob and cleared by SendJob whenever that id changes, so a
    // resubmission under a NEW job is never turned away, while a replayed
    // (job_id, nonce) pair under the same job is refused before it can be
    // credited twice into roundShareCount -- the figure the payout split
    // divides by.
    std::set<uint32_t> seenNonces;
    std::string        seenNoncesJob;

    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;

// #19 instrumentation: the "broadcast to N miner(s)" line counts connected
// sockets, which says nothing about how many are actually mining. These split
// the connection census into connected / subscribed / authorized / recently
// active, so an operator can tell a real workforce from a pile of half-open
// sessions at a glance.
static const int64 POOL_CENSUS_SECS       = 60;  // how often to log the breakdown
static const int64 POOL_ACTIVE_WINDOW     = 300; // a miner is "active" if it
                                                 // authorized and was seen within
                                                 // this window (one payout-rotate
                                                 // cycle)
static const int64 POOL_UNAUTH_GRACE_SECS = 60;  // a connection still not
                                                 // authorized after this long is
                                                 // idle, not merely starting up

// 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;
    // Every path that hands a miner a new job funnels through here, so this
    // is the one place the replay guard needs to follow the job id: nonces
    // credited under the previous job mean nothing under the new one. Keying
    // the clear to the id (rather than clearing unconditionally) keeps a
    // same-job re-notify -- the vardiff retarget and stall watchdog both
    // re-send the CURRENT job -- from reopening the duplicate window it
    // exists to close.
    if (m->seenNoncesJob != gCurrentJob.jobId) {
        m->seenNonces.clear();
        m->seenNoncesJob = gCurrentJob.jobId;
    }
    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);
}

// ---------------------------------------------------------------------------
// #19: connection census. "broadcast to N miner(s)" only counts authorized
// sockets that got a job; it says nothing about the rest of the connection
// table. This logs the full breakdown so an operator can tell a real workforce
// apart from half-open / never-authorized / idle sessions, and flags when the
// single select() loop is running out of room (issue #26).
//
// "Out of room" is not the same question on both platforms, and FD_SETSIZE
// answers a different one on each. On POSIX it bounds the *value* of any
// descriptor handed to FD_SET, so what matters is the highest fd we hold. On
// Windows an fd_set is a counted array of SOCKET handles whose numeric values
// are arbitrary, so what matters is how many we hold, not how large they are.
// FD_SETSIZE comes from <sys/select.h> or winsock2.h; this file must not define
// it, because redefining it after those headers changes nothing.
// ---------------------------------------------------------------------------
static void LogPoolConnectionCensus()
{
    int connected = 0, subscribed = 0, authorized = 0, active = 0, neverAuthIdle = 0;
    int64 now = GetTime();
#ifndef _WIN32
    int maxfd = 0;
#endif
    for (Miner* m : gMiners) {
        connected++;
#ifndef _WIN32
        if ((int)m->fd > maxfd) maxfd = (int)m->fd;
#endif
        if (m->subscribed) subscribed++;
        if (m->authorised) {
            authorized++;
            if (m->lastSeen > 0 && now - m->lastSeen <= POOL_ACTIVE_WINDOW)
                active++;
        } else if (now - m->connectTime >= POOL_UNAUTH_GRACE_SECS) {
            neverAuthIdle++; // past the grace period and still not authorized
        }
    }

#ifndef _WIN32
    int headroom = (int)FD_SETSIZE - maxfd;
    LogPrint("pool", "[pool] census: connected=%d subscribed=%d authorized=%d"
             " active(last %llds)=%d never-auth-idle=%d"
             "  (cap=%zu, maxfd=%d, FD_SETSIZE=%d)\n",
             connected, subscribed, authorized, (long long)POOL_ACTIVE_WINDOW,
             active, neverAuthIdle, MAX_POOL_MINERS, maxfd, (int)FD_SETSIZE);
#else
    int headroom = (int)FD_SETSIZE - connected;
    LogPrint("pool", "[pool] census: connected=%d subscribed=%d authorized=%d"
             " active(last %llds)=%d never-auth-idle=%d"
             "  (cap=%zu, FD_SETSIZE=%d sockets)\n",
             connected, subscribed, authorized, (long long)POOL_ACTIVE_WINDOW,
             active, neverAuthIdle, MAX_POOL_MINERS, (int)FD_SETSIZE);
#endif
    if (headroom <= 16)
        LogPrint("pool", "[pool] WARNING: %d slots left before FD_SETSIZE=%d --"
                 " the select() loop is about to walk into undefined behaviour"
                 " (#26). Refuse new miners or rebuild with poll/epoll.\n",
                 headroom, (int)FD_SETSIZE);
}

// ---------------------------------------------------------------------------
// 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);
        m->subscribed = true;
        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);

        // Reject a replayed (job, nonce) before it costs a rebuild and a
        // RandomX hash: nothing about the work below distinguishes a second
        // submission of a valid share from the first, so without this check
        // it passes every test and is credited again -- and roundShareCount
        // is what the payout split divides by, so each replay multiplies the
        // address's cut of the round. The set is per-connection and cleared
        // by SendJob on every job change, so only a same-job repeat is
        // refused.
        if (m->seenNonces.count(nNonce)) {
            LogPrint("worker", "[worker->pool] duplicate share from %s"
                     " nonce=%u job=%s -- already credited, not counting again\n",
                     m->address.c_str(), nNonce, submitJobId.c_str());
            reply(false, "Duplicate share");
            return true;
        }

        // 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++;
        m->seenNonces.insert(nNonce); // replay guard: credit each nonce once per job
        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.
            // Passed as a uint256, not hex: the ledger key must stay a real
            // hash. The remaining arguments are upstream's round-proof data.
            QueuePayouts(foundHeight, b.GetHash(), b.vtx[0].GetHash().GetHex(),
                         b.vtx[0].vout[0].nValue, m->address,
                         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;
}

// ---------------------------------------------------------------------------
// payout-ledger self-test
//
// Regression cover for the payout defects fixed in this series. Everything
// here runs against a scratch data directory and an in-memory wallet map; it
// never opens wallet.dat, never touches a real datadir, and never sends
// anything.
// ---------------------------------------------------------------------------

static bool PayoutTestReadFile(const std::string& path, std::string& out)
{
    FILE* f = fopen(path.c_str(), "r");
    if (!f) return false;
    fseek(f, 0, SEEK_END);
    long sz = ftell(f);
    fseek(f, 0, SEEK_SET);
    out.assign((size_t)(sz > 0 ? sz : 0), '\0');
    if (sz > 0 && fread(&out[0], 1, (size_t)sz, f) != (size_t)sz) {
        fclose(f);
        return false;
    }
    fclose(f);
    return true;
}

// Delete the scratch data directory and everything in it. A sweep and not a
// list of known names: LogPrint binds debug.log to whatever GetAppDir() is
// when it first fires, so the test's own log lands in here too, and a
// directory left behind in src/ after `make tests` is litter.
static void PayoutTestRemoveDir(const std::string& dir)
{
#ifdef _WIN32
    WIN32_FIND_DATAA findData;
    HANDLE hFind = FindFirstFileA((dir + "\\*").c_str(), &findData);
    if (hFind != INVALID_HANDLE_VALUE) {
        do {
            std::string name = findData.cFileName;
            if (name == "." || name == "..") continue;
            DeleteFileA((dir + "\\" + name).c_str());
        } while (FindNextFileA(hFind, &findData));
        FindClose(hFind);
    }
    RemoveDirectoryA(dir.c_str());
#else
    // One level of recursion, because CreateTransaction() opens a Berkeley DB
    // environment and BDB puts its logs in a "database" subdirectory. Without
    // this the scratch directory cannot be removed and the test leaves one
    // behind on every run, which is what d5b3503 was for.
    DIR* d = opendir(dir.c_str());
    if (d) {
        struct dirent* ent;
        while ((ent = readdir(d)) != NULL) {
            std::string name = ent->d_name;
            if (name == "." || name == "..") continue;
            const std::string path = dir + "/" + name;
            if (unlink(path.c_str()) != 0)
                PayoutTestRemoveDir(path);
        }
        closedir(d);
    }
    rmdir(dir.c_str());
#endif
}

static bool PayoutTestWriteFile(const std::string& path, const std::string& data)
{
    FILE* f = fopen(path.c_str(), "w");
    if (!f) return false;
    bool ok = data.empty() || fwrite(data.c_str(), 1, data.size(), f) == data.size();
    return fclose(f) == 0 && ok;
}

// Wipe every piece of payout state this test touches, so each section starts
// from the same place and a "restart" can be simulated by reloading from disk.
static void PayoutTestResetState()
{
    {
        std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
        gPendingPayouts.clear();
    }
    {
        std::lock_guard<std::mutex> lk(gPayoutExecMutex);
        gPaying.clear();
        gQueuedPayouts.clear();
    }
    {
        std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
        gPoolRounds.clear();
    }
    {
        std::lock_guard<std::mutex> lk(gClaimedTxidMutex);
        gLegacyClaimedTxids.clear();
    }
    {
        // A quarantine is meant to last the life of the process; each section
        // here is a fresh process as far as the code under test is concerned.
        std::lock_guard<std::mutex> lk(gRefusedFilesMutex);
        gRefusedFiles.clear();
    }
}

// A BTF address whose hash160 is entirely under the test's control, so the
// output script the probe compares against can be built without a wallet.
static std::string PayoutTestAddress(unsigned char seed)
{
    uint160 h = 0;
    for (int i = 0; i < 20; i++)
        ((unsigned char*)&h)[i] = (unsigned char)(seed + i);
    return Hash160ToAddress(h);
}

static CScript PayoutTestScriptFor(const std::string& address)
{
    uint160 h160;
    CScript sc;
    if (AddressToHash160(address, h160))
        sc << OP_DUP << OP_HASH160 << h160 << OP_EQUALVERIFY << OP_CHECKSIG;
    return sc;
}

// Block index entries fabricated by PayoutTestAddWalletTx, so they can be
// freed and unhooked when the test finishes. gPayoutTestChainTip is the end of
// that chain: blocks are linked onto it rather than each replacing the tip,
// because IsInMainChain() is "pnext is set, or I am the tip".
static std::vector<CBlockIndex*> gPayoutTestBlockIndexes;
static CBlockIndex*              gPayoutTestChainTip = NULL;

static void PayoutTestClearBlockIndex()
{
    CRITICAL_BLOCK(cs_main)
    {
        for (CBlockIndex* p : gPayoutTestBlockIndexes) {
            for (std::map<uint256, CBlockIndex*>::iterator it = mapBlockIndex.begin();
                 it != mapBlockIndex.end(); ) {
                if (it->second == p) mapBlockIndex.erase(it++);
                else ++it;
            }
            delete p;
        }
        gPayoutTestBlockIndexes.clear();
        gPayoutTestChainTip = NULL;
        pindexBest = NULL;
    }
}

// Put a transaction in the wallet map that looks exactly like one this pool
// sent. `nonce` only exists to keep the hashes distinct.
//
// fConfirmed decides what GetDepthInMainChain() will answer for it, which is
// what the legacy probe now gates on. A confirmed one is given a block index
// entry of its own: an empty merkle branch hashes to the transaction itself,
// so a block whose merkle root IS that hash accepts it at index 0, and
// pointing pindexBest at that block puts it in the main chain one deep. An
// unconfirmed one keeps hashBlock 0, which is depth 0 -- the shape a payout
// left behind by a reorg has.
static std::string PayoutTestAddWalletTx(const std::string& comment,
                                         const std::string& address,
                                         int64 amount,
                                         int nonce,
                                         const std::string& structuredId,
                                         bool fConfirmed = true)
{
    CWalletTx wtx;
    wtx.vin.resize(1);
    wtx.vin[0].prevout.SetNull();
    wtx.vin[0].scriptSig << nonce;
    wtx.vout.resize(1);
    wtx.vout[0].nValue = amount;
    wtx.vout[0].scriptPubKey = PayoutTestScriptFor(address);
    if (!comment.empty())
        wtx.mapValue["comment"] = comment;
    if (!structuredId.empty()) {
        wtx.mapValue["pool_payout_id"]      = structuredId;
        wtx.mapValue["pool_payout_address"] = address;
        wtx.mapValue["pool_payout_amount"]  = strprintf("%lld", amount);
    }
    uint256 hash = wtx.GetHash();
    if (fConfirmed) {
        // A block hash distinct from the transaction's own.
        uint256 blockHash = hash;
        ((unsigned char*)&blockHash)[0] ^= 0xff;
        CRITICAL_BLOCK(cs_main)
        {
            CBlockIndex* pindex = new CBlockIndex();
            pindex->hashMerkleRoot = hash;
            // Linked onto the end of the fabricated chain rather than simply
            // becoming the new tip: IsInMainChain() is "pnext is set, or I am
            // the tip", so a second confirmed transaction would otherwise take
            // the first one out of the chain again.
            pindex->pprev   = gPayoutTestChainTip;
            pindex->nHeight = gPayoutTestChainTip ? gPayoutTestChainTip->nHeight + 1 : 1;
            if (gPayoutTestChainTip)
                gPayoutTestChainTip->pnext = pindex;
            gPayoutTestChainTip = pindex;
            std::map<uint256, CBlockIndex*>::iterator mi =
                mapBlockIndex.insert(std::make_pair(blockHash, pindex)).first;
            pindex->phashBlock = &mi->first;
            pindexBest = pindex;
            gPayoutTestBlockIndexes.push_back(pindex);
        }
        wtx.hashBlock = blockHash;
        wtx.nIndex    = 0;
    }
    CRITICAL_BLOCK(cs_mapWallet)
        mapWallet[hash] = wtx;
    return hash.GetHex();
}

// A real key, in the keystore, so a test wallet's outputs are IsMine() and can
// actually be signed -- both are needed for CreateTransaction to build the
// large transaction the fee schedule is being tested against. Inserted
// straight into the key maps rather than through AddKey(), which would also
// write it to a wallet.dat.
static std::string PayoutTestAddSpendableKey()
{
    CKey key;
    key.MakeNewKey();
    std::vector<unsigned char> vchPubKey = key.GetPubKey();
    CRITICAL_BLOCK(cs_mapKeys)
    {
        mapKeys[vchPubKey]                = key.GetPrivKey();
        mapPubKeys[Hash160(vchPubKey)]    = vchPubKey;
    }
    return Hash160ToAddress(Hash160(vchPubKey));
}

// Give the wallet `n` spendable outputs of `each`, one transaction apiece --
// which is the granularity SelectCoins works at, and therefore the number of
// inputs the transaction spending them will carry.
//
// Deliberately not coinbases: a null prevout would make CWalletTx::GetCredit
// return 0 until COINBASE_SPENDABLE_DEPTH, and there is no chain here.
static void PayoutTestFundWallet(int n, int64 each, const std::string& address)
{
    const CScript sc = PayoutTestScriptFor(address);
    CRITICAL_BLOCK(cs_mapWallet)
    {
        for (int i = 0; i < n; i++) {
            CWalletTx wtx;
            wtx.vin.resize(1);
            wtx.vin[0].prevout.hash = (uint64)(i + 1);
            wtx.vin[0].prevout.n    = 0;
            wtx.vout.resize(1);
            wtx.vout[0].nValue       = each;
            wtx.vout[0].scriptPubKey = sc;
            mapWallet[wtx.GetHash()] = wtx;
        }
    }
}

// Defined below with the event loop; the status-file publication test drives
// it directly.
static void RefreshStats(int blocksFoundThisSession, uint64 roundShareTotal);

int RunPayoutLedgerSelfTest()
{
    int nFail = 0;
    auto check = [&](bool ok, const char* msg) {
        fprintf(stdout, "  %s   %s\n", ok ? "ok" : "FAIL", msg);
        if (!ok) nFail++;
    };

    fprintf(stdout, "payout-ledger self-test\n");

    const std::string savedDataDir = strSetDataDir;
    const int savedBestHeight = nBestHeight;

    // Scratch data directory in the current working directory. GetAppDir()
    // follows strSetDataDir, so this is what PayoutFilePath() and
    // PayingFilePath() resolve against for the rest of the test.
#ifdef _WIN32
    unsigned long pid = (unsigned long)GetCurrentProcessId();
#else
    unsigned long pid = (unsigned long)getpid();
#endif
    // Absolute, and the process moves into it.
    //
    // Both are needed by the CreateTransaction section: Berkeley DB resolves
    // its log directory relative to the environment home, so a relative
    // datadir makes dbenv.open() fail with DB_RUNRECOVERY; and BDB opens
    // db.log relative to the working directory, which is src/ during `make
    // tests`. Moving in puts it inside the scratch directory, where it dies
    // with everything else.
    std::string cwd = ".";
    {
#ifdef _WIN32
        char buf[MAX_PATH];
        if (GetCurrentDirectoryA(MAX_PATH, buf) > 0) cwd = buf;
#else
        char buf[4096];
        if (getcwd(buf, sizeof(buf)) != NULL) cwd = buf;
#endif
    }
    std::string dir = cwd + "/" +
                      strprintf("payout_selftest_%d_%lu", (int)GetTime(), pid);
#ifdef _WIN32
    CreateDirectoryA(dir.c_str(), NULL);
    SetCurrentDirectoryA(dir.c_str());
#else
    mkdir(dir.c_str(), 0700);
    if (chdir(dir.c_str()) != 0) { /* stay put; the paths are absolute anyway */ }
#endif
    strSetDataDir = dir;
    fprintf(stdout, "  scratch datadir: %s\n", dir.c_str());

    const std::string ledgerPath = PayoutFilePath();
    const std::string payingPath = PayingFilePath();
    const std::string addrA = PayoutTestAddress(0x11);
    const std::string addrB = PayoutTestAddress(0x55);

    // -- 1. A restart must not duplicate an in-flight record -----------------
    //
    // The ledger entry is grandfathered (zero hash) on purpose: it keeps
    // PayoutBlockStillOnChain out of the picture, which needs a block index,
    // and it exercises the height fallback at the same time.
    {
        PayoutTestResetState();
        remove(ledgerPath.c_str());
        remove(payingPath.c_str());

        PendingPayout pp;
        pp.blockHeight    = 4242;
        pp.blockHash      = 0;
        pp.matureAtHeight = 4242;
        pp.amounts[addrA] = 10 * COIN;
        pp.amounts[addrB] = 20 * COIN;
        VersionedBlob snap;
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            gPendingPayouts.push_back(pp);
            snap = SerializePendingPayoutsLocked();
        }
        PublishPendingPayouts(snap);

        nBestHeight = 5000;
        FlushMaturePayouts();
        size_t afterFirstFlush = 0;
        {
            std::lock_guard<std::mutex> lk(gPayoutExecMutex);
            afterFirstFlush = gPaying.size();
        }
        check(afterFirstFlush == 2,
              "a matured round explodes into one record per unpaid recipient");

        // Restart: drop everything in memory and come back from the files.
        for (int restart = 1; restart <= 2; restart++) {
            PayoutTestResetState();
            LoadPendingPayouts();
            LoadPaying();
            ReconcilePayouts();

            size_t queued = 0;
            {
                std::lock_guard<std::mutex> lk(gPayoutExecMutex);
                queued = gQueuedPayouts.size();
            }
            if (restart == 1)
                check(queued == 1,
                      "startup rebuilds the exploded-round set from the restored"
                      " in-flight list");

            FlushMaturePayouts();
            size_t n = 0;
            {
                std::lock_guard<std::mutex> lk(gPayoutExecMutex);
                n = gPaying.size();
            }
            check(n == 2, restart == 1
                  ? "a restart does not duplicate an in-flight record"
                  : "a second restart does not duplicate one either");
        }

        // The other half of the guard: even with the round-level dedupe set
        // deliberately emptied, the per-recipient check against gPaying must
        // stop a second standing order for the same money.
        {
            std::lock_guard<std::mutex> lk(gPayoutExecMutex);
            gQueuedPayouts.clear();
        }
        FlushMaturePayouts();
        size_t n = 0;
        {
            std::lock_guard<std::mutex> lk(gPayoutExecMutex);
            n = gPaying.size();
        }
        check(n == 2,
              "the flush refuses a second record for a (round, address) already"
              " in flight even with the round set lost");
    }

    // -- 2. Duplicates already on disk are collapsed as they are read --------
    {
        PayoutTestResetState();
        nlohmann::json arr = nlohmann::json::array();
        for (int i = 0; i < 2; i++) {
            nlohmann::json o;
            o["blockHeight"]    = 4242;
            o["blockHash"]      = uint256(0).ToString();
            o["matureAtHeight"] = 4242;
            o["address"]        = addrA;
            o["amount"]         = (int64)(10 * COIN);
            o["retries"]        = i == 0 ? 7 : 0;
            o["deferrals"]      = i == 0 ? 0 : 3;
            arr.push_back(o);
        }
        check(PayoutTestWriteFile(payingPath, arr.dump(2)),
              "wrote a paying file containing the duplicate the old build made");
        LoadPaying();
        std::lock_guard<std::mutex> lk(gPayoutExecMutex);
        check(gPaying.size() == 1,
              "loading collapses a duplicate in-flight record from disk");
        check(gPaying.size() == 1 && gPaying[0].retries == 7 && gPaying[0].deferrals == 3,
              "collapsing merges the attempt counts upwards, not resetting them");
    }

    // -- 3. State updates address a slot, not a value -----------------------
    //
    // Build the exact pathology the old code produced -- two records equal in
    // every field -- and prove the update lands on the one asked for. Under
    // value matching the first record absorbed everything and the second was
    // untouchable.
    {
        PayoutTestResetState();
        PayingRecord a;
        a.blockHeight = 4242; a.blockHash = 0; a.matureAtHeight = 4242;
        a.address = addrA; a.amount = 10 * COIN;
        PayingRecord b = a;
        {
            std::lock_guard<std::mutex> lk(gPayoutExecMutex);
            a.id = ++gPayingNextId;
            b.id = ++gPayingNextId;
            gPaying.push_back(a);
            gPaying.push_back(b);
        }
        PayingRecord upd = b;
        upd.retries = 5;
        upd.retryAfter = 1234567;
        ReschedulePayment(upd);
        {
            std::lock_guard<std::mutex> lk(gPayoutExecMutex);
            check(gPaying.size() == 2 &&
                  gPaying[0].retries == 0 && gPaying[1].retries == 5,
                  "rescheduling updates the record it was given, not the first"
                  " one that looks like it");
        }
        ForgetPayment(a);
        {
            std::lock_guard<std::mutex> lk(gPayoutExecMutex);
            check(gPaying.size() == 1 && gPaying[0].id == b.id,
                  "forgetting erases the record it was given, not the first"
                  " one that looks like it");
        }
    }

    // -- 4. retryAfter survives a restart ------------------------------------
    {
        PayoutTestResetState();
        remove(payingPath.c_str());
        PayingRecord r;
        r.blockHeight = 77; r.blockHash = 0; r.matureAtHeight = 77;
        r.address = addrA; r.amount = 3 * COIN;
        r.retryAfter = 1700000000LL;
        r.deferrals = 9;
        VersionedBlob snap;
        {
            std::lock_guard<std::mutex> lk(gPayoutExecMutex);
            r.id = ++gPayingNextId;
            gPaying.push_back(r);
            snap = SerializePayingLocked();
        }
        PublishPaying(snap);
        PayoutTestResetState();
        LoadPaying();
        std::lock_guard<std::mutex> lk(gPayoutExecMutex);
        check(gPaying.size() == 1 && gPaying[0].retryAfter == 1700000000LL &&
              gPaying[0].deferrals == 9,
              "the retry clock that is actually used survives a restart");
    }

    // -- 5. The ledger file cannot be published from a stale snapshot --------
    {
        PayoutTestResetState();
        remove(ledgerPath.c_str());

        PendingPayout one;
        one.blockHeight = 100; one.blockHash = 0; one.matureAtHeight = 100;
        one.amounts[addrA] = 1 * COIN;
        VersionedBlob older, newer;
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            gPendingPayouts.push_back(one);
            older = SerializePendingPayoutsLocked();
            PendingPayout two;
            two.blockHeight = 200; two.blockHash = 0; two.matureAtHeight = 200;
            two.amounts[addrB] = 2 * COIN;
            gPendingPayouts.push_back(two);
            newer = SerializePendingPayoutsLocked();
        }
        check(newer.seq > older.seq,
              "a later serialisation carries a strictly higher version");

        // The order the race produced: the payout thread's older image
        // arriving after the event loop's newer one.
        PublishPendingPayouts(newer);
        PublishPendingPayouts(older);
        std::string onDisk;
        check(PayoutTestReadFile(ledgerPath, onDisk) && onDisk == newer.blob,
              "an older snapshot cannot overwrite a newer published ledger");

        // And the ordinary direction still works.
        VersionedBlob newest;
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            gPendingPayouts.clear();
            newest = SerializePendingPayoutsLocked();
        }
        PublishPendingPayouts(newest);
        check(PayoutTestReadFile(ledgerPath, onDisk) && onDisk == newest.blob,
              "a newer snapshot does replace the published ledger");
    }

    // -- 6. Concurrent publishers cannot interleave a file -------------------
    //
    // Two threads opening and renaming the same "<path>.tmp" over the same
    // target was the original defect. This test was mutation-tested and did
    // not catch it: with the file mutex removed the suite still went green 2
    // runs in 5, and the "never leave it unparseable" assertion fired 0 times
    // in 5. Two reasons, both fixed here.
    //
    // First, every image had the same structure and near enough the same
    // length, so an interleaving of two of them was still valid JSON. The
    // images now alternate between ONE entry and FIVE THOUSAND, so a write
    // that lands on top of another leaves either a truncated array or a small
    // image followed by the tail of a large one -- neither of which parses.
    //
    // Second, and worse, almost nothing was being written. Each thread walked
    // a permutation of the images, and PublishVersioned drops any blob that is
    // not newer than what is on disk, so only the record highs in each
    // permutation ever reached the disk -- on the order of ten writes in
    // total, four threads, no contention to speak of. Every publish here is
    // stamped from a shared counter at the moment it is made, so every one of
    // them is newer than the last and every one of them writes.
    //
    // And the file is watched while the writers run, not only inspected after
    // they finish. "The published file is never observed unparseable" is the
    // actual invariant -- rename(2) is atomic, so a reader can only ever see a
    // completely written image -- and checking it once at the end samples the
    // one moment when a torn file is least likely to still be there.
    {
        const std::string racePath = dir + "/publish_race_selftest.json";
        std::mutex raceMutex;
        uint64 racePublished = 0;

        std::string tiny, huge;
        {
            nlohmann::json one = nlohmann::json::array();
            nlohmann::json o;
            o["image"]   = 0;
            o["entry"]   = 0;
            o["address"] = addrA;
            o["amount"]  = (int64)COIN;
            one.push_back(o);
            tiny = one.dump(2);

            nlohmann::json many = nlohmann::json::array();
            for (int k = 0; k < 5000; k++) {
                nlohmann::json e;
                e["image"]   = 1;
                e["entry"]   = k;
                e["address"] = addrA;
                e["amount"]  = (int64)(k + 1) * COIN;
                many.push_back(e);
            }
            huge = many.dump(2);
        }

        const int kThreads   = 4;
        const int kPerThread = 100;
        const uint64 kTotal  = (uint64)kThreads * kPerThread;

        std::atomic<uint64> nextSeq(0);
        std::atomic<bool>   racing(true);
        std::atomic<int>    tornReads(0);
        std::atomic<int>    goodReads(0);

        std::thread watcher([&]() {
            bool seenOnce = false;
            while (racing) {
                std::string s;
                if (!PayoutTestReadFile(racePath, s)) {
                    // Only meaningful once the file exists at all.
                    if (seenOnce) tornReads++;
                    continue;
                }
                seenOnce = true;
                try {
                    if (nlohmann::json::parse(s).is_array()) goodReads++;
                    else tornReads++;
                } catch (...) {
                    tornReads++;
                }
            }
        });

        std::vector<std::thread> threads;
        for (int t = 0; t < kThreads; t++) {
            threads.push_back(std::thread([&]() {
                for (int i = 0; i < kPerThread; i++) {
                    VersionedBlob v;
                    v.seq  = ++nextSeq;
                    v.blob = (v.seq & 1) ? tiny : huge;
                    PublishVersioned(racePath, v, raceMutex, racePublished,
                                     "publish race selftest");
                }
            }));
        }
        for (std::thread& th : threads)
            th.join();
        racing = false;
        watcher.join();

        check(goodReads > 0,
              "the race test actually observed the file while it was being"
              " written");
        check(tornReads == 0,
              "four threads publishing the same file never leave it"
              " unparseable, at any instant while they are doing it");

        // The highest sequence number issued is the one the file must settle
        // on: PublishVersioned refuses anything older, so whichever order the
        // writes complete in, that image is the last one that can land.
        const std::string& expected = (kTotal & 1) ? tiny : huge;
        std::string onDisk;
        bool read = PayoutTestReadFile(racePath, onDisk);
        check(read && onDisk == expected,
              "the published file settles on the highest version, whatever the"
              " completion order");
        remove(racePath.c_str());
        remove((racePath + ".tmp").c_str());
    }

    // -- 7. Grandfathered zero-hash rounds do not answer for each other ------
    {
        uint256 h1 = 0, h2 = 0;
        h1.SetHex("00000000000000000000000000000000000000000000000000000000000000aa");
        h2.SetHex("00000000000000000000000000000000000000000000000000000000000000bb");
        check(!SamePayoutRound(100, 0, 200, 0),
              "two zero-hash entries at different heights are different rounds");
        check(SamePayoutRound(100, 0, 100, 0),
              "two zero-hash entries at the same height are the same round");
        check(SamePayoutRound(100, h1, 200, h1),
              "a real hash identifies the round on its own");
        check(!SamePayoutRound(100, h1, 100, h2),
              "different hashes at one height are different rounds");
        check(!SamePayoutRound(100, h1, 100, 0),
              "a grandfathered entry never matches a hash-keyed one");
        check(!PayoutRoundIdentifiable(0, 0) && PayoutRoundIdentifiable(0, h1) &&
              PayoutRoundIdentifiable(7, 0),
              "an entry with neither hash nor height is not adjudicable");

        // What actually makes a grandfathered same-height pair safe, since the
        // comment on FindRecordedPoolPayout used to claim it was the orphan
        // check. It is not: the orphan check passes every zero-hash entry, and
        // every pre-upgrade entry has a zero hash.
        check(PayoutBlockStillOnChain(0),
              "the orphan check passes every grandfathered entry, so it cannot"
              " be what voids a losing twin");
        PayoutTestResetState();
        remove(ledgerPath.c_str());
        remove(payingPath.c_str());
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            PendingPayout twin;
            twin.blockHeight = 300; twin.blockHash = 0; twin.matureAtHeight = 300;
            twin.amounts[addrA] = 1 * COIN;
            gPendingPayouts.push_back(twin);
            gPendingPayouts.push_back(twin);   // the losing side of a reorg
        }
        nBestHeight = 5000;
        FlushMaturePayouts();
        {
            std::lock_guard<std::mutex> lk(gPayoutExecMutex);
            check(gPaying.size() == 1,
                  "two grandfathered rounds at one height are one round, so the"
                  " pair presents as one debt and is paid at most once");
        }

        // End to end: recording a payment against one grandfathered round must
        // not settle the other's debt.
        PayoutTestResetState();
        remove(ledgerPath.c_str());
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            PendingPayout first;
            first.blockHeight = 100; first.blockHash = 0; first.matureAtHeight = 100;
            first.amounts[addrA] = 1 * COIN;
            PendingPayout second;
            second.blockHeight = 200; second.blockHash = 0; second.matureAtHeight = 200;
            second.amounts[addrA] = 1 * COIN;
            gPendingPayouts.push_back(first);
            gPendingPayouts.push_back(second);
        }
        PayingRecord r;
        r.blockHeight = 200; r.blockHash = 0; r.matureAtHeight = 200;
        r.address = addrA; r.amount = 1 * COIN;
        RecordPayoutTxid(r, "txid-for-the-second-round");
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            // The second round was fully paid and retired; the first is
            // untouched and still owed.
            check(gPendingPayouts.size() == 1 &&
                  gPendingPayouts[0].blockHeight == 100 &&
                  gPendingPayouts[0].paidTxids.empty(),
                  "a txid recorded against one zero-hash round leaves the other"
                  " still owed");
        }
    }

    // -- 8. A payment with only a legacy comment is recognised as paid -------
    {
        PayoutTestResetState();
        const int    height = 8080;
        const int64  amount = 7 * COIN;
        const std::string id = PayoutIdForBlock(height, 0);
        const std::string comment = PayoutComment(height);
        std::set<std::string> none;
        std::string found;

        check(FindRecordedPoolPayout(id, height, addrA, amount, none, found)
                  == PAYOUT_PROBE_NONE,
              "an empty wallet is not evidence that anything was paid");

        const std::string legacyTxid =
            PayoutTestAddWalletTx(comment, addrA, amount, 1, "");
        check(FindRecordedPoolPayout(id, height, addrA, amount, none, found)
                  == PAYOUT_PROBE_LEGACY && found == legacyTxid,
              "a payment carrying only a pre-upgrade comment is recognised as"
              " already paid");

        check(FindRecordedPoolPayout(id, height, addrB, amount, none, found)
                  == PAYOUT_PROBE_NONE,
              "the legacy match requires the output to pay THIS recipient");
        check(FindRecordedPoolPayout(id, height, addrA, amount + COIN, none, found)
                  == PAYOUT_PROBE_NONE,
              "the legacy match requires the output to pay exactly this amount");
        check(FindRecordedPoolPayout(id, height + 1, addrA, amount, none, found)
                  == PAYOUT_PROBE_NONE,
              "the legacy match requires the comment for this block height");

        std::set<std::string> spent;
        spent.insert(legacyTxid);
        check(FindRecordedPoolPayout(id, height, addrA, amount, spent, found)
                  == PAYOUT_PROBE_NONE,
              "one legacy transaction cannot settle a second debt");

        // A transaction written by THIS build carries pool_payout_id, and must
        // never be offered as legacy evidence for a different round -- that is
        // exactly what the structured metadata exists to prevent.
        const std::string otherId = PayoutIdForBlock(height, 0) + "-other";
        PayoutTestAddWalletTx(comment, addrB, amount, 2, otherId);
        check(FindRecordedPoolPayout(id, height, addrB, amount, none, found)
                  == PAYOUT_PROBE_NONE,
              "another round's structured payment is not legacy evidence for"
              " this one");

        // The structured path still wins where it applies.
        const std::string structuredTxid =
            PayoutTestAddWalletTx(comment, addrB, amount, 3, id);
        check(FindRecordedPoolPayout(id, height, addrB, amount, none, found)
                  == PAYOUT_PROBE_STRUCTURED && found == structuredTxid,
              "structured payout metadata is matched in preference to a comment");

        // A comment-matching transaction that never made it into the chain is
        // evidence of NON-payment. This tree has seen a 280-block reorg; a
        // pre-upgrade payout invalidated by one is still in mapWallet with its
        // comment intact, and the first boot after the upgrade is exactly when
        // it would be matched.
        CRITICAL_BLOCK(cs_mapWallet)
            mapWallet.clear();
        PayoutTestClearBlockIndex();
        const std::string addrC = PayoutTestAddress(0x33);
        PayoutTestAddWalletTx(comment, addrC, amount, 4, "", false);
        check(FindRecordedPoolPayout(id, height, addrC, amount, none, found)
                  == PAYOUT_PROBE_NONE,
              "a pre-upgrade payout that never confirmed is not read as"
              " payment");

        // The same transaction, confirmed, is.
        CRITICAL_BLOCK(cs_mapWallet)
            mapWallet.clear();
        const std::string confirmedTxid =
            PayoutTestAddWalletTx(comment, addrC, amount, 4, "", true);
        check(FindRecordedPoolPayout(id, height, addrC, amount, none, found)
                  == PAYOUT_PROBE_LEGACY && found == confirmedTxid,
              "the same payout, confirmed, is");

        // And an unconfirmed one does not shadow a confirmed one that would
        // have matched: the probe used to stop at the first comment match.
        const std::string alsoConfirmed =
            PayoutTestAddWalletTx(comment, addrC, amount, 6, "", true);
        PayoutTestAddWalletTx(comment, addrC, amount, 5, "", false);
        std::set<std::string> spentFirst;
        spentFirst.insert(confirmedTxid);
        check(FindRecordedPoolPayout(id, height, addrC, amount, spentFirst, found)
                  == PAYOUT_PROBE_LEGACY && found == alsoConfirmed,
              "an unconfirmed transaction does not hide a confirmed one behind"
              " it");

        CRITICAL_BLOCK(cs_mapWallet)
            mapWallet.clear();
        PayoutTestClearBlockIndex();
    }

    // -- 9. On-disk compatibility with the live production ledger ------------
    {
        PayoutTestResetState();
        // Exactly the shape of the file on the pool node: amounts, blockHash,
        // blockHeight, matureAtHeight, and nothing else.
        const char* live =
            "[{\"amounts\":{\"%s\":1234500000},"
            "\"blockHash\":\"00000000000000000000000000000000000000000000000000000000000000cd\","
            "\"blockHeight\":1000,"
            "\"matureAtHeight\":1120}]";
        check(PayoutTestWriteFile(ledgerPath, strprintf(live, addrA.c_str())),
              "wrote a ledger file in the live production shape");
        LoadPendingPayouts();
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            check(gPendingPayouts.size() == 1 &&
                  gPendingPayouts[0].blockHeight == 1000 &&
                  gPendingPayouts[0].matureAtHeight == 1120 &&
                  gPendingPayouts[0].blockHash != 0 &&
                  gPendingPayouts[0].amounts.size() == 1 &&
                  gPendingPayouts[0].amounts[addrA] == 1234500000LL,
                  "the live production pending_payouts.json still loads");
        }

        // And the shape the merged build writes, retryAfter included, which
        // this build no longer writes but must still read past.
        PayoutTestResetState();
        const char* merged =
            "[{\"amounts\":{\"%s\":100000000},"
            "\"blockHash\":\"00000000000000000000000000000000000000000000000000000000000000ef\","
            "\"blockHeight\":2000,\"matureAtHeight\":2120,\"retryAfter\":0,"
            "\"paidTxids\":{\"%s\":\"abc123\"}}]";
        check(PayoutTestWriteFile(ledgerPath,
                                  strprintf(merged, addrA.c_str(), addrA.c_str())),
              "wrote a ledger file in the merged build's shape");
        LoadPendingPayouts();
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            check(gPendingPayouts.size() == 1 &&
                  gPendingPayouts[0].blockHeight == 2000 &&
                  gPendingPayouts[0].paidTxids.size() == 1 &&
                  gPendingPayouts[0].paidTxids[addrA] == "abc123",
                  "a ledger file carrying retryAfter and paidTxids still loads");
        }

        // A round trip through this build must not lose anything either.
        VersionedBlob snap;
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            snap = SerializePendingPayoutsLocked();
        }
        PublishPendingPayouts(snap);
        PayoutTestResetState();
        LoadPendingPayouts();
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            check(gPendingPayouts.size() == 1 &&
                  gPendingPayouts[0].blockHeight == 2000 &&
                  gPendingPayouts[0].matureAtHeight == 2120 &&
                  gPendingPayouts[0].paidTxids[addrA] == "abc123",
                  "what this build writes, this build reads back unchanged");
        }
    }

    // -- 10. A forfeit is published, not just dropped ------------------------
    {
        PayoutTestResetState();
        remove(ledgerPath.c_str());
        const std::string roundsPath = dir + "/pool_rounds_selftest.json";
        const std::string savedRoundsFile = strPoolRoundsFile;
        strPoolRoundsFile = roundsPath;
        remove(roundsPath.c_str());

        const int height = 606;
        PoolRoundProof round;
        round.id = PayoutIdForBlock(height, 0);
        round.blockHeight = height;
        round.matureAtHeight = height;
        round.payouts[addrA] = 5 * COIN;
        round.payouts[addrB] = 6 * COIN;
        RecordPoolRound(round);

        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            PendingPayout pp;
            pp.blockHeight = height; pp.blockHash = 0; pp.matureAtHeight = height;
            pp.amounts[addrA] = 5 * COIN;
            pp.amounts[addrB] = 6 * COIN;
            gPendingPayouts.push_back(pp);
        }

        PayingRecord r;
        r.blockHeight = height; r.blockHash = 0; r.matureAtHeight = height;
        r.address = addrA; r.amount = 5 * COIN;
        ForfeitPayoutRecipient(r, "selftest reason");

        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            check(gPendingPayouts.size() == 1 &&
                  gPendingPayouts[0].amounts.count(addrA) == 0 &&
                  gPendingPayouts[0].amounts.count(addrB) == 1,
                  "a forfeit removes only that recipient from the ledger entry");
        }

        std::string onDisk;
        bool proofOk = false;
        if (PayoutTestReadFile(roundsPath, onDisk)) {
            try {
                nlohmann::json root = nlohmann::json::parse(onDisk);
                const nlohmann::json& p = root["rounds"][0]["payouts"];
                proofOk = p[addrA].value("forfeited", false) &&
                          p[addrA].value("forfeitReason", "") == "selftest reason" &&
                          p[addrA]["txid"].is_null() &&
                          !p[addrB].contains("forfeited");
            } catch (...) {
                proofOk = false;
            }
        }
        check(proofOk,
              "the published round proof records the forfeit instead of"
              " advertising the address as still owed");

        // And it survives a reload of the proof file.
        {
            std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
            gPoolRounds.clear();
        }
        LoadPoolRounds();
        {
            std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
            check(gPoolRounds.size() == 1 &&
                  gPoolRounds[0].forfeited.count(addrA) == 1 &&
                  gPoolRounds[0].forfeited[addrA] == "selftest reason",
                  "the forfeit is read back from pool_rounds.json");
        }

        remove(roundsPath.c_str());
        remove((roundsPath + ".tmp").c_str());
        strPoolRoundsFile = savedRoundsFile;
    }

    // -- 11. A ledger that failed to load is never written back --------------
    //
    // The loaders push entries into the live vector inside the try, so an
    // exception part way down a file leaves a PARTIAL ledger in memory: the
    // entries before the bad one and nothing after it. Publishing that -- and
    // ReconcilePayouts publishes unconditionally, milliseconds into startup --
    // writes the partial state over the file and destroys the only record of
    // what was owed.
    {
        PayoutTestResetState();
        remove(ledgerPath.c_str());

        // Two entries, the second with a matureAtHeight that is not a number.
        // The first loads; the second throws where it is read.
        const std::string corrupt = strprintf(
            "[{\"amounts\":{\"%s\":100000000},\"blockHash\":\"%s\","
            "\"blockHeight\":10,\"matureAtHeight\":10},"
            "{\"amounts\":{\"%s\":900000000},\"blockHash\":\"%s\","
            "\"blockHeight\":20,\"matureAtHeight\":\"twenty\"}]",
            addrA.c_str(), uint256(0).ToString().c_str(),
            addrB.c_str(), uint256(0).ToString().c_str());
        check(PayoutTestWriteFile(ledgerPath, corrupt),
              "wrote a ledger whose second entry does not parse");

        const int64 t0 = GetTime();
        LoadPendingPayouts();
        const int64 t1 = GetTime();

        size_t loaded = 0;
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            loaded = gPendingPayouts.size();
        }
        check(loaded == 1,
              "a mid-file parse error does leave a partial ledger in memory --"
              " which is the reason for everything below");

        std::string gone;
        check(!PayoutTestReadFile(ledgerPath, gone),
              "the file that did not parse is moved out of the way");

        // The rename stamps GetTime(); it was taken between t0 and t1.
        std::string kept, keptPath;
        for (int64 t = t0; t <= t1 && keptPath.empty(); t++) {
            const std::string candidate =
                strprintf("%s.bad.%lld", ledgerPath.c_str(), (long long)t);
            if (PayoutTestReadFile(candidate, kept))
                keptPath = candidate;
        }
        check(!keptPath.empty() && kept == corrupt,
              "it is kept, byte for byte, as <path>.bad.<timestamp>");

        // The publish that used to destroy it.
        VersionedBlob partial;
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            partial = SerializePendingPayoutsLocked();
        }
        PublishPendingPayouts(partial);
        check(!PayoutTestReadFile(ledgerPath, gone),
              "and the partial ledger is not published over it");

        // The refusal is per path and lasts the process: a second attempt,
        // with more state, is refused too.
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            PendingPayout pp;
            pp.blockHeight = 30; pp.blockHash = 0; pp.matureAtHeight = 30;
            pp.amounts[addrA] = 5 * COIN;
            gPendingPayouts.push_back(pp);
            partial = SerializePendingPayoutsLocked();
        }
        PublishPendingPayouts(partial);
        check(!PayoutTestReadFile(ledgerPath, gone),
              "the refusal is not a one-shot -- the path stays unwritable");

        if (!keptPath.empty())
            remove(keptPath.c_str());
        PayoutTestResetState();
        remove(ledgerPath.c_str());

        // And a file that parses is not quarantined, so the ordinary path is
        // still the ordinary path.
        check(PayoutTestWriteFile(ledgerPath, "[]"),
              "wrote a ledger that parses");
        LoadPendingPayouts();
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            PendingPayout pp;
            pp.blockHeight = 40; pp.blockHash = 0; pp.matureAtHeight = 40;
            pp.amounts[addrA] = 2 * COIN;
            gPendingPayouts.push_back(pp);
            partial = SerializePendingPayoutsLocked();
        }
        PublishPendingPayouts(partial);
        std::string written;
        check(PayoutTestReadFile(ledgerPath, written) && written == partial.blob,
              "a ledger that loaded cleanly is still published normally");
    }

    // -- 12. A send that failed for want of coin is not a fault --------------
    //
    // The defect this replaces: the classifier decided "was this a funding
    // problem?" by re-reading GetBalance() and allowing amount +
    // nTransactionFee + CENT. nTransactionFee is 0 on a headless node, and
    // GetMinFee() is a step function -- 0 below 10,000 bytes, 11 * CENT at it
    // -- so the allowance was either unnecessary or eleven times too small,
    // and the ledger's own retry budget then wrote the debt off after 1440
    // repeats of the same stable condition.
    //
    // This is tested against a real wallet and the real CreateTransaction, not
    // against injected values: the honest question is not "does my enum
    // switch work" but "does a wallet holding many small coins actually report
    // the state I think it does". 60 outputs of 1 BTF give ~180 signed P2PKH
    // bytes each, which is where the fee schedule steps. Nothing is committed
    // or broadcast -- CreateTransaction only builds.
    {
        PayoutTestResetState();
        CRITICAL_BLOCK(cs_mapWallet)
            mapWallet.clear();

        // CreateTransaction opens CTxDB("r"), which throws if blkindex.dat is
        // not there. Create it in the scratch datadir first.
        bool dbOk = true;
        try {
            CTxDB txdb("cr");
        } catch (const std::exception&) {
            dbOk = false;
        }
        check(dbOk, "a scratch block index exists for CreateTransaction to open");

        const std::string walletAddr = PayoutTestAddSpendableKey();
        const std::string payee      = PayoutTestAddress(0x77);
        const CScript     payeeScript = PayoutTestScriptFor(payee);

        // -- the >= 10 KB case: 60 small coins, balance = amount + 0.05 ------
        PayoutTestFundWallet(60, 1 * COIN, walletAddr);
        const int64 balance = GetBalance();
        check(balance == 60 * COIN,
              "the test wallet holds 60 separately spendable coins");

        const int64 amount = balance - 5 * CENT;
        // The pre-send gate in PayoutThreadFn passes: the balance covers the
        // amount and the only fee that can be known in advance.
        check(balance >= amount + nTransactionFee,
              "the pre-send gate would let this payment through");

        CWalletTx        wtx;
        SendMoneyOutcome outcome;
        int64            feeRequired = 0;
        WalletSendResult result      = WALLET_SEND_OK;
        bool             built       = true;
        if (dbOk) {
            try {
                built = CreateTransaction(payeeScript, amount, wtx, feeRequired,
                                          &result);
            } catch (const std::exception&) {
                built = true;   // report as a failure below
            }
        }
        check(dbOk && !built,
              "a wallet of many small coins cannot fund amount + the fee its"
              " own size demands");
        check(dbOk && result == WALLET_SEND_INSUFFICIENT_FUNDS,
              "the wallet reports that as a funding shortfall, not as a"
              " failure to build");
        // GetMinFee(fDiscount) returns 0 below 10,000 bytes and
        // (1 + bytes/1000) * CENT at or above, so a required fee of 11 CENT or
        // more IS the statement that the transaction crossed 10 KB. Measured
        // through the real serialiser rather than asserted.
        check(dbOk && feeRequired >= 11 * CENT,
              "the transaction it built crossed 10,000 bytes, so the fee"
              " stepped from 0 to at least 11 CENT");

        // The defect, stated as arithmetic. The old classifier asked whether
        // the balance covered amount + nTransactionFee + CENT; it did, so it
        // called a permanent funding shortfall a fault and started counting
        // towards a write-off.
        check(!(balance < amount + nTransactionFee + CENT),
              "the reserve this replaces would have judged this send a fault");
        check(balance < amount + feeRequired,
              "while the fee actually required was more than the balance had");

        outcome.result       = result;
        outcome.nFeeRequired = feeRequired;
        check(dbOk && ClassifyPayoutSendFailure(outcome.result, false)
                          == PAYOUT_FAILURE_DEFER,
              "a >= 10 KB send that ran out of coin is deferred, not counted"
              " against the payment");

        // -- the < 10 KB case: no reserve is needed at all -------------------
        //
        // The other half of the arithmetic. Below the step the required fee is
        // 0, so a reserve of any size is a reason to refuse a payment the
        // wallet could have made -- which is its own way of not paying a
        // miner.
        CRITICAL_BLOCK(cs_mapWallet)
            mapWallet.clear();
        PayoutTestFundWallet(3, 1 * COIN, walletAddr);
        const int64 smallAmount = 3 * COIN - 5 * CENT;
        CWalletTx   smallTx;
        int64       smallFee    = -1;
        WalletSendResult smallResult = WALLET_SEND_FAILED;
        bool        smallBuilt  = false;
        if (dbOk) {
            try {
                smallBuilt = CreateTransaction(payeeScript, smallAmount, smallTx,
                                               smallFee, &smallResult);
            } catch (const std::exception&) {
                smallBuilt = false;
            }
        }
        check(dbOk && smallBuilt && smallResult == WALLET_SEND_OK,
              "the same balance pays a small transaction with no headroom at"
              " all");
        check(dbOk && smallFee == 0,
              "because below 10,000 bytes the required fee is zero");

        // -- and the classification of every other answer --------------------
        check(ClassifyPayoutSendFailure(WALLET_SEND_INSUFFICIENT_FUNDS, false)
                  == PAYOUT_FAILURE_DEFER &&
              ClassifyPayoutSendFailure(WALLET_SEND_INSUFFICIENT_FUNDS, true)
                  == PAYOUT_FAILURE_DEFER &&
              ClassifyPayoutSendFailure(WALLET_SEND_FAILED, true)
                  == PAYOUT_FAILURE_DEFER,
              "short of coin or unable to sign is always a deferral");
        check(ClassifyPayoutSendFailure(WALLET_SEND_FAILED, false)
                  == PAYOUT_FAILURE_FAULT,
              "funded and unlocked and still refused is a fault");

        CRITICAL_BLOCK(cs_mapWallet)
            mapWallet.clear();
    }

    // -- 13. No payout mutex is held across the status-file write ------------
    //
    // RefreshStats runs on the event loop -- the thread that owns every
    // miner's socket -- and used to hold gStatsMutex and gPayoutLedgerMutex at
    // function scope across WriteFileAtomic. Once that write gained fsync() on
    // the file and on its directory, publishing pool_status.json parked the
    // payout thread behind two disk flushes every ten seconds.
    //
    // Timing assertions would be flaky, so this does not measure anything. It
    // makes the write block for as long as the test likes: strPoolStatusFile
    // points at a path whose ".tmp" is a FIFO, and open(2) for writing on a
    // FIFO does not return until a reader arrives. While RefreshStats is
    // parked in there, every mutex it must NOT be holding has to be
    // acquirable. Against the previous code this test does not merely fail, it
    // cannot pass: the writer is inside fopen() with both mutexes held and
    // nothing will release them until the test lets the write finish.
#ifndef _WIN32
    {
        PayoutTestResetState();
        const std::string statusPath = dir + "/pool_status_selftest.json";
        const std::string statusTmp  = statusPath + ".tmp";
        const std::string savedStatusFile = strPoolStatusFile;
        const int64       savedLastWrite  = gLastPoolStatusWrite;
        remove(statusPath.c_str());
        remove(statusTmp.c_str());
        strPoolStatusFile    = statusPath;
        gLastPoolStatusWrite = 0;   // due a publish on the next refresh

        // Give the status file something to report, so the ledger mutex is
        // genuinely taken on the way through rather than skipped over.
        {
            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
            PendingPayout pp;
            pp.blockHeight = 909; pp.blockHash = 0; pp.matureAtHeight = 909;
            pp.amounts[addrA] = 3 * COIN;
            gPendingPayouts.push_back(pp);
        }

        bool fifoOk = (mkfifo(statusTmp.c_str(), 0600) == 0);
        check(fifoOk, "a FIFO stands in for the status file's temp file, so the"
                      " write can be held open");
        if (fifoOk) {
            std::atomic<bool> done(false);
            std::thread writer([&done]() {
                RefreshStats(0, 0);
                done = true;
            });

            // Settle: the work before the write is a handful of vector
            // operations, so anything past this point is inside fopen(). The
            // !done check below is what actually proves it -- with no reader
            // there is no other way for this thread to still be running.
            Sleep(500);
            bool blocked = !done;
            bool gotLedger = false, gotStats = false, gotExec = false,
                 gotRounds = false;
            for (int i = 0; i < 1000 && blocked; i++) {
                if (!gotLedger && gPayoutLedgerMutex.try_lock()) {
                    gotLedger = true; gPayoutLedgerMutex.unlock();
                }
                if (!gotStats && gStatsMutex.try_lock()) {
                    gotStats = true; gStatsMutex.unlock();
                }
                if (!gotExec && gPayoutExecMutex.try_lock()) {
                    gotExec = true; gPayoutExecMutex.unlock();
                }
                if (!gotRounds && gPoolRoundsMutex.try_lock()) {
                    gotRounds = true; gPoolRoundsMutex.unlock();
                }
                if (gotLedger && gotStats && gotExec && gotRounds)
                    break;
                Sleep(2);
            }

            check(blocked,
                  "the status-file write is held open for the duration of this"
                  " test");
            check(!blocked || gotLedger,
                  "gPayoutLedgerMutex is free while pool_status.json is being"
                  " written");
            check(!blocked || (gotExec && gotRounds),
                  "gPayoutExecMutex and gPoolRoundsMutex are free while"
                  " pool_status.json is being written");
            check(!blocked || gotStats,
                  "gStatsMutex is free while pool_status.json is being written");

            // Let the write finish. Drain to EOF rather than closing early:
            // the writer is still filling the pipe and a closed read end would
            // deliver it SIGPIPE.
            std::thread reader([&statusTmp]() {
                FILE* rf = fopen(statusTmp.c_str(), "r");
                if (!rf) return;
                char buf[4096];
                while (fread(buf, 1, sizeof(buf), rf) > 0) { }
                fclose(rf);
            });
            writer.join();
            reader.join();
        }
        remove(statusTmp.c_str());

        // And the ordinary path still produces the file it always did.
        gLastPoolStatusWrite = 0;
        RefreshStats(0, 0);
        std::string onDisk;
        bool statusOk = false;
        if (PayoutTestReadFile(statusPath, onDisk)) {
            try {
                nlohmann::json j = nlohmann::json::parse(onDisk);
                statusOk = j.value("schema", "") == "bitflash-pool-status-1" &&
                           j["pendingPayouts"].is_array() &&
                           j["pendingPayouts"].size() == 1 &&
                           j["pendingPayouts"][0].value("matureAtHeight", 0) == 909 &&
                           j["pendingPayouts"][0].value("totalAmountSatoshis",
                                                        (int64)0) == 3 * COIN;
            } catch (...) {
                statusOk = false;
            }
        }
        check(statusOk,
              "splitting the serialise from the write leaves pool_status.json"
              " saying the same thing");

        remove(statusPath.c_str());
        remove((statusPath + ".tmp").c_str());
        strPoolStatusFile    = savedStatusFile;
        gLastPoolStatusWrite = savedLastWrite;
    }
#endif

    // -- clean up ------------------------------------------------------------
    PayoutTestResetState();
    CRITICAL_BLOCK(cs_mapWallet)
        mapWallet.clear();
    PayoutTestClearBlockIndex();
    // Out of the scratch directory before removing it, and out of the key maps
    // too: the keys this test generated are process-local, but leaving them in
    // the keystore would let a later test think a script was its own.
    CRITICAL_BLOCK(cs_mapKeys)
    {
        mapKeys.clear();
        mapPubKeys.clear();
    }
#ifdef _WIN32
    SetCurrentDirectoryA(cwd.c_str());
#else
    if (chdir(cwd.c_str()) != 0) { /* nothing useful to do about it */ }
#endif
    PayoutTestRemoveDir(dir);
    strSetDataDir = savedDataDir;
    nBestHeight = savedBestHeight;

    fprintf(stdout, "%s (%d failure%s)\n", nFail == 0 ? "ALL TESTS PASSED" : "TESTS FAILED",
            nFail, nFail == 1 ? "" : "s");
    fflush(stdout);
    return nFail == 0 ? 0 : 1;
}

int RunPoolStratumSelfTest()
{
    int nFail = 0;
    auto check = [&](bool ok, const char* msg) {
        fprintf(stdout, "  %s   %s\n", ok ? "ok" : "FAIL", msg);
        if (!ok) nFail++;
    };

    fprintf(stdout, "pool-stratum self-test\n");

    StratumJob savedJob = gCurrentJob;
    bool savedHaveJob = gHaveJob;

    CTransaction coinbase;
    coinbase.vin.resize(1);
    coinbase.vin[0].prevout.SetNull();
    coinbase.vin[0].scriptSig << 1 << (CBigNum)1;
    coinbase.vout.resize(1);
    coinbase.vout[0].nValue = 50 * COIN;

    CBlock block;
    block.vtx.push_back(coinbase);
    block.nVersion = 1;
    block.nBits = 0x1f0fffff;
    block.nTime = 1;
    block.nNonce = 1;
    block.hashMerkleRoot = block.BuildMerkleTree();

    gCurrentJob.jobId = "selftest";
    gCurrentJob.block = block;
    gCurrentJob.target = 0; // accepted share, not a block candidate
    gCurrentJob.height = 1;
    gHaveJob = true;

    Miner miner(INVALID_SOCKET);
    miner.authorised = true;
    miner.address = "selftest-address";
    miner.worker = "worker";
    miner.shareTarget = ~uint256(0); // every RandomX hash is a valid share
    miner.difficulty = 0.001;
    miner.vardiffWindowStart = GetTime();

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

    uint256 targetFromNotify;
    CBlock minerBlock = BuildMinerBlock(gCurrentJob, miner.extranonce2);
    json params = MakeNotifyParams(gCurrentJob, minerBlock, miner.shareTarget, true);
    std::vector<unsigned char> targetBytes = FromHex(params[3].get<std::string>());
    if (targetBytes.size() == 32)
        memcpy(&targetFromNotify, &targetBytes[0], 32);
    check(targetFromNotify == miner.shareTarget,
          "mining.notify exposes the same share target the pool validates");

    json submit = {
        {"id", 7},
        {"method", "mining.submit"},
        {"params", json::array({miner.address, "selftest", ToHex(&minerBlock.nNonce, 4)})}
    };
    HandleLine(&miner, submit.dump(), roundShareCount, roundShareTotal, blocksFound);

    check(roundShareTotal == 1, "fake low-difficulty share increments the round total");
    check(roundShareCount[miner.address] == 1, "fake share is credited to the submitting address");
    check(miner.sessionShares == 1 && miner.roundShares == 1 && miner.totalShares == 1,
          "miner share counters advance together");
    check(blocksFound == 0, "share below block target is not counted as a found block");

    json stale = {
        {"id", 8},
        {"method", "mining.submit"},
        {"params", json::array({miner.address, "stale", ToHex(&minerBlock.nNonce, 4)})}
    };
    HandleLine(&miner, stale.dump(), roundShareCount, roundShareTotal, blocksFound);
    check(roundShareTotal == 1, "stale submit does not increment shares");

    std::vector<PoolRoundProof> savedRounds = gPoolRounds;
    std::string savedRoundsFile = strPoolRoundsFile;
    std::string proofPath = "pool_rounds_selftest.json";
    remove(proofPath.c_str());
    remove((proofPath + ".tmp").c_str());
    gPoolRounds.clear();
    strPoolRoundsFile = proofPath;

    PoolRoundProof proof;
    proof.id = "selftest-round";
    proof.blockHeight = 42;
    proof.matureAtHeight = 42 + COINBASE_MATURITY;
    proof.createdAt = GetTime();
    proof.blockHash = block.GetHash().GetHex();
    proof.coinbaseTxid = coinbase.GetHash().GetHex();
    proof.foundBy = miner.address;
    proof.feePercent = 1.25;
    proof.totalShares = 3;
    proof.coinbaseValue = 50 * COIN;
    proof.subsidy = 50 * COIN;
    proof.minerReward = 49 * COIN;
    proof.operatorReward = 1 * COIN;
    proof.shares[miner.address] = 3;
    proof.payouts[miner.address] = 49 * COIN;
    RecordPoolRound(proof);
    MarkPoolRoundPaid(proof.id, miner.address, "selftest-txid");

    bool proofOk = false;
    FILE* pf = fopen(proofPath.c_str(), "r");
    if (pf) {
        fseek(pf, 0, SEEK_END);
        long sz = ftell(pf);
        fseek(pf, 0, SEEK_SET);
        std::string s(sz, '\0');
        fread(&s[0], 1, sz, pf);
        fclose(pf);
        try {
            json root = json::parse(s);
            proofOk =
                root.value("schema", "") == "bitflash-pool-rounds-1" &&
                root["rounds"].is_array() &&
                root["rounds"].size() == 1 &&
                root["rounds"][0].value("id", "") == proof.id &&
                root["rounds"][0]["shares"].value(miner.address, (uint64)0) == 3 &&
                root["rounds"][0]["payouts"][miner.address].value("txid", "") == "selftest-txid";
        } catch (...) {
            proofOk = false;
        }
    }
    check(proofOk, "pool round proof ledger is written and paid txid is recorded");

    remove(proofPath.c_str());
    remove((proofPath + ".tmp").c_str());
    strPoolRoundsFile = savedRoundsFile;
    gPoolRounds = savedRounds;

    gCurrentJob = savedJob;
    gHaveJob = savedHaveJob;

    fprintf(stdout, "%s (%d failure%s)\n", nFail == 0 ? "ALL TESTS PASSED" : "TESTS FAILED",
            nFail, nFail == 1 ? "" : "s");
    fflush(stdout);
    return nFail == 0 ? 0 : 1;
}

// ---------------------------------------------------------------------------
// Stats refresh -- called every 2s from event loop
// ---------------------------------------------------------------------------
static void RefreshStats(int blocksFoundThisSession, uint64 roundShareTotal)
{
    std::string statusBlob;
    bool        fPublish = false;
    {
        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;
            });

        // Scoped, not held to the end of the function. At function scope this
        // lock covered the status-file write below, so the event loop did two
        // fsyncs holding the mutex the payout thread takes to record that a
        // miner has been paid.
        gPayoutViews.clear();
        {
            std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);
            for (const PendingPayout& pp : gPendingPayouts) {
                PendingPayoutView pv;
                pv.matureAtHeight = pp.matureAtHeight;
                pv.recipients     = 0;
                pv.totalAmount    = 0;
                for (auto& kv : pp.amounts) {
                    if (pp.paidTxids.count(kv.first))
                        continue;
                    pv.recipients++;
                    pv.totalAmount += kv.second;
                }
                if (pv.recipients > 0)
                    gPayoutViews.push_back(pv);
            }
        }

        int64 now = GetTime();
        if (now - gLastPoolStatusWrite >= 10) {
            statusBlob = SerializePoolStatusLocked();
            gLastPoolStatusWrite = now;
            fPublish = true;
        }
    }
    // Every payout mutex is released here, and gStatsMutex with them. This is
    // where the two fsyncs happen, and no other thread is waiting behind them.
    if (fPublish)
        PublishPoolStatus(statusBlob);
}

// ---------------------------------------------------------------------------
// Event loop
// ---------------------------------------------------------------------------
static void PoolEventLoop()
{
    int   lastHeight            = -1;
    int64 lastRefresh           = GetTime();
    int64 lastStatsRefresh      = 0;
    int64 lastCensus            = 0;   // #19 connection census throttle
    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;
                }
                // #26: select() + FD_SET goes undefined once we exceed what an
                // fd_set can carry, and MAX_POOL_MINERS (1000) alone does not
                // prevent it -- a Miner is only removed on clean disconnect, so
                // half-open zombies accumulate and the OS can hand out fds past
                // the limit long before the count reaches 1000. Refuse rather
                // than corrupt the fd_set. Which limit applies is platform
                // specific: POSIX bounds the descriptor value, Windows the
                // number of SOCKET handles an fd_set can hold.
#ifndef _WIN32
                if ((int)fd >= (int)FD_SETSIZE) {
                    LogPrint("worker", "[pool] refusing miner fd=%d -- at/over"
                             " FD_SETSIZE=%d, select() would be undefined (#26)\n",
                             (int)fd, (int)FD_SETSIZE);
                    sock_close(fd);
                    continue;
                }
#else
                if (gMiners.size() >= (size_t)FD_SETSIZE) {
                    LogPrint("worker", "[pool] refusing miner -- %zu sockets held,"
                             " an fd_set carries at most FD_SETSIZE=%d (#26)\n",
                             gMiners.size(), (int)FD_SETSIZE);
                    sock_close(fd);
                    continue;
                }
#endif
                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
            }
        }

        // 3.6. Hashrate decay (#7) -- see the policy comment at
        // HASHRATE_IDLE_MISSES. The EWMA is only re-sampled on share arrival,
        // so without this pass a miner that stops hashing keeps its last
        // estimate forever and the summed pool hashrate RefreshStats (and
        // the nostr announcement built from it) publishes is fiction.
        for (Miner* m : gMiners) {
            if (m->hashRate <= 0.0 || m->lastShareTime <= 0) continue;
            double expected = ExpectedHashesForDifficulty(m->difficulty) /
                              std::max(m->hashRate, 1e-9);
            if (expected < HASHRATE_MIN_INTERVAL) expected = HASHRATE_MIN_INTERVAL;
            if (expected > HASHRATE_MAX_INTERVAL) expected = HASHRATE_MAX_INTERVAL;
            double silent = (double)(GetTime() - m->lastShareTime);
            double grace  = HASHRATE_IDLE_MISSES * expected;
            if (silent <= grace) continue;
            // Half-life of one expected interval past the grace period.
            // Written off absolute time, not per-tick, so the decay curve is
            // the same however often this loop runs.
            m->hashRate *= pow(0.5, (silent - grace) / expected);
        }

        // 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();
        }

        // 6. #19: log the connection census periodically. Cheap (one pass over
        // gMiners) and it is the single number an operator reads to judge pool
        // health, so it should not be an overstatement.
        if (GetTime() - lastCensus >= POOL_CENSUS_SECS) {
            LogPoolConnectionCensus();
            lastCensus = 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();
    LoadPoolRounds();
    LoadPaying();
    // Must run before the payout thread starts, so nothing is resent that the
    // ledger already shows as sent, and every restored in-flight record is
    // armed for its one wallet probe before it is allowed to send.
    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;
}