Release version 3.2.1

Four defects in the snapshot mode 3.2.0 introduced.

The resolver dropped the trailing separator get_storage_path puts on a
volume path, because os.path.abspath strips it. rsync reads "dir" as
"copy the directory" where "dir/" means "copy its contents", so every
snapshot generation landed at <volume>/files/_data/... while the live
path lands at <volume>/files/... . Restores read the live layout, and
--link-dest found nothing to match against the previous generation. The
e2e never caught it because its driver appended the separator by hand.

Snapshot teardown was fatal and masking. A busy `btrfs subvolume delete`
raised out of the finally, which skipped the generation stamp and the
compose handling on a run whose data was already complete, and replaced
whatever the body had raised. The leftover is reported instead; removing
it is a cleanup problem, not a reason to discard a good generation.

A volume created after the snapshot was taken aborted the whole run: the
volume list is enumerated inside the snapshot context, and nothing is
stopped in snapshot mode, so the host keeps creating volumes for the
duration of the copy. Such a volume is now copied live with a warning,
which is exactly what the pre-snapshot code did for it.

The snapshot pass compares by content again. 3.2.0 dropped --checksum
because a snapshot source cannot move, which is true, but the comparison
that matters is against --link-dest: a file that changed while keeping
its size and whole-second mtime was hard-linked stale out of the previous
generation, and the single snapshot pass had no authoritative pass to
repair it the way the live path does. It is still one pass against two.

--hard-restart-projects is refused alongside --snapshot, the same way
--shutdown already is: the flag exists for stacks whose database cannot
be backed up hot, which is what a snapshot removes.

Tests: the trailing separator, both teardown behaviours, the new refusal,
and app.main driving the snapshot branch - the caller that runs in
production, which no test had exercised and where the layout defect
therefore stayed invisible. The e2e driver now feeds the resolver the
string shape get_storage_path really produces.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

.claude/.gitignore

@@ -0,0 +1,3 @@
+*
+!.gitignore
+!settings.json

.claude/settings.json

@@ -0,0 +1,10 @@
+{
+  "permissions": {
+    "ask": [
+      "Edit(CHANGELOG.md)",
+      "Write(CHANGELOG.md)",
+      "Edit(pyproject.toml)",
+      "Write(pyproject.toml)"
+    ]
+  }
+}

.github/workflows/ci.yml

@@ -29,7 +29,16 @@ jobs:
           docker version
           docker info
 
+      - name: Provide zfs so the snapshot suite covers every filesystem
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y --no-install-recommends zfsutils-linux
+          sudo modprobe zfs
+          zpool version
+
       - name: Run all tests via Makefile
+        env:
+          E2E_REQUIRE_FILESYSTEMS: "btrfs ext4 zfs"
         run: |
           make test
 

.gitignore

@@ -3,3 +3,4 @@ artifacts/
 *.egg-info
 dist/
 build/
+.mcp.json

CHANGELOG.md

@@ -1,5 +1,179 @@
 # Changelog
 
+## [3.2.1] - 2026-07-31
+
+- Backup: the snapshot resolver keeps the trailing separator *get_storage_path*
+  puts on a volume path — *os.path.abspath* stripped it. rsync reads *dir* as
+  "copy the directory" where *dir/* means "copy its contents", so every snapshot
+  generation landed at *<volume>/files/_data/...* while the live path lands at
+  *<volume>/files/...*. Restores read the live layout, and *--link-dest* had
+  nothing to match against the previous generation.
+- Backup: snapshot teardown no longer fails a completed run. A busy
+  *btrfs subvolume delete* raised out of the *finally*, skipping the generation
+  stamp and the compose handling on a run whose data was already copied, and
+  masking whatever the body had raised. The leftover is reported instead.
+- Backup: a volume created after the snapshot was taken is copied live with a
+  warning instead of aborting the run. Nothing is stopped in snapshot mode, so
+  the host keeps creating volumes for the whole copy.
+- Backup: the snapshot pass compares by content (*--checksum*) again. 3.2.0
+  dropped it because a snapshot source cannot move, which is true, but the
+  comparison that matters is against *--link-dest*: a file that changed while
+  keeping its size and whole-second mtime was hard-linked stale out of the
+  previous generation, and the single pass had no authoritative pass to repair
+  it. Still one pass where the live path takes two.
+- Backup: *--hard-restart-projects* is refused alongside *--snapshot*, like
+  *--shutdown* already is. It exists for stacks whose database cannot be backed
+  up hot, which is what a snapshot removes.
+- Tests: the trailing separator, both teardown behaviours, the new refusal, and
+  *app.main* driving the snapshot branch — the caller that runs in production,
+  which no test had exercised, which is why the layout defect shipped.
+
+## [3.2.0] - 2026-07-31
+
+- Backup: *--snapshot {btrfs,zfs}* with *--snapshot-subject* captures every
+  volume from one atomic filesystem snapshot. The subject — the btrfs subvolume
+  or zfs dataset holding the docker volumes, e.g. */var/lib/docker* — is frozen
+  once per run, so a generation shares a single point in time and no container
+  is stopped. Restoring such a copy is an ordinary crash recovery, which every
+  supported engine performs at startup. This is the mode 3.1.4 pointed to for
+  volumes where two rsync passes over a live tree stop being affordable.
+- Backup: snapshot passes copy once and drop *--checksum*. Verification exists
+  because a hot pass writes its destination from a moving source; a snapshot
+  source cannot move, so size and mtime cannot race and the second full read is
+  pure cost.
+- Backup: an unsupported filesystem or unknown kind fails with *SnapshotError*.
+  The kind is stated, not probed — an inconclusive probe would fall back to a
+  live copy and hand out the torn backup the mode prevents. *--shutdown* is
+  rejected alongside *--snapshot* rather than ignored, since nothing is stopped.
+- Refactor: *backup/app.py* splits into *layout.py*, *policy.py* and *dumps.py*
+  along the lines it already had; 276 lines down to 124.
+- Tests: unit coverage for snapshot, layout, policy, volume and CLI validation.
+  E2E cases drive real btrfs, zfs and ext4 on loop devices, one proving the
+  loud refusal; another writes a MariaDB across the snapshot and requires the
+  restored server to recover on its own with every committed row and none of
+  the later ones. CI installs zfs and sets *E2E_REQUIRE_FILESYSTEMS*, so a
+  missing kernel module fails the build instead of skipping a filesystem.
+
+## [3.1.4] - 2026-07-31
+
+- Backup: the post-stop volume pass now compares by content (*--checksum*), so
+  it can no longer skip a file the hot pass had copied from a live source. Each
+  volume is rsynced twice into the same destination — once with the container
+  running, once after it is stopped — and the second pass used rsync's quick
+  check (size plus whole-second mtime). A pre-allocated 16 MiB WAL segment never
+  changes size, so when its last pre-stop write and postgres' shutdown
+  checkpoint fell in the same whole second, the cold pass skipped it while still
+  replacing *global/pg_control*, whose previous write was seconds earlier. The
+  generation then paired a post-shutdown *pg_control* with a WAL segment still
+  zeroed at the recorded checkpoint LSN, and restoring it crash-looped postgres
+  with "invalid record length … expected at least 24, got 0" followed by "PANIC:
+  could not locate a valid checkpoint record". *backup_volume* takes
+  *authoritative* as a required keyword rather than an optional flag, so each of
+  the four call sites states which pass it is; the hot passes keep the quick
+  check. *--ignore-times* was rejected because it destroys the *--link-dest*
+  hardlink dedup (measured 20/20 to 0/20, generation size doubled), and
+  *--modify-window=-1* because it makes correctness depend on the destination
+  filesystem preserving sub-second mtimes, which degrades silently on NFS or
+  ext3.
+- Cost: the quick check scales with file count, *--checksum* with bytes read on
+  both sides, and it runs while the container is stopped. The extra stop time
+  stays under a minute up to roughly 4 GB on spinning disk, 15 GB on a SATA SSD
+  and 60 GB on NVMe; at 1 TB it is 17 minutes on NVMe and over three hours on
+  spinning disk. No size threshold is built in, since a guessed one would drop
+  the guarantee exactly where an unrestorable backup costs most — volumes at
+  that scale want filesystem snapshots or *pg_basebackup* rather than two rsync
+  passes over a live tree.
+
+## [3.1.3] - 2026-07-20
+
+- Restore: the postgres dump replay now runs under *--single-transaction*,
+  so a concurrent writer on a live database can no longer interleave a row
+  between the replay's table re-create and its *COPY* and trip a "duplicate
+  key value violates unique constraint" abort under ON_ERROR_STOP. This is
+  the discourse restore-drill race (*mini_scheduler* upserting
+  *scheduler_stats(id=1)* mid-restore) that failed the whole restore. The
+  *--empty* pre-clean stays multi-statement (*\gexec*, one DROP per
+  statement) because a single DROP transaction exhausts
+  *max_locks_per_transaction* on large schemas (e.g. gitlab).
+- Refactor: the *--empty* pre-clean SQL moves out of the inline Python
+  string into *src/baudolo/restore/db/empty_preclean.sql* (loaded via
+  *dirname(__file__)*, declared as package-data so it ships in the wheel).
+- Tests: a unit test guards the single-transaction / multi-statement split
+  (replay carries *--single-transaction*, pre-clean does not); a new e2e
+  reproduces the live-writer race and asserts the restore survives it.
+
+## [3.1.2] - 2026-07-18
+
+- Restore: the postgres *--empty* pre-clean also drops user-owned text
+  search configurations and dictionaries (*pg_ts_config*, *pg_ts_dict*),
+  so a schema shipping a custom dictionary (e.g. taiga's
+  *english_stem_nostop*) no longer aborts the replay with "duplicate key
+  value violates unique constraint pg_ts_dict_dictname_index" under
+  ON_ERROR_STOP.
+- Tests: the string-assertion unit test for the pre-clean SQL is replaced
+  by real scenario data in the e2e: the seeded schema contains an
+  overloaded *f()/f(int)* pair and the nostop dictionary plus
+  configuration, and the restored database is queried to prove each
+  survives the backup, pre-clean and replay cycle exactly once.
+
+## [3.1.1] - 2026-07-17
+
+- Restore: the postgres *--empty* pre-clean drops functions and procedures
+  by their identity signature (*pg_get_function_identity_arguments*), so a
+  schema that overloads a function name (e.g. discourse) no longer aborts
+  the replay with "function name is not unique" under ON_ERROR_STOP.
+  Identifier quoting moves from the outer DROP format into each object
+  branch, since the *name(args)* compound must not be quoted as a whole; a
+  unit test pins the per-branch *%I* quoting so future branches cannot
+  regress unquoted.
+
+## [3.1.0] - 2026-07-15
+
+- Restore: the postgres *--empty* pre-clean emits one DROP per object and
+  runs them via *\gexec* instead of a single DO-block, so large schemas
+  (e.g. gitlab) no longer exhaust *max_locks_per_transaction* in one
+  transaction. It also drops user-owned non-public schemas, so dumps that
+  CREATE SCHEMA (e.g. discourse's *discourse_functions*) no longer abort
+  on the already-existing schema under ON_ERROR_STOP.
+- Backup: *--database-containers* and *--images-no-stop-required* are now
+  optional and default to an empty list, so a pure file backup needs no
+  dummy arguments; an empty stop whitelist keeps the conservative
+  stop-all behavior.
+- Tests: new e2e test restores *--empty* against a fully populated
+  database containing a non-public schema and every dropped object class.
+  *make test* runs the three suites concurrently after a single
+  clean+build; *E2E_TEST_PATTERN* runs an e2e subset.
+
+## [3.0.0] - 2026-07-12
+
+- Backup: *--images-no-stop-required* and *--images-no-backup-required* now
+  match a container's exact *.Config.Image* (full *repo:tag*, registry
+  prefix included) instead of a substring, so a near-miss image name no
+  longer flips the stop/skip decision. Callers must pass exact image
+  references. **Breaking.**
+- Backup: renamed *--hard-compose-restart* to *--hard-restart-projects*
+  (its value stays a list of compose project dir names). **Breaking:** the
+  old flag name is removed.
+
+## [2.0.0] - 2026-07-12
+
+- Backup: renamed *--docker-compose-hard-restart-required* to
+  *--hard-compose-restart* and changed its default from *["mailu"]* to *[]*
+  (nargs="*"). The compose down/up is now opt-in: compose hosts pass
+  *mailu* explicitly, while swarm hosts pass nothing, since there the dir is
+  a stack whose overlay network collides with *compose up*. **Breaking:** the
+  old flag name is removed and the implicit mailu default is gone.
+- Backup: *--backups-dir* is now required (no */var/lib/backup/* default) so
+  a run can never silently target the wrong backup root. **Breaking.**
+- Restore: volume files are rsynced directly into the target volume's
+  mountpoint (resolved via *docker volume inspect*), mirroring the backup
+  path; the *alpine-rsync* helper image and the *--rsync-image* flag are
+  gone. The caller needs write access to the docker volume root (root on the
+  host, baudolo's normal privilege). **Breaking:** the restore *files*
+  subcommand no longer accepts *--rsync-image*.
+- Tests: the e2e suite tracks *postgres:alpine* (18+, mounted at
+  */var/lib/postgresql*) and *mariadb:latest* from a single source of truth.
+
 ## [1.8.1] - 2026-07-12
 
 - Restore: the postgres empty mode also drops user-owned collations in

Makefile

@@ -1,5 +1,6 @@
-.PHONY: install build \
+.PHONY: install build clean \
-		test-e2e test test-unit test-integration
+		test test-unit test-integration test-e2e \
+		test-unit-run test-integration-run test-e2e-run
 
 # Default python if no venv is active
 PY_DEFAULT ?= python3
@@ -33,25 +34,32 @@ build:
 clean:
 	git clean -fdX .
 
-# ------------------------------------------------------------
+# clean + build run once and in order, then the three suites run concurrently
-# Run E2E tests inside the container (Docker socket required)
+# via -j3; the *-run targets carry no clean/build prereq so the sub-make cannot
-# ------------------------------------------------------------
+# race a second clean against build.
-# E2E via isolated Docker-in-Docker (DinD)
+test:
-# - depends on local image build
+	@$(MAKE) clean
-# - starts a DinD daemon container on a dedicated network
+	@$(MAKE) build
-# - loads the freshly built image into DinD
+	@$(MAKE) -j3 test-unit-run test-integration-run test-e2e-run
-# - runs the unittest suite inside a container that talks to DinD via DOCKER_HOST
-test-e2e: clean build
-	@bash scripts/test-e2e.sh
 
-test: test-unit test-integration test-e2e
+test-unit: clean build test-unit-run
 
-test-unit: clean build
+test-integration: clean build test-integration-run
+
+test-e2e: clean build test-e2e-run
+
+test-unit-run:
 	@echo ">> Running unit tests"
 	@docker run --rm -t $(IMAGE) \
 	  bash -lc 'python -m unittest discover -t . -s tests/unit -p "test_*.py" -v'
 
-test-integration: clean build
+test-integration-run:
 	@echo ">> Running integration tests"
 	@docker run --rm -t $(IMAGE) \
 	  bash -lc 'python -m unittest discover -t . -s tests/integration -p "test_*.py" -v'
+
+# E2E via isolated Docker-in-Docker (DinD): starts a DinD daemon on a dedicated
+# network, loads the freshly built image into it, and runs tests/e2e inside a
+# container that talks to DinD via DOCKER_HOST.
+test-e2e-run:
+	@bash scripts/test-e2e.sh

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "backup-docker-to-local"
-version = "1.8.1"
+version = "3.2.1"
 description = "Backup Docker volumes to local with rsync and optional DB dumps."
 readme = "README.md"
 requires-python = ">=3.9"
@@ -27,3 +27,6 @@ package-dir = { "" = "src" }
 [tool.setuptools.packages.find]
 where = ["src"]
 exclude = ["tests*"]
+
+[tool.setuptools.package-data]
+"baudolo.restore.db" = ["*.sql"]

scripts/test-e2e.sh

@@ -27,7 +27,6 @@ dind() { docker exec "${DIND}" docker "$@"; }
 dind_stdin() { docker exec -i "${DIND}" docker "$@"; }
 
 IMG="${E2E_IMAGE:-baudolo:local}"
