vm: run VM replays side by side; keep installed boxes across replay-only changes

.github/workflows/client.yml

@@ -96,6 +96,9 @@ jobs:
       - name: primary metadata however it's compressed (gz, EPEL 9's xz, EPEL 10's zst), by behaviour (runs the PR's own tools, so not in the gate)
         working-directory: .
         run: WEND_ALL_DECODERS=1 python3 -B tools/test_primary.py
+      - name: VM replays side by side (per-guest ports, the install lock, the box cache key), by behaviour with a stand-in QEMU (runs the PR's own vm.py, so not in the gate)
+        working-directory: .
+        run: python3 -B tools/test_vm_parallel.py
       - name: the atlas passes the client's own check
         working-directory: .
         run: client/target/x86_64-unknown-linux-musl/release/wend check

SPEC.md

@@ -1609,7 +1609,13 @@ The rules:
   runner that a pull request can trigger. Its runs go to `.wend/vm/` (schema `wend.vmrun/0`, `runner.kind = "vm"`),
   not to `runs/`.
 - QEMU user-mode networking only: nothing on the host changes. The guest reaches the mirror shim at `10.0.2.2`, and its
-  ssh and http are forwarded to fixed ports on `127.0.0.1`, so a host runs one VM replay at a time.
+  ssh is forwarded to a free port on `127.0.0.1` of its own, so a host runs several VM replays side by side. They
+  share the mirror shim and the installed boxes: a box is installed once per cache key (the box file, the adapter,
+  and the part of `tools/vm.py` the install runs), and a run that needs a box another run is installing waits for it.
+  Every run holds a shared lock on `/var/tmp/wend-shim/<port>.users` for each shim port while it lives, whatever its
+  `$WEND_WORK` (a directory only root or the running user can write, or it's refused), and `tools/vm.py shim-stop PID`
+  stops the shim only when no run holds its port's lock.
+  Every run records the commit it ran as `head` (`commit`, and `dirty` when tracked files differ from it).
 - Box: an optional `[vm]` table with `ssh_port` (default 22), `skip` (roles that start their own daemons, which a
   running box already has) and `live` (role name to shell: the live forms of those checks; every skipped role needs
   one, so skipping never drops a check), `pockets` (`updates`, and `security` and/or `backports` beside it, once each: the box

tools/sandbox.py

@@ -6,6 +6,7 @@
 a hostname of the box's own (never the host's), daemons never start, and nothing is deleted until every mount under
 it is gone.
 """
+import fcntl
 import hashlib
 import os
 import subprocess
@@ -257,6 +258,15 @@ def build_box(box_id, box_doc, adapter, box_toml_bytes, adapter_bytes, log=print
     root = f"{WORK}/boxes/{box_id.replace('/', '_')}-{key}"
     if os.path.exists(f"{root}/.wend-ready"):
         return Box(root)
+    os.makedirs(f"{WORK}/boxes", exist_ok=True)
+    with open(f"{root}.lock", "a") as lock:  # runs side by side (tools/vm.py): one builds, the others wait for it
+        fcntl.flock(lock, fcntl.LOCK_EX)  # closing the file drops it
+        if not os.path.exists(f"{root}/.wend-ready"):
+            build_unlocked(root, box_id, box_doc, adapter, log)
+    return Box(root)
+
+
+def build_unlocked(root, box_id, box_doc, adapter, log):
     if os.path.exists(root):
         safe_rm(root)
     os.makedirs(root)
@@ -269,4 +279,3 @@ def build_box(box_id, box_doc, adapter, box_toml_bytes, adapter_bytes, log=print
         raise RuntimeError(f"box {box_id} fails its own checks at the start: {failed}\n" +
                            "\n".join(tail(h["checks"][c]["log"], 800) for c in failed))
     open(f"{root}/.wend-ready", "w").close()
-    return Box(root)

tools/test_vm_parallel.py

@@ -0,0 +1,190 @@
+"""VM replays side by side (tools/vm.py), by behaviour, the parts that don't need KVM: two guests started at once get
+ssh forwards on different host ports and both run; a guest whose port another process takes first asks again for
+another one; the mirror shim can't be stopped while a run uses it, whatever its WEND_WORK (shim-stop); a second run
+that needs a box another run is installing waits for it; and the box cache key follows what the install runs, not the
+whole file (build_source). QEMU is a stand-in that binds its hostfwd port as QEMU does and says what QEMU says when
+it can't. It runs the PR's own vm.py, so the gate's schema job doesn't run it: the client job does.
+python3 tools/test_vm_parallel.py"""
+import os
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+
+sys.dont_write_bytecode = True
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, os.path.join(ROOT, "tools"))
+tmp = tempfile.mkdtemp(prefix="wend-test-vm-")
+os.environ["WEND_WORK"] = os.path.join(tmp, "work")  # before vm reads it
+import vm  # noqa: E402
+
+bad = 0
+
+
+def say(ok, what, detail=""):
+    global bad
+    print(("passed  " if ok else "WRONG   ") + what + ("" if ok or not detail else f"\n        {detail}"))
+    bad += not ok
+
+
+bin_dir = os.path.join(tmp, "bin")
+os.makedirs(bin_dir)
+open(os.path.join(bin_dir, "qemu-system-x86_64"), "w").write(f"""#!{sys.executable}
+import re, socket, sys, time
+net = sys.argv[sys.argv.index("-netdev") + 1]
+port = int(re.search(r"hostfwd=tcp:127\\.0\\.0\\.1:(\\d+)-", net).group(1))
+s = socket.socket()
+try:
+    s.bind(("127.0.0.1", port))
+    s.listen()
+except OSError:
+    sys.stderr.write(f"qemu-system-x86_64: -netdev {{net}}: Could not set up host forwarding rule 'tcp:127.0.0.1:{{port}}'\\n")
+    sys.exit(1)
+time.sleep(60)
+""")
+os.chmod(os.path.join(bin_dir, "qemu-system-x86_64"), 0o755)
+os.environ["PATH"] = bin_dir + os.pathsep + os.environ["PATH"]
+
+try:
+    # two guests at once: each its own port, both up, and ssh asks for that port
+    g1, g2 = (vm.Guest(os.path.join(tmp, f"g{i}.qcow2"), "key", 22) for i in (1, 2))
+    g1.start()
+    g2.start()
+    say(g1.running() and g2.running(), "two guests run side by side")
+    say(g1.fwd != g2.fwd, "each guest forwards ssh from a host port of its own", f"{g1.fwd} and {g2.fwd}")
+    argv = g1.ssh_argv("true")
+    say(argv[argv.index("-p") + 1] == str(g1.fwd), "ssh goes to the guest's own forwarded port", str(argv))
+    say(not re.search(r"hostfwd=tcp:127\.0\.0\.1:10022", " ".join(g1.argv())), "no fixed port is left in qemu's line")
+
+    # a port taken between the pick and QEMU's bind: the guest asks again
+    real, picks = vm.free_port, [g1.fwd]
+    vm.free_port = lambda: picks.pop() if picks else real()
+    g3 = vm.Guest(os.path.join(tmp, "g3.qcow2"), "key", 22)
+    try:
+        g3.start()
+        say(g3.running() and g3.fwd not in (g1.fwd, g2.fwd), "a guest whose port was taken first starts on another one",
+            f"port {g3.fwd}")
+    except RuntimeError as e:
+        say(False, "a guest whose port was taken first starts on another one", str(e))
+    finally:
+        vm.free_port = real
+
+    # anything else QEMU refuses is not retried
+    g4 = vm.Guest(os.path.join(tmp, "g4.qcow2"), "key", 22)
+    open(os.path.join(bin_dir, "qemu-system-x86_64"), "w").write(
+        f"#!{sys.executable}\nimport sys\nsys.stderr.write('could not open disk image\\n')\nsys.exit(1)\n")
+    t0 = time.time()
+    try:
+        g4.start()
+        say(False, "qemu failing for another reason fails at once, not five times")
+    except RuntimeError as e:
+        say("could not open disk image" in str(e) and time.time() - t0 < 6,
+            "qemu failing for another reason fails at once, not five times", f"{time.time() - t0:.0f}s: {e}")
+    for g in vm.GUESTS:
+        g.kill()
+        shutil.rmtree(os.path.dirname(g.monitor_path), ignore_errors=True)
+
+    # one run installs a box, a second waits for it, then finds it installed
+    lock, order = os.path.join(tmp, "work", "box.lock"), []
+
+    def holder(ready):
+        with vm.Locked(lock, "installing the box"):
+            ready.set()
+            time.sleep(1)
+            order.append("first done")
+
+    ready = threading.Event()
+    t = threading.Thread(target=holder, args=(ready,))
+    t.start()
+    ready.wait()
+    with vm.Locked(lock, "installing the box"):
+        order.append("second in")
+    t.join()
+    say(order == ["first done", "second in"], "a run waits while another installs the same box", str(order))
+
+    # the shim: shim-stop refuses while a run holds it, stops it once none does, and stops nothing else
+    fake = os.path.join(tmp, "checkout", "tools", "mirror.py")  # a checkout's shim: vm.py and sandbox.py beside it
+    os.makedirs(os.path.dirname(fake))
+    open(fake, "w").write("import time\ntime.sleep(60)\n")
+    for f in ("vm.py", "sandbox.py"):
+        open(os.path.join(os.path.dirname(fake), f), "w").close()
+    port = str(vm.free_port())  # a port no real shim uses, so its lock is this test's alone
+    shim = subprocess.Popen([sys.executable, fake, port, os.path.join(tmp, "cache")])
+    # the run holds the port's lock under another WEND_WORK than shim-stop's: the lock follows the port
+    run = subprocess.Popen([sys.executable, "-B", "-c", "import sys, time; sys.path.insert(0, sys.argv[1]); import vm; "
+                            "vm.use_shim([int(sys.argv[2])]); print('holding', flush=True); time.sleep(60)",
+                            os.path.join(ROOT, "tools"), port], stdout=subprocess.PIPE, text=True,
+                           env={**os.environ, "WEND_WORK": os.path.join(tmp, "other-work")})
+    run.stdout.readline()
+    stop = lambda pid: subprocess.run([sys.executable, "-B", os.path.join(ROOT, "tools", "vm.py"), "shim-stop", str(pid)],
+                                      capture_output=True, text=True)  # noqa: E731
+    p = stop(shim.pid)
+    say(p.returncode == 3 and shim.poll() is None, "shim-stop leaves the shim running while a VM run uses it",
+        f"rc {p.returncode}: {p.stdout}{p.stderr}")
+    run.kill()
+    run.wait()
+    p = stop(shim.pid)
+    time.sleep(0.5)
+    say(p.returncode == 0 and shim.poll() is not None, "shim-stop stops the shim once no run uses it",
+        f"rc {p.returncode}: {p.stdout}{p.stderr}")
+    elsewhere = os.path.join(tmp, "srv", "mirror.py")  # a mirror.py that isn't a wend checkout's
+    os.makedirs(os.path.dirname(elsewhere))
+    shutil.copyfile(fake, elsewhere)
+    for what, argv in [("a process that isn't the shim", ["-c", "import time; time.sleep(60)"]),
+                       ("another program's mirror.py", [elsewhere, port])]:
+        other = subprocess.Popen([sys.executable, *argv])
+        p = stop(other.pid)
+        say(p.returncode == 2 and other.poll() is None, f"shim-stop refuses {what}",
+            f"rc {p.returncode}: {p.stdout}{p.stderr}")
+        other.kill()
+    shim.kill()
+    os.remove(os.path.join(vm.SHIM_LOCKS, f"{port}.users"))
+
+    # a lock directory someone else could swap files in is refused, not used
+    real_locks = vm.SHIM_LOCKS
+    vm.SHIM_LOCKS = os.path.join(tmp, "open-locks")
+    os.mkdir(vm.SHIM_LOCKS)
+    os.chmod(vm.SHIM_LOCKS, 0o1777)
+    try:
+        vm.shim_lock(1)
+        say(False, "a world-writable lock directory is refused")
+    except SystemExit as e:
+        say("writable by nobody else" in str(e), "a world-writable lock directory is refused", str(e))
+    vm.SHIM_LOCKS = real_locks
+
+    # the cache key: what the install runs, and nothing it doesn't
+    src = open(os.path.join(ROOT, "tools", "vm.py")).read()
+    key = vm.build_source()
+
+    def key_after(old, new):
+        assert src.count(old) == 1, old
+        p = os.path.join(tmp, "vm-changed.py")
+        open(p, "w").write(src.replace(old, new))
+        return vm.build_source(p)
+
+    for what, old in [("the hop walk", "def walk_vm("), ("the shelter's confinement proof", "def shelter_confine("),
+                      ("relocate", "def relocate("), ("prune", "def prune("), ("main", "def main(")]:
+        say(key_after(f"\n{old}", f"\nUNUSED_BY_BUILD = 1\n\n\n{old}") == key,
+            f"a change beside {what} keeps the installed boxes")
+    say(key_after("    hops, start_net, efi_start = [], None, None\n",
+                  "    hops, start_net, efi_start = [], None, None  # changed\n") == key,
+        "a change inside the hop walk keeps the installed boxes")
+    for what, old, new in [("the kickstart", 'def kickstart(box_doc, adapter, key_pub):\n',
+                            'def kickstart(box_doc, adapter, key_pub):\n    pass\n'),
+                           ("the install", "def install(box_id, box_doc, d):\n",
+                            "def install(box_id, box_doc, d):\n    pass\n"),
+                           ("Guest's reboot", "    def reboot(self):\n", "    def reboot(self):\n        pass\n"),
+                           ("a constant the install uses", 'MAC = "52:54:00:12:34:56"', 'MAC = "52:54:00:12:34:57"'),
+                           ("the box's checks", "def checks(g, box_doc, adapter):\n",
+                            "def checks(g, box_doc, adapter):\n    pass\n")]:
+        say(key_after(old, new) != key, f"a change to {what} installs the boxes again")
+finally:
+    for g in vm.GUESTS:
+        g.kill()
+    shutil.rmtree(tmp, ignore_errors=True)
+
+print("ok" if not bad else f"{bad} WRONG")
+sys.exit(1 if bad else 0)