From 0b01d6cfce8be68150f5e06c11e13e0443372c4d Mon Sep 17 00:00:00 2001 From: Nsable-sh Date: Fri, 7 Aug 2026 11:25:45 -0300 Subject: pool: make payout ledger crash-safe --- src/rpc.cpp | 247 +++++++++++++++++++++++++++++++++++++++++++++++------------- 1 file changed, 196 insertions(+), 51 deletions(-) (limited to 'src') diff --git a/src/rpc.cpp b/src/rpc.cpp index d73b09c..8f3740d 100644 --- a/src/rpc.cpp +++ b/src/rpc.cpp @@ -251,36 +251,72 @@ static bool RebuildJob() // Payouts -- event loop thread only // --------------------------------------------------------------------------- struct PendingPayout { + std::string id; + int blockHeight; int matureAtHeight; std::map amounts; // address -> satoshis + std::map paidTxids; // address -> txid + int64 nextAttempt; + + PendingPayout() : blockHeight(0), matureAtHeight(0), nextAttempt(0) {} }; static std::vector gPendingPayouts; +static std::mutex gPayoutLedgerMutex; static std::string PayoutFilePath() { return GetAppDir() + "/pending_payouts.json"; } -static void SavePendingPayouts() +static std::string PayoutIdForBlock(int blockHeight) +{ + return strprintf("block-%d", blockHeight); +} + +static bool PayoutFullyPaid(const PendingPayout& pp) +{ + for (const auto& kv : pp.amounts) + if (pp.paidTxids.count(kv.first) == 0) + return false; + return true; +} + +static void SavePendingPayoutsLocked() { nlohmann::json arr = nlohmann::json::array(); for (const auto& pp : gPendingPayouts) { nlohmann::json obj; + obj["id"] = pp.id; + obj["blockHeight"] = pp.blockHeight; obj["matureAtHeight"] = pp.matureAtHeight; + obj["nextAttempt"] = pp.nextAttempt; 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); } std::string path = PayoutFilePath(); - FILE* f = fopen(path.c_str(), "w"); + std::string tmp = path + ".tmp"; + FILE* f = fopen(tmp.c_str(), "w"); if (f) { std::string s = arr.dump(2); fwrite(s.c_str(), 1, s.size(), f); - fclose(f); + if (fclose(f) != 0) { + LogPrint("payout", "[payout] WARNING: could not close %s\n", tmp.c_str()); + return; + } +#ifdef _WIN32 + remove(path.c_str()); +#endif + if (rename(tmp.c_str(), path.c_str()) != 0) + LogPrint("payout", "[payout] WARNING: could not replace %s\n", path.c_str()); } else { - LogPrint("payout", "[payout] WARNING: could not write %s\n", path.c_str()); + LogPrint("payout", "[payout] WARNING: could not write %s\n", tmp.c_str()); } } @@ -297,11 +333,25 @@ static void LoadPendingPayouts() fclose(f); try { nlohmann::json arr = nlohmann::json::parse(s); + std::lock_guard lk(gPayoutLedgerMutex); for (const auto& obj : arr) { PendingPayout pp; + if (obj.contains("blockHeight")) + pp.blockHeight = obj["blockHeight"].get(); pp.matureAtHeight = obj["matureAtHeight"].get(); + if (obj.contains("id") && obj["id"].is_string()) + pp.id = obj["id"].get(); + if (pp.blockHeight == 0) + pp.blockHeight = pp.matureAtHeight - COINBASE_MATURITY; + if (pp.id.empty()) + pp.id = PayoutIdForBlock(pp.blockHeight); + if (obj.contains("nextAttempt")) + pp.nextAttempt = obj["nextAttempt"].get(); for (auto it = obj["amounts"].begin(); it != obj["amounts"].end(); ++it) pp.amounts[it.key()] = it.value().get(); + 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(); gPendingPayouts.push_back(pp); } LogPrint("payout", "[payout] loaded %zu pending payout(s) from disk\n", @@ -319,6 +369,35 @@ static std::vector gPayoutExecQueue; static std::mutex gPayoutExecMutex; static std::atomic gPayoutThreadRunning{false}; +static std::set gQueuedPayouts; + +static bool FindRecordedPoolPayout(const std::string& id, + const std::string& address, + int64 amount, + std::string& txidRet) +{ + CRITICAL_BLOCK(cs_mapWallet) + { + for (map::const_iterator it = mapWallet.begin(); + it != mapWallet.end(); ++it) + { + const CWalletTx& wtx = it->second; + map::const_iterator idIt = wtx.mapValue.find("pool_payout_id"); + if (idIt == wtx.mapValue.end() || idIt->second != id) + continue; + map::const_iterator addrIt = wtx.mapValue.find("pool_payout_address"); + if (addrIt == wtx.mapValue.end() || addrIt->second != address) + continue; + map::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 true; + } + } + return false; +} + static void QueuePayouts(int blockHeight, const std::map& shareCount, uint64 total) @@ -341,6 +420,8 @@ static void QueuePayouts(int blockHeight, int64 minerReward = (int64)((double)nSubsidy * (1.0 - feeFrac)); PendingPayout pp; + pp.id = PayoutIdForBlock(blockHeight); + pp.blockHeight = blockHeight; pp.matureAtHeight = blockHeight + COINBASE_MATURITY; int skipped = 0; @@ -364,8 +445,11 @@ static void QueuePayouts(int blockHeight, pp.amounts[kv.first] = amount; } - gPendingPayouts.push_back(pp); - SavePendingPayouts(); + { + std::lock_guard lk(gPayoutLedgerMutex); + gPendingPayouts.push_back(pp); + SavePendingPayoutsLocked(); + } int64 operatorCut = nSubsidy - minerReward; LogPrint("payout", @@ -383,42 +467,33 @@ static void QueuePayouts(int blockHeight, : ""); } -// Called from event loop: move matured entries to the exec queue (fast). -// NOT crash-safe, despite the ordering below looking like it is. -// -// A matured payout is erased from gPendingPayouts, the file is rewritten -// without it, and only then does it reach the exec queue -- which lives in -// memory alone. SavePendingPayouts writes gPendingPayouts and nothing else; -// there is no exec-queue section of the file to recover from. Die anywhere -// between the rewrite and SendMoney and those miners are simply never paid, -// with no record that they were owed anything. -// -// Making this safe needs the payment to be idempotent -- record the txid and -// check the wallet for it on restart -- rather than a reordering, since -// re-queueing blindly would double-pay whoever was already sent to. +// Called from event loop: copy matured entries to the exec queue (fast). +// The ledger entry stays on disk until every recipient has a txid recorded. +// On restart, FindRecordedPoolPayout() checks wallet metadata before sending, +// so a crash after SendMoney but before the ledger update does not double-pay. static void FlushMaturePayouts() { std::vector ready; - for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ) { - if (nBestHeight < it->matureAtHeight) { ++it; continue; } - LogPrint("payout", "[payout] block height %d matured -- queuing %zu payments\n", - it->matureAtHeight, it->amounts.size()); - ready.push_back(*it); // copy first; erase after save - ++it; - } - if (!ready.empty()) { - // Erase from pending then save before handing off to exec thread, - // so the file never contains an entry that is already being sent. - for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ) { - if (nBestHeight >= it->matureAtHeight) - it = gPendingPayouts.erase(it); - else - ++it; + { + std::lock_guard ledger(gPayoutLedgerMutex); + std::lock_guard queue(gPayoutExecMutex); + int64 now = GetTime(); + for (const PendingPayout& pp : gPendingPayouts) { + if (nBestHeight < pp.matureAtHeight) + continue; + if (pp.nextAttempt > now) + continue; + if (PayoutFullyPaid(pp)) + continue; + if (gQueuedPayouts.count(pp.id)) + continue; + LogPrint("payout", "[payout] block height %d matured -- queuing payout %s (%zu recipients)\n", + pp.matureAtHeight, pp.id.c_str(), pp.amounts.size()); + ready.push_back(pp); + gQueuedPayouts.insert(pp.id); } - SavePendingPayouts(); - std::lock_guard lk(gPayoutExecMutex); for (auto& pp : ready) - gPayoutExecQueue.push_back(std::move(pp)); + gPayoutExecQueue.push_back(pp); } } @@ -437,10 +512,21 @@ static void PayoutThreadFn(void*) batch.swap(gPayoutExecQueue); } - for (auto& pp : batch) { + for (auto& ppQueued : batch) { LogPrint("payout", "[payout] executing payments for block matured at %d" - " (%zu miners)\n", pp.matureAtHeight, pp.amounts.size()); - for (auto& kv : pp.amounts) { + " (%zu miners)\n", ppQueued.matureAtHeight, ppQueued.amounts.size()); + bool fAbortBatch = false; + for (auto& kv : ppQueued.amounts) { + bool alreadyPaid = false; + { + std::lock_guard lk(gPayoutLedgerMutex); + for (const PendingPayout& pp : gPendingPayouts) + if (pp.id == ppQueued.id && pp.paidTxids.count(kv.first)) + alreadyPaid = true; + } + if (alreadyPaid) + continue; + uint160 h160; if (!AddressToHash160(kv.first, h160)) { // Should not happen -- bad addresses are filtered in QueuePayouts @@ -451,16 +537,70 @@ static void PayoutThreadFn(void*) CScript sc; sc << OP_DUP << OP_HASH160 << h160 << OP_EQUALVERIFY << OP_CHECKSIG; CWalletTx wtx; - wtx.mapValue["comment"] = strprintf("Pool payout height %d", - pp.matureAtHeight); - if (SendMoney(sc, kv.second, wtx)) + wtx.mapValue["comment"] = strprintf("Pool payout height %d", ppQueued.matureAtHeight); + wtx.mapValue["pool_payout_id"] = ppQueued.id; + wtx.mapValue["pool_payout_address"] = kv.first; + wtx.mapValue["pool_payout_amount"] = strprintf("%lld", kv.second); + + std::string txid; + if (FindRecordedPoolPayout(ppQueued.id, kv.first, kv.second, txid)) + { + LogPrint("payout", "[payout] recovered recorded payout %s to %s" + " from wallet tx %s\n", + FormatMoney(kv.second).c_str(), kv.first.c_str(), txid.c_str()); + } + else if (SendMoney(sc, kv.second, wtx)) + { + txid = wtx.GetHash().GetHex(); LogPrint("payout", "[payout] paid %s to %s\n", - FormatMoney(kv.second).c_str(), kv.first.c_str()); + FormatMoney(kv.second).c_str(), kv.first.c_str()); + } else + { LogPrint("payout", "[payout] FAILED to pay %s to %s" - " (insufficient funds or wallet locked?)\n", - FormatMoney(kv.second).c_str(), kv.first.c_str()); + " (insufficient funds or wallet locked?)\n", + FormatMoney(kv.second).c_str(), kv.first.c_str()); + std::lock_guard lk(gPayoutLedgerMutex); + for (PendingPayout& pp : gPendingPayouts) + if (pp.id == ppQueued.id) + pp.nextAttempt = GetTime() + 60; + SavePendingPayoutsLocked(); + fAbortBatch = true; + break; + } + + { + std::lock_guard lk(gPayoutLedgerMutex); + for (PendingPayout& pp : gPendingPayouts) { + if (pp.id != ppQueued.id) + continue; + pp.paidTxids[kv.first] = txid; + pp.nextAttempt = 0; + SavePendingPayoutsLocked(); + break; + } + } } + + { + std::lock_guard ledger(gPayoutLedgerMutex); + for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ) { + if (it->id == ppQueued.id && PayoutFullyPaid(*it)) { + LogPrint("payout", "[payout] payout %s complete -- removing from ledger\n", + it->id.c_str()); + it = gPendingPayouts.erase(it); + SavePendingPayoutsLocked(); + } else { + ++it; + } + } + } + { + std::lock_guard lk(gPayoutExecMutex); + gQueuedPayouts.erase(ppQueued.id); + } + if (fAbortBatch) + continue; } Sleep(500); // check for new work every 500ms @@ -1073,16 +1213,21 @@ static void RefreshStats(int blocksFoundThisSession, uint64 roundShareTotal) : a.lastSeen > b.lastSeen; }); - // gPendingPayouts is event-loop-only data (no lock needed here since - // RefreshStats is always called from the event loop thread). gPayoutViews.clear(); + std::lock_guard ledger(gPayoutLedgerMutex); for (const PendingPayout& pp : gPendingPayouts) { PendingPayoutView pv; pv.matureAtHeight = pp.matureAtHeight; - pv.recipients = (int)pp.amounts.size(); + pv.recipients = 0; pv.totalAmount = 0; - for (auto& kv : pp.amounts) pv.totalAmount += kv.second; - gPayoutViews.push_back(pv); + 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); } } -- cgit v1.2.3