perf(cli): archive snapshot volumes with multi-threaded zstd in one container (#1061)

`abi stack snapshot create` archived each stateful volume with a separate
`alpine tar czf` (single-thread gzip) container, run sequentially. On a real
3.88GB TDB2 (fuseki) volume this took ~146s and produced a 1.06GB archive,
with the wall-clock dominated by single-thread gzip on the one large volume.

Two changes:

- Compress with multi-threaded zstd instead of gzip. A tiny helper image
  (alpine + zstd, ~6MB) is baked once via `ensure_snapshot_helper_image()`
  before the stack is stopped, so its one-time build never lands in the
  downtime window and there is no per-snapshot package-mirror dependency.
  Compression runs inside the container, so only the compressed bytes cross
  the docker VM->host boundary. Measured: ~146s -> ~23s on the fuseki volume,
  and the archive roughly halves (1.06GB -> 501MB).

- Archive every volume in a SINGLE helper container (`archive_volumes`)
  instead of one container per volume, paying container/image startup once.
  Volumes are mounted read-only at /from/<key> and streamed
  `tar cf - | zstd -T0 -3` to /to/<key>.tar.zst. `set -eo pipefail` so a
  failed tar aborts the run rather than being masked by zstd's exit code.

Volume archives are now `.tar.zst` (centralised in `volume_archive_name`);
`extract_volume` decompresses with `zstd -dc | tar xf -`. No backward-compat
shim is needed (no existing snapshots). Cross-volume parallelism was
deliberately not added: one 3.88GB volume is ~81% of the bytes, so parallel
archiving caps at ~1.2x, and a per-archive `-T0` already saturates all cores.
The clean `docker compose stop` is kept (Fuseki/TDB2 needs a clean checkpoint;
`docker pause` only yields a crash-consistent, in-flight-lossy image).

`storage/` intentionally stays on host-side gzip to preserve the audited
`_safe_extract`/`_assert_safe_members` guard and avoid a host-zstd dependency.

Tests: 56 pass; batched archive + per-volume extract verified byte-identical
end-to-end on real volumes.

Author: Dr0p42

libs/naas-abi-cli/naas_abi_cli/cli/snapshot_runtime.py

@@ -8,8 +8,8 @@
 What is captured:
   * the stateful Docker volumes (``postgres_data``, ``minio_data``,
     ``fuseki_data``, ``qdrant_storage``, ``redis_data``, ``rabbitmq_data``,
-    ``headscale_data``) -- copied cold as raw tarballs so each engine's on-disk
-    state is consistent;
+    ``headscale_data``) -- copied cold as zstd-compressed tarballs so each
+    engine's on-disk state is consistent;
   * the host ``storage/`` directory (the durable SQLite event log + local
     datastore files).
 
@@ -54,8 +54,19 @@
 MANIFEST_NAME = "manifest.json"
 STORAGE_DIRNAME = "storage"
 VOLUMES_DIRNAME = "volumes"
-# Tiny, ubiquitous image used purely to tar/untar volume contents.
+# Tiny, ubiquitous image used purely to reset (empty) volume contents.
 HELPER_IMAGE = "alpine:3.20"
+# Volume archives are tar + multi-threaded zstd. busybox alpine ships no zstd,
+# so we bake a tiny helper image once (alpine + zstd, ~6MB) instead of shelling
+# out to a package mirror on every snapshot. Compressing inside the container
+# keeps only the compressed bytes crossing the docker VM->host boundary, and
+# using all cores (-T0) archives the large TDB2 volume ~6x faster than the
+# single-thread gzip busybox tar would use.
+SNAPSHOT_HELPER_IMAGE = "abi-snapshot-helper:zstd1"
+SNAPSHOT_HELPER_DOCKERFILE = "FROM alpine:3.20\nRUN apk add --no-cache zstd\n"
+# zstd's default level; benchmarked as the best time/size point on TDB2 data.
+ZSTD_LEVEL = 3
+VOLUME_ARCHIVE_SUFFIX = ".tar.zst"
 # Files whose content we fingerprint so we can warn about drift on restore.
 TRACKED_CONFIG_FILES = ("config.local.yaml", ".env")
 MANIFEST_FORMAT_VERSION = 1
@@ -141,6 +152,11 @@ def volume_full_name(project: str, key: str) -> str:
     return f"{project}_{key}"
 
 
+def volume_archive_name(key: str) -> str:
+    """Filename for a volume's archive inside a snapshot's ``volumes/`` dir."""
+    return f"{key}{VOLUME_ARCHIVE_SUFFIX}"
+
+
 def select_prunable(manifests: list[SnapshotManifest], keep: int) -> list[str]:
     """Return ids of snapshots to delete, keeping the ``keep`` newest."""
     if keep < 0:
@@ -295,6 +311,41 @@ def run_docker(
         ) from error
 
 
