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 {
 };
 static std::vector<PendingPayout> gPendingPayouts;
 static std::mutex                 gPayoutLedgerMutex;
+// Bumped on every serialisation, under gPayoutLedgerMutex. See PublishVersioned.
+static uint64                     gPayoutLedgerSeq = 0;
+// Serialises writers of pending_payouts.json, and guards the sequence number
+// of the image currently on disk. Never taken while a data mutex is held.
+static std::mutex                 gPayoutLedgerFileMutex;
+static uint64                     gPayoutLedgerPublishedSeq = 0;
 
 struct PoolRoundProof {
     std::string                  id;
@@ -388,6 +394,9 @@ struct PoolRoundProof {
 };
 static std::vector<PoolRoundProof> gPoolRounds;
 static std::mutex                  gPoolRoundsMutex;
+static uint64                      gPoolRoundsSeq = 0; // guarded by gPoolRoundsMutex
+static std::mutex                  gPoolRoundsFileMutex;
+static uint64                      gPoolRoundsPublishedSeq = 0; // guarded by gPoolRoundsFileMutex
 static const size_t                MAX_POOL_ROUNDS = 200;
 
 static std::string PayoutFilePath()
@@ -402,6 +411,104 @@ static std::string PoolRoundsFilePath()
     return GetAppDir() + "/pool_rounds.json";
 }
 
+// ---------------------------------------------------------------------------
+// Atomic, ordered publication of the payout state files
+// ---------------------------------------------------------------------------
+//
+// Every one of these files is a full dump of an in-memory structure, written
+// tmp-then-rename so a crash cannot truncate it. Two things have to be true
+// for that to actually be safe, and only the first was:
+//
+//   1. the rename must be atomic (it is), and
+//   2. exactly one thread at a time may be writing the temp file, and a write
+//      must never publish an OLDER image over a newer one.
+//
+// (2) was broken for pending_payouts.json. The event loop wrote it with
+// gPayoutLedgerMutex held, the payout thread wrote it with the mutex
+// released, and nothing serialised the two: both opened the same
+// "<path>.tmp" with independent handles and both renamed it. The interleaved
+// file that lands is either a mixture of two images (LoadPendingPayouts then
+// throws mid-loop and comes up with a nearly empty ledger -- every obligation
+// not yet exploded, gone) or a whole newly-queued round silently overwritten
+// by an older snapshot that predates it.
+//
+// The fix is a per-file write mutex, held only across the I/O and never
+// across a data-structure mutex, plus a version on the snapshot. Serialising
+// happens under the data mutex and stamps a monotonically increasing sequence
+// number; publishing happens with the data mutex released and refuses any
+// blob whose sequence number is not newer than what is already on disk. So
+// writers can queue in any order and finish in any order -- the file only
+// ever moves forwards. Crucially, no thread ever waits for I/O while holding
+// gPayoutLedgerMutex / gPayoutExecMutex / gPoolRoundsMutex, which is the
+// contention the merge was right to remove from the event loop.
+
+// A serialised file image plus the version of the in-memory state it was
+// taken from.
+struct VersionedBlob {
+    std::string blob;
+    uint64      seq = 0;
+};
+
+// Replace `path` with `data` atomically. Caller must hold the write mutex for
+// `path`: the temp name is derived from the target, so two threads in here at
+// once for one path is precisely the race this exists to prevent.
+static bool WriteFileAtomic(const std::string& path, const std::string& data,
+                            const char* what)
+{
+    std::string tmp = path + ".tmp";
+    FILE* f = fopen(tmp.c_str(), "w");
+    if (!f) {
+        LogPrint("payout", "[payout] WARNING: could not write %s (%s)\n",
+                 tmp.c_str(), what);
+        return false;
+    }
+    if (!data.empty() && fwrite(data.c_str(), 1, data.size(), f) != data.size()) {
+        LogPrint("payout", "[payout] WARNING: short write to %s (%s)\n",
+                 tmp.c_str(), what);
+        fclose(f);
+        return false;
+    }
+    // fclose() is where buffered data actually reaches the OS, so a failure
+    // here means the .tmp is short -- renaming it over the real file would
+    // destroy the good copy, which is exactly what these files exist to
+    // prevent.
+    if (fclose(f) != 0) {
+        LogPrint("payout", "[payout] WARNING: could not close %s (%s)\n",
+                 tmp.c_str(), what);
+        return false;
+    }
+#ifdef _WIN32
+    // rename() on Windows will not replace an existing file.
+    remove(path.c_str());
+#endif
+    if (rename(tmp.c_str(), path.c_str()) != 0) {
+        LogPrint("payout", "[payout] WARNING: could not replace %s (%s)\n",
+                 path.c_str(), what);
+        return false;
+    }
+    return true;
+}
+
+// Publish a versioned image of some in-memory state to `path`.
+//
+// `fileMutex` serialises writers of `path` and guards `publishedSeq`. A blob
+// older than what is already published is dropped: these files are whole-state
+// dumps, so a higher sequence number is by construction a later state of the
+// same structure and publishing it loses nothing. On a write failure
+// `publishedSeq` is deliberately left alone, so the next (newer) write still
+// gets a chance to land.
+static void PublishVersioned(const std::string& path, const VersionedBlob& v,
+                             std::mutex& fileMutex, uint64& publishedSeq,
+                             const char* what)
+{
+    std::lock_guard<std::mutex> lk(fileMutex);
+    if (v.seq <= publishedSeq)
+        return;
+    if (!WriteFileAtomic(path, v.blob, what))
+        return;
+    publishedSeq = v.seq;
+}
+
 // Identity of a payout round. Used as the pool-round proof id and, more
 // importantly, as the `pool_payout_id` written into wallet transaction
 // metadata and matched by FindRecordedPoolPayout().
