upgrade-test: watch memory after the readback (harness v2, R-635/R-462)

After an edge reads back, --soak seconds (default 600) of light load while
the kernel's own oom_kill counter is read host-side from the container's
cgroup. A kill or restart turns proven into failed; a peak over 80% of the
limit adds the memory_tight mark. New Romm fixture; edges M1 / M1old.

Red-proof on scratch 9202: M1old (template as promoted, 512M, 4 workers)
OOM-killed at +76 s -> failed. M1 (current, 768M, 2 workers) proven, 0
kills in 608.5 s, peak 81% -> memory_tight.

Test code only; no template changed.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS

scripts/upgrade-test.py

@@ -13,6 +13,9 @@ Method, per EDGE (one app, one FROM image set, one TO image set):
      nothing after.
   4. re-render with the TO images, `up -d`, settle
   5. VERIFY the seed reads back AGAIN  ← THE RESULT
+  5b. WATCH MEMORY for --soak seconds under light load (R-635): peak against the compose limit, the
+      kernel's own OOM-kill counter and the restart count. A kill or a restart turns `proven` into
+      `failed`; a peak above 80 % of the limit adds the `memory_tight` mark.
   6. ABORT: put the FROM images back, `up -d`, and record what happens
 
 WHAT SUCCESS IS, AND THE RULE THAT DOES *NOT* CARRY OVER FROM survive2.py.
@@ -32,15 +35,16 @@ The undo is an ABORT, never a "rollback". The word is struck — see
 felhom.eu/documentation/architecture/09-update-architecture.md §4: once a migration has run, the old
 image refuses to start on the migrated data, so there is no rollback to speak of.
 
-Usage:  python3 upgrade-test.py <edge-id> [<edge-id> …]      (see EDGES)
+Usage:  python3 upgrade-test.py [--soak SECONDS] <edge-id> [<edge-id> …]   (see EDGES)
         python3 upgrade-test.py --list
+        --soak: how long the memory watch runs after a successful readback (default 600; 0 = off)
 Layout: templates under /opt/upg/templates, evidence under /opt/upg/evidence
 """
 import importlib.util, json, os, re, shutil, subprocess, sys, time
 from datetime import datetime, timezone
 from pathlib import Path
 
-HARNESS_VERSION = 1
+HARNESS_VERSION = 2   # 2: the memory watch after the readback (R-635, 2026-09-23)
 ROOT = Path("/opt/upg")
 TEMPLATES = ROOT / "templates"
 EVIDENCE = ROOT / "evidence"
@@ -113,18 +117,44 @@ EDGES = {
     "U7": dict(app="vikunja", note="real upstream move 2.3.0 -> 2.6.0; the app migrates its own SQLite",
                frm={"vikunja": "vikunja/vikunja:2.3.0"},
                to={"vikunja": "vikunja/vikunja:2.6.0"}),
+
+    # --- added 2026-09-23: the memory watch (R-635) --------------------------------------------
+    # The edge that passed its walk on 2026-09-21, was promoted, and then OOM-looped on demo-hp for
+    # six hours. M1 runs it on the CURRENT template (768M, two workers — the fix); M1old runs it on
+    # the template AS PROMOTED (512M, four workers), taken from catalog commit 15f9ebf and placed in
+    # the templates dir as `romm@15f9ebf`. M1old is the memory watch's RED-PROOF: it must fail.
+    "M1": dict(app="romm", note="catalog move 15f9ebf 5.0.0 -> 5.3.0 on the CURRENT template (768M, 2 workers)",
+               frm={"romm": "rommapp/romm:5.0.0"},
+               to={"romm": "rommapp/romm:5.3.0"}),
+    "M1old": dict(app="romm", template="romm@15f9ebf",
+                  note="the same move on the template AS PROMOTED (512M, 4 workers) — must FAIL the memory watch",
+                  frm={"romm": "rommapp/romm:5.0.0"},
+                  to={"romm": "rommapp/romm:5.3.0"}),
 }
 
+# Light load for the memory watch: paths that make the app's own workers do work. An app not listed
+# gets its front page only. Unauthenticated API calls still reach a worker (they answer 401/403),
+# which is the point — RomM's workers were killed while nginx in front of them answered 200.
+LOAD_PATHS = {
+    "romm": ["/api/heartbeat", "/api/platforms", "/api/roms?limit=50", "/api/collections",
+             "/api/stats", "/api/users/me", "/api/config", "/"],
+}
+LOAD_CONCURRENCY = 4
+MEMORY_TIGHT = 0.80
+
 
 # --- compose plumbing --------------------------------------------------------------------------
 
-def render(app: str, images: dict, workdir: Path, env: dict) -> Path:
+def render(app: str, images: dict, workdir: Path, env: dict, template: str = None) -> Path:
     """Write the catalog template into workdir with `images` substituted per SERVICE.
 
