fix(fuse): keep a move that lands while its create is read back (B161)

adopt_real_uid asked whether a move was queued only after reading the
node back. A move that landed in between left no op, and the answer still
had the old name, so the mount showed the file under it. Ask first.

The fake Drive can hold a read's answer (hold_answer_to_next_read) so the
test can land the move while the answer is on its way.

crates/pdfs-fuse/src/drain.rs

@@ -1587,15 +1587,17 @@ impl Core {
         real: &NodeUid,
         adopt: impl FnOnce(&Node),
     ) -> Result<(), Box<dyn std::error::Error>> {
+        // A rename made while the create was on the wire is queued against the
+        // real uid (`Db::finish_create`). Until it lands, the tree keeps the
+        // name and folder the user gave it. Asked before the read-back: a
+        // rename that lands while Drive answers leaves no op behind, and the
+        // answer may still have the old name (`docs/BUGS.md` B161).
+        let real_key = real.to_string();
+        let renaming = self.db.has_pending_op(&real_key, OP_RENAME)?;
         // Drive answers a create before it lists the new node.
         let mut node = self
             .read_back(real)
             .map_err(|e| self.errno_error(e, "fetch node"))?;
-        // A rename made while the create was on the wire is queued against the
-        // real uid (`Db::finish_create`). Until it lands, the tree keeps the
-        // name and folder the user gave it.
-        let real_key = real.to_string();
-        let renaming = self.db.has_pending_op(&real_key, OP_RENAME)?;
         // Before open handles are rebased onto the node, so one released in
         // between finds what `adopt` keeps (`docs/BUGS.md` B123).
         adopt(&node);

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

@@ -422,6 +422,78 @@ mod tests {
         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() {
+        // The move queued behind the create landed while the drain read the
+        // new node back. The answer still had the old name, and the drain no
+        // longer found the move queued, so the mount showed the old name in
+        // the new folder (B161).
+        let drive = FakeDrive::new();
+        let dir = scratch("moved-while-created");
+        let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap();
+        let mnt = daemon.mountpoint.clone();
+        std::fs::create_dir(mnt.join("x")).unwrap();
+        std::fs::create_dir(mnt.join("y")).unwrap();
+        assert!(
+            wait_until(Duration::from_secs(30), || {
+                daemon.pending().is_ok_and(|items| items.is_empty()) && drive.lookup("y").is_some()
+            }),
+            "{:?} {:?}",
+            daemon.pending(),
+            drive.tree()
+        );
+
+        let create = daemon.client.hold_reply_to_next_create();
+        std::fs::write(mnt.join("x/c.bin"), b"abc").unwrap();
+        assert!(wait_until(Duration::from_secs(30), || create.reached()));
+        std::fs::rename(mnt.join("x/c.bin"), mnt.join("y/d")).unwrap();
+        let uid = drive.lookup("x/c.bin").unwrap();
+        // Each read-back is answered in turn until the create is retired; the
+        // answer to the next one is held until the move has landed. A new
+        // folder wakes the drain for the move.
+        let mut held = create;
+        let adopting = loop {
+            let next = daemon.client.hold_answer_to_next_read(&uid);
+            drop(held);
+            assert!(wait_until(Duration::from_secs(30), || next.reached()));
+            if daemon
+                .pending()
+                .is_ok_and(|items| items.iter().all(|op| op.kind != "create"))
+            {
+                break next;
+            }
+            held = next;
+        };
+        std::fs::create_dir(mnt.join("z")).unwrap();
+        assert!(
+            wait_until(Duration::from_secs(10), || drive.lookup("y/d").is_some()),
+            "{:?} {:?}",
+            daemon.pending(),
+            drive.tree()
+        );
+        drop(adopting);
+
+        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("y/d").cloned() == file
+            }),
+            "{:?} {:?}",
+            daemon.pending(),
+            drive.tree()
+        );
+        let names: Vec<_> = std::fs::read_dir(mnt.join("y"))
+            .unwrap()
+            .map(|e| e.unwrap().file_name().into_string().unwrap())
+            .collect();
+        assert_eq!(names, ["d"]);
+
+        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

@@ -33,7 +33,8 @@
 //!
 //! A test can also hold the next read of one node on its way to Drive
 //! ([`FakeClient::hold_next_read`]), to make a change while the daemon waits
