payout: build pool_status.json under the locks, write it after releasing them

RefreshStats runs on the event loop -- the one thread that reads and
writes every miner's socket -- and took gStatsMutex and then
gPayoutLedgerMutex at function scope, then called
WritePoolStatusJsonLocked() with both still held. Since 86b39b9 that
write ends in fsync() on the file and fsync() on its directory. So every
ten seconds the event loop performed two synchronous disk flushes
holding the mutex the payout thread must take to record that a miner has
been paid, and the payout thread waited behind them. Measured here on
ext4/LVM/SSD: 3.78 ms mean per publish. On the pool's shared virtual
disk it is higher and it has no upper bound under load.

That is worse than the code the merge replaced, which wrote the same
file under the same locks but buffered and without fsync. It also made
33f2e8a's headline claim -- "no thread waits on I/O holding
gPayoutLedgerMutex, gPayoutExecMutex or gPoolRoundsMutex" -- false on
the commit that asserted it.

Split the same way as the three ledgers: SerializePoolStatusLocked()
returns the image, PublishPoolStatus() writes it, and RefreshStats calls
the second one after leaving the scope that holds the first. The ledger
lock is also scoped to the loop that actually needs it, which is the
gPayoutViews rebuild; the serialiser never needed it, since those rows
were already copied.

No version guard is added and none is needed: RefreshStats is still the
only caller of either half, it still runs only on the event loop, and it
now calls them in sequence on that one thread, so pool_status.json still
has a single writer by construction. The comment saying so has been
rechecked against the new shape rather than carried over.

The claim is now tested rather than asserted. The self-test points
strPoolStatusFile at a path whose ".tmp" is a FIFO, so the publish parks
in open(2) until the test chooses to read, and every mutex the write
must not be holding is required to be acquirable while it is parked.
Against the previous code that test cannot pass: the writer sits inside
fopen() holding gStatsMutex and gPayoutLedgerMutex, and nothing releases
them until the write is allowed to finish. Verified by reintroducing the
defect -- two assertions fail, deterministically.

Property: pool_status.json is serialised under the locks and written
with all four released, so no payout mutex is ever held across I/O.

1 files changed, 237 insertions, 53 deletions

diff --git a/src/rpc.cpp b/src/rpc.cpp
index be265db..403cb14 100644
--- a/src/rpc.cpp
+++ b/src/rpc.cpp
@@ -451,9 +451,20 @@ static std::string PoolRoundsFilePath()
 // 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.
+// ever moves forwards.
+//
+// The other half of the rule is that no thread waits for I/O while holding
+// gPayoutLedgerMutex, gPayoutExecMutex or gPoolRoundsMutex, which is the
+// contention the merge was right to remove from the event loop. That was
+// claimed here before it was true: pool_status.json is written by the same
+// WriteFileAtomic and RefreshStats held gPayoutLedgerMutex at function scope
+// across the call, so once the writes gained fsync() the event loop was doing
+// two of them every ten seconds with the payout thread's mutex held -- worse
+// than the buffered write the merge replaced. The status file now goes through
+// the same serialise-under-the-lock / publish-outside-it split as the ledgers,
+// and the rule holds for all four files. It is checked, not asserted: see the
+// "no payout mutex is held across a publish" section of the payout-ledger
+// self-test.
 
 // A serialised file image plus the version of the in-memory state it was
 // taken from.
@@ -2100,7 +2111,22 @@ static std::string PoolStatusFilePath()
     return GetAppDir() + "/pool_status.json";
 }
 
-static void WritePoolStatusJsonLocked()
+// Serialise the operator status file. Caller must hold gStatsMutex (this reads
+// gStats / gWorkerStats / gPayoutViews); gPoolRoundsMutex is taken and
+// released inside.
+//
+// Split from the write for the same reason as the three ledgers, and for a
+// worse instance of the same problem. This used to be called from RefreshStats
+// with gPayoutLedgerMutex held at function scope, and once the ledger writes
+// gained fsync() that call became two fsyncs -- one on the file, one on its
+// directory -- performed by the event loop, the thread that owns every miner's
+// socket, with the mutex the payout thread needs in its hand. Measured on the
+// build host (ext4 on LVM on SSD): 3.78 ms per publish, every 10 seconds, and
+// unbounded on a shared virtual disk under load.
+//
+// Nothing here needs the ledger mutex at all: the pending-payout rows were
+// already copied into gPayoutViews under it, further up RefreshStats.
+static std::string SerializePoolStatusLocked()
 {
     nlohmann::json j;
     j["schema"] = "bitflash-pool-status-1";
@@ -2184,11 +2210,20 @@ static void WritePoolStatusJsonLocked()
         }
     }
 
