fix(recover): read the generation's own layout instead of guessing it

A recent refactor imported baudolo into utils.recovery for three layout strings, and cli.administration.recover.__main__ imports full unconditionally - so the whole recovery CLI, --help included, died at import on every host without baudolo. That is most of them: recover device|nfs|secrets runs on swarm nodes, the NFS server and bare rescue hosts, while only the backup node carries svc-bkp-volume-2-local. utils/recovery had been deliberately stdlib-only for exactly that reason.

baudolo 5.0.0 writes a manifest into every generation stating the layout it used, so the names are now asked of the generation being read rather than of the installed package. A tree written by any version is walked with the names that version used, and utils.recovery.layout answers only for generations older than the manifest, which cannot describe themselves. That also retires two of core's guesses: the '<entity>_database' naming convention now yields to the engine the run actually recorded.

A repo whose manifest cannot be read is skipped rather than guessed at, but no longer takes the other repos with it - recovering two of three beats recovering none because one manifest is corrupt. The exit code reports the gap.

The pin lint compared two declared version strings and was therefore green while a different distribution was installed. It now checks the installed distribution and the layout literals against baudolo's own BackupPaths.

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

cli/administration/recover/full.py

@@ -19,7 +19,7 @@ import sys
 from pathlib import Path
 
 from cli.administration.recover import paths, recoverers, remote