-RSYNC_IMG="${E2E_RSYNC_IMAGE:-ghcr.io/kevinveenbirkenbach/alpine-rsync}"
 
 READY_TIMEOUT_SECONDS="${E2E_READY_TIMEOUT_SECONDS:-120}"
 ARTIFACTS_DIR="${E2E_ARTIFACTS_DIR:-./artifacts}"
@@ -38,6 +37,9 @@ KEEP_ON_FAIL="${E2E_KEEP_ON_FAIL:-0}"
 KEEP_VOLUMES="${E2E_KEEP_VOLUMES:-0}"
 DEBUG_SHELL="${E2E_DEBUG_SHELL:-0}"
 
+# Override to run a subset, e.g. E2E_TEST_PATTERN=test_e2e_postgres_empty_drop_hard.py
+TEST_PATTERN="${E2E_TEST_PATTERN:-test_*.py}"
+
 FAILED=0
 TS="$(date +%Y%m%d%H%M%S)"
 
@@ -164,10 +166,6 @@ for i in $(seq 1 "${READY_TIMEOUT_SECONDS}"); do
   fi
 done
 
-log "Pre-pulling helper images in DinD..."
-log " - Pulling: ${RSYNC_IMG}"
-dind pull "${RSYNC_IMG}"
-
 log "Ensuring alpine exists in DinD (for debug helpers)"
 dind pull alpine:3.20 >/dev/null
 
@@ -181,8 +179,7 @@ if [ "${DEBUG_SHELL}" = "1" ]; then
   docker run --rm -it \
     --network "${NET}" \
     -e DOCKER_HOST="${DIND_HOST_IN_NET}" \
-    -e E2E_RSYNC_IMAGE="${RSYNC_IMG}" \
+    -v "${DIND_VOL}:/var/lib/docker" \
-    -v "${DIND_VOL}:/var/lib/docker:ro" \
     -v "${E2E_TMP_VOL}:/tmp" \
     "${IMG}" \
     bash -lc '
@@ -200,8 +197,9 @@ else
   docker run --rm \
     --network "${NET}" \
     -e DOCKER_HOST="${DIND_HOST_IN_NET}" \
-    -e E2E_RSYNC_IMAGE="${RSYNC_IMG}" \
+    -e E2E_TEST_PATTERN="${TEST_PATTERN}" \
-    -v "${DIND_VOL}:/var/lib/docker:ro" \
+    -e E2E_REQUIRE_FILESYSTEMS="${E2E_REQUIRE_FILESYSTEMS:-}" \
+    -v "${DIND_VOL}:/var/lib/docker" \
     -v "${E2E_TMP_VOL}:/tmp" \
     "${IMG}" \
     bash -lc '
@@ -216,7 +214,7 @@ else
         cat /proc/sys/kernel/random/uuid > /etc/machine-id
     fi
 
-    python -m unittest discover -t . -s tests/e2e -p "test_*.py" -v -f
+    python -m unittest discover -t . -s tests/e2e -p "${E2E_TEST_PATTERN}" -v -f
     '
   rc=$?
 fi

src/baudolo/backup/app.py

@@ -1,171 +1,29 @@
+"""Back up every Docker volume of a host into a timestamped generation."""
+
 from __future__ import annotations
 
 import os
-import pathlib
+from contextlib import ExitStack
-import sys
 from datetime import datetime
 
-import pandas
-from dirval import create_stamp_file
-from pandas.errors import EmptyDataError
-
 from .cli import parse_args
 from .compose import handle_docker_compose_services
-from .db import backup_database
 from .docker import (
     change_containers_status,
     containers_using_volume,
     docker_volume_names,
     filter_stoppable,
-    get_image_info,
-    has_image,
-    is_swarm_task,
 )
-from .shell import execute_shell_command
+from .dumps import backup_dumps_for_volume, load_databases_df
-from .volume import backup_volume
+from .layout import (
-
+    create_version_directory,
-
+    create_volume_directory,
-def get_machine_id() -> str:
+    get_machine_id,
-    return execute_shell_command("sha256sum /etc/machine-id")[0][0:64]
+    stamp_directory,
-
-
-def stamp_directory(version_dir: str) -> None:
-    """
-    Use dirval as a Python library to stamp the directory (no CLI dependency).
-    """
-    create_stamp_file(version_dir)
-
-
-def create_version_directory(versions_dir: str, backup_time: str) -> str:
-    version_dir = os.path.join(versions_dir, backup_time)
-    pathlib.Path(version_dir).mkdir(parents=True, exist_ok=True)
-    return version_dir
-
-
-def create_volume_directory(version_dir: str, volume_name: str) -> str:
-    path = os.path.join(version_dir, volume_name)
-    pathlib.Path(path).mkdir(parents=True, exist_ok=True)
-    return path
-
-
-def is_image_ignored(container: str, images_no_backup_required: list[str]) -> bool:
-    if not images_no_backup_required:
-        return False
-    img = get_image_info(container)
-    return any(pat in img for pat in images_no_backup_required)
-
-
-def volume_is_fully_ignored(
-    containers: list[str], images_no_backup_required: list[str]
-) -> bool:
-    """
-    Skip file backup only if all containers linked to the volume are ignored.
-    """
-    if not containers:
-        return False
-    return all(is_image_ignored(c, images_no_backup_required) for c in containers)
-
-
-def requires_stop(containers: list[str], images_no_stop_required: list[str]) -> bool:
-    """
-    Stop is required if ANY stoppable container image is NOT in the
-    whitelist patterns. Swarm task containers never count: baudolo must
-    not cycle them (see docker.is_swarm_task).
-    """
-    for c in containers:
-        if is_swarm_task(c):
-            continue
-        img = get_image_info(c)
-        if not any(pat in img for pat in images_no_stop_required):
-            return True
-    return False
-
-
-def backup_mariadb_or_postgres(
-    *,
-    container: str,
-    volume_dir: str,
-    databases_df: "pandas.DataFrame",
-    database_containers: list[str],
-) -> tuple[bool, bool]:
-    """
-    Returns (is_db_container, dumped_any)
-    """
-    for img in ["mariadb", "postgres"]:
-        if has_image(container, img):
-            dumped = backup_database(
-                container=container,
-                volume_dir=volume_dir,
-                db_type=img,
-                databases_df=databases_df,
-                database_containers=database_containers,
 )
-            return True, dumped
+from .policy import requires_stop, volume_is_fully_ignored
-    return False, False
+from .snapshot import volume_snapshot
-
+from .volume import backup_volume, get_storage_path
-
-def _empty_databases_df() -> "pandas.DataFrame":
-    """
-    Create an empty DataFrame with the expected schema for databases.csv.
-
-    This allows the backup to continue without DB dumps when the CSV is missing
-    or empty (pandas EmptyDataError).
-    """
-    return pandas.DataFrame(columns=["instance", "database", "username", "password"])
-
-
-def _load_databases_df(csv_path: str) -> "pandas.DataFrame":
-    """
-    Load databases.csv robustly.
-
-    - Missing file     -> warn, continue with empty df
-    - Empty file       -> warn, continue with empty df
-    - Valid CSV        -> return dataframe
-    """
-    try:
-        return pandas.read_csv(csv_path, sep=";", keep_default_na=False, dtype=str)
-    except FileNotFoundError:
-        print(
-            f"WARNING: databases.csv not found: {csv_path}. Continuing without database dumps.",
-            file=sys.stderr,
-            flush=True,
-        )
-        return _empty_databases_df()
-    except EmptyDataError:
-        print(
-            f"WARNING: databases.csv exists but is empty: {csv_path}. Continuing without database dumps.",
-            file=sys.stderr,
-            flush=True,
-        )
-        return _empty_databases_df()
-
-
-def _backup_dumps_for_volume(
-    *,
-    containers: list[str],
-    vol_dir: str,
-    databases_df: "pandas.DataFrame",
-    database_containers: list[str],
-) -> tuple[bool, bool]:
-    """
-    Returns (found_db_container, dumped_any)
-    """
-    found_db = False
-    dumped_any = False
-
-    for c in containers:
-        is_db, dumped = backup_mariadb_or_postgres(
-            container=c,
-            volume_dir=vol_dir,
-            databases_df=databases_df,
-            database_containers=database_containers,
-        )
-        if is_db:
-            found_db = True
-        if dumped:
-            dumped_any = True
-
-    return found_db, dumped_any
 
 
 def main() -> int:
@@ -183,15 +41,21 @@ def main() -> int:
     #
     # Robust behavior:
     # - if the file is missing or empty, we continue without DB dumps.
-    databases_df = _load_databases_df(args.databases_csv)
+    databases_df = load_databases_df(args.databases_csv)
 
     print("💾 Start volume backups...", flush=True)
 
+    with ExitStack() as stack:
+        resolve_source = None
+        if args.snapshot:
+            resolve_source = stack.enter_context(
+                volume_snapshot(args.snapshot, args.snapshot_subject, backup_time)
+            )
+
         for volume_name in docker_volume_names():
             print(f"Start backup routine for volume: {volume_name}", flush=True)
             containers = containers_using_volume(volume_name)
 
-        # EARLY SKIP: if all linked containers are ignored, do not create any dirs
             if volume_is_fully_ignored(containers, args.images_no_backup_required):
                 print(
                     f"Skipping volume '{volume_name}' entirely (all linked containers are ignored).",
@@ -201,14 +65,13 @@ def main() -> int:
 
             vol_dir = create_volume_directory(version_dir, volume_name)
 
-        found_db, dumped_any = _backup_dumps_for_volume(
+            found_db, dumped_any = backup_dumps_for_volume(
                 containers=containers,
                 vol_dir=vol_dir,
                 databases_df=databases_df,
                 database_containers=args.database_containers,
             )
 
-        # dump-only-sql logic:
             if args.dump_only_sql:
                 if found_db:
                     if not dumped_any:
@@ -217,38 +80,54 @@ def main() -> int:
                             "Falling back to file backup.",
                             flush=True,
                         )
-                    # fall through to file backup below
                     else:
-                    # DB volume successfully dumped -> skip file backup
                         continue
-            # Non-DB volume -> always do file backup (fall through)
+
+            live_source = get_storage_path(volume_name)
+
+            def copy(*, authoritative: bool, source: str = live_source) -> None:
+                backup_volume(
+                    versions_dir,
+                    volume_name,
+                    vol_dir,
+                    authoritative=authoritative,
+                    source=source,
+                )
+
+            if resolve_source is not None:
+                snapshot_source = resolve_source(live_source)
+                if os.path.isdir(snapshot_source):
+                    copy(authoritative=True, source=snapshot_source)
+                else:
+                    print(
+                        f"WARNING: volume '{volume_name}' is not in the snapshot "
+                        "(created after it was taken); copying it live instead.",
+                        flush=True,
+                    )
+                    copy(authoritative=False)
+                continue
 
             if args.everything:
-            # "everything": always do pre-rsync, then stop + rsync again
                 stoppable = filter_stoppable(containers)
-            backup_volume(versions_dir, volume_name, vol_dir)
+                copy(authoritative=False)
                 change_containers_status(stoppable, "stop")
-            backup_volume(versions_dir, volume_name, vol_dir)
+                copy(authoritative=True)
                 if not args.shutdown:
                     change_containers_status(stoppable, "start")
                 continue
 
-        # default: rsync, and if needed stop + rsync
+            copy(authoritative=False)
-        backup_volume(versions_dir, volume_name, vol_dir)
             if requires_stop(containers, args.images_no_stop_required):
                 stoppable = filter_stoppable(containers)
                 change_containers_status(stoppable, "stop")
-            backup_volume(versions_dir, volume_name, vol_dir)
+                copy(authoritative=True)
                 if not args.shutdown:
                     change_containers_status(stoppable, "start")
 
-    # Stamp the backup version directory using dirval (python lib)
     stamp_directory(version_dir)
     print("Finished volume backups.", flush=True)
 
     print("Handling Docker Compose services...", flush=True)
-    handle_docker_compose_services(
+    handle_docker_compose_services(args.compose_dir, args.hard_restart_projects)
-        args.compose_dir, args.docker_compose_hard_restart_required
-    )
 
     return 0

src/baudolo/backup/cli.py

@@ -17,10 +17,10 @@ def parse_args() -> argparse.Namespace:
         help="Path to the parent directory containing docker-compose setups",
     )
     p.add_argument(
-        "--docker-compose-hard-restart-required",
+        "--hard-restart-projects",
-        nargs="+",
+        nargs="*",
-        default=["mailu"],
+        default=[],
-        help="Compose dir names that require 'docker-compose down && up -d' (default: mailu)",
+        help="Compose dir names that require 'docker-compose down && up -d' (default: none; pass e.g. 'mailu' under compose where the DB cannot be backed up hot)",
     )
 
     p.add_argument(
@@ -35,27 +35,37 @@ def parse_args() -> argparse.Namespace:
     )
     p.add_argument(
         "--backups-dir",
-        default="/var/lib/backup/",
+        required=True,
-        help="Backup root directory (default: /var/lib/backup/)",
+        help="Backup root directory (e.g. /var/lib/backup/)",
+    )
+
+    p.add_argument(
+        "--snapshot",
+        choices=["btrfs", "zfs"],
+        help="Capture every volume from one atomic filesystem snapshot instead of copying the live tree. Containers are not stopped, and the copy is a single pass. Requires --snapshot-subject. Omit to keep the live two-pass copy.",
+    )
+    p.add_argument(
+        "--snapshot-subject",
+        help="Btrfs subvolume or zfs dataset mountpoint holding the docker volumes, e.g. /var/lib/docker. Required with --snapshot.",
     )
 
     p.add_argument(
         "--database-containers",
         nargs="+",
-        required=True,
+        default=[],
         help="Container names treated as special instances for database backups",
     )
     p.add_argument(
         "--images-no-stop-required",
         nargs="+",
-        required=True,
+        default=[],
-        help="Image name patterns for which containers should not be stopped during file backup",
+        help="Exact image references (repo:tag, incl. any registry prefix) whose containers must not be stopped during file backup",
     )
     p.add_argument(
         "--images-no-backup-required",
         nargs="+",
         default=[],
-        help="Image name patterns for which no backup should be performed",
+        help="Exact image references (repo:tag, incl. any registry prefix) for which no backup should be performed",
     )
 
     p.add_argument(
@@ -79,4 +89,14 @@ def parse_args() -> argparse.Namespace:
             "If a DB dump cannot be produced, baudolo falls back to a file backup."
         ),
     )
-    return p.parse_args()
+    args = p.parse_args()
+    if bool(args.snapshot) != bool(args.snapshot_subject):
+        p.error("--snapshot and --snapshot-subject must be given together")
+    if args.snapshot and args.shutdown:
+        p.error("--shutdown is meaningless with --snapshot: containers are never stopped")
+    if args.snapshot and args.hard_restart_projects:
+        p.error(
+            "--hard-restart-projects is meaningless with --snapshot: the flag exists "
+            "for stacks whose database cannot be backed up hot, which a snapshot solves"
+        )
+    return args

src/baudolo/backup/dumps.py

