diff --git a/docs/development/OpenNexus生产化实施进度-2026-09-08.md b/docs/development/OpenNexus生产化实施进度-2026-09-08.md index 5219545..410c656 100644 --- a/docs/development/OpenNexus生产化实施进度-2026-09-08.md +++ b/docs/development/OpenNexus生产化实施进度-2026-09-08.md @@ -605,3 +605,13 @@ Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写 - 新增实际文件在 open_verified 后被同长度修改的复制测试，要求 copy_verified 返回 SYNC_SPOOL_CORRUPT，调用者仅在复制校验成功时发布临时文件。该证据针对复制内容校验，不代表所有重解析点/目录替换攻击均已覆盖。 - rename/delete 的 file_ops payload 恢复、冲突保留/解决、外部变更扫描及部分 API 仍有整块读取；因此完整大附件低内存与全崩溃边界验收继续未完成。 - 最终 desktop 全目标 133 通过、12 ignored（库 118、Host 8、其余集成 7），包括真实 Sync 服务有序推送/重复提交及故障恢复集成；日志 `.build/sync-stream-journal-full.log`。desktop 全目标 Clippy -D warnings 通过，日志 `.build/sync-stream-journal-clippy.log`。ignored 长时沙箱验收本轮未重跑，完整生产化目标保持未完成。 + + +## 增量：重命名与删除的流式快照和恢复 + +- 新增 store_payload_file：以有界摘要校验源文件，通过同一已校验句柄将快照流式写入 spool 临时文件，复制时复核摘要与长度，fsync/无覆盖发布后才登记 payload 引用。已有 spool 仍做有界校验；源文件超出 100 MiB 或内容不符合预期时不接受该快照。 +- prepare_file_op 的源摘要和 payload 保存不再使用整块 Vec。apply_file_op 保持旧 inline 日志迁移兼容，随后只使用摘要/长度引用与已校验文件句柄；源/目标 CAS 检查改为有界 hash_file，目标不存在时按 64 KiB 缓冲复制并复核内容，再按原顺序发布目标、删除源文件、提交身份/观察索引/outbox/回执事务。持久化引用摘要必须与 file_ops 的预期摘要一致。 +- 原有 Workspace 测试 10 项通过，日志 `.build/file-ops-stream-tests.log`。新增实际 100 MiB 文件测试，以分块写入建立源文件，依次模拟重命名前目标不存在、目标已复制且源仍在、目标已复制且源已删除；每次重开后核验摘要、原 file_id、已提交回执及仅一条 outbox，再次重开 revision 不递增。删除场景在源已删除后从 payload 恢复 trash，最后验证准备删除后发生的外部修改保持原样并进入 conflict，不产生额外 outbox。 +- 大文件测试通过，耗时 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、断电或完整生产化验收通过。 diff --git a/frontend/src-tauri/src/payloads.rs b/frontend/src-tauri/src/payloads.rs index 83b8b22..bf33325 100644 --- a/frontend/src-tauri/src/payloads.rs +++ b/frontend/src-tauri/src/payloads.rs @@ -24,6 +24,40 @@ impl Workspace { } self.register_payload(operation, &digest, content.len() as u64) } + pub(crate) fn store_payload_file( + &self, + operation: &str, + source: &Path, + expected: &str, + ) -> Result<()> { + let size = fs::metadata(source)?.len(); + if size > 100 * 1024 * 1024 { + return Err(HostError::new("FILE_TOO_LARGE")); + } + let mut input = open_verified(source, expected, size).map_err(|error| { + if error.code == "SYNC_SPOOL_CORRUPT" { + HostError::new("REVISION_CONFLICT") + } else { + error + } + })?; + let target = self.sync_spool(expected)?; + if target.exists() { + verify(&target, expected, size)?; + } else { + let parent = target + .parent() + .ok_or_else(|| HostError::new("UNSAFE_PATH"))?; + let mut temp = tempfile::NamedTempFile::new_in(parent)?; + copy_verified(&mut input, &mut temp, expected, size)?; + temp.as_file().sync_all()?; + temp.persist_noclobber(&target) + .map_err(|_| HostError::new("SYNC_SPOOL_FAILED"))?; + #[cfg(unix)] + fs::File::open(parent)?.sync_all()?; + } + self.register_payload(operation, expected, size) + } fn register_payload(&self, operation: &str, digest: &str, size: u64) -> Result<()> { let old: Option<(String, i64)> = self .db diff --git a/frontend/src-tauri/src/workspace.rs b/frontend/src-tauri/src/workspace.rs index a76f9b6..13971ab 100644 --- a/frontend/src-tauri/src/workspace.rs +++ b/frontend/src-tauri/src/workspace.rs @@ -5,6 +5,7 @@ use rusqlite::{params, Connection, OptionalExtension}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use std::fs::{self, File, OpenOptions}; +#[cfg(test)] use std::io::Write; use std::path::{Component, Path, PathBuf}; use uuid::Uuid; @@ -872,13 +873,12 @@ impl Workspace { if !source.is_file() { return Err(HostError::new("FILE_NOT_FOUND")); } - let content = fs::read(source)?; - if hash(&content) != expected { + if crate::payloads::hash_file(&source)? != expected { return Err(HostError::new("REVISION_CONFLICT")); } self.entry(path)? .ok_or_else(|| HostError::new("FILE_NOT_FOUND"))?; - self.store_payload(id, &content)?; + self.store_payload_file(id, &source, expected)?; let tx = self.db.transaction()?; tx.execute( "INSERT INTO operations VALUES (?1,?2,'pending',NULL)", @@ -922,13 +922,22 @@ impl Workspace { )) }, )?; - let content = self.payload(id, &content)?; - self.store_payload(id, &content)?; + if self.payload_ref(id)?.is_none() { + self.store_payload(id, &content)?; + } + let (digest, size) = self + .payload_ref(id)? + .ok_or_else(|| HostError::new("SYNC_SPOOL_CORRUPT"))?; + if digest != expected || !(0..=100 * 1024 * 1024).contains(&size) { + return Err(HostError::new("SYNC_SPOOL_CORRUPT")); + } + let mut payload = + crate::payloads::open_verified(&self.sync_spool(&digest)?, &digest, size as u64)?; let source = self.resolve(&path)?; let previous = self .entry(&path)? .ok_or_else(|| HostError::new("FILE_NOT_FOUND"))?; - let source_conflict = source.exists() && hash(&fs::read(&source)?) != expected; + let source_conflict = source.exists() && crate::payloads::hash_file(&source)? != expected; let target = if kind == "rename" { self.resolve(&destination)? } else { @@ -939,8 +948,8 @@ impl Workspace { fs::create_dir_all(&trash)?; trash.join(id) }; - let target_conflict = - target.exists() && (linked(&target)? || hash(&fs::read(&target)?) != expected); + let target_conflict = target.exists() + && (linked(&target)? || crate::payloads::hash_file(&target)? != expected); if source_conflict || target_conflict { let tx = self.db.transaction()?; tx.execute("UPDATE file_ops SET state='conflict' WHERE id=?1", [id])?; @@ -951,14 +960,14 @@ impl Workspace { tx.commit()?; return Err(HostError::new("RECOVERY_CONFLICT")); } - // journal 保留完整内容，目标刷盘后才删除来源；两处崩溃均可幂等重放。 + // The durable payload remains available after removing the source. if !target.exists() { let parent = target .parent() .ok_or_else(|| HostError::new("UNSAFE_PATH"))?; fs::create_dir_all(parent)?; let mut temp = tempfile::NamedTempFile::new_in(parent)?; - temp.write_all(&content)?; + crate::payloads::copy_verified(&mut payload, &mut temp, &digest, size as u64)?; temp.as_file().sync_all()?; temp.persist_noclobber(&target) .map_err(|_| HostError::new("PATH_CONFLICT"))?; @@ -1043,6 +1052,108 @@ mod tests { assert_eq!(ws.pending_count().unwrap(), 3); } + #[test] + fn hundred_mib_file_operations_recover_each_copy_stage_without_duplicate_events() { + let dir = tempfile::tempdir().unwrap(); + let mut ws = Workspace::open(dir.path()).unwrap(); + fs::create_dir(dir.path().join("attachments")).unwrap(); + let mut path = "attachments/source.bin".to_owned(); + let mut source = File::create(dir.path().join(&path)).unwrap(); + let block = vec![23u8; 64 * 1024]; + for _ in 0..1600 { + source.write_all(&block).unwrap(); + } + source.sync_all().unwrap(); + drop(source); + let digest = crate::payloads::hash_file(&dir.path().join(&path)).unwrap(); + let seed = Uuid::new_v4().to_string(); + ws.store_payload_file(&seed, &dir.path().join(&path), &digest) + .unwrap(); + let identity = Uuid::new_v4().to_string(); + ws.write_spooled_with_identity( + &path, + &digest, + (&digest, 100 * 1024 * 1024), + "remote", + &Uuid::new_v4().to_string(), + Some(&identity), + ) + .unwrap(); + for stage in 0..3 { + let target = format!("attachments/stage-{stage}.bin"); + let operation = ws + .prepare_file_op("rename", &path, &target, &digest) + .unwrap(); + if stage >= 1 { + fs::copy(dir.path().join(&path), dir.path().join(&target)).unwrap(); + } + if stage == 2 { + fs::remove_file(dir.path().join(&path)).unwrap(); + } + drop(ws); + ws = Workspace::open(dir.path()).unwrap(); + assert!(!dir.path().join(&path).exists()); + assert_eq!( + crate::payloads::hash_file(&dir.path().join(&target)).unwrap(), + digest + ); + assert_eq!(ws.entry(&target).unwrap().unwrap().file_id, identity); + assert_eq!( + ws.operation(&operation).unwrap().unwrap()["state"], + "committed" + ); + let revision = ws.entry(&target).unwrap().unwrap().revision; + drop(ws); + ws = Workspace::open(dir.path()).unwrap(); + assert_eq!(ws.entry(&target).unwrap().unwrap().revision, revision); + assert_eq!( + ws.db + .query_row( + "SELECT COUNT(*) FROM outbox WHERE operation_id=?1", + [&operation], + |r| r.get::<_, i64>(0) + ) + .unwrap(), + 1 + ); + path = target; + } + let deletion = ws.prepare_file_op("delete", &path, "", &digest).unwrap(); + fs::remove_file(dir.path().join(&path)).unwrap(); + drop(ws); + ws = Workspace::open(dir.path()).unwrap(); + assert_eq!( + crate::payloads::hash_file(&dir.path().join(".ainote/trash").join(&deletion)).unwrap(), + digest + ); + assert!(ws.entry(&path).unwrap().unwrap().deleted); + ws.recover().unwrap(); + assert_eq!(ws.pending_count().unwrap(), 4); + let restored = "attachments/conflict.bin"; + ws.write_spooled_with_identity( + restored, + "", + (&digest, 100 * 1024 * 1024), + "remote", + &Uuid::new_v4().to_string(), + None, + ) + .unwrap(); + let conflict = ws.prepare_file_op("delete", restored, "", &digest).unwrap(); + fs::write(dir.path().join(restored), b"external edit").unwrap(); + drop(ws); + ws = Workspace::open(dir.path()).unwrap(); + assert_eq!( + fs::read(dir.path().join(restored)).unwrap(), + b"external edit" + ); + assert_eq!( + ws.operation(&conflict).unwrap().unwrap()["state"], + "conflict" + ); + assert_eq!(ws.pending_count().unwrap(), 4); + } + #[test] fn saves_conflict_and_reopen() { let dir = tempfile::tempdir().unwrap();