Make exec-server file reads cancellable between chunks (#49696)

## What changed

Run direct `read_file` operations in a blocking task with bounded read chunks and cancellation checks. Dropping the async caller cancels further reads between chunks. Preserve the 512 MiB size limit, symlink-following options, and regular-file validation.

## Testing

Add regression coverage for 4 MiB reads and oversized-file rejection with both symlink-following options, cancellation after the first chunk, and FIFO rejection without waiting for a writer.

GitOrigin-RevId: 19f2816de15a03bd89b463de6b002367e6a692a9

codex-rs/exec-server/src/lib.rs

@@ -21,6 +21,7 @@ mod fs_helper;
 mod fs_helper_main;
 mod fs_sandbox;
 mod local_file_system;
+mod local_file_system_read;
 mod local_process;
 mod network_policy_decisions;
 mod no_follow;

codex-rs/exec-server/src/local_file_system.rs

@@ -14,7 +14,6 @@ use std::sync::LazyLock;
 use std::time::SystemTime;
 use std::time::UNIX_EPOCH;
 use tokio::io;
-use tokio::io::AsyncReadExt;
 use tokio_util::io::ReaderStream;
 use tokio_util::sync::CancellationToken;
 
@@ -42,15 +41,6 @@ use crate::no_follow;
 use crate::regular_file;
 use crate::sandboxed_file_system::SandboxedFileSystem;
 
-const MAX_READ_FILE_BYTES: u64 = 512 * 1024 * 1024;
-
-fn file_too_large_error() -> io::Error {
-    io::Error::new(
-        io::ErrorKind::InvalidInput,
-        format!("file is too large to read: limit is {MAX_READ_FILE_BYTES} bytes"),
-    )
-}
-
 pub static LOCAL_FS: LazyLock<Arc<dyn ExecutorFileSystem>> =
     LazyLock::new(|| -> Arc<dyn ExecutorFileSystem> { Arc::new(LocalFileSystem::unsandboxed()) });
 
@@ -620,23 +610,8 @@ impl DirectFileSystem {
         sandbox: Option<&FileSystemSandboxContext>,
     ) -> FileSystemResult<Vec<u8>> {
         reject_sandbox_context(sandbox)?;
-        let file = if options.follow_symlinks {
-            self.open_file_for_read(path, /*sandbox*/ None).await?
-        } else {
-            no_follow::open_file(path.to_abs_path()?.as_path()).await?
-        };
-        let metadata = file.metadata().await?;
-        if metadata.len() > MAX_READ_FILE_BYTES {
-            return Err(file_too_large_error());
-        }
-        let mut bytes = Vec::with_capacity(metadata.len() as usize);
-        file.take(MAX_READ_FILE_BYTES + 1)
-            .read_to_end(&mut bytes)
-            .await?;
-        if bytes.len() as u64 > MAX_READ_FILE_BYTES {
-            return Err(file_too_large_error());
-        }
-        Ok(bytes)
+        let path = path.to_abs_path()?;
+        crate::local_file_system_read::read_file(path, options).await
     }
 
     async fn read_file_stream(

codex-rs/exec-server/src/local_file_system_read.rs

@@ -0,0 +1,73 @@
+//! Bounded blocking file reads that stop between chunks when their async caller is dropped.
+
+use crate::FILE_READ_CHUNK_SIZE;
+use crate::ReadFileOptions;
+use crate::no_follow;
+use crate::regular_file;
+use codex_utils_absolute_path::AbsolutePathBuf;
+use std::io;
+use std::io::Read;
+use tokio_util::sync::CancellationToken;
+
+const MAX_READ_FILE_BYTES: u64 = 512 * 1024 * 1024;
+
+pub(super) async fn read_file(
+    path: AbsolutePathBuf,
+    options: ReadFileOptions,
+) -> io::Result<Vec<u8>> {
+    let cancelled = CancellationToken::new();
+    let _cancel_on_drop = cancelled.clone().drop_guard();
+    tokio::task::spawn_blocking(move || {
+        if cancelled.is_cancelled() {
+            return Err(io::Error::other("file read cancelled"));
+        }
+        let file = if options.follow_symlinks {
+            regular_file::open_sync(path.as_path())?
+        } else {
+            no_follow::open_file(path.as_path())?
+        };
+        let metadata = file.metadata()?;
+        if metadata.len() > MAX_READ_FILE_BYTES {
+            return Err(file_too_large_error());
+        }
+        let mut bytes = Vec::with_capacity(metadata.len() as usize);
+        CancellableReader {
+            reader: file.take(MAX_READ_FILE_BYTES + 1),
+            cancelled,
+        }
+        .read_to_end(&mut bytes)?;
+        if bytes.len() as u64 > MAX_READ_FILE_BYTES {
+            return Err(file_too_large_error());
+        }
+        Ok(bytes)
+    })
+    .await
+    .map_err(|error| io::Error::other(format!("filesystem task failed: {error}")))?
+}
+
+struct CancellableReader<R> {
+    reader: R,
+    cancelled: CancellationToken,
+}
+
+impl<R: Read> Read for CancellableReader<R> {
+    fn read(&mut self, bytes: &mut [u8]) -> io::Result<usize> {
+        if self.cancelled.is_cancelled() {
+            // read_to_end retries Interrupted errors, so cancellation must use another kind.
+            return Err(io::Error::other("file read cancelled"));
+        }
+        let count = bytes.len().min(FILE_READ_CHUNK_SIZE);
+        self.reader.read(&mut bytes[..count])
+    }
+}
+
+fn file_too_large_error() -> io::Error {
+    io::Error::new(
+        io::ErrorKind::InvalidInput,
+        format!("file is too large to read: limit is {MAX_READ_FILE_BYTES} bytes"),
+    )
+}
+
+#[cfg(test)]
+#[path = "local_file_system_read_tests.rs"]
+mod tests;

codex-rs/exec-server/src/local_file_system_read_tests.rs

@@ -0,0 +1,91 @@
+//! Regression coverage for bounded, cancellable reads and regular-file validation.
+
+#[cfg(unix)]
+use super::CancellableReader;
+use super::MAX_READ_FILE_BYTES;
+use crate::ReadFileOptions;
+use crate::local_file_system::LOCAL_FS;
+use codex_utils_path_uri::PathUri;
+use pretty_assertions::assert_eq;
+use tempfile::TempDir;
+use test_case::test_case;
+
+#[test_case(true ; "follow")]
+#[test_case(false ; "no_follow")]
+#[tokio::test]
+async fn reads_file_larger_than_tokio_buffer(follow_symlinks: bool) -> anyhow::Result<()> {
+    let directory = TempDir::new()?;
+    let path = directory.path().canonicalize()?.join("large.bin");
+    let contents = vec![0x5a; 4 * 1024 * 1024];
+    std::fs::write(&path, &contents)?;
+    assert_eq!(
+        LOCAL_FS
+            .read_file(
+                &PathUri::from_host_native_path(&path)?,
+                ReadFileOptions { follow_symlinks },
+                /*sandbox*/ None,
+            )
+            .await?,
+        contents,
+    );
+    Ok(())
+}
+
+#[test_case(true ; "follow")]
+#[test_case(false ; "no_follow")]
+#[tokio::test]
+async fn rejects_oversized_file(follow_symlinks: bool) -> anyhow::Result<()> {
+    let directory = TempDir::new()?;
+    let path = directory.path().canonicalize()?.join("oversized.bin");
+    std::fs::File::create(&path)?.set_len(MAX_READ_FILE_BYTES + 1)?;
+    let error = LOCAL_FS
+        .read_file(
+            &PathUri::from_host_native_path(&path)?,
+            ReadFileOptions { follow_symlinks },
+            /*sandbox*/ None,
+        )
+        .await
+        .expect_err("oversized file must be rejected before allocation");
+    assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput);
+    Ok(())
+}
+
+#[cfg(unix)]
+#[test]
+fn cancellation_stops_a_reader_after_its_first_chunk() -> anyhow::Result<()> {
+    use std::io::Read;
+    use std::io::Write;
+    use std::os::unix::net::UnixStream;
+    use tokio_util::sync::CancellationToken;
+
+    let (socket, mut writer) = UnixStream::pair()?;
+    let cancelled = CancellationToken::new();
+    let (reached_tx, reached_rx) = std::sync::mpsc::channel();
+    let (resume_tx, resume_rx) = std::sync::mpsc::channel::<()>();
+    let worker = std::thread::spawn({
+        let cancelled = cancelled.clone();
+        move || -> std::io::Result<()> {
+            let mut reader = CancellableReader {
+                reader: socket,
+                cancelled,
+            };
+            let mut first = [0];
+            reader.read_exact(&mut first)?;
+            let _ = reached_tx.send(first);
+            let _ = resume_rx.recv();
+            reader.read_to_end(&mut Vec::new()).map(|_| ())
+        }
+    });
+    writer.write_all(b"x")?;
+    let first = reached_rx.recv()?;
+    cancelled.cancel();
+    // Closing the peer also makes a missing cancellation check fail without hanging the test.
+    drop(writer);
+    drop(resume_tx);
+    let error = worker
+        .join()
+        .unwrap()
+        .expect_err("cancelled read must stop");
+    assert_eq!((first, error.kind()), (*b"x", std::io::ErrorKind::Other));
+    Ok(())
+}

codex-rs/exec-server/src/no_follow/mod.rs

@@ -16,8 +16,8 @@ use unix as imp;
 #[cfg(windows)]
 use windows as imp;
 
-pub(crate) async fn open_file(path: &Path) -> io::Result<tokio::fs::File> {
-    imp::open_file(path.to_path_buf()).await
+pub(crate) fn open_file(path: &Path) -> io::Result<std::fs::File> {
+    imp::open_file_sync(path)
 }
 
 pub(crate) async fn write_file(path: &Path, contents: Vec<u8>) -> io::Result<()> {

codex-rs/exec-server/src/no_follow/unix.rs

@@ -102,7 +102,7 @@ fn open_entry_sync(path: &Path) -> io::Result<std::fs::File> {
     Ok(std::fs::File::from(file))
 }
 
-fn open_file_sync(path: &Path) -> io::Result<std::fs::File> {
+pub(super) fn open_file_sync(path: &Path) -> io::Result<std::fs::File> {
     let file = open_entry_sync(path)?;
     if !file.metadata()?.is_file() {
         return Err(io::Error::new(
@@ -113,12 +113,6 @@ fn open_file_sync(path: &Path) -> io::Result<std::fs::File> {
     Ok(file)
 }
 
-pub(super) async fn open_file(path: PathBuf) -> io::Result<tokio::fs::File> {
-    tokio::task::spawn_blocking(move || open_file_sync(&path).map(tokio::fs::File::from_std))
-        .await
-        .map_err(|error| io::Error::other(format!("filesystem task failed: {error}")))?
-}
-
 pub(super) async fn write_file(path: PathBuf, contents: Vec<u8>) -> io::Result<()> {
     tokio::task::spawn_blocking(move || {
         let (parent, leaf) = parent(&path)?;

codex-rs/exec-server/src/no_follow/windows.rs

@@ -200,7 +200,7 @@ fn open_entry(path: &Path) -> io::Result<std::fs::File> {
     Ok(std::fs::File::from(handle))
 }
 
-fn open_file_sync(path: &Path) -> io::Result<std::fs::File> {
+pub(super) fn open_file_sync(path: &Path) -> io::Result<std::fs::File> {
     let handle = open_handle(path, FILE_GENERIC_READ, FILE_OPEN, FILE_NON_DIRECTORY_FILE)?;
     let file = std::fs::File::from(handle);
     validate_regular_file(&file, path)?;
@@ -217,12 +217,6 @@ fn validate_regular_file(file: &std::fs::File, path: &Path) -> io::Result<()> {
     Ok(())
 }
 
-pub(super) async fn open_file(path: PathBuf) -> io::Result<tokio::fs::File> {
-    tokio::task::spawn_blocking(move || open_file_sync(&path).map(tokio::fs::File::from_std))
-        .await
-        .map_err(|error| io::Error::other(format!("filesystem task failed: {error}")))?
-}
-
 pub(super) async fn write_file(path: PathBuf, contents: Vec<u8>) -> io::Result<()> {
     tokio::task::spawn_blocking(move || {
         let handle = open_handle(

codex-rs/exec-server/src/regular_file.rs

@@ -1,14 +1,26 @@
 use std::io;
+#[cfg(unix)]
+use std::os::unix::fs::OpenOptionsExt;
+#[cfg(windows)]
+use std::os::windows::fs::OpenOptionsExt;
 use std::path::Path;
 use tokio::io::AsyncReadExt;
 
 pub(crate) async fn open(path: &Path) -> io::Result<tokio::fs::File> {
-    let mut options = tokio::fs::OpenOptions::new();
+    let path = path.to_path_buf();
+    tokio::task::spawn_blocking(move || open_sync(&path).map(tokio::fs::File::from_std))
+        .await
+        .map_err(|error| io::Error::other(format!("filesystem task failed: {error}")))?
+}
+
+/// Opens a regular file without blocking on Unix FIFOs or impersonating Windows pipe servers.
+pub(crate) fn open_sync(path: &Path) -> io::Result<std::fs::File> {
+    let mut options = std::fs::OpenOptions::new();
     options.read(true);
     configure_open(&mut options);
 
-    let file = options.open(path).await?;
-    if !is_disk_file(&file) || !file.metadata().await?.is_file() {
+    let file = options.open(path)?;
+    if !is_disk_file(&file) || !file.metadata()?.is_file() {
         return Err(io::Error::new(
             io::ErrorKind::InvalidInput,
             format!("path `{}` is not a file", path.display()),
@@ -19,7 +31,7 @@ pub(crate) async fn open(path: &Path) -> io::Result<tokio::fs::File> {
 
 /// Reads a regular UTF-8 file without following a symlink at its final path component.
 pub async fn read_sensitive_file_to_string(path: &Path) -> io::Result<String> {
-    let mut options = tokio::fs::OpenOptions::new();
+    let mut options = std::fs::OpenOptions::new();
     options.read(true);
     configure_open(&mut options);
 
@@ -33,7 +45,7 @@ pub async fn read_sensitive_file_to_string(path: &Path) -> io::Result<String> {
         options.custom_flags(FILE_FLAG_OPEN_REPARSE_POINT);
     }
 
-    let mut file = options.open(path).await?;
+    let mut file = tokio::fs::OpenOptions::from(options).open(path).await?;
     let metadata = file.metadata().await?;
     if !is_disk_file(&file) || !metadata.is_file() || metadata.file_type().is_symlink() {
         return Err(io::Error::new(
@@ -48,19 +60,19 @@ pub async fn read_sensitive_file_to_string(path: &Path) -> io::Result<String> {
 }
 
 #[cfg(unix)]
-fn configure_open(options: &mut tokio::fs::OpenOptions) {
+fn configure_open(options: &mut std::fs::OpenOptions) {
     options.custom_flags(libc::O_NONBLOCK);
 }
 
 #[cfg(windows)]
-fn configure_open(options: &mut tokio::fs::OpenOptions) {
+fn configure_open(options: &mut std::fs::OpenOptions) {
     use windows_sys::Win32::Storage::FileSystem::SECURITY_IDENTIFICATION;
 
     options.security_qos_flags(SECURITY_IDENTIFICATION);
 }
 
 #[cfg(not(any(unix, windows)))]
-fn configure_open(_options: &mut tokio::fs::OpenOptions) {}
+fn configure_open(_options: &mut std::fs::OpenOptions) {}
 
 #[cfg(windows)]
 pub(crate) fn is_disk_file(file: &impl std::os::windows::io::AsRawHandle) -> bool {
@@ -73,7 +85,7 @@ pub(crate) fn is_disk_file(file: &impl std::os::windows::io::AsRawHandle) -> boo
 }
 
 #[cfg(not(windows))]
-fn is_disk_file(_file: &tokio::fs::File) -> bool {
+fn is_disk_file<T>(_file: &T) -> bool {
     true
 }
 

codex-rs/exec-server/tests/file_stream.rs

@@ -1,5 +1,7 @@
 mod common;
 
+#[cfg(unix)]
+use anyhow::Context;
 use anyhow::Result;
 use codex_exec_server::Environment;
 use codex_exec_server::ExecServerClient;
@@ -77,12 +79,14 @@ async fn completed_streams_release_handle_capacity() -> Result<()> {
 }
 
 #[cfg(unix)]
+#[test_case::test_case(true ; "follow")]
+#[test_case::test_case(false ; "no_follow")]
 #[tokio::test]
-async fn file_reads_reject_fifo_without_waiting_for_a_writer() -> Result<()> {
+async fn file_reads_reject_fifo_without_waiting_for_a_writer(follow_symlinks: bool) -> Result<()> {
     let server = exec_server().await?;
     let file_system = connect_file_system(server.websocket_url())?;
     let tmp = TempDir::new()?;
-    let path = tmp.path().join("named-pipe");
+    let path = tmp.path().canonicalize()?.join("named-pipe");
     let output = std::process::Command::new("mkfifo").arg(&path).output()?;
     if !output.status.success() {
         anyhow::bail!(
@@ -93,26 +97,37 @@ async fn file_reads_reject_fifo_without_waiting_for_a_writer() -> Result<()> {
     }
 
     let path_uri = PathUri::from_host_native_path(&path)?;
-    let read_error = timeout(
+    let read_result = timeout(
         Duration::from_secs(1),
-        file_system.read_file(&path_uri, ReadFileOptions::default(), /*sandbox*/ None),
+        file_system.read_file(
+            &path_uri,
+            ReadFileOptions { follow_symlinks },
+            /*sandbox*/ None,
+        ),
     )
     .await
-    .expect("reading a FIFO should not wait for a writer")
-    .expect_err("reading a FIFO should be rejected");
+    .context("reading a FIFO should not wait for a writer")?;
+    let Err(read_error) = read_result else {
+        panic!("reading a FIFO should be rejected");
+    };
     let stream_result = timeout(
         Duration::from_secs(1),
         file_system.read_file_stream(&path_uri, /*sandbox*/ None),
     )
     .await
-    .expect("streaming a FIFO should not wait for a writer");
+    .context("streaming a FIFO should not wait for a writer")?;
     let Err(stream_error) = stream_result else {
         panic!("streaming a FIFO should be rejected");
     };
     let expected = format!("path `{}` is not a file", path.display());
+    let expected_read = if follow_symlinks {
+        expected.clone()
+    } else {
+        "path is not a regular file".to_string()
+    };
     assert_eq!(
         (read_error.to_string(), stream_error.to_string()),
-        (expected.clone(), expected)
+        (expected_read, expected)
     );
     Ok(())
 }