fix(fuse): cache an upload whose read-back failed, and know it as ours (B157)

When the read-back after an upload went unanswered, the drain let go of
the staged blob without caching it. A partial write opened next had
nothing to fill its gaps from and queued an incomplete revision, so the
write open after it was refused with EIO.

The bytes are now cached under the size and mtime the tree still has,
which is the base the next write takes. Their SHA-1 stands in for the
revision id the read-back did not bring, so that write chains onto the
upload and does not fork a conflict copy of itself.

crates/pdfs-fuse/src/drain.rs

@@ -1752,6 +1752,22 @@ impl Core {
             debug!(%uid, ?remote_rev, "the remote already holds the queued write; not a conflict");
             return Ok(None);
         }
+        // An upload of ours landed and its read-back went unanswered, so its
+        // revision id was never recorded as ours. The bytes it sent say the
+        // same: the remote holds them, so no other device wrote since
+        // (docs/BUGS.md B157).
+        let sealed = self.sealed_unread.lock().get(uid).cloned();
+        if meta.complete
+            && let Some(sealed) = sealed
+            && let NodeKind::File {
+                content_sha1: Some(remote),
+                ..
+            } = &node.kind
+            && remote.eq_ignore_ascii_case(&sealed)
+        {
+            debug!(%uid, ?remote_rev, "queued write chains onto our own unread revision; not a conflict");
+            return Ok(None);
+        }
         Ok(Some(reason))
     }
 
@@ -2178,13 +2194,11 @@ impl Core {
         // B119). Adopted by link, not by value: the blob is the whole file, and
         // a video read into a `Vec` here is an OOM. `discard_staged` below drops
         // the staging name only — the cache keeps the inode.
-        self.refresh_after_upload(&uid, &blob, |node| {
-            if node_size(node) == meta.len
+        self.refresh_after_upload(&uid, &blob, |mtime, size| {
+            if size == meta.len
                 && keeps_landed_upload(self.cache.is_pinned(&uid), meta.len, self.cache.budget())
             {
-                let _ = self
-                    .cache
-                    .store_file(&uid, node.modification_time, meta.len, &blob);
+                let _ = self.cache.store_file(&uid, mtime, meta.len, &blob);
             }
         });
         self.cache.discard_staged(&blob);
@@ -2222,11 +2236,16 @@ impl Core {
     /// that conflicted with itself and left a full-size duplicate behind.
     ///
     /// `blob` is the revision that landed, which stays the pending one until
-    /// open write handles are rebased off it, and `adopt` gets the node read
-    /// back before that, so the uploaded bytes can be cached under its
-    /// revision first. A handle released in between fills its gaps from one
-    /// or the other; with the blob let go first and the cache filled last, it
-    /// found neither, and the next write open was refused (docs/BUGS.md B123).
+    /// open write handles are rebased off it, and `adopt` gets the mtime and
+    /// size of the node read back before that, so the uploaded bytes can be
+    /// cached under its revision first. A handle released in between fills its
+    /// gaps from one or the other; with the blob let go first and the cache
+    /// filled last, it found neither, and the next write open was refused
+    /// (docs/BUGS.md B123). When the node cannot be read back, `adopt` gets
+    /// the stamp the tree still has, which is the base a write opened next
+    /// takes: without it the blob went and nothing replaced it (B157). The
+    /// blob's digest then stands in for the revision id the read-back did
+    /// not bring, so the next write knows the revision as its own.
     ///
     /// Best effort: a failure here costs a spurious conflict copy on the next
     /// write, not this upload, which has already landed. Returns the node as
@@ -2235,25 +2254,50 @@ impl Core {
         &self,
         uid: &NodeUid,
         blob: &Path,
-        adopt: impl FnOnce(&Node),
+        adopt: impl FnOnce(i64, u64),
     ) -> Option<Node> {
         let fetched = self.fetch_node_remote(uid);
-        if !matches!(fetched, Ok(Some(_))) {
-            self.release_pending(uid, blob);
+        if fetched.is_ok() {
+            self.sealed_unread.lock().remove(uid);
         }
         let node = match fetched {
             Ok(Some(node)) => node,
             // Trashed or deleted under us: the tree will hear it from the event
             // sync, which is better placed to unhook the inode than we are.
-            Ok(None) => return None,
+            Ok(None) => {
+                self.release_pending(uid, blob);
+                return None;
+            }
             Err(e) => {
                 warn!(%uid, error = %e,
                       "refreshing metadata after an upload failed; \
                        the next write to this file may conflict with itself");
+                match staged_sha1(blob) {
+                    Ok(sha) => {
+                        self.sealed_unread.lock().insert(uid.clone(), sha);
+                    }
+                    Err(_) => {
+                        self.sealed_unread.lock().remove(uid);
+                    }
+                }
+                // No newer write took over, so the tree's stamp is this upload's.
+                if self.pending_blob(uid).as_deref() == Some(blob) {
+                    let mut stamp = None;
+                    self.for_each_state(|st| {
+                        if let Some(entry) = st.by_uid.get(uid).and_then(|ino| st.entries.get(ino))
+                        {
+                            stamp = Some((entry.node.modification_time, node_size(&entry.node)));
+                        }
+                    });
+                    if let Some((mtime, size)) = stamp {
+                        adopt(mtime, size);
+                    }
+                }
+                self.release_pending(uid, blob);
                 return None;
             }
         };
-        adopt(&node);
+        adopt(node.modification_time, node_size(&node));
         // This function's whole job is to bring the tree level with the revision
         // we just sealed, so the feed's report of that revision has nothing left
         // to tell us. Claimed here rather than at the upload call so it is only

crates/pdfs-fuse/src/lib.rs

@@ -576,6 +576,9 @@ struct Core {
     /// 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>>>>,
+    /// The SHA-1 of the last revision this daemon uploaded to a file and could
+    /// not read back, so its revision id is not known (`docs/BUGS.md` B157).
+    sealed_unread: Arc<Mutex<HashMap<NodeUid, 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

@@ -435,6 +435,7 @@ pub(crate) fn mount_with(
         hidden: Arc::new(Mutex::new(HashSet::new())),
         creating: Arc::new(Mutex::new(HashMap::new())),
         create_sent: Arc::new(Mutex::new(HashMap::new())),
+        sealed_unread: 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

@@ -358,6 +358,46 @@ 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_whose_upload_was_not_read_back_takes_two_writes_at_once() {
+        // The read-back after an upload went unanswered, and the drain let go of
+        // the uploaded bytes without caching them. A partial write had nothing
+        // to fill its gaps from, and the next open was refused with EIO (B157).
+        let drive = FakeDrive::new();
+        let dir = scratch("upload-not-read-back");
+        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_read_back_of_next_revision();
+        std::fs::write(&path, b"0123456789").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"0123456789")
+        }));
+        write_at(&path, 2, b"ab").unwrap();
+        write_at(&path, 6, b"cd").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"01ab45cd89")
+            }),
+            "{:?} {:?}",
+            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 the_echo_of_our_own_rename_keeps_the_cached_bytes() {

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

@@ -334,6 +334,8 @@ impl FakeDrive {
             lose_create_reply: AtomicBool::new(false),
             hold_create_reply: Mutex::new(None),
             lose_revision_reply: AtomicBool::new(false),
+            lose_revision_read_back: AtomicBool::new(false),
+            lose_read: Mutex::new(None),
         }))
     }
 