+def ensure_snapshot_helper_image() -> None:
+    """Build the zstd-capable helper image if it is not already present.
+
+    Idempotent and cheap on the hot path: a present image short-circuits on the
+    ``docker image inspect``. Callers run this *before* stopping the stack so the
+    one-time build (alpine + zstd) never counts against snapshot downtime.
+    """
+    inspect = subprocess.run(
+        ["docker", "image", "inspect", SNAPSHOT_HELPER_IMAGE],
+        check=False,
+        text=True,
+        capture_output=True,
+    )
+    if inspect.returncode == 0:
+        return
+    try:
+        subprocess.run(
+            ["docker", "build", "-t", SNAPSHOT_HELPER_IMAGE, "-"],
+            input=SNAPSHOT_HELPER_DOCKERFILE,
+            check=True,
+            text=True,
+            capture_output=True,
+        )
+    except FileNotFoundError as error:
+        raise click.ClickException(
+            "Docker is not installed or not available in PATH."
+        ) from error
+    except subprocess.CalledProcessError as error:
+        detail = (error.stderr or "").strip() or f"exit code {error.returncode}"
+        raise click.ClickException(
+            f"Could not build the snapshot helper image "
+            f"'{SNAPSHOT_HELPER_IMAGE}': {detail}."
+        ) from error
+
+
 def compose_project_name() -> str:
     """Resolve the compose project name (the volume prefix).
 
@@ -352,29 +403,41 @@ def volume_exists(name: str) -> bool:
     )
 
 
-def archive_volume(volume: str, dest_dir: Path, key: str) -> None:
+def archive_volumes(volumes: dict[str, str], dest_dir: Path) -> None:
+    """Archive every volume in ``volumes`` in a single helper container.
+
+    ``volumes`` maps snapshot key -> full docker volume name. Each is mounted
+    read-only at ``/from/<key>`` and compressed to ``/to/<key>.tar.zst`` with a
+    multi-threaded zstd. Using one container instead of one-per-volume pays the
+    image/container startup cost once; the archives still run sequentially
+    inside it (a single zstd -T0 already saturates the cores, so concurrency
+    would only oversubscribe them). The stack is stopped, so the cold reads are
+    consistent. ``set -eo pipefail`` so a failed tar aborts the whole run rather
+    than being masked by zstd's exit code; create discards the partial snapshot.
+    Keys come from the hardcoded ``DATA_VOLUMES`` set, so they are safe to
+    interpolate into the shell script and mount paths. --mount long-form (not
+    -v) so host paths containing ':' stay unambiguous.
+    """
+    if not volumes:
+        return
     dest_dir.mkdir(parents=True, exist_ok=True)
-    # --mount long-form (not -v) so host paths containing ':' are unambiguous.
-    run_docker(
-        [
-            "run",
-            "--rm",
-            "--mount",
-            f"type=volume,source={volume},destination=/from,readonly",
+    args = ["run", "--rm"]
+    for key, full in volumes.items():
+        args += [
             "--mount",
-            f"type=bind,source={dest_dir.resolve()},destination=/to",
-            HELPER_IMAGE,
-            "tar",
-            "czf",
-            f"/to/{key}.tar.gz",
-            "-C",
-            "/from",
-            ".",
+            f"type=volume,source={full},destination=/from/{key},readonly",
         ]
+    args += ["--mount", f"type=bind,source={dest_dir.resolve()},destination=/to"]
+    steps = "; ".join(
+        f"tar cf - -C /from/{key} . | zstd -q -T0 -{ZSTD_LEVEL} -o /to/{volume_archive_name(key)}"
+        for key in volumes
     )
+    args += [SNAPSHOT_HELPER_IMAGE, "sh", "-c", f"set -eo pipefail; {steps}"]
+    run_docker(args)
 
 
 def extract_volume(volume: str, src_dir: Path, key: str) -> None:
+    archive = volume_archive_name(key)
     run_docker(
         [
             "run",
@@ -383,10 +446,10 @@ def extract_volume(volume: str, src_dir: Path, key: str) -> None:
             f"type=volume,source={volume},destination=/to",
             "--mount",
             f"type=bind,source={src_dir.resolve()},destination=/from,readonly",
-            HELPER_IMAGE,
+            SNAPSHOT_HELPER_IMAGE,
             "sh",
             "-c",
-            f"cd /to && tar xzf /from/{key}.tar.gz",
+            f"set -eo pipefail; zstd -dc /from/{archive} | tar xf - -C /to",
         ]
     )
 
@@ -491,7 +554,7 @@ def _snapshot_artifacts_ok(snap_dir: Path, manifest: SnapshotManifest) -> list[s
     """Return a list of missing-artifact messages for a snapshot dir (empty=ok)."""
     problems: list[str] = []
     for key in manifest.volumes:
-        tarball = snap_dir / VOLUMES_DIRNAME / f"{key}.tar.gz"
+        tarball = snap_dir / VOLUMES_DIRNAME / volume_archive_name(key)
         if not tarball.exists():
             problems.append(f"missing volume archive for '{key}' ({tarball.name})")
     if manifest.storage_included and not (snap_dir / "storage.tar.gz").exists():
@@ -639,6 +702,10 @@ def create_snapshot(
     if missing:
         emit(f"Note: skipping volumes not present on this stack: {', '.join(missing)}")
 
+    # Build the zstd helper now, while the stack is still up, so its one-time
+    # build never lands inside the stopped (downtime) window below.
+    ensure_snapshot_helper_image()
+
     snapshot_id = generate_snapshot_id(now, slug)
     snap_dir = snapshots_root(root) / snapshot_id
     if snap_dir.exists():
@@ -650,9 +717,11 @@ def create_snapshot(
     run_compose(["stop"])
     try:
         try:
-            for key in present:
-                emit(f"  - archiving volume {key}")
-                archive_volume(volume_full_name(project, key), volumes_dir, key)
+            emit(f"  - archiving {len(present)} volume(s): {', '.join(present)}")
+            archive_volumes(
+                {key: volume_full_name(project, key) for key in present},
+                volumes_dir,
+            )
 
             storage_dir = root / STORAGE_DIRNAME
             storage_included = storage_dir.exists()
@@ -736,6 +805,10 @@ def restore_snapshot(
         else:
             emit("No existing stack data found; skipping safety snapshot.")
 
+    # Build the zstd helper (if absent) before tearing the stack down so the
+    # one-time build stays outside the downtime window.
+    ensure_snapshot_helper_image()
+
     emit(f"Restoring snapshot {snapshot_id}...")
     run_compose(["down"])
     try: