fix(fuse,core): adopt a twin holding the create's bytes (B127); trash the node of a withdrawn create that lost its answer (B151)

A retried create now adopts the file its lost answer made when Drive's
SHA-1 matches the staged blob, not only when the file is empty. A delete
after a sent create keeps the op as a trash that finds and trashes the
node that create may have made; a new create of that name waits for it.
The simulation's settle now reads the log, so known damage is explained.

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

@@ -476,6 +476,65 @@ mod tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    #[test]
+    #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"]
+    fn a_create_whose_answer_was_lost_lands_once() {
+        // Drive made the file, but its answer was lost. The retry found the
+        // name taken by a node holding the file's bytes, which was not empty
+        // and so not adopted, and uploaded the file again as a conflict copy
+        // (B127).
+        let drive = FakeDrive::new();
+        let dir = scratch("lost-create-reply");
+        let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap();
+        let file = |bytes: &[u8]| Some(Entry::File(Arc::new(bytes.to_vec())));
+
+        daemon.client.lose_reply_to_next_create();
+        std::fs::write(daemon.mountpoint.join("f.txt"), b"once").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"once")
+            }),
+            "{:?}",
+            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_file_deleted_after_its_create_lost_its_answer_leaves_drive() {
+        // Drive made the file, but its answer was lost, and the file was
+        // deleted before the retry. The delete dropped the queued create and
+        // left the node Drive had made in place (B151).
+        let drive = FakeDrive::new();
+        let dir = scratch("lost-create-reply-deleted");
+        let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap();
+        let path = daemon.mountpoint.join("f.txt");
+
+        daemon.client.lose_reply_to_next_create();
+        std::fs::write(&path, b"gone").unwrap();
+        assert!(wait_until(Duration::from_secs(30), || {
+            drive.tree().contains_key("f.txt")
+        }));
+        std::fs::remove_file(&path).unwrap();
+        assert!(
+            wait_until(Duration::from_secs(30), || {
+                daemon.pending().is_ok_and(|items| items.is_empty()) && drive.tree().is_empty()
+            }),
+            "{:?} {:?}",
+            daemon.pending(),
+            drive.tree()
+        );
+        assert!(!path.exists());
+
+        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

@@ -331,6 +331,7 @@ impl FakeDrive {
             requests: AtomicU64::new(0),
             hold: Mutex::new(None),
             drop_after_create: AtomicBool::new(false),
+            lose_create_reply: AtomicBool::new(false),
         }))
     }
 
@@ -580,6 +581,8 @@ struct ClientInner {
     hold: Mutex<Option<Arc<HoldState>>>,
     /// Set by [`FakeClient::drop_link_after_next_create`].
     drop_after_create: AtomicBool,
+    /// Set by [`FakeClient::lose_reply_to_next_create`].
+    lose_create_reply: AtomicBool,
 }
 
 struct HoldState {
@@ -625,6 +628,12 @@ impl FakeClient {
         self.0.drop_after_create.store(true, Ordering::SeqCst);
     }
 
+    /// Make the next file create on Drive and lose its answer, as a link that
+    /// drops it on the way back does.
+    pub(crate) fn lose_reply_to_next_create(&self) {
+        self.0.lose_create_reply.store(true, Ordering::SeqCst);
+    }
+
     /// Hold the next read of `uid` alone, before it reaches Drive, until the
     /// answer is dropped. Only one read is held at a time.
     pub(crate) fn hold_next_read(&self, uid: &NodeUid) -> Held {
@@ -1061,6 +1070,9 @@ impl DriveApi for FakeClient {
         if made.is_ok() && self.0.drop_after_create.swap(false, Ordering::SeqCst) {
             self.set_online(false);
         }
+        if made.is_ok() && self.0.lose_create_reply.swap(false, Ordering::SeqCst) {
+            return Err(no_answer());
+        }
         made
     }
 

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

@@ -363,7 +363,11 @@ impl Run {
     fn settle(&mut self, budget: Duration) -> Result<(), String> {
         self.note("settle".into());
         let start = Instant::now();
-        if !wait_until(budget, || self.landed()) {
+        // What the drain logs meanwhile is what explains a known bug's damage.
+        if !wait_until(budget, || {
+            self.logged.extend(take_logged());
+            self.landed()
+        }) {
             return Err(format!(
                 "liveness: the queues did not drain within {budget:?}\n{}",
                 self.difference()