upleb.uk

Public git repos — served from a NIP-34 GRASP relay at git.upleb.uk

summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorbitflash-maintainer <noreply@bitflash.invalid>2026-08-23 07:38:39 +0000
committerbitflash-maintainer <noreply@bitflash.invalid>2026-08-23 07:38:39 +0000
commit489be141ee3be1c355650e1429c3f830eb8e442d (patch)
tree45ee1d997d78298addd2308082850e569c63087d /src
parent63740a1b00d78441713a2b909aa40429098184d2 (diff)
payout: one in-flight record per payment, and give it a real identity
A restart duplicated every in-flight payment, deterministically, and the duplicates were load-bearing. gQueuedPayouts is memory-only. ReconcilePayouts restored gPaying from disk but left that set empty, and FlushMaturePayouts deduped only against it and against the ledger's paidTxids -- never against gPaying itself. So a round that had been exploded but not yet paid got a SECOND PayingRecord for every unpaid recipient on the first flush after every restart. Another restart added a third. The crash test saw exactly that: two byte-identical records for one (block, address). ForgetPayment and ReschedulePayment then found their record with SamePayment(), a field-by-field comparison, which stops being an identity the moment two records are equal. Every update landed in the first slot and the second was never touched: - the duplicate's retryAfter stayed 0, so it ran in every 500 ms batch instead of once a minute; - each pass wrote retries = 0+1 into the FIRST slot, resetting its count, so the 1440-retry budget never completed and forfeit was unreachable; - the duplicate's fProbeWallet was never cleared, so FindRecordedPoolPayout -- a full mapWallet scan under cs_mapWallet -- ran twice a second per stuck recipient, forever, along with GetBalance() and, whenever funds allowed, SendMoney/CreateTransaction under cs_main. That is a permanent hot loop holding cs_mapWallet and cs_main on a pool node, starving RebuildJob() so job refresh and block submission stall. It is reached by the ordinary state of a payment waiting for coin or for an unlocked wallet. It did not double-pay only because a slot can reach SendMoney unprobed only when fProbeWallet is already false, and any successful payment calls ForgetPayment, which erases exactly the one slot that could be in that state. That invariant is emergent, undocumented, and dies the moment anyone erases by index or gives the record a stable identity -- which is precisely what this commit does, so it is replaced by a structural one. Both halves are fixed. Not creating duplicates: - ReconcilePayouts rebuilds gQueuedPayouts from the restored gPaying, so the first flush after a restart no longer re-explodes rounds already in flight; - FlushMaturePayouts additionally refuses to append a record for a (round, address) that gPaying already holds. That check is the invariant proper -- the round-level set is an optimisation that can be stale, gPaying is the list being protected; - LoadPaying collapses duplicates already on disk as it reads them, merging the retry and deferral counts upwards, so a file written by the broken build does not inherit the pathology into the fixed one. Real identity: PayingRecord gains an id, assigned under gPayoutExecMutex, and ForgetPayment/ReschedulePayment match on it instead of on values. SamePayment is deleted. The id is process-local and deliberately not persisted -- gPaying is rebuilt from disk at every start, so it only has to name a slot within one run, and reading ids back would mean trusting a file to contain no duplicates. The exactly-once property is now the conjunction of four checks that each hold on their own, rather than one accident: 1. at most one gPaying record per (round, address), enforced at insert; 2. the ledger's paidTxids, re-checked at the top of every attempt; 3. the wallet probe, once per record, before its first send; 4. state updates addressed by record id, so no slot can be skipped. No on-disk format change: the id is not written, and duplicate records in an existing file are read and collapsed.
Diffstat (limited to 'src')
-rw-r--r--src/rpc.cpp118
1 files changed, 97 insertions, 21 deletions
diff --git a/src/rpc.cpp b/src/rpc.cpp
index b6b26f1..7badb7c 100644
--- a/src/rpc.cpp
+++ b/src/rpc.cpp
@@ -680,6 +680,18 @@ static void LoadPendingPayouts()
680// and reconciling against the wallet on startup, is what makes a restart 680// and reconciling against the wallet on startup, is what makes a restart
681// survivable. 681// survivable.
682struct PayingRecord { 682struct PayingRecord {
683 // Identity of this SLOT, distinct from the identity of the payment it
684 // describes. ForgetPayment and ReschedulePayment used to find their record
685 // by comparing every field, which is only an identity while no two records
686 // can be equal -- and a restart used to produce byte-identical duplicates
687 // (see FlushMaturePayouts), so the first match absorbed every update and
688 // the second was never touched again.
689 //
690 // Process-local and deliberately not persisted: gPaying is rebuilt from
691 // disk at every start, so an id only ever has to name a slot within one
692 // run, and reading ids back would mean trusting a file to contain no
693 // duplicates. Assigned under gPayoutExecMutex from gPayingNextId.
694 uint64 id = 0;
683 int blockHeight = 0; 695 int blockHeight = 0;
684 uint256 blockHash = 0; // zero = predates hash keying, see PendingPayout 696 uint256 blockHash = 0; // zero = predates hash keying, see PendingPayout
685 int matureAtHeight = 0; 697 int matureAtHeight = 0;
@@ -702,11 +714,37 @@ struct PayingRecord {
702}; 714};
703static std::vector<PayingRecord> gPaying; // guarded by gPayoutExecMutex 715static std::vector<PayingRecord> gPaying; // guarded by gPayoutExecMutex
704static std::mutex gPayoutExecMutex; 716static std::mutex gPayoutExecMutex;
717static uint64 gPayingNextId = 0; // guarded by gPayoutExecMutex
705static uint64 gPayingSeq = 0; // guarded by gPayoutExecMutex 718static uint64 gPayingSeq = 0; // guarded by gPayoutExecMutex
706static std::mutex gPayingFileMutex; 719static std::mutex gPayingFileMutex;
707static uint64 gPayingPublishedSeq = 0; // guarded by gPayingFileMutex 720static uint64 gPayingPublishedSeq = 0; // guarded by gPayingFileMutex
708static std::atomic<bool> gPayoutThreadRunning{false}; 721static std::atomic<bool> gPayoutThreadRunning{false};
709 722
723// Upstream's gPayoutExecQueue/gPayoutExecMutex/gPayoutThreadRunning are
724// declared above with gPaying: the in-flight list IS the exec queue in this
725// tree, at per-recipient rather than per-round granularity, so a second queue
726// would be a second answer to "who is owed money". Only the dedupe set is new.
727//
728// Ids of rounds currently exploded into gPaying, so a flush cannot queue the
729// same round twice. Guarded by gPayoutExecMutex. Seeded from the restored
730// in-flight list by ReconcilePayouts -- it is memory-only, and leaving it
731// empty at startup was half of why a restart duplicated everything in flight.
732static std::set<std::string> gQueuedPayouts;
733
734// Is this exact payment already in flight? gPaying is keyed on
735// (round, address): the ledger holds one amount per address per round, so two
736// records for that pair are two standing orders to send the same money.
737// Caller must hold gPayoutExecMutex.
738static bool PaymentInFlightLocked(int blockHeight, const uint256& blockHash,
739 const std::string& address)
740{
741 for (const PayingRecord& e : gPaying)
742 if (e.address == address
743 && SamePayoutRound(e.blockHeight, e.blockHash, blockHeight, blockHash))
744 return true;
745 return false;
746}
747
710static std::string PayingFilePath() 748static std::string PayingFilePath()
711{ 749{
712 return GetAppDir() + "/paying_payouts.json"; 750 return GetAppDir() + "/paying_payouts.json";
@@ -774,6 +812,30 @@ static void LoadPaying()
774 r.retries = obj["retries"].get<int>(); 812 r.retries = obj["retries"].get<int>();
775 if (obj.contains("deferrals") && obj["deferrals"].is_number()) 813 if (obj.contains("deferrals") && obj["deferrals"].is_number())
776 r.deferrals = obj["deferrals"].get<int>(); 814 r.deferrals = obj["deferrals"].get<int>();
815 // Collapse duplicates as they are read. A build without the
816 // ReconcilePayouts seeding below wrote a second (and after another
817 // restart, a third) byte-identical record for every unpaid
818 // recipient, so a file written by one of those has to be cleaned
819 // up here or the pathology is inherited. The retry and deferral
820 // counts are merged upwards so collapsing does not hand the
821 // survivor a fresh budget.
822 PayingRecord* dup = NULL;
823 for (PayingRecord& e : gPaying)
824 if (e.address == r.address
825 && SamePayoutRound(e.blockHeight, e.blockHash,
826 r.blockHeight, r.blockHash)) {
827 dup = &e;
828 break;
829 }
830 if (dup) {
831 dup->retries = std::max(dup->retries, r.retries);
832 dup->deferrals = std::max(dup->deferrals, r.deferrals);
833 LogPrint("payout", "[payout] collapsing duplicate in-flight"
834 " record for %s to %s -- one payment, one record\n",
835 FormatMoney(r.amount).c_str(), r.address.c_str());
836 continue;
837 }
838 r.id = ++gPayingNextId;
777 gPaying.push_back(r); 839 gPaying.push_back(r);
778 } 840 }
779 LogPrint("payout", "[payout] loaded %zu in-flight payment(s) from disk\n", 841 LogPrint("payout", "[payout] loaded %zu in-flight payment(s) from disk\n",
@@ -962,6 +1024,15 @@ static std::string PayoutComment(int blockHeight)
962// What is left is the useful part: the in-flight list restored from disk may 1024// What is left is the useful part: the in-flight list restored from disk may
963// contain records the ledger already shows as paid (we died between the send 1025// contain records the ledger already shows as paid (we died between the send
964// and the in-flight file being rewritten). The ledger wins. 1026// and the in-flight file being rewritten). The ledger wins.
1027//
1028// And the part that was missing: gQueuedPayouts is memory-only, so it came up
1029// empty while gPaying came up full. FlushMaturePayouts deduped against that
1030// empty set and against the ledger's paidTxids, and against nothing else --
1031// so a round that was exploded but not yet paid got a SECOND record per
1032// unpaid recipient on the first flush after every restart, deterministically.
1033// (Observed live: two byte-identical records for one (block, address).) The
1034// surviving in-flight list is exactly the statement of which rounds are
1035// already exploded, so it is what the set is rebuilt from.
965static void ReconcilePayouts() 1036static void ReconcilePayouts()
966{ 1037{
967 VersionedBlob payingSnap; 1038 VersionedBlob payingSnap;
@@ -997,6 +1068,13 @@ static void ReconcilePayouts()
997 " already recorded paid in the ledger -- not resending\n", 1068 " already recorded paid in the ledger -- not resending\n",
998 settled, FormatMoney(settledAmount).c_str()); 1069 settled, FormatMoney(settledAmount).c_str());
999 } 1070 }
1071
1072 // Rebuild the "already exploded" set from what survived. Without this the
1073 // first flush after a restart re-explodes every round still in flight.
1074 gQueuedPayouts.clear();
1075 for (const PayingRecord& r : gPaying)
1076 gQueuedPayouts.insert(PayoutIdForBlock(r.blockHeight, r.blockHash));
1077
1000 payingSnap = SerializePayingLocked(); 1078 payingSnap = SerializePayingLocked();
1001 1079
1002 if (!gPaying.empty()) { 1080 if (!gPaying.empty()) {
@@ -1054,15 +1132,6 @@ static void MarkPoolRoundPaid(const std::string& id,
1054 PublishPoolRounds(snap); 1132 PublishPoolRounds(snap);
1055} 1133}
1056 1134
1057// Upstream's gPayoutExecQueue/gPayoutExecMutex/gPayoutThreadRunning are
1058// declared above with gPaying: the in-flight list IS the exec queue in this
1059// tree, at per-recipient rather than per-round granularity, so a second queue
1060// would be a second answer to "who is owed money". Only the dedupe set is new.
1061//
1062// Ids of rounds currently exploded into gPaying, so a flush cannot queue the
1063// same round twice. Guarded by gPayoutExecMutex.
1064static std::set<std::string> gQueuedPayouts;
1065
1066static bool FindRecordedPoolPayout(const std::string& id, 1135static bool FindRecordedPoolPayout(const std::string& id,
1067 const std::string& address, 1136 const std::string& address,
1068 int64 amount, 1137 int64 amount,
@@ -1311,7 +1380,17 @@ static void FlushMaturePayouts()
1311 // txid is done, even though the entry as a whole is not. 1380 // txid is done, even though the entry as a whole is not.
1312 if (pp.paidTxids.count(kv.first)) 1381 if (pp.paidTxids.count(kv.first))
1313 continue; 1382 continue;
1383 // The invariant, checked against the thing it is an invariant of:
1384 // at most one in-flight record per (round, address). The round-level
1385 // set above is an optimisation that can be wrong -- it is memory
1386 // only, it is rebuilt from gPaying at startup, and a stale `ready`
1387 // snapshot can race a round finishing -- whereas gPaying is the
1388 // list being protected. A second record here is a second standing
1389 // order to send the same money.
1390 if (PaymentInFlightLocked(pp.blockHeight, pp.blockHash, kv.first))
1391 continue;
1314 PayingRecord r; 1392 PayingRecord r;
1393 r.id = ++gPayingNextId;
1315 r.blockHeight = pp.blockHeight; 1394 r.blockHeight = pp.blockHeight;
1316 r.blockHash = pp.blockHash; 1395 r.blockHash = pp.blockHash;
1317 r.matureAtHeight = pp.matureAtHeight; 1396 r.matureAtHeight = pp.matureAtHeight;
@@ -1331,27 +1410,24 @@ static void FlushMaturePayouts()
1331 PublishPaying(payingSnap); 1410 PublishPaying(payingSnap);
1332} 1411}
1333 1412
1334// Identity of a payment, for finding its record again after SendMoney.
1335// amounts is a map keyed by address, so (block, address) is already unique.
1336static bool SamePayment(const PayingRecord& a, const PayingRecord& b)
1337{
1338 return SamePayoutRound(a.blockHeight, a.blockHash, b.blockHeight, b.blockHash)
1339 && a.matureAtHeight == b.matureAtHeight
1340 && a.address == b.address
1341 && a.amount == b.amount;
1342}
1343
1344// Settle a payment: it is either done or abandoned, either way it stops being 1413// Settle a payment: it is either done or abandoned, either way it stops being
1345// in flight. Note this only retires the working record -- whether the miner is 1414// in flight. Note this only retires the working record -- whether the miner is
1346// still OWED the money is the ledger's business, and the ledger is updated 1415// still OWED the money is the ledger's business, and the ledger is updated
1347// separately by RecordPayoutTxid(). 1416// separately by RecordPayoutTxid().
1417//
1418// Erases by record id, not by comparing fields. Value-matching was only an
1419// identity for as long as no two records could be equal, and the restart bug
1420// above produced records that were equal in every field: the first match then
1421// absorbed every erase and every state update, and the second was never
1422// touched again. The id names the slot the caller took its copy from, so the
1423// update lands where it belongs whatever else is in the list.
1348static void ForgetPayment(const PayingRecord& r) 1424static void ForgetPayment(const PayingRecord& r)
1349{ 1425{
1350 VersionedBlob snap; 1426 VersionedBlob snap;
1351 { 1427 {
1352 std::lock_guard<std::mutex> lk(gPayoutExecMutex); 1428 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
1353 for (auto it = gPaying.begin(); it != gPaying.end(); ++it) 1429 for (auto it = gPaying.begin(); it != gPaying.end(); ++it)
1354 if (SamePayment(*it, r)) { gPaying.erase(it); break; } 1430 if (it->id == r.id) { gPaying.erase(it); break; }
1355 1431
1356 // When the last record for a round is gone the round is no longer in 1432 // When the last record for a round is gone the round is no longer in
1357 // flight, so a later flush may explode it again if the ledger still 1433 // flight, so a later flush may explode it again if the ledger still
@@ -1379,7 +1455,7 @@ static void ReschedulePayment(const PayingRecord& r)
1379 { 1455 {
1380 std::lock_guard<std::mutex> lk(gPayoutExecMutex); 1456 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
1381 for (auto& e : gPaying) 1457 for (auto& e : gPaying)
1382 if (SamePayment(e, r)) { 1458 if (e.id == r.id) {
1383 e.retries = r.retries; 1459 e.retries = r.retries;
1384 e.deferrals = r.deferrals; 1460 e.deferrals = r.deferrals;
1385 e.retryAfter = r.retryAfter; 1461 e.retryAfter = r.retryAfter;