fix(fuse,core): fix B114 to B120 from the account runs

- B114: `pdfs rm`, `pdfs rename` and `pdfs move` tell each kernel
  session that the old name is gone, so it stops resolving at once
  instead of for up to 30 seconds (`Core::forget_everywhere`,
  `Core::relocate_everywhere`).
- B115: `pdfs ls` and the app's file browser wait up to 10 seconds for
  the listing's size fetch, as `stat` does, so they show a file's real
  size instead of its encrypted one.
- B116: trashing a folder also drops the creates queued in its
  subfolders. A parked one stayed queued with its bytes, and a due one
  would have landed in the root.
- B117: `pdfs move` and `pdfs rename` move the node in each mount
  instead of forgetting it, so its queued upload keeps its path in
  `pdfs sync queue` and is not put off once after the resume.
- B118: when a create or mkdir reads back nothing, the mount asks again
  and then queues it as it does offline, instead of failing with ENOENT
  for something Drive has made.
- B119: a landed upload keeps its staged bytes as the file's cached
  content, unless it is larger than a sixteenth of the cache, so
  reading a file just written does not need the network.
- B120: a landed folder's row passes to its real uid in the same
  transaction, so a file queued inside it is not put off for five
  seconds.

docs/BUGS.md records B117 to B120. The account run of 2026-10-03
verified B114 to B117 and B119 in My files.

crates/pdfs-core/src/db/nodes.rs

@@ -599,6 +599,50 @@ struct PriorRow {
     trashed: bool,
 }
 
