feat(sync): 添加已验证的备份与恢复操作

2026-09-09 15:16:52 +08:00
parent e44ead1342
commit a0ba7f55f7
9 changed files with 1141 additions and 27 deletions
@@ -24,4 +24,4 @@ python scripts/phase3-production-acceptance.py `

 报告目录包含 `summary.json`、`case-manifest.json`、`junit.xml`、`cases/<ID>.json` 和脱敏的 `logs/<ID>.log`。摘要记录 commit、各锁文件 SHA-256、配置摘要与已提供安装产物摘要。日志将仓库、数据根、报告根、用户主目录和配置声明的秘密值替换为占位符，并限制为 10 MiB。报告目录必须为空，避免单例复跑覆盖原始证据。

-当前 runner 与失败闭合行为已实现，A-02/A-03 Sidecar、B-01/B-02 凭据、D-01 扩展包、S-01/S-02/S-03/S-08 Sync 客户端以及 S-04/S-05/S-06 Sync 服务 driver 已登记；其余 18 个生产验收 ID 尚未登记，运行时会生成 `NOT_IMPLEMENTED` 证据并退出 1。这用于阻止误报，不是这些用例的验收通过。
+当前 runner 与失败闭合行为已实现，A-02/A-03 Sidecar、B-01/B-02 凭据、D-01 扩展包、S-01/S-02/S-03/S-08 Sync 客户端以及 S-04/S-05/S-06/S-07 Sync 服务 driver 已登记；其余 17 个生产验收 ID 尚未登记，运行时会生成 `NOT_IMPLEMENTED` 证据并退出 1。这用于阻止误报，不是这些用例的验收通过。
@@ -0,0 +1,560 @@
+"""S-07 empty deployment, backup/restore, and migration safety acceptance."""
+
+from __future__ import annotations
+
+import argparse
+from concurrent.futures import ThreadPoolExecutor
+import hashlib
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import time
+
+from sync_production_stack import (
+    ROOT,
+    SERVICE,
+    SyncProductionStack,
+    call,
+    sha256,
+    wait_http,
+)
+
+FILE_COUNT = 10_000
+NOTE_COUNT = 9_990
+NOTE_BYTES = 4 * 1024
+ATTACHMENT_COUNT = 10
+ATTACHMENT_BYTES = 100 * 1024 * 1024
+ONE_GIB = 1024**3
+
+
+def result(case_id: str, status: str, reason: str, facts: dict) -> dict:
+    passed = status == "PASSED"
+    assertions = [
+        (
+            "an empty PostgreSQL and MinIO instance initializes automatically",
+            facts.get("automatic_initialization") is True,
+        ),
+        (
+            "repeated initialization preserves every existing database row and object",
+            facts.get("initialization_runs") == 2
+            and facts.get("initialization_digest_preserved") is True,
+        ),
+        (
+            "the backup contains at least one GiB and exactly 10000 historical files",
+            facts.get("file_count") == FILE_COUNT
+            and facts.get("object_count") == FILE_COUNT
+            and facts.get("object_bytes", 0) >= ONE_GIB,
+        ),
+        (
+            "the source instance is deleted and all restored object hashes verify",
+            facts.get("source_deleted") is True
+            and facts.get("verified_objects") == FILE_COUNT
+            and facts.get("restored_digest_matches") is True,
+        ),
+        (
+            "restore reaches a ready service within 30 minutes from a backup under 24 hours old",
+            facts.get("rto_ms", 1_800_001) <= 1_800_000
+            and facts.get("backup_age_seconds", 86_401) <= 86_400,
+        ),
+        (
+            "a rejected migration leaves the complete old-data digest unchanged",
+            facts.get("migration_failures") == 1
+            and facts.get("migration_digest_preserved") is True,
+        ),
+    ]
+    evidence = (
+        "real PostgreSQL 17, fixed MinIO, production backup/restore commands, "
+        "and full SHA-256 verification"
+    )
+    return {
+        "schema": 1,
+        "case_id": case_id,
+        "status": status,
+        "reason": reason,
+        "assertions": [
+            {
+                "name": name,
+                "status": "PASSED" if passed and actual else "FAILED",
+                "evidence": evidence,
+            }
+            for name, actual in assertions
+        ],
+        "metrics": {
+            "file_count": facts.get("file_count", 0),
+            "object_count": facts.get("object_count", 0),
+            "object_bytes": facts.get("object_bytes", 0),
+            "verified_objects": facts.get("verified_objects", 0),
+            "rto_ms": facts.get("rto_ms", 0),
+            "backup_age_seconds": facts.get("backup_age_seconds", 0),
+            "initialization_runs": facts.get("initialization_runs", 0),
+            "migration_failures": facts.get("migration_failures", 0),
+        },
+        "files": [
+            {"path": relative, "sha256": sha256(ROOT / relative)}
+            for relative in (
+                "server sync/compose.yaml",
+                "server sync/sync_server/__main__.py",
+                "server sync/sync_server/database.py",
+                "server sync/sync_server/operations.py",
+                "server sync/sync_server/storage.py",
+                "scripts/acceptance_cases/sync_production_stack.py",
+                "scripts/acceptance_cases/s07_sync_backup.py",
+            )
+        ],
+        "revisions": [
+            {
+                "scope": "runtime",
+                "postgres": facts.get("postgres_version"),
+                "minio": facts.get("minio_version"),
+            },
+            {
+                "scope": "dataset",
+                "files": facts.get("file_count", 0),
+                "objects": facts.get("object_count", 0),
+                "bytes": facts.get("object_bytes", 0),
+            },
+            {
+                "scope": "recovery",
+                "rto_ms": facts.get("rto_ms", 0),
+                "backup_age_seconds": facts.get("backup_age_seconds", 0),
+                "verified_objects": facts.get("verified_objects", 0),
+            },
+            {
+                "scope": "last checkpoint",
+                "stage": facts.get("active_stage"),
+                "iteration": facts.get("active_iteration", -1),
+            },
+        ],
+    }
+
+
+def _note_data(index: int) -> bytes:
+    line = f"# OpenNexus S-07 note {index:05d}\nseed=20260908\n".encode()
+    return (line * (NOTE_BYTES // len(line) + 1))[:NOTE_BYTES]
+
+
+def _stable_id(kind: str, index: int, length: int = 32) -> str:
+    return hashlib.sha256(f"OpenNexus:S-07:{kind}:{index}".encode()).hexdigest()[:length]
+
+
+def seed_worker() -> int:
+    sys.path.insert(0, str(SERVICE))
+    from sqlalchemy import text
+
+    from sync_server.database import Database
+    from sync_server.storage import S3Objects
+
+    db = Database(os.environ["SYNC_DATABASE_URL"])
+    objects = S3Objects(os.environ["SYNC_S3_ENDPOINT"], os.environ["SYNC_S3_BUCKET"])
+    vault_id = os.environ["S07_VAULT_ID"]
+    device_id = os.environ["S07_DEVICE_ID"]
+    created = int(time.time())
+
+    def put_note(index: int) -> dict:
+        data = _note_data(index)
+        digest = hashlib.sha256(data).hexdigest()
+        objects.put(f"{vault_id}/{digest}", data)
+        return {
+            "hash": digest,
+            "size": len(data),
+            "path": f"notes/{index:05d}.md",
+        }
+
+    with ThreadPoolExecutor(max_workers=16) as pool:
+        records = list(pool.map(put_note, range(NOTE_COUNT)))
+
+    scratch = Path(os.environ["S07_SCRATCH_FILE"])
+    try:
+        for index in range(ATTACHMENT_COUNT):
+            block = hashlib.sha256(f"OpenNexus:S-07:attachment:{index}".encode()).digest()
+            block = block * (1024 * 1024 // len(block))
+            digest = hashlib.sha256()
+            with scratch.open("wb") as output:
+                for _ in range(ATTACHMENT_BYTES // len(block)):
+                    output.write(block)
+                    digest.update(block)
+            content_hash = digest.hexdigest()
+            objects.put_file(f"{vault_id}/{content_hash}", scratch, content_hash)
+            records.append(
+                {
+                    "hash": content_hash,
+                    "size": ATTACHMENT_BYTES,
+                    "path": f"attachments/{index:02d}.bin",
+                }
+            )
+    finally:
+        scratch.unlink(missing_ok=True)
+
+    object_rows = [
+        {"vault": vault_id, "hash": item["hash"], "size": item["size"], "created": created}
+        for item in records
+    ]
+    revision_rows = []
+    file_rows = []
+    for sequence, item in enumerate(records, start=1):
+        file_id = _stable_id("file", sequence)
+        operation_id = _stable_id("operation", sequence)
+        revision_rows.append(
+            {
+                "vault": vault_id,
+                "sequence": sequence,
+                "file": file_id,
+                "base": 0,
+                "path": item["path"],
+                "path_key": item["path"].casefold(),
+                "hash": item["hash"],
+                "size": item["size"],
+                "device": device_id,
+                "operation": operation_id,
+                "fingerprint": _stable_id("fingerprint", sequence, 64),
+            }
+        )
+        file_rows.append(
+            {
+                "vault": vault_id,
+                "file": file_id,
+                "sequence": sequence,
+                "path_key": item["path"].casefold(),
+            }
+        )
+    total_bytes = sum(item["size"] for item in records)
+    with db.transaction() as conn:
+        existing = conn.execute(
+            text("SELECT COUNT(*) FROM objects WHERE vault_id=:vault"), {"vault": vault_id}
+        ).scalar_one()
+        if existing:
+            raise RuntimeError("S07_DATASET_NOT_EMPTY")
+        for start in range(0, len(records), 1000):
+            conn.execute(
+                text(
+                    "INSERT INTO objects(vault_id,hash,size,created) "
+                    "VALUES (:vault,:hash,:size,:created)"
+                ),
+                object_rows[start : start + 1000],
+            )
+            conn.execute(
+                text(
+                    "INSERT INTO revisions(vault_id,sequence,file_id,base_revision,path,path_key,"
+                    "operation,hash,size,device_id,operation_id,fingerprint) VALUES "
+                    "(:vault,:sequence,:file,:base,:path,:path_key,'put',:hash,:size,:device,"
+                    ":operation,:fingerprint)"
+                ),
+                revision_rows[start : start + 1000],
+            )
+            conn.execute(
+                text(
+                    "INSERT INTO files(vault_id,file_id,sequence,path_key,deleted) "
+                    "VALUES (:vault,:file,:sequence,:path_key,0)"
+                ),
+                file_rows[start : start + 1000],
+            )
+        conn.execute(
+            text("UPDATE vaults SET sequence=:sequence,used=:used,quota=:quota WHERE id=:vault"),
+            {
+                "sequence": FILE_COUNT,
+                "used": total_bytes,
+                "quota": total_bytes + ONE_GIB,
+                "vault": vault_id,
+            },
+        )
+    print(json.dumps({"files": len(records), "objects": len(records), "bytes": total_bytes}))
+    return 0
+
+
+def digest_worker(mode: str) -> int:
+    sys.path.insert(0, str(SERVICE))
+    from sqlalchemy import text
+
+    from sync_server.database import Database
+    from sync_server.storage import S3Objects
+
+    db = Database(os.environ["SYNC_DATABASE_URL"])
+    objects = S3Objects(os.environ["SYNC_S3_ENDPOINT"], os.environ["SYNC_S3_BUCKET"])
+    database_digest = hashlib.sha256()
+    with db.engine.connect() as conn:
+        tables = [
+            row[0]
+            for row in conn.execute(
+                text(
+                    "SELECT table_name FROM information_schema.tables "
+                    "WHERE table_schema='public' ORDER BY table_name"
+                )
+            )
+        ]
+        for table in tables:
+            if not table.replace("_", "").isalnum():
+                raise RuntimeError("S07_TABLE_NAME_INVALID")
+            rows = [
+                json.dumps(dict(row), ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+                for row in conn.execute(text(f'SELECT * FROM "{table}"')).mappings()
+            ]
+            database_digest.update(table.encode())
+            for encoded in sorted(rows):
+                database_digest.update(encoded.encode())
+        catalog = [
+            {"vault_id": row.vault_id, "hash": row.hash, "size": int(row.size)}
+            for row in conn.execute(
+                text("SELECT vault_id,hash,size FROM objects ORDER BY vault_id,hash")
+            )
+        ]
+
+    def verify(item: dict) -> str:
+        key = f"{item['vault_id']}/{item['hash']}"
+        if mode == "full":
+            response = objects.client.get_object(Bucket=objects.bucket, Key=key)
+            digest = hashlib.sha256()
+            size = 0
+            with response["Body"] as body:
+                for chunk in iter(lambda: body.read(1024 * 1024), b""):
+                    digest.update(chunk)
+                    size += len(chunk)
+            if digest.hexdigest() != item["hash"] or size != item["size"]:
+                raise RuntimeError("S07_OBJECT_DIGEST_MISMATCH")
+        else:
+            response = objects.client.head_object(Bucket=objects.bucket, Key=key)
+            if (
+                int(response["ContentLength"]) != item["size"]
+                or response.get("Metadata", {}).get("sha256") != item["hash"]
+            ):
+                raise RuntimeError("S07_OBJECT_HEAD_MISMATCH")
+        return f"{key}:{item['size']}"
+
+    with ThreadPoolExecutor(max_workers=16) as pool:
+        verified = list(pool.map(verify, catalog))
+    catalog_digest = hashlib.sha256("\n".join(verified).encode()).hexdigest()
+    combined = hashlib.sha256(
+        (database_digest.hexdigest() + ":" + catalog_digest).encode()
+    ).hexdigest()
+    print(
+        json.dumps(
+            {
+                "digest": combined,
+                "database_digest": database_digest.hexdigest(),
+                "catalog_digest": catalog_digest,
+                "verified_objects": len(verified),
+                "object_bytes": sum(item["size"] for item in catalog),
+            }
+        )
+    )
+    return 0
+
+
+def worker_entry() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--worker", choices=["seed", "digest-head", "digest-full"], required=True)
+    args = parser.parse_args()
+    if args.worker == "seed":
+        return seed_worker()
+    return digest_worker("full" if args.worker == "digest-full" else "head")
+
+
+def run_worker(stack: SyncProductionStack, worker: str, **environment: str) -> dict:
+    child_env = dict(stack.service_env)
+    child_env.update(environment)
+    completed = subprocess.run(
+        [str(stack.server_python), str(Path(__file__).resolve()), "--worker", worker],
+        cwd=ROOT,
+        env=child_env,
+        capture_output=True,
+        text=True,
+        check=True,
+        timeout=1800,
+    )
+    return json.loads(completed.stdout.splitlines()[-1])
+
+
+def run_operation(stack: SyncProductionStack, command: str, directory: Path) -> dict:
+    completed = subprocess.run(
+        [
+            str(stack.server_python),
+            "-m",
+            "sync_server",
+            command,
+            "--directory",
+            str(directory),
+            "--io-workers",
+            "16",
+        ],
+        cwd=SERVICE,
+        env=stack.service_env,
+        capture_output=True,
+        text=True,
+        check=True,
+        timeout=2400,
+    )
+    return json.loads(completed.stdout.splitlines()[-1])
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--config", required=True)
+    parser.add_argument("--output", required=True)
+    args = parser.parse_args()
+    output = Path(args.output)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    case_id = os.environ.get("OPENNEXUS_ACCEPTANCE_CASE_ID", "")
+    facts: dict = {}
+    reason = ""
+    status = "FAILED"
+    source_stack = None
+    restored_stack = None
+    source_username = ""
+    source_password = ""
+    try:
+        config = json.loads(Path(args.config).read_text(encoding="utf-8"))
+        data_root = Path(os.environ["OPENNEXUS_ACCEPTANCE_DATA_ROOT"]).resolve()
+        data_root.mkdir(parents=True, exist_ok=True)
+        backup = data_root / "s07-backup"
+
+        facts["active_stage"] = "initialize-empty"
+        source_stack = SyncProductionStack(config, data_root, "s07-source").start_dependencies()
+        source_stack.initialize()
+        source_username = source_stack.username
+        source_password = source_stack.password
+        facts.update(
+            {
+                "automatic_initialization": source_stack.sql_scalar(
+                    "SELECT version FROM schema_version"
+                )
+                == 1,
+                "postgres_version": source_stack.postgres_version,
+                "minio_version": source_stack.minio_version,
+                "initialization_runs": 1,
+            }
+        )
+        source_stack.add_user_with_password(source_stack.username, source_stack.password)
+        source_stack.start_sync()
+        wait_http(source_stack.origin + "/ready", timeout=60)
+        owner = source_stack.login(source_stack.username, source_stack.password, "S-07 owner")
+        owner_auth = {"Authorization": "Bearer " + owner["access_token"]}
+        _, _, vault_id = source_stack.create_vault_for_auth(owner_auth, "S-07 dataset")
+        source_stack.stop_sync()
+
+        facts["active_stage"] = "seed-dataset"
+        seeded = run_worker(
+            source_stack,
+            "seed",
+            S07_VAULT_ID=vault_id,
+            S07_DEVICE_ID=owner["device_id"],
+            S07_SCRATCH_FILE=str(source_stack.stack / "s07-seed-object.bin"),
+        )
+        facts.update(
+            {
+                "file_count": seeded["files"],
+                "object_count": seeded["objects"],
+                "object_bytes": seeded["bytes"],
+            }
+        )
+        assert seeded["files"] == FILE_COUNT and seeded["objects"] == FILE_COUNT
+        assert seeded["bytes"] >= ONE_GIB
+
+        facts["active_stage"] = "repeat-initialize"
+        before_initialize = run_worker(source_stack, "digest-head")
+        source_stack.initialize()
+        facts["initialization_runs"] = 2
+        after_initialize = run_worker(source_stack, "digest-head")
+        facts["initialization_digest_preserved"] = (
+            before_initialize["digest"] == after_initialize["digest"]
+        )
+        assert facts["initialization_digest_preserved"]
+
+        facts["active_stage"] = "backup"
+        backup_info = run_operation(source_stack, "backup", backup)
+        manifest = json.loads((backup / "manifest.json").read_text(encoding="utf-8"))
+        assert backup_info["status"] == "BACKUP_COMPLETE"
+        assert manifest["object_count"] == FILE_COUNT and manifest["object_bytes"] >= ONE_GIB
+
+        facts["active_stage"] = "delete-source"
+        source_root = source_stack.stack.resolve()
+        source_stack.stop()
+        source_stack = None
+        if source_root.parent != data_root or source_root.name != "s07-source-production-stack":
+            raise RuntimeError("S07_SOURCE_DELETE_BOUNDARY")
+        shutil.rmtree(source_root)
+        facts["source_deleted"] = not source_root.exists()
+        assert facts["source_deleted"]
+
+        facts["active_stage"] = "restore-empty"
+        rto_started = time.monotonic()
+        restored_stack = SyncProductionStack(
+            config, data_root, "s07-restored"
+        ).start_dependencies()
+        restore_info = run_operation(restored_stack, "restore", backup)
+        restored_stack.start_sync()
+        wait_http(restored_stack.origin + "/ready", timeout=60)
+        facts["rto_ms"] = int((time.monotonic() - rto_started) * 1000)
+        facts["backup_age_seconds"] = restore_info["backup_age_seconds"]
+        facts["verified_objects"] = restore_info["verified_objects"]
+        assert facts["rto_ms"] <= 1_800_000 and facts["backup_age_seconds"] <= 86_400
+        assert facts["verified_objects"] == FILE_COUNT
+        restored_digest = run_worker(restored_stack, "digest-head")
+        facts["restored_digest_matches"] = (
+            restored_digest["digest"] == before_initialize["digest"]
+        )
+        assert facts["restored_digest_matches"]
+
+        restored_owner = restored_stack.login(
+            source_username, source_password, "S-07 restored owner"
+        )
+        restored_auth = {"Authorization": "Bearer " + restored_owner["access_token"]}
+        restored_base = restored_stack.origin + "/sync/v1/vaults/" + vault_id
+        sample = next(item for item in manifest["objects"] if item["size"] == NOTE_BYTES)
+        downloaded = call(
+            "GET", restored_base + "/objects/" + sample["hash"], headers=restored_auth
+        )
+        assert downloaded[0] == 200 and hashlib.sha256(downloaded[1]).hexdigest() == sample["hash"]
+
+        facts["active_stage"] = "migration-failure"
+        restored_stack.stop_sync()
+        changed = restored_stack.sql_scalar(
+            "WITH changed AS (UPDATE schema_version SET version=999 RETURNING 1) "
+            "SELECT COUNT(*) FROM changed"
+        )
+        assert changed == 1
+        before_migration = run_worker(restored_stack, "digest-full")
+        migration = subprocess.run(
+            [str(restored_stack.server_python), "-m", "sync_server", "migrate"],
+            cwd=SERVICE,
+            env=restored_stack.service_env,
+            capture_output=True,
+            text=True,
+            timeout=90,
+        )
+        facts["migration_failures"] = int(migration.returncode != 0)
+        after_migration = run_worker(restored_stack, "digest-head")
+        facts["migration_digest_preserved"] = (
+            before_migration["digest"] == after_migration["digest"]
+        )
+        assert facts["migration_failures"] == 1 and facts["migration_digest_preserved"]
+        restored_stack.sql_scalar(
+            "WITH changed AS (UPDATE schema_version SET version=1 RETURNING 1) "
+            "SELECT COUNT(*) FROM changed"
+        )
+        restored_stack.start_sync()
+        wait_http(restored_stack.origin + "/ready", timeout=60)
+        facts.update({"active_stage": "complete", "active_iteration": FILE_COUNT})
+        status = "PASSED"
+    except BaseException as error:
+        reason = "S07_ORACLE_FAILED:" + type(error).__name__
+    finally:
+        if source_stack is not None:
+            source_stack.stop()
+        if restored_stack is not None:
+            restored_stack.stop()
+    payload = result(
+        case_id,
+        status if case_id == "S-07" else "FAILED",
+        reason or ("" if case_id == "S-07" else "CASE_ID_MISMATCH"),
+        facts,
+    )
+    output.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    return 0 if payload["status"] == "PASSED" else 1
+
+
+if __name__ == "__main__":
+    if "--worker" in sys.argv:
+        raise SystemExit(worker_entry())
+    raise SystemExit(main())
@@ -147,6 +147,14 @@ class SyncProductionStack:
        self.service_env = {}

    def start(self) -> "SyncProductionStack":
+        self.start_dependencies()
+        self.initialize()
+        self.add_user_with_password(self.username, self.password)
+        self.start_sync()
+        wait_http(self.origin + "/ready", timeout=60)
+        return self
+
+    def start_dependencies(self) -> "SyncProductionStack":
        self.stack.mkdir()
        self.minio_data.mkdir()
        self.staging.mkdir()
@@ -214,22 +222,22 @@ class SyncProductionStack:
                "ACCEPTANCE_PASSWORD": self.password,
            }
        )
+        return self
+
+    def initialize(self) -> None:
        subprocess.run(
-            [
-                str(self.server_python),
-                "-c",
-                "import os,boto3; from sync_server.database import Database; "
-                "d=Database(os.environ['SYNC_DATABASE_URL']); d.migrate(); "
-                "d.add_user(os.environ['ACCEPTANCE_USERNAME'],os.environ['ACCEPTANCE_PASSWORD']); "
-                "boto3.client('s3',endpoint_url=os.environ['SYNC_S3_ENDPOINT']).create_bucket(Bucket=os.environ['SYNC_S3_BUCKET'])",
-            ],
+            [str(self.server_python), "-m", "sync_server", "initialize"],
            cwd=SERVICE,
            env=self.service_env,
            stdout=self.sync_log,
            stderr=subprocess.STDOUT,
            check=True,
-            timeout=60,
+            timeout=90,
        )
+
+    def start_sync(self) -> None:
+        if self.sync is not None and self.sync.poll() is None:
+            return
        self.sync = subprocess.Popen(
            [str(self.server_python), "-m", "sync_server", "serve", "--workers", "2"],
            cwd=SERVICE,
@@ -237,8 +245,10 @@ class SyncProductionStack:
            stdout=self.sync_log,
            stderr=subprocess.STDOUT,
        )
-        wait_http(self.origin + "/ready", timeout=60)
-        return self
+
+    def stop_sync(self) -> None:
+        stop_tree(self.sync)
+        self.sync = None

    def login(self, username: str, password: str, label: str) -> dict:
        session = None
@@ -293,6 +303,10 @@ class SyncProductionStack:
    def add_user(self, prefix: str) -> tuple[str, str]:
        username = prefix + "-" + secrets.token_hex(8)
        password = secrets.token_urlsafe(32)
+        self.add_user_with_password(username, password)
+        return username, password
+
+    def add_user_with_password(self, username: str, password: str) -> None:
        environment = dict(self.service_env)
        environment.update(
            {"ACCEPTANCE_EXTRA_USERNAME": username, "ACCEPTANCE_EXTRA_PASSWORD": password}
@@ -312,7 +326,6 @@ class SyncProductionStack:
            check=True,
            timeout=30,
        )
-        return username, password

    def start_postgres(self) -> None:
        subprocess.run(
@@ -419,7 +432,7 @@ class SyncProductionStack:
            raise RuntimeError("PRODUCTION_STACK_EXITED")

    def stop(self) -> None:
-        stop_tree(self.sync)
+        self.stop_sync()
        self.stop_minio()
        if self.postgres_started:
            try:
@@ -110,6 +110,21 @@ CASE_DRIVERS: dict[str, dict[str, Any]] = {
        ),
        "required_artifacts": ("postgres_initdb", "minio_server"),
    },
+    "S-07": {
+        "driver": "scripts/acceptance_cases/s07_sync_backup.py",
+        "timeout_seconds": 3600,
+        "required_metrics": (
+            "file_count",
+            "object_count",
+            "object_bytes",
+            "verified_objects",
+            "rto_ms",
+            "backup_age_seconds",
+            "initialization_runs",
+            "migration_failures",
+        ),
+        "required_artifacts": ("postgres_initdb", "minio_server"),
+    },
 }
 ENV_NAME = re.compile(r"[A-Z][A-Z0-9_]{2,127}")
 RUN_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{2,63}")
@@ -15,16 +15,30 @@ uv run pytest
 ## 自托管准备

 1. 将 `.env.example` 复制为 `.env`，生成独立数据库、MinIO 管理和同步访问凭据。数据库 URL 使用 `postgresql+psycopg://…`，其中密码须 URL 编码。
