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.cpp431
1 files changed, 272 insertions, 159 deletions
diff --git a/src/rpc.cpp b/src/rpc.cpp
index 8af2fbe..76867ab 100644
--- a/src/rpc.cpp
+++ b/src/rpc.cpp
@@ -360,6 +360,12 @@ struct PendingPayout {
360}; 360};
361static std::vector<PendingPayout> gPendingPayouts; 361static std::vector<PendingPayout> gPendingPayouts;
362static std::mutex gPayoutLedgerMutex; 362static std::mutex gPayoutLedgerMutex;
363// Bumped on every serialisation, under gPayoutLedgerMutex. See PublishVersioned.
364static uint64 gPayoutLedgerSeq = 0;
365// Serialises writers of pending_payouts.json, and guards the sequence number
366// of the image currently on disk. Never taken while a data mutex is held.
367static std::mutex gPayoutLedgerFileMutex;
368static uint64 gPayoutLedgerPublishedSeq = 0;
363 369
364struct PoolRoundProof { 370struct PoolRoundProof {
365 std::string id; 371 std::string id;
@@ -388,6 +394,9 @@ struct PoolRoundProof {
388}; 394};
389static std::vector<PoolRoundProof> gPoolRounds; 395static std::vector<PoolRoundProof> gPoolRounds;
390static std::mutex gPoolRoundsMutex; 396static std::mutex gPoolRoundsMutex;
397static uint64 gPoolRoundsSeq = 0; // guarded by gPoolRoundsMutex
398static std::mutex gPoolRoundsFileMutex;
399static uint64 gPoolRoundsPublishedSeq = 0; // guarded by gPoolRoundsFileMutex
391static const size_t MAX_POOL_ROUNDS = 200; 400static const size_t MAX_POOL_ROUNDS = 200;
392 401
393static std::string PayoutFilePath() 402static std::string PayoutFilePath()
@@ -402,6 +411,104 @@ static std::string PoolRoundsFilePath()
402 return GetAppDir() + "/pool_rounds.json"; 411 return GetAppDir() + "/pool_rounds.json";
403} 412}
404 413
414// ---------------------------------------------------------------------------
415// Atomic, ordered publication of the payout state files
416// ---------------------------------------------------------------------------
417//
418// Every one of these files is a full dump of an in-memory structure, written
419// tmp-then-rename so a crash cannot truncate it. Two things have to be true
420// for that to actually be safe, and only the first was:
421//
422// 1. the rename must be atomic (it is), and
423// 2. exactly one thread at a time may be writing the temp file, and a write
424// must never publish an OLDER image over a newer one.
425//
426// (2) was broken for pending_payouts.json. The event loop wrote it with
427// gPayoutLedgerMutex held, the payout thread wrote it with the mutex
428// released, and nothing serialised the two: both opened the same
429// "<path>.tmp" with independent handles and both renamed it. The interleaved
430// file that lands is either a mixture of two images (LoadPendingPayouts then
431// throws mid-loop and comes up with a nearly empty ledger -- every obligation
432// not yet exploded, gone) or a whole newly-queued round silently overwritten
433// by an older snapshot that predates it.
434//
435// The fix is a per-file write mutex, held only across the I/O and never
436// across a data-structure mutex, plus a version on the snapshot. Serialising
437// happens under the data mutex and stamps a monotonically increasing sequence
438// number; publishing happens with the data mutex released and refuses any
439// blob whose sequence number is not newer than what is already on disk. So
440// writers can queue in any order and finish in any order -- the file only
441// ever moves forwards. Crucially, no thread ever waits for I/O while holding
442// gPayoutLedgerMutex / gPayoutExecMutex / gPoolRoundsMutex, which is the
443// contention the merge was right to remove from the event loop.
444
445// A serialised file image plus the version of the in-memory state it was
446// taken from.
447struct VersionedBlob {
448 std::string blob;
449 uint64 seq = 0;
450};
451
452// Replace `path` with `data` atomically. Caller must hold the write mutex for
453// `path`: the temp name is derived from the target, so two threads in here at
454// once for one path is precisely the race this exists to prevent.
455static bool WriteFileAtomic(const std::string& path, const std::string& data,
456 const char* what)
457{
458 std::string tmp = path + ".tmp";
459 FILE* f = fopen(tmp.c_str(), "w");
460 if (!f) {
461 LogPrint("payout", "[payout] WARNING: could not write %s (%s)\n",
462 tmp.c_str(), what);
463 return false;
464 }
465 if (!data.empty() && fwrite(data.c_str(), 1, data.size(), f) != data.size()) {
466 LogPrint("payout", "[payout] WARNING: short write to %s (%s)\n",
467 tmp.c_str(), what);
468 fclose(f);
469 return false;
470 }
471 // fclose() is where buffered data actually reaches the OS, so a failure
472 // here means the .tmp is short -- renaming it over the real file would
473 // destroy the good copy, which is exactly what these files exist to
474 // prevent.
475 if (fclose(f) != 0) {
476 LogPrint("payout", "[payout] WARNING: could not close %s (%s)\n",
477 tmp.c_str(), what);
478 return false;
479 }
480#ifdef _WIN32
481 // rename() on Windows will not replace an existing file.
482 remove(path.c_str());
483#endif
484 if (rename(tmp.c_str(), path.c_str()) != 0) {
485 LogPrint("payout", "[payout] WARNING: could not replace %s (%s)\n",
486 path.c_str(), what);
487 return false;
488 }
489 return true;
490}
491
492// Publish a versioned image of some in-memory state to `path`.
493//
494// `fileMutex` serialises writers of `path` and guards `publishedSeq`. A blob
495// older than what is already published is dropped: these files are whole-state
496// dumps, so a higher sequence number is by construction a later state of the
497// same structure and publishing it loses nothing. On a write failure
498// `publishedSeq` is deliberately left alone, so the next (newer) write still
499// gets a chance to land.
500static void PublishVersioned(const std::string& path, const VersionedBlob& v,
501 std::mutex& fileMutex, uint64& publishedSeq,
502 const char* what)
503{
504 std::lock_guard<std::mutex> lk(fileMutex);
505 if (v.seq <= publishedSeq)
506 return;
507 if (!WriteFileAtomic(path, v.blob, what))
508 return;
509 publishedSeq = v.seq;
510}
511
405// Identity of a payout round. Used as the pool-round proof id and, more 512// Identity of a payout round. Used as the pool-round proof id and, more
406// importantly, as the `pool_payout_id` written into wallet transaction 513// importantly, as the `pool_payout_id` written into wallet transaction
407// metadata and matched by FindRecordedPoolPayout(). 514// metadata and matched by FindRecordedPoolPayout().
@@ -431,14 +538,16 @@ static bool PayoutFullyPaid(const PendingPayout& pp)
431 return true; 538 return true;
432} 539}
433 540
434// Serialise the ledger. Caller must hold gPayoutLedgerMutex. 541// Serialise the ledger and stamp it with the version it was taken at. Caller
542// must hold gPayoutLedgerMutex.
435// 543//
436// Split from the file write so the payout thread can hold the mutex just long 544// Split from the file write so no thread holds the ledger mutex across the
437// enough to read the structure and then do the I/O with it released. Upstream 545// I/O. Upstream writes the file inside the per-recipient loop with
438// writes the file inside the per-recipient loop with gPayoutLedgerMutex held, 546// gPayoutLedgerMutex held, while the event loop's FlushMaturePayouts blocks on
439// while the event loop's FlushMaturePayouts blocks on that same mutex -- one 547// that same mutex -- one slow fsync per recipient, on the thread the miners'
440// slow fsync per recipient, on the thread the miners' I/O shares. 548// I/O shares. The sequence number is what lets the write happen later, and
441static std::string SerializePendingPayoutsLocked() 549// possibly out of order, without a stale image reaching the disk.
550static VersionedBlob SerializePendingPayoutsLocked()
442{ 551{
443 nlohmann::json arr = nlohmann::json::array(); 552 nlohmann::json arr = nlohmann::json::array();
444 for (const auto& pp : gPendingPayouts) { 553 for (const auto& pp : gPendingPayouts) {
@@ -457,38 +566,18 @@ static std::string SerializePendingPayoutsLocked()
457 obj["paidTxids"] = paid; 566 obj["paidTxids"] = paid;
458 arr.push_back(obj); 567 arr.push_back(obj);
459 } 568 }
460 return arr.dump(2); 569 VersionedBlob v;
461} 570 v.blob = arr.dump(2);
462 571 v.seq = ++gPayoutLedgerSeq;
463// Write the serialised ledger. Holds no lock; safe to call with the mutex 572 return v;
464// released. Atomic tmp+rename so a crash mid-write cannot truncate the record
465// of who is still owed money.
466static void WritePendingPayoutsFile(const std::string& s)
467{
468 std::string path = PayoutFilePath();
469 std::string tmp = path + ".tmp";
470 FILE* f = fopen(tmp.c_str(), "w");
471 if (!f) {
472 LogPrint("payout", "[payout] WARNING: could not write %s\n", tmp.c_str());
473 return;
474 }
475 fwrite(s.c_str(), 1, s.size(), f);
476 if (fclose(f) != 0) {
477 LogPrint("payout", "[payout] WARNING: could not close %s\n", tmp.c_str());
478 return;
479 }
480#ifdef _WIN32
481 remove(path.c_str());
482#endif
483 if (rename(tmp.c_str(), path.c_str()) != 0)
484 LogPrint("payout", "[payout] WARNING: could not replace %s\n", path.c_str());
485} 573}
486 574
487// Convenience for the cold paths (startup, queueing) where holding the mutex 575// Publish a ledger snapshot. Holds no data mutex; safe -- and intended -- to
488// across the write costs nothing. Caller must hold gPayoutLedgerMutex. 576// call with gPayoutLedgerMutex released.
489static void SavePendingPayoutsLocked() 577static void PublishPendingPayouts(const VersionedBlob& v)
490{ 578{
491 WritePendingPayoutsFile(SerializePendingPayoutsLocked()); 579 PublishVersioned(PayoutFilePath(), v, gPayoutLedgerFileMutex,
580 gPayoutLedgerPublishedSeq, "pending payout ledger");
492} 581}
493 582
494static void LoadPendingPayouts() 583static void LoadPendingPayouts()
@@ -577,6 +666,9 @@ struct PayingRecord {
577}; 666};
578static std::vector<PayingRecord> gPaying; // guarded by gPayoutExecMutex 667static std::vector<PayingRecord> gPaying; // guarded by gPayoutExecMutex
579static std::mutex gPayoutExecMutex; 668static std::mutex gPayoutExecMutex;
669static uint64 gPayingSeq = 0; // guarded by gPayoutExecMutex
670static std::mutex gPayingFileMutex;
671static uint64 gPayingPublishedSeq = 0; // guarded by gPayingFileMutex
580static std::atomic<bool> gPayoutThreadRunning{false}; 672static std::atomic<bool> gPayoutThreadRunning{false};
581 673
582static std::string PayingFilePath() 674static std::string PayingFilePath()
@@ -584,8 +676,16 @@ static std::string PayingFilePath()
584 return GetAppDir() + "/paying_payouts.json"; 676 return GetAppDir() + "/paying_payouts.json";
585} 677}
586 678
587// Caller must hold gPayoutExecMutex. 679// Serialise the in-flight list. Caller must hold gPayoutExecMutex.
588static void SavePayingLocked() 680//
681// gPayoutExecMutex already serialises every writer of this file, so it never
682// had the interleaving race pending_payouts.json had. It is still split the
683// same way for a different reason: FlushMaturePayouts runs on the event loop
684// and holds gPayoutExecMutex, so writing (and, since this series, fsyncing)
685// the file under that mutex parks miner I/O behind the payout thread's disk.
686// The version stamp keeps the file monotonic now that the write is outside
687// the mutex that orders the writers.
688static VersionedBlob SerializePayingLocked()
589{ 689{
590 nlohmann::json arr = nlohmann::json::array(); 690 nlohmann::json arr = nlohmann::json::array();
591 for (const auto& r : gPaying) { 691 for (const auto& r : gPaying) {
@@ -599,33 +699,17 @@ static void SavePayingLocked()
599 obj["deferrals"] = r.deferrals; 699 obj["deferrals"] = r.deferrals;
600 arr.push_back(obj); 700 arr.push_back(obj);
601 } 701 }
602 std::string path = PayingFilePath(); 702 VersionedBlob v;
603 // Write to a temporary file and rename, so a crash mid-write cannot leave 703 v.blob = arr.dump(2);
604 // a truncated list of who is still owed money. 704 v.seq = ++gPayingSeq;
605 std::string tmp = path + ".tmp"; 705 return v;
606 FILE* f = fopen(tmp.c_str(), "w"); 706}
607 if (!f) { 707
608 LogPrint("payout", "[payout] WARNING: could not write %s\n", tmp.c_str()); 708// Publish an in-flight-list snapshot. Call with gPayoutExecMutex released.
609 return; 709static void PublishPaying(const VersionedBlob& v)
610 } 710{
611 std::string s = arr.dump(2); 711 PublishVersioned(PayingFilePath(), v, gPayingFileMutex,
612 fwrite(s.c_str(), 1, s.size(), f); 712 gPayingPublishedSeq, "in-flight payment list");
613 // fclose() is where buffered data actually reaches the OS, so a failure
614 // here means the .tmp is short -- renaming it over the real file would
615 // destroy the good copy. Upstream's check, taken because a truncated
616 // list of who is owed money is exactly what this function exists to
617 // prevent. rename() on Windows will not replace an existing file, hence
618 // the remove() first.
619 if (fclose(f) != 0) {
620 LogPrint("payout", "[payout] WARNING: could not close %s\n", tmp.c_str());
621 return;
622 }
623#ifdef _WIN32
624 remove(path.c_str());
625#endif
626 if (rename(tmp.c_str(), path.c_str()) != 0)
627 LogPrint("payout", "[payout] WARNING: could not rename %s -> %s\n",
628 tmp.c_str(), path.c_str());
629} 713}
630 714
631static void LoadPaying() 715static void LoadPaying()
@@ -713,7 +797,14 @@ static nlohmann::json PoolRoundToJson(const PoolRoundProof& r)
713 return j; 797 return j;
714} 798}
715 799
716static void SavePoolRoundsLocked() 800// Serialise the published round proofs. Caller must hold gPoolRoundsMutex.
801//
802// Like the in-flight list, gPoolRoundsMutex already serialises the writers
803// (RecordPoolRound on the event loop, MarkPoolRoundPaid on the payout thread),
804// so this file never had the interleaving race either -- but the event loop
805// should not be waiting on the payout thread's fsync to publish a round, so
806// the write moves outside the mutex on the same versioned terms.
807static VersionedBlob SerializePoolRoundsLocked()
717{ 808{
718 nlohmann::json root; 809 nlohmann::json root;
719 root["schema"] = "bitflash-pool-rounds-1"; 810 root["schema"] = "bitflash-pool-rounds-1";
@@ -722,24 +813,17 @@ static void SavePoolRoundsLocked()
722 for (const PoolRoundProof& r : gPoolRounds) 813 for (const PoolRoundProof& r : gPoolRounds)
723 root["rounds"].push_back(PoolRoundToJson(r)); 814 root["rounds"].push_back(PoolRoundToJson(r));
724 815
725 std::string path = PoolRoundsFilePath(); 816 VersionedBlob v;
726 std::string tmp = path + ".tmp"; 817 v.blob = root.dump(2);
727 FILE* f = fopen(tmp.c_str(), "w"); 818 v.seq = ++gPoolRoundsSeq;
728 if (!f) { 819 return v;
729 LogPrint("payout", "[payout] WARNING: could not write %s\n", tmp.c_str()); 820}
730 return; 821
731 } 822// Publish a round-proof snapshot. Call with gPoolRoundsMutex released.
732 std::string s = root.dump(2); 823static void PublishPoolRounds(const VersionedBlob& v)
733 fwrite(s.c_str(), 1, s.size(), f); 824{
734 if (fclose(f) != 0) { 825 PublishVersioned(PoolRoundsFilePath(), v, gPoolRoundsFileMutex,
735 LogPrint("payout", "[payout] WARNING: could not close %s\n", tmp.c_str()); 826 gPoolRoundsPublishedSeq, "pool round proofs");
736 return;
737 }
738#ifdef _WIN32
739 remove(path.c_str());
740#endif
741 if (rename(tmp.c_str(), path.c_str()) != 0)
742 LogPrint("payout", "[payout] WARNING: could not replace %s\n", path.c_str());
743} 827}
744 828
745static void LoadPoolRounds() 829static void LoadPoolRounds()
@@ -844,6 +928,8 @@ static std::string PayoutComment(int blockHeight)
844// and the in-flight file being rewritten). The ledger wins. 928// and the in-flight file being rewritten). The ledger wins.
845static void ReconcilePayouts() 929static void ReconcilePayouts()
846{ 930{
931 VersionedBlob payingSnap;
932 {
847 std::lock_guard<std::mutex> exec(gPayoutExecMutex); 933 std::lock_guard<std::mutex> exec(gPayoutExecMutex);
848 std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex); 934 std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);
849 935
@@ -873,7 +959,7 @@ static void ReconcilePayouts()
873 " already recorded paid in the ledger -- not resending\n", 959 " already recorded paid in the ledger -- not resending\n",
874 settled, FormatMoney(settledAmount).c_str()); 960 settled, FormatMoney(settledAmount).c_str());
875 } 961 }
876 SavePayingLocked(); 962 payingSnap = SerializePayingLocked();
877 963
878 if (!gPaying.empty()) { 964 if (!gPaying.empty()) {
879 int64 owed = 0; 965 int64 owed = 0;
@@ -881,36 +967,53 @@ static void ReconcilePayouts()
881 LogPrint("payout", "[payout] resuming %zu in-flight payment(s) totalling %s\n", 967 LogPrint("payout", "[payout] resuming %zu in-flight payment(s) totalling %s\n",
882 gPaying.size(), FormatMoney(owed).c_str()); 968 gPaying.size(), FormatMoney(owed).c_str());
883 } 969 }
970 }
971 PublishPaying(payingSnap);
884} 972}
885 973
886static void RecordPoolRound(const PoolRoundProof& round) 974static void RecordPoolRound(const PoolRoundProof& round)
887{ 975{
888 std::lock_guard<std::mutex> lk(gPoolRoundsMutex); 976 VersionedBlob snap;
889 for (PoolRoundProof& existing : gPoolRounds) { 977 {
890 if (existing.id == round.id) { 978 std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
891 existing = round; 979 bool replaced = false;
892 SavePoolRoundsLocked(); 980 for (PoolRoundProof& existing : gPoolRounds) {
893 return; 981 if (existing.id == round.id) {
982 existing = round;
983 replaced = true;
984 break;
985 }
894 } 986 }
987 if (!replaced) {
988 gPoolRounds.push_back(round);
989 while (gPoolRounds.size() > MAX_POOL_ROUNDS)
990 gPoolRounds.erase(gPoolRounds.begin());
991 }
992 snap = SerializePoolRoundsLocked();
895 } 993 }
896 gPoolRounds.push_back(round); 994 PublishPoolRounds(snap);
897 while (gPoolRounds.size() > MAX_POOL_ROUNDS)
898 gPoolRounds.erase(gPoolRounds.begin());
899 SavePoolRoundsLocked();
900} 995}
901 996
902static void MarkPoolRoundPaid(const std::string& id, 997static void MarkPoolRoundPaid(const std::string& id,
903 const std::string& address, 998 const std::string& address,
904 const std::string& txid) 999 const std::string& txid)
905{ 1000{
906 std::lock_guard<std::mutex> lk(gPoolRoundsMutex); 1001 VersionedBlob snap;
907 for (PoolRoundProof& r : gPoolRounds) { 1002 {
908 if (r.id != id) 1003 std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
909 continue; 1004 bool found = false;
910 r.paidTxids[address] = txid; 1005 for (PoolRoundProof& r : gPoolRounds) {
911 SavePoolRoundsLocked(); 1006 if (r.id != id)
912 return; 1007 continue;
1008 r.paidTxids[address] = txid;
1009 found = true;
1010 break;
1011 }
1012 if (!found)
1013 return;
1014 snap = SerializePoolRoundsLocked();
913 } 1015 }
1016 PublishPoolRounds(snap);
914} 1017}
915 1018
916// Upstream's gPayoutExecQueue/gPayoutExecMutex/gPayoutThreadRunning are 1019// Upstream's gPayoutExecQueue/gPayoutExecMutex/gPayoutThreadRunning are
@@ -1039,9 +1142,13 @@ static void QueuePayouts(int blockHeight,
1039 round.forfeitedBadAddress = burned; 1142 round.forfeitedBadAddress = burned;
1040 1143
1041 { 1144 {
1042 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex); 1145 VersionedBlob snap;
1043 gPendingPayouts.push_back(pp); 1146 {
1044 SavePendingPayoutsLocked(); 1147 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
1148 gPendingPayouts.push_back(pp);
1149 snap = SerializePendingPayoutsLocked();
1150 }
1151 PublishPendingPayouts(snap);
1045 } 1152 }
1046 RecordPoolRound(round); 1153 RecordPoolRound(round);
1047 1154
@@ -1098,6 +1205,7 @@ static void FlushMaturePayouts()
1098 } 1205 }
1099 1206
1100 std::vector<PendingPayout> ready; 1207 std::vector<PendingPayout> ready;
1208 VersionedBlob ledgerSnap;
1101 { 1209 {
1102 std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex); 1210 std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);
1103 1211
@@ -1119,7 +1227,7 @@ static void FlushMaturePayouts()
1119 it = gPendingPayouts.erase(it); 1227 it = gPendingPayouts.erase(it);
1120 voided = true; 1228 voided = true;
1121 } 1229 }
1122 if (voided) SavePendingPayoutsLocked(); 1230 if (voided) ledgerSnap = SerializePendingPayoutsLocked();
1123 1231
1124 int64 now = GetTime(); 1232 int64 now = GetTime();
1125 for (const PendingPayout& pp : gPendingPayouts) { 1233 for (const PendingPayout& pp : gPendingPayouts) {
@@ -1132,10 +1240,15 @@ static void FlushMaturePayouts()
1132 ready.push_back(pp); 1240 ready.push_back(pp);
1133 } 1241 }
1134 } 1242 }
1243 // Outside the ledger mutex: publishing is I/O and the event loop is the
1244 // one thread that must never wait on the payout thread's disk.
1245 PublishPendingPayouts(ledgerSnap);
1135 1246
1136 if (ready.empty()) 1247 if (ready.empty())
1137 return; 1248 return;
1138 1249
1250 VersionedBlob payingSnap;
1251 {
1139 std::lock_guard<std::mutex> queue(gPayoutExecMutex); 1252 std::lock_guard<std::mutex> queue(gPayoutExecMutex);
1140 size_t added = 0; 1253 size_t added = 0;
1141 for (const PendingPayout& pp : ready) { 1254 for (const PendingPayout& pp : ready) {
@@ -1163,7 +1276,9 @@ static void FlushMaturePayouts()
1163 " (%zu of %zu recipients still owed)\n", 1276 " (%zu of %zu recipients still owed)\n",
1164 pp.matureAtHeight, id.c_str(), n, pp.amounts.size()); 1277 pp.matureAtHeight, id.c_str(), n, pp.amounts.size());
1165 } 1278 }
1166 if (added) SavePayingLocked(); 1279 if (added) payingSnap = SerializePayingLocked();
1280 }
1281 PublishPaying(payingSnap);
1167} 1282}
1168 1283
1169// Identity of a payment, for finding its record again after SendMoney. 1284// Identity of a payment, for finding its record again after SendMoney.
@@ -1182,49 +1297,60 @@ static bool SamePayment(const PayingRecord& a, const PayingRecord& b)
1182// separately by RecordPayoutTxid(). 1297// separately by RecordPayoutTxid().
1183static void ForgetPayment(const PayingRecord& r) 1298static void ForgetPayment(const PayingRecord& r)
1184{ 1299{
1185 std::lock_guard<std::mutex> lk(gPayoutExecMutex); 1300 VersionedBlob snap;
1186 for (auto it = gPaying.begin(); it != gPaying.end(); ++it) 1301 {
1187 if (SamePayment(*it, r)) { gPaying.erase(it); break; } 1302 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
1188 1303 for (auto it = gPaying.begin(); it != gPaying.end(); ++it)
1189 // When the last record for a round is gone the round is no longer in 1304 if (SamePayment(*it, r)) { gPaying.erase(it); break; }
1190 // flight, so a later flush may explode it again if the ledger still shows 1305
1191 // recipients owed (a give-up, for instance, leaves the entry standing). 1306 // When the last record for a round is gone the round is no longer in
1192 bool stillInFlight = false; 1307 // flight, so a later flush may explode it again if the ledger still
1193 for (const auto& e : gPaying) 1308 // shows recipients owed (a give-up, for instance, leaves the entry
1194 if (e.blockHash == r.blockHash) { stillInFlight = true; break; } 1309 // standing).
1195 if (!stillInFlight) 1310 bool stillInFlight = false;
1196 gQueuedPayouts.erase(PayoutIdForBlock(r.blockHeight, r.blockHash)); 1311 for (const auto& e : gPaying)
1197 1312 if (e.blockHash == r.blockHash) { stillInFlight = true; break; }
1198 SavePayingLocked(); 1313 if (!stillInFlight)
1314 gQueuedPayouts.erase(PayoutIdForBlock(r.blockHeight, r.blockHash));
1315
1316 snap = SerializePayingLocked();
1317 }
1318 PublishPaying(snap);
1199} 1319}
1200 1320
1201// Persist an updated attempt count / backoff for a payment still owed. 1321// Persist an updated attempt count / backoff for a payment still owed.
1202static void ReschedulePayment(const PayingRecord& r) 1322static void ReschedulePayment(const PayingRecord& r)
1203{ 1323{
1204 std::lock_guard<std::mutex> lk(gPayoutExecMutex); 1324 VersionedBlob snap;
1205 for (auto& e : gPaying) 1325 {
1206 if (SamePayment(e, r)) { 1326 std::lock_guard<std::mutex> lk(gPayoutExecMutex);
1207 e.retries = r.retries; 1327 for (auto& e : gPaying)
1208 e.deferrals = r.deferrals; 1328 if (SamePayment(e, r)) {
1209 e.retryAfter = r.retryAfter; 1329 e.retries = r.retries;
1210 // Carry the cleared probe flag back, or every retry re-scans the 1330 e.deferrals = r.deferrals;
1211 // whole wallet -- the cost this merge is specifically removing. 1331 e.retryAfter = r.retryAfter;
1212 e.fProbeWallet = r.fProbeWallet; 1332 // Carry the cleared probe flag back, or every retry re-scans
1213 break; 1333 // the whole wallet -- the cost this merge is specifically
1214 } 1334 // removing.
1215 SavePayingLocked(); 1335 e.fProbeWallet = r.fProbeWallet;
1336 break;
1337 }
1338 snap = SerializePayingLocked();
1339 }
1340 PublishPaying(snap);
1216} 1341}
1217 1342
1218// Record that a recipient has been paid, in the ledger, which is the only 1343// Record that a recipient has been paid, in the ledger, which is the only
1219// durable statement that the debt is settled. 1344// durable statement that the debt is settled.
1220// 1345//
1221// The file write happens with gPayoutLedgerMutex RELEASED. Upstream calls 1346// The file write happens with gPayoutLedgerMutex RELEASED. Upstream writes the
1222// SavePendingPayoutsLocked() inside the per-recipient loop with the mutex 1347// file inside the per-recipient loop with the mutex held, so every recipient
1223// held, so every recipient costs the event loop a full serialise+fsync it has 1348// costs the event loop a full serialise+fsync it has to wait for in
1224// to wait for in FlushMaturePayouts. 1349// FlushMaturePayouts. Ordering against the event loop's own writes is the
1350// snapshot version's job, not the mutex's -- see PublishVersioned.
1225static void RecordPayoutTxid(const PayingRecord& r, const std::string& txid) 1351static void RecordPayoutTxid(const PayingRecord& r, const std::string& txid)
1226{ 1352{
1227 std::string blob; 1353 VersionedBlob blob;
1228 bool complete = false; 1354 bool complete = false;
1229 { 1355 {
1230 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex); 1356 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
@@ -1243,7 +1369,7 @@ static void RecordPayoutTxid(const PayingRecord& r, const std::string& txid)
1243 } 1369 }
1244 blob = SerializePendingPayoutsLocked(); 1370 blob = SerializePendingPayoutsLocked();
1245 } 1371 }
1246 WritePendingPayoutsFile(blob); 1372 PublishPendingPayouts(blob);
1247 // The round proof records a txid per address, so this is per recipient and 1373 // The round proof records a txid per address, so this is per recipient and
1248 // not only on completion. 1374 // not only on completion.
1249 MarkPoolRoundPaid(PayoutIdForBlock(r.blockHeight, r.blockHash), r.address, txid); 1375 MarkPoolRoundPaid(PayoutIdForBlock(r.blockHeight, r.blockHash), r.address, txid);
@@ -1259,7 +1385,7 @@ static void RecordPayoutTxid(const PayingRecord& r, const std::string& txid)
1259// is not reported as sent. Once nothing is left owed the entry retires. 1385// is not reported as sent. Once nothing is left owed the entry retires.
1260static void ForfeitPayoutRecipient(const PayingRecord& r) 1386static void ForfeitPayoutRecipient(const PayingRecord& r)
1261{ 1387{
1262 std::string blob; 1388 VersionedBlob blob;
1263 { 1389 {
1264 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex); 1390 std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
1265 for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ++it) { 1391 for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ++it) {
@@ -1272,7 +1398,7 @@ static void ForfeitPayoutRecipient(const PayingRecord& r)
1272 } 1398 }
1273 blob = SerializePendingPayoutsLocked(); 1399 blob = SerializePendingPayoutsLocked();
1274 } 1400 }
1275 WritePendingPayoutsFile(blob); 1401 PublishPendingPayouts(blob);
1276} 1402}
1277 1403
1278// Payout execution thread: works through gPaying, calls SendMoney. 1404// Payout execution thread: works through gPaying, calls SendMoney.
@@ -1554,24 +1680,11 @@ static void WritePoolStatusJsonLocked()
1554 } 1680 }
1555 } 1681 }
1556 1682
1557 std::string path = PoolStatusFilePath(); 1683 // No version guard and no write mutex here, deliberately: RefreshStats is
1558 std::string tmp = path + ".tmp"; 1684 // the only caller and it only ever runs on the event loop, so there is a
1559 FILE* f = fopen(tmp.c_str(), "w"); 1685 // single writer of pool_status.json by construction. It shares
1560 if (!f) { 1686 // WriteFileAtomic so the durability rules stay in one place.
1561 LogPrint("pool", "[pool] WARNING: could not write status file %s\n", tmp.c_str()); 1687 WriteFileAtomic(PoolStatusFilePath(), j.dump(2), "pool status");
1562 return;
1563 }
1564 std::string s = j.dump(2);
1565 fwrite(s.c_str(), 1, s.size(), f);
1566 if (fclose(f) != 0) {
1567 LogPrint("pool", "[pool] WARNING: could not close status file %s\n", tmp.c_str());
1568 return;
1569 }
1570#ifdef _WIN32
1571 remove(path.c_str());
1572#endif
1573 if (rename(tmp.c_str(), path.c_str()) != 0)
1574 LogPrint("pool", "[pool] WARNING: could not replace status file %s\n", path.c_str());
1575} 1688}
1576 1689
1577// --------------------------------------------------------------------------- 1690// ---------------------------------------------------------------------------