-    // 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");
+    return j.dump(2);
+}
+
+// Publish the operator status file. Call with every payout mutex released.
+//
+// Still no version guard and no write mutex, deliberately, and the split does
+// not change the argument: RefreshStats remains the only caller, it still runs
+// only on the event loop, and it now calls this after leaving the scope that
+// serialised -- so the serialise and the write stay strictly ordered on one
+// thread and pool_status.json still has a single writer by construction. It
+// shares WriteFileAtomic so the durability rules stay in one place.
+static void PublishPoolStatus(const std::string& blob)
+{
+    WriteFileAtomic(PoolStatusFilePath(), blob, "pool status");
 }
 
 // ---------------------------------------------------------------------------
@@ -2907,6 +2942,10 @@ static std::string PayoutTestAddWalletTx(const std::string& comment,
     return hash.GetHex();
 }
 
+// Defined below with the event loop; the status-file publication test drives
+// it directly.
+static void RefreshStats(int blocksFoundThisSession, uint64 roundShareTotal);
+
 int RunPayoutLedgerSelfTest()
 {
     int nFail = 0;
@@ -3455,6 +3494,136 @@ int RunPayoutLedgerSelfTest()
         strPoolRoundsFile = savedRoundsFile;
     }
 
+    // -- 11. No payout mutex is held across the status-file write ------------
+    //
+    // RefreshStats runs on the event loop -- the thread that owns every
+    // miner's socket -- and used to hold gStatsMutex and gPayoutLedgerMutex at
+    // function scope across WriteFileAtomic. Once that write gained fsync() on
+    // the file and on its directory, publishing pool_status.json parked the
+    // payout thread behind two disk flushes every ten seconds.
+    //
+    // Timing assertions would be flaky, so this does not measure anything. It
+    // makes the write block for as long as the test likes: strPoolStatusFile
+    // points at a path whose ".tmp" is a FIFO, and open(2) for writing on a
+    // FIFO does not return until a reader arrives. While RefreshStats is
+    // parked in there, every mutex it must NOT be holding has to be
+    // acquirable. Against the previous code this test does not merely fail, it
+    // cannot pass: the writer is inside fopen() with both mutexes held and
+    // nothing will release them until the test lets the write finish.
+#ifndef _WIN32
+    {
+        PayoutTestResetState();
+        const std::string statusPath = dir + "/pool_status_selftest.json";
+        const std::string statusTmp  = statusPath + ".tmp";
+        const std::string savedStatusFile = strPoolStatusFile;
+        const int64       savedLastWrite  = gLastPoolStatusWrite;
+        remove(statusPath.c_str());
+        remove(statusTmp.c_str());
+        strPoolStatusFile    = statusPath;
+        gLastPoolStatusWrite = 0;   // due a publish on the next refresh
+
+        // Give the status file something to report, so the ledger mutex is
+        // genuinely taken on the way through rather than skipped over.
+        {
+            std::lock_guard<std::mutex> lk(gPayoutLedgerMutex);
+            PendingPayout pp;
+            pp.blockHeight = 909; pp.blockHash = 0; pp.matureAtHeight = 909;
+            pp.amounts[addrA] = 3 * COIN;
+            gPendingPayouts.push_back(pp);
+        }
+
+        bool fifoOk = (mkfifo(statusTmp.c_str(), 0600) == 0);
+        check(fifoOk, "a FIFO stands in for the status file's temp file, so the"
+                      " write can be held open");
+        if (fifoOk) {
+            std::atomic<bool> done(false);
+            std::thread writer([&done]() {
+                RefreshStats(0, 0);
+                done = true;
+            });
+
+            // Settle: the work before the write is a handful of vector
+            // operations, so anything past this point is inside fopen(). The
+            // !done check below is what actually proves it -- with no reader
+            // there is no other way for this thread to still be running.
+            Sleep(500);
+            bool blocked = !done;
+            bool gotLedger = false, gotStats = false, gotExec = false,
+                 gotRounds = false;
+            for (int i = 0; i < 1000 && blocked; i++) {
+                if (!gotLedger && gPayoutLedgerMutex.try_lock()) {
+                    gotLedger = true; gPayoutLedgerMutex.unlock();
+                }
+                if (!gotStats && gStatsMutex.try_lock()) {
+                    gotStats = true; gStatsMutex.unlock();
+                }
+                if (!gotExec && gPayoutExecMutex.try_lock()) {
+                    gotExec = true; gPayoutExecMutex.unlock();
+                }
+                if (!gotRounds && gPoolRoundsMutex.try_lock()) {
+                    gotRounds = true; gPoolRoundsMutex.unlock();
+                }
+                if (gotLedger && gotStats && gotExec && gotRounds)
+                    break;
+                Sleep(2);
+            }
+
+            check(blocked,
+                  "the status-file write is held open for the duration of this"
+                  " test");
+            check(!blocked || gotLedger,
+                  "gPayoutLedgerMutex is free while pool_status.json is being"
+                  " written");
+            check(!blocked || (gotExec && gotRounds),
+                  "gPayoutExecMutex and gPoolRoundsMutex are free while"
+                  " pool_status.json is being written");
+            check(!blocked || gotStats,
+                  "gStatsMutex is free while pool_status.json is being written");
+
+            // Let the write finish. Drain to EOF rather than closing early:
+            // the writer is still filling the pipe and a closed read end would
+            // deliver it SIGPIPE.
+            std::thread reader([&statusTmp]() {
+                FILE* rf = fopen(statusTmp.c_str(), "r");
+                if (!rf) return;
+                char buf[4096];
+                while (fread(buf, 1, sizeof(buf), rf) > 0) { }
+                fclose(rf);
+            });
+            writer.join();
+            reader.join();
+        }
+        remove(statusTmp.c_str());
+
+        // And the ordinary path still produces the file it always did.
+        gLastPoolStatusWrite = 0;
+        RefreshStats(0, 0);
+        std::string onDisk;
+        bool statusOk = false;
+        if (PayoutTestReadFile(statusPath, onDisk)) {
+            try {
+                nlohmann::json j = nlohmann::json::parse(onDisk);
+                statusOk = j.value("schema", "") == "bitflash-pool-status-1" &&
+                           j["pendingPayouts"].is_array() &&
+                           j["pendingPayouts"].size() == 1 &&
+                           j["pendingPayouts"][0].value("matureAtHeight", 0) == 909 &&
+                           j["pendingPayouts"][0].value("totalAmountSatoshis",
+                                                        (int64)0) == 3 * COIN;
+            } catch (...) {
+                statusOk = false;
+            }
+        }
+        check(statusOk,
+              "splitting the serialise from the write leaves pool_status.json"
+              " saying the same thing");
+
+        remove(statusPath.c_str());
+        remove((statusPath + ".tmp").c_str());
+        strPoolStatusFile    = savedStatusFile;
+        gLastPoolStatusWrite = savedLastWrite;
+    }
+#endif
+
     // -- clean up ------------------------------------------------------------
     PayoutTestResetState();
     CRITICAL_BLOCK(cs_mapWallet)
