perf(workspace): 流式处理重命名与删除恢复快照

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、断电或完整生产化验收通过。

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

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 crate::payloads::hash_file(&source)? != expected {
-        if hash(&content) != 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)?;
+        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 =
+        let target_conflict = target.exists()
-            target.exists() && (linked(&target)? || hash(&fs::read(&target)?) != expected);
+            && (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();