feat(fuse): answer copy_file_range from bytes already on this machine

A copy inside the mount went through the kernel's read/write loop, one
FUSE round trip per page, because copy_file_range answered ENOSYS. The
daemon now copies a range whose source bytes are local (an open write
handle, a queued write, the cache) into the destination's scratch file,
the same way write stages them. A range it would have to download gets
EOPNOTSUPP, which makes the kernel copy that one call itself and keeps
the operation enabled. The copy still uploads: Drive cannot copy into a
file cp has already created.

In the simulation, a warm cp of a 256 MiB file went from about
100 MiB/s to about 450 MiB/s (debug build).

crates/pdfs-fuse/src/filesystem.rs

@@ -430,58 +430,10 @@ impl Filesystem for ProtonFs {
             reply.error(error);
             return;
         }
-        let fh = fh.0;
-        // Stage the bytes straight into the scratch file (no base download): only
-        // the untouched remainder is pulled from the remote, and only at commit.
-        let file = {
-            let st = self.core.state();
-            match st.handles.get(&fh).and_then(|&i| st.active_writes.get(&i)) {
-                Some(aw) => aw.file.clone(),
-                None => {
-                    reply.error(Errno::EBADF);
-                    return;
-                }
-            }
-        };
-        if let Err(e) = file.write_all_at(data, offset) {
-            error!(ino = ino.0, fh, error = %e, "scratch write failed");
-            // A full local disk is not a broken file: answer `ENOSPC` so the
-            // application can say so (and so `cp`/editors take their own
-            // out-of-space path), and free what the cache is holding — the
-            // scratch file the next write goes to shares that filesystem.
-            if e.raw_os_error() == Some(libc::ENOSPC) {
-                self.core.cache.emergency_evict();
-                reply.error(Errno::ENOSPC);
-            } else {
-                reply.error(Errno::EIO);
-            }
-            return;
+        match self.stage_write(ino.0, fh.0, offset, data) {
+            Ok(()) => reply.written(data.len() as u32),
+            Err(error) => reply.error(error),
         }
-        let new_len = {
-            let mut st = self.core.state();
-            let Some(aw) = st
-                .handles
-                .get(&fh)
-                .copied()
-                .and_then(|i| st.active_writes.get_mut(&i))
-            else {
-                reply.error(Errno::EBADF);
-                return;
-            };
-            aw.record_write(offset, offset + data.len() as u64);
-            let len = aw.len;
-            st.set_size(ino.0, len);
-            len
-        };
-        debug!(
-            ino = ino.0,
-            fh,
-            offset,
-            len = data.len(),
-            new_len,
-            "staged write"
-        );
-        reply.written(data.len() as u32);
     }
 
     #[allow(clippy::too_many_arguments)]
@@ -1016,20 +968,59 @@ impl Filesystem for ProtonFs {
     fn copy_file_range(
         &self,
         _req: &Request,
-        _ino_in: INodeNo,
+        ino_in: INodeNo,
         _fh_in: FileHandle,
-        _offset_in: u64,
-        _ino_out: INodeNo,
-        _fh_out: FileHandle,
-        _offset_out: u64,
-        _len: u64,
-        _flags: CopyFileRangeFlags,
+        offset_in: u64,
+        ino_out: INodeNo,
+        fh_out: FileHandle,
+        offset_out: u64,
+        len: u64,
+        flags: CopyFileRangeFlags,
         reply: ReplyWrite,
     ) {
-        // Also must stay ENOSYS: there is no server-side copy, and this is the
-        // errno that sends coreutils (and the kernel) down the read/write path
-        // instead of failing the copy outright.
-        reply.error(Errno::ENOSYS);
+        // Copied here when the source bytes are on this machine, as `write`
+        // would stage them, minus a FUSE round trip per page: `cp` of a cached
+        // 256 MiB file ran at about 100 MiB/s through read and write. Anything
+        // that would have to be downloaded gets `EOPNOTSUPP`, which makes the
+        // kernel copy that one call through read and write itself; `ENOSYS`
+        // would turn the op off for the whole mount. There is no server-side
+        // copy into a file `cp` has already created, so the copy still uploads.
+        //
+        // On the dispatch loop, like `write`, so it keeps its order with the
+        // reads and writes of the same handle. [`COPY_RANGE_MAX`] bounds how
+        // long one call holds the loop.
+        let (ino_in, ino_out) = (self.tree(ino_in), self.tree(ino_out));
+        if !flags.is_empty() || ino_in == ino_out {
+            reply.error(Errno::EOPNOTSUPP);
+            return;
+        }
+        if let Err(error) = self.core.require_writable(ino_out.0) {
+            reply.error(error);
+            return;
+        }
+        let bytes = match self
+            .core
+            .read_local(ino_in.0, offset_in, len.min(COPY_RANGE_MAX))
+        {
+            Ok(Some(bytes)) => bytes,
+            Ok(None) => {
+                debug!(ino_in = ino_in.0, offset_in, "copy range not local");
+                reply.error(Errno::EOPNOTSUPP);
+                return;
+            }
+            Err(error) => {
+                reply.error(error);
+                return;
+            }
+        };
+        if bytes.is_empty() {
+            reply.written(0);
+            return;
+        }
+        match self.stage_write(ino_out.0, fh_out.0, offset_out, &bytes) {
+            Ok(()) => reply.written(bytes.len() as u32),
+            Err(error) => reply.error(error),
+        }
     }
 }
 
