sync refactor

This commit is contained in:
nvrl committed 2026-07-22 02:22:40 +02:00
1 parent 0a1fa95eb4
commit bef39f75cf
2 files changed
+248 -106

No files matched your search

+71 -106
View File
@@ -56,8 +56,8 @@ use local_scan::open_for_write_set;
use model::{LocalItem, Outcome, PassAbort, Pending, RemoteItem};
pub(crate) use planner::base_name;
use planner::{
classification_order, conflict_path, filter_baseline, guard_local_wipe, join_rel, parent_rel,
rel_to_path, remote_sig, unchanged_remote_size,
FilePlan, classification_order, conflict_path, filter_baseline, guard_local_wipe, join_rel,
parent_rel, plan_file, rel_to_path, remote_sig, unchanged_remote_size,
};
/// What a [`Pending`] op did, so the reconcile can update shared state on the
/// engine thread (never inside a task).
@@ -666,117 +666,82 @@ impl Core {
continue;
}
// File.
// A file still open for writing is treated as unchanged on the local
// side: uploading it now would capture a torn snapshot, and the next
// pass — after the writer has closed it — will see the final content.
if l.is_some_and(|l| l.open_for_write) {
debug!(
rel,
"sync: file open for write by another process; deferring upload"
);
outcome.deferred += 1;
continue;
}
let base = baseline.get(rel);
let local_changed = l.is_some_and(|l| {
base.is_none_or(|b| b.local_mtime != l.mtime || b.local_size != l.size)
});
let remote_changed =
r.is_some_and(|r| base.is_none_or(|b| remote_sig(b) != Some((r.mtime, r.size))));
match (l, r) {
(Some(_), Some(r)) => {
// With no baseline, both sides read as "changed" and fall to the
// conflict arm — the safe default for a folder re-added over
// existing remote content.
if !local_changed && !remote_changed {
// Both untouched; baseline already matches.
} else if local_changed && !remote_changed {
batch.push(Pending::UploadRevision {
rel: rel.clone(),
uid: r.uid.clone(),
});
} else if remote_changed && !local_changed {
batch.push(Pending::Download {
rel: rel.clone(),
uid: r.uid.clone(),
mtime: r.mtime,
size: r.size,
});
} else {
batch.push(Pending::Conflict {
rel: rel.clone(),
uid: r.uid.clone(),
mtime: r.mtime,
size: r.size,
});
}
match plan_file(l, r, baseline.get(rel)) {
FilePlan::Unchanged => {}
FilePlan::Deferred => {
debug!(
rel,
"sync: file open for write by another process; deferring upload"
);
outcome.deferred += 1;
}
(Some(_), None) => {
if base.is_none() || local_changed {
// New local file, or remote deleted while local changed →
// (re)upload it as a new remote file.
match remote_dirs.get(parent_rel(rel)) {
Some(parent) => batch.push(Pending::UploadNew {
rel: rel.clone(),
parent: parent.clone(),
}),
None => {
warn!(rel, "sync: no remote parent for file; continuing");
outcome.errors += 1;
}
}
} else {
// Remote deleted, local untouched → delete local.
let _ = std::fs::remove_file(local_root.join(rel_to_path(rel)));
let _ = self.db.sync_entry_remove(folder_id, rel);
FilePlan::UploadRevision => {
let remote = r.expect("upload revision requires a remote file");
batch.push(Pending::UploadRevision {
rel: rel.clone(),
uid: remote.uid.clone(),
});
}
FilePlan::Download => {
let remote = r.expect("download requires a remote file");
batch.push(Pending::Download {
rel: rel.clone(),
uid: remote.uid.clone(),
mtime: remote.mtime,
size: remote.size,
});
}
FilePlan::Conflict => {
let remote = r.expect("conflict requires a remote file");
batch.push(Pending::Conflict {
rel: rel.clone(),
uid: remote.uid.clone(),
mtime: remote.mtime,
size: remote.size,
});
}
FilePlan::UploadNew => match remote_dirs.get(parent_rel(rel)) {
Some(parent) => batch.push(Pending::UploadNew {
rel: rel.clone(),
parent: parent.clone(),
}),
None => {
warn!(rel, "sync: no remote parent for file; continuing");
outcome.errors += 1;
}
},
FilePlan::DeleteLocal => {
let _ = std::fs::remove_file(local_root.join(rel_to_path(rel)));
let _ = self.db.sync_entry_remove(folder_id, rel);
self.log_activity(ActivityKind::Trash, base_name(rel), "removed locally", true);
outcome.deleted += 1;
}
FilePlan::DeleteRemote => {
let remote = r.expect("remote deletion requires a remote file");
if let Err(e) = self
.rt
.block_on(self.client.trash_nodes(std::slice::from_ref(&remote.uid)))
{
warn!(rel, error = %e, "sync: trash remote failed");
self.log_activity(
ActivityKind::Trash,
base_name(rel),
"removed locally",
true,
e.to_string(),
false,
);
outcome.deleted += 1;
outcome.errors += 1;
continue;
}
let _ = self.db.sync_entry_remove(folder_id, rel);
self.log_activity(
ActivityKind::Trash,
base_name(rel),
"removed on Drive",
true,
);
outcome.deleted += 1;
}
(None, Some(r)) => {
if base.is_none() || remote_changed {
// New remote file, or local deleted while remote changed →
// (re)download it.
batch.push(Pending::Download {
rel: rel.clone(),
uid: r.uid.clone(),
mtime: r.mtime,
size: r.size,
});
} else {
// Local deleted, remote untouched → delete remote.
if let Err(e) = self
.rt
.block_on(self.client.trash_nodes(std::slice::from_ref(&r.uid)))
{
warn!(rel, error = %e, "sync: trash remote failed");
self.log_activity(
ActivityKind::Trash,
base_name(rel),
e.to_string(),
false,
);
outcome.errors += 1;
continue;
}
let _ = self.db.sync_entry_remove(folder_id, rel);
self.log_activity(
ActivityKind::Trash,
base_name(rel),
"removed on Drive",
true,
);
outcome.deleted += 1;
}
}
(None, None) => {
FilePlan::ForgetBaseline => {
let _ = self.db.sync_entry_remove(folder_id, rel);
}
}
+177
View File
@@ -174,3 +174,180 @@ pub(super) fn conflict_path(path: &Path, stamp: i64) -> PathBuf {
None => PathBuf::from(name),
}
}
/// A pure decision for one file-shaped path. Transfer identities and filesystem
/// effects are attached by the executor after this classification.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum FilePlan {
Unchanged,
Deferred,
UploadNew,
UploadRevision,
Download,
Conflict,
DeleteLocal,
DeleteRemote,
ForgetBaseline,
}
pub(super) fn plan_file(
local: Option<&LocalItem>,
remote: Option<&RemoteItem>,
baseline: Option<&StoredSyncEntry>,
) -> FilePlan {
if local.is_some_and(|item| item.open_for_write) {
return FilePlan::Deferred;
}
let local_changed = local.is_some_and(|item| {
baseline.is_none_or(|base| base.local_mtime != item.mtime || base.local_size != item.size)
});
let remote_changed = remote.is_some_and(|item| {
baseline.is_none_or(|base| remote_sig(base) != Some((item.mtime, item.size)))
});
match (local, remote) {
(Some(_), Some(_)) => match (local_changed, remote_changed) {
(false, false) => FilePlan::Unchanged,
(true, false) => FilePlan::UploadRevision,
(false, true) => FilePlan::Download,
(true, true) => FilePlan::Conflict,
},
(Some(_), None) if baseline.is_none() || local_changed => FilePlan::UploadNew,
(Some(_), None) => FilePlan::DeleteLocal,
(None, Some(_)) if baseline.is_none() || remote_changed => FilePlan::Download,
(None, Some(_)) => FilePlan::DeleteRemote,
(None, None) => FilePlan::ForgetBaseline,
}
}
#[cfg(test)]
mod file_plan_tests {
use super::*;
use proton_drive_rs::proton_sdk::ids::{LinkId, VolumeId};
fn local(mtime: i64, size: i64) -> LocalItem {
LocalItem {
is_dir: false,
mtime,
size,
open_for_write: false,
}
}
fn remote(mtime: i64, size: i64) -> RemoteItem {
RemoteItem {
uid: NodeUid::new(VolumeId::from("v"), LinkId::from("l")),
is_dir: false,
mtime,
size,
}
}
fn baseline(
local_mtime: i64,
local_size: i64,
remote_mtime: i64,
remote_size: i64,
) -> StoredSyncEntry {
StoredSyncEntry {
rel_path: "file".into(),
remote_uid: Some("v~l".into()),
local_mtime,
local_size,
remote_rev: Some(remote_mtime.to_string()),
remote_hash: Some(remote_size.to_string()),
}
}
#[test]
fn file_plan_covers_every_presence_and_change_shape() {
let same_local = local(10, 20);
let changed_local = local(11, 20);
let same_remote = remote(30, 40);
let changed_remote = remote(31, 40);
let base = baseline(10, 20, 30, 40);
let cases = [
(None, None, None, FilePlan::ForgetBaseline),
(None, None, Some(&base), FilePlan::ForgetBaseline),
(Some(&same_local), None, None, FilePlan::UploadNew),
(Some(&same_local), None, Some(&base), FilePlan::DeleteLocal),
(Some(&changed_local), None, Some(&base), FilePlan::UploadNew),
(None, Some(&same_remote), None, FilePlan::Download),
(
None,
Some(&same_remote),
Some(&base),
FilePlan::DeleteRemote,
),
(None, Some(&changed_remote), Some(&base), FilePlan::Download),
(
Some(&same_local),
Some(&same_remote),
None,
FilePlan::Conflict,
),
(
Some(&same_local),
Some(&same_remote),
Some(&base),
FilePlan::Unchanged,
),
(
Some(&changed_local),
Some(&same_remote),
Some(&base),
FilePlan::UploadRevision,
),
(
Some(&same_local),
Some(&changed_remote),
Some(&base),
FilePlan::Download,
),
(
Some(&changed_local),
Some(&changed_remote),
Some(&base),
FilePlan::Conflict,
),
];
for (local, remote, baseline, expected) in cases {
assert_eq!(plan_file(local, remote, baseline), expected);
}
}
#[test]
fn an_open_writer_overrides_every_other_file_decision() {
let mut local = local(11, 20);
local.open_for_write = true;
let remote = remote(31, 40);
let base = baseline(10, 20, 30, 40);
for (remote, baseline) in [
(None, None),
(None, Some(&base)),
(Some(&remote), None),
(Some(&remote), Some(&base)),
] {
assert_eq!(
plan_file(Some(&local), remote, baseline),
FilePlan::Deferred
);
}
}
#[test]
fn a_missing_remote_signature_is_treated_as_changed() {
let local = local(10, 20);
let remote = remote(30, 40);
let mut base = baseline(10, 20, 30, 40);
base.remote_hash = None;
assert_eq!(
plan_file(Some(&local), Some(&remote), Some(&base)),
FilePlan::Download
);
}
}