From b15e7a10a28ff7dae7f23acbae9bc2e3f7720109 Mon Sep 17 00:00:00 2001 From: KiriAky 107 Date: Wed, 9 Sep 2026 06:05:10 +0800 Subject: [PATCH] =?UTF-8?q?perf(sync):=20=E9=80=9A=E8=BF=87=E5=86=99?= =?UTF-8?q?=E5=85=A5=E6=97=A5=E5=BF=97=E6=81=A2=E5=A4=8D=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E4=BC=A0=E8=BE=93=E6=8C=81=E4=B9=85=E8=BD=BD=E8=8D=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../OpenNexus生产化实施进度-2026-09-08.md | 11 + frontend/src-tauri/src/payloads.rs | 109 ++++++++- frontend/src-tauri/src/sync_inbox.rs | 18 +- frontend/src-tauri/src/workspace.rs | 213 ++++++++++++++++-- 4 files changed, 321 insertions(+), 30 deletions(-) diff --git a/docs/development/OpenNexus生产化实施进度-2026-09-08.md b/docs/development/OpenNexus生产化实施进度-2026-09-08.md index 3bf8006..5219545 100644 --- a/docs/development/OpenNexus生产化实施进度-2026-09-08.md +++ b/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 长时沙箱验收本轮未重跑，完整生产化目标保持未完成。 diff --git a/frontend/src-tauri/src/payloads.rs b/frontend/src-tauri/src/payloads.rs index 571fff1..83b8b22 100644 --- a/frontend/src-tauri/src/payloads.rs +++ b/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 { + 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 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"); diff --git a/frontend/src-tauri/src/sync_inbox.rs b/frontend/src-tauri/src/sync_inbox.rs index fb7a61d..f83d1db 100644 --- a/frontend/src-tauri/src/sync_inbox.rs +++ b/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( - 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_id, Some(&revision.file_id), diff --git a/frontend/src-tauri/src/workspace.rs b/frontend/src-tauri/src/workspace.rs index fbf67c3..a76f9b6 100644 --- a/frontend/src-tauri/src/workspace.rs +++ b/frontend/src-tauri/src/workspace.rs @@ -448,23 +448,57 @@ impl Workspace { origin: &str, operation_id: &str, authorization: (Option<&str>, &dyn Fn() -> Result<()>), + ) -> Result { + 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 { + 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 { 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) { - crate::records::validate(path, content)?; - } - if content.len() > 100 * 1024 * 1024 { - return Err(HostError::new("FILE_TOO_LARGE")); - } + content.validate(self, path)?; 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 = 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)?; - match self.apply_journal(&op, &id, &path, &expected, &content, &origin) { + let result = if self.payload_ref(&op)?.is_some() { + 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();