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) {
-            return Some(node);
+        if self.rebaseline_pending(uid, 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 {