+    `template` names a directory under TEMPLATES other than the app's own — an older revision of
+    the same template kept for a red-proof (`romm@15f9ebf`).
+
     Substitution is per service and only on that service's own `image:` line — never a blind
     string replace, which would also rewrite an image name that appears in a comment or an env var.
     """
-    src = (TEMPLATES / app / "docker-compose.yml").read_text()
+    src = (TEMPLATES / (template or app) / "docker-compose.yml").read_text()
     out, cur = [], None
     for line in src.splitlines():
         m = re.match(r"^  ([A-Za-z0-9_-]+):\s*$", line)
@@ -235,6 +265,156 @@ def engine_state(images: dict):
     return out
 
 
+# --- the memory watch (R-635) ---------------------------------------------------------------------
+#
+# WHY THIS EXISTS. `romm 5.0.0 -> 5.3.0` was walked on 2026-09-21: seeded, read back, `proven`. It was
+# promoted, applied on demo-hp, reached `done` — and two hours later its workers began to be
+# OOM-killed, 4,530 times in six hours, while nginx in front of them answered 200. `proven` meant
+# "the update applied and the data survived", never "the new version runs". This step is the first
+# half of closing that gap: it runs the new version under light load and watches memory.
+#
+# THE INSTRUMENT, and why not `docker inspect`'s OOMKilled. On an LXC guest OOMKilled has been
+# measured FALSE for real kills (2026-09-15, R-528) and TRUE on another box (2026-09-22). The
+# kernel's own counter is the cgroup's `memory.events` `oom_kill`, read HOST-SIDE so it works on
+# images that ship no shell (vikunja's does not). A worker killed inside a container that keeps
+# running moves that counter and moves nothing else — exactly RomM's failure. OOMKilled and the
+# restart count are recorded beside it, never instead of it.
+
+def _cgroup_dir(cid: str):
+    for p in (f"/sys/fs/cgroup/system.slice/docker-{cid}.scope", f"/sys/fs/cgroup/docker/{cid}"):
+        if os.path.isdir(p):
+            return p
+    return None
+
+
+def _read_int(path: str):
+    try:
+        v = open(path).read().strip()
+        return None if v == "max" else int(v)
+    except (OSError, ValueError):
+        return None
+
+
+def _oom_kills(cg: str):
+    try:
+        for line in open(os.path.join(cg, "memory.events")):
+            k, _, v = line.partition(" ")
+            if k == "oom_kill":
+                return int(v)
+    except (OSError, ValueError):
+        pass
+    return None
+
+
+def memory_snapshot(project: str, workdir: Path):
+    """One reading per container: usage, peak, limit, kernel OOM kills, restarts, OOMKilled."""
+    out = {}
+    r = compose(workdir, project, "ps", "-aq", timeout=120)
+    for cid in [c for c in r.stdout.split() if c]:
+        info = cvp._inspect(cid)
+        if not info:
+            continue
+        full = info["Id"]
+        name = info["Name"].lstrip("/")
+        cg = _cgroup_dir(full)
+        limit = (info.get("HostConfig") or {}).get("Memory") or None
+        out[name] = {
+            "limit": limit or (_read_int(os.path.join(cg, "memory.max")) if cg else None),
+            "current": _read_int(os.path.join(cg, "memory.current")) if cg else None,
+            "peak": _read_int(os.path.join(cg, "memory.peak")) if cg else None,
+            "oom_kill": _oom_kills(cg) if cg else None,
+            "restarts": (info.get("State") or {}).get("RestartCount", 0),
+            "oomkilled_flag": (info.get("State") or {}).get("OOMKilled"),
+            "status": (info.get("State") or {}).get("Status"),
+            "cgroup": bool(cg),
+        }
+    return out
+
+
+def memory_watch(app: str, project: str, workdir: Path, seconds: int, say, ev: Path):
+    """Run light load for `seconds`, sampling every 15 s. Returns the `memory` record and the marks.
+
+    Stops EARLY at the first kernel OOM kill or restart: the verdict is decided then, and the time
+    to the first kill is the number worth having. A container whose cgroup cannot be read is
+    reported `unmeasured` — "we could not look" never reads as "nothing happened" (R-96 rule 3).
+    """
+    import threading, urllib.error, urllib.request
+    paths = LOAD_PATHS.get(app, ["/"])
+    target = container_ip(getattr(fx.FIXTURES.get(app), "container", app))
+    port = getattr(fx.FIXTURES.get(app), "port", 80)
+    stop = threading.Event()
+    hits = {"n": 0, "codes": {}}
+    lock = threading.Lock()
+
+    def worker(i):
+        k = 0
+        while not stop.is_set():
+            url = f"http://{target}:{port}{paths[(i + k) % len(paths)]}"
+            k += 1
+            try:
+                code = urllib.request.urlopen(url, timeout=20).status
+            except urllib.error.HTTPError as e:
+                code = e.code
+            except Exception:
+                code = "err"
+            with lock:
+                hits["n"] += 1
+                hits["codes"][str(code)] = hits["codes"].get(str(code), 0) + 1
+            time.sleep(0.2)
+
+    base0 = memory_snapshot(project, workdir)
+    unmeasured = [n for n, v in base0.items() if not v["cgroup"]]
+    say(f"memory watch: {seconds}s, {LOAD_CONCURRENCY} callers on {len(paths)} path(s) at {target}:{port}"
+        + (f"; UNMEASURED (no cgroup): {unmeasured}" if unmeasured else ""))
+    threads = [threading.Thread(target=worker, args=(i,), daemon=True) for i in range(LOAD_CONCURRENCY)]
+    for t in threads:
+        t.start()
+    t0, samples, first_bad = time.time(), [], None
+    try:
+        while time.time() - t0 < seconds:
+            time.sleep(15)
+            snap = memory_snapshot(project, workdir)
+            el = round(time.time() - t0, 1)
+            samples.append({"t": el, "containers": snap, "requests": hits["n"]})
+            line = "  ".join(
+                f"{n}={(v['current'] or 0) // 2**20}M/{(v['limit'] or 0) // 2**20}M"
+                f" peak={(v['peak'] or 0) // 2**20}M kills={v['oom_kill']} rs={v['restarts']}"
+                for n, v in snap.items())
+            say(f"  +{el:5.0f}s  {line}  reqs={hits['n']}")
+            for n, v in snap.items():
+                b = base0.get(n, {})
+                dk = (v["oom_kill"] or 0) - (b.get("oom_kill") or 0)
+                dr = (v["restarts"] or 0) - (b.get("restarts") or 0)
+                if (dk > 0 or dr > 0) and first_bad is None:
+                    first_bad = {"t": el, "container": n, "oom_kills": dk, "restarts": dr}
+            if first_bad:
+                say(f"memory watch: STOPPING EARLY — {first_bad}")
+                break
+    finally:
+        stop.set()
+        for t in threads:
+            t.join(timeout=25)
+    end = memory_snapshot(project, workdir)
+    per = {}
+    for n, v in end.items():
+        b = base0.get(n, {})
+        pk, lim = v["peak"], v["limit"]
+        per[n] = {"limit": lim, "peak": pk,
+                  "peak_pct": round(pk / lim, 3) if (pk and lim) else None,
+                  "oom_kills": None if v["oom_kill"] is None else v["oom_kill"] - (b.get("oom_kill") or 0),
+                  "restarts": (v["restarts"] or 0) - (b.get("restarts") or 0),
+                  "oomkilled_flag": v["oomkilled_flag"], "measured": v["cgroup"]}
+    (ev / "memory-samples.json").write_text(json.dumps(samples, indent=2))
+    killed = any((p["oom_kills"] or 0) > 0 or p["restarts"] > 0 or p["oomkilled_flag"] for p in per.values())
+    tight = [n for n, p in per.items() if p["peak_pct"] is not None and p["peak_pct"] > MEMORY_TIGHT]
+    rec = {"soak_s": round(time.time() - t0, 1), "requested_s": seconds, "requests": hits["n"],
+           "codes": hits["codes"], "first_kill": first_bad, "containers": per,
+           "unmeasured": [n for n, p in per.items() if not p["measured"]]}
+    marks = ["memory_tight"] if tight and not killed else []
+    say(f"memory watch: killed={killed} tight={tight} requests={hits['n']} codes={hits['codes']}")
+    return rec, killed, marks
+
+
 MIGRATION_RE = re.compile(
     r"migrat|upgrad|schema|alter table|CREATE TABLE|InnoDB: Upgrad|mysql_upgrade|"
     r"mariadb-upgrade|Running .* migration|Applying|db:migrate",
@@ -260,8 +440,9 @@ def run_edge(edge_id: str) -> dict:
     ev.mkdir(parents=True, exist_ok=True)
     shutil.rmtree(workdir, ignore_errors=True)
 
-    felhom = (TEMPLATES / app / ".felhom.yml").read_text()
-    compose_text = (TEMPLATES / app / "docker-compose.yml").read_text()
+    tdir = e.get("template") or app
+    felhom = (TEMPLATES / tdir / ".felhom.yml").read_text()
+    compose_text = (TEMPLATES / tdir / "docker-compose.yml").read_text()
     env = cvp.build_env(app, felhom, compose_text)
 
     rec = {"harness_version": HARNESS_VERSION, "edge": edge_id, "app": app, "note": e["note"],
@@ -270,6 +451,9 @@ def run_edge(edge_id: str) -> dict:
            "migration_observed": None, "abort": "not-attempted", "abort_detail": None,
            # engine_state is a REPORT, not a judgement — see ENGINE_PROBES.
            "engine_state_after": None,
+           # memory is a MEASUREMENT beside the verdict; a kill or a restart during it DOES decide
+           # the verdict, a tight peak only marks it (R-635, 2026-09-23).
+           "memory": None, "marks": [],
            "duration_s": 0, "measured_at": None, "evidence": f"evidence/{edge_id}"}
     t0 = time.time()
     log = []
@@ -282,7 +466,7 @@ def run_edge(edge_id: str) -> dict:
     try:
         # --- 1. FROM ---
         say(f"{edge_id}: deploying {app} at FROM {e['frm']}")
-        render(app, e["frm"], workdir, env)
+        render(app, e["frm"], workdir, env, e.get("template"))
         up = compose(workdir, project, "up", "-d")
         if up.returncode != 0:
             rec["abort_detail"] = f"FROM deploy failed rc={up.returncode}: {up.stderr[-800:]}"
@@ -317,7 +501,7 @@ def run_edge(edge_id: str) -> dict:
         # --- 4. TO ---
         swap_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
         say(f"{edge_id}: swapping to TO {e['to']}")
-        render(app, e["to"], workdir, env)
+        render(app, e["to"], workdir, env, e.get("template"))
         up2 = compose(workdir, project, "up", "-d")
         say(f"TO up -d rc={up2.returncode}")
         ok2, secs2, states2 = settle(project, workdir)
@@ -351,9 +535,17 @@ def run_edge(edge_id: str) -> dict:
         say(f"RESULT (seed reads back AFTER): {rec['seed_read_after']}")
         rec["verdict"] = "proven" if (ok2 and rec["seed_read_after"]) else "failed"
 
+        # --- 5b. the MEMORY WATCH — only for an edge that just read back; a failed one is decided ---
+        if rec["verdict"] == "proven" and SOAK_SECONDS > 0:
+            mem, killed, marks = memory_watch(app, project, workdir, SOAK_SECONDS, say, ev)
+            rec["memory"], rec["marks"] = mem, marks
+            if killed:
+                rec["verdict"] = "failed"
+                say("VERDICT -> failed: the new version was OOM-killed or restarted under light load")
+
         # --- 6. the ABORT ---
         say(f"{edge_id}: ABORT — putting the FROM images back")
-        render(app, e["frm"], workdir, env)
+        render(app, e["frm"], workdir, env, e.get("template"))
         up3 = compose(workdir, project, "up", "-d")
         ok3, secs3, states3 = settle(project, workdir, wait=180)
         (ev / "abort-states.json").write_text(json.dumps(states3, indent=2))
@@ -383,7 +575,14 @@ def run_edge(edge_id: str) -> dict:
         compose(workdir, project, "down", "-v", "--remove-orphans", timeout=900)
 
 
+SOAK_SECONDS = 600
+
+
 def main(argv):
+    global SOAK_SECONDS
+    if argv and argv[0] == "--soak":
+        SOAK_SECONDS = int(argv[1])
+        argv = argv[2:]
     if not argv or argv[0] == "--list":
         for k, v in EDGES.items():
             print(f"{k:5s} {v['app']:12s} {v['note']}")

scripts/upgrade_fixtures.py

@@ -488,6 +488,77 @@ class Vikunja:
         return ok
 
 
+# ---------------------------------------------------------------------------------------------
+class Romm:
+    """RomM's own user API, driven the way RomM's own front end drives it (added 2026-09-23 for the
+    memory watch's red-proof, R-635). Ported from the box-side fixture that walked the 5.0.0 -> 5.3.0
+    edge on guest 9202 on 2026-09-21, where each of these was measured rather than guessed:
+
+    - RomM sets a `romm_csrftoken` cookie on any GET and requires it back as an `x-csrftoken`
+      header; a bare POST is `403 CSRF token verification failed`, which reads like an auth fault.
+    - The fields go in the JSON BODY; a query-string POST is `422 Field required`.
+    - The first `POST /api/users` on a fresh install is accepted unauthenticated; afterwards it is
+      not — which is what makes `POST /api/login` as that user a real authentication.
+
+    LIMITATION: the DATABASE half only. RomM's other half is the ROM library on the drive.
+    """
+    port = 8080
+    container = "romm"
+
+    def _base(self, ipfn):
+        ip = ipfn(self.container)
+        return f"http://{ip}:{self.port}" if ip else ""
+
+    def _csrf(self, base):
+        jar = f"/tmp/romm-{secrets.token_hex(4)}.jar"
+        _curl(base + "/api/heartbeat", "-c", jar)
+        tok = ""
+        try:
+            for line in open(jar):
+                if "csrf" in line.lower():
+                    tok = line.split()[-1]
+        except OSError:
+            pass
+        return jar, tok
+
+    def seed(self, ipfn, say):
+        base = self._base(ipfn)
+        if not base or not _wait_http(base + "/api/heartbeat", {"200"}, tries=72, say=say):
+            say("    romm: no container IP, or the app never answered /api/heartbeat")
+            return None
+        jar, tok = self._csrf(base)
+        if not tok:
+            say("    romm: no romm_csrftoken cookie was set on /api/heartbeat")
+            return None
+        user = "spike" + secrets.token_hex(3)
+        pw = "Spike-" + secrets.token_hex(10)
+        rc, code, out = _curl(base + "/api/users", "-b", jar, "-H", f"x-csrftoken: {tok}",
+                              "-H", "Content-Type: application/json",
+                              data=json.dumps({"username": user, "email": f"{user}@gate.invalid",
+                                               "password": pw, "role": "admin"}), method="POST")
+        say(f"    romm: POST /api/users http={code}")
+        if code not in ("200", "201"):
+            say(f"    romm: refused {out[:200]}")
+            return None
+        return {"user": user, "pw": pw}
+
+    def verify(self, ipfn, seeded, say):
+        base = self._base(ipfn)
+        if not base or not _wait_http(base + "/api/heartbeat", {"200"}, tries=36, say=say):
+            return False
+        jar, tok = self._csrf(base)
+        rc, code, _ = _curl(base + "/api/login", "-b", jar, "-H", f"x-csrftoken: {tok}",
+                            "-u", f"{seeded['user']}:wrong-{secrets.token_hex(5)}", method="POST")
+        if code == "200":
+            say("    romm: READBACK IS UNUSABLE — a wrong password authenticated")
+            return False
+        rc, code, out = _curl(base + "/api/login", "-b", jar, "-H", f"x-csrftoken: {tok}",
+                              "-u", f"{seeded['user']}:{seeded['pw']}", method="POST")
+        ok = code == "200"
+        say(f"    romm: login as the seeded user http={code} ok={ok}")
+        return ok
+
+
 FIXTURES = {
     "privatebin": PrivateBin(),
     "docmost": Docmost(),
@@ -496,4 +567,5 @@ FIXTURES = {
     "navidrome": Navidrome(),
     "audiobookshelf": AudiobookShelf(),
     "vikunja": Vikunja(),
+    "romm": Romm(),
 }