perf(sync): 通过写入日志恢复流式传输持久载荷

docs/development/OpenNexus生产化实施进度-2026-09-08.md

@@ -594,3 +594,14 @@ Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写
 - 新增生成式 100 MiB 读取测试，检查每次请求不超过 64 KiB；增长输入精确在声明长度加 1 字节时拒绝，截断、同长度内容损坏均拒绝，空对象正常。另用实际文件检查返回句柄从头读取与同长度内容修改被拒绝。两项通过，日志 `.build/sync-spool-bounded-tests.log`；生成流测试不是操作系统 RSS 基准。
 - 尚有 sync_apply_pending、Workspace journal 恢复/应用、冲突保留和外部变更扫描等路径读取整块内容；因此大附件端到端固定内存目标仍未完成，接下来仍需将已有持久化日志/操作回执/CAS 语义保持不变地接到流式 payload 应用路径。
 - desktop 全目标累计 131 通过、12 ignored（库 116、Host 8、其他集成 7），包含真实 Sync 服务有序推送/重复提交及故障恢复集成，日志 `.build/sync-spool-bounded-full.log`；desktop 全目标 Clippy -D warnings 通过，日志 `.build/sync-spool-bounded-clippy.log`。本轮未将这组结果视为大附件端到端 RSS 或完整生产化验收通过。
+
+
+## 增量：持久化 payload 的流式日志应用与恢复
+
+- 新增 WritePayload 的内联字节/既有 spool 两种输入，write_source 保留现有操作 fingerprint、CAS、授权复核、持久化 intent 和回执事务；普通内联写入仍先持久化 payload，远端 sync_apply_pending 则直接引用已下载的摘要/长度，不再读取整个附件 Vec。
+- apply_stored_journal 从 payloads 引用打开并校验同一文件句柄，将内容以固定 64 KiB 缓冲复制到目标旁的临时文件；复制期间再次核对长度和摘要，成功后 fsync 与原子替换，再执行既有 files/sync_observed/outbox/operations/journal 事务。普通写入日志的 recover 直接走该路径，不再把 spool 读入 Vec。旧 inline journal 保持迁移兼容，先建立 payload 引用再按同一路径应用。
+- 结构化记录仍按 1 MiB 上限读取和校验，不将 JSON 规则放宽为任意文件流；文件上限保留 100 MiB。用于新写入 CAS、日志应用 CAS 及远端应用本地摘要的读取也改为有界 hash_file。路径安全仍依赖现有 resolve，未声称已解决校验与替换之间的所有文件系统竞争。
+- 真实 100 MiB 文件测试由 64 KiB 块创建 spool：验证远端身份、无远端回声 outbox、重复调用回执、同 operation 不同 payload 冲突；分别模拟目标尚未替换、目标已替换但数据库未提交两个日志恢复状态，重开后摘要正确、revision=1、每个本地 operation 只有一条 outbox，再次重开仍不重复。测试通过，日志 `.build/sync-stream-journal-large.log`，耗时 1.52 秒；这不是完整进程强杀矩阵或 RSS 基准。
+- 新增实际文件在 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 长时沙箱验收本轮未重跑，完整生产化目标保持未完成。

frontend/src-tauri/src/payloads.rs

@@ -22,6 +22,9 @@ impl Workspace {
             #[cfg(unix)]
             fs::File::open(target.parent().unwrap())?.sync_all()?;
         }
+        self.register_payload(operation, &digest, content.len() as u64)
+    }
+    fn register_payload(&self, operation: &str, digest: &str, size: u64) -> Result<()> {
         let old: Option<(String, i64)> = self
             .db
             .query_row(
@@ -30,12 +33,12 @@ impl Workspace {
                 |r| Ok((r.get(0)?, r.get(1)?)),
             )
             .optional()?;
-        if old.is_some_and(|v| v != (digest.clone(), content.len() as i64)) {
+        if old.is_some_and(|v| v != (digest.to_owned(), size as i64)) {
             return Err(HostError::new("OPERATION_PAYLOAD_CONFLICT"));
         }
         self.db.execute(
             "INSERT OR IGNORE INTO payloads VALUES (?1,?2,?3)",
-            params![operation, digest, content.len() as i64],
+            params![operation, digest, size as i64],
         )?;
         Ok(())
     }
@@ -58,6 +61,84 @@ impl Workspace {
             .optional()?)
     }
 }