@@ -0,0 +1,98 @@
+"""Database dumps taken before a volume's files are copied."""
+
+from __future__ import annotations
+
+import sys
+
+import pandas
+from pandas.errors import EmptyDataError
+
+from .db import backup_database
+from .docker import has_image
+
+
+def backup_mariadb_or_postgres(
+    *,
+    container: str,
+    volume_dir: str,
+    databases_df: "pandas.DataFrame",
+    database_containers: list[str],
+) -> tuple[bool, bool]:
+    """
+    Returns (is_db_container, dumped_any)
+    """
+    for img in ["mariadb", "postgres"]:
+        if has_image(container, img):
+            dumped = backup_database(
+                container=container,
+                volume_dir=volume_dir,
+                db_type=img,
+                databases_df=databases_df,
+                database_containers=database_containers,
+            )
+            return True, dumped
+    return False, False
+
+
+def _empty_databases_df() -> "pandas.DataFrame":
+    """
+    Create an empty DataFrame with the expected schema for databases.csv.
+
+    This allows the backup to continue without DB dumps when the CSV is missing
+    or empty (pandas EmptyDataError).
+    """
+    return pandas.DataFrame(columns=["instance", "database", "username", "password"])
+
+
+def load_databases_df(csv_path: str) -> "pandas.DataFrame":
+    """
+    Load databases.csv robustly.
+
+    - Missing file     -> warn, continue with empty df
+    - Empty file       -> warn, continue with empty df
+    - Valid CSV        -> return dataframe
+    """
+    try:
+        return pandas.read_csv(csv_path, sep=";", keep_default_na=False, dtype=str)
+    except FileNotFoundError:
+        print(
+            f"WARNING: databases.csv not found: {csv_path}. Continuing without database dumps.",
+            file=sys.stderr,
+            flush=True,
+        )
+        return _empty_databases_df()
+    except EmptyDataError:
+        print(
+            f"WARNING: databases.csv exists but is empty: {csv_path}. Continuing without database dumps.",
+            file=sys.stderr,
+            flush=True,
+        )
+        return _empty_databases_df()
+
+
+def backup_dumps_for_volume(
+    *,
+    containers: list[str],
+    vol_dir: str,
+    databases_df: "pandas.DataFrame",
+    database_containers: list[str],
+) -> tuple[bool, bool]:
+    """
+    Returns (found_db_container, dumped_any)
+    """
+    found_db = False
+    dumped_any = False
+
+    for c in containers:
+        is_db, dumped = backup_mariadb_or_postgres(
+            container=c,
+            volume_dir=vol_dir,
+            databases_df=databases_df,
+            database_containers=database_containers,
+        )
+        if is_db:
+            found_db = True
+        if dumped:
+            dumped_any = True
+
+    return found_db, dumped_any

src/baudolo/backup/layout.py

@@ -0,0 +1,33 @@
+"""Where a backup run puts its files, and how a finished run is stamped."""
+
+from __future__ import annotations
+
+import os
+import pathlib
+
+from dirval import create_stamp_file
+
+from .shell import execute_shell_command
+
+
+def get_machine_id() -> str:
+    return execute_shell_command("sha256sum /etc/machine-id")[0][0:64]
+
+
+def stamp_directory(version_dir: str) -> None:
+    """
+    Use dirval as a Python library to stamp the directory (no CLI dependency).
+    """
+    create_stamp_file(version_dir)
+
+
+def create_version_directory(versions_dir: str, backup_time: str) -> str:
+    version_dir = os.path.join(versions_dir, backup_time)
+    pathlib.Path(version_dir).mkdir(parents=True, exist_ok=True)
+    return version_dir
+
+
+def create_volume_directory(version_dir: str, volume_name: str) -> str:
+    path = os.path.join(version_dir, volume_name)
+    pathlib.Path(path).mkdir(parents=True, exist_ok=True)
+    return path

src/baudolo/backup/policy.py

@@ -0,0 +1,38 @@
+"""Which volumes are backed up, and which containers must stop for it."""
+
+from __future__ import annotations
+
+from .docker import get_image_info, is_swarm_task
+
+
+def is_image_ignored(container: str, images_no_backup_required: list[str]) -> bool:
+    if not images_no_backup_required:
+        return False
+    img = get_image_info(container)
+    return img in images_no_backup_required
+
+
+def volume_is_fully_ignored(
+    containers: list[str], images_no_backup_required: list[str]
+) -> bool:
+    """
+    Skip file backup only if all containers linked to the volume are ignored.
+    """
+    if not containers:
+        return False
+    return all(is_image_ignored(c, images_no_backup_required) for c in containers)
+
+
+def requires_stop(containers: list[str], images_no_stop_required: list[str]) -> bool:
+    """
+    Stop is required if ANY stoppable container image is NOT in the exact
+    image whitelist. Swarm task containers never count: baudolo must
+    not cycle them (see docker.is_swarm_task).
+    """
+    for c in containers:
+        if is_swarm_task(c):
+            continue
+        img = get_image_info(c)
+        if img not in images_no_stop_required:
+            return True
+    return False

src/baudolo/backup/snapshot.py