@@ -1057,6 +1048,11 @@ pub struct ProtonFs {
 /// worker busy indefinitely.
 const OPEN_BASE_ATTEMPTS: u32 = 3;
 
+/// The most one `copy_file_range` call copies; the caller loops for the rest.
+/// A local copy of this much takes milliseconds, which is how long it holds
+/// the dispatch loop.
+const COPY_RANGE_MAX: u64 = 16 * 1024 * 1024;
+
 /// One directory's entries as `readdir` reports them: `.` and `..` first, then
 /// the interned children in `children` order.
 ///
@@ -1641,6 +1637,48 @@ impl ProtonFs {
     }
 
     /// The body of [`Filesystem::create`], run off the dispatch loop.
+    /// Stage `data` at `offset` of the write handle `fh` on `ino`, straight into
+    /// its scratch file (no base download): only the untouched remainder is
+    /// pulled from the remote, and only at commit.
+    fn stage_write(&self, ino: u64, fh: u64, offset: u64, data: &[u8]) -> Result<(), Errno> {
+        let file = {
+            let st = self.core.state();
+            match st.handles.get(&fh).and_then(|&i| st.active_writes.get(&i)) {
+                Some(aw) => aw.file.clone(),
+                None => return Err(Errno::EBADF),
+            }
+        };
+        if let Err(e) = file.write_all_at(data, offset) {
+            error!(ino, fh, error = %e, "scratch write failed");
+            // A full local disk is not a broken file: answer `ENOSPC` so the
+            // application can say so (and so `cp`/editors take their own
+            // out-of-space path), and free what the cache is holding — the
+            // scratch file the next write goes to shares that filesystem.
+            if e.raw_os_error() == Some(libc::ENOSPC) {
+                self.core.cache.emergency_evict();
+                return Err(Errno::ENOSPC);
+            }
+            return Err(Errno::EIO);
+        }
+        let new_len = {
+            let mut st = self.core.state();
+            let Some(aw) = st
+                .handles
+                .get(&fh)
+                .copied()
+                .and_then(|i| st.active_writes.get_mut(&i))
+            else {
+                return Err(Errno::EBADF);
+            };
+            aw.record_write(offset, offset + data.len() as u64);
+            let len = aw.len;
+            st.set_size(ino, len);
+            len
+        };
+        debug!(ino, fh, offset, len = data.len(), new_len, "staged write");
+        Ok(())
+    }
+
     fn serve_create(&self, parent: u64, name: &str, reply: ReplyCreate) {
         if let Err(error) = self.core.require_writable(parent) {
             reply.error(error);

crates/pdfs-fuse/src/reads.rs

@@ -1101,6 +1101,92 @@ impl Core {
         })
     }
 
+    /// Bytes `[offset, offset + size)` of file `ino` as `read` would serve
+    /// them, but only from this machine: an open write handle, a queued write,
+    /// or the caches. `Ok(None)` the moment any of it would have to be
+    /// downloaded, and for a size still provisional (B100), which only the
+    /// network settles. Short or empty past the end of the file.
+    ///
+    /// `copy_file_range` answers with this and leaves anything else to the
+    /// kernel's read/write copy.
+    pub(super) fn read_local(
+        &self,
+        ino: u64,
+        offset: u64,
+        size: u64,
+    ) -> Result<Option<Vec<u8>>, Errno> {
+        let handle = self.state().active_writes.get(&ino).map(|h| {
+            (
+                h.file.clone(),
+                h.len,
+                h.uid.clone(),
+                h.base_mtime,
+                h.base_size,
+                h.written.clone(),
+            )
+        });
+        if let Some((file, len, uid, base_mtime, base_size, written)) = handle {
+            return self.serve_open_read_local(
+                &file, len, &uid, base_mtime, base_size, &written, offset, size,
+            );
+        }
+        let (uid, mtime, fsize, provisional) = {
+            let st = self.state();
+            match st.entries.get(&ino) {
+                Some(e) if e.node.is_file() => (
+                    e.uid.clone(),
+                    e.node.modification_time,
+                    node_size(&e.node),
+                    matches!(
+                        &e.node.kind,
+                        NodeKind::File {
+                            claimed_size: None,
+                            ..
+                        }
+                    ),
+                ),
+                Some(_) => return Err(Errno::EISDIR),
+                None => return Err(Errno::ENOENT),
+            }
+        };
+        let uid = match is_local_uid(&uid) {
+            true => self.landed_stand_in_uid(&uid).unwrap_or(uid),
+            false => uid,
+        };
+        if let Some(pending) = self.pending.lock().get(&uid).cloned() {
+            let m = &pending.meta;
+            let file = File::open(&pending.path).map_err(|e| {
+                error!(%uid, path = %pending.path.display(), error = %e, "open staged blob failed");
+                Errno::EIO
+            })?;
+            let mut written = Intervals::default();
+            for &(s, e) in &m.authored {
+                written.add(s, e);
+            }
+            return self.serve_open_read_local(
+                &Arc::new(file),
+                m.len,
+                &uid,
+                m.base_mtime,
+                m.base_size,
+                &written,
+                offset,
+                size,
+            );
+        }
+        // Created here and never written: an empty file.
+        if is_local_uid(&uid) {
+            return Ok(Some(Vec::new()));
+        }
+        if provisional {
+            return Ok(None);
+        }
+        if offset >= fsize || size == 0 {
+            return Ok(Some(Vec::new()));
+        }
+        Ok(self.cached_range(&uid, mtime, fsize, offset, size.min(fsize - offset)))
+    }
+
     /// The largest amount [`Core::fill_gaps_cached`] will copy out of the local
     /// caches while a `release(2)` is in progress.
     ///

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

@@ -2433,6 +2433,152 @@ mod tests {
         eprintln!("wifi: syscalls {wifi_syscalls:?}, drain {wifi_drain:?}");
     }
 
+    /// `copy_file_range` inside the mount is answered from the bytes the daemon
+    /// holds, and left to the kernel for bytes it would have to download. Every
+    /// source kind has to come out byte for byte: cached, queued, open with
+    /// unsaved writes, and on Drive only.
+    #[test]
+    #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"]
+    fn a_copy_inside_the_mount_has_the_source_bytes() {
+        use std::os::fd::AsRawFd as _;
+        use std::os::unix::fs::FileExt as _;
+
+        fn copy_range(from: &std::fs::File, at: u64, to: &std::fs::File, to_at: u64, len: usize) {
+            let (mut at, mut to_at) = (at as i64, to_at as i64);
+            let mut left = len;
+            while left > 0 {
+                // SAFETY: both descriptors are open; the offsets are ours.
+                let n = unsafe {
+                    libc::copy_file_range(
+                        from.as_raw_fd(),
+                        &mut at,
+                        to.as_raw_fd(),
+                        &mut to_at,
+                        left,
+                        0,
+                    )
+                };
+                assert!(n > 0, "copy_file_range: {}", io::Error::last_os_error());
+                left -= n as usize;
+            }
+        }
+
+        let pattern = |len: usize, seed: u8| -> Vec<u8> {
+            (0..len).map(|i| (i % 253) as u8 ^ seed).collect()
+        };
+        let drive = FakeDrive::new();
+        // On Drive before the mount knows it, so nothing of it is cached.
+        let remote = pattern(9 << 20, 1);
+        drive.device().write("remote.bin", &remote);
+        let dir = scratch("copy-range");
+        let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap();
+        let mount = &daemon.mountpoint;
+
+        // Downloaded through the kernel's own copy, then served from the cache.
+        std::fs::copy(mount.join("remote.bin"), mount.join("remote-1.bin")).unwrap();
+        assert!(std::fs::read(mount.join("remote-1.bin")).unwrap() == remote);
+        std::fs::copy(mount.join("remote.bin"), mount.join("remote-2.bin")).unwrap();
+        assert!(std::fs::read(mount.join("remote-2.bin")).unwrap() == remote);
+
+        // Queued and not on Drive yet: the staged blob is the content.
+        let paused = |paused| Request::SetSyncPaused {
+            paused,
+            until: None,
+        };
+        daemon.request(&paused(true)).unwrap();
+        let queued = pattern(20 << 20, 2);
+        std::fs::write(mount.join("queued.bin"), &queued).unwrap();
+        std::fs::copy(mount.join("queued.bin"), mount.join("queued-copy.bin")).unwrap();
+        assert!(std::fs::read(mount.join("queued-copy.bin")).unwrap() == queued);
+
+        // Open with writes not saved yet: the copy sees them, as a read would.
+        let open = std::fs::OpenOptions::new()
+            .read(true)
+            .write(true)
+            .open(mount.join("queued.bin"))
+            .unwrap();
+        open.write_all_at(b"UNSAVED", 100).unwrap();
+        let target = std::fs::File::create(mount.join("unsaved-copy.bin")).unwrap();
+        copy_range(&open, 96, &target, 0, 16);
+        drop(target);
+        drop(open);
+        let mut expected = queued[96..112].to_vec();
+        expected[4..11].copy_from_slice(b"UNSAVED");
+        assert_eq!(
+            std::fs::read(mount.join("unsaved-copy.bin")).unwrap(),
+            expected
+        );
+
+        // A range into the middle of a file keeps the bytes around it.
+        std::fs::write(mount.join("middle.bin"), b"0123456789").unwrap();
+        let source = std::fs::File::open(mount.join("remote.bin")).unwrap();
+        let target = std::fs::OpenOptions::new()
+            .write(true)
+            .open(mount.join("middle.bin"))
+            .unwrap();
+        copy_range(&source, 5, &target, 3, 4);
+        drop(target);
+        let mut expected = b"0123456789".to_vec();
+        expected[3..7].copy_from_slice(&remote[5..9]);
+        assert_eq!(std::fs::read(mount.join("middle.bin")).unwrap(), expected);
+
+        daemon.request(&paused(false)).unwrap();
+        let mut edited = queued.clone();
+        edited[100..107].copy_from_slice(b"UNSAVED");
+        let file = |bytes: &[u8]| Some(Entry::File(Arc::new(bytes.to_vec())));
+        assert!(wait_until(Duration::from_secs(120), || {
+            let tree = drive.tree();
+            daemon.pending().is_ok_and(|ops| ops.is_empty())
+                && tree.get("queued.bin").cloned() == file(&edited)
+                && tree.get("queued-copy.bin").cloned() == file(&queued)
+                && tree.get("remote-2.bin").cloned() == file(&remote)
+                && tree.get("middle.bin").cloned() == file(&expected)
+        }));
+        assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown)));
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
+    /// How long `cp` inside the mount takes for a file the daemon already
+    /// holds, cold and from the page cache. It decides whether the daemon
+    /// should answer `copy_file_range` itself rather than leave the kernel to
+    /// copy through read and write. Runs when `PDFS_SIM_MEASURE` is set.
+    #[test]
+    #[ignore = "mounts FUSE: run with `PDFS_SIM_MEASURE=1 cargo test -p pdfs-fuse sim:: -- --ignored`"]
+    fn a_copy_inside_the_mount_measures_its_speed() {
+        if std::env::var_os("PDFS_SIM_MEASURE").is_none() {
+            eprintln!("skipped: set PDFS_SIM_MEASURE to measure the copy");
+            return;
+        }
+        const SIZE: usize = 256 << 20;
+        let drive = FakeDrive::new();
+        let dir = scratch("copy-speed");
+        let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap();
+        let mount = &daemon.mountpoint;
+        let bytes: Vec<u8> = (0..SIZE).map(|i| (i % 251) as u8).collect();
+        std::fs::write(mount.join("big.bin"), &bytes).unwrap();
+        assert!(wait_until(Duration::from_secs(120), || {
+            daemon.pending().is_ok_and(|ops| ops.is_empty())
+        }));
+        for (round, name) in ["copy-1.bin", "copy-2.bin", "copy-3.bin"]
+            .into_iter()
+            .enumerate()
+        {
+            let start = Instant::now();
+            let status = std::process::Command::new("cp")
+                .arg(mount.join("big.bin"))
+                .arg(mount.join(name))
+                .status()
+                .unwrap();
+            let took = start.elapsed();
+            assert!(status.success());
+            let mib_per_s = (SIZE >> 20) as f64 / took.as_secs_f64();
+            eprintln!("cp round {round}: {took:?}, {mib_per_s:.0} MiB/s");
+            assert!(std::fs::read(mount.join(name)).unwrap() == bytes);
+        }
+        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 half_of_a_wide_folder_unlinked_at_once_leaves_the_other_half() {