fuzz(restore): diskfull app create probe runs New Workspace as a person

scripts/fuzz/restore/restore_fuzz.py

@@ -1211,6 +1211,13 @@ def fill(self, margin_mb):
             os.fsync(f.fileno())
         return self.free()
 
+    def trim(self, margin_mb):
+        """Shrinks the filler so the image has at least margin_mb free again (after a fill to margin 0)."""
+        need = margin_mb * (1 << 20) - self.free()
+        if need > 0 and os.path.exists(self.filler):
+            os.truncate(self.filler, max(0, os.path.getsize(self.filler) - need))
+        return self.free()
+
     def unfill(self):
         try:
             os.remove(self.filler)
@@ -1433,6 +1440,154 @@ def recover_after_space(target, old_daemons):
     return out
 
 
+def app_view():
+    """{"readable": both topologies were read, "app": workspace ids the app shows, "daemon": workspace ids the daemon
+    lists, "notices": toasts + dialogs shown, "texts": their text, "raw": the raw answers}. An unreadable view never
+    counts as "no ghost": the probe makes it "not run"."""
+    snap = rpc("snapshot.get")
+    topo = snap.get("topology") if isinstance(snap, dict) else None
+    app_ids = {w.get("id") for w in (topo or {}).get("workspaces", []) if isinstance(w, dict) and w.get("id")}
+    wsl, _ = cli("workspace", "list")
+    daemon_read = isinstance(value(wsl), list) or (isinstance(wsl, dict) and "error" not in wsl)
+    daemon_ids = {first(w, "id", "workspace_id") for w in listed(wsl, "workspaces")} - {None}
+    toast = rpc("debug.toast", {})
+    dialog = rpc("debug.dialog", {})
+    items = []
+    if isinstance(toast, dict):
+        items += list(toast.get("refusals") or [])
+        for window in toast.get("windows") or []:
+            items += list(window.get("toasts") or [])
+    if isinstance(dialog, dict):
+        items += list(dialog.get("dialogs") or [])
+    texts, count = [json.dumps(item)[:300] for item in items], len(items)
+    return {"readable": isinstance(topo, dict) and daemon_read, "app": app_ids, "daemon": daemon_ids,
+            "notices": count, "texts": texts,
+            "raw": {"toast": json.dumps(toast)[:600], "dialog": json.dumps(dialog)[:600]}}
+
+
+CREATE_REFUSAL_WORDS = re.compile(r"workspace|disk|space|create|storage", re.I)
+DISK_FULL = re.compile(r"disk is full|no space left|ENOSPC|SQLITE_FULL|disk full", re.I)
+
+
+def new_refusal_texts(before, after):
+    """Notice texts that are new after the create and are about it (a workspace create or the full disk); a toast
+    for another cause, or an old refusal still in the history, does not count."""
+    old = {}
+    for t in before.get("texts") or []:
+        old[t] = old.get(t, 0) + 1
+    fresh = []
+    for t in after.get("texts") or []:  # a multiset difference: the same refusal shown again is new
+        if old.get(t):
+            old[t] -= 1
+        elif CREATE_REFUSAL_WORDS.search(t):
+            fresh.append(t)
+    return fresh
+
+
+def judge_app_create(before, after, relaunched):
+    """Problems of the app-path create under a full disk: a workspace the app shows and the daemon does not (a ghost),
+    now or after a relaunch (a saved record); a refusal with no new toast or dialog about it for the person."""
+    problems = []
+    for label, view in (("after the create", after), ("after the relaunch", relaunched)):
+        ghost = (view["app"] - view["daemon"]) - (before["app"] - before["daemon"])
+        if ghost:
+            problems.append(f"{label}: ghost workspace(s) {sorted(ghost)} in the app, not in the daemon")
+    made = after["daemon"] - before["daemon"]
+    if not made and not new_refusal_texts(before, after):
+        problems.append("the create was refused, but no visible refusal (no new toast or dialog about it) for the person")
+    return problems
+
+
+def disk_full_refusal(answer):
+    """Is a CLI create answer the daemon's typed disk-full refusal (not a dead socket, a stall or bad output)?"""
+    return (isinstance(answer, dict) and "error" in answer and typed_error(answer)
+            and bool(DISK_FULL.search(json.dumps(answer.get("error")))))
+
+
+PROBE_SETTLE = 5.0  # seconds from the action to the after view (the refusal toast shows)
+PROBE_BOUND = 240  # seconds the probe may take at most: fills, CLI creates, the action, a relaunch and 3 view reads
+
+
+def probe_record(verdict, problems, **extra):
+    rec = {"probe": "app_create", "verdict": verdict, "problems": problems}
+    for key, view in extra.items():
+        rec[key] = {k: sorted(v) if isinstance(v, set) else v for k, v in view.items()} if isinstance(view, dict) else view
+    with open(os.path.join(opts.out, "app-create-probe.json"), "w") as f:
+        json.dump(rec, f, indent=1, default=str)
+    say(f"  app create probe: {verdict} {'; '.join(problems)[:300]}")
+    return rec
+
+
+def app_create_probe():
+    """The person's create (New Workspace, debug.toast perform "newTab" as origin user) on the full disk, judged by
+    judge_app_create. Runs once per diskfull run, before the iterations. A fail only when every view was read and the
+    daemon really refused a CLI create for the full disk just before; otherwise "not run". Leaves the image at
+    --disk-margin-mb free again."""
+    vol = DISK["volume"]
+    try:
+        return run_app_create_probe(vol)
+    finally:
+        vol.trim(opts.disk_margin_mb)
+
+
+def run_app_create_probe(vol):
+    if time.time() > T0 + opts.max_seconds - PROBE_BOUND:
+        return probe_record("not run", ["no time left in the lease for the probe"])
+    # First make sure the daemon refuses a create for the full disk now: fill again (margin 0 on the second try)
+    # until a CLI create answers a typed disk-full error. A create that still works is closed again.
+    refused = None
+    for margin in (opts.disk_margin_mb, 0, 0):
+        vol.fill(margin)
+        answer, _ = cli("workspace", "create", "--name", "fz-probe-check")
+        if disk_full_refusal(answer):
+            refused = answer
+            break
+        made = value(answer).get("workspace_id") if isinstance(value(answer), dict) else None
+        if made:
+            cli("workspace", made, "close")
+    if refused is None:
+        return probe_record("not run", ["the daemon gave no typed disk-full refusal of a CLI create on the full disk"])
+    before = app_view()
+    if not before["readable"]:
+        return probe_record("not run", ["the app or daemon topology could not be read before the create"], before=before)
+    # The person's New Workspace: action id "newTab", titled "New Workspace" (CmuxNextActions WorkspaceActionCatalog).
+    # As a person runs it (debug.toast perform: origin user), so it gets the refusal a person gets.
+    answer = rpc("debug.toast", {"perform": "newTab"}, timeout=20)
+    err = answer.get("error") if isinstance(answer, dict) else None
+    if err is not None and not isinstance(err, dict):
+        # rpc() gives a bare string for a socket timeout, a closed socket or bad output: the app hung or went away.
+        return probe_record("fail", [f"the app gave no answer to New Workspace on the full disk: {str(err)[:200]}"],
+                            before=before, action_answer=json.dumps(answer)[:400])
+    if isinstance(answer, dict) and answer.get("performed") is False:
+        return probe_record("not run", [f"the action did not run: {json.dumps(answer)[:200]}"])
+    if isinstance(err, dict) and err.get("code") == "not_found":
+        return probe_record("not run", [f"the action is unknown: {json.dumps(answer)[:200]}"])
+    time.sleep(PROBE_SETTLE)
+    after = app_view()
+    if not after["readable"]:
+        return probe_record("not run", ["the app or daemon topology could not be read after the create"],
+                            before=before, after=after)
+    made = after["daemon"] - before["daemon"]
+    if made:
+        for ws_id in made:
+            cli("workspace", ws_id, "close")
+        return probe_record("not run", ["the app's create was not refused (the CLI create just before was): "
+                                        "no refusal to judge"], action_answer=json.dumps(answer)[:400])
+    stop_build_processes(exclude=set(APP.daemon_pids()) | set(APP.acpmux_pids()))
+    relaunched_ok = APP.relaunch()
+    loaded = relaunched_ok and wait(lambda: app_state() and app_state()[1], 30, 0.5)
+    relaunched = app_view() if loaded else None
+    if not relaunched_ok:
+        return probe_record("fail", ["the app did not answer after the relaunch"], before=before, after=after)
+    if not relaunched or not relaunched["readable"]:
+        return probe_record("not run", ["the topology could not be read after the relaunch: the saved-record check "
+                                        "was not done"], before=before, after=after)
+    problems = judge_app_create(before, after, relaunched)
+    return probe_record("fail" if problems else "pass", problems, action_answer=json.dumps(answer)[:400],
+                        before=before, after=after, relaunched=relaunched,
+                        new_refusals=new_refusal_texts(before, after))
+
+
 def teardown_diskfull():
     """Detaches the image, puts the original state dir back, removes the image file. Signals are ignored meanwhile
     and each step runs even if one before it failed; nothing here uses wait() or needs stdout."""