-2. 启动 `docker compose up -d database objects`。由管理员在 MinIO 建立 `notesagent` 私有 Bucket，并创建仅能访问该 Bucket 的同步账号；服务不使用 MinIO root 身份。
+2. 执行 `docker compose up -d`。一次性 `initialize` 服务等待依赖后幂等创建 schema 与 `opennexus` Bucket；重复运行只检查并补齐缺失资源，不覆盖已有行或对象。长期运行的 `sync` 服务继续使用只限该 Bucket 的同步账号，不使用 MinIO root 身份。
 3. 执行 `docker compose run --rm sync /service/.venv/bin/python -m sync_server create-user`，密码交互输入，不放命令参数。
-4. 启动 `docker compose up -d sync`；使用 Caddy 示例配置 TLS。8080 仅绑定本机，不直接公开明文 HTTP。
+4. 使用 Caddy 示例配置 TLS。8080 仅绑定本机，不直接公开明文 HTTP。
 5. 检查 `/health`、`/ready` 及经过授权的上传/读取；`/ready` 探测数据库 schema、staging 读写和对象存储测试前缀。

-Dockerfile/Compose 是待实测部署配置，不能视为已验证安装程序。未提供自动 MinIO 初始化和备份恢复命令，发布前必须完成。
+`initialize` 命令已通过真实 PostgreSQL/MinIO 的空实例与重复运行验证，并由 Compose 的一次性服务调用。MinIO 同步账号仍须由管理员创建并限制到 `opennexus` Bucket，`.env` 中的 root 与同步凭据必须不同。

 2026-09-08 已在独立 Docker 项目完成真实 PostgreSQL/MinIO 双 worker 测试部署，修正基础镜像中的 `sync` 系统用户名冲突。测试专用 HTTP 地址、故障检查、完整验收缺口与运维入口见[验收报告](../docs/development/OpenNexus验收报告-2026-09-08.md)。仓库通用 Compose 的生产 TLS 与备份恢复仍未通过发布验收。