-//! for the answer.
+//! for the answer, or on its way back ([`FakeClient::hold_answer_to_next_read`]),
+//! so the daemon gets an answer a change made meanwhile has overtaken.
 
 use std::collections::{BTreeMap, HashMap};
 use std::io::{Read, Write};
@@ -330,6 +331,7 @@ impl FakeDrive {
             online: AtomicBool::new(true),
             requests: AtomicU64::new(0),
             hold: Mutex::new(None),
+            hold_answer: Mutex::new(None),
             drop_after_create: AtomicBool::new(false),
             lose_create_reply: AtomicBool::new(false),
             hold_create_reply: Mutex::new(None),
@@ -583,6 +585,8 @@ struct ClientInner {
     requests: AtomicU64,
     /// The read [`FakeClient::hold_next_read`] is waiting for.
     hold: Mutex<Option<Arc<HoldState>>>,
+    /// The read [`FakeClient::hold_answer_to_next_read`] is waiting for.
+    hold_answer: Mutex<Option<Arc<HoldState>>>,
     /// Set by [`FakeClient::drop_link_after_next_create`].
     drop_after_create: AtomicBool,
     /// Set by [`FakeClient::lose_reply_to_next_create`].
@@ -667,6 +671,19 @@ impl FakeClient {
         Held(state)
     }
 
+    /// Answer the next read of `uid` alone with what Drive holds now, then
+    /// hold the answer until it is dropped, as a slow link does. Only one
+    /// answer is held at a time.
+    pub(crate) fn hold_answer_to_next_read(&self, uid: &NodeUid) -> Held {
+        let state = Arc::new(HoldState {
+            uid: uid.clone(),
+            reached: AtomicBool::new(false),
+            released: AtomicBool::new(false),
+        });
+        *self.0.hold_answer.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 {
@@ -691,10 +708,10 @@ impl FakeClient {
         self.0.lose_revision_read_back.store(true, Ordering::SeqCst);
     }
 
-    /// Wait here if `uids` is the read a [`Held`] is for.
-    async fn wait_if_held(&self, uids: &[NodeUid]) {
+    /// Wait here if `uids` is the read a [`Held`] in `slot` is for.
+    async fn wait_if_held(&self, slot: &Mutex<Option<Arc<HoldState>>>, uids: &[NodeUid]) {
         let state = {
-            let mut hold = self.0.hold.lock();
+            let mut hold = slot.lock();
             match hold.as_ref() {
                 Some(state) if uids == std::slice::from_ref(&state.uid) => hold.take(),
                 _ => None,
@@ -928,7 +945,7 @@ impl DriveApi for FakeClient {
     }
 
     async fn enumerate_nodes(&self, uids: &[NodeUid]) -> Result<Vec<Node>> {
-        self.wait_if_held(uids).await;
+        self.wait_if_held(&self.0.hold, uids).await;
         {
             let mut lose = self.0.lose_read.lock();
             if lose
@@ -949,6 +966,7 @@ impl DriveApi for FakeClient {
                     .collect())
             })
             .await?;
+        self.wait_if_held(&self.0.hold_answer, uids).await;
         self.cache(&nodes);
         Ok(nodes)
     }
@@ -1645,6 +1663,29 @@ mod tests {
         });
     }
 
+    #[test]
+    fn a_held_answer_is_what_drive_held_when_read() {
+        rt().block_on(async {
+            let drive = FakeDrive::new();
+            let client = drive.client(1, Faults::none());
+            let root = drive.root();
+            let file = client.upload_file(&root, "a", "", b"1").await.unwrap();
+            let held = client.hold_answer_to_next_read(&file);
+            let read = tokio::spawn({
+                let client = client.clone();
+                let file = file.clone();
+                async move { client.enumerate_nodes(std::slice::from_ref(&file)).await }
+            });
+            tokio::time::sleep(Duration::from_millis(50)).await;
+            assert!(held.reached());
+            assert!(!read.is_finished());
+            drive.device().trash("a");
+            drop(held);
+            let nodes = read.await.unwrap().unwrap();
+            assert!(!nodes[0].trashed);
+        });
+    }
+
     #[test]
     fn a_rename_lags_only_by_the_update_lag() {
         rt().block_on(async {