1 files changed, 96 insertions, 32 deletions

diff --git a/src/rpc.cpp b/src/rpc.cpp
index 7ea0597..79947df 100644
--- a/src/rpc.cpp
+++ b/src/rpc.cpp
@@ -3542,54 +3542,118 @@ int RunPayoutLedgerSelfTest()
 
     // -- 6. Concurrent publishers cannot interleave a file -------------------
     //
-    // Two threads renaming the same "<path>.tmp" over the same target was the
-    // original defect. The images are made big so a torn write shows up as
-    // unparseable JSON rather than by luck.
+    // Two threads opening and renaming the same "<path>.tmp" over the same
+    // target was the original defect. This test was mutation-tested and did
+    // not catch it: with the file mutex removed the suite still went green 2
+    // runs in 5, and the "never leave it unparseable" assertion fired 0 times
+    // in 5. Two reasons, both fixed here.
+    //
+    // First, every image had the same structure and near enough the same
+    // length, so an interleaving of two of them was still valid JSON. The
+    // images now alternate between ONE entry and FIVE THOUSAND, so a write
+    // that lands on top of another leaves either a truncated array or a small
+    // image followed by the tail of a large one -- neither of which parses.
+    //
+    // Second, and worse, almost nothing was being written. Each thread walked
+    // a permutation of the images, and PublishVersioned drops any blob that is
+    // not newer than what is on disk, so only the record highs in each
+    // permutation ever reached the disk -- on the order of ten writes in
+    // total, four threads, no contention to speak of. Every publish here is
+    // stamped from a shared counter at the moment it is made, so every one of
+    // them is newer than the last and every one of them writes.
+    //
+    // And the file is watched while the writers run, not only inspected after
+    // they finish. "The published file is never observed unparseable" is the
+    // actual invariant -- rename(2) is atomic, so a reader can only ever see a
+    // completely written image -- and checking it once at the end samples the
+    // one moment when a torn file is least likely to still be there.
     {
         const std::string racePath = dir + "/publish_race_selftest.json";
         std::mutex raceMutex;
         uint64 racePublished = 0;
 
-        std::vector<VersionedBlob> images;
-        for (int i = 0; i < 64; i++) {
-            nlohmann::json arr = nlohmann::json::array();
-            for (int k = 0; k < 200; k++) {
-                nlohmann::json o;
-                o["image"] = i;
-                o["entry"] = k;
-                o["address"] = addrA;
-                o["amount"] = (int64)(k + 1) * COIN;
-                arr.push_back(o);
+        std::string tiny, huge;
+        {
+            nlohmann::json one = nlohmann::json::array();
+            nlohmann::json o;
+            o["image"]   = 0;
+            o["entry"]   = 0;
+            o["address"] = addrA;
+            o["amount"]  = (int64)COIN;
+            one.push_back(o);
+            tiny = one.dump(2);
+
+            nlohmann::json many = nlohmann::json::array();
+            for (int k = 0; k < 5000; k++) {
+                nlohmann::json e;
+                e["image"]   = 1;
+                e["entry"]   = k;
+                e["address"] = addrA;
+                e["amount"]  = (int64)(k + 1) * COIN;
+                many.push_back(e);
             }
-            VersionedBlob v;
-            v.blob = arr.dump(2);
-            v.seq  = (uint64)i + 1;
-            images.push_back(v);
+            huge = many.dump(2);
         }
 
+        const int kThreads   = 4;
+        const int kPerThread = 100;
+        const uint64 kTotal  = (uint64)kThreads * kPerThread;
+
+        std::atomic<uint64> nextSeq(0);
+        std::atomic<bool>   racing(true);
+        std::atomic<int>    tornReads(0);
+        std::atomic<int>    goodReads(0);
+
+        std::thread watcher([&]() {
+            bool seenOnce = false;
+            while (racing) {
+                std::string s;
+                if (!PayoutTestReadFile(racePath, s)) {
+                    // Only meaningful once the file exists at all.
+                    if (seenOnce) tornReads++;
+                    continue;
+                }
+                seenOnce = true;
+                try {
+                    if (nlohmann::json::parse(s).is_array()) goodReads++;
+                    else tornReads++;
+                } catch (...) {
+                    tornReads++;
+                }
+            }
+        });
+
         std::vector<std::thread> threads;
-        for (int t = 0; t < 4; t++) {
-            threads.push_back(std::thread([&, t]() {
-                for (size_t i = 0; i < images.size(); i++) {
-                    size_t idx = (i * 7 + (size_t)t * 13) % images.size();
-                    PublishVersioned(racePath, images[idx], raceMutex,
-                                     racePublished, "publish race selftest");
+        for (int t = 0; t < kThreads; t++) {
+            threads.push_back(std::thread([&]() {
+                for (int i = 0; i < kPerThread; i++) {
+                    VersionedBlob v;
+                    v.seq  = ++nextSeq;
+                    v.blob = (v.seq & 1) ? tiny : huge;
+                    PublishVersioned(racePath, v, raceMutex, racePublished,
+                                     "publish race selftest");
                 }
             }));
         }
         for (std::thread& th : threads)
             th.join();
-
+        racing = false;
+        watcher.join();
+
+        check(goodReads > 0,
+              "the race test actually observed the file while it was being"
+              " written");
+        check(tornReads == 0,
+              "four threads publishing the same file never leave it"
+              " unparseable, at any instant while they are doing it");
+
+        // The highest sequence number issued is the one the file must settle
+        // on: PublishVersioned refuses anything older, so whichever order the
+        // writes complete in, that image is the last one that can land.
+        const std::string& expected = (kTotal & 1) ? tiny : huge;
         std::string onDisk;
         bool read = PayoutTestReadFile(racePath, onDisk);
-        bool parses = false;
-        if (read) {
-            try { parses = nlohmann::json::parse(onDisk).is_array(); }
-            catch (...) { parses = false; }
-        }
-        check(read && parses,
-              "four threads publishing the same file never leave it unparseable");
-        check(read && onDisk == images.back().blob,
+        check(read && onDisk == expected,
               "the published file settles on the highest version, whatever the"
               " completion order");
         remove(racePath.c_str());