Batch rollout listing scans on cancellable blocking workers (#49694)

## What changed

Move directory enumeration and candidate metadata lookups into blocking workers, batching updated-time candidate collection into one worker per scan. Check cancellation between entries and metadata lookups so dropping the async waiter stops further scanning.

Share synchronous rollout metadata resolution with the async helper, preserving plain-file precedence over compressed siblings. Retain scan limits, stable file ordering, and directory-level early stopping for creation-time listings.

## Testing

Add tests for cancellation during directory enumeration and candidate discovery in both flat and date-partitioned layouts, including plain-sibling metadata precedence and filtering of unrelated files, directories, and symlinks.

GitOrigin-RevId: 324289003b29a7d2b3220da03e3c002ab5e23eec

codex-rs/rollout/src/compression.rs

@@ -18,9 +18,11 @@ use std::os::unix::fs::PermissionsExt;
 
 mod blocking_reader;
 mod error_metrics;
+mod path_metadata;
 mod read_metrics;
 
 use error_metrics::FailureMetric;
+pub(crate) use path_metadata::existing_rollout_with_metadata_sync;
 use read_metrics::ReadFailureSource;
 use read_metrics::ReadMetrics;
 
@@ -1329,19 +1331,11 @@ mod path {
     ///
     /// Returning the metadata lets callers inspect the selected file without a second stat.
     pub(super) async fn existing_rollout_with_metadata(path: &Path) -> Option<(PathBuf, Metadata)> {
-        let plain_path = plain_rollout_path(path);
-        if let Ok(metadata) = tokio::fs::metadata(plain_path.as_path()).await
-            && metadata.is_file()
-        {
-            return Some((plain_path, metadata));
-        }
-        let compressed_path = compressed_rollout_path(plain_path.as_path());
-        if let Ok(metadata) = tokio::fs::metadata(compressed_path.as_path()).await
-            && metadata.is_file()
-        {
-            return Some((compressed_path, metadata));
-        }
-        None
+        let path = path.to_path_buf();
+        tokio::task::spawn_blocking(move || super::existing_rollout_with_metadata_sync(&path))
+            .await
+            .ok()
+            .flatten()
     }
 }
 

codex-rs/rollout/src/compression/path_metadata.rs

@@ -0,0 +1,18 @@
+//! Resolves rollout metadata with shared plain-before-compressed precedence.
+
+use std::fs::Metadata;
+use std::path::Path;
+use std::path::PathBuf;
+
+use super::path::compressed_rollout_path;
+use super::path::plain_rollout_path;
+
+/// Resolves a regular rollout file and its metadata without leaving the blocking worker.
+pub(crate) fn existing_rollout_with_metadata_sync(path: &Path) -> Option<(PathBuf, Metadata)> {
+    let plain_path = plain_rollout_path(path);
+    let compressed_path = compressed_rollout_path(&plain_path);
+    [plain_path, compressed_path].into_iter().find_map(|path| {
+        let metadata = std::fs::metadata(&path).ok()?;
+        metadata.is_file().then_some((path, metadata))
+    })
+}

codex-rs/rollout/src/list.rs

@@ -16,6 +16,14 @@ use time::format_description::well_known::Rfc3339;
 use time::macros::format_description;
 use uuid::Uuid;
 
+#[path = "list_files.rs"]
+mod files;
+use files::collect_dirs_desc;
+use files::collect_files_by_updated_at;
+use files::collect_flat_files_by_updated_at;
+use files::collect_flat_rollout_files;
+use files::collect_rollout_day_files;
+
 use super::ARCHIVED_SESSIONS_SUBDIR;
 use super::SESSIONS_SUBDIR;
 use super::compression;
@@ -290,30 +298,6 @@ impl<'a> RolloutFileVisitor for FilesByCreatedAtVisitor<'a> {
     }
 }
 