+/// A new write either supplies bytes or references an existing immutable spool.
+pub(crate) enum WritePayload<'a> {
+    Inline(&'a [u8]),
+    Stored { digest: &'a str, size: u64 },
+}
+impl WritePayload<'_> {
+    pub(crate) fn size(&self) -> u64 {
+        match self {
+            Self::Inline(bytes) => bytes.len() as u64,
+            Self::Stored { size, .. } => *size,
+        }
+    }
+    pub(crate) fn digest(&self) -> String {
+        match self {
+            Self::Inline(bytes) => hash(bytes),
+            Self::Stored { digest, .. } => (*digest).to_owned(),
+        }
+    }
+    pub(crate) fn validate(&self, ws: &Workspace, path: &str) -> Result<()> {
+        if crate::records::is_record(path) && self.size() > 1024 * 1024 {
+            return Err(HostError::new("RECORD_TOO_LARGE"));
+        }
+        if self.size() > 100 * 1024 * 1024 {
+            return Err(HostError::new("FILE_TOO_LARGE"));
+        }
+        match self {
+            Self::Inline(bytes) => {
+                if crate::records::is_record(path) {
+                    crate::records::validate(path, bytes)?;
+                }
+            }
+            Self::Stored { digest, size } => {
+                let mut file = open_verified(&ws.sync_spool(digest)?, digest, *size)?;
+                validate_record(&mut file, path, *size)?;
+            }
+        }
+        Ok(())
+    }
+    pub(crate) fn store(&self, ws: &Workspace, operation: &str) -> Result<()> {
+        match self {
+            Self::Inline(bytes) => ws.store_payload(operation, bytes),
+            Self::Stored { digest, size } => {
+                verify(&ws.sync_spool(digest)?, digest, *size)?;
+                ws.register_payload(operation, digest, *size)
+            }
+        }
+    }
+}
+pub(crate) fn validate_record(file: &mut fs::File, path: &str, size: u64) -> Result<()> {
+    if crate::records::is_record(path) {
+        if size > 1024 * 1024 {
+            return Err(HostError::new("RECORD_TOO_LARGE"));
+        }
+        let mut bytes = Vec::new();
+        file.take(1024 * 1024 + 1).read_to_end(&mut bytes)?;
+        crate::records::validate(path, &bytes)?;
+        file.seek(SeekFrom::Start(0))?;
+    }
+    Ok(())
+}
+pub(crate) fn hash_file(path: &Path) -> Result<String> {
+    let mut file = fs::File::open(path)?;
+    let mut buffer = vec![0u8; VERIFY_BUFFER_BYTES];
+    let mut hasher = Sha256::new();
+    let mut total = 0u64;
+    loop {
+        let count = file.read(&mut buffer)?;
+        if count == 0 {
+            break;
+        }
+        total += count as u64;
+        if total > 100 * 1024 * 1024 {
+            return Err(HostError::new("FILE_TOO_LARGE"));
+        }
+        hasher.update(&buffer[..count]);
+    }
+    Ok(format!("{:x}", hasher.finalize()))
+}
 pub(crate) fn verify(path: &Path, digest: &str, size: u64) -> Result<()> {
     open_verified(path, digest, size).map(drop)
 }
@@ -85,6 +166,14 @@ pub(crate) fn open_verified(path: &Path, digest: &str, size: u64) -> Result<fs::
 }
 const VERIFY_BUFFER_BYTES: usize = 64 * 1024;
 fn verify_reader(stream: &mut impl Read, digest: &str, size: u64) -> Result<()> {
+    copy_verified(stream, &mut std::io::sink(), digest, size)
+}
+pub(crate) fn copy_verified(
+    stream: &mut impl Read,
+    target: &mut impl Write,
+    digest: &str,
+    size: u64,
+) -> Result<()> {
     let mut hasher = Sha256::new();
     let mut buffer = vec![0; VERIFY_BUFFER_BYTES];
     let mut length = 0u64;
@@ -104,6 +193,7 @@ fn verify_reader(stream: &mut impl Read, digest: &str, size: u64) -> Result<()>
             return Err(HostError::new("SYNC_SPOOL_CORRUPT"));
         }
         hasher.update(&buffer[..count]);
