test(host): run the connectivity watchdog under real systemd in the KVM harness

tests/nm-isolated/vm_harness/README.md

@@ -1,3 +1,5 @@
 Host-side harness that produced the first real-daemon results for the Update2 fixture. `build_guest_layer_and_run.sh` runs on a KVM Linux host: it installs this directory's fixture files and `held_update2_scenario.sh` into a copy of a read-only golden Ubuntu 26.04 image (offline, via a cloud-init seed), then boots the result through `qemu_vm.disposable_vm` and prints the single JSON line the guest writes to its virtio port. The golden image must be built from a signature-verified Ubuntu cloud image with `dbus`, `network-manager`, `python3`, `python3-dbus`, `python3-gi` and `systemd` installed.
 
 The scenario starts the private bus, the PolicyKit stub and NetworkManager as root, adds in-memory profiles, runs the non-root `update2_client.py` with `--disconnect-after-held`, optionally changes the profile as root while the request is held, then releases authorisation and reads the final route metric. It is evidence for one NetworkManager version on in-memory profiles with the stub as authority; persistent profiles, a NetworkManager restart and the checkpoint expiry race are not covered.
+
+`watchdog_scenario.sh` is the second scenario. It runs the real transaction and the host-owned watchdog worker under the guest's own systemd, with the real `connectivity_transaction.py`, `connectivity_watchdog.py` and the unit from the role's template placed under `wd-modules/` next to the builder. A change whose owner crashes before committing is restored by the worker after its deadline; the same holds when the worker is killed as well and systemd restarts it; a change committed in time is left alone. It does not cover a reboot (the boot-id path), and the guest's NetworkManager reports connectivity checking off from the start, so the recorded `rolled_back` phase, which is written only after the reload returns, is the evidence of the reload. Pass the scenario to the builder by copying it to `smoke.sh` beside the golden image.

tests/nm-isolated/vm_harness/build_guest_layer_and_run.sh

@@ -12,12 +12,15 @@ for name in gp.FILES:
     digests[name] = hashlib.sha256(data).hexdigest()
     files.append({"path": f"{gp.DESTINATION}/{name}", "encoding": "b64", "content": base64.b64encode(data).decode(), "permissions": "0755" if name.endswith(".sh") else "0644"})
 smoke = open(os.path.expanduser('~/Developer/ExaDev/Internal/github-runner-worktrees/nm-isolated-vm/smoke.sh')).read()
+for mod in ("connectivity_transaction.py", "connectivity_watchdog.py"):
+    files.append({"path": "/usr/local/lib/github-runner-connectivity/" + mod, "encoding": "b64", "content": base64.b64encode(open(os.path.join(root, "wd-modules", mod), "rb").read()).decode(), "permissions": "0644"})
+files.append({"path": "/etc/systemd/system/github-runner-connectivity-watchdog.service", "content": open(os.path.join(root, "wd-modules", "connectivity-watchdog.service")).read(), "permissions": "0644"})
 files.append({"path": gp.DESTINATION + "/DIGESTS", "content": "".join(f"{n} {h}\n" for n, h in digests.items()), "permissions": "0644"})
 files.append({"path": gp.DESTINATION + "/smoke.sh", "content": smoke, "permissions": "0755"})
 files.append({"path": "/etc/systemd/system/nm-smoke.service", "content": "[Unit]\nDescription=Fixture smoke\nAfter=multi-user.target\n[Service]\nType=oneshot\nExecStart=/opt/nm-isolated-test/smoke.sh\n[Install]\nWantedBy=multi-user.target\n", "permissions": "0644"})
 ud = {"write_files": files, "runcmd": [["systemctl", "enable", "nm-smoke.service"]], "power_state": {"mode": "poweroff", "timeout": 30, "condition": True}}
 open("user-data", "w").write("#cloud-config\n" + json.dumps(ud))
-open("meta-data", "w").write("instance-id: nm-golden-23\nlocal-hostname: nm-golden\n")
+open("meta-data", "w").write("instance-id: nm-golden-30\nlocal-hostname: nm-golden\n")
 print("files", len(files))
 PY
 cloud-localds seed2.img user-data meta-data
