fix(fuse,core): key conflict detection on revision id, stop transient-name seals (B69, B70)

B69 — spurious `(sync-conflict)` copies came from `(mtime, size)` being used as a
revision identity. The server re-stamps a sealed revision with its own commit
time, so the drain diverted a write into a conflict copy of its own base; and
`plan_file` classified a no-baseline (local, remote) pair as `Conflict`, forking
every pre-existing file on first reconcile.

- `Baseline` carries `revision_id` (from the SDK's new `active_revision_id`),
  captured at write-open and preserved across supersede/rebaseline.
  `revision_changed` trusts the id and only falls back to `(mtime, size)` for
  old sidecars.
- Planner gains `FilePlan::AdoptBaseline`: equal-size no-baseline pairs are
  recorded, not conflicted. Divergent sizes stay a real conflict.
- New `sweep.rs` (`pdfs-conflict-sweep`, 5-min cadence) removes conflict copies
  proven identical by size + `content_sha1` and surfaces divergent ones once as
  an `ActivityKind::Conflict` entry. Never removes what it cannot prove.

B70 — browsers downloading into an on-demand mount had their in-flight temps
sealed as canonical revisions, exiling the finished file to a conflict copy.
`syncignore::is_transient_name` now recognises download/scratch temps, and the
on-demand `release`/`rename` paths hold them (`set_create_hold`) instead of
sealing until the name is finalized.

Workspace deps bumped to proton-sdk/proton-drive-rs 0.2.2 (registry, no patch).

Live validation against a real account still pending for both.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01V8P42asymU158CPnhjRwri
This commit is contained in:
nvrlandClaude Opus 5 committed 2026-07-25 15:24:31 +02:00
1 parent 6df757006a
commit bca0fe34bc
19 files changed
+978 -115

No files matched your search

+1
View File
@@ -1002,6 +1002,7 @@ fn activity_verb(kind: pdfs_core::control::ActivityKind) -> &'static str {
Share => "share",
PublicLink => "link",
Unshare => "unshare",
Conflict => "conflict",
}
}
+12 -3
View File
@@ -150,15 +150,23 @@ pub struct StagedWrite {
pub based_on: Option<Baseline>,
}
/// The identity of a remote revision, as far as we can observe it: a file whose
/// size and mtime both still match the ones a queued change was made against has
/// not been rewritten by anyone else in the meantime.
/// The identity of a remote revision, as far as we can observe it.
///
/// `revision_id` is the authoritative identity when present: the server assigns a
/// fresh id to every sealed revision, so an unchanged id proves no one rewrote
/// the file — even when the server re-stamped its `mtime` on the *same* revision,
/// which is the drift that used to cut spurious conflict copies (B16/B25).
/// `mtime`/`size` remain the fallback for a baseline recorded before the id was
/// observable, or a node read through a surface that does not carry it.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
pub struct Baseline {
pub mtime: i64,
pub size: u64,
#[serde(default)]
pub hash: Option<String>,
/// Server revision id this change was made against (`None` if unknown).
#[serde(default)]
pub revision_id: Option<String>,
}
/// Content cache rooted at a directory, with a sibling pin-registry file.
@@ -1873,6 +1881,7 @@ mod tests {
mtime: 100,
size: 40,
hash: None,
revision_id: None,
}),
};
let staged = c.stage_write(&meta, &scratch).unwrap();
+5
View File
@@ -573,6 +573,11 @@ pub enum ActivityKind {
Share,
PublicLink,
Unshare,
/// A `(sync-conflict …)` copy that needs the user's attention: it diverges
/// from the live file (or has no sibling), so the auto-sweep left it in place
/// rather than removing it. An auto-*removed* identical copy is logged as a
/// [`ActivityKind::Trash`] instead, since that is what happened to it.
Conflict,
}
/// One line in the daemon's activity log (in [`Response::Activity`]). Newest
+1 -1
View File
@@ -42,7 +42,7 @@ pub use local::LocalFileHit;
pub use maintenance::{DbStats, VacuumOutcome};
pub use nodes::{SearchHit, StoredNode};
pub use ops::{
AttachedBlob, LOCAL_VOLUME, OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH,
AttachedBlob, LOCAL_VOLUME, OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PARK_UNTIL,
PendingCounts, PendingOp, op_supersedes,
};
pub use photos::{StoredPhoto, THUMB_HAVE, THUMB_NONE, THUMB_UNKNOWN};
+38 -2
View File
@@ -63,6 +63,16 @@ pub fn op_supersedes(kind: &str) -> bool {
/// drain replaces it with the uid the server assigns.
pub const LOCAL_VOLUME: &str = "local";
/// A `next_attempt_at` far enough in the future that [`Db::next_due_op`] never
/// selects the op (ms since epoch, ≈ year 2223). Used to *park* a queued create
/// for a transient file — a browser's in-flight `*.crdownload`/`*.part`, an
/// editor's `*.swp` — so its bytes never upload while it wears that name. A
/// rename to the finished name un-parks it (see [`Db::set_create_hold`]), which
/// is the only moment the completed file is meant to reach Drive. Without the
/// park, every partial and every abandoned temp would upload and, on a
/// stall+resume, fork a `(sync-conflict)` copy (docs/BUGS.md B70).
pub const PARK_UNTIL: i64 = 8_000_000_000_000;
/// A mutation that has been accepted locally but not yet performed against the
/// API — the durable half of the write-back queue (offline.md Phase 3).
///
@@ -183,11 +193,18 @@ impl Db {
let Some((id, superseded)) = existing else {
return Ok(None);
};
// Clearing the backoff lets the fresh bytes be tried promptly — except
// when the create is *parked* (a transient file, next_attempt_at at the
// PARK_UNTIL sentinel). Attaching a new revision to a still-growing
// download must not wake it; only the finalize rename does. So preserve a
// park and reset an ordinary backoff.
conn.execute(
"UPDATE pending_op
SET blob_path = ?2, meta_json = ?3, attempts = 0, next_attempt_at = 0
SET blob_path = ?2, meta_json = ?3, attempts = 0,
next_attempt_at = CASE WHEN next_attempt_at >= ?4
THEN next_attempt_at ELSE 0 END
WHERE id = ?1",
params![id, blob_path, meta_json],
params![id, blob_path, meta_json, PARK_UNTIL],
)?;
Ok(Some(AttachedBlob { id, superseded }))
}
@@ -233,6 +250,25 @@ impl Db {
Ok(n > 0)
}
/// Park or un-park a queued create/mkdir for `uid`, returning whether a row
/// was touched.
///
/// Parking sets `next_attempt_at` to [`PARK_UNTIL`] so [`Db::next_due_op`]
/// skips it indefinitely; un-parking sets it due now. Used to keep a
/// transient file's bytes off Drive until it is renamed to its finished name
/// (docs/BUGS.md B70). Un-parking a create that was never parked is harmless:
/// it just makes the create due immediately, which for a normal local node it
/// already is.
pub fn set_create_hold(&self, uid: &str, held: bool) -> Result<bool> {
let conn = self.conn.lock();
let n = conn.execute(
"UPDATE pending_op SET next_attempt_at = ?2
WHERE uid = ?1 AND kind IN (?3, ?4)",
params![uid, if held { PARK_UNTIL } else { 0 }, OP_CREATE, OP_MKDIR],
)?;
Ok(n > 0)
}
/// Check if a queued create or mkdir op exists for `uid`.
pub fn has_create_op(&self, uid: &str) -> Result<bool> {
let conn = self.conn.lock();
+54
View File
@@ -1615,6 +1615,58 @@ fn next_due_op_skips_ops_blocked_on_a_local_parent() {
assert_eq!(db.next_due_op(10).unwrap().unwrap().uid, "vol~link3");
}
/// B70: a transient file's create is parked so its bytes never upload while it
/// wears that name, the park survives a write attaching its blob, and the
/// finalize rename (un-park) is what finally makes the completed file due.
#[test]
fn a_parked_transient_create_stays_off_the_drain_until_finalized() {
let db = Db::open_in_memory().unwrap();
let node = uid("dl").to_string();
let parent = uid("root").to_string();
db.enqueue_op(&PendingOp {
id: 0,
kind: OP_CREATE.to_string(),
uid: node.clone(),
parent_uid: Some(parent),
name: Some("movie.mkv.part".to_string()),
blob_path: None,
meta_json: None,
created_at: 1,
attempts: 0,
last_error: None,
next_attempt_at: PARK_UNTIL,
})
.unwrap();
// Parked: never selected, however far the clock advances.
assert!(
db.next_due_op(PARK_UNTIL - 1).unwrap().is_none(),
"a parked create is never due"
);
// A write attaches its blob to the create but must not wake it.
let attached = db
.attach_blob_to_create(&node, "/staging/blob1", "{}")
.unwrap();
assert!(attached.is_some(), "the blob attached to the create");
assert!(
db.next_due_op(PARK_UNTIL - 1).unwrap().is_none(),
"still parked after a write attaches"
);
// Finalize: un-park, and the completed file is due to upload with its bytes.
assert!(
db.set_create_hold(&node, false).unwrap(),
"a create row was un-parked"
);
let due = db
.next_due_op(1_000)
.unwrap()
.expect("the un-parked create is due");
assert_eq!(due.uid, node);
assert_eq!(due.blob_path.as_deref(), Some("/staging/blob1"));
}
/// Backoff still gates: nothing is returned before an op is due.
#[test]
fn next_due_op_respects_backoff() {
@@ -1806,6 +1858,8 @@ fn test_db_has_children_and_trashed_filtering() {
media_type: "text/plain".into(),
total_size_on_storage: 10,
active_revision_state: None,
active_revision_id: None,
content_sha1: None,
claimed_size: Some(10),
claimed_modification_time: None,
},
+72
View File
@@ -45,6 +45,45 @@ pub const DEFAULT_IGNORE_PATTERNS: &[&str] = &[
"Thumbs.db",
];
/// Suffixes of files that are, by convention, incomplete or throwaway while they
/// wear the name: a download in flight, an editor's swap or backup, an atomic
/// write's scratch copy. On an on-demand mount these should never seal a remote
/// revision — the app renames the finished file to its real name, and only that
/// belongs on Drive. See [`is_transient_name`] and docs/BUGS.md B70.
const TRANSIENT_SUFFIXES: &[&str] = &[
".crdownload", // Chromium/Brave/Edge
".part", // Firefox, wget, aria2c
".partial", // some downloaders / older IE
".download", // Safari / macOS
".tmp", // generic scratch
".temp", // generic scratch
".swp", // vim swap
".swx", // vim swap (secondary)
];
/// Whether `name` is a transient scratch/in-flight file whose bytes should not
/// be uploaded on an on-demand mount until it is renamed to its finished name.
///
/// Recognises browser download temps, editor swap/backup files, and lock files.
/// Case-insensitive on the suffix so `FOO.CRDOWNLOAD` counts too. This is the
/// on-demand-write counterpart to [`DEFAULT_IGNORE_PATTERNS`] (which only covers
/// mirror-folder reconcile); keep the two conceptually in step.
pub fn is_transient_name(name: &str) -> bool {
let lower = name.to_ascii_lowercase();
if TRANSIENT_SUFFIXES.iter().any(|s| lower.ends_with(s)) {
return true;
}
// Emacs/gedit/kate backup copies.
if name.ends_with('~') {
return true;
}
// LibreOffice lock files (`.~lock.<doc>#`) and MS Office temps (`~$doc`).
if name.starts_with(".~lock.") || name.starts_with("~$") {
return true;
}
false
}
/// Filenames accepted as a folder's ignore file, in precedence order.
const IGNORE_FILE_NAMES: &[&str] = &[".pdfsignore", ".protonignore"];
@@ -272,4 +311,37 @@ mod tests {
// have nothing to reconcile against.
assert!(!rules.is_ignored("", true));
}
#[test]
fn transient_download_and_editor_names_are_recognised() {
for name in [
"Unconfirmed 540477.crdownload",
"movie.mkv.part",
"big.iso.partial",
"photo.jpg.download",
"export.xml.tmp",
"notes.TEMP",
".notes.txt.swp",
"doc.txt~",
".~lock.report.odt#",
"~$budget.xlsx",
"FOO.CRDOWNLOAD",
] {
assert!(is_transient_name(name), "{name} should be transient");
}
}
#[test]
fn finished_names_are_not_transient() {
for name in [
"teamspeak.tar.gz",
"nils.zip",
"report.xml",
"archive.partly.zip", // ".partly" is not ".part"
"template.dotx",
"readme",
] {
assert!(!is_transient_name(name), "{name} should not be transient");
}
}
}
+179 -10
View File
@@ -36,8 +36,8 @@ use super::state::Intervals;
use super::transfers::CountingReader;
use super::{
Core, DRAIN_BACKOFF_MAX, DRAIN_BACKOFF_MIN, DRAIN_IDLE_POLL, ROOT_INO, conflict_name,
is_already_exists, is_gone, is_local_uid_str, media_type_for, node_size, now_millis, now_secs,
parse_node_uid,
is_already_exists, is_gone, is_local_uid_str, media_type_for, node_revision_id, node_size,
now_millis, now_secs, parse_node_uid,
};
impl Core {
@@ -252,6 +252,7 @@ impl Core {
Err(e) => return Err(e.into()),
}
self.db.delete_op(op.id)?;
self.own_sealed_revs.lock().remove(&uid);
self.invalidate_trash();
self.log_activity(ActivityKind::Trash, &name, "trashed", true);
info!(%uid, name, "pending trash landed");
@@ -503,15 +504,23 @@ impl Core {
if node.trashed {
return Ok(Some("the file was trashed remotely".into()));
}
let (mtime, size) = (node.modification_time, node_size(&node));
if mtime != base.mtime || size != base.size {
return Ok(Some(format!(
"the remote revision changed under the queued write \
(expected {} bytes at mtime {}, found {size} at {mtime})",
base.size, base.mtime
)));
let Some(reason) = revision_changed(base, &node) else {
return Ok(None);
};
// The remote moved on from the baseline — normally a conflict. But if it
// sits at a revision *this daemon* sealed, no other device changed the
// file: it is a single-writer stall→resume (a browser that closed and
// reopened its download fd; B70 layer B). Chain onto it — supersede
// instead of forking — so the finished bytes win the name rather than
// being exiled to a `(sync-conflict)` copy.
let remote_rev = node_revision_id(&node);
let own_rev = self.own_sealed_revs.lock().get(uid).cloned();
if is_own_self_supersede(meta.complete, remote_rev.as_deref(), own_rev.as_deref()) {
debug!(%uid, ?remote_rev,
"queued write chains onto our own sealed revision; not a conflict");
return Ok(None);
}
Ok(None)
Ok(Some(reason))
}
/// Land a queued write that can no longer be applied to its own node as a
@@ -826,11 +835,20 @@ impl Core {
{
let sealed_mtime = node.modification_time;
let sealed_size = node_size(&node);
let sealed_rev = node_revision_id(&node);
// Remember it as ours, so a queued write that opened over an earlier
// revision recognises this as a self-supersede rather than a foreign
// change and does not fork (B70 layer B; consulted in
// `revision_conflict`).
if let Some(rev) = sealed_rev.clone() {
self.own_sealed_revs.lock().insert(uid.clone(), rev);
}
let mut st = self.state.lock();
for aw in st.active_writes.values_mut() {
if aw.uid == *uid {
aw.base_mtime = sealed_mtime;
aw.base_size = sealed_size;
aw.base_revision_id = sealed_rev.clone();
debug!(%uid, mtime = sealed_mtime, size = sealed_size,
"rebased open write handle onto the revision just uploaded");
}
@@ -875,6 +893,7 @@ impl Core {
mtime: sealed.modification_time,
size: node_size(sealed),
hash: None,
revision_id: node_revision_id(sealed),
};
let mut pending = self.pending.lock();
let Some(p) = pending.get_mut(uid) else {
@@ -896,3 +915,153 @@ impl Core {
true
}
}
/// Whether the remote `node` moved on from the revision a queued write was
/// `base`d on — the reason string when it did, `None` when the write can still be
/// applied. Pure so it is testable without a live remote (the trashed/gone checks
/// stay in [`Core::revision_conflict`], which does the fetch).
///
/// The revision id is the authoritative identity: it advances iff a *new* revision
/// was sealed. When both sides carry one, it alone decides — the server re-stamps
/// the *same* revision's mtime, and reading that drift as a change is exactly what
/// diverted a write into a conflict copy of its own base (B16/B25). Only when
/// there is no id to compare (an old sidecar, or a surface that omits it) does it
/// fall back to the observable `(mtime, size)` tuple.
fn revision_changed(base: &Baseline, node: &Node) -> Option<String> {
if let (Some(base_rev), Some(remote_rev)) = (&base.revision_id, node_revision_id(node)) {
return (*base_rev != remote_rev).then(|| {
format!(
"the remote revision changed under the queued write \
(based on revision {base_rev}, remote now at {remote_rev})"
)
});
}
let (mtime, size) = (node.modification_time, node_size(node));
(mtime != base.mtime || size != base.size).then(|| {
format!(
"the remote revision changed under the queued write \
(expected {} bytes at mtime {}, found {size} at {mtime})",
base.size, base.mtime
)
})
}
/// Whether a queued write that found the remote moved on ([`revision_changed`]
/// fired) is nonetheless a single-writer self-supersede rather than a foreign
/// conflict — the remote sits at a revision this daemon sealed itself, so
/// chaining onto it is safe and forking would be wrong (B70 layer B).
///
/// `remote_rev` is the id the remote currently holds, `own_rev` this node's
/// latest own-sealed id. Requires a `complete` blob: an incomplete one's gaps
/// still refer to the stale base, so it must take the non-destructive
/// conflict-copy path instead of overwriting. A `None` remote id never matches
/// (there is nothing to prove it was ours), so absence falls through to a fork.
fn is_own_self_supersede(complete: bool, remote_rev: Option<&str>, own_rev: Option<&str>) -> bool {
complete && remote_rev.is_some() && remote_rev == own_rev
}
#[cfg(test)]
mod tests {
use super::*;
use proton_drive_rs::NodeKind;
use proton_drive_rs::proton_sdk::ids::{LinkId, VolumeId};
fn file_node(mtime: i64, size: i64, rev: Option<&str>) -> Node {
Node {
uid: NodeUid::new(VolumeId::from("v"), LinkId::from("l")),
parent_uid: None,
kind: NodeKind::File {
media_type: "text/plain".into(),
total_size_on_storage: size,
active_revision_state: None,
active_revision_id: rev.map(String::from),
claimed_size: Some(size),
claimed_modification_time: None,
content_sha1: None,
},
name: "f.txt".into(),
creation_time: 0,
modification_time: mtime,
trashed: false,
is_shared: false,
is_shared_publicly: false,
signature_email: None,
verification: Default::default(),
}
}
fn baseline(mtime: i64, size: u64, rev: Option<&str>) -> Baseline {
Baseline {
mtime,
size,
hash: None,
revision_id: rev.map(String::from),
}
}
#[test]
fn same_revision_id_is_not_a_conflict_despite_mtime_drift() {
// The exact shape of the reported bug: same revision, server re-stamped
// its mtime (1784897777 -> 1784897780), same size. Must NOT conflict.
let base = baseline(1_784_897_777, 87_438, Some("rev-A"));
let remote = file_node(1_784_897_780, 87_438, Some("rev-A"));
assert_eq!(revision_changed(&base, &remote), None);
}
#[test]
fn a_new_revision_id_is_a_conflict_even_at_the_same_size() {
// A same-size edit by another device advances the revision id: this is a
// real conflict the id catches where (mtime, size) alone could miss it.
let base = baseline(100, 40, Some("rev-A"));
let remote = file_node(100, 40, Some("rev-B"));
assert!(revision_changed(&base, &remote).is_some());
}
#[test]
fn without_ids_it_falls_back_to_mtime_and_size() {
// Old sidecar (no revision id): the (mtime, size) tuple still governs.
let base = baseline(100, 40, None);
assert_eq!(revision_changed(&base, &file_node(100, 40, None)), None);
assert!(revision_changed(&base, &file_node(101, 40, None)).is_some());
assert!(revision_changed(&base, &file_node(100, 41, None)).is_some());
}
#[test]
fn a_complete_write_over_our_own_sealed_revision_chains_not_forks() {
// B70 layer B: the partial we sealed ourselves (rev-mine) is what the
// remote holds; the resume rewrote the whole file. Not a foreign change.
assert!(is_own_self_supersede(
true,
Some("rev-mine"),
Some("rev-mine")
));
}
#[test]
fn a_move_to_a_revision_we_did_not_seal_still_forks() {
// Another device wrote rev-theirs; we only ever sealed rev-mine. Fork.
assert!(!is_own_self_supersede(
true,
Some("rev-theirs"),
Some("rev-mine")
));
}
#[test]
fn an_incomplete_blob_never_self_supersedes() {
// Its gaps still refer to the stale base; overwriting would mix
// revisions, so it must take the conflict-copy path even over our seal.
assert!(!is_own_self_supersede(
false,
Some("rev-mine"),
Some("rev-mine")
));
}
#[test]
fn a_missing_remote_id_never_self_supersedes() {
// No id to prove the revision was ours: fall through to a fork rather
// than matching None==None.
assert!(!is_own_self_supersede(true, None, None));
}
}
+77 -35
View File
@@ -110,12 +110,17 @@ impl Filesystem for ProtonFs {
}
fn open(&self, _req: &Request, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) {
let (uid, base_mtime, base_size) = {
let (uid, base_mtime, base_size, base_revision_id) = {
let mut st = self.core.state.lock();
match st.entries.get_mut(&ino.0) {
Some(e) if e.node.is_file() => {
e.open_count = e.open_count.saturating_add(1);
(e.uid.clone(), e.node.modification_time, node_size(&e.node))
(
e.uid.clone(),
e.node.modification_time,
node_size(&e.node),
node_revision_id(&e.node),
)
}
Some(_) => {
reply.error(Errno::EISDIR);
@@ -194,6 +199,7 @@ impl Filesystem for ProtonFs {
len: base_size,
base_size,
base_mtime,
base_revision_id,
dirty: false,
open_count: 0,
}
@@ -330,44 +336,57 @@ impl Filesystem for ProtonFs {
}
}
};
// A transient scratch name — a browser's in-flight `*.crdownload`, an
// editor's `*.swp` — is kept purely local and *parked*: no remote node is
// minted and its bytes never upload while it wears that name. The app
// renames the finished file to its real name, and that rename un-parks the
// create so only the completed file reaches Drive. Minting an empty remote
// node here instead is what let a stalled+resumed download fork a
// `(sync-conflict)` copy (docs/BUGS.md B70).
let transient = is_transient_name(&name);
// Offline the server cannot mint a uid, so invent one and queue the
// create. The file is real to the caller either way; only its identity is
// provisional until the drain (offline.md Phase 3b).
//
// A parent that is itself still queued forces the same path even when we
// are online: the API has no folder to put this in yet.
let node = if self.core.online.load(Ordering::Relaxed) && !is_local_uid(&parent_uid) {
// Create an empty file on the remote so it has a real uid immediately;
// written bytes are buffered and sealed as a new revision on close.
let new_uid = match self.core.rt.block_on(self.core.client.upload_file(
&parent_uid,
&name,
media_type_for(&name),
b"",
)) {
Ok(u) => u,
Err(e) => {
error!(%parent_uid, name, error = %e, "create file failed");
reply.error(Errno::EIO);
return;
let node =
if !transient && self.core.online.load(Ordering::Relaxed) && !is_local_uid(&parent_uid)
{
// Create an empty file on the remote so it has a real uid immediately;
// written bytes are buffered and sealed as a new revision on close.
let new_uid = match self.core.rt.block_on(self.core.client.upload_file(
&parent_uid,
&name,
media_type_for(&name),
b"",
)) {
Ok(u) => u,
Err(e) => {
error!(%parent_uid, name, error = %e, "create file failed");
reply.error(Errno::EIO);
return;
}
};
match self.core.fetch_node(&new_uid) {
Ok(n) => n,
Err(e) => {
reply.error(e);
return;
}
}
} else {
match self
.core
.queue_local_node(&parent_uid, &name, false, transient)
{
Ok(n) => n,
Err(e) => {
reply.error(e);
return;
}
}
};
match self.core.fetch_node(&new_uid) {
Ok(n) => n,
Err(e) => {
reply.error(e);
return;
}
}
} else {
match self.core.queue_local_node(&parent_uid, &name, false) {
Ok(n) => n,
Err(e) => {
reply.error(e);
return;
}
}
};
let new_uid = node.uid.clone();
// The base this handle writes over is the node as it actually exists —
// the empty file the server just minted — so its modification time comes
@@ -409,6 +428,8 @@ impl Filesystem for ProtonFs {
len: 0,
base_size: 0,
base_mtime,
// A brand-new file has no sealed remote revision to conflict with.
base_revision_id: None,
dirty: false,
open_count: 0,
}
@@ -655,10 +676,13 @@ impl Filesystem for ProtonFs {
aw.len,
aw.base_size,
aw.base_mtime,
aw.base_revision_id.clone(),
)
})
};
let Some((f, path, uid, dirty, written, len, base_size, base_mtime)) = handle else {
let Some((f, path, uid, dirty, written, len, base_size, base_mtime, base_revision_id)) =
handle
else {
reply.error(Errno::EBADF);
return;
};
@@ -694,7 +718,9 @@ impl Filesystem for ProtonFs {
base_mtime,
complete: authored == [(0, len)],
authored,
based_on: self.core.remote_baseline(&uid, base_mtime, base_size),
based_on: self
.core
.remote_baseline(&uid, base_mtime, base_size, base_revision_id),
};
// A failed sidecar means the write is not durable, and `fsync` promising
// otherwise is the defect this exists to fix — so it is an error, not a
@@ -887,7 +913,7 @@ impl Filesystem for ProtonFs {
}
}
} else {
match self.core.queue_local_node(&parent_uid, &name, true) {
match self.core.queue_local_node(&parent_uid, &name, true, false) {
Ok(n) => n,
Err(e) => {
reply.error(e);
@@ -1069,6 +1095,22 @@ impl Filesystem for ProtonFs {
&new_parent_uid,
&newname,
);
// Finalize: a transient scratch file (its create parked, bytes
// held back) renamed to a finished name is the moment the
// completed file is meant to reach Drive. Un-park the create so
// the drain uploads it once (docs/BUGS.md B70). Renaming to
// another transient name leaves it parked.
if is_transient_name(&name) && !is_transient_name(&newname) {
match self.core.db.set_create_hold(&uid.to_string(), false) {
Ok(_) => {
self.core.wake_drain();
debug!(%uid, newname, "un-parked a finalized transient create");
}
Err(e) => {
warn!(%uid, error = %e, "un-parking a finalized transient create failed")
}
}
}
debug!(%uid, newname, "renamed a node whose create is still queued");
reply.ok();
}
+75 -10
View File
@@ -60,11 +60,12 @@ use pdfs_core::control::{
TransferDirection,
};
use pdfs_core::db::{
Db, LOCAL_VOLUME, OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PendingOp, StoredNode,
StoredSyncFolder, StoredTrash,
Db, LOCAL_VOLUME, OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PARK_UNTIL, PendingOp,
StoredNode, StoredSyncFolder, StoredTrash,
};
use pdfs_core::localindex;
use pdfs_core::search::relevance_score;
use pdfs_core::syncignore::is_transient_name;
use pdfs_core::{CoreError, CoreResult};
use proton_drive_rs::proton_sdk::api::ResponseCode;
use proton_drive_rs::proton_sdk::error::ProtonError;
@@ -87,6 +88,7 @@ mod profile;
mod reads;
mod sharing;
mod state;
mod sweep;
mod sync;
mod transfers;
mod upload;
@@ -237,6 +239,22 @@ struct Core {
/// off one refresh rather than one per request.
timeline_refreshing: Arc<AtomicBool>,
trash_refreshing: Arc<AtomicBool>,
/// Conflict copies the sweep has already flagged as needing attention this
/// run, so a divergent `(sync-conflict …)` file is logged once rather than
/// on every sweep pass. See [`Core::run_conflict_sweep_loop`].
conflict_notified: Arc<Mutex<HashSet<NodeUid>>>,
/// The latest server revision id this daemon has itself sealed, keyed by
/// node. A queued write whose baseline names an *earlier* revision would
/// normally fork into a `(sync-conflict)` copy when the drain finds the
/// remote already moved on — but if the remote sits at a revision *we*
/// sealed, no other device touched the file: it is a single-writer
/// stall→resume (a browser download that closed and reopened its fd, then
/// rewrote the whole file). [`Core::revision_conflict`] consults this to
/// adopt our own revision as the base and supersede instead of forking
/// (docs/BUGS.md B70, layer B). One entry per written node, overwritten as
/// the node advances and dropped when it is trashed; lost on restart, which
/// only widens the (already narrow) fork window across a daemon bounce.
own_sealed_revs: Arc<Mutex<HashMap<NodeUid, String>>>,
/// Photos whose missing thumbnail is being generated right now. A tile that is
/// still on screen asks for its thumbnail again every few seconds, and each of
/// those downloads is a full-size photo — so an in-flight uid is never started
@@ -1146,7 +1164,12 @@ impl Core {
base_mtime: h.base_mtime,
complete: authored == [(0, h.len)],
authored,
based_on: self.remote_baseline(&h.uid, h.base_mtime, h.base_size),
based_on: self.remote_baseline(
&h.uid,
h.base_mtime,
h.base_size,
h.base_revision_id.clone(),
),
};
self.enqueue_staged_write(&h.uid, h.ino, &h.path, meta)?;
debug!(uid = %h.uid, len = h.len, complete = filled, "queued revision upload");
@@ -1167,7 +1190,13 @@ impl Core {
///
/// `None` for a node that has never existed remotely: there is no revision
/// to conflict with until its create drains.
fn remote_baseline(&self, uid: &NodeUid, base_mtime: i64, base_size: u64) -> Option<Baseline> {
fn remote_baseline(
&self,
uid: &NodeUid,
base_mtime: i64,
base_size: u64,
base_revision_id: Option<String>,
) -> Option<Baseline> {
if is_local_uid(uid) {
return None;
}
@@ -1177,6 +1206,7 @@ impl Core {
mtime: base_mtime,
size: base_size,
hash: None,
revision_id: base_revision_id,
}),
}
}
@@ -1293,12 +1323,15 @@ impl Core {
/// - Shrinking authors nothing: every remaining byte still comes from the
/// base, so it is the drain that has to fetch it.
fn queue_truncate(&self, ino: u64, size: u64) -> Result<(), Errno> {
let (uid, base_mtime, base_size) = {
let (uid, base_mtime, base_size, base_revision_id) = {
let st = self.state.lock();
match st.entries.get(&ino) {
Some(e) if e.node.is_file() => {
(e.uid.clone(), e.node.modification_time, node_size(&e.node))
}
Some(e) if e.node.is_file() => (
e.uid.clone(),
e.node.modification_time,
node_size(&e.node),
node_revision_id(&e.node),
),
Some(_) => return Err(Errno::EISDIR),
None => return Err(Errno::ENOENT),
}
@@ -1374,7 +1407,7 @@ impl Core {
base_mtime,
authored,
complete,
based_on: self.remote_baseline(&uid, base_mtime, base_size),
based_on: self.remote_baseline(&uid, base_mtime, base_size, base_revision_id),
},
};
self.enqueue_staged_write(&uid, ino, &path, meta)?;
@@ -1394,6 +1427,7 @@ impl Core {
parent_uid: &NodeUid,
name: &str,
is_dir: bool,
hold: bool,
) -> Result<Node, Errno> {
let uid = mint_local_uid();
let op = PendingOp {
@@ -1407,7 +1441,10 @@ impl Core {
created_at: now_millis(),
attempts: 0,
last_error: None,
next_attempt_at: 0,
// Parked when the name is transient: its bytes ride on this create
// (via `attach_blob_to_create`) and must not upload until a rename to
// the finished name un-parks it (docs/BUGS.md B70).
next_attempt_at: if hold { PARK_UNTIL } else { 0 },
};
self.db.enqueue_op(&op).map_err(|e| {
error!(%parent_uid, name, error = %e, "queueing local node failed");
@@ -2725,6 +2762,8 @@ fn local_node(uid: NodeUid, parent_uid: NodeUid, name: String, is_dir: bool) ->
total_size_on_storage: 0,
// No revision has been sealed: nothing has been uploaded yet.
active_revision_state: None,
active_revision_id: None,
content_sha1: None,
claimed_size: Some(0),
claimed_modification_time: None,
}
@@ -2933,6 +2972,27 @@ fn node_size(node: &Node) -> u64 {
}
}
/// The server revision id of a node's active revision, if it is a file that has
/// one. The stable identity the drain conflict-checks against (see [`Baseline`]).
fn node_revision_id(node: &Node) -> Option<String> {
match &node.kind {
NodeKind::Folder => None,
NodeKind::File {
active_revision_id, ..
} => active_revision_id.clone(),
}
}
/// The plaintext content SHA-1 of a file node, if its active revision carried one.
/// A download-free content fingerprint the conflict sweep uses to prove two files
/// hold identical bytes before it removes one.
fn node_content_sha1(node: &Node) -> Option<String> {
match &node.kind {
NodeKind::Folder => None,
NodeKind::File { content_sha1, .. } => content_sha1.clone(),
}
}
/// `"1 file"` / `"3 files"` — a count with a correctly pluralised noun, for
/// human-readable activity-log lines.
fn count_noun(n: usize, one: &str, many: &str) -> String {
@@ -3290,6 +3350,8 @@ mod pending_size_tests {
media_type: "text/plain".into(),
total_size_on_storage: 0,
active_revision_state: None,
active_revision_id: None,
content_sha1: None,
claimed_size: Some(claimed),
claimed_modification_time: None,
},
@@ -3486,6 +3548,7 @@ mod tests {
mtime: 42,
size: 3,
hash: None,
revision_id: None,
}),
},
}
@@ -3661,6 +3724,8 @@ mod tests {
media_type: "text/plain".into(),
total_size_on_storage: 0,
active_revision_state: None,
active_revision_id: None,
content_sha1: None,
claimed_size: Some(0),
claimed_modification_time: None,
}
+12
View File
@@ -196,6 +196,8 @@ pub fn mount(
drain_wake: Arc::new((Mutex::new(false), Condvar::new())),
timeline_refreshing: Arc::new(AtomicBool::new(false)),
trash_refreshing: Arc::new(AtomicBool::new(false)),
conflict_notified: Arc::new(Mutex::new(HashSet::new())),
own_sealed_revs: Arc::new(Mutex::new(HashMap::new())),
thumb_gen: Arc::new(Mutex::new(HashSet::new())),
no_thumbnail: Arc::new(Mutex::new(HashMap::new())),
size_upgrades: Arc::new(Mutex::new(HashMap::new())),
@@ -226,6 +228,16 @@ pub fn mount(
// remotes, and reconciles on its own thread — never in front of a FUSE call.
sync::spawn(core.clone(), sync_rx);
// Reconcile leftover `(sync-conflict …)` copies: drop the ones that turned
// out identical to the live file, surface the ones that genuinely diverge.
// Its own thread — it enumerates and trashes, and must never block a FUSE call.
{
let core = core.clone();
std::thread::Builder::new()
.name("pdfs-conflict-sweep".into())
.spawn(move || core.run_conflict_sweep_loop())?;
}
// Mounted from the cache: watch for the network coming back so the mount can
// stop being read-only-ish without the user restarting the daemon.
if !online {
+8
View File
@@ -106,6 +106,12 @@ pub(crate) struct WriteHandle {
pub(crate) base_size: u64,
/// Modification time of the remote base at open, validating its block cache.
pub(crate) base_mtime: i64,
/// Server revision id of the remote base at open, if it had one. This is the
/// stable identity the drain conflict-checks against: `base_mtime` alone
/// drifts when the server re-stamps the *same* revision, which reads as a
/// spurious conflict (see B16/B25). `None` for a file with no sealed remote
/// revision, or one read through a surface that does not carry the id.
pub(crate) base_revision_id: Option<String>,
/// Whether anything diverged from the remote and needs an upload.
pub(crate) dirty: bool,
/// Number of file handles currently sharing this scratch state.
@@ -438,6 +444,8 @@ mod tests {
media_type: "text/plain".into(),
total_size_on_storage: 0,
active_revision_state: None,
active_revision_id: None,
content_sha1: None,
claimed_size: Some(0),
claimed_modification_time: None,
}
+223
View File
@@ -0,0 +1,223 @@
//! Background conflict sweep.
//!
//! A `(sync-conflict <ts>)` copy is only ever a fallback: the drain and the sync
//! engine cut one when they cannot prove a queued write is safe to apply, so the
//! user never loses bytes. Most of those copies turn out to be byte-identical to
//! the live file — a spurious conflict from a re-stamped revision (B16/B25) or a
//! first-sync of an already-identical file. This sweep reconciles them after the
//! fact:
//!
//! - **Identical** to the live sibling (same size *and* same content SHA-1): the
//! copy is redundant, so it is trashed and logged as a [`ActivityKind::Trash`].
//! - **Divergent** (different bytes) or **orphaned** (no live sibling): left in
//! place and surfaced once as an [`ActivityKind::Conflict`], because only the
//! user can decide which version wins.
//!
//! Identity is proved without downloading anything: the SHA-1 comes from the
//! revision's decrypted extended attributes, which enumeration already reads. A
//! copy whose sibling has no recorded SHA-1 is treated as divergent — the sweep
//! never removes a file it cannot prove is a duplicate.
use std::collections::HashMap;
use std::time::Duration;
use pdfs_core::control::ActivityKind;
use proton_drive_rs::proton_sdk::ids::NodeUid;
use tracing::{debug, info, warn};
use super::{Core, node_content_sha1, node_size};
/// Time between sweeps. Conflicts are rare and non-urgent; a slow cadence keeps
/// the sweep off the hot path while still cleaning up within a few minutes.
const CONFLICT_SWEEP_INTERVAL: Duration = Duration::from_secs(300);
/// Delay before the first sweep, so it does not compete with the burst of work a
/// fresh mount does at startup (hydrate, recover, initial enumeration).
const CONFLICT_SWEEP_WARMUP: Duration = Duration::from_secs(30);
/// Strip a ` (sync-conflict <stamp>)` (optionally `-<n>`) marker from `name`,
/// returning the original file name it was a copy of. `None` when `name` is not a
/// conflict copy. Inverse of the `conflict_name` / `conflict_path_with_suffix`
/// formats, so the two halves of the product agree on what a conflict is.
pub(crate) fn conflict_base_name(name: &str) -> Option<String> {
const MARKER: &str = " (sync-conflict ";
let start = name.find(MARKER)?;
let after = start + MARKER.len();
let close_rel = name[after..].find(')')?;
let close = after + close_rel;
let inner = &name[after..close];
// The token is `<digits>` or `<digits>-<digits>`; reject anything else so a
// file that merely happens to contain the phrase is not mistaken for a copy.
let mut parts = inner.splitn(2, '-');
let stamp = parts.next().unwrap_or_default();
let suffix = parts.next();
let is_digits = |s: &str| !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit());
if !is_digits(stamp) || suffix.is_some_and(|s| !is_digits(s)) {
return None;
}
let mut base = String::with_capacity(start + name.len() - close);
base.push_str(&name[..start]);
base.push_str(&name[close + 1..]);
Some(base)
}
impl Core {
/// Sweep for stale conflict copies forever, on its own thread. See the module
/// docs for the identical/divergent policy.
pub(crate) fn run_conflict_sweep_loop(&self) {
std::thread::sleep(CONFLICT_SWEEP_WARMUP);
loop {
self.sweep_conflicts_once();
std::thread::sleep(CONFLICT_SWEEP_INTERVAL);
}
}
/// One sweep pass over every node the daemon currently knows.
fn sweep_conflicts_once(&self) {
let stored = match self.db.load_all() {
Ok(s) => s,
Err(e) => {
warn!(error = ?e, "conflict sweep: load nodes failed");
return;
}
};
let nodes: Vec<_> = stored.into_iter().map(|s| s.node).collect();
// Index live (non-trashed) files by (parent, name) for sibling lookup.
let mut by_parent_name: HashMap<(String, &str), &_> = HashMap::new();
for node in &nodes {
if node.trashed || node.is_folder() {
continue;
}
let parent = node
.parent_uid
.as_ref()
.map(|u| u.to_string())
.unwrap_or_default();
by_parent_name.insert((parent, node.name.as_str()), node);
}
let online = self.online.load(std::sync::atomic::Ordering::Relaxed);
for node in &nodes {
if node.trashed || node.is_folder() {
continue;
}
let Some(base) = conflict_base_name(&node.name) else {
continue;
};
let parent = node
.parent_uid
.as_ref()
.map(|u| u.to_string())
.unwrap_or_default();
let sibling = by_parent_name.get(&(parent, base.as_str())).copied();
match sibling {
Some(sib) if is_identical(node, sib) => {
// Redundant duplicate — remove it, but only while online (the
// remote trash is the point). Offline, it waits for a later pass.
if online {
self.remove_conflict_copy(&node.uid, &node.name, &base);
}
}
Some(_) => {
self.flag_conflict(&node.uid, &node.name, format!("differs from {base}"))
}
None => self.flag_conflict(
&node.uid,
&node.name,
format!("no original {base} to reconcile against"),
),
}
}
}
/// Trash a conflict copy proven identical to its live sibling, and unhook it
/// locally — the same sequence the interactive unlink path uses after a trash.
fn remove_conflict_copy(&self, uid: &NodeUid, name: &str, base: &str) {
match self
.rt
.block_on(self.client.trash_nodes(std::slice::from_ref(uid)))
{
Ok(()) => {}
Err(e) if super::is_gone(&e) => {}
Err(e) => {
warn!(%uid, name, error = %e, "conflict sweep: trashing duplicate failed");
return;
}
}
self.state.lock().forget_or_unlink(uid);
self.discard_queued_ops(uid);
self.cache.evict(uid);
self.evict_reader(uid);
if let Err(e) = self.db.delete_node(uid) {
warn!(%uid, error = ?e, "conflict sweep: db delete after trash failed");
}
self.invalidate_trash();
self.conflict_notified.lock().remove(uid);
self.log_activity(
ActivityKind::Trash,
name,
format!("auto-removed duplicate of {base}"),
true,
);
info!(%uid, name, base, "conflict sweep: removed identical conflict copy");
}
/// Surface a conflict copy that needs the user's attention, at most once per
/// daemon run so the activity feed is not spammed each pass.
fn flag_conflict(&self, uid: &NodeUid, name: &str, detail: String) {
if !self.conflict_notified.lock().insert(uid.clone()) {
return;
}
debug!(%uid, name, detail, "conflict sweep: flagged divergent conflict copy");
self.log_activity(ActivityKind::Conflict, name, detail, false);
}
}
/// Whether two file nodes provably hold the same bytes: equal size and equal,
/// present content SHA-1. A missing digest on either side is not proof, so the
/// sweep treats it as divergent rather than risk removing a real difference.
fn is_identical(a: &proton_drive_rs::Node, b: &proton_drive_rs::Node) -> bool {
if node_size(a) != node_size(b) {
return false;
}
match (node_content_sha1(a), node_content_sha1(b)) {
(Some(x), Some(y)) => x == y,
_ => false,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_conflict_names_and_leaves_others_alone() {
assert_eq!(
conflict_base_name("test_export (sync-conflict 1784898786).xml").as_deref(),
Some("test_export.xml")
);
assert_eq!(
conflict_base_name("README (sync-conflict 42)").as_deref(),
Some("README")
);
// The sync engine's de-duplicated suffix form.
assert_eq!(
conflict_base_name("notes (sync-conflict 1700-3).txt").as_deref(),
Some("notes.txt")
);
// Not a conflict copy.
assert_eq!(conflict_base_name("report.xml"), None);
// Looks close but the token is not a stamp.
assert_eq!(conflict_base_name("x (sync-conflict abc).txt"), None);
}
#[test]
fn a_dotted_stem_keeps_its_inner_dots() {
assert_eq!(
conflict_base_name("archive.tar (sync-conflict 7).gz").as_deref(),
Some("archive.tar.gz")
);
}
}
+21
View File
@@ -670,6 +670,27 @@ impl Core {
match plan_file(l, r, baseline.get(rel)) {
FilePlan::Unchanged => {}
FilePlan::AdoptBaseline => {
// First sync of an already-identical file: record the baseline
// in place so future divergence is detected, without any
// transfer or a spurious conflict copy.
let local = l.expect("adopt baseline requires a local file");
let remote = r.expect("adopt baseline requires a remote file");
if let Err(e) = self.db.sync_entry_upsert(
folder_id,
&StoredSyncEntry {
rel_path: rel.clone(),
remote_uid: Some(remote.uid.to_string()),
local_mtime: local.mtime,
local_size: local.size,
remote_rev: Some(remote.mtime.to_string()),
remote_hash: Some(remote.size.to_string()),
},
) {
warn!(rel, error = ?e, "sync: adopt baseline failed");
outcome.errors += 1;
}
}
FilePlan::Deferred => {
debug!(
rel,
+38 -1
View File
@@ -200,6 +200,12 @@ pub(super) enum FilePlan {
DeleteLocal,
DeleteRemote,
ForgetBaseline,
/// First sync of a file that already exists on both sides with identical
/// content: record the baseline without transferring anything. Without this,
/// a folder synced for the first time cuts a `(sync-conflict)` copy of every
/// pre-existing file, because a missing baseline makes both sides read as
/// "changed" (see the mass first-sync conflict storm in docs/BUGS.md).
AdoptBaseline,
}
pub(super) fn plan_file(
@@ -219,10 +225,16 @@ pub(super) fn plan_file(
});
match (local, remote) {
(Some(_), Some(_)) => match (local_changed, remote_changed) {
(Some(l), Some(r)) => match (local_changed, remote_changed) {
(false, false) => FilePlan::Unchanged,
(true, false) => FilePlan::UploadRevision,
(false, true) => FilePlan::Download,
// With a baseline, both-changed is a genuine concurrent edit and must
// be preserved as a conflict. Without one, "both changed" only means
// this is the first sync and there was nothing to compare to: adopt
// the baseline when the sizes already match rather than conflict-copy
// an already-identical file. Divergent sizes stay a real conflict.
(true, true) if baseline.is_none() && l.size == r.size => FilePlan::AdoptBaseline,
(true, true) => FilePlan::Conflict,
},
(Some(_), None) if baseline.is_none() || local_changed => FilePlan::UploadNew,
@@ -350,6 +362,31 @@ mod file_plan_tests {
}
}
#[test]
fn first_sync_of_an_identical_file_adopts_the_baseline_not_a_conflict() {
// No baseline yet (first sync). Same size on both sides is a pre-existing
// identical file, not a concurrent edit: adopt it rather than cut a
// conflict copy of every file in the folder.
let local = local(10, 40);
let remote = remote(30, 40);
assert_eq!(
plan_file(Some(&local), Some(&remote), None),
FilePlan::AdoptBaseline
);
}
#[test]
fn first_sync_of_differing_sizes_is_still_a_conflict() {
// No baseline and the sizes disagree: a genuine both-sides divergence,
// preserved non-destructively as a conflict.
let local = local(10, 20);
let remote = remote(30, 40);
assert_eq!(
plan_file(Some(&local), Some(&remote), None),
FilePlan::Conflict
);
}
#[test]
fn a_missing_remote_signature_is_treated_as_changed() {
let local = local(10, 20);
@@ -281,6 +281,7 @@ pub(crate) fn activity_verb(kind: ActivityKind) -> &'static str {
ActivityKind::Share => "Shared",
ActivityKind::PublicLink => "Linked",
ActivityKind::Unshare => "Unshared",
ActivityKind::Conflict => "Conflict",
}
}
@@ -299,6 +300,7 @@ pub(crate) fn activity_icon(kind: ActivityKind) -> &'static str {
ActivityKind::Restore => "edit-undo-symbolic",
ActivityKind::Share | ActivityKind::PublicLink => "emblem-shared-symbolic",
ActivityKind::Unshare => "action-unavailable-symbolic",
ActivityKind::Conflict => "dialog-warning-symbolic",
}
}