Support ZFS snapshots on datasets with child datasets

TrueCloud Backup's "Take Snapshot" option is rejected on any path containing
child datasets:

  This option is only available for datasets that have no further nesting

That excludes every pool running Apps, where each app is its own dataset and
often has config/pgdata children. Without the option the backup reads live
files, so databases are captured mid-write and an app that continuously
rewrites its files can stall a run as restic chases a moving target.

The stock guard is correct and must not simply be removed. create_snapshot()
already takes a recursive ZFS snapshot, but points the backup tool at the
parent dataset's .zfs/snapshot/, and ZFS does not expose child datasets there:

  /mnt/Tap/.zfs/snapshot/<snap>/apps/                -> 0 entries
  /mnt/Tap/apps/lidarr/config/.zfs/snapshot/<snap>/  -> the real data

Deleting the check would make restic walk a near-empty tree, report success,
and upload almost nothing.

Implement the missing traversal instead. After the recursive snapshot is taken,
each descendant dataset's own .zfs/snapshot/<snap> is bind-mounted into a
staging tree mirroring the original layout, and the backup tool is pointed at
the staging root. The guard is relaxed only after that machinery is in place.

Safety properties:
- staging failure aborts the backup; a partial tree is never handed to restic
- a post-mount pass asserts every target is a mountpoint and the root is
  non-empty, so this cannot regress into the empty backup it exists to prevent
- apply.sh patches crud.py last, so a partial failure leaves the guard intact
  rather than exposing "guard removed, traversal missing"
- every injected block no-ops when _truecloud_nested is absent
- unmountable/locked datasets are skipped and reported, never dropped silently
- scoped to cloud_backup; cloudsync has no teardown wired in, so its guard stays

The staging root is stable per task, so restic can find its parent snapshot
between runs; stock's timestamped .zfs path changes every run and forces a
full re-scan.

Add CI (shellcheck, bash -n, ruff, pytest on 3.11-3.13), including tests that
compile the *_BLOCK strings, which are Python source appended to live
middlewared modules and were previously unchecked.

Also: sync stale version strings, untrack a committed .pyc, gitignore
__pycache__.

patch/apply.sh

@@ -7,13 +7,21 @@
 # is already up and has already imported the stock modules — the on-disk
 # patch alone cannot reach the running process.
 #
-# TrueNAS updates replace /usr/ entirely; this script re-applies two patches:
+# TrueNAS updates replace /usr/ entirely; this script re-applies three patches:
 #
 #   1. Backend — b2.py and restic.py are patched directly in the overlay.
 #      On a boot run, a single detached middlewared restart is scheduled
 #      (Step 3) so the patched modules actually get loaded.
 #
-#   2. Angular JS bundle — Widens the TrueCloud Backup credential dropdown
+#   2. Nested-dataset snapshots — installs _truecloud_nested.py and patches
+#      plugins/cloud/{snapshot,crud}.py + plugins/cloud_backup/sync.py so the
+#      "Take Snapshot" option works on a dataset that has child datasets.
+#      Stock middleware refuses that config, because it points the backup tool
+#      at the PARENT's .zfs/snapshot/ where children are invisible — it would
+#      silently back up a near-empty tree. We stage a complete tree of
+#      per-dataset bind mounts and only then relax the guard.
+#
+#   3. Angular JS bundle — Widens the TrueCloud Backup credential dropdown
 #                          from Storj-only to include S3 and B2. Served from
 #                          disk per request, so no restart is needed for it.
 #
@@ -24,7 +32,7 @@
 # Derive PATCH_DIR from this script's location (parent of the patch/ directory).
 PATCH_DIR="$(cd "$(dirname "$0")/.." && pwd)"
 LOG="$PATCH_DIR/apply.log"
-VERSION="0.2.1"
+VERSION="0.3.0"
 
 # Rotate log at 512 KB to avoid unbounded growth on a system volume.
 # Keep two prior generations (.1 and .2) so the last three boots are always available.
@@ -175,12 +183,18 @@ elif [ -z "$_MW_DIR" ]; then
 else
     _B2_PY="$_MW_DIR/rclone/remote/b2.py"
     _RESTIC_PY="$_MW_DIR/plugins/cloud_backup/restic.py"
+    _CLOUD_DIR="$_MW_DIR/plugins/cloud"
+    _SYNC_PY="$_MW_DIR/plugins/cloud_backup/sync.py"
+    _NESTED_SRC="$PATCH_DIR/patch/truecloud_nested.py"
 
