diff --git a/docs/contracts/Sync-v1契约.md b/docs/contracts/Sync-v1契约.md index 76df0c8..810e1a2 100644 --- a/docs/contracts/Sync-v1契约.md +++ b/docs/contracts/Sync-v1契约.md @@ -1,6 +1,6 @@ # Sync Protocol v1 -状态：服务端协议原型已实现，生产运维与桌面客户端尚未验收。入口为 `server sync/sync_server`；运行时 `/openapi.json` 是字段约束来源。不能将 SQLite Fixture 结果称为 PostgreSQL / MinIO 验收。 +状态：服务端协议已实现，PostgreSQL 17、MinIO 与双 Worker 的 S-04 生产并发验收已通过；其余服务运维门和发布端到端验收仍在推进。入口为 `server sync/sync_server`；运行时 `/openapi.json` 是字段约束来源。SQLite Fixture 只用于协议回归，不能代替生产依赖验收。 ## Rust 上传客户端增量 @@ -40,7 +40,7 @@ Workspace schema 4 增加持久 inbox、分页 boundary 和冲突记录，并为 | GET `/vaults/{id}/history/{file_id}` | before/limit 倒序历史 | | GET `/vaults/{id}/objects/{hash}` | 经鉴权读取与摘要复核 | -上传分块最大 1 MiB，对象最大 100 MiB。offset 冲突为 409 `UPLOAD_OFFSET`，响应给出已确认偏移；客户端查询偏移后续传。完成请求响应丢失时重新预申请同一摘要即可确认对象已存在。未完成上传不允许提交 Revision。 +上传分块最大 1 MiB，对象最大 100 MiB。offset 冲突为 409 `UPLOAD_OFFSET`，响应给出已确认偏移；客户端查询偏移后续传。暂存文件长于数据库 offset 时先截断并 `fsync`，短于 offset 或无法读取时删除损坏上传并释放预留，再返回 409 `UPLOAD_DAMAGED` 与 `restart_required=true`。完成请求响应丢失时用同一 upload ID 重试，持久回执返回原摘要且不会重复计费；重新预申请同一摘要也会确认对象已存在。未完成上传不允许提交 Revision。 提交字段：operation_id、file_id、base_revision、path、operation (`put`/`delete`)、content_hash、size。device_id 从会话取得，客户端不能冒充。移动使用同一 file_id、新 path 及当前对象；恢复历史使用历史摘要、新 operation_id 和当前 base_revision，生成新 Revision。删除必须携带当前基线、空摘要和 size=0。 @@ -50,7 +50,7 @@ Vault 行锁内执行幂等键验证、CAS、路径检查、序列分配及历 ## 保留与未交付边界 -当前历史和 tombstone **永久保留**，游标不主动过期；不启用 GC，以免缺少恢复演练时删除历史对象。配额包含全部历史对象，未完成上传预留额度。通知尚未实现，客户端必须主动按 cursor 拉取。上传过期清理、对象 GC、限速指标、完整备份恢复工具与生产负载验收仍未交付。`/ready` 目前只检查数据库 schema，不证明 S3 就绪。 +当前历史和 tombstone **永久保留**，游标不主动过期；不启用对象 GC，以免缺少恢复演练时删除历史对象。配额包含全部历史对象与未完成上传的预留额度。每个 Worker 每 60 秒清理已过期上传；清理与 PUT/complete 使用相同 Vault 行锁顺序，先删除暂存文件再删除预留记录，失败可在下一轮重试。通知尚未实现，客户端必须主动按 cursor 拉取。对象 GC、限速指标、完整备份恢复工具与生产负载验收仍未交付。`/ready` 会在受控缓存周期内检查数据库 schema、暂存目录写入/`fsync`/读回及 S3 临时对象写入/读回/删除，任一依赖异常返回 503。 验证向量见 `server sync/tests/test_protocol.py`：双设备 CAS、幂等重放、移动与删除、历史恢复、稳定分页、隔离、撤销、刷新、配额、路径冲突、摘要与断点。 diff --git a/docs/development/OpenNexus生产验收Runner.md b/docs/development/OpenNexus生产验收Runner.md index 837ce62..deae8c5 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 Sync 服务 driver 已登记；其余 20 个生产验收 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 Sync 服务 driver 已登记；其余 19 个生产验收 ID 尚未登记，运行时会生成 `NOT_IMPLEMENTED` 证据并退出 1。这用于阻止误报，不是这些用例的验收通过。 diff --git a/scripts/acceptance_cases/s05_sync_uploads.py b/scripts/acceptance_cases/s05_sync_uploads.py new file mode 100644 index 0000000..3f065a1 --- /dev/null +++ b/scripts/acceptance_cases/s05_sync_uploads.py @@ -0,0 +1,438 @@ +"""S-05 upload durability and cleanup races on the production stack.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import socket +import subprocess +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from urllib.parse import urlsplit + +from sync_production_stack import ROOT, SERVICE, SyncProductionStack, call, sha256, stop_tree + +ROUNDS = 100 + + +def result(case_id: str, status: str, reason: str, facts: dict) -> dict: + passed = status == "PASSED" + assertions = [ + ( + "the oracle uses real PostgreSQL, MinIO, and both production workers", + facts.get("dependencies") is True and facts.get("worker_count") == 2, + ), + ( + "100 same-offset races never acknowledge more than durable staging bytes", + facts.get("offset_races") == ROUNDS and facts.get("durable_offsets") == ROUNDS, + ), + ( + "100 disk-ahead and 100 disk-behind uploads reconcile or restart safely", + facts.get("disk_ahead") == ROUNDS and facts.get("disk_behind") == ROUNDS, + ), + ( + "100 lost complete responses recover from durable receipts", + facts.get("response_loss_recoveries") == ROUNDS, + ), + ( + "complete retries charge every unique object exactly once", + facts.get("quota_exact") is True and facts.get("receipt_count_exact") is True, + ), + ( + "100 expiry cleanup races release every reservation within 15 minutes", + facts.get("cleanup_races") == ROUNDS + and facts.get("cleanup_split") == {"complete": 50, "cleanup": 50} + and facts.get("remaining_uploads") == 0 + and facts.get("reserved_bytes") == 0 + and facts.get("max_cleanup_latency_ms", 900001) < 900000, + ), + ( + "all referenced objects remain readable and cleaned uploads are not referenced", + facts.get("referenced_objects_verified") == facts.get("object_count") + and facts.get("cleaned_objects_absent") == 50 + and facts.get("staging_files") == 0, + ), + ] + evidence = "real PostgreSQL 17, MinIO S3, two Uvicorn workers, shared staging, and black-box HTTP" + 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": { + "offset_races": facts.get("offset_races", 0), + "response_loss_recoveries": facts.get("response_loss_recoveries", 0), + "cleanup_races": facts.get("cleanup_races", 0), + "max_cleanup_latency_ms": facts.get("max_cleanup_latency_ms", 0), + "worker_count": facts.get("worker_count", 0), + }, + "files": [ + {"path": relative, "sha256": sha256(ROOT / relative)} + for relative in ( + "server sync/sync_server/app.py", + "server sync/sync_server/database.py", + "server sync/sync_server/maintenance.py", + "server sync/sync_server/storage.py", + "server sync/tests/test_production_storage.py", + "scripts/acceptance_cases/sync_production_stack.py", + "scripts/acceptance_cases/s05_sync_uploads.py", + ) + ], + "revisions": [ + { + "scope": "runtime", + "postgres": facts.get("postgres_version"), + "minio": facts.get("minio_version"), + }, + { + "scope": "upload fault matrix", + "same_offset": facts.get("offset_races", 0), + "disk_ahead": facts.get("disk_ahead", 0), + "disk_behind": facts.get("disk_behind", 0), + "response_loss": facts.get("response_loss_recoveries", 0), + "cleanup_races": facts.get("cleanup_races", 0), + }, + { + "scope": "cleanup outcomes", + **facts.get("cleanup_split", {}), + "max_latency_ms": facts.get("max_cleanup_latency_ms", 0), + }, + { + "scope": "final integrity", + "objects": facts.get("object_count", 0), + "used_bytes": facts.get("used_bytes", 0), + "reserved_bytes": facts.get("reserved_bytes", -1), + }, + ], + } + + +def content(vector: str, index: int) -> bytes: + return f"OpenNexus-S05-{vector}-{index:03d}".encode() + + +def begin(base: str, auth: dict, data: bytes): + digest = hashlib.sha256(data).hexdigest() + code, info, _ = call( + "POST", base + "/uploads", headers=auth, body={"content_hash": digest, "size": len(data)} + ) + assert code == 200 and info == { + "complete": False, + "upload_id": info["upload_id"], + "offset": 0, + } + return digest, info["upload_id"], base + "/uploads/" + info["upload_id"] + + +def complete(path: str, auth: dict, digest: str) -> None: + code, body, _ = call("POST", path + "/complete", headers=auth) + assert code == 200 and body == {"complete": True, "content_hash": digest} + + +def drop_complete_response(url: str, authorization: str) -> None: + parsed = urlsplit(url) + target = parsed.path + (("?" + parsed.query) if parsed.query else "") + request = ( + f"POST {target} HTTP/1.1\r\n" + f"Host: {parsed.hostname}:{parsed.port}\r\n" + f"Authorization: {authorization}\r\n" + "Content-Length: 0\r\n" + "Connection: close\r\n\r\n" + ).encode("ascii") + with socket.create_connection((parsed.hostname, parsed.port), timeout=10) as stream: + stream.sendall(request) + stream.shutdown(socket.SHUT_WR) + # The request is complete, but the client deliberately never reads its response. + time.sleep(0.01) + + +class CleanupClient: + def __init__(self, stack: SyncProductionStack): + program = ( + "import json,os,sys,time; from pathlib import Path; " + "from sync_server.database import Database; " + "from sync_server.maintenance import cleanup_expired_uploads; " + "d=Database(os.environ['SYNC_DATABASE_URL']); s=Path(os.environ['SYNC_STAGING_DIR']); " + "\nfor line in sys.stdin:\n" + " r=json.loads(line); time.sleep(r['delay_ms']/1000); " + " print(json.dumps(cleanup_expired_uploads(d,s,now=r['now'],limit=500)),flush=True)" + ) + self.process = subprocess.Popen( + [str(stack.server_python), "-u", "-c", program], + cwd=SERVICE, + env=stack.service_env, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=stack.handles[2], + text=True, + bufsize=1, + ) + self.lock = threading.Lock() + + def cleanup(self, *, now: int, delay_ms: int) -> dict: + with self.lock: + if self.process.poll() is not None or self.process.stdin is None or self.process.stdout is None: + raise RuntimeError("CLEANUP_HELPER_EXITED") + self.process.stdin.write(json.dumps({"now": now, "delay_ms": delay_ms}) + "\n") + self.process.stdin.flush() + line = self.process.stdout.readline() + if not line: + raise RuntimeError("CLEANUP_HELPER_NO_RESPONSE") + return json.loads(line) + + def close(self) -> None: + if self.process.stdin is not None: + self.process.stdin.close() + try: + self.process.wait(timeout=10) + except subprocess.TimeoutExpired: + stop_tree(self.process) + if self.process.stdout is not None: + self.process.stdout.close() + + +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" + stack = cleanup_client = None + try: + config = json.loads(Path(args.config).read_text(encoding="utf-8")) + stack = SyncProductionStack( + config, Path(os.environ["OPENNEXUS_ACCEPTANCE_DATA_ROOT"]), "s05" + ).start() + facts.update( + { + "dependencies": True, + "postgres_version": stack.postgres_version, + "minio_version": stack.minio_version, + "worker_count": len(stack.worker_ids()), + } + ) + assert facts["worker_count"] == 2 + auth, base, vault_id = stack.create_vault("S-05 isolated") + committed: dict[str, bytes] = {} + cleaned: dict[str, bytes] = {} + expected_used = 0 + + offset_races = durable_offsets = 0 + for index in range(ROUNDS): + data = content("offset", index) + digest, upload_id, path = begin(base, auth, data) + with ThreadPoolExecutor(max_workers=2) as pool: + attempts = list( + pool.map( + lambda _: call("PUT", path + "?offset=0", headers=auth, body=data), + range(2), + ) + ) + codes = sorted(item[0] for item in attempts) + rejected = next(item for item in attempts if item[0] == 409) + assert codes == [200, 409] + assert rejected[1]["error"] == { + "code": "UPLOAD_OFFSET", + "details": {"offset": len(data)}, + } + state = call("GET", path, headers=auth) + assert state[0] == 200 and state[1]["offset"] == len(data) + assert (stack.staging / upload_id).stat().st_size == state[1]["offset"] + complete(path, auth, digest) + committed[digest] = data + expected_used += len(data) + offset_races += 1 + durable_offsets += 1 + facts.update({"offset_races": offset_races, "durable_offsets": durable_offsets}) + + disk_ahead = 0 + for index in range(ROUNDS): + data = content("ahead", index) + digest, upload_id, path = begin(base, auth, data) + local = stack.staging / upload_id + local.write_bytes(data + b"-uncommitted-tail") + state = call("GET", path, headers=auth) + assert state[0] == 200 and state[1]["offset"] == 0 and local.stat().st_size == 0 + assert call("PUT", path + "?offset=0", headers=auth, body=data)[1] == { + "offset": len(data) + } + complete(path, auth, digest) + committed[digest] = data + expected_used += len(data) + disk_ahead += 1 + facts["disk_ahead"] = disk_ahead + + disk_behind = 0 + for index in range(ROUNDS): + data = content("behind", index) + digest, upload_id, path = begin(base, auth, data) + assert call("PUT", path + "?offset=0", headers=auth, body=data[:1])[1] == { + "offset": 1 + } + local = stack.staging / upload_id + local.write_bytes(b"") + damaged = call("GET", path, headers=auth) + assert damaged[0] == 409 and damaged[1]["error"] == { + "code": "UPLOAD_DAMAGED", + "details": {"restart_required": True}, + } + assert not local.exists() + expired = call("GET", path, headers=auth) + assert expired[0] == 404 and expired[1]["error"]["code"] == "UPLOAD_EXPIRED" + replacement_digest, replacement_id, replacement = begin(base, auth, data) + assert replacement_digest == digest and replacement_id != upload_id + assert call("PUT", replacement + "?offset=0", headers=auth, body=data)[1] == { + "offset": len(data) + } + complete(replacement, auth, digest) + committed[digest] = data + expected_used += len(data) + disk_behind += 1 + facts["disk_behind"] = disk_behind + + response_loss_recoveries = 0 + for index in range(ROUNDS): + data = content("lost-complete", index) + digest, upload_id, path = begin(base, auth, data) + assert call("PUT", path + "?offset=0", headers=auth, body=data)[1] == { + "offset": len(data) + } + drop_complete_response(path + "/complete", auth["Authorization"]) + deadline = time.monotonic() + 10 + while (stack.staging / upload_id).exists() and time.monotonic() < deadline: + time.sleep(0.01) + assert not (stack.staging / upload_id).exists() + complete(path, auth, digest) + complete(path, auth, digest) + committed[digest] = data + expected_used += len(data) + response_loss_recoveries += 1 + facts["response_loss_recoveries"] = response_loss_recoveries + + cleanup_client = CleanupClient(stack) + cleanup_complete = cleanup_removed = 0 + cleanup_latencies = [] + for index in range(ROUNDS): + data = content("cleanup", index) + digest, upload_id, path = begin(base, auth, data) + assert call("PUT", path + "?offset=0", headers=auth, body=data)[1] == { + "offset": len(data) + } + barrier = threading.Barrier(2) + complete_delay = 0 if index % 2 == 0 else 250 + cleanup_delay = 250 if index % 2 == 0 else 0 + + def race_complete(): + barrier.wait(timeout=5) + time.sleep(complete_delay / 1000) + return call("POST", path + "/complete", headers=auth) + + def race_cleanup(): + barrier.wait(timeout=5) + started = time.monotonic() + outcome = cleanup_client.cleanup( + now=int(time.time()) + 7200, delay_ms=cleanup_delay + ) + return outcome, int((time.monotonic() - started) * 1000) + + with ThreadPoolExecutor(max_workers=2) as pool: + complete_future = pool.submit(race_complete) + cleanup_future = pool.submit(race_cleanup) + complete_result = complete_future.result(timeout=30) + cleanup_result, latency = cleanup_future.result(timeout=30) + cleanup_latencies.append(latency) + if index % 2 == 0: + assert complete_result[0] == 200 and cleanup_result["expired_uploads_removed"] == 0 + cleanup_complete += 1 + committed[digest] = data + expected_used += len(data) + else: + assert complete_result[0] == 404 + assert complete_result[1]["error"]["code"] == "UPLOAD_EXPIRED" + assert cleanup_result["expired_uploads_removed"] == 1 + cleanup_removed += 1 + cleaned[digest] = data + assert not (stack.staging / upload_id).exists() + cleanup_client.close() + cleanup_client = None + facts.update( + { + "cleanup_races": ROUNDS, + "cleanup_split": {"complete": cleanup_complete, "cleanup": cleanup_removed}, + "max_cleanup_latency_ms": max(cleanup_latencies), + } + ) + + vaults = call("GET", stack.origin + "/sync/v1/vaults", headers=auth)[1]["items"] + current = next(item for item in vaults if item["id"] == vault_id) + facts["used_bytes"] = current["used"] + facts["quota_exact"] = current["used"] == expected_used + facts["remaining_uploads"] = stack.sql_scalar( + f"SELECT COUNT(*) FROM uploads WHERE vault_id='{vault_id}'" + ) + facts["reserved_bytes"] = stack.sql_scalar( + f"SELECT COALESCE(SUM(size),0) FROM uploads WHERE vault_id='{vault_id}'" + ) + facts["object_count"] = stack.sql_scalar( + f"SELECT COUNT(*) FROM objects WHERE vault_id='{vault_id}'" + ) + receipt_count = stack.sql_scalar( + f"SELECT COUNT(*) FROM upload_receipts WHERE vault_id='{vault_id}'" + ) + facts["receipt_count_exact"] = receipt_count == len(committed) + assert facts["object_count"] == len(committed) + + verified = 0 + for digest, data in committed.items(): + response = call("GET", base + "/objects/" + digest, headers=auth) + assert response[0] == 200 and response[1] == data + verified += 1 + absent = 0 + for digest in cleaned: + response = call("GET", base + "/objects/" + digest, headers=auth) + assert response[0] == 404 and response[1]["error"]["code"] == "OBJECT_NOT_FOUND" + absent += 1 + facts["referenced_objects_verified"] = verified + facts["cleaned_objects_absent"] = absent + facts["staging_files"] = len(list(stack.staging.iterdir())) + stack.assert_running() + assert len(stack.worker_ids()) == 2 + assert facts["quota_exact"] and facts["receipt_count_exact"] + assert facts["remaining_uploads"] == facts["reserved_bytes"] == facts["staging_files"] == 0 + status = "PASSED" + except BaseException as error: + reason = "S05_ORACLE_FAILED:" + type(error).__name__ + finally: + if cleanup_client is not None: + cleanup_client.close() + if stack is not None: + stack.stop() + payload = result( + case_id, + status if case_id == "S-05" else "FAILED", + reason or ("" if case_id == "S-05" 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__": + raise SystemExit(main()) diff --git a/scripts/acceptance_cases/sync_production_stack.py b/scripts/acceptance_cases/sync_production_stack.py new file mode 100644 index 0000000..f9400bb --- /dev/null +++ b/scripts/acceptance_cases/sync_production_stack.py @@ -0,0 +1,351 @@ +"""Isolated PostgreSQL, MinIO, and multi-worker Sync acceptance stack.""" + +from __future__ import annotations + +import hashlib +import json +import os +import secrets +import socket +import subprocess +import time +from pathlib import Path +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + +ROOT = Path(__file__).resolve().parents[2] +SERVICE = ROOT / "server sync" +MINIO_RELEASE = "RELEASE.2025-09-07T16-13-09Z" +MINIO_SHA256 = "af709e6ba68488404e85acdd22a3030d0f5e56a108d4b27d744f18ceb50861b4" + + +def sha256(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 free_port() -> int: + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + return listener.getsockname()[1] + + +def call(method: str, url: str, *, headers=None, body=None, timeout=30): + request_headers = dict(headers or {}) + payload = body + if isinstance(body, (dict, list)): + payload = json.dumps(body, separators=(",", ":")).encode() + request_headers["Content-Type"] = "application/json" + request = Request(url, data=payload, headers=request_headers, method=method) + try: + response = urlopen(request, timeout=timeout) + except HTTPError as error: + response = error + data = response.read() + media_type = response.headers.get("Content-Type", "") + value = json.loads(data) if data and "json" in media_type else data + return response.status, value, {key.lower(): item for key, item in response.headers.items()} + + +def wait_http(url: str, *, expected=200, timeout=30) -> None: + until = time.monotonic() + timeout + while time.monotonic() < until: + try: + if call("GET", url)[0] == expected: + return + except (OSError, URLError): + pass + time.sleep(0.1) + raise RuntimeError("SERVICE_START_TIMEOUT") + + +def stop_tree(process: subprocess.Popen | None) -> None: + if process is None or process.poll() is not None: + return + if os.name == "nt": + subprocess.run( + ["taskkill", "/PID", str(process.pid), "/T", "/F"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + ) + else: + process.terminate() + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=10) + + +class SyncProductionStack: + """Own a disposable production dependency stack for one acceptance case.""" + + def __init__(self, config: dict, data_root: Path, case_tag: str): + self.config = config + self.data_root = data_root + self.case_tag = case_tag.lower() + self.postgres_started = False + self.minio = None + self.sync = None + self.handles = [] + + initdb = Path(config["artifacts"]["postgres_initdb"]).resolve() + self.pg_bin = initdb.parent + executable = lambda name: self.pg_bin / (name + ".exe" if os.name == "nt" else name) + self.initdb = initdb + self.pg_ctl = executable("pg_ctl") + self.createdb = executable("createdb") + self.postgres = executable("postgres") + self.psql = executable("psql") + self.minio_server = Path(config["artifacts"]["minio_server"]).resolve() + self.server_python = SERVICE / (".venv/Scripts/python.exe" if os.name == "nt" else ".venv/bin/python") + required = ( + self.initdb, + self.pg_ctl, + self.createdb, + self.postgres, + self.psql, + self.minio_server, + self.server_python, + ) + if not all(path.is_file() for path in required): + raise RuntimeError("PRODUCTION_RUNTIME_MISSING") + + self.postgres_version = subprocess.check_output( + [str(self.postgres), "--version"], text=True, timeout=10 + ).strip() + self.minio_version = subprocess.check_output( + [str(self.minio_server), "--version"], text=True, timeout=10 + ).splitlines()[0] + if " 17." not in self.postgres_version or MINIO_RELEASE not in self.minio_version: + raise RuntimeError("PRODUCTION_RUNTIME_VERSION_MISMATCH") + if sha256(self.minio_server) != MINIO_SHA256: + raise RuntimeError("MINIO_ARTIFACT_INTEGRITY") + + self.stack = data_root / f"{self.case_tag}-production-stack" + self.postgres_data = self.stack / "postgres" + self.minio_data = self.stack / "objects" + self.staging = self.stack / "staging" + self.pg_port, self.minio_port, self.console_port, self.sync_port = ( + free_port() for _ in range(4) + ) + self.database_url = f"postgresql+psycopg://postgres@127.0.0.1:{self.pg_port}/opennexus" + self.origin = f"http://127.0.0.1:{self.sync_port}" + self.username = self.case_tag + "-" + secrets.token_hex(8) + self.password = secrets.token_urlsafe(32) + self.minio_user = self.case_tag + secrets.token_hex(8) + self.minio_password = secrets.token_urlsafe(32) + self.bucket = "opennexus-" + self.case_tag + self.service_env = {} + + def start(self) -> "SyncProductionStack": + self.stack.mkdir() + self.minio_data.mkdir() + self.staging.mkdir() + pg_log = (self.stack / "postgres.log").open("wb") + minio_log = (self.stack / "minio.log").open("wb") + sync_log = (self.stack / "sync.log").open("wb") + self.handles.extend([pg_log, minio_log, sync_log]) + + subprocess.run( + [ + str(self.initdb), + "-D", + str(self.postgres_data), + "-U", + "postgres", + "-A", + "trust", + "--no-locale", + "-E", + "UTF8", + ], + stdout=pg_log, + stderr=subprocess.STDOUT, + check=True, + timeout=120, + ) + subprocess.run( + [ + str(self.pg_ctl), + "-D", + str(self.postgres_data), + "-l", + str(self.stack / "postgres-server.log"), + "-o", + f"-p {self.pg_port} -h 127.0.0.1", + "-w", + "start", + ], + stdout=pg_log, + stderr=subprocess.STDOUT, + check=True, + timeout=60, + ) + self.postgres_started = True + subprocess.run( + [ + str(self.createdb), + "-h", + "127.0.0.1", + "-p", + str(self.pg_port), + "-U", + "postgres", + "opennexus", + ], + stdout=pg_log, + stderr=subprocess.STDOUT, + check=True, + timeout=30, + ) + + minio_env = os.environ.copy() + minio_env.update( + {"MINIO_ROOT_USER": self.minio_user, "MINIO_ROOT_PASSWORD": self.minio_password} + ) + self.minio = subprocess.Popen( + [ + str(self.minio_server), + "server", + str(self.minio_data), + "--address", + f"127.0.0.1:{self.minio_port}", + "--console-address", + f"127.0.0.1:{self.console_port}", + ], + stdout=minio_log, + stderr=subprocess.STDOUT, + env=minio_env, + ) + wait_http(f"http://127.0.0.1:{self.minio_port}/minio/health/ready") + + self.service_env = os.environ.copy() + self.service_env.update( + { + "SYNC_DATABASE_URL": self.database_url, + "SYNC_S3_ENDPOINT": f"http://127.0.0.1:{self.minio_port}", + "SYNC_S3_BUCKET": self.bucket, + "SYNC_STAGING_DIR": str(self.staging), + "SYNC_HOST": "127.0.0.1", + "SYNC_PORT": str(self.sync_port), + "AWS_ACCESS_KEY_ID": self.minio_user, + "AWS_SECRET_ACCESS_KEY": self.minio_password, + "AWS_DEFAULT_REGION": "us-east-1", + "ACCEPTANCE_USERNAME": self.username, + "ACCEPTANCE_PASSWORD": self.password, + } + ) + 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'])", + ], + cwd=SERVICE, + env=self.service_env, + stdout=sync_log, + stderr=subprocess.STDOUT, + check=True, + timeout=60, + ) + self.sync = subprocess.Popen( + [str(self.server_python), "-m", "sync_server", "serve", "--workers", "2"], + cwd=SERVICE, + env=self.service_env, + stdout=sync_log, + stderr=subprocess.STDOUT, + ) + wait_http(self.origin + "/ready", timeout=60) + return self + + def create_vault(self, label: str): + code, session, _ = call( + "POST", + self.origin + "/sync/v1/auth/sessions", + body={ + "username": self.username, + "password": self.password, + "device_name": label, + }, + ) + if code != 200: + raise RuntimeError("ACCEPTANCE_LOGIN_FAILED") + auth = {"Authorization": "Bearer " + session["access_token"]} + code, vault, _ = call( + "POST", self.origin + "/sync/v1/vaults", headers=auth, body={"name": label} + ) + if code != 200: + raise RuntimeError("ACCEPTANCE_VAULT_FAILED") + return auth, self.origin + "/sync/v1/vaults/" + vault["vault_id"], vault["vault_id"] + + def sql_scalar(self, statement: str) -> int: + output = subprocess.check_output( + [ + str(self.psql), + "-X", + "-A", + "-t", + "-v", + "ON_ERROR_STOP=1", + "-h", + "127.0.0.1", + "-p", + str(self.pg_port), + "-U", + "postgres", + "-d", + "opennexus", + "-c", + statement, + ], + text=True, + timeout=30, + ).strip() + return int(output) + + def worker_ids(self, attempts=200) -> set[str]: + from concurrent.futures import ThreadPoolExecutor + + def probe(_): + return call( + "GET", self.origin + "/health", headers={"Connection": "close"} + )[2].get("x-opennexus-worker") + + with ThreadPoolExecutor(max_workers=40) as pool: + return {value for value in pool.map(probe, range(attempts)) if value} + + def assert_running(self) -> None: + if self.sync is None or self.minio is None: + raise RuntimeError("PRODUCTION_STACK_NOT_STARTED") + if self.sync.poll() is not None or self.minio.poll() is not None: + raise RuntimeError("PRODUCTION_STACK_EXITED") + + def stop(self) -> None: + stop_tree(self.sync) + stop_tree(self.minio) + if self.postgres_started: + subprocess.run( + [ + str(self.pg_ctl), + "-D", + str(self.postgres_data), + "-m", + "fast", + "-w", + "stop", + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + timeout=30, + ) + for handle in self.handles: + handle.close() diff --git a/scripts/phase3_acceptance.py b/scripts/phase3_acceptance.py index c06b32a..6e0c7a2 100644 --- a/scripts/phase3_acceptance.py +++ b/scripts/phase3_acceptance.py @@ -83,6 +83,18 @@ CASE_DRIVERS: dict[str, dict[str, Any]] = { "required_metrics": ("successful_commits", "conflict_responses", "worker_count"), "required_artifacts": ("postgres_initdb", "minio_server"), }, + "S-05": { + "driver": "scripts/acceptance_cases/s05_sync_uploads.py", + "timeout_seconds": 900, + "required_metrics": ( + "offset_races", + "response_loss_recoveries", + "cleanup_races", + "max_cleanup_latency_ms", + "worker_count", + ), + "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/sync_server/app.py b/server sync/sync_server/app.py index 85ecf52..b790c39 100644 --- a/server sync/sync_server/app.py +++ b/server sync/sync_server/app.py @@ -27,6 +27,15 @@ class SyncError(Exception): self.status, self.code, self.details = status, code, details or {} +class StagingDamaged(Exception): + """Internal signal used to commit cleanup before returning UPLOAD_DAMAGED.""" + + def __init__(self, upload): + self.upload_id = upload["id"] + self.vault_id = upload["vault_id"] + self.device_id = upload["device_id"] + + def digest(value: str) -> str: return hashlib.sha256(value.encode()).hexdigest() @@ -232,22 +241,48 @@ def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=tim try: length = path.stat().st_size if length < upload["offset_bytes"]: - raise SyncError(409, "UPLOAD_DAMAGED", {"restart_required": True}) + raise StagingDamaged(upload) if length > upload["offset_bytes"]: with path.open("r+b") as stream: stream.truncate(upload["offset_bytes"]) stream.flush() os.fsync(stream.fileno()) except OSError: - raise SyncError(409, "UPLOAD_DAMAGED", {"restart_required": True}) from None + raise StagingDamaged(upload) from None return path + def discard_damaged_upload(damaged): + # Re-open after the failed operation released its transaction. Deleting + # inside the failed transaction would be rolled back with the response. + with db.transaction() as conn: + suffix = " FOR UPDATE" if not db.sqlite else "" + row(conn, "SELECT id FROM vaults WHERE id=:v" + suffix, v=damaged.vault_id) + upload = row( + conn, + "SELECT id FROM uploads WHERE id=:id AND vault_id=:v AND device_id=:device", + id=damaged.upload_id, + v=damaged.vault_id, + device=damaged.device_id, + ) + if upload: + # File first: interruption leaves a row whose quota reservation + # can still be released by expiry maintenance. + (staging / upload["id"]).unlink(missing_ok=True) + run(conn, "DELETE FROM uploads WHERE id=:id", id=upload["id"]) + + def upload_damaged(damaged): + discard_damaged_upload(damaged) + raise SyncError(409, "UPLOAD_DAMAGED", {"restart_required": True}) + @app.get("/sync/v1/vaults/{vault_id}/uploads/{upload_id}") def upload_status(vault_id: str, upload_id: str, authorization: str = Header(default="")): - with db.transaction() as conn: - upload = authorized_upload(conn, vault_id, upload_id, authorization) - reconcile_staging(upload) - return {"offset": upload["offset_bytes"], "size": upload["size"], "expires": upload["expires"]} + try: + with db.transaction() as conn: + upload = authorized_upload(conn, vault_id, upload_id, authorization) + reconcile_staging(upload) + return {"offset": upload["offset_bytes"], "size": upload["size"], "expires": upload["expires"]} + except StagingDamaged as damaged: + upload_damaged(damaged) @app.put("/sync/v1/vaults/{vault_id}/uploads/{upload_id}") async def upload_chunk(vault_id: str, upload_id: str, request: Request, @@ -263,23 +298,26 @@ def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=tim authorization, offset, data) def persist_upload_chunk(vault_id, upload_id, authorization, offset, data): - with db.transaction() as conn: - upload = authorized_upload(conn, vault_id, upload_id, authorization) - reconcile_staging(upload) - if offset != upload["offset_bytes"]: - raise SyncError(409, "UPLOAD_OFFSET", {"offset": upload["offset_bytes"]}) - if offset + len(data) > upload["size"]: - raise SyncError(413, "OBJECT_TOO_LARGE") - # 先刷盘后提交 offset；崩溃重试覆盖未确认尾部，不能重复追加。 - import os - with (staging / upload_id).open("r+b") as stream: - stream.seek(offset) - stream.write(data) - stream.truncate() - stream.flush() - os.fsync(stream.fileno()) - run(conn, "UPDATE uploads SET offset_bytes=:offset WHERE id=:id", id=upload_id, offset=offset + len(data)) - return {"offset": offset + len(data)} + try: + with db.transaction() as conn: + upload = authorized_upload(conn, vault_id, upload_id, authorization) + reconcile_staging(upload) + if offset != upload["offset_bytes"]: + raise SyncError(409, "UPLOAD_OFFSET", {"offset": upload["offset_bytes"]}) + if offset + len(data) > upload["size"]: + raise SyncError(413, "OBJECT_TOO_LARGE") + # 先刷盘后提交 offset；崩溃重试覆盖未确认尾部，不能重复追加。 + import os + with (staging / upload_id).open("r+b") as stream: + stream.seek(offset) + stream.write(data) + stream.truncate() + stream.flush() + os.fsync(stream.fileno()) + run(conn, "UPDATE uploads SET offset_bytes=:offset WHERE id=:id", id=upload_id, offset=offset + len(data)) + return {"offset": offset + len(data)} + except StagingDamaged as damaged: + upload_damaged(damaged) @app.delete("/sync/v1/vaults/{vault_id}/uploads/{upload_id}", status_code=204) def cancel_upload(vault_id: str, upload_id: str, authorization: str = Header(default="")): @@ -290,26 +328,29 @@ def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=tim @app.post("/sync/v1/vaults/{vault_id}/uploads/{upload_id}/complete") def complete_upload(vault_id: str, upload_id: str, authorization: str = Header(default="")): - with db.transaction() as conn: - session, _ = vault(conn, vault_id, authorization, lock=True) - receipt = row(conn, "SELECT hash FROM upload_receipts WHERE id=:id AND vault_id=:v AND device_id=:d", - id=upload_id, v=vault_id, d=session["device_id"]) - if receipt: - return {"complete": True, "content_hash": receipt["hash"]} - upload = authorized_upload(conn, vault_id, upload_id, authorization) - path = reconcile_staging(upload) - with path.open("rb") as stream: - content_hash = hashlib.file_digest(stream, "sha256").hexdigest() - if upload["size"] != upload["offset_bytes"] or content_hash != upload["hash"]: - raise SyncError(422, "OBJECT_INTEGRITY") - exists = row(conn, "SELECT hash FROM objects WHERE vault_id=:v AND hash=:h", v=vault_id, h=upload["hash"]) - if not exists: - objects.put_file(vault_id + "/" + upload["hash"], path, upload["hash"]) - run(conn, "INSERT INTO objects VALUES (:v,:h,:size,:now)", v=vault_id, h=upload["hash"], size=upload["size"], now=int(clock())) - run(conn, "UPDATE vaults SET used=used+:size WHERE id=:v", size=upload["size"], v=vault_id) - run(conn, "INSERT INTO upload_receipts VALUES (:id,:v,:d,:h,:now)", id=upload_id, - v=vault_id, d=session["device_id"], h=upload["hash"], now=int(clock())) - run(conn, "DELETE FROM uploads WHERE id=:id", id=upload_id) + try: + with db.transaction() as conn: + session, _ = vault(conn, vault_id, authorization, lock=True) + receipt = row(conn, "SELECT hash FROM upload_receipts WHERE id=:id AND vault_id=:v AND device_id=:d", + id=upload_id, v=vault_id, d=session["device_id"]) + if receipt: + return {"complete": True, "content_hash": receipt["hash"]} + upload = authorized_upload(conn, vault_id, upload_id, authorization) + path = reconcile_staging(upload) + with path.open("rb") as stream: + content_hash = hashlib.file_digest(stream, "sha256").hexdigest() + if upload["size"] != upload["offset_bytes"] or content_hash != upload["hash"]: + raise SyncError(422, "OBJECT_INTEGRITY") + exists = row(conn, "SELECT hash FROM objects WHERE vault_id=:v AND hash=:h", v=vault_id, h=upload["hash"]) + if not exists: + objects.put_file(vault_id + "/" + upload["hash"], path, upload["hash"]) + run(conn, "INSERT INTO objects VALUES (:v,:h,:size,:now)", v=vault_id, h=upload["hash"], size=upload["size"], now=int(clock())) + run(conn, "UPDATE vaults SET used=used+:size WHERE id=:v", size=upload["size"], v=vault_id) + run(conn, "INSERT INTO upload_receipts VALUES (:id,:v,:d,:h,:now)", id=upload_id, + v=vault_id, d=session["device_id"], h=upload["hash"], now=int(clock())) + run(conn, "DELETE FROM uploads WHERE id=:id", id=upload_id) + except StagingDamaged as damaged: + upload_damaged(damaged) (staging / upload_id).unlink(missing_ok=True) return {"complete": True, "content_hash": upload["hash"]} diff --git a/server sync/tests/test_production_storage.py b/server sync/tests/test_production_storage.py index f86919a..7134b58 100644 --- a/server sync/tests/test_production_storage.py +++ b/server sync/tests/test_production_storage.py @@ -23,7 +23,7 @@ def env(tmp_path): def test_offset_reconciliation_never_acknowledges_missing_disk_bytes(env): - client, _, _, staging, _ = env + client, db, _, staging, _ = env auth, _, base = setup(client) info = client.post(base + "/uploads", headers=auth, json={"content_hash": hashlib.sha256(b"abc").hexdigest(), "size": 3}).json() @@ -35,7 +35,17 @@ def test_offset_reconciliation_never_acknowledges_missing_disk_bytes(env): assert local.read_bytes() == b"a" local.write_bytes(b"") assert client.get(path, headers=auth).json()["error"]["code"] == "UPLOAD_DAMAGED" - assert client.put(path + "?offset=1", headers=auth, content=b"bc").status_code == 409 + assert not local.exists() + assert client.put(path + "?offset=1", headers=auth, content=b"bc").status_code == 404 + with db.transaction() as conn: + assert row(conn, "SELECT COUNT(*) AS n FROM uploads")["n"] == 0 + replacement = client.post( + base + "/uploads", + headers=auth, + json={"content_hash": hashlib.sha256(b"abc").hexdigest(), "size": 3}, + ).json() + assert replacement["complete"] is False + assert replacement["upload_id"] != info["upload_id"] def test_complete_retry_has_durable_receipt_and_charges_once(env):