-升级前同时备份 PostgreSQL 和 Bucket，停止提交以取得一致切点。Schema v1 拒绝未知数据库版本，不自动降级。当前历史永久保留，容量管理不能手动删除被历史引用的对象。
+## 备份与空实例恢复
+
+`backup` 在 PostgreSQL 的同一只读 repeatable-read 事务中导出 schema v1 的全部表和对象清单，再流式下载清单中的不可变对象并核对长度/SHA-256。存在未完成上传或缺失历史对象时失败；目标目录必须不存在。备份目录含认证哈希和会话哈希，必须使用受限 ACL、加密磁盘及异机副本保护。
+
+```powershell
+python -m sync_server backup --directory D:/OpenNexus-backups/2026-09-09 --io-workers 8
+```
+
+`restore` 默认拒绝超过 24 小时的备份，只允许空 PostgreSQL 数据库和空/不存在 Bucket。命令先校验完整备份，再上传并回读核对全部对象，最后在单个 PostgreSQL 事务中创建 schema、导入并复核对象目录；数据库不会引用只恢复一部分的对象。恢复演练应使用新实例，成功后再启动服务并检查 `/ready`。
+
+```powershell
+python -m sync_server restore --directory D:/OpenNexus-backups/2026-09-09 --io-workers 8
+```
+
+生产定时任务至少每日生成一次新目录并检查命令退出码、`created_utc`、对象数与总字节；保留策略和异机复制由部署维护者配置。升级前执行新备份并完成抽样恢复。Schema v1 拒绝未知数据库版本，不自动降级；当前历史永久保留，容量管理不能手动删除被历史引用的对象。


 ## 四并发大附件传输探针
