The sidecar must carry EVERY pending tree, not just the newest

Found live, in the code written to prevent exactly this.

The sidecar held ONE snapshot. So a run that reclaimed an older tree, failed to
finish reclaiming it, and then recorded its own snapshot OVERWROTE the only record of
the survivor -- orphaning it permanently.

Observed: job 24 left one snapshot busy and kept the sidecar (correct). Job 46
reclaimed it, hit ZFS's 300s automount window (the runs were minutes apart), left it
behind again, and then wrote its own snapshot over the record. Permanent orphan,
created by the safety net.

The sidecar is now a list. stage_nested carries forward whatever a reclaim could not
delete; cleanup_task sweeps every pending tree and writes back only the survivors.
cleanup_all reports them one per line instead of formatting a list into an f-string
at the user during uninstall.

Job 46 also confirms the automount fix itself: it swept all 256 of its own snapshots
with no straggler.

.github/workflows/compat.yml

@@ -170,14 +170,15 @@ jobs:
         if: ${{ steps.report.outputs.broken == '1' && contains(github.server_url, 'github.com') }}
         env:
           GH_TOKEN: ${{ github.token }}
-          TITLE: "Incompatible with upcoming TrueNAS: ${{ steps.report.outputs.refs }}"
+          TITLE: "TrueNAS compatibility: the patch's assumptions no longer hold"
         run: |
           # One issue per set of broken refs, reopened/updated rather than duplicated
           # daily -- a bot that files the same issue every morning gets muted, and
           # then it is not a warning system any more.
+          # Lowest-numbered match, for the same reason as the Gitea step below.
           existing="$(gh issue list --state all --search "$TITLE" \
                         --json number,title \
-                        --jq '.[] | select(.title == env.TITLE) | .number' | head -1)"
+                        --jq '[.[] | select(.title == env.TITLE) | .number] | min // empty')"
 
           if [ -n "$existing" ]; then
             gh issue comment "$existing" --body-file /tmp/issue.md
@@ -191,7 +192,7 @@ jobs:
         env:
           TOKEN: ${{ secrets.GITEA_TOKEN || github.token }}
           API: ${{ github.server_url }}/api/v1/repos/${{ github.repository }}
-          TITLE: "Incompatible with upcoming TrueNAS: ${{ steps.report.outputs.refs }}"
+          TITLE: "TrueNAS compatibility: the patch's assumptions no longer hold"
         run: |
           # python3, not jq: jq is not guaranteed on a self-hosted runner, and a bug
           # report that dies on a missing tool is a warning system that does not warn.
@@ -214,8 +215,16 @@ jobs:
           # Same title => same issue. Comment on it rather than filing a new one every
           # morning: a bot that duplicates itself daily gets muted, and then it is not
           # a warning system any more.
+          # LOWEST-numbered match, not "whichever the API returns first". Two issues
+          # with the same title already existed once (the old title embedded the ref
+          # list, so the identity changed when that set changed), and an
+          # order-dependent pick would have alternated between them, reopening one and
+          # commenting on the other. Lowest number is stable no matter what the API
+          # sorts by.
           issues = call(f"{api}/issues?state=all&type=issues", "GET")
-          match = next((i for i in issues if i["title"] == title), None)
+          matches = sorted((i for i in issues if i["title"] == title),
+                           key=lambda i: i["number"])
+          match = matches[0] if matches else None
 
           if match:
               n = match["number"]

CHANGELOG.md

@@ -66,6 +66,38 @@ worse than no alert, because one day it carries a security fix.
 
 ### Fixed
 
