diff --git a/crates/pdfs-fuse/src/background.rs b/crates/pdfs-fuse/src/background.rs index 7c1835a..d907b9b 100644 --- a/crates/pdfs-fuse/src/background.rs +++ b/crates/pdfs-fuse/src/background.rs @@ -43,7 +43,7 @@ fn forget_and_notify(st: &mut State, notifier: Option<&Notifier>, uid: &NodeUid) /// read-side twin of `docs/BUGS.md` B74. Inode numbers are per-mount, so each /// mount must be notified through its **own** channel; that pairing is what /// [`Core::for_each_mount`] exists to preserve. -fn apply_event(core: &Core, event: &DriveEvent) { +fn apply_event(core: &Core, event: &DriveEvent, dirty: &mut DirtyParents) { match event { DriveEvent::NodeUpdated { node_uid, @@ -56,9 +56,16 @@ fn apply_event(core: &Core, event: &DriveEvent) { // create, and re-fetching would replace the size and mtime of the // write we just accepted with the stale revision's — making a file // that was copied in seconds ago read as empty until its upload - // lands (offline.md Phase 3). The parent listing is still refreshed - // below; it is only the node itself that is ours to define. - let ours = !*is_trashed && core.pending.lock().contains_key(node_uid); + // lands (offline.md Phase 3). + // + // A write that has already drained is the same story one step later: + // the drain brought the tree level with the revision it sealed, so + // the feed's report of that revision has nothing to add, and acting + // on it evicts the very bytes we uploaded. Claimed unconditionally, + // even for a trash, so an echo we no longer care about does not + // linger to suppress a later foreign change. + let echo = core.take_self_change(node_uid); + let ours = !*is_trashed && (echo || core.pending.lock().contains_key(node_uid)); if ours { debug!(uid = %node_uid, "ignoring remote event for a node with a queued write"); } @@ -77,22 +84,24 @@ fn apply_event(core: &Core, event: &DriveEvent) { let _ = notifier.inval_inode(INodeNo(ino), 0, 0); } } - // A create (or move-in) shows up as a change to the parent - // listing; drop it so the new child is picked up on the next - // readdir. - if let Some(parent_uid) = parent_node_uid - && let Some(&parent) = st.by_uid.get(parent_uid) - { - st.invalidate_listing(parent); - if let Some(notifier) = notifier { - let _ = notifier.inval_inode(INodeNo(parent), 0, 0); - } - } }); // Once, not once per mount: the content cache is per-daemon. if had_node { core.cache.evict(node_uid); } + // A create, rename or move-in shows up as a change to the parent + // listing too, which is recorded for the end of the batch rather + // than acted on here — see [`DirtyParents`]. + if let Some(parent_uid) = parent_node_uid { + match ours { + true => dirty + .ours + .entry(parent_uid.clone()) + .or_default() + .insert(node_uid.clone()), + false => dirty.foreign.insert(parent_uid.clone()), + }; + } } DriveEvent::NodeDeleted { node_uid, .. } => { core.cache.evict(node_uid); @@ -121,6 +130,88 @@ fn apply_event(core: &Core, event: &DriveEvent) { } } +/// The folders one batch of events changed, collected so the listings are +/// dropped once at the end of the batch rather than once per event. +/// +/// Both halves of that matter. A burst of uploads produces one event per file +/// all naming the *same* folder, and dropping the listing per event made every +/// interleaved lookup re-enumerate that folder from the API — quadratic in its +/// size, and slow enough to stall a busy mount. Splitting foreign changes from +/// our own then removes the re-enumeration entirely for the common case: a file +/// this daemon created is already in the tree under the right parent, so there +/// is nothing for a re-enumeration to discover. +/// +/// Neither is a weakening. The state after the batch is what it always was. +#[derive(Default)] +struct DirtyParents { + /// Folders changed by someone else. Their listings must go: an event names + /// no name, so a rename inside a folder is indistinguishable from a write to + /// a file in it, and only re-enumerating can tell us which happened. + foreign: HashSet, + /// Folders changed by this daemon, and the children the changes were about. + /// Checked per mount — a mount that already lists every one of those + /// children under that parent keeps its listing. + ours: HashMap>, +} + +/// Whether this mount's cached listing of `parent_uid` already accounts for +/// `child_uid`, so a change we made to that child leaves nothing to re-read. +/// +/// "Nothing cached" counts as current: an absent listing is re-enumerated on the +/// next `readdir` regardless, so there is nothing to drop. +fn listing_accounts_for(st: &State, parent_uid: &NodeUid, child_uid: &NodeUid) -> bool { + let Some(&parent) = st.by_uid.get(parent_uid) else { + return true; + }; + let Some(children) = st.children.get(&parent) else { + return true; + }; + let Some(&child) = st.by_uid.get(child_uid) else { + return false; + }; + children.contains(&child) && st.entries.get(&child).is_some_and(|e| e.parent == parent) +} + +/// Drop the cached listing of every folder a batch of events changed, in every +/// mount that holds one, so the next `readdir`/`lookup` re-enumerates it. +/// +/// Runs strictly after the whole batch is applied: doing it per event would +/// leave a window where a concurrent `readdir` re-populates a listing that a +/// later event in the same batch then has to drop again. +fn flush_dirty_parents(core: &Core, dirty: &DirtyParents) { + if dirty.foreign.is_empty() && dirty.ours.is_empty() { + return; + } + core.for_each_mount(|st, notifier| { + let drop_listing = |st: &mut State, parent_uid: &NodeUid| { + let Some(&parent) = st.by_uid.get(parent_uid) else { + return; + }; + st.invalidate_listing(parent); + if let Some(notifier) = notifier { + let _ = notifier.inval_inode(INodeNo(parent), 0, 0); + } + }; + for parent_uid in &dirty.foreign { + drop_listing(st, parent_uid); + } + for (parent_uid, children) in &dirty.ours { + if dirty.foreign.contains(parent_uid) { + continue; + } + // One child this mount has not placed is enough: the listing it is + // serving is missing something the user just made. + if children + .iter() + .all(|child| listing_accounts_for(st, parent_uid, child)) + { + continue; + } + drop_listing(st, parent_uid); + } + }); +} + /// Poll the remote event cursor forever, applying each batch to the shared /// state. Resumes from the cursor persisted in the DB so changes made while /// unmounted are applied; only a first-ever mount seeds from the server head. @@ -190,6 +281,11 @@ pub(super) async fn run_event_sync( continue; } debug!(count = events.len(), "applying remote events"); + // Folders whose listing this batch invalidates, applied once at the end + // rather than per event — see [`DirtyParents`]. Every `break` below falls + // through to that flush, so an interrupted batch still publishes what it + // did apply. + let mut dirty = DirtyParents::default(); for event in &events { // Converge the SDK's own caches (folder keys, entity cache) on the // server before applying the event to our tree. Without this, a node @@ -200,7 +296,7 @@ pub(super) async fn run_event_sync( warn!(error = %e, "sdk cache invalidation for event failed; retaining cursor for retry"); break; } - apply_event(&core, event); + apply_event(&core, event, &mut dirty); let applied = event.id().clone(); if let Err(e) = db.set_event_cursor(applied.as_str()) { warn!(error = %e, "persist event cursor failed; retaining prior cursor for retry"); @@ -211,6 +307,7 @@ pub(super) async fn run_event_sync( // that was not already made visible locally. cursor = Some(applied); } + flush_dirty_parents(&core, &dirty); } } diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 069b6b4..fff2e29 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -229,6 +229,7 @@ impl Core { } } self.db.delete_op(op.id)?; + self.note_self_change(&uid); info!(%uid, name = %landed, "pending rename landed"); Ok(()) } @@ -344,6 +345,9 @@ impl Core { // sync, whereas a surviving op would create the file a second time. self.db.delete_op(op.id)?; self.adopt_real_uid(&local, &real)?; + // The feed will report this create back to us; the tree already has it + // under its real uid, so that event is ours to ignore (`Core::self_changes`). + self.note_self_change(&real); if let Some(blob) = op.blob_path.as_deref() { self.cache.discard_staged(Path::new(blob)); } @@ -845,6 +849,11 @@ impl Core { return; } }; + // This function's whole job is to bring the tree level with the revision + // we just sealed, so the feed's report of that revision has nothing left + // to tell us. Claimed here rather than at the upload call so it is only + // recorded when the node was actually re-read (`Core::self_changes`). + self.note_self_change(uid); // Rebase any *open* write handles targeting this node. A handle opened // before this upload carries a stale base — its `base_mtime`/`base_size` // name the revision we just replaced. Without this update, diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs index fc1cd70..27e03d0 100644 --- a/crates/pdfs-fuse/src/filesystem.rs +++ b/crates/pdfs-fuse/src/filesystem.rs @@ -315,10 +315,6 @@ impl Filesystem for ProtonFs { reply: ReplyCreate, ) { let parent = parent.0; - if let Err(e) = self.core.ensure_children(parent) { - reply.error(e); - return; - } let name = match fuse_name(name) { Ok(name) => name, Err(error) => { @@ -326,124 +322,16 @@ impl Filesystem for ProtonFs { return; } }; - let parent_uid = { - let st = self.core.state.lock(); - match st.entries.get(&parent) { - Some(e) => e.uid.clone(), - None => { - reply.error(Errno::ENOENT); - return; - } - } - }; - // 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 !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; - } - } - }; - 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 - // from the node, never from the local clock. `queue_revision` turns this - // into `StagedWrite::based_on` and the drain compares that against the - // remote: a `now_secs()` here differs from the server's stamp by however - // long the create round-trip took, so the first write to a brand-new file - // conflicts with its own create whenever the second happens to tick in - // between. - let base_mtime = node.modification_time; - let (file, path) = match self.core.cache.create_scratch() { - Ok(x) => x, - Err(e) => { - error!(%new_uid, error = %e, "create scratch file failed"); - reply.error(Errno::EIO); - return; - } - }; - let mut st = self.core.state.lock(); - let ino = st.intern(parent, node); - if let Some(entry) = st.entries.get_mut(&ino) { - entry.open_count = entry.open_count.saturating_add(1); - } - if let Some(kids) = st.children.get_mut(&parent) - && !kids.contains(&ino) - { - kids.push(ino); - } - let fh = st.next_fh; - st.next_fh += 1; - let aw = st.active_writes.entry(ino).or_insert_with(|| { - // A brand-new file: empty base, everything written is authored. - WriteHandle { - ino, - uid: new_uid, - file: Arc::new(file), - path, - written: Intervals::default(), - 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, - } - }); - aw.open_count += 1; - st.handles.insert(fh, ino); - let attr = self.attr(ino, &st.entries.get(&ino).unwrap().node); - reply.created( - &TTL, - &attr, - Generation(0), - FileHandle(fh), - FopenFlags::empty(), - ); + // Minting a node on the remote is one round trip plus a `fetch_node`, + // and a cold parent adds an enumeration before either. Answering that + // inline stalls the whole mount: fuser's dispatch loop is + // non-concurrent, so eight applications each creating a file were served + // one create at a time, with `ls` on the same mount timing out for as + // long as the burst lasted. See [`Workers`]. + let fs = self.clone(); + self.core + .workers + .run(Lane::Meta, move || fs.serve_create(parent, &name, reply)); } fn write( @@ -556,22 +444,26 @@ impl Filesystem for ProtonFs { // queueing it is what lets a redirect work offline at all: it // never reaches `release`, so without this it failed before any // byte was written (offline.md Phase 2/3b). - if let Err(e) = self.core.queue_truncate(ino.0, size) { - reply.error(e); - return; - } - } else { - self.core.state.lock().set_size(ino.0, size); + // + // The one branch of `setattr` that can reach the network: + // shrinking keeps the prefix, which `queue_truncate` gap-fills + // from the remote. So it answers from a worker, while the + // handle-backed resize above stays inline (it is a `set_len` on + // the scratch file). + let ino = ino.0; + let fs = self.clone(); + self.core.workers.run(Lane::Transfer, move || { + if let Err(e) = fs.core.queue_truncate(ino, size) { + reply.error(e); + return; + } + fs.reply_attr(ino, reply); + }); + return; } + self.core.state.lock().set_size(ino.0, size); } - let st = self.core.state.lock(); - match st.entries.get(&ino.0) { - Some(e) => { - let attr = self.attr(ino.0, &e.node); - reply.attr(&TTL, &attr); - } - None => reply.error(Errno::ENOENT), - } + self.reply_attr(ino.0, reply); } fn fallocate( @@ -769,36 +661,24 @@ impl Filesystem for ProtonFs { }); (h, unlinked_uid) }; - let was_unlinked = unlinked_uid.is_some(); - if let Some(uid) = unlinked_uid { - self.core.discard_queued_ops(&uid); - self.core.cache.evict(&uid); - self.core.evict_reader(&uid); - } - // Hand the bytes to the queue rather than uploading them here: the - // scratch file is the only copy of what was just written, and blocking - // the caller on the network is what made a copy into the mount run at - // upload speed (and fail outright offline). - match (handle, was_unlinked) { - (Some(h), true) => { - // POSIX keeps an unlinked file usable until the last close, but - // closing it must not resurrect those bytes as a new revision. - if let Err(e) = std::fs::remove_file(&h.path) - && e.kind() != std::io::ErrorKind::NotFound - { - warn!(path = %h.path.display(), error = %e, "discarding unlinked scratch file failed"); - } - reply.ok(); - } - (Some(h), false) => match self.core.queue_revision(&h) { - Ok(()) => reply.ok(), - Err(e) => reply.error(e), - }, - (None, _) => reply.ok(), - } + // Deliberately **not** handed to a worker, unlike the other slow + // handlers here. The kernel does not wait for a `release` reply before + // letting `close(2)` return, so the only thing sequencing this against + // the `open`+`read` an application issues immediately afterwards is that + // both are served by the same dispatch loop. Off a worker, the read + // overtakes the staging and sees the *remote's* older revision: the + // acceptance suite's concurrency case failed with "concurrent file 1 + // mismatch" on all three runs the moment this moved. + // + // The cost is real — `queue_revision` gap-fills a partial write from the + // remote — so moving it off still wants doing, but it needs the ordering + // made explicit first (a per-node "staging in flight" barrier that reads + // wait on) rather than inherited from the dispatch loop. + self.finish_release(handle, unlinked_uid, reply); } fn unlink(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) { + let parent = parent.0; let name_str = match fuse_name(name) { Ok(name) => name, Err(error) => { @@ -806,19 +686,15 @@ impl Filesystem for ProtonFs { return; } }; - if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) { - let st = self.core.state.lock(); - if let Some(entry) = st.entries.get(&ino) - && entry.node.is_folder() - { - reply.error(Errno::EISDIR); - return; - } - } - self.trash_child(parent.0, &name_str, reply); + // `trash_child` is a remote call; see `serve_create`. + let fs = self.clone(); + self.core.workers.run(Lane::Meta, move || { + fs.serve_unlink(parent, &name_str, reply) + }); } fn rmdir(&self, _req: &Request, parent: INodeNo, name: &OsStr, reply: ReplyEmpty) { + let parent = parent.0; let name_str = match fuse_name(name) { Ok(name) => name, Err(error) => { @@ -826,31 +702,12 @@ impl Filesystem for ProtonFs { return; } }; - if let Ok((ino, _)) = self.core.lookup_child(parent.0, &name_str) { - { - let st = self.core.state.lock(); - if st - .entries - .get(&ino) - .is_some_and(|entry| !entry.node.is_folder()) - { - reply.error(Errno::ENOTDIR); - return; - } - } - // A missing cached listing says "unknown", not "empty". Proton's - // trash call is recursive, so enumerate before applying rmdir's - // POSIX emptiness rule. - if let Err(e) = self.core.ensure_children(ino) { - reply.error(e); - return; - } - if self.core.state.lock().has_children(ino) { - reply.error(Errno::ENOTEMPTY); - return; - } - } - self.trash_child(parent.0, &name_str, reply); + // Enumerates the folder to apply rmdir's emptiness rule, then trashes + // it: two remote calls. See `serve_create`. + let fs = self.clone(); + self.core + .workers + .run(Lane::Meta, move || fs.serve_rmdir(parent, &name_str, reply)); } fn mkdir( @@ -863,10 +720,6 @@ impl Filesystem for ProtonFs { reply: ReplyEntry, ) { let parent = parent.0; - if let Err(e) = self.core.ensure_children(parent) { - reply.error(e); - return; - } let name = match fuse_name(name) { Ok(name) => name, Err(error) => { @@ -874,73 +727,13 @@ impl Filesystem for ProtonFs { return; } }; - let parent_uid = { - let st = self.core.state.lock(); - match st.entries.get(&parent) { - Some(e) => e.uid.clone(), - None => { - reply.error(Errno::ENOENT); - return; - } - } - }; - let now = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.as_secs() as i64) - .ok(); - // As in `create`: offline — or under a parent that is itself still - // queued — the folder becomes a placeholder that the drain turns into a - // real one (offline.md Phase 3b). - let node = if self.core.online.load(Ordering::Relaxed) && !is_local_uid(&parent_uid) { - let new_uid = - match self - .core - .rt - .block_on(self.core.client.create_folder(&parent_uid, &name, now)) - { - Ok(u) => u, - Err(e) => { - error!(%parent_uid, name, error = %e, "create folder 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, true, false) { - Ok(n) => n, - Err(e) => { - reply.error(e); - return; - } - } - }; - let mut st = self.core.state.lock(); - let local = is_local_uid(&node.uid); - let uid = node.uid.clone(); - let ino = st.intern(parent, node); - if let Some(kids) = st.children.get_mut(&parent) - && !kids.contains(&ino) - { - kids.push(ino); - } - // A folder that exists only here has nothing to enumerate, and asking the - // API to enumerate a `local~` uid would 404. Record it as fully listed and - // empty, which it is, so reads stay offline-clean across a restart too. - if local { - st.children.insert(ino, Vec::new()); - if let Err(e) = self.core.db.set_listed(&uid, true) { - warn!(%uid, error = %e, "db set_listed(true) failed for local folder"); - } - } - let attr = self.attr(ino, &st.entries.get(&ino).unwrap().node); - reply.entry(&TTL, &attr, Generation(0)); + // Off the dispatch loop for the same reason as `create`: this is a + // remote round trip, and inline it blocks every other request on the + // mount until it answers. + let fs = self.clone(); + self.core + .workers + .run(Lane::Meta, move || fs.serve_mkdir(parent, &name, reply)); } fn rename( @@ -969,269 +762,15 @@ impl Filesystem for ProtonFs { return; } }; - let (ino, uid) = match self.core.lookup_child(parent, &name) { - Ok(x) => x, - Err(e) => { - reply.error(e); - return; - } - }; - // Ancestor cycle check: moving a directory into itself or one of its descendants is forbidden by POSIX (EINVAL). - { - let st = self.core.state.lock(); - if let Some(entry) = st.entries.get(&ino) - && entry.node.is_folder() - && st.is_ancestor_of(ino, newparent) - { - reply.error(Errno::EINVAL); - return; - } - } - // The destination has to be listed either way: the queued path pushes - // the node into that listing, and the online one drops it to force a - // re-enumeration. - if newparent != parent - && let Err(e) = self.core.ensure_children(newparent) - { - reply.error(e); - return; - } - let new_parent_uid = { - let st = self.core.state.lock(); - match st.entries.get(&newparent) { - Some(e) => e.uid.clone(), - None => { - reply.error(Errno::ENOENT); - return; - } - } - }; - // `rename(2)` replaces an existing destination; Proton has no replacing - // rename, so the victim has to be removed first and the operation stops - // being atomic. Without this the 422 surfaced as a blanket EIO and every - // write-to-temp-then-rename tool — rsync, atomic editor saves — failed - // at the very end of its transfer (bugs.md B13). - // - // `RENAME_EXCHANGE` has no Proton primitive and cannot be emulated - // without a window in which one of the two names does not exist. - if flags.contains(RenameFlags::RENAME_EXCHANGE) { - reply.error(Errno::EINVAL); - return; - } - let victim = match self.core.lookup_child(newparent, &newname) { - // Renaming a node onto its own name is a no-op, never a self-replace. - Ok((_, vuid)) if vuid == uid => None, - Ok(x) => Some(x), - // Nothing to replace: the overwhelmingly common case. - Err(e) if e.code() == libc::ENOENT => None, - Err(e) => { - reply.error(e); - return; - } - }; - if let Some((victim_ino, victim_uid)) = &victim { - if flags.contains(RenameFlags::RENAME_NOREPLACE) { - reply.error(Errno::EEXIST); - return; - } - // POSIX requires both ends to agree on being a directory, and the - // check has to happen before anything is trashed: getting it wrong - // turns a refusal into the destruction of the destination. - let (src_dir, dst_dir) = { - let st = self.core.state.lock(); - let is_dir = |i: &u64| { - st.entries - .get(i) - .map(|e| matches!(e.node.kind, NodeKind::Folder)) - }; - match (is_dir(&ino), is_dir(victim_ino)) { - (Some(a), Some(b)) => (a, b), - _ => { - reply.error(Errno::ENOENT); - return; - } - } - }; - // Replacing a directory means trashing it, and Proton trashes a - // folder with its whole subtree — so the destination's emptiness has - // to be known before the decision is made. - let dst_empty = if dst_dir { - if let Err(e) = self.core.ensure_children(*victim_ino) { - reply.error(e); - return; - } - self.core - .state - .lock() - .children - .get(victim_ino) - .is_none_or(|kids| kids.is_empty()) - } else { - true - }; - if let Err(e) = check_replaceable(src_dir, dst_dir, dst_empty) { - reply.error(e); - return; - } - if let Err(e) = self.core.remove_replaced(victim_uid, &newname) { - reply.error(e); - return; - } - } - // A node whose own creation is still queued has no server-side identity - // to rename: the queued op *is* the node, so rewriting its target is the - // whole rename. Nothing reaches the API, which is why this works offline - // (offline.md Phase 3b). - if is_local_uid(&uid) { - match self.core.db.rewrite_op_target( - &uid.to_string(), - &new_parent_uid.to_string(), - &newname, - ) { - Ok(true) => { - self.core.state.lock().rename_in_place( - ino, - newparent, - &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(); - } - // The create drained underneath us, so the node has a real uid - // now and this handle's is stale. A retry resolves it. - Ok(false) => { - warn!(%uid, name, newname, "queued create vanished under a rename"); - self.core.restore_replaced(victim.as_ref(), &newname); - reply.error(Errno::EBUSY); - } - Err(e) => { - error!(%uid, error = %e, "rewriting a queued create's target failed"); - self.core.restore_replaced(victim.as_ref(), &newname); - reply.error(Errno::EIO); - } - } - return; - } - // Offline, or into a folder that does not exist remotely yet: queue the - // rename rather than 404 or fail. Online moves into a real parent still - // take the synchronous path, so a genuine API refusal (permissions, a - // name clash) surfaces to the caller instead of becoming a queued op - // that can only ever conflict. - if rename_needs_queue( - self.core.online.load(Ordering::Relaxed), - is_local_uid(&new_parent_uid), - newparent != parent, - newname != name, - ) { - match self - .core - .queue_rename(ino, &uid, newparent, &new_parent_uid, &newname) - { - Ok(()) => { - // Send the rename response first. Kernel-side rename cache - // updates happen while processing that response; notifying - // before it lets those updates overwrite our invalidation - // and retain an empty destination readdir page. - reply.ok(); - if let Some(notifier) = self.core.notifier.get() { - let _ = notifier.inval_entry(INodeNo(parent), OsStr::new(&name)); - let _ = notifier.inval_entry(INodeNo(newparent), OsStr::new(&newname)); - let _ = notifier.inval_inode(INodeNo(parent), 0, 0); - if newparent != parent { - let _ = notifier.inval_inode(INodeNo(newparent), 0, 0); - } - } - } - Err(e) => { - self.core.restore_replaced(victim.as_ref(), &newname); - reply.error(e); - } - } - return; - } - // Rename first if both halves change. Moving first makes the encrypted - // name requirements stale and repeatedly failed with InvalidRequirements. - // A failure past this point has already trashed any node the rename was - // replacing, so put it back rather than leaving the caller with neither - // the source moved nor the destination intact. - if newname != name - && let Err(e) = self - .core - .rt - .block_on(self.core.client.rename_node(&uid, &newname, None)) - { - error!(%uid, error = %e, "rename failed"); - self.core.restore_replaced(victim.as_ref(), &newname); - reply.error(Errno::EIO); - return; - } - if newparent != parent { - let mut attempts = 0u32; - let moved = loop { - match self - .core - .rt - .block_on(self.core.client.move_node(&uid, &new_parent_uid)) - { - Ok(()) => break Ok(()), - Err(e) - if newname != name - && api_code(&e) == Some(ResponseCode::InvalidRequirements) - && attempts < 20 => - { - attempts += 1; - // Requirement propagation routinely takes longer than - // three seconds. Keep a bounded ten-second window; a - // successful POSIX reply must mean both remote halves - // landed, not merely that the second half was queued. - std::thread::sleep(Duration::from_millis(500)); - } - Err(e) => break Err(e), - } - }; - if let Err(e) = moved { - error!(%uid, attempts, error = %e, "move after rename failed"); - // The rename half landed in the source directory. - let mut state = self.core.state.lock(); - let old_parent_uid = state - .entries - .get(&parent) - .map(|entry| entry.uid.clone()) - .unwrap_or_else(|| uid.clone()); - state.rename_in_place(ino, parent, &old_parent_uid, &newname); - drop(state); - self.core.restore_replaced(victim.as_ref(), &newname); - reply.error(Errno::EIO); - return; - } - } - self.core - .state - .lock() - .relocate(ino, parent, newparent, &new_parent_uid, &newname); - reply.ok(); + // A rename that reaches the remote is up to two calls (rename, then + // move), and the replaced-destination path adds a trash on top. See + // `serve_create` for why none of that may run on the dispatch loop. + let fs = self.clone(); + self.core.workers.run(Lane::Meta, move || { + fs.serve_rename(parent, &name, newparent, &newname, flags, reply) + }); } - /// Expose a file's server-side thumbnail/preview as an extended attribute, so - /// a previewing client can fetch it without downloading the whole file. The - /// bytes are fetched on demand and cached; absence yields `ENODATA`. fn getxattr(&self, _req: &Request, ino: INodeNo, name: &OsStr, size: u32, reply: ReplyXattr) { // Always off the dispatch loop: a miss goes to the wire, and a lister // that stats a directory issues one of these per file per advertised @@ -1556,6 +1095,573 @@ impl ProtonFs { /// Trash the child `name` under `parent` on the remote, then drop it from the /// local cache. Backs both `unlink` and `rmdir` (Proton trashes whole /// subtrees, so an `rmdir` of a non-empty dir behaves the same). + /// The body of [`Filesystem::create`], run off the dispatch loop. + fn serve_create(&self, parent: u64, name: &str, reply: ReplyCreate) { + if let Err(e) = self.core.ensure_children(parent) { + reply.error(e); + return; + } + let parent_uid = { + let st = self.core.state.lock(); + match st.entries.get(&parent) { + Some(e) => e.uid.clone(), + None => { + reply.error(Errno::ENOENT); + return; + } + } + }; + // 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 !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; + } + } + }; + 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 + // from the node, never from the local clock. `queue_revision` turns this + // into `StagedWrite::based_on` and the drain compares that against the + // remote: a `now_secs()` here differs from the server's stamp by however + // long the create round-trip took, so the first write to a brand-new file + // conflicts with its own create whenever the second happens to tick in + // between. + let base_mtime = node.modification_time; + let (file, path) = match self.core.cache.create_scratch() { + Ok(x) => x, + Err(e) => { + error!(%new_uid, error = %e, "create scratch file failed"); + reply.error(Errno::EIO); + return; + } + }; + let mut st = self.core.state.lock(); + let ino = st.intern(parent, node); + if let Some(entry) = st.entries.get_mut(&ino) { + entry.open_count = entry.open_count.saturating_add(1); + } + if let Some(kids) = st.children.get_mut(&parent) + && !kids.contains(&ino) + { + kids.push(ino); + } + let fh = st.next_fh; + st.next_fh += 1; + let aw = st.active_writes.entry(ino).or_insert_with(|| { + // A brand-new file: empty base, everything written is authored. + WriteHandle { + ino, + uid: new_uid, + file: Arc::new(file), + path, + written: Intervals::default(), + 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, + } + }); + aw.open_count += 1; + st.handles.insert(fh, ino); + let attr = self.attr(ino, &st.entries.get(&ino).unwrap().node); + reply.created( + &TTL, + &attr, + Generation(0), + FileHandle(fh), + FopenFlags::empty(), + ); + } + + /// The body of [`Filesystem::mkdir`], run off the dispatch loop. + fn serve_mkdir(&self, parent: u64, name: &str, reply: ReplyEntry) { + if let Err(e) = self.core.ensure_children(parent) { + reply.error(e); + return; + } + let parent_uid = { + let st = self.core.state.lock(); + match st.entries.get(&parent) { + Some(e) => e.uid.clone(), + None => { + reply.error(Errno::ENOENT); + return; + } + } + }; + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .ok(); + // As in `create`: offline — or under a parent that is itself still + // queued — the folder becomes a placeholder that the drain turns into a + // real one (offline.md Phase 3b). + let node = if self.core.online.load(Ordering::Relaxed) && !is_local_uid(&parent_uid) { + let new_uid = + match self + .core + .rt + .block_on(self.core.client.create_folder(&parent_uid, name, now)) + { + Ok(u) => u, + Err(e) => { + error!(%parent_uid, name, error = %e, "create folder 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, true, false) { + Ok(n) => n, + Err(e) => { + reply.error(e); + return; + } + } + }; + let mut st = self.core.state.lock(); + let local = is_local_uid(&node.uid); + let uid = node.uid.clone(); + let ino = st.intern(parent, node); + if let Some(kids) = st.children.get_mut(&parent) + && !kids.contains(&ino) + { + kids.push(ino); + } + // A folder that exists only here has nothing to enumerate, and asking the + // API to enumerate a `local~` uid would 404. Record it as fully listed and + // empty, which it is, so reads stay offline-clean across a restart too. + if local { + st.children.insert(ino, Vec::new()); + if let Err(e) = self.core.db.set_listed(&uid, true) { + warn!(%uid, error = %e, "db set_listed(true) failed for local folder"); + } + } + let attr = self.attr(ino, &st.entries.get(&ino).unwrap().node); + reply.entry(&TTL, &attr, Generation(0)); + } + + /// The body of [`Filesystem::unlink`], run off the dispatch loop. + fn serve_unlink(&self, parent: u64, name_str: &str, reply: ReplyEmpty) { + if let Ok((ino, _)) = self.core.lookup_child(parent, name_str) { + let st = self.core.state.lock(); + if let Some(entry) = st.entries.get(&ino) + && entry.node.is_folder() + { + reply.error(Errno::EISDIR); + return; + } + } + self.trash_child(parent, name_str, reply); + } + + /// The body of [`Filesystem::rmdir`], run off the dispatch loop. + fn serve_rmdir(&self, parent: u64, name_str: &str, reply: ReplyEmpty) { + if let Ok((ino, _)) = self.core.lookup_child(parent, name_str) { + { + let st = self.core.state.lock(); + if st + .entries + .get(&ino) + .is_some_and(|entry| !entry.node.is_folder()) + { + reply.error(Errno::ENOTDIR); + return; + } + } + // A missing cached listing says "unknown", not "empty". Proton's + // trash call is recursive, so enumerate before applying rmdir's + // POSIX emptiness rule. + if let Err(e) = self.core.ensure_children(ino) { + reply.error(e); + return; + } + if self.core.state.lock().has_children(ino) { + reply.error(Errno::ENOTEMPTY); + return; + } + } + self.trash_child(parent, name_str, reply); + } + + /// The tail of [`Filesystem::release`], run off the dispatch loop: retire + /// what the closed handle leaves behind. + fn finish_release( + &self, + handle: Option, + unlinked_uid: Option, + reply: ReplyEmpty, + ) { + let was_unlinked = unlinked_uid.is_some(); + if let Some(uid) = unlinked_uid { + self.core.discard_queued_ops(&uid); + self.core.cache.evict(&uid); + self.core.evict_reader(&uid); + } + // Hand the bytes to the queue rather than uploading them here: the + // scratch file is the only copy of what was just written, and blocking + // the caller on the network is what made a copy into the mount run at + // upload speed (and fail outright offline). + match (handle, was_unlinked) { + (Some(h), true) => { + // POSIX keeps an unlinked file usable until the last close, but + // closing it must not resurrect those bytes as a new revision. + if let Err(e) = std::fs::remove_file(&h.path) + && e.kind() != std::io::ErrorKind::NotFound + { + warn!(path = %h.path.display(), error = %e, "discarding unlinked scratch file failed"); + } + reply.ok(); + } + (Some(h), false) => match self.core.queue_revision(&h) { + Ok(()) => reply.ok(), + Err(e) => reply.error(e), + }, + (None, _) => reply.ok(), + } + } + + /// The body of [`Filesystem::rename`], run off the dispatch loop. + fn serve_rename( + &self, + parent: u64, + name: &str, + newparent: u64, + newname: &str, + flags: RenameFlags, + reply: ReplyEmpty, + ) { + let (ino, uid) = match self.core.lookup_child(parent, name) { + Ok(x) => x, + Err(e) => { + reply.error(e); + return; + } + }; + // Ancestor cycle check: moving a directory into itself or one of its descendants is forbidden by POSIX (EINVAL). + { + let st = self.core.state.lock(); + if let Some(entry) = st.entries.get(&ino) + && entry.node.is_folder() + && st.is_ancestor_of(ino, newparent) + { + reply.error(Errno::EINVAL); + return; + } + } + // The destination has to be listed either way: the queued path pushes + // the node into that listing, and the online one drops it to force a + // re-enumeration. + if newparent != parent + && let Err(e) = self.core.ensure_children(newparent) + { + reply.error(e); + return; + } + let new_parent_uid = { + let st = self.core.state.lock(); + match st.entries.get(&newparent) { + Some(e) => e.uid.clone(), + None => { + reply.error(Errno::ENOENT); + return; + } + } + }; + // `rename(2)` replaces an existing destination; Proton has no replacing + // rename, so the victim has to be removed first and the operation stops + // being atomic. Without this the 422 surfaced as a blanket EIO and every + // write-to-temp-then-rename tool — rsync, atomic editor saves — failed + // at the very end of its transfer (bugs.md B13). + // + // `RENAME_EXCHANGE` has no Proton primitive and cannot be emulated + // without a window in which one of the two names does not exist. + if flags.contains(RenameFlags::RENAME_EXCHANGE) { + reply.error(Errno::EINVAL); + return; + } + let victim = match self.core.lookup_child(newparent, newname) { + // Renaming a node onto its own name is a no-op, never a self-replace. + Ok((_, vuid)) if vuid == uid => None, + Ok(x) => Some(x), + // Nothing to replace: the overwhelmingly common case. + Err(e) if e.code() == libc::ENOENT => None, + Err(e) => { + reply.error(e); + return; + } + }; + if let Some((victim_ino, victim_uid)) = &victim { + if flags.contains(RenameFlags::RENAME_NOREPLACE) { + reply.error(Errno::EEXIST); + return; + } + // POSIX requires both ends to agree on being a directory, and the + // check has to happen before anything is trashed: getting it wrong + // turns a refusal into the destruction of the destination. + let (src_dir, dst_dir) = { + let st = self.core.state.lock(); + let is_dir = |i: &u64| { + st.entries + .get(i) + .map(|e| matches!(e.node.kind, NodeKind::Folder)) + }; + match (is_dir(&ino), is_dir(victim_ino)) { + (Some(a), Some(b)) => (a, b), + _ => { + reply.error(Errno::ENOENT); + return; + } + } + }; + // Replacing a directory means trashing it, and Proton trashes a + // folder with its whole subtree — so the destination's emptiness has + // to be known before the decision is made. + let dst_empty = if dst_dir { + if let Err(e) = self.core.ensure_children(*victim_ino) { + reply.error(e); + return; + } + self.core + .state + .lock() + .children + .get(victim_ino) + .is_none_or(|kids| kids.is_empty()) + } else { + true + }; + if let Err(e) = check_replaceable(src_dir, dst_dir, dst_empty) { + reply.error(e); + return; + } + if let Err(e) = self.core.remove_replaced(victim_uid, newname) { + reply.error(e); + return; + } + } + // A node whose own creation is still queued has no server-side identity + // to rename: the queued op *is* the node, so rewriting its target is the + // whole rename. Nothing reaches the API, which is why this works offline + // (offline.md Phase 3b). + if is_local_uid(&uid) { + match self.core.db.rewrite_op_target( + &uid.to_string(), + &new_parent_uid.to_string(), + newname, + ) { + Ok(true) => { + self.core.state.lock().rename_in_place( + ino, + newparent, + &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(); + } + // The create drained underneath us, so the node has a real uid + // now and this handle's is stale. A retry resolves it. + Ok(false) => { + warn!(%uid, name, newname, "queued create vanished under a rename"); + self.core.restore_replaced(victim.as_ref(), newname); + reply.error(Errno::EBUSY); + } + Err(e) => { + error!(%uid, error = %e, "rewriting a queued create's target failed"); + self.core.restore_replaced(victim.as_ref(), newname); + reply.error(Errno::EIO); + } + } + return; + } + // Offline, or into a folder that does not exist remotely yet: queue the + // rename rather than 404 or fail. Online moves into a real parent still + // take the synchronous path, so a genuine API refusal (permissions, a + // name clash) surfaces to the caller instead of becoming a queued op + // that can only ever conflict. + if rename_needs_queue( + self.core.online.load(Ordering::Relaxed), + is_local_uid(&new_parent_uid), + newparent != parent, + newname != name, + ) { + match self + .core + .queue_rename(ino, &uid, newparent, &new_parent_uid, newname) + { + Ok(()) => { + // Send the rename response first. Kernel-side rename cache + // updates happen while processing that response; notifying + // before it lets those updates overwrite our invalidation + // and retain an empty destination readdir page. + reply.ok(); + if let Some(notifier) = self.core.notifier.get() { + let _ = notifier.inval_entry(INodeNo(parent), OsStr::new(&name)); + let _ = notifier.inval_entry(INodeNo(newparent), OsStr::new(&newname)); + let _ = notifier.inval_inode(INodeNo(parent), 0, 0); + if newparent != parent { + let _ = notifier.inval_inode(INodeNo(newparent), 0, 0); + } + } + } + Err(e) => { + self.core.restore_replaced(victim.as_ref(), newname); + reply.error(e); + } + } + return; + } + // Rename first if both halves change. Moving first makes the encrypted + // name requirements stale and repeatedly failed with InvalidRequirements. + // A failure past this point has already trashed any node the rename was + // replacing, so put it back rather than leaving the caller with neither + // the source moved nor the destination intact. + if newname != name + && let Err(e) = self + .core + .rt + .block_on(self.core.client.rename_node(&uid, newname, None)) + { + error!(%uid, error = %e, "rename failed"); + self.core.restore_replaced(victim.as_ref(), newname); + reply.error(Errno::EIO); + return; + } + if newparent != parent { + let mut attempts = 0u32; + let moved = loop { + match self + .core + .rt + .block_on(self.core.client.move_node(&uid, &new_parent_uid)) + { + Ok(()) => break Ok(()), + Err(e) + if newname != name + && api_code(&e) == Some(ResponseCode::InvalidRequirements) + && attempts < 20 => + { + attempts += 1; + // Requirement propagation routinely takes longer than + // three seconds. Keep a bounded ten-second window; a + // successful POSIX reply must mean both remote halves + // landed, not merely that the second half was queued. + std::thread::sleep(Duration::from_millis(500)); + } + Err(e) => break Err(e), + } + }; + if let Err(e) = moved { + error!(%uid, attempts, error = %e, "move after rename failed"); + // The rename half landed in the source directory. + let mut state = self.core.state.lock(); + let old_parent_uid = state + .entries + .get(&parent) + .map(|entry| entry.uid.clone()) + .unwrap_or_else(|| uid.clone()); + state.rename_in_place(ino, parent, &old_parent_uid, newname); + drop(state); + self.core.restore_replaced(victim.as_ref(), newname); + reply.error(Errno::EIO); + return; + } + } + self.core + .state + .lock() + .relocate(ino, parent, newparent, &new_parent_uid, newname); + reply.ok(); + } + + /// Expose a file's server-side thumbnail/preview as an extended attribute, so + /// a previewing client can fetch it without downloading the whole file. The + /// bytes are fetched on demand and cached; absence yields `ENODATA`. + /// Answer with an inode's current attributes, or `ENOENT` if it went away + /// while the caller was doing something slower. + fn reply_attr(&self, ino: u64, reply: ReplyAttr) { + let st = self.core.state.lock(); + match st.entries.get(&ino) { + Some(e) => { + let attr = self.attr(ino, &e.node); + reply.attr(&TTL, &attr); + } + None => reply.error(Errno::ENOENT), + } + } + fn trash_child(&self, parent: u64, name: &str, reply: ReplyEmpty) { let (ino, uid) = match self.core.lookup_child(parent, name) { Ok(x) => x, diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 1d38004..880df87 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -320,6 +320,26 @@ struct Core { /// uploading that same tree — the engine would upload files as they vanish /// and then walk the FUSE mount as if it were local. sync_locks: Arc>>>>, + /// Nodes this daemon changed on the remote itself, and how many echoes of + /// those changes the event feed still owes us. + /// + /// The feed replays our own writes back at us seconds later, and a + /// `NodeUpdated` carries no revision id — nothing in it distinguishes "another + /// device changed this file" from "you changed this file". Treating the echo + /// as foreign is not free: it evicts the content blob we just uploaded, so the + /// next read re-downloads a revision the API may not be serving yet. That is + /// how a SQLite database written on the mount came back `malformed`, and how a + /// mount doing bulk uploads spent its time re-fetching its own bytes. + /// + /// So the first event for a node we changed ourselves, within + /// [`SELF_CHANGE_TTL_MS`], is consumed rather than applied: the tree already + /// holds what that event describes, because [`Core::refresh_after_upload`] put + /// it there. Anything beyond that applies normally, which bounds the staleness + /// a genuine remote change can suffer to one event. + /// + /// Shared across forks like `pending`: the drain that records the change and + /// the event task that consumes the echo are both per-daemon. + self_changes: Arc>>, /// Every live inode space: this mount's own `state` plus one per on-demand /// fork ([`Core::fork_state`]). Shared by every fork, unlike `state` itself. /// @@ -337,6 +357,58 @@ struct Core { states: Arc, } +/// How long a change this daemon made stays recognisable as its own. Generous +/// against a feed that can lag, but finite: past it, an event for the node is +/// applied like any other. See [`Core::self_changes`]. +const SELF_CHANGE_TTL_MS: i64 = 120_000; + +/// A remote change this daemon made, awaiting its echo from the event feed. +struct SelfChange { + /// When we made it, in the same epoch millis as [`now_millis`]. + at_ms: i64, + /// Echoes not yet seen. More than one because a single file can be created + /// and then have a revision uploaded, and the feed reports both. + echoes: u32, +} + +/// Record one remote change this daemon made. Split from [`Core`] so the +/// expiry and echo-counting rules can be tested against a clock the test owns. +fn note_self_change(changes: &mut HashMap, uid: &NodeUid, now_ms: i64) { + // Echoes that never arrived — a node trashed before its event came round, a + // feed that skipped it — would otherwise accumulate for the life of the + // daemon. Pruning on write keeps the map the size of the recent queue. + changes.retain(|_, c| now_ms - c.at_ms < SELF_CHANGE_TTL_MS); + let change = changes.entry(uid.clone()).or_insert(SelfChange { + at_ms: now_ms, + echoes: 0, + }); + change.at_ms = now_ms; + change.echoes += 1; +} + +/// Claim one echo, per [`Core::take_self_change`]. +fn take_self_change( + changes: &mut HashMap, + uid: &NodeUid, + now_ms: i64, +) -> bool { + let Some(change) = changes.get_mut(uid) else { + return false; + }; + // Too old to attribute: the feed is far enough behind that this is as likely + // to be someone else's change as ours, and guessing wrong here costs the user + // a stale file. + if now_ms - change.at_ms >= SELF_CHANGE_TTL_MS { + changes.remove(uid); + return false; + } + change.echoes -= 1; + if change.echoes == 0 { + changes.remove(uid); + } + true +} + /// One live inode space, as published to the shared registry. /// /// The notifier travels with the state because the two are only meaningful @@ -436,6 +508,21 @@ impl Core { } } + /// Record that this daemon just changed `uid` on the remote, so the event + /// feed's echo of that change is recognised instead of re-applied. Called + /// from the drain once the remote has actually accepted the change and the + /// tree has been brought level with it. See [`Core::self_changes`]. + pub(crate) fn note_self_change(&self, uid: &NodeUid) { + note_self_change(&mut self.self_changes.lock(), uid, now_millis()); + } + + /// Claim one outstanding echo for `uid`, reporting whether this event is one + /// this daemon caused. Consuming rather than merely testing is what bounds + /// the suppression: the next event for the node applies normally. + fn take_self_change(&self, uid: &NodeUid) -> bool { + take_self_change(&mut self.self_changes.lock(), uid, now_millis()) + } + /// Rehydrate the in-memory `State` maps from the DB on mount, so a cold /// start serves previously-seen metadata (stable inodes, instant listings) /// without re-hitting the API. The root inode is already installed by @@ -3548,8 +3635,9 @@ mod pending_size_tests { #[cfg(test)] mod tests { use super::{ - Intervals, PendingRevision, StateRegistry, conflict_name, copy_pending_for_truncate, - fuse_name, is_stale_mount, node_visible, rename_needs_queue, + HashMap, Intervals, PendingRevision, SELF_CHANGE_TTL_MS, StateRegistry, conflict_name, + copy_pending_for_truncate, fuse_name, is_stale_mount, node_visible, note_self_change, + parse_node_uid, rename_needs_queue, take_self_change, }; use pdfs_core::cache::{Baseline, StagedWrite}; @@ -3600,6 +3688,67 @@ mod tests { ); } + /// An event feed carries no revision id, so the only thing that tells a + /// remote change apart from the echo of one this daemon just made is this + /// bookkeeping. Getting it wrong is expensive in both directions: too eager + /// and we evict the bytes we uploaded and re-download them from an API that + /// may still be serving the previous revision (a SQLite file on the mount + /// read back `malformed`); too generous and a real change from another device + /// is ignored. + #[test] + fn a_self_change_is_claimed_once_and_expires() { + let mut changes = HashMap::new(); + let uid = parse_node_uid("vol~link").unwrap(); + let other = parse_node_uid("vol~other").unwrap(); + let t0 = 1_000_000; + + assert!( + !take_self_change(&mut changes, &uid, t0), + "a node we never touched is never ours" + ); + + note_self_change(&mut changes, &uid, t0); + assert!( + take_self_change(&mut changes, &uid, t0 + 5_000), + "the echo of our own change is claimed" + ); + assert!( + !take_self_change(&mut changes, &uid, t0 + 6_000), + "and only once, so a later foreign change still applies" + ); + + // A create followed by a revision upload is two changes to one node, and + // the feed reports both. + note_self_change(&mut changes, &uid, t0); + note_self_change(&mut changes, &uid, t0); + assert!(take_self_change(&mut changes, &uid, t0 + 1)); + assert!(take_self_change(&mut changes, &uid, t0 + 2)); + assert!(!take_self_change(&mut changes, &uid, t0 + 3)); + + // Past the window the attribution is a guess, and a wrong guess leaves + // the user looking at a stale file. + note_self_change(&mut changes, &uid, t0); + assert!(!take_self_change( + &mut changes, + &uid, + t0 + SELF_CHANGE_TTL_MS + )); + assert!( + !changes.contains_key(&uid), + "an expired change is dropped, not left to be re-tested" + ); + + // An echo that never arrives must not accumulate for the life of the + // daemon; the next recorded change prunes it. + note_self_change(&mut changes, &uid, t0); + note_self_change(&mut changes, &other, t0 + SELF_CHANGE_TTL_MS); + assert_eq!( + changes.len(), + 1, + "recording a change prunes the ones whose echo never came" + ); + } + /// The predicate must answer *only* for a dead FUSE connection. A healthy /// directory and an absent path are both "not stale" — widening it to any /// `metadata` error would make the on-demand restore lazily unmount paths diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index 052173c..64e5713 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -212,6 +212,7 @@ pub fn mount( conflict_notified: Arc::new(Mutex::new(HashSet::new())), sweep_mode, own_sealed_revs: Arc::new(Mutex::new(HashMap::new())), + self_changes: 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())), diff --git a/docs/BUGS.md b/docs/BUGS.md index 09db95e..cc397ae 100644 --- a/docs/BUGS.md +++ b/docs/BUGS.md @@ -2504,9 +2504,9 @@ seconds after `pending create landed`, and at the same moment uid at all. `/proc//task/*/comm` showed 1 `pdfs-drain` against 8 fuser sessions. -**Fix:** `Core` gained a `states: Arc>>>>` registry -that every mount publishes itself into (`Core::register_state`, called from -`mount.rs` for the primary and from `fork_state` for each fork), plus +**Fix:** `Core` gained a `states: Arc` — a `Vec` of weakly-held +mounts that every mount publishes itself into (`Core::register_state`, called +from `mount.rs` for the primary and from `fork_state` for each fork), plus `Core::for_each_state`, which walks every live inode space one lock at a time. A uid is unique across mounts, so at most one state matches and the rest are no-ops. Five drain sites were converted: `adopt_real_uid`, `adopt_drained_name`, @@ -2523,13 +2523,126 @@ diagnosis that relies on the journal or on queue state will not see this. live runs before the fix and passes after it. The pre-existing B70 case failed on the same cause and now passes too. -**Remaining work.** `drain.rs` still has fork-blind *reads* (`node_place`, -`root_uid`, `parent_is_gone`) and `lib.rs` has ~31 direct `state.lock()` sites; -any of them reached from a background thread has the same shape of problem. -Worth an audit, and worth a unit test over `for_each_state` so a future fork -cannot silently drop out of the registry. +**Audit of the rest of the daemon.** Every background-thread `state.lock()` was +reviewed for the same shape. The rule that came out of it: **a uid is unique +across mounts, so uid-keyed work must be broadcast; an inode number is only +meaningful to the mount that minted it, so inode-keyed work must not be.** What +changed: + +| file | site | why | +|---|---|---| +| `background.rs` | `apply_event` | the read-side twin of this bug — remote deletes, trashes and renames only ever reached the primary mount's tree | +| `sweep.rs` | conflict-copy `forget_or_unlink` | a conflict copy inside a sync folder lives in that fork's inode space | +| `sweep.rs` | `is_busy` | answered "idle" for every file open on a fork, letting the sweep delete one with a writer attached | +| `sharing.rs` | `rel_path_for_uid` | a share of a node in a sync folder could not be named | +| `drain.rs` | `node_place` | fell through to a DB lookup for fork-resident nodes | +| `lib.rs` | `queue_trash`, `rename`, `remove_replaced` (×2), `drop_local`, `restore` | uid-keyed `forget`/invalidate | + +Deliberately left on `self.state`: `upload.rs` (`parent_ino`/`pino` are inode +numbers), the size-upgrade path in `lib.rs`, and `drain.rs`'s `root_uid`, which +is correctly primary-only. The 38 `filesystem.rs` sites are FUSE callbacks — +they already run on the right mount's session thread. `parent_is_gone` never +touches state at all; an earlier note here saying it did was wrong. + +The registry is now a `StateRegistry` type with a unit test +(`state_registry_walks_every_live_mount_and_reaps_the_rest`) covering the case +this bug was: a fork that fails to appear in the walk, or one that lingers after +unmount. + +**One regression came out of the audit, and is fixed.** Broadcasting +`apply_event`'s parent-listing invalidation meant forks started receiving +invalidations they had never received — and it invalidated a folder once per +event, while a burst of our own uploads produces one event per file all naming +the *same* folder. Every following lookup in that folder then re-enumerated it +from the API: quadratic in the folder's size. + +Two mitigations, both in `background.rs`. A batch is collapsed to one +invalidation per folder — `apply_event` records the parent in `DirtyParents` and +`flush_dirty_parents` publishes it once after the batch loop (every `break` path +falls through to it). And a change the daemon made *itself* is recognised as its +own echo rather than treated as foreign: the drain records it +(`Core::note_self_change`, consumed once within `SELF_CHANGE_TTL_MS`), which +stops the feed's report of our own upload from evicting the content blob we just +sent. Neither weakens the result — the state after a batch is what it always was. + +**The hang that showed up alongside this was a different bug, and this entry +originally misattributed it.** `readdir stability` at 133.94 s and a concurrency +case that blew past its timeout were blamed here on the invalidation storm. They +were not: the mount was freezing because network work runs on fuser's single +dispatch thread, which is B75. Fixing that took the concurrency case from two +timeouts in three runs to 16.9 / 24.1 / 29.1 s and `readdir stability` to 62.9 s. +The coalescing above is still worth having — it is a real amplification — but it +was never the stall. **Note on B70:** B70's own fix was never implicated. The `.crdownload` park/un-park behaves correctly in isolation; the empty read arrived with the rename, on the same path a plain temp file takes. B70's live validation was blocked on this and is now unblocked — its acceptance case passes. + +## B75 — Network work on fuser's single dispatch thread freezes the whole mount + +**Status:** Fixed in-tree 2026-07-26 — live-validated. Found 2026-07-26 while +chasing what was thought to be a B74 regression. +**Impact:** any burst of concurrent writes froze the entire mount — not the +files being written, the *mount*. `ls` on an unrelated directory of it timed out +for minutes. The daemon itself stayed healthy throughout: the primary mount and +the control socket answered normally while a sync-folder mount was unreachable. + +**Root cause.** `mount.rs` and `devices.rs` both build their session from +`Config::default()`, which leaves `n_threads` unset, and fuser 0.17 defaults it +to 1 (`session.rs:254`). One dispatch thread per mount reads a request, runs the +handler to completion, and only then reads the next. Six handlers were handed to +the [`Workers`] pool — `lookup`, `getattr`, `readdir`, `read`, `getxattr` and the +slow `serve_lookup` — and everything else ran inline, including handlers that +block on the API: + +| handler | blocking work | +|---|---| +| `create` | `upload_file` to mint the node, then `fetch_node` — two round trips | +| `mkdir` | `create_folder`, then `fetch_node` | +| `rename` | `rename_node`, then `move_node`, plus a trash on the replaced-destination path | +| `unlink` / `rmdir` | `trash_nodes`; `rmdir` enumerates the folder first | +| `setattr` (path truncate) | `queue_truncate` → `fill_gaps`, which reads the kept prefix from the remote | + +So eight applications each creating a file served one create at a time, and +every `read`, `open` and `lookup` on that mount queued behind them. + +**Measured**, acceptance suite's `independent and shared-file concurrency` (8 +threads × 16 files), and a 3-second `ls` of the mount root sampled every 4 s +throughout: + +| | before | after | +|---|---|---| +| concurrency case | 200 s watchdog ×2, 37 s ×1 | 16.9 / 24.1 / 29.1 s, all pass | +| `ls` of the mount during the run | 33 of 55 probes timed out | 0 of 85 | +| `readdir stability` | 148.8 s | 62.9 s | +| `namespace operations` | 31.3 s | 15.7 s | + +**Fix:** each of those handlers now parses its arguments inline — cheap, and it +keeps the error paths prompt — and hands the body to the worker pool as a +`serve_*` method, the pattern `lookup`/`readdir` already used. Metadata work +takes `Lane::Meta`; the truncate path takes `Lane::Transfer` because it can pull +a whole file. + +**`release` was deliberately left inline, and this is load-bearing.** Moving it +too made all three concurrency runs fail with `concurrent file 1 mismatch`. The +kernel does not wait for a `release` reply before letting `close(2)` return, so +the only thing sequencing the staging of the written bytes against the `open` + +`read` an application issues immediately afterwards is that both are served by +the same dispatch loop. From a worker, the read overtakes the staging and is +answered from the remote's older revision. Taking `release` off the loop — worth +doing, since `queue_revision` gap-fills a partial write from the network — needs +a per-node "staging in flight" barrier that reads wait on, not the dispatch +loop's accidental ordering. + +**Not fixed here.** `n_threads` is still 1; with the blocking handlers moved off, +the loop is short enough that raising it is a separate question. Related findings +from the same audit, still open: cache/reader validity is keyed on +`(mtime, size)` while the revision identity is `active_revision_id` +(`cache.rs:84` — two same-size revisions within one mtime second collide, which +is what a SQLite page rewrite looks like); `local_finish_scan` holds the daemon's +one SQLite connection across a full FTS5 trigram rebuild (`db/local.rs:79`); +`earliest_due_at` has no `<= now` filter where `next_due_op` does, so the drain +busy-spins when offline with a due op (`db/ops.rs:385` vs `:421`); and +`drain_local_node` deletes its op before the fallible `adopt_real_uid` +(`drain.rs:346`).