@@ -2,13 +2,13 @@ services:
  database:
    image: postgres:17.6
    environment:
-      POSTGRES_USER: notesagent
-      POSTGRES_DB: notesagent
+      POSTGRES_USER: opennexus
+      POSTGRES_DB: opennexus
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?required}
    volumes:
      - postgres:/var/lib/postgresql/data
    healthcheck:
-      test: [CMD-SHELL, "pg_isready -U notesagent -d notesagent"]
+      test: [CMD-SHELL, "pg_isready -U opennexus -d opennexus"]
      interval: 5s
      retries: 20
  objects:
@@ -19,18 +19,34 @@ services:
      MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD:?required}
    volumes:
      - objects:/data
+  initialize:
+    build: .
+    command: ["/service/.venv/bin/python", "-m", "sync_server", "initialize"]
+    environment:
+      SYNC_DATABASE_URL: ${SYNC_DATABASE_URL:?required}
+      SYNC_S3_ENDPOINT: http://objects:9000
+      SYNC_S3_BUCKET: opennexus
+      AWS_ACCESS_KEY_ID: ${MINIO_ROOT_USER:?required}
+      AWS_SECRET_ACCESS_KEY: ${MINIO_ROOT_PASSWORD:?required}
+      AWS_DEFAULT_REGION: us-east-1
+    depends_on:
+      database:
+        condition: service_healthy
+      objects:
+        condition: service_started
+    restart: "no"
  sync:
    build: .
    environment:
      SYNC_DATABASE_URL: ${SYNC_DATABASE_URL:?required}
      SYNC_S3_ENDPOINT: http://objects:9000
