fix: fourth audit — two regressions from the last fix, and the boot preflight had no test

Two of these were mine, from the previous round.

- mounted_snapshots still swallowed OSError. I said I had fixed it and had not: the
  edit never matched, and I did not read it back. With the mount table unreadable the
  GC loses its in-use protection entirely and can destroy the snapshots of a backup
  that is still uploading (a first upload easily outlives the 1h age floor). It raises
  now, and both behaviours are tested.

- The foreign-dataset check added last round had two bugs of its own. It ignored
  `mounted`, so a locked/encrypted dataset from a sibling tree turned a working nightly
  backup into a permanent failure — it belongs in `skipped`, exactly as an in-tree one
  does. And it tested `mp.startswith(path + "/")`, so a foreign dataset mounted EXACTLY
  at the backup path slipped through — the very hole the check was added to close, one
  character wide, and the worse case of the two because it SHADOWS the base dataset's
  own directory.

- _read_sidecar's new raise broke cleanup_all, which is what recover.sh and
  uninstall.sh call — i.e. the code that must work when the box is ALREADY stuck. One
  unreadable sidecar aborted it before it unmounted anything, leaving the staging tree
  mounted, which pins the snapshots, which is the state recover.sh exists to escape. It
  now reports and carries on — and does not delete a record it could not read.