@@ -431,14 +538,16 @@ static bool PayoutFullyPaid(const PendingPayout& pp)
     return true;
 }
 
-// Serialise the ledger. Caller must hold gPayoutLedgerMutex.
+// Serialise the ledger and stamp it with the version it was taken at. Caller
+// must hold gPayoutLedgerMutex.
 //
-// Split from the file write so the payout thread can hold the mutex just long
-// enough to read the structure and then do the I/O with it released. Upstream
-// writes the file inside the per-recipient loop with gPayoutLedgerMutex held,
-// while the event loop's FlushMaturePayouts blocks on that same mutex -- one
-// slow fsync per recipient, on the thread the miners' I/O shares.
-static std::string SerializePendingPayoutsLocked()
+// Split from the file write so no thread holds the ledger mutex across the
+// I/O. Upstream writes the file inside the per-recipient loop with
+// gPayoutLedgerMutex held, while the event loop's FlushMaturePayouts blocks on
+// that same mutex -- one slow fsync per recipient, on the thread the miners'
+// I/O shares. The sequence number is what lets the write happen later, and
+// possibly out of order, without a stale image reaching the disk.
+static VersionedBlob SerializePendingPayoutsLocked()
 {
     nlohmann::json arr = nlohmann::json::array();
     for (const auto& pp : gPendingPayouts) {
@@ -457,38 +566,18 @@ static std::string SerializePendingPayoutsLocked()
         obj["paidTxids"] = paid;
         arr.push_back(obj);
     }
-    return arr.dump(2);
-}
-
-// Write the serialised ledger. Holds no lock; safe to call with the mutex
-// released. Atomic tmp+rename so a crash mid-write cannot truncate the record
-// of who is still owed money.
-static void WritePendingPayoutsFile(const std::string& s)
-{
-    std::string path = PayoutFilePath();
-    std::string tmp = path + ".tmp";
-    FILE* f = fopen(tmp.c_str(), "w");
-    if (!f) {
-        LogPrint("payout", "[payout] WARNING: could not write %s\n", tmp.c_str());
-        return;
-    }
-    fwrite(s.c_str(), 1, s.size(), f);
-    if (fclose(f) != 0) {
-        LogPrint("payout", "[payout] WARNING: could not close %s\n", tmp.c_str());
-        return;
-    }
-#ifdef _WIN32
-    remove(path.c_str());
-#endif
-    if (rename(tmp.c_str(), path.c_str()) != 0)
-        LogPrint("payout", "[payout] WARNING: could not replace %s\n", path.c_str());
+    VersionedBlob v;
+    v.blob = arr.dump(2);
+    v.seq  = ++gPayoutLedgerSeq;
+    return v;
 }
 
-// Convenience for the cold paths (startup, queueing) where holding the mutex
-// across the write costs nothing. Caller must hold gPayoutLedgerMutex.
-static void SavePendingPayoutsLocked()
+// Publish a ledger snapshot. Holds no data mutex; safe -- and intended -- to
+// call with gPayoutLedgerMutex released.
+static void PublishPendingPayouts(const VersionedBlob& v)
 {
-    WritePendingPayoutsFile(SerializePendingPayoutsLocked());
+    PublishVersioned(PayoutFilePath(), v, gPayoutLedgerFileMutex,
+                     gPayoutLedgerPublishedSeq, "pending payout ledger");
 }
 
 static void LoadPendingPayouts()
