Updated Changelog

A volume with a backing store of its own is not in a snapshot of the docker data root: it appears there as an existing empty directory, so the copy succeeds, the generation is stamped complete, and the volume is empty in it. The existing check only asked whether the path was inside the snapshot, which that empty directory answers with yes.

The driver, its options and the filesystem the mountpoint sits on now decide, per volume. An uncaptured volume is copied live - correct data without the point in time - while every other volume of the same run keeps its snapshot. One NFS volume no longer costs the whole host its consistent backup.

A volume resolving outside the subject degrades the same way instead of aborting the run.

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

CHANGELOG.md

@@ -1,5 +1,22 @@
 # Changelog
 
+## [3.6.0] - 2026-08-17
+
+- Restore: *--empty* drops the schema in one session and replays in the next,
+  with no rollback across the two, so a dump the engine could not parse left an
+  emptied database behind. The dump's header is now checked against the running
+  engine before anything is dropped, and a newer dump is refused.
+  Forward across a major version stays allowed; *--no-version-check* is the way out.
+- Restore: a volume with driver options — NFS, a bind device, tmpfs — keeps the
+  usual *_data* path, but docker mounts its real storage over it only while a
+  container holds it. Restoring meanwhile landed under the mount, stayed hidden
+  there, and rsync reported success. That volume is now refused until something
+  mounts it.
+- Backup: the same volume sits in a snapshot as an empty directory, so it was
+  copied empty and the generation stamped complete. Capture is decided per volume
+  now — an uncaptured one is copied live, the rest keep their snapshot. A single
+  NFS volume no longer costs the whole host its consistent backup.
+
 ## [3.5.0] - 2026-08-17
 
 - Restore: a *database = '*'* row makes the backup write

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "backup-docker-to-local"
-version = "3.5.0"
+version = "3.6.0"
 description = "Backup Docker volumes to local with rsync and optional DB dumps."
 readme = "README.md"
 requires-python = ">=3.9"

src/baudolo/backup/app.py

@@ -22,8 +22,8 @@ from .layout import (
     stamp_directory,
 )
 from .policy import requires_stop, volume_is_fully_ignored
-from .snapshot import volume_snapshot
+from .snapshot import snapshot_source, volume_snapshot
-from .volume import backup_volume, get_storage_path
+from .volume import backup_volume, inspect_backing
 
 
 def main() -> int:
@@ -86,7 +86,8 @@ def main() -> int:
                 else:
                     continue
 
-            live_source = get_storage_path(volume_name)
+            backing = inspect_backing(volume_name)
+            live_source = backing.source
 
             def copy(
                 *,
@@ -104,13 +105,15 @@ def main() -> int:
                 )
 
             if resolve_source is not None:
-                snapshot_source = resolve_source(live_source)
+                source, reason = snapshot_source(
-                if os.path.isdir(snapshot_source):
+                    resolve_source, backing, args.snapshot_subject
-                    copy(authoritative=True, source=snapshot_source)
+                )
+                if source is not None:
+                    copy(authoritative=True, source=source)
                 else:
                     print(
                         f"WARNING: volume '{volume_name}' is not in the snapshot "
-                        "(created after it was taken); copying it live instead.",
+                        f"({reason}); copying it live instead.",
                         flush=True,
                     )
                     copy(authoritative=False)

src/baudolo/backup/snapshot.py

@@ -9,6 +9,13 @@ 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.
+
+Which volumes a snapshot of the subject contains is a different question, and
+it is decided per volume: a volume with a backing store of its own appears
+inside the snapshot as an existing empty directory, so copying from there
+succeeds and stores nothing. Such a volume is copied live instead - correct
+data without the point in time - while every other volume of the same run
+keeps its snapshot.
 """
 
 from __future__ import annotations
@@ -18,6 +25,7 @@ from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 
 from .shell import BackupException, execute_shell_command
+from .volume import Backing
 
 KINDS = ("btrfs", "zfs")
 
@@ -62,6 +70,60 @@ def _zfs(subject: str, name: str, run: Callable[[str], list[str]]) -> tuple[str,
 _CREATE = {"btrfs": _btrfs, "zfs": _zfs}
 
 
+def unsnapshotted(backing: Backing, subject: str) -> str | None:
+    """Return why a snapshot of ``subject`` does not hold this volume's data.
+
+    Docker mounts a volume's own backing store lazily and unmounts it when the
+    last consumer stops, so the declaration is what gets checked: it is true at
+    every moment, where the mount table is only true while a container happens
+    to hold the volume.
+
+    Args:
+        backing: the volume as the daemon describes it.
+        subject: the snapshot subject, e.g. ``/var/lib/docker``.
+
+    Returns:
+        The reason, or None when the snapshot holds the volume.
+    """
+    if backing.driver != "local":
+        return f"it uses the {backing.driver} driver"
+    if backing.options:
+        return f"it declares its own backing store {backing.options}"
+    if not backing.mountpoint:
+        return "it reports no mountpoint"
+    real = os.path.realpath(backing.mountpoint)
+    if os.path.ismount(real):
+        return f"its mountpoint {backing.mountpoint} sits on its own mount"
+    try:
+        crosses = os.stat(real).st_dev != os.stat(os.path.realpath(subject)).st_dev
+    except OSError as error:
+        return f"its mountpoint {backing.mountpoint} could not be read: {error}"
+    if crosses:
+        return f"its mountpoint {backing.mountpoint} crosses a filesystem boundary"
+    return None
+
+
+def snapshot_source(
+    resolve: Callable[[str], str], backing: Backing, subject: str
+) -> tuple[str | None, str]:
+    """Resolve where to read a volume from, and why if not from the snapshot.
+
+    Returns:
+        ``(path, "")`` to copy from the snapshot, or ``(None, reason)`` to copy
+        it live.
+    """
+    reason = unsnapshotted(backing, subject)
+    if reason:
+        return None, reason
+    try:
+        source = resolve(backing.source)
+    except SnapshotError as error:
+        return None, str(error)
+    if not os.path.isdir(source):
+        return None, "it was created after the snapshot was taken"
+    return source, ""
+
+
 @contextmanager
 def volume_snapshot(
     kind: str,

src/baudolo/backup/volume.py

@@ -1,16 +1,43 @@
 from __future__ import annotations
 
+import json
 import os
 import pathlib
+from dataclasses import dataclass, field
 
 from .shell import BackupException, execute_shell_command
 
 
-def get_storage_path(volume_name: str) -> str:
+@dataclass(frozen=True)
-    path = execute_shell_command(
+class Backing:
-        f"docker volume inspect --format '{{{{ .Mountpoint }}}}' {volume_name}"
+    """Where a docker volume actually keeps its data.
+
+    Args:
+        mountpoint: the path the daemon reports.
+        driver: the volume driver, ``local`` for the built-in one.
+        options: the driver options; a non-empty map means the mountpoint is a
+            mount target rather than the storage itself.
+    """
+
+    mountpoint: str
+    driver: str = "local"
+    options: dict = field(default_factory=dict)
+
+    @property
+    def source(self) -> str:
+        return f"{self.mountpoint}/"
+
+
+def inspect_backing(volume_name: str) -> Backing:
+    reported = execute_shell_command(
+        f"docker volume inspect --format '{{{{json .}}}}' {volume_name}"
     )[0]
-    return f"{path}/"
+    data = json.loads(reported)
+    return Backing(
+        data.get("Mountpoint") or "",
+        data.get("Driver") or "",
+        data.get("Options") or {},
+    )
 
 
 def get_last_backup_dir(

src/baudolo/restore/__main__.py

@@ -27,6 +27,20 @@ def _add_common_backup_args(p: argparse.ArgumentParser) -> None:
     )
 
 
+def _add_common_engine_args(p: argparse.ArgumentParser) -> None:
+    p.add_argument("--container", required=True)
+    p.add_argument("--db-password", required=True)
+    p.add_argument("--empty", action="store_true")
+    p.add_argument(
+        "--no-version-check",
+        action="store_true",
+        help=(
+            "Replay even if the dump comes from a newer engine than the target. "
+            "With --empty this can leave an emptied database behind."
+        ),
+    )
+
+
 def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(
         prog="baudolo-restore",
@@ -48,17 +62,15 @@ def main(argv: list[str] | None = None) -> int:
 
     p_pg = sub.add_parser("postgres", help="Restore a single PostgreSQL database dump")
     _add_common_backup_args(p_pg)
-    p_pg.add_argument("--container", required=True)
+    _add_common_engine_args(p_pg)
     p_pg.add_argument("--db-name", required=True)
     p_pg.add_argument("--db-user", default=None, help="Defaults to db-name if omitted")
-    p_pg.add_argument("--db-password", required=True)
-    p_pg.add_argument("--empty", action="store_true")
 
     p_cluster = sub.add_parser(
         "cluster", help="Restore a full PostgreSQL cluster dump (pg_dumpall)"
     )
     _add_common_backup_args(p_cluster)
-    p_cluster.add_argument("--container", required=True)
+    _add_common_engine_args(p_cluster)
     p_cluster.add_argument(
         "--instance",
         required=True,
@@ -69,18 +81,14 @@ def main(argv: list[str] | None = None) -> int:
         required=True,
         help="Superuser of the instance; the dump creates roles and databases",
     )
-    p_cluster.add_argument("--db-password", required=True)
-    p_cluster.add_argument("--empty", action="store_true")
 
     p_mdb = sub.add_parser(
         "mariadb", help="Restore a single MariaDB/MySQL-compatible dump"
     )
     _add_common_backup_args(p_mdb)
-    p_mdb.add_argument("--container", required=True)
+    _add_common_engine_args(p_mdb)
     p_mdb.add_argument("--db-name", required=True)
     p_mdb.add_argument("--db-user", default=None, help="Defaults to db-name if omitted")
-    p_mdb.add_argument("--db-password", required=True)
-    p_mdb.add_argument("--empty", action="store_true")
 
     args = parser.parse_args(argv)
 
@@ -116,6 +124,7 @@ def main(argv: list[str] | None = None) -> int:
                     backups_dir=args.backups_dir,
                 ).sql_file(args.db_name),
                 empty=args.empty,
+                check_version=not args.no_version_check,
             )
             return 0
 
@@ -132,6 +141,7 @@ def main(argv: list[str] | None = None) -> int:
                     backups_dir=args.backups_dir,
                 ).cluster_file(args.instance),
                 empty=args.empty,
+                check_version=not args.no_version_check,
             )
             return 0
 
@@ -150,6 +160,7 @@ def main(argv: list[str] | None = None) -> int:
                     backups_dir=args.backups_dir,
                 ).sql_file(args.db_name),
                 empty=args.empty,
+                check_version=not args.no_version_check,
             )
             return 0
 

src/baudolo/restore/db/cluster.py

@@ -25,6 +25,7 @@ import tempfile
 from collections.abc import Iterable, Iterator
 
 from ..run import docker_exec
+from .version import guard
 
 CONTROL_DB = "postgres"
 _CLUSTER_PRECLEAN_SQL = os.path.join(os.path.dirname(__file__), "cluster_preclean.sql")
@@ -65,6 +66,7 @@ def restore_cluster_sql(
     password: str,
     sql_path: str,
     empty: bool,
+    check_version: bool = True,
 ) -> None:
     """Replay a pg_dumpall stream into a running instance.
 