- compat could report a FALSE OK: `defined` was collected by walking the whole file, so
  any function named `delete` anywhere in it — on an unrelated class, or nested inside
  another method — satisfied "this namespace defines delete". The runtime is stricter
  (a plugin class on the service's MRO), so the two could disagree in the ok direction.
  compat now looks in the class that declares the namespace. Same question on both
  sides, which is what pick_snapshot_service's docstring has been claiming all along.

- apply.sh's compat preflight — the guard that refuses to patch a middleware whose
  assumptions no longer hold, on every boot, on a live NAS — had no test at all. It
  could be turned into a no-op eight different ways with the suite still green. The
  SHIPPED heredoc is now extracted and driven directly against fake verdicts.

Also pinned: the Tap/Tap2 prefix collisions (a sweep that treats "Tap2/data@snap" as
part of Tap's tree DESTROYS another pool's snapshot), and the GC's in_use wiring.

355 tests. Verified on TrueNAS 26.0.0-BETA.1: 292-dataset backup, 0 orphans, 0 leaked
mounts, byte-identical restore of a 4-deep child dataset.

patch/truecloud_nested.py

@@ -690,19 +690,36 @@ def plan_staging(base_dataset, base_mountpoint, path, snapshot_name, datasets,
     # exists to prevent. Stock has the same blind spot, but stock also REFUSES the
     # nested config outright -- we are the ones relaxing that guard, so the hole is
     # ours to close.
-    foreign = sorted(
+    foreign = []
-        ds.get("name", "")
+    for ds in datasets:
-        for ds in datasets
+        name = ds.get("name", "")
-        if (mp := ds.get("properties", {}).get("mountpoint", {}).get("value", ""))
+        if name.startswith(ds_prefix) or name == base_dataset:
-        and mp.startswith(path_prefix)
+            continue                                   # in-tree: handled above
-        and not ds.get("name", "").startswith(ds_prefix)
+
-        and ds.get("name", "") != base_dataset
+        props = ds.get("properties", {})
-    )
+        mp = props.get("mountpoint", {}).get("value", "")
+        # `mp == path` as well as below it. A foreign dataset mounted exactly AT the
+        # backup path is the same hole -- and it is worse, because it shadows the base
+        # dataset's own directory, so we would stage what is hidden underneath instead
+        # of the data actually visible there.
+        if not mp or (mp != path and not mp.startswith(path_prefix)):
+            continue                                   # not inside the backed-up path
+
+        if props.get("mounted", {}).get("value", "yes") == "no":
+            # An unmounted (locked/encrypted) dataset contributes nothing to the live
+            # tree, so its absence is not a hole -- exactly as for an in-tree one, 40
+            # lines above. Raising here would turn a working nightly backup into a
+            # permanent failure the first time somebody locked a dataset.
+            skipped.append((name, "dataset is not mounted (locked/encrypted?)"))
+            continue
+
+        foreign.append(name)
+
     if foreign:
         raise StagingError(
             "dataset(s) outside " + repr(base_dataset) + " are mounted inside the "
             "backup path and cannot be captured by its recursive snapshot: "
-            + ", ".join(repr(f) for f in foreign)
+            + ", ".join(repr(f) for f in sorted(foreign))
             + ". Refusing to back up an incomplete tree -- move them, or back up "
             "their own dataset separately."
         )
@@ -1118,7 +1135,12 @@ def mounted_snapshots(mounts_file=None):
                 if "@" in dev:
                     live.add(dev.replace("\\040", " "))
     except OSError:
-        return set()
+        # An empty set says "nothing is mounted", which silently switches OFF the GC's
+        # protection for snapshots a CONCURRENT run is using -- leaving only the age
+        # floor between us and destroying a snapshot out from under a backup that is
+        # still uploading. If we cannot read the mount table we do not KNOW what is in
+        # use, and must not pretend we do.
+        raise
     return live
 
 
@@ -1418,8 +1440,21 @@ def cleanup_all(base=None, runner=None, mounts_file=None,
     # Report orphaned snapshots BEFORE removing the sidecars that name them --
     # a sidecar is the only record that an interrupted run's snapshot tree (one
     # snapshot per descendant dataset) is still on disk.
+    unreadable = set()
     for sc in sorted(glob_fn(os.path.join(base, "*.snapshot"))):
-        for snap in read_sidecar(sc[: -len(".snapshot")]):
+        try:
+            recorded = read_sidecar(sc[: -len(".snapshot")])
+        except OSError as e:
+            unreadable.add(sc)
+            # REPORT it and carry on. This function's job is to get the mounts off, and
+            # it is called precisely when the box is already in a bad state
+            # (recover.sh, uninstall.sh). Letting one unreadable sidecar abort the run
+            # would leave the staging tree mounted -- which pins the snapshots, which is
+            # the exact situation the caller is trying to escape.
+            lines.append(f"  WARNING: could not read the snapshot record {sc} ({e}).")
+            lines.append("           It may name snapshots nothing else can find.")
+            continue
+        for snap in recorded:
             lines.append(f"  NOTE: an interrupted backup left snapshot '{snap}' behind.")
             lines.append(f"        Remove it and its children:  zfs destroy -r '{snap}'")
 
@@ -1435,6 +1470,12 @@ def cleanup_all(base=None, runner=None, mounts_file=None,
 
     if not errors:
         for sc in glob_fn(os.path.join(base, "*.snapshot")):
+            if sc in unreadable:
+                # Do NOT delete a record we could not read. We have no idea what it
+                # names, and it may be the only thing that knows those snapshots exist.
+                # Removing it here would be the very bug the read guard was added for,
+                # committed by the cleanup path instead of the backup path.
+                continue
             with contextlib.suppress(OSError):
                 os.unlink(sc)
         with contextlib.suppress(OSError):

tests/test_apply_blocks.py

@@ -642,3 +642,112 @@ def test_the_flavour_mapping_is_not_inverted():
             f"the {block} flavour mapping is missing or inverted: _flavour is True for "
             f"an ASYNC middleware, so it must select {block}_ASYNC"
         )
+
+
+# ── the compat preflight ─────────────────────────────────────────────────────
+#
+# This is the guard that stands between a broken middleware and a live NAS: at every
+# boot, apply.sh checks the patch's assumptions against the middlewared actually
+# installed, and REFUSES to apply a module whose assumptions no longer hold.
+#
+# It had no test. An audit turned it into a no-op eight different ways -- `verdict()`
+# always returning 'ok', the broken branch never firing, the kill switch never honoured
+# -- and the suite stayed green every time. The most consequential safety net in the
+# project was unguarded.
+
+def preflight_heredoc():
+    """The preflight's Python, lifted out of apply.sh and made runnable.
+
+    Extracted, not reimplemented: a reimplementation would happily pass while the
+    SHIPPED preflight stayed broken, which is exactly the failure being guarded.
+    """
+    with open(APPLY_SH, encoding="utf-8") as fh:
+        sh = fh.read()
+    # Line-based: the compat heredoc opens with `<<'PYEOF'` on the _tc_compat line and
+    # closes at the next bare PYEOF. (A regex that matched `<< 'PYEOF'` silently found
+    # the OTHER heredoc and ran a different script entirely.)
+    lines = sh.splitlines()
+    start = next(
+        i for i, ln in enumerate(lines)
+        if ln.startswith("_tc_compat=$(") and "<<'PYEOF'" in ln
+    )
+    end = next(i for i in range(start + 1, len(lines)) if lines[i].strip() == "PYEOF")
+    m = "\n".join(lines[start + 1:end])
+    assert m, "could not find the compat preflight heredoc in apply.sh"
+    return m
+
+
+def run_preflight(result, tmp_path):
+    """Run the SHIPPED preflight against a fake compat.check_tree result.
+
+    The heredoc does `import sys`, so a fake `sys` in the namespace is immediately
+    rebound to the real module -- drive the real one instead.
+    """
+    import contextlib
+    import io
+    import sys
+    import types
+
+    src = preflight_heredoc()
+    fake = types.ModuleType("compat")
+    fake.check_tree = lambda _mw: result
+
+    saved_mod = sys.modules.get("compat")
+    saved_argv = sys.argv
+    sys.modules["compat"] = fake
+    sys.argv = ["x", "/patch", "/mw", str(tmp_path / "compat.json")]
+
+    buf = io.StringIO()
+    try:
+        with contextlib.redirect_stdout(buf):
+            exec(compile(src, "apply.sh:preflight", "exec"), {"__name__": "__main__"})  # noqa: S102
+    except SystemExit:
+        pass
+    finally:
+        sys.argv = saved_argv
+        if saved_mod is not None:
+            sys.modules["compat"] = saved_mod
+        else:
+            sys.modules.pop("compat", None)
+    return buf.getvalue().splitlines()
+
+
+def _mod(ok=True, native=False, unknown=False, problems=()):
+    return {"ok": ok, "native": native, "unknown": unknown, "problems": list(problems)}
+
+
+class TestTheBootPreflightRefusesABrokenMiddleware:
+    def test_a_healthy_tree_is_ok(self, tmp_path):
+        out = run_preflight({"providers": _mod(), "nested": _mod()}, tmp_path)
+        assert out[:2] == ["ok", "ok"]
+
+    def test_a_broken_module_is_reported_broken(self, tmp_path):
+        out = run_preflight({
+            "providers": _mod(),
+            "nested": _mod(ok=False, problems=[
+                {"id": "x", "detail": "gone", "why": "orphans every run"},
+            ]),
+        }, tmp_path)
+        assert "broken" in out, (
+            "the preflight did not report a module whose assumptions FAILED. It would "
+            "be injected into a middleware it does not fit -- broken backups, "
+            "discovered at restore time."
+        )
+
+    def test_a_module_that_went_NATIVE_is_also_not_applied(self, tmp_path):
+        # 'native' answers "do we still need it?", 'ok' answers "is it safe to inject?".
+        # Applying a module TrueNAS now implements itself is not safe either.
+        out = run_preflight({
+            "providers": _mod(),
+            "nested": _mod(ok=False, native=True),
+        }, tmp_path)
+        assert "broken" in out
+
+    def test_an_UNKNOWN_verdict_is_not_reported_as_broken(self, tmp_path):
+        # A network error or an unreadable file is not iX deleting our symbols. Calling
+        # it broken would switch a working module off on a healthy box.
+        out = run_preflight({
+            "providers": _mod(unknown=True),
+            "nested": _mod(unknown=True),
+        }, tmp_path)
+        assert "broken" not in out

tests/test_truecloud_nested.py

@@ -1869,3 +1869,150 @@ class TestAnUnreadableZfsDuringTheByNameSweep:
             "with ZFS unreadable we cannot confirm anything was deleted. Returning [] "
             "makes cleanup_task drop the sidecar and orphans the tree forever."
         )
+
+
+class TestTheSnapshotRecordIsNeverLostToAnUnreadableFile:
+    """"There is no sidecar" and "I could not READ the sidecar" are different facts.
+
+    Conflating them let `cleanup_task` take its `if not pinned` branch and UNLINK the
+    only record of a tree it had just failed to read.
+    """
+
+    def test_an_unreadable_sidecar_raises_rather_than_reading_as_empty(self, tmp_path):
+        sc = tmp_path / "cloud_backup-5.snapshot"
+        sc.write_text("Tap@snap\n")
+        sc.chmod(0)
+        try:
+            with pytest.raises(OSError):
+                tn._read_sidecar(str(tmp_path / "cloud_backup-5"))
+        finally:
+            sc.chmod(0o600)
+
+    def test_a_missing_sidecar_is_simply_empty(self, tmp_path):
+        assert tn._read_sidecar(str(tmp_path / "nope")) == []
+
+    def test_cleanup_all_still_UNMOUNTS_when_a_sidecar_cannot_be_read(self, tmp_path):
+        # cleanup_all is what recover.sh and uninstall.sh call, i.e. it runs precisely
+        # when the box is already stuck. Its job is to get the mounts off. Aborting on
+        # one unreadable sidecar leaves the staging tree mounted -- which pins the
+        # snapshots, which is the state the caller is trying to escape.
+        sc = tmp_path / "cloud_backup-5.snapshot"
+        sc.write_text("Tap@snap\n")
+        sc.chmod(0)
+
+        unmounted = []
+
+        class R:
+            returncode = 0
+            stdout = ""
+            stderr = ""
+
+        def runner(cmd):
+            unmounted.append(cmd)
+            return R()
+
+        mounts = tmp_path / "mounts"
+        mounts.write_text(f"Tap@s {tmp_path}/cloud_backup-5 zfs ro 0 0\n")
+        out, _errors = tn.cleanup_all(
+            base=str(tmp_path), runner=runner, mounts_file=str(mounts),
+            glob_fn=lambda pat: [str(sc)],
+        )
+
+        assert sc.exists(), (
+            "cleanup_all DELETED the sidecar it had just failed to read. It has no idea "
+            "what that file names, and it may be the only record those snapshots exist."
+        )
+        sc.chmod(0o600)
+
+        assert any("umount" in " ".join(c) for c in unmounted), (
+            "cleanup_all stopped at the unreadable sidecar and never unmounted. The "
+            "staging tree stays mounted, pinning the snapshots -- which is exactly the "
+            "state recover.sh exists to escape."
+        )
+        assert "could not read the snapshot record" in "\n".join(out).lower()
+
+
+class TestTheMountTableIsNeverGuessedAt:
+    def test_an_unreadable_mount_table_raises_rather_than_saying_nothing_is_mounted(self):
+        # Returning an empty set means "no snapshot is in use", which silently switches
+        # OFF the GC's only protection for a CONCURRENTLY RUNNING backup's snapshots.
+        # A first upload can easily run past the 1h age floor.
+        with pytest.raises(OSError):
+            tn.mounted_snapshots(mounts_file="/nonexistent/mounts")
+
+    def test_the_GC_actually_passes_in_use_through(self, tmp_path, monkeypatch):
+        # The `in_use` guard is well tested as a pure function -- but nothing checked
+        # that gc_stale_snapshots WIRES it in. Deleting the argument passed the suite.
+        monkeypatch.setattr(tn, "STAGING_BASE", str(tmp_path))
+        import datetime as dt
+
+        old = "Tap/apps/x@cloud_backup-5-20260713030000"
+        mounts = tmp_path / "mounts"
+        mounts.write_text(f"{old} /some/where zfs ro 0 0\n")   # ...it is MOUNTED
+
+        mw = FakeMiddleware(["Tap@cloud_backup-5-20260714115900", old])
+        remaining = tn.gc_stale_snapshots(
+            mw, "cloud_backup-5", "Tap@cloud_backup-5-20260714115900",
+            now=dt.datetime(2026, 7, 14, 12, 0, 0, tzinfo=dt.UTC),
+            mounts_file=str(mounts), list_snapshots=mw.list_snapshots,
+        )
+        assert remaining == []
+        assert old in mw.snapshots, (
+            "the GC destroyed a snapshot that is still MOUNTED -- i.e. one a "
+            "concurrently running backup is reading from"
+        )
+
+
+class TestPrefixCollisionsBetweenPools:
+    """`Tap` and `Tap2` are different pools. String prefixes do not know that."""
+
+    def test_the_sweep_does_not_touch_a_similarly_named_pool(self):
+        mw = FakeMiddleware(["Tap@snap", "Tap/apps@snap", "Tap2/data@snap"])
+        tn.delete_snapshot_tree(mw, "Tap@snap", list_snapshots=lambda _d: list(mw.snapshots))
+        assert mw.snapshots == ["Tap2/data@snap"], (
+            "the sweep destroyed a snapshot in Tap2, a DIFFERENT POOL, because 'Tap2' "
+            "starts with 'Tap'"
+        )
+
+    def test_the_planner_does_not_stage_a_similarly_named_dataset(self):
+        noisy = DATASETS + [ds("Tap2/apps", "/mnt/Tap2/apps")]
+        mounts, skipped = plan(datasets=noisy)
+        assert len(mounts) == 6
+        assert skipped == []
+
+    def test_a_similarly_named_MOUNTPOINT_does_not_trip_the_foreign_check(self):
+        # /mnt/Tap2/x is not inside /mnt/Tap.
+        noisy = DATASETS + [ds("Tank/x", "/mnt/Tap2/x")]
+        mounts, skipped = plan(datasets=noisy)
+        assert len(mounts) == 6
+
+    def test_Tap2_is_not_a_child_of_Tap_even_when_mounted_inside_it(self):
+        # The one that actually bites: `Tap2/x` starts with the STRING "Tap", so a
+        # scope test that forgets the trailing "/" treats it as an in-tree descendant
+        # and STAGES it -- from a snapshot `zfs snapshot -r Tap@...` never took, since
+        # Tap2 is a different pool. It must be seen as FOREIGN and refused.
+        noisy = DATASETS + [ds("Tap2/x", "/mnt/Tap/apps/x")]
+        with pytest.raises(StagingError, match="Tap2/x"):
+            plan(datasets=noisy)
+
+
+class TestTheForeignCheckDoesNotBreakWorkingConfigs:
+    def test_an_UNMOUNTED_foreign_dataset_is_skipped_not_refused(self):
+        # A locked/encrypted dataset contributes nothing to the live tree, so its
+        # absence is not a hole -- exactly as for an in-tree one. Raising here would
+        # turn a working nightly backup into a permanent failure the first time
+        # somebody locked a dataset.
+        datasets = DATASETS + [
+            ds("Tank/photos", "/mnt/Tap/apps/photos", mounted="no"),
+        ]
+        mounts, skipped = plan(datasets=datasets)
+        assert len(mounts) == 6
+        assert ("Tank/photos", "dataset is not mounted (locked/encrypted?)") in skipped
+
+    def test_a_foreign_dataset_mounted_EXACTLY_AT_the_path_is_refused(self):
+        # One character wide: `mp.startswith(path + "/")` misses `mp == path`. And this
+        # case is the worse one -- it SHADOWS the base dataset's own directory, so we
+        # would stage what is hidden underneath instead of the data actually there.
+        datasets = DATASETS + [ds("Tank/x", "/mnt/Tap")]
+        with pytest.raises(StagingError, match="Tank/x"):
+            plan(datasets=datasets)

tools/compat.py

@@ -301,37 +301,61 @@ def check_call(c: MiddlewareCall, src: str | None,
     except SyntaxError as e:
         return "unknown", f"{path} does not parse: {e}"
 
-    # namespace = 'zfs.snapshot' on some Service class in this file...
+    # Find the CLASS that declares this namespace, and look for the method THERE.
-    namespaces = {
+    #
-        n.value.value
+    # Not anywhere in the file. `ast.walk` over the whole module made *any* function
-        for n in ast.walk(tree)
+    # called `delete` satisfy the check -- one on an unrelated class, or even a nested
-        if isinstance(n, ast.Assign)
+    # local function inside `do_query`. That is a FALSE OK, and it breaks the one
-        and isinstance(n.value, ast.Constant)
+    # invariant this checker and the runtime share: `_defines_delete()` looks in
-        and isinstance(n.value.value, str)
+    # `vars(klass)` for a PLUGIN class on the service's MRO. If iX gutted
-        and any(isinstance(t, ast.Name) and t.id == "namespace" for t in n.targets)
+    # `PoolSnapshotService.do_delete` while some other class in the same file still had
-    }
+    # a `delete`, compat would say ok, apply.sh would patch, and the runtime would then
-    if namespace not in namespaces:
+    # correctly refuse `pool.snapshot`, fall through to a `zfs.snapshot` that does not
-        return "broken", (
+    # exist on 26, and fail every nested backup on a box the preflight called healthy.
-            f"{path} no longer declares namespace {namespace!r} "
+    #
-            f"(found: {sorted(namespaces) or 'none'}), so `{method}` is gone"
+    # Same question on both sides: does the class that OWNS this namespace define the
-        )
+    # method?
-
-    # ...and it defines the method.
     #
     # A CRUDService exposes `create`/`update`/`delete` from methods NAMED
-    # `do_create`/`do_update`/`do_delete`. Both spellings are live right now:
+    # `do_create`/`do_update`/`do_delete`. Both spellings are live: 24.10 and 25.04
-    # 24.10 and 25.04 declare `do_delete`, 25.10 renamed it to `delete`, and all
+    # declare `do_delete`, 25.10 renamed it to `delete`, and all answer to
-    # three answer to `zfs.snapshot.delete`. Accepting only the literal name reported
+    # `<ns>.delete`. Accepting only the literal name reported working releases as
-    # the two older releases as broken -- a false BROKEN that would have switched off
+    # broken.
-    # nested snapshots on boxes where they work.
+    owners = []
-    defined = {
+    all_namespaces = set()
-        n.name for n in ast.walk(tree)
+    for cls in (n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)):
-        if isinstance(n, ast.FunctionDef | ast.AsyncFunctionDef)
+        declared = {
-    }
+            n.value.value
-    if not any(sp in defined for sp in accepted_spellings(name)):
+            for n in ast.walk(cls)
-        return "broken", f"{path} no longer defines `{method}`"
+            if isinstance(n, ast.Assign)
+            and isinstance(n.value, ast.Constant)
+            and isinstance(n.value.value, str)
+            and any(isinstance(t, ast.Name) and t.id == "namespace" for t in n.targets)
+        }
+        all_namespaces |= declared
+        if namespace in declared:
+            owners.append(cls)
 
-    return "ok", None
+    if not owners:
+        return "broken", (
+            f"{path} no longer declares namespace {namespace!r} "
+            f"(found: {sorted(all_namespaces) or 'none'}), so `{method}` is gone"
+        )
+
+    wanted = accepted_spellings(name)
+    for cls in owners:
+        # Direct members of the class, not its nested scopes: a `def delete` inside
+        # another method is a local function, not a service method.
+        if any(
+            isinstance(n, ast.FunctionDef | ast.AsyncFunctionDef) and n.name in wanted
+            for n in cls.body
+        ):
+            return "ok", None
+
+    return "broken", (
+        f"{path} still declares namespace {namespace!r}, but its class no longer "
+        f"defines `{'` or `'.join(wanted)}` -- so `{method}` is gone"
+    )
 
 
 #: Things that mean iX has done the job themselves and the module should RETIRE,