@@ -577,6 +666,9 @@ struct PayingRecord {
 };
 static std::vector<PayingRecord> gPaying; // guarded by gPayoutExecMutex
 static std::mutex                gPayoutExecMutex;
+static uint64                    gPayingSeq = 0; // guarded by gPayoutExecMutex
+static std::mutex                gPayingFileMutex;
+static uint64                    gPayingPublishedSeq = 0; // guarded by gPayingFileMutex
 static std::atomic<bool>         gPayoutThreadRunning{false};
 
 static std::string PayingFilePath()
@@ -584,8 +676,16 @@ static std::string PayingFilePath()
     return GetAppDir() + "/paying_payouts.json";
 }
 
-// Caller must hold gPayoutExecMutex.
-static void SavePayingLocked()
+// Serialise the in-flight list. Caller must hold gPayoutExecMutex.
+//
+// gPayoutExecMutex already serialises every writer of this file, so it never
+// had the interleaving race pending_payouts.json had. It is still split the
+// same way for a different reason: FlushMaturePayouts runs on the event loop
+// and holds gPayoutExecMutex, so writing (and, since this series, fsyncing)
+// the file under that mutex parks miner I/O behind the payout thread's disk.
+// The version stamp keeps the file monotonic now that the write is outside
+// the mutex that orders the writers.
+static VersionedBlob SerializePayingLocked()
 {
     nlohmann::json arr = nlohmann::json::array();
     for (const auto& r : gPaying) {
@@ -599,33 +699,17 @@ static void SavePayingLocked()
         obj["deferrals"]      = r.deferrals;
         arr.push_back(obj);
     }
-    std::string path = PayingFilePath();
-    // Write to a temporary file and rename, so a crash mid-write cannot leave
-    // a truncated list of who is still owed money.
-    std::string tmp = path + ".tmp";
-    FILE* f = fopen(tmp.c_str(), "w");
-    if (!f) {
-        LogPrint("payout", "[payout] WARNING: could not write %s\n", tmp.c_str());
-        return;
-    }
-    std::string s = arr.dump(2);
-    fwrite(s.c_str(), 1, s.size(), f);
-    // fclose() is where buffered data actually reaches the OS, so a failure
-    // here means the .tmp is short -- renaming it over the real file would
-    // destroy the good copy. Upstream's check, taken because a truncated
-    // list of who is owed money is exactly what this function exists to
-    // prevent. rename() on Windows will not replace an existing file, hence
-    // the remove() first.
-    if (fclose(f) != 0) {
-        LogPrint("payout", "[payout] WARNING: could not close %s\n", tmp.c_str());
-        return;
-    }
-#ifdef _WIN32
-    remove(path.c_str());
-#endif
-    if (rename(tmp.c_str(), path.c_str()) != 0)
-        LogPrint("payout", "[payout] WARNING: could not rename %s -> %s\n",
-                 tmp.c_str(), path.c_str());
+    VersionedBlob v;
+    v.blob = arr.dump(2);
+    v.seq  = ++gPayingSeq;
+    return v;
+}
+
+// Publish an in-flight-list snapshot. Call with gPayoutExecMutex released.
+static void PublishPaying(const VersionedBlob& v)
+{
+    PublishVersioned(PayingFilePath(), v, gPayingFileMutex,
+                     gPayingPublishedSeq, "in-flight payment list");
 }
 
 static void LoadPaying()
@@ -713,7 +797,14 @@ static nlohmann::json PoolRoundToJson(const PoolRoundProof& r)
     return j;
 }
 
