fix(fuse,core): count a new file's landing until its bytes are cached

A create that carries a file's bytes retired its op, then cached the
bytes and gave the tree the real uid. Nothing counted that window, so
`pdfs status` read empty while it ran, which B119 promises means the
file reads without the network.

A delete in that window found the placeholder uid in the tree and
trashed the node by its real uid. `queue_trash` asked the tree whether
that uid was open, found no entry, and dropped the cached bytes the open
handle reads. If the cache store ran at the same moment, its tag was
written after the eviction and never removed. CI caught that as a
stray .meta file.

The create's landing is now counted like an upload's, from the op's
retirement until the tree has the real uid. The delete and the replace
pass whether they found the file open. Opening the cache removes a tag
whose bytes are gone, which also covers a crash between an eviction's
two removals. The simulations wait for `Daemon::drained`, which counts
the landing, before they go offline.

B255 and B256 move to fixed: the account run checked both.

crates/pdfs-core/src/cache.rs

@@ -621,10 +621,20 @@ impl ContentCache {
             for entry in rd.flatten() {
                 let name = entry.file_name();
                 let Some(name) = name.to_str() else { continue };
+                // A `.meta` tags the blob named like it without the suffix. One
+                // whose blob is gone is what an eviction leaves between its two
+                // removals, by a crash or by a store of the same uid landing in
+                // between, and nothing else would ever remove it.
+                if let Some(blob) = name.strip_suffix(".meta") {
+                    if !dir.join(blob).exists() {
+                        let _ = std::fs::remove_file(entry.path());
+                    }
+                    continue;
+                }
                 // `.geom` is a block-geometry sidecar (see `GeometryMeta`), tens
                 // of bytes and not a cached block; like `.meta` it is neither
                 // budgeted nor evictable on its own.
-                if name.ends_with(".meta") || name.ends_with(".geom") {
+                if name.ends_with(".geom") {
                     continue;
                 }
                 // A `.tmp` is a staging file from a store that never completed
@@ -2250,6 +2260,28 @@ mod tests {
         assert!(total <= 8, "budget enforced after reconcile, total={total}");
     }
 
+    /// An eviction removes the blob, then its tag. A crash in between, or a
+    /// store of the same uid landing in between, left the tag on its own, and
+    /// nothing removed it.
+    #[test]
+    fn a_tag_without_its_blob_is_swept_at_open() {
+        let dir = TempDir::new();
+        let content = dir.path().join("content");
+        let pins = dir.path().join("pins.json");
+        let db = Arc::new(Db::open_in_memory().unwrap());
+        {
+            let c = ContentCache::open(content.clone(), pins.clone(), 0, db.clone()).unwrap();
+            c.store(&uid("kept"), 1, 4, b"kept").unwrap();
+            c.store(&uid("gone"), 1, 4, b"gone").unwrap();
+            std::fs::remove_file(c.blob_path(&uid("gone"))).unwrap();
+        }
+
+        let c = ContentCache::open(content, pins, 0, db).unwrap();
+        assert!(!c.meta_path(&uid("gone")).exists());
+        assert!(c.meta_path(&uid("kept")).exists());
+        assert_eq!(c.read_range(&uid("kept"), 1, 4, 0, 4).unwrap(), b"kept");
+    }
+
     #[test]
     fn stale_metadata_is_a_miss() {
         let (c, _d) = cache();

crates/pdfs-fuse/src/drain.rs

@@ -1281,6 +1281,13 @@ impl Core {
             sent: (&sent.0, &sent.1),
             landed: (&home.to_string(), &name),
         };
+        // Counted from the op's retirement until the bytes are cached and the
+        // tree has the real uid, as an upload is. Uncounted, the queue read
+        // empty while that still ran: a delete in between took an open file
+        // for closed and dropped what its handle reads, and a script that
+        // waited for an empty queue read the file before it was cached
+        // (`docs/BUGS.md` B257).
+        let _landing = LandingUpload::begin(&self.landing_uploads);
         let newer = match self.retire_create(op, uploaded, &landing, &local, &real)? {
             CreateRetired::Withdrawn { trash } => {
                 return self.trash_withdrawn_create(op, trash, &local, &real, &name);

crates/pdfs-fuse/src/filesystem.rs

@@ -2368,7 +2368,7 @@ impl ProtonFs {
         // regardless — the file is gone from their point of view the moment the
         // command returns. The same when the network goes
         // away under the remote call below.
-        let queue_it = |reply: ReplyEmpty| match self.core.queue_trash(&uid, name) {
+        let queue_it = |reply: ReplyEmpty| match self.core.queue_trash(&uid, name, open_now) {
             Ok(()) => reply.ok(),
             Err(e) => reply.error(e),
         };

crates/pdfs-fuse/src/lib.rs

@@ -3997,7 +3997,12 @@ impl Core {
     /// the cached ones, as when the trash is sent at once: a partial write
     /// fills its gaps from them, offline too (`docs/BUGS.md` B186). The final
     /// close or the trash landing, whichever comes last, lets go of them.
-    fn queue_trash(&self, uid: &NodeUid, name: &str) -> Result<(), Errno> {
+    ///
+    /// `open` is whether the caller found the file open. A create that landed
+    /// under a delete is trashed by its real uid before the tree has it, so
+    /// asking the tree alone took an open file for closed (`docs/BUGS.md`
+    /// B257).
+    fn queue_trash(&self, uid: &NodeUid, name: &str, open: bool) -> Result<(), Errno> {
         self.require_uid_writable(uid)?;
         // The bytes below are about to be discarded, and a drain worker may be
         // reading them onto the wire right now. Stop it first: with several
@@ -4012,7 +4017,7 @@ impl Core {
                 error!(%uid, error = %e, "queueing trash failed");
                 Errno::EIO
             })?;
-        let open_now = self.is_open_anywhere(uid);
+        let open_now = open || self.is_open_anywhere(uid);
         if open_now {
             let read_from = self.pending_blob(uid);
             self.release_dropped_below(
@@ -5381,9 +5386,12 @@ impl Core {
     /// `unlink` under an open handle. Its queued ops go now, or the drain would
     /// upload to the node after it was trashed (`docs/BUGS.md` B106).
     fn remove_replaced(&self, ino: u64, uid: &NodeUid, name: &str) -> Result<(), Errno> {
+        // Asked under the uid the tree has now: a landed create's real uid is
+        // not in the tree yet (`docs/BUGS.md` B257).
+        let open = self.is_open_anywhere(uid);
         let landed;
         let uid = if is_local_uid(uid) {
-            if self.is_open_anywhere(uid) {
+            if open {
                 self.withdraw_queued_ops(uid)?;
             } else {
                 self.discard_queued_ops(uid)?;
@@ -5408,7 +5416,7 @@ impl Core {
         // Local-first or offline, or the network goes away under the call:
         // queue the trash. The rename that follows queues as well, behind it.
         if !self.sends_inline() {
-            return self.queue_trash(uid, name);
+            return self.queue_trash(uid, name, open);
         }
         self.require_uid_writable(uid)?;
         if let Err(e) = self
@@ -5416,7 +5424,7 @@ impl Core {
             .and_then(batch::into_unit)
         {
             if self.lost_link(&e, "trash a replaced node") {
-                return self.queue_trash(uid, name);
+                return self.queue_trash(uid, name, open);
             }
             error!(%uid, name, error = %e, "trashing the node a rename replaces failed");
             self.log_activity(ActivityKind::Trash, name, e.to_string(), false);
@@ -9255,7 +9263,7 @@ mod tests {
         assert!(!replaced.contains("st.forget(uid)"));
         assert_before(
             replaced,
-            "if self.is_open_anywhere(uid)",
+            "let open = self.is_open_anywhere(uid)",
             "self.discard_queued_ops(uid)",
         );
         assert_before(replaced, "if open_now", "self.cache.evict(uid)");

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

@@ -130,6 +130,22 @@ impl Daemon {
         }
     }
 
+    /// Whether nothing is queued and no upload is still landing, as `pdfs
+    /// status` counts it. An empty [`Self::pending`] is not enough before a
+    /// read that needs the cache: the op goes before the uploaded bytes are
+    /// cached, so a test that went offline at once raced the drain.
+    pub(crate) fn drained(&self) -> bool {
+        self.pending().is_ok_and(|items| items.is_empty())
+            && matches!(
+                self.request(&Request::Status),
+                Ok(Response::Status {
+                    pending_uploads: 0,
+                    pending_changes: 0,
+                    ..
+                })
+            )
+    }
+
     /// Whether the mount believes Proton is reachable, as `pdfs status` shows
     /// it.
     pub(crate) fn online(&self) -> io::Result<bool> {
@@ -565,7 +581,7 @@ mod tests {
 
         std::fs::write(&path, b"0123456789").unwrap();
         assert!(wait_until(Duration::from_secs(30), || {
-            daemon.pending().is_ok_and(|items| items.is_empty())
+            daemon.drained()
                 && drive.tree().get("f.txt") == Some(&Entry::File(Arc::new(b"0123456789".to_vec())))
         }));
         daemon.client.set_online(false);
@@ -684,6 +700,55 @@ mod tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    #[test]
+    #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"]
+    fn a_new_file_deleted_while_open_as_its_create_lands_still_reads() {
+        // The create's op was retired and the tree still had the placeholder.
+        // The queue read empty, and the delete trashed the node by its real
+        // uid, which the tree did not have yet: it took the open file for
+        // closed and dropped the bytes its handle reads (B257).
+        use std::io::{Read, Seek, SeekFrom, Write};
+        let drive = FakeDrive::new();
+        let dir = scratch("delete-open-as-create-lands");
+        let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap();
+        let path = daemon.mountpoint.join("f.txt");
+
+        let landing = daemon.client.hold_next_landing_of_any();
+        std::fs::write(&path, b"0123456789").unwrap();
+        assert!(wait_until(Duration::from_secs(30), || landing.reached()));
+        assert!(!daemon.drained(), "the landing create is counted");
+        let mut file = std::fs::OpenOptions::new()
+            .read(true)
+            .write(true)
+            .open(&path)
+            .unwrap();
+        drop(landing);
+        assert!(wait_until(Duration::from_secs(30), || daemon.drained()));
+
+        daemon.client.set_online(false);
+        std::fs::remove_file(&path).unwrap();
+        file.seek(SeekFrom::End(0)).unwrap();
+        file.write_all(b"ab").unwrap();
+        let mut read = Vec::new();
+        file.seek(SeekFrom::Start(0)).unwrap();
+        file.read_to_end(&mut read).unwrap();
+        assert_eq!(read, b"0123456789ab");
+        drop(file);
+
+        daemon.client.set_online(true);
+        assert!(
+            wait_until(Duration::from_secs(30), || {
+                daemon.drained() && drive.tree().is_empty()
+            }),
+            "{:?} {:?}",
+            daemon.pending(),
+            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_restart_does_not_bring_back_the_write_of_a_file_deleted_while_open() {
@@ -698,7 +763,7 @@ mod tests {
 
         std::fs::write(&path, b"0123456789").unwrap();
         assert!(wait_until(Duration::from_secs(30), || {
-            daemon.pending().is_ok_and(|items| items.is_empty())
+            daemon.drained()
                 && drive.tree().get("f.txt") == Some(&Entry::File(Arc::new(b"0123456789".to_vec())))
         }));
         daemon.client.set_online(false);