fix(fuse): a write opened as an upload lands takes the landed base (B170)

An upload that landed let go of its blob before the tree had its node. A
write open in between took the base it sampled at its start, the revision
the upload replaced, found none of its bytes cached, and queued an
incomplete revision; the next write open failed with EIO.

The drain keeps the landing node aside until the tree has it, and a write
open with no queued revision takes its base from that node, or from the
tree as it is when the handle goes in. The fake Drive can hold the drain
in that window (`hold_next_landing`).

crates/pdfs-fuse/src/drain.rs

@@ -2365,14 +2365,29 @@ impl Core {
                     warn!(%uid, error = %e, "persisting our sealed revision failed");
                 }
             }
+            // A write open from here until the tree has the node takes it as
+            // its base: the blob it would have copied is let go of below, and
+            // the tree's stamp names the revision this one replaced
+            // (docs/BUGS.md B170).
+            self.landed.lock().insert(uid.clone(), node.clone());
             self.rebase_open_writes(uid, &node);
         }
         self.release_pending(uid, blob);
+        #[cfg(test)]
+        self.drive.landing(uid);
+        self.place_landed(uid, &node);
+        self.landed.lock().remove(uid);
+        Some(node)
+    }
+
+    /// Put a node whose upload just landed into the tree, unless a newer write
+    /// is queued for it.
+    fn place_landed(&self, uid: &NodeUid, node: &Node) {
         // Ordered so that a write queued *during* the fetch above is still
         // 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) {
+        if self.rebaseline_pending(uid, node) {
-            return Some(node);
+            return;
         }
         // A rename queued behind the upload has not reached Drive yet, and
         // until it lands the tree keeps the name and folder the user gave the
@@ -2402,7 +2417,6 @@ impl Core {
             }
             st.intern(parent, node);
         });
-        Some(node)
     }
 
     /// Rebase every write handle open on `uid` onto `node`, a revision this

crates/pdfs-fuse/src/drive.rs

@@ -153,6 +153,12 @@ pub(crate) trait DriveApi: Send + Sync {
 
     /// The account's storage, across every Proton product.
     async fn quota(&self) -> Result<Quota>;
+
+    /// Called by the drain once it has let go of the blob of an upload to
+    /// `uid` that landed, before the tree has the node, so a simulation can
+    /// hold it there. Blocks the drain worker it is called on.
+    #[cfg(test)]
+    fn landing(&self, _uid: &NodeUid) {}
 }
 
 /// The generic signatures of the client, over the object-safe ones above.

crates/pdfs-fuse/src/filesystem.rs