-/// Collects lightweight file candidates (path + id + mtime).
-/// Sorting after mtime happens after all files are collected.
-struct FilesByUpdatedAtVisitor<'a> {
-    candidates: &'a mut Vec<ThreadCandidate>,
-}
-
-impl<'a> RolloutFileVisitor for FilesByUpdatedAtVisitor<'a> {
-    async fn visit(
-        &mut self,
-        _ts: OffsetDateTime,
-        id: Uuid,
-        path: PathBuf,
-        _scanned: usize,
-    ) -> ControlFlow<()> {
-        let updated_at = file_modified_time(&path).await.unwrap_or(None);
-        self.candidates.push(ThreadCandidate {
-            path,
-            id,
-            updated_at,
-        });
-        ControlFlow::Continue(())
-    }
-}
-
 impl serde::Serialize for Cursor {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
     where
@@ -901,102 +885,6 @@ pub async fn read_thread_item_from_rollout(path: PathBuf) -> Option<ThreadItem>
     .await
 }
 
-/// Collects immediate subdirectories of `parent`, parses their (string) names with `parse`,
-/// and returns them sorted descending by the parsed key.
-async fn collect_dirs_desc<T, F>(parent: &Path, parse: F) -> io::Result<Vec<(T, PathBuf)>>
-where
-    T: Ord + Copy,
-    F: Fn(&str) -> Option<T>,
-{
-    let mut dir = tokio::fs::read_dir(parent).await?;
-    let mut vec: Vec<(T, PathBuf)> = Vec::new();
-    while let Some(entry) = dir.next_entry().await? {
-        if entry
-            .file_type()
-            .await
-            .map(|ft| ft.is_dir())
-            .unwrap_or(false)
-            && let Some(s) = entry.file_name().to_str()
-            && let Some(v) = parse(s)
-        {
-            vec.push((v, entry.path()));
-        }
-    }
-    vec.sort_by_key(|(v, _)| Reverse(*v));
-    Ok(vec)
-}
-
-/// Collects files in a directory and parses them with `parse`.
-async fn collect_files<T, F>(parent: &Path, parse: F) -> io::Result<Vec<T>>
-where
-    F: Fn(&str, &Path) -> Option<T>,
-{
-    let mut dir = tokio::fs::read_dir(parent).await?;
-    let mut collected: Vec<T> = Vec::new();
-    while let Some(entry) = dir.next_entry().await? {
-        if entry
-            .file_type()
-            .await
-            .map(|ft| ft.is_file())
-            .unwrap_or(false)
-            && let Some(s) = entry.file_name().to_str()
-            && let Some(v) = parse(s, &entry.path())
-        {
-            collected.push(v);
-        }
-    }
-    Ok(collected)
-}
-
-async fn collect_flat_rollout_files(
-    root: &Path,
-    scanned_files: &mut usize,
-) -> io::Result<Vec<(OffsetDateTime, Uuid, PathBuf)>> {
-    let mut dir = tokio::fs::read_dir(root).await?;
-    let mut collected = Vec::new();
-    while let Some(entry) = dir.next_entry().await? {
-        if *scanned_files >= MAX_SCAN_FILES {
-            break;
-        }
-        if !entry
-            .file_type()
-            .await
-            .map(|ft| ft.is_file())
-            .unwrap_or(false)
-        {
-            continue;
-        }
-        let Some(rollout_file) = compression::RolloutFile::from_path(entry.path()) else {
-            continue;
-        };
-        let Some((ts, id)) = parse_timestamp_uuid_from_filename(rollout_file.plain_file_name())
-        else {
-            continue;
-        };
-        *scanned_files += 1;
-        if *scanned_files > MAX_SCAN_FILES {
-            break;
-        }
-        collected.push((ts, id, rollout_file.into_path()));
-    }
-    collected.sort_by_key(|(ts, sid, _path)| (Reverse(*ts), Reverse(*sid)));
-    Ok(collected)
-}
-
-async fn collect_rollout_day_files(
-    day_path: &Path,
-) -> io::Result<Vec<(OffsetDateTime, Uuid, PathBuf)>> {
-    let mut day_files = collect_files(day_path, |_name_str, path| {
-        let rollout_file = compression::RolloutFile::from_path(path.to_path_buf())?;
-        parse_timestamp_uuid_from_filename(rollout_file.plain_file_name())
-            .map(|(ts, id)| (ts, id, rollout_file.into_path()))
-    })
-    .await?;
-    // Stable ordering within the same second: (timestamp desc, uuid desc)
-    day_files.sort_by_key(|(ts, sid, _path)| (Reverse(*ts), Reverse(*sid)));
-    Ok(day_files)
-}
-
 pub(crate) fn parse_timestamp_uuid_from_filename(name: &str) -> Option<(OffsetDateTime, Uuid)> {
     let file_name = RolloutFileName::parse(name)?;
     let thread_id = Uuid::parse_str(&file_name.thread_id().to_string()).ok()?;
@@ -1009,61 +897,6 @@ struct ThreadCandidate {
     updated_at: Option<OffsetDateTime>,
 }
 
-async fn collect_files_by_updated_at(
-    root: &Path,
-    scanned_files: &mut usize,
-) -> io::Result<Vec<ThreadCandidate>> {
-    let mut candidates = Vec::new();
-    let mut visitor = FilesByUpdatedAtVisitor {
-        candidates: &mut candidates,
-    };
-    walk_rollout_files(root, scanned_files, &mut visitor).await?;
-
-    Ok(candidates)
-}
-
-async fn collect_flat_files_by_updated_at(
-    root: &Path,
-    scanned_files: &mut usize,
-) -> io::Result<Vec<ThreadCandidate>> {
-    let mut candidates = Vec::new();
-    let mut dir = tokio::fs::read_dir(root).await?;
-    while let Some(entry) = dir.next_entry().await? {
-        if *scanned_files >= MAX_SCAN_FILES {
-            break;
-        }
-        if !entry
-            .file_type()
-            .await
-            .map(|ft| ft.is_file())
-            .unwrap_or(false)
-        {
-            continue;
-        }
-        let Some(rollout_file) = compression::RolloutFile::from_path(entry.path()) else {
-            continue;
-        };
-        let Some((_ts, id)) = parse_timestamp_uuid_from_filename(rollout_file.plain_file_name())
-        else {
-            continue;
-        };
-        *scanned_files += 1;
-        if *scanned_files > MAX_SCAN_FILES {
-            break;
-        }
-        let updated_at = file_modified_time(rollout_file.path())
-            .await
-            .unwrap_or(None);
-        candidates.push(ThreadCandidate {
-            path: rollout_file.into_path(),
-            id,
-            updated_at,
-        });
-    }
-
-    Ok(candidates)
-}
-
 async fn walk_rollout_files(
     root: &Path,
     scanned_files: &mut usize,

codex-rs/rollout/src/list_files.rs

@@ -0,0 +1,283 @@
+//! Filesystem candidate collection for rollout listing.
+//!
+//! Keep directory scans and metadata lookups on blocking workers. Updated-time listings need
+//! all candidates before sorting; creation-time listings retain directory-level early stopping.
+//! Scans check cancellation between directory entries and candidate metadata lookups.
+
+use std::cmp::Reverse;
+use std::io;
+use std::path::Path;
+use std::path::PathBuf;
+use time::OffsetDateTime;
+use uuid::Uuid;
+
+use super::MAX_SCAN_FILES;
+use super::ThreadCandidate;
+use super::parse_timestamp_uuid_from_filename;
+use super::truncate_to_millis;
+use crate::compression;
+
+/// Runs a scan whose worker observes cancellation when its async waiter is dropped.
+async fn run_scan<T, F>(read: F) -> io::Result<T>
+where
+    T: Send + 'static,
+    F: FnOnce(&Scan) -> io::Result<T> + Send + 'static,
+{
+    let (stop, keep_running) = tokio::sync::oneshot::channel::<()>();
+    let result = tokio::task::spawn_blocking(move || {
+        let scan = Scan { stop };
+        scan.check_cancelled()?;
+        let result = read(&scan);
+        scan.check_cancelled()?;
+        result
+    })
+    .await
+    .map_err(io::Error::other)?;
+    drop(keep_running);
+    result
+}
+
+struct Scan {
+    stop: tokio::sync::oneshot::Sender<()>,
+}
+
+impl Scan {
+    fn check_cancelled(&self) -> io::Result<()> {
+        if self.stop.is_closed() {
+            return Err(io::Error::new(
+                io::ErrorKind::Interrupted,
+                "rollout scan cancelled",
+            ));
+        }
+        Ok(())
+    }
+}
+
+pub(super) async fn collect_dirs_desc<T, F>(
+    parent: &Path,
+    parse: F,
+) -> io::Result<Vec<(T, PathBuf)>>
+where
+    T: Ord + Copy + Send + 'static,
+    F: Fn(&str) -> Option<T> + Send + 'static,
+{
+    let parent = parent.to_path_buf();
+    run_scan(move |scan| collect_dirs_desc_sync(&parent, parse, scan)).await
+}
+
+pub(super) async fn collect_rollout_day_files(
+    day_path: &Path,
+) -> io::Result<Vec<(OffsetDateTime, Uuid, PathBuf)>> {
+    let day_path = day_path.to_path_buf();
+    run_scan(move |scan| collect_rollout_day_files_sync(&day_path, scan)).await
+}
+
+pub(super) async fn collect_flat_rollout_files(
+    root: &Path,
+    scanned_files: &mut usize,
+) -> io::Result<Vec<(OffsetDateTime, Uuid, PathBuf)>> {
+    let root = root.to_path_buf();
+    let mut scanned = *scanned_files;
+    let (result, scanned) = run_scan(move |scan| {
+        let result = collect_flat_rollout_files_sync(&root, &mut scanned, scan);
+        let result = result.and_then(|mut files| {
+            scan.check_cancelled()?;
+            files.sort_by_key(|(ts, sid, _path)| (Reverse(*ts), Reverse(*sid)));
+            Ok(files)
+        });
+        Ok((result, scanned))
+    })
+    .await?;
+    *scanned_files = scanned;
+    result
+}
+
+pub(super) async fn collect_files_by_updated_at(
+    root: &Path,
+    scanned_files: &mut usize,
+) -> io::Result<Vec<ThreadCandidate>> {
+    let root = root.to_path_buf();
+    let mut scanned = *scanned_files;
+    let (result, scanned) = run_scan(move |scan| {
+        let result = (|| {
+            let mut candidates = Vec::new();
+            'outer: for (_, year_path) in
+                collect_dirs_desc_sync(&root, |s| s.parse::<u16>().ok(), scan)?
+            {
+                if scanned >= MAX_SCAN_FILES {
+                    break;
+                }
+                for (_, month_path) in
+                    collect_dirs_desc_sync(&year_path, |s| s.parse::<u8>().ok(), scan)?
+                {
+                    if scanned >= MAX_SCAN_FILES {
+                        break 'outer;
+                    }
+                    for (_, day_path) in
+                        collect_dirs_desc_sync(&month_path, |s| s.parse::<u8>().ok(), scan)?
+                    {
+                        if scanned >= MAX_SCAN_FILES {
+                            break 'outer;
+                        }
+                        for (_, id, path) in collect_rollout_day_files_sync(&day_path, scan)? {
+                            scan.check_cancelled()?;
+                            scanned += 1;
+                            if scanned > MAX_SCAN_FILES {
+                                break 'outer;
+                            }
+                            candidates.push(thread_candidate(path, id));
+                        }
+                    }
+                }
+            }
+            Ok(candidates)
+        })();
+        Ok((result, scanned))
+    })
+    .await?;
+    *scanned_files = scanned;
+    result
+}
+
+pub(super) async fn collect_flat_files_by_updated_at(
+    root: &Path,
+    scanned_files: &mut usize,
+) -> io::Result<Vec<ThreadCandidate>> {
+    let root = root.to_path_buf();
+    let mut scanned = *scanned_files;
+    let (result, scanned) = run_scan(move |scan| {
+        let result = collect_flat_rollout_files_sync(&root, &mut scanned, scan).and_then(|files| {
+            files
+                .into_iter()
+                .map(|(_, id, path)| {
+                    scan.check_cancelled()?;
+                    Ok(thread_candidate(path, id))
+                })
+                .collect()
+        });
+        Ok((result, scanned))
+    })
+    .await?;
+    *scanned_files = scanned;
+    result
+}
+
+/// Resolves metadata in the same plain-before-compressed order as rollout readers.
+fn thread_candidate(path: PathBuf, id: Uuid) -> ThreadCandidate {
+    let updated_at = compression::existing_rollout_with_metadata_sync(&path)
+        .map(|(_, metadata)| metadata)
+        .and_then(|metadata| metadata.modified().ok())
+        .map(OffsetDateTime::from)
+        .and_then(truncate_to_millis);
+    ThreadCandidate {
+        path,
+        id,
+        updated_at,
+    }
+}
+
+/// Collects immediate subdirectories of `parent`, parses their (string) names with `parse`,
+/// and returns them sorted descending by the parsed key.
+fn collect_dirs_desc_sync<T, F>(
+    parent: &Path,
+    parse: F,
+    scan: &Scan,
+) -> io::Result<Vec<(T, PathBuf)>>
+where
+    T: Ord + Copy,
+    F: Fn(&str) -> Option<T>,
+{
+    scan.check_cancelled()?;
+    let dir = std::fs::read_dir(parent)?;
+    let mut vec: Vec<(T, PathBuf)> = Vec::new();
+    for entry in dir {
+        scan.check_cancelled()?;
+        let entry = entry?;
+        if entry.file_type().map(|ft| ft.is_dir()).unwrap_or(false)
+            && let Some(s) = entry.file_name().to_str()
+            && let Some(v) = parse(s)
+        {
+            vec.push((v, entry.path()));
+        }
+    }
+    scan.check_cancelled()?;
+    vec.sort_by_key(|(v, _)| Reverse(*v));
+    Ok(vec)
+}
+
+/// Collects files in a directory and parses them with `parse`.
+fn collect_files<T, F>(parent: &Path, mut parse: F, scan: &Scan) -> io::Result<Vec<T>>
+where
+    F: FnMut(&str, &Path) -> Option<T>,
+{
+    scan.check_cancelled()?;
+    let dir = std::fs::read_dir(parent)?;
+    let mut collected: Vec<T> = Vec::new();
+    for entry in dir {
+        scan.check_cancelled()?;
+        let entry = entry?;
+        if entry.file_type().map(|ft| ft.is_file()).unwrap_or(false)
+            && let Some(s) = entry.file_name().to_str()
+            && let Some(v) = parse(s, &entry.path())
+        {
+            collected.push(v);
+        }
+    }
+    Ok(collected)
+}
+
+fn collect_flat_rollout_files_sync(
+    root: &Path,
+    scanned_files: &mut usize,
+    scan: &Scan,
+) -> io::Result<Vec<(OffsetDateTime, Uuid, PathBuf)>> {
+    scan.check_cancelled()?;
+    let dir = std::fs::read_dir(root)?;
+    let mut collected = Vec::new();
+    for entry in dir {
+        scan.check_cancelled()?;
+        let entry = entry?;
+        if *scanned_files >= MAX_SCAN_FILES {
+            break;
+        }
+        if !entry.file_type().map(|ft| ft.is_file()).unwrap_or(false) {
+            continue;
+        }
+        let Some(rollout_file) = compression::RolloutFile::from_path(entry.path()) else {
+            continue;
+        };
+        let Some((ts, id)) = parse_timestamp_uuid_from_filename(rollout_file.plain_file_name())
+        else {
+            continue;
+        };
+        *scanned_files += 1;
+        if *scanned_files > MAX_SCAN_FILES {
+            break;
+        }
+        collected.push((ts, id, rollout_file.into_path()));
+    }
+    Ok(collected)
+}
+
+fn collect_rollout_day_files_sync(
+    day_path: &Path,
+    scan: &Scan,
+) -> io::Result<Vec<(OffsetDateTime, Uuid, PathBuf)>> {
+    let mut day_files = collect_files(
+        day_path,
+        |_name_str, path| {
+            let rollout_file = compression::RolloutFile::from_path(path.to_path_buf())?;
+            parse_timestamp_uuid_from_filename(rollout_file.plain_file_name())
+                .map(|(ts, id)| (ts, id, rollout_file.into_path()))
+        },
+        scan,
+    )?;
+    // Stable ordering within the same second: (timestamp desc, uuid desc)
+    scan.check_cancelled()?;
+    day_files.sort_by_key(|(ts, sid, _path)| (Reverse(*ts), Reverse(*sid)));
+    Ok(day_files)
+}
+
+#[cfg(test)]
+#[path = "list_files_tests.rs"]
+mod tests;

codex-rs/rollout/src/list_files_tests.rs

@@ -0,0 +1,107 @@
+//! Candidate discovery coverage shared by flat and date-partitioned rollout layouts.
+
+use super::*;
+use crate::list::ThreadListLayout;
+use pretty_assertions::assert_eq;
+use std::fs::File;
+use std::sync::Arc;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+use std::time::UNIX_EPOCH;
+use tempfile::TempDir;
+
+#[tokio::test]
+async fn dropping_scan_stops_directory_enumeration() -> anyhow::Result<()> {
+    let root = TempDir::new()?;
+    for name in ["first", "second"] {
+        File::create(root.path().join(name))?;
+    }
+    let path = root.path().to_path_buf();
+    let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+    let (resume_tx, resume_rx) = std::sync::mpsc::channel();
+    let (finished_tx, finished_rx) = tokio::sync::oneshot::channel();
+    let visited = Arc::new(AtomicUsize::new(0));
+    let worker_visited = Arc::clone(&visited);
+    let task = tokio::spawn(run_scan(move |scan| {
+        let mut started_tx = Some(started_tx);
+        let result = collect_files(
+            &path,
+            |name, _| {
+                worker_visited.fetch_add(1, Ordering::SeqCst);
+                if let Some(started_tx) = started_tx.take() {
+                    let _ = started_tx.send(());
+                    resume_rx.recv().expect("resume scan");
+                }
+                Some(name.to_string())
+            },
+            scan,
+        );
+        let _ = finished_tx.send(result.as_ref().err().map(io::Error::kind));
+        result
+    }));
+    started_rx.await?;
+    task.abort();
+    assert!(task.await.unwrap_err().is_cancelled());
+    resume_tx.send(())?;
+    assert_eq!(
+        (finished_rx.await?, visited.load(Ordering::SeqCst)),
+        (Some(io::ErrorKind::Interrupted), 1),
+    );
+    Ok(())
+}
+
+#[tokio::test]
+async fn updated_candidates_preserve_layout_and_plain_sibling_preference() -> io::Result<()> {
+    for layout in [ThreadListLayout::Flat, ThreadListLayout::NestedByDate] {
+        let root = TempDir::new()?;
+        let directory = match layout {
+            ThreadListLayout::Flat => root.path().to_path_buf(),
+            ThreadListLayout::NestedByDate => root.path().join("2026/09/26"),
+        };
+        std::fs::create_dir_all(&directory)?;
+        let id = Uuid::from_u128(1);
+        let path = directory.join(format!("rollout-2026-09-26T10-00-00-{id}.jsonl"));
+        File::create(&path)?.set_modified(UNIX_EPOCH + Duration::from_secs(100))?;
+        File::create(compression::compressed_rollout_path(&path))?
+            .set_modified(UNIX_EPOCH + Duration::from_secs(200))?;
+        File::create(directory.join("unrelated.jsonl"))?;
+        std::fs::create_dir(directory.join("rollout-2026-09-26T10-00-01-invalid.jsonl"))?;
+        #[cfg(unix)]
+        std::os::unix::fs::symlink(
+            &path,
+            directory.join(format!(
+                "rollout-2026-09-26T10-00-02-{}.jsonl",
+                Uuid::from_u128(2)
+            )),
+        )?;
+
+        let mut scanned = 0;
+        let candidates = match layout {
+            ThreadListLayout::Flat => {
+                collect_flat_files_by_updated_at(root.path(), &mut scanned).await?
+            }
+            ThreadListLayout::NestedByDate => {
+                collect_files_by_updated_at(root.path(), &mut scanned).await?
+            }
+        };
+        assert_eq!(
+            (
+                scanned,
+                candidates
+                    .into_iter()
+                    .map(|candidate| (candidate.id, candidate.path, candidate.updated_at))
+                    .collect::<Vec<_>>()
+            ),
+            (
+                1,
+                vec![(
+                    id,
+                    path,
+                    Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(100))
+                )]
+            ),
+        );
+    }
+    Ok(())
+}