-static void SavePoolRoundsLocked()
+// Serialise the published round proofs. Caller must hold gPoolRoundsMutex.
+//
+// Like the in-flight list, gPoolRoundsMutex already serialises the writers
+// (RecordPoolRound on the event loop, MarkPoolRoundPaid on the payout thread),
+// so this file never had the interleaving race either -- but the event loop
+// should not be waiting on the payout thread's fsync to publish a round, so
+// the write moves outside the mutex on the same versioned terms.
+static VersionedBlob SerializePoolRoundsLocked()
 {
     nlohmann::json root;
     root["schema"] = "bitflash-pool-rounds-1";
@@ -722,24 +813,17 @@ static void SavePoolRoundsLocked()
     for (const PoolRoundProof& r : gPoolRounds)
         root["rounds"].push_back(PoolRoundToJson(r));
 
-    std::string path = PoolRoundsFilePath();
-    std::string tmp = path + ".tmp";
-    FILE* f = fopen(tmp.c_str(), "w");
-    if (!f) {
-        LogPrint("payout", "[payout] WARNING: could not write %s\n", tmp.c_str());
-        return;
-    }
-    std::string s = root.dump(2);
-    fwrite(s.c_str(), 1, s.size(), f);
-    if (fclose(f) != 0) {
-        LogPrint("payout", "[payout] WARNING: could not close %s\n", tmp.c_str());
-        return;
-    }
-#ifdef _WIN32
-    remove(path.c_str());
-#endif
-    if (rename(tmp.c_str(), path.c_str()) != 0)
-        LogPrint("payout", "[payout] WARNING: could not replace %s\n", path.c_str());
+    VersionedBlob v;
+    v.blob = root.dump(2);
+    v.seq  = ++gPoolRoundsSeq;
+    return v;
+}
+
+// Publish a round-proof snapshot. Call with gPoolRoundsMutex released.
+static void PublishPoolRounds(const VersionedBlob& v)
+{
+    PublishVersioned(PoolRoundsFilePath(), v, gPoolRoundsFileMutex,
+                     gPoolRoundsPublishedSeq, "pool round proofs");
 }
 
 static void LoadPoolRounds()
@@ -844,6 +928,8 @@ static std::string PayoutComment(int blockHeight)
 // and the in-flight file being rewritten). The ledger wins.
 static void ReconcilePayouts()
 {
+    VersionedBlob payingSnap;
+    {
     std::lock_guard<std::mutex> exec(gPayoutExecMutex);
     std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);
 
@@ -873,7 +959,7 @@ static void ReconcilePayouts()
                  " already recorded paid in the ledger -- not resending\n",
                  settled, FormatMoney(settledAmount).c_str());
     }
-    SavePayingLocked();
+    payingSnap = SerializePayingLocked();
 
     if (!gPaying.empty()) {
         int64 owed = 0;
@@ -881,36 +967,53 @@ static void ReconcilePayouts()
         LogPrint("payout", "[payout] resuming %zu in-flight payment(s) totalling %s\n",
                  gPaying.size(), FormatMoney(owed).c_str());
     }
