Prevent sync data loss from incomplete tracked files (#725)

Co-authored-by: Zhizhen He <hezhizhen.yi@gmail.com>
This commit is contained in:
Zhizhen He authored and GitHub committed 2026-10-06 21:45:26 -07:00
1 parent b4417add33
commit 919e58fc9d
7 files changed
+419 -48

No files matched your search

Generated
+1
View File
@@ -11667,6 +11667,7 @@ dependencies = [
"serde_path_to_error",
"serde_yaml",
"sha1 0.11.0",
"tempfile",
"thiserror 2.0.20",
"tokio",
"ts-rs",
+2 -2
View File
@@ -29,7 +29,7 @@ pub(crate) async fn cmd_sync_calculate<R: Runtime>(
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<R: Runtime>(
pub(crate) async fn cmd_sync_calculate_fs(dir: &Path) -> Result<Vec<SyncOp>> {
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))
}
+2 -2
View File
@@ -135,7 +135,7 @@ pub fn git_status(dir: &Path) -> crate::error::Result<GitStatusSummary> {
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<GitStatusSummary> {
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 {
+3
View File
@@ -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"
+12 -25
View File
@@ -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<u8>, file_path: &Path) -> Result<Option<(SyncModel, String)>> {
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::<SyncModel>(&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::<SyncModel>(&content_str)?
} else if ext == "json" {
Ok(match serde_json::from_str::<SyncModel>(&content_str) {
Ok(m) => Some((m, checksum)),
Err(e) => {
warn!("Error parsing {:?} {:?}", file_path.file_name(), e);
None
}
})
serde_json::from_str::<SyncModel>(&content_str)?
} else {
Ok(None)
}
return Ok(None);
};
Ok(Some((model, checksum)))
}
pub fn from_file(file_path: &Path) -> Result<Option<(SyncModel, String)>> {
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<u8>, String)> {
+182 -19
View File
@@ -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<Vec<FsCandidate>> {
/// 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<Vec<FsCandidate>> {
// Ensure the root directory exists
fs::create_dir_all(dir)?;
let tracked_paths: HashMap<PathBuf, (&SyncModel, &SyncState)> = 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
);
}
}
+217
View File
@@ -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<Vec<SyncOp>> {
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<SyncOp>) {
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<SyncState> {
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());
}