diff --git a/codex-rs/rollout/src/compression.rs b/codex-rs/rollout/src/compression.rs index 47c86520f4..3065b05373 100644 --- a/codex-rs/rollout/src/compression.rs +++ b/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() } } diff --git a/codex-rs/rollout/src/compression/path_metadata.rs b/codex-rs/rollout/src/compression/path_metadata.rs new file mode 100644 index 0000000000..6ea0d0deee --- /dev/null +++ b/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)) + }) +} diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index 13c94617a1..bfed9a7097 100644 --- a/codex-rs/rollout/src/list.rs +++ b/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, -} - -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(&self, serializer: S) -> Result where @@ -901,102 +885,6 @@ pub async fn read_thread_item_from_rollout(path: PathBuf) -> Option .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(parent: &Path, parse: F) -> io::Result> -where - T: Ord + Copy, - F: Fn(&str) -> Option, -{ - 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(parent: &Path, parse: F) -> io::Result> -where - F: Fn(&str, &Path) -> Option, -{ - let mut dir = tokio::fs::read_dir(parent).await?; - let mut collected: Vec = 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> { - 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> { - 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, } -async fn collect_files_by_updated_at( - root: &Path, - scanned_files: &mut usize, -) -> io::Result> { - 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> { - 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, diff --git a/codex-rs/rollout/src/list_files.rs b/codex-rs/rollout/src/list_files.rs new file mode 100644 index 0000000000..38b62f86b8 --- /dev/null +++ b/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(read: F) -> io::Result +where + T: Send + 'static, + F: FnOnce(&Scan) -> io::Result + 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( + parent: &Path, + parse: F, +) -> io::Result> +where + T: Ord + Copy + Send + 'static, + F: Fn(&str) -> Option + 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> { + 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> { + 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> { + 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::().ok(), scan)? + { + if scanned >= MAX_SCAN_FILES { + break; + } + for (_, month_path) in + collect_dirs_desc_sync(&year_path, |s| s.parse::().ok(), scan)? + { + if scanned >= MAX_SCAN_FILES { + break 'outer; + } + for (_, day_path) in + collect_dirs_desc_sync(&month_path, |s| s.parse::().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> { + 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( + parent: &Path, + parse: F, + scan: &Scan, +) -> io::Result> +where + T: Ord + Copy, + F: Fn(&str) -> Option, +{ + 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(parent: &Path, mut parse: F, scan: &Scan) -> io::Result> +where + F: FnMut(&str, &Path) -> Option, +{ + scan.check_cancelled()?; + let dir = std::fs::read_dir(parent)?; + let mut collected: Vec = 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> { + 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> { + 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; diff --git a/codex-rs/rollout/src/list_files_tests.rs b/codex-rs/rollout/src/list_files_tests.rs new file mode 100644 index 0000000000..25f6033f2e --- /dev/null +++ b/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::>() + ), + ( + 1, + vec![( + id, + path, + Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(100)) + )] + ), + ); + } + Ok(()) +}