Release version 3.2.0

Backing up a live volume with rsync copies a moving target: a database
written to mid-copy lands on disk in a state no engine ever committed.
Stopping the container avoids that at the cost of downtime.

A snapshot removes both. `--snapshot {btrfs,zfs}` with `--snapshot-subject`
freezes the docker root once per run, and every volume copy is then read
from that frozen tree while the containers keep serving. A restore of such
a copy is an ordinary crash recovery, which every supported engine performs
on its own at startup.

An unsupported filesystem or an unknown snapshot kind fails loudly rather
than degrading to a live copy, since a silent fallback would return exactly
the torn backup the mode exists to prevent. `--shutdown` is rejected
alongside `--snapshot` instead of being ignored: under a snapshot no
container is ever stopped, so accepting the flag would promise downtime
semantics the run does not deliver.

Copies out of a snapshot skip rsync's --checksum verification. The source
is immutable for the lifetime of the copy, so size-and-mtime cannot race,
and dropping the second full read roughly halves the I/O per volume.

backup/app.py grew past what one module could carry and is split into
layout, policy and dumps along the lines it already had internally.

Tests: unit coverage for the new snapshot, layout, policy, volume and cli
units; e2e cases drive real btrfs, zfs and ext4 filesystems on loop devices
in a privileged container, including a MariaDB that is written to across
the snapshot and must recover from the restored copy without losing a
committed row. CI installs zfs and sets E2E_REQUIRE_FILESYSTEMS so a
missing kernel module fails the build instead of silently skipping a
filesystem.

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,121 @@
 # Changelog
 
+## [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

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 = "3.0.0"
+version = "3.2.0"
 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

@@ -37,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)"
 
@@ -194,6 +197,8 @@ else
   docker run --rm \
     --network "${NET}" \
     -e DOCKER_HOST="${DIND_HOST_IN_NET}" \
+    -e E2E_TEST_PATTERN="${TEST_PATTERN}" \
+    -e E2E_REQUIRE_FILESYSTEMS="${E2E_REQUIRE_FILESYSTEMS:-}" \
     -v "${DIND_VOL}:/var/lib/docker" \
     -v "${E2E_TMP_VOL}:/tmp" \
     "${IMG}" \
@@ -209,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,
-
+)
-
+from .policy import requires_stop, volume_is_fully_ignored
-def stamp_directory(version_dir: str) -> None:
+from .snapshot import volume_snapshot
-    """
+from .volume import backup_volume, get_storage_path
-    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 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
-
-
-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
 
 
 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,32 +80,41 @@ 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:
+                copy(authoritative=False, source=resolve_source(live_source))
+                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)
 

src/baudolo/backup/cli.py

@@ -39,16 +39,26 @@ def parse_args() -> argparse.Namespace:
         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="Exact image references (repo:tag, incl. any registry prefix) whose containers must not be stopped during file backup",
     )
     p.add_argument(
@@ -79,4 +89,9 @@ 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")
+    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,86 @@
+"""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 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}")
+        return os.path.join(root, relative) if relative != "." else root
+
+    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.
+    """
+    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:
+        run(remove)

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/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,
         )

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,19 +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
-
-# SPOT for the database images and their in-container data dirs the e2e
-# suite runs against. postgres:alpine tracks latest (18+), which mounts at
-# /var/lib/postgresql (not /var/lib/postgresql/data); bump here only.
-POSTGRES_IMAGE = "postgres:alpine"
-POSTGRES_DATA_DIR = "/var/lib/postgresql"
-MARIADB_IMAGE = "mariadb:latest"
-MARIADB_DATA_DIR = "/var/lib/mysql"
 
 
 def run(
@@ -163,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,
-        "--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/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=False,
+        source=resolve(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_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_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_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_cli.py

@@ -0,0 +1,65 @@
+"""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)
+
+
+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,63 +211,3 @@ class TestCompose(unittest.TestCase):
                     ),
                 ],
             )
-
-
-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_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,107 @@
+"""Contract of the filesystem snapshot used to capture volumes atomically."""
+
+from __future__ import annotations
+
+import unittest
+
+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_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")
+
+
+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,12 +1,12 @@
 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
     ):
@@ -18,7 +18,7 @@ class TestRequiresStop(unittest.TestCase):
         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
     ):
@@ -30,7 +30,7 @@ class TestRequiresStop(unittest.TestCase):
         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
     ):
@@ -38,7 +38,7 @@ class TestRequiresStop(unittest.TestCase):
         self.assertTrue(requires_stop(["c1"], ["mastodon"]))
         self.assertTrue(requires_stop(["c1"], ["repo/mastodon:v4"]))
 
-    @patch("baudolo.backup.app.get_image_info")
+    @patch("baudolo.backup.policy.get_image_info")
     def test_requires_stop_true_when_whitelist_empty(
         self, mock_get_image_info, _mock_is_swarm_task
     ):
@@ -47,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
     ):