fix(fuse): download the base of a file open for writing on a worker

A read on a write handle was answered on the dispatch loop to keep its
order with the handle's writes, and when the bytes it asked for were
neither written nor cached it downloaded them there, holding every request
of the mount for up to the read deadline. The loop now answers from the
scratch file and the local caches only, and hands a read that needs a
download to a transfer worker with the handle as it was when the read came
in (B227).

crates/pdfs-fuse/src/filesystem.rs

@@ -261,10 +261,13 @@ impl Filesystem for ProtonFs {
         // in-flight (possibly unsaved) content: authored bytes come from the
         // scratch file, untouched bytes from the base.
         //
-        // This path stays on the dispatch loop. `write` runs there too, so
-        // keeping its reads there as well preserves the ordering the write path
-        // has always had between a write and a read of the same handle; a mostly
-        // local read is not worth reasoning about that for.
+        // Answered on the dispatch loop when everything it needs is on this
+        // machine, which keeps the order `write` has with reads of the same
+        // handle. A base range that has to be downloaded goes to a worker
+        // instead, with the handle as it was when the read came in: the kernel
+        // issued the read before any write that comes after it, so that is
+        // the content it asked for. On the loop, the download held every
+        // request of the mount (`docs/BUGS.md` B227).
         let handle = {
             let st = self.core.state();
             st.active_writes.get(&ino.0).map(|h| {
@@ -279,7 +282,7 @@ impl Filesystem for ProtonFs {
             })
         };
         if let Some((file, len, uid, base_mtime, base_size, written)) = handle {
-            match self.core.serve_open_read(
+            match self.core.serve_open_read_local(
                 &file,
                 len,
                 &uid,
@@ -289,7 +292,25 @@ impl Filesystem for ProtonFs {
                 offset,
                 size as u64,
             ) {
-                Ok(bytes) => reply.data(&bytes),
+                Ok(Some(bytes)) => reply.data(&bytes),
+                Ok(None) => {
+                    let fs = self.clone();
+                    self.core.workers.run(Lane::Transfer, "read", move || {
+                        match fs.core.serve_open_read(
+                            &file,
+                            len,
+                            &uid,
+                            base_mtime,
+                            base_size,
+                            &written,
+                            offset,
+                            size as u64,
+                        ) {
+                            Ok(bytes) => reply.data(&bytes),
+                            Err(e) => reply.error(e),
+                        }
+                    });
+                }
                 Err(e) => reply.error(e),
             }
             return;

crates/pdfs-fuse/src/reads.rs

@@ -1070,37 +1070,35 @@ impl Core {
         offset: u64,
         size: u64,
     ) -> Result<Vec<u8>, Errno> {
-        if offset >= len || size == 0 {
-            return Ok(Vec::new());
-        }
-        let end = offset.saturating_add(size).min(len);
-        let mut out = Vec::with_capacity((end - offset) as usize);
-        for (s, e, authored) in written.segments(offset, end) {
-            if authored {
-                let mut buf = vec![0u8; (e - s) as usize];
-                file.read_exact_at(&mut buf, s).map_err(|err| {
-                    warn!(%uid, error = %err, "scratch read failed");
-                    Errno::EIO
-                })?;
-                out.extend_from_slice(&buf);
-            } else {
-                let bend = e.min(base_size);
-                if s < bend {
-                    out.extend_from_slice(&self.read_range(
-                        uid,
-                        base_mtime,
-                        base_size,
-                        s,
-                        bend - s,
-                        true,
-                    )?);
-                }
-                // Anything past the base is a hole: zero-fill.
-                let zeros = e.saturating_sub(s.max(base_size));
-                out.resize(out.len() + zeros as usize, 0);
-            }
-        }
-        Ok(out)
+        let read = assemble_open_read(file, len, uid, base_size, written, offset, size, |s, n| {
+            self.read_range(uid, base_mtime, base_size, s, n, true)
+                .map(Some)
+        })?;
+        Ok(read.unwrap_or_default())
+    }
+
+    /// [`Core::serve_open_read`] from bytes already on this machine only:
+    /// `Ok(None)` the moment a range of the base would have to be downloaded.
+    ///
+    /// The dispatch loop answers a read on a write handle with this, and hands
+    /// it to a worker only when it says `None`. Fetching the base there held
+    /// every request of the mount for as long as the download took, up to the
+    /// read deadline (`docs/BUGS.md` B227).
+    #[allow(clippy::too_many_arguments)]
+    pub(super) fn serve_open_read_local(
+        &self,
+        file: &Arc<File>,
+        len: u64,
+        uid: &NodeUid,
+        base_mtime: i64,
+        base_size: u64,
+        written: &Intervals,
+        offset: u64,
+        size: u64,
+    ) -> Result<Option<Vec<u8>>, Errno> {
+        assemble_open_read(file, len, uid, base_size, written, offset, size, |s, n| {
+            Ok(self.cached_range(uid, base_mtime, base_size, s, n))
+        })
     }
 
     /// The largest amount [`Core::fill_gaps_cached`] will copy out of the local
@@ -1242,6 +1240,49 @@ impl Core {
     }
 }
 
+/// The bytes `[offset, offset + size)` of a file open for writing: authored
+/// ranges from its scratch `file`, the rest from its base through `base`,
+/// and zeros past the base. `Ok(None)` when `base` says it cannot answer.
+#[allow(clippy::too_many_arguments)]
+fn assemble_open_read(
+    file: &File,
+    len: u64,
+    uid: &NodeUid,
+    base_size: u64,
+    written: &Intervals,
+    offset: u64,
+    size: u64,
+    mut base: impl FnMut(u64, u64) -> Result<Option<Vec<u8>>, Errno>,
+) -> Result<Option<Vec<u8>>, Errno> {
+    if offset >= len || size == 0 {
+        return Ok(Some(Vec::new()));
+    }
+    let end = offset.saturating_add(size).min(len);
+    let mut out = Vec::with_capacity((end - offset) as usize);
+    for (s, e, authored) in written.segments(offset, end) {
+        if authored {
+            let mut buf = vec![0u8; (e - s) as usize];
+            file.read_exact_at(&mut buf, s).map_err(|err| {
+                warn!(%uid, error = %err, "scratch read failed");
+                Errno::EIO
+            })?;
+            out.extend_from_slice(&buf);
+        } else {
+            let bend = e.min(base_size);
+            if s < bend {
+                match base(s, bend - s)? {
+                    Some(bytes) => out.extend_from_slice(&bytes),
+                    None => return Ok(None),
+                }
+            }
+            // Anything past the base is a hole: zero-fill.
+            let zeros = e.saturating_sub(s.max(base_size));
+            out.resize(out.len() + zeros as usize, 0);
+        }
+    }
+    Ok(Some(out))
+}
+
 #[cfg(test)]
 mod block_ring_tests {
     use super::*;
@@ -1394,3 +1435,54 @@ mod prefetch_tests {
         assert!(prefetch.streams.lock().len() <= MAX_PREFETCH_STREAMS + 1);
     }
 }
+
+#[cfg(test)]
+mod open_read_tests {
+    use super::*;
+    use proton_drive_rs::proton_sdk::ids::{LinkId, VolumeId};
+
+    fn scratch(bytes: &[u8]) -> (File, PathBuf) {
+        let path = std::env::temp_dir().join(format!(
+            "pdfs-open-read-{}-{}",
+            std::process::id(),
+            std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_nanos()
+        ));
+        let mut file = File::options()
+            .create_new(true)
+            .read(true)
+            .write(true)
+            .open(&path)
+            .unwrap();
+        file.write_all(bytes).unwrap();
+        (file, path)
+    }
+
+    /// The dispatch loop answers from what is on this machine and says so
+    /// when the base would have to be downloaded, for a worker to do it
+    /// (`docs/BUGS.md` B227).
+    #[test]
+    fn a_read_needing_the_base_says_so_instead_of_fetching() {
+        let uid = NodeUid::new(VolumeId::from("v"), LinkId::from("f"));
+        let (file, path) = scratch(b"ABCDEFGH");
+        let mut written = Intervals::default();
+        written.add(0, 4);
+
+        let authored = assemble_open_read(&file, 8, &uid, 8, &written, 0, 4, |_, _| {
+            panic!("an authored range never asks the base")
+        });
+        assert_eq!(authored.unwrap().as_deref(), Some(&b"ABCD"[..]));
+
+        let missing = assemble_open_read(&file, 8, &uid, 8, &written, 2, 6, |_, _| Ok(None));
+        assert_eq!(missing.unwrap(), None);
+
+        let base = assemble_open_read(&file, 10, &uid, 8, &written, 2, 8, |s, n| {
+            assert_eq!((s, n), (4, 4));
+            Ok(Some(b"efgh".to_vec()))
+        });
+        assert_eq!(base.unwrap().as_deref(), Some(&b"CDefgh\0\0"[..]));
+        let _ = std::fs::remove_file(path);
+    }
+}