+- **A few snapshots leaked on every nested run, forever.** Found on real hardware, in
+  the one place it could be: a 256-snapshot backup of `/mnt/Tap` swept 253 cleanly and
+  left **3 behind** with `dataset is busy`.
+
+  The cause is ZFS's own automount. Reading anything under
+  `<dataset>/.zfs/snapshot/<snap>/` makes ZFS **automount that snapshot**, and it stays
+  mounted for `zfs_expire_snapshot` seconds (**300** by default) after the last access.
+  `teardown()` unmounts *our* bind mounts — but not the automount underneath — so
+  `zfs destroy` refuses for exactly the datasets restic read most recently. Then
+  `cleanup_task()` removed the sidecar anyway, destroying the only record that those
+  snapshots existed. Nothing would ever have reclaimed them.
+
+  Three changes, and the third is the one that makes it safe rather than merely
+  unlikely:
+  - `release_snapdirs()` unmounts ZFS's own `.zfs/snapshot` automounts (deepest first)
+    before deleting, so the snapshots are not busy in the first place.
+  - `delete_snapshot_tree()` **retries** the transient busy, and **returns the
+    snapshots it could not delete** instead of swallowing them.
+  - **The sidecar is now removed only on a confirmed-clean sweep** — including on the
+    staging-failure path, which used to remove it *before* the caller swept. The
+    asymmetry is deliberate: a sidecar left behind when the tree is already gone costs
+    one no-op delete on the next run, while a sidecar removed while the tree still
+    exists is unrecoverable. Survivors are reclaimed by the next run.
+
+  **Expect the occasional straggler, and expect it to clean itself up.** On a
+  256-snapshot tree this reliably sweeps ~255 immediately and may leave **one**: it is
+  whatever restic read last, so its 300-second window has barely opened. That one is
+  logged, its sidecar is kept, and the next run reclaims it before doing anything else.
+  The leak is bounded at a single cycle rather than growing without limit — which is
+  the property that actually matters. Blocking a backup job for five minutes to chase
+  the last snapshot would be a worse trade, so it is not made.
+
 - **Installing the patch permanently blocked updating it.** `install.sh` does
   `chmod +x update.sh`, and git recorded `update.sh` as `100644` — so the chmod was a
   *tracked modification*, and `update.sh` refuses to run over a dirty tree. Install

docs/nested-snapshots.md

@@ -95,6 +95,25 @@ re-scan each time.
 
 ### Snapshot lifecycle
 
+> **A snapshot may survive a run, and that is expected.** ZFS **automounts**
+> `<dataset>/.zfs/snapshot/<snap>` the moment it is read, and holds it for
+> `zfs_expire_snapshot` seconds (**300** by default) after the last access. So
+> whatever restic read *last* is still pinned when we try to destroy it, and
+> `zfs destroy` refuses with `dataset is busy`.
+>
+> The patch unmounts those automounts itself and retries, which clears ~255 of 256 on
+> a real pool. The one that remains is **logged, its sidecar is kept, and the next run
+> reclaims it before doing anything else** — so the leak is bounded at a single cycle
+> instead of growing forever. Seeing one `could not delete snapshot … it will be
+> reclaimed on the next run` in the log is normal. Seeing the count *grow* run over run
+> is not, and would be a bug.
+>
+> This is why the sidecar is removed **only on a confirmed-clean sweep**: it is the
+> only record those snapshots exist, and a run that dropped it while they were still
+> around would orphan them permanently. That is precisely what happened before this was
+> fixed.
+
+
 `zfs.snapshot.delete` defaults to **`recursive=False`**, and stock
 `restic_backup()` calls it with no options. Stock is safe only because its
 validation means a *recursive* snapshot never actually happens in the field.

patch/truecloud_nested.py

@@ -58,6 +58,7 @@ import contextlib
 import os
 import stat
 import subprocess