@@ -0,0 +1,95 @@
+"""Capture every volume from one atomic filesystem snapshot.
+
+Copying a live tree file by file cannot produce a point in time: a database can
+write between two files and leave a control file and its write-ahead log
+disagreeing, which no recovery can repair. A snapshot freezes the whole subject
+at once, so a database reads it as a crash and replays its log - a case it is
+built for. That also removes the reason to stop containers at all.
+
+The snapshot kind is stated by the caller rather than probed, because falling
+back to a live copy when a probe is inconclusive would hand out backups that
+look consistent and are not.
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
+
+from .shell import BackupException, execute_shell_command
+
+KINDS = ("btrfs", "zfs")
+
+
+class SnapshotError(RuntimeError):
+    """A snapshot could not be created, resolved or removed."""
+
+
+def _resolver(subject: str, root: str) -> Callable[[str], str]:
+    def resolve(path: str) -> str:
+        relative = os.path.relpath(os.path.abspath(path), os.path.abspath(subject))
+        if relative.startswith(".."):
+            raise SnapshotError(f"{path} lies outside the snapshot subject {subject}")
+        resolved = root if relative == "." else os.path.join(root, relative)
+
+        # abspath drops a trailing separator, and rsync reads "dir/" as its
+        # contents where "dir" means the directory itself.
+        return resolved + os.sep if path.endswith(os.sep) else resolved
+
+    return resolve
+
+
+def _btrfs(subject: str, name: str, run: Callable[[str], list[str]]) -> tuple[str, str]:
+    target = os.path.join(os.path.dirname(os.path.abspath(subject)), f".{name}")
+    run(f"btrfs subvolume snapshot -r {subject} {target}")
+    return target, f"btrfs subvolume delete {target}"
+
+
+def _zfs(subject: str, name: str, run: Callable[[str], list[str]]) -> tuple[str, str]:
+    output = run(f"zfs list -H -o name {subject}")
+    dataset = (output[0] if output else "").strip()
+    if not dataset:
+        raise SnapshotError(f"no zfs dataset is mounted at {subject}")
+    run(f"zfs snapshot {dataset}@{name}")
+    root = os.path.join(subject, ".zfs", "snapshot", name)
+    return root, f"zfs destroy {dataset}@{name}"
+
+
+_CREATE = {"btrfs": _btrfs, "zfs": _zfs}
+
+
+@contextmanager
+def volume_snapshot(
+    kind: str,
+    subject: str,
+    tag: str,
+    run: Callable[[str], list[str]] = execute_shell_command,
+) -> Iterator[Callable[[str], str]]:
+    """Yield a resolver mapping a path under ``subject`` into a snapshot of it.
+
+    Args:
+        kind: ``btrfs`` or ``zfs``; the caller states it, nothing is probed.
+        subject: the btrfs subvolume or zfs dataset mountpoint holding the
+            volumes, e.g. ``/var/lib/docker``.
+        tag: unique suffix for the snapshot name, e.g. the backup timestamp.
+        run: shell runner, injected so the mechanics are testable.
+
+    Raises:
+        SnapshotError: the kind is unknown, or the snapshot cannot be created.
+            Removal failure is reported, not raised: a leftover snapshot is a
+            cleanup problem and must not discard a generation that is complete.
+    """
+    create = _CREATE.get(kind)
+    if create is None:
+        raise SnapshotError(f"unknown snapshot kind {kind!r}; expected one of {KINDS}")
+
+    root, remove = create(subject, f"baudolo-{tag}", run)
+    try:
+        yield _resolver(subject, root)
+    finally:
+        try:
+            run(remove)
+        except BackupException as error:
+            # Raising here would also mask whatever the body raised.
+            print(f"WARNING: {root} could not be removed: {error}", flush=True)

src/baudolo/backup/volume.py

@@ -24,16 +24,32 @@ def get_last_backup_dir(
     return None
 
 
-def backup_volume(versions_dir: str, volume_name: str, volume_dir: str) -> None:
+def backup_volume(
-    """Perform incremental file backup of a Docker volume."""
+    versions_dir: str,
+    volume_name: str,
+    volume_dir: str,
+    *,
+    authoritative: bool,
+    source: str,
+) -> None:
+    """Perform incremental file backup of a Docker volume.
+
+    Args:
+        authoritative: compare source and destination by content instead of by
+            size and whole-second mtime. Required on a pass whose destination was
+            already written from a live source, where a file can differ while
+            both attributes still agree.
+        source: directory to read from - the volume's mountpoint, or its path
+            inside a snapshot.
+    """
     dest = os.path.join(volume_dir, "files") + "/"
     pathlib.Path(dest).mkdir(parents=True, exist_ok=True)
 
     last = get_last_backup_dir(versions_dir, volume_name, dest)
     link_dest = f"--link-dest='{last}'" if last else ""
-    source = get_storage_path(volume_name)
+    verify = "--checksum " if authoritative else ""
 
-    cmd = f"rsync -abP --delete --delete-excluded {link_dest} {source} {dest}"
+    cmd = f"rsync -abP --delete --delete-excluded {verify}{link_dest} {source} {dest}"
 
     try:
         execute_shell_command(cmd)

src/baudolo/restore/__main__.py

@@ -38,10 +38,6 @@ def main(argv: list[str] | None = None) -> int:
     # ------------------------------------------------------------------
     p_files = sub.add_parser("files", help="Restore files into a docker volume")
     _add_common_backup_args(p_files)
-    p_files.add_argument(
-        "--rsync-image",
-        default="ghcr.io/kevinveenbirkenbach/alpine-rsync",
-    )
     p_files.add_argument(
         "--source-volume",
         default=None,
@@ -95,7 +91,6 @@ def main(argv: list[str] | None = None) -> int:
             return restore_volume_files(
                 args.volume_name,
                 bp_files.files_dir(),
-                rsync_image=args.rsync_image,
             )
 
         if args.cmd == "postgres":

src/baudolo/restore/db/empty_preclean.sql

@@ -0,0 +1,64 @@
+-- Owner-filtered pre-clean for `restore --empty`. Emitted as one DROP per row and
+-- run via \gexec so each executes as its own top-level statement: a single DO-block
+-- would run every DROP in one transaction and exhaust max_locks_per_transaction on
+-- large schemas (e.g. gitlab). Also drops user-owned non-public schemas so a dump
+-- that CREATE SCHEMAs (e.g. discourse's discourse_functions) does not fail on an
+-- already-existing schema. Extension members (pg_trgm's set_limit) are
+-- superuser-owned; IF EXISTS absorbs the CASCADE fallout.
+SELECT format('DROP %s IF EXISTS public.%s CASCADE', obj.type, obj.name)
+  FROM (
+    SELECT format('%I', c.relname) AS name,
+           CASE c.relkind
+             WHEN 'v' THEN 'VIEW'
+             WHEN 'm' THEN 'MATERIALIZED VIEW'
+             WHEN 'f' THEN 'FOREIGN TABLE'
+             ELSE 'TABLE'
+           END AS type
+      FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
+     WHERE n.nspname = 'public' AND c.relkind IN ('r', 'p', 'v', 'm', 'f')
+       AND pg_get_userbyid(c.relowner) = current_user
+    UNION ALL
+    -- Overloaded functions share a proname; DROP needs the identity
+    -- signature or psql aborts with "function name is not unique".
+    SELECT format('%I(%s)', p.proname, pg_get_function_identity_arguments(p.oid)) AS name,
+           CASE p.prokind WHEN 'p' THEN 'PROCEDURE' ELSE 'FUNCTION' END AS type
+      FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace
+     WHERE n.nspname = 'public' AND p.prokind IN ('f', 'p', 'w')
+       AND pg_get_userbyid(p.proowner) = current_user
+    UNION ALL
+    SELECT format('%I', c.relname) AS name, 'SEQUENCE' AS type
+      FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
+     WHERE n.nspname = 'public' AND c.relkind = 'S'
+       AND pg_get_userbyid(c.relowner) = current_user
+    UNION ALL
+    SELECT format('%I', t.typname) AS name, 'TYPE' AS type
+      FROM pg_type t JOIN pg_namespace n ON n.oid = t.typnamespace
+     WHERE n.nspname = 'public'
+       AND pg_get_userbyid(t.typowner) = current_user
+       AND (t.typtype IN ('e', 'd')
+            OR (t.typtype = 'c' AND EXISTS (
+                  SELECT 1 FROM pg_class c2
+                   WHERE c2.oid = t.typrelid AND c2.relkind = 'c')))
+    UNION ALL
+    SELECT format('%I', col.collname) AS name, 'COLLATION' AS type
+      FROM pg_collation col JOIN pg_namespace n ON n.oid = col.collnamespace
+     WHERE n.nspname = 'public'
+       AND pg_get_userbyid(col.collowner) = current_user
+    UNION ALL
+    SELECT format('%I', ts.cfgname) AS name, 'TEXT SEARCH CONFIGURATION' AS type
+      FROM pg_ts_config ts JOIN pg_namespace n ON n.oid = ts.cfgnamespace
+     WHERE n.nspname = 'public'
+       AND pg_get_userbyid(ts.cfgowner) = current_user
+    UNION ALL
+    SELECT format('%I', d.dictname) AS name, 'TEXT SEARCH DICTIONARY' AS type
+      FROM pg_ts_dict d JOIN pg_namespace n ON n.oid = d.dictnamespace
+     WHERE n.nspname = 'public'
+       AND pg_get_userbyid(d.dictowner) = current_user
+  ) obj
+UNION ALL
+SELECT format('DROP SCHEMA IF EXISTS %I CASCADE', n.nspname)
+  FROM pg_namespace n
+ WHERE NOT starts_with(n.nspname, 'pg_')
+   AND n.nspname NOT IN ('public', 'information_schema')
+   AND pg_get_userbyid(n.nspowner) = current_user
+\gexec

src/baudolo/restore/db/postgres.py

@@ -7,6 +7,7 @@ from collections.abc import Iterable, Iterator
 from ..run import docker_exec
 
 _SUPERUSER_ONLY_PREFIXES = (b"COMMENT ON EXTENSION", b"ALTER DEFAULT PRIVILEGES")
+_EMPTY_PRECLEAN_SQL = os.path.join(os.path.dirname(__file__), "empty_preclean.sql")
 
 
 def filter_superuser_only_lines(lines: Iterable[bytes]) -> Iterator[bytes]:
@@ -53,51 +54,8 @@ def restore_postgres_sql(
     docker_env = {"PGPASSWORD": password}
 
     if empty:
-        # Owner-filtered: extension members (pg_trgm's set_limit) are superuser-owned; IF EXISTS absorbs CASCADE fallout.
+        with open(_EMPTY_PRECLEAN_SQL, encoding="utf-8") as preclean:
-        drop_sql = r"""
+            drop_sql = preclean.read()
-DO $$ DECLARE r RECORD;
-BEGIN
-  FOR r IN (
-    SELECT c.relname AS name,
-           CASE c.relkind
-             WHEN 'v' THEN 'VIEW'
-             WHEN 'm' THEN 'MATERIALIZED VIEW'
-             WHEN 'f' THEN 'FOREIGN TABLE'
-             ELSE 'TABLE'
-           END AS type
-      FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
-     WHERE n.nspname = 'public' AND c.relkind IN ('r', 'p', 'v', 'm', 'f')
-       AND pg_get_userbyid(c.relowner) = current_user
-    UNION ALL
-    SELECT p.proname AS name,
-           CASE p.prokind WHEN 'p' THEN 'PROCEDURE' ELSE 'FUNCTION' END AS type
-      FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace
-     WHERE n.nspname = 'public' AND p.prokind IN ('f', 'p', 'w')
-       AND pg_get_userbyid(p.proowner) = current_user
-    UNION ALL
-    SELECT c.relname AS name, 'SEQUENCE' AS type
-      FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
-     WHERE n.nspname = 'public' AND c.relkind = 'S'
-       AND pg_get_userbyid(c.relowner) = current_user
-    UNION ALL
-    SELECT t.typname AS name, 'TYPE' AS type
-      FROM pg_type t JOIN pg_namespace n ON n.oid = t.typnamespace
-     WHERE n.nspname = 'public'
-       AND pg_get_userbyid(t.typowner) = current_user
-       AND (t.typtype IN ('e', 'd')
-            OR (t.typtype = 'c' AND EXISTS (
-                  SELECT 1 FROM pg_class c2
-                   WHERE c2.oid = t.typrelid AND c2.relkind = 'c')))
-    UNION ALL
-    SELECT col.collname AS name, 'COLLATION' AS type
-      FROM pg_collation col JOIN pg_namespace n ON n.oid = col.collnamespace
-     WHERE n.nspname = 'public'
-       AND pg_get_userbyid(col.collowner) = current_user
-  ) LOOP
-    EXECUTE format('DROP %s IF EXISTS public.%I CASCADE', r.type, r.name);
-  END LOOP;
-END $$;
-"""
         docker_exec(
             container,
             ["psql", "-v", "ON_ERROR_STOP=1", "-U", user, "-d", db_name],
@@ -114,7 +72,16 @@ END $$;
         filtered.seek(0)
         docker_exec(
             container,
-            ["psql", "-v", "ON_ERROR_STOP=1", "-U", user, "-d", db_name],
+            [
+                "psql",
+                "--single-transaction",
+                "-v",
+                "ON_ERROR_STOP=1",
+                "-U",
+                user,
+                "-d",
+                db_name,
+            ],
             stdin=filtered,
             docker_env=docker_env,
         )

src/baudolo/restore/files.py

@@ -3,12 +3,10 @@ from __future__ import annotations
 import os
 import sys
 
-from .run import run, docker_volume_exists
+from .run import docker_volume_exists, run
 
 
-def restore_volume_files(
+def restore_volume_files(volume_name: str, backup_files_dir: str) -> int:
-    volume_name: str, backup_files_dir: str, *, rsync_image: str
-) -> int:
     if not os.path.isdir(backup_files_dir):
         print(f"ERROR: backup files dir not found: {backup_files_dir}", file=sys.stderr)
         return 2
@@ -19,21 +17,21 @@ def restore_volume_files(
     else:
         print(f"Volume {volume_name} already exists.")
 
-    # Keep behavior close to the old script: rsync -avv --delete
+    cp = run(
-    run(
+        ["docker", "volume", "inspect", "--format", "{{ .Mountpoint }}", volume_name],
-        [
+        capture=True,
-            "docker",
-            "run",
-            "--rm",
-            "-v",
-            f"{volume_name}:/recover/",
-            "-v",
-            f"{backup_files_dir}:/backup/",
-            rsync_image,
-            "sh",
-            "-lc",
-            "rsync -avv --delete /backup/ /recover/",
-        ]
     )
+    raw = cp.stdout or b""
+    mountpoint = (raw.decode() if isinstance(raw, bytes) else raw).strip()
+    if not mountpoint:
+        print(
+            f"ERROR: could not resolve mountpoint for volume {volume_name}",
+            file=sys.stderr,
+        )
+        return 2
+
+    src = os.path.join(backup_files_dir, "")
+    dest = os.path.join(mountpoint, "")
+    run(["rsync", "-avv", "--delete", src, dest])
     print("File restore complete.")
     return 0

tests/e2e/helpers/__init__.py

@@ -0,0 +1,4 @@
+"""Shared e2e helpers, re-exported so tests import one name."""
+
+from .fixtures import *  # noqa: F401,F403
+from .process import *  # noqa: F401,F403

tests/e2e/helpers/fixtures.py

@@ -0,0 +1,114 @@
+"""Fixtures and paths the e2e suite builds its scenarios from."""
+from __future__ import annotations
+
+import shutil
+import subprocess
+from pathlib import Path
+
+from .process import machine_hash, run
+
+# postgres 18+ mounts at /var/lib/postgresql, not /var/lib/postgresql/data.
+POSTGRES_IMAGE = "postgres:alpine"
+POSTGRES_DATA_DIR = "/var/lib/postgresql"
+MARIADB_IMAGE = "mariadb:latest"
+MARIADB_DATA_DIR = "/var/lib/mysql"
+
+
+def backup_run(
+    *,
+    backups_dir: str,
+    repo_name: str,
+    compose_dir: str,
+    databases_csv: str,
+    database_containers: list[str],
+    images_no_stop_required: list[str],
+    images_no_backup_required: list[str] | None = None,
+    dump_only_sql: bool = False,
+) -> None:
+    cmd = [
+        "baudolo",
+        "--compose-dir",
+        compose_dir,
+        "--hard-restart-projects",
+        "mailu",
+        "--repo-name",
+        repo_name,
+        "--databases-csv",
+        databases_csv,
+        "--backups-dir",
+        backups_dir,
+        "--database-containers",
+        *database_containers,
+        "--images-no-stop-required",
+        *images_no_stop_required,
+    ]
+    if images_no_backup_required:
+        cmd += ["--images-no-backup-required", *images_no_backup_required]
+    if dump_only_sql:
+        cmd += ["--dump-only-sql"]
+
+    try:
+        run(cmd, capture=True, check=True)
+    except subprocess.CalledProcessError as e:
+        print(">>> baudolo failed (exit code:", e.returncode, ")")
+        if e.stdout:
+            print(">>> baudolo STDOUT:\n" + e.stdout)
+        if e.stderr:
+            print(">>> baudolo STDERR:\n" + e.stderr)
+        raise
+
+
+def latest_version_dir(backups_dir: str, repo_name: str) -> tuple[str, str]:
+    """
+    Returns (hash, version) for the latest backup.
+    """
+    h = machine_hash()
+    root = Path(backups_dir) / h / repo_name
+    if not root.is_dir():
+        raise FileNotFoundError(str(root))
+
+    versions = sorted([p.name for p in root.iterdir() if p.is_dir()])
+    if not versions:
+        raise RuntimeError(f"No versions found under {root}")
+    return h, versions[-1]
+
+
+def backup_path(backups_dir: str, repo_name: str, version: str, volume: str) -> Path:
+    h = machine_hash()
+    return Path(backups_dir) / h / repo_name / version / volume
+
+
+def create_minimal_compose_dir(base: str) -> str:
+    """
+    baudolo requires --compose-dir. Create an empty dir with one non-compose subdir.
+    """
+    p = Path(base) / "compose-root"
+    p.mkdir(parents=True, exist_ok=True)
+    (p / "noop").mkdir(parents=True, exist_ok=True)
+    return str(p)
+
+
+def write_databases_csv(path: str, rows: list[tuple[str, str, str, str]]) -> None:
+    """
+    rows: (instance, database, username, password)
+    database may be '' (empty) to trigger pg_dumpall behavior if you want, but here we use db name.
+    """
+    Path(path).parent.mkdir(parents=True, exist_ok=True)
+    with open(path, "w", encoding="utf-8") as f:
+        f.write("instance;database;username;password\n")
+        for inst, db, user, pw in rows:
+            f.write(f"{inst};{db};{user};{pw}\n")
+
+
+def cleanup_docker(*, containers: list[str], volumes: list[str]) -> None:
+    for c in containers:
+        run(["docker", "rm", "-f", c], capture=True, check=False)
+    for v in volumes:
+        run(["docker", "volume", "rm", "-f", v], capture=True, check=False)
+
+
+def ensure_empty_dir(path: str) -> None:
+    p = Path(path)
+    if p.exists():
+        shutil.rmtree(p)
+    p.mkdir(parents=True, exist_ok=True)

tests/e2e/helpers/process.py

@@ -1,11 +1,9 @@
-# tests/e2e/helpers.py
+"""Process, docker and readiness helpers for the e2e suite."""
 from __future__ import annotations
 
-import shutil
 import subprocess
 import time
 import uuid
-from pathlib import Path
 
 
 def run(
@@ -155,103 +153,3 @@ def wait_for_mariadb_sql(
     raise TimeoutError(
         f"Timed out waiting for MariaDB SQL login readiness in container {container}"
     )
-
-
-def backup_run(
-    *,
-    backups_dir: str,
-    repo_name: str,
-    compose_dir: str,
-    databases_csv: str,
-    database_containers: list[str],
-    images_no_stop_required: list[str],
-    images_no_backup_required: list[str] | None = None,
-    dump_only_sql: bool = False,
-) -> None:
-    cmd = [
-        "baudolo",
-        "--compose-dir",
-        compose_dir,
-        "--docker-compose-hard-restart-required",
-        "mailu",
-        "--repo-name",
-        repo_name,
-        "--databases-csv",
-        databases_csv,
-        "--backups-dir",
-        backups_dir,
-        "--database-containers",
-        *database_containers,
-        "--images-no-stop-required",
-        *images_no_stop_required,
-    ]
-    if images_no_backup_required:
-        cmd += ["--images-no-backup-required", *images_no_backup_required]
-    if dump_only_sql:
-        cmd += ["--dump-only-sql"]
-
-    try:
-        run(cmd, capture=True, check=True)
-    except subprocess.CalledProcessError as e:
-        print(">>> baudolo failed (exit code:", e.returncode, ")")
-        if e.stdout:
-            print(">>> baudolo STDOUT:\n" + e.stdout)
-        if e.stderr:
-            print(">>> baudolo STDERR:\n" + e.stderr)
-        raise
-
-
-def latest_version_dir(backups_dir: str, repo_name: str) -> tuple[str, str]:
-    """
-    Returns (hash, version) for the latest backup.
-    """
-    h = machine_hash()
-    root = Path(backups_dir) / h / repo_name
-    if not root.is_dir():
-        raise FileNotFoundError(str(root))
-
-    versions = sorted([p.name for p in root.iterdir() if p.is_dir()])
-    if not versions:
-        raise RuntimeError(f"No versions found under {root}")
-    return h, versions[-1]
-
-
-def backup_path(backups_dir: str, repo_name: str, version: str, volume: str) -> Path:
-    h = machine_hash()
-    return Path(backups_dir) / h / repo_name / version / volume
-
-
-def create_minimal_compose_dir(base: str) -> str:
-    """
-    baudolo requires --compose-dir. Create an empty dir with one non-compose subdir.
-    """
-    p = Path(base) / "compose-root"
-    p.mkdir(parents=True, exist_ok=True)
-    (p / "noop").mkdir(parents=True, exist_ok=True)
-    return str(p)
-
-
-def write_databases_csv(path: str, rows: list[tuple[str, str, str, str]]) -> None:
-    """
-    rows: (instance, database, username, password)
-    database may be '' (empty) to trigger pg_dumpall behavior if you want, but here we use db name.
-    """
-    Path(path).parent.mkdir(parents=True, exist_ok=True)
-    with open(path, "w", encoding="utf-8") as f:
-        f.write("instance;database;username;password\n")
-        for inst, db, user, pw in rows:
-            f.write(f"{inst};{db};{user};{pw}\n")
-
-
-def cleanup_docker(*, containers: list[str], volumes: list[str]) -> None:
-    for c in containers:
-        run(["docker", "rm", "-f", c], capture=True, check=False)
-    for v in volumes:
-        run(["docker", "volume", "rm", "-f", v], capture=True, check=False)
-
-
-def ensure_empty_dir(path: str) -> None:
-    p = Path(path)
-    if p.exists():
-        shutil.rmtree(p)
-    p.mkdir(parents=True, exist_ok=True)

tests/e2e/snapshot_db_driver.py

@@ -0,0 +1,41 @@
+"""Copy a live database's volume out of a snapshot, using the real backup path.
+
+Runs inside the privileged container built by test_e2e_snapshot_db.py, where a
+database is mid-write on a btrfs subvolume. Exercises volume_snapshot and
+backup_volume exactly as a backup run would.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+
+sys.path.insert(0, "/src")
+
+from baudolo.backup.snapshot import SnapshotError, volume_snapshot  # noqa: E402
+from baudolo.backup.volume import backup_volume  # noqa: E402
+
+SUBJECT = "/subject/docker"
+VOLUME = "mariadb_data"
+DATADIR = f"{SUBJECT}/volumes/{VOLUME}/_data"
+VERSIONS = "/backups"
+GENERATION = f"{VERSIONS}/20260731"
+
+
+def shell(command: str) -> list[str]:
+    proc = subprocess.run(command, shell=True, capture_output=True, text=True)
+    if proc.returncode != 0:
+        raise SnapshotError(f"{command} exited {proc.returncode}: {proc.stderr.strip()}")
+    return proc.stdout.splitlines()
+
+
+with volume_snapshot("btrfs", SUBJECT, "dbtest", run=shell) as resolve:
+    backup_volume(
+        VERSIONS,
+        VOLUME,
+        f"{GENERATION}/{VOLUME}",
+        authoritative=True,
+        source=resolve(f"{DATADIR}/"),
+    )
+
+print("SNAPSHOT COPY DONE", flush=True)

tests/e2e/snapshot_driver.py

