diff --git a/docs/development/OpenNexus生产化实施进度-2026-09-08.md b/docs/development/OpenNexus生产化实施进度-2026-09-08.md index 410c656..ec3a8d3 100644 --- a/docs/development/OpenNexus生产化实施进度-2026-09-08.md +++ b/docs/development/OpenNexus生产化实施进度-2026-09-08.md @@ -615,3 +615,13 @@ Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写 - 大文件测试通过，耗时 4.36 秒，日志 `.build/file-ops-stream-large.log`。该测试覆盖人工构造的持久化恢复阶段，不等于完整强杀/断电矩阵，也不是 RSS 测量；文件系统路径替换与全部持久化保证仍按原验收要求继续验证。 - 冲突保留/解决、外部变更发现和部分记录/API 仍有整块内容读取；大附件端到端内存验收和整个生产化目标仍未完成。 - desktop 全目标累计 134 通过、12 ignored（库 119、Host 8、其余集成 7），包括真实 Sync 服务及故障恢复集成，日志 `.build/file-ops-stream-full.log`。Clippy 首次指出流式改造后 workspace.rs 的 Write 导入仅供测试使用，已收为 cfg(test)；desktop 全目标 Clippy -D warnings 随后通过，日志 `.build/file-ops-stream-clippy.log`。未把这组结果当作全部大附件 RSS、断电或完整生产化验收通过。 + + +## 增量：冲突快照与解决的流式 payload + +- 从 store_payload_file 抽出共用 snapshot_file，新 sync_store_file 先有界计算源摘要，再用已有校验/流式复制/fsync/无覆盖发布流程保存快照，不为没有 operation 的冲突快照创建多余 payload 映射。冲突保留和用户决策前快照均改用此入口。 +- 冲突列表当前摘要、解决时本地 CAS 摘要改用有界 hash_file。“保留本地”“保留远端”“保留双方”的文件应用均使用 write_spooled_with_identity；本地快照长度从 spool 元数据取得并由写入入口复核，远端使用已存修订的摘要/长度。既有用户决定持久化、operation/rename/copy ID、版本复核、身份/别名及队列归档事务保留。 +- 原同步回归 16 项通过，日志 `.build/sync-conflict-stream-tests.log`。新增本地和远端各 100 MiB 的真实文件测试，三种选择分别验证过期 expected 被拒绝、冲突列表摘要、最终内容、主文件身份、copy 独立身份、重开并重复解决后待发送操作数不增加，以及旧提交仍被判定 SYNC_OPERATION_SUPERSEDED。大文件测试通过，耗时 14.35 秒，日志 `.build/sync-conflict-stream-large.log`。 +- 首轮全目标编译发现删除决策 fingerprint 仍需要生产 hash 导入，已恢复其非 test 范围，同时移除 sync_inbox 中不再使用的 fs 导入；该问题在全目标回归期间发现，未将仅 lib test 编译通过当作全部目标通过。 +- 外部变更扫描、初始绑定对账和部分逻辑记录/API 仍有整块读取；未将这些功能测试当作 RSS 基准或完整强杀/断电矩阵。大附件端到端内存与完整生产化目标继续未完成。 +- 最终 desktop 全目标累计 135 通过、12 ignored（库 120、Host 8、其余集成 7），包括实际 Sync 服务与故障恢复集成；日志 `.build/sync-conflict-stream-full.log`。desktop 全目标 Clippy -D warnings 通过，日志 `.build/sync-conflict-stream-clippy.log`。本轮没有重新执行 ignored 沙箱长时验收，全部生产化条件仍未满足。 diff --git a/frontend/src-tauri/src/payloads.rs b/frontend/src-tauri/src/payloads.rs index bf33325..c6cc029 100644 --- a/frontend/src-tauri/src/payloads.rs +++ b/frontend/src-tauri/src/payloads.rs @@ -30,6 +30,15 @@ impl Workspace { source: &Path, expected: &str, ) -> Result<()> { + let size = self.snapshot_file(source, expected)?; + self.register_payload(operation, expected, size) + } + pub(crate) fn sync_store_file(&self, source: &Path) -> Result { + let digest = hash_file(source)?; + self.snapshot_file(source, &digest)?; + Ok(digest) + } + fn snapshot_file(&self, source: &Path, expected: &str) -> Result { let size = fs::metadata(source)?.len(); if size > 100 * 1024 * 1024 { return Err(HostError::new("FILE_TOO_LARGE")); @@ -56,7 +65,7 @@ impl Workspace { #[cfg(unix)] fs::File::open(parent)?.sync_all()?; } - self.register_payload(operation, expected, size) + Ok(size) } fn register_payload(&self, operation: &str, digest: &str, size: u64) -> Result<()> { let old: Option<(String, i64)> = self diff --git a/frontend/src-tauri/src/sync_inbox.rs b/frontend/src-tauri/src/sync_inbox.rs index f83d1db..3d7c67e 100644 --- a/frontend/src-tauri/src/sync_inbox.rs +++ b/frontend/src-tauri/src/sync_inbox.rs @@ -6,7 +6,7 @@ use crate::{ use rusqlite::{params, OptionalExtension}; use serde::{Deserialize, Serialize}; use serde_json::Value; -use std::{fs, io::Write}; +use std::io::Write; use uuid::Uuid; #[derive(Clone, Serialize, Deserialize)] @@ -223,7 +223,7 @@ impl Workspace { ) -> Result<()> { let path = self.resolve(local_path)?; let digest = if path.is_file() { - self.sync_store_bytes(&fs::read(path)?)? + self.sync_store_file(&path)? } else { String::new() }; @@ -363,7 +363,7 @@ impl Workspace { let remote: Value = serde_json::from_str(&remote).map_err(|_| HostError::new("SYNC_RESPONSE_INVALID"))?; let current_path=self.path_for_id(&file_id).unwrap_or_else(|_|local_path.clone()); let source=self.resolve(¤t_path)?; - let current_hash=if source.is_file() {hash(&fs::read(source)?)} else {String::new()}; + let current_hash=if source.is_file() {crate::payloads::hash_file(&source)?} else {String::new()}; Ok(serde_json::json!({"sequence":sequence,"file_id":file_id,"local_path":local_path,"local_hash":local_hash,"current_path":current_path,"current_hash":current_hash,"remote":remote})) }).collect() } diff --git a/frontend/src-tauri/src/sync_resolution.rs b/frontend/src-tauri/src/sync_resolution.rs index 19f885f..27e64dd 100644 --- a/frontend/src-tauri/src/sync_resolution.rs +++ b/frontend/src-tauri/src/sync_resolution.rs @@ -1,7 +1,8 @@ //! User decisions are durable before changing files; journal IDs make restart replay safe. +use crate::workspace::hash; use crate::{ sync_inbox::RemoteRevision, - workspace::{hash, Entry, HostError, Result, Workspace}, + workspace::{Entry, HostError, Result, Workspace}, }; use rusqlite::{params, OptionalExtension}; use std::fs; @@ -38,7 +39,7 @@ impl Workspace { let path = self.path_for_id(&revision.file_id).unwrap_or(path); let source = self.resolve(&path)?; let current = if source.is_file() { - self.sync_store_bytes(&fs::read(source)?)? + self.sync_store_file(&source)? } else { String::new() }; @@ -110,7 +111,7 @@ impl Workspace { { let source = self.resolve(&path)?; let mut current = if source.is_file() { - hash(&fs::read(source)?) + crate::payloads::hash_file(&source)? } else { String::new() }; @@ -121,8 +122,15 @@ impl Workspace { return Err(HostError::new("REVISION_CONFLICT")); } if choice == "copy" { - let content = fs::read(self.sync_spool(&expected)?)?; - self.write_operation(&destination, "", &content, "local", ©)?; + let size = fs::metadata(self.sync_spool(&expected)?)?.len(); + self.write_spooled_with_identity( + &destination, + "", + (&expected, size), + "local", + ©, + None, + )?; } if self .entry(&path)? @@ -135,11 +143,11 @@ impl Workspace { if expected.is_empty() { self.sync_delete_intent(&revision, &operation)?; } else { - let content = fs::read(self.sync_spool(&expected)?)?; - self.write_with_identity( + let size = fs::metadata(self.sync_spool(&expected)?)?.len(); + self.write_spooled_with_identity( &path, ¤t, - &content, + (&expected, size), "local", &operation, Some(&revision.file_id), @@ -160,18 +168,14 @@ impl Workspace { "remote", )?; } - let content = fs::read( - self.sync_spool( - revision - .hash - .as_deref() - .ok_or_else(|| HostError::new("SYNC_RESPONSE_INVALID"))?, - )?, - )?; - self.write_with_identity( + let digest = revision + .hash + .as_deref() + .ok_or_else(|| HostError::new("SYNC_RESPONSE_INVALID"))?; + self.write_spooled_with_identity( &revision.path, ¤t, - &content, + (digest, revision.size as u64), "remote", &operation, Some(&revision.file_id), @@ -261,6 +265,110 @@ mod tests { ws.sync_apply_pending(binding).unwrap(); } #[test] + fn hundred_mib_conflicts_resolve_all_choices_and_reopen_without_duplicate_jobs() { + use std::io::Write; + let fixtures = tempfile::tempdir().unwrap(); + for (name, value) in [("local", 31u8), ("remote", 47u8)] { + let mut file = fs::File::create(fixtures.path().join(name)).unwrap(); + let block = vec![value; 64 * 1024]; + for _ in 0..1600 { + file.write_all(&block).unwrap(); + } + file.sync_all().unwrap(); + } + for choice in ["local", "remote", "copy"] { + let root = tempfile::tempdir().unwrap(); + let mut ws = Workspace::open(root.path()).unwrap(); + let binding = ws + .sync_bind_download("https://sync.example", "remote-vault", "account") + .unwrap(); + let mut revision = RemoteRevision { + vault_id: "remote-vault".into(), + sequence: 1, + file_id: Uuid::new_v4().to_string(), + base_revision: 0, + path: "attachments/large.bin".into(), + operation: "put".into(), + hash: Some(ws.sync_store_bytes(b"base").unwrap()), + size: 4, + operation_id: Uuid::new_v4().to_string(), + }; + receive(&mut ws, &binding.id, &revision); + let local_hash = ws.sync_store_file(&fixtures.path().join("local")).unwrap(); + let remote_hash = ws.sync_store_file(&fixtures.path().join("remote")).unwrap(); + ws.write_spooled_with_identity( + &revision.path, + revision.hash.as_deref().unwrap(), + (&local_hash, 100 * 1024 * 1024), + "local", + &Uuid::new_v4().to_string(), + Some(&revision.file_id), + ) + .unwrap(); + ws.sync_capture(&binding.id).unwrap(); + let stale = ws.sync_next(&binding.id).unwrap().unwrap(); + revision.sequence = 2; + revision.base_revision = 1; + revision.hash = Some(remote_hash.clone()); + revision.size = 100 * 1024 * 1024; + revision.operation_id = Uuid::new_v4().to_string(); + receive(&mut ws, &binding.id, &revision); + assert_eq!( + ws.sync_conflicts(&binding.id).unwrap()[0]["current_hash"], + local_hash + ); + let destination = if choice == "copy" { + "attachments/copy.bin" + } else { + "" + }; + assert_eq!( + ws.sync_resolve(&binding.id, 2, choice, destination, &hash(b"stale")) + .unwrap_err() + .code, + "REVISION_CONFLICT" + ); + ws.sync_resolve(&binding.id, 2, choice, destination, &local_hash) + .unwrap(); + drop(ws); + let mut ws = Workspace::open(root.path()).unwrap(); + ws.sync_resume_resolutions(&binding.id).unwrap(); + ws.sync_resolve(&binding.id, 2, choice, destination, &local_hash) + .unwrap(); + assert!(ws.sync_conflicts(&binding.id).unwrap().is_empty()); + assert_eq!( + crate::payloads::hash_file(&root.path().join(&revision.path)).unwrap(), + if choice == "local" { + local_hash.clone() + } else { + remote_hash + } + ); + assert_eq!( + ws.pending_count().unwrap(), + if choice == "remote" { 0 } else { 1 } + ); + assert_eq!( + ws.entry(&revision.path).unwrap().unwrap().file_id, + revision.file_id + ); + if choice == "copy" { + assert_eq!( + crate::payloads::hash_file(&root.path().join(destination)).unwrap(), + local_hash + ); + assert_ne!( + ws.entry(destination).unwrap().unwrap().file_id, + revision.file_id + ); + } + assert_eq!( + ws.sync_commit_payload(&stale).unwrap_err().code, + "SYNC_OPERATION_SUPERSEDED" + ); + } + } + #[test] fn decisions_recover_after_file_commit_without_duplicate_outbox() { for choice in ["local", "remote", "copy"] { let root = tempfile::tempdir().unwrap();