upleb.uk

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

summaryrefslogtreecommitdiff
path: root/src/rpc.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/rpc.cpp')
-rw-r--r--src/rpc.cpp3444
1 files changed, 3175 insertions, 269 deletions
diff --git a/src/rpc.cpp b/src/rpc.cpp
index 813f1d1..7a6b6a4 100644
--- a/src/rpc.cpp
+++ b/src/rpc.cpp
@@ -6,8 +6,8 @@
6// Architecture: 6// Architecture:
7// PoolEventLoop -- single thread, owns all miner fds, all job/share/payout 7// PoolEventLoop -- single thread, owns all miner fds, all job/share/payout
8// state. No locks needed on any of that data. 8// state. No locks needed on any of that data.
9// AcceptOneFn -- one thread per relay. Blocks on RvServiceRegister + 9// AcceptOneFn -- one thread per relay. Registers, waits to be paired,
10// BtfServiceWrap, pushes finished socket to gReadyQueue. 10// wraps the tunnel, pushes finished socket to gReadyQueue.
11// AcceptLoopFn -- spawns AcceptOneFn threads, re-polls relay list every 11// AcceptLoopFn -- spawns AcceptOneFn threads, re-polls relay list every
12// 60s so dynamically discovered relays get picked up. 12// 60s so dynamically discovered relays get picked up.
13// 13//
@@ -31,10 +31,15 @@
31#include <netinet/in.h> 31#include <netinet/in.h>
32#include <arpa/inet.h> 32#include <arpa/inet.h>
33#include <unistd.h> 33#include <unistd.h>
34#include <fcntl.h>
35#include <dirent.h>
36#else
37#include <io.h>
34#endif 38#endif
35 39
36#include <atomic> 40#include <atomic>
37#include <mutex> 41#include <mutex>
42#include <thread>
38#include <map> 43#include <map>
39#include <set> 44#include <set>
40#include <deque> 45#include <deque>
@@ -143,6 +148,75 @@ static bool RebuildJob()
143 return false; 148 return false;
144 } 149 }
145 150
151 // Don't issue jobs while we are behind the network: mining against a
152 // stale tip produces blocks the network will orphan, and payout
153 // entries that can never be paid (their coinbases never exist on the
154 // main chain). Require at least one peer, and require that no peer
155 // claims a longer chain than ours.
156 if (vNodes.empty()) {
157 LogPrint("pool", "RebuildJob: no peers yet -- not issuing jobs\n");
158 return false;
159 }
160 // This used to compare the tip's timestamp against 30 minutes, which
161 // was wrong in both directions and cost real money.
162 //
163 // It cannot tell "we are not synced" from "the network is slow". A
164 // pool mining its own fork keeps its tip timestamp fresh forever, so
165 // the check passed for the whole of heights 3966-4085 -- 120
166 // consecutive blocks, every one of them orphaned, 5890.50 BTF of
167 // payouts voided. Then, once the pool stopped finding blocks, the
168 // same check refused to issue jobs against a tip that was perfectly
169 // valid and was the network's own best, and the pool -- roughly 90%
170 // of the hashrate -- could not restart the chain it was the majority
171 // of. Stale tip, so no jobs; no jobs, so the tip stays stale.
172 //
173 // The height peers announce in their version message answers the
174 // question the age check was only guessing at. Median, so one peer
175 // claiming an absurd height cannot stall us, and one block of slack
176 // to match WarnIfBehind, because somebody is always mid-relay. A
177 // peer that predates the announcement reports -1 and is ignored; if
178 // nobody has told us anything we mine our best chain, which is what
179 // the node would do without a pool attached.
180 int nPeerHeight = GetPeerMedianHeight();
181 if (nPeerHeight >= 0 && nBestHeight < nPeerHeight - 1) {
182 LogPrint("pool", "RebuildJob: %d blocks behind the network "
183 "(height %d, peers report %d) -- not issuing jobs\n",
184 nPeerHeight - nBestHeight, nBestHeight, nPeerHeight);
185 return false;
186 }
187
188 // The comparison above is blind in the one case that matters most.
189 //
190 // When the node goes deaf -- peers still in vNodes, nothing arriving --
191 // no peer announces anything, GetPeerMedianHeight() returns -1, and the
192 // fallback ("nobody has told us anything, so mine our best chain") let
193 // the pool build a private fork for as long as the deafness lasted.
194 // That fallback is right for a node that has simply not been told yet
195 // and wrong for a node that has stopped being able to hear, and a
196 // height nobody sent cannot tell those apart. Time can.
197 //
198 // Deliberately not behind a debug category, and deliberately not a
199 // silent refusal: an operator whose pool has stopped issuing jobs needs
200 // the reason in the log. /poolmineblind is the escape hatch for a node
201 // that really is alone -- without one, a genuinely isolated pool that is
202 // the majority of the hashrate could never restart its own chain, which
203 // is the deadlock the tip-age check used to cause.
204 if (nPeerHeight < 0 && !fPoolMineBlind) {
205 int64 nAge = GetPeerHeightAgeSecs();
206 if (nAge >= BTF_NO_PEER_HEIGHT_MAX_SECS) {
207 static int64 nLastSaid = 0;
208 if (GetTime() - nLastSaid > 60) {
209 nLastSaid = GetTime();
210 printf("RebuildJob: no peer has reported a height for %d minutes "
211 "-- not issuing jobs (our height %d, %d peers connected). "
212 "This node is deaf; mining now would fork it. "
213 "Pass /poolmineblind to override.\n",
214 (int)(nAge / 60), nBestHeight, (int)vNodes.size());
215 }
216 return false;
217 }
218 }
219
146 CBlockIndex* pindexPrev = pindexBest; 220 CBlockIndex* pindexPrev = pindexBest;
147 unsigned int nBits = GetNextWorkRequired(pindexPrev); 221 unsigned int nBits = GetNextWorkRequired(pindexPrev);
148 222
@@ -194,7 +268,15 @@ static bool RebuildJob()
194 vector<char> vfAlreadyAdded(mapTransactions.size()); 268 vector<char> vfAlreadyAdded(mapTransactions.size());
195 bool fFoundSomething = true; 269 bool fFoundSomething = true;
196 unsigned int sz = 0; 270 unsigned int sz = 0;
197 while (fFoundSomething && sz < MAX_SIZE / 2) 271 // The cap is MAX_BLOCK_SIZE/2, not MAX_SIZE/2. MAX_SIZE is the
272 // 32 MB serialisation ceiling, so half of it is 16 MB -- this
273 // builder would happily assemble a template that its own
274 // CheckBlock() rejects at the 1 MB consensus cap. This is the
275 // pool's job builder, so a block over the cap is not one
276 // wasted solve, it is every miner on the pool working a
277 // template the network will reject. The inner break bounds the
278 // overshoot to a single transaction within a pass.
279 while (fFoundSomething && sz < MAX_BLOCK_SIZE / 2)
198 { 280 {
199 fFoundSomething = false; 281 fFoundSomething = false;
200 unsigned int n = 0; 282 unsigned int n = 0;
@@ -214,6 +296,7 @@ static bool RebuildJob()
214 sz += ::GetSerializeSize(tx, SER_NETWORK); 296 sz += ::GetSerializeSize(tx, SER_NETWORK);
215 vfAlreadyAdded[n] = true; 297 vfAlreadyAdded[n] = true;
216 fFoundSomething = true; 298 fFoundSomething = true;
299 if (sz > MAX_BLOCK_SIZE / 2) break;
217 } 300 }
218 } 301 }
219 } 302 }
@@ -251,17 +334,45 @@ static bool RebuildJob()
251// Payouts -- event loop thread only 334// Payouts -- event loop thread only
252// --------------------------------------------------------------------------- 335// ---------------------------------------------------------------------------
253struct PendingPayout { 336struct PendingPayout {
254 std::string id; 337 // The block that funds this payout, identified by hash and not by height
255 int blockHeight; 338 // alone. A reorg replaces the block at a height without changing the
256 int matureAtHeight; 339 // height, so a height-keyed obligation outlives the coinbase that was
340 // supposed to pay it: a 280-block reorg once left ~14.4k BTF of debt
341 // owed against blocks that no longer existed, which then drained the
342 // wallet paying fork-era miners out of real-chain coin.
343 //
344 // Upstream keys these on the string "block-<height>". That is the same
345 // bug this fork already paid for, and height keying additionally lets
346 // FindRecordedPoolPayout match the *previous* round's transaction after a
347 // reorg and silently skip a miner, so their `id` is deliberately dropped.
348 int blockHeight = 0;
349 // Must be explicitly zeroed: uint256's default constructor is empty
350 // (uint256.h:514), so leaving this unassigned gives every ledger entry
351 // loaded from an older file a garbage hash, which then matches no block
352 // and gets voided as "orphaned". That mistake voided a live payout.
353 uint256 blockHash = 0; // zero = entry predates hash keying
354 int matureAtHeight = 0;
257 std::map<std::string, int64> amounts; // address -> satoshis 355 std::map<std::string, int64> amounts; // address -> satoshis
356 // From upstream: which recipients already have a transaction. This is what
357 // lets an entry stay on disk until every address in `amounts` is paid,
358 // instead of being erased the moment the batch is handed off.
258 std::map<std::string, std::string> paidTxids; // address -> txid 359 std::map<std::string, std::string> paidTxids; // address -> txid
259 int64 nextAttempt; 360 // Retry and backoff state does NOT live here. It used to be declared here
260 361 // as well as on PayingRecord, and the copy on this side was never written:
261 PendingPayout() : blockHeight(0), matureAtHeight(0), nextAttempt(0) {} 362 // `retryAfter` was declared, serialised, loaded and tested at the flush
363 // gate, and the only assignment to it anywhere was `= 0`, so the gate could
364 // not fire. `retries` and `deferrals` were the same story. The real state
365 // is per (round, recipient) -- which is the granularity a payment can fail
366 // at -- and it lives on PayingRecord, which is now what persists it.
262}; 367};
263static std::vector<PendingPayout> gPendingPayouts; 368static std::vector<PendingPayout> gPendingPayouts;
264static std::mutex gPayoutLedgerMutex; 369static std::mutex gPayoutLedgerMutex;
370// Bumped on every serialisation, under gPayoutLedgerMutex. See PublishVersioned.
371static uint64 gPayoutLedgerSeq = 0;
372// Serialises writers of pending_payouts.json, and guards the sequence number
373// of the image currently on disk. Never taken while a data mutex is held.
374static std::mutex gPayoutLedgerFileMutex;
375static uint64 gPayoutLedgerPublishedSeq = 0;
265 376
266struct PoolRoundProof { 377struct PoolRoundProof {
267 std::string id; 378 std::string id;
@@ -282,6 +393,12 @@ struct PoolRoundProof {
282 std::map<std::string, uint64> shares; 393 std::map<std::string, uint64> shares;
283 std::map<std::string, int64> payouts; 394 std::map<std::string, int64> payouts;
284 std::map<std::string, std::string> paidTxids; 395 std::map<std::string, std::string> paidTxids;
396 // Recipients the ledger has written off, and why. Without this the public
397 // round proof goes on advertising an address as owed with "txid": null
398 // forever, while the ledger has already dropped the obligation -- the two
399 // records of the same round permanently disagreeing, with the operator
400 // holding the money.
401 std::map<std::string, std::string> forfeited; // address -> reason
285 402
286 PoolRoundProof() 403 PoolRoundProof()
287 : blockHeight(0), matureAtHeight(0), createdAt(0), feePercent(0.0), 404 : blockHeight(0), matureAtHeight(0), createdAt(0), feePercent(0.0),
@@ -290,6 +407,9 @@ struct PoolRoundProof {
290}; 407};
291static std::vector<PoolRoundProof> gPoolRounds; 408static std::vector<PoolRoundProof> gPoolRounds;
292static std::mutex gPoolRoundsMutex; 409static std::mutex gPoolRoundsMutex;
410static uint64 gPoolRoundsSeq = 0; // guarded by gPoolRoundsMutex
411static std::mutex gPoolRoundsFileMutex;
412static uint64 gPoolRoundsPublishedSeq = 0; // guarded by gPoolRoundsFileMutex
293static const size_t MAX_POOL_ROUNDS = 200; 413static const size_t MAX_POOL_ROUNDS = 200;
294 414
295static std::string PayoutFilePath() 415static std::string PayoutFilePath()
@@ -304,9 +424,302 @@ static std::string PoolRoundsFilePath()
304 return GetAppDir() + "/pool_rounds.json"; 424 return GetAppDir() + "/pool_rounds.json";
305} 425}
306 426
307static std::string PayoutIdForBlock(int blockHeight) 427// ---------------------------------------------------------------------------
428// Atomic, ordered publication of the payout state files
429// ---------------------------------------------------------------------------
430//
431// Every one of these files is a full dump of an in-memory structure, written
432// tmp-then-rename so a crash cannot truncate it. Two things have to be true
433// for that to actually be safe, and only the first was:
434//
435// 1. the rename must be atomic (it is), and
436// 2. exactly one thread at a time may be writing the temp file, and a write
437// must never publish an OLDER image over a newer one.
438//
439// (2) was broken for pending_payouts.json. The event loop wrote it with
440// gPayoutLedgerMutex held, the payout thread wrote it with the mutex
441// released, and nothing serialised the two: both opened the same
442// "<path>.tmp" with independent handles and both renamed it. The interleaved
443// file that lands is either a mixture of two images (LoadPendingPayouts then
444// throws mid-loop and comes up with a nearly empty ledger -- every obligation
445// not yet exploded, gone) or a whole newly-queued round silently overwritten
446// by an older snapshot that predates it.
447//
448// The fix is a per-file write mutex, held only across the I/O and never
449// across a data-structure mutex, plus a version on the snapshot. Serialising
450// happens under the data mutex and stamps a monotonically increasing sequence
451// number; publishing happens with the data mutex released and refuses any
452// blob whose sequence number is not newer than what is already on disk. So
453// writers can queue in any order and finish in any order -- the file only
454// ever moves forwards.
455//
456// The other half of the rule is that no thread waits for I/O while holding
457// gPayoutLedgerMutex, gPayoutExecMutex or gPoolRoundsMutex, which is the
458// contention the merge was right to remove from the event loop. That was
459// claimed here before it was true: pool_status.json is written by the same
460// WriteFileAtomic and RefreshStats held gPayoutLedgerMutex at function scope
461// across the call, so once the writes gained fsync() the event loop was doing
462// two of them every ten seconds with the payout thread's mutex held -- worse
463// than the buffered write the merge replaced. The status file now goes through
464// the same serialise-under-the-lock / publish-outside-it split as the ledgers,
465// and the rule holds for all four files. It is checked, not asserted: see the
466// "no payout mutex is held across a publish" section of the payout-ledger
467// self-test.
468
469// A serialised file image plus the version of the in-memory state it was
470// taken from.
471struct VersionedBlob {
472 std::string blob;
473 uint64 seq = 0;
474};
475
476// Push a file's contents out of the C library's buffer AND out of the kernel's
477// page cache, so the bytes exist on the device before anything points at them.
478static bool SyncFileHandle(FILE* f)
479{
480 if (fflush(f) != 0)
481 return false;
482#ifdef _WIN32
483 return _commit(_fileno(f)) == 0;
484#else
485 return fsync(fileno(f)) == 0;
486#endif
487}
488
489// Make the directory entry that rename() created durable too.
490//
491// A tmp+rename is only atomic with respect to a crash of the process. Against
492// a power cut or a hard VM reset the rename can reach the platter while the
493// data it points at has not, and the file that comes back is short or empty.
494// For pending_payouts.json that is not a corrupt file, it is a silent one:
495// LoadPendingPayouts catches the parse error, logs a single warning, and the
496// node comes up owing nobody anything.
497//
498// Best effort -- a filesystem that does not allow the directory to be opened
499// for this is not a reason to refuse to write the ledger.
500static void SyncDirectoryOf(const std::string& path)
501{
502#ifndef _WIN32
503 size_t slash = path.find_last_of('/');
504 std::string dir = (slash == std::string::npos) ? std::string(".")
505 : path.substr(0, slash);
506 if (dir.empty())
507 dir = "/";
508 int fd = open(dir.c_str(), O_RDONLY);
509 if (fd < 0)
510 return;
511 fsync(fd);
512 close(fd);
513#endif
514}
515
516// Replace `path` with `data` atomically and durably. Caller must hold the
517// write mutex for `path`: the temp name is derived from the target, so two
518// threads in here at once for one path is precisely the race this exists to
519// prevent.
520//
521// Every failure removes the temp file before returning. It used to be left
522// where it fell, so a disk that filled up once left a short, valid-looking
523// "pending_payouts.json.tmp" next to the good ledger for as long as the node
524// ran. That file is a trap: it has the name of a half-finished save, and an
525// operator cleaning up after an incident is exactly the person who will move
526// it into place.
527//
528// Failures print unconditionally rather than through LogPrint(). These three
529// files are written under two different debug categories ("payout" for the
530// ledgers, "pool" for the status file), and a write failure is not a trace of
531// normal operation -- it is the file the operator is about to read not being
532// there. Gating it behind a category that had to be chosen before the failure
533// happened is how "could not replace pool_status.json" went missing.
534static bool WriteFileAtomic(const std::string& path, const std::string& data,
535 const char* what)
536{
537 std::string tmp = path + ".tmp";
538 FILE* f = fopen(tmp.c_str(), "w");
539 if (!f) {
540 printf("[payout] WARNING: could not write %s (%s)\n",
541 tmp.c_str(), what);
542 return false;
543 }
544 if (!data.empty() && fwrite(data.c_str(), 1, data.size(), f) != data.size()) {
545 printf("[payout] WARNING: short write to %s (%s)\n",
546 tmp.c_str(), what);
547 fclose(f);
548 remove(tmp.c_str());
549 return false;
550 }
551 // Before the rename, not after: publishing a name that points at nothing
552 // is the failure this ordering exists to prevent. If the data cannot be
553 // made durable, do not rename -- the previous file is still intact and
554 // still correct, and the next write will try again.
555 if (!SyncFileHandle(f)) {
556 printf("[payout] WARNING: could not flush %s to disk (%s)"
557 " -- keeping the previous file\n", tmp.c_str(), what);
558 fclose(f);
559 remove(tmp.c_str());
560 return false;
561 }
562 // fclose() is where buffered data actually reaches the OS, so a failure
563 // here means the .tmp is short -- renaming it over the real file would
564 // destroy the good copy, which is exactly what these files exist to
565 // prevent.
566 if (fclose(f) != 0) {
567 printf("[payout] WARNING: could not close %s (%s)\n",
568 tmp.c_str(), what);
569 remove(tmp.c_str());
570 return false;
571 }
572#ifdef _WIN32
573 // rename() on Windows will not replace an existing file.
574 remove(path.c_str());
575#endif
576 if (rename(tmp.c_str(), path.c_str()) != 0) {
577 printf("[payout] WARNING: could not replace %s (%s)\n",
578 path.c_str(), what);
579 remove(tmp.c_str());
580 return false;
581 }
582 SyncDirectoryOf(path);
583 return true;
584}
585
586// Paths whose loader threw part-way through, and which must therefore never be
587// written again by this process.
588//
589// All three loaders push entries into the live structure INSIDE the try, so an
590// exception half way down a file leaves a PARTIAL ledger in memory -- the
591// entries before the bad one, and nothing after it. The first publish after
592// that then writes the partial state over the file, and the only record of
593// what was owed is gone. ReconcilePayouts publishes unconditionally at
594// startup, so "the first publish" is a few milliseconds later, before an
595// operator could possibly intervene. a8e2f58 named this failure mode in its
596// own commit message and then only fixed the write side.
597//
598// The file is renamed out of the way instead of being left in place, so the
599// evidence survives even if the operator restarts into the same fault, and so
600// a node that comes up owing nobody anything is visibly missing its ledger
601// rather than quietly appearing to have none.
602static std::set<std::string> gRefusedFiles;
603static std::mutex gRefusedFilesMutex;
604
605static bool FileIsRefused(const std::string& path)
606{
607 std::lock_guard<std::mutex> lk(gRefusedFilesMutex);
608 return gRefusedFiles.count(path) > 0;
609}
610
611// Quarantine `path` and refuse to write it for the rest of the process.
612//
613// Prints unconditionally: an operator who has just lost a payout ledger must
614// be told at whatever log settings they were already running with, and the
615// name it was moved to is the only way back to the data.
616static void RefuseFileAfterLoadFailure(const std::string& path, const char* what,
617 const char* why)
618{
619 {
620 std::lock_guard<std::mutex> lk(gRefusedFilesMutex);
621 gRefusedFiles.insert(path);
622 }
623 std::string bad = strprintf("%s.bad.%lld", path.c_str(), (long long)GetTime());
624 bool moved = (rename(path.c_str(), bad.c_str()) == 0);
625 printf("[payout] ERROR: %s (%s) did not parse: %s\n"
626 "[payout] ERROR: what loaded before the error is a PARTIAL picture,"
627 " so this file will not be written again by this process --"
628 " restart once the cause is fixed\n"
629 "[payout] ERROR: the file %s\n",
630 path.c_str(), what, why,
631 moved ? strprintf("has been kept as %s", bad.c_str()).c_str()
632 : "could NOT be renamed out of the way -- copy it now");
633}
634
635// Publish a versioned image of some in-memory state to `path`.
636//
637// `fileMutex` serialises writers of `path` and guards `publishedSeq`. A blob
638// older than what is already published is dropped: these files are whole-state
639// dumps, so a higher sequence number is by construction a later state of the
640// same structure and publishing it loses nothing. On a write failure
641// `publishedSeq` is deliberately left alone, so the next (newer) write still
642// gets a chance to land.
643static void PublishVersioned(const std::string& path, const VersionedBlob& v,
644 std::mutex& fileMutex, uint64& publishedSeq,
645 const char* what)
646{
647 // A default-constructed blob is not a snapshot of anything; it is a
648 // VersionedBlob nobody filled in. Today they are already dropped, because
649 // every serialiser stamps ++gSeq and so never produces 0, and 0 fails the
650 // seq <= publishedSeq test below -- but only for as long as publishedSeq
651 // starts at 0 and nobody writes `seq = gSeq++`. One character, and the
652 // ledger file becomes "[]". Say it out loud instead of relying on it.
653 if (!v.seq)
654 return;
655 // A file whose loader threw is not a file this process knows the contents
656 // of. Writing it would replace a full record of what was owed with
657 // whatever happened to parse before the error.
658 if (FileIsRefused(path))
659 return;
660 std::lock_guard<std::mutex> lk(fileMutex);
661 if (v.seq <= publishedSeq)
662 return;
663 if (!WriteFileAtomic(path, v.blob, what))
664 return;
665 publishedSeq = v.seq;
666}
667
668// Identity of a payout round. Used as the pool-round proof id and, more
669// importantly, as the `pool_payout_id` written into wallet transaction
670// metadata and matched by FindRecordedPoolPayout().
671//
672// Derived from the block HASH, not from its height alone. Upstream uses
673// "block-<height>", which names two different blocks identically across a
674// reorg -- and because this id is what the crash-recovery probe matches on, a
675// height-keyed id lets that probe find the *previous* round's transaction and
676// conclude a miner has already been paid when they have not. The height stays
677// in the string for readability only.
678//
679// A zero hash means the entry predates hash keying; those keep the old form so
680// anything already recorded against them still matches.
681static std::string PayoutIdForBlock(int blockHeight, const uint256& blockHash)
682{
683 if (blockHash == 0)
684 return strprintf("block-%d", blockHeight);
685 return strprintf("block-%d-%s", blockHeight,
686 blockHash.ToString().substr(0, 16).c_str());
687}
688
689// Do these two references name the same payout round?
690//
691// The block hash alone does not decide it, and every ledger lookup used to
692// assume it did. Entries grandfathered in from a file written before payouts
693// were keyed by hash all carry hash 0, so under a hash-only comparison the
694// first zero-hash entry answers for every other one: RecordPayoutTxid writes
695// a txid against the wrong round (which is then erased as "fully paid" while
696// the real debt still stands), and ReconcilePayouts drops a live in-flight
697// record because a *different* zero-hash round happened to pay that address.
698//
699// A non-zero hash is unique and decides on its own -- that is the whole point
700// of keying on it. A zero hash falls back to the height, which
701// LoadPendingPayouts supplies for any entry that can name its own maturity.
702// Mixed hash-ness never matches: refusing to match is the safe direction for
703// every caller here, because each of them treats a match as "this debt is
704// already dealt with".
705static bool SamePayoutRound(int aHeight, const uint256& aHash,
706 int bHeight, const uint256& bHash)
707{
708 if (aHash != bHash)
709 return false;
710 if (aHash != 0)
711 return true;
712 return aHeight == bHeight;
713}
714
715// An entry with neither a hash nor a height cannot be told apart from any
716// other entry in the same state, so nothing may be recorded against it: a
717// txid written to the wrong one reports a debt as settled that is still owed.
718// Such an entry is never paid and is logged instead -- refusing to act is the
719// safe failure, and no file this tree has ever written can produce one.
720static bool PayoutRoundIdentifiable(int height, const uint256& hash)
308{ 721{
309 return strprintf("block-%d", blockHeight); 722 return hash != 0 || height != 0;
310} 723}
311 724
312static bool PayoutFullyPaid(const PendingPayout& pp) 725static bool PayoutFullyPaid(const PendingPayout& pp)
@@ -317,15 +730,28 @@ static bool PayoutFullyPaid(const PendingPayout& pp)
317 return true; 730 return true;
318} 731}
319 732
320static void SavePendingPayoutsLocked() 733// Serialise the ledger and stamp it with the version it was taken at. Caller
734// must hold gPayoutLedgerMutex.
735//
736// Split from the file write so no thread holds the ledger mutex across the
737// I/O. Upstream writes the file inside the per-recipient loop with
738// gPayoutLedgerMutex held, while the event loop's FlushMaturePayouts blocks on
739// that same mutex -- one slow fsync per recipient, on the thread the miners'
740// I/O shares. The sequence number is what lets the write happen later, and
741// possibly out of order, without a stale image reaching the disk.
742static VersionedBlob SerializePendingPayoutsLocked()
321{ 743{
322 nlohmann::json arr = nlohmann::json::array(); 744 nlohmann::json arr = nlohmann::json::array();
323 for (const auto& pp : gPendingPayouts) { 745 for (const auto& pp : gPendingPayouts) {
324 nlohmann::json obj; 746 nlohmann::json obj;
325 obj["id"] = pp.id; 747 obj["blockHeight"] = pp.blockHeight;
326 obj["blockHeight"] = pp.blockHeight; 748 obj["blockHash"] = pp.blockHash.ToString();
327 obj["matureAtHeight"] = pp.matureAtHeight; 749 obj["matureAtHeight"] = pp.matureAtHeight;
328 obj["nextAttempt"] = pp.nextAttempt; 750 // No retryAfter: it was never written with anything but zero, and the
751 // retry clock that is actually used is on PayingRecord. Files that
752 // still carry the key load fine (it is ignored), and a build that
753 // expects it defaults an absent one to 0, which means "ready now" --
754 // the same answer it always got.
329 nlohmann::json amounts = nlohmann::json::object(); 755 nlohmann::json amounts = nlohmann::json::object();
330 for (const auto& kv : pp.amounts) 756 for (const auto& kv : pp.amounts)
331 amounts[kv.first] = kv.second; 757 amounts[kv.first] = kv.second;
@@ -336,24 +762,18 @@ static void SavePendingPayoutsLocked()
336 obj["paidTxids"] = paid; 762 obj["paidTxids"] = paid;
337 arr.push_back(obj); 763 arr.push_back(obj);
338 } 764 }
339 std::string path = PayoutFilePath(); 765 VersionedBlob v;
340 std::string tmp = path + ".tmp"; 766 v.blob = arr.dump(2);
341 FILE* f = fopen(tmp.c_str(), "w"); 767 v.seq = ++gPayoutLedgerSeq;
342 if (f) { 768 return v;
343 std::string s = arr.dump(2); 769}
344 fwrite(s.c_str(), 1, s.size(), f); 770
345 if (fclose(f) != 0) { 771// Publish a ledger snapshot. Holds no data mutex; safe -- and intended -- to
346 LogPrint("payout", "[payout] WARNING: could not close %s\n", tmp.c_str()); 772// call with gPayoutLedgerMutex released.
347 return; 773static void PublishPendingPayouts(const VersionedBlob& v)
348 } 774{
349#ifdef _WIN32 775 PublishVersioned(PayoutFilePath(), v, gPayoutLedgerFileMutex,
350 remove(path.c_str()); 776 gPayoutLedgerPublishedSeq, "pending payout ledger");
351#endif
352 if (rename(tmp.c_str(), path.c_str()) != 0)
353 LogPrint("payout", "[payout] WARNING: could not replace %s\n", path.c_str());
354 } else {
355 LogPrint("payout", "[payout] WARNING: could not write %s\n", tmp.c_str());
356 }
357} 777}
358 778
359static void LoadPendingPayouts() 779static void LoadPendingPayouts()
@@ -372,17 +792,26 @@ static void LoadPendingPayouts()
372 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex); 792 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
373 for (const auto& obj : arr) { 793 for (const auto& obj : arr) {
374 PendingPayout pp; 794 PendingPayout pp;
375 if (obj.contains("blockHeight"))
376 pp.blockHeight = obj["blockHeight"].get<int>();
377 pp.matureAtHeight = obj["matureAtHeight"].get<int>(); 795 pp.matureAtHeight = obj["matureAtHeight"].get<int>();
378 if (obj.contains("id") && obj["id"].is_string()) 796 // Entries written before payouts were keyed by hash carry neither
379 pp.id = obj["id"].get<std::string>(); 797 // field. Leave blockHash zero and they are grandfathered past the
380 if (pp.blockHeight == 0) 798 // orphan check rather than voided on a guess.
799 if (obj.contains("blockHash") && obj["blockHash"].is_string())
800 pp.blockHash.SetHex(obj["blockHash"].get<std::string>());
801 if (obj.contains("blockHeight") && obj["blockHeight"].is_number())
802 pp.blockHeight = obj["blockHeight"].get<int>();
803 // Upstream's tolerance for an entry with no usable height, kept
804 // because it costs nothing and an entry that cannot name its block
805 // is worse than one derived from its maturity. COINBASE_MATURITY
806 // and not COINBASE_SPENDABLE_DEPTH: this reverses whatever the
807 // writer of *that* file used, and only files old enough to lack
808 // blockHeight are affected -- they were written when maturity was
809 // still COINBASE_MATURITY.
810 if (pp.blockHeight == 0 && pp.matureAtHeight > COINBASE_MATURITY)
381 pp.blockHeight = pp.matureAtHeight - COINBASE_MATURITY; 811 pp.blockHeight = pp.matureAtHeight - COINBASE_MATURITY;
382 if (pp.id.empty()) 812 // "retryAfter" is deliberately not read. Files written by the
383 pp.id = PayoutIdForBlock(pp.blockHeight); 813 // merged build carry it, always as 0, and there is nowhere left to
384 if (obj.contains("nextAttempt")) 814 // put it -- the retry clock is PayingRecord's.
385 pp.nextAttempt = obj["nextAttempt"].get<int64>();
386 for (auto it = obj["amounts"].begin(); it != obj["amounts"].end(); ++it) 815 for (auto it = obj["amounts"].begin(); it != obj["amounts"].end(); ++it)
387 pp.amounts[it.key()] = it.value().get<int64>(); 816 pp.amounts[it.key()] = it.value().get<int64>();
388 if (obj.contains("paidTxids") && obj["paidTxids"].is_object()) 817 if (obj.contains("paidTxids") && obj["paidTxids"].is_object())
@@ -393,11 +822,211 @@ static void LoadPendingPayouts()
393 LogPrint("payout", "[payout] loaded %zu pending payout(s) from disk\n", 822 LogPrint("payout", "[payout] loaded %zu pending payout(s) from disk\n",
394 gPendingPayouts.size()); 823 gPendingPayouts.size());
395 } catch (const std::exception& e) { 824 } catch (const std::exception& e) {
396 LogPrint("payout", "[payout] WARNING: failed to parse %s: %s\n", 825 RefuseFileAfterLoadFailure(path, "pending payout ledger", e.what());
397 path.c_str(), e.what()); 826 }
827}
828
829// A payment in flight: one record per (payout entry, address) pair, which is
830// the granularity SendMoney works at and therefore the granularity at which a
831// payment can be half-done when the process dies.
832//
833// This list is persisted. It used to be a memory-only queue, and matured
834// entries were erased from pending_payouts.json *before* being handed to it,
835// so any restart in between lost those payments with no record that anyone
836// was owed anything: a container restart on 2026-07-30 silently dropped
837// ~1,156 BTF that way. Writing the records before removing the ledger entry,
838// and reconciling against the wallet on startup, is what makes a restart
839// survivable.
840struct PayingRecord {
841 // Identity of this SLOT, distinct from the identity of the payment it
842 // describes. ForgetPayment and ReschedulePayment used to find their record
843 // by comparing every field, which is only an identity while no two records
844 // can be equal -- and a restart used to produce byte-identical duplicates
845 // (see FlushMaturePayouts), so the first match absorbed every update and
846 // the second was never touched again.
847 //
848 // Process-local and deliberately not persisted: gPaying is rebuilt from
849 // disk at every start, so an id only ever has to name a slot within one
850 // run, and reading ids back would mean trusting a file to contain no
851 // duplicates. Assigned under gPayoutExecMutex from gPayingNextId.
852 uint64 id = 0;
853 int blockHeight = 0;
854 uint256 blockHash = 0; // zero = predates hash keying, see PendingPayout
855 int matureAtHeight = 0;
856 std::string address;
857 int64 amount = 0;
858 int retries = 0;
859 int deferrals = 0;
860 int64 retryAfter = 0;
861 // Ask the wallet whether this payment already went out, once, before the
862 // first send attempt. Upstream asks on every attempt: FindRecordedPoolPayout
863 // is a linear scan of mapWallet under cs_mapWallet, per recipient, so with
864 // 2000 rotating payout addresses and a retry budget of 1440 that is a
865 // scan per address per minute for a day. The question it answers -- "did a
866 // previous process send this before it died?" -- can only change across a
867 // restart, and every restart re-sets this flag (ReconcilePayouts), so
868 // asking once per record is the whole of the answer.
869 //
870 // Not persisted: a record read back from disk is by definition a restart.
871 bool fProbeWallet = true;
872};
873static std::vector<PayingRecord> gPaying; // guarded by gPayoutExecMutex
874static std::mutex gPayoutExecMutex;
875static uint64 gPayingNextId = 0; // guarded by gPayoutExecMutex
876static uint64 gPayingSeq = 0; // guarded by gPayoutExecMutex
877static std::mutex gPayingFileMutex;
878static uint64 gPayingPublishedSeq = 0; // guarded by gPayingFileMutex
879static std::atomic<bool> gPayoutThreadRunning{false};
880
881// Upstream's gPayoutExecQueue/gPayoutExecMutex/gPayoutThreadRunning are
882// declared above with gPaying: the in-flight list IS the exec queue in this
883// tree, at per-recipient rather than per-round granularity, so a second queue
884// would be a second answer to "who is owed money". Only the dedupe set is new.
885//
886// Ids of rounds currently exploded into gPaying, so a flush cannot queue the
887// same round twice. Guarded by gPayoutExecMutex. Seeded from the restored
888// in-flight list by ReconcilePayouts -- it is memory-only, and leaving it
889// empty at startup was half of why a restart duplicated everything in flight.
890static std::set<std::string> gQueuedPayouts;
891
892// Is this exact payment already in flight? gPaying is keyed on
893// (round, address): the ledger holds one amount per address per round, so two
894// records for that pair are two standing orders to send the same money.
895// Caller must hold gPayoutExecMutex.
896static bool PaymentInFlightLocked(int blockHeight, const uint256& blockHash,
897 const std::string& address)
898{
899 for (const PayingRecord& e : gPaying)
900 if (e.address == address
901 && SamePayoutRound(e.blockHeight, e.blockHash, blockHeight, blockHash))
902 return true;
903 return false;
904}
905
906static std::string PayingFilePath()
907{
908 return GetAppDir() + "/paying_payouts.json";
909}
910
911// Serialise the in-flight list. Caller must hold gPayoutExecMutex.
912//
913// gPayoutExecMutex already serialises every writer of this file, so it never
914// had the interleaving race pending_payouts.json had. It is still split the
915// same way for a different reason: FlushMaturePayouts runs on the event loop
916// and holds gPayoutExecMutex, so writing (and, since this series, fsyncing)
917// the file under that mutex parks miner I/O behind the payout thread's disk.
918// The version stamp keeps the file monotonic now that the write is outside
919// the mutex that orders the writers.
920static VersionedBlob SerializePayingLocked()
921{
922 nlohmann::json arr = nlohmann::json::array();
923 for (const auto& r : gPaying) {
924 nlohmann::json obj;
925 obj["blockHeight"] = r.blockHeight;
926 obj["blockHash"] = r.blockHash.ToString();
927 obj["matureAtHeight"] = r.matureAtHeight;
928 obj["address"] = r.address;
929 obj["amount"] = r.amount;
930 obj["retries"] = r.retries;
931 obj["deferrals"] = r.deferrals;
932 // The live retry clock, which used not to be written at all: every
933 // restart handed every backing-off payment a retryAfter of 0 and put
934 // the whole set into the very next 500 ms batch, so a node restarted
935 // in a loop retried continuously instead of once a minute.
936 obj["retryAfter"] = r.retryAfter;
937 arr.push_back(obj);
938 }
939 VersionedBlob v;
940 v.blob = arr.dump(2);
941 v.seq = ++gPayingSeq;
942 return v;
943}
944
945// Publish an in-flight-list snapshot. Call with gPayoutExecMutex released.
946static void PublishPaying(const VersionedBlob& v)
947{
948 PublishVersioned(PayingFilePath(), v, gPayingFileMutex,
949 gPayingPublishedSeq, "in-flight payment list");
950}
951
952static void LoadPaying()
953{
954 std::string path = PayingFilePath();
955 FILE* f = fopen(path.c_str(), "r");
956 if (!f) return;
957 fseek(f, 0, SEEK_END);
958 long sz = ftell(f);
959 fseek(f, 0, SEEK_SET);
960 std::string s(sz, '\0');
961 fread(&s[0], 1, sz, f);
962 fclose(f);
963 try {
964 nlohmann::json arr = nlohmann::json::parse(s);
965 // gPaying and gPayingNextId are both annotated as guarded by
966 // gPayoutExecMutex and this was mutating them without it. Safe today
967 // -- LoadPaying only runs at startup, before the payout thread is
968 // spawned -- but "safe because of where it is called from" is a fact
969 // about the caller, not about this function, and the annotation is
970 // what every other reader will trust. The mutex is uncontended here by
971 // construction, so taking it costs nothing.
972 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
973 for (const auto& obj : arr) {
974 PayingRecord r;
975 r.address = obj["address"].get<std::string>();
976 r.amount = obj["amount"].get<int64>();
977 r.matureAtHeight = obj["matureAtHeight"].get<int>();
978 if (obj.contains("blockHash") && obj["blockHash"].is_string())
979 r.blockHash.SetHex(obj["blockHash"].get<std::string>());
980 if (obj.contains("blockHeight") && obj["blockHeight"].is_number())
981 r.blockHeight = obj["blockHeight"].get<int>();
982 if (obj.contains("retries") && obj["retries"].is_number())
983 r.retries = obj["retries"].get<int>();
984 if (obj.contains("deferrals") && obj["deferrals"].is_number())
985 r.deferrals = obj["deferrals"].get<int>();
986 // Absent in files written before this change; 0 means ready now,
987 // which is exactly the behaviour those files used to get.
988 if (obj.contains("retryAfter") && obj["retryAfter"].is_number())
989 r.retryAfter = obj["retryAfter"].get<int64>();
990 // Collapse duplicates as they are read. A build without the
991 // ReconcilePayouts seeding below wrote a second (and after another
992 // restart, a third) byte-identical record for every unpaid
993 // recipient, so a file written by one of those has to be cleaned
994 // up here or the pathology is inherited. The retry and deferral
995 // counts are merged upwards so collapsing does not hand the
996 // survivor a fresh budget.
997 PayingRecord* dup = NULL;
998 for (PayingRecord& e : gPaying)
999 if (e.address == r.address
1000 && SamePayoutRound(e.blockHeight, e.blockHash,
1001 r.blockHeight, r.blockHash)) {
1002 dup = &e;
1003 break;
1004 }
1005 if (dup) {
1006 dup->retries = std::max(dup->retries, r.retries);
1007 dup->deferrals = std::max(dup->deferrals, r.deferrals);
1008 dup->retryAfter = std::max(dup->retryAfter, r.retryAfter);
1009 LogPrint("payout", "[payout] collapsing duplicate in-flight"
1010 " record for %s to %s -- one payment, one record\n",
1011 FormatMoney(r.amount).c_str(), r.address.c_str());
1012 continue;
1013 }
1014 r.id = ++gPayingNextId;
1015 gPaying.push_back(r);
1016 }
1017 LogPrint("payout", "[payout] loaded %zu in-flight payment(s) from disk\n",
1018 gPaying.size());
1019 } catch (const std::exception& e) {
1020 RefuseFileAfterLoadFailure(path, "in-flight payment list", e.what());
398 } 1021 }
399} 1022}
400 1023
1024// ---------------------------------------------------------------------------
1025// Pool round proofs (upstream cbdd06d). Disjoint from the in-flight ledger
1026// above -- this is the operator's published record of how a round was split,
1027// not an obligation to pay. Both sides added code at this offset; both are
1028// kept.
1029// ---------------------------------------------------------------------------
401static nlohmann::json PoolRoundToJson(const PoolRoundProof& r) 1030static nlohmann::json PoolRoundToJson(const PoolRoundProof& r)
402{ 1031{
403 nlohmann::json j; 1032 nlohmann::json j;
@@ -436,12 +1065,26 @@ static nlohmann::json PoolRoundToJson(const PoolRoundProof& r)
436 p["txid"] = tx->second; 1065 p["txid"] = tx->second;
437 else 1066 else
438 p["txid"] = nullptr; 1067 p["txid"] = nullptr;
1068 // Only emitted when it happened, so the common shape of the file is
1069 // unchanged for anything already parsing it.
1070 std::map<std::string, std::string>::const_iterator fo = r.forfeited.find(kv.first);
1071 if (fo != r.forfeited.end()) {
1072 p["forfeited"] = true;
1073 p["forfeitReason"] = fo->second;
1074 }
439 j["payouts"][kv.first] = p; 1075 j["payouts"][kv.first] = p;
440 } 1076 }
441 return j; 1077 return j;
442} 1078}
443 1079
444static void SavePoolRoundsLocked() 1080// Serialise the published round proofs. Caller must hold gPoolRoundsMutex.
1081//
1082// Like the in-flight list, gPoolRoundsMutex already serialises the writers
1083// (RecordPoolRound on the event loop, MarkPoolRoundPaid on the payout thread),
1084// so this file never had the interleaving race either -- but the event loop
1085// should not be waiting on the payout thread's fsync to publish a round, so
1086// the write moves outside the mutex on the same versioned terms.
1087static VersionedBlob SerializePoolRoundsLocked()
445{ 1088{
446 nlohmann::json root; 1089 nlohmann::json root;
447 root["schema"] = "bitflash-pool-rounds-1"; 1090 root["schema"] = "bitflash-pool-rounds-1";
@@ -450,24 +1093,17 @@ static void SavePoolRoundsLocked()
450 for (const PoolRoundProof& r : gPoolRounds) 1093 for (const PoolRoundProof& r : gPoolRounds)
451 root["rounds"].push_back(PoolRoundToJson(r)); 1094 root["rounds"].push_back(PoolRoundToJson(r));
452 1095
453 std::string path = PoolRoundsFilePath(); 1096 VersionedBlob v;
454 std::string tmp = path + ".tmp"; 1097 v.blob = root.dump(2);
455 FILE* f = fopen(tmp.c_str(), "w"); 1098 v.seq = ++gPoolRoundsSeq;
456 if (!f) { 1099 return v;
457 LogPrint("payout", "[payout] WARNING: could not write %s\n", tmp.c_str()); 1100}
458 return; 1101
459 } 1102// Publish a round-proof snapshot. Call with gPoolRoundsMutex released.
460 std::string s = root.dump(2); 1103static void PublishPoolRounds(const VersionedBlob& v)
461 fwrite(s.c_str(), 1, s.size(), f); 1104{
462 if (fclose(f) != 0) { 1105 PublishVersioned(PoolRoundsFilePath(), v, gPoolRoundsFileMutex,
463 LogPrint("payout", "[payout] WARNING: could not close %s\n", tmp.c_str()); 1106 gPoolRoundsPublishedSeq, "pool round proofs");
464 return;
465 }
466#ifdef _WIN32
467 remove(path.c_str());
468#endif
469 if (rename(tmp.c_str(), path.c_str()) != 0)
470 LogPrint("payout", "[payout] WARNING: could not replace %s\n", path.c_str());
471} 1107}
472 1108
473static void LoadPoolRounds() 1109static void LoadPoolRounds()
@@ -512,6 +1148,9 @@ static void LoadPoolRounds()
512 r.payouts[it.key()] = it.value().value("amountSatoshis", (int64)0); 1148 r.payouts[it.key()] = it.value().value("amountSatoshis", (int64)0);
513 if (it.value().contains("txid") && it.value()["txid"].is_string()) 1149 if (it.value().contains("txid") && it.value()["txid"].is_string())
514 r.paidTxids[it.key()] = it.value()["txid"].get<std::string>(); 1150 r.paidTxids[it.key()] = it.value()["txid"].get<std::string>();
1151 if (it.value().value("forfeited", false))
1152 r.forfeited[it.key()] =
1153 it.value().value("forfeitReason", std::string("unspecified"));
515 } else if (it.value().is_number_integer()) { 1154 } else if (it.value().is_number_integer()) {
516 r.payouts[it.key()] = it.value().get<int64>(); 1155 r.payouts[it.key()] = it.value().get<int64>();
517 } 1156 }
@@ -524,55 +1163,283 @@ static void LoadPoolRounds()
524 LogPrint("payout", "[payout] loaded %zu pool round proof(s) from disk\n", 1163 LogPrint("payout", "[payout] loaded %zu pool round proof(s) from disk\n",
525 gPoolRounds.size()); 1164 gPoolRounds.size());
526 } catch (const std::exception& e) { 1165 } catch (const std::exception& e) {
527 LogPrint("payout", "[payout] WARNING: failed to parse %s: %s\n", 1166 RefuseFileAfterLoadFailure(path, "pool round proofs", e.what());
528 path.c_str(), e.what());
529 } 1167 }
530} 1168}
531 1169
1170// Is the block that funds this payout still on the main chain?
1171//
1172// A zero hash means the entry was written before payouts recorded one, so
1173// there is nothing to check and we assume it is good -- voiding those would
1174// mean refusing to pay miners on the strength of a missing field.
1175static bool PayoutBlockStillOnChain(const uint256& blockHash)
1176{
1177 if (blockHash == 0)
1178 return true;
1179 CRITICAL_BLOCK(cs_main)
1180 {
1181 std::map<uint256, CBlockIndex*>::iterator mi = mapBlockIndex.find(blockHash);
1182 if (mi == mapBlockIndex.end() || mi->second == NULL)
1183 return false;
1184 return mi->second->IsInMainChain();
1185 }
1186 return false;
1187}
1188
1189static std::string PayoutComment(int blockHeight)
1190{
1191 return strprintf("Pool payout height %d", blockHeight);
1192}
1193
1194// Startup reconciliation.
1195//
1196// This used to be the crash-recovery mechanism: it scanned the wallet with
1197// WalletAlreadyPaid() and then dropped ledger entries that had already been
1198// exploded into in-flight records. Neither job is right any more.
1199//
1200// The ledger is now authoritative and an entry survives until every recipient
1201// has a txid recorded against it (upstream cbdd06d), so dropping ledger
1202// entries here would reintroduce exactly the loss this merge is adopting that
1203// model to prevent. And the wallet scan is now FindRecordedPoolPayout(), done
1204// once per record on its first attempt, which is the same question asked in
1205// the one place that can act on the answer. Running both would be two
1206// authorities for the same debt.
1207//
1208// What is left is the useful part: the in-flight list restored from disk may
1209// contain records the ledger already shows as paid (we died between the send
1210// and the in-flight file being rewritten). The ledger wins.
1211//
1212// And the part that was missing: gQueuedPayouts is memory-only, so it came up
1213// empty while gPaying came up full. FlushMaturePayouts deduped against that
1214// empty set and against the ledger's paidTxids, and against nothing else --
1215// so a round that was exploded but not yet paid got a SECOND record per
1216// unpaid recipient on the first flush after every restart, deterministically.
1217// (Observed live: two byte-identical records for one (block, address).) The
1218// surviving in-flight list is exactly the statement of which rounds are
1219// already exploded, so it is what the set is rebuilt from.
1220static void ReconcilePayouts()
1221{
1222 VersionedBlob payingSnap;
1223 {
1224 std::lock_guard<std::mutex> exec(gPayoutExecMutex);
1225 std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);
1226
1227 size_t settled = 0;
1228 int64 settledAmount = 0;
1229 for (auto it = gPaying.begin(); it != gPaying.end(); ) {
1230 bool paid = false;
1231 for (const PendingPayout& pp : gPendingPayouts)
1232 if (SamePayoutRound(pp.blockHeight, pp.blockHash,
1233 it->blockHeight, it->blockHash)
1234 && pp.paidTxids.count(it->address)) {
1235 paid = true;
1236 break;
1237 }
1238 if (paid) {
1239 settledAmount += it->amount;
1240 settled++;
1241 it = gPaying.erase(it);
1242 } else {
1243 // Every restored record gets one wallet probe before it is
1244 // allowed to send: this is precisely the window where the
1245 // wallet may hold a transaction the ledger has not recorded.
1246 it->fProbeWallet = true;
1247 ++it;
1248 }
1249 }
1250 if (settled) {
1251 LogPrint("payout", "[payout] %zu in-flight payment(s) totalling %s"
1252 " are already recorded paid in the ledger -- not"
1253 " resending\n",
1254 settled, FormatMoney(settledAmount).c_str());
1255 }
1256
1257 // Rebuild the "already exploded" set from what survived. Without this
1258 // the first flush after a restart re-explodes every round in flight.
1259 gQueuedPayouts.clear();
1260 for (const PayingRecord& r : gPaying)
1261 gQueuedPayouts.insert(PayoutIdForBlock(r.blockHeight, r.blockHash));
1262
1263 payingSnap = SerializePayingLocked();
1264
1265 if (!gPaying.empty()) {
1266 int64 owed = 0;
1267 for (const auto& r : gPaying) owed += r.amount;
1268 LogPrint("payout", "[payout] resuming %zu in-flight payment(s)"
1269 " totalling %s\n",
1270 gPaying.size(), FormatMoney(owed).c_str());
1271 }
1272 }
1273 PublishPaying(payingSnap);
1274}
1275
532static void RecordPoolRound(const PoolRoundProof& round) 1276static void RecordPoolRound(const PoolRoundProof& round)
533{ 1277{
534 std::lock_guard<std::mutex> lk(gPoolRoundsMutex); 1278 VersionedBlob snap;
535 for (PoolRoundProof& existing : gPoolRounds) { 1279 {
536 if (existing.id == round.id) { 1280 std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
537 existing = round; 1281 bool replaced = false;
538 SavePoolRoundsLocked(); 1282 for (PoolRoundProof& existing : gPoolRounds) {
539 return; 1283 if (existing.id == round.id) {
1284 existing = round;
1285 replaced = true;
1286 break;
1287 }
540 } 1288 }
1289 if (!replaced) {
1290 gPoolRounds.push_back(round);
1291 while (gPoolRounds.size() > MAX_POOL_ROUNDS)
1292 gPoolRounds.erase(gPoolRounds.begin());
1293 }
1294 snap = SerializePoolRoundsLocked();
541 } 1295 }
542 gPoolRounds.push_back(round); 1296 PublishPoolRounds(snap);
543 while (gPoolRounds.size() > MAX_POOL_ROUNDS)
544 gPoolRounds.erase(gPoolRounds.begin());
545 SavePoolRoundsLocked();
546} 1297}
547 1298
548static void MarkPoolRoundPaid(const std::string& id, 1299static void MarkPoolRoundPaid(const std::string& id,
549 const std::string& address, 1300 const std::string& address,
550 const std::string& txid) 1301 const std::string& txid)
551{ 1302{
552 std::lock_guard<std::mutex> lk(gPoolRoundsMutex); 1303 VersionedBlob snap;
553 for (PoolRoundProof& r : gPoolRounds) { 1304 {
554 if (r.id != id) 1305 std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
555 continue; 1306 bool found = false;
556 r.paidTxids[address] = txid; 1307 for (PoolRoundProof& r : gPoolRounds) {
557 SavePoolRoundsLocked(); 1308 if (r.id != id)
558 return; 1309 continue;
1310 r.paidTxids[address] = txid;
1311 found = true;
1312 break;
1313 }
1314 if (!found)
1315 return;
1316 snap = SerializePoolRoundsLocked();
559 } 1317 }
1318 PublishPoolRounds(snap);
560} 1319}
561 1320
562// Payouts ready to execute are moved here under gPayoutExecMutex. 1321// Publish the fact that a recipient was written off, so the round proof and
563// A dedicated thread drains this list so SendMoney (which takes cs_main) 1322// the ledger agree. The ledger erases the amount (it is not owed any more and
564// never runs on the event loop thread and can't stall miner I/O. 1323// must not be reported as sent); the proof keeps the amount visible and marks
565static std::vector<PendingPayout> gPayoutExecQueue; 1324// it forfeited with a reason, because a miner who was promised a share is
566static std::mutex gPayoutExecMutex; 1325// entitled to see what happened to it.
567static std::atomic<bool> gPayoutThreadRunning{false}; 1326static void MarkPoolRoundForfeited(const std::string& id,
1327 const std::string& address,
1328 const std::string& reason)
1329{
1330 VersionedBlob snap;
1331 {
1332 std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
1333 bool found = false;
1334 for (PoolRoundProof& r : gPoolRounds) {
1335 if (r.id != id)
1336 continue;
1337 r.forfeited[address] = reason;
1338 found = true;
1339 break;
1340 }
1341 if (!found)
1342 return;
1343 snap = SerializePoolRoundsLocked();
1344 }
1345 PublishPoolRounds(snap);
1346}
568 1347
569static std::set<std::string> gQueuedPayouts; 1348// Which wallet transaction, if any, already paid this?
1349//
1350// Two kinds of evidence, in order of quality.
1351//
1352// Structured (pool_payout_id / pool_payout_address / pool_payout_amount) is
1353// upstream's, and it is exact: the id is derived from the block hash, so it
1354// names one round and one round only. Every transaction this build sends
1355// carries it.
1356//
1357// But NO transaction this pool has ever sent carries it. Every payout in the
1358// existing wallet has only a "comment". So on the first boot after this
1359// upgrade -- precisely the boot where the binary is swapped after an unclean
1360// shutdown with payments in flight -- a structured-only probe finds nothing
1361// and every one of those recipients is paid a second time. The deleted
1362// WalletAlreadyPaid() matched on what the payment WAS, and that evidence is
1363// still there; it just has to be read.
1364//
1365// The legacy fallback is therefore restored, with three constraints that
1366// together stop it answering for a round that is not this one:
1367//
1368// 1. it is only consulted when the structured lookup found nothing;
1369// 2. it ignores any transaction carrying pool_payout_id. That key means this
1370// build wrote the transaction, and this build's transactions are
1371// adjudicated by the exact path -- so a NEW round's payment can never be
1372// offered as evidence for an old one. Only pre-upgrade transactions,
1373// which is exactly the population the fallback exists for, are eligible;
1374// 3. `excluded` holds every txid already credited to some OTHER
1375// (round, address) pair -- from the ledger's paidTxids, which is durable,
1376// and from what this process has already claimed. One legacy transaction
1377// can therefore settle at most one debt, so N transactions can never
1378// satisfy more than N obligations.
1379//
1380// What is left unresolved by construction is two pre-upgrade rounds at the
1381// SAME height paying the same address the same amount, since the comment
1382// carries only the height. That needs a reorg.
1383//
1384// This used to say the losing twin's ledger entry is voided by the orphan
1385// check before it can be paid. That is wrong: PayoutBlockStillOnChain()
1386// returns true for a zero hash, and EVERY pre-upgrade entry has a zero hash,
1387// so the orphan check is a no-op for exactly the population this fallback
1388// serves.
1389//
1390// The outcome is still safe, by a different mechanism. SamePayoutRound()
1391// treats two zero-hash entries at the same height as the SAME round, so the
1392// pair is indistinguishable everywhere, not only here: FlushMaturePayouts
1393// explodes one of them and skips the other as already queued,
1394// PaymentInFlightLocked refuses a second record for that (round, address), and
1395// RecordPayoutTxid settles against the first match. The pair therefore
1396// presents as ONE debt and can be paid at most once -- which is the direction
1397// that matters for a fallback whose failure mode is treating one payment as
1398// proof of two.
1399//
1400// The cost of that mechanism, stated plainly: only one of a genuine
1401// same-height pair would ever be paid. It takes a reorg AND both rounds
1402// predating hash keying, and no file this tree has written since can produce
1403// one, so it is a property of history rather than a live risk.
1404//
1405// Where the evidence genuinely runs out the fallback simply does not match,
1406// and the payment is made -- paying a debt we cannot prove was settled is the
1407// correct default for a pool.
1408enum PayoutProbeResult {
1409 PAYOUT_PROBE_NONE = 0, // the wallet holds no record of this payment
1410 PAYOUT_PROBE_STRUCTURED, // matched this build's pool_payout_* metadata
1411 PAYOUT_PROBE_LEGACY // matched a pre-upgrade comment and output
1412};
570 1413
571static bool FindRecordedPoolPayout(const std::string& id, 1414static PayoutProbeResult FindRecordedPoolPayout(const std::string& id,
572 const std::string& address, 1415 int blockHeight,
573 int64 amount, 1416 const std::string& address,
574 std::string& txidRet) 1417 int64 amount,
1418 const std::set<std::string>& excluded,
1419 std::string& txidRet)
575{ 1420{
1421 // Built before cs_mapWallet is taken: an invalid address has no script to
1422 // compare against, and the caller has already rejected those.
1423 uint160 h160;
1424 bool fHaveScript = AddressToHash160(address, h160);
1425 CScript want;
1426 if (fHaveScript)
1427 want << OP_DUP << OP_HASH160 << h160 << OP_EQUALVERIFY << OP_CHECKSIG;
1428 const std::string comment = PayoutComment(blockHeight);
1429
1430 // Candidate legacy matches, adjudicated after cs_mapWallet is released.
1431 //
1432 // The depth check cannot happen inside the scan: GetDepthInMainChain()
1433 // reads mapBlockIndex and pindexBest, which belong to cs_main, and cs_main
1434 // is taken BEFORE cs_mapWallet everywhere else in this tree
1435 // (CreateTransaction, ConnectBlock). Taking it the other way round here
1436 // would be a lock-order inversion in the payout thread. There is at most a
1437 // handful of these: a candidate has to carry this exact height's comment
1438 // and an output of exactly this amount to this address.
1439 std::vector<uint256> legacyCandidates;
1440
1441 // One pass, not two: this scan runs under cs_mapWallet, which the whole
1442 // node contends on, and the merge is already paying for it once per record.
576 CRITICAL_BLOCK(cs_mapWallet) 1443 CRITICAL_BLOCK(cs_mapWallet)
577 { 1444 {
578 for (map<uint256, CWalletTx>::const_iterator it = mapWallet.begin(); 1445 for (map<uint256, CWalletTx>::const_iterator it = mapWallet.begin();
@@ -580,23 +1447,130 @@ static bool FindRecordedPoolPayout(const std::string& id,
580 { 1447 {
581 const CWalletTx& wtx = it->second; 1448 const CWalletTx& wtx = it->second;
582 map<string, string>::const_iterator idIt = wtx.mapValue.find("pool_payout_id"); 1449 map<string, string>::const_iterator idIt = wtx.mapValue.find("pool_payout_id");
583 if (idIt == wtx.mapValue.end() || idIt->second != id) 1450 if (idIt != wtx.mapValue.end()) {
1451 // Written by this build. Adjudicated exactly, and never
1452 // offered to the fallback below.
1453 if (idIt->second != id)
1454 continue;
1455 map<string, string>::const_iterator addrIt = wtx.mapValue.find("pool_payout_address");
1456 if (addrIt == wtx.mapValue.end() || addrIt->second != address)
1457 continue;
1458 map<string, string>::const_iterator amtIt = wtx.mapValue.find("pool_payout_amount");
1459 if (amtIt == wtx.mapValue.end() || amtIt->second != strprintf("%lld", amount))
1460 continue;
1461 txidRet = wtx.GetHash().GetHex();
1462 return PAYOUT_PROBE_STRUCTURED;
1463 }
1464
1465 if (!fHaveScript)
584 continue; 1466 continue;
585 map<string, string>::const_iterator addrIt = wtx.mapValue.find("pool_payout_address"); 1467 // The comment is written only by the payout path, so its presence
586 if (addrIt == wtx.mapValue.end() || addrIt->second != address) 1468 // already identifies the transaction as one of ours.
1469 map<string, string>::const_iterator ci = wtx.mapValue.find("comment");
1470 if (ci == wtx.mapValue.end() || ci->second != comment)
587 continue; 1471 continue;
588 map<string, string>::const_iterator amtIt = wtx.mapValue.find("pool_payout_amount"); 1472 if (excluded.count(wtx.GetHash().GetHex()))
589 if (amtIt == wtx.mapValue.end() || amtIt->second != strprintf("%lld", amount)) 1473 continue; // already credited to a different debt
1474 foreach (const CTxOut& txout, wtx.vout)
1475 if (txout.nValue == amount && txout.scriptPubKey == want) {
1476 legacyCandidates.push_back(it->first);
1477 break;
1478 }
1479 }
1480 }
1481
1482 // A legacy match must have CONFIRMED.
1483 //
1484 // The structured path does not need this and deliberately does not have
1485 // it: this process sent that transaction moments ago, and a send that
1486 // failed leaves nothing behind, because SendMoney calls
1487 // RollbackTransactionSpent, which calls EraseFromWallet.
1488 //
1489 // The legacy path is the opposite case in every respect. Those
1490 // transactions are days or weeks old, so whether they confirmed is a
1491 // settled fact rather than an open question, and a pre-upgrade payout that
1492 // never confirmed is evidence of NON-payment about to be read as payment.
1493 // This chain has had a 280-block reorg (see FlushMaturePayouts): a payout
1494 // invalidated by it is still sitting in mapWallet with its comment intact,
1495 // and the first boot after this upgrade is exactly when it gets matched.
1496 // The cost of being wrong here is a miner who is never paid and no record
1497 // that they were owed.
1498 CRITICAL_BLOCK(cs_main)
1499 CRITICAL_BLOCK(cs_mapWallet)
1500 {
1501 for (const uint256& hash : legacyCandidates) {
1502 map<uint256, CWalletTx>::const_iterator it = mapWallet.find(hash);
1503 if (it == mapWallet.end())
1504 continue; // erased under us; it is not evidence of anything
1505 int depth = it->second.GetDepthInMainChain();
1506 if (depth < 1) {
1507 // Loud: this is the difference between a miner having been
1508 // paid and a miner about to be paid twice, decided on a fact
1509 // that is easy to check and impossible to recover afterwards.
1510 LogPrint("payout", "[payout] wallet tx %s carries the payout"
1511 " comment for height %d and an output of exactly %s to"
1512 " %s, but it is NOT in the main chain (depth %d) --"
1513 " refusing to read it as payment; this recipient will"
1514 " be paid\n",
1515 hash.GetHex().c_str(), blockHeight,
1516 FormatMoney(amount).c_str(), address.c_str(), depth);
590 continue; 1517 continue;
591 txidRet = wtx.GetHash().GetHex(); 1518 }
592 return true; 1519 txidRet = hash.GetHex();
1520 return PAYOUT_PROBE_LEGACY;
593 } 1521 }
594 } 1522 }
595 return false; 1523 return PAYOUT_PROBE_NONE;
596} 1524}
597 1525
1526// txid -> "<round id>|<address>", for payments this run has already credited
1527// to a debt on legacy evidence. Stops one pre-upgrade transaction being read
1528// as proof that two different obligations were met. The ledger's paidTxids
1529// covers the same ground durably; this covers the window before the recovered
1530// txid reaches the ledger, and the case where the ledger entry has already
1531// been retired.
1532static std::map<std::string, std::string> gLegacyClaimedTxids;
1533static std::mutex gClaimedTxidMutex;
1534
1535// Every txid already spoken for by some payment OTHER than this one. Collected
1536// before cs_mapWallet is taken: the ledger mutex must not be acquired under a
1537// wallet lock.
1538static void CollectForeignPayoutTxids(int blockHeight, const uint256& blockHash,
1539 const std::string& address,
1540 const std::string& claimKey,
1541 std::set<std::string>& out)
1542{
1543 {
1544 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
1545 for (const PendingPayout& pp : gPendingPayouts) {
1546 bool sameRound = SamePayoutRound(pp.blockHeight, pp.blockHash,
1547 blockHeight, blockHash);
1548 for (const auto& kv : pp.paidTxids)
1549 if (!(sameRound && kv.first == address))
1550 out.insert(kv.second);
1551 }
1552 }
1553 {
1554 std::lock_guard<std::mutex> lk(gClaimedTxidMutex);
1555 for (const auto& kv : gLegacyClaimedTxids)
1556 if (kv.second != claimKey)
1557 out.insert(kv.first);
1558 }
1559}
1560
1561static void ClaimLegacyPayoutTxid(const std::string& txid, const std::string& claimKey)
1562{
1563 std::lock_guard<std::mutex> lk(gClaimedTxidMutex);
1564 gLegacyClaimedTxids[txid] = claimKey;
1565}
1566
1567// blockHash stays a uint256 and not upstream's display string: it is the key
1568// this fork's payout ledger is built on, and a key that has been through
1569// ToString() is a key that can be compared wrong. It is converted to text only
1570// where the round proof needs text. The other three arguments are upstream's,
1571// carried for the round-proof data.
598static void QueuePayouts(int blockHeight, 1572static void QueuePayouts(int blockHeight,
599 const std::string& blockHash, 1573 const uint256& blockHash,
600 const std::string& coinbaseTxid, 1574 const std::string& coinbaseTxid,
601 int64 coinbaseValue, 1575 int64 coinbaseValue,
602 const std::string& foundBy, 1576 const std::string& foundBy,
@@ -604,11 +1578,11 @@ static void QueuePayouts(int blockHeight,
604 uint64 total) 1578 uint64 total)
605{ 1579{
606 PoolRoundProof round; 1580 PoolRoundProof round;
607 round.id = PayoutIdForBlock(blockHeight); 1581 round.id = PayoutIdForBlock(blockHeight, blockHash);
608 round.blockHeight = blockHeight; 1582 round.blockHeight = blockHeight;
609 round.matureAtHeight = blockHeight + COINBASE_MATURITY; 1583 round.matureAtHeight = blockHeight + COINBASE_SPENDABLE_DEPTH;
610 round.createdAt = GetTime(); 1584 round.createdAt = GetTime();
611 round.blockHash = blockHash; 1585 round.blockHash = blockHash.ToString();
612 round.coinbaseTxid = coinbaseTxid; 1586 round.coinbaseTxid = coinbaseTxid;
613 round.coinbaseValue = coinbaseValue; 1587 round.coinbaseValue = coinbaseValue;
614 round.foundBy = foundBy; 1588 round.foundBy = foundBy;
@@ -639,9 +1613,21 @@ static void QueuePayouts(int blockHeight,
639 round.operatorReward = operatorCut; 1613 round.operatorReward = operatorCut;
640 1614
641 PendingPayout pp; 1615 PendingPayout pp;
642 pp.id = round.id;
643 pp.blockHeight = blockHeight; 1616 pp.blockHeight = blockHeight;
644 pp.matureAtHeight = blockHeight + COINBASE_MATURITY; 1617 pp.blockHash = blockHash;
1618 // COINBASE_SPENDABLE_DEPTH, not COINBASE_MATURITY. Paying at depth 100
1619 // means the block's own coinbase is still 20 blocks short of spendable, so
1620 // the payment has to come out of some *other* mature coin -- the pool runs
1621 // a permanent ~19-block float and every payout fails until unrelated coin
1622 // matures. The earlier "96% of 934 payments went through at depth 100"
1623 // measurement that justified the shorter wait was taken while this node was
1624 // alone on a private fork paying itself, so it says nothing about a pool
1625 // competing on the real chain. Waiting the extra 20 blocks makes each
1626 // payout funded by construction, and puts the maturity gate past the
1627 // reorg depth that orphaned 280 blocks of obligations. Upstream uses
1628 // COINBASE_MATURITY here; this is a considered disagreement, not an
1629 // oversight -- see the merge commit message.
1630 pp.matureAtHeight = blockHeight + COINBASE_SPENDABLE_DEPTH;
645 1631
646 int skipped = 0; 1632 int skipped = 0;
647 int64 burned = 0; // amount owed to miners with bad/empty addresses 1633 int64 burned = 0; // amount owed to miners with bad/empty addresses
@@ -668,9 +1654,13 @@ static void QueuePayouts(int blockHeight,
668 round.forfeitedBadAddress = burned; 1654 round.forfeitedBadAddress = burned;
669 1655
670 { 1656 {
671 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex); 1657 VersionedBlob snap;
672 gPendingPayouts.push_back(pp); 1658 {
673 SavePendingPayoutsLocked(); 1659 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
1660 gPendingPayouts.push_back(pp);
1661 snap = SerializePendingPayoutsLocked();
1662 }
1663 PublishPendingPayouts(snap);
674 } 1664 }
675 RecordPoolRound(round); 1665 RecordPoolRound(round);
676 1666
@@ -689,37 +1679,338 @@ static void QueuePayouts(int blockHeight,
689 : ""); 1679 : "");
690} 1680}
691 1681
692// Called from event loop: copy matured entries to the exec queue (fast). 1682// Called from the event loop: explode matured ledger entries into per-recipient
693// The ledger entry stays on disk until every recipient has a txid recorded. 1683// in-flight records (fast; no wallet or disk work beyond one small write).
694// On restart, FindRecordedPoolPayout() checks wallet metadata before sending, 1684//
695// so a crash after SendMoney but before the ledger update does not double-pay. 1685// The ledger entry SURVIVES here. That is upstream's model and it is the
1686// crash-safe one: this fork used to erase the entry as soon as its payments
1687// were handed off, so a death in that window left the debt described by
1688// nothing at all -- a container restart on 2026-07-30 silently dropped
1689// ~1,156 BTF exactly that way. An entry is removed only once every recipient
1690// in it has a txid recorded (PayoutFullyPaid), which the payout thread does.
1691//
1692// gPaying is kept on top of that as the in-flight working set: it carries the
1693// per-recipient retry/deferral state that a per-round `nextAttempt` cannot
1694// express, and it is what the payout thread iterates. It is a cache of work
1695// derived from the ledger, never an independent record of debt -- if it were
1696// lost entirely the ledger would rebuild it on the next flush.
696static void FlushMaturePayouts() 1697static void FlushMaturePayouts()
697{ 1698{
1699 // Which funding blocks are still on the main chain? Asked with no payout
1700 // lock held: PayoutBlockStillOnChain takes cs_main, and cs_main is held by
1701 // ProcessBlock and SendMoney for long stretches. Holding the ledger mutex
1702 // across it would let the event loop park the payout thread behind a chain
1703 // lock it has no interest in. (It is not a deadlock either way -- nothing
1704 // takes the ledger mutex while holding cs_main -- but it is exactly the
1705 // kind of stall this merge is removing from the ledger path.)
1706 std::vector<uint256> orphaned;
1707 {
1708 std::vector<uint256> hashes;
1709 {
1710 std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);
1711 for (const PendingPayout& pp : gPendingPayouts)
1712 hashes.push_back(pp.blockHash);
1713 }
1714 for (const uint256& h : hashes)
1715 if (!PayoutBlockStillOnChain(h))
1716 orphaned.push_back(h);
1717 }
1718
698 std::vector<PendingPayout> ready; 1719 std::vector<PendingPayout> ready;
1720 VersionedBlob ledgerSnap;
699 { 1721 {
700 std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex); 1722 std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);
701 std::lock_guard<std::mutex> queue(gPayoutExecMutex); 1723
702 int64 now = GetTime(); 1724 // Void obligations whose funding block lost a reorg. The coin that was
1725 // meant to pay them does not exist, so paying anyway spends real-chain
1726 // coin the pool owes to somebody else.
1727 bool voided = false;
1728 for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ) {
1729 if (std::find(orphaned.begin(), orphaned.end(), it->blockHash) == orphaned.end()) {
1730 ++it;
1731 continue;
1732 }
1733 int64 total = 0;
1734 for (auto& kv : it->amounts) total += kv.second;
1735 LogPrint("payout", "[payout] block %d (%s) was orphaned -- voiding %zu"
1736 " payment(s) totalling %s\n",
1737 it->blockHeight, it->blockHash.ToString().substr(0, 14).c_str(),
1738 it->amounts.size(), FormatMoney(total).c_str());
1739 it = gPendingPayouts.erase(it);
1740 voided = true;
1741 }
1742 if (voided) ledgerSnap = SerializePendingPayoutsLocked();
1743
703 for (const PendingPayout& pp : gPendingPayouts) { 1744 for (const PendingPayout& pp : gPendingPayouts) {
704 if (nBestHeight < pp.matureAtHeight) 1745 if (nBestHeight < pp.matureAtHeight)
705 continue; 1746 continue;
706 if (pp.nextAttempt > now)
707 continue;
708 if (PayoutFullyPaid(pp)) 1747 if (PayoutFullyPaid(pp))
709 continue; 1748 continue;
710 if (gQueuedPayouts.count(pp.id)) 1749 // An entry with neither a hash nor a height cannot be told apart
1750 // from another in the same state, so a txid recorded against it
1751 // could settle the wrong debt. Paying it is the unsafe move;
1752 // leaving it standing and saying so is not.
1753 if (!PayoutRoundIdentifiable(pp.blockHeight, pp.blockHash)) {
1754 LogPrint("payout", "[payout] WARNING: ledger entry maturing at"
1755 " height %d names neither a block hash nor a height --"
1756 " refusing to pay %zu recipient(s) that cannot be told"
1757 " apart from another round's\n",
1758 pp.matureAtHeight, pp.amounts.size());
711 continue; 1759 continue;
712 LogPrint("payout", "[payout] block height %d matured -- queuing payout %s (%zu recipients)\n", 1760 }
713 pp.matureAtHeight, pp.id.c_str(), pp.amounts.size());
714 ready.push_back(pp); 1761 ready.push_back(pp);
715 gQueuedPayouts.insert(pp.id);
716 } 1762 }
717 for (auto& pp : ready)
718 gPayoutExecQueue.push_back(pp);
719 } 1763 }
1764 // Outside the ledger mutex: publishing is I/O and the event loop is the
1765 // one thread that must never wait on the payout thread's disk.
1766 PublishPendingPayouts(ledgerSnap);
1767
1768 if (ready.empty())
1769 return;
1770
1771 VersionedBlob payingSnap;
1772 {
1773 std::lock_guard<std::mutex> queue(gPayoutExecMutex);
1774 size_t added = 0;
1775 for (const PendingPayout& pp : ready) {
1776 const std::string id = PayoutIdForBlock(pp.blockHeight, pp.blockHash);
1777 if (gQueuedPayouts.count(id))
1778 continue; // already exploded into gPaying and still in flight
1779 size_t n = 0;
1780 for (const auto& kv : pp.amounts) {
1781 // Upstream's per-recipient dedupe: a recipient already carrying
1782 // a txid is done, even though the entry as a whole is not.
1783 if (pp.paidTxids.count(kv.first))
1784 continue;
1785 // The invariant, checked against the thing it is an invariant
1786 // of: at most one in-flight record per (round, address). The
1787 // round-level set above is an optimisation that can be wrong --
1788 // it is memory only, it is rebuilt from gPaying at startup, and
1789 // a stale `ready` snapshot can race a round finishing --
1790 // whereas gPaying is the list being protected. A second record
1791 // here is a second standing order to send the same money.
1792 if (PaymentInFlightLocked(pp.blockHeight, pp.blockHash, kv.first))
1793 continue;
1794 PayingRecord r;
1795 r.id = ++gPayingNextId;
1796 r.blockHeight = pp.blockHeight;
1797 r.blockHash = pp.blockHash;
1798 r.matureAtHeight = pp.matureAtHeight;
1799 r.address = kv.first;
1800 r.amount = kv.second;
1801 gPaying.push_back(r);
1802 n++;
1803 }
1804 gQueuedPayouts.insert(id);
1805 added += n;
1806 LogPrint("payout", "[payout] block height %d matured -- queuing"
1807 " payout %s (%zu of %zu recipients still owed)\n",
1808 pp.matureAtHeight, id.c_str(), n, pp.amounts.size());
1809 }
1810 if (added) payingSnap = SerializePayingLocked();
1811 }
1812 PublishPaying(payingSnap);
720} 1813}
721 1814
722// Payout execution thread: drains gPayoutExecQueue, calls SendMoney. 1815// Settle a payment: it is either done or abandoned, either way it stops being
1816// in flight. Note this only retires the working record -- whether the miner is
1817// still OWED the money is the ledger's business, and the ledger is updated
1818// separately by RecordPayoutTxid().
1819//
1820// Erases by record id, not by comparing fields. Value-matching was only an
1821// identity for as long as no two records could be equal, and the restart bug
1822// above produced records that were equal in every field: the first match then
1823// absorbed every erase and every state update, and the second was never
1824// touched again. The id names the slot the caller took its copy from, so the
1825// update lands where it belongs whatever else is in the list.
1826static void ForgetPayment(const PayingRecord& r)
1827{
1828 VersionedBlob snap;
1829 {
1830 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
1831 for (auto it = gPaying.begin(); it != gPaying.end(); ++it)
1832 if (it->id == r.id) { gPaying.erase(it); break; }
1833
1834 // When the last record for a round is gone the round is no longer in
1835 // flight, so a later flush may explode it again if the ledger still
1836 // shows recipients owed. That is the intended behaviour and it is now
1837 // the only way a stalled payment ever leaves this list: a payment that
1838 // keeps failing is retried forever rather than given up on, so its
1839 // record stays here, and the flush that would re-explode it is stopped
1840 // by PaymentInFlightLocked.
1841 bool stillInFlight = false;
1842 for (const auto& e : gPaying)
1843 if (SamePayoutRound(e.blockHeight, e.blockHash,
1844 r.blockHeight, r.blockHash)) {
1845 stillInFlight = true;
1846 break;
1847 }
1848 if (!stillInFlight)
1849 gQueuedPayouts.erase(PayoutIdForBlock(r.blockHeight, r.blockHash));
1850
1851 snap = SerializePayingLocked();
1852 }
1853 PublishPaying(snap);
1854}
1855
1856// Persist an updated attempt count / backoff for a payment still owed.
1857static void ReschedulePayment(const PayingRecord& r)
1858{
1859 VersionedBlob snap;
1860 {
1861 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
1862 for (auto& e : gPaying)
1863 if (e.id == r.id) {
1864 e.retries = r.retries;
1865 e.deferrals = r.deferrals;
1866 e.retryAfter = r.retryAfter;
1867 // Carry the cleared probe flag back, or every retry re-scans
1868 // the whole wallet -- the cost this merge is specifically
1869 // removing.
1870 e.fProbeWallet = r.fProbeWallet;
1871 break;
1872 }
1873 snap = SerializePayingLocked();
1874 }
1875 PublishPaying(snap);
1876}
1877
1878// Record that a recipient has been paid, in the ledger, which is the only
1879// durable statement that the debt is settled.
1880//
1881// The file write happens with gPayoutLedgerMutex RELEASED. Upstream writes the
1882// file inside the per-recipient loop with the mutex held, so every recipient
1883// costs the event loop a full serialise+fsync it has to wait for in
1884// FlushMaturePayouts. Ordering against the event loop's own writes is the
1885// snapshot version's job, not the mutex's -- see PublishVersioned.
1886static void RecordPayoutTxid(const PayingRecord& r, const std::string& txid)
1887{
1888 VersionedBlob blob;
1889 bool complete = false;
1890 {
1891 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
1892 for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ++it) {
1893 if (!SamePayoutRound(it->blockHeight, it->blockHash,
1894 r.blockHeight, r.blockHash))
1895 continue;
1896 it->paidTxids[r.address] = txid;
1897 if (PayoutFullyPaid(*it)) {
1898 LogPrint("payout", "[payout] payout for block %d complete --"
1899 " removing from ledger\n", it->blockHeight);
1900 gPendingPayouts.erase(it);
1901 complete = true;
1902 }
1903 break;
1904 }
1905 blob = SerializePendingPayoutsLocked();
1906 }
1907 PublishPendingPayouts(blob);
1908 // The round proof records a txid per address, so this is per recipient and
1909 // not only on completion.
1910 MarkPoolRoundPaid(PayoutIdForBlock(r.blockHeight, r.blockHash), r.address, txid);
1911 if (complete)
1912 LogPrint("payout", "[payout] block %d fully settled\n", r.blockHeight);
1913}
1914
1915// Write a recipient off.
1916//
1917// Only ever called for a state that is DECIDABLE -- currently one: the ledger
1918// names an address that is not a valid address, so no transaction paying it
1919// can be constructed by this or any other build. It is not called, and must
1920// not be, because a payment has failed some number of times. "The wallet said
1921// no N times" is not a decidable state, it is a description of how long we
1922// have been asking; a debt written off on the strength of it is a debt the
1923// operator keeps, and the arithmetic that decided WHY the wallet said no was
1924// wrong for a year (see PayoutThreadFn).
1925//
1926// The amount is removed from the entry rather than marked paid, so the money
1927// is not reported as sent. Once nothing is left owed the entry retires. The
1928// round proof is told separately, because dropping the obligation from the
1929// ledger while pool_rounds.json goes on publishing the address as owed with
1930// "txid": null leaves the pool's two records of the same round permanently
1931// disagreeing.
1932static void ForfeitPayoutRecipient(const PayingRecord& r, const std::string& reason)
1933{
1934 LogPrint("payout", "[payout] writing off %s owed to %s for block %d: %s\n",
1935 FormatMoney(r.amount).c_str(), r.address.c_str(), r.blockHeight,
1936 reason.c_str());
1937 VersionedBlob blob;
1938 {
1939 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
1940 for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ++it) {
1941 if (!SamePayoutRound(it->blockHeight, it->blockHash,
1942 r.blockHeight, r.blockHash))
1943 continue;
1944 it->amounts.erase(r.address);
1945 // PayoutFullyPaid() returns true for an empty amounts map -- there
1946 // is no recipient in it without a txid -- so the
1947 // `amounts.empty()` disjunct that used to be here decided nothing.
1948 if (PayoutFullyPaid(*it))
1949 gPendingPayouts.erase(it);
1950 break;
1951 }
1952 blob = SerializePendingPayoutsLocked();
1953 }
1954 PublishPendingPayouts(blob);
1955 MarkPoolRoundForfeited(PayoutIdForBlock(r.blockHeight, r.blockHash),
1956 r.address, reason);
1957}
1958
1959// What a failed send means for the payment's retry budget.
1960//
1961// A deferral is "the wallet's condition is temporarily wrong": wait, and do
1962// not hold it against this payment. A fault is "the wallet is funded and
1963// unlocked and still would not do it", which is worth logging loudly and
1964// retrying with backoff.
1965//
1966// This used to be decided by re-reading GetBalance() after the failure and
1967// comparing it against `amount + nTransactionFee + CENT`. Every term of that
1968// was wrong:
1969//
1970// * nTransactionFee is 0 in production. It is only ever assigned from the
1971// GUI preferences dialog (gui.cpp), which a headless pool node does not
1972// have, so the whole reserve was the CENT.
1973//
1974// * The fee the reserve is for is GetMinFee(fDiscount=true), a STEP
1975// function of the serialised size: 0 below 10,000 bytes, and
1976// (1 + bytes/1000) * CENT at or above it -- so 11 * CENT at the step.
1977// There is no required-fee value between 1 and 11 CENT. A one-CENT
1978// reserve is therefore never the right number: below the step nothing
1979// needs reserving, and at the step it is short by a factor of eleven.
1980//
1981// * A pool wallet close to broke selects many small inputs BECAUSE it is
1982// close to broke, and ~56 signed P2PKH inputs is 10 KB. The trigger and
1983// the failure are positively correlated: the guess is wrong precisely
1984// when the pool is nearly out of money.
1985//
1986// Concretely, with a balance of amount + 0.05: the pre-send gate passes,
1987// CreateTransaction builds a >10 KB transaction, the fee becomes 0.11,
1988// SelectCoins(amount + 0.11) fails, SendMoney returns false -- and the old
1989// arithmetic said (amount + 0.05) < (amount + 0.01) is false, so "funded",
1990// so a fault. The condition is stable, so it recurred every 60 s, and 1440 of
1991// them erased the debt with the coin still in the operator's wallet.
1992//
1993// So stop guessing. The wallet is asked to say what it did, and this only
1994// classifies the answer. WALLET_SEND_INSUFFICIENT_FUNDS is coin selection
1995// failing to cover amount plus whatever fee that size of transaction
1996// required, whatever that turned out to be -- no reserve to get wrong.
1997enum PayoutFailureClass {
1998 PAYOUT_FAILURE_DEFER = 0, // conditions are wrong; wait for them to change
1999 PAYOUT_FAILURE_FAULT // funded and unlocked, and it still would not go
2000};
2001
2002static PayoutFailureClass ClassifyPayoutSendFailure(WalletSendResult result,
2003 bool fWalletLocked)
2004{
2005 // A locked wallet cannot sign. Nothing about this payment is wrong.
2006 if (fWalletLocked)
2007 return PAYOUT_FAILURE_DEFER;
2008 if (result == WALLET_SEND_INSUFFICIENT_FUNDS)
2009 return PAYOUT_FAILURE_DEFER;
2010 return PAYOUT_FAILURE_FAULT;
2011}
2012
2013// Payout execution thread: works through gPaying, calls SendMoney.
723// Runs independently of the event loop so slow wallet operations 2014// Runs independently of the event loop so slow wallet operations
724// (SendMoney takes cs_main) never stall miner I/O. 2015// (SendMoney takes cs_main) never stall miner I/O.
725static void PayoutThreadFn(void*) 2016static void PayoutThreadFn(void*)
@@ -728,103 +2019,259 @@ static void PayoutThreadFn(void*)
728 LogPrint("payout", "[payout] payout thread started\n"); 2019 LogPrint("payout", "[payout] payout thread started\n");
729 2020
730 while (!fShutdown && gPoolRunning) { 2021 while (!fShutdown && gPoolRunning) {
731 std::vector<PendingPayout> batch; 2022 std::vector<PayingRecord> batch;
732 { 2023 {
733 std::lock_guard<std::mutex> lk(gPayoutExecMutex); 2024 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
734 batch.swap(gPayoutExecQueue); 2025 int64 now = GetTime();
2026 for (const auto& r : gPaying)
2027 if (r.retryAfter <= now)
2028 batch.push_back(r);
735 } 2029 }
736 2030
737 for (auto& ppQueued : batch) { 2031 for (PayingRecord r : batch) {
738 LogPrint("payout", "[payout] executing payments for block matured at %d" 2032 const std::string id = PayoutIdForBlock(r.blockHeight, r.blockHash);
739 " (%zu miners)\n", ppQueued.matureAtHeight, ppQueued.amounts.size()); 2033
740 bool fAbortBatch = false; 2034 // Re-check here and not only at maturity: a payment can sit in this
741 for (auto& kv : ppQueued.amounts) { 2035 // list for hours across retries, and the reorg that kills its
2036 // funding block can land at any point in that window. Upstream has
2037 // no equivalent -- it decides once, when the entry matures.
2038 if (!PayoutBlockStillOnChain(r.blockHash)) {
2039 LogPrint("payout", "[payout] block %d (%s) orphaned while queued"
2040 " -- dropping %s to %s\n",
2041 r.blockHeight, r.blockHash.ToString().substr(0, 14).c_str(),
2042 FormatMoney(r.amount).c_str(), r.address.c_str());
2043 ForgetPayment(r);
2044 continue;
2045 }
2046
2047 // Upstream's per-recipient dedupe, against the ledger rather than
2048 // against the batch: if a txid is already recorded for this
2049 // address, the money went out and nothing more is owed.
2050 {
742 bool alreadyPaid = false; 2051 bool alreadyPaid = false;
743 { 2052 {
744 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex); 2053 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
745 for (const PendingPayout& pp : gPendingPayouts) 2054 for (const PendingPayout& pp : gPendingPayouts)
746 if (pp.id == ppQueued.id && pp.paidTxids.count(kv.first)) 2055 if (SamePayoutRound(pp.blockHeight, pp.blockHash,
2056 r.blockHeight, r.blockHash)
2057 && pp.paidTxids.count(r.address)) {
747 alreadyPaid = true; 2058 alreadyPaid = true;
2059 break;
2060 }
748 } 2061 }
749 if (alreadyPaid) 2062 if (alreadyPaid) {
2063 ForgetPayment(r);
750 continue; 2064 continue;
751
752 uint160 h160;
753 if (!AddressToHash160(kv.first, h160)) {
754 // Should not happen -- bad addresses are filtered in QueuePayouts
755 LogPrint("payout", "[payout] SKIPPED bad address '%s'\n",
756 kv.first.c_str());
757 continue;
758 }
759 CScript sc;
760 sc << OP_DUP << OP_HASH160 << h160 << OP_EQUALVERIFY << OP_CHECKSIG;
761 CWalletTx wtx;
762 wtx.mapValue["comment"] = strprintf("Pool payout height %d", ppQueued.matureAtHeight);
763 wtx.mapValue["pool_payout_id"] = ppQueued.id;
764 wtx.mapValue["pool_payout_address"] = kv.first;
765 wtx.mapValue["pool_payout_amount"] = strprintf("%lld", kv.second);
766
767 std::string txid;
768 if (FindRecordedPoolPayout(ppQueued.id, kv.first, kv.second, txid))
769 {
770 LogPrint("payout", "[payout] recovered recorded payout %s to %s"
771 " from wallet tx %s\n",
772 FormatMoney(kv.second).c_str(), kv.first.c_str(), txid.c_str());
773 MarkPoolRoundPaid(ppQueued.id, kv.first, txid);
774 }
775 else if (SendMoney(sc, kv.second, wtx))
776 {
777 txid = wtx.GetHash().GetHex();
778 LogPrint("payout", "[payout] paid %s to %s\n",
779 FormatMoney(kv.second).c_str(), kv.first.c_str());
780 MarkPoolRoundPaid(ppQueued.id, kv.first, txid);
781 }
782 else
783 {
784 LogPrint("payout", "[payout] FAILED to pay %s to %s"
785 " (insufficient funds or wallet locked?)\n",
786 FormatMoney(kv.second).c_str(), kv.first.c_str());
787 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
788 for (PendingPayout& pp : gPendingPayouts)
789 if (pp.id == ppQueued.id)
790 pp.nextAttempt = GetTime() + 60;
791 SavePendingPayoutsLocked();
792 fAbortBatch = true;
793 break;
794 }
795
796 {
797 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
798 for (PendingPayout& pp : gPendingPayouts) {
799 if (pp.id != ppQueued.id)
800 continue;
801 pp.paidTxids[kv.first] = txid;
802 pp.nextAttempt = 0;
803 SavePendingPayoutsLocked();
804 break;
805 }
806 } 2065 }
807 } 2066 }
808 2067
809 { 2068 uint160 h160;
810 std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex); 2069 if (!AddressToHash160(r.address, h160)) {
811 for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ) { 2070 // Should not happen -- bad addresses are filtered in
812 if (it->id == ppQueued.id && PayoutFullyPaid(*it)) { 2071 // QueuePayouts. If one is here anyway it is decidable and
813 LogPrint("payout", "[payout] payout %s complete -- removing from ledger\n", 2072 // permanent: there is no transaction that pays a string which
814 it->id.c_str()); 2073 // is not an address, so retrying is not waiting for anything.
815 it = gPendingPayouts.erase(it); 2074 //
816 SavePendingPayoutsLocked(); 2075 // This used to only ForgetPayment(), which drops the in-flight
2076 // record and leaves the ledger entry claiming the debt and
2077 // pool_rounds.json advertising the recipient as unpaid
2078 // forever, with the next flush re-exploding it. Write it off
2079 // properly and say why.
2080 const std::string reason =
2081 strprintf("'%s' is not a valid address, so no transaction"
2082 " can pay it", r.address.c_str());
2083 LogPrint("payout", "[payout] SKIPPED bad address '%s' --"
2084 " %s owed for block %d cannot be paid\n",
2085 r.address.c_str(), FormatMoney(r.amount).c_str(),
2086 r.blockHeight);
2087 ForfeitPayoutRecipient(r, reason);
2088 ForgetPayment(r);
2089 continue;
2090 }
2091 CScript sc;
2092 sc << OP_DUP << OP_HASH160 << h160 << OP_EQUALVERIFY << OP_CHECKSIG;
2093
2094 // Did a previous process send this and die before recording it?
2095 // Only asked once per record -- see PayingRecord::fProbeWallet.
2096 if (r.fProbeWallet) {
2097 std::string recovered;
2098 const std::string claimKey = id + "|" + r.address;
2099 std::set<std::string> excluded;
2100 CollectForeignPayoutTxids(r.blockHeight, r.blockHash, r.address,
2101 claimKey, excluded);
2102 r.fProbeWallet = false;
2103 PayoutProbeResult probe = FindRecordedPoolPayout(
2104 id, r.blockHeight, r.address, r.amount, excluded, recovered);
2105 if (probe != PAYOUT_PROBE_NONE) {
2106 if (probe == PAYOUT_PROBE_LEGACY) {
2107 // Loud on purpose: this is the upgrade path, where the
2108 // only evidence a payment went out is a comment written
2109 // by a binary that is no longer running.
2110 LogPrint("payout", "[payout] %s to %s was already paid by"
2111 " a pre-upgrade build (wallet tx %s carries the"
2112 " payout comment for height %d and an output of"
2113 " exactly this amount, and no structured payout"
2114 " metadata) -- not resending\n",
2115 FormatMoney(r.amount).c_str(), r.address.c_str(),
2116 recovered.c_str(), r.blockHeight);
2117 ClaimLegacyPayoutTxid(recovered, claimKey);
817 } else { 2118 } else {
818 ++it; 2119 LogPrint("payout", "[payout] recovered recorded payout %s to %s"
2120 " from wallet tx %s -- not resending\n",
2121 FormatMoney(r.amount).c_str(), r.address.c_str(),
2122 recovered.c_str());
819 } 2123 }
2124 RecordPayoutTxid(r, recovered);
2125 ForgetPayment(r);
2126 continue;
820 } 2127 }
821 } 2128 }
822 { 2129
823 std::lock_guard<std::mutex> lk(gPayoutExecMutex); 2130 // Waiting for coin to mature is not a failure, so it must not
824 gQueuedPayouts.erase(ppQueued.id); 2131 // consume the retry budget. Burning a retry every 60s means a
2132 // payment the pool fully intends to honour can hit GIVING UP
2133 // 24h later purely because the subsidy funding it had not
2134 // matured yet -- the miner is then silently never paid.
2135 //
2136 // The comparison includes the fee. It used to be against the
2137 // amount alone, so a balance anywhere between amount and
2138 // amount+fee sailed past this gate, failed in SendMoney, and was
2139 // counted as a send FAILURE rather than a wait -- 1440 of those is
2140 // 24 hours, and the recipient was erased with the money left in
2141 // the operator's wallet and nothing on disk saying it was owed.
2142 //
2143 // nTransactionFee is the part of the fee that can be known before
2144 // the transaction exists, and in production it is 0 -- it is only
2145 // ever assigned from the GUI preferences dialog, which a pool node
2146 // does not have -- so including it here changes nothing on this
2147 // pool and is kept only because a GUI operator can set it. The
2148 // rest is GetMinFee(), a function of the serialised size and
2149 // therefore of how many inputs SelectCoins ends up picking, which
2150 // cannot be predicted from here -- so this gate deliberately does
2151 // NOT try to reserve it. Refusing to attempt a payment the wallet
2152 // could actually have made would be its own way of not paying a
2153 // miner. That residual is handled after the fact, by asking the
2154 // wallet why it refused -- see ClassifyPayoutSendFailure.
2155 int64 nSpendable = GetBalance();
2156 if (nSpendable < r.amount + nTransactionFee) {
2157 r.deferrals++;
2158 r.retryAfter = GetTime() + 60;
2159 // Hourly, not every minute: a dry pool would otherwise
2160 // write one line per payment per minute forever.
2161 if (r.deferrals == 1 || r.deferrals % 60 == 0)
2162 LogPrint("payout", "[payout] deferring %s to %s -- only %s"
2163 " spendable, need %s with fee (waiting %d min so far)\n",
2164 FormatMoney(r.amount).c_str(), r.address.c_str(),
2165 FormatMoney(nSpendable).c_str(),
2166 FormatMoney(r.amount + nTransactionFee).c_str(),
2167 r.deferrals);
2168 ReschedulePayment(r);
2169 continue;
825 } 2170 }
826 if (fAbortBatch) 2171
2172 // Likewise a locked wallet: it cannot sign, so every attempt fails
2173 // for a reason that has nothing to do with this payment being
2174 // payable. Left to the retry budget, an operator who locks the
2175 // wallet overnight comes back to erased debts.
2176 if (IsWalletLocked()) {
2177 r.deferrals++;
2178 r.retryAfter = GetTime() + 60;
2179 if (r.deferrals == 1 || r.deferrals % 60 == 0)
2180 LogPrint("payout", "[payout] deferring %s to %s -- wallet is"
2181 " locked (waiting %d min so far)\n",
2182 FormatMoney(r.amount).c_str(), r.address.c_str(),
2183 r.deferrals);
2184 ReschedulePayment(r);
827 continue; 2185 continue;
2186 }
2187
2188 CWalletTx wtx;
2189 wtx.mapValue["comment"] = PayoutComment(r.blockHeight);
2190 // Upstream's structured metadata. This is what makes the wallet
2191 // probe above possible at all -- the old comment-only form could
2192 // not distinguish two rounds paying the same address the same
2193 // amount. Keyed by the hash-derived id, so a reorg cannot make one
2194 // round's transaction answer for another's.
2195 wtx.mapValue["pool_payout_id"] = id;
2196 wtx.mapValue["pool_payout_address"] = r.address;
2197 wtx.mapValue["pool_payout_amount"] = strprintf("%lld", r.amount);
2198
2199 SendMoneyOutcome outcome;
2200 if (SendMoney(sc, r.amount, wtx, outcome)) {
2201 // The wallet holds the transaction before this returns, so a
2202 // crash between here and the ledger write is recoverable: the
2203 // probe above finds it by id+address+amount and does not
2204 // resend. Ledger first, in-flight record second -- the debt is
2205 // described by at least one of them at every instant.
2206 RecordPayoutTxid(r, wtx.GetHash().GetHex());
2207 ForgetPayment(r);
2208 LogPrint("payout", "[payout] paid %s to %s (tx %s)\n",
2209 FormatMoney(r.amount).c_str(), r.address.c_str(),
2210 wtx.GetHash().ToString().substr(0, 12).c_str());
2211 } else {
2212 // Why did it fail? The answer used to decide whether the
2213 // payment eventually got erased, and it was decided by
2214 // guessing from a balance snapshot -- see
2215 // ClassifyPayoutSendFailure for the arithmetic and what it
2216 // cost. The wallet now says what it did, and nothing here
2217 // erases anything.
2218 //
2219 // The lock state is re-read rather than trusted from before
2220 // the attempt: SendMoney takes cs_main and can block for a
2221 // long time, so it may have moved underneath us. The funding
2222 // answer does not need re-reading -- it is a statement about
2223 // the attempt that was actually made.
2224 bool fLocked = IsWalletLocked();
2225 int64 nAfter = GetBalance();
2226
2227 if (ClassifyPayoutSendFailure(outcome.result, fLocked)
2228 == PAYOUT_FAILURE_DEFER) {
2229 // Conditions are temporarily wrong. Wait.
2230 r.deferrals++;
2231 r.retryAfter = GetTime() + 60;
2232 if (r.deferrals == 1 || r.deferrals % 60 == 0)
2233 LogPrint("payout", "[payout] deferring %s to %s -- send"
2234 " failed with %s (%s spendable; the wallet"
2235 " needed a fee of %s for a transaction this"
2236 " size); waiting %d min so far\n",
2237 FormatMoney(r.amount).c_str(), r.address.c_str(),
2238 fLocked ? "the wallet locked" : "too little coin",
2239 FormatMoney(nAfter).c_str(),
2240 FormatMoney(outcome.nFeeRequired).c_str(),
2241 r.deferrals);
2242 ReschedulePayment(r);
2243 } else {
2244 // Funded, unlocked, block on the main chain, and the
2245 // wallet still would not build, sign or accept the
2246 // transaction. That is worth shouting about; it is not
2247 // grounds to keep the money.
2248 //
2249 // There is deliberately no give-up branch. It used to
2250 // forfeit the debt after 1440 of these, which is not a
2251 // decidable state -- it is a count of how long we have
2252 // been asking, and it was reached by misclassifying a
2253 // funding shortfall as a fault. A debt that is never
2254 // written off cannot be written off wrongly. The payment
2255 // stays in flight, keeps its ledger entry, keeps being
2256 // reported in pool_status.json as owed, and keeps
2257 // retrying once a minute until the operator or the wallet
2258 // makes it possible.
2259 r.retries++;
2260 r.retryAfter = GetTime() + 60;
2261 ReschedulePayment(r);
2262 // Every minute at first, then hourly: a fault that is not
2263 // going away must not turn the log into one line a minute
2264 // forever, and must not go quiet either.
2265 if (r.retries <= 5 || r.retries % 60 == 0)
2266 LogPrint("payout", "[payout] FAILED to pay %s to %s"
2267 " (%s spendable, wallet unlocked, block %d on"
2268 " the main chain) -- attempt %d, retrying in"
2269 " 60s; this debt stands until it is paid\n",
2270 FormatMoney(r.amount).c_str(), r.address.c_str(),
2271 FormatMoney(nAfter).c_str(), r.blockHeight,
2272 r.retries);
2273 }
2274 }
828 } 2275 }
829 2276
830 Sleep(500); // check for new work every 500ms 2277 Sleep(500); // check for new work every 500ms
@@ -877,7 +2324,22 @@ static std::string PoolStatusFilePath()
877 return GetAppDir() + "/pool_status.json"; 2324 return GetAppDir() + "/pool_status.json";
878} 2325}
879 2326
880static void WritePoolStatusJsonLocked() 2327// Serialise the operator status file. Caller must hold gStatsMutex (this reads
2328// gStats / gWorkerStats / gPayoutViews); gPoolRoundsMutex is taken and
2329// released inside.
2330//
2331// Split from the write for the same reason as the three ledgers, and for a
2332// worse instance of the same problem. This used to be called from RefreshStats
2333// with gPayoutLedgerMutex held at function scope, and once the ledger writes
2334// gained fsync() that call became two fsyncs -- one on the file, one on its
2335// directory -- performed by the event loop, the thread that owns every miner's
2336// socket, with the mutex the payout thread needs in its hand. Measured on the
2337// build host (ext4 on LVM on SSD): 3.78 ms per publish, every 10 seconds, and
2338// unbounded on a shared virtual disk under load.
2339//
2340// Nothing here needs the ledger mutex at all: the pending-payout rows were
2341// already copied into gPayoutViews under it, further up RefreshStats.
2342static std::string SerializePoolStatusLocked()
881{ 2343{
882 nlohmann::json j; 2344 nlohmann::json j;
883 j["schema"] = "bitflash-pool-status-1"; 2345 j["schema"] = "bitflash-pool-status-1";
@@ -933,10 +2395,19 @@ static void WritePoolStatusJsonLocked()
933 size_t start = gPoolRounds.size() > 10 ? gPoolRounds.size() - 10 : 0; 2395 size_t start = gPoolRounds.size() > 10 ? gPoolRounds.size() - 10 : 0;
934 for (size_t i = start; i < gPoolRounds.size(); i++) { 2396 for (size_t i = start; i < gPoolRounds.size(); i++) {
935 const PoolRoundProof& r = gPoolRounds[i]; 2397 const PoolRoundProof& r = gPoolRounds[i];
2398 // A forfeited recipient is not "unpaid and pending" -- the ledger
2399 // has already dropped the obligation. Counting it as unpaid is
2400 // what made the round proof and the ledger disagree for good.
936 int unpaid = 0; 2401 int unpaid = 0;
937 for (const auto& kv : r.payouts) 2402 int forfeited = 0;
938 if (r.paidTxids.count(kv.first) == 0) 2403 for (const auto& kv : r.payouts) {
2404 if (r.paidTxids.count(kv.first))
2405 continue;
2406 if (r.forfeited.count(kv.first))
2407 forfeited++;
2408 else
939 unpaid++; 2409 unpaid++;
2410 }
940 j["recentRounds"].push_back({ 2411 j["recentRounds"].push_back({
941 {"id", r.id}, 2412 {"id", r.id},
942 {"blockHeight", r.blockHeight}, 2413 {"blockHeight", r.blockHeight},
@@ -945,30 +2416,34 @@ static void WritePoolStatusJsonLocked()
945 {"totalShares", r.totalShares}, 2416 {"totalShares", r.totalShares},
946 {"payouts", (int)r.payouts.size()}, 2417 {"payouts", (int)r.payouts.size()},
947 {"unpaidPayouts", unpaid}, 2418 {"unpaidPayouts", unpaid},
2419 {"forfeitedPayouts", forfeited},
948 {"coinbaseValueBtf", FormatMoney(r.coinbaseValue)}, 2420 {"coinbaseValueBtf", FormatMoney(r.coinbaseValue)},
949 {"minerRewardBtf", FormatMoney(r.minerReward)} 2421 {"minerRewardBtf", FormatMoney(r.minerReward)}
950 }); 2422 });
951 } 2423 }
952 } 2424 }
953 2425
954 std::string path = PoolStatusFilePath(); 2426 return j.dump(2);
955 std::string tmp = path + ".tmp"; 2427}
956 FILE* f = fopen(tmp.c_str(), "w"); 2428
957 if (!f) { 2429// Publish the operator status file. Call with every payout mutex released.
958 LogPrint("pool", "[pool] WARNING: could not write status file %s\n", tmp.c_str()); 2430//
959 return; 2431// Still no version guard and no write mutex, deliberately, and the split does
960 } 2432// not change the argument: RefreshStats remains the only caller of either
961 std::string s = j.dump(2); 2433// half, in production it runs only on the event loop, and it now calls this
962 fwrite(s.c_str(), 1, s.size(), f); 2434// after leaving the scope that serialised -- so the serialise and the write
963 if (fclose(f) != 0) { 2435// stay strictly ordered on one thread and pool_status.json still has a single
964 LogPrint("pool", "[pool] WARNING: could not close status file %s\n", tmp.c_str()); 2436// writer by construction. It shares WriteFileAtomic so the durability rules
965 return; 2437// stay in one place.
966 } 2438//
967#ifdef _WIN32 2439// "In production" is doing real work in that sentence. The payout-ledger
968 remove(path.c_str()); 2440// self-test also calls RefreshStats, from one thread at a time and with the
969#endif 2441// pool not running, which is why it is still a single writer there too. A
970 if (rename(tmp.c_str(), path.c_str()) != 0) 2442// second concurrent caller anywhere would invalidate this paragraph and want
971 LogPrint("pool", "[pool] WARNING: could not replace status file %s\n", path.c_str()); 2443// the same versioned publish the ledgers use.
2444static void PublishPoolStatus(const std::string& blob)
2445{
2446 WriteFileAtomic(PoolStatusFilePath(), blob, "pool status");
972} 2447}
973 2448
974// --------------------------------------------------------------------------- 2449// ---------------------------------------------------------------------------
@@ -1044,6 +2519,45 @@ static void PushReady(btf_socket_t s)
1044 gReadyQueue.push_back(s); 2519 gReadyQueue.push_back(s);
1045} 2520}
1046 2521
2522// Loopback stratum listener: lets a miner running on the SAME machine reach
2523// this pool over plain TCP (127.0.0.1:8334), skipping the .btf rendezvous
2524// entirely. The rendezvous relay only allows one service registration per
2525// pubkey, so the pool service and the node's own peer-accept fight over the
2526// node identity at the relay -- loopback avoids that fight for local miners.
2527static void LocalAcceptFn(void*)
2528{
2529 SOCKET ls = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
2530 if (ls == INVALID_SOCKET)
2531 return;
2532 int yes = 1;
2533 setsockopt(ls, SOL_SOCKET, SO_REUSEADDR, (char*)&yes, sizeof(yes));
2534 struct sockaddr_in addr;
2535 memset(&addr, 0, sizeof(addr));
2536 addr.sin_family = AF_INET;
2537 addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
2538 addr.sin_port = htons(8334);
2539 if (bind(ls, (struct sockaddr*)&addr, sizeof(addr)) != 0 || listen(ls, 16) != 0)
2540 {
2541 LogPrint("pool", "[pool] loopback stratum bind/listen failed on 127.0.0.1:8334\n");
2542 closesocket(ls);
2543 return;
2544 }
2545 LogPrint("pool", "[pool] loopback stratum listening on 127.0.0.1:8334\n");
2546 while (!fShutdown && gPoolRunning)
2547 {
2548 SOCKET c = accept(ls, NULL, NULL);
2549 if (c == INVALID_SOCKET)
2550 {
2551 if (fShutdown || !gPoolRunning)
2552 break;
2553 continue;
2554 }
2555 LogPrint("pool", "[pool] miner arrived via loopback -- pushing to event loop\n");
2556 PushReady((btf_socket_t)c);
2557 }
2558 closesocket(ls);
2559}
2560
1047// --------------------------------------------------------------------------- 2561// ---------------------------------------------------------------------------
1048// I/O helpers -- event loop thread only 2562// I/O helpers -- event loop thread only
1049// --------------------------------------------------------------------------- 2563// ---------------------------------------------------------------------------
@@ -1480,7 +2994,11 @@ static bool HandleLine(Miner* m, const std::string& rawLine,
1480 if (accepted) { 2994 if (accepted) {
1481 // Snapshot height before RebuildJob() (which increments it) 2995 // Snapshot height before RebuildJob() (which increments it)
1482 int foundHeight = gCurrentJob.height; 2996 int foundHeight = gCurrentJob.height;
1483 QueuePayouts(foundHeight, b.GetHash().GetHex(), b.vtx[0].GetHash().GetHex(), 2997 // b is the block we submitted, untouched by ProcessBlock (which
2998 // took a copy), so its hash is what the payout must be keyed on.
2999 // Passed as a uint256, not hex: the ledger key must stay a real
3000 // hash. The remaining arguments are upstream's round-proof data.
3001 QueuePayouts(foundHeight, b.GetHash(), b.vtx[0].GetHash().GetHex(),
1484 b.vtx[0].vout[0].nValue, m->address, 3002 b.vtx[0].vout[0].nValue, m->address,
1485 roundShareCount, roundShareTotal); 3003 roundShareCount, roundShareTotal);
1486 // Reset round shares 3004 // Reset round shares
@@ -1508,6 +3026,1352 @@ static bool HandleLine(Miner* m, const std::string& rawLine,
1508 return true; 3026 return true;
1509} 3027}
1510 3028
3029// ---------------------------------------------------------------------------
3030// payout-ledger self-test
3031//
3032// Regression cover for the payout defects fixed in this series. Everything
3033// here runs against a scratch data directory and an in-memory wallet map; it
3034// never opens wallet.dat, never touches a real datadir, and never sends
3035// anything.
3036// ---------------------------------------------------------------------------
3037
3038static bool PayoutTestReadFile(const std::string& path, std::string& out)
3039{
3040 FILE* f = fopen(path.c_str(), "r");
3041 if (!f) return false;
3042 fseek(f, 0, SEEK_END);
3043 long sz = ftell(f);
3044 fseek(f, 0, SEEK_SET);
3045 out.assign((size_t)(sz > 0 ? sz : 0), '\0');
3046 if (sz > 0 && fread(&out[0], 1, (size_t)sz, f) != (size_t)sz) {
3047 fclose(f);
3048 return false;
3049 }
3050 fclose(f);
3051 return true;
3052}
3053
3054// Delete the scratch data directory and everything in it. A sweep and not a
3055// list of known names: LogPrint binds debug.log to whatever GetAppDir() is
3056// when it first fires, so the test's own log lands in here too, and a
3057// directory left behind in src/ after `make tests` is litter.
3058static void PayoutTestRemoveDir(const std::string& dir)
3059{
3060#ifdef _WIN32
3061 WIN32_FIND_DATAA findData;
3062 HANDLE hFind = FindFirstFileA((dir + "\\*").c_str(), &findData);
3063 if (hFind != INVALID_HANDLE_VALUE) {
3064 do {
3065 std::string name = findData.cFileName;
3066 if (name == "." || name == "..") continue;
3067 DeleteFileA((dir + "\\" + name).c_str());
3068 } while (FindNextFileA(hFind, &findData));
3069 FindClose(hFind);
3070 }
3071 RemoveDirectoryA(dir.c_str());
3072#else
3073 // One level of recursion, because CreateTransaction() opens a Berkeley DB
3074 // environment and BDB puts its logs in a "database" subdirectory. Without
3075 // this the scratch directory cannot be removed and the test leaves one
3076 // behind on every run, which is what d5b3503 was for.
3077 DIR* d = opendir(dir.c_str());
3078 if (d) {
3079 struct dirent* ent;
3080 while ((ent = readdir(d)) != NULL) {
3081 std::string name = ent->d_name;
3082 if (name == "." || name == "..") continue;
3083 const std::string path = dir + "/" + name;
3084 if (unlink(path.c_str()) != 0)
3085 PayoutTestRemoveDir(path);
3086 }
3087 closedir(d);
3088 }
3089 rmdir(dir.c_str());
3090#endif
3091}
3092
3093static bool PayoutTestWriteFile(const std::string& path, const std::string& data)
3094{
3095 FILE* f = fopen(path.c_str(), "w");
3096 if (!f) return false;
3097 bool ok = data.empty() || fwrite(data.c_str(), 1, data.size(), f) == data.size();
3098 return fclose(f) == 0 && ok;
3099}
3100
3101// Wipe every piece of payout state this test touches, so each section starts
3102// from the same place and a "restart" can be simulated by reloading from disk.
3103static void PayoutTestResetState()
3104{
3105 {
3106 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
3107 gPendingPayouts.clear();
3108 }
3109 {
3110 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
3111 gPaying.clear();
3112 gQueuedPayouts.clear();
3113 }
3114 {
3115 std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
3116 gPoolRounds.clear();
3117 }
3118 {
3119 std::lock_guard<std::mutex> lk(gClaimedTxidMutex);
3120 gLegacyClaimedTxids.clear();
3121 }
3122 {
3123 // A quarantine is meant to last the life of the process; each section
3124 // here is a fresh process as far as the code under test is concerned.
3125 std::lock_guard<std::mutex> lk(gRefusedFilesMutex);
3126 gRefusedFiles.clear();
3127 }
3128}
3129
3130// A BTF address whose hash160 is entirely under the test's control, so the
3131// output script the probe compares against can be built without a wallet.
3132static std::string PayoutTestAddress(unsigned char seed)
3133{
3134 uint160 h = 0;
3135 for (int i = 0; i < 20; i++)
3136 ((unsigned char*)&h)[i] = (unsigned char)(seed + i);
3137 return Hash160ToAddress(h);
3138}
3139
3140static CScript PayoutTestScriptFor(const std::string& address)
3141{
3142 uint160 h160;
3143 CScript sc;
3144 if (AddressToHash160(address, h160))
3145 sc << OP_DUP << OP_HASH160 << h160 << OP_EQUALVERIFY << OP_CHECKSIG;
3146 return sc;
3147}
3148
3149// Block index entries fabricated by PayoutTestAddWalletTx, so they can be
3150// freed and unhooked when the test finishes. gPayoutTestChainTip is the end of
3151// that chain: blocks are linked onto it rather than each replacing the tip,
3152// because IsInMainChain() is "pnext is set, or I am the tip".
3153static std::vector<CBlockIndex*> gPayoutTestBlockIndexes;
3154static CBlockIndex* gPayoutTestChainTip = NULL;
3155
3156static void PayoutTestClearBlockIndex()
3157{
3158 CRITICAL_BLOCK(cs_main)
3159 {
3160 for (CBlockIndex* p : gPayoutTestBlockIndexes) {
3161 for (std::map<uint256, CBlockIndex*>::iterator it = mapBlockIndex.begin();
3162 it != mapBlockIndex.end(); ) {
3163 if (it->second == p) mapBlockIndex.erase(it++);
3164 else ++it;
3165 }
3166 delete p;
3167 }
3168 gPayoutTestBlockIndexes.clear();
3169 gPayoutTestChainTip = NULL;
3170 pindexBest = NULL;
3171 }
3172}
3173
3174// Put a transaction in the wallet map that looks exactly like one this pool
3175// sent. `nonce` only exists to keep the hashes distinct.
3176//
3177// fConfirmed decides what GetDepthInMainChain() will answer for it, which is
3178// what the legacy probe now gates on. A confirmed one is given a block index
3179// entry of its own: an empty merkle branch hashes to the transaction itself,
3180// so a block whose merkle root IS that hash accepts it at index 0, and
3181// pointing pindexBest at that block puts it in the main chain one deep. An
3182// unconfirmed one keeps hashBlock 0, which is depth 0 -- the shape a payout
3183// left behind by a reorg has.
3184static std::string PayoutTestAddWalletTx(const std::string& comment,
3185 const std::string& address,
3186 int64 amount,
3187 int nonce,
3188 const std::string& structuredId,
3189 bool fConfirmed = true)
3190{
3191 CWalletTx wtx;
3192 wtx.vin.resize(1);
3193 wtx.vin[0].prevout.SetNull();
3194 wtx.vin[0].scriptSig << nonce;
3195 wtx.vout.resize(1);
3196 wtx.vout[0].nValue = amount;
3197 wtx.vout[0].scriptPubKey = PayoutTestScriptFor(address);
3198 if (!comment.empty())
3199 wtx.mapValue["comment"] = comment;
3200 if (!structuredId.empty()) {
3201 wtx.mapValue["pool_payout_id"] = structuredId;
3202 wtx.mapValue["pool_payout_address"] = address;
3203 wtx.mapValue["pool_payout_amount"] = strprintf("%lld", amount);
3204 }
3205 uint256 hash = wtx.GetHash();
3206 if (fConfirmed) {
3207 // A block hash distinct from the transaction's own.
3208 uint256 blockHash = hash;
3209 ((unsigned char*)&blockHash)[0] ^= 0xff;
3210 CRITICAL_BLOCK(cs_main)
3211 {
3212 CBlockIndex* pindex = new CBlockIndex();
3213 pindex->hashMerkleRoot = hash;
3214 // Linked onto the end of the fabricated chain rather than simply
3215 // becoming the new tip: IsInMainChain() is "pnext is set, or I am
3216 // the tip", so a second confirmed transaction would otherwise take
3217 // the first one out of the chain again.
3218 pindex->pprev = gPayoutTestChainTip;
3219 pindex->nHeight = gPayoutTestChainTip ? gPayoutTestChainTip->nHeight + 1 : 1;
3220 if (gPayoutTestChainTip)
3221 gPayoutTestChainTip->pnext = pindex;
3222 gPayoutTestChainTip = pindex;
3223 std::map<uint256, CBlockIndex*>::iterator mi =
3224 mapBlockIndex.insert(std::make_pair(blockHash, pindex)).first;
3225 pindex->phashBlock = &mi->first;
3226 pindexBest = pindex;
3227 gPayoutTestBlockIndexes.push_back(pindex);
3228 }
3229 wtx.hashBlock = blockHash;
3230 wtx.nIndex = 0;
3231 }
3232 CRITICAL_BLOCK(cs_mapWallet)
3233 mapWallet[hash] = wtx;
3234 return hash.GetHex();
3235}
3236
3237// A real key, in the keystore, so a test wallet's outputs are IsMine() and can
3238// actually be signed -- both are needed for CreateTransaction to build the
3239// large transaction the fee schedule is being tested against. Inserted
3240// straight into the key maps rather than through AddKey(), which would also
3241// write it to a wallet.dat.
3242static std::string PayoutTestAddSpendableKey()
3243{
3244 CKey key;
3245 key.MakeNewKey();
3246 std::vector<unsigned char> vchPubKey = key.GetPubKey();
3247 CRITICAL_BLOCK(cs_mapKeys)
3248 {
3249 mapKeys[vchPubKey] = key.GetPrivKey();
3250 mapPubKeys[Hash160(vchPubKey)] = vchPubKey;
3251 }
3252 return Hash160ToAddress(Hash160(vchPubKey));
3253}
3254
3255// Give the wallet `n` spendable outputs of `each`, one transaction apiece --
3256// which is the granularity SelectCoins works at, and therefore the number of
3257// inputs the transaction spending them will carry.
3258//
3259// Deliberately not coinbases: a null prevout would make CWalletTx::GetCredit
3260// return 0 until COINBASE_SPENDABLE_DEPTH, and there is no chain here.
3261static void PayoutTestFundWallet(int n, int64 each, const std::string& address)
3262{
3263 const CScript sc = PayoutTestScriptFor(address);
3264 CRITICAL_BLOCK(cs_mapWallet)
3265 {
3266 for (int i = 0; i < n; i++) {
3267 CWalletTx wtx;
3268 wtx.vin.resize(1);
3269 wtx.vin[0].prevout.hash = (uint64)(i + 1);
3270 wtx.vin[0].prevout.n = 0;
3271 wtx.vout.resize(1);
3272 wtx.vout[0].nValue = each;
3273 wtx.vout[0].scriptPubKey = sc;
3274 mapWallet[wtx.GetHash()] = wtx;
3275 }
3276 }
3277}
3278
3279// Defined below with the event loop; the status-file publication test drives
3280// it directly.
3281static void RefreshStats(int blocksFoundThisSession, uint64 roundShareTotal);
3282
3283int RunPayoutLedgerSelfTest()
3284{
3285 int nFail = 0;
3286 auto check = [&](bool ok, const char* msg) {
3287 fprintf(stdout, " %s %s\n", ok ? "ok" : "FAIL", msg);
3288 if (!ok) nFail++;
3289 };
3290
3291 fprintf(stdout, "payout-ledger self-test\n");
3292
3293 const std::string savedDataDir = strSetDataDir;
3294 const int savedBestHeight = nBestHeight;
3295
3296 // Scratch data directory in the current working directory. GetAppDir()
3297 // follows strSetDataDir, so this is what PayoutFilePath() and
3298 // PayingFilePath() resolve against for the rest of the test.
3299#ifdef _WIN32
3300 unsigned long pid = (unsigned long)GetCurrentProcessId();
3301#else
3302 unsigned long pid = (unsigned long)getpid();
3303#endif
3304 // Absolute, and the process moves into it.
3305 //
3306 // Both are needed by the CreateTransaction section: Berkeley DB resolves
3307 // its log directory relative to the environment home, so a relative
3308 // datadir makes dbenv.open() fail with DB_RUNRECOVERY; and BDB opens
3309 // db.log relative to the working directory, which is src/ during `make
3310 // tests`. Moving in puts it inside the scratch directory, where it dies
3311 // with everything else.
3312 std::string cwd = ".";
3313 {
3314#ifdef _WIN32
3315 char buf[MAX_PATH];
3316 if (GetCurrentDirectoryA(MAX_PATH, buf) > 0) cwd = buf;
3317#else
3318 char buf[4096];
3319 if (getcwd(buf, sizeof(buf)) != NULL) cwd = buf;
3320#endif
3321 }
3322 std::string dir = cwd + "/" +
3323 strprintf("payout_selftest_%d_%lu", (int)GetTime(), pid);
3324#ifdef _WIN32
3325 CreateDirectoryA(dir.c_str(), NULL);
3326 SetCurrentDirectoryA(dir.c_str());
3327#else
3328 mkdir(dir.c_str(), 0700);
3329 if (chdir(dir.c_str()) != 0) { /* stay put; the paths are absolute anyway */ }
3330#endif
3331 strSetDataDir = dir;
3332 fprintf(stdout, " scratch datadir: %s\n", dir.c_str());
3333
3334 const std::string ledgerPath = PayoutFilePath();
3335 const std::string payingPath = PayingFilePath();
3336 const std::string addrA = PayoutTestAddress(0x11);
3337 const std::string addrB = PayoutTestAddress(0x55);
3338
3339 // -- 1. A restart must not duplicate an in-flight record -----------------
3340 //
3341 // The ledger entry is grandfathered (zero hash) on purpose: it keeps
3342 // PayoutBlockStillOnChain out of the picture, which needs a block index,
3343 // and it exercises the height fallback at the same time.
3344 {
3345 PayoutTestResetState();
3346 remove(ledgerPath.c_str());
3347 remove(payingPath.c_str());
3348
3349 PendingPayout pp;
3350 pp.blockHeight = 4242;
3351 pp.blockHash = 0;
3352 pp.matureAtHeight = 4242;
3353 pp.amounts[addrA] = 10 * COIN;
3354 pp.amounts[addrB] = 20 * COIN;
3355 VersionedBlob snap;
3356 {
3357 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
3358 gPendingPayouts.push_back(pp);
3359 snap = SerializePendingPayoutsLocked();
3360 }
3361 PublishPendingPayouts(snap);
3362
3363 nBestHeight = 5000;
3364 FlushMaturePayouts();
3365 size_t afterFirstFlush = 0;
3366 {
3367 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
3368 afterFirstFlush = gPaying.size();
3369 }
3370 check(afterFirstFlush == 2,
3371 "a matured round explodes into one record per unpaid recipient");
3372
3373 // Restart: drop everything in memory and come back from the files.
3374 for (int restart = 1; restart <= 2; restart++) {
3375 PayoutTestResetState();
3376 LoadPendingPayouts();
3377 LoadPaying();
3378 ReconcilePayouts();
3379
3380 size_t queued = 0;
3381 {
3382 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
3383 queued = gQueuedPayouts.size();
3384 }
3385 if (restart == 1)
3386 check(queued == 1,
3387 "startup rebuilds the exploded-round set from the restored"
3388 " in-flight list");
3389
3390 FlushMaturePayouts();
3391 size_t n = 0;
3392 {
3393 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
3394 n = gPaying.size();
3395 }
3396 check(n == 2, restart == 1
3397 ? "a restart does not duplicate an in-flight record"
3398 : "a second restart does not duplicate one either");
3399 }
3400
3401 // The other half of the guard: even with the round-level dedupe set
3402 // deliberately emptied, the per-recipient check against gPaying must
3403 // stop a second standing order for the same money.
3404 {
3405 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
3406 gQueuedPayouts.clear();
3407 }
3408 FlushMaturePayouts();
3409 size_t n = 0;
3410 {
3411 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
3412 n = gPaying.size();
3413 }
3414 check(n == 2,
3415 "the flush refuses a second record for a (round, address) already"
3416 " in flight even with the round set lost");
3417 }
3418
3419 // -- 2. Duplicates already on disk are collapsed as they are read --------
3420 {
3421 PayoutTestResetState();
3422 nlohmann::json arr = nlohmann::json::array();
3423 for (int i = 0; i < 2; i++) {
3424 nlohmann::json o;
3425 o["blockHeight"] = 4242;
3426 o["blockHash"] = uint256(0).ToString();
3427 o["matureAtHeight"] = 4242;
3428 o["address"] = addrA;
3429 o["amount"] = (int64)(10 * COIN);
3430 o["retries"] = i == 0 ? 7 : 0;
3431 o["deferrals"] = i == 0 ? 0 : 3;
3432 arr.push_back(o);
3433 }
3434 check(PayoutTestWriteFile(payingPath, arr.dump(2)),
3435 "wrote a paying file containing the duplicate the old build made");
3436 LoadPaying();
3437 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
3438 check(gPaying.size() == 1,
3439 "loading collapses a duplicate in-flight record from disk");
3440 check(gPaying.size() == 1 && gPaying[0].retries == 7 && gPaying[0].deferrals == 3,
3441 "collapsing merges the attempt counts upwards, not resetting them");
3442 }
3443
3444 // -- 3. State updates address a slot, not a value -----------------------
3445 //
3446 // Build the exact pathology the old code produced -- two records equal in
3447 // every field -- and prove the update lands on the one asked for. Under
3448 // value matching the first record absorbed everything and the second was
3449 // untouchable.
3450 {
3451 PayoutTestResetState();
3452 PayingRecord a;
3453 a.blockHeight = 4242; a.blockHash = 0; a.matureAtHeight = 4242;
3454 a.address = addrA; a.amount = 10 * COIN;
3455 PayingRecord b = a;
3456 {
3457 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
3458 a.id = ++gPayingNextId;
3459 b.id = ++gPayingNextId;
3460 gPaying.push_back(a);
3461 gPaying.push_back(b);
3462 }
3463 PayingRecord upd = b;
3464 upd.retries = 5;
3465 upd.retryAfter = 1234567;
3466 ReschedulePayment(upd);
3467 {
3468 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
3469 check(gPaying.size() == 2 &&
3470 gPaying[0].retries == 0 && gPaying[1].retries == 5,
3471 "rescheduling updates the record it was given, not the first"
3472 " one that looks like it");
3473 }
3474 ForgetPayment(a);
3475 {
3476 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
3477 check(gPaying.size() == 1 && gPaying[0].id == b.id,
3478 "forgetting erases the record it was given, not the first"
3479 " one that looks like it");
3480 }
3481 }
3482
3483 // -- 4. retryAfter survives a restart ------------------------------------
3484 {
3485 PayoutTestResetState();
3486 remove(payingPath.c_str());
3487 PayingRecord r;
3488 r.blockHeight = 77; r.blockHash = 0; r.matureAtHeight = 77;
3489 r.address = addrA; r.amount = 3 * COIN;
3490 r.retryAfter = 1700000000LL;
3491 r.deferrals = 9;
3492 VersionedBlob snap;
3493 {
3494 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
3495 r.id = ++gPayingNextId;
3496 gPaying.push_back(r);
3497 snap = SerializePayingLocked();
3498 }
3499 PublishPaying(snap);
3500 PayoutTestResetState();
3501 LoadPaying();
3502 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
3503 check(gPaying.size() == 1 && gPaying[0].retryAfter == 1700000000LL &&
3504 gPaying[0].deferrals == 9,
3505 "the retry clock that is actually used survives a restart");
3506 }
3507
3508 // -- 5. The ledger file cannot be published from a stale snapshot --------
3509 {
3510 PayoutTestResetState();
3511 remove(ledgerPath.c_str());
3512
3513 PendingPayout one;
3514 one.blockHeight = 100; one.blockHash = 0; one.matureAtHeight = 100;
3515 one.amounts[addrA] = 1 * COIN;
3516 VersionedBlob older, newer;
3517 {
3518 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
3519 gPendingPayouts.push_back(one);
3520 older = SerializePendingPayoutsLocked();
3521 PendingPayout two;
3522 two.blockHeight = 200; two.blockHash = 0; two.matureAtHeight = 200;
3523 two.amounts[addrB] = 2 * COIN;
3524 gPendingPayouts.push_back(two);
3525 newer = SerializePendingPayoutsLocked();
3526 }
3527 check(newer.seq > older.seq,
3528 "a later serialisation carries a strictly higher version");
3529
3530 // The order the race produced: the payout thread's older image
3531 // arriving after the event loop's newer one.
3532 PublishPendingPayouts(newer);
3533 PublishPendingPayouts(older);
3534 std::string onDisk;
3535 check(PayoutTestReadFile(ledgerPath, onDisk) && onDisk == newer.blob,
3536 "an older snapshot cannot overwrite a newer published ledger");
3537
3538 // And the ordinary direction still works.
3539 VersionedBlob newest;
3540 {
3541 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
3542 gPendingPayouts.clear();
3543 newest = SerializePendingPayoutsLocked();
3544 }
3545 PublishPendingPayouts(newest);
3546 check(PayoutTestReadFile(ledgerPath, onDisk) && onDisk == newest.blob,
3547 "a newer snapshot does replace the published ledger");
3548 }
3549
3550 // -- 6. Concurrent publishers cannot interleave a file -------------------
3551 //
3552 // Two threads opening and renaming the same "<path>.tmp" over the same
3553 // target was the original defect. This test was mutation-tested and did
3554 // not catch it: with the file mutex removed the suite still went green 2
3555 // runs in 5, and the "never leave it unparseable" assertion fired 0 times
3556 // in 5. Two reasons, both fixed here.
3557 //
3558 // First, every image had the same structure and near enough the same
3559 // length, so an interleaving of two of them was still valid JSON. The
3560 // images now alternate between ONE entry and FIVE THOUSAND, so a write
3561 // that lands on top of another leaves either a truncated array or a small
3562 // image followed by the tail of a large one -- neither of which parses.
3563 //
3564 // Second, and worse, almost nothing was being written. Each thread walked
3565 // a permutation of the images, and PublishVersioned drops any blob that is
3566 // not newer than what is on disk, so only the record highs in each
3567 // permutation ever reached the disk -- on the order of ten writes in
3568 // total, four threads, no contention to speak of. Every publish here is
3569 // stamped from a shared counter at the moment it is made, so every one of
3570 // them is newer than the last and every one of them writes.
3571 //
3572 // And the file is watched while the writers run, not only inspected after
3573 // they finish. "The published file is never observed unparseable" is the
3574 // actual invariant -- rename(2) is atomic, so a reader can only ever see a
3575 // completely written image -- and checking it once at the end samples the
3576 // one moment when a torn file is least likely to still be there.
3577 {
3578 const std::string racePath = dir + "/publish_race_selftest.json";
3579 std::mutex raceMutex;
3580 uint64 racePublished = 0;
3581
3582 std::string tiny, huge;
3583 {
3584 nlohmann::json one = nlohmann::json::array();
3585 nlohmann::json o;
3586 o["image"] = 0;
3587 o["entry"] = 0;
3588 o["address"] = addrA;
3589 o["amount"] = (int64)COIN;
3590 one.push_back(o);
3591 tiny = one.dump(2);
3592
3593 nlohmann::json many = nlohmann::json::array();
3594 for (int k = 0; k < 5000; k++) {
3595 nlohmann::json e;
3596 e["image"] = 1;
3597 e["entry"] = k;
3598 e["address"] = addrA;
3599 e["amount"] = (int64)(k + 1) * COIN;
3600 many.push_back(e);
3601 }
3602 huge = many.dump(2);
3603 }
3604
3605 const int kThreads = 4;
3606 const int kPerThread = 100;
3607 const uint64 kTotal = (uint64)kThreads * kPerThread;
3608
3609 std::atomic<uint64> nextSeq(0);
3610 std::atomic<bool> racing(true);
3611 std::atomic<int> tornReads(0);
3612 std::atomic<int> goodReads(0);
3613
3614 std::thread watcher([&]() {
3615 bool seenOnce = false;
3616 while (racing) {
3617 std::string s;
3618 if (!PayoutTestReadFile(racePath, s)) {
3619 // Only meaningful once the file exists at all.
3620 if (seenOnce) tornReads++;
3621 continue;
3622 }
3623 seenOnce = true;
3624 try {
3625 if (nlohmann::json::parse(s).is_array()) goodReads++;
3626 else tornReads++;
3627 } catch (...) {
3628 tornReads++;
3629 }
3630 }
3631 });
3632
3633 std::vector<std::thread> threads;
3634 for (int t = 0; t < kThreads; t++) {
3635 threads.push_back(std::thread([&]() {
3636 for (int i = 0; i < kPerThread; i++) {
3637 VersionedBlob v;
3638 v.seq = ++nextSeq;
3639 v.blob = (v.seq & 1) ? tiny : huge;
3640 PublishVersioned(racePath, v, raceMutex, racePublished,
3641 "publish race selftest");
3642 }
3643 }));
3644 }
3645 for (std::thread& th : threads)
3646 th.join();
3647 racing = false;
3648 watcher.join();
3649
3650 check(goodReads > 0,
3651 "the race test actually observed the file while it was being"
3652 " written");
3653 check(tornReads == 0,
3654 "four threads publishing the same file never leave it"
3655 " unparseable, at any instant while they are doing it");
3656
3657 // The highest sequence number issued is the one the file must settle
3658 // on: PublishVersioned refuses anything older, so whichever order the
3659 // writes complete in, that image is the last one that can land.
3660 const std::string& expected = (kTotal & 1) ? tiny : huge;
3661 std::string onDisk;
3662 bool read = PayoutTestReadFile(racePath, onDisk);
3663 check(read && onDisk == expected,
3664 "the published file settles on the highest version, whatever the"
3665 " completion order");
3666 remove(racePath.c_str());
3667 remove((racePath + ".tmp").c_str());
3668 }
3669
3670 // -- 7. Grandfathered zero-hash rounds do not answer for each other ------
3671 {
3672 uint256 h1 = 0, h2 = 0;
3673 h1.SetHex("00000000000000000000000000000000000000000000000000000000000000aa");
3674 h2.SetHex("00000000000000000000000000000000000000000000000000000000000000bb");
3675 check(!SamePayoutRound(100, 0, 200, 0),
3676 "two zero-hash entries at different heights are different rounds");
3677 check(SamePayoutRound(100, 0, 100, 0),
3678 "two zero-hash entries at the same height are the same round");
3679 check(SamePayoutRound(100, h1, 200, h1),
3680 "a real hash identifies the round on its own");
3681 check(!SamePayoutRound(100, h1, 100, h2),
3682 "different hashes at one height are different rounds");
3683 check(!SamePayoutRound(100, h1, 100, 0),
3684 "a grandfathered entry never matches a hash-keyed one");
3685 check(!PayoutRoundIdentifiable(0, 0) && PayoutRoundIdentifiable(0, h1) &&
3686 PayoutRoundIdentifiable(7, 0),
3687 "an entry with neither hash nor height is not adjudicable");
3688
3689 // What actually makes a grandfathered same-height pair safe, since the
3690 // comment on FindRecordedPoolPayout used to claim it was the orphan
3691 // check. It is not: the orphan check passes every zero-hash entry, and
3692 // every pre-upgrade entry has a zero hash.
3693 check(PayoutBlockStillOnChain(0),
3694 "the orphan check passes every grandfathered entry, so it cannot"
3695 " be what voids a losing twin");
3696 PayoutTestResetState();
3697 remove(ledgerPath.c_str());
3698 remove(payingPath.c_str());
3699 {
3700 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
3701 PendingPayout twin;
3702 twin.blockHeight = 300; twin.blockHash = 0; twin.matureAtHeight = 300;
3703 twin.amounts[addrA] = 1 * COIN;
3704 gPendingPayouts.push_back(twin);
3705 gPendingPayouts.push_back(twin); // the losing side of a reorg
3706 }
3707 nBestHeight = 5000;
3708 FlushMaturePayouts();
3709 {
3710 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
3711 check(gPaying.size() == 1,
3712 "two grandfathered rounds at one height are one round, so the"
3713 " pair presents as one debt and is paid at most once");
3714 }
3715
3716 // End to end: recording a payment against one grandfathered round must
3717 // not settle the other's debt.
3718 PayoutTestResetState();
3719 remove(ledgerPath.c_str());
3720 {
3721 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
3722 PendingPayout first;
3723 first.blockHeight = 100; first.blockHash = 0; first.matureAtHeight = 100;
3724 first.amounts[addrA] = 1 * COIN;
3725 PendingPayout second;
3726 second.blockHeight = 200; second.blockHash = 0; second.matureAtHeight = 200;
3727 second.amounts[addrA] = 1 * COIN;
3728 gPendingPayouts.push_back(first);
3729 gPendingPayouts.push_back(second);
3730 }
3731 PayingRecord r;
3732 r.blockHeight = 200; r.blockHash = 0; r.matureAtHeight = 200;
3733 r.address = addrA; r.amount = 1 * COIN;
3734 RecordPayoutTxid(r, "txid-for-the-second-round");
3735 {
3736 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
3737 // The second round was fully paid and retired; the first is
3738 // untouched and still owed.
3739 check(gPendingPayouts.size() == 1 &&
3740 gPendingPayouts[0].blockHeight == 100 &&
3741 gPendingPayouts[0].paidTxids.empty(),
3742 "a txid recorded against one zero-hash round leaves the other"
3743 " still owed");
3744 }
3745 }
3746
3747 // -- 8. A payment with only a legacy comment is recognised as paid -------
3748 {
3749 PayoutTestResetState();
3750 const int height = 8080;
3751 const int64 amount = 7 * COIN;
3752 const std::string id = PayoutIdForBlock(height, 0);
3753 const std::string comment = PayoutComment(height);
3754 std::set<std::string> none;
3755 std::string found;
3756
3757 check(FindRecordedPoolPayout(id, height, addrA, amount, none, found)
3758 == PAYOUT_PROBE_NONE,
3759 "an empty wallet is not evidence that anything was paid");
3760
3761 const std::string legacyTxid =
3762 PayoutTestAddWalletTx(comment, addrA, amount, 1, "");
3763 check(FindRecordedPoolPayout(id, height, addrA, amount, none, found)
3764 == PAYOUT_PROBE_LEGACY && found == legacyTxid,
3765 "a payment carrying only a pre-upgrade comment is recognised as"
3766 " already paid");
3767
3768 check(FindRecordedPoolPayout(id, height, addrB, amount, none, found)
3769 == PAYOUT_PROBE_NONE,
3770 "the legacy match requires the output to pay THIS recipient");
3771 check(FindRecordedPoolPayout(id, height, addrA, amount + COIN, none, found)
3772 == PAYOUT_PROBE_NONE,
3773 "the legacy match requires the output to pay exactly this amount");
3774 check(FindRecordedPoolPayout(id, height + 1, addrA, amount, none, found)
3775 == PAYOUT_PROBE_NONE,
3776 "the legacy match requires the comment for this block height");
3777
3778 std::set<std::string> spent;
3779 spent.insert(legacyTxid);
3780 check(FindRecordedPoolPayout(id, height, addrA, amount, spent, found)
3781 == PAYOUT_PROBE_NONE,
3782 "one legacy transaction cannot settle a second debt");
3783
3784 // A transaction written by THIS build carries pool_payout_id, and must
3785 // never be offered as legacy evidence for a different round -- that is
3786 // exactly what the structured metadata exists to prevent.
3787 const std::string otherId = PayoutIdForBlock(height, 0) + "-other";
3788 PayoutTestAddWalletTx(comment, addrB, amount, 2, otherId);
3789 check(FindRecordedPoolPayout(id, height, addrB, amount, none, found)
3790 == PAYOUT_PROBE_NONE,
3791 "another round's structured payment is not legacy evidence for"
3792 " this one");
3793
3794 // The structured path still wins where it applies.
3795 const std::string structuredTxid =
3796 PayoutTestAddWalletTx(comment, addrB, amount, 3, id);
3797 check(FindRecordedPoolPayout(id, height, addrB, amount, none, found)
3798 == PAYOUT_PROBE_STRUCTURED && found == structuredTxid,
3799 "structured payout metadata is matched in preference to a comment");
3800
3801 // A comment-matching transaction that never made it into the chain is
3802 // evidence of NON-payment. This tree has seen a 280-block reorg; a
3803 // pre-upgrade payout invalidated by one is still in mapWallet with its
3804 // comment intact, and the first boot after the upgrade is exactly when
3805 // it would be matched.
3806 CRITICAL_BLOCK(cs_mapWallet)
3807 mapWallet.clear();
3808 PayoutTestClearBlockIndex();
3809 const std::string addrC = PayoutTestAddress(0x33);
3810 PayoutTestAddWalletTx(comment, addrC, amount, 4, "", false);
3811 check(FindRecordedPoolPayout(id, height, addrC, amount, none, found)
3812 == PAYOUT_PROBE_NONE,
3813 "a pre-upgrade payout that never confirmed is not read as"
3814 " payment");
3815
3816 // The same transaction, confirmed, is.
3817 CRITICAL_BLOCK(cs_mapWallet)
3818 mapWallet.clear();
3819 const std::string confirmedTxid =
3820 PayoutTestAddWalletTx(comment, addrC, amount, 4, "", true);
3821 check(FindRecordedPoolPayout(id, height, addrC, amount, none, found)
3822 == PAYOUT_PROBE_LEGACY && found == confirmedTxid,
3823 "the same payout, confirmed, is");
3824
3825 // And an unconfirmed one does not shadow a confirmed one that would
3826 // have matched: the probe used to stop at the first comment match.
3827 const std::string alsoConfirmed =
3828 PayoutTestAddWalletTx(comment, addrC, amount, 6, "", true);
3829 PayoutTestAddWalletTx(comment, addrC, amount, 5, "", false);
3830 std::set<std::string> spentFirst;
3831 spentFirst.insert(confirmedTxid);
3832 check(FindRecordedPoolPayout(id, height, addrC, amount, spentFirst, found)
3833 == PAYOUT_PROBE_LEGACY && found == alsoConfirmed,
3834 "an unconfirmed transaction does not hide a confirmed one behind"
3835 " it");
3836
3837 CRITICAL_BLOCK(cs_mapWallet)
3838 mapWallet.clear();
3839 PayoutTestClearBlockIndex();
3840 }
3841
3842 // -- 9. On-disk compatibility with the live production ledger ------------
3843 {
3844 PayoutTestResetState();
3845 // Exactly the shape of the file on the pool node: amounts, blockHash,
3846 // blockHeight, matureAtHeight, and nothing else.
3847 const char* live =
3848 "[{\"amounts\":{\"%s\":1234500000},"
3849 "\"blockHash\":\"00000000000000000000000000000000000000000000000000000000000000cd\","
3850 "\"blockHeight\":1000,"
3851 "\"matureAtHeight\":1120}]";
3852 check(PayoutTestWriteFile(ledgerPath, strprintf(live, addrA.c_str())),
3853 "wrote a ledger file in the live production shape");
3854 LoadPendingPayouts();
3855 {
3856 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
3857 check(gPendingPayouts.size() == 1 &&
3858 gPendingPayouts[0].blockHeight == 1000 &&
3859 gPendingPayouts[0].matureAtHeight == 1120 &&
3860 gPendingPayouts[0].blockHash != 0 &&
3861 gPendingPayouts[0].amounts.size() == 1 &&
3862 gPendingPayouts[0].amounts[addrA] == 1234500000LL,
3863 "the live production pending_payouts.json still loads");
3864 }
3865
3866 // And the shape the merged build writes, retryAfter included, which
3867 // this build no longer writes but must still read past.
3868 PayoutTestResetState();
3869 const char* merged =
3870 "[{\"amounts\":{\"%s\":100000000},"
3871 "\"blockHash\":\"00000000000000000000000000000000000000000000000000000000000000ef\","
3872 "\"blockHeight\":2000,\"matureAtHeight\":2120,\"retryAfter\":0,"
3873 "\"paidTxids\":{\"%s\":\"abc123\"}}]";
3874 check(PayoutTestWriteFile(ledgerPath,
3875 strprintf(merged, addrA.c_str(), addrA.c_str())),
3876 "wrote a ledger file in the merged build's shape");
3877 LoadPendingPayouts();
3878 {
3879 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
3880 check(gPendingPayouts.size() == 1 &&
3881 gPendingPayouts[0].blockHeight == 2000 &&
3882 gPendingPayouts[0].paidTxids.size() == 1 &&
3883 gPendingPayouts[0].paidTxids[addrA] == "abc123",
3884 "a ledger file carrying retryAfter and paidTxids still loads");
3885 }
3886
3887 // A round trip through this build must not lose anything either.
3888 VersionedBlob snap;
3889 {
3890 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
3891 snap = SerializePendingPayoutsLocked();
3892 }
3893 PublishPendingPayouts(snap);
3894 PayoutTestResetState();
3895 LoadPendingPayouts();
3896 {
3897 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
3898 check(gPendingPayouts.size() == 1 &&
3899 gPendingPayouts[0].blockHeight == 2000 &&
3900 gPendingPayouts[0].matureAtHeight == 2120 &&
3901 gPendingPayouts[0].paidTxids[addrA] == "abc123",
3902 "what this build writes, this build reads back unchanged");
3903 }
3904 }
3905
3906 // -- 10. A forfeit is published, not just dropped ------------------------
3907 {
3908 PayoutTestResetState();
3909 remove(ledgerPath.c_str());
3910 const std::string roundsPath = dir + "/pool_rounds_selftest.json";
3911 const std::string savedRoundsFile = strPoolRoundsFile;
3912 strPoolRoundsFile = roundsPath;
3913 remove(roundsPath.c_str());
3914
3915 const int height = 606;
3916 PoolRoundProof round;
3917 round.id = PayoutIdForBlock(height, 0);
3918 round.blockHeight = height;
3919 round.matureAtHeight = height;
3920 round.payouts[addrA] = 5 * COIN;
3921 round.payouts[addrB] = 6 * COIN;
3922 RecordPoolRound(round);
3923
3924 {
3925 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
3926 PendingPayout pp;
3927 pp.blockHeight = height; pp.blockHash = 0; pp.matureAtHeight = height;
3928 pp.amounts[addrA] = 5 * COIN;
3929 pp.amounts[addrB] = 6 * COIN;
3930 gPendingPayouts.push_back(pp);
3931 }
3932
3933 PayingRecord r;
3934 r.blockHeight = height; r.blockHash = 0; r.matureAtHeight = height;
3935 r.address = addrA; r.amount = 5 * COIN;
3936 ForfeitPayoutRecipient(r, "selftest reason");
3937
3938 {
3939 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
3940 check(gPendingPayouts.size() == 1 &&
3941 gPendingPayouts[0].amounts.count(addrA) == 0 &&
3942 gPendingPayouts[0].amounts.count(addrB) == 1,
3943 "a forfeit removes only that recipient from the ledger entry");
3944 }
3945
3946 std::string onDisk;
3947 bool proofOk = false;
3948 if (PayoutTestReadFile(roundsPath, onDisk)) {
3949 try {
3950 nlohmann::json root = nlohmann::json::parse(onDisk);
3951 const nlohmann::json& p = root["rounds"][0]["payouts"];
3952 proofOk = p[addrA].value("forfeited", false) &&
3953 p[addrA].value("forfeitReason", "") == "selftest reason" &&
3954 p[addrA]["txid"].is_null() &&
3955 !p[addrB].contains("forfeited");
3956 } catch (...) {
3957 proofOk = false;
3958 }
3959 }
3960 check(proofOk,
3961 "the published round proof records the forfeit instead of"
3962 " advertising the address as still owed");
3963
3964 // And it survives a reload of the proof file.
3965 {
3966 std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
3967 gPoolRounds.clear();
3968 }
3969 LoadPoolRounds();
3970 {
3971 std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
3972 check(gPoolRounds.size() == 1 &&
3973 gPoolRounds[0].forfeited.count(addrA) == 1 &&
3974 gPoolRounds[0].forfeited[addrA] == "selftest reason",
3975 "the forfeit is read back from pool_rounds.json");
3976 }
3977
3978 remove(roundsPath.c_str());
3979 remove((roundsPath + ".tmp").c_str());
3980 strPoolRoundsFile = savedRoundsFile;
3981 }
3982
3983 // -- 11. A ledger that failed to load is never written back --------------
3984 //
3985 // The loaders push entries into the live vector inside the try, so an
3986 // exception part way down a file leaves a PARTIAL ledger in memory: the
3987 // entries before the bad one and nothing after it. Publishing that -- and
3988 // ReconcilePayouts publishes unconditionally, milliseconds into startup --
3989 // writes the partial state over the file and destroys the only record of
3990 // what was owed.
3991 {
3992 PayoutTestResetState();
3993 remove(ledgerPath.c_str());
3994
3995 // Two entries, the second with a matureAtHeight that is not a number.
3996 // The first loads; the second throws where it is read.
3997 const std::string corrupt = strprintf(
3998 "[{\"amounts\":{\"%s\":100000000},\"blockHash\":\"%s\","
3999 "\"blockHeight\":10,\"matureAtHeight\":10},"
4000 "{\"amounts\":{\"%s\":900000000},\"blockHash\":\"%s\","
4001 "\"blockHeight\":20,\"matureAtHeight\":\"twenty\"}]",
4002 addrA.c_str(), uint256(0).ToString().c_str(),
4003 addrB.c_str(), uint256(0).ToString().c_str());
4004 check(PayoutTestWriteFile(ledgerPath, corrupt),
4005 "wrote a ledger whose second entry does not parse");
4006
4007 const int64 t0 = GetTime();
4008 LoadPendingPayouts();
4009 const int64 t1 = GetTime();
4010
4011 size_t loaded = 0;
4012 {
4013 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
4014 loaded = gPendingPayouts.size();
4015 }
4016 check(loaded == 1,
4017 "a mid-file parse error does leave a partial ledger in memory --"
4018 " which is the reason for everything below");
4019
4020 std::string gone;
4021 check(!PayoutTestReadFile(ledgerPath, gone),
4022 "the file that did not parse is moved out of the way");
4023
4024 // The rename stamps GetTime(); it was taken between t0 and t1.
4025 std::string kept, keptPath;
4026 for (int64 t = t0; t <= t1 && keptPath.empty(); t++) {
4027 const std::string candidate =
4028 strprintf("%s.bad.%lld", ledgerPath.c_str(), (long long)t);
4029 if (PayoutTestReadFile(candidate, kept))
4030 keptPath = candidate;
4031 }
4032 check(!keptPath.empty() && kept == corrupt,
4033 "it is kept, byte for byte, as <path>.bad.<timestamp>");
4034
4035 // The publish that used to destroy it.
4036 VersionedBlob partial;
4037 {
4038 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
4039 partial = SerializePendingPayoutsLocked();
4040 }
4041 PublishPendingPayouts(partial);
4042 check(!PayoutTestReadFile(ledgerPath, gone),
4043 "and the partial ledger is not published over it");
4044
4045 // The refusal is per path and lasts the process: a second attempt,
4046 // with more state, is refused too.
4047 {
4048 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
4049 PendingPayout pp;
4050 pp.blockHeight = 30; pp.blockHash = 0; pp.matureAtHeight = 30;
4051 pp.amounts[addrA] = 5 * COIN;
4052 gPendingPayouts.push_back(pp);
4053 partial = SerializePendingPayoutsLocked();
4054 }
4055 PublishPendingPayouts(partial);
4056 check(!PayoutTestReadFile(ledgerPath, gone),
4057 "the refusal is not a one-shot -- the path stays unwritable");
4058
4059 if (!keptPath.empty())
4060 remove(keptPath.c_str());
4061 PayoutTestResetState();
4062 remove(ledgerPath.c_str());
4063
4064 // And a file that parses is not quarantined, so the ordinary path is
4065 // still the ordinary path.
4066 check(PayoutTestWriteFile(ledgerPath, "[]"),
4067 "wrote a ledger that parses");
4068 LoadPendingPayouts();
4069 {
4070 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
4071 PendingPayout pp;
4072 pp.blockHeight = 40; pp.blockHash = 0; pp.matureAtHeight = 40;
4073 pp.amounts[addrA] = 2 * COIN;
4074 gPendingPayouts.push_back(pp);
4075 partial = SerializePendingPayoutsLocked();
4076 }
4077 PublishPendingPayouts(partial);
4078 std::string written;
4079 check(PayoutTestReadFile(ledgerPath, written) && written == partial.blob,
4080 "a ledger that loaded cleanly is still published normally");
4081 }
4082
4083 // -- 12. A send that failed for want of coin is not a fault --------------
4084 //
4085 // The defect this replaces: the classifier decided "was this a funding
4086 // problem?" by re-reading GetBalance() and allowing amount +
4087 // nTransactionFee + CENT. nTransactionFee is 0 on a headless node, and
4088 // GetMinFee() is a step function -- 0 below 10,000 bytes, 11 * CENT at it
4089 // -- so the allowance was either unnecessary or eleven times too small,
4090 // and the ledger's own retry budget then wrote the debt off after 1440
4091 // repeats of the same stable condition.
4092 //
4093 // This is tested against a real wallet and the real CreateTransaction, not
4094 // against injected values: the honest question is not "does my enum
4095 // switch work" but "does a wallet holding many small coins actually report
4096 // the state I think it does". 60 outputs of 1 BTF give ~180 signed P2PKH
4097 // bytes each, which is where the fee schedule steps. Nothing is committed
4098 // or broadcast -- CreateTransaction only builds.
4099 {
4100 PayoutTestResetState();
4101 CRITICAL_BLOCK(cs_mapWallet)
4102 mapWallet.clear();
4103
4104 // CreateTransaction opens CTxDB("r"), which throws if blkindex.dat is
4105 // not there. Create it in the scratch datadir first.
4106 bool dbOk = true;
4107 try {
4108 CTxDB txdb("cr");
4109 } catch (const std::exception&) {
4110 dbOk = false;
4111 }
4112 check(dbOk, "a scratch block index exists for CreateTransaction to open");
4113
4114 const std::string walletAddr = PayoutTestAddSpendableKey();
4115 const std::string payee = PayoutTestAddress(0x77);
4116 const CScript payeeScript = PayoutTestScriptFor(payee);
4117
4118 // -- the >= 10 KB case: 60 small coins, balance = amount + 0.05 ------
4119 PayoutTestFundWallet(60, 1 * COIN, walletAddr);
4120 const int64 balance = GetBalance();
4121 check(balance == 60 * COIN,
4122 "the test wallet holds 60 separately spendable coins");
4123
4124 const int64 amount = balance - 5 * CENT;
4125 // The pre-send gate in PayoutThreadFn passes: the balance covers the
4126 // amount and the only fee that can be known in advance.
4127 check(balance >= amount + nTransactionFee,
4128 "the pre-send gate would let this payment through");
4129
4130 CWalletTx wtx;
4131 SendMoneyOutcome outcome;
4132 int64 feeRequired = 0;
4133 WalletSendResult result = WALLET_SEND_OK;
4134 bool built = true;
4135 if (dbOk) {
4136 try {
4137 built = CreateTransaction(payeeScript, amount, wtx, feeRequired,
4138 &result);
4139 } catch (const std::exception&) {
4140 built = true; // report as a failure below
4141 }
4142 }
4143 check(dbOk && !built,
4144 "a wallet of many small coins cannot fund amount + the fee its"
4145 " own size demands");
4146 check(dbOk && result == WALLET_SEND_INSUFFICIENT_FUNDS,
4147 "the wallet reports that as a funding shortfall, not as a"
4148 " failure to build");
4149 // GetMinFee(fDiscount) returns 0 below 10,000 bytes and
4150 // (1 + bytes/1000) * CENT at or above, so a required fee of 11 CENT or
4151 // more IS the statement that the transaction crossed 10 KB. Measured
4152 // through the real serialiser rather than asserted.
4153 check(dbOk && feeRequired >= 11 * CENT,
4154 "the transaction it built crossed 10,000 bytes, so the fee"
4155 " stepped from 0 to at least 11 CENT");
4156
4157 // The defect, stated as arithmetic. The old classifier asked whether
4158 // the balance covered amount + nTransactionFee + CENT; it did, so it
4159 // called a permanent funding shortfall a fault and started counting
4160 // towards a write-off.
4161 check(!(balance < amount + nTransactionFee + CENT),
4162 "the reserve this replaces would have judged this send a fault");
4163 check(balance < amount + feeRequired,
4164 "while the fee actually required was more than the balance had");
4165
4166 outcome.result = result;
4167 outcome.nFeeRequired = feeRequired;
4168 check(dbOk && ClassifyPayoutSendFailure(outcome.result, false)
4169 == PAYOUT_FAILURE_DEFER,
4170 "a >= 10 KB send that ran out of coin is deferred, not counted"
4171 " against the payment");
4172
4173 // -- the < 10 KB case: no reserve is needed at all -------------------
4174 //
4175 // The other half of the arithmetic. Below the step the required fee is
4176 // 0, so a reserve of any size is a reason to refuse a payment the
4177 // wallet could have made -- which is its own way of not paying a
4178 // miner.
4179 CRITICAL_BLOCK(cs_mapWallet)
4180 mapWallet.clear();
4181 PayoutTestFundWallet(3, 1 * COIN, walletAddr);
4182 const int64 smallAmount = 3 * COIN - 5 * CENT;
4183 CWalletTx smallTx;
4184 int64 smallFee = -1;
4185 WalletSendResult smallResult = WALLET_SEND_FAILED;
4186 bool smallBuilt = false;
4187 if (dbOk) {
4188 try {
4189 smallBuilt = CreateTransaction(payeeScript, smallAmount, smallTx,
4190 smallFee, &smallResult);
4191 } catch (const std::exception&) {
4192 smallBuilt = false;
4193 }
4194 }
4195 check(dbOk && smallBuilt && smallResult == WALLET_SEND_OK,
4196 "the same balance pays a small transaction with no headroom at"
4197 " all");
4198 check(dbOk && smallFee == 0,
4199 "because below 10,000 bytes the required fee is zero");
4200
4201 // -- and the classification of every other answer --------------------
4202 check(ClassifyPayoutSendFailure(WALLET_SEND_INSUFFICIENT_FUNDS, false)
4203 == PAYOUT_FAILURE_DEFER &&
4204 ClassifyPayoutSendFailure(WALLET_SEND_INSUFFICIENT_FUNDS, true)
4205 == PAYOUT_FAILURE_DEFER &&
4206 ClassifyPayoutSendFailure(WALLET_SEND_FAILED, true)
4207 == PAYOUT_FAILURE_DEFER,
4208 "short of coin or unable to sign is always a deferral");
4209 check(ClassifyPayoutSendFailure(WALLET_SEND_FAILED, false)
4210 == PAYOUT_FAILURE_FAULT,
4211 "funded and unlocked and still refused is a fault");
4212
4213 CRITICAL_BLOCK(cs_mapWallet)
4214 mapWallet.clear();
4215 }
4216
4217 // -- 13. No payout mutex is held across the status-file write ------------
4218 //
4219 // RefreshStats runs on the event loop -- the thread that owns every
4220 // miner's socket -- and used to hold gStatsMutex and gPayoutLedgerMutex at
4221 // function scope across WriteFileAtomic. Once that write gained fsync() on
4222 // the file and on its directory, publishing pool_status.json parked the
4223 // payout thread behind two disk flushes every ten seconds.
4224 //
4225 // Timing assertions would be flaky, so this does not measure anything. It
4226 // makes the write block for as long as the test likes: strPoolStatusFile
4227 // points at a path whose ".tmp" is a FIFO, and open(2) for writing on a
4228 // FIFO does not return until a reader arrives. While RefreshStats is
4229 // parked in there, every mutex it must NOT be holding has to be
4230 // acquirable. Against the previous code this test does not merely fail, it
4231 // cannot pass: the writer is inside fopen() with both mutexes held and
4232 // nothing will release them until the test lets the write finish.
4233#ifndef _WIN32
4234 {
4235 PayoutTestResetState();
4236 const std::string statusPath = dir + "/pool_status_selftest.json";
4237 const std::string statusTmp = statusPath + ".tmp";
4238 const std::string savedStatusFile = strPoolStatusFile;
4239 const int64 savedLastWrite = gLastPoolStatusWrite;
4240 remove(statusPath.c_str());
4241 remove(statusTmp.c_str());
4242 strPoolStatusFile = statusPath;
4243 gLastPoolStatusWrite = 0; // due a publish on the next refresh
4244
4245 // Give the status file something to report, so the ledger mutex is
4246 // genuinely taken on the way through rather than skipped over.
4247 {
4248 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
4249 PendingPayout pp;
4250 pp.blockHeight = 909; pp.blockHash = 0; pp.matureAtHeight = 909;
4251 pp.amounts[addrA] = 3 * COIN;
4252 gPendingPayouts.push_back(pp);
4253 }
4254
4255 bool fifoOk = (mkfifo(statusTmp.c_str(), 0600) == 0);
4256 check(fifoOk, "a FIFO stands in for the status file's temp file, so the"
4257 " write can be held open");
4258 if (fifoOk) {
4259 std::atomic<bool> done(false);
4260 std::thread writer([&done]() {
4261 RefreshStats(0, 0);
4262 done = true;
4263 });
4264
4265 // Settle: the work before the write is a handful of vector
4266 // operations, so anything past this point is inside fopen(). The
4267 // !done check below is what actually proves it -- with no reader
4268 // there is no other way for this thread to still be running.
4269 Sleep(500);
4270 bool blocked = !done;
4271 bool gotLedger = false, gotStats = false, gotExec = false,
4272 gotRounds = false;
4273 for (int i = 0; i < 1000 && blocked; i++) {
4274 if (!gotLedger && gPayoutLedgerMutex.try_lock()) {
4275 gotLedger = true; gPayoutLedgerMutex.unlock();
4276 }
4277 if (!gotStats && gStatsMutex.try_lock()) {
4278 gotStats = true; gStatsMutex.unlock();
4279 }
4280 if (!gotExec && gPayoutExecMutex.try_lock()) {
4281 gotExec = true; gPayoutExecMutex.unlock();
4282 }
4283 if (!gotRounds && gPoolRoundsMutex.try_lock()) {
4284 gotRounds = true; gPoolRoundsMutex.unlock();
4285 }
4286 if (gotLedger && gotStats && gotExec && gotRounds)
4287 break;
4288 Sleep(2);
4289 }
4290
4291 check(blocked,
4292 "the status-file write is held open for the duration of this"
4293 " test");
4294 check(!blocked || gotLedger,
4295 "gPayoutLedgerMutex is free while pool_status.json is being"
4296 " written");
4297 check(!blocked || (gotExec && gotRounds),
4298 "gPayoutExecMutex and gPoolRoundsMutex are free while"
4299 " pool_status.json is being written");
4300 check(!blocked || gotStats,
4301 "gStatsMutex is free while pool_status.json is being written");
4302
4303 // Let the write finish. Drain to EOF rather than closing early:
4304 // the writer is still filling the pipe and a closed read end would
4305 // deliver it SIGPIPE.
4306 std::thread reader([&statusTmp]() {
4307 FILE* rf = fopen(statusTmp.c_str(), "r");
4308 if (!rf) return;
4309 char buf[4096];
4310 while (fread(buf, 1, sizeof(buf), rf) > 0) { }
4311 fclose(rf);
4312 });
4313 writer.join();
4314 reader.join();
4315 }
4316 remove(statusTmp.c_str());
4317
4318 // And the ordinary path still produces the file it always did.
4319 gLastPoolStatusWrite = 0;
4320 RefreshStats(0, 0);
4321 std::string onDisk;
4322 bool statusOk = false;
4323 if (PayoutTestReadFile(statusPath, onDisk)) {
4324 try {
4325 nlohmann::json j = nlohmann::json::parse(onDisk);
4326 statusOk = j.value("schema", "") == "bitflash-pool-status-1" &&
4327 j["pendingPayouts"].is_array() &&
4328 j["pendingPayouts"].size() == 1 &&
4329 j["pendingPayouts"][0].value("matureAtHeight", 0) == 909 &&
4330 j["pendingPayouts"][0].value("totalAmountSatoshis",
4331 (int64)0) == 3 * COIN;
4332 } catch (...) {
4333 statusOk = false;
4334 }
4335 }
4336 check(statusOk,
4337 "splitting the serialise from the write leaves pool_status.json"
4338 " saying the same thing");
4339
4340 remove(statusPath.c_str());
4341 remove((statusPath + ".tmp").c_str());
4342 strPoolStatusFile = savedStatusFile;
4343 gLastPoolStatusWrite = savedLastWrite;
4344 }
4345#endif
4346
4347 // -- clean up ------------------------------------------------------------
4348 PayoutTestResetState();
4349 CRITICAL_BLOCK(cs_mapWallet)
4350 mapWallet.clear();
4351 PayoutTestClearBlockIndex();
4352 // Out of the scratch directory before removing it, and out of the key maps
4353 // too: the keys this test generated are process-local, but leaving them in
4354 // the keystore would let a later test think a script was its own.
4355 CRITICAL_BLOCK(cs_mapKeys)
4356 {
4357 mapKeys.clear();
4358 mapPubKeys.clear();
4359 }
4360#ifdef _WIN32
4361 SetCurrentDirectoryA(cwd.c_str());
4362#else
4363 if (chdir(cwd.c_str()) != 0) { /* nothing useful to do about it */ }
4364#endif
4365 PayoutTestRemoveDir(dir);
4366 strSetDataDir = savedDataDir;
4367 nBestHeight = savedBestHeight;
4368
4369 fprintf(stdout, "%s (%d failure%s)\n", nFail == 0 ? "ALL TESTS PASSED" : "TESTS FAILED",
4370 nFail, nFail == 1 ? "" : "s");
4371 fflush(stdout);
4372 return nFail == 0 ? 0 : 1;
4373}
4374
1511int RunPoolStratumSelfTest() 4375int RunPoolStratumSelfTest()
1512{ 4376{
1513 int nFail = 0; 4377 int nFail = 0;
@@ -1654,56 +4518,71 @@ int RunPoolStratumSelfTest()
1654// --------------------------------------------------------------------------- 4518// ---------------------------------------------------------------------------
1655static void RefreshStats(int blocksFoundThisSession, uint64 roundShareTotal) 4519static void RefreshStats(int blocksFoundThisSession, uint64 roundShareTotal)
1656{ 4520{
1657 std::lock_guard<std::mutex> lk(gStatsMutex); 4521 std::string statusBlob;
4522 bool fPublish = false;
4523 {
4524 std::lock_guard<std::mutex> lk(gStatsMutex);
1658 4525
1659 gStats.authorizedMiners = 0; 4526 gStats.authorizedMiners = 0;
1660 gStats.blocksFound = blocksFoundThisSession; 4527 gStats.blocksFound = blocksFoundThisSession;
1661 gStats.roundShares = roundShareTotal; 4528 gStats.roundShares = roundShareTotal;
1662 gStats.totalHashRate = 0.0; 4529 gStats.totalHashRate = 0.0;
1663 for (Miner* m : gMiners) 4530 for (Miner* m : gMiners)
1664 if (m->authorised) gStats.authorizedMiners++; 4531 if (m->authorised) gStats.authorizedMiners++;
1665 4532
1666 gWorkerStats.clear(); 4533 gWorkerStats.clear();
1667 for (Miner* m : gMiners) { 4534 for (Miner* m : gMiners) {
1668 if (!m->authorised || m->lastSeen <= 0) continue; 4535 if (!m->authorised || m->lastSeen <= 0) continue;
1669 PoolWorkerStatView v; 4536 PoolWorkerStatView v;
1670 v.address = m->address; 4537 v.address = m->address;
1671 v.worker = m->worker; 4538 v.worker = m->worker;
1672 v.roundShares = m->roundShares; 4539 v.roundShares = m->roundShares;
1673 v.totalShares = m->totalShares; 4540 v.totalShares = m->totalShares;
1674 v.lastSeen = m->lastSeen; 4541 v.lastSeen = m->lastSeen;
1675 v.hashRate = m->hashRate; 4542 v.hashRate = m->hashRate;
1676 gWorkerStats.push_back(v); 4543 gWorkerStats.push_back(v);
1677 gStats.totalHashRate += m->hashRate; 4544 gStats.totalHashRate += m->hashRate;
1678 } 4545 }
1679 std::sort(gWorkerStats.begin(), gWorkerStats.end(), 4546 std::sort(gWorkerStats.begin(), gWorkerStats.end(),
1680 [](const PoolWorkerStatView& a, const PoolWorkerStatView& b) { 4547 [](const PoolWorkerStatView& a, const PoolWorkerStatView& b) {
1681 return a.roundShares != b.roundShares ? a.roundShares > b.roundShares 4548 return a.roundShares != b.roundShares ? a.roundShares > b.roundShares
1682 : a.lastSeen > b.lastSeen; 4549 : a.lastSeen > b.lastSeen;
1683 }); 4550 });
1684 4551
1685 gPayoutViews.clear(); 4552 // Scoped, not held to the end of the function. At function scope this
1686 std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex); 4553 // lock covered the status-file write below, so the event loop did two
1687 for (const PendingPayout& pp : gPendingPayouts) { 4554 // fsyncs holding the mutex the payout thread takes to record that a
1688 PendingPayoutView pv; 4555 // miner has been paid.
1689 pv.matureAtHeight = pp.matureAtHeight; 4556 gPayoutViews.clear();
1690 pv.recipients = 0; 4557 {
1691 pv.totalAmount = 0; 4558 std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);
1692 for (auto& kv : pp.amounts) { 4559 for (const PendingPayout& pp : gPendingPayouts) {
1693 if (pp.paidTxids.count(kv.first)) 4560 PendingPayoutView pv;
1694 continue; 4561 pv.matureAtHeight = pp.matureAtHeight;
1695 pv.recipients++; 4562 pv.recipients = 0;
1696 pv.totalAmount += kv.second; 4563 pv.totalAmount = 0;
4564 for (auto& kv : pp.amounts) {
4565 if (pp.paidTxids.count(kv.first))
4566 continue;
4567 pv.recipients++;
4568 pv.totalAmount += kv.second;
4569 }
4570 if (pv.recipients > 0)
4571 gPayoutViews.push_back(pv);
4572 }
1697 } 4573 }
1698 if (pv.recipients > 0)
1699 gPayoutViews.push_back(pv);
1700 }
1701 4574
1702 int64 now = GetTime(); 4575 int64 now = GetTime();
1703 if (now - gLastPoolStatusWrite >= 10) { 4576 if (now - gLastPoolStatusWrite >= 10) {
1704 WritePoolStatusJsonLocked(); 4577 statusBlob = SerializePoolStatusLocked();
1705 gLastPoolStatusWrite = now; 4578 gLastPoolStatusWrite = now;
4579 fPublish = true;
4580 }
1706 } 4581 }
4582 // Every payout mutex is released here, and gStatsMutex with them. This is
4583 // where the two fsyncs happen, and no other thread is waiting behind them.
4584 if (fPublish)
4585 PublishPoolStatus(statusBlob);
1707} 4586}
1708 4587
1709// --------------------------------------------------------------------------- 4588// ---------------------------------------------------------------------------
@@ -1962,7 +4841,11 @@ static void AcceptOneFn(void* arg)
1962 LogPrint("pool", "[pool] accept thread started for relay %s\n", relay.c_str()); 4841 LogPrint("pool", "[pool] accept thread started for relay %s\n", relay.c_str());
1963 4842
1964 while (!fShutdown && gPoolRunning) { 4843 while (!fShutdown && gPoolRunning) {
1965 // Blocks until a miner connects to us at this relay 4844 // Advertise this relay as the pool's meeting node BEFORE registering,
4845 // so that miners which resolve our descriptor find a relay where the
4846 // pool (not the node's peer-accept) holds the registration.
4847 BtfSetActiveRelay(relay);
4848 // Returns as soon as the relay lists us -- registered, not yet paired.
1966 btf::RvSocket rv = btf::RvServiceRegister(host.c_str(), (unsigned short)port, pk); 4849 btf::RvSocket rv = btf::RvServiceRegister(host.c_str(), (unsigned short)port, pk);
1967 4850
1968 if (fShutdown || !gPoolRunning) { 4851 if (fShutdown || !gPoolRunning) {
@@ -1976,6 +4859,23 @@ static void AcceptOneFn(void* arg)
1976 continue; 4859 continue;
1977 } 4860 }
1978 4861
4862 // Block for a miner to actually arrive. RvServiceRegister used to do
4863 // this itself and returned with the relay's 0x01 pairing byte already
4864 // consumed; it no longer does. Without waiting here, BtfServiceWrap
4865 // reads that 0x01 as the first byte of the miner's 32-byte ephemeral
4866 // key -- every subsequent byte is shifted by one, so the X25519
4867 // handshake completes against a garbage shared secret (it has no
4868 // confirmation step to catch it), the tunnel is handed to the event
4869 // loop, and the miner's mining.subscribe never decrypts into a valid
4870 // line. The miner sees the pool close on it; the pool sees a peer that
4871 // never authorized.
4872 if (!btf::RvServiceWaitPaired(rv)) {
4873 btf::RvClose(rv);
4874 LogPrint("pool", "[pool] relay %s: dropped us before any miner dialed"
4875 " -- re-registering\n", relay.c_str());
4876 continue;
4877 }
4878
1979 // Echan handshake -- blocking but fast (local + relay round-trip) 4879 // Echan handshake -- blocking but fast (local + relay round-trip)
1980 btf_socket_t fd = btf::BtfServiceWrap(rv, sk); 4880 btf_socket_t fd = btf::BtfServiceWrap(rv, sk);
1981 if (fd == INVALID_SOCKET) { 4881 if (fd == INVALID_SOCKET) {
@@ -2062,8 +4962,14 @@ void ThreadRPCServer(void*)
2062 4962
2063 LoadPendingPayouts(); 4963 LoadPendingPayouts();
2064 LoadPoolRounds(); 4964 LoadPoolRounds();
4965 LoadPaying();
4966 // Must run before the payout thread starts, so nothing is resent that the
4967 // ledger already shows as sent, and every restored in-flight record is
4968 // armed for its one wallet probe before it is allowed to send.
4969 ReconcilePayouts();
2065 4970
2066 _beginthread(AcceptLoopFn, 0, NULL); 4971 _beginthread(AcceptLoopFn, 0, NULL);
4972 _beginthread(LocalAcceptFn, 0, NULL);
2067 _beginthread(ThreadBtfPoolAnnouncer, 0, NULL); 4973 _beginthread(ThreadBtfPoolAnnouncer, 0, NULL);
2068 _beginthread(PayoutThreadFn, 0, NULL); 4974 _beginthread(PayoutThreadFn, 0, NULL);
2069 4975