perf(sync): 流式处理冲突快照与解决载荷

2026-09-09 06:16:21 +08:00
parent bd6ccc4237
commit 6373d8ee81
4 changed files with 149 additions and 22 deletions
@@ -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 沙箱长时验收，全部生产化条件仍未满足。
@@ -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<String> {
+        let digest = hash_file(source)?;
+        self.snapshot_file(source, &digest)?;
+        Ok(digest)
+    }
+    fn snapshot_file(&self, source: &Path, expected: &str) -> Result<u64> {
        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
@@ -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(&current_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()
    }
@@ -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", &copy)?;
+                let size = fs::metadata(self.sync_spool(&expected)?)?.len();
+                self.write_spooled_with_identity(
+                    &destination,
+                    "",
+                    (&expected, size),
+                    "local",
+                    &copy,
+                    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,
                        &current,
-                        &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,
                    &current,
-                    &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();