@@ -1488,6 +1643,7 @@ def restore_moved_state(state, orig):
 # ---- chaos -------------------------------------------------------------------------------------
 
 CHAOS_PROCS = []
+PROBE = []  # the diskfull app-path create probe record
 
 
 def start_chaos():
@@ -1519,6 +1675,11 @@ def start_chaos():
         say("chaos slowdisk: every app relaunch runs under taskpolicy -d throttle (disk I/O throttled)")
     elif opts.chaos == "diskfull":
         why = setup_diskfull()
+        if not why:
+            try:
+                PROBE.append(app_create_probe())
+            except HarnessInvalid as error:
+                why = str(error)
         if why:
             teardown_diskfull()
             say(f"RESTORE-FUZZ SUMMARY seed=none chaos=diskfull iterations=0 judged=0 not_run=0 violations=0 "
@@ -1639,14 +1800,21 @@ def main():
     degraded = [r for r in records if r["verdict"] == "degraded"]
     kills = kill_counts(records)
     settle = sorted(r["timings"]["settled_s"] for r in judged if r.get("timings", {}).get("settled_s") is not None)
+    for p in PROBE:
+        if p["verdict"] == "fail":
+            violations.append({"record": {"seed": seed, "iteration": "probe", "target": "app-create", "point": "full disk",
+                                          "problems": p["problems"]}, "shrink": None})
+    if any(p["verdict"] == "fail" for p in PROBE):
+        with open(os.path.join(opts.out, "violations.json"), "w") as f:
+            json.dump(violations, f, indent=1, default=str)
     for v in violations:
         r = v["record"]
         say(f"RESTORE-FUZZ VIOLATION seed={r['seed']} iter={r['iteration']} chaos={opts.chaos} target={r.get('target')} "
             f"point={r.get('point')} keep={(v.get('shrink') or {}).get('keep')} what={r['problems'][0][:300]}")
     p = lambda q: settle[min(len(settle) - 1, int(q * len(settle)))] if settle else None
     say(f"RESTORE-FUZZ SUMMARY seed={seed} chaos={opts.chaos} iterations={len(records)} judged={len(judged)} "
         f"not_run={len(records) - len(judged)} kills={json.dumps(kills, sort_keys=True)} violations={len(violations)} "
-        f"chaos_lost={chaos_lost[0]} degraded={len(degraded)} app_daemon_differ={sum(1 for r in records if r.get('app_daemon_differ'))} errors_untyped={sum(r.get('errors_untyped', 0) for r in records)} "
+        f"chaos_lost={chaos_lost[0]} degraded={len(degraded)} app_create_probe={PROBE[0]['verdict'] if PROBE else 'none'} app_daemon_differ={sum(1 for r in records if r.get('app_daemon_differ'))} errors_untyped={sum(r.get('errors_untyped', 0) for r in records)} "
         f"hosts_max={HOSTS_MAX[0]} listed_minus_shown_max={EXCESS_MAX[0]} "
         f"listed_minus_shown_minus_closed30s_max={EXCESS_MAX[1]} "
         f"settle_s p50={p(0.5)} p95={p(0.95)} max={settle[-1] if settle else None}")

scripts/fuzz/restore/test_restore_fuzz.py

@@ -406,6 +406,162 @@ def test_stall_or_loss_in_not_run_fails(self):
         self.assertEqual(rf.not_run_outcome({"verdict": "not run", "problems": [reason]}, []), "not run")
 
 
+class AppCreateProbe(unittest.TestCase):
+    """diskfull: the app's own create path (New Workspace: debug.toast perform "newTab") under a full disk."""
+
+    def test_judge(self):
+        before = {"app": {"ws_1"}, "daemon": {"ws_1"}, "notices": 0, "texts": []}
+        ok = {"app": {"ws_1"}, "daemon": {"ws_1"}, "notices": 1, "texts": ["The disk is full"]}
+        self.assertEqual(rf.judge_app_create(before, ok, ok), [])
+        ghost = {"app": {"ws_1", "ws_2"}, "daemon": {"ws_1"}, "notices": 1, "texts": ["x"]}
+        self.assertTrue(any("ghost" in p for p in rf.judge_app_create(before, ghost, ok)))
+        self.assertTrue(any("after the relaunch" in p for p in rf.judge_app_create(before, ok, ghost)))
+        silent = {"app": {"ws_1"}, "daemon": {"ws_1"}, "notices": 0, "texts": []}
+        self.assertTrue(any("no visible refusal" in p for p in rf.judge_app_create(before, silent, silent)))
+
+    def test_app_view_reads_window_toasts_and_refusals(self):
+        saved_rpc, saved_cli = rf.rpc, rf.cli
+        answers = {"snapshot.get": {"topology": {"workspaces": [{"id": "ws_1"}]}},
+                   "debug.toast": {"refusals": [{"text": "disk is full"}], "windows": [{"toasts": [{"text": "a"}]}]},
+                   "debug.dialog": {"dialogs": []}}
+        rf.rpc = lambda m, p=None, timeout=20: answers.get(m)
+        rf.cli = lambda *a, timeout=10: ([{"id": "ws_1"}], 0.0)
+        try:
+            view = rf.app_view()
+        finally:
+            rf.rpc, rf.cli = saved_rpc, saved_cli
+        self.assertEqual(view["notices"], 2)
+        self.assertTrue(any("disk is full" in t for t in view["texts"]))
+
+    def test_created_for_real_is_not_a_ghost(self):
+        before = {"app": {"ws_1"}, "daemon": {"ws_1"}, "notices": 0, "texts": []}
+        made = {"app": {"ws_1", "ws_2"}, "daemon": {"ws_1", "ws_2"}, "notices": 0, "texts": []}
+        self.assertEqual(rf.judge_app_create(before, made, made), [])
+
+
+class AppCreateProbeDecisions(unittest.TestCase):
+    """Review FIX at 0c648467267 (#30531): every probe path, with fake answers (no app)."""
+
+    FULL = {"error": '{"code":"operation.failed","details":{"reason":"database or disk is full"}}'}
+    VIEW = {"readable": True, "app": {"ws_1"}, "daemon": {"ws_1"}, "notices": 0, "texts": []}
+
+    class Vol:
+        def __init__(self, mounted=True):
+            self.mounted, self.fills, self.trims = mounted, [], []
+
+        def fill(self, margin):
+            if not self.mounted:
+                raise rf.HarnessInvalid("not mounted")
+            self.fills.append(margin)
+            return 0
+
+        def trim(self, margin):
+            self.trims.append(margin)
+            return margin << 20
+
+    class App:
+        def __init__(self, relaunch_ok=True):
+            self.relaunch_ok, self.relaunched = relaunch_ok, 0
+
+        def daemon_pids(self):
+            return []
+
+        def acpmux_pids(self):
+            return []
+
+        def relaunch(self):
+            self.relaunched += 1
+            return self.relaunch_ok
+
+    def run_probe(self, create=None, action=None, views=None, vol=None, app=None, loaded=True):
+        saved = {k: getattr(rf, k, None) for k in ("cli", "rpc", "app_view", "stop_build_processes", "APP", "app_state",
+                                                   "wait", "PROBE_SETTLE")}
+        self.closed = []
+        views = list(views if views is not None else [self.VIEW, dict(self.VIEW, notices=1, texts=['"New Workspace did not apply: the disk is full"']), self.VIEW])
+
+        def fake_cli(*args, timeout=10):
+            if args[:2] == ("workspace", "create"):
+                return (create if create is not None else self.FULL), 0.1
+            if args[-1:] == ("close",):
+                self.closed.append(args[1])
+            return {}, 0.1
+
+        rf.cli = fake_cli
+        rf.rpc = lambda m, p=None, timeout=20: action if action is not None else {"performed": True}
+        rf.app_view = lambda: views.pop(0)
+        rf.stop_build_processes = lambda exclude=(): None
+        rf.APP = app or self.App()
+        rf.app_state = lambda: ({}, loaded)
+        rf.wait = lambda pred, seconds, step=0.25: pred()
+        rf.PROBE_SETTLE = 0
+        rf.DISK["volume"] = vol or self.Vol()
+        try:
+            return rf.app_create_probe()
+        finally:
+            for k, v in saved.items():
+                setattr(rf, k, v)
+            rf.DISK.pop("volume", None)
+
+    def test_pass_and_fill_trimmed_back(self):
+        vol = self.Vol()
+        rec = self.run_probe(vol=vol)
+        self.assertEqual(rec["verdict"], "pass", rec)
+        self.assertEqual(vol.trims, [rf.opts.disk_margin_mb])
+
+    def test_untyped_or_dead_daemon_error_is_not_a_refusal(self):
+        for bad in ({"error": "no daemon socket"}, {"error": "no answer within 10 s", "stall": True},
+                    {"error": "garbage output"}, {"error": '{"code":"not_found"}'}):
+            rec = self.run_probe(create=bad)
+            self.assertEqual(rec["verdict"], "not run", (bad, rec))
+
+    def test_app_timeout_is_a_fail_not_a_crash(self):
+        rec = self.run_probe(action={"error": "timed out"})
+        self.assertEqual(rec["verdict"], "fail")
+        self.assertIn("no answer", rec["problems"][0])
+
+    def test_unknown_action_and_not_performed_are_not_run(self):
+        self.assertEqual(self.run_probe(action={"error": {"code": "not_found"}})["verdict"], "not run")
+        self.assertEqual(self.run_probe(action={"performed": False})["verdict"], "not run")
+
+    def test_accepted_create_is_not_run_and_closed(self):
+        made = dict(self.VIEW, app={"ws_1", "ws_2"}, daemon={"ws_1", "ws_2"})
+        rec = self.run_probe(views=[self.VIEW, made])
+        self.assertEqual(rec["verdict"], "not run")
+        self.assertEqual(self.closed, ["ws_2"])
+
+    def test_unreadable_relaunch_view_is_not_run(self):
+        shown = dict(self.VIEW, notices=1, texts=['"disk is full"'])
+        unreadable = dict(self.VIEW, readable=False, app=set())
+        self.assertEqual(self.run_probe(views=[self.VIEW, shown, unreadable])["verdict"], "not run")
+        self.assertEqual(self.run_probe(views=[self.VIEW, shown], loaded=False)["verdict"], "not run")
+
+    def test_failed_relaunch_is_a_fail(self):
+        shown = dict(self.VIEW, notices=1, texts=['"disk is full"'])
+        self.assertEqual(self.run_probe(views=[self.VIEW, shown], app=self.App(False))["verdict"], "fail")
+
+    def test_unmounted_image_raises_harness_invalid(self):
+        with self.assertRaises(rf.HarnessInvalid):
+            self.run_probe(vol=self.Vol(mounted=False))
+
+    def test_unrelated_or_old_notice_is_no_visible_refusal(self):
+        old = dict(self.VIEW, notices=1, texts=['"disk is full"'])
+        other = dict(self.VIEW, notices=1, texts=['"Copied to clipboard"'])
+        rec = self.run_probe(views=[self.VIEW, other, self.VIEW])
+        self.assertEqual(rec["verdict"], "fail")
+        rec = self.run_probe(views=[old, old, old])
+        self.assertEqual(rec["verdict"], "fail")
+        again = dict(self.VIEW, notices=2, texts=['"disk is full"', '"disk is full"'])
+        self.assertEqual(self.run_probe(views=[old, again, old])["verdict"], "pass")
+
+    def test_no_time_left_is_not_run(self):
+        saved = rf.opts.max_seconds
+        rf.opts.max_seconds = 1
+        try:
+            self.assertEqual(self.run_probe()["verdict"], "not run")
+        finally:
+            rf.opts.max_seconds = saved
+
+
 class Answers(unittest.TestCase):
     def test_bare_list_answer_is_listed(self):
         # `cmux --json workspace list` answers a bare list (fleet run seed 1004, cmux21).