fix(sync): 强化 S-05 上传持久性

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())

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()

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}")

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="")):
+        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,6 +298,7 @@ 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):
+        try:
             with db.transaction() as conn:
                 upload = authorized_upload(conn, vault_id, upload_id, authorization)
                 reconcile_staging(upload)
@@ -280,6 +316,8 @@ def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=tim
                     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,6 +328,7 @@ 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="")):
+        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",
@@ -310,6 +349,8 @@ def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=tim
                 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"]}
 

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):