tools/dings-voices/README.md

@@ -18,8 +18,8 @@ Run: `node src/polyphony.test.mjs && node src/build.mjs raw && python src/finish
 One dev, one URL: the listening page is **https://minds.gamecult.org/aqua/dings/** (`/srv/minds-web/aqua/dings/`), always the latest version, overwritten in place. Do not publish per-hash pages any more. After building a publish dir (the `Run` line, then `gen_page.py`), run
 
 ```
-python3 tools/publish-stable/publish_stable.py dings <publishDir>    # idempotent; replaces the stable dir by renaming a finished staging copy
+python3 tools/publish-stable/publish_stable.py dings <publishDir>    # idempotent; swaps the stable dir with a finished staging copy in one atomic exchange (renameat2, Linux): the URL is never missing or half-written, even if the run is killed; stale staging is removed by the next run
 python3 tools/publish-stable/publish_stable.py banners               # old per-version pages get a pointer banner; deletes nothing
 ```
 
-The script stamps a banner and a `VERSION` file (publish dir name and AquaSynth revision) into the stable copy. The aqua-ear examples are published the same way to `/srv/minds-web/aqua/ear/` (`publish_stable.py ear <dir of runs>`). Tests: `python3 tools/publish-stable/test_publish_stable.py`. Nobody has opened these pages in a browser; every link and audio file on them was checked with curl over https.
+The script stamps a banner and a `VERSION` file (publish dir name and AquaSynth revision) into the stable copy. The aqua-ear examples are published the same way to `/srv/minds-web/aqua/ear/` (`publish_stable.py ear <dir of runs>`). Tests: `python3 tools/publish-stable/test_publish_stable.py` and `test_interruption.py` (kill at every mutation, concurrent reader). Nobody has opened these pages in a browser; every link and audio file on them was checked with curl over https.

tools/publish-stable/publish_stable.py

@@ -7,11 +7,11 @@
 
 SRC_DIR for `dings` is a built publish dir (index.html, out/, findings/, metrics.json; see tools/dings-voices/README.md);
 for `ear` a dir of aqua-ear outputs (one subdir per run, each with report.txt/png files, as `aqua_ear.py --out` writes).
-Stdlib only. Idempotent: running it again with the same input leaves the same tree; the stable dir is replaced by renaming
-a fully written staging dir, so a reader never sees a half-copied page. Deletes nothing outside the stable dir it owns.
+Stdlib only. Idempotent: running it again with the same input leaves the same tree; the stable dir is swapped with a fully
+written staging dir in one atomic exchange (renameat2), so it is never missing or half-copied, even if the process is killed. Deletes nothing outside the stable dir it owns.
 --root and --base are for tests. Nothing here listens to anything: it moves files and stamps a header.
 """
-import argparse, html, os, re, shutil, subprocess, sys, tempfile
+import argparse, ctypes, fcntl, glob, html, os, re, shutil, subprocess, sys, tempfile
 
 ROOT = "/srv/minds-web/aqua"
 BASE = "https://minds.gamecult.org/aqua"
@@ -41,24 +41,44 @@ def put_banner(path, inner):
     open(path, "w", encoding="utf-8").write(s)
 
 
+def _exchange(a, b):
+    """Atomically swap two directories (renameat2 RENAME_EXCHANGE). There is no instant at which either path is absent."""
+    libc = ctypes.CDLL(None, use_errno=True)
+    fn = getattr(libc, "renameat2", None)
+    if fn is None:
+        raise OSError("renameat2 is not available; refusing a non-atomic swap")
+    AT_FDCWD, RENAME_EXCHANGE = -100, 2
+    if fn(AT_FDCWD, os.fsencode(a), AT_FDCWD, os.fsencode(b), RENAME_EXCHANGE) != 0:
+        e = ctypes.get_errno()
+        raise OSError(e, f"renameat2(RENAME_EXCHANGE) {a} <-> {b}: {os.strerror(e)}")
+
+
 def swap_in(src, dest, prepare):
-    """Copy src to a staging dir beside dest, let prepare(stage) stamp it, then replace dest by rename."""
-    parent = os.path.dirname(dest)
+    """Make dest a complete copy of src (stamped by prepare) without a moment when dest is missing or half-written.
+
+    The copy is built in a staging dir beside dest; then dest and the staging dir are EXCHANGED in one syscall (a plain rename
+    when dest does not exist yet) and the staging dir, now holding the old tree, is deleted. An interrupted run leaves dest as
+    the old complete tree or the new complete tree, plus at most a staging dir, which the next run removes under a lock
+    that serialises publishers of the same target."""
+    parent, name = os.path.dirname(dest), os.path.basename(dest)
     os.makedirs(parent, exist_ok=True)
-    stage = tempfile.mkdtemp(prefix=".stage-", dir=parent)
-    try:
-        shutil.copytree(src, stage, dirs_exist_ok=True)
-        prepare(stage)
-        os.chmod(stage, 0o755)
-        old = dest + ".old"
-        if os.path.exists(dest):
-            shutil.rmtree(old, ignore_errors=True)
-            os.rename(dest, old)
-        os.rename(stage, dest)
-        shutil.rmtree(old, ignore_errors=True)
-    except BaseException:
-        shutil.rmtree(stage, ignore_errors=True)
-        raise
+    with open(os.path.join(parent, f".{name}.lock"), "w") as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX)
+        for leftover in glob.glob(os.path.join(parent, f".stage-{name}-*")):
+            shutil.rmtree(leftover, ignore_errors=True)
+        stage = tempfile.mkdtemp(prefix=f".stage-{name}-", dir=parent)
+        try:
+            shutil.copytree(src, stage, dirs_exist_ok=True)
+            prepare(stage)
+            os.chmod(stage, 0o755)
+            if os.path.isdir(dest):
+                _exchange(stage, dest)       # stage now holds the old tree
+            else:
+                os.rename(stage, dest)
+                stage = None
+        finally:
+            if stage:
+                shutil.rmtree(stage, ignore_errors=True)
 
 
 def publish_dings(src, root, base):

tools/publish-stable/test_interruption.py

@@ -0,0 +1,69 @@
+"""A publish that is killed at ANY filesystem mutation, or read concurrently, never leaves the live dir missing or half-written.
+Implementation-agnostic: a sys.audit hook counts every mutating event (rename, mkdir, rmtree, remove, copyfile, ctypes call) and
+os._exit()s the publisher at the Nth one; the live dir is then checked against the old and new trees."""
+import os, subprocess, sys, tempfile, threading, unittest
+HERE = os.path.dirname(os.path.abspath(__file__)); S = os.path.join(HERE, "publish_stable.py")
+MUTATING = {"os.rename", "os.mkdir", "os.rmdir", "os.remove", "shutil.rmtree", "shutil.copyfile", "shutil.copymode", "ctypes.call_function", "os.chmod"}
+RUNNER = r'''
+import sys, os, runpy
+kill_at = int(sys.argv[1]); counter = [0]
+MUT = %r
+def hook(ev, args):
+    if ev in MUT:
+        counter[0] += 1
+        if kill_at and counter[0] == kill_at: os._exit(9)
+sys.addaudithook(hook)
+sys.argv = [%r, "dings", sys.argv[2], "--root", sys.argv[3]]
+try: runpy.run_path(%r, run_name="__main__")
+finally:
+    if not kill_at: print("EVENTS", counter[0])
+''' % (MUTATING, S, S)
+def make_src(d, tag):
+    os.makedirs(d + "/out/senders"); open(d + "/index.html", "w").write(f"<html><body>VERSION-{tag}</body></html>"); open(d + "/out/senders/a.ogg", "w").write(f"OggS-{tag}" * 5000)
+def snapshot(d): return sorted((os.path.relpath(os.path.join(r, f), d), open(os.path.join(r, f)).read()) for r, _, fs in os.walk(d) for f in fs if f != "VERSION")
+class T(unittest.TestCase):
+    def setUp(self):
+        self.t = tempfile.TemporaryDirectory(); self.root = self.t.name + "/aqua"; os.makedirs(self.root)
+        self.a, self.b = self.t.name + "/dings-voices-aaa", self.t.name + "/dings-voices-bbb"; make_src(self.a, "A"); make_src(self.b, "B")
+        self.run_runner = lambda kill, src: subprocess.run([sys.executable, "-c", RUNNER, str(kill), src, self.root], capture_output=True, text=True)
+        self.live = self.root + "/dings"
+    def tearDown(self): self.t.cleanup()
+    def stamped(self, src):   # what the live dir holds after a clean publish of src
+        with tempfile.TemporaryDirectory() as r: self.assertEqual(self.run_runner(0, src).returncode, 0) if False else subprocess.run([sys.executable, S, "dings", src, "--root", r], check=True, capture_output=True); return snapshot(r + "/dings")
+    def test_kill_at_every_step_over_an_existing_page_keeps_a_complete_tree(self):
+        old, new = self.stamped(self.a), self.stamped(self.b)
+        r = self.run_runner(0, self.a); self.assertEqual(r.returncode, 0, r.stderr)
+        n = int([l for l in r.stdout.splitlines() if l.startswith("EVENTS")][0].split()[1]); self.assertGreater(n, 5)
+        for kill in range(1, n + 1):
+            subprocess.run([sys.executable, S, "dings", self.a, "--root", self.root], check=True, capture_output=True)
+            self.assertEqual(self.run_runner(kill, self.b).returncode, 9)
+            self.assertTrue(os.path.isdir(self.live), f"live dir missing after a kill at mutation {kill} of {n}")
+            self.assertIn(snapshot(self.live), (old, new), f"live dir half-written after a kill at mutation {kill} of {n}")
+            subprocess.run([sys.executable, S, "dings", self.b, "--root", self.root], check=True, capture_output=True)       # the next run heals
+            self.assertEqual(snapshot(self.live), new); self.assertEqual([f for f in os.listdir(self.root) if f.startswith(".stage-") or f.endswith(".old")], [], f"stale staging after healing run (kill {kill})")
+    def test_kill_during_the_very_first_publish_never_leaves_a_half_written_page(self):
+        new = self.stamped(self.b)
+        r = self.run_runner(0, self.b); n = int([l for l in r.stdout.splitlines() if l.startswith("EVENTS")][0].split()[1])
+        for kill in range(1, n + 1):
+            subprocess.run(["rm", "-rf", self.live]); self.assertEqual(self.run_runner(kill, self.b).returncode, 9)
+            if os.path.isdir(self.live): self.assertEqual(snapshot(self.live), new, f"half-written first publish after kill {kill}")
+    def test_a_raising_step_keeps_the_old_page_and_cleans_its_staging(self):
+        subprocess.run([sys.executable, S, "dings", self.a, "--root", self.root], check=True, capture_output=True); old = snapshot(self.live)
+        os.remove(self.b + "/index.html")
+        self.assertNotEqual(subprocess.run([sys.executable, S, "dings", self.b, "--root", self.root], capture_output=True).returncode, 0)
+        self.assertEqual(snapshot(self.live), old); self.assertEqual([f for f in os.listdir(self.root) if f.startswith(".stage-")], [])
+    def test_a_concurrent_reader_never_finds_the_page_missing_or_cut_short(self):
+        subprocess.run([sys.executable, S, "dings", self.a, "--root", self.root], check=True, capture_output=True)
+        bad, stop = [], threading.Event()
+        def reader():
+            while not stop.is_set():
+                try:
+                    idx = open(self.live + "/index.html").read(); ogg = open(self.live + "/out/senders/a.ogg").read()
+                    if "VERSION-" not in idx or not (ogg.startswith("OggS-A") or ogg.startswith("OggS-B")) or len(ogg) not in (len("OggS-A") * 5000,): bad.append("partial")
+                except OSError as e: bad.append(repr(e))
+        th = [threading.Thread(target=reader) for _ in range(4)]; [x.start() for x in th]
+        try:
+            for i in range(30): subprocess.run([sys.executable, S, "dings", self.b if i % 2 == 0 else self.a, "--root", self.root], check=True, capture_output=True)
+        finally: stop.set(); [x.join() for x in th]
+        self.assertEqual(bad, [])
+if __name__ == "__main__": unittest.main()

tools/publish-stable/test_publish_stable.py

@@ -15,7 +15,7 @@ class T(unittest.TestCase):
         self.assertEqual(first, tree(self.root + "/dings"))
         page = open(self.root + "/dings/index.html").read(); self.assertEqual(page.count('<div class="aqua-stable-banner"'), 1)
         self.assertTrue(os.path.exists(self.root + "/dings/out/senders/a.ogg")); self.assertIn("dings-voices-abc1234", open(self.root + "/dings/VERSION").read())
-        self.assertEqual(sorted(os.listdir(self.root)), ["dings"])          # no staging or .old leftovers
+        self.assertEqual(sorted(f for f in os.listdir(self.root) if not f.endswith(".lock")), ["dings"])   # no staging leftovers
     def test_new_version_replaces_old_files(self):
         run("dings", self.src, "--root", self.root); os.remove(self.src + "/out/senders/a.ogg"); open(self.src + "/out/senders/b.ogg", "wb").write(b"OggS")
         run("dings", self.src, "--root", self.root); self.assertEqual(os.listdir(self.root + "/dings/out/senders"), ["b.ogg"])