From 640e45196f4c6008b135e729d4b7414ccb8533f7 Mon Sep 17 00:00:00 2001 From: KiriAky 107 Date: Wed, 9 Sep 2026 09:22:47 +0800 Subject: [PATCH] =?UTF-8?q?test(sync):=20=E7=99=BB=E8=AE=B0=20S-01=20?= =?UTF-8?q?=E7=94=9F=E4=BA=A7=E9=AA=8C=E6=94=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/development/OpenNexus生产验收Runner.md | 2 +- frontend/src-tauri/tests/sync_push.rs | 102 ++++++++++++++++++-- scripts/acceptance_cases/s01_sync_client.py | 96 ++++++++++++++++++ scripts/phase3_acceptance.py | 5 + server sync/tests/host_fixture.py | 6 ++ 5 files changed, 203 insertions(+), 8 deletions(-) create mode 100644 scripts/acceptance_cases/s01_sync_client.py diff --git a/docs/development/OpenNexus生产验收Runner.md b/docs/development/OpenNexus生产验收Runner.md index 113f244..00ff0d6 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 与失败闭合行为已实现，B-02 Fernet 迁移 driver 已登记；其余 29 个生产验收 ID 尚未登记，运行时会生成 `NOT_IMPLEMENTED` 证据并退出 1。这用于阻止误报，不是这些用例的验收通过。 +当前 runner 与失败闭合行为已实现，B-02 Fernet 迁移和 S-01 Sync 客户端 driver 已登记；其余 28 个生产验收 ID 尚未登记，运行时会生成 `NOT_IMPLEMENTED` 证据并退出 1。这用于阻止误报，不是这些用例的验收通过。 diff --git a/frontend/src-tauri/tests/sync_push.rs b/frontend/src-tauri/tests/sync_push.rs index a1c7e22..53daece 100644 --- a/frontend/src-tauri/tests/sync_push.rs +++ b/frontend/src-tauri/tests/sync_push.rs @@ -18,7 +18,7 @@ impl Drop for Server { } #[tokio::test] -async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicates() { +async fn s01_actual_service_preserves_offline_chains_and_response_loss_idempotency() { let root = tempfile::tempdir().unwrap(); std::fs::write(root.path().join(".opennexus-test"), b"fixture").unwrap(); let service = Path::new(env!("CARGO_MANIFEST_DIR")) @@ -100,13 +100,55 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate ws.sync_bind_empty(&endpoint, remote, "rust-fixture") .unwrap() }; + assert_eq!(workspace.lock().unwrap().pending_count().unwrap(), 20); let first = workspace .lock() .unwrap() .sync_next(&binding.id) .unwrap() .unwrap(); - assert!(client.push_one(&workspace, &binding).await.unwrap()); + // Commit the first revision, then kill the client before the response can + // acknowledge the local journal. Reopen must keep all pending operations. + std::fs::write( + root.path().join("interrupt-revision"), + b"controlled-fixture", + ) + .unwrap(); + drop(workspace); + let mut lost_response = Server( + Command::new(std::env::current_exe().unwrap()) + .args(["--ignored", "--exact", "revision_response_loss_worker"]) + .env("OPENNEXUS_SYNC_WORKER_ROOT", local.path()) + .stdin(Stdio::piped()) + .stdout(Stdio::inherit()) + .stderr(Stdio::inherit()) + .spawn() + .unwrap(), + ); + use std::io::Write; + lost_response + .0 + .stdin + .take() + .unwrap() + .write_all(session.access_token.as_bytes()) + .unwrap(); + let committed = root.path().join("revision-committed"); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + while !committed.exists() { + assert!(lost_response.0.try_wait().unwrap().is_none()); + assert!( + std::time::Instant::now() < deadline, + "revision commit timeout" + ); + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + lost_response.0.kill().unwrap(); + lost_response.0.wait().unwrap(); + std::fs::remove_file(root.path().join("interrupt-revision")).unwrap(); + std::fs::remove_file(committed).unwrap(); + let workspace = Arc::new(Mutex::new(Workspace::open(local.path()).unwrap())); + assert_eq!(workspace.lock().unwrap().pending_count().unwrap(), 20); let payload = workspace .lock() .unwrap() @@ -123,6 +165,9 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate .unwrap(); assert_eq!(replay["sequence"], 1); } + assert_eq!(workspace.lock().unwrap().pending_count().unwrap(), 20); + assert!(client.push_one(&workspace, &binding).await.unwrap()); + assert_eq!(workspace.lock().unwrap().pending_count().unwrap(), 19); for _ in 1..20 { assert!(client.push_one(&workspace, &binding).await.unwrap()); } @@ -182,6 +227,22 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate "fixture-19" ); assert_eq!(workspace_b.lock().unwrap().pending_count().unwrap(), 0); + assert_eq!( + workspace_b + .lock() + .unwrap() + .read("note.md") + .unwrap() + .entry + .hash, + workspace + .lock() + .unwrap() + .read("note.md") + .unwrap() + .entry + .hash + ); assert_eq!( workspace_b .lock() @@ -207,9 +268,20 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate { let mut ws = workspace_b.lock().unwrap(); let current = ws.read("note.md").unwrap(); - ws.write("note.md", ¤t.entry.hash, b"offline-b", "local") - .unwrap(); + let mut digest = current.entry.hash; + for index in 0..20 { + digest = ws + .write( + "note.md", + &digest, + format!("offline-b-{index}").as_bytes(), + "local", + ) + .unwrap() + .hash; + } } + assert_eq!(workspace_b.lock().unwrap().pending_count().unwrap(), 20); client.push_one(&workspace, &binding).await.unwrap(); assert_eq!( client_b.pull_page(&workspace_b, &binding_b).await.unwrap(), @@ -217,7 +289,7 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate ); assert_eq!( workspace_b.lock().unwrap().read("note.md").unwrap().content, - "offline-b" + "offline-b-19" ); let conflicts = workspace_b .lock() @@ -291,7 +363,7 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate client.pull_page(&workspace, &binding).await.unwrap(); client_b.pull_page(&workspace_b, &binding_b).await.unwrap(); let expected = if choice == "local" { - "offline-b" + "offline-b-19" } else { "next-a" }; @@ -781,7 +853,6 @@ async fn actual_service_accepts_ordered_push_and_repeat_commit_without_duplicate // Kill the actual client process after each durable 10 MiB server offset, // before its response reaches the client. The next process must query offset. use sha2::{Digest, Sha256}; - use std::io::Write; let large_remote = client .json( reqwest::Method::POST, @@ -1069,3 +1140,20 @@ async fn resumable_upload_worker() { let client = SyncClient::new(&binding.endpoint, token, true).unwrap(); client.push_one(&ws, &binding).await.unwrap(); } + +#[tokio::test] +#[ignore = "helper process killed after the server commits but before response delivery"] +async fn revision_response_loss_worker() { + let root = std::env::var("OPENNEXUS_SYNC_WORKER_ROOT").expect("controlled fixture root"); + let ws = Arc::new(Mutex::new(Workspace::open(Path::new(&root)).unwrap())); + let binding = ws.lock().unwrap().sync_binding().unwrap().unwrap(); + assert!(binding.endpoint.starts_with("http://127.0.0.1:")); + use std::io::Read; + let mut token = Zeroizing::new(String::new()); + std::io::stdin() + .take(4096) + .read_to_string(&mut token) + .unwrap(); + let client = SyncClient::new(&binding.endpoint, token, true).unwrap(); + client.push_one(&ws, &binding).await.unwrap(); +} diff --git a/scripts/acceptance_cases/s01_sync_client.py b/scripts/acceptance_cases/s01_sync_client.py new file mode 100644 index 0000000..a96cf82 --- /dev/null +++ b/scripts/acceptance_cases/s01_sync_client.py @@ -0,0 +1,96 @@ +"""S-01 two-client offline chain and idempotent response-loss driver.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import shutil +import subprocess +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +TEST = "s01_actual_service_preserves_offline_chains_and_response_loss_idempotency" + + +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 result(case_id: str, status: str, reason: str, assertions: list[dict]) -> dict: + evidence = f"cargo exact integration test {TEST}" + for assertion in assertions: + assertion.update({"status": status, "evidence": evidence}) + files = [] + for relative in ("frontend/src-tauri/tests/sync_push.rs", "server sync/tests/host_fixture.py"): + files.append({"path": relative, "sha256": sha256(ROOT / relative)}) + return { + "schema": 1, + "case_id": case_id, + "status": status, + "reason": reason, + "assertions": assertions, + "metrics": {"peak_rss_bytes": None, "max_process_count": None, "denied_access_count": None}, + "files": files, + "revisions": [ + {"scope": "isolated ephemeral Vault", "offline_chain_count": 20, "idempotent_replays": 100}, + ], + } + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--config", required=True) + parser.add_argument("--output", required=True) + args = parser.parse_args() + case_id = os.environ.get("OPENNEXUS_ACCEPTANCE_CASE_ID", "") + output = Path(args.output) + output.parent.mkdir(parents=True, exist_ok=True) + assertions = [ + {"name": "both Rust clients retain twenty offline edits without silent pending loss"}, + {"name": "server revisions form a contiguous base and sequence chain"}, + {"name": "client kill after commit preserves pending work for reopen"}, + {"name": "one hundred identical operation retries return the original revision"}, + {"name": "both clients converge on content hash and stable file identity"}, + ] + cargo = shutil.which("cargo") + if case_id != "S-01" or cargo is None: + payload = result(case_id, "FAILED", "S-01 requires the registered case ID and Cargo.", assertions) + output.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + return 1 + command = [ + cargo, + "test", + "--manifest-path", + str(ROOT / "frontend" / "src-tauri" / "Cargo.toml"), + "--locked", + "--features", + "desktop", + "--test", + "sync_push", + TEST, + "--", + "--exact", + "--nocapture", + ] + completed = subprocess.run(command, cwd=ROOT, capture_output=True, text=True, check=False) + print(completed.stdout, end="") + print(completed.stderr, end="") + passed = completed.returncode == 0 and "1 passed; 0 failed" in completed.stdout + payload = result( + case_id, + "PASSED" if passed else "FAILED", + "" if passed else "The exact Rust S-01 acceptance oracle failed.", + assertions, + ) + output.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + return 0 if passed else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/phase3_acceptance.py b/scripts/phase3_acceptance.py index 2e795b5..83e95bf 100644 --- a/scripts/phase3_acceptance.py +++ b/scripts/phase3_acceptance.py @@ -37,6 +37,11 @@ CASE_DRIVERS: dict[str, dict[str, Any]] = { "timeout_seconds": 900, "required_metrics": (), }, + "S-01": { + "driver": "scripts/acceptance_cases/s01_sync_client.py", + "timeout_seconds": 900, + "required_metrics": (), + }, } 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/tests/host_fixture.py b/server sync/tests/host_fixture.py index 6cc867e..dc5b217 100644 --- a/server sync/tests/host_fixture.py +++ b/server sync/tests/host_fixture.py @@ -29,6 +29,12 @@ def main(): for _ in range(600): if not marker.exists(): break await asyncio.sleep(.05) + if (root / 'interrupt-revision').exists() and request.method == 'POST' and request.url.path.endswith('/revisions') and response.status_code == 200: + marker = root / 'revision-committed' + marker.write_text('committed', encoding='ascii') + for _ in range(600): + if not (root / 'interrupt-revision').exists(): break + await asyncio.sleep(.05) return response sock = socket.socket() sock.bind(('127.0.0.1', 0))