-      SYNC_S3_BUCKET: notesagent
+      SYNC_S3_BUCKET: opennexus
      AWS_ACCESS_KEY_ID: ${SYNC_ACCESS_KEY_ID:?required}
      AWS_SECRET_ACCESS_KEY: ${SYNC_SECRET_ACCESS_KEY:?required}
      AWS_DEFAULT_REGION: us-east-1
    depends_on:
-      database:
-        condition: service_healthy
+      initialize:
+        condition: service_completed_successfully
    volumes:
      - staging:/staging
    ports:
@@ -2,8 +2,10 @@

 import argparse
 import getpass
+import json
 import os
 from pathlib import Path
+import time

 from .app import create_app
 from .database import Database
@@ -20,18 +22,66 @@ def application():

 def main():
    parser = argparse.ArgumentParser()
-    parser.add_argument("command", choices=["serve", "migrate", "create-user", "cleanup-uploads"])
+    parser.add_argument(
+        "command",
+        choices=["serve", "initialize", "migrate", "create-user", "cleanup-uploads", "backup", "restore"],
+    )
    parser.add_argument("--workers", type=int, choices=[1, 2], default=2)
    parser.add_argument("--username")
+    parser.add_argument("--directory", type=Path)
+    parser.add_argument("--io-workers", type=int, choices=range(1, 33), default=8)
+    parser.add_argument("--max-age-hours", type=float, default=24)
    args = parser.parse_args()
    url = os.environ["SYNC_DATABASE_URL"]
    if not url.startswith("postgresql+psycopg://"):
        raise SystemExit("生产入口只支持 PostgreSQL")
    db = Database(url)