+import time
 
 __all__ = [
     "STAGING_BASE",
@@ -113,21 +114,39 @@ def sidecar_for(staging_root: str) -> str:
     return staging_root + ".snapshot"
 
 
-def _write_sidecar(staging_root: str, snapshot: str) -> None:
-    """Record the pinned snapshot on disk. Blocking; call via run_in_thread."""
+def _write_sidecar(staging_root: str, snapshots) -> None:
+    """Record every snapshot tree this task still owns. One per line.
+
+    A LIST, not a single name -- and that is not over-engineering, it is a bug fix.
+
+    The sidecar used to hold one snapshot, so a run that reclaimed an older tree,
+    FAILED to finish reclaiming it, and then recorded its own snapshot would
+    **overwrite the only record of the survivor** -- orphaning it permanently, which is
+    exactly the outcome the sidecar exists to prevent. Observed live: a snapshot
+    survived one run, the next run's reclaim also failed (ZFS's 300s automount window
+    had not elapsed, because the runs were minutes apart), and the record was
+    destroyed anyway.
+
+    Now every still-pending tree is carried forward until it is actually gone.
+    """
+    if isinstance(snapshots, str):
+        snapshots = [snapshots]
     with contextlib.suppress(OSError):
         os.makedirs(os.path.dirname(staging_root), exist_ok=True)
         with open(sidecar_for(staging_root), "w", encoding="utf-8") as fh:
-            fh.write(snapshot)
+            fh.write("\n".join(dict.fromkeys(snapshots)))   # de-duped, order kept
 
 
-def _read_sidecar(staging_root: str) -> str | None:
-    """The snapshot a previous run recorded here, if any."""
+def _read_sidecar(staging_root: str):
+    """Every snapshot tree a previous run recorded here. [] if none.
+
+    Tolerates the old single-line format, which is just a one-element list.
+    """
     try:
         with open(sidecar_for(staging_root), encoding="utf-8") as fh:
-            return fh.read().strip() or None
+            return [ln.strip() for ln in fh if ln.strip()]
     except OSError:
-        return None
+        return []
 
 
 def _remove_sidecar(staging_root: str) -> None:
@@ -362,6 +381,48 @@ def teardown(staging_root, runner=_run, mounts_file="/proc/self/mounts"):
     return errors
 
 
+def snapdir_automounts(snapshot_name, mounts_file="/proc/self/mounts"):
+    """Every ``<dataset>/.zfs/snapshot/<snap>`` ZFS automount for this snapshot."""
+    suffix = "/.zfs/snapshot/" + snapshot_name
+    found = []
+    try:
+        with open(mounts_file, encoding="utf-8") as fh:
+            for line in fh:
+                parts = line.split()
+                if len(parts) > 1:
+                    mp = parts[1].replace("\\040", " ")
+                    if mp.endswith(suffix):
+                        found.append(mp)
+    except OSError:
+        return []
+    return sorted(found, key=_depth, reverse=True)      # deepest first
+
+
+def release_snapdirs(snapshot_name, runner=_run, mounts_file="/proc/self/mounts"):
+    """Unmount ZFS's OWN snapshot automounts, so the snapshots can be destroyed.
+
+    Reading anything under ``<dataset>/.zfs/snapshot/<snap>/`` makes ZFS **automount**
+    that snapshot, and it stays mounted for ``zfs_expire_snapshot`` seconds (300 by
+    default) after the last access. teardown() unmounts OUR bind mounts -- but the
+    automount underneath them survives, and while it exists ``zfs destroy`` refuses
+    with *"dataset is busy"*.
+
+    Proven on a real pool: a 256-snapshot recursive tree swept cleanly except for the
+    three datasets restic had read most recently. Those failed with EBUSY, and because
+    cleanup_task removed the sidecar anyway, they were orphaned **permanently** -- a
+    small leak, but a growing one, and exactly the failure this module exists to
+    prevent.
+
+    Deepest first, so a child's automount is released before its parent's.
+    """
+    errors = []
+    for mp in snapdir_automounts(snapshot_name, mounts_file=mounts_file):
+        res = runner(["umount", mp])
+        if res.returncode != 0:
+            errors.append(f"{mp}: {(res.stderr or '').strip()}")
+    return errors
+
+
 # ── orchestration (middleware is duck-typed; no middlewared import) ───────────
 #
 # These are SYNCHRONOUS and talk to middlewared via `middleware.call_sync`, which
@@ -424,15 +485,31 @@ def get_dataset_recursive(datasets, directory):
     )
 
 
-def delete_snapshot_tree(middleware, snapshot, logger=None):
+def delete_snapshot_tree(middleware, snapshot, logger=None, attempts=4,
+                         sleep=time.sleep):
     """Delete the parent snapshot AND every child created by ``zfs snapshot -r``.
 
+    Returns the snapshots it could NOT delete -- callers must not throw that away.
+
     ``zfs.snapshot.delete`` is non-recursive by default and stock calls it with
     no options, so relying on stock would orphan one snapshot per descendant
     dataset on every run. Idempotent: tolerates the parent already being gone
     (stock's ``finally`` may have won the race once our mounts were released).
+
+    "dataset is busy" is EXPECTED here and is TRANSIENT. ZFS automounts
+    ``<dataset>/.zfs/snapshot/<snap>`` when it is read and keeps it mounted for
+    ``zfs_expire_snapshot`` seconds (300 by default) afterwards. So the datasets restic
+    touched last are still pinned when we try to destroy them. We release the
+    automounts explicitly and then retry -- on a real 256-snapshot tree, exactly three
+    snapshots hit this, and before the fix they were orphaned permanently.
     """
-    dataset = snapshot.partition("@")[0]
+    dataset, _, snapname = snapshot.partition("@")
+
+    # Release ZFS's own automounts first, or `zfs destroy` refuses with EBUSY on
+    # everything restic read in the last few minutes.
+    for err in release_snapdirs(snapname):
+        if logger:
+            logger.debug("truecloud-patch: could not release snapdir %s", err)
 
     # Fast path: ONE recursive delete removes the parent and every child that
     # `zfs snapshot -r` created (252 on a real pool). Deleting them individually
@@ -441,7 +518,7 @@ def delete_snapshot_tree(middleware, snapshot, logger=None):
     # exists to prevent.
     try:
         middleware.call_sync("zfs.snapshot.delete", snapshot, {"recursive": True})
-        return
+        return []
     except Exception as e:  # noqa: BLE001 - fall through to the explicit sweep
         # Usually just "parent already gone" (stock's finally won the race once our
         # mounts were released), which the sweep below handles. Log it rather than
