fix(fuse): never adopt a node we trashed as an unanswered create's twin (B162)

A folder removed and made again met the old folder while its trash was on
the wire and adopted it, so the new folder's contents went into the old
one and were trashed with it. The twin search now skips hidden nodes, as
listings do.

The fake Drive can hold a trash (hold_next_trash).

crates/pdfs-fuse/src/drain.rs

@@ -1337,9 +1337,14 @@ impl Core {
         }
         let uids = self.block_on_bounded(self.drive.enumerate_folder_children_node_uids(parent))?;
         let children = self.block_on_bounded(self.drive.enumerate_nodes_light(&uids))?;
+        // A node this mount trashed is not one an unanswered create made, even
+        // while its trash is on the wire or a listing still shows it. Taking a
+        // folder removed and made again put the new one's contents into the
+        // old one, and its trash took them along (docs/BUGS.md B162).
+        let hidden = self.hidden.lock().clone();
         let Some(twin) = children
             .into_iter()
-            .find(|node| node.name == name && !node.trashed)
+            .find(|node| node.name == name && !node.trashed && !hidden.contains(&node.uid))
         else {
             return Ok(None);
         };

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

@@ -494,6 +494,56 @@ mod tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    #[test]
+    #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"]
+    fn a_folder_made_again_while_the_old_one_is_trashed_is_a_new_folder() {
+        // A folder removed and made again at once: the new one's create met
+        // the old folder, its trash still on the wire, and took it for the
+        // node an unanswered create of its own had made. What went into the
+        // new folder went into the old one, and the trash took it all along
+        // (B162).
+        let drive = FakeDrive::new();
+        let dir = scratch("remade-while-trashed");
+        let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap();
+        let mnt = daemon.mountpoint.clone();
+        std::fs::create_dir(mnt.join("z")).unwrap();
+        assert!(wait_until(Duration::from_secs(30), || {
+            daemon.pending().is_ok_and(|items| items.is_empty()) && drive.lookup("z").is_some()
+        }));
+        let old = drive.lookup("z").unwrap();
+
+        let trash = daemon.client.hold_next_trash(&old);
+        std::fs::remove_dir(mnt.join("z")).unwrap();
+        assert!(wait_until(Duration::from_secs(30), || trash.reached()));
+        std::fs::create_dir(mnt.join("z")).unwrap();
+        std::fs::write(mnt.join("z/f"), b"abc").unwrap();
+        // Sent, and either waiting for the name or landed.
+        assert!(wait_until(Duration::from_secs(30), || {
+            daemon.pending().is_ok_and(|items| {
+                !items
+                    .iter()
+                    .any(|op| op.kind == "mkdir" && op.attempts == 0)
+            })
+        }));
+        drop(trash);
+
+        let file = Some(Entry::File(Arc::new(b"abc".to_vec())));
+        assert!(
+            wait_until(Duration::from_secs(30), || {
+                daemon.pending().is_ok_and(|items| items.is_empty())
+                    && drive.tree().get("z/f").cloned() == file
+            }),
+            "{:?} {:?}",
+            daemon.pending(),
+            drive.tree()
+        );
+        assert_ne!(drive.lookup("z"), Some(old));
+        assert_eq!(std::fs::read(mnt.join("z/f")).unwrap(), b"abc");
+
+        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 the_echo_of_our_own_rename_keeps_the_cached_bytes() {

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

@@ -332,6 +332,7 @@ impl FakeDrive {
             requests: AtomicU64::new(0),
             hold: Mutex::new(None),
             hold_answer: Mutex::new(None),
+            hold_trash: Mutex::new(None),
             drop_after_create: AtomicBool::new(false),
             lose_create_reply: AtomicBool::new(false),
             hold_create_reply: Mutex::new(None),
@@ -587,6 +588,8 @@ struct ClientInner {
     hold: Mutex<Option<Arc<HoldState>>>,
     /// The read [`FakeClient::hold_answer_to_next_read`] is waiting for.
     hold_answer: Mutex<Option<Arc<HoldState>>>,
+    /// The trash [`FakeClient::hold_next_trash`] is waiting for.
+    hold_trash: Mutex<Option<Arc<HoldState>>>,
     /// Set by [`FakeClient::drop_link_after_next_create`].
     drop_after_create: AtomicBool,
     /// Set by [`FakeClient::lose_reply_to_next_create`].
@@ -684,6 +687,18 @@ impl FakeClient {
         Held(state)
     }
 
+    /// Hold the next trash of `uid` alone, before it reaches Drive, until the
+    /// answer is dropped.
+    pub(crate) fn hold_next_trash(&self, uid: &NodeUid) -> Held {
+        let state = Arc::new(HoldState {
+            uid: uid.clone(),
+            reached: AtomicBool::new(false),
+            released: AtomicBool::new(false),
+        });
+        *self.0.hold_trash.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 {
@@ -1313,6 +1328,7 @@ impl DriveApi for FakeClient {
     }
 
     async fn trash_nodes(&self, uids: &[NodeUid]) -> Result<NodeOutcomes> {
+        self.wait_if_held(&self.0.hold_trash, uids).await;
         self.batch(uids, trash_one).await
     }