+    }
+    PublishPaying(payingSnap);
 }
 
 static void RecordPoolRound(const PoolRoundProof& round)
 {
-    std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
-    for (PoolRoundProof& existing : gPoolRounds) {
-        if (existing.id == round.id) {
-            existing = round;
-            SavePoolRoundsLocked();
-            return;
+    VersionedBlob snap;
+    {
+        std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
+        bool replaced = false;
+        for (PoolRoundProof& existing : gPoolRounds) {
+            if (existing.id == round.id) {
+                existing = round;
+                replaced = true;
+                break;
+            }
         }
+        if (!replaced) {
+            gPoolRounds.push_back(round);
+            while (gPoolRounds.size() > MAX_POOL_ROUNDS)
+                gPoolRounds.erase(gPoolRounds.begin());
+        }
+        snap = SerializePoolRoundsLocked();
     }
-    gPoolRounds.push_back(round);
-    while (gPoolRounds.size() > MAX_POOL_ROUNDS)
-        gPoolRounds.erase(gPoolRounds.begin());
-    SavePoolRoundsLocked();
+    PublishPoolRounds(snap);
 }
 
 static void MarkPoolRoundPaid(const std::string& id,
                               const std::string& address,
                               const std::string& txid)
 {
-    std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
-    for (PoolRoundProof& r : gPoolRounds) {
-        if (r.id != id)
-            continue;
-        r.paidTxids[address] = txid;
-        SavePoolRoundsLocked();
-        return;
+    VersionedBlob snap;
+    {
+        std::lock_guard<std::mutex> lk(gPoolRoundsMutex);
+        bool found = false;
+        for (PoolRoundProof& r : gPoolRounds) {
+            if (r.id != id)
+                continue;
+            r.paidTxids[address] = txid;
+            found = true;
+            break;
+        }
+        if (!found)
+            return;
+        snap = SerializePoolRoundsLocked();
     }
+    PublishPoolRounds(snap);
 }
 
 // Upstream's gPayoutExecQueue/gPayoutExecMutex/gPayoutThreadRunning are
@@ -1039,9 +1142,13 @@ static void QueuePayouts(int blockHeight,
     round.forfeitedBadAddress = burned;
 
     {
-        std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
-        gPendingPayouts.push_back(pp);
-        SavePendingPayoutsLocked();
+        VersionedBlob snap;
+        {
+            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
+            gPendingPayouts.push_back(pp);
+            snap = SerializePendingPayoutsLocked();
+        }
+        PublishPendingPayouts(snap);
     }
     RecordPoolRound(round);
 
@@ -1098,6 +1205,7 @@ static void FlushMaturePayouts()
     }
 
     std::vector<PendingPayout> ready;
+    VersionedBlob ledgerSnap;
     {
         std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);
 
@@ -1119,7 +1227,7 @@ static void FlushMaturePayouts()
             it = gPendingPayouts.erase(it);
             voided = true;
         }
-        if (voided) SavePendingPayoutsLocked();
+        if (voided) ledgerSnap = SerializePendingPayoutsLocked();
 
         int64 now = GetTime();
         for (const PendingPayout& pp : gPendingPayouts) {
@@ -1132,10 +1240,15 @@ static void FlushMaturePayouts()
             ready.push_back(pp);
         }
     }
+    // Outside the ledger mutex: publishing is I/O and the event loop is the
+    // one thread that must never wait on the payout thread's disk.
+    PublishPendingPayouts(ledgerSnap);
 
     if (ready.empty())
         return;
 
+    VersionedBlob payingSnap;
+    {
     std::lock_guard<std::mutex> queue(gPayoutExecMutex);
     size_t added = 0;
     for (const PendingPayout& pp : ready) {
@@ -1163,7 +1276,9 @@ static void FlushMaturePayouts()
                  " (%zu of %zu recipients still owed)\n",
                  pp.matureAtHeight, id.c_str(), n, pp.amounts.size());
     }
-    if (added) SavePayingLocked();
+    if (added) payingSnap = SerializePayingLocked();
+    }
+    PublishPaying(payingSnap);
 }
 
 // Identity of a payment, for finding its record again after SendMoney.
@@ -1182,49 +1297,60 @@ static bool SamePayment(const PayingRecord& a, const PayingRecord& b)
 // separately by RecordPayoutTxid().
 static void ForgetPayment(const PayingRecord& r)
 {
-    std::lock_guard<std::mutex> lk(gPayoutExecMutex);
-    for (auto it = gPaying.begin(); it != gPaying.end(); ++it)
-        if (SamePayment(*it, r)) { gPaying.erase(it); break; }
-
-    // When the last record for a round is gone the round is no longer in
-    // flight, so a later flush may explode it again if the ledger still shows
-    // recipients owed (a give-up, for instance, leaves the entry standing).
-    bool stillInFlight = false;
-    for (const auto& e : gPaying)
-        if (e.blockHash == r.blockHash) { stillInFlight = true; break; }
-    if (!stillInFlight)
-        gQueuedPayouts.erase(PayoutIdForBlock(r.blockHeight, r.blockHash));
-
-    SavePayingLocked();
+    VersionedBlob snap;
+    {
+        std::lock_guard<std::mutex> lk(gPayoutExecMutex);
+        for (auto it = gPaying.begin(); it != gPaying.end(); ++it)
+            if (SamePayment(*it, r)) { gPaying.erase(it); break; }
+
+        // When the last record for a round is gone the round is no longer in
+        // flight, so a later flush may explode it again if the ledger still
+        // shows recipients owed (a give-up, for instance, leaves the entry
+        // standing).
+        bool stillInFlight = false;
+        for (const auto& e : gPaying)
+            if (e.blockHash == r.blockHash) { stillInFlight = true; break; }
+        if (!stillInFlight)
+            gQueuedPayouts.erase(PayoutIdForBlock(r.blockHeight, r.blockHash));
+
+        snap = SerializePayingLocked();
+    }
+    PublishPaying(snap);
 }
 
 // Persist an updated attempt count / backoff for a payment still owed.
 static void ReschedulePayment(const PayingRecord& r)
 {
-    std::lock_guard<std::mutex> lk(gPayoutExecMutex);
-    for (auto& e : gPaying)
-        if (SamePayment(e, r)) {
-            e.retries      = r.retries;
-            e.deferrals    = r.deferrals;
-            e.retryAfter   = r.retryAfter;
-            // Carry the cleared probe flag back, or every retry re-scans the
-            // whole wallet -- the cost this merge is specifically removing.
-            e.fProbeWallet = r.fProbeWallet;
-            break;
-        }
-    SavePayingLocked();
+    VersionedBlob snap;
+    {
+        std::lock_guard<std::mutex> lk(gPayoutExecMutex);
+        for (auto& e : gPaying)
+            if (SamePayment(e, r)) {
+                e.retries      = r.retries;
+                e.deferrals    = r.deferrals;
+                e.retryAfter   = r.retryAfter;
+                // Carry the cleared probe flag back, or every retry re-scans
+                // the whole wallet -- the cost this merge is specifically
+                // removing.
+                e.fProbeWallet = r.fProbeWallet;
+                break;
+            }
+        snap = SerializePayingLocked();
+    }
+    PublishPaying(snap);
 }
 
 // Record that a recipient has been paid, in the ledger, which is the only
 // durable statement that the debt is settled.
 //
