fix(fuse): retry a write open whose queued revision landed under the copy

A write open copies the queued revision's blob into its scratch file.
When the upload landed between the sample and the copy, the drain had
already deleted the blob, and the open failed with EIO (B274, seen once
in CI). The drain drops the pending entry before the blob, so a failed
copy whose entry moved on now goes around again, as the re-check after
the copy already does.

Two simulations raced the daemon rather than it:
- a_file_deleted_after_a_restart_leaves_drive_with_what_its_create_sent
  let the create's reply through before the kernel's release attached
  the second bytes. That attach resets the attempts, so the retry
  landed before the test could see the failure. It now waits for the
  bytes to be attached.
- a_file_deleted_offline_while_open_still_reads_and_takes_writes read
  the cache as the queue emptied, which is a moment before the
  eviction. It now waits for the cache to empty.

crates/pdfs-fuse/src/filesystem.rs

@@ -1434,6 +1434,14 @@ impl ProtonFs {
                     // (`ContentCache::clone_or_copy`).
                     if let Err(e) = ContentCache::clone_or_copy(&pending.path, &path) {
                         let _ = std::fs::remove_file(&path);
+                        // The drain lets go of the pending entry before it
+                        // discards the blob, so a blob that went away under the
+                        // copy is a revision that landed or was replaced: the
+                        // same race the re-check below catches, only earlier.
+                        if self.core.pending_blob(&uid).as_ref() != Some(&pending.path) {
+                            warn!(%uid, attempt, "queued revision went away while opening for write; retrying");
+                            continue;
+                        }
                         error!(%uid, source = %pending.path.display(), error = %e,
                             "copy queued revision into write scratch failed");
                         break 'install Err(Errno::EIO);

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

@@ -592,14 +592,22 @@ mod tests {
             daemon.pending(),
             drive.tree()
         );
-        // The bytes kept for the handle went once the trash landed with it closed.
-        let cached: Vec<_> = std::fs::read_dir(dir.join("cache"))
-            .unwrap()
-            .filter_map(|e| e.ok())
-            .filter(|e| e.file_type().is_ok_and(|t| t.is_file()))
-            .map(|e| e.file_name())
-            .collect();
-        assert!(cached.is_empty(), "{cached:?}");
+        // The bytes kept for the handle went once the trash landed with it
+        // closed. The queue empties as the trash's row goes, a moment before
+        // the eviction, so this waits too.
+        let cached = || -> Vec<_> {
+            std::fs::read_dir(dir.join("cache"))
+                .unwrap()
+                .filter_map(|e| e.ok())
+                .filter(|e| e.file_type().is_ok_and(|t| t.is_file()))
+                .map(|e| e.file_name())
+                .collect()
+        };
+        assert!(
+            wait_until(Duration::from_secs(30), || cached().is_empty()),
+            "{:?}",
+            cached()
+        );
 
         assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown)));
         let _ = std::fs::remove_dir_all(&dir);
@@ -1998,6 +2006,21 @@ mod tests {
         daemon.write_new(&path, b"first").unwrap();
         assert!(wait_until(Duration::from_secs(30), || held.reached()));
         std::fs::write(&path, b"second").unwrap();
+        // The kernel sends the release after `close` returns. Attaching the
+        // bytes resets the create's attempts, so a release that came after the
+        // lost reply had it retried at once, and the retry landed before
+        // anything here saw it fail. The bytes it replaced go only once they
+        // are attached.
+        let staged = dir.join("cache").join("staging");
+        assert!(wait_until(Duration::from_secs(30), || {
+            let blobs: Vec<Vec<u8>> = std::fs::read_dir(&staged)
+                .into_iter()
+                .flatten()
+                .filter_map(|e| e.ok())
+                .filter_map(|e| std::fs::read(e.path()).ok())
+                .collect();
+            blobs.iter().any(|b| b == b"second") && !blobs.iter().any(|b| b == b"first")
+        }));
         drop(held);
         assert!(wait_until(Duration::from_secs(30), || {
             daemon