From 54af75e90ea9ee90b06104cd7282d3ee91ec29a8 Mon Sep 17 00:00:00 2001 From: KiriAky 107 Date: Wed, 9 Sep 2026 06:10:49 +0800 Subject: [PATCH] =?UTF-8?q?perf(workspace):=20=E6=B5=81=E5=BC=8F=E5=A4=84?= =?UTF-8?q?=E7=90=86=E9=87=8D=E5=91=BD=E5=90=8D=E4=B8=8E=E5=88=A0=E9=99=A4?= =?UTF-8?q?=E6=81=A2=E5=A4=8D=E5=BF=AB=E7=85=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- frontend/src-tauri/src/payloads.rs | 34 ++++++++ frontend/src-tauri/src/workspace.rs | 131 +++++++++++++++++++++++++--- 2 files changed, 155 insertions(+), 10 deletions(-) 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();