fix(fuse): adopt a twin holding bytes an earlier attempt of the create sent (B156)

crates/pdfs-fuse/src/drain.rs

@@ -927,6 +927,7 @@ impl Core {
                 None => debug!(%uid, name, "withdrawn create never landed; trash op satisfied"),
             }
             self.retire_trash_op(op, &uid)?;
+            self.create_sent.lock().remove(&op.id);
             return Ok(());
         }
         match self
@@ -1018,11 +1019,12 @@ impl Core {
             },
         };
         // A blob gone from staging matches nothing, and leaves the node be.
-        let sent = match op.blob_path.as_deref() {
+        let current = match op.blob_path.as_deref() {
             Some(blob) if node_size(&twin) > 0 => staged_sha1(Path::new(blob)).ok(),
             _ => None,
         };
-        Ok(adoptable(is_dir, &twin, op.created_at, sent.as_deref()).then_some(twin.uid))
+        let sent = self.create_sent_digests(op.id, current);
+        Ok(adoptable(is_dir, &twin, op.created_at, &sent).then_some(twin.uid))
     }
 
     /// Record the name the remote actually gave a node, after a conflict forced
@@ -1062,6 +1064,11 @@ impl Core {
         }
         let created = self.create_local_node(op);
         self.creating.lock().remove(&op.id);
+        // A withdrawn create goes on as a trash under the same id, which still
+        // looks for the bytes it sent (`Core::withdrawn_twin`).
+        if !self.db.op_exists(op.id).unwrap_or(true) {
+            self.create_sent.lock().remove(&op.id);
+        }
         created
     }
 
@@ -1346,15 +1353,32 @@ impl Core {
             },
         };
         // A create uploads its bytes in the same call, so the node a lost
-        // answer leaves holds them (docs/BUGS.md B127).
-        let sent = match op.blob_path.as_deref() {
+        // answer leaves holds them (docs/BUGS.md B127): the ones the op holds
+        // now, or ones a write has replaced since (B156), which still go up.
+        let current = match op.blob_path.as_deref() {
             Some(blob) if node_size(&twin) > 0 => Some(staged_sha1(Path::new(blob))?),
             _ => None,
         };
-        Ok(
-            adoptable(op.kind == OP_MKDIR, &twin, op.created_at, sent.as_deref())
-                .then(|| (twin.uid, sent.is_some())),
-        )
+        let holds_blob = current
+            .as_deref()
+            .is_some_and(|sha| holds_bytes(&twin, sha));
+        let sent = self.create_sent_digests(op.id, current);
+        Ok(adoptable(op.kind == OP_MKDIR, &twin, op.created_at, &sent)
+            .then_some((twin.uid, holds_blob)))
+    }
+
+    /// The SHA-1s of the bytes queued create `id` may have made its file with:
+    /// `current`, those of the blob it holds now, and those each earlier
+    /// attempt uploaded.
+    fn create_sent_digests(&self, id: i64, current: Option<String>) -> Vec<String> {
+        let mut sent = self
+            .create_sent
+            .lock()
+            .get(&id)
+            .cloned()
+            .unwrap_or_default();
+        sent.extend(current);
+        sent
     }
 
     /// Make one queued `create`/`mkdir` real under a given name, and hand back
