payout: serialise ledger file writes and order them by snapshot version

Splitting SavePendingPayoutsLocked() into a serialise step that needs
gPayoutLedgerMutex and a write step that does not left the write step with
no serialisation at all. The event loop wrote pending_payouts.json from
QueuePayouts and the orphan-void path with the ledger mutex HELD; the payout
thread wrote it from RecordPayoutTxid and ForfeitPayoutRecipient with the
mutex RELEASED. Both fopen()ed the same "<path>.tmp" with independent handles
and both rename()d it.

Two ways that ends, both live money:

  - the renamed file is a mixture of two images. LoadPendingPayouts throws
    inside its parse loop, logs one warning, and the node comes up with a
    nearly empty ledger -- every obligation not yet exploded into an in-flight
    record is simply gone, with nothing on disk saying anyone was owed it.

  - a lost update: the payout thread's older snapshot lands after the event
    loop's newer one and silently erases a whole round that had just been
    queued.

Neither parent had this. Upstream writes under the mutex everywhere; the
pre-merge tree had one writer thread and no ledger mutex. The merge invented
the window while fixing a real problem -- the event loop should not block in
FlushMaturePayouts behind a per-recipient fsync on the payout thread -- so
putting the write back under the ledger mutex is not the fix.

Instead: a per-file write mutex that is never held together with a data
mutex, plus a version on every snapshot.

  - serialising happens under the data mutex and stamps a monotonically
    increasing sequence number (gPayoutLedgerSeq).
  - publishing happens with the data mutex released, under the file mutex,
    and drops any blob whose sequence number is not strictly newer than the
    one already on disk.

Because each blob is a whole-state dump of a single in-memory structure and
the sequence number is assigned under the mutex guarding that structure, a
higher sequence number is by construction a later state -- so discarding an
older blob loses nothing, and the file can only move forwards. Writers may
queue and complete in any order and the published file is still the newest
state anyone has serialised. A failed write deliberately does not advance the
published sequence, so the next write still tries.

The same treatment is applied to paying_payouts.json and pool_rounds.json.
Neither had the interleaving race -- gPayoutExecMutex and gPoolRoundsMutex
already serialise their writers -- but both did their I/O under a mutex the
event loop takes (FlushMaturePayouts, RecordPoolRound), so both were a place
for the event loop to wait on the payout thread's disk. They are now written
outside those mutexes on the same versioned terms, which is also what makes
adding fsync to these paths safe.

pool_status.json keeps a plain atomic write with no version guard: RefreshStats
is its only caller and only ever runs on the event loop, so it has a single
writer by construction. It shares the same WriteFileAtomic helper so the
durability rules live in one place.

No on-disk format change.

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");
 }
 
 // ---------------------------------------------------------------------------