-        # ── patch b2.py + restic.py + hook_status.json (single subprocess) ──────
-        if "$PYTHON" - "$_B2_PY" "$_RESTIC_PY" "$PATCH_DIR/hook_status.json" << 'PYEOF'
-import json, os, sys, time
+        # ── patch b2.py + restic.py + nested-snapshot + hook_status.json ────────
+        # (single subprocess: PREINIT has a tight timeout budget)
+        if "$PYTHON" - "$_B2_PY" "$_RESTIC_PY" "$PATCH_DIR/hook_status.json" \
+                      "$_CLOUD_DIR" "$_SYNC_PY" "$_NESTED_SRC" << 'PYEOF'
+import json, os, shutil, sys, time
 
 b2_path, restic_path, status_path = sys.argv[1], sys.argv[2], sys.argv[3]
+cloud_dir, sync_path, nested_src = sys.argv[4], sys.argv[5], sys.argv[6]
 
 B2_BLOCK = """
 # TRUECLOUD_PATCH — added by truenas-truecloud-patch/patch/apply.sh
@@ -240,6 +254,116 @@ else:
     get_restic_config._truecloud_patched = True
 """
 
+# ── nested-dataset snapshot support ───────────────────────────────────────────
+# Stock middleware refuses `snapshot=true` on a path containing child datasets,
+# because it points the backup tool at the PARENT dataset's .zfs/snapshot/,
+# where child datasets are INVISIBLE -- it would silently back up a near-empty
+# tree. That guard is correct. We implement the missing traversal (a staging
+# tree of per-dataset bind mounts) and only then relax the guard.
+#
+# Fail-safe direction: if any of these three blocks fails to apply, the stock
+# guard remains and the option simply stays unavailable. We never end up with
+# the guard removed but the traversal missing -- that would be a silently empty
+# backup, the worst possible outcome.
+
+SNAPSHOT_BLOCK = """
+# TRUECLOUD_PATCH — added by truenas-truecloud-patch/patch/apply.sh
+try:
+    from middlewared.plugins.cloud import _truecloud_nested as _tc_nested
+except ImportError:
+    _tc_nested = None
+
+if _tc_nested is not None:
+    _tc_orig_create_snapshot = create_snapshot
+
+    async def create_snapshot(middleware, path, name="cloud_task-onetime"):
+        # Determine nesting BEFORE delegating. We must never silently fall back
+        # to stock behaviour on a nested path: stock returns the parent's
+        # .zfs/snapshot/ path, where children are invisible, which is exactly
+        # the near-empty backup this feature exists to prevent. If we cannot
+        # tell, we raise -- loud failure beats a backup that lies.
+        datasets = await middleware.call("zfs.dataset.query", [["type", "=", "FILESYSTEM"]])
+        dataset, nested = get_dataset_recursive(datasets, path)
+
+        # Stock takes the (already recursive) snapshot; we only replace the PATH.
+        snapshot, snap_path = await _tc_orig_create_snapshot(middleware, path, name)
+
+        if not nested:
+            return snapshot, snap_path  # no children: stock behaviour, untouched
+
+        staging_root = await _tc_nested.stage_nested(
+            middleware, path, snapshot,
+            dataset["properties"]["mountpoint"]["value"], name,
+            logger=getattr(middleware, "logger", None),
+        )
+        return snapshot, staging_root
+
+    create_snapshot._truecloud_patched = True
+"""
+
+CRUD_BLOCK = """
+# TRUECLOUD_PATCH — added by truenas-truecloud-patch/patch/apply.sh
+try:
+    from middlewared.plugins.cloud import _truecloud_nested as _tc_nested
+except ImportError:
+    _tc_nested = None
+
+if _tc_nested is not None:
+    _tc_orig_validate = CloudTaskServiceMixin._validate
+
+    async def _tc_validate(self, app, verrors, name, data):
+        await _tc_orig_validate(self, app, verrors, name, data)
+
+        # Only cloud_backup: staging teardown is wired into cloud_backup.sync's
+        # finally. cloudsync would leak bind mounts, so leave its guard intact.
+        if getattr(getattr(self, "_config", None), "namespace", "") != "cloud_backup":
+            return
+
+        # Drop ONLY the nested-dataset guard. If iX ever rewords the message the
+        # filter stops matching, the guard survives, and the option merely stays
+        # unavailable -- the safe direction to fail.
+        verrors.errors = [
+            e for e in verrors.errors
+            if not (
+                getattr(e, "attribute", "") == f"{name}.snapshot"
+                and "no further nesting" in getattr(e, "errmsg", "")
+            )
+        ]
+
+    CloudTaskServiceMixin._validate = _tc_validate
+    CloudTaskServiceMixin._validate._truecloud_patched = True
+"""
+
+SYNC_BLOCK = """
+# TRUECLOUD_PATCH — added by truenas-truecloud-patch/patch/apply.sh
+try:
+    from middlewared.plugins.cloud import _truecloud_nested as _tc_nested
+except ImportError:
+    _tc_nested = None
+
+if _tc_nested is not None:
+    _tc_orig_restic_backup = restic_backup
+
+    async def restic_backup(middleware, job, cloud_backup, dry_run=False, rate_limit=None):
+        # Our bind mounts pin the ZFS snapshot, so stock's `finally` cannot
+        # destroy it (EBUSY) and logs one benign warning. We unmount here and
+        # then delete the snapshot for real.
+        try:
+            return await _tc_orig_restic_backup(middleware, job, cloud_backup, dry_run, rate_limit)
+        finally:
+            try:
+                await _tc_nested.cleanup_task(
+                    middleware,
+                    f"cloud_backup-{cloud_backup.get('id', 'onetime')}",
+                    logger=getattr(middleware, "logger", None),
+                )
+            except Exception as e:
+                middleware.logger.warning("truecloud-patch: staging cleanup failed: %r", e)
+
+    restic_backup._truecloud_patched = True
+"""
+
+
 def patch_file(path, block):
     with open(path, encoding="utf-8") as fh:
         content = fh.read()