@@ -589,6 +591,10 @@ struct ClientInner {
     hold_create_reply: Mutex<Option<Arc<HoldState>>>,
     /// Set by [`FakeClient::lose_reply_to_next_revision`].
     lose_revision_reply: AtomicBool,
+    /// Set by [`FakeClient::lose_read_back_of_next_revision`].
+    lose_revision_read_back: AtomicBool,
+    /// The file whose next read goes unanswered, once its revision is made.
+    lose_read: Mutex<Option<NodeUid>>,
 }
 
 struct HoldState {
@@ -678,6 +684,13 @@ impl FakeClient {
         self.0.lose_revision_reply.store(true, Ordering::SeqCst);
     }
 
+    /// Make the next revision of a file on Drive and answer it, then lose the
+    /// answer to the next read of that file alone, as the read-back after an
+    /// upload meets on a link that drops it.
+    pub(crate) fn lose_read_back_of_next_revision(&self) {
+        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]) {
         let state = {
@@ -916,6 +929,16 @@ impl DriveApi for FakeClient {
 
     async fn enumerate_nodes(&self, uids: &[NodeUid]) -> Result<Vec<Node>> {
         self.wait_if_held(uids).await;
+        {
+            let mut lose = self.0.lose_read.lock();
+            if lose
+                .as_ref()
+                .is_some_and(|uid| uids == std::slice::from_ref(uid))
+            {
+                *lose = None;
+                return Err(no_answer());
+            }
+        }
         let lag = self.lag();
         let nodes: Vec<Node> = self
             .request(false, |server| {
@@ -1168,6 +1191,9 @@ impl DriveApi for FakeClient {
         if self.0.lose_revision_reply.swap(false, Ordering::SeqCst) {
             return Err(no_answer());
         }
+        if self.0.lose_revision_read_back.swap(false, Ordering::SeqCst) {
+            *self.0.lose_read.lock() = Some(file_uid.clone());
+        }
         Ok(())
     }