@@ -3615,56 +3784,71 @@ int RunPoolStratumSelfTest()
 // ---------------------------------------------------------------------------
 static void RefreshStats(int blocksFoundThisSession, uint64 roundShareTotal)
 {
-    std::lock_guard<std::mutex> lk(gStatsMutex);
+    std::string statusBlob;
+    bool        fPublish = false;
+    {
+        std::lock_guard<std::mutex> lk(gStatsMutex);
 
-    gStats.authorizedMiners = 0;
-    gStats.blocksFound      = blocksFoundThisSession;
-    gStats.roundShares      = roundShareTotal;
-    gStats.totalHashRate    = 0.0;
-    for (Miner* m : gMiners)
-        if (m->authorised) gStats.authorizedMiners++;
+        gStats.authorizedMiners = 0;
+        gStats.blocksFound      = blocksFoundThisSession;
+        gStats.roundShares      = roundShareTotal;
+        gStats.totalHashRate    = 0.0;
+        for (Miner* m : gMiners)
+            if (m->authorised) gStats.authorizedMiners++;
 
-    gWorkerStats.clear();
-    for (Miner* m : gMiners) {
-        if (!m->authorised || m->lastSeen <= 0) continue;
-        PoolWorkerStatView v;
-        v.address     = m->address;
-        v.worker      = m->worker;
-        v.roundShares = m->roundShares;
-        v.totalShares = m->totalShares;
-        v.lastSeen    = m->lastSeen;
-        v.hashRate    = m->hashRate;
-        gWorkerStats.push_back(v);
-        gStats.totalHashRate += m->hashRate;
-    }
-    std::sort(gWorkerStats.begin(), gWorkerStats.end(),
-        [](const PoolWorkerStatView& a, const PoolWorkerStatView& b) {
-            return a.roundShares != b.roundShares ? a.roundShares > b.roundShares
-                                                  : a.lastSeen > b.lastSeen;
-        });
+        gWorkerStats.clear();
+        for (Miner* m : gMiners) {
+            if (!m->authorised || m->lastSeen <= 0) continue;
+            PoolWorkerStatView v;
+            v.address     = m->address;
+            v.worker      = m->worker;
+            v.roundShares = m->roundShares;
+            v.totalShares = m->totalShares;
+            v.lastSeen    = m->lastSeen;
+            v.hashRate    = m->hashRate;
+            gWorkerStats.push_back(v);
+            gStats.totalHashRate += m->hashRate;
+        }
+        std::sort(gWorkerStats.begin(), gWorkerStats.end(),
+            [](const PoolWorkerStatView& a, const PoolWorkerStatView& b) {
+                return a.roundShares != b.roundShares ? a.roundShares > b.roundShares
+                                                      : a.lastSeen > b.lastSeen;
+            });
 
-    gPayoutViews.clear();
-    std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);
-    for (const PendingPayout& pp : gPendingPayouts) {
-        PendingPayoutView pv;
-        pv.matureAtHeight = pp.matureAtHeight;
-        pv.recipients     = 0;
-        pv.totalAmount    = 0;
-        for (auto& kv : pp.amounts) {
-            if (pp.paidTxids.count(kv.first))
-                continue;
-            pv.recipients++;
-            pv.totalAmount += kv.second;
+        // Scoped, not held to the end of the function. At function scope this
+        // lock covered the status-file write below, so the event loop did two
+        // fsyncs holding the mutex the payout thread takes to record that a
+        // miner has been paid.
+        gPayoutViews.clear();
+        {
+            std::lock_guard<std::mutex> ledger(gPayoutLedgerMutex);
+            for (const PendingPayout& pp : gPendingPayouts) {
+                PendingPayoutView pv;
+                pv.matureAtHeight = pp.matureAtHeight;
+                pv.recipients     = 0;
+                pv.totalAmount    = 0;
+                for (auto& kv : pp.amounts) {
+                    if (pp.paidTxids.count(kv.first))
+                        continue;
+                    pv.recipients++;
+                    pv.totalAmount += kv.second;
+                }
+                if (pv.recipients > 0)
+                    gPayoutViews.push_back(pv);
+            }
         }
-        if (pv.recipients > 0)
-            gPayoutViews.push_back(pv);
-    }
 
-    int64 now = GetTime();
-    if (now - gLastPoolStatusWrite >= 10) {
-        WritePoolStatusJsonLocked();
-        gLastPoolStatusWrite = now;
+        int64 now = GetTime();
+        if (now - gLastPoolStatusWrite >= 10) {
+            statusBlob = SerializePoolStatusLocked();
+            gLastPoolStatusWrite = now;
+            fPublish = true;
+        }
     }
+    // Every payout mutex is released here, and gStatsMutex with them. This is
+    // where the two fsyncs happen, and no other thread is waiting behind them.
+    if (fPublish)
+        PublishPoolStatus(statusBlob);
 }
 
 // ---------------------------------------------------------------------------