-    db.migrate()
-    if args.command == "create-user":
+    if args.command == "initialize":
+        objects = S3Objects(os.environ["SYNC_S3_ENDPOINT"], os.environ["SYNC_S3_BUCKET"])
+        deadline = time.monotonic() + 60
+        while True:
+            try:
+                db.migrate()
+                created = objects.ensure_bucket()
+                print(json.dumps({"schema": 1, "bucket_created": created}))
+                break
+            except Exception:
+                if time.monotonic() >= deadline:
+                    raise
+                time.sleep(1)
+    elif args.command == "backup":
+        if args.directory is None:
+            raise SystemExit("backup 需要 --directory")
+        from .operations import create_backup
+
+        objects = S3Objects(os.environ["SYNC_S3_ENDPOINT"], os.environ["SYNC_S3_BUCKET"])
+        print(
+            json.dumps(
+                create_backup(db, objects, args.directory, workers=args.io_workers)
+            )
+        )
+    elif args.command == "restore":
+        if args.directory is None:
+            raise SystemExit("restore 需要 --directory")
+        from .operations import restore_backup
+
+        objects = S3Objects(os.environ["SYNC_S3_ENDPOINT"], os.environ["SYNC_S3_BUCKET"])
+        print(
+            json.dumps(
+                restore_backup(
+                    db,
+                    objects,
+                    args.directory,
+                    workers=args.io_workers,
+                    max_age_hours=args.max_age_hours,
+                )
+            )
+        )
+    elif args.command == "create-user":
+        db.migrate()
        db.add_user(args.username or input("用户名: "), getpass.getpass("密码（至少12字符）: "))
    elif args.command == "serve":
+        db.migrate()
        import uvicorn
        host = os.environ.get("SYNC_HOST", "0.0.0.0")
        if host not in {"0.0.0.0", "127.0.0.1", "::1"}:
@@ -45,8 +95,11 @@ def main():
        uvicorn.run("sync_server.__main__:application", factory=True, workers=args.workers,
                    host=host, port=port, access_log=False)
    elif args.command == "cleanup-uploads":
+        db.migrate()
        from .maintenance import cleanup_expired_uploads
        print(cleanup_expired_uploads(db, Path(os.environ["SYNC_STAGING_DIR"])))
+    elif args.command == "migrate":
+        db.migrate()


 if __name__ == "__main__":
