Delete the snapshot tree atomically instead of 252 calls

delete_snapshot_tree removed the parent and every child snapshot individually.
On a real pool `zfs snapshot -r` creates one snapshot per descendant dataset --
252 on Tap -- so cleanup was 252 sequential middleware calls.

Slow, but the real problem is that it is not atomic: a job killed part-way
through the sweep leaves behind exactly the orphaned snapshots this function
exists to prevent.

zfs.snapshot.delete accepts {"recursive": True}, which destroys the parent and
all children in one call. Use that as the fast path and keep the name-by-name
sweep as the fallback -- it is still needed when the parent is already gone
(stock's finally can win the race once our mounts are released), which makes a
recursive delete fail while the children survive.

The test fake now emulates real `zfs destroy -r` semantics, so a test cannot pass
while the shipped code deletes only the parent.

76 tests, ruff and shellcheck clean.

CHANGELOG.md

@@ -165,6 +165,17 @@
   middlewared; there is now a regression test that executes apply.sh's own probe
   code against the real wrapped source.
 
+### Changed (production audit)
+
+- **`delete_snapshot_tree` now uses a single recursive delete.** It previously
+  removed the parent and each child snapshot one at a time — 252 sequential
+  middleware calls on a real pool. That is slow, but the real problem is that it
+  is **not atomic**: a run killed part-way through the sweep leaves exactly the
+  orphaned snapshots the function exists to prevent. It now issues one
+  `zfs.snapshot.delete(..., {"recursive": True})` and falls back to the
+  name-by-name sweep only when that fails (e.g. stock's `finally` already removed
+  the parent, which leaves the children behind).
+
 ### Refactored
 
 - Staging teardown had been copy-pasted into `uninstall.sh` and `recover.sh` —

patch/truecloud_nested.py

@@ -371,6 +371,20 @@ async def delete_snapshot_tree(middleware, snapshot, logger=None):
     """
     dataset = snapshot.partition("@")[0]
 
+    # Fast path: ONE recursive delete removes the parent and every child that
+    # `zfs snapshot -r` created (252 on a real pool). Deleting them individually
+    # also works, but it is neither cheap nor atomic -- a run killed part-way
+    # through 252 sequential deletes leaves exactly the orphans this function
+    # exists to prevent.
+    try:
+        await middleware.call("zfs.snapshot.delete", snapshot, {"recursive": True})
+        return
+    except Exception:  # noqa: BLE001 - fall through to the explicit sweep
+        pass
+
+    # The parent may already be gone -- stock's `finally` can win the race once
+    # our mounts are released -- which fails the recursive delete while the
+    # children survive. Sweep them by name.
     try:
         snaps = await middleware.call(
             "zfs.snapshot.query", [["name", "^", dataset]], {"select": ["name"]}

tests/test_truecloud_nested.py

@@ -209,9 +209,16 @@ class FakeMiddleware:
         if method == "zfs.snapshot.query":
             return [{"name": n} for n in self.snapshots]
         if method == "zfs.snapshot.delete":
-            if args[0] not in self.snapshots:
+            name = args[0]
+            opts = args[1] if len(args) > 1 else {}
+            if name not in self.snapshots:
                 raise RuntimeError("does not exist")
-            self.snapshots.remove(args[0])
+            if opts.get("recursive"):
+                # Real `zfs destroy -r` takes the parent and every child snapshot.
+                for n in snapshot_tree_names(name, list(self.snapshots)):
+                    self.snapshots.remove(n)
+            else:
+                self.snapshots.remove(name)
             return True
         raise AssertionError(f"unexpected call {method}")
 
@@ -233,24 +240,35 @@ class TestDeleteSnapshotTree:
         asyncio.run(delete_snapshot_tree(mw, "Tap@snap"))
         assert mw.snapshots == []
 
-    def test_survives_query_failure_by_deleting_at_least_the_parent(self):
+    def test_uses_a_single_recursive_delete_not_252_individual_ones(self):
+        # 252 sequential deletes are slow AND not atomic: a run killed part-way
+        # through leaves exactly the orphans this function exists to prevent.
+        mw = FakeMiddleware(["Tap@snap", "Tap/apps@snap", "Tap/apps/lidarr@snap"])
+        asyncio.run(delete_snapshot_tree(mw, "Tap@snap"))
+        assert mw.snapshots == []
+        deletes = [a for m, a in mw.calls if m == "zfs.snapshot.delete"]
+        assert len(deletes) == 1, "should be ONE recursive call, not one per snapshot"
+        assert deletes[0][1] == {"recursive": True}
+        assert not [m for m, _a in mw.calls if m == "zfs.snapshot.query"], (
+            "no enumeration needed on the fast path"
+        )
+
+    def test_survives_recursive_and_query_failure_by_deleting_the_parent(self):
         class Broken(FakeMiddleware):
             async def call(self, method, *args):
                 if method == "zfs.snapshot.query":
                     raise RuntimeError("boom")
+                if method == "zfs.snapshot.delete" and len(args) > 1:
+                    raise RuntimeError("recursive delete unavailable")
                 return await super().call(method, *args)
 
         mw = Broken(["Tap@snap"])
         asyncio.run(delete_snapshot_tree(mw, "Tap@snap"))
         assert mw.snapshots == []
 
-    def test_attempts_no_delete_when_the_tree_is_already_gone(self):
-        # A successful query returning nothing means there is nothing to do.
-        # Falling back to the parent here would log a spurious "does not exist"
-        # warning on every clean run.
+    def test_leaves_unrelated_snapshots_alone_when_the_tree_is_gone(self):
         mw = FakeMiddleware(["Tap@unrelated"])
         asyncio.run(delete_snapshot_tree(mw, "Tap@snap"))
-        assert [m for m, _a in mw.calls if m == "zfs.snapshot.delete"] == []
         assert mw.snapshots == ["Tap@unrelated"]