diff --git a/docs/development/OpenNexus生产验收Runner.md b/docs/development/OpenNexus生产验收Runner.md index df9d335..c12fcaa 100644 --- a/docs/development/OpenNexus生产验收Runner.md +++ b/docs/development/OpenNexus生产验收Runner.md @@ -24,4 +24,4 @@ python scripts/phase3-production-acceptance.py ` 报告目录包含 `summary.json`、`case-manifest.json`、`junit.xml`、`cases/.json` 和脱敏的 `logs/.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。这用于阻止误报，不是这些用例的验收通过。 diff --git a/scripts/acceptance_cases/s07_sync_backup.py b/scripts/acceptance_cases/s07_sync_backup.py new file mode 100644 index 0000000..1de483d --- /dev/null +++ b/scripts/acceptance_cases/s07_sync_backup.py @@ -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()) diff --git a/scripts/acceptance_cases/sync_production_stack.py b/scripts/acceptance_cases/sync_production_stack.py index f5d26a7..6e7d7de 100644 --- a/scripts/acceptance_cases/sync_production_stack.py +++ b/scripts/acceptance_cases/sync_production_stack.py @@ -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: diff --git a/scripts/phase3_acceptance.py b/scripts/phase3_acceptance.py index e016a11..c4771a1 100644 --- a/scripts/phase3_acceptance.py +++ b/scripts/phase3_acceptance.py @@ -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}") diff --git a/server sync/README.md b/server sync/README.md index 4266bcc..a1a637d 100644 --- a/server sync/README.md +++ b/server sync/README.md @@ -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 拒绝未知数据库版本，不自动降级；当前历史永久保留，容量管理不能手动删除被历史引用的对象。 ## 四并发大附件传输探针 diff --git a/server sync/compose.yaml b/server sync/compose.yaml index 9c3c0fc..14d6b6b 100644 --- a/server sync/compose.yaml +++ b/server sync/compose.yaml @@ -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: diff --git a/server sync/sync_server/__main__.py b/server sync/sync_server/__main__.py index 748d44c..06c8ed1 100644 --- a/server sync/sync_server/__main__.py +++ b/server sync/sync_server/__main__.py @@ -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__": diff --git a/server sync/sync_server/operations.py b/server sync/sync_server/operations.py new file mode 100644 index 0000000..cab080d --- /dev/null +++ b/server sync/sync_server/operations.py @@ -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"]), + } diff --git a/server sync/sync_server/storage.py b/server sync/sync_server/storage.py index c453242..622ae15 100644 --- a/server sync/sync_server/storage.py +++ b/server sync/sync_server/storage.py @@ -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()})