@@ -0,0 +1,412 @@
+"""Consistent PostgreSQL/S3 backup and empty-instance restore operations."""
+
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import datetime, timezone
+import hashlib
+import json
+import math
+import os
+from pathlib import Path
+import re
+import shutil
+import tempfile
+from typing import Any
+
+from sqlalchemy import text
+
+from .database import SCHEMA
+
+
+MANIFEST_SCHEMA = 1
+VAULT_ID = re.compile(r"[0-9a-f]{32}")
+CONTENT_HASH = re.compile(r"[0-9a-f]{64}")
+TABLES: dict[str, tuple[str, ...]] = {
+    "schema_version": ("version",),
+    "users": ("id", "username", "password"),
+    "devices": ("id", "user_id", "name", "revoked"),
+    "sessions": ("token", "refresh", "device_id", "expires", "refresh_expires"),
+    "vaults": ("id", "user_id", "name", "sequence", "quota", "used"),
+    "uploads": ("id", "vault_id", "device_id", "hash", "size", "offset_bytes", "expires"),
+    "objects": ("vault_id", "hash", "size", "created"),
+    "revisions": (
+        "vault_id",
+        "sequence",
+        "file_id",
+        "base_revision",
+        "path",
+        "path_key",
+        "operation",
+        "hash",
+        "size",
+        "device_id",
+        "operation_id",
+        "fingerprint",
+    ),
+    "files": ("vault_id", "file_id", "sequence", "path_key", "deleted"),
+    "login_limits": ("key", "started", "attempts"),
+    "upload_receipts": ("id", "vault_id", "device_id", "hash", "completed"),
+}
+
+
+class OperationsError(RuntimeError):
+    """Stable operator-facing failure without credentials or response bodies."""
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _object_path(root: Path, vault_id: str, content_hash: str) -> Path:
+    if not VAULT_ID.fullmatch(vault_id) or not CONTENT_HASH.fullmatch(content_hash):
+        raise OperationsError("OBJECT_ID_INVALID")
+    return root / "objects" / vault_id / content_hash
+
+
+def _manifest_objects(conn) -> list[dict[str, Any]]:
+    pending = conn.execute(text("SELECT COUNT(*) FROM uploads")).scalar_one()
+    if pending:
+        raise OperationsError("BACKUP_PENDING_UPLOADS")
+    missing = conn.execute(
+        text(
+            "SELECT COUNT(*) FROM revisions r LEFT JOIN objects o "
+            "ON o.vault_id=r.vault_id AND o.hash=r.hash "
+            "WHERE r.hash IS NOT NULL AND o.hash IS NULL"
+        )
+    ).scalar_one()
+    if missing:
+        raise OperationsError("HISTORICAL_OBJECT_MISSING")
+    return [
+        {"vault_id": row.vault_id, "hash": row.hash, "size": int(row.size)}
+        for row in conn.execute(
+            text("SELECT vault_id,hash,size FROM objects ORDER BY vault_id,hash")
+        )
+    ]
+
+
+def _write_database_snapshot(conn, destination: Path) -> dict[str, int]:
+    counts = {}
+    with destination.open("x", encoding="utf-8", newline="\n") as output:
+        for table, columns in TABLES.items():
+            projection = ",".join(f'"{column}"' for column in columns)
+            ordering = ",".join(f'"{column}"' for column in columns)
+            rows = conn.execute(
+                text(f'SELECT {projection} FROM "{table}" ORDER BY {ordering}')
+            )
+            count = 0
+            for row in rows:
+                output.write(
+                    json.dumps(
+                        {"table": table, "values": list(row)},
+                        ensure_ascii=False,
+                        separators=(",", ":"),
+                    )
+                    + "\n"
+                )
+                count += 1
+            counts[table] = count
+        output.flush()
+        os.fsync(output.fileno())
+    try:
+        destination.chmod(0o600)
+    except OSError:
+        pass
+    return counts
+
+
+def _download(objects, root: Path, item: dict[str, Any]) -> None:
+    target = _object_path(root, item["vault_id"], item["hash"])
+    target.parent.mkdir(parents=True, exist_ok=True)
+    response = objects.client.get_object(
+        Bucket=objects.bucket,
+        Key=f"{item['vault_id']}/{item['hash']}",
+    )
+    digest = hashlib.sha256()
+    size = 0
+    try:
+        with response["Body"] as body, target.open("xb") as destination:
+            for chunk in iter(lambda: body.read(1024 * 1024), b""):
+                destination.write(chunk)
+                digest.update(chunk)
+                size += len(chunk)
+            destination.flush()
+            os.fsync(destination.fileno())
+    except BaseException:
+        target.unlink(missing_ok=True)
+        raise
+    if size != item["size"] or digest.hexdigest() != item["hash"]:
+        target.unlink(missing_ok=True)
+        raise OperationsError("OBJECT_INTEGRITY_FAILED")
+
+
+def create_backup(db, objects, destination: Path, *, workers: int = 8) -> dict[str, Any]:
+    destination = destination.resolve(strict=False)
+    if destination.exists():
+        raise OperationsError("BACKUP_DESTINATION_EXISTS")
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    temporary = Path(
+        tempfile.mkdtemp(prefix=destination.name + ".incomplete-", dir=destination.parent)
+    )
+    database = temporary / "database.jsonl"
+    try:
+        with db.engine.connect() as conn:
+            transaction = conn.begin()
+            try:
+                conn.exec_driver_sql(
+                    "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY"
+                )
+                version = conn.execute(text("SELECT version FROM schema_version")).scalar_one()
+                if version != 1:
+                    raise OperationsError("SCHEMA_INCOMPATIBLE")
+                catalog = _manifest_objects(conn)
+                table_rows = _write_database_snapshot(conn, database)
+                transaction.commit()
+            except BaseException:
+                if transaction.is_active:
+                    transaction.rollback()
+                raise
+        with ThreadPoolExecutor(max_workers=workers) as pool:
+            list(pool.map(lambda item: _download(objects, temporary, item), catalog))
+        manifest = {
+            "schema": MANIFEST_SCHEMA,
+            "created_utc": datetime.now(timezone.utc).isoformat(),
+            "database_sha256": sha256_file(database),
+            "database_rows": table_rows,
+            "object_count": len(catalog),
+            "object_bytes": sum(item["size"] for item in catalog),
+            "objects": catalog,
+        }
+        manifest_path = temporary / "manifest.json"
+        manifest_path.write_text(
+            json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
+        )
+        os.replace(temporary, destination)
+        return {
+            "status": "BACKUP_COMPLETE",
+            "created_utc": manifest["created_utc"],
+            "object_count": manifest["object_count"],
+            "object_bytes": manifest["object_bytes"],
+        }
+    except BaseException:
+        shutil.rmtree(temporary, ignore_errors=True)
+        raise
+
+
+def _load_manifest(source: Path, max_age_hours: float) -> dict[str, Any]:
+    try:
+        manifest = json.loads((source / "manifest.json").read_text(encoding="utf-8"))
+        created = datetime.fromisoformat(manifest["created_utc"])
+    except (OSError, ValueError, KeyError, json.JSONDecodeError) as error:
+        raise OperationsError("BACKUP_MANIFEST_INVALID") from error
+    if manifest.get("schema") != MANIFEST_SCHEMA or created.tzinfo is None:
+        raise OperationsError("BACKUP_MANIFEST_INVALID")
+    age = (datetime.now(timezone.utc) - created.astimezone(timezone.utc)).total_seconds()
+    if age < -300 or age > max_age_hours * 3600:
+        raise OperationsError("BACKUP_AGE_INVALID")
+    catalog = manifest.get("objects")
+    if not isinstance(catalog, list) or manifest.get("database_rows") is None:
+        raise OperationsError("BACKUP_MANIFEST_INVALID")
+    normalized = []
+    for item in catalog:
+        if (
+            not isinstance(item, dict)
+            or not isinstance(item.get("vault_id"), str)
+            or not isinstance(item.get("hash"), str)
+            or not isinstance(item.get("size"), int)
+            or isinstance(item.get("size"), bool)
+            or item["size"] < 0
+        ):
+            raise OperationsError("BACKUP_MANIFEST_INVALID")
+        _object_path(source, item["vault_id"], item["hash"])
+        normalized.append(
+            {"vault_id": item["vault_id"], "hash": item["hash"], "size": item["size"]}
+        )
+    rows = manifest["database_rows"]
+    if (
+        normalized != sorted(normalized, key=lambda item: (item["vault_id"], item["hash"]))
+        or len({(item["vault_id"], item["hash"]) for item in normalized}) != len(normalized)
+        or manifest.get("object_count") != len(normalized)
+        or manifest.get("object_bytes") != sum(item["size"] for item in normalized)
+        or not CONTENT_HASH.fullmatch(str(manifest.get("database_sha256", "")))
+        or not isinstance(rows, dict)
+        or set(rows) != set(TABLES)
+        or any(not isinstance(rows[name], int) or rows[name] < 0 for name in TABLES)
+    ):
+        raise OperationsError("BACKUP_MANIFEST_INVALID")
+    manifest["objects"] = normalized
+    manifest["age_seconds"] = max(0, int(age))
+    return manifest
+
+
+def _load_database_snapshot(source: Path, manifest: dict[str, Any]) -> dict[str, list[list]]:
+    database = source / "database.jsonl"
+    if not database.is_file() or sha256_file(database) != manifest["database_sha256"]:
+        raise OperationsError("BACKUP_DATABASE_INTEGRITY_FAILED")
+    restored = {table: [] for table in TABLES}
+    try:
+        with database.open("r", encoding="utf-8") as stream:
+            for line in stream:
+                item = json.loads(line)
+                table = item.get("table")
+                values = item.get("values")
+                if table not in TABLES or not isinstance(values, list):
+                    raise OperationsError("BACKUP_DATABASE_INVALID")
+                if len(values) != len(TABLES[table]):
+                    raise OperationsError("BACKUP_DATABASE_INVALID")
+                restored[table].append(values)
+    except (OSError, json.JSONDecodeError) as error:
+        raise OperationsError("BACKUP_DATABASE_INVALID") from error
+    if any(len(restored[table]) != manifest["database_rows"][table] for table in TABLES):
+        raise OperationsError("BACKUP_DATABASE_INVALID")
+    if restored["schema_version"] != [[1]]:
+        raise OperationsError("BACKUP_DATABASE_INVALID")
+    object_columns = TABLES["objects"]
+    indexes = {name: object_columns.index(name) for name in ("vault_id", "hash", "size")}
+    catalog = sorted(
+        (
+            {
+                "vault_id": row[indexes["vault_id"]],
+                "hash": row[indexes["hash"]],
+                "size": row[indexes["size"]],
+            }
+            for row in restored["objects"]
+        ),
+        key=lambda item: (item["vault_id"], item["hash"]),
+    )
+    if catalog != manifest["objects"]:
+        raise OperationsError("BACKUP_DATABASE_INVALID")
+    return restored
+
+
+def _verify_backup_file(source: Path, item: dict[str, Any]) -> None:
+    path = _object_path(source, item["vault_id"], item["hash"])
+    try:
+        size = path.stat().st_size
+    except OSError as error:
+        raise OperationsError("BACKUP_OBJECT_MISSING") from error
+    if size != item["size"] or sha256_file(path) != item["hash"]:
+        raise OperationsError("BACKUP_OBJECT_INTEGRITY_FAILED")
+
+
+def _upload_and_verify(objects, source: Path, item: dict[str, Any]) -> str:
+    key = f"{item['vault_id']}/{item['hash']}"
+    path = _object_path(source, item["vault_id"], item["hash"])
+    objects.put_file(key, path, item["hash"])
+    try:
+        response = objects.client.get_object(Bucket=objects.bucket, Key=key)
+        digest = hashlib.sha256()
+        size = 0
+        with response["Body"] as body:
+            for chunk in iter(lambda: body.read(1024 * 1024), b""):
+                digest.update(chunk)
+                size += len(chunk)
+        if size != item["size"] or digest.hexdigest() != item["hash"]:
+            raise OperationsError("RESTORED_OBJECT_INTEGRITY_FAILED")
+    except BaseException:
+        try:
+            objects.delete(key)
+        except BaseException:
+            pass
+        raise
+    return key
+
+
+def _database_is_empty(conn) -> bool:
+    return (
+        conn.execute(
+            text(
+                "SELECT COUNT(*) FROM information_schema.tables "
+                "WHERE table_schema='public'"
+            )
+        ).scalar_one()
+        == 0
+    )
+
+
+def _restore_database(
+    db, snapshot: dict[str, list[list]], expected_objects: list[dict[str, Any]]
+) -> None:
+    with db.transaction() as conn:
+        conn.execute(text("SELECT pg_advisory_xact_lock(1330534488)"))
+        if not _database_is_empty(conn):
+            raise OperationsError("RESTORE_DATABASE_NOT_EMPTY")
+        for statement in SCHEMA:
+            conn.execute(text(statement))
+        for table, columns in TABLES.items():
+            rows = snapshot[table]
+            if not rows:
+                continue
+            names = ",".join(f'"{column}"' for column in columns)
+            values = ",".join(f":v{index}" for index in range(len(columns)))
+            parameters = [
+                {f"v{index}": value for index, value in enumerate(row)} for row in rows
+            ]
+            conn.execute(text(f'INSERT INTO "{table}" ({names}) VALUES ({values})'), parameters)
+        version = conn.execute(text("SELECT version FROM schema_version")).scalar_one()
+        if version != 1 or _manifest_objects(conn) != expected_objects:
+            raise OperationsError("RESTORED_DATABASE_INTEGRITY_FAILED")
+
+
+def restore_backup(
+    db,
+    objects,
+    source: Path,
+    *,
+    workers: int = 8,
+    max_age_hours: float = 24,
+) -> dict[str, Any]:
+    if not math.isfinite(max_age_hours) or max_age_hours <= 0:
+        raise OperationsError("BACKUP_AGE_LIMIT_INVALID")
+    source = source.resolve(strict=True)
+    manifest = _load_manifest(source, max_age_hours)
+    snapshot = _load_database_snapshot(source, manifest)
+    with ThreadPoolExecutor(max_workers=workers) as pool:
+        list(pool.map(lambda item: _verify_backup_file(source, item), manifest["objects"]))
+    with db.engine.connect() as conn:
+        if not _database_is_empty(conn):
+            raise OperationsError("RESTORE_DATABASE_NOT_EMPTY")
+    created_bucket = objects.ensure_bucket()
+    if not objects.is_empty():
+        raise OperationsError("RESTORE_BUCKET_NOT_EMPTY")
+    uploaded: list[str] = []
+    database_restored = False
+    try:
+        failure = None
+        with ThreadPoolExecutor(max_workers=workers) as pool:
+            futures = [
+                pool.submit(_upload_and_verify, objects, source, item)
+                for item in manifest["objects"]
+            ]
+            for future in as_completed(futures):
+                try:
+                    uploaded.append(future.result())
+                except BaseException as error:
+                    failure = failure or error
+        if failure is not None:
+            raise failure
+        _restore_database(db, snapshot, manifest["objects"])
+        database_restored = True
+    except BaseException:
+        if not database_restored:
+            try:
+                objects.delete_many(uploaded)
+            except BaseException:
+                pass
+            if created_bucket:
+                try:
+                    objects.delete_bucket()
+                except BaseException:
+                    pass
+        raise
+    return {
+        "status": "RESTORE_COMPLETE",
+        "backup_age_seconds": manifest["age_seconds"],
+        "object_count": manifest["object_count"],
+        "object_bytes": manifest["object_bytes"],
+        "verified_objects": len(manifest["objects"]),
+    }
@@ -66,6 +66,37 @@ class S3Objects:
                                                 retries={"max_attempts": 0}))
        self.bucket = bucket

