fix(sync): 将上传持久化移出 ASGI 事件循环

docs/development/OpenNexus生产化实施进度-2026-09-08.md

@@ -644,3 +644,11 @@ Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写
 - 对照（生成式 100 MiB 流）、真实文件写入/恢复、重命名/删除、三种冲突解决、发现/重绑定各运行 3 次，全部通过。四个实际文件工作负载三次最大工作集分别约 12.70、12.63、12.72、12.91 MiB；最大提交内存分别约 2.75、2.85、2.82、2.92 MiB。没有改造前实测基线，不报告下降百分比。
 - 可审阅报告 [Windows Sync 组件内存观测](benchmarks/sync-memory-windows-5d3c214.md) 及 [原始 JSON](benchmarks/sync-memory-windows-5d3c214.json)。原始逐次测试日志留在 `.build/sync-memory-5d3c214/`。脚本 py_compile 通过，测量器和实际工作负载均已运行；本轮未改变 Rust 生产代码，未重复无关全量测试。
 - scope 仅为单个 Rust 组件测试进程，不包括 WebView/Core/服务端/系统文件缓存；没有测量规划基准的磁盘/网络/10000 笔记/双账号双 Vault，也不是 S-09 四并发上传服务总 RSS 的证明。报告明确 NOT_ASSESSED，完整基准、部署和生产化目标保持未完成。
+
+
+## 增量：Sync 分块上传不阻塞 ASGI 事件循环
+
+- 复核服务端现状：complete 已用 hashlib.file_digest 与 put_file，download 已逐块校验到临时文件再 FileResponse；未重复改造已存在的流式路径。
+- upload_chunk 原先在 async 路由中直接执行同步数据库锁等待、文件写入和 fsync，会阻塞该 worker 事件循环。现将整段事务/文件持久化交给 Starlette run_in_threadpool，连接在同一工作线程创建和提交，仍保持先刷盘再确认 offset 的协议顺序。接收块先检查剩余额度再扩展 bytearray，避免把超限块复制进应用缓冲区；ASGI 自身收到的块不计入该应用缓冲上限证明。
+- 新增故障测试以 Event 持续阻塞 fsync，在同一 TestClient ASGI 事件循环中要求 /health 在 2 秒内响应且上传仍在等待；释放后检查偏移与 complete。另测 1 MiB 加 1 字节拒绝、fsync 异常事务回滚、未确认尾部协调截断、恰好 1 MiB 重试成功。
+- uv run --project "server sync" pytest "server sync/tests" -q 全套 25 通过，耗时 7.80 秒，JUnit：.build/sync-server-upload-threadpool.xml。依赖报告 2 项 TestClient/AnyIO 弃用警告，无测试失败。环境为 SQLite/DiskObjects 受控测试，不是 PostgreSQL/MinIO 两 worker 或 S-09 四并发 100 MiB 服务总 RSS/30 分钟负载证明；线程池饱和与真实部署验收仍需继续。完整生产化目标未完成。

server sync/sync_server/app.py

@@ -15,6 +15,7 @@ from fastapi import FastAPI, Header, Query, Request
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse, FileResponse
 from starlette.background import BackgroundTask
+from starlette.concurrency import run_in_threadpool
 
 from .database import Database, password_hash, row, rows, run
 from .models import Commit, Login, Refresh, Upload, VaultCreate
@@ -257,9 +258,15 @@ def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=tim
                            offset: int = Query(ge=0), authorization: str = Header(default="")):
         data = bytearray()
         async for chunk in request.stream():
-            data.extend(chunk)
+            if len(chunk) > 1048576 - len(data):
-            if len(data) > 1048576:
                 raise SyncError(413, "CHUNK_TOO_LARGE")
+            data.extend(chunk)
+        # Keep the transaction and durable write on one worker thread. A slow
+        # database lock or fsync must not block this worker's ASGI event loop.
+        return await run_in_threadpool(persist_upload_chunk, vault_id, upload_id,
+                                       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)

server sync/tests/test_production_storage.py

@@ -97,3 +97,66 @@ def test_readiness_fails_closed_when_object_storage_unavailable(env):
     store.put = unavailable
     assert client.get("/ready").status_code == 503
     assert client.get("/health").status_code == 200
+
+
+def test_slow_upload_fsync_does_not_block_worker_health(env, monkeypatch):
+    from concurrent.futures import ThreadPoolExecutor
+    from threading import Event
+    import os
+
+    client, _, _, _, _ = env
+    auth, _, base = setup(client)
+    info = client.post(base + "/uploads", headers=auth,
+                       json={"content_hash": hashlib.sha256(b"abc").hexdigest(), "size": 3}).json()
+    path = base + "/uploads/" + info["upload_id"]
+    entered, release = Event(), Event()
+    original = os.fsync
+
+    def delayed(fd):
+        entered.set()
+        if not release.wait(10):
+            raise TimeoutError("test did not release fsync")
+        original(fd)
+
+    monkeypatch.setattr(os, "fsync", delayed)
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        pending = pool.submit(client.put, path + "?offset=0", headers=auth, content=b"abc")
+        try:
+            assert entered.wait(5)
+            # Both requests use this TestClient's single ASGI event loop.
+            health = pool.submit(client.get, "/health").result(timeout=2)
+            assert health.status_code == 200
+            assert not pending.done()
+        finally:
+            release.set()
+        assert pending.result(timeout=5).json() == {"offset": 3}
+    assert client.get(path, headers=auth).json()["offset"] == 3
+    assert client.post(path + "/complete", headers=auth).status_code == 200
+
+
+def test_upload_limits_and_failed_fsync_preserve_durable_offset(env, monkeypatch):
+    import os
+
+    client, _, _, staging, _ = env
+    auth, _, base = setup(client)
+    data = b"a" * 1048576
+    info = client.post(base + "/uploads", headers=auth,
+                       json={"content_hash": hashlib.sha256(data).hexdigest(), "size": len(data)}).json()
+    path = base + "/uploads/" + info["upload_id"]
+    rejected = client.put(path + "?offset=0", headers=auth, content=data + b"x")
+    assert rejected.status_code == 413
+    assert rejected.json()["error"]["code"] == "CHUNK_TOO_LARGE"
+    assert (staging / info["upload_id"]).stat().st_size == 0
+
+    def failed(_fd):
+        raise OSError("controlled fsync failure")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(os, "fsync", failed)
+        with pytest.raises(OSError, match="controlled fsync failure"):
+            client.put(path + "?offset=0", headers=auth, content=data)
+    # Failure propagated from the thread; the SQL transaction did not advance.
+    assert client.get(path, headers=auth).json()["offset"] == 0
+    assert (staging / info["upload_id"]).stat().st_size == 0
+    assert client.put(path + "?offset=0", headers=auth, content=data).json() == {"offset": len(data)}
+    assert client.post(path + "/complete", headers=auth).status_code == 200