+        target.write_all(&buffer[..count])?;
     }
     if length != size || format!("{:x}", hasher.finalize()) != digest {
         return Err(HostError::new("SYNC_SPOOL_CORRUPT"));
@@ -173,6 +263,21 @@ mod tests {
         verify_reader(&mut &[][..], &hash(&[]), 0).unwrap();
     }
     #[test]
+    fn copying_rechecks_source_changed_after_initial_verification() {
+        let temp = tempfile::tempdir().unwrap();
+        let path = temp.path().join("source");
+        fs::write(&path, b"original").unwrap();
+        let mut source = open_verified(&path, &hash(b"original"), 8).unwrap();
+        fs::write(&path, b"modified").unwrap();
+        let mut destination = tempfile::NamedTempFile::new_in(temp.path()).unwrap();
+        assert_eq!(
+            copy_verified(&mut source, &mut destination, &hash(b"original"), 8)
+                .unwrap_err()
+                .code,
+            "SYNC_SPOOL_CORRUPT"
+        );
+    }
+    #[test]
     fn verified_file_is_rewound_and_corruption_is_rejected() {
         let temp = tempfile::tempdir().unwrap();
         let path = temp.path().join("payload");

frontend/src-tauri/src/sync_inbox.rs

@@ -290,7 +290,7 @@ impl Workspace {
         let path = local_path.as_deref().unwrap_or(&revision.path);
         let local = self.resolve(path)?;
         let current = if local.is_file() {
-            hash(&fs::read(&local)?)
+            crate::payloads::hash_file(&local)?
         } else {
             String::new()
         };
@@ -329,18 +329,14 @@ impl Workspace {
                     "remote",
                 )?;
             }
-            let content = fs::read(
+            let digest = revision
-                self.sync_spool(
+                .hash
-                    revision
+                .as_deref()
-                        .hash
+                .ok_or_else(|| HostError::new("SYNC_RESPONSE_INVALID"))?;
-                        .as_deref()
+            self.write_spooled_with_identity(
-                        .ok_or_else(|| HostError::new("SYNC_RESPONSE_INVALID"))?,
-                )?,
-            )?;
-            self.write_with_identity(
                 &revision.path,
                 &current,
-                &content,
+                (digest, revision.size as u64),
                 "remote",
                 &operation_id,
                 Some(&revision.file_id),

frontend/src-tauri/src/workspace.rs

@@ -448,23 +448,57 @@ impl Workspace {
         origin: &str,
         operation_id: &str,
         authorization: (Option<&str>, &dyn Fn() -> Result<()>),
+    ) -> Result<Entry> {
+        self.write_source(
+            path,
+            expected,
+            crate::payloads::WritePayload::Inline(content),
+            origin,
+            operation_id,
+            authorization,
+        )
+    }
+    pub(crate) fn write_spooled_with_identity(
+        &mut self,
+        path: &str,
+        expected: &str,
+        payload: (&str, u64),
+        origin: &str,
+        operation_id: &str,
+        identity: Option<&str>,
+    ) -> Result<Entry> {
+        self.write_source(
+            path,
+            expected,
+            crate::payloads::WritePayload::Stored {
+                digest: payload.0,
+                size: payload.1,
+            },
+            origin,
+            operation_id,
+            (identity, &|| Ok(())),
+        )
+    }
+    fn write_source(
+        &mut self,
+        path: &str,
+        expected: &str,
+        content: crate::payloads::WritePayload<'_>,
+        origin: &str,
+        operation_id: &str,
+        authorization: (Option<&str>, &dyn Fn() -> Result<()>),
     ) -> Result<Entry> {
         let (identity, authorize) = authorization;
         authorize()?;
         if Uuid::parse_str(operation_id).is_err() {
             return Err(HostError::new("OPERATION_ID_INVALID"));
         }
-        if crate::records::is_record(path) {
+        content.validate(self, path)?;
-            crate::records::validate(path, content)?;
-        }
-        if content.len() > 100 * 1024 * 1024 {
-            return Err(HostError::new("FILE_TOO_LARGE"));
-        }
         if origin != "local" && origin != "remote" {
             return Err(HostError::new("INVALID_ORIGIN"));
         }
         let fingerprint = hash(
-            &serde_json::to_vec(&(path, expected, hash(content), origin))
+            &serde_json::to_vec(&(path, expected, content.digest(), origin))
                 .map_err(|_| HostError::new("INVALID_OPERATION"))?,
         );
         let previous: Option<String> = self
@@ -491,7 +525,7 @@ impl Workspace {
         }
         let target = self.resolve(path)?;
         let current = if target.exists() {
-            hash(&fs::read(&target)?)
+            crate::payloads::hash_file(&target)?
         } else {
             String::new()
         };
@@ -544,7 +578,7 @@ impl Workspace {
             },
             |entry| entry.file_id,
         );
-        self.store_payload(operation_id, content)?;
+        content.store(self, operation_id)?;
         let tx = self.db.transaction()?;
         tx.execute(
             "INSERT INTO operations VALUES (?1,?2,'pending',NULL)",
@@ -565,7 +599,7 @@ impl Workspace {
         // outbox entry is published. An unreferenced payload is never replayed.
         authorize()?;
         tx.commit()?;
-        self.apply_journal(operation_id, &file_id, path, expected, content, origin)?;
+        self.apply_stored_journal(operation_id, &file_id, path, expected, origin)?;
         self.entry(path)?
             .ok_or_else(|| HostError::new("FILE_NOT_FOUND"))
     }
@@ -582,10 +616,29 @@ impl Workspace {
         if crate::records::is_record(path) {
             crate::records::validate(path, content)?;
         }
+        self.store_payload(operation_id, content)?;
+        self.apply_stored_journal(operation_id, file_id, path, expected, origin)
+    }
+    fn apply_stored_journal(
+        &mut self,
+        operation_id: &str,
+        file_id: &str,
+        path: &str,
+        expected: &str,
+        origin: &str,
+    ) -> Result<()> {
+        let (digest, size) = self
+            .payload_ref(operation_id)?
+            .ok_or_else(|| HostError::new("SYNC_SPOOL_CORRUPT"))?;
+        if !(0..=100 * 1024 * 1024).contains(&size) {
+            return Err(HostError::new("SYNC_SPOOL_CORRUPT"));
+        }
+        let mut source =
+            crate::payloads::open_verified(&self.sync_spool(&digest)?, &digest, size as u64)?;
+        crate::payloads::validate_record(&mut source, path, size as u64)?;
         let target = self.resolve(path)?;
-        let digest = hash(content);
         let current = if target.exists() {
-            hash(&fs::read(&target)?)
+            crate::payloads::hash_file(&target)?
         } else {
             String::new()
         };
@@ -608,15 +661,13 @@ impl Workspace {
                 .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 source, &mut temp, &digest, size as u64)?;
             temp.as_file().sync_all()?;
             temp.persist(&target)
                 .map_err(|_| HostError::new("ATOMIC_REPLACE_FAILED"))?;
             #[cfg(unix)]
             File::open(parent)?.sync_all()?;
         }
-        // Upgrade legacy inline journal payloads before publishing an outbox reference.
-        self.store_payload(operation_id, content)?;
         // 文件成功但 DB 未提交时，重启凭 journal 补齐同一 operation_id，避免丢 outbox。
         let tx = self.db.transaction()?;
         tx.execute("INSERT INTO files VALUES (?1,?2,?3,1,0) ON CONFLICT(path) DO UPDATE SET hash=excluded.hash,revision=files.revision+1,deleted=0", params![file_id,path,digest])?;
@@ -681,8 +732,12 @@ impl Workspace {
             result
         };
         for (op, id, path, expected, content, origin) in pending {
-            let content = self.payload(&op, &content)?;
+            let result = if self.payload_ref(&op)?.is_some() {
-            match self.apply_journal(&op, &id, &path, &expected, &content, &origin) {
+                self.apply_stored_journal(&op, &id, &path, &expected, &origin)
+            } else {
+                self.apply_journal(&op, &id, &path, &expected, &content, &origin)
+            };
+            match result {
                 Err(e) if e.code == "RECOVERY_CONFLICT" => {}
                 result => result?,
             }
@@ -1043,6 +1098,130 @@ mod tests {
         );
     }
 
+    #[test]
+    fn hundred_mib_spooled_writes_and_journal_recovery_keep_receipts_exactly_once() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut ws = Workspace::open(dir.path()).unwrap();
+        let block = vec![19u8; 64 * 1024];
+        let size = 100 * 1024 * 1024u64;
+        let mut hasher = Sha256::new();
+        for _ in 0..size / block.len() as u64 {
+            hasher.update(&block);
+        }
+        let digest = format!("{:x}", hasher.finalize());
+        let spool = ws.sync_spool(&digest).unwrap();
+        let mut file = File::create(&spool).unwrap();
+        for _ in 0..size / block.len() as u64 {
+            file.write_all(&block).unwrap();
+        }
+        file.sync_all().unwrap();
+        drop(file);
+        let operation = Uuid::new_v4().to_string();
+        let identity = Uuid::new_v4().to_string();
+        let first = ws
+            .write_spooled_with_identity(
+                "attachments/direct.bin",
+                "",
+                (&digest, size),
+                "remote",
+                &operation,
+                Some(&identity),
+            )
+            .unwrap();
+        assert_eq!(first.file_id, identity);
+        assert_eq!(ws.pending_count().unwrap(), 0);
+        assert_eq!(
+            ws.write_spooled_with_identity(
+                "attachments/direct.bin",
+                "",
+                (&digest, size),
+                "remote",
+                &operation,
+                Some(&identity)
+            )
+            .unwrap()
+            .revision,
+            first.revision
+        );
+        assert_eq!(
+            crate::payloads::hash_file(&dir.path().join("attachments/direct.bin")).unwrap(),
+            digest
+        );
+        let other = ws.sync_store_bytes(b"different payload").unwrap();
+        assert_eq!(
+            ws.write_spooled_with_identity(
+                "attachments/direct.bin",
+                "",
+                (&other, 17),
+                "remote",
+                &operation,
+                Some(&identity)
+            )
+            .unwrap_err()
+            .code,
+            "OPERATION_PAYLOAD_CONFLICT"
+        );
+        for already_replaced in [false, true] {
+            let operation = Uuid::new_v4().to_string();
+            let identity = Uuid::new_v4().to_string();
+            let path = format!("attachments/recovery-{already_replaced}.bin");
+            let fingerprint = hash(&serde_json::to_vec(&(&path, "", &digest, "local")).unwrap());
+            ws.db
+                .execute(
+                    "INSERT INTO payloads VALUES (?1,?2,?3)",
+                    params![operation, digest, size],
+                )
+                .unwrap();
+            ws.db
+                .execute(
+                    "INSERT INTO operations VALUES (?1,?2,'pending',NULL)",
+                    params![operation, fingerprint],
+                )
+                .unwrap();
+            ws.db
+                .execute(
+                    "INSERT INTO journal VALUES (?1,?2,?3,'',?4,'local','pending')",
+                    params![operation, identity, path, b"".as_slice()],
+                )
+                .unwrap();
+            if already_replaced {
+                fs::copy(&spool, dir.path().join(&path)).unwrap();
+            }
+            drop(ws);
+            ws = Workspace::open(dir.path()).unwrap();
+            assert_eq!(
+                ws.operation(&operation).unwrap().unwrap()["state"],
+                "committed"
+            );
+            assert_eq!(ws.entry(&path).unwrap().unwrap().revision, 1);
+            assert_eq!(
+                crate::payloads::hash_file(&dir.path().join(&path)).unwrap(),
+                digest
+            );
+            let count: i64 = ws
+                .db
+                .query_row(
+                    "SELECT COUNT(*) FROM outbox WHERE operation_id=?1",
+                    [&operation],
+                    |row| row.get(0),
+                )
+                .unwrap();
+            assert_eq!(count, 1);
+            drop(ws);
+            ws = Workspace::open(dir.path()).unwrap();
+            assert_eq!(ws.entry(&path).unwrap().unwrap().revision, 1);
+            let count: i64 = ws
+                .db
+                .query_row(
+                    "SELECT COUNT(*) FROM outbox WHERE operation_id=?1",
+                    [&operation],
+                    |row| row.get(0),
+                )
+                .unwrap();
+            assert_eq!(count, 1);
+        }
+    }
+
     #[test]
     fn schema_upgrade_preserves_a_readable_previous_database() {
         let dir = tempfile::tempdir().unwrap();