@@ -472,14 +549,53 @@ def delete_snapshot_tree(middleware, snapshot, logger=None):
             )
         names = [snapshot]
 
-    for name in names:
+    def confirm_gone(failed):
+        """Drop any name ZFS no longer has, even though its delete raised.
+
+        A delete that raised "does not exist" SUCCEEDED as far as we care, and must
+        not be retried or reported. The query is only a refinement: if it cannot be
+        answered we keep the delete's own verdict, rather than inventing survivors --
+        a false survivor keeps the sidecar forever and is reported as a leak that
+        isn't there.
+        """
+        if not failed:
+            return []
         try:
-            middleware.call_sync("zfs.snapshot.delete", name)
-        except Exception as e:  # noqa: BLE001 - already gone is fine
-            if logger:
-                logger.warning(
-                    "truecloud-patch: could not delete snapshot %s: %r", name, e
-                )
+            live = middleware.call_sync(
+                "zfs.snapshot.query", [["name", "^", dataset]], {"select": ["name"]}
+            )
+        except Exception:  # noqa: BLE001 - cannot refine; trust the delete's verdict
+            return list(failed)
+        live = {s["name"] for s in live}
+        return [n for n in failed if n in live]
+
+    remaining = list(names)
+    for attempt in range(attempts):
+        failed = []
+        for name in remaining:
+            try:
+                middleware.call_sync("zfs.snapshot.delete", name)
+            except Exception:  # noqa: BLE001 - busy, or already gone; sorted out below
+                failed.append(name)
+
+        remaining = confirm_gone(failed)
+        if not remaining:
+            return []
+
+        if attempt < attempts - 1:
+            # EBUSY is the automount expiring. Release again (anything that walks
+            # .zfs can re-automount a snapshot) and give it a moment.
+            release_snapdirs(snapname)
+            sleep(5)
+
+    for name in remaining:
+        if logger:
+            logger.warning(
+                "truecloud-patch: could not delete snapshot %s after %d attempts "
+                "(still busy?) -- it will be reclaimed on the next run",
+                name, attempts,
+            )
+    return remaining
 
 
 def stage_nested(middleware, path, snapshot, base_dataset, base_mountpoint,
@@ -508,24 +624,38 @@ def stage_nested(middleware, path, snapshot, base_dataset, base_mountpoint,
     # A previous run may have crashed mid-flight; never build on top of that.
     teardown(staging_root)
 
-    # ...and if it left a sidecar behind, that snapshot tree is still on disk and
-    # nothing else will ever reclaim it. Sweep it before we overwrite the record,
-    # or a single crashed run orphans 160+ snapshots permanently.
-    stale = _read_sidecar(staging_root)
-    if stale and stale != snapshot:
+    # ...and if it left snapshot trees behind, they are still on disk and nothing else
+    # will ever reclaim them. Sweep them before recording our own, or a single crashed
+    # run orphans 160+ snapshots permanently.
+    #
+    # Anything a reclaim FAILS to delete is carried forward, not dropped. Overwriting
+    # the sidecar with only our own snapshot is what destroyed the record of a survivor
+    # once already: the reclaim ran, hit ZFS's 300-second automount window (the runs
+    # were minutes apart), left one snapshot behind, and then the record of it was
+    # overwritten -- a permanent orphan, created by the very code meant to prevent one.
+    pending = []
+    for stale in _read_sidecar(staging_root):
+        if stale == snapshot:
+            continue
         if logger:
             logger.warning(
                 "truecloud-patch: reclaiming snapshot tree from an earlier "
-                "interrupted run: %s", stale,
+                "run: %s", stale,
             )
-        delete_snapshot_tree(middleware, stale, logger=logger)
+        pending.extend(delete_snapshot_tree(middleware, stale, logger=logger))
+
+    if pending and logger:
+        logger.warning(
+            "truecloud-patch: %d snapshot(s) from an earlier run are still busy; "
+            "carrying them forward to the next run", len(pending),
+        )
 
     # Record the snapshot BEFORE mounting anything, not after. middlewared can
     # die at any point (this patch even schedules a restart at boot), and the
     # sidecar is the only thing that survives it -- an in-process dict would take
     # the sole record of a 160-snapshot tree with it. Writing it after apply_plan
     # would leave exactly the crash window the sidecar exists to close.
-    _write_sidecar(staging_root, snapshot)
+    _write_sidecar(staging_root, [*pending, snapshot])
 
     try:
         mounts, skipped = plan_staging(
@@ -541,8 +671,18 @@ def stage_nested(middleware, path, snapshot, base_dataset, base_mountpoint,
         apply_plan(mounts)
         verify_staged(mounts)
     except Exception:
+        # Tear down the mounts, but KEEP the sidecar.
+        #
+        # The caller (SNAPSHOT_BLOCK) sweeps the snapshot tree on the way out, and if
+        # any of it is still busy it will survive -- and the sidecar is the only record
+        # that it exists. Removing it here would orphan those snapshots permanently.
+        #
+        # The asymmetry is deliberate: a sidecar left behind when the tree is already
+        # gone is harmless (the next run tries to delete a tree that is not there,
+        # finds nothing, and moves on), while a sidecar removed while the tree still
+        # exists is unrecoverable. Only a confirmed-clean sweep removes it -- see
+        # cleanup_task().
         teardown(staging_root)
-        _remove_sidecar(staging_root)
         raise
 
     if logger:
@@ -559,9 +699,9 @@ def cleanup_task(middleware, task_name, logger=None):
     Safe to call unconditionally: a no-op when the task was never staged.
     """
     staging_root = staging_root_for(task_name)
-    snapshot = _read_sidecar(staging_root)
+    pinned = _read_sidecar(staging_root)
 
-    if snapshot is None and not os.path.isdir(staging_root):
+    if not pinned and not os.path.isdir(staging_root):
         return  # never staged; nothing to do
 
     errors = teardown(staging_root)
@@ -569,8 +709,39 @@ def cleanup_task(middleware, task_name, logger=None):
         for err in errors:
             logger.warning("truecloud-patch: staging teardown: %s", err)
 
-    if snapshot is not None:
-        delete_snapshot_tree(middleware, snapshot, logger=logger)
+    if not pinned:
+        _remove_sidecar(staging_root)
+        return
+
+    # Every tree this task still owns -- ours, plus anything an earlier run could not
+    # finish reclaiming.
+    survivors = []
+    for snapshot in pinned:
+        survivors.extend(delete_snapshot_tree(middleware, snapshot, logger=logger))
+
+    # KEEP the sidecar if anything survived. It is the only record that those
+    # snapshots exist, and removing it orphans them permanently.
+    #
+    # That is not theoretical: on a real 256-snapshot tree, three snapshots were still
+    # pinned by ZFS's own .zfs/snapshot automount (which lingers for 300s after the
+    # last read), failed to delete with "dataset is busy", and the sidecar was removed
+    # anyway -- so nothing would ever have reclaimed them. A small leak, but one that
+    # grows by a few snapshots on every single run, forever.
+    #
+    # Left in place, the next run's stage_nested() sees a stale sidecar naming a
+    # different snapshot and sweeps that tree first -- by which time the automounts are
+    # long gone and the delete succeeds.
+    if survivors:
+        if logger:
+            logger.warning(
+                "truecloud-patch: %d snapshot(s) could not be deleted (still busy); "
+                "recording them so the next run reclaims them: %s",
+                len(survivors), ", ".join(survivors),
+            )
+        # The SURVIVORS, not the trees we asked to delete. Writing the original list
+        # back would keep re-sweeping trees that are already gone.
+        _write_sidecar(staging_root, survivors)
+        return
 
     _remove_sidecar(staging_root)
 
@@ -599,8 +770,7 @@ def cleanup_all(base=None, runner=_run, mounts_file="/proc/self/mounts",
     # a sidecar is the only record that an interrupted run's snapshot tree (one
     # snapshot per descendant dataset) is still on disk.
     for sc in sorted(glob_fn(os.path.join(base, "*.snapshot"))):
-        snap = read_sidecar(sc[: -len(".snapshot")])
-        if snap:
+        for snap in read_sidecar(sc[: -len(".snapshot")]):
             lines.append(f"  NOTE: an interrupted backup left snapshot '{snap}' behind.")
             lines.append(f"        Remove it and its children:  zfs destroy -r '{snap}'")
 

tests/test_truecloud_nested.py

@@ -348,7 +348,14 @@ class TestStageNestedOrdering:
 
         assert mw.snapshots == [], "the crashed run's snapshot tree must be reclaimed"
 
-    def test_sidecar_is_removed_when_staging_fails(self, tmp_path, monkeypatch):
+    def test_sidecar_is_KEPT_when_staging_fails(self, tmp_path, monkeypatch):
+        # The caller sweeps the snapshot tree on the way out, and anything still busy
+        # SURVIVES that sweep -- with the sidecar as its only record. Removing the
+        # sidecar here would orphan those snapshots permanently.
+        #
+        # The asymmetry is the point: a sidecar left behind when the tree is already
+        # gone costs one no-op delete on the next run; a sidecar removed while the tree
+        # still exists is unrecoverable.
         import truecloud_nested as tn
 
         monkeypatch.setattr(tn, "STAGING_BASE", str(tmp_path))
@@ -361,7 +368,12 @@ class TestStageNestedOrdering:
                 "cloud_backup-5", DATASETS,
             )
 
-        assert not os.path.exists(sidecar_for(root))
+        assert os.path.exists(sidecar_for(root)), (
+            "sidecar removed on staging failure — any snapshot the caller's sweep "
+            "cannot delete is now orphaned forever"
+        )
+        with open(sidecar_for(root), encoding="utf-8") as fh:
+            assert fh.read().strip() == "Tap@snap"
 
 
 class TestCleanupTask:
@@ -609,3 +621,275 @@ class TestStagingRootFor:
         # os.path.join(BASE, "..") normalises to /run — teardown would rmdir it.
         root = staging_root_for(name)
         assert os.path.normpath(root).startswith("/run/truecloud-nested/")
+
+
+class TestZfsAutomountKeepsSnapshotsBusy:
+    """"dataset is busy" is EXPECTED, TRANSIENT, and used to orphan snapshots forever.
+
+    Reading `<dataset>/.zfs/snapshot/<snap>/` makes ZFS **automount** that snapshot,
+    and it stays mounted for zfs_expire_snapshot seconds (300 by default) after the
+    last access. teardown() unmounts OUR bind mounts, but not the automount underneath
+    -- so `zfs destroy` refuses with EBUSY for everything restic read recently.
+
+    Observed on a real 256-snapshot tree: 253 swept cleanly, and the 3 datasets restic
+    had touched last failed with "dataset is busy". cleanup_task then removed the
+    sidecar anyway, so nothing would ever reclaim them. A few snapshots leaked per run,
+    forever.
+    """
+
+    MOUNTS = (
+        "tmpfs /run tmpfs rw 0 0\n"
+        "Tap/apps/prometheus /mnt/Tap/apps/prometheus/.zfs/snapshot/snap1 zfs ro 0 0\n"
+        "Tap/apps/standing/data /mnt/Tap/apps/standing/data/.zfs/snapshot/snap1 zfs ro 0 0\n"
+        "Tap /mnt/Tap/.zfs/snapshot/snap1 zfs ro 0 0\n"
+        "Tap/other /mnt/Tap/other/.zfs/snapshot/OTHER zfs ro 0 0\n"
+    )
+
+    def _mounts_file(self, tmp_path):
+        p = tmp_path / "mounts"
+        p.write_text(self.MOUNTS)
+        return str(p)
+
+    def test_it_finds_the_automounts_for_this_snapshot_only(self, tmp_path):
+        import truecloud_nested as tn
+        found = tn.snapdir_automounts("snap1", mounts_file=self._mounts_file(tmp_path))
+        assert "/mnt/Tap/other/.zfs/snapshot/OTHER" not in found
+        assert len(found) == 3
+
+    def test_deepest_first(self, tmp_path):
+        # A child's automount must be released before its parent's.
+        import truecloud_nested as tn
+        found = tn.snapdir_automounts("snap1", mounts_file=self._mounts_file(tmp_path))
+        assert found[-1] == "/mnt/Tap/.zfs/snapshot/snap1"
+
+    def test_release_snapdirs_unmounts_them(self, tmp_path):
+        import truecloud_nested as tn
+        called = []
+
+        class R:
+            returncode = 0
+            stderr = ""
+
+        def runner(cmd):
+            called.append(cmd)
+            return R()
+
+        errs = tn.release_snapdirs("snap1", runner=runner,
+                                   mounts_file=self._mounts_file(tmp_path))
+        assert errs == []
+        assert all(c[0] == "umount" for c in called)
+        assert len(called) == 3
+
+
+class BusyMiddleware(FakeMiddleware):
+    """Deletes fail with EBUSY until `busy_until_attempt` passes -- like a ZFS
+    automount expiring."""
+
+    def __init__(self, snapshots, busy, busy_for=2):
+        super().__init__(snapshots)
+        self.busy = set(busy)
+        self.busy_for = busy_for
+        self.attempts = 0
+
+    def call_sync(self, method, *args):
+        if method == "zfs.snapshot.delete":
+            name = args[0]
+            opts = args[1] if len(args) > 1 else {}
+            if opts.get("recursive"):
+                raise RuntimeError("cannot destroy snapshot: dataset is busy")
+            if name in self.busy:
+                self.attempts += 1
+                if self.attempts <= self.busy_for * len(self.busy):
+                    raise RuntimeError(f"cannot destroy '{name}': dataset is busy")
+        return super().call_sync(method, *args)
+
+
+class TestDeleteRetriesAndReportsSurvivors:
+    def test_a_transient_busy_is_retried_and_wins(self, monkeypatch):
+        import truecloud_nested as tn
+        monkeypatch.setattr(tn, "release_snapdirs", lambda *a, **k: [])
+
+        mw = BusyMiddleware(
+            ["Tap@snap", "Tap/apps@snap", "Tap/apps/prometheus@snap"],
+            busy=["Tap/apps/prometheus@snap"], busy_for=1,
+        )
+        survivors = tn.delete_snapshot_tree(mw, "Tap@snap", sleep=lambda _s: None)
+        assert survivors == []
+        assert mw.snapshots == []
+
+    def test_a_permanently_busy_snapshot_is_REPORTED_not_swallowed(self, monkeypatch):
+        import truecloud_nested as tn
+        monkeypatch.setattr(tn, "release_snapdirs", lambda *a, **k: [])
+
+        mw = BusyMiddleware(
+            ["Tap@snap", "Tap/apps/prometheus@snap"],
+            busy=["Tap/apps/prometheus@snap"], busy_for=99,
+        )
+        survivors = tn.delete_snapshot_tree(mw, "Tap@snap", sleep=lambda _s: None)
+        assert survivors == ["Tap/apps/prometheus@snap"]
+        assert mw.snapshots == ["Tap/apps/prometheus@snap"]
+
+    def test_the_automounts_are_released_before_deleting(self, monkeypatch):
+        import truecloud_nested as tn
+        order = []
+        monkeypatch.setattr(tn, "release_snapdirs",
+                            lambda name, **k: order.append(("release", name)) or [])
+        mw = FakeMiddleware(["Tap@snap"])
+        real = mw.call_sync
+
+        def spy(method, *args):
+            order.append((method, args[0] if args else None))
+            return real(method, *args)
+
+        mw.call_sync = spy
+        tn.delete_snapshot_tree(mw, "Tap@snap", sleep=lambda _s: None)
+        assert order[0] == ("release", "snap"), order
+
+
+class TestSidecarSurvivesAnIncompleteSweep:
+    def test_the_sidecar_is_KEPT_when_snapshots_could_not_be_deleted(
+        self, tmp_path, monkeypatch
+    ):
+        # It is the ONLY record those snapshots exist. Removing it orphans them
+        # permanently -- which is exactly what happened on the real box.
+        import truecloud_nested as tn
+
+        monkeypatch.setattr(tn, "STAGING_BASE", str(tmp_path))
+        monkeypatch.setattr(tn, "release_snapdirs", lambda *a, **k: [])
+        root = tn.staging_root_for("cloud_backup-5")
+        os.makedirs(root, exist_ok=True)
+        with open(sidecar_for(root), "w", encoding="utf-8") as fh:
+            fh.write("Tap@snap")
+
+        mw = BusyMiddleware(["Tap@snap", "Tap/apps/prometheus@snap"],
+                            busy=["Tap/apps/prometheus@snap"], busy_for=99)
+        monkeypatch.setattr(tn, "delete_snapshot_tree",
+                            lambda m, s, logger=None: ["Tap/apps/prometheus@snap"])
+
+        tn.cleanup_task(mw, "cloud_backup-5")
+        assert os.path.exists(sidecar_for(root)), (
+            "sidecar removed despite survivors — they are now orphaned forever"
+        )
+
+    def test_the_sidecar_is_removed_on_a_clean_sweep(self, tmp_path, monkeypatch):
+        import truecloud_nested as tn
+
+        monkeypatch.setattr(tn, "STAGING_BASE", str(tmp_path))
+        root = tn.staging_root_for("cloud_backup-5")
+        os.makedirs(root, exist_ok=True)
+        with open(sidecar_for(root), "w", encoding="utf-8") as fh:
+            fh.write("Tap@snap")
+
+        monkeypatch.setattr(tn, "delete_snapshot_tree", lambda m, s, logger=None: [])
+        tn.cleanup_task(FakeMiddleware(), "cloud_backup-5")
+        assert not os.path.exists(sidecar_for(root))
+
+
+class TestTheSidecarCarriesEveryPendingTree:
+    """The sidecar holds a LIST, and that is a bug fix, not a generalisation.
+
+    It used to hold ONE snapshot. So a run that reclaimed an older tree, FAILED to
+    finish reclaiming it, and then recorded its own snapshot would **overwrite the only
+    record of the survivor** — orphaning it permanently, via the exact code written to
+    prevent orphans.
+
+    Observed live: a snapshot survived one run; the next run's reclaim also failed
+    (ZFS's 300s automount window had not elapsed, because the two runs were minutes
+    apart); the record was overwritten; the snapshot was orphaned for good.
+    """
+
+    def test_round_trips_a_list(self, tmp_path):
+        import truecloud_nested as tn
+        root = str(tmp_path / "cloud_backup-5")
+        tn._write_sidecar(root, ["Tap@a", "Tap@b"])
+        assert tn._read_sidecar(root) == ["Tap@a", "Tap@b"]
+
+    def test_reads_the_old_single_line_format(self, tmp_path):
+        # Boxes upgrading from an older version have a one-line sidecar on disk.
+        import truecloud_nested as tn
+        root = str(tmp_path / "cloud_backup-5")
+        os.makedirs(os.path.dirname(sidecar_for(root)), exist_ok=True)
+        with open(sidecar_for(root), "w", encoding="utf-8") as fh:
+            fh.write("Tap@legacy")
+        assert tn._read_sidecar(root) == ["Tap@legacy"]
+
+    def test_a_failed_reclaim_is_carried_forward_not_overwritten(
+        self, tmp_path, monkeypatch
+    ):
+        # THE bug. stage_nested reclaims an old tree, cannot finish, then records its
+        # own snapshot -- the survivor must still be in the sidecar afterwards.
+        import truecloud_nested as tn
+
+        monkeypatch.setattr(tn, "STAGING_BASE", str(tmp_path))
+        root = tn.staging_root_for("cloud_backup-5")
+        os.makedirs(os.path.dirname(root), exist_ok=True)
+        tn._write_sidecar(root, ["Tap@old"])
+
+        # The reclaim of Tap@old leaves one snapshot behind (still busy).
+        monkeypatch.setattr(
+            tn, "delete_snapshot_tree",
+            lambda m, s, logger=None: ["Tap/apps/x@old"] if s == "Tap@old" else [],
+        )
+        stub_core(monkeypatch, tn, plan=([("/src", root)], []))
+
+        tn.stage_nested(FakeMiddleware(), "/mnt/Tap", "Tap@new", "Tap", "/mnt/Tap",
+                        "cloud_backup-5", DATASETS)
+
+        recorded = tn._read_sidecar(root)
+        assert "Tap/apps/x@old" in recorded, (
+            "the failed reclaim's survivor was dropped — orphaned forever"
+        )
+        assert "Tap@new" in recorded, "our own snapshot must also be recorded"
+
+    def test_cleanup_sweeps_every_pending_tree_and_records_only_survivors(
+        self, tmp_path, monkeypatch
+    ):
+        import truecloud_nested as tn
+
+        monkeypatch.setattr(tn, "STAGING_BASE", str(tmp_path))
+        root = tn.staging_root_for("cloud_backup-5")
+        os.makedirs(root, exist_ok=True)
+        tn._write_sidecar(root, ["Tap@old", "Tap@new"])
+
+        swept = []
+
+        def fake_delete(m, s, logger=None):
+            swept.append(s)
+            return ["Tap/apps/x@new"] if s == "Tap@new" else []
+
+        monkeypatch.setattr(tn, "delete_snapshot_tree", fake_delete)
+        tn.cleanup_task(FakeMiddleware(), "cloud_backup-5")
+
+        assert swept == ["Tap@old", "Tap@new"], "both pending trees must be swept"
+        # Only the SURVIVOR is written back -- re-recording Tap@old would make every
+        # future run re-sweep a tree that is already gone.
+        assert tn._read_sidecar(root) == ["Tap/apps/x@new"]
+
+    def test_a_fully_clean_sweep_removes_the_sidecar(self, tmp_path, monkeypatch):
+        import truecloud_nested as tn
+
+        monkeypatch.setattr(tn, "STAGING_BASE", str(tmp_path))
+        root = tn.staging_root_for("cloud_backup-5")
+        os.makedirs(root, exist_ok=True)
+        tn._write_sidecar(root, ["Tap@a", "Tap@b"])
+        monkeypatch.setattr(tn, "delete_snapshot_tree", lambda m, s, logger=None: [])
+
+        tn.cleanup_task(FakeMiddleware(), "cloud_backup-5")
+        assert not os.path.exists(sidecar_for(root))
+
+    def test_cleanup_all_reports_each_pending_snapshot_on_its_own_line(self, tmp_path):
+        # It formats them for a human during uninstall. A list rendered into an
+        # f-string would print "['Tap@a', 'Tap@b']" at them.
+        import truecloud_nested as tn
+        root = str(tmp_path / "cloud_backup-5")
+        tn._write_sidecar(root, ["Tap@a", "Tap@b"])
+
+        lines, _errors = tn.cleanup_all(
+            base=str(tmp_path),
+            glob_fn=lambda _p: [sidecar_for(root)],
+            mounts_file=os.devnull,
+        )
+        notes = [ln for ln in lines if "left snapshot" in ln]
+        assert len(notes) == 2
+        assert "'Tap@a'" in notes[0] and "'Tap@b'" in notes[1]
+        assert "[" not in "".join(notes)