+/// Hand a drained placeholder's row, and the rows below it, to the real uid it
+/// landed as.
+///
+/// Dropping the placeholder row and waiting for the server's copy of the real
+/// node left a window with no row for either uid. A queued child drained in
+/// that window found no authority for its parent, re-read it over the network
+/// and was deferred five seconds behind a folder that had just landed
+/// (`docs/BUGS.md` B120). The placeholder's own node, readdressed, stands in
+/// until the server's copy is upserted over it. A row the real uid already
+/// has is newer than the placeholder and is kept.
+pub(super) fn adopt_placeholder_row_tx(
+    tx: &Transaction<'_>,
+    local: &str,
+    real: &str,
+) -> Result<()> {
+    tx.execute(
+        "UPDATE nodes SET parent_uid = ?2 WHERE parent_uid = ?1",
+        params![local, real],
+    )?;
+    let row: Option<(i64, Option<String>)> = tx
+        .query_row(
+            "SELECT rowid, node_json FROM nodes WHERE uid = ?1",
+            params![local],
+            |row| Ok((row.get(0)?, row.get(1)?)),
+        )
+        .optional()?;
+    let Some((rowid, json)) = row else {
+        return Ok(());
+    };
+    tx.execute("DELETE FROM nodes WHERE uid = ?1", params![local])?;
+    tx.execute("DELETE FROM nodes_fts WHERE rowid = ?1", params![rowid])?;
+    let known: bool = tx.query_row(
+        "SELECT EXISTS(SELECT 1 FROM nodes WHERE uid = ?1)",
+        params![real],
+        |row| row.get(0),
+    )?;
+    let (false, Some(json), Some(real)) = (known, json, parse_node_uid(real)) else {
+        return Ok(());
+    };
+    let mut node: Node = serde_json::from_str(&json)?;
+    node.uid = real;
+    upsert_node_tx(tx, &node)
+}
+
 fn upsert_node_tx(tx: &Transaction<'_>, node: &Node) -> Result<()> {
     let json = serde_json::to_string(node)?;
     let uid = node.uid.to_string();

crates/pdfs-core/src/db/ops.rs

@@ -6,6 +6,7 @@ use rusqlite::{OptionalExtension, params};
 use serde::{Deserialize, Serialize};
 
 use super::Db;
+use super::nodes::adopt_placeholder_row_tx;
 use crate::Result;
 
 /// The `kind` of a [`PendingOp`] that uploads a staged file as a new revision.
@@ -75,22 +76,33 @@ pub fn op_supersedes(kind: &str) -> bool {
 /// a removed folder: the upload outlived the folder, failed forever against a
 /// trashed node, and kept its blob on disk (`docs/BUGS.md` B102).
 ///
-/// Only revisions are taken from the node tree. A queued rename that moved a
-/// node *into* the folder has to land, or the node would stay where it was
-/// on the server instead of going to the trash with the folder.
+/// "Queued beneath it" starts from every folder the node tree places there,
+/// not only from `uid`. A create in a subfolder that already exists on Drive
+/// names that subfolder as its parent, so following the queue from `uid` alone
+/// missed it: a parked one kept its blob for good, and a due one would have
+/// been made in the root once its folder turned out trashed (`docs/BUGS.md`
+/// B116).
+///
+/// Other ops are taken from the node tree only for revisions. A queued rename
+/// that moved a node *into* the folder has to land, or the node would stay
+/// where it was on the server instead of going to the trash with the folder.
 fn drop_doomed_ops(tx: &rusqlite::Transaction<'_>, uid: &str) -> Result<Vec<String>> {
     const DOOMED: &str = "
         WITH RECURSIVE
-          queued(uid) AS (
-            SELECT ?1
-            UNION
-            SELECT p.uid FROM pending_op p JOIN queued q ON p.parent_uid = q.uid
-          ),
           below(uid) AS (
             SELECT ?1
             UNION
             SELECT n.uid FROM nodes n JOIN below b ON n.parent_uid = b.uid
           ),
+          queued(uid) AS (
+            SELECT ?1
+            UNION
+            SELECT p.uid FROM pending_op p
+            WHERE p.kind IN ('create', 'mkdir')
+              AND p.parent_uid IN (SELECT uid FROM below)
+            UNION
+            SELECT p.uid FROM pending_op p JOIN queued q ON p.parent_uid = q.uid
+          ),
           doomed(id) AS (
             SELECT id FROM pending_op
             WHERE uid IN (SELECT uid FROM queued)
@@ -424,11 +436,7 @@ impl Db {
             "UPDATE pending_op SET parent_uid = ?2 WHERE parent_uid = ?1",
             params![local, real],
         )?;
-        tx.execute(
-            "UPDATE nodes SET parent_uid = ?2 WHERE parent_uid = ?1",
-            params![local, real],
-        )?;
-        tx.execute("DELETE FROM nodes WHERE uid = ?1", params![local])?;
+        adopt_placeholder_row_tx(&tx, local, real)?;
         tx.commit()?;
         Ok(newer)
     }
@@ -547,7 +555,8 @@ impl Db {
 
     /// Rewrite every queued op that points at a placeholder parent, once that
     /// parent has drained and has a real uid. Also moves the node rows whose
-    /// parent column still names the placeholder, so listings keep resolving.
+    /// parent column still names the placeholder, so listings keep resolving,
+    /// and readdresses the placeholder's own row.
     pub fn remap_local_uid(&self, local: &str, real: &str) -> Result<()> {
         let mut conn = self.conn.lock();
         let tx = conn.transaction()?;
@@ -555,11 +564,7 @@ impl Db {
             "UPDATE pending_op SET parent_uid = ?2 WHERE parent_uid = ?1",
             params![local, real],
         )?;
-        tx.execute(
-            "UPDATE nodes SET parent_uid = ?2 WHERE parent_uid = ?1",
-            params![local, real],
-        )?;
-        tx.execute("DELETE FROM nodes WHERE uid = ?1", params![local])?;
+        adopt_placeholder_row_tx(&tx, local, real)?;
         tx.commit()?;
         Ok(())
     }

crates/pdfs-core/src/db/tests.rs

@@ -2579,6 +2579,74 @@ fn trashing_a_folder_drops_the_queued_revisions_of_files_below_it() {
     );
 }
 
+/// A create in a subfolder that already exists on Drive names that subfolder,
+/// not the folder being trashed, as its parent. It has to go with the folder
+/// all the same, along with whatever is queued inside it, or a parked one
+/// keeps its bytes for good and a due one is made in the root (B116).
+#[test]
+fn trashing_a_folder_drops_the_queued_creates_in_its_subfolders() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_nodes(&[
+        folder("root", None, "My Files"),
+        folder("doomed", Some("root"), "doomed"),
+        folder("sub", Some("doomed"), "sub"),
+        folder("other", Some("root"), "other"),
+    ])
+    .unwrap();
+    let op = |kind: &str, node: &str, parent: &str, blob: Option<&str>| PendingOp {
+        id: 0,
+        kind: kind.to_string(),
+        uid: node.to_string(),
+        parent_uid: Some(parent.to_string()),
+        name: Some(format!("{node}.bin")),
+        blob_path: blob.map(str::to_string),
+        meta_json: None,
+        created_at: 1,
+        attempts: 0,
+        last_error: None,
+        next_attempt_at: 0,
+    };
+    let sub = uid("sub").to_string();
+    db.enqueue_op(&op(OP_CREATE, "local~deep", &sub, Some("/staging/deep")))
+        .unwrap();
+    db.enqueue_op(&op(OP_MKDIR, "local~dir", &sub, None))
+        .unwrap();
+    db.enqueue_op(&op(
+        OP_CREATE,
+        "local~deeper",
+        "local~dir",
+        Some("/staging/deeper"),
+    ))
+    .unwrap();
+    db.enqueue_op(&op(
+        OP_CREATE,
+        "local~kept",
+        &uid("other").to_string(),
+        Some("/staging/kept"),
+    ))
+    .unwrap();
+
+    let doomed = uid("doomed").to_string();
+    let (_, mut blobs) = db.replace_ops_with_trash(&doomed, "doomed", 2).unwrap();
+    blobs.sort();
+    assert_eq!(blobs, vec!["/staging/deep", "/staging/deeper"]);
+
+    let mut left: Vec<(String, String)> = db
+        .pending_ops()
+        .unwrap()
+        .into_iter()
+        .map(|op| (op.kind, op.uid))
+        .collect();
+    left.sort();
+    assert_eq!(
+        left,
+        vec![
+            (OP_CREATE.to_string(), "local~kept".to_string()),
+            (OP_TRASH.to_string(), doomed),
+        ]
+    );
+}
+
 /// Deleting a folder that was created offline must take the ops queued
 /// underneath it with it: they name a placeholder parent that will now never
 /// become real, so nothing could ever drain them and nothing is left to
@@ -2902,6 +2970,92 @@ fn draining_a_folder_repoints_its_queued_children() {
     assert_eq!(ops[0].parent_uid.as_deref(), Some(real_dir.as_str()));
 }
 
+/// B120: a child queued inside a folder that has just landed is checked
+/// against the folder's real uid before the server's copy of it arrives.
+#[test]
+fn a_landed_folder_keeps_a_row_until_the_server_copy_replaces_it() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_node(&folder("root", None, "My Files")).unwrap();
+    let local = NodeUid::new(VolumeId::from("local"), LinkId::from("dir"));
+    let mut placeholder = folder("dir", Some("root"), "New folder");
+    placeholder.uid = local.clone();
+    db.upsert_node(&placeholder).unwrap();
+    let mut child = file("child", "root", "inside.txt", 1);
+    child.parent_uid = Some(local.clone());
+    db.upsert_node(&child).unwrap();
+    let (id, _) = db
+        .enqueue_op(&PendingOp {
+            id: 0,
+            kind: OP_MKDIR.to_string(),
+            uid: local.to_string(),
+            parent_uid: Some(uid("root").to_string()),
+            name: Some("New folder".to_string()),
+            blob_path: None,
+            meta_json: None,
+            created_at: 1,
+            attempts: 0,
+            last_error: None,
+            next_attempt_at: 0,
+        })
+        .unwrap();
+
+    db.finish_create(
+        id,
+        None,
+        &local.to_string(),
+        &uid("real").to_string(),
+        |_| None,
+    )
+    .unwrap();
+
+    assert!(db.node_by_uid(&local.to_string()).unwrap().is_none());
+    let landed = db
+        .node_by_uid(&uid("real").to_string())
+        .unwrap()
+        .expect("the landed folder has a row before the server's copy is read");
+    assert_eq!(landed.uid, uid("real"));
+    assert_eq!(landed.name, "New folder");
+    assert_eq!(
+        db.effective_node_access(&uid("child")).unwrap(),
+        Some(crate::Access::Owner),
+        "the child's authority resolves without a network re-read"
+    );
+    let names: Vec<_> = db
+        .known_children(&uid("real"))
+        .unwrap()
+        .into_iter()
+        .map(|n| n.name)
+        .collect();
+    assert_eq!(names, ["inside.txt"]);
+    let found = db.search("inside", 10).unwrap();
+    assert_eq!(found.len(), 1);
+    assert_eq!(found[0].path, "New folder/inside.txt");
+}
+
+#[test]
+fn a_landed_placeholder_never_overwrites_the_server_copy() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_node(&folder("root", None, "My Files")).unwrap();
+    let local = NodeUid::new(VolumeId::from("local"), LinkId::from("dir"));
+    let mut placeholder = folder("dir", Some("root"), "New folder");
+    placeholder.uid = local.clone();
+    db.upsert_node(&placeholder).unwrap();
+    // The event stream reported the folder before the drain got round to it.
+    db.upsert_node(&folder("real", Some("root"), "New folder (1)"))
+        .unwrap();
+
+    db.remap_local_uid(&local.to_string(), &uid("real").to_string())
+        .unwrap();
+
+    assert!(db.node_by_uid(&local.to_string()).unwrap().is_none());
+    assert_eq!(
+        db.node_by_uid(&uid("real").to_string())
+            .unwrap()
+            .map(|n| n.name),
+        Some("New folder (1)".to_string())
+    );
+}
+
 #[test]
 fn a_failed_op_stays_queued_with_backoff() {
     let db = Db::open_in_memory().unwrap();

crates/pdfs-fuse/src/drain.rs

@@ -1106,7 +1106,7 @@ impl Core {
         let node = self
             .fetch_node(real)
             .map_err(|e| self.errno_error(e, "fetch node"))?;
-        // Repoints queued children and node rows, and drops the placeholder row.
+        // Repoints queued children and node rows, and readdresses the placeholder row.
         // A no-op after `retire_create`, which has already done both.
         self.db
             .remap_local_uid(&local.to_string(), &real.to_string())?;
@@ -1638,18 +1638,25 @@ impl Core {
         // op pointing at nothing.
         self.db.delete_op(op.id)?;
         self.release_pending(&uid, &blob);
+        self.evict_reader(&uid);
+        let landed = self.refresh_after_upload(&uid);
 
-        // The staged blob now matches the sealed revision, so a pinned file keeps
-        // it as its cached content rather than re-downloading what we just sent.
-        // Adopted by link, not by value: the blob is the whole file, and a pinned
-        // video read into a `Vec` here is an OOM. `discard_staged` below drops the
-        // staging name only — the cache keeps the inode.
-        if self.cache.is_pinned(&uid) {
-            let _ = self.cache.store_file(&uid, now_secs(), meta.len, &blob);
+        // The staged blob now matches the sealed revision, so it becomes the
+        // file's cached content, stamped like the revision just read back.
+        // Reading the file again then needs no network, which an editor or a
+        // build does at once and a flaky link cannot always give (`docs/BUGS.md`
+        // B119). Adopted by link, not by value: the blob is the whole file, and
+        // a video read into a `Vec` here is an OOM. `discard_staged` below drops
+        // the staging name only — the cache keeps the inode.
+        if let Some(node) = landed
+            && node_size(&node) == meta.len
+            && keeps_landed_upload(self.cache.is_pinned(&uid), meta.len, self.cache.budget())
+        {
+            let _ = self
+                .cache
+                .store_file(&uid, node.modification_time, meta.len, &blob);
         }
         self.cache.discard_staged(&blob);
-        self.evict_reader(&uid);
-        self.refresh_after_upload(&uid);
         self.log_activity(ActivityKind::Upload, &name, "uploaded", true);
         // Both times, so a slow drain shows whether the upload itself or the
         // checks and gap-fill around it are where the time goes.
@@ -1684,18 +1691,19 @@ impl Core {
     /// that conflicted with itself and left a full-size duplicate behind.
     ///
     /// Best effort: a failure here costs a spurious conflict copy on the next
-    /// write, not this upload, which has already landed.
-    pub(crate) fn refresh_after_upload(&self, uid: &NodeUid) {
+    /// write, not this upload, which has already landed. Returns the node as
+    /// the server now has it, or `None` when it could not be read back.
+    pub(crate) fn refresh_after_upload(&self, uid: &NodeUid) -> Option<Node> {
         let node = match self.fetch_node_remote(uid) {
             Ok(Some(node)) => node,
             // Trashed or deleted under us: the tree will hear it from the event
             // sync, which is better placed to unhook the inode than we are.
-            Ok(None) => return,
+            Ok(None) => return None,
             Err(e) => {
                 warn!(%uid, error = %e,
                       "refreshing metadata after an upload failed; \
                        the next write to this file may conflict with itself");
-                return;
+                return None;
             }
         };
         // This function's whole job is to bring the tree level with the revision
@@ -1735,7 +1743,7 @@ impl Core {
         // caught: it took its baseline from the node's optimistic stamp, and
         // this overwrites it with the revision the server actually holds.
         if self.rebaseline_pending(uid, &node) {
-            return;
+            return Some(node);
         }
         self.for_each_state(|st| {
             let Some(parent) = st
@@ -1748,6 +1756,7 @@ impl Core {
             };
             st.intern(parent, node.clone());
         });
+        Some(node)
     }
 
     /// Rebase every write handle open on `uid` onto `node`, a revision this
@@ -1895,6 +1904,19 @@ fn adoptable(is_dir: bool, twin: &Node, op_created_ms: i64) -> bool {
     }
 }
 
+/// What share of the cache budget an unpinned upload may take and still be kept
+/// as cached content once it lands: a sixteenth, 320 MiB of the default 5 GiB.
+/// A larger file would push out much of what the user read lately, for bytes
+/// they may never read back.
+const LANDED_KEEP_SHARE: u64 = 16;
+
+/// Whether an upload of `len` bytes that just landed stays as the file's cached
+/// content. A pinned file always does, and so does any file when the budget is
+/// unlimited (`0`).
+fn keeps_landed_upload(pinned: bool, len: u64, budget: u64) -> bool {
+    pinned || budget == 0 || len <= budget / LANDED_KEEP_SHARE
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -2298,6 +2320,21 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn a_landed_upload_stays_cached_unless_it_would_crowd_the_cache_out() {
+        const GIB: u64 = 1 << 30;
+        let budget = 5 * GIB;
+        // An edited document or a source file: read back from disk, not Drive.
+        assert!(keeps_landed_upload(false, 40_000, budget));
+        assert!(keeps_landed_upload(false, budget / 16, budget));
+        // A video the user copied in would push out what they read lately.
+        assert!(!keeps_landed_upload(false, budget / 16 + 1, budget));
+        // Pinned content is kept whatever its size, and so is everything when
+        // the cache has no cap.
+        assert!(keeps_landed_upload(true, 4 * GIB, budget));
+        assert!(keeps_landed_upload(false, 4 * GIB, 0));
+    }
+
     fn baseline(mtime: i64, size: u64, rev: Option<&str>) -> Baseline {
         Baseline {
             mtime,

crates/pdfs-fuse/src/filesystem.rs

@@ -1517,16 +1517,9 @@ impl ProtonFs {
                 media_type_for(name),
                 b"",
             )) {
-                Ok(new_uid) => match self.core.fetch_node(&new_uid) {
-                    Ok(n) => Some(n),
-                    // The file exists remotely but we could not read it back.
-                    // The queued create finds it by name and adopts it.
-                    Err(e) if e == Errno::EIO && !self.core.is_online() => None,
-                    Err(e) => {
-                        reply.error(e);
-                        return;
-                    }
-                },
+                // A file that exists remotely but could not be read back is
+                // queued, and the queued create finds it by name and adopts it.
+                Ok(new_uid) => self.core.fetch_minted_node(&new_uid, name),
                 Err(e) if self.core.lost_link(&e, "create") => None,
                 Err(e) => {
                     error!(%parent_uid, name, error = %e, "create file failed");
@@ -1626,14 +1619,7 @@ impl ProtonFs {
                 .core
                 .block_on_bounded(self.core.client.create_folder(&parent_uid, name, now))
             {
-                Ok(new_uid) => match self.core.fetch_node(&new_uid) {
-                    Ok(n) => Some(n),
-                    Err(e) if e == Errno::EIO && !self.core.is_online() => None,
-                    Err(e) => {
-                        reply.error(e);
-                        return;
-                    }
-                },
+                Ok(new_uid) => self.core.fetch_minted_node(&new_uid, name),
                 Err(e) if self.core.lost_link(&e, "mkdir") => None,
                 Err(e) => {
                     error!(%parent_uid, name, error = %e, "create folder failed");

crates/pdfs-fuse/src/lib.rs

@@ -111,7 +111,7 @@ mod transfers;
 mod upload;
 mod r#virtual;
 mod workers;
-use background::{run_event_sync, run_local_index, run_photos_event_sync};
+use background::{forget_and_notify, run_event_sync, run_local_index, run_photos_event_sync};
 pub(crate) use mount::is_stale_mount;
 pub use mount::{MountOptions, MountOutcome, mount};
 use mount::{
@@ -169,6 +169,9 @@ const DRAIN_IDLE_POLL: Duration = Duration::from_secs(30);
 /// its inode (`Core::landed_uid`): 200 polls of 10 ms, two seconds in all.
 const LANDED_UID_POLLS: u32 = 200;
 const LANDED_UID_POLL: Duration = Duration::from_millis(10);
+/// How long the mount waits before asking again for a node it has just created
+/// when Drive does not list it yet (`Core::fetch_minted_node`).
+const MINTED_READ_BACK_DELAYS: [Duration; 2] = [Duration::from_millis(250), Duration::from_secs(1)];
 /// Grace period before a queued revision becomes eligible for draining.
 ///
 /// Tools like aria2c preallocate a file (truncate to target size) and then write
@@ -1275,6 +1278,41 @@ fn uid_write_authority(db: &Db, uid: &NodeUid, live_access: &[Access]) -> WriteA
     }
 }
 
+/// Move `uid` to `name` in `new_parent` within one mount, for a rename or move
+/// Drive has made, and record what the kernel has to forget. Returns the name
+/// the mount knew the node by, and whether the node was moved rather than
+/// forgotten, which writes its row through. `None` when the mount does not
+/// hold the node.
+///
+/// The inode is kept wherever the mount shows the new parent, as the mount's
+/// own rename keeps it (`State::relocate`). A mount that does not show the new
+/// parent only forgets the node; to it, the node is gone. The row stays either
+/// way (`docs/BUGS.md` B117).
+fn relocate_in_mount(
+    st: &mut State,
+    notify: &mut NotifyBatch,
+    uid: &NodeUid,
+    new_parent: &NodeUid,
+    name: &str,
+) -> Option<(String, bool)> {
+    let ino = *st.by_uid.get(uid)?;
+    let entry = st.entries.get(&ino)?;
+    let (from_parent, from_name) = (entry.parent, entry.node.name.clone());
+    match st.by_uid.get(new_parent).copied() {
+        Some(to_parent) => {
+            st.relocate(ino, from_parent, to_parent, new_parent, name);
+            notify.inval_entry(from_parent, from_name.clone());
+            notify.inval_entry(to_parent, name.to_string());
+            Some((from_name, true))
+        }
+        None => {
+            st.forget_mem(uid);
+            notify.delete(from_parent, ino, from_name.clone());
+            Some((from_name, false))
+        }
+    }
+}
+
 /// One kernel notification, recorded while the State lock is held so it can be
 /// sent once it is not.
 ///
@@ -2694,6 +2732,46 @@ impl Core {
         }
     }
 
+    /// Wait, up to [`SizeUpgrade::WAIT`] as a `stat` would, for the real size
+    /// of every file in `ino`'s listing that still carries a provisional one.
+    ///
+    /// For callers that read a whole listing's sizes straight from `state` and
+    /// do not run on a FUSE thread: `pdfs ls` and the app's file browser. A
+    /// listing just fetched has the ciphertext size of every file in it until
+    /// its upgrade lands, and they showed those (`docs/BUGS.md` B115).
+    fn settle_listing_sizes(&self, ino: u64) {
+        // Offline the fetch cannot succeed; the wait would only delay the
+        // answer it ends up giving anyway.
+        if !self.is_online() {
+            return;
+        }
+        let missing: Vec<NodeUid> = {
+            let st = self.state();
+            st.children
+                .get(&ino)
+                .map(|kids| {
+                    kids.iter()
+                        .filter_map(|k| st.entries.get(k))
+                        .filter(|e| {
+                            !is_local_uid(&e.uid)
+                                && matches!(
+                                    &e.node.kind,
+                                    NodeKind::File {
+                                        claimed_size: None,
+                                        ..
+                                    }
+                                )
+                        })
+                        .map(|e| e.uid.clone())
+                        .collect()
+                })
+                .unwrap_or_default()
+        };
+        if let Some(slot) = self.upgrade_sizes(ino, missing) {
+            slot.wait_finished(Instant::now() + SizeUpgrade::WAIT);
+        }
+    }
+
     /// Whether `ino` has a real size — the condition a waiter is waiting on.
     /// A node that vanished counts as resolved; there is nothing left to wait
     /// for, and its caller will find the `ENOENT` for itself.
@@ -2917,6 +2995,29 @@ impl Core {
         }
     }
 
+    /// Read back a node just created on the remote.
+    ///
+    /// Drive can answer a create before a lookup lists the new node, so an
+    /// empty answer is asked again for a moment. `None` means the node exists
+    /// but could not be read back. The mount then queues the create, and the
+    /// drain finds the node by name and adopts it: the application gets its
+    /// file or folder either way, never an `ENOENT` for something that exists.
+    fn fetch_minted_node(&self, uid: &NodeUid, name: &str) -> Option<Node> {
+        let mut delays = MINTED_READ_BACK_DELAYS.iter();
+        loop {
+            match self.fetch_node(uid) {
+                Ok(node) => return Some(node),
+                Err(e) if e == Errno::ENOENT => match delays.next() {
+                    Some(delay) => std::thread::sleep(*delay),
+                    None => break,
+                },
+                Err(_) => break,
+            }
+        }
+        warn!(%uid, name, "Drive made the node but did not hand it back");
+        None
+    }
+
     /// [`Core::fetch_node`] without the collapse to an `Errno`, for the drain:
     /// resolving a conflict turns on *why* a call failed, and "the node is not
     /// there" (`Ok(None)`) is a different outcome from "we could not ask".
@@ -3964,6 +4065,7 @@ impl Core {
         let (ino, _uid) = self.resolve(rel)?;
         self.ensure_children(ino)
             .map_err(|e| self.errno_error(e, "enumerate"))?;
+        self.settle_listing_sizes(ino);
         // Snapshot the listing, then drop the lock before touching the on-disk
         // pin registry so a slow disk read doesn't block FUSE metadata ops.
         let rows: Vec<(String, bool, u64, i64, NodeUid)> = {
@@ -4401,9 +4503,9 @@ impl Core {
     }
 
     /// Rename a file or folder to `new_name`. `rel` is mountpoint-relative.
-    /// Mirrors the FUSE `rename` write path: rename on the remote, forget the
-    /// node so it re-interns under its new name, and drop the parent listing so
-    /// the next `ListDir` re-enumerates.
+    /// Mirrors the FUSE `rename` write path: rename on the remote, then move the
+    /// node to its new name in every mount and drop the parent listing so the
+    /// next `ListDir` re-enumerates.
     fn rename(&self, rel: &Path, new_name: &str) -> CoreResult<String> {
         if new_name.is_empty() || new_name.contains('/') {
             return Err(CoreError::invalid(format!("invalid name: {new_name:?}")));
@@ -4419,18 +4521,85 @@ impl Core {
         .map_err(|error| self.errno_error(error, "rename access"))?;
         self.rename_remote(&uid, new_name)
             .map_err(|e| CoreError::from_api(&e, "rename"))?;
-        // Every mount, so a fork showing the same node re-interns it under the
-        // new name instead of keeping the old one (`docs/BUGS.md` B74).
-        self.for_each_state(|st| {
-            st.forget(&uid);
-        });
+        self.relocate_everywhere(&uid, &old_parent_uid, new_name);
         self.invalidate_parent_listing(rel);
         Ok(new_name.to_string())
     }
 
+    /// Bring every mount level with a rename or move Drive has made of `uid`,
+    /// to `name` in `new_parent`, and have each kernel session drop the name it
+    /// cached for it. Returns the name a mount knew the node by.
+    ///
+    /// For changes that do not come through the mount. The kernel made a
+    /// rename through the mount itself, but one made over the control socket
+    /// leaves the old name cached for [`TTL`] (`docs/BUGS.md` B114).
+    ///
+    /// The node is moved, not forgotten, and so is its row. Forgetting it, as
+    /// this used to, deleted the row, and until a listing brought it back
+    /// nothing local knew the node: `pdfs sync queue` showed its queued upload
+    /// by raw id, and the drain re-read the node and put the upload off
+    /// (`docs/BUGS.md` B117).
+    fn relocate_everywhere(
+        &self,
+        uid: &NodeUid,
+        new_parent: &NodeUid,
+        name: &str,
+    ) -> Option<String> {
+        let mut old_name = None;
+        let mut moved_row = false;
+        self.for_each_mount(|st, notify| {
+            if let Some((from_name, relocated)) =
+                relocate_in_mount(st, notify, uid, new_parent, name)
+            {
+                old_name = Some(from_name);
+                moved_row |= relocated;
+            }
+        });
+        if !moved_row {
+            self.move_row(uid, new_parent, name);
+        }
+        old_name
+    }
+
+    /// Point the persisted row of `uid` at `name` in `new_parent`, for a node
+    /// no mount holds where it now is.
+    fn move_row(&self, uid: &NodeUid, new_parent: &NodeUid, name: &str) {
+        let moved = self.db.node_by_uid(&uid.to_string()).and_then(|node| {
+            let Some(mut node) = node else {
+                return Ok(());
+            };
+            node.name = name.to_string();
+            node.parent_uid = Some(new_parent.clone());
+            self.db.upsert_node(&node)
+        });
+        if let Err(error) = moved {
+            warn!(%uid, ?error, "could not move the node's row along with it");
+        }
+    }
+
+    /// Forget `uid` in every mount, and have each kernel session drop the name
+    /// it cached for it. Returns the name a mount knew the node by.
+    ///
+    /// For a removal that does not come through the mount. The kernel made an
+    /// unlink through the mount itself, but one made over the control socket
+    /// leaves the old name cached for [`TTL`], pointing at an inode the mount no
+    /// longer has (`docs/BUGS.md` B114). Every mount, not only this one, so a
+    /// fork showing the same node does not keep serving it (`docs/BUGS.md`
+    /// B74).
+    fn forget_everywhere(&self, uid: &NodeUid) -> Option<String> {
+        let mut name = None;
+        self.for_each_mount(|st, notify| {
+            if let Some(entry) = st.by_uid.get(uid).and_then(|ino| st.entries.get(ino)) {
+                name = Some(entry.node.name.clone());
+            }
+            forget_and_notify(st, notify, uid);
+        });
+        name
+    }
+
     /// Move a file or folder into the folder at `new_parent_rel`. Both paths are
-    /// mountpoint-relative. Forgets the node and invalidates both the source and
-    /// destination listings so each re-enumerates on next access.
+    /// mountpoint-relative. Moves the node in every mount and invalidates both
+    /// the source and destination listings so each re-enumerates on next access.
     fn move_to(&self, rel: &Path, new_parent_rel: &Path) -> CoreResult<String> {
         let (ino, uid) = self.resolve(rel)?;
         let old_parent_uid = self.source_parent_uid(ino, rel)?;
@@ -4444,15 +4613,16 @@ impl Core {
             &new_parent_uid,
         )
         .map_err(|error| self.errno_error(error, "move access"))?;
+        let name = self
+            .state()
+            .entries
+            .get(&ino)
+            .map(|entry| entry.node.name.clone())
+            .unwrap_or_default();
         self.rt
             .block_on(self.client.move_node(&uid, &new_parent_uid))
             .map_err(|e| CoreError::from_api(&e, "move"))?;
-        let name = self
-            .state
-            .lock()
-            .forget(&uid)
-            .map(|(_, n)| n)
-            .unwrap_or_default();
+        self.relocate_everywhere(&uid, &new_parent_uid, &name);
         self.invalidate_parent_listing(rel);
         self.state().invalidate_listing(pino);
         Ok(name)
@@ -4478,12 +4648,7 @@ impl Core {
         if let Err(error) = self.discard_queued_ops(&uid) {
             warn!(%uid, ?error, "remote trash landed but queued-op cleanup failed");
         }
-        let name = self
-            .state
-            .lock()
-            .forget(&uid)
-            .map(|(_, n)| n)
-            .unwrap_or_default();
+        let name = self.forget_everywhere(&uid).unwrap_or_default();
         self.cache.evict(&uid);
         self.evict_reader(&uid);
         self.invalidate_parent_listing(rel);
@@ -5015,8 +5180,8 @@ impl Core {
             )
             .map_err(|e| CoreError::from_api(&e, "create folder"))?;
         let node = self
-            .fetch_node(&new_uid)
-            .map_err(|e| self.errno_error(e, "fetch node"))?;
+            .fetch_minted_node(&new_uid, name)
+            .ok_or_else(|| minted_but_unlisted(name))?;
         let mut st = self.state();
         let ino = st.intern(pino, node);
         if let Some(kids) = st.children.get_mut(&pino)
@@ -5381,6 +5546,15 @@ fn is_already_exists(e: &(dyn std::error::Error + 'static)) -> bool {
 }
 
 /// Whether a call failed because the node it addressed is not there.
+/// The error for a folder Drive created but did not hand back, where there is
+/// no queue to adopt it later. Saying it was created keeps the caller from
+/// taking a retry's "already exists" for someone else's folder.
+fn minted_but_unlisted(name: &str) -> CoreError {
+    CoreError::remote(format!(
+        "created folder {name}, but Drive did not list it yet; it shows on the next refresh"
+    ))
+}
+
 fn is_gone(e: &(dyn std::error::Error + 'static)) -> bool {
     api_code(e) == Some(ResponseCode::DoesNotExist)
 }
@@ -5626,6 +5800,18 @@ impl SizeUpgrade {
         self.ready.notify_all();
     }
 
+    /// Block until the batch has ended or `deadline` has passed. Only for a
+    /// caller that is not on a FUSE or [`Workers`] thread; those park their
+    /// reply with [`Core::await_size`] instead.
+    fn wait_finished(&self, deadline: Instant) {
+        let mut progress = self.inner.lock();
+        while !progress.done {
+            if self.ready.wait_until(&mut progress, deadline).timed_out() {
+                return;
+            }
+        }
+    }
+
     /// Release every waiter for good. The worker must reach this on all paths.
     fn finish(&self) {
         let mut progress = self.inner.lock();
@@ -6270,6 +6456,32 @@ mod size_upgrade_tests {
         t.join().unwrap();
     }
 
+    /// `pdfs ls` waits for the whole batch, so it lists no ciphertext size
+    /// (`docs/BUGS.md` B115), but no longer than a `stat` would.
+    #[test]
+    fn a_listing_waits_for_the_whole_size_batch_but_not_forever() {
+        let slot = Arc::new(SizeUpgrade::default());
+        let worker = slot.clone();
+        let t = std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(30));
+            worker.chunk_done();
+            std::thread::sleep(Duration::from_millis(60));
+            worker.finish();
+        });
+        let started = Instant::now();
+        slot.wait_finished(started + Duration::from_secs(5));
+        // Past the first chunk, up to the end of the batch.
+        assert!(started.elapsed() >= Duration::from_millis(90));
+        assert!(slot.is_finished());
+        t.join().unwrap();
+
+        let stuck = SizeUpgrade::default();
+        let started = Instant::now();
+        stuck.wait_finished(started + Duration::from_millis(50));
+        assert!(started.elapsed() < Duration::from_secs(1));
+        assert!(!stuck.is_finished());
+    }
+
     /// A chunk that does not resolve this waiter must not answer it.
     #[test]
     fn an_unrelated_chunk_does_not_release_a_waiter() {
@@ -8128,26 +8340,32 @@ mod tests {
             "require_uid_writable",
             ".create_folder(&parent_uid",
         );
-        for (signature, gate, remote_call) in [
+        for (signature, gate, remote_call, settle) in [
             (
                 "fn rename(&self, rel: &Path",
                 "require_rename_access",
                 ".rename_remote",
+                "self.relocate_everywhere(&uid",
             ),
             (
                 "fn move_to(&self, rel: &Path",
                 "require_rename_access",
                 ".move_node",
+                "self.relocate_everywhere(&uid",
             ),
             (
                 "fn delete(&self, rel: &Path",
                 "require_node_parent_access",
                 ".trash_nodes",
+                "self.forget_everywhere(&uid)",
             ),
         ] {
             let function = function_source(source, signature);
             assert_before(function, "source_parent_uid", gate);
             assert_before(function, gate, remote_call);
+            // The kernel did not see this change, so it has to be told the
+            // old name is gone (`docs/BUGS.md` B114).
+            assert_before(function, remote_call, settle);
         }
 
         let upload = include_str!("upload.rs");
@@ -8159,6 +8377,19 @@ mod tests {
         assert_before(folder, "require_uid_writable(parent_uid)", ".create_folder");
     }
 
+    #[test]
+    fn a_listing_for_the_cli_settles_sizes_before_reading_them() {
+        // A fresh listing holds ciphertext sizes until its upgrade lands
+        // (`docs/BUGS.md` B115).
+        let list = function_source(include_str!("lib.rs"), "fn list_dir(&self, rel: &Path)");
+        assert_before(
+            list,
+            "self.ensure_children(ino)",
+            "self.settle_listing_sizes(ino)",
+        );
+        assert_before(list, "self.settle_listing_sizes(ino)", "node_size(&e.node)");
+    }
+
     #[test]
     fn a_replaced_node_survives_its_open_handles() {
         // A rename over a file someone is reading must leave that reader on the
@@ -8475,6 +8706,60 @@ mod tests {
         node
     }
 
+    /// A rename or move made over the control socket keeps the node and its
+    /// row, so the queue and the drain still know it (`docs/BUGS.md` B117),
+    /// and tells the kernel the old name is gone (`docs/BUGS.md` B114).
+    #[test]
+    fn a_control_move_keeps_the_node_and_its_row_and_drops_the_old_name() {
+        use super::{KernelNotice, NotifyBatch, relocate_in_mount};
+
+        let (mut state, _dir) = state_test_helper();
+        let src = state.intern(0, node_helper("src", "none", "src", true));
+        let dst = state.intern(0, node_helper("dst", "none", "dst", true));
+        let moved = node_helper("moved", "src", "a.txt", false);
+        let uid = moved.uid.clone();
+        let ino = state.intern(src, moved);
+        state.children.insert(src, vec![ino]);
+        state.children.insert(dst, vec![]);
+        let dst_uid = state.entries[&dst].uid.clone();
+
+        let mut notify = NotifyBatch::default();
+        let outcome = relocate_in_mount(&mut state, &mut notify, &uid, &dst_uid, "b.txt");
+        assert_eq!(outcome, Some(("a.txt".to_string(), true)));
+        assert_eq!(state.by_uid.get(&uid), Some(&ino), "the inode survives");
+        assert_eq!(state.entries[&ino].parent, dst);
+        let row = state
+            .flushed_db()
+            .node_by_uid(&uid.to_string())
+            .unwrap()
+            .expect("the row survives the move");
+        assert_eq!(row.name, "b.txt");
+        assert_eq!(row.parent_uid.as_ref(), Some(&dst_uid));
+        assert!(notify.0.iter().any(|notice| matches!(
+            notice,
+            KernelNotice::InvalEntry { parent, name } if *parent == src && name == "a.txt"
+        )));
+
+        // A mount that does not show the new parent forgets the node, and the
+        // kernel the name, but the row stays for the caller to move.
+        let elsewhere = node_helper("elsewhere", "none", "elsewhere", true).uid;
+        let mut notify = NotifyBatch::default();
+        let outcome = relocate_in_mount(&mut state, &mut notify, &uid, &elsewhere, "c.txt");
+        assert_eq!(outcome, Some(("b.txt".to_string(), false)));
+        assert!(!state.by_uid.contains_key(&uid));
+        assert!(notify.0.iter().any(|notice| matches!(
+            notice,
+            KernelNotice::Delete { parent, child, name } if *parent == dst && *child == ino && name == "b.txt"
+        )));
+        assert!(
+            state
+                .flushed_db()
+                .node_by_uid(&uid.to_string())
+                .unwrap()
+                .is_some()
+        );
+    }
+
     #[test]
     fn queued_trash_of_open_file_survives_last_release_until_atomic_completion() {
         let (mut state, _dir) = state_test_helper();

crates/pdfs-fuse/src/relocate.rs

@@ -25,7 +25,6 @@ use proton_drive_rs::NodeKind;
 use proton_drive_rs::proton_sdk::ids::NodeUid;
 use tracing::{info, warn};
 
-use super::background::forget_and_notify;
 use super::sync::mirror_subtree_unsynced;
 use super::{Core, is_local_uid, parse_uid};
 
@@ -265,7 +264,15 @@ impl Core {
             return Err(CoreError::from_api(&e, "move"));
         }
 
-        self.settle_source(&source, &src_uid, renamed.is_some(), &dest, &name, src);
+        self.settle_source(
+            &source,
+            &src_uid,
+            renamed.is_some(),
+            &dest,
+            &dest_uid,
+            &name,
+            src,
+        );
         self.settle_dest(&dest, &dest_uid, &name);
         drop(guards);
         if renamed.is_none()
@@ -429,22 +436,23 @@ impl Core {
     }
 
     /// Bring the source location level with a move Drive has made.
+    #[allow(clippy::too_many_arguments)]
     fn settle_source(
         &self,
         source: &Place,
         uid: &NodeUid,
         renamed: bool,
         dest: &Place,
+        dest_uid: &NodeUid,
         name: &str,
         src: &Path,
     ) {
         match source {
             Place::Mounted { core, rel } => {
-                // Forget it everywhere, so no mount keeps serving it under the
-                // old path. The content cache is keyed by uid and stays valid.
-                self.for_each_mount(|st, notify| {
-                    forget_and_notify(st, notify, uid);
-                });
+                // Moved everywhere, so no mount keeps serving it under the old
+                // path and its row stays (`docs/BUGS.md` B117). The content
+                // cache is keyed by uid and stays valid.
+                self.relocate_everywhere(uid, dest_uid, name);
                 core.invalidate_parent_listing(rel);
             }
             Place::Mirror { folder, rel } if renamed => {

crates/pdfs-fuse/src/upload.rs

@@ -244,8 +244,8 @@ impl Core {
             )
             .map_err(|e| CoreError::from_api(&e, &format!("create folder {name}")))?;
         let node = self
-            .fetch_node(&new_uid)
-            .map_err(|e| self.errno_error(e, "fetch node"))?;
+            .fetch_minted_node(&new_uid, name)
+            .ok_or_else(|| minted_but_unlisted(name))?;
         let mut st = self.state();
         let ino = st.intern(pino, node);
         if let Some(kids) = st.children.get_mut(&pino)