-from utils.recovery.databases import DUMP_GLOB, FILES_DIR, FILES_GLOB
+from utils.recovery import manifest
 
 _REPOS: tuple[tuple[str, str], ...] = (
     ("nfs", "backup-nfs-to-local"),
@@ -35,25 +35,39 @@ def _newest_gen(root: Path, repo: str) -> Path | None:
     return generations[-1] if generations else None
 
 
-def plan(root: Path) -> list[tuple[str, str]]:
+def plan(root: Path) -> tuple[list[tuple[str, str]], list[str]]:
     """(type, source) for each present repo, in order; volume per-volume.
 
     The volume repo contributes a database step after its file steps: the
     dumps have to land once the volumes are back and while the consumers are
     still down, which is where the caller has them.
+
+    Returns:
+        The steps, and the repos whose manifest could not be read. A repo that
+        cannot state its layout is skipped rather than guessed at, but it must
+        not take the others with it: recovering two repos of three beats
+        recovering none because one manifest is corrupt.
     """
     steps: list[tuple[str, str]] = []
+    unreadable: list[str] = []
     for rtype, repo in _REPOS:
         generation = _newest_gen(root, repo)
         if generation is None:
             continue
+        try:
+            names = manifest.layout_of(generation)
+        except ValueError as error:
+            unreadable.append(f"{repo}: {error}")
+            continue
+        files_glob, dump_glob = manifest.globs_of(names)
+        files_dir = names["files_dir"]
         if rtype == "volume":
-            steps += [(rtype, str(vol)) for vol in sorted(generation.glob(FILES_GLOB))]
-            if any(generation.glob(DUMP_GLOB)):
+            steps += [(rtype, str(vol)) for vol in sorted(generation.glob(files_glob))]
+            if any(generation.glob(dump_glob)):
                 steps.append(("database", str(generation)))
-        elif (generation / FILES_DIR).is_dir():
-            steps.append((rtype, str(generation / FILES_DIR)))
-    return steps
+        elif (generation / files_dir).is_dir():
+            steps.append((rtype, str(generation / files_dir)))
+    return steps, unreadable
 
 
 def _pull_cmd(source: str) -> list[str]:
@@ -78,20 +92,22 @@ def _device_restore_root(source: str) -> str:
 def _recover_under(
     root: Path, target: str, *, preview: bool, service_backup: bool
 ) -> int:
-    steps = plan(root)
+    steps, unreadable = plan(root)
+    for problem in unreadable:
+        print(f"# full: SKIPPED {problem}", file=sys.stderr)
     if not steps:
         print(f"# full: no backup repos found under {root}")
         return 1
-    returncode = 0
+    returncode = 2 if unreadable else 0
     for rtype, files_dir in steps:
         cmd = recoverers.RECOVERERS[rtype].command(
             files_dir, service_backup=service_backup, target=target
         )
         print(f"# full {rtype}  {target}: {shlex.join(cmd)}")
         if not preview:
-            returncode = subprocess.run(cmd, check=False).returncode
-            if returncode != 0:
-                return returncode
+            failed = subprocess.run(cmd, check=False).returncode
+            if failed != 0:
+                return failed
     return returncode
 
 

pyproject.toml

@@ -13,7 +13,7 @@ license = { file = "LICENSE.md" }
 dependencies = [
   # Pinned: utils/recovery reads the generation layout from baudolo's own
   # BackupPaths, so the version here and in svc-bkp-volume-2-local must match.
-  "backup-docker-to-local==4.0.0",
+  "backup-docker-to-local==5.0.0",
   "bcrypt",
   "cryptography",
   "humanfriendly",

roles/svc-bkp-volume-2-local/files/test/seed/marker.py

@@ -30,7 +30,7 @@ from pathlib import Path
 sys.path.insert(0, os.environ["BKP_TEST_REPO_ROOT"])
 
 from utils.paths import FILE_DATABASE_SECRETS
-from utils.recovery import databases
+from utils.recovery import databases, manifest
 from utils.recovery import docker as recovery_docker
 
 MARKER_FILE = ".dr-drill-marker"
@@ -98,9 +98,8 @@ def volume_mountpoint(volume: str) -> Path:
 
 def file_volumes(generation_dir: Path) -> list[str]:
     """The volumes this generation stores as a file tree."""
-    return sorted(
-        path.parent.name for path in generation_dir.glob(databases.FILES_GLOB)
-    )
+    files_glob, _dump_glob = manifest.globs_of(manifest.layout_of(generation_dir))
+    return sorted(path.parent.name for path in generation_dir.glob(files_glob))
 
 
 def sql(
@@ -124,7 +123,7 @@ def each_database(generation_dir: Path):
     per database it recreates: the marker has to sit in each of them, or a
     replay that brings back only part of an instance would still pass.
     """
-    engines = databases.engine_by_key()
+    engines = databases.engine_by_key(generation_dir=generation_dir)
     credentials = databases.credentials_of(FILE_DATABASE_SECRETS)
     dumps, clusters = databases.dumps_of(generation_dir)
     for dump in dumps:
@@ -171,8 +170,9 @@ def captured(generation_dir: Path, token: str) -> int:
     the row, fails here - long before anything is torn down.
     """
     missing: list[str] = []
+    files_dir = manifest.layout_of(generation_dir)["files_dir"]
     for volume in file_volumes(generation_dir):
-        marker = generation_dir / volume / databases.FILES_DIR / MARKER_FILE
+        marker = generation_dir / volume / files_dir / MARKER_FILE
         if not marker.is_file() or marker.read_text(encoding="utf-8").strip() != token:
             missing.append(
                 f"volume {volume}: the generation holds no current {MARKER_FILE}"

roles/svc-bkp-volume-2-local/tasks/01_install.yml

@@ -3,7 +3,7 @@
   include_role:
     name: sys-pip-install
   vars:
-    package_name: "backup-docker-to-local==4.0.0"
+    package_name: "backup-docker-to-local==5.0.0"
     global_install: true
 
 - name: ✅ Assert required baudolo commands are available

tests/lint/repository/dependencies/test_baudolo_pin.py

@@ -1,10 +1,14 @@
-"""Lint guard: baudolo is pinned to one version, not two.
+"""Lint guard: baudolo is pinned to one version, and core agrees with it.
 
 `svc-bkp-volume-2-local` pip-installs baudolo on the target host, while
-`[project.dependencies]` declares it for the control host, whose
-`utils.recovery` imports `baudolo.restore.paths` to read the generation layout.
-Two pins mean the code that reads a backup can be a different version from the
-one that wrote it, and the mismatch surfaces as a restore that finds nothing.
+`[project.dependencies]` declares it for the control host. Two pins mean the
+code that reads a backup can be a different version from the one that wrote it,
+and the mismatch surfaces as a restore that finds nothing.
+
+`utils.recovery.layout` spells the generation layout out as literals, because
+`recover device|nfs|secrets|volume` runs on hosts that carry a git checkout and
+no baudolo. This is where that copy is held to baudolo's own `BackupPaths`, so
+the literals cannot drift from the tool that writes them.
 """
 
 from __future__ import annotations
@@ -12,8 +16,11 @@ from __future__ import annotations
 import re
 import tomllib
 import unittest
+from importlib.metadata import PackageNotFoundError, version
+from pathlib import PurePath
 
 from utils.cache.files import read_text
+from utils.recovery import layout
 
 from . import PROJECT_ROOT
 
@@ -47,6 +54,63 @@ class TestBaudoloPin(unittest.TestCase):
             f"'{declared}'. Both must name the same version.",
         )
 
+    def test_baudolo_is_installed_here(self) -> None:
+        try:
+            version(_PACKAGE)
+        except PackageNotFoundError:
+            self.fail(
+                f"{_PACKAGE} is declared in {_PYPROJECT} but not installed for this "
+                f"interpreter, so the layout assertion below would be vacuous. Run "
+                f"'make install-python-dev'."
+            )
+
+    def test_the_manifest_contract_matches_baudolo(self) -> None:
+        from baudolo import generation
+
+        self.assertEqual(layout.MANIFEST_FILE, generation.MANIFEST_FILE)
+        self.assertEqual(layout.MANIFEST_SCHEMA, generation.MANIFEST_SCHEMA)
+
+    def test_the_layout_literals_match_baudolo(self) -> None:
+        from baudolo.restore.paths import BackupPaths
+
+        installed = version(_PACKAGE)
+        paths = BackupPaths("", "", "", repo_name="", backups_dir="")
+        for name, ours, theirs in (
+            ("SQL_DIR", layout.SQL_DIR, PurePath(paths.sql_file("")).parent.name),
+            ("FILES_DIR", layout.FILES_DIR, PurePath(paths.files_dir()).name),
+            ("DUMP_SUFFIX", layout.DUMP_SUFFIX, PurePath(paths.sql_file("")).name),
+            (
+                "CLUSTER_SUFFIX",
+                layout.CLUSTER_SUFFIX,
+                PurePath(paths.cluster_file("")).name,
+            ),
+        ):
+            with self.subTest(constant=name):
+                self.assertEqual(
+                    ours,
+                    theirs,
+                    f"utils.recovery.layout.{name} is '{ours}', while "
+                    f"{_PACKAGE} {installed} writes '{theirs}'. A restore would "
+                    f"look in the wrong place; update the literal, do not import "
+                    f"baudolo here.",
+                )
+
+    def test_the_layout_module_imports_nothing(self) -> None:
+        source = read_text(str(PROJECT_ROOT / "utils/recovery/layout.py"))
+        offenders = [
+            line
+            for line in source.splitlines()
+            if re.match(r"^\s*(import|from)\s", line)
+            and "__future__" not in line
+        ]
+        self.assertEqual(
+            offenders,
+            [],
+            "utils/recovery/layout.py must stay import-free: it is read on swarm "
+            "nodes, the NFS server and rescue hosts that have nothing but a git "
+            f"checkout. Found {offenders}.",
+        )
+
 
 if __name__ == "__main__":  # pragma: no cover
     unittest.main()

tests/unit/cli/administration/recover/test_full.py

@@ -27,7 +27,7 @@ class PlanTest(unittest.TestCase):
         with tempfile.TemporaryDirectory() as td:
             root = Path(td)
             self._tree(root)
-            steps = full.plan(root)
+            steps, _unreadable = full.plan(root)
             types = [t for t, _ in steps]
             self.assertEqual(types, ["nfs", "volume", "volume", "secrets"])
             self.assertTrue(
@@ -42,13 +42,13 @@ class PlanTest(unittest.TestCase):
             repo = root / "HASH" / "backup-nfs-to-local"
             (repo / "20260101000000" / "files").mkdir(parents=True)
             (repo / "20260709120000" / "files").mkdir(parents=True)
-            steps = full.plan(root)
+            steps, _unreadable = full.plan(root)
             self.assertEqual(len(steps), 1)
             self.assertIn("20260709120000", steps[0][1])
 
     def test_plan_empty_root(self):
         with tempfile.TemporaryDirectory() as td:
-            self.assertEqual(full.plan(Path(td)), [])
+            self.assertEqual(full.plan(Path(td)), ([], []))
 
 
 class DeviceDetectTest(unittest.TestCase):
@@ -89,3 +89,28 @@ class PullCmdTest(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class TestUnreadableManifest(unittest.TestCase):
+    """One corrupt manifest must not block the repos that are intact."""
+
+    def _root(self) -> Path:
+        root = Path(tempfile.mkdtemp())
+        bad = root / "hash" / "backup-nfs-to-local" / "20260101000000"
+        bad.mkdir(parents=True)
+        (bad / "manifest.json").write_text(
+            '{"schema": 99, "layout": {}, "volumes": {}}', encoding="utf-8"
+        )
+        (
+            root / "hash" / "backup-docker-to-local" / "20260101000000" / "v" / "files"
+        ).mkdir(parents=True)
+        return root
+
+    def test_the_intact_repo_is_still_planned(self) -> None:
+        steps, _unreadable = full.plan(self._root())
+        self.assertEqual([rtype for rtype, _src in steps], ["volume"])
+
+    def test_the_skipped_repo_is_named(self) -> None:
+        _steps, unreadable = full.plan(self._root())
+        self.assertEqual(len(unreadable), 1)
+        self.assertIn("backup-nfs-to-local", unreadable[0])

tests/unit/utils/recovery/test_manifest.py

@@ -0,0 +1,132 @@
+"""What the generation manifest tells a reader, and what it refuses to guess."""
+
+from __future__ import annotations
+
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from utils.recovery import layout, manifest
+from utils.recovery.layout import MANIFEST_FILE, MANIFEST_SCHEMA
+
+
+def generation(
+    volumes: dict, schema: int = MANIFEST_SCHEMA, layout_names: dict | None = None
+) -> Path:
+    directory = Path(tempfile.mkdtemp())
+    (directory / MANIFEST_FILE).write_text(
+        json.dumps(
+            {
+                "schema": schema,
+                "layout": layout_names or {},
+                "volumes": volumes,
+            }
+        ),
+        encoding="utf-8",
+    )
+    return directory
+
+
+class TestWithoutManifest(unittest.TestCase):
+    """A generation written before the manifest existed."""
+
+    def test_read_reports_absence_rather_than_an_empty_verdict(self) -> None:
+        self.assertIsNone(manifest.read(tempfile.mkdtemp()))
+
+    def test_the_cli_exits_two_so_a_caller_can_fall_back(self) -> None:
+        self.assertEqual(manifest.main([tempfile.mkdtemp()]), 2)
+
+
+class TestUndumped(unittest.TestCase):
+    def test_a_database_volume_without_a_dump_is_reported(self) -> None:
+        directory = generation(
+            {
+                "discourse_database": {
+                    "database": True,
+                    "dumped": False,
+                    "engine": "postgres",
+                }
+            }
+        )
+        self.assertEqual(
+            manifest.undumped(directory), [("discourse_database", "postgres")]
+        )
+
+    def test_a_dumped_database_volume_is_not(self) -> None:
+        directory = generation(
+            {"pgdata": {"database": True, "dumped": True, "engine": "postgres"}}
+        )
+        self.assertEqual(manifest.undumped(directory), [])
+
+    def test_a_plain_volume_is_not(self) -> None:
+        directory = generation({"assets": {"database": False, "dumped": False}})
+        self.assertEqual(manifest.undumped(directory), [])
+
+    def test_a_database_volume_of_unknown_engine_is_still_reported(self) -> None:
+        directory = generation({"x": {"database": True, "dumped": False}})
+        self.assertEqual(manifest.undumped(directory), [("x", "unknown")])
+
+
+class TestEngineByVolume(unittest.TestCase):
+    def test_it_names_the_engine_the_run_detected(self) -> None:
+        directory = generation(
+            {
+                "pgdata": {"database": True, "dumped": True, "engine": "postgres"},
+                "mysqldata": {"database": True, "dumped": True, "engine": "mariadb"},
+                "assets": {"database": False, "dumped": False},
+            }
+        )
+        self.assertEqual(
+            manifest.engine_by_volume(directory),
+            {"pgdata": "postgres", "mysqldata": "mariadb"},
+        )
+
+
+class TestLayoutOf(unittest.TestCase):
+    """The generation states its own shape; the literals are only a fallback."""
+
+    def test_a_generation_is_walked_with_the_names_it_recorded(self) -> None:
+        directory = generation(
+            {},
+            layout_names={
+                "files_dir": "payload",
+                "sql_dir": "dumps",
+                "dump_suffix": ".sql.gz",
+                "cluster_suffix": ".all.sql.gz",
+            },
+        )
+        self.assertEqual(manifest.layout_of(directory)["files_dir"], "payload")
+        self.assertEqual(
+            manifest.globs_of(manifest.layout_of(directory)),
+            ("*/payload", "*/dumps/*.sql.gz"),
+        )
+
+    def test_a_generation_without_a_manifest_falls_back_to_the_literals(self) -> None:
+        names = manifest.layout_of(tempfile.mkdtemp())
+        self.assertEqual(names["files_dir"], layout.FILES_DIR)
+        self.assertEqual(names["sql_dir"], layout.SQL_DIR)
+
+    def test_the_fallback_globs_match_the_literal_layout(self) -> None:
+        files_glob, dump_glob = manifest.globs_of(
+            manifest.layout_of(tempfile.mkdtemp())
+        )
+        self.assertEqual(files_glob, f"*/{layout.FILES_DIR}")
+        self.assertEqual(dump_glob, f"*/{layout.SQL_DIR}/*{layout.DUMP_SUFFIX}")
+
+
+class TestSchemaGuard(unittest.TestCase):
+    def test_a_newer_schema_is_refused_rather_than_guessed_at(self) -> None:
+        directory = generation({}, schema=MANIFEST_SCHEMA + 1)
+        with self.assertRaises(ValueError):
+            manifest.read(directory)
+
+    def test_a_corrupt_manifest_is_refused(self) -> None:
+        directory = Path(tempfile.mkdtemp())
+        (directory / MANIFEST_FILE).write_text("{not json", encoding="utf-8")
+        with self.assertRaises(ValueError):
+            manifest.read(directory)
+
+
+if __name__ == "__main__":
+    unittest.main()

utils/recovery/databases.py

@@ -10,10 +10,10 @@ Whether a dump is readable by that engine is decided by ``baudolo-restore``
 itself, before its pre-clean drops anything.
 
 The generation layout belongs to baudolo, which writes it, so the file names
-and subdirectory below are asked of its ``BackupPaths`` instead of restated
-here. Restating them would let a rename on its side leave the globs matching
-nothing, which reads as "this generation holds no databases" rather than as a
-failure.
+and subdirectory below are read from the generation's own manifest rather than
+restated here. Restating them would let a rename on its side leave the globs
+matching nothing, which reads as "this generation holds no databases" rather
+than as a failure.
 
 A ``database = '*'`` row dumps a whole instance with ``pg_dumpall``. It goes
 back through ``baudolo-restore cluster`` with the instance's superuser; its
@@ -23,15 +23,14 @@ project of the engine, so both are checked.
 
 from __future__ import annotations
 
-from pathlib import PurePath
 from typing import TYPE_CHECKING, NamedTuple
 
 from baudolo.databases import CLUSTER_ROW, DatabasesCsvError, Row, read_rows
 from baudolo.restore.db.cluster import dump_inventory
-from baudolo.restore.paths import BackupPaths
 
 from utils.cache.applications import get_application_defaults
 from utils.recovery import docker as recovery_docker
+from utils.recovery import manifest
 from utils.recovery.docker import RecoveryError
 from utils.roles.applications.services.database import (
     RDBMS_SERVICE_KEYS,
@@ -45,14 +44,6 @@ if TYPE_CHECKING:
 
 RESTORE_BIN = "baudolo-restore"
 
-_LAYOUT = BackupPaths("", "", "", repo_name="", backups_dir="")
-DUMP_SUFFIX = PurePath(_LAYOUT.sql_file("")).name
-CLUSTER_SUFFIX = PurePath(_LAYOUT.cluster_file("")).name
-SQL_DIR = PurePath(_LAYOUT.sql_file("")).parent.name
-FILES_DIR = PurePath(_LAYOUT.files_dir()).name
-DUMP_GLOB = f"*/{SQL_DIR}/*{DUMP_SUFFIX}"
-FILES_GLOB = f"*/{FILES_DIR}"
-
 DEDICATED_VOLUME_SUFFIX = "_database"
 CLUSTER_ENGINE = "postgres"
 CONTROL_DATABASES = frozenset({"postgres", "template0", "template1"})
@@ -83,12 +74,19 @@ class Generation(NamedTuple):
     name: str
 
 
-def engine_by_key(applications: Mapping[str, object] | None = None) -> dict[str, str]:
+def engine_by_key(
+    applications: Mapping[str, object] | None = None,
+    generation_dir: Path | None = None,
+) -> dict[str, str]:
     """Map every name a dump can be addressed by to its database engine.
 
     Args:
         applications: the materialised application payload; loaded from the
             repository when omitted.
+        generation_dir: a generation whose manifest states, per volume, the
+            engine the run actually detected. Those entries win: they are
+            observed, where everything derived from the repository is inferred
+            from naming. A generation without a manifest contributes nothing.
 
     Returns:
         A mapping of docker volume names and database names to ``postgres``
@@ -106,6 +104,8 @@ def engine_by_key(applications: Mapping[str, object] | None = None) -> dict[str,
         entity = get_entity_name(application_id)
         mapping[f"{entity}{DEDICATED_VOLUME_SUFFIX}"] = engine
         mapping[entity] = engine
+    if generation_dir is not None:
+        mapping.update(manifest.engine_by_volume(generation_dir))
     return mapping
 
 
@@ -130,12 +130,16 @@ def dumps_of(generation_dir: Path) -> tuple[list[Dump], list[Cluster]]:
     """
     dumps: list[Dump] = []
     clusters: list[Cluster] = []
-    for path in sorted(generation_dir.glob(DUMP_GLOB)):
+    names = manifest.layout_of(generation_dir)
+    cluster_suffix = names["cluster_suffix"]
+    dump_suffix = names["dump_suffix"]
+    _files_glob, dump_glob = manifest.globs_of(names)
+    for path in sorted(generation_dir.glob(dump_glob)):
         volume = path.parent.parent.name
-        if path.name.endswith(CLUSTER_SUFFIX):
-            clusters.append(Cluster(volume, path.name[: -len(CLUSTER_SUFFIX)], path))
+        if path.name.endswith(cluster_suffix):
+            clusters.append(Cluster(volume, path.name[: -len(cluster_suffix)], path))
             continue
-        dumps.append(Dump(volume, path.name[: -len(DUMP_SUFFIX)], path))
+        dumps.append(Dump(volume, path.name[: -len(dump_suffix)], path))
     return dumps, clusters
 
 

utils/recovery/layout.py

@@ -0,0 +1,23 @@
+"""Fallback shape for a generation that cannot state its own.
+
+baudolo owns these values and writes them into every generation's manifest, so
+:func:`utils.recovery.manifest.layout_of` is what callers ask. These literals
+answer only for generations written before the manifest existed - those cannot
+describe themselves, and nothing else can tell a reader what they used.
+
+Literals rather than an import because the code that reads a generation runs on
+swarm nodes, the NFS server and bare rescue hosts, which carry a git checkout
+and no baudolo; this module must stay import-free for that reason. The
+agreement is enforced by
+``tests/lint/repository/dependencies/test_baudolo_pin.py``.
+"""
+
+from __future__ import annotations
+
+SQL_DIR = "sql"
+FILES_DIR = "files"
+DUMP_SUFFIX = ".backup.sql"
+CLUSTER_SUFFIX = ".cluster.backup.sql"
+
+MANIFEST_FILE = "manifest.json"
+MANIFEST_SCHEMA = 1

utils/recovery/manifest.py

@@ -0,0 +1,119 @@
+"""Read the manifest a baudolo generation carries, with nothing but ``json``.
+
+Import-light for the same reason as :mod:`utils.recovery.layout`: this runs on
+hosts that have no baudolo.
+
+Generations written before the manifest existed carry none, so :func:`read`
+reports absence rather than an empty verdict and callers can fall back.
+
+Exit codes as a module: 0 with one ``<volume>\\t<engine>`` line per undumped
+database volume, 2 when the generation carries no manifest, 64 on bad usage.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+from utils.recovery import layout as fallback
+from utils.recovery.layout import MANIFEST_FILE, MANIFEST_SCHEMA
+
+
+def read(generation_dir: Path | str) -> dict | None:
+    """The manifest document, or None when the generation carries none.
+
+    Args:
+        generation_dir: the ``<backups>/<hash>/<repo>/<generation>`` directory.
+
+    Raises:
+        ValueError: the manifest is unreadable, or states a schema this code
+            does not know. A newer schema must not be guessed at: the fields
+            this reader relies on may mean something else in it.
+    """
+    path = Path(generation_dir) / MANIFEST_FILE
+    if not path.is_file():
+        return None
+    try:
+        document = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as error:
+        raise ValueError(f"{path} is unreadable: {error}") from error
+    schema = document.get("schema")
+    if schema != MANIFEST_SCHEMA:
+        raise ValueError(
+            f"{path} states schema {schema!r}, this reader knows {MANIFEST_SCHEMA}"
+        )
+    return document
+
+
+def layout_of(generation_dir: Path | str) -> dict[str, str]:
+    """The names payload is stored under in THIS generation.
+
+    Read from the generation's own manifest, so a tree written by any version
+    is walked with the names that version used. Generations older than the
+    manifest cannot say, and only for those does the fallback in
+    :mod:`utils.recovery.layout` apply.
+    """
+    document = read(generation_dir)
+    if document is not None and document.get("layout"):
+        return document["layout"]
+    return {
+        "files_dir": fallback.FILES_DIR,
+        "sql_dir": fallback.SQL_DIR,
+        "dump_suffix": fallback.DUMP_SUFFIX,
+        "cluster_suffix": fallback.CLUSTER_SUFFIX,
+    }
+
+
+def globs_of(names: dict[str, str]) -> tuple[str, str]:
+    """``(files_glob, dump_glob)`` for the layout *names* describe.
+
+    Takes the names rather than the directory so a caller that needs both
+    reads the manifest once.
+    """
+    return (
+        f"*/{names['files_dir']}",
+        f"*/{names['sql_dir']}/*{names['dump_suffix']}",
+    )
+
+
+def volumes(generation_dir: Path | str) -> dict[str, dict]:
+    """Every volume the generation recorded, empty for an old generation."""
+    document = read(generation_dir)
+    return {} if document is None else document.get("volumes", {})
+
+
+def undumped(generation_dir: Path | str) -> list[tuple[str, str]]:
+    """``(volume, engine)`` for every database volume that carries no dump."""
+    return sorted(
+        (name, state.get("engine") or "unknown")
+        for name, state in volumes(generation_dir).items()
+        if state.get("database") and not state.get("dumped")
+    )
+
+
+def engine_by_volume(generation_dir: Path | str) -> dict[str, str]:
+    """The engine each database volume was dumped from, as the run detected it."""
+    return {
+        name: state["engine"]
+        for name, state in volumes(generation_dir).items()
+        if state.get("engine")
+    }
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = sys.argv[1:] if argv is None else argv
+    if len(args) != 1:
+        print(
+            "usage: python -m utils.recovery.manifest <generation-dir>", file=sys.stderr
+        )
+        return 64
+    if read(args[0]) is None:
+        return 2
+    for volume, engine in undumped(args[0]):
+        print(f"{volume}\t{engine}")
+    return 0
+
+
+if __name__ == "__main__":  # pragma: no cover
+    sys.exit(main())