fix(fuse): list a file whose create lands while its folder is listed (B166)

A listing from Drive was overlaid with the queue as it was after Drive
answered. A create that landed in between was in neither, and a wide
folder on Wi-Fi dropped files that were never unlinked. The queue is now
read before Drive is asked too; a node that left it meanwhile is listed
as the DB has it.

crates/pdfs-fuse/src/lib.rs

@@ -2340,6 +2340,39 @@ impl Core {
         Ok(())
     }
 
+    /// The nodes of `before`, queued in `folder_uid` when its listing was asked
+    /// of Drive, that are no longer queued in `after`, as the DB has them now.
+    ///
+    /// Their create landed while the listing was on the wire, after Drive took
+    /// it, and so is in neither. One unlinked or moved away meanwhile is not
+    /// under the folder by that name any more and stays out.
+    fn landed_meanwhile(
+        &self,
+        folder_uid: &NodeUid,
+        before: Vec<Node>,
+        after: &[Node],
+    ) -> Vec<Node> {
+        let still: HashSet<&str> = after.iter().map(|node| node.name.as_str()).collect();
+        let landed: HashSet<String> = before
+            .into_iter()
+            .map(|node| node.name)
+            .filter(|name| !still.contains(name.as_str()))
+            .collect();
+        if landed.is_empty() {
+            return Vec::new();
+        }
+        match self.db.known_children(folder_uid) {
+            Ok(known) => known
+                .into_iter()
+                .filter(|node| landed.contains(&node.name))
+                .collect(),
+            Err(e) => {
+                warn!(%folder_uid, error = %e, "db known_children failed");
+                Vec::new()
+            }
+        }
+    }
+
     /// Serve a listing without the network: the one already in memory if any,
     /// else whatever children the DB still knows. The fallback only answers
     /// until the link is back, when [`Core::relist_offline_listings`] drops it:
@@ -2501,6 +2534,12 @@ impl Core {
 
         let refresh_generation =
             refresh_foreign.then(|| self.shared_generation.load(Ordering::SeqCst));
+        // What is queued before Drive is asked: a create that lands while the
+        // listing is on the wire is in neither Drive's answer nor the queue read
+        // after it (`docs/BUGS.md` B166).
+        let queued_before = (!foreign)
+            .then(|| self.db.queued_children(&folder_uid).ok())
+            .flatten();
         let uids = match self
             .rt
             .block_on(self.drive.enumerate_folder_children_node_uids(&folder_uid))
@@ -2537,7 +2576,12 @@ impl Core {
         // A shared folder's listing is published as Drive has it.
         if !foreign {
             match self.db.queued_children(&folder_uid) {
-                Ok((queued, gone)) => overlay_queued(&mut nodes, queued, &gone),
+                Ok((mut queued, gone)) => {
+                    if let Some((before, _)) = queued_before {
+                        queued.extend(self.landed_meanwhile(&folder_uid, before, &queued));
+                    }
+                    overlay_queued(&mut nodes, queued, &gone)
+                }
                 Err(e) => warn!(%folder_uid, error = %e, "db queued_children failed"),
             }
         }

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

@@ -1084,4 +1084,58 @@ mod tests {
         eprintln!("lan:  syscalls {lan_syscalls:?}, drain {lan_drain:?}");
         eprintln!("wifi: syscalls {wifi_syscalls:?}, drain {wifi_drain:?}");
     }
+
+    #[test]
+    #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"]
+    fn half_of_a_wide_folder_unlinked_at_once_leaves_the_other_half() {
+        // The acceptance suite's wide directory on Wi-Fi. A create that landed
+        // while the folder's listing was on the wire was in neither Drive's
+        // answer nor the queue read after it, and a file that was never
+        // unlinked dropped out of the listing until the next one (B166).
+        let drive = FakeDrive::new();
+        let dir = scratch("wide-unlink");
+        let daemon = Daemon::start(&dir, drive.client(1, Faults::wifi())).unwrap();
+        let root = daemon.mountpoint.join("wide");
+        std::fs::create_dir(&root).unwrap();
+        let names: Vec<String> = (0..128).map(|i| format!("entry-{i:04}.txt")).collect();
+        let in_parallel = |op: &(dyn Fn(&str) + Sync), names: &[String]| {
+            std::thread::scope(|scope| {
+                for chunk in names.chunks(names.len().div_ceil(8)) {
+                    scope.spawn(move || chunk.iter().for_each(|name| op(name)));
+                }
+            });
+        };
+        in_parallel(
+            &|name| {
+                let mut file = std::fs::File::create(root.join(name)).unwrap();
+                std::io::Write::write_all(&mut file, name.as_bytes()).unwrap();
+                file.sync_all().unwrap();
+            },
+            &names,
+        );
+        let listed = |root: &std::path::Path| {
+            let mut names: Vec<String> = std::fs::read_dir(root)
+                .unwrap()
+                .map(|e| e.unwrap().file_name().into_string().unwrap())
+                .collect();
+            names.sort();
+            names
+        };
+        assert_eq!(listed(&root), names);
+        in_parallel(
+            &|name| std::fs::remove_file(root.join(name)).unwrap(),
+            &names[..64],
+        );
+        assert_eq!(listed(&root), names[64..]);
+        let mut settled = None;
+        while settled.is_none_or(|at: Instant| at.elapsed() < crate::POLL_INTERVAL * 2) {
+            assert_eq!(listed(&root), names[64..]);
+            if settled.is_none() && daemon.pending().is_ok_and(|ops| ops.is_empty()) {
+                settled = Some(Instant::now());
+            }
+            std::thread::sleep(Duration::from_millis(50));
+        }
+        assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown)));
+        let _ = std::fs::remove_dir_all(&dir);
+    }
 }