fix(fuse): wait for a create on the wire that holds the name (B154); a write whose answer was lost is not a conflict (B155)

B154 is a regression from the B153 fix: after a failure on a held name the
op woke the ops waiting for it, but the database cannot see a create on the
wire, so the op was due again at once and spent its waits in milliseconds.
No wake while another create on the wire holds the name.

B155: a revision upload that reached Drive but lost its answer left the
file at a revision this daemon never heard of, and the retry kept the write
as a conflict copy of itself. A complete write whose bytes the remote holds,
by size and SHA-1, is sent again instead.

The fake Drive gains hold_reply_to_next_create and
lose_reply_to_next_revision for the two new sim tests.

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

@@ -535,6 +535,84 @@ 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_made_under_the_name_of_a_create_on_the_wire_waits_for_it() {
+        // A file renamed while its create was on the wire, and a new one made
+        // under its old name. The new file's create found the name held by the
+        // first and woke itself again after every try: it spent its waits in
+        // milliseconds and landed under a conflict name, which the mount then
+        // showed instead of the user's (B154).
+        let drive = FakeDrive::new();
+        let dir = scratch("create-waits-for-create");
+        let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap();
+        let mount = &daemon.mountpoint;
+        let file = |bytes: &[u8]| Some(Entry::File(Arc::new(bytes.to_vec())));
+
+        let held = daemon.client.hold_reply_to_next_create();
+        std::fs::write(mount.join("c.bin"), b"first").unwrap();
+        assert!(wait_until(Duration::from_secs(30), || held.reached()));
+        std::fs::rename(mount.join("c.bin"), mount.join("d")).unwrap();
+        std::fs::write(mount.join("c.bin"), b"second").unwrap();
+        assert!(wait_until(Duration::from_secs(30), || {
+            daemon
+                .pending()
+                .is_ok_and(|items| items.iter().any(|op| op.attempts > 0))
+        }));
+        // Long enough for the waits to run out, were they not waits.
+        std::thread::sleep(Duration::from_millis(300));
+        drop(held);
+        assert!(
+            wait_until(Duration::from_secs(30), || {
+                daemon.pending().is_ok_and(|items| items.is_empty())
+                    && drive.tree().get("d").cloned() == file(b"first")
+                    && drive.tree().get("c.bin").cloned() == file(b"second")
+            }),
+            "{:?} {:?}",
+            daemon.pending(),
+            drive.tree()
+        );
+        assert_eq!(drive.tree().len(), 2, "{:?}", drive.tree());
+        assert_eq!(std::fs::read(mount.join("c.bin")).unwrap(), b"second");
+
+        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_write_whose_answer_was_lost_lands_once() {
+        // Drive made the revision, but its answer was lost. The retry found
+        // the file at a revision it did not know and kept the write as a
+        // conflict copy of itself (B155).
+        let drive = FakeDrive::new();
+        let dir = scratch("lost-revision-reply");
+        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")
+        }));
+        daemon.client.lose_reply_to_next_revision();
+        std::fs::write(&path, b"second").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"second")
+            }),
+            "{:?} {:?}",
+            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_pdfs_rename_after_a_queued_rename_is_not_undone() {

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

@@ -332,6 +332,8 @@ impl FakeDrive {
             hold: Mutex::new(None),
             drop_after_create: AtomicBool::new(false),
             lose_create_reply: AtomicBool::new(false),
+            hold_create_reply: Mutex::new(None),
+            lose_revision_reply: AtomicBool::new(false),
         }))
     }
 
@@ -583,6 +585,10 @@ struct ClientInner {
     drop_after_create: AtomicBool,
     /// Set by [`FakeClient::lose_reply_to_next_create`].
     lose_create_reply: AtomicBool,
+    /// The create [`FakeClient::hold_reply_to_next_create`] is waiting for.
+    hold_create_reply: Mutex<Option<Arc<HoldState>>>,
+    /// Set by [`FakeClient::lose_reply_to_next_revision`].
+    lose_revision_reply: AtomicBool,
 }
 
 struct HoldState {
@@ -591,8 +597,17 @@ struct HoldState {
     released: AtomicBool,
 }
 
-/// A read held on its way to Drive, from [`FakeClient::hold_next_read`]. It
-/// goes on when this is dropped.
+/// Wait while `state` holds a call.
+async fn wait_until_released(state: &HoldState) {
+    state.reached.store(true, Ordering::SeqCst);
+    while !state.released.load(Ordering::SeqCst) {
+        tokio::time::sleep(Duration::from_millis(5)).await;
+    }
+}
+
+/// A call held on its way to or from Drive, from [`FakeClient::hold_next_read`]
+/// or [`FakeClient::hold_reply_to_next_create`]. It goes on when this is
+/// dropped.
 pub(crate) struct Held(Arc<HoldState>);
 
 impl Held {
@@ -646,6 +661,23 @@ impl FakeClient {
         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 {
+        let state = Arc::new(HoldState {
+            uid: self.0.drive.root(),
+            reached: AtomicBool::new(false),
+            released: AtomicBool::new(false),
+        });
+        *self.0.hold_create_reply.lock() = Some(state.clone());
+        Held(state)
+    }
+
+    /// Make the next revision of a file on Drive and lose its answer.
+    pub(crate) fn lose_reply_to_next_revision(&self) {
+        self.0.lose_revision_reply.store(true, Ordering::SeqCst);
+    }
+
     /// Wait here if `uids` is the read a [`Held`] is for.
     async fn wait_if_held(&self, uids: &[NodeUid]) {
         let state = {
@@ -656,10 +688,7 @@ impl FakeClient {
             }
         };
         let Some(state) = state else { return };
-        state.reached.store(true, Ordering::SeqCst);
-        while !state.released.load(Ordering::SeqCst) {
-            tokio::time::sleep(Duration::from_millis(5)).await;
-        }
+        wait_until_released(&state).await;
     }
 
     /// Requests made through this client so far.
@@ -1073,6 +1102,12 @@ impl DriveApi for FakeClient {
         if made.is_ok() && self.0.lose_create_reply.swap(false, Ordering::SeqCst) {
             return Err(no_answer());
         }
+        let held = self.0.hold_create_reply.lock().take();
+        if made.is_ok()
+            && let Some(state) = held
+        {
+            wait_until_released(&state).await;
+        }
         made
     }
 
@@ -1129,7 +1164,11 @@ impl DriveApi for FakeClient {
             server.put(file_uid, Some(remote));
             Ok(())
         })
-        .await
+        .await?;
+        if self.0.lose_revision_reply.swap(false, Ordering::SeqCst) {
+            return Err(no_answer());
+        }
+        Ok(())
     }
 
     async fn open_revision(&self, uid: &NodeUid) -> Result<Arc<dyn RevisionRead>> {