-// The file write happens with gPayoutLedgerMutex RELEASED. Upstream calls
-// SavePendingPayoutsLocked() inside the per-recipient loop with the mutex
-// held, so every recipient costs the event loop a full serialise+fsync it has
-// to wait for in FlushMaturePayouts.
+// The file write happens with gPayoutLedgerMutex RELEASED. Upstream writes the
+// file inside the per-recipient loop with the mutex held, so every recipient
+// costs the event loop a full serialise+fsync it has to wait for in
+// FlushMaturePayouts. Ordering against the event loop's own writes is the
+// snapshot version's job, not the mutex's -- see PublishVersioned.
 static void RecordPayoutTxid(const PayingRecord& r, const std::string& txid)
 {
-    std::string blob;
+    VersionedBlob blob;
     bool complete = false;
     {
         std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
@@ -1243,7 +1369,7 @@ static void RecordPayoutTxid(const PayingRecord& r, const std::string& txid)
         }
         blob = SerializePendingPayoutsLocked();
     }
-    WritePendingPayoutsFile(blob);
+    PublishPendingPayouts(blob);
     // The round proof records a txid per address, so this is per recipient and
     // not only on completion.
     MarkPoolRoundPaid(PayoutIdForBlock(r.blockHeight, r.blockHash), r.address, txid);
@@ -1259,7 +1385,7 @@ static void RecordPayoutTxid(const PayingRecord& r, const std::string& txid)
 // is not reported as sent. Once nothing is left owed the entry retires.
 static void ForfeitPayoutRecipient(const PayingRecord& r)
 {
-    std::string blob;
+    VersionedBlob blob;
     {
         std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
         for (auto it = gPendingPayouts.begin(); it != gPendingPayouts.end(); ++it) {
@@ -1272,7 +1398,7 @@ static void ForfeitPayoutRecipient(const PayingRecord& r)
         }
         blob = SerializePendingPayoutsLocked();
     }
-    WritePendingPayoutsFile(blob);
+    PublishPendingPayouts(blob);
 }
 
 // Payout execution thread: works through gPaying, calls SendMoney.
@@ -1554,24 +1680,11 @@ static void WritePoolStatusJsonLocked()
         }
     }
 
-    std::string path = PoolStatusFilePath();
-    std::string tmp = path + ".tmp";
-    FILE* f = fopen(tmp.c_str(), "w");
-    if (!f) {
-        LogPrint("pool", "[pool] WARNING: could not write status file %s\n", tmp.c_str());
-        return;
-    }
-    std::string s = j.dump(2);
-    fwrite(s.c_str(), 1, s.size(), f);
-    if (fclose(f) != 0) {
-        LogPrint("pool", "[pool] WARNING: could not close status file %s\n", tmp.c_str());
-        return;
-    }
-#ifdef _WIN32
-    remove(path.c_str());
-#endif
-    if (rename(tmp.c_str(), path.c_str()) != 0)
-        LogPrint("pool", "[pool] WARNING: could not replace status file %s\n", path.c_str());
+    // No version guard and no write mutex here, deliberately: RefreshStats is
+    // the only caller and it only ever runs on the event loop, so there is a
+    // single writer of pool_status.json by construction. It shares
+    // WriteFileAtomic so the durability rules stay in one place.
+    WriteFileAtomic(PoolStatusFilePath(), j.dump(2), "pool status");
 }
 
 // ---------------------------------------------------------------------------