selftest: make a torn publish structurally invalid, not merely unlucky

Mutation-tested, the concurrency test did not test concurrency. With the
per-file write mutex removed from PublishVersioned, `make tests` still
went green 2 runs out of 5, and the assertion named "four threads
publishing the same file never leave it unparseable" fired 0 times out
of 5. It was asserting something true for reasons unrelated to the code
it was pointed at.

Two causes.

The images were 64 arrays of 200 entries, identical in structure and
within bytes of each other in length, so an interleaving of any two of
them was still a valid JSON array. There was nothing structural for a
torn write to break. They now alternate between one entry and five
thousand: a write landing on top of another leaves either a truncated
array or a small image followed by the tail of a large one, and neither
parses.

Worse, almost nothing was being written. Each thread walked a
permutation of the images and PublishVersioned drops any blob not newer
than what is on disk, so only the record highs in each permutation ever
reached the disk -- about ten writes in total across four threads, which
is not enough contention to produce a race however the images are
shaped. Every publish now stamps its sequence number from a shared
counter at the moment it is made, so every publish is newer than the
last one issued and every publish writes: 400 of them, half of them
600 KB.

And the file is now watched while the writers run rather than inspected
once after they stop. "The published file is never observed
unparseable" is the real invariant -- rename(2) is atomic, so a reader
can only ever see a completely written image -- and sampling it only at
the end looks at the one moment a torn file is least likely to have
survived.

Measured on the build host, five runs with the file mutex removed: 5
detections out of 5, every one of them on the unparseable-file
assertion, which is the assertion that used to fire never. Five runs
with the mutex in place: 5 passes. The whole payout-ledger self-test
takes 1.4 s.

Property: removing the serialisation this test exists to check fails the
test, on every run.

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());