@@ -250,6 +374,7 @@ def patch_file(path, block):
         fh.write(base.rstrip("\n") + "\n" + block)
 
 b2_ok = restic_ok = False
+nested_ok = False
 
 if os.path.exists(b2_path):
     try:
@@ -271,6 +396,35 @@ if os.path.exists(restic_path):
 else:
     print(f"WARNING: restic.py not found at {restic_path}")
 
+# ── nested-dataset snapshot support ───────────────────────────────────────────
+# Order matters: install the traversal machinery FIRST, relax the validation
+# guard LAST. If anything fails partway, the guard is still in place and the
+# option stays unavailable -- we never expose "guard removed, traversal missing".
+nested_detail = ''
+try:
+    snapshot_py = os.path.join(cloud_dir, 'snapshot.py')
+    crud_py = os.path.join(cloud_dir, 'crud.py')
+    nested_dst = os.path.join(cloud_dir, '_truecloud_nested.py')
+
+    missing = [p for p in (snapshot_py, crud_py, sync_path, nested_src) if not os.path.exists(p)]
+    if missing:
+        raise FileNotFoundError('missing: ' + ', '.join(missing))
+
+    shutil.copyfile(nested_src, nested_dst)   # 1. traversal implementation
+    patch_file(snapshot_py, SNAPSHOT_BLOCK)   # 2. build the staging tree
+    patch_file(sync_path, SYNC_BLOCK)         # 3. tear it down afterwards
+    patch_file(crud_py, CRUD_BLOCK)           # 4. ONLY NOW allow nested tasks
+
+    nested_ok = True
+    nested_detail = 'nested-dataset snapshots enabled (staging tree)'
+    print(f'OK: Installed nested-snapshot support → {nested_dst}')
+    print(f'OK: Patched snapshot.py, sync.py, crud.py → {cloud_dir}')
+except Exception as e:
+    nested_detail = f'not applied: {e}'
+    print(f'WARNING: Failed to apply nested-snapshot patch: {e}')
+    print('WARNING: Stock nesting guard remains; snapshot option stays unavailable')
+    print('WARNING: for nested datasets. Existing backups are unaffected.')
+
 patches = {
     'middlewared.rclone.remote.b2': {
         'ok': b2_ok,
@@ -280,6 +434,10 @@ patches = {
         'ok': restic_ok,
         'detail': 'patched on disk in overlay at boot' if restic_ok else 'restic.py not found or write failed',
     },
+    'middlewared.plugins.cloud.nested_snapshot': {
+        'ok': nested_ok,
+        'detail': nested_detail,
+    },
 }
 payload = {'patched_at': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()), 'patches': patches}
 tmp = status_path + '.tmp'

patch/patch_ui.py