@@ -1217,7 +1217,7 @@ impl ProtonFs {
     /// open.
     fn serve_open(&self, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) {
         let write_requested = flags.acc_mode() != OpenAccMode::O_RDONLY;
-        let (uid, base_mtime, base_size, provisional, base_revision_id) = {
+        let (uid, mut base_mtime, mut base_size, mut provisional, mut base_revision_id) = {
             let mut st = self.core.state();
             match st.entries.get_mut(&ino.0) {
                 Some(e) if e.node.is_file() => {
@@ -1283,22 +1283,24 @@ impl ProtonFs {
         // `reply` is consumed by whichever arm answers, so the loop yields an
         // outcome and the answer is sent once, after it.
         let outcome = 'install: {
-            // The handle starts at the base's size. A provisional one is the
-            // ciphertext size, so the file would grow by the encryption overhead
-            // and reading the base would fail (bugs.md B100). A queued revision
-            // stamps a real size, so this only goes to the network for a file
-            // the listing has not sized yet.
-            let base_size = match provisional && self.core.pending_blob(&uid).is_none() {
-                false => base_size,
-                true => match self.core.resolve_provisional_size(ino.0, &uid) {
-                    Some(size) => size,
-                    None => {
-                        error!(%uid, "could not settle the size of the base for a write open");
-                        break 'install Err(Errno::EIO);
-                    }
-                },
-            };
             for attempt in 0..OPEN_BASE_ATTEMPTS {
+                // The handle starts at the base's size. A provisional one is the
+                // ciphertext size, so the file would grow by the encryption
+                // overhead and reading the base would fail (bugs.md B100). A
+                // queued revision stamps a real size, so this only goes to the
+                // network for a file the listing has not sized yet.
+                if provisional && self.core.pending_blob(&uid).is_none() {
+                    match self.core.resolve_provisional_size(ino.0, &uid) {
+                        Some(size) => {
+                            base_size = size;
+                            provisional = false;
+                        }
+                        None => {
+                            error!(%uid, "could not settle the size of the base for a write open");
+                            break 'install Err(Errno::EIO);
+                        }
+                    }
+                }
                 let pending_base = self.core.pending.lock().get(&uid).cloned();
                 if pending_base.as_ref().is_some_and(|p| !p.meta.complete) {
                     error!(%uid, "refusing write over incomplete queued revision");
@@ -1332,17 +1334,49 @@ impl ProtonFs {
                 // a lock-tight comparison — against the multi-gigabyte copy
                 // above, which is the window that actually loses data.
                 let current = self.core.pending_blob(&uid);
+                let landed = self.core.landed.lock().get(&uid).cloned();
                 let mut st = self.core.state();
                 // Only a handle this open is about to *create* can carry a stale
                 // base; if one already exists, that handle is authoritative and
                 // the scratch prepared above is discarded either way.
                 let creating = !st.active_writes.contains_key(&ino.0);
-                if creating && current != pending_base.map(|p| p.path) {
+                if creating && current.as_ref() != pending_base.as_ref().map(|p| &p.path) {
                     drop(st);
                     let _ = std::fs::remove_file(&path);
                     warn!(%uid, attempt, "queued revision changed while opening for write; retrying");
                     continue;
                 }
+                // With no queued revision, the base's bytes come from the cache,
+                // keyed by the base's mtime and size. The ones sampled above can
+                // name a revision this daemon has since replaced: an upload that
+                // lands lets go of its blob before the tree has its node, and
+                // the cache holds its bytes under the node only. So the base is
+                // the landing node while there is one, and the tree's otherwise
+                // (docs/BUGS.md B170).
+                if creating && pending_base.is_none() {
+                    let now = landed.or_else(|| st.entries.get(&ino.0).map(|e| e.node.clone()));
+                    if let Some(node) = now
+                        && (node.modification_time != base_mtime
+                            || node_revision_id(&node) != base_revision_id)
+                    {
+                        base_mtime = node.modification_time;
+                        base_size = node_size(&node);
+                        base_revision_id = node_revision_id(&node);
+                        provisional = matches!(
+                            &node.kind,
+                            NodeKind::File {
+                                claimed_size: None,
+                                ..
+                            }
+                        );
+                        debug!(%uid, base_mtime, base_size, "write open took the base the file has now");
+                        if provisional {
+                            drop(st);
+                            let _ = std::fs::remove_file(&path);
+                            continue;
+                        }
+                    }
+                }
                 let fh = st.next_fh;
                 st.next_fh += 1;
                 let aw = st.active_writes.entry(ino.0).or_insert_with(|| {

crates/pdfs-fuse/src/lib.rs

@@ -579,6 +579,9 @@ struct Core {
     /// The SHA-1 of the last revision this daemon uploaded to a file and could
     /// not read back, so its revision id is not known (`docs/BUGS.md` B157).
     sealed_unread: Arc<Mutex<HashMap<NodeUid, String>>>,
+    /// The node an upload has just put on Drive, from when the drain lets go
+    /// of its blob until the tree has the node (`docs/BUGS.md` B170).
+    landed: Arc<Mutex<HashMap<NodeUid, Node>>>,
     /// Creates that landed but could not be read back, by the uid Drive gave
     /// them, with the placeholder's. The drain adopts one before it sends
     /// anything else for it.

crates/pdfs-fuse/src/mount.rs

@@ -436,6 +436,7 @@ pub(crate) fn mount_with(
         creating: Arc::new(Mutex::new(HashMap::new())),
         create_sent: Arc::new(Mutex::new(HashMap::new())),
         sealed_unread: Arc::new(Mutex::new(HashMap::new())),
+        landed: Arc::new(Mutex::new(HashMap::new())),
         unadopted: Arc::new(Mutex::new(HashMap::new())),
         drain_wake: Arc::new((Mutex::new(false), Condvar::new())),
         shutdown: Arc::new(crate::shutdown::Shutdown::default()),

crates/pdfs-fuse/src/sim/daemon.rs

@@ -422,6 +422,47 @@ mod tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    #[test]
+    #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"]
+    fn a_write_opened_as_an_upload_lands_takes_the_landed_revision_as_its_base() {
+        // An upload landed and the drain let go of its blob before the tree had
+        // the node. A partial write opened in between took the revision the
+        // upload replaced as its base, found none of its bytes cached, and
+        // queued an incomplete revision: the next write open failed with EIO
+        // (B170).
+        let drive = FakeDrive::new();
+        let dir = scratch("write-opened-as-upload-lands");
+        let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap();
+        let path = daemon.mountpoint.join("f.txt");
+        let file = |bytes: &[u8]| Some(Entry::File(Arc::new(bytes.to_vec())));
+
+        std::fs::write(&path, b"first").unwrap();
+        assert!(wait_until(Duration::from_secs(30), || {
+            daemon.pending().is_ok_and(|items| items.is_empty())
+                && drive.tree().get("f.txt").cloned() == file(b"first")
+        }));
+        let uid = drive.lookup("f.txt").unwrap();
+        let landing = daemon.client.hold_next_landing(&uid);
+        std::fs::write(&path, b"0123456789").unwrap();
+        assert!(wait_until(Duration::from_secs(30), || landing.reached()));
+        write_at(&path, 2, b"ab").unwrap();
+        drop(landing);
+        write_at(&path, 6, b"cd").unwrap();
+        assert!(
+            wait_until(Duration::from_secs(30), || {
+                daemon.pending().is_ok_and(|items| items.is_empty())
+                    && drive.tree().get("f.txt").cloned() == file(b"01ab45cd89")
+            }),
+            "{:?} {:?}",
+            daemon.pending(),
+            drive.tree()
+        );
+        assert_eq!(drive.tree().len(), 1, "{:?}", drive.tree());
+
+        assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown)));
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
     #[test]
     #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"]
     fn a_file_moved_while_its_create_was_on_the_wire_keeps_its_new_name() {

crates/pdfs-fuse/src/sim/fake_drive.rs

@@ -338,6 +338,7 @@ impl FakeDrive {
             hold: Mutex::new(None),
             hold_answer: Mutex::new(None),
             hold_trash: Mutex::new(None),
+            hold_landing: Mutex::new(None),
             drop_after_create: AtomicBool::new(false),
             lose_create_reply: AtomicBool::new(false),
             hold_create_reply: Mutex::new(None),
@@ -615,6 +616,8 @@ struct ClientInner {
     hold_answer: Mutex<Option<Arc<HoldState>>>,
     /// The trash [`FakeClient::hold_next_trash`] is waiting for.
     hold_trash: Mutex<Option<Arc<HoldState>>>,
+    /// The landing [`FakeClient::hold_next_landing`] is waiting for.
+    hold_landing: Mutex<Option<Arc<HoldState>>>,
     /// Set by [`FakeClient::drop_link_after_next_create`].
     drop_after_create: AtomicBool,
     /// Set by [`FakeClient::lose_reply_to_next_create`].
@@ -752,6 +755,19 @@ impl FakeClient {
         Held(state)
     }
 
+    /// Hold the drain once the next upload to `uid` has landed and its blob
+    /// is let go of, before the tree has the node (`DriveApi::landing`), until
+    /// the answer is dropped.
+    pub(crate) fn hold_next_landing(&self, uid: &NodeUid) -> Held {
+        let state = Arc::new(HoldState {
+            uid: uid.clone(),
+            reached: AtomicBool::new(false),
+            released: AtomicBool::new(false),
+        });
+        *self.0.hold_landing.lock() = Some(state.clone());
+        Held(state)
+    }
+
     /// Make the next file create on Drive, then hold its answer until the
     /// answer is dropped, as a slow link does.
     pub(crate) fn hold_reply_to_next_create(&self) -> Held {
@@ -1465,6 +1481,21 @@ impl DriveApi for FakeClient {
         outcome_stream(self.batch(uids, delete_one))
     }
 
+    fn landing(&self, uid: &NodeUid) {
+        let state = {
+            let mut hold = self.0.hold_landing.lock();
+            match hold.as_ref() {
+                Some(state) if state.uid == *uid => hold.take(),
+                _ => None,
+            }
+        };
+        let Some(state) = state else { return };
+        state.reached.store(true, Ordering::SeqCst);
+        while !state.released.load(Ordering::SeqCst) {
+            std::thread::sleep(Duration::from_millis(5));
+        }
+    }
+
     async fn quota(&self) -> Result<Quota> {
         self.request(false, |server| {
             Ok(Quota {

docs/BUGS.md

@@ -12,6 +12,37 @@ Conventions:
 
 ---
 
+## B170 — A write opened as an upload lands fails the next write with `EIO`
+
+**Status:** Fixed (unverified).
+**Found:** 2026-10-04, by CI on 9bbc994 and by three parallel copies of
+`one_client_on_a_good_link` (seed 4; seed 5 in the local suite). A write at an offset was
+refused with `EIO`, and the daemon logged "refusing write over incomplete queued revision".
+Just before, an open had retried with "queued revision changed while opening for write", and
+the queued write it then made was incomplete.
+
+**Where:** `serve_open` in `crates/pdfs-fuse/src/filesystem.rs`, and `Core::refresh_after_upload`
+in `crates/pdfs-fuse/src/drain.rs`.
+
+**Cause.** When an upload lands, the drain caches its bytes under the node as Drive has it,
+rebases the open write handles onto it, lets go of the blob, and then puts the node in the
+tree. A write open in between found no blob to copy and took the base it had sampled at its
+start: the mtime and size the tree had before the upload. The release of its partial write
+filled the gaps from the cache under that key, found nothing, and queued an incomplete
+revision, and a write open over one is refused (B123). Under load, the open retried because
+of the landing and met that window.
+
+**Fix.** From before the drain lets go of the blob until the tree has the node, the landing
+node is kept aside. A write open with no queued revision takes its base from that node, or
+else from the tree as it is when the handle goes in, not from its first sample.
+
+**Test:** `a_write_opened_as_an_upload_lands_takes_the_landed_revision_as_its_base` in
+`sim/daemon.rs` holds the drain in that window with the fake Drive's new `hold_next_landing`,
+writes at an offset, and writes again after it. Without the fix the second write fails with
+`EIO`.
+
+---
+
 ## B169 — A stop lets a mirror pass go on until the runtime ends under it
 
 **Status:** Fixed.

docs/CHANGELOG.md

@@ -68,6 +68,8 @@ by its local id, in the queue, the stored tree and pins alike.
   to upload them to the top of your Drive instead (B168).
 - **Stopping the daemon during a folder sync no longer waits for the sync.** The uploads in
   progress finish, and the rest waits for the next start (B169).
+- **Writing to a file again just as its upload finishes no longer fails.** A write right after
+  it failed with "Input/output error" (B170).
 - **A file renamed just as its upload finished keeps its new name.** The mount no longer showed
   the old name, and a second rename no longer failed with "No such file or directory" (B152).
 - **A change that waits for a name to be freed no longer waits seconds after it is.** A rename or

docs/MILESTONE-3.0.0-STATUS.md

@@ -9,7 +9,7 @@ Tracks `docs/MILESTONE-3.0.0.md`. Update it with every commit that moves a phase
 - **Phase 2**: local ids, inode = local id, one tree for every mount (2.10.0).
 - **Phase 3, first part**: mkdir, create, rename, unlink and rmdir on the mount are recorded
   locally and sent from the queue; `"local_first": false` brings back the old path (5e0f30c).
-- Fixes from the simulation and account runs since then: B127 and B141 to B169.
+- Fixes from the simulation and account runs since then: B127 and B141 to B170.
 - **Sync issues, first cut** (§6.2): refusals sorted into issues, shown at once in the app,
   tray and CLI; Export button and `pdfs sync export`; `pdfs sync issues`.
 
@@ -110,7 +110,9 @@ Tracks `docs/MILESTONE-3.0.0.md`. Update it with every commit that moves a phase
          Looking into it found the start spending 1.7 s rebuilding listings (B167, fixed).
          Its journal check caught a file removed with its folder while uploading being
          sent again, to the root (B168, fixed), and a stop that let a mirror pass go on
-         until the runtime ended under it (B169, fixed).
+         until the runtime ended under it (B169, fixed). CI on 9bbc994 failed
+         `one_client_on_a_good_link`: a write opened as an upload landed took the revision
+         it replaced as its base, and the next write failed with `EIO` (B170, fixed).
 5. **Release**: 2.10.0 is tagged; the 2.9.0 tag still waits for the user's go-ahead.
 
 ## After 3.0.0