@@ -0,0 +1,59 @@
+"""Exercise volume_snapshot against a real filesystem, from inside a container.
+
+Runs where loop devices exist. Prints one PASS/FAIL line per assertion and exits
+non-zero on the first failure, so the calling test can surface the reason.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+sys.path.insert(0, "/src")
+
+from baudolo.backup.snapshot import SnapshotError, volume_snapshot  # noqa: E402
+
+KIND = sys.argv[1]
+SUBJECT = sys.argv[2]
+EXPECT = sys.argv[3]
+
+
+def shell(command: str) -> list[str]:
+    proc = subprocess.run(command, shell=True, capture_output=True, text=True)
+    if proc.returncode != 0:
+        raise SnapshotError(f"{command} exited {proc.returncode}: {proc.stderr.strip()}")
+    return proc.stdout.splitlines()
+
+
+def check(label: str, condition: bool) -> None:
+    print(f"{'PASS' if condition else 'FAIL'} {label}", flush=True)
+    if not condition:
+        sys.exit(1)
+
+
+volume = Path(SUBJECT) / "volumes" / "demo" / "_data"
+volume.mkdir(parents=True, exist_ok=True)
+(volume / "state").write_text("before\n")
+
+if EXPECT == "unsupported":
+    try:
+        with volume_snapshot(KIND, SUBJECT, "e2e", run=shell):
+            check("snapshot on an unsupported filesystem must not succeed", False)
+    except SnapshotError as exc:
+        check(f"refused loudly: {str(exc)[:60]}", True)
+    sys.exit(0)
+
+with volume_snapshot(KIND, SUBJECT, "e2e", run=shell) as resolve:
+    frozen = Path(resolve(str(volume))) / "state"
+    check("the snapshot exposes the volume", frozen.is_file())
+    check("the snapshot carries the content", frozen.read_text() == "before\n")
+
+    (volume / "state").write_text("after\n")
+    check("a later write does not reach the snapshot", frozen.read_text() == "before\n")
+    check("the live tree did change", (volume / "state").read_text() == "after\n")
+
+    root = Path(resolve(SUBJECT))
+
+check("the snapshot is removed afterwards", not root.exists() or not (root / "volumes").exists())
+print("ALL OK", flush=True)

tests/e2e/test_e2e_dump_only_fallback_to_files.py

@@ -2,6 +2,8 @@
 import unittest
 
 from .helpers import (
+    POSTGRES_IMAGE,
+    POSTGRES_DATA_DIR,
     backup_path,
     cleanup_docker,
     create_minimal_compose_dir,
@@ -50,8 +52,8 @@ class TestE2EDumpOnlyFallbackToFiles(unittest.TestCase):
                 "-e",
                 "POSTGRES_USER=postgres",
                 "-v",
-                f"{cls.pg_volume}:/var/lib/postgresql/data",
+                f"{cls.pg_volume}:{POSTGRES_DATA_DIR}",
-                "postgres:16",
+                POSTGRES_IMAGE,
             ]
         )
         wait_for_postgres(cls.pg_container, user="postgres", timeout_s=90)
@@ -65,7 +67,7 @@ class TestE2EDumpOnlyFallbackToFiles(unittest.TestCase):
                 cls.pg_container,
                 "sh",
                 "-lc",
-                f"echo '{cls.marker}' > /var/lib/postgresql/data/marker.txt",
+                f"echo '{cls.marker}' > {POSTGRES_DATA_DIR}/marker.txt",
             ]
         )
 
@@ -79,7 +81,7 @@ class TestE2EDumpOnlyFallbackToFiles(unittest.TestCase):
             "baudolo",
             "--compose-dir",
             cls.compose_dir,
-            "--docker-compose-hard-restart-required",
+            "--hard-restart-projects",
             "mailu",
             "--repo-name",
             cls.repo_name,
@@ -90,10 +92,7 @@ class TestE2EDumpOnlyFallbackToFiles(unittest.TestCase):
             "--database-containers",
             cls.pg_container,
             "--images-no-stop-required",
-            "postgres",
+            POSTGRES_IMAGE,
-            "mariadb",
-            "mysql",
-            "alpine",
             "--dump-only-sql",
         ]
         cp = run(cmd, capture=True, check=True)
@@ -116,8 +115,6 @@ class TestE2EDumpOnlyFallbackToFiles(unittest.TestCase):
                 cls.repo_name,
                 "--source-volume",
                 cls.pg_volume,
-                "--rsync-image",
-                "ghcr.io/kevinveenbirkenbach/alpine-rsync",
             ]
         )
 

tests/e2e/test_e2e_dump_only_sql_mixed_run.py

@@ -1,6 +1,8 @@
 import unittest
 
 from .helpers import (
+    POSTGRES_IMAGE,
+    POSTGRES_DATA_DIR,
     backup_path,
     cleanup_docker,
     create_minimal_compose_dir,
@@ -70,8 +72,8 @@ class TestE2EDumpOnlySqlMixedRun(unittest.TestCase):
                 "-e",
                 f"POSTGRES_PASSWORD={cls.pg_password}",
                 "-v",
-                f"{cls.db_volume}:/var/lib/postgresql/data",
+                f"{cls.db_volume}:{POSTGRES_DATA_DIR}",
-                "postgres:16-alpine",
+                POSTGRES_IMAGE,
             ]
         )
         wait_for_postgres(cls.pg_container, user="postgres", timeout_s=90)
@@ -122,10 +124,7 @@ class TestE2EDumpOnlySqlMixedRun(unittest.TestCase):
             "--database-containers",
             cls.pg_container,
             "--images-no-stop-required",
-            "alpine",
+            POSTGRES_IMAGE,
-            "postgres",
-            "mariadb",
-            "mysql",
             "--dump-only-sql",
             "--backups-dir",
             cls.backups_dir,

tests/e2e/test_e2e_files_full.py

@@ -57,7 +57,7 @@ class TestE2EFilesFull(unittest.TestCase):
             compose_dir=cls.compose_dir,
             databases_csv=cls.databases_csv,
             database_containers=["dummy-db"],
-            images_no_stop_required=["alpine", "postgres", "mariadb", "mysql"],
+            images_no_stop_required=["alpine:3.20"],
         )
 
         cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)
@@ -94,8 +94,6 @@ class TestE2EFilesFull(unittest.TestCase):
                 self.repo_name,
                 "--source-volume",
                 self.volume_src,
-                "--rsync-image",
-                "ghcr.io/kevinveenbirkenbach/alpine-rsync",
             ]
         )
 

tests/e2e/test_e2e_files_no_copy.py

@@ -55,7 +55,7 @@ class TestE2EFilesNoCopy(unittest.TestCase):
             compose_dir=cls.compose_dir,
             databases_csv=cls.databases_csv,
             database_containers=["dummy-db"],
-            images_no_stop_required=["alpine", "postgres", "mariadb", "mysql"],
+            images_no_stop_required=["alpine:3.20"],
             dump_only_sql=True,
         )
 

tests/e2e/test_e2e_images_no_backup_required_early_skip.py

@@ -76,7 +76,7 @@ class TestE2EImagesNoBackupRequiredEarlySkip(unittest.TestCase):
             "baudolo",
             "--compose-dir",
             cls.compose_dir,
-            "--docker-compose-hard-restart-required",
+            "--hard-restart-projects",
             "mailu",
             "--repo-name",
             cls.repo_name,
@@ -87,13 +87,9 @@ class TestE2EImagesNoBackupRequiredEarlySkip(unittest.TestCase):
             "--database-containers",
             "dummy-db",
             "--images-no-stop-required",
-            "alpine",
+            "redis:alpine",
-            "redis",
-            "postgres",
-            "mariadb",
-            "mysql",
             "--images-no-backup-required",
-            "redis",
+            "redis:alpine",
         ]
         cp = run(cmd, capture=True, check=True)
         cls.stdout = cp.stdout or ""

tests/e2e/test_e2e_mariadb_anonymous_preemption.py

@@ -30,6 +30,8 @@ import pandas
 from baudolo.backup import db as db_mod
 
 from .helpers import (
+    MARIADB_IMAGE,
+    MARIADB_DATA_DIR,
     cleanup_docker,
     require_docker,
     run,
@@ -69,8 +71,8 @@ class TestE2EMariaDBAnonymousPreemption(unittest.TestCase):
                 "-e",
                 f"MARIADB_ROOT_PASSWORD={cls.root_password}",
                 "-v",
-                f"{cls.db_volume}:/var/lib/mysql",
+                f"{cls.db_volume}:{MARIADB_DATA_DIR}",
-                "mariadb:12.2",
+                MARIADB_IMAGE,
             ]
         )
 

tests/e2e/test_e2e_mariadb_full.py

@@ -2,6 +2,8 @@
 import unittest
 
 from .helpers import (
+    MARIADB_IMAGE,
+    MARIADB_DATA_DIR,
     backup_run,
     backup_path,
     cleanup_docker,
@@ -56,8 +58,8 @@ class TestE2EMariaDBFull(unittest.TestCase):
                 "-e",
                 f"MARIADB_PASSWORD={cls.db_password}",
                 "-v",
-                f"{cls.db_volume}:/var/lib/mysql",
+                f"{cls.db_volume}:{MARIADB_DATA_DIR}",
-                "mariadb:11",
+                MARIADB_IMAGE,
             ]
         )
 
@@ -97,7 +99,7 @@ class TestE2EMariaDBFull(unittest.TestCase):
             compose_dir=cls.compose_dir,
             databases_csv=cls.databases_csv,
             database_containers=[cls.db_container],
-            images_no_stop_required=["mariadb", "mysql", "alpine", "postgres"],
+            images_no_stop_required=[MARIADB_IMAGE],
         )
 
         cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)

tests/e2e/test_e2e_mariadb_no_copy.py

@@ -2,6 +2,8 @@
 import unittest
 
 from .helpers import (
+    MARIADB_IMAGE,
+    MARIADB_DATA_DIR,
     backup_run,
     backup_path,
     cleanup_docker,
@@ -55,8 +57,8 @@ class TestE2EMariaDBNoCopy(unittest.TestCase):
                 "-e",
                 f"MARIADB_PASSWORD={cls.db_password}",
                 "-v",
-                f"{cls.db_volume}:/var/lib/mysql",
+                f"{cls.db_volume}:{MARIADB_DATA_DIR}",
-                "mariadb:11",
+                MARIADB_IMAGE,
             ]
         )
 
@@ -94,7 +96,7 @@ class TestE2EMariaDBNoCopy(unittest.TestCase):
             compose_dir=cls.compose_dir,
             databases_csv=cls.databases_csv,
             database_containers=[cls.db_container],
-            images_no_stop_required=["mariadb", "mysql", "alpine", "postgres"],
+            images_no_stop_required=[MARIADB_IMAGE],
             dump_only_sql=True,
         )
 

tests/e2e/test_e2e_postgres_empty_drop_hard.py

@@ -0,0 +1,186 @@
+# tests/e2e/test_e2e_postgres_empty_drop_hard.py
+import unittest
+
+from .helpers import (
+    POSTGRES_IMAGE,
+    POSTGRES_DATA_DIR,
+    backup_run,
+    cleanup_docker,
+    create_minimal_compose_dir,
+    ensure_empty_dir,
+    latest_version_dir,
+    require_docker,
+    run,
+    unique,
+    wait_for_postgres,
+    write_databases_csv,
+)
+
+# The scenario the --empty pre-clean must survive: a non-public user schema
+# plus one object of every class the discovery SELECT enumerates. Restore
+# --empty runs against the still-populated DB (no wipe), so the pre-clean must
+# drop discourse_functions too or the dump's CREATE SCHEMA aborts the replay
+# under ON_ERROR_STOP; the old public-only DROP left it and broke discourse.
+# The f()/f(int) pair reproduces discourse's overload abort ("function name
+# is not unique") and english_stem_nostop reproduces taiga's text search
+# dictionary abort (duplicate pg_ts_dict_dictname_index).
+SCENARIO_SQL = (
+    "CREATE SCHEMA discourse_functions;"
+    "CREATE TABLE discourse_functions.helper (id int);"
+    "INSERT INTO discourse_functions.helper VALUES (1);"
+    "CREATE TABLE public.t (id int primary key, v text);"
+    "INSERT INTO public.t VALUES (1, 'ok');"
+    "CREATE VIEW public.t_view AS SELECT * FROM public.t;"
+    "CREATE SEQUENCE public.s;"
+    "CREATE TYPE public.mood AS ENUM ('ok', 'bad');"
+    "CREATE FUNCTION public.f() RETURNS int LANGUAGE sql AS 'SELECT 1';"
+    "CREATE FUNCTION public.f(i int) RETURNS int LANGUAGE sql AS 'SELECT i';"
+    "CREATE COLLATION public.c (locale = 'C');"
+    "CREATE TEXT SEARCH DICTIONARY public.english_stem_nostop"
+    " (Template = snowball, Language = english);"
+    "CREATE TEXT SEARCH CONFIGURATION public.english_nostop (COPY = english);"
+    "ALTER TEXT SEARCH CONFIGURATION public.english_nostop"
+    " ALTER MAPPING FOR asciiword WITH public.english_stem_nostop;"
+)
+
+
+class TestE2EPostgresEmptyDropHard(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+        cls.prefix = unique("baudolo-e2e-postgres-empty-drop-hard")
+        cls.backups_dir = f"/tmp/{cls.prefix}/Backups"
+        ensure_empty_dir(cls.backups_dir)
+        cls.compose_dir = create_minimal_compose_dir(f"/tmp/{cls.prefix}")
+        cls.repo_name = cls.prefix
+
+        cls.pg_container = f"{cls.prefix}-pg"
+        cls.pg_volume = f"{cls.prefix}-pg-vol"
+        cls.containers = [cls.pg_container]
+        cls.volumes = [cls.pg_volume]
+
+        run(["docker", "volume", "create", cls.pg_volume])
+        run(
+            [
+                "docker",
+                "run",
+                "-d",
+                "--name",
+                cls.pg_container,
+                "-e",
+                "POSTGRES_PASSWORD=pgpw",
+                "-e",
+                "POSTGRES_DB=appdb",
+                "-e",
+                "POSTGRES_USER=postgres",
+                "-v",
+                f"{cls.pg_volume}:{POSTGRES_DATA_DIR}",
+                POSTGRES_IMAGE,
+            ]
+        )
+        wait_for_postgres(cls.pg_container, user="postgres", timeout_s=90)
+
+        run(
+            [
+                "docker",
+                "exec",
+                cls.pg_container,
+                "sh",
+                "-lc",
+                f'psql -U postgres -d appdb -v ON_ERROR_STOP=1 -c "{SCENARIO_SQL}"',
+            ]
+        )
+
+        cls.databases_csv = f"/tmp/{cls.prefix}/databases.csv"
+        write_databases_csv(
+            cls.databases_csv, [(cls.pg_container, "appdb", "postgres", "pgpw")]
+        )
+        backup_run(
+            backups_dir=cls.backups_dir,
+            repo_name=cls.repo_name,
+            compose_dir=cls.compose_dir,
+            databases_csv=cls.databases_csv,
+            database_containers=[cls.pg_container],
+            images_no_stop_required=[POSTGRES_IMAGE],
+        )
+        cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)
+
+        # No wipe: restore --empty must pre-clean the fully-populated DB
+        # (incl. the non-public schema) before replaying the dump.
+        run(
+            [
+                "baudolo-restore",
+                "postgres",
+                cls.pg_volume,
+                cls.hash,
+                cls.version,
+                "--backups-dir",
+                cls.backups_dir,
+                "--repo-name",
+                cls.repo_name,
+                "--container",
+                cls.pg_container,
+                "--db-name",
+                "appdb",
+                "--db-user",
+                "postgres",
+                "--db-password",
+                "pgpw",
+                "--empty",
+            ]
+        )
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cleanup_docker(containers=cls.containers, volumes=cls.volumes)
+
+    def _scalar(self, sql: str) -> str:
+        p = run(
+            [
+                "docker",
+                "exec",
+                self.pg_container,
+                "sh",
+                "-lc",
+                f'psql -U postgres -d appdb -t -A -c "{sql}"',
+            ]
+        )
+        return (p.stdout or "").strip()
+
+    def test_public_data_restored(self) -> None:
+        self.assertEqual(self._scalar("SELECT v FROM public.t WHERE id=1;"), "ok")
+
+    def test_view_restored(self) -> None:
+        self.assertEqual(self._scalar("SELECT count(*) FROM public.t_view;"), "1")
+
+    def test_non_public_schema_restored(self) -> None:
+        self.assertEqual(
+            self._scalar(
+                "SELECT count(*) FROM pg_namespace WHERE nspname='discourse_functions';"
+            ),
+            "1",
+        )
+
+    def test_overloaded_functions_restored_once_each(self) -> None:
+        self.assertEqual(
+            self._scalar("SELECT count(*) FROM pg_proc WHERE proname='f';"), "2"
+        )
+        self.assertEqual(self._scalar("SELECT public.f(41) + public.f();"), "42")
+
+    def test_text_search_dictionary_restored_once(self) -> None:
+        self.assertEqual(
+            self._scalar(
+                "SELECT count(*) FROM pg_ts_dict WHERE dictname='english_stem_nostop';"
+            ),
+            "1",
+        )
+        self.assertEqual(
+            self._scalar(
+                "SELECT count(*) FROM pg_ts_config WHERE cfgname='english_nostop';"
+            ),
+            "1",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/e2e/test_e2e_postgres_full.py

@@ -2,6 +2,8 @@
 import unittest
 
 from .helpers import (
+    POSTGRES_IMAGE,
+    POSTGRES_DATA_DIR,
     backup_run,
     backup_path,
     cleanup_docker,
@@ -47,8 +49,8 @@ class TestE2EPostgresFull(unittest.TestCase):
                 "-e",
                 "POSTGRES_USER=postgres",
                 "-v",
-                f"{cls.pg_volume}:/var/lib/postgresql/data",
+                f"{cls.pg_volume}:{POSTGRES_DATA_DIR}",
-                "postgres:16",
+                POSTGRES_IMAGE,
             ]
         )
         wait_for_postgres(cls.pg_container, user="postgres", timeout_s=90)
@@ -76,7 +78,7 @@ class TestE2EPostgresFull(unittest.TestCase):
             compose_dir=cls.compose_dir,
             databases_csv=cls.databases_csv,
             database_containers=[cls.pg_container],
-            images_no_stop_required=["postgres", "mariadb", "mysql", "alpine"],
+            images_no_stop_required=[POSTGRES_IMAGE],
         )
 
         cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)

tests/e2e/test_e2e_postgres_no_copy.py

@@ -2,6 +2,8 @@
 import unittest
 
 from .helpers import (
+    POSTGRES_IMAGE,
+    POSTGRES_DATA_DIR,
     backup_run,
     backup_path,
     cleanup_docker,
@@ -46,8 +48,8 @@ class TestE2EPostgresNoCopy(unittest.TestCase):
                 "-e",
                 "POSTGRES_USER=postgres",
                 "-v",
-                f"{cls.pg_volume}:/var/lib/postgresql/data",
+                f"{cls.pg_volume}:{POSTGRES_DATA_DIR}",
-                "postgres:16",
+                POSTGRES_IMAGE,
             ]
         )
         wait_for_postgres(cls.pg_container, user="postgres", timeout_s=90)
@@ -74,7 +76,7 @@ class TestE2EPostgresNoCopy(unittest.TestCase):
             compose_dir=cls.compose_dir,
             databases_csv=cls.databases_csv,
             database_containers=[cls.pg_container],
-            images_no_stop_required=["postgres", "mariadb", "mysql", "alpine"],
+            images_no_stop_required=[POSTGRES_IMAGE],
             dump_only_sql=True,
         )
 

tests/e2e/test_e2e_postgres_single_transaction_live_writer.py

@@ -0,0 +1,185 @@
+# tests/e2e/test_e2e_postgres_single_transaction_live_writer.py
+import unittest
+
+from .helpers import (
+    POSTGRES_IMAGE,
+    POSTGRES_DATA_DIR,
+    backup_run,
+    cleanup_docker,
+    create_minimal_compose_dir,
+    ensure_empty_dir,
+    latest_version_dir,
+    require_docker,
+    run,
+    unique,
+    wait_for_postgres,
+    write_databases_csv,
+)
+
+# The discourse restore-drill race: `restore --empty` replays the dump into a
+# LIVE database while a background writer keeps touching a primary-key row
+# (discourse's mini_scheduler upserts scheduler_stats(id=1)). Without a
+# single-transaction replay, the pre-clean drops the table, the replay recreates
+# it and auto-commits, the writer wins the gap and inserts id=1, and the dump's
+# COPY of the same id then aborts with a duplicate-key violation under
+# ON_ERROR_STOP -> the whole restore fails. The --single-transaction replay keeps
+# the recreated table invisible until commit, so the writer can never insert the
+# racing row and the restore completes. A wide filler table makes the COPY slow
+# enough that the non-transactional variant loses the race deterministically.
+SEED_SQL = (
+    "CREATE TABLE public.scheduler_stats (id int primary key, v text);"
+    "INSERT INTO public.scheduler_stats VALUES (1, 'from-dump');"
+    "CREATE TABLE public.filler (id serial primary key, blob text);"
+    "INSERT INTO public.filler (blob)"
+    " SELECT repeat('x', 512) FROM generate_series(1, 100000);"
+)
+
+WRITER_LOOP = (
+    "while true; do "
+    "psql -h 127.0.0.1 -U postgres -d appdb "
+    "-c \"INSERT INTO public.scheduler_stats(id, v) VALUES (1, 'live') "
+    'ON CONFLICT (id) DO NOTHING;" >/dev/null 2>&1; '
+    "done"
+)
+
+
+class TestE2EPostgresSingleTransactionLiveWriter(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+        cls.prefix = unique("baudolo-e2e-pg-single-txn")
+        cls.backups_dir = f"/tmp/{cls.prefix}/Backups"
+        ensure_empty_dir(cls.backups_dir)
+        cls.compose_dir = create_minimal_compose_dir(f"/tmp/{cls.prefix}")
+        cls.repo_name = cls.prefix
+
+        cls.pg_container = f"{cls.prefix}-pg"
+        cls.pg_volume = f"{cls.prefix}-pg-vol"
+        cls.writer = f"{cls.prefix}-writer"
+        cls.containers = [cls.pg_container, cls.writer]
+        cls.volumes = [cls.pg_volume]
+
+        run(["docker", "volume", "create", cls.pg_volume])
+        run(
+            [
+                "docker",
+                "run",
+                "-d",
+                "--name",
+                cls.pg_container,
+                "-e",
+                "POSTGRES_PASSWORD=pgpw",
+                "-e",
+                "POSTGRES_DB=appdb",
+                "-e",
+                "POSTGRES_USER=postgres",
+                "-v",
+                f"{cls.pg_volume}:{POSTGRES_DATA_DIR}",
+                POSTGRES_IMAGE,
+            ]
+        )
+        wait_for_postgres(cls.pg_container, user="postgres", timeout_s=90)
+
+        run(
+            [
+                "docker",
+                "exec",
+                cls.pg_container,
+                "sh",
+                "-lc",
+                f'psql -U postgres -d appdb -v ON_ERROR_STOP=1 -c "{SEED_SQL}"',
+            ]
+        )
+
+        cls.databases_csv = f"/tmp/{cls.prefix}/databases.csv"
+        write_databases_csv(
+            cls.databases_csv, [(cls.pg_container, "appdb", "postgres", "pgpw")]
+        )
+        backup_run(
+            backups_dir=cls.backups_dir,
+            repo_name=cls.repo_name,
+            compose_dir=cls.compose_dir,
+            databases_csv=cls.databases_csv,
+            database_containers=[cls.pg_container],
+            images_no_stop_required=[POSTGRES_IMAGE],
+        )
+        cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)
+
+        run(
+            [
+                "docker",
+                "run",
+                "-d",
+                "--name",
+                cls.writer,
+                "--network",
+                f"container:{cls.pg_container}",
+                "-e",
+                "PGPASSWORD=pgpw",
+                POSTGRES_IMAGE,
+                "sh",
+                "-lc",
+                WRITER_LOOP,
+            ]
+        )
+
+        cls.restore = run(
+            [
+                "baudolo-restore",
+                "postgres",
+                cls.pg_volume,
+                cls.hash,
+                cls.version,
+                "--backups-dir",
+                cls.backups_dir,
+                "--repo-name",
+                cls.repo_name,
+                "--container",
+                cls.pg_container,
+                "--db-name",
+                "appdb",
+                "--db-user",
+                "postgres",
+                "--db-password",
+                "pgpw",
+                "--empty",
+            ],
+            capture=True,
+            check=False,
+        )
+
+        run(["docker", "rm", "-f", cls.writer], capture=True, check=False)
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cleanup_docker(containers=cls.containers, volumes=cls.volumes)
+
+    def _scalar(self, sql: str) -> str:
+        p = run(
+            [
+                "docker",
+                "exec",
+                self.pg_container,
+                "sh",
+                "-lc",
+                f'psql -U postgres -d appdb -t -A -c "{sql}"',
+            ]
+        )
+        return (p.stdout or "").strip()
+
+    def test_restore_survived_the_live_writer(self) -> None:
+        self.assertEqual(
+            self.restore.returncode,
+            0,
+            f"restore aborted (duplicate-key race not contained):\n{self.restore.stderr}",
+        )
+
+    def test_primary_key_row_restored(self) -> None:
+        self.assertEqual(
+            self._scalar("SELECT count(*) FROM public.scheduler_stats WHERE id=1;"),
+            "1",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/e2e/test_e2e_seed_star_and_db_entries_backup_postgres.py

@@ -1,6 +1,8 @@
 import unittest
 
 from .helpers import (
+    POSTGRES_IMAGE,
+    POSTGRES_DATA_DIR,
     backup_path,
     cleanup_docker,
     create_minimal_compose_dir,
@@ -66,8 +68,8 @@ class TestE2ESeedStarAndDbEntriesBackupPostgres(unittest.TestCase):
                 "-e",
                 f"POSTGRES_PASSWORD={cls.pg_password}",
                 "-v",
-                f"{cls.db_volume}:/var/lib/postgresql/data",
+                f"{cls.db_volume}:{POSTGRES_DATA_DIR}",
-                "postgres:16-alpine",
+                POSTGRES_IMAGE,
             ]
         )
         wait_for_postgres(cls.pg_container, user="postgres", timeout_s=90)
@@ -165,10 +167,7 @@ class TestE2ESeedStarAndDbEntriesBackupPostgres(unittest.TestCase):
             "--database-containers",
             cls.pg_container,
             "--images-no-stop-required",
-            "alpine",
+            POSTGRES_IMAGE,
-            "postgres",
-            "mariadb",
-            "mysql",
             "--dump-only-sql",
             "--backups-dir",
             cls.backups_dir,

tests/e2e/test_e2e_snapshot.py

@@ -0,0 +1,141 @@
+"""Snapshot capture against real filesystems.
+
+Loop devices are only available to a privileged container, so each case builds
+its filesystem inside one and drives snapshot_driver.py there. A filesystem
+without snapshot support must fail loudly rather than degrade to a live copy,
+which is the property that makes the mode safe to offer at all.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import unittest
+from pathlib import Path
+
+from .helpers import require_docker, run, unique
+
+REPO_SRC = Path(__file__).resolve().parents[2] / "src"
+DRIVER = Path(__file__).resolve().parent / "snapshot_driver.py"
+IMAGE = "alpine:3.20"
+PACKAGES = "apk add -q btrfs-progs e2fsprogs zfs python3 util-linux"
+ATTACH = (
+    "LOOP=$(losetup -f | awk '{print $1}') "
+    '&& { [ -b "$LOOP" ] || mknod "$LOOP" b 7 "${LOOP#/dev/loop}"; }; '
+    'losetup "$LOOP" /img'
+)
+LOOP_FS = {
+    "btrfs": "btrfs subvolume create /subject/docker >/dev/null",
+    "ext4": "mkdir -p /subject/docker",
+}
+# A container carries no /lib/modules, so modprobe fails even on a loaded module.
+ZFS_READY = '{ [ -c /dev/zfs ] || modprobe zfs 2>/dev/null; }; [ -c /dev/zfs ]'
+
+
+def mount_script(fstype: str) -> str:
+    """Build a filesystem on a loop device and carve out the snapshot subject."""
+    if fstype == "zfs":
+        return (
+            f"{PACKAGES} && {ZFS_READY} && truncate -s 400M /img "
+            "&& zpool create -m none baudolo /img "
+            "&& zfs create -o mountpoint=/subject/docker baudolo/docker"
+        )
+    return (
+        f"{PACKAGES} && truncate -s 400M /img && mkfs.{fstype} -q /img "
+        f'&& mkdir -p /subject && {ATTACH} && mount -t {fstype} "$LOOP" /subject '
+        f"&& {LOOP_FS[fstype]}"
+    )
+
+
+def zfs_usable() -> bool:
+    """Whether this host's kernel can serve zfs to a privileged container."""
+    proc = run(
+        [
+            "docker", "run", "--rm", "--privileged", IMAGE, "sh", "-lc",
+            f"apk add -q zfs >/dev/null 2>&1 && {ZFS_READY}",
+        ],
+        capture=True,
+        check=False,
+    )
+    return proc.returncode == 0
+
+
+def required(fstype: str) -> bool:
+    """Whether this run must cover ``fstype`` instead of skipping it.
+
+    CI sets E2E_REQUIRE_FILESYSTEMS so a missing kernel module fails the build
+    rather than passing it with a filesystem silently untested.
+    """
+    demanded = os.environ.get("E2E_REQUIRE_FILESYSTEMS", "")
+    return fstype in demanded.replace(",", " ").split()
+
+
+def stage() -> Path:
+    """Copy source and driver under /tmp, the only path the DinD daemon shares."""
+    staged = Path("/tmp") / unique("baudolo-e2e-snapshot")
+    shutil.copytree(REPO_SRC, staged / "src")
+    shutil.copy(DRIVER, staged / "driver.py")
+    return staged
+
+
+def drive(fstype: str, kind: str, expect: str) -> str:
+    staged = stage()
+    script = (
+        f"set -e; {mount_script(fstype)}; "
+        f"python3 /driver.py {kind} /subject/docker {expect}"
+    )
+    try:
+        proc = run(
+            [
+                "docker", "run", "--rm", "--privileged",
+                "--name", staged.name,
+                "-v", f"{staged / 'src'}:/src:ro",
+                "-v", f"{staged / 'driver.py'}:/driver.py:ro",
+                IMAGE, "sh", "-lc", script,
+            ],
+            capture=True,
+            check=False,
+        )
+    finally:
+        shutil.rmtree(staged, ignore_errors=True)
+    if proc.returncode != 0:
+        raise AssertionError(f"{fstype}/{kind} driver failed:\n{proc.stdout}\n{proc.stderr}")
+    return proc.stdout
+
+
+class TestE2ESnapshot(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+
+    def assert_freezes(self, fstype: str) -> None:
+        output = drive(fstype, fstype, "supported")
+        self.assertIn("PASS the snapshot exposes the volume", output)
+        self.assertIn("PASS a later write does not reach the snapshot", output)
+        self.assertIn("PASS the snapshot is removed afterwards", output)
+        self.assertIn("ALL OK", output)
+
+    def test_btrfs_snapshot_freezes_the_volume(self) -> None:
+        self.assert_freezes("btrfs")
+
+    def test_zfs_snapshot_freezes_the_volume(self) -> None:
+        if not zfs_usable():
+            if required("zfs"):
+                self.fail(
+                    "E2E_REQUIRE_FILESYSTEMS demands zfs, but this kernel provides no "
+                    "zfs module; load it before running the suite"
+                )
+            self.skipTest("this kernel provides no zfs module, so no pool can be created")
+        self.assert_freezes("zfs")
+
+    def test_ext4_has_no_snapshot_and_says_so(self) -> None:
+        output = drive("ext4", "btrfs", "unsupported")
+        self.assertIn("PASS refused loudly", output)
+
+    def test_an_unknown_kind_is_refused_before_touching_the_filesystem(self) -> None:
+        output = drive("ext4", "lvm", "unsupported")
+        self.assertIn("PASS refused loudly", output)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/e2e/test_e2e_snapshot_db.py

@@ -0,0 +1,103 @@
+"""A live database survives being captured from a snapshot.
+
+The database is written to while the snapshot is taken and keeps writing
+afterwards, so the copy can only be a point in time - never a clean shutdown.
+A second server is then started on that copy: it must recover on its own and
+still hold every row committed before the snapshot.
+"""
+
+from __future__ import annotations
+
+import shutil
+import unittest
+from pathlib import Path
+
+from .helpers import require_docker, run, unique
+
+REPO_SRC = Path(__file__).resolve().parents[2] / "src"
+DRIVER = Path(__file__).resolve().parent / "snapshot_db_driver.py"
+IMAGE = "alpine:3.20"
+SOCKET = "/tmp/live.sock"
+RESTORED_SOCKET = "/tmp/restored.sock"
+DATADIR = "/subject/docker/volumes/mariadb_data/_data"
+RESTORED = "/restored"
+
+SCRIPT = f"""set -e
+apk add -q btrfs-progs util-linux python3 mariadb mariadb-client rsync
+truncate -s 900M /img
+mkfs.btrfs -q /img
+mkdir -p /subject
+LOOP=$(losetup -f | awk '{{print $1}}')
+{{ [ -b "$LOOP" ] || mknod "$LOOP" b 7 "${{LOOP#/dev/loop}}"; }}
+losetup "$LOOP" /img
+mount -t btrfs "$LOOP" /subject
+btrfs subvolume create /subject/docker >/dev/null
+mkdir -p {DATADIR}
+
+mariadb-install-db --user=root --datadir={DATADIR} >/dev/null 2>&1
+mariadbd --user=root --datadir={DATADIR} --socket={SOCKET} --skip-networking &
+for i in $(seq 1 60); do mariadb-admin --socket={SOCKET} ping >/dev/null 2>&1 && break; sleep 1; done
+
+mariadb --socket={SOCKET} -e "CREATE DATABASE demo;
+  CREATE TABLE demo.t (id INT PRIMARY KEY, v VARCHAR(32)) ENGINE=InnoDB;
+  INSERT INTO demo.t VALUES (1,'committed'),(2,'committed');"
+
+mariadb --socket={SOCKET} -e "INSERT INTO demo.t VALUES (3,'committed');"
+python3 /driver.py
+mariadb --socket={SOCKET} -e "INSERT INTO demo.t VALUES (4,'after-snapshot');"
+
+mkdir -p {RESTORED}
+rsync -a /backups/20260731/mariadb_data/files/ {RESTORED}/
+mariadbd --user=root --datadir={RESTORED} --socket={RESTORED_SOCKET} --skip-networking &
+for i in $(seq 1 60); do mariadb-admin --socket={RESTORED_SOCKET} ping >/dev/null 2>&1 && break; sleep 1; done
+
+echo "RESTORED_ROWS=$(mariadb --socket={RESTORED_SOCKET} -N -B -e 'SELECT COUNT(*) FROM demo.t;')"
+echo "RESTORED_AFTER=$(mariadb --socket={RESTORED_SOCKET} -N -B -e \\
+  "SELECT COUNT(*) FROM demo.t WHERE v='after-snapshot';")"
+echo DB_OK
+"""
+
+
+class TestE2ESnapshotDatabase(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+        staged = Path("/tmp") / unique("baudolo-e2e-snapshot-db")
+        shutil.copytree(REPO_SRC, staged / "src")
+        shutil.copy(DRIVER, staged / "driver.py")
+        try:
+            proc = run(
+                [
+                    "docker", "run", "--rm", "--privileged",
+                    "--name", staged.name,
+                    "-v", f"{staged / 'src'}:/src:ro",
+                    "-v", f"{staged / 'driver.py'}:/driver.py:ro",
+                    IMAGE, "sh", "-lc", SCRIPT,
+                ],
+                capture=True,
+                check=False,
+            )
+        finally:
+            shutil.rmtree(staged, ignore_errors=True)
+        cls.output = proc.stdout + proc.stderr
+        cls.returncode = proc.returncode
+
+    def test_the_run_completed(self) -> None:
+        self.assertEqual(self.returncode, 0, self.output)
+        self.assertIn("DB_OK", self.output)
+
+    def test_the_backup_came_from_the_snapshot(self) -> None:
+        self.assertIn("SNAPSHOT COPY DONE", self.output)
+
+    def test_the_restored_server_recovered_on_its_own(self) -> None:
+        self.assertIn("RESTORED_ROWS=3", self.output)
+
+    def test_writes_after_the_snapshot_are_absent(self) -> None:
+        self.assertIn("RESTORED_AFTER=0", self.output)
+
+    def test_the_copy_was_an_unclean_state_the_engine_had_to_repair(self) -> None:
+        self.assertRegex(self.output, r"(?i)crash recovery|rolling back|log sequence")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/backup/compose_fixture.py

@@ -0,0 +1,44 @@
+"""Builds the on-disk compose directories the compose tests discover."""
+
+from __future__ import annotations
+
+import shutil
+from pathlib import Path
+
+
+def touch(p: Path) -> None:
+    p.parent.mkdir(parents=True, exist_ok=True)
+
+    # ".env/env" leaves ".env" behind as a directory, which blocks a later ".env" file.
+    if p.exists() and p.is_dir():
+        shutil.rmtree(p)
+
+    p.write_text("x", encoding="utf-8")
+
+
+def setup_compose_dir(
+    tmp_path: Path,
+    name: str = "mailu",
+    *,
+    compose_name: str = "docker-compose.yml",
+    with_override: bool = False,
+    with_ca_override: bool = False,
+    env_layout: str | None = None,  # None | ".env" | ".env/env"
+) -> Path:
+    d = tmp_path / name
+    d.mkdir(parents=True, exist_ok=True)
+
+    touch(d / compose_name)
+
+    if with_override:
+        touch(d / "docker-compose.override.yml")
+
+    if with_ca_override:
+        touch(d / "docker-compose.ca.override.yml")
+
+    if env_layout == ".env":
+        touch(d / ".env")
+    elif env_layout == ".env/env":
+        touch(d / ".env" / "env")
+
+    return d

tests/unit/backup/test_app_databases_csv.py

@@ -7,7 +7,7 @@ from contextlib import redirect_stderr
 import pandas as pd
 
 # Adjust if your package name/import path differs.
-from baudolo.backup.app import _load_databases_df
+from baudolo.backup.dumps import load_databases_df
 
 
 EXPECTED_COLUMNS = ["instance", "database", "username", "password"]
@@ -20,7 +20,7 @@ class TestLoadDatabasesDf(unittest.TestCase):
 
             buf = io.StringIO()
             with redirect_stderr(buf):
-                df = _load_databases_df(missing_path)
+                df = load_databases_df(missing_path)
 
             stderr = buf.getvalue()
             self.assertIn("WARNING:", stderr)
@@ -39,7 +39,7 @@ class TestLoadDatabasesDf(unittest.TestCase):
 
             buf = io.StringIO()
             with redirect_stderr(buf):
-                df = _load_databases_df(empty_path)
+                df = load_databases_df(empty_path)
 
             stderr = buf.getvalue()
             self.assertIn("WARNING:", stderr)
@@ -59,7 +59,7 @@ class TestLoadDatabasesDf(unittest.TestCase):
 
             buf = io.StringIO()
             with redirect_stderr(buf):
-                df = _load_databases_df(csv_path)
+                df = load_databases_df(csv_path)
 
             stderr = buf.getvalue()
             self.assertEqual(stderr, "")  # no warning expected

tests/unit/backup/test_app_snapshot.py

@@ -0,0 +1,79 @@
+"""Contract of app.main's snapshot branch - the caller that runs in production."""
+
+from __future__ import annotations
+
+import unittest
+from unittest import mock
+
+from baudolo.backup import app
+from baudolo.backup.snapshot import volume_snapshot
+
+
+def stubbed_snapshot(kind: str, subject: str, tag: str):
+    return volume_snapshot(kind, subject, tag, run=lambda command: [])
+
+ARGV = [
+    "baudolo",
+    "--compose-dir",
+    "/compose",
+    "--backups-dir",
+    "/backups",
+    "--snapshot",
+    "btrfs",
+    "--snapshot-subject",
+    "/var/lib/docker",
+]
+
+
+def drive(*, present: bool) -> list[dict]:
+    calls: list[dict] = []
+
+    def record(versions_dir, volume_name, volume_dir, *, authoritative, source):
+        calls.append(
+            {"volume": volume_name, "authoritative": authoritative, "source": source}
+        )
+
+    with (
+        mock.patch("sys.argv", ARGV),
+        mock.patch.object(app, "get_machine_id", return_value="machine"),
+        mock.patch.object(app, "create_version_directory", return_value="/gen"),
+        mock.patch.object(app, "create_volume_directory", return_value="/gen/vol"),
+        mock.patch.object(app, "load_databases_df", return_value=None),
+        mock.patch.object(app, "docker_volume_names", return_value=["vol"]),
+        mock.patch.object(app, "containers_using_volume", return_value=[]),
+        mock.patch.object(app, "volume_is_fully_ignored", return_value=False),
+        mock.patch.object(app, "backup_dumps_for_volume", return_value=(False, False)),
+        mock.patch.object(
+            app, "get_storage_path", return_value="/var/lib/docker/volumes/vol/_data/"
+        ),
+        mock.patch.object(app, "stamp_directory"),
+        mock.patch.object(app, "handle_docker_compose_services"),
+        mock.patch.object(app.os.path, "isdir", return_value=present),
+        mock.patch.object(app, "backup_volume", side_effect=record),
+        mock.patch.object(app, "volume_snapshot", stubbed_snapshot),
+    ):
+        app.main()
+    return calls
+
+
+class TestSnapshotBranch(unittest.TestCase):
+    def test_it_passes_a_path_ending_in_a_separator(self) -> None:
+        source = drive(present=True)[0]["source"]
+        self.assertTrue(source.endswith("/volumes/vol/_data/"), source)
+        self.assertNotIn("/var/lib/docker/volumes", source)
+
+    def test_it_reads_from_the_snapshot_and_not_from_the_live_tree(self) -> None:
+        source = drive(present=True)[0]["source"]
+        self.assertTrue(source.startswith("/var/lib/.baudolo-"), source)
+
+    def test_it_compares_by_content_against_the_previous_generation(self) -> None:
+        self.assertTrue(drive(present=True)[0]["authoritative"])
+
+    def test_a_volume_missing_from_the_snapshot_is_copied_live(self) -> None:
+        call = drive(present=False)[0]
+        self.assertEqual(call["source"], "/var/lib/docker/volumes/vol/_data/")
+        self.assertFalse(call["authoritative"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/backup/test_cli.py

@@ -0,0 +1,81 @@
+"""Contract of the backup CLI, in particular the snapshot flag pairing."""
+
+from __future__ import annotations
+
+import unittest
+from unittest import mock
+
+from baudolo.backup.cli import parse_args
+
+REQUIRED = ["--compose-dir", "/compose", "--backups-dir", "/backups"]
+
+
+def parse(*extra: str):
+    with mock.patch("sys.argv", ["baudolo", *REQUIRED, *extra]):
+        return parse_args()
+
+
+class TestSnapshotFlags(unittest.TestCase):
+    def test_no_snapshot_by_default(self) -> None:
+        args = parse()
+        self.assertIsNone(args.snapshot)
+        self.assertIsNone(args.snapshot_subject)
+
+    def test_both_flags_together_are_accepted(self) -> None:
+        args = parse("--snapshot", "btrfs", "--snapshot-subject", "/var/lib/docker")
+        self.assertEqual(args.snapshot, "btrfs")
+        self.assertEqual(args.snapshot_subject, "/var/lib/docker")
+
+    def test_the_kind_alone_is_rejected(self) -> None:
+        with self.assertRaises(SystemExit):
+            parse("--snapshot", "btrfs")
+
+    def test_the_subject_alone_is_rejected(self) -> None:
+        with self.assertRaises(SystemExit):
+            parse("--snapshot-subject", "/var/lib/docker")
+
+    def test_an_unsupported_kind_is_rejected(self) -> None:
+        with self.assertRaises(SystemExit):
+            parse("--snapshot", "ext4", "--snapshot-subject", "/var/lib/docker")
+
+    def test_zfs_is_accepted(self) -> None:
+        self.assertEqual(parse("--snapshot", "zfs", "--snapshot-subject", "/d").snapshot, "zfs")
+
+    def test_shutdown_is_rejected_because_nothing_is_stopped(self) -> None:
+        with self.assertRaises(SystemExit):
+            parse("--snapshot", "btrfs", "--snapshot-subject", "/d", "--shutdown")
+
+    def test_shutdown_stays_available_without_a_snapshot(self) -> None:
+        self.assertTrue(parse("--shutdown").shutdown)
+
+    def test_hard_restart_is_rejected_because_nothing_is_stopped(self) -> None:
+        with self.assertRaises(SystemExit):
+            parse(
+                "--snapshot",
+                "btrfs",
+                "--snapshot-subject",
+                "/d",
+                "--hard-restart-projects",
+                "mailu",
+            )
+
+    def test_hard_restart_stays_available_without_a_snapshot(self) -> None:
+        self.assertEqual(
+            parse("--hard-restart-projects", "mailu").hard_restart_projects, ["mailu"]
+        )
+
+
+class TestRequiredFlags(unittest.TestCase):
+    def test_backups_dir_is_required(self) -> None:
+        with mock.patch("sys.argv", ["baudolo", "--compose-dir", "/compose"]):
+            with self.assertRaises(SystemExit):
+                parse_args()
+
+    def test_compose_dir_is_required(self) -> None:
+        with mock.patch("sys.argv", ["baudolo", "--backups-dir", "/backups"]):
+            with self.assertRaises(SystemExit):
+                parse_args()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/backup/test_compose.py

@@ -1,50 +1,12 @@
 from __future__ import annotations
 
-import shutil
 import tempfile
 import unittest
 from pathlib import Path
 from typing import List
 from unittest.mock import patch
 
-
+from .compose_fixture import setup_compose_dir as _setup_compose_dir
-def _touch(p: Path) -> None:
-    p.parent.mkdir(parents=True, exist_ok=True)
-
-    # If the path already exists as a directory (e.g. ".env" created by ".env/env"),
-    # remove it so we can create a file with the same name.
-    if p.exists() and p.is_dir():
-        shutil.rmtree(p)
-
-    p.write_text("x", encoding="utf-8")
-
-
-def _setup_compose_dir(
-    tmp_path: Path,
-    name: str = "mailu",
-    *,
-    compose_name: str = "docker-compose.yml",
-    with_override: bool = False,
-    with_ca_override: bool = False,
-    env_layout: str | None = None,  # None | ".env" | ".env/env"
-) -> Path:
-    d = tmp_path / name
-    d.mkdir(parents=True, exist_ok=True)
-
-    _touch(d / compose_name)
-
-    if with_override:
-        _touch(d / "docker-compose.override.yml")
-
-    if with_ca_override:
-        _touch(d / "docker-compose.ca.override.yml")
-
-    if env_layout == ".env":
-        _touch(d / ".env")
-    elif env_layout == ".env/env":
-        _touch(d / ".env" / "env")
-
-    return d
 
 
 class TestCompose(unittest.TestCase):
@@ -249,7 +211,3 @@ class TestCompose(unittest.TestCase):
                     ),
                 ],
             )
-
-
-if __name__ == "__main__":
-    unittest.main(verbosity=2)

tests/unit/backup/test_compose_hard_restart.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import unittest
+from typing import List
+from unittest.mock import patch
+
+
+class HardRestartArgTests(unittest.TestCase):
+    """The hard-restart list defaults to empty (no compose down/up); callers
+    opt in per dir, e.g. compose hosts pass 'mailu' while swarm hosts, where
+    the dir is a stack whose overlay network collides with compose up, pass
+    nothing."""
+
+    def _parse(self, extra: List[str]):
+        import sys
+
+        from baudolo.backup import cli
+
+        argv = [
+            "baudolo",
+            "--compose-dir",
+            "/tmp",
+            "--backups-dir",
+            "/tmp/backup",
+            "--database-containers",
+            "postgres",
+            "--images-no-stop-required",
+            "redis",
+            *extra,
+        ]
+        with patch.object(sys, "argv", argv):
+            return cli.parse_args()
+
+    def test_default_is_empty(self) -> None:
+        args = self._parse([])
+        self.assertEqual(args.hard_restart_projects, [])
+
+    def test_empty_flag_stays_empty(self) -> None:
+        args = self._parse(["--hard-restart-projects"])
+        self.assertEqual(args.hard_restart_projects, [])
+
+    def test_explicit_names_preserved(self) -> None:
+        args = self._parse(["--hard-restart-projects", "mailu", "foo"])
+        self.assertEqual(args.hard_restart_projects, ["mailu", "foo"])
+
+    def test_backups_dir_is_required(self) -> None:
+        import sys
+
+        from baudolo.backup import cli
+
+        argv = [
+            "baudolo",
+            "--compose-dir",
+            "/tmp",
+            "--database-containers",
+            "postgres",
+            "--images-no-stop-required",
+            "redis",
+        ]
+        with patch.object(sys, "argv", argv), self.assertRaises(SystemExit):
+            cli.parse_args()
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

tests/unit/backup/test_layout.py

@@ -0,0 +1,51 @@
+"""Contract of where a backup run puts its directories."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from baudolo.backup import layout as mod
+
+
+class TestVersionDirectory(unittest.TestCase):
+    def test_it_creates_the_generation_directory(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            created = mod.create_version_directory(tmp, "20260731020304")
+            self.assertTrue(Path(created).is_dir())
+            self.assertEqual(Path(created).name, "20260731020304")
+
+    def test_it_is_idempotent(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            first = mod.create_version_directory(tmp, "20260731")
+            second = mod.create_version_directory(tmp, "20260731")
+            self.assertEqual(first, second)
+
+    def test_it_creates_missing_parents(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            nested = str(Path(tmp) / "machine" / "repo")
+            created = mod.create_version_directory(nested, "20260731")
+            self.assertTrue(Path(created).is_dir())
+
+
+class TestVolumeDirectory(unittest.TestCase):
+    def test_it_nests_the_volume_under_the_generation(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            created = mod.create_volume_directory(tmp, "postgres_data")
+            self.assertEqual(Path(created).parent, Path(tmp))
+            self.assertTrue(Path(created).is_dir())
+
+
+class TestMachineId(unittest.TestCase):
+    def test_it_takes_the_hash_without_the_filename(self) -> None:
+        digest = "a" * 64
+        with mock.patch.object(
+            mod, "execute_shell_command", return_value=[f"{digest}  /etc/machine-id"]
+        ):
+            self.assertEqual(mod.get_machine_id(), digest)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/backup/test_policy.py

@@ -0,0 +1,68 @@
+"""Contract of the rules deciding what is backed up and what must stop."""
+
+from __future__ import annotations
+
+import unittest
+from unittest import mock
+
+from baudolo.backup import policy as mod
+
+
+class TestIsImageIgnored(unittest.TestCase):
+    def test_an_empty_whitelist_ignores_nothing(self) -> None:
+        self.assertFalse(mod.is_image_ignored("c1", []))
+
+    def test_a_listed_image_is_ignored(self) -> None:
+        with mock.patch.object(mod, "get_image_info", return_value="alpine:3.20"):
+            self.assertTrue(mod.is_image_ignored("c1", ["alpine:3.20"]))
+
+    def test_matching_is_exact(self) -> None:
+        with mock.patch.object(mod, "get_image_info", return_value="alpine:3.21"):
+            self.assertFalse(mod.is_image_ignored("c1", ["alpine:3.20"]))
+
+
+class TestVolumeIsFullyIgnored(unittest.TestCase):
+    def test_a_volume_without_containers_is_kept(self) -> None:
+        self.assertFalse(mod.volume_is_fully_ignored([], ["alpine:3.20"]))
+
+    def test_it_needs_every_container_to_be_ignored(self) -> None:
+        with mock.patch.object(mod, "get_image_info", side_effect=["a", "b"]):
+            self.assertFalse(mod.volume_is_fully_ignored(["c1", "c2"], ["a"]))
+
+    def test_all_ignored_skips_the_volume(self) -> None:
+        with mock.patch.object(mod, "get_image_info", side_effect=["a", "a"]):
+            self.assertTrue(mod.volume_is_fully_ignored(["c1", "c2"], ["a"]))
+
+
+class TestRequiresStop(unittest.TestCase):
+    def test_no_containers_means_no_stop(self) -> None:
+        self.assertFalse(mod.requires_stop([], []))
+
+    def test_a_swarm_task_never_forces_a_stop(self) -> None:
+        with mock.patch.object(mod, "is_swarm_task", return_value=True):
+            self.assertFalse(mod.requires_stop(["c1"], []))
+
+    def test_a_whitelisted_image_does_not_force_a_stop(self) -> None:
+        with (
+            mock.patch.object(mod, "is_swarm_task", return_value=False),
+            mock.patch.object(mod, "get_image_info", return_value="alpine:3.20"),
+        ):
+            self.assertFalse(mod.requires_stop(["c1"], ["alpine:3.20"]))
+
+    def test_an_unlisted_image_forces_a_stop(self) -> None:
+        with (
+            mock.patch.object(mod, "is_swarm_task", return_value=False),
+            mock.patch.object(mod, "get_image_info", return_value="postgres:17"),
+        ):
+            self.assertTrue(mod.requires_stop(["c1"], ["alpine:3.20"]))
+
+    def test_one_unlisted_container_is_enough(self) -> None:
+        with (
+            mock.patch.object(mod, "is_swarm_task", return_value=False),
+            mock.patch.object(mod, "get_image_info", side_effect=["alpine:3.20", "x"]),
+        ):
+            self.assertTrue(mod.requires_stop(["c1", "c2"], ["alpine:3.20"]))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/backup/test_snapshot.py

@@ -0,0 +1,134 @@
+"""Contract of the filesystem snapshot used to capture volumes atomically."""
+
+from __future__ import annotations
+
+import unittest
+
+from baudolo.backup.shell import BackupException
+from baudolo.backup.snapshot import SnapshotError, volume_snapshot
+
+
+class Runner:
+    def __init__(self, replies: dict[str, list[str]] | None = None) -> None:
+        self.calls: list[str] = []
+        self.replies = replies or {}
+
+    def __call__(self, command: str) -> list[str]:
+        self.calls.append(command)
+        for prefix, reply in self.replies.items():
+            if command.startswith(prefix):
+                return reply
+        return []
+
+
+class TestBtrfs(unittest.TestCase):
+    def test_it_creates_a_read_only_snapshot_beside_the_subject(self) -> None:
+        run = Runner()
+        with volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=run):
+            pass
+        self.assertEqual(
+            run.calls[0],
+            "btrfs subvolume snapshot -r /var/lib/docker /var/lib/.baudolo-20260731",
+        )
+
+    def test_it_removes_the_snapshot_afterwards(self) -> None:
+        run = Runner()
+        with volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=run):
+            pass
+        self.assertEqual(run.calls[-1], "btrfs subvolume delete /var/lib/.baudolo-20260731")
+
+    def test_it_maps_a_volume_path_into_the_snapshot(self) -> None:
+        run = Runner()
+        with volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=run) as resolve:
+            self.assertEqual(
+                resolve("/var/lib/docker/volumes/postgres_data/_data"),
+                "/var/lib/.baudolo-20260731/volumes/postgres_data/_data",
+            )
+
+    def test_it_keeps_the_trailing_slash_rsync_reads_as_contents(self) -> None:
+        run = Runner()
+        with volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=run) as resolve:
+            self.assertEqual(
+                resolve("/var/lib/docker/volumes/postgres_data/_data/"),
+                "/var/lib/.baudolo-20260731/volumes/postgres_data/_data/",
+            )
+
+    def test_it_removes_the_snapshot_even_when_the_body_raises(self) -> None:
+        run = Runner()
+        with self.assertRaises(ZeroDivisionError):
+            with volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=run):
+                raise ZeroDivisionError
+        self.assertTrue(run.calls[-1].startswith("btrfs subvolume delete"))
+
+
+class TestZfs(unittest.TestCase):
+    def _run(self) -> Runner:
+        return Runner({"zfs list": ["tank/docker"]})
+
+    def test_it_snapshots_the_dataset_mounted_at_the_subject(self) -> None:
+        run = self._run()
+        with volume_snapshot("zfs", "/var/lib/docker", "20260731", run=run):
+            pass
+        self.assertIn("zfs snapshot tank/docker@baudolo-20260731", run.calls)
+
+    def test_it_destroys_the_snapshot_afterwards(self) -> None:
+        run = self._run()
+        with volume_snapshot("zfs", "/var/lib/docker", "20260731", run=run):
+            pass
+        self.assertEqual(run.calls[-1], "zfs destroy tank/docker@baudolo-20260731")
+
+    def test_it_maps_a_volume_path_through_the_dot_zfs_directory(self) -> None:
+        run = self._run()
+        with volume_snapshot("zfs", "/var/lib/docker", "20260731", run=run) as resolve:
+            self.assertEqual(
+                resolve("/var/lib/docker/volumes/postgres_data/_data"),
+                "/var/lib/docker/.zfs/snapshot/baudolo-20260731/volumes/postgres_data/_data",
+            )
+
+    def test_an_unmounted_dataset_is_an_error(self) -> None:
+        run = Runner({"zfs list": [""]})
+        with self.assertRaises(SnapshotError):
+            with volume_snapshot("zfs", "/var/lib/docker", "20260731", run=run):
+                pass
+
+
+class TestRejections(unittest.TestCase):
+    def test_an_unknown_kind_is_rejected(self) -> None:
+        run = Runner()
+        with self.assertRaises(SnapshotError):
+            with volume_snapshot("ext4", "/var/lib/docker", "20260731", run=run):
+                pass
+        self.assertEqual(run.calls, [])
+
+    def test_a_path_outside_the_subject_is_rejected(self) -> None:
+        run = Runner()
+        with volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=run) as resolve:
+            with self.assertRaises(SnapshotError):
+                resolve("/etc/passwd")
+
+    def test_the_subject_itself_resolves_to_the_snapshot_root(self) -> None:
+        run = Runner()
+        with volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=run) as resolve:
+            self.assertEqual(resolve("/var/lib/docker"), "/var/lib/.baudolo-20260731")
+
+
+class Busy(Runner):
+    def __call__(self, command: str) -> list[str]:
+        if command.startswith("btrfs subvolume delete"):
+            raise BackupException("target is busy")
+        return super().__call__(command)
+
+
+class TestRemovalFailure(unittest.TestCase):
+    def test_a_failed_removal_does_not_fail_a_completed_run(self) -> None:
+        with volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=Busy()):
+            pass
+
+    def test_a_failed_removal_does_not_mask_the_body(self) -> None:
+        with self.assertRaises(ZeroDivisionError):
+            with volume_snapshot("btrfs", "/var/lib/docker", "20260731", run=Busy()):
+                raise ZeroDivisionError
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/backup/test_volume.py

@@ -0,0 +1,87 @@
+"""Contract of the rsync invocation that copies a volume."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from baudolo.backup import volume as mod
+
+
+class TestBackupVolume(unittest.TestCase):
+    def copy(self, **kwargs) -> str:
+        with tempfile.TemporaryDirectory() as tmp:
+            defaults = {
+                "versions_dir": tmp,
+                "volume_name": "demo",
+                "volume_dir": str(Path(tmp) / "gen" / "demo"),
+                "authoritative": False,
+                "source": "/var/lib/docker/volumes/demo/_data/",
+            }
+            defaults.update(kwargs)
+            with mock.patch.object(mod, "execute_shell_command") as run:
+                mod.backup_volume(
+                    defaults.pop("versions_dir"),
+                    defaults.pop("volume_name"),
+                    defaults.pop("volume_dir"),
+                    **defaults,
+                )
+            return run.call_args[0][0]
+
+    def test_the_quick_check_pass_carries_no_checksum(self) -> None:
+        self.assertNotIn("--checksum", self.copy(authoritative=False))
+
+    def test_the_authoritative_pass_compares_by_content(self) -> None:
+        self.assertIn("--checksum", self.copy(authoritative=True))
+
+    def test_it_reads_from_the_given_source(self) -> None:
+        command = self.copy(source="/snapshot/volumes/demo/_data/")
+        self.assertIn("/snapshot/volumes/demo/_data/", command)
+
+    def test_it_always_deletes_what_the_source_no_longer_has(self) -> None:
+        self.assertIn("--delete", self.copy())
+
+    def test_it_creates_the_destination(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            dest = Path(tmp) / "gen" / "demo"
+            with mock.patch.object(mod, "execute_shell_command"):
+                mod.backup_volume(
+                    tmp, "demo", str(dest), authoritative=False, source="/src/"
+                )
+            self.assertTrue((dest / "files").is_dir())
+
+    def test_source_is_required(self) -> None:
+        with self.assertRaises(TypeError):
+            mod.backup_volume("/v", "demo", "/d", authoritative=False)
+
+    def test_authoritative_is_required(self) -> None:
+        with self.assertRaises(TypeError):
+            mod.backup_volume("/v", "demo", "/d", source="/src/")
+
+
+class TestLastBackupDir(unittest.TestCase):
+    def test_it_ignores_the_generation_being_written(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            current = Path(tmp) / "20260101" / "demo" / "files"
+            current.mkdir(parents=True)
+            found = mod.get_last_backup_dir(tmp, "demo", str(current) + "/")
+            self.assertIsNone(found)
+
+    def test_it_finds_the_previous_generation(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            older = Path(tmp) / "20260101" / "demo" / "files"
+            older.mkdir(parents=True)
+            current = Path(tmp) / "20260102" / "demo" / "files"
+            current.mkdir(parents=True)
+            found = mod.get_last_backup_dir(tmp, "demo", str(current) + "/")
+            self.assertEqual(found, str(older) + "/")
+
+    def test_a_first_run_has_no_predecessor(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            self.assertIsNone(mod.get_last_backup_dir(tmp, "demo", f"{tmp}/x/"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/restore/test_postgres_empty_drop.py

@@ -1,43 +0,0 @@
-"""The --empty pre-clean drop must cover every object class a dump can
-re-CREATE ahead of tables; a missed class aborts the replay under
-ON_ERROR_STOP (OpenProject's ICU collation public.versions_name did)."""
-
-import tempfile
-import unittest
-from pathlib import Path
-from unittest import mock
-
-from baudolo.restore.db import postgres
-
-
-class TestPostgresEmptyDrop(unittest.TestCase):
-    def _drop_sql(self) -> str:
-        with tempfile.NamedTemporaryFile(suffix=".sql") as fh:
-            Path(fh.name).write_bytes(b"SELECT 1;\n")
-            with mock.patch.object(postgres, "docker_exec") as run:
-                postgres.restore_postgres_sql(
-                    container="c",
-                    db_name="db",
-                    user="u",
-                    password="p",
-                    sql_path=fh.name,
-                    empty=True,
-                )
-        first_call = run.call_args_list[0]
-        return first_call.kwargs["stdin"].decode()
-
-    def test_drop_covers_collations(self) -> None:
-        sql = self._drop_sql()
-        self.assertIn("pg_collation", sql)
-        self.assertIn("'COLLATION' AS type", sql)
-        self.assertIn("pg_get_userbyid(col.collowner) = current_user", sql)
-
-    def test_drop_still_covers_the_other_classes(self) -> None:
-        sql = self._drop_sql()
-        for marker in ("pg_class", "pg_proc", "'SEQUENCE' AS type", "'TYPE' AS type"):
-            self.assertIn(marker, sql)
-        self.assertIn("DROP %s IF EXISTS public.%I CASCADE", sql)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/restore/test_postgres_single_transaction.py

@@ -0,0 +1,44 @@
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+from baudolo.restore.db import postgres as pg_mod
+
+
+class TestPostgresSingleTransaction(unittest.TestCase):
+    def test_replay_is_single_transaction_but_preclean_is_not(self) -> None:
+        calls = []
+
+        def _capture(container, argv, **kwargs):
+            calls.append(argv)
+            return MagicMock()
+
+        with tempfile.NamedTemporaryFile(suffix=".sql") as sql:
+            sql.write(b"CREATE TABLE t (id int);\nINSERT INTO t VALUES (1);\n")
+            sql.flush()
+            with patch.object(pg_mod, "docker_exec", side_effect=_capture):
+                pg_mod.restore_postgres_sql(
+                    container="db",
+                    db_name="discourse",
+                    user="discourse",
+                    password="pw",
+                    sql_path=sql.name,
+                    empty=True,
+                )
+
+        self.assertEqual(len(calls), 2, f"expected pre-clean + replay: {calls}")
+        preclean, replay = calls[0], calls[1]
+        self.assertNotIn(
+            "--single-transaction",
+            preclean,
+            "pre-clean must stay multi-statement or it exhausts max_locks on large schemas",
+        )
+        self.assertIn(
+            "--single-transaction",
+            replay,
+            "dump replay must be atomic so a live concurrent writer cannot trip a duplicate-key abort",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_backup.py

@@ -1,25 +1,24 @@
 import unittest
 from unittest.mock import patch
 
-from baudolo.backup.app import requires_stop
+from baudolo.backup.policy import requires_stop
 
 
-@patch("baudolo.backup.app.is_swarm_task", return_value=False)
+@patch("baudolo.backup.policy.is_swarm_task", return_value=False)
 class TestRequiresStop(unittest.TestCase):
-    @patch("baudolo.backup.app.get_image_info")
+    @patch("baudolo.backup.policy.get_image_info")
     def test_requires_stop_false_when_all_images_are_whitelisted(
         self, mock_get_image_info, _mock_is_swarm_task
     ):
-        # All containers use images containing allowed substrings
         mock_get_image_info.side_effect = [
             "repo/mastodon:v4",
             "repo/wordpress:latest",
         ]
         containers = ["c1", "c2"]
-        whitelist = ["mastodon", "wordpress"]
+        whitelist = ["repo/mastodon:v4", "repo/wordpress:latest"]
         self.assertFalse(requires_stop(containers, whitelist))
 
-    @patch("baudolo.backup.app.get_image_info")
+    @patch("baudolo.backup.policy.get_image_info")
     def test_requires_stop_true_when_any_image_is_not_whitelisted(
         self, mock_get_image_info, _mock_is_swarm_task
     ):
@@ -28,10 +27,18 @@ class TestRequiresStop(unittest.TestCase):
             "repo/nginx:latest",
         ]
         containers = ["c1", "c2"]
-        whitelist = ["mastodon", "wordpress"]
+        whitelist = ["repo/mastodon:v4", "repo/wordpress:latest"]
         self.assertTrue(requires_stop(containers, whitelist))
 
-    @patch("baudolo.backup.app.get_image_info")
+    @patch("baudolo.backup.policy.get_image_info")
+    def test_requires_stop_true_on_substring_only_match(
+        self, mock_get_image_info, _mock_is_swarm_task
+    ):
+        mock_get_image_info.return_value = "reg:5000/repo/mastodon:v4"
+        self.assertTrue(requires_stop(["c1"], ["mastodon"]))
+        self.assertTrue(requires_stop(["c1"], ["repo/mastodon:v4"]))
+
+    @patch("baudolo.backup.policy.get_image_info")
     def test_requires_stop_true_when_whitelist_empty(
         self, mock_get_image_info, _mock_is_swarm_task
     ):
@@ -40,8 +47,8 @@ class TestRequiresStop(unittest.TestCase):
 
 
 class TestRequiresStopSwarm(unittest.TestCase):
-    @patch("baudolo.backup.app.get_image_info")
+    @patch("baudolo.backup.policy.get_image_info")
-    @patch("baudolo.backup.app.is_swarm_task", return_value=True)
+    @patch("baudolo.backup.policy.is_swarm_task", return_value=True)
     def test_swarm_tasks_never_require_stop(
         self, _mock_is_swarm_task, mock_get_image_info
     ):