@@ -19,6 +19,7 @@ Safe to run multiple times — a marker string detects an already-patched file.
 Exits 0 in all cases (warnings are printed to stdout and logged by apply.sh).
 """
 
+import contextlib
 import os
 import re
 import shutil
@@ -131,10 +132,8 @@ def main():
         os.replace(tmp, path)
     except OSError as exc:
         print(f"[truecloud-patch] ERROR: Could not write {path}: {exc}")
-        try:
+        with contextlib.suppress(OSError):
             os.unlink(tmp)
-        except OSError:
-            pass
         return
 
     print(f"[truecloud-patch] UI bundle patched ({count} replacement(s)): {path}")

patch/truecloud_nested.py

@@ -0,0 +1,322 @@
+"""Nested-dataset snapshot support for TrueCloud Backup / Cloud Sync.
+
+Why this exists
+---------------
+Stock TrueNAS refuses ``snapshot = true`` when the backup path contains child
+datasets::
+
+    This option is only available for datasets that have no further nesting
+
+That guard is *correct* and it is not laziness. ``plugins/cloud/snapshot.py``
+already takes a **recursive** ZFS snapshot, but it then points the backup tool
+at the *parent* dataset's ``.zfs/snapshot/<snap>/`` directory -- and ZFS does
+not expose child datasets through a parent's snapshot directory::
+
+    /mnt/Tap/.zfs/snapshot/<snap>/apps/   ->  0 entries   (children invisible)
+    /mnt/Tap/apps/lidarr/config/.zfs/snapshot/<snap>/  ->  the real data
+
+So without the guard, the backup tool would walk a near-empty tree, report
+SUCCESS, and upload almost nothing. A backup that lies about succeeding is the
+worst failure a backup system can have, so middleware refuses the config
+instead.
+
+This module implements the missing half: after the (already recursive) snapshot
+is taken, every descendant dataset's *own* ``.zfs/snapshot/<snap>`` directory is
+bind-mounted into a staging tree that mirrors the original layout. The backup
+tool is then pointed at the staging root, which is a complete, consistent,
+point-in-time view of the whole subtree.
+
+Cardinal safety rule
+--------------------
+**If the tree cannot be staged completely, fail loudly.** Never return a partial
+tree. Silently backing up an incomplete tree is precisely the failure this
+feature exists to prevent, and it would be worse than not having the feature.
+
+Notes
+-----
+* ZFS snapshots are immutable, so a plain ``mount --bind`` is inherently
+  read-only; no remount dance is needed.
+* Bind-mounting ``.zfs/snapshot/<snap>`` pins the snapshot, so ``zfs destroy``
+  of that snapshot returns EBUSY until we unmount. Stock ``restic_backup()``
+  deletes the snapshot in its ``finally``, which therefore logs one benign
+  "Error deleting snapshot ... busy" warning; :func:`cleanup_task` then unmounts
+  and deletes the snapshot for real. See ``patch/apply.sh``.
+* Staging roots live under a stable, per-task path so that the backup tool sees
+  the *same* path every run. Stock's ``.zfs/snapshot/<name>-<timestamp>/`` path
+  changes every run, which defeats restic's parent-snapshot detection; the
+  staging tree is an improvement on that.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import os
+import subprocess
+
+__all__ = [
+    "StagingError",
+    "STAGING_BASE",
+    "ACTIVE",
+    "staging_root_for",
+    "plan_staging",
+    "current_mounts_under",
+    "apply_plan",
+    "verify_staged",
+    "teardown",
+]
+
+#: Where staging trees are assembled. tmpfs; bind mounts consume no space.
+STAGING_BASE = "/run/truecloud-nested"
+
+#: staging_root -> zfs snapshot name ("pool/ds@snap"), for cleanup.
+ACTIVE: dict[str, str] = {}
+
+
+class StagingError(Exception):
+    """Staging could not produce a complete tree. The backup must not proceed."""
+
+
+def staging_root_for(name: str, base: str = STAGING_BASE) -> str:
+    """Stable staging root for a task name (e.g. ``cloud_backup-5``)."""
+    safe = "".join(c if (c.isalnum() or c in "-_.") else "_" for c in name) or "task"
+    return os.path.join(base, safe)
+
+
+def _depth(path: str) -> int:
+    return len([p for p in path.split("/") if p])
+
+
+def plan_staging(base_mountpoint, path, snapshot_name, datasets, staging_root,
+                 isdir=os.path.isdir):
+    """Compute the bind-mount plan for staging a nested tree. Pure function.
+
+    ``datasets`` is a list of dicts shaped like ``zfs.dataset.query`` results:
+    ``{"name": str, "properties": {"mountpoint": {"value": str},
+    "mounted": {"value": "yes"|"no"}}}``.
+
+    Returns ``(mounts, skipped)`` where ``mounts`` is an ordered list of
+    ``(source, target)`` pairs (parents before children) and ``skipped`` is a
+    list of ``(dataset_name, reason)``.
+
+    Raises StagingError if a descendant holds data we would silently omit.
+    """
+    def snapdir(mountpoint):
+        return os.path.join(mountpoint, ".zfs", "snapshot", snapshot_name)
+
+    # Root of the staging tree: the backup path as seen inside the base
+    # dataset's own snapshot.
+    rel = os.path.relpath(path, base_mountpoint)
+    root_src = snapdir(base_mountpoint)
+    if rel != ".":
+        root_src = os.path.join(root_src, rel)
+
+    mounts = [(root_src, staging_root)]
+    skipped = []
+
+    prefix = path.rstrip("/") + "/"
+    for ds in datasets:
+        props = ds.get("properties", {})
+        mp = props.get("mountpoint", {}).get("value", "")
+        name = ds.get("name", "?")
+
+        if not mp or mp in ("none", "legacy", "-"):
+            skipped.append((name, f"mountpoint is {mp or 'unset'}"))
+            continue
+        if not mp.startswith(prefix):
+            continue  # not a descendant of the backup path
+
+        mounted = props.get("mounted", {}).get("value", "yes")
+        if mounted == "no":
+            # An unmounted (e.g. locked/encrypted) dataset contributes nothing to
+            # the live tree either, so skipping matches stock semantics -- but it
+            # is a real gap and must be visible, never silent.
+            skipped.append((name, "dataset is not mounted (locked/encrypted?)"))
+            continue
+
+        src = snapdir(mp)
+        if not isdir(src):
+            # The recursive snapshot should have covered every descendant. If it
+            # did not, this dataset's data would be silently omitted. Refuse.
+            raise StagingError(
+                f"dataset {name!r} has no snapshot {snapshot_name!r} at {src!r}; "
+                f"refusing to back up an incomplete tree"
+            )
+
+        target = os.path.join(staging_root, os.path.relpath(mp, path))
+        mounts.append((src, target))
+
+    # Parents before children, so each mountpoint exists before we mount onto it.
+    mounts.sort(key=lambda m: _depth(m[1]))
+    return mounts, skipped
+
+
+def current_mounts_under(root, mounts_file="/proc/self/mounts"):
+    """Mountpoints at or under ``root``, deepest first. Used for teardown."""
+    found = []
+    try:
+        with open(mounts_file, encoding="utf-8") as fh:
+            for line in fh:
+                parts = line.split()
+                if len(parts) < 2:
+                    continue
+                mp = parts[1].replace("\\040", " ").replace("\\011", "\t")
+                if mp == root or mp.startswith(root.rstrip("/") + "/"):
+                    found.append(mp)
+    except OSError:
+        return []
+    found.sort(key=_depth, reverse=True)
+    return found
+
+
+def _run(cmd):
+    return subprocess.run(cmd, capture_output=True, text=True, check=False)
+
+
+def apply_plan(mounts, runner=_run):
+    """Execute the bind-mount plan. Blocking; call via ``run_in_thread``.
+
+    Raises StagingError on the first failure, after rolling back what was
+    mounted -- a half-built tree must never be handed to the backup tool.
+    """
+    if not mounts:
+        raise StagingError("empty staging plan")
+
+    staging_root = mounts[0][1]
+    done = []
+    try:
+        os.makedirs(staging_root, exist_ok=True)
+        for src, target in mounts:
+            if not os.path.isdir(target):
+                # Child mountpoint dirs come from the parent snapshot, which is
+                # read-only -- we cannot mkdir them. Only the root is ours.
+                raise StagingError(f"staging target {target!r} does not exist")
+            res = runner(["mount", "--bind", src, target])
+            if res.returncode != 0:
+                raise StagingError(
+                    f"bind-mount {src!r} -> {target!r} failed: "
+                    f"{(res.stderr or '').strip() or res.returncode}"
+                )
+            done.append(target)
+    except Exception:
+        for target in reversed(done):
+            runner(["umount", "-l", target])
+        with contextlib.suppress(OSError):
+            os.rmdir(staging_root)
+        raise
+    return staging_root
+
+
+def verify_staged(mounts, runner=_run, ismount=os.path.ismount, listdir=os.listdir):
+    """Assert the staged tree is real and complete. Raises StagingError if not.
+
+    This is the anti-regression guard: it is what stops this feature from ever
+    degrading back into the silently-empty backup that the stock validation
+    refuses to allow.
+    """
+    if not mounts:
+        raise StagingError("nothing was staged")
+
+    staging_root = mounts[0][1]
+    for _src, target in mounts:
+        if not ismount(target):
+            raise StagingError(f"staging target {target!r} is not a mountpoint")
+
+    try:
+        if not listdir(staging_root):
+            raise StagingError(f"staging root {staging_root!r} is empty")
+    except OSError as e:
+        raise StagingError(f"staging root {staging_root!r} unreadable: {e}") from e
+
+    return True
+
+
+def teardown(staging_root, runner=_run, mounts_file="/proc/self/mounts"):
+    """Unmount the staging tree (deepest first) and remove the root.
+
+    Idempotent, and does not depend on an in-memory plan -- so it also cleans up
+    leftovers from a crashed run.
+    """
+    errors = []
+    for mp in current_mounts_under(staging_root, mounts_file=mounts_file):
+        res = runner(["umount", mp])
+        if res.returncode != 0:
+            res = runner(["umount", "-l", mp])  # lazy: better than leaking
+            if res.returncode != 0:
+                errors.append(f"{mp}: {(res.stderr or '').strip()}")
+    with contextlib.suppress(OSError):
+        os.rmdir(staging_root)
+    return errors
+
+
+# ── async orchestration (middleware is duck-typed; no middlewared import) ─────
+
+
+async def stage_nested(middleware, path, snapshot, base_mountpoint, task_name, logger=None):
+    """Build a complete staging tree for `path` from the already-taken `snapshot`.
+
+    `snapshot` is a full ZFS snapshot name ("Tap@cloud_backup-5-2026...").
+    Returns the staging root to hand to the backup tool.
+
+    Raises StagingError if the tree cannot be staged completely -- the caller
+    must let that propagate so the backup fails instead of silently uploading a
+    partial tree.
+    """
+    snapshot_name = snapshot.split("@", 1)[1]
+    staging_root = staging_root_for(task_name)
+
+    # A previous run may have crashed mid-flight; never build on top of that.
+    await middleware.run_in_thread(teardown, staging_root)
+
+    datasets = await middleware.call("zfs.dataset.query", [["type", "=", "FILESYSTEM"]])
+
+    mounts, skipped = await middleware.run_in_thread(
+        plan_staging, base_mountpoint, path, snapshot_name, datasets, staging_root
+    )
+    if logger:
+        for name, reason in skipped:
+            logger.warning("truecloud-patch: not staging dataset %r: %s", name, reason)
+
+    await middleware.run_in_thread(apply_plan, mounts)
+    try:
+        await middleware.run_in_thread(verify_staged, mounts)
+    except Exception:
+        await middleware.run_in_thread(teardown, staging_root)
+        raise
+
+    ACTIVE[staging_root] = snapshot
+    if logger:
+        logger.info(
+            "truecloud-patch: staged %d dataset(s) from %s at %s",
+            len(mounts), snapshot, staging_root,
+        )
+    return staging_root
+
+
+async def cleanup_task(middleware, task_name, logger=None):
+    """Tear down a task's staging tree and delete the snapshot it pinned.
+
+    Safe to call unconditionally: a no-op when the task was never staged.
+
+    Stock `restic_backup()` deletes the snapshot in its own `finally`, which
+    fails with EBUSY while our bind mounts pin it (it logs a warning and moves
+    on). We unmount here and then delete the snapshot for real.
+    """
+    staging_root = staging_root_for(task_name)
+    snapshot = ACTIVE.pop(staging_root, None)
+
+    if snapshot is None and not os.path.isdir(staging_root):
+        return  # never staged; nothing to do
+
+    errors = await middleware.run_in_thread(teardown, staging_root)
+    if errors and logger:
+        for err in errors:
+            logger.warning("truecloud-patch: staging teardown: %s", err)
+
+    if snapshot is not None:
+        try:
+            await middleware.call("zfs.snapshot.delete", snapshot)
+        except Exception as e:  # noqa: BLE001 - cleanup must never mask the real error
+            if logger:
+                logger.warning(
+                    "truecloud-patch: could not delete snapshot %s: %r", snapshot, e
+                )