@@ -78,10 +80,21 @@ def restore_cluster_sql(
             replay stops at the first object that already exists, which is the
             honest outcome: recreating a cluster over a populated one is a
             decision, not a default.
+        check_version: refuse a dump from a newer major version than the
+            running engine before anything is dropped.
     """
     if not os.path.isfile(sql_path):
         raise FileNotFoundError(sql_path)
 
+    if check_version:
+        guard(
+            sql_path=sql_path,
+            engine="postgres",
+            container=container,
+            user=user,
+            password=password,
+        )
+
     docker_env = {"PGPASSWORD": password}
 
     if empty:

src/baudolo/restore/db/mariadb.py

@@ -4,6 +4,7 @@ import os
 import sys
 
 from ..run import docker_exec, docker_exec_sh
+from .version import guard
 
 
 def _pick_client(container: str) -> str:
@@ -37,12 +38,23 @@ def restore_mariadb_sql(
     password: str,
     sql_path: str,
     empty: bool,
+    check_version: bool = True,
 ) -> None:
     client = _pick_client(container)
 
     if not os.path.isfile(sql_path):
         raise FileNotFoundError(sql_path)
 
+    if check_version:
+        guard(
+            sql_path=sql_path,
+            engine="mariadb",
+            container=container,
+            user=user,
+            password=password,
+            client=client,
+        )
+
     if empty:
         # Do not hardcode 'mysql': MariaDB 11 images may not ship that binary.
         result = docker_exec(

src/baudolo/restore/db/postgres.py

@@ -5,6 +5,7 @@ import tempfile
 from collections.abc import Iterable, Iterator
 
 from ..run import docker_exec
+from .version import guard
 
 _SUPERUSER_ONLY_PREFIXES = (b"COMMENT ON EXTENSION", b"ALTER DEFAULT PRIVILEGES")
 _EMPTY_PRECLEAN_SQL = os.path.join(os.path.dirname(__file__), "empty_preclean.sql")
@@ -46,10 +47,20 @@ def restore_postgres_sql(
     password: str,
     sql_path: str,
     empty: bool,
+    check_version: bool = True,
 ) -> None:
     if not os.path.isfile(sql_path):
         raise FileNotFoundError(sql_path)
 
+    if check_version:
+        guard(
+            sql_path=sql_path,
+            engine="postgres",
+            container=container,
+            user=user,
+            password=password,
+        )
+
     docker_env = {"PGPASSWORD": password}
 
     if empty:

src/baudolo/restore/db/version.py

@@ -0,0 +1,146 @@
+"""Refuse a dump the target engine is too old to read.
+
+A restore with ``--empty`` destroys before it replays: the pre-clean drops the
+schema in one session and the dump goes in the next, with no rollback across
+the two. A dump the engine cannot parse therefore does not fail harmlessly -
+it leaves an emptied database behind. Comparing the two versions first turns
+that into a refusal.
+
+Both engines state their origin in the dump's own header, and they do not
+state it the same way. Postgres writes ``-- Dumped from database version``
+around line seven. MariaDB opens line two with ``-- MariaDB dump 10.19-11.8.8``,
+where the first number is mariadb-dump's own version, and names the server only
+further down on the tab-separated ``-- Server version`` line. Matching the first
+number in the header would read the tool on one engine and the server on the
+other, so each engine gets its own pattern.
+
+A ``pg_dumpall`` cluster dump has no version line of its own: its header opens
+with the cluster banner and the roles section, and the first
+``-- Dumped from database version`` belongs to the first database's embedded
+``pg_dump`` output, arbitrarily far down. Hence the scan runs to
+``SCAN_LINES`` rather than to a header-sized handful.
+"""
+
+from __future__ import annotations
+
+import re
+
+from ..run import docker_exec, stdout_of
+
+SCAN_LINES = 2000
+DUMP_VERSION = {
+    "postgres": re.compile(r"^-- Dumped from database version (\S+)"),
+    "mariadb": re.compile(r"^-- Server version\s+(\S+)"),
+}
+
+
+class VersionMismatch(Exception):
+    """The dump cannot be replayed into this engine."""
+
+
+def major_of(version: str) -> int:
+    """The major number of an engine version string.
+
+    Args:
+        version: as the engine spells it, e.g. ``17.11`` or
+            ``11.8.8-MariaDB-ubu2404``.
+
+    Raises:
+        VersionMismatch: the string does not start with a number.
+    """
+    leading = re.match(r"(\d+)", version)
+    if not leading:
+        raise VersionMismatch(f"cannot read a major version from '{version}'")
+    return int(leading.group(1))
+
+
+def dump_version(sql_path: str, engine: str) -> str:
+    """Read the engine version a dump was taken from, out of its own header.
+
+    Args:
+        sql_path: the dump to read.
+        engine: ``postgres`` or ``mariadb``.
+
+    Returns:
+        The version string as the dump spells it.
+
+    Raises:
+        VersionMismatch: no version line within the first ``SCAN_LINES``.
+    """
+    pattern = DUMP_VERSION[engine]
+    with open(sql_path, encoding="utf-8", errors="replace") as handle:
+        for _ in range(SCAN_LINES):
+            line = handle.readline()
+            if not line:
+                break
+            found = pattern.search(line)
+            if found:
+                return found.group(1)
+    raise VersionMismatch(
+        f"{sql_path} carries no {engine} version header in its first {SCAN_LINES} lines"
+    )
+
+
+def server_version(
+    container: str, engine: str, user: str, password: str, client: str = ""
+) -> str:
+    """Ask the running engine which version it is."""
+    if engine == "postgres":
+        return stdout_of(
+            docker_exec(
+                container,
+                ["psql", "-U", user, "-tAc", "SHOW server_version"],
+                capture=True,
+                docker_env={"PGPASSWORD": password},
+            )
+        )
+    return stdout_of(
+        docker_exec(
+            container,
+            [
+                client or "mariadb",
+                "-u",
+                user,
+                f"--password={password}",
+                "-N",
+                "-B",
+                "-e",
+                "SELECT VERSION()",
+            ],
+            capture=True,
+        )
+    )
+
+
+def assert_replayable(sql_path: str, engine: str, dumped: str, serving: str) -> None:
+    """Refuse a dump from a newer major version than the target engine.
+
+    Restoring forward across a major version is the upgrade path and stays
+    allowed; backward is refused, because a newer dump uses syntax an older
+    server rejects and the pre-clean would already have dropped the schema.
+
+    Raises:
+        VersionMismatch: the dump is newer than the engine.
+    """
+    if major_of(dumped) > major_of(serving):
+        raise VersionMismatch(
+            f"{sql_path} came from {engine} {dumped} but {serving} is running; "
+            "a newer dump does not replay into an older engine, and --empty "
+            "would drop the schema before finding out"
+        )
+
+
+def guard(
+    *,
+    sql_path: str,
+    engine: str,
+    container: str,
+    user: str,
+    password: str,
+    client: str = "",
+) -> None:
+    """Compare the dump's origin against the running engine before replaying."""
+    dumped = dump_version(sql_path, engine)
+    serving = server_version(container, engine, user, password, client)
+    assert_replayable(sql_path, engine, dumped, serving)
+    print(f"OK: dump is from {engine} {dumped}, {serving} is serving.")

src/baudolo/restore/files.py

@@ -1,9 +1,23 @@
+"""Restore a volume's file tree by writing into its mountpoint.
+
+That shortcut only holds for a plain local volume, where the mountpoint *is*
+the storage. A volume with driver options - NFS, a bind device, tmpfs - keeps
+the same ``/var/lib/docker/volumes/<name>/_data`` path, but docker mounts the
+real backing store over it on demand and unmounts it again when the last
+consumer stops. Writing there while nothing has it mounted lands in the empty
+directory underneath, is hidden by the next mount, and rsync reports success.
+"""
+
 from __future__ import annotations
 
 import os
 import sys
 
-from .run import docker_volume_exists, run
+from .run import docker_volume_exists, run, stdout_of
+
+INSPECT_FORMAT = (
+    "{{ .Mountpoint }}|{{ .Driver }}|{{ if .Options }}opts{{ else }}plain{{ end }}"
+)
 
 
 def restore_volume_files(volume_name: str, backup_files_dir: str) -> int:
@@ -18,11 +32,11 @@ def restore_volume_files(volume_name: str, backup_files_dir: str) -> int:
         print(f"Volume {volume_name} already exists.")
 
     cp = run(
-        ["docker", "volume", "inspect", "--format", "{{ .Mountpoint }}", volume_name],
+        ["docker", "volume", "inspect", "--format", INSPECT_FORMAT, volume_name],
         capture=True,
     )
-    raw = cp.stdout or b""
+    fields = stdout_of(cp).split("|")
-    mountpoint = (raw.decode() if isinstance(raw, bytes) else raw).strip()
+    mountpoint = fields[0] if fields else ""
     if not mountpoint:
         print(
             f"ERROR: could not resolve mountpoint for volume {volume_name}",
@@ -30,6 +44,17 @@ def restore_volume_files(volume_name: str, backup_files_dir: str) -> int:
         )
         return 2
 
+    driver, options = (fields + ["local", "plain"])[1:3]
+    if (driver != "local" or options == "opts") and not os.path.ismount(mountpoint):
+        print(
+            f"ERROR: volume {volume_name} has a backing store of its own "
+            f"(driver {driver}) but nothing has it mounted; writing to "
+            f"{mountpoint} now would land under the mount and be lost. "
+            "Start a container that mounts the volume, then restore again.",
+            file=sys.stderr,
+        )
+        return 2
+
     src = os.path.join(backup_files_dir, "")
     dest = os.path.join(mountpoint, "")
     run(["rsync", "-avv", "--delete", src, dest])

src/baudolo/restore/run.py

@@ -40,6 +40,12 @@ def run(
         raise
 
 
+def stdout_of(completed: subprocess.CompletedProcess) -> str:
+    """The captured stdout as stripped text, whether it came back bytes or str."""
+    raw = completed.stdout or b""
+    return (raw.decode() if isinstance(raw, bytes) else raw).strip()
+
+
 def docker_exec(
     container: str,
     argv: list[str],

tests/e2e/faithful_driver.py

@@ -0,0 +1,92 @@
+"""Show what a real snapshot holds for a volume that has its own storage.
+
+Runs inside the privileged container that built the btrfs subject. Prints one
+PASS/FAIL line per assertion and exits non-zero on the first failure.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+sys.path.insert(0, "/src")
+
+from baudolo.backup.snapshot import (
+    SnapshotError,
+    snapshot_source,
+    unsnapshotted,
+    volume_snapshot,
+)
+from baudolo.backup.volume import Backing
+
+SUBJECT = sys.argv[1]
+
+
+def shell(command: str) -> list[str]:
+    proc = subprocess.run(
+        command, shell=True, capture_output=True, text=True, check=False
+    )
+    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)
+
+
+def volume(name: str, payload: str) -> Path:
+    path = Path(SUBJECT) / "volumes" / name / "_data"
+    path.mkdir(parents=True, exist_ok=True)
+    (path / "state").write_text(payload)
+    return path
+
+
+plain = volume("plain", "plain-payload")
+own = Path(SUBJECT) / "volumes" / "own" / "_data"
+own.mkdir(parents=True, exist_ok=True)
+shell(f"mount -t tmpfs tmpfs {own}")
+(own / "state").write_text("own-payload")
+
+check("a plain volume is captured", unsnapshotted(Backing(str(plain)), SUBJECT) is None)
+check(
+    "a volume on a mount of its own is not",
+    unsnapshotted(Backing(str(own)), SUBJECT) is not None,
+)
+check(
+    "a declared backing store is not, mounted or not",
+    unsnapshotted(Backing(str(plain), options={"type": "nfs"}), SUBJECT) is not None,
+)
+check(
+    "a foreign driver is not",
+    unsnapshotted(Backing(str(plain), driver="rexray"), SUBJECT) is not None,
+)
+
+with volume_snapshot("btrfs", SUBJECT, "e2e", run=shell) as resolve:
+    frozen_plain = Path(resolve(str(plain)))
+    frozen_own = Path(resolve(str(own)))
+
+    check(
+        "the snapshot carries the plain volume",
+        (frozen_plain / "state").read_text() == "plain-payload",
+    )
+    check(
+        "the snapshot shows the other volume as an empty directory",
+        frozen_own.is_dir() and not any(frozen_own.iterdir()),
+    )
+
+    source, reason = snapshot_source(resolve, Backing(str(plain)), SUBJECT)
+    check(
+        "the plain volume is read from the snapshot",
+        source is not None and source.rstrip("/") == str(frozen_plain),
+    )
+
+    source, reason = snapshot_source(resolve, Backing(str(own)), SUBJECT)
+    check(f"the other volume degrades to live: {reason[:60]}", source is None)
+
+print("ALL OK", flush=True)

tests/e2e/test_e2e_restore_files_backing_store.py

@@ -0,0 +1,120 @@
+"""Restoring files into a volume that has a backing store of its own.
+
+Docker keeps the same ``/var/lib/docker/volumes/<name>/_data`` path for such a
+volume and mounts the real storage over it only while a container holds it.
+Writing there unmounted lands in the empty directory underneath, is hidden by
+the next mount, and rsync reports success - so the restore has to refuse.
+"""
+
+import unittest
+from pathlib import Path
+
+from .helpers import (
+    backup_path,
+    cleanup_docker,
+    ensure_empty_dir,
+    machine_hash,
+    require_docker,
+    run,
+    unique,
+)
+
+MARKER = "restored-payload"
+VERSION = "20260817000000"
+
+
+def mountpoint_of(volume: str) -> Path:
+    return Path("/var/lib/docker/volumes") / volume / "_data"
+
+
+def contents(directory: Path) -> list[str]:
+    return sorted(p.name for p in directory.iterdir()) if directory.is_dir() else []
+
+
+class TestE2ERestoreFilesBackingStore(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+        cls.prefix = unique("baudolo-e2e-backing")
+        cls.repo_name = cls.prefix
+        cls.backups_dir = f"/tmp/{cls.prefix}/Backups"
+        cls.backing = Path(f"/tmp/{cls.prefix}/backing")
+        ensure_empty_dir(cls.backups_dir)
+        ensure_empty_dir(str(cls.backing))
+
+        cls.bound_volume = f"{cls.prefix}-bound"
+        cls.plain_volume = f"{cls.prefix}-plain"
+        cls.volumes = [cls.bound_volume, cls.plain_volume]
+
+        for volume in cls.volumes:
+            files = (
+                backup_path(cls.backups_dir, cls.repo_name, VERSION, volume) / "files"
+            )
+            files.mkdir(parents=True, exist_ok=True)
+            (files / "marker.txt").write_text(MARKER, encoding="utf-8")
+
+        run(
+            [
+                "docker",
+                "volume",
+                "create",
+                "--driver",
+                "local",
+                "--opt",
+                "type=none",
+                "--opt",
+                "o=bind",
+                "--opt",
+                f"device={cls.backing}",
+                cls.bound_volume,
+            ]
+        )
+        run(["docker", "volume", "create", cls.plain_volume])
+
+        cls.refused = cls.restore(cls.bound_volume)
+        cls.accepted = cls.restore(cls.plain_volume)
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cleanup_docker(containers=[], volumes=cls.volumes)
+
+    @classmethod
+    def restore(cls, volume: str):
+        return run(
+            [
+                "baudolo-restore",
+                "files",
+                volume,
+                machine_hash(),
+                VERSION,
+                "--backups-dir",
+                cls.backups_dir,
+                "--repo-name",
+                cls.repo_name,
+            ],
+            check=False,
+        )
+
+    def test_a_volume_with_its_own_backing_store_is_refused(self) -> None:
+        self.assertEqual(self.refused.returncode, 2, self.refused.stdout)
+        self.assertIn("backing store of its own", self.refused.stderr)
+
+    def test_nothing_was_written_into_the_backing_store(self) -> None:
+        self.assertEqual(contents(self.backing), [])
+
+    def test_nothing_was_written_under_the_mount_either(self) -> None:
+        self.assertEqual(
+            contents(mountpoint_of(self.bound_volume)),
+            [],
+            "the copy landed in the directory the next mount hides",
+        )
+
+    def test_a_plain_volume_is_still_restored(self) -> None:
+        self.assertEqual(self.accepted.returncode, 0, self.accepted.stderr)
+        restored = mountpoint_of(self.plain_volume) / "marker.txt"
+        self.assertTrue(restored.is_file(), f"{restored} missing")
+        self.assertEqual(restored.read_text(encoding="utf-8"), MARKER)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/e2e/test_e2e_snapshot.py

@@ -17,6 +17,7 @@ from .helpers import require_docker, run, unique
 
 REPO_SRC = Path(__file__).resolve().parents[2] / "src"
 DRIVER = Path(__file__).resolve().parent / "snapshot_driver.py"
+FAITHFUL_DRIVER = Path(__file__).resolve().parent / "faithful_driver.py"
 IMAGE = "alpine:3.20"
 PACKAGES = "apk add -q btrfs-progs e2fsprogs zfs python3 util-linux"
 ATTACH = (
@@ -76,20 +77,17 @@ def required(fstype: str) -> bool:
     return fstype in demanded.replace(",", " ").split()
 
 
-def stage() -> Path:
+def stage(driver: Path) -> 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")
+    shutil.copy(driver, staged / "driver.py")
     return staged
 
 
-def drive(fstype: str, kind: str, expect: str) -> str:
+def drive(fstype: str, arguments: str, *, driver: Path = DRIVER) -> str:
-    staged = stage()
+    staged = stage(driver)
-    script = (
+    script = f"set -e; {mount_script(fstype)}; python3 /driver.py {arguments}"
-        f"set -e; {mount_script(fstype)}; "
-        f"python3 /driver.py {kind} /subject/docker {expect}"
-    )
     try:
         proc = run(
             [
@@ -115,7 +113,7 @@ def drive(fstype: str, kind: str, expect: str) -> str:
         shutil.rmtree(staged, ignore_errors=True)
     if proc.returncode != 0:
         raise AssertionError(
-            f"{fstype}/{kind} driver failed:\n{proc.stdout}\n{proc.stderr}"
+            f"{fstype} driver failed on {arguments}:\n{proc.stdout}\n{proc.stderr}"
         )
     return proc.stdout
 
@@ -126,7 +124,7 @@ class TestE2ESnapshot(unittest.TestCase):
         require_docker()
 
     def assert_freezes(self, fstype: str) -> None:
-        output = drive(fstype, fstype, "supported")
+        output = drive(fstype, f"{fstype} /subject/docker 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)
@@ -148,13 +146,24 @@ class TestE2ESnapshot(unittest.TestCase):
         self.assert_freezes("zfs")
 
     def test_ext4_has_no_snapshot_and_says_so(self) -> None:
-        output = drive("ext4", "btrfs", "unsupported")
+        output = drive("ext4", "btrfs /subject/docker 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")
+        output = drive("ext4", "lvm /subject/docker unsupported")
         self.assertIn("PASS refused loudly", output)
 
+    def test_a_volume_with_its_own_storage_is_copied_live_not_from_the_snapshot(
+        self,
+    ) -> None:
+        output = drive("btrfs", "/subject/docker", driver=FAITHFUL_DRIVER)
+        self.assertIn(
+            "PASS the snapshot shows the other volume as an empty directory", output
+        )
+        self.assertIn("PASS the other volume degrades to live", output)
+        self.assertIn("PASS the plain volume is read from the snapshot", output)
+        self.assertIn("ALL OK", output)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/e2e/test_e2e_version_gate.py

@@ -0,0 +1,283 @@
+"""A dump from a newer engine must be refused before --empty destroys anything.
+
+The pre-clean and the replay are two separate sessions with no rollback across
+them, so a dump the engine cannot parse leaves an emptied database behind. The
+decisive assertion here is not the non-zero exit - it is that the payload is
+still readable afterwards.
+"""
+
+import re
+import unittest
+from pathlib import Path
+
+from .helpers import (
+    MARIADB_DATA_DIR,
+    MARIADB_IMAGE,
+    POSTGRES_DATA_DIR,
+    POSTGRES_IMAGE,
+    backup_path,
+    backup_run,
+    cleanup_docker,
+    create_minimal_compose_dir,
+    ensure_empty_dir,
+    latest_version_dir,
+    require_docker,
+    run,
+    unique,
+    wait_for_mariadb,
+    wait_for_mariadb_sql,
+    wait_for_postgres,
+    write_databases_csv,
+)
+
+PAYLOAD = "gate-payload"
+FUTURE = "99.0"
+
+
+def rewrite_version(dump: Path, pattern: str, version: str) -> str:
+    """Make the dump claim ``version``; return what it claimed before."""
+    text = dump.read_text(encoding="utf-8", errors="replace")
+    found = re.search(pattern, text)
+    if not found:
+        raise AssertionError(f"{dump} carries no version header matching {pattern}")
+    claimed = found.group(1)
+    dump.write_text(
+        text.replace(found.group(0), found.group(0).replace(claimed, version), 1),
+        encoding="utf-8",
+    )
+    return claimed
+
+
+class GateCase:
+    """Drive one engine through refusal, escape hatch and truthful replay."""
+
+    engine = ""
+    pattern = ""
+
+    @classmethod
+    def restore(cls, *extra: str):
+        return run(
+            [
+                "baudolo-restore",
+                cls.engine,
+                cls.volume,
+                cls.hash,
+                cls.version,
+                "--backups-dir",
+                cls.backups_dir,
+                "--repo-name",
+                cls.repo_name,
+                "--container",
+                cls.container,
+                "--db-name",
+                cls.db_name,
+                "--db-user",
+                cls.db_user,
+                "--db-password",
+                cls.db_password,
+                "--empty",
+                *extra,
+            ],
+            check=False,
+        )
+
+    @classmethod
+    def prepare(cls) -> None:
+        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.databases_csv = f"/tmp/{cls.prefix}/databases.csv"
+        write_databases_csv(
+            cls.databases_csv,
+            [(cls.container, cls.db_name, cls.db_user, cls.db_password)],
+        )
+        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.container],
+            images_no_stop_required=[cls.image],
+        )
+        cls.hash, cls.version = latest_version_dir(cls.backups_dir, cls.repo_name)
+        cls.dump = (
+            backup_path(cls.backups_dir, cls.repo_name, cls.version, cls.volume)
+            / "sql"
+            / f"{cls.db_name}.backup.sql"
+        )
+
+        cls.truthful_version = rewrite_version(cls.dump, cls.pattern, FUTURE)
+        cls.refused = cls.restore()
+        cls.payload_after_refusal = cls.read_payload()
+
+        cls.forced = cls.restore("--no-version-check")
+        cls.payload_after_force = cls.read_payload()
+
+        rewrite_version(cls.dump, cls.pattern, cls.truthful_version)
+        cls.replayed = cls.restore()
+        cls.payload_after_replay = cls.read_payload()
+
+    def test_the_dump_states_the_engine_it_came_from(self) -> None:
+        self.assertRegex(self.truthful_version, r"^\d+")
+
+    def test_a_newer_dump_is_refused(self) -> None:
+        self.assertNotEqual(self.refused.returncode, 0, self.refused.stdout)
+
+    def test_the_refusal_names_the_version_it_refused(self) -> None:
+        self.assertIn(FUTURE, self.refused.stderr)
+        self.assertIn("older engine", self.refused.stderr)
+
+    def test_the_refusal_left_the_data_untouched(self) -> None:
+        self.assertEqual(
+            self.payload_after_refusal,
+            PAYLOAD,
+            "--empty pre-cleaned before the version was checked",
+        )
+
+    def test_the_escape_hatch_replays_anyway(self) -> None:
+        self.assertEqual(self.forced.returncode, 0, self.forced.stderr)
+        self.assertEqual(self.payload_after_force, PAYLOAD)
+
+    def test_a_truthful_dump_replays(self) -> None:
+        self.assertEqual(self.replayed.returncode, 0, self.replayed.stderr)
+        self.assertEqual(self.payload_after_replay, PAYLOAD)
+
+
+class TestE2EPostgresVersionGate(GateCase, unittest.TestCase):
+    engine = "postgres"
+    pattern = r"-- Dumped from database version (\S+)"
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+        cls.prefix = unique("baudolo-e2e-pg-gate")
+        cls.container = f"{cls.prefix}-pg"
+        cls.volume = f"{cls.prefix}-pg-vol"
+        cls.image = POSTGRES_IMAGE
+        cls.db_name = "appdb"
+        cls.db_user = "postgres"
+        cls.db_password = "pgpw"
+
+        run(["docker", "volume", "create", cls.volume])
+        run(
+            [
+                "docker",
+                "run",
+                "-d",
+                "--name",
+                cls.container,
+                "-e",
+                f"POSTGRES_PASSWORD={cls.db_password}",
+                "-v",
+                f"{cls.volume}:{POSTGRES_DATA_DIR}",
+                POSTGRES_IMAGE,
+            ]
+        )
+        wait_for_postgres(cls.container, user=cls.db_user)
+        cls.sql("postgres", f"CREATE DATABASE {cls.db_name}")
+        cls.sql(
+            cls.db_name,
+            f"CREATE TABLE t (v text); INSERT INTO t VALUES ('{PAYLOAD}');",
+        )
+        cls.prepare()
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cleanup_docker(containers=[cls.container], volumes=[cls.volume])
+
+    @classmethod
+    def sql(cls, database: str, statement: str) -> str:
+        p = run(
+            [
+                "docker",
+                "exec",
+                cls.container,
+                "sh",
+                "-lc",
+                f'psql -U {cls.db_user} -d {database} -t -A -c "{statement}"',
+            ],
+            check=False,
+        )
+        return (p.stdout or "").strip()
+
+    @classmethod
+    def read_payload(cls) -> str:
+        return cls.sql(cls.db_name, "SELECT v FROM t")
+
+
+class TestE2EMariadbVersionGate(GateCase, unittest.TestCase):
+    engine = "mariadb"
+    pattern = r"-- Server version\s+(\S+)"
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        require_docker()
+        cls.prefix = unique("baudolo-e2e-mdb-gate")
+        cls.container = f"{cls.prefix}-mdb"
+        cls.volume = f"{cls.prefix}-mdb-vol"
+        cls.image = MARIADB_IMAGE
+        cls.db_name = "appdb"
+        cls.db_user = "test"
+        cls.db_password = "testpw"
+
+        run(["docker", "volume", "create", cls.volume])
+        run(
+            [
+                "docker",
+                "run",
+                "-d",
+                "--name",
+                cls.container,
+                "-e",
+                "MARIADB_ROOT_PASSWORD=rootpw",
+                "-e",
+                f"MARIADB_DATABASE={cls.db_name}",
+                "-e",
+                f"MARIADB_USER={cls.db_user}",
+                "-e",
+                f"MARIADB_PASSWORD={cls.db_password}",
+                "-v",
+                f"{cls.volume}:{MARIADB_DATA_DIR}",
+                MARIADB_IMAGE,
+            ]
+        )
+        wait_for_mariadb(cls.container, root_password="rootpw", timeout_s=90)
+        wait_for_mariadb_sql(
+            cls.container, user=cls.db_user, password=cls.db_password, timeout_s=90
+        )
+        cls.sql(
+            f"CREATE TABLE {cls.db_name}.t (v VARCHAR(50)); "
+            f"INSERT INTO {cls.db_name}.t VALUES ('{PAYLOAD}');"
+        )
+        cls.prepare()
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cleanup_docker(containers=[cls.container], volumes=[cls.volume])
+
+    @classmethod
+    def sql(cls, statement: str) -> str:
+        p = run(
+            [
+                "docker",
+                "exec",
+                cls.container,
+                "sh",
+                "-lc",
+                (
+                    f"mariadb -h 127.0.0.1 -u{cls.db_user} -p{cls.db_password} "
+                    f'-N -B -e "{statement}"'
+                ),
+            ],
+            check=False,
+        )
+        return (p.stdout or "").strip()
+
+    @classmethod
+    def read_payload(cls) -> str:
+        return cls.sql(f"SELECT v FROM {cls.db_name}.t")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/backup/test_app_snapshot.py

@@ -6,7 +6,9 @@ import unittest
 from unittest import mock
 
 from baudolo.backup import app
+from baudolo.backup import snapshot as snapshot_mod
 from baudolo.backup.snapshot import volume_snapshot
+from baudolo.backup.volume import Backing
 
 
 def stubbed_snapshot(kind: str, subject: str, tag: str):
@@ -26,7 +28,7 @@ ARGV = [
 ]
 
 
-def drive(*, present: bool) -> list[dict]:
+def drive(*, present: bool = True, reason: str | None = None) -> list[dict]:
     calls: list[dict] = []
 
     def record(versions_dir, volume_name, volume_dir, *, authoritative, source):
@@ -45,8 +47,11 @@ def drive(*, present: bool) -> list[dict]:
         mock.patch.object(app, "volume_is_fully_ignored", return_value=False),
         mock.patch.object(app, "backup_dumps_for_volume", return_value=(False, False)),
         mock.patch.object(
-            app, "get_storage_path", return_value="/var/lib/docker/volumes/vol/_data/"
+            app,
+            "inspect_backing",
+            return_value=Backing("/var/lib/docker/volumes/vol/_data"),
         ),
+        mock.patch.object(snapshot_mod, "unsnapshotted", return_value=reason),
         mock.patch.object(app, "stamp_directory"),
         mock.patch.object(app, "handle_docker_compose_services"),
         mock.patch.object(app.os.path, "isdir", return_value=present),
@@ -75,6 +80,14 @@ class TestSnapshotBranch(unittest.TestCase):
         self.assertEqual(call["source"], "/var/lib/docker/volumes/vol/_data/")
         self.assertFalse(call["authoritative"])
 
+    def test_a_volume_with_its_own_backing_store_is_copied_live(self) -> None:
+        call = drive(reason="it declares its own backing store")[0]
+        self.assertEqual(call["source"], "/var/lib/docker/volumes/vol/_data/")
+        self.assertFalse(
+            call["authoritative"],
+            "the snapshot holds an empty directory for it, not its data",
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/backup/test_app_volumes_no_backup_required.py

@@ -7,6 +7,7 @@ import unittest
 from unittest import mock
 
 from baudolo.backup import app
+from baudolo.backup.volume import Backing
 
 ARGV = [
     "baudolo",
@@ -47,7 +48,7 @@ def drive() -> tuple[list[str], list[str], list[str]]:
         ),
         mock.patch.object(app, "volume_is_fully_ignored", return_value=False),
         mock.patch.object(app, "backup_dumps_for_volume", return_value=(False, False)),
-        mock.patch.object(app, "get_storage_path", return_value="/data/"),
+        mock.patch.object(app, "inspect_backing", return_value=Backing("/data")),
         mock.patch.object(app, "stamp_directory"),
         mock.patch.object(app, "handle_docker_compose_services"),
         mock.patch.object(app.os.path, "isdir", return_value=True),

tests/unit/backup/test_snapshot_faithfulness.py

@@ -0,0 +1,126 @@
+"""Which volumes a snapshot of the subject actually contains.
+
+The failure this guards against is silent: a volume with a backing store of
+its own is present inside the snapshot as an empty directory, so rsync
+succeeds, the generation is stamped complete, and the volume is empty in it.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from unittest import mock
+
+from baudolo.backup.snapshot import SnapshotError, snapshot_source, unsnapshotted
+from baudolo.backup.volume import Backing
+
+
+class TestUnsnapshotted(unittest.TestCase):
+    def setUp(self) -> None:
+        self.subject = tempfile.mkdtemp()
+        self.mountpoint = os.path.join(self.subject, "volumes", "app", "_data")
+        os.makedirs(self.mountpoint)
+
+    def backing(self, **kwargs) -> Backing:
+        return Backing(kwargs.pop("mountpoint", self.mountpoint), **kwargs)
+
+    def test_a_plain_local_volume_is_captured(self) -> None:
+        self.assertIsNone(unsnapshotted(self.backing(), self.subject))
+
+    def test_a_foreign_driver_is_not(self) -> None:
+        reason = unsnapshotted(self.backing(driver="rexray"), self.subject)
+        self.assertIn("rexray", reason)
+
+    def test_declared_driver_options_are_not(self) -> None:
+        reason = unsnapshotted(
+            self.backing(options={"type": "nfs", "device": ":/exports/app"}),
+            self.subject,
+        )
+        self.assertIn("backing store", reason)
+
+    def test_the_declaration_decides_not_the_mount_table(self) -> None:
+        """Docker unmounts an NFS volume when its last container stops."""
+        with mock.patch.object(os.path, "ismount", return_value=False):
+            reason = unsnapshotted(self.backing(options={"type": "nfs"}), self.subject)
+        self.assertIsNotNone(reason)
+
+    def test_a_volume_without_a_mountpoint_is_not(self) -> None:
+        reason = unsnapshotted(Backing(""), self.subject)
+        self.assertIn("no mountpoint", reason)
+
+    def test_an_own_mount_is_not(self) -> None:
+        with mock.patch.object(os.path, "ismount", return_value=True):
+            reason = unsnapshotted(self.backing(), self.subject)
+        self.assertIn("own mount", reason)
+
+    def test_a_filesystem_boundary_is_not(self) -> None:
+        real = os.stat
+
+        def crossing(path, *args, **kwargs):
+            info = real(path, *args, **kwargs)
+            if os.path.realpath(path) == os.path.realpath(self.mountpoint):
+                return os.stat_result(
+                    (info.st_mode, info.st_ino, info.st_dev + 1, *tuple(info)[3:])
+                )
+            return info
+
+        with mock.patch.object(os, "stat", side_effect=crossing):
+            reason = unsnapshotted(self.backing(), self.subject)
+        self.assertIn("filesystem boundary", reason)
+
+    def test_an_unreadable_mountpoint_is_not(self) -> None:
+        reason = unsnapshotted(
+            self.backing(mountpoint=os.path.join(self.subject, "gone")), self.subject
+        )
+        self.assertIn("could not be read", reason)
+
+
+class TestSnapshotSource(unittest.TestCase):
+    def setUp(self) -> None:
+        self.subject = tempfile.mkdtemp()
+        self.mountpoint = os.path.join(self.subject, "volumes", "app", "_data")
+        os.makedirs(self.mountpoint)
+        self.snapshot = os.path.join(
+            self.subject, ".baudolo-tag", "volumes", "app", "_data"
+        )
+        os.makedirs(self.snapshot)
+        self.backing = Backing(self.mountpoint)
+
+    def test_a_captured_volume_reads_from_the_snapshot(self) -> None:
+        source, reason = snapshot_source(
+            lambda path: self.snapshot + "/", self.backing, self.subject
+        )
+        self.assertEqual(source, self.snapshot + "/")
+        self.assertEqual(reason, "")
+
+    def test_an_uncaptured_volume_is_refused_before_the_resolver_runs(self) -> None:
+        def resolve(path):
+            raise AssertionError("must not resolve a volume the snapshot misses")
+
+        source, reason = snapshot_source(
+            resolve, Backing(self.mountpoint, options={"type": "nfs"}), self.subject
+        )
+        self.assertIsNone(source)
+        self.assertIn("backing store", reason)
+
+    def test_a_volume_outside_the_subject_degrades_instead_of_raising(self) -> None:
+        def resolve(path):
+            raise SnapshotError(f"{path} lies outside the snapshot subject")
+
+        source, reason = snapshot_source(resolve, self.backing, self.subject)
+        self.assertIsNone(source)
+        self.assertIn("lies outside", reason)
+
+    def test_a_volume_created_after_the_snapshot_degrades(self) -> None:
+        source, reason = snapshot_source(
+            lambda path: os.path.join(self.subject, "absent") + "/",
+            self.backing,
+            self.subject,
+        )
+        self.assertIsNone(source)
+        self.assertIn("created after", reason)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/restore/test_cli_version_flag.py

@@ -0,0 +1,52 @@
+import unittest
+from unittest.mock import patch
+
+from baudolo.restore import __main__ as cli
+
+ENGINES = {
+    "postgres": ("restore_postgres_sql", ["--db-name", "app"]),
+    "mariadb": ("restore_mariadb_sql", ["--db-name", "app"]),
+    "cluster": ("restore_cluster_sql", ["--instance", "central", "--db-user", "root"]),
+}
+
+
+class TestVersionFlagReachesEveryEngine(unittest.TestCase):
+    def call(self, engine: str, extra: list) -> dict:
+        target, required = ENGINES[engine]
+        argv = [
+            engine,
+            "app_vol",
+            "hash",
+            "20260817000000",
+            "--container",
+            "db",
+            "--db-password",
+            "pw",
+            *required,
+            *extra,
+        ]
+        with patch.object(cli, target) as restore:
+            self.assertEqual(cli.main(argv), 0)
+        return restore.call_args.kwargs
+
+    def test_the_gate_is_on_by_default(self) -> None:
+        for engine in ENGINES:
+            with self.subTest(engine=engine):
+                self.assertTrue(self.call(engine, [])["check_version"])
+
+    def test_the_flag_turns_it_off(self) -> None:
+        for engine in ENGINES:
+            with self.subTest(engine=engine):
+                kwargs = self.call(engine, ["--no-version-check"])
+                self.assertFalse(kwargs["check_version"])
+
+    def test_empty_stays_independent_of_the_gate(self) -> None:
+        for engine in ENGINES:
+            with self.subTest(engine=engine):
+                kwargs = self.call(engine, ["--empty"])
+                self.assertTrue(kwargs["empty"])
+                self.assertTrue(kwargs["check_version"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/restore/test_cluster_replay.py

@@ -24,6 +24,7 @@ class TestClusterReplay(unittest.TestCase):
                     password="pw",
                     sql_path=sql.name,
                     empty=empty,
+                    check_version=False,
                 )
         return calls
 
@@ -107,6 +108,7 @@ class TestClusterReplay(unittest.TestCase):
                 password="pw",
                 sql_path="/nonexistent/x.cluster.backup.sql",
                 empty=False,
+                check_version=False,
             )
 
     def test_the_path_helper_names_the_dumpall_file(self) -> None:

tests/unit/restore/test_files_backing_store.py

@@ -0,0 +1,62 @@
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+from baudolo.restore import files as files_mod
+
+
+class TestBackingStoreGuard(unittest.TestCase):
+    def restore(self, inspect: str, mounted: bool) -> tuple[int, list]:
+        calls = []
+
+        def _run(cmd, **kwargs):
+            calls.append(cmd)
+            return MagicMock(stdout=inspect.encode())
+
+        with (
+            patch.object(files_mod, "docker_volume_exists", return_value=True),
+            patch.object(files_mod.os.path, "ismount", return_value=mounted),
+            patch.object(files_mod, "run", side_effect=_run),
+        ):
+            code = files_mod.restore_volume_files("app_data", tempfile.mkdtemp())
+        return code, calls
+
+    def rsynced(self, calls: list) -> bool:
+        return any(cmd[0] == "rsync" for cmd in calls)
+
+    def test_plain_local_volume_is_restored_unmounted(self) -> None:
+        code, calls = self.restore("/var/lib/docker/volumes/a/_data|local|plain", False)
+        self.assertEqual(code, 0)
+        self.assertTrue(self.rsynced(calls))
+
+    def test_volume_with_driver_options_is_refused_while_unmounted(self) -> None:
+        code, calls = self.restore("/var/lib/docker/volumes/a/_data|local|opts", False)
+        self.assertEqual(code, 2)
+        self.assertFalse(
+            self.rsynced(calls),
+            "an NFS or bind volume writes under the mount and reports success",
+        )
+
+    def test_volume_with_driver_options_is_restored_once_mounted(self) -> None:
+        code, calls = self.restore("/var/lib/docker/volumes/a/_data|local|opts", True)
+        self.assertEqual(code, 0)
+        self.assertTrue(self.rsynced(calls))
+
+    def test_foreign_driver_is_refused_while_unmounted(self) -> None:
+        code, calls = self.restore("/mnt/gluster/a|glusterfs|plain", False)
+        self.assertEqual(code, 2)
+        self.assertFalse(self.rsynced(calls))
+
+    def test_an_unresolvable_mountpoint_still_fails_first(self) -> None:
+        code, calls = self.restore("|local|plain", False)
+        self.assertEqual(code, 2)
+        self.assertFalse(self.rsynced(calls))
+
+    def test_a_format_without_the_new_fields_is_treated_as_plain(self) -> None:
+        code, calls = self.restore("/var/lib/docker/volumes/a/_data", False)
+        self.assertEqual(code, 0)
+        self.assertTrue(self.rsynced(calls))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/restore/test_mariadb_empty_drop.py

@@ -29,6 +29,7 @@ class TestMariadbEmptyDrop(unittest.TestCase):
                     password="pw",
                     sql_path=sql.name,
                     empty=True,
+                    check_version=False,
                 )
 
         drop_calls = [argv for argv in calls if any("DROP TABLE" in a for a in argv)]

tests/unit/restore/test_postgres_single_transaction.py

@@ -24,6 +24,7 @@ class TestPostgresSingleTransaction(unittest.TestCase):
                     password="pw",
                     sql_path=sql.name,
                     empty=True,
+                    check_version=False,
                 )
 
         self.assertEqual(len(calls), 2, f"expected pre-clean + replay: {calls}")

tests/unit/restore/test_version_gate.py

@@ -0,0 +1,244 @@
+import os
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+from baudolo.restore.db import cluster as cluster_mod
+from baudolo.restore.db import mariadb as mdb_mod
+from baudolo.restore.db import postgres as pg_mod
+from baudolo.restore.db import version as ver
+
+POSTGRES_HEADER = """--
+-- PostgreSQL database dump
+--
+
+\\restrict BbyzwODc1rWKL3rDyLhEjgCF0Kf2TU5ma7gcTs8eQI7copLtydXkc61zdULsPav
+
+-- Dumped from database version 17.11
+-- Dumped by pg_dump version 17.11
+
+SET statement_timeout = 0;
+"""
+
+MARIADB_HEADER = """/*M!999999\\- enable the sandbox mode */
+-- MariaDB dump 10.19-11.8.8-MariaDB, for debian-linux-gnu (x86_64)
+--
+-- Host: 127.0.0.1    Database: mysql
+-- ------------------------------------------------------
+-- Server version\t11.8.8-MariaDB-ubu2404
+
+/*!40101 SET @OLD_CHARACTER_SET_CLIENT=@@CHARACTER_SET_CLIENT */;
+"""
+
+
+def cluster_header(roles: int) -> str:
+    """A pg_dumpall stream: banner, N roles, then the first database's dump."""
+    head = [
+        "--",
+        "-- PostgreSQL database cluster dump",
+        "--",
+        "",
+        "SET default_transaction_read_only = off;",
+        "",
+        "--",
+        "-- Roles",
+        "--",
+    ]
+    for i in range(roles):
+        head.append(f'CREATE ROLE "app{i}";')
+        head.append(f'ALTER ROLE "app{i}" WITH NOSUPERUSER INHERIT LOGIN;')
+    head.append("\\connect app")
+    head.append("")
+    return "\n".join(head) + "\n" + POSTGRES_HEADER
+
+
+def dump_file(text: str) -> str:
+    path = os.path.join(tempfile.mkdtemp(), "app.backup.sql")
+    with open(path, "w", encoding="utf-8") as handle:
+        handle.write(text)
+    return path
+
+
+class TestDumpVersion(unittest.TestCase):
+    """Headers captured from postgres:17-alpine and mariadb:11 themselves."""
+
+    def test_postgres_reads_the_source_server(self) -> None:
+        self.assertEqual(
+            ver.dump_version(dump_file(POSTGRES_HEADER), "postgres"), "17.11"
+        )
+
+    def test_mariadb_reads_the_server_not_the_dump_tool(self) -> None:
+        found = ver.dump_version(dump_file(MARIADB_HEADER), "mariadb")
+        self.assertEqual(found, "11.8.8-MariaDB-ubu2404")
+        self.assertNotEqual(
+            ver.major_of(found),
+            10,
+            "10.19 is mariadb-dump's own version, not the server's",
+        )
+
+    def test_cluster_dump_states_its_version_far_below_the_header(self) -> None:
+        path = dump_file(cluster_header(roles=200))
+        with open(path, encoding="utf-8") as handle:
+            offset = next(i for i, line in enumerate(handle) if "Dumped from" in line)
+        self.assertGreater(offset, 100, "fixture must exercise the deep scan")
+        self.assertEqual(ver.dump_version(path, "postgres"), "17.11")
+
+    def test_a_version_beyond_the_scan_limit_is_refused_not_ignored(self) -> None:
+        path = dump_file(cluster_header(roles=ver.SCAN_LINES))
+        with self.assertRaises(ver.VersionMismatch):
+            ver.dump_version(path, "postgres")
+
+    def test_a_dump_without_a_version_header_is_refused(self) -> None:
+        path = dump_file("CREATE TABLE t (id int);\n")
+        with self.assertRaises(ver.VersionMismatch):
+            ver.dump_version(path, "postgres")
+
+
+class TestMajorOf(unittest.TestCase):
+    def test_reads_the_leading_number(self) -> None:
+        self.assertEqual(ver.major_of("17.11"), 17)
+        self.assertEqual(ver.major_of("11.8.8-MariaDB-ubu2404"), 11)
+        self.assertEqual(ver.major_of("9.6.24"), 9)
+        self.assertEqual(ver.major_of("18beta1"), 18)
+
+    def test_refuses_an_unreadable_version(self) -> None:
+        with self.assertRaises(ver.VersionMismatch):
+            ver.major_of("unknown")
+
+
+class TestAssertReplayable(unittest.TestCase):
+    def test_newer_dump_into_older_engine_is_refused(self) -> None:
+        with self.assertRaises(ver.VersionMismatch) as caught:
+            ver.assert_replayable("/b/app.sql", "postgres", "17.11", "15.6")
+        self.assertIn("17.11", str(caught.exception))
+        self.assertIn("15.6", str(caught.exception))
+
+    def test_same_major_passes(self) -> None:
+        ver.assert_replayable("/b/app.sql", "postgres", "17.4", "17.11")
+
+    def test_older_dump_into_newer_engine_passes(self) -> None:
+        ver.assert_replayable("/b/app.sql", "postgres", "15.6", "17.11")
+
+
+class TestServerVersion(unittest.TestCase):
+    def test_postgres_asks_over_pgpassword(self) -> None:
+        with patch.object(ver, "docker_exec") as exec_:
+            exec_.return_value = MagicMock(stdout=b" 17.11 \n")
+            found = ver.server_version("db", "postgres", "app", "pw")
+        self.assertEqual(found, "17.11")
+        argv = exec_.call_args.args[1]
+        self.assertIn("SHOW server_version", argv)
+        self.assertEqual(exec_.call_args.kwargs["docker_env"], {"PGPASSWORD": "pw"})
+
+    def test_mariadb_asks_through_the_detected_client(self) -> None:
+        with patch.object(ver, "docker_exec") as exec_:
+            exec_.return_value = MagicMock(stdout=b"11.8.8-MariaDB-ubu2404\n")
+            found = ver.server_version("db", "mariadb", "app", "pw", client="mysql")
+        self.assertEqual(found, "11.8.8-MariaDB-ubu2404")
+        self.assertEqual(exec_.call_args.args[1][0], "mysql")
+
+
+class TestGateStopsBeforeDestroying(unittest.TestCase):
+    """--empty drops in one session and replays in the next, with no rollback
+    between them, so the refusal has to land before the first session."""
+
+    def setUp(self) -> None:
+        self.serving = patch.object(ver, "docker_exec").start()
+        self.addCleanup(patch.stopall)
+
+    def serve(self, version: str) -> None:
+        self.serving.return_value = MagicMock(stdout=version.encode())
+
+    def test_postgres_refuses_without_running_the_preclean(self) -> None:
+        self.serve("15.6")
+        path = dump_file(POSTGRES_HEADER)
+        with (
+            patch.object(pg_mod, "docker_exec") as replay,
+            self.assertRaises(ver.VersionMismatch),
+        ):
+            pg_mod.restore_postgres_sql(
+                container="db",
+                db_name="app",
+                user="app",
+                password="pw",
+                sql_path=path,
+                empty=True,
+            )
+        replay.assert_not_called()
+
+    def test_cluster_refuses_without_running_the_preclean(self) -> None:
+        self.serve("15.6")
+        path = dump_file(cluster_header(roles=3))
+        with (
+            patch.object(cluster_mod, "docker_exec") as replay,
+            self.assertRaises(ver.VersionMismatch),
+        ):
+            cluster_mod.restore_cluster_sql(
+                container="db",
+                user="postgres",
+                password="pw",
+                sql_path=path,
+                empty=True,
+            )
+        replay.assert_not_called()
+
+    def test_mariadb_refuses_without_dropping_tables(self) -> None:
+        self.serve("10.11.6-MariaDB")
+        path = dump_file(MARIADB_HEADER)
+        with (
+            patch.object(mdb_mod, "_pick_client", return_value="mariadb"),
+            patch.object(mdb_mod, "docker_exec") as replay,
+            self.assertRaises(ver.VersionMismatch),
+        ):
+            mdb_mod.restore_mariadb_sql(
+                container="db",
+                db_name="app",
+                user="app",
+                password="pw",
+                sql_path=path,
+                empty=True,
+            )
+        replay.assert_not_called()
+
+    def test_matching_versions_let_the_replay_through(self) -> None:
+        self.serve("17.11")
+        path = dump_file(POSTGRES_HEADER)
+        with patch.object(pg_mod, "docker_exec") as replay:
+            pg_mod.restore_postgres_sql(
+                container="db",
+                db_name="app",
+                user="app",
+                password="pw",
+                sql_path=path,
+                empty=False,
+            )
+        replay.assert_called_once()
+
+    def test_the_escape_hatch_asks_the_engine_nothing(self) -> None:
+        path = dump_file("CREATE TABLE t (id int);\n")
+        with patch.object(pg_mod, "docker_exec"):
+            pg_mod.restore_postgres_sql(
+                container="db",
+                db_name="app",
+                user="app",
+                password="pw",
+                sql_path=path,
+                empty=False,
+                check_version=False,
+            )
+        self.serving.assert_not_called()
+
+    def test_a_missing_dump_is_reported_as_missing_not_as_a_mismatch(self) -> None:
+        with self.assertRaises(FileNotFoundError):
+            pg_mod.restore_postgres_sql(
+                container="db",
+                db_name="app",
+                user="app",
+                password="pw",
+                sql_path=os.path.join(tempfile.mkdtemp(), "absent.sql"),
+                empty=True,
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()