@@ -32,7 +35,7 @@ with qemu_vm.disposable_vm(root + '/golden2-nm-fixture.qcow2', w, qemu='/usr/bin
     for _ in range(100):
         if os.path.exists(w + '/guest.sock'): break
         time.sleep(0.1)
-    s = socket.socket(socket.AF_UNIX); s.settimeout(90); s.connect(w + '/guest.sock')
+    s = socket.socket(socket.AF_UNIX); s.settimeout(300); s.connect(w + '/guest.sock')
     buf = b''
     while b'\n' not in buf:
         c = s.recv(4096)

tests/nm-isolated/vm_harness/watchdog_scenario.sh

@@ -0,0 +1,80 @@
+#!/bin/bash
+# Real systemd, real NetworkManager and the real transaction and watchdog modules: does the host-owned worker roll back a change whose owner crashed?
+set -u
+PORT=/dev/virtio-ports/test.nm.fixture
+STATE=/var/lib/github-runner-connectivity
+UNIT=github-runner-connectivity-watchdog
+mkdir -p /run/nm-isolated-test && touch /run/nm-isolated-test/disposable-vm
+systemctl daemon-reload
+systemctl start NetworkManager
+for _ in $(seq 1 60); do nmcli general status >/dev/null 2>&1 && break; sleep 0.5; done
+systemctl start $UNIT
+for _ in $(seq 1 40); do [ -S $STATE/watchdog.sock ] && break; sleep 0.5; done
+python3 - >/run/nm-isolated-test/wd.json 2>/run/nm-isolated-test/wd.err <<'PY'
+import json, os, subprocess, sys, time
+sys.path.insert(0, "/usr/local/lib/github-runner-connectivity")
+from pathlib import Path
+from connectivity_transaction import ConnectivityTransaction
+from connectivity_watchdog import probe_worker
+STATE = Path("/var/lib/github-runner-connectivity")
+DROPIN = Path("/etc/NetworkManager/conf.d/zz-test-connectivity.conf")
+CONTENT = b"[connectivity]\nenabled=false\n"
+UNIT = "github-runner-connectivity-watchdog"
+out = {}
+clock = lambda: time.clock_gettime(time.CLOCK_BOOTTIME)
+boot_id = Path("/proc/sys/kernel/random/boot_id").read_text().strip()
+def reload():
+    subprocess.run(["nmcli", "general", "reload", "conf"], check=True, timeout=30)
+class Scheduler:
+    def arm(self, tid, gen, deadline):
+        pass
+    def is_armed(self, tid, gen, deadline):
+        return probe_worker(STATE, tid, gen, deadline)
+def phase():
+    try:
+        return json.loads((STATE / "connectivity-transaction.json").read_text())["phase"]
+    except Exception as e:
+        return "unreadable:" + type(e).__name__
+def main_pid():
+    return subprocess.run(["systemctl", "show", "-p", "MainPID", "--value", UNIT], capture_output=True, text=True).stdout.strip()
+def begin(tid, seconds):
+    ConnectivityTransaction(STATE, reload, exclusive_writer=True).begin(tid, DROPIN, CONTENT, deadline=clock() + seconds, scheduler=Scheduler(), now=clock, boot_id=boot_id)
+def wait(cond, limit):
+    end = time.time() + limit
+    while time.time() < end:
+        if cond():
+            return True
+        time.sleep(0.5)
+    return False
+def runtime():
+    r = subprocess.run(["busctl", "get-property", "org.freedesktop.NetworkManager", "/org/freedesktop/NetworkManager", "org.freedesktop.NetworkManager", "ConnectivityCheckEnabled"], capture_output=True, text=True)
+    return r.stdout.strip()
+out["worker_active"] = subprocess.run(["systemctl", "is-active", UNIT], capture_output=True, text=True).stdout.strip()
+out["worker_pid_0"] = main_pid()
+out["runtime_before"] = runtime()
+# A: the owner of the change crashes before committing; the worker must restore.
+begin("tA", 12)
+out["A_after_begin"] = {"dropin": DROPIN.exists(), "phase": phase(), "runtime": runtime()}
+t0 = time.time()
+out["A_restored"] = wait(lambda: not DROPIN.exists() and phase() == "rolled_back", 60)
+out["A_seconds"] = round(time.time() - t0)
+out["A_runtime_after"] = runtime()
+# B: the owner crashes and the worker is killed too; systemd restarts it and it must still restore.
+begin("tB", 15)
+pid1 = main_pid()
+subprocess.run(["systemctl", "kill", "-s", "KILL", UNIT])
+out["B_restarted"] = wait(lambda: main_pid() not in ("", "0", pid1), 20)
+out["B_restored"] = wait(lambda: not DROPIN.exists() and phase() == "rolled_back", 60)
+# C: the owner commits in time; the worker must leave the change alone.
+begin("tC", 12)
+ConnectivityTransaction(STATE, reload, exclusive_writer=True).commit("tC", now=clock, boot_id=boot_id)
+time.sleep(20)
+out["C_after_deadline"] = {"dropin": DROPIN.exists(), "phase": phase()}
+if DROPIN.exists():
+    DROPIN.unlink()
+print(json.dumps(out))
+PY
+sc=$(tr '\012\042\134' '  _' < /run/nm-isolated-test/wd.json | cut -c1-1800)
+er=$(tail -c 900 /run/nm-isolated-test/wd.err 2>/dev/null | tr '\012\042\134' '  _')
+jl=$(journalctl -u $UNIT --no-pager 2>/dev/null | tail -4 | tr '\012\042\134' '  _' | cut -c1-500)
+printf '{"watchdog":"%s","err":"%s","unit_log":"%s"}\n' "$sc" "$er" "$jl" > $PORT