From 919e58fc9df84654a6e9a6d64ff7211d85f21e11 Mon Sep 17 00:00:00 2001 From: Zhizhen He Date: Wed, 7 Oct 2026 12:45:26 +0800 Subject: [PATCH] Prevent sync data loss from incomplete tracked files (#725) Co-authored-by: Zhizhen He --- Cargo.lock | 1 + crates-tauri/yaak-app-client/src/sync_ext.rs | 4 +- crates/yaak-git/src/status.rs | 4 +- crates/yaak-sync/Cargo.toml | 3 + crates/yaak-sync/src/models.rs | 37 +--- crates/yaak-sync/src/sync.rs | 201 +++++++++++++++-- crates/yaak-sync/tests/sync_files.rs | 217 +++++++++++++++++++ 7 files changed, 419 insertions(+), 48 deletions(-) create mode 100644 crates/yaak-sync/tests/sync_files.rs diff --git a/Cargo.lock b/Cargo.lock index bbff0f5e6..abbb7d28a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11667,6 +11667,7 @@ dependencies = [ "serde_path_to_error", "serde_yaml", "sha1 0.11.0", + "tempfile", "thiserror 2.0.20", "tokio", "ts-rs", diff --git a/crates-tauri/yaak-app-client/src/sync_ext.rs b/crates-tauri/yaak-app-client/src/sync_ext.rs index ab26fccc0..6ba538096 100644 --- a/crates-tauri/yaak-app-client/src/sync_ext.rs +++ b/crates-tauri/yaak-app-client/src/sync_ext.rs @@ -29,7 +29,7 @@ pub(crate) async fn cmd_sync_calculate( let db = app_handle.db(); let version = app_handle.package_info().version.to_string(); let db_candidates = get_db_candidates(&db, &version, workspace_id, sync_dir)?; - let fs_candidates = get_fs_candidates(sync_dir)? + let fs_candidates = get_fs_candidates(sync_dir, &db_candidates)? .into_iter() // Only keep items in the same workspace .filter(|fs| fs.model.workspace_id() == workspace_id) @@ -39,7 +39,7 @@ pub(crate) async fn cmd_sync_calculate( pub(crate) async fn cmd_sync_calculate_fs(dir: &Path) -> Result> { let db_candidates = Vec::new(); - let fs_candidates = get_fs_candidates(dir)?; + let fs_candidates = get_fs_candidates(dir, &db_candidates)?; Ok(compute_sync_ops(db_candidates, fs_candidates)) } diff --git a/crates/yaak-git/src/status.rs b/crates/yaak-git/src/status.rs index ad3002e05..39c969952 100644 --- a/crates/yaak-git/src/status.rs +++ b/crates/yaak-git/src/status.rs @@ -135,7 +135,7 @@ pub fn git_status(dir: &Path) -> crate::error::Result { let obj = entry.to_object(&repo)?; let content = obj.as_blob().unwrap().content(); let name = Path::new(entry.name().unwrap_or_default()); - SyncModel::from_bytes(content.into(), name)?.map(|m| m.0) + SyncModel::from_bytes(content.into(), name).ok().flatten().map(|m| m.0) } Err(_) => None, }, @@ -143,7 +143,7 @@ pub fn git_status(dir: &Path) -> crate::error::Result { let next = { let full_path = repo.workdir().unwrap().join(rela_path.clone()); - SyncModel::from_file(full_path.as_path())?.map(|m| m.0) + SyncModel::from_file(full_path.as_path()).ok().flatten().map(|m| m.0) }; entries.push(GitStatusEntry { diff --git a/crates/yaak-sync/Cargo.toml b/crates/yaak-sync/Cargo.toml index c2c721800..c74ebb8d4 100644 --- a/crates/yaak-sync/Cargo.toml +++ b/crates/yaak-sync/Cargo.toml @@ -18,3 +18,6 @@ tokio = { workspace = true, features = ["fs", "sync", "macros", "rt"] } ts-rs = { workspace = true, features = ["chrono-impl", "serde-json-impl"] } yaak-models = { workspace = true } serde_path_to_error = "0.1.20" + +[dev-dependencies] +tempfile = "3" diff --git a/crates/yaak-sync/src/models.rs b/crates/yaak-sync/src/models.rs index c90e3036a..b79c6bed9 100644 --- a/crates/yaak-sync/src/models.rs +++ b/crates/yaak-sync/src/models.rs @@ -1,7 +1,7 @@ use crate::error::Error::UnknownModel; use crate::error::Result; use chrono::NaiveDateTime; -use log::{debug, warn}; +use log::debug; use serde::{Deserialize, Deserializer, Serialize}; use serde_yaml::{Mapping, Value}; use sha1::{Digest, Sha1}; @@ -98,11 +98,13 @@ fn migrate_environment(obj: &mut Mapping) { } impl SyncModel { + /// Returns `Ok(None)` for files that don't look like Yaak models, and an error for ones that + /// do but can't be parsed. pub fn from_bytes(content: Vec, file_path: &Path) -> Result> { let mut hasher = Sha1::new(); hasher.update(&content); let checksum = hex::encode(hasher.finalize()); - let content_str = String::from_utf8(content.clone()).unwrap_or_default(); + let content_str = String::from_utf8(content).unwrap_or_default(); // Check for some strings that will be in a model file for sure. If these strings // don't exist, then it's probably not a Yaak file. @@ -111,34 +113,19 @@ impl SyncModel { } let ext = file_path.extension().unwrap_or_default(); - if ext == "yml" || ext == "yaml" { - Ok(match serde_yaml::from_str::(&content_str) { - Ok(m) => Some((m, checksum)), - Err(e) => { - warn!("Error parsing {:?} {:?}", file_path.file_name(), e); - None - } - }) + let model = if ext == "yml" || ext == "yaml" { + serde_yaml::from_str::(&content_str)? } else if ext == "json" { - Ok(match serde_json::from_str::(&content_str) { - Ok(m) => Some((m, checksum)), - Err(e) => { - warn!("Error parsing {:?} {:?}", file_path.file_name(), e); - None - } - }) + serde_json::from_str::(&content_str)? } else { - Ok(None) - } + return Ok(None); + }; + + Ok(Some((model, checksum))) } pub fn from_file(file_path: &Path) -> Result> { - let content = match fs::read(file_path) { - Ok(c) => c, - Err(_) => return Ok(None), - }; - - Self::from_bytes(content, file_path) + Self::from_bytes(fs::read(file_path)?, file_path) } pub fn to_file_contents(&self, rel_path: &Path) -> Result<(Vec, String)> { diff --git a/crates/yaak-sync/src/sync.rs b/crates/yaak-sync/src/sync.rs index cc9faa1b9..b24f47728 100644 --- a/crates/yaak-sync/src/sync.rs +++ b/crates/yaak-sync/src/sync.rs @@ -1,4 +1,4 @@ -use crate::error::Result; +use crate::error::{Error, Result}; use crate::models::SyncModel; use chrono::Utc; use log::{info, warn}; @@ -7,8 +7,9 @@ use std::collections::HashMap; use std::fmt::{Display, Formatter}; use std::fs; use std::fs::File; -use std::io::Write; +use std::io::{self, Write}; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; use ts_rs::TS; use yaak_models::blob_manager::BlobManager; use yaak_models::client_db::{ClientDb, WriteDb}; @@ -178,33 +179,77 @@ pub fn get_db_candidates( Ok(candidates) } -pub fn get_fs_candidates(dir: &Path) -> Result> { +/// Read sync files, validating paths tracked by existing database models. +/// +/// Workspace sync callers must pass all candidates from [`get_db_candidates`] for +/// the same workspace and directory. An empty slice is for import discovery +/// without database sync state; it does not validate tracked model identities. +pub fn get_fs_candidates(dir: &Path, db_candidates: &[DbCandidate]) -> Result> { // Ensure the root directory exists fs::create_dir_all(dir)?; + let tracked_paths: HashMap = db_candidates + .iter() + .filter_map(|candidate| match candidate { + DbCandidate::Added(_) | DbCandidate::Deleted(_) => None, + DbCandidate::Modified(model, state) | DbCandidate::Unmodified(model, state) => { + Some((PathBuf::from(&state.rel_path), (model, state))) + } + }) + .collect(); + let mut candidates = Vec::new(); let entries = fs::read_dir(dir)?; for dir_entry in entries { - let dir_entry = match dir_entry { - Ok(v) => v, - Err(_) => continue, - }; + let dir_entry = dir_entry?; + let path = dir_entry.path(); + let rel_path = PathBuf::from(dir_entry.file_name()); + let tracked = tracked_paths.get(&rel_path); + let invalid = + |reason: String| Err(Error::InvalidSyncFile(format!("{}: {reason}", path.display()))); if !dir_entry.file_type()?.is_file() { + if tracked.is_some() { + return invalid("expected a regular sync file".into()); + } continue; }; - let path = dir_entry.path(); - let (model, checksum) = match SyncModel::from_file(&path) { - Ok(Some(m)) => m, - Ok(None) => continue, - Err(e) => { - warn!("Failed to parse sync file {e}"); - return Err(e); + // A present tracked file must not disappear from the candidates: that + // would turn a read/parse failure into a database deletion. + let (model, checksum) = match (SyncModel::from_file(&path), tracked) { + (Ok(Some(m)), _) => m, + (Ok(None), None) => continue, + (Err(e), None) => { + warn!("Skipping invalid sync file {}: {e}", path.display()); + continue; } + (Ok(None), Some(_)) => return invalid("file does not contain a Yaak model".into()), + (Err(e), Some(_)) => return invalid(e.to_string()), }; - let rel_path = Path::new(&dir_entry.file_name()).to_path_buf(); + if let Some((previous, state)) = tracked { + if model.id() != state.model_id { + return invalid(format!( + "expected model ID {}, found {}", + state.model_id, + model.id() + )); + } + // A wrong/defaulted workspace ID would be filtered out by the caller, + // making this tracked model look deleted even if its ID is intact. + if model.workspace_id() != state.workspace_id { + return invalid(format!( + "expected workspace ID {}, found {}", + state.workspace_id, + model.workspace_id() + )); + } + if std::mem::discriminant(&model) != std::mem::discriminant(*previous) { + return invalid("model type does not match the tracked model".into()); + } + } + candidates.push(FsCandidate { rel_path, model, checksum }) } @@ -391,8 +436,7 @@ pub fn apply_fs_sync_ops( let rel_path = derive_model_filename(&model); let abs_path = sync_dir.join(rel_path.clone()); let (content, checksum) = model.to_file_contents(&rel_path)?; - let mut f = File::create(&abs_path)?; - f.write_all(&content)?; + write_sync_file(&abs_path, |file| file.write_all(&content))?; SyncStateOp::Create { model_id: model.id(), checksum, rel_path } } SyncOp::FsUpdate { model, state } => { @@ -400,8 +444,7 @@ pub fn apply_fs_sync_ops( let rel_path = Path::new(&state.rel_path); let abs_path = Path::new(&state.sync_dir).join(&rel_path); let (content, checksum) = model.to_file_contents(&rel_path)?; - let mut f = File::create(&abs_path)?; - f.write_all(&content)?; + write_sync_file(&abs_path, |file| file.write_all(&content))?; SyncStateOp::Update { state: state.to_owned(), checksum, @@ -447,6 +490,39 @@ pub fn apply_fs_sync_ops( Ok(pending) } +/// Publish a complete file without exposing partial writes to sync readers. +/// +/// The temporary file's extension keeps it out of sync candidates. A crash may +/// leave it behind, but it is never read as a model. +fn write_sync_file(path: &Path, write: impl FnOnce(&mut File) -> io::Result<()>) -> Result<()> { + static COUNTER: AtomicUsize = AtomicUsize::new(0); + + // Opening the existing file for writing keeps read-only files from being replaced + let permissions = match fs::OpenOptions::new().write(true).open(path) { + Ok(file) => Some(file.metadata()?.permissions()), + Err(error) if error.kind() == io::ErrorKind::NotFound => None, + Err(error) => return Err(error.into()), + }; + let tmp_path = path.with_file_name(format!( + ".{}.{}-{}.tmp", + path.file_name().unwrap_or_default().to_string_lossy(), + std::process::id(), + COUNTER.fetch_add(1, Ordering::Relaxed), + )); + let result = File::create_new(&tmp_path).and_then(|mut file| { + if let Some(permissions) = permissions { + file.set_permissions(permissions)?; + } + write(&mut file)?; + drop(file); + fs::rename(&tmp_path, path) + }); + if result.is_err() { + let _ = fs::remove_file(&tmp_path); + } + Ok(result?) +} + impl PendingDbSyncOps { /// Upserts are collected per model type and written in one batch so /// foreign keys are satisfied. @@ -603,3 +679,90 @@ fn delete_model(db: &WriteDb, blobs: &BlobManager, model: &SyncModel) -> Result< }; Ok(()) } + +#[cfg(test)] +mod write_tests { + use super::*; + + #[test] + fn failed_partial_write_keeps_previous_file() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("request.yaml"); + fs::write(&path, b"complete previous contents").unwrap(); + let result = write_sync_file(&path, |file| { + file.write_all(b"partial new contents")?; + Err(io::Error::new(io::ErrorKind::WriteZero, "simulated write failure")) + }); + assert!( + matches!(result, Err(Error::IoError(error)) if error.kind() == io::ErrorKind::WriteZero) + ); + assert_eq!(fs::read(&path).unwrap(), b"complete previous contents"); + assert_eq!(fs::read_dir(dir.path()).unwrap().count(), 1); + } + + #[test] + fn readers_ignore_the_in_progress_file() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("request.yaml"); + let request = HttpRequest { + id: "rq_write_test".into(), + workspace_id: "wk_write_test".into(), + name: "Previous request".into(), + ..Default::default() + }; + let model = SyncModel::HttpRequest(request.clone()); + let (old_contents, checksum) = model.to_file_contents(&path).unwrap(); + fs::write(&path, &old_contents).unwrap(); + let db_candidates = vec![DbCandidate::Unmodified( + model, + SyncState { + model_id: request.id.clone(), + workspace_id: request.workspace_id.clone(), + rel_path: "request.yaml".into(), + checksum, + ..Default::default() + }, + )]; + let model = SyncModel::HttpRequest(HttpRequest { name: "New request".into(), ..request }); + let (new_contents, _) = model.to_file_contents(&path).unwrap(); + let split = new_contents.len() / 2; + write_sync_file(&path, |file| { + file.write_all(&new_contents[..split])?; + assert_eq!(fs::read(&path).unwrap(), old_contents); + let fs_candidates = get_fs_candidates(dir.path(), &db_candidates).unwrap(); + assert_eq!(fs_candidates.len(), 1); + assert!(compute_sync_ops(db_candidates.clone(), fs_candidates).is_empty()); + file.write_all(&new_contents[split..]) + }) + .unwrap(); + assert_eq!(fs::read(&path).unwrap(), new_contents); + assert_eq!(fs::read_dir(dir.path()).unwrap().count(), 1); + } + + #[cfg(unix)] + #[test] + fn writes_preserve_existing_permissions_and_new_file_defaults() { + use std::os::unix::fs::PermissionsExt; + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("request.yaml"); + fs::write(&path, b"old contents").unwrap(); + fs::set_permissions(&path, fs::Permissions::from_mode(0o640)).unwrap(); + write_sync_file(&path, |file| { + // Keep unpublished data at least as private as the original file. + assert_eq!(file.metadata()?.permissions().mode() & 0o777 & !0o640, 0); + file.write_all(b"new contents") + }) + .unwrap(); + assert_eq!(fs::metadata(&path).unwrap().permissions().mode() & 0o777, 0o640); + + let reference = dir.path().join("reference.yaml"); + File::create(&reference).unwrap(); + let new_path = dir.path().join("new.yaml"); + write_sync_file(&new_path, |file| file.write_all(b"new contents")).unwrap(); + assert_eq!( + fs::metadata(&new_path).unwrap().permissions().mode() & 0o777, + fs::metadata(&reference).unwrap().permissions().mode() & 0o777 + ); + } +} diff --git a/crates/yaak-sync/tests/sync_files.rs b/crates/yaak-sync/tests/sync_files.rs new file mode 100644 index 000000000..12e1b66e7 --- /dev/null +++ b/crates/yaak-sync/tests/sync_files.rs @@ -0,0 +1,217 @@ +use std::fs; +use std::path::PathBuf; +use tempfile::TempDir; +use yaak_models::blob_manager::BlobManager; +use yaak_models::models::{HttpRequest, SyncState, Workspace}; +use yaak_models::query_manager::QueryManager; +use yaak_sync::error::{Error, Result}; +use yaak_sync::models::SyncModel; +use yaak_sync::sync::{ + SyncOp, apply_db_sync_ops, apply_fs_sync_ops, apply_sync_state_ops, compute_sync_ops, + get_db_candidates, get_fs_candidates, +}; + +struct Fixture { + db: QueryManager, + blobs: BlobManager, + dir: TempDir, + workspace_id: String, + request: HttpRequest, + request_path: PathBuf, +} + +impl Fixture { + fn new(request_filename: &str) -> Self { + let (db, blobs, _events) = yaak_models::init_in_memory().unwrap(); + let dir = tempfile::tempdir().unwrap(); + let workspace = Workspace { id: "wk_sync_test".into(), ..Default::default() }; + let request = HttpRequest { + id: "rq_sync_test".into(), + workspace_id: workspace.id.clone(), + name: "Keep this request".into(), + ..Default::default() + }; + let request_path = dir.path().join(request_filename); + let fixture = Self { + db, + blobs, + dir, + workspace_id: workspace.id.clone(), + request: request.clone(), + request_path, + }; + for (model, path) in [ + (SyncModel::Workspace(workspace), fixture.dir.path().join("workspace.yaml")), + (SyncModel::HttpRequest(request), fixture.request_path.clone()), + ] { + let (contents, _) = model.to_file_contents(&path).unwrap(); + fs::write(path, contents).unwrap(); + } + let fs_candidates = get_fs_candidates(fixture.dir.path(), &[]).unwrap(); + fixture.apply(compute_sync_ops(Vec::new(), fs_candidates)); + assert!(fixture.calculate().unwrap().is_empty()); + fixture + } + + fn calculate(&self) -> Result> { + let db_candidates = + get_db_candidates(&self.db.connect(), "test", &self.workspace_id, self.dir.path())?; + let fs_candidates = get_fs_candidates(self.dir.path(), &db_candidates)? + .into_iter() + .filter(|candidate| candidate.model.workspace_id() == self.workspace_id) + .collect(); + Ok(compute_sync_ops(db_candidates, fs_candidates)) + } + + fn apply(&self, ops: Vec) { + let pending = apply_fs_sync_ops(&self.workspace_id, self.dir.path(), ops).unwrap(); + let workspace = &self.workspace_id; + let dir = self.dir.path(); + self.db + .with_tx::<_, Error>(|tx| { + let states = apply_db_sync_ops(tx, &self.blobs, workspace, dir, pending)?; + apply_sync_state_ops(tx, workspace, dir, states) + }) + .unwrap(); + } + + fn states(&self) -> Vec { + let mut states = self + .db + .connect() + .list_sync_states_for_workspace(&self.workspace_id, self.dir.path()) + .unwrap(); + states.sort_by(|a, b| a.id.cmp(&b.id)); + states + } + + fn db_request(&self) -> HttpRequest { + self.db.connect().get_http_request(&self.request.id).unwrap() + } + + fn edit(&self, edit: impl FnOnce(&mut serde_json::Value)) { + let mut value = serde_json::to_value(SyncModel::HttpRequest(self.request.clone())).unwrap(); + edit(&mut value); + let content = if self.request_path.extension().unwrap() == "json" { + serde_json::to_string(&value).unwrap() + } else { + serde_yaml::to_string(&value).unwrap() + }; + fs::write(&self.request_path, content).unwrap(); + } + + fn assert_invalid_file(&self) -> String { + let before_request = self.db_request(); + let before_states = serde_json::to_value(self.states()).unwrap(); + let error = self.calculate().unwrap_err(); + assert!(matches!(&error, Error::InvalidSyncFile(_)), "{error:?}"); + assert!(error.to_string().contains(self.request_path.to_str().unwrap())); + assert_eq!(self.db_request(), before_request); + assert_eq!(serde_json::to_value(self.states()).unwrap(), before_states); + error.to_string() + } +} + +#[test] +fn tracked_model_type_changes_abort_sync() { + let fixture = Fixture::new("request.yaml"); + fixture.edit(|value| { + value["model"] = "websocket_request".into(); + }); + fixture.assert_invalid_file(); +} + +#[test] +fn invalid_tracked_files_abort_without_changing_data() { + for (filename, contents) in [ + ("request.yaml", b"model: http_request\nid: [broken\n".as_slice()), + ("request.json", b"{\"model\":\"http_request\",\"id\":".as_slice()), + ("custom-request.yml", b"name: half-written\n".as_slice()), + ("custom-request.yml", b"model: http_request\n".as_slice()), + ("custom-request.yml", [0xff, 0xfe].as_slice()), + ] { + let fixture = Fixture::new(filename); + fs::write(&fixture.request_path, contents).unwrap(); + fixture.assert_invalid_file(); + assert_eq!(fs::read(&fixture.request_path).unwrap(), contents); + } +} + +#[test] +fn tracked_file_with_git_conflict_markers_aborts_sync() { + let fixture = Fixture::new("request.yaml"); + let contents = fs::read_to_string(&fixture.request_path).unwrap().replace( + "name: Keep this request\n", + "<<<<<<< HEAD\nname: Mine\n=======\nname: Theirs\n>>>>>>> origin/main\n", + ); + fs::write(&fixture.request_path, contents).unwrap(); + let error = fixture.assert_invalid_file(); + assert!(error.contains("line "), "{error}"); +} + +#[test] +fn empty_tracked_file_aborts_sync() { + let fixture = Fixture::new("request.yaml"); + fs::write(&fixture.request_path, "").unwrap(); + fixture.assert_invalid_file(); +} + +#[cfg(unix)] +#[test] +fn unreadable_tracked_file_aborts_sync() { + use std::os::unix::fs::PermissionsExt; + let fixture = Fixture::new("request.yaml"); + fs::set_permissions(&fixture.request_path, fs::Permissions::from_mode(0o000)).unwrap(); + // Root bypasses ordinary POSIX permission checks. + if fs::read(&fixture.request_path).is_ok() { + return; + } + let error = fixture.assert_invalid_file(); + assert!(error.contains("Permission denied"), "{error}"); +} + +#[test] +fn mismatched_identities_abort_sync_for_yaml_and_json() { + for filename in ["request.yaml", "request.json"] { + for field in ["id", "workspaceId"] { + let fixture = Fixture::new(filename); + fixture.edit(|value| value[field] = "different".into()); + fixture.assert_invalid_file(); + } + } +} + +#[test] +fn tracked_path_replaced_by_a_directory_aborts_sync() { + let fixture = Fixture::new("request.yaml"); + fs::remove_file(&fixture.request_path).unwrap(); + fs::create_dir(&fixture.request_path).unwrap(); + fixture.assert_invalid_file(); +} + +#[test] +fn deleting_a_tracked_file_still_deletes_the_request() { + let fixture = Fixture::new("request.yaml"); + fs::remove_file(&fixture.request_path).unwrap(); + let ops = fixture.calculate().unwrap(); + assert!( + matches!(ops.as_slice(), [SyncOp::DbDelete { model, .. }] if model.id() == fixture.request.id) + ); + fixture.apply(ops); + assert!(fixture.db.connect().get_http_request(&fixture.request.id).is_err()); +} + +#[test] +fn unrelated_files_are_still_ignored() { + let fixture = Fixture::new("request.yaml"); + for (name, contents) in [ + ("README.md", "model: http_request\nid: [broken\n"), + ("config.yaml", "model: unrelated\nid: example\n"), + ("unrelated.json", "{\"model\":\"config\",\"id\":"), + ("empty.yaml", ""), + ] { + fs::write(fixture.dir.path().join(name), contents).unwrap(); + } + fs::create_dir(fixture.dir.path().join("other-files")).unwrap(); + assert!(fixture.calculate().unwrap().is_empty()); +}