+    def ensure_bucket(self) -> bool:
+        """Create the configured bucket when absent; never alter an existing bucket."""
+        from botocore.exceptions import ClientError
+
+        try:
+            self.client.head_bucket(Bucket=self.bucket)
+            return False
+        except ClientError as error:
+            code = str(error.response.get("Error", {}).get("Code", ""))
+            status = error.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
+            if code not in {"404", "NoSuchBucket", "NotFound"} and status != 404:
+                raise
+        self.client.create_bucket(Bucket=self.bucket)
+        return True
+
+    def is_empty(self) -> bool:
+        response = self.client.list_objects_v2(Bucket=self.bucket, MaxKeys=1)
+        return not response.get("Contents")
+
+    def delete_many(self, keys: list[str]) -> None:
+        for start in range(0, len(keys), 1000):
+            batch = keys[start : start + 1000]
+            if batch:
+                self.client.delete_objects(
+                    Bucket=self.bucket,
+                    Delete={"Objects": [{"Key": key} for key in batch], "Quiet": True},
+                )
+
+    def delete_bucket(self) -> None:
+        self.client.delete_bucket(Bucket=self.bucket)
+
    def put(self, key: str, data: bytes):
        self.client.put_object(Bucket=self.bucket, Key=key, Body=data,
                               Metadata={"sha256": hashlib.sha256(data).hexdigest()})