@@ -1402,6 +1426,15 @@ impl Core {
         if !meta.complete {
             return Err("queued create holds an incomplete blob".into());
         }
+        // The call may make the file and lose its answer, and the op may hold
+        // a newer blob by the time it is retried (`Core::adoptable_twin`).
+        let sha = staged_sha1(Path::new(blob))?;
+        let mut sent = self.create_sent.lock();
+        let digests = sent.entry(op.id).or_default();
+        if !digests.contains(&sha) {
+            digests.push(sha);
+        }
+        drop(sent);
         let thumbnails = self.upload_thumbnails(Path::new(blob), name);
         let guard = self
             .transfers
@@ -2503,30 +2536,26 @@ const ADOPT_WINDOW_MS: i64 = 2 * 60 * 1000;
 /// Whether `twin`, found holding the name a queued create wants, is the node an
 /// earlier, unanswered attempt at that create made: the same kind, created
 /// around when the op was queued, and for a file still empty, as the mount's
-/// create leaves it, or holding exactly the bytes the op uploads, whose SHA-1
-/// is `sent`.
-fn adoptable(is_dir: bool, twin: &Node, op_created_ms: i64, sent: Option<&str>) -> bool {
+/// create leaves it, or holding exactly bytes the op uploaded, whose SHA-1 is
+/// one of `sent`.
+fn adoptable(is_dir: bool, twin: &Node, op_created_ms: i64, sent: &[String]) -> bool {
     if twin.creation_time.saturating_mul(1000) < op_created_ms - ADOPT_WINDOW_MS {
         return false;
     }
     match &twin.kind {
         NodeKind::Folder => is_dir,
-        NodeKind::File {
-            claimed_size,
-            content_sha1,
-            ..
-        } => {
-            !is_dir
-                && (*claimed_size == Some(0)
-                    || sent.is_some_and(|sent| {
-                        content_sha1
-                            .as_deref()
-                            .is_some_and(|sha| sha.eq_ignore_ascii_case(sent))
-                    }))
+        NodeKind::File { claimed_size, .. } => {
+            !is_dir && (*claimed_size == Some(0) || sent.iter().any(|sha| holds_bytes(twin, sha)))
         }
     }
 }
 
+/// Whether `node` is a file whose content has SHA-1 `sha`.
+fn holds_bytes(node: &Node, sha: &str) -> bool {
+    matches!(&node.kind, NodeKind::File { content_sha1: Some(have), .. }
+        if have.eq_ignore_ascii_case(sha))
+}
+
 /// The lowercase-hex SHA-1 of a staged blob, as Drive reports a file's.
 fn staged_sha1(blob: &Path) -> Result<String, Box<dyn std::error::Error>> {
     Ok(sha1_of(&mut File::open(blob)?)?)
@@ -3033,20 +3062,20 @@ mod tests {
             false,
             &made(file_node(0, 0, None), 990),
             queued,
-            None
+            &[]
         ));
         assert!(adoptable(
             true,
             &made(folder_node("d", None), 990),
             queued,
-            None
+            &[]
         ));
         // Made long before the op: someone else's file, not our lost create.
         assert!(!adoptable(
             false,
             &made(file_node(0, 0, None), 800),
             queued,
-            None
+            &[]
         ));
         // Our create leaves a file empty; one with other content is someone
         // else's.
@@ -3054,20 +3083,20 @@ mod tests {
             false,
             &made(file_node(0, 5, None), 990),
             queued,
-            None
+            &[]
         ));
         // The kind has to match what the op makes.
         assert!(!adoptable(
             true,
             &made(file_node(0, 0, None), 990),
             queued,
-            None
+            &[]
         ));
         assert!(!adoptable(
             false,
             &made(folder_node("d", None), 990),
             queued,
-            None
+            &[]
         ));
     }
 
@@ -3080,10 +3109,15 @@ mod tests {
         if let NodeKind::File { content_sha1, .. } = &mut twin.kind {
             *content_sha1 = Some("AA55".into());
         }
-        assert!(adoptable(false, &twin, queued, Some("aa55")));
-        assert!(!adoptable(false, &twin, queued, Some("bb66")));
-        assert!(!adoptable(false, &twin, queued, None));
-        assert!(!adoptable(true, &twin, queued, Some("aa55")));
+        let sent = |shas: &[&str]| shas.iter().map(|sha| sha.to_string()).collect::<Vec<_>>();
+        assert!(adoptable(false, &twin, queued, &sent(&["aa55"])));
+        assert!(!adoptable(false, &twin, queued, &sent(&["bb66"])));
+        assert!(!adoptable(false, &twin, queued, &[]));
+        assert!(!adoptable(true, &twin, queued, &sent(&["aa55"])));
+        // An attempt sent these bytes before a write replaced them (B156).
+        assert!(adoptable(false, &twin, queued, &sent(&["aa55", "bb66"])));
+        assert!(holds_bytes(&twin, "aa55"));
+        assert!(!holds_bytes(&twin, "bb66"));
     }
 
     #[test]

crates/pdfs-fuse/src/lib.rs

@@ -573,6 +573,10 @@ struct Core {
     /// Drive holds that name until the create lands and a rename or trash
     /// queued behind it frees it again.
     creating: Arc<Mutex<HashMap<i64, (NodeUid, String)>>>,
+    /// The SHA-1 of each blob an attempt at a queued create uploaded, by op
+    /// id. An attempt whose answer was lost may have made the file with bytes
+    /// a later write has since replaced in the op (`docs/BUGS.md` B156).
+    create_sent: Arc<Mutex<HashMap<i64, Vec<String>>>>,
     /// Creates that landed but could not be read back, by the uid Drive gave
     /// them, with the placeholder's. The drain adopts one before it sends
     /// anything else for it.

crates/pdfs-fuse/src/mount.rs

@@ -434,6 +434,7 @@ pub(crate) fn mount_with(
         pending: Arc::new(Mutex::new(HashMap::new())),
         hidden: Arc::new(Mutex::new(HashSet::new())),
         creating: Arc::new(Mutex::new(HashMap::new())),
+        create_sent: Arc::new(Mutex::new(HashMap::new())),
         unadopted: Arc::new(Mutex::new(HashMap::new())),
         drain_wake: Arc::new((Mutex::new(false), Condvar::new())),
         shutdown: Arc::new(crate::shutdown::Shutdown::default()),

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

@@ -504,6 +504,41 @@ 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_written_again_after_its_create_lost_its_answer_lands_once() {
+        // Drive made the file, but its answer was lost, and the file was
+        // written again before the retry. The retry found the name taken by a
+        // node holding the bytes the first attempt sent, not the ones the op
+        // held now, and uploaded the file as a conflict copy (B156).
+        let drive = FakeDrive::new();
+        let dir = scratch("lost-create-reply-rewritten");
+        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())));
+
+        let held = daemon.client.hold_reply_to_next_create();
+        daemon.client.lose_reply_to_next_create();
+        std::fs::write(&path, b"first").unwrap();
+        assert!(wait_until(Duration::from_secs(30), || held.reached()));
+        std::fs::write(&path, b"second").unwrap();
+        drop(held);
+        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_eq!(std::fs::read(&path).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_file_deleted_after_its_create_lost_its_answer_leaves_drive() {

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

@@ -1099,15 +1099,15 @@ 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());
-        }
         let held = self.0.hold_create_reply.lock().take();
         if made.is_ok()
             && let Some(state) = held
         {
             wait_until_released(&state).await;
         }
+        if made.is_ok() && self.0.lose_create_reply.swap(false, Ordering::SeqCst) {
+            return Err(no_answer());
+        }
         made
     }