codex_rollout/
compression.rs

1use std::ffi::OsStr;
2use std::fs::File;
3use std::fs::Permissions;
4use std::io;
5use std::io::Read;
6use std::io::Write;
7use std::path::Path;
8use std::path::PathBuf;
9use std::sync::atomic::AtomicU64;
10use std::sync::atomic::Ordering;
11use std::time::Duration;
12
13#[cfg(unix)]
14use std::os::unix::fs::OpenOptionsExt;
15#[cfg(unix)]
16use std::os::unix::fs::PermissionsExt;
17
18const COMPRESSED_SUFFIX: &str = ".zst";
19const MAX_NOT_FOUND_RETRIES: usize = 3;
20const OPEN_ROLLOUT_LINE_READER_RETRY_DELAY: Duration = Duration::from_millis(50);
21const TEMP_SUFFIX: &str = ".tmp";
22static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0);
23
24/// Starts a best-effort background job that compresses cold local rollout files.
25///
26/// The worker is fire-and-forget: failures are logged, startup is not blocked,
27/// and a run marker under `codex_home` prevents overlapping or too-frequent
28/// compression runs from the same local store.
29pub fn spawn_rollout_compression_worker(codex_home: PathBuf) {
30    worker::spawn(codex_home)
31}
32
33/// Returns the modified time for the existing plain or compressed rollout file.
34pub(crate) async fn file_modified_time(path: &Path) -> io::Result<Option<time::OffsetDateTime>> {
35    let Some(path) = path::existing_rollout_path(path).await else {
36        return Ok(None);
37    };
38    let meta = tokio::fs::metadata(path).await?;
39    let modified = meta.modified().ok();
40    Ok(modified.map(time::OffsetDateTime::from))
41}
42
43/// Opens a rollout line reader that transparently handles plain `.jsonl` and `.jsonl.zst` files.
44///
45/// If the requested path disappears during a representation transition, this briefly retries
46/// resolution so callers do not need to know which representation is on disk.
47pub async fn open_rollout_line_reader(path: &Path) -> io::Result<RolloutLineReader> {
48    for _ in 0..MAX_NOT_FOUND_RETRIES {
49        match reader::open_once(path).await {
50            Ok(reader) => return Ok(reader),
51            Err(err) if err.kind() == io::ErrorKind::NotFound => {
52                tokio::time::sleep(OPEN_ROLLOUT_LINE_READER_RETRY_DELAY).await;
53            }
54            Err(err) => return Err(err),
55        }
56    }
57    reader::open_once(path).await
58}
59
60/// Returns the compressed `.jsonl.zst` path for a rollout path.
61#[cfg(test)]
62pub(crate) fn compressed_rollout_path(path: &Path) -> PathBuf {
63    path::compressed_rollout_path(path)
64}
65
66/// Materializes a compressed rollout back to plain `.jsonl` for async append paths.
67pub(crate) async fn materialize_rollout_for_append(path: &Path) -> io::Result<PathBuf> {
68    let path = path.to_path_buf();
69    tokio::task::spawn_blocking(move || materialize_rollout_for_append_blocking(path.as_path()))
70        .await
71        .map_err(io::Error::other)?
72}
73
74/// Materializes a compressed rollout back to plain `.jsonl` for blocking append paths.
75pub(crate) fn materialize_rollout_for_append_blocking(path: &Path) -> io::Result<PathBuf> {
76    let plain_path = plain_rollout_path(path);
77    if plain_path.exists() {
78        metrics::materialize("plain_exists");
79        return Ok(plain_path);
80    }
81    let compressed_path = path::compressed_rollout_path(plain_path.as_path());
82    if !compressed_path.exists() {
83        metrics::materialize("missing");
84        return Ok(plain_path);
85    }
86
87    let temp_path = temp_path_for(plain_path.as_path(), "decompress");
88    if let Some(parent) = plain_path.parent() {
89        std::fs::create_dir_all(parent)?;
90    }
91    let result: io::Result<()> = (|| {
92        let permissions = std::fs::metadata(compressed_path.as_path())?.permissions();
93        {
94            let input = File::open(compressed_path.as_path())?;
95            let mut decoder = zstd::stream::read::Decoder::new(input)?;
96            let mut output = create_file_with_permissions(temp_path.as_path(), &permissions)?;
97            io::copy(&mut decoder, &mut output)?;
98            output.flush()?;
99            output.sync_all()?;
100        }
101        match std::fs::hard_link(temp_path.as_path(), plain_path.as_path()) {
102            Ok(()) => {}
103            Err(err) if err.kind() == io::ErrorKind::AlreadyExists => {}
104            Err(_) => persist_temp_file_noclobber(temp_path.as_path(), plain_path.as_path())?,
105        }
106        let _ = std::fs::remove_file(temp_path.as_path());
107        match std::fs::remove_file(compressed_path.as_path()) {
108            Ok(()) => {}
109            Err(err) if err.kind() == io::ErrorKind::NotFound => {}
110            Err(err) => return Err(err),
111        }
112        Ok(())
113    })();
114    if result.is_err() {
115        let _ = std::fs::remove_file(temp_path.as_path());
116        metrics::materialize("failed");
117    }
118    result?;
119    metrics::materialize("decompressed");
120    Ok(plain_path)
121}
122
123fn persist_temp_file_noclobber(temp_path: &Path, destination: &Path) -> io::Result<()> {
124    let temp_path = tempfile::TempPath::try_from_path(temp_path)?;
125    match temp_path.persist_noclobber(destination) {
126        Ok(()) => Ok(()),
127        Err(err) if err.error.kind() == io::ErrorKind::AlreadyExists => Ok(()),
128        Err(err) => Err(err.error),
129    }
130}
131
132/// Returns the plain `.jsonl` path for a plain or compressed rollout path.
133pub fn plain_rollout_path(path: &Path) -> PathBuf {
134    path::plain_rollout_path(path)
135}
136
137/// Parses a rollout file name, returning its plain `.jsonl` name when valid.
138pub(crate) fn parse_rollout_file_name(name: &str) -> Option<&str> {
139    file_name::parse_rollout_file_name(name)
140}
141
142/// A discovered rollout file, represented by exactly one physical path.
143///
144/// This keeps directory walkers from reimplementing the plain/compressed
145/// precedence rules. The physical path may point at either `.jsonl` or
146/// `.jsonl.zst`, while `plain_file_name` is always the canonical `.jsonl`
147/// filename used for timestamp and id parsing.
148#[derive(Debug, Clone, PartialEq, Eq)]
149pub(crate) struct RolloutFile {
150    path: PathBuf,
151    plain_file_name: String,
152}
153
154impl RolloutFile {
155    /// Creates a logical rollout file from a physical path found during discovery.
156    ///
157    /// Returns `None` for non-rollout names and for compressed siblings hidden by
158    /// an existing plain `.jsonl` file.
159    pub(crate) fn from_path(path: PathBuf) -> Option<Self> {
160        let file_name = path.file_name().and_then(|name| name.to_str())?;
161        let plain_file_name = file_name::parse_rollout_file_name(file_name)?.to_string();
162        if path::should_skip_compressed_sibling(path.as_path()) {
163            return None;
164        }
165
166        Some(Self {
167            path,
168            plain_file_name,
169        })
170    }
171
172    /// Returns the physical path that should be opened for reads.
173    pub(crate) fn path(&self) -> &Path {
174        self.path.as_path()
175    }
176
177    /// Returns the canonical `.jsonl` filename for timestamp and id parsing.
178    pub(crate) fn plain_file_name(&self) -> &str {
179        self.plain_file_name.as_str()
180    }
181
182    /// Returns whether the physical path is the compressed representation.
183    pub(crate) fn is_compressed(&self) -> bool {
184        path::is_compressed_rollout_path(self.path.as_path())
185    }
186
187    /// Consumes the entry and returns the physical path that should be read.
188    pub(crate) fn into_path(self) -> PathBuf {
189        self.path
190    }
191}
192
193/// Line-oriented rollout reader returned by [`open_rollout_line_reader`].
194pub struct RolloutLineReader {
195    inner: RolloutLineReaderInner,
196}
197
198enum RolloutLineReaderInner {
199    Plain(tokio::io::Lines<tokio::io::BufReader<tokio::fs::File>>),
200    Blocking(Option<BlockingLineReader>),
201}
202
203impl RolloutLineReader {
204    /// Reads the next JSONL record from the rollout.
205    pub async fn next_line(&mut self) -> io::Result<Option<String>> {
206        match &mut self.inner {
207            RolloutLineReaderInner::Plain(lines) => lines.next_line().await,
208            RolloutLineReaderInner::Blocking(slot) => {
209                let Some(mut reader) = slot.take() else {
210                    return Err(io::Error::other("compressed rollout reader is busy"));
211                };
212                let (line, reader) =
213                    tokio::task::spawn_blocking(move || (reader.next().transpose(), reader))
214                        .await
215                        .map_err(io::Error::other)?;
216                *slot = Some(reader);
217                line
218            }
219        }
220    }
221}
222
223type BlockingLineReader = std::io::Lines<std::io::BufReader<Box<dyn Read + Send>>>;
224
225mod worker {
226    use std::ffi::OsStr;
227    use std::fs::File;
228    use std::fs::FileTimes;
229    use std::fs::Permissions;
230    use std::io;
231    use std::io::Write;
232    use std::path::Path;
233    use std::path::PathBuf;
234    use std::time::Duration;
235    use std::time::Instant;
236    use std::time::SystemTime;
237
238    use tracing::debug;
239    use tracing::info;
240    use tracing::warn;
241
242    use tokio::task::JoinSet;
243
244    use crate::ARCHIVED_SESSIONS_SUBDIR;
245    use crate::RolloutReferenceIndex;
246    use crate::SESSIONS_SUBDIR;
247
248    use super::RolloutFile;
249    use super::metrics;
250    use super::path;
251
252    const TEMP_SUFFIX: &str = ".tmp";
253    const COMPRESSION_LEVEL: i32 = 3;
254    const MIN_ROLLOUT_AGE: Duration = Duration::from_secs(7 * 24 * 60 * 60);
255    const RUN_MARKER_STALE_AFTER: Duration = Duration::from_secs(6 * 60 * 60);
256    const TEMP_FILE_STALE_AFTER: Duration = RUN_MARKER_STALE_AFTER;
257    const WORKER_MAX_RUNTIME: Duration = Duration::from_secs(5 * 60 * 60);
258    const RUN_MARKER_FILE_NAME: &str = "rollout-compression.lock";
259    const MAX_CONCURRENT_COMPRESSION_JOBS: usize = 2;
260
261    #[derive(Default)]
262    struct CompressionStats {
263        scanned: usize,
264        compressed: usize,
265        skipped: usize,
266        failed: usize,
267    }
268
269    pub(super) struct CompressionRunMarker {
270        path: PathBuf,
271        remove_on_drop: bool,
272    }
273
274    impl CompressionRunMarker {
275        pub(super) fn try_claim(codex_home: &Path) -> io::Result<Option<Self>> {
276            let marker_dir = codex_home.join(".tmp");
277            std::fs::create_dir_all(marker_dir.as_path())?;
278            let path = marker_dir.join(RUN_MARKER_FILE_NAME);
279            match create_run_marker_file(path.as_path()) {
280                Ok(()) => return Ok(Some(Self::new(path))),
281                Err(err) if err.kind() == io::ErrorKind::AlreadyExists => {}
282                Err(err) => return Err(err),
283            }
284
285            let stale = std::fs::metadata(path.as_path())
286                .and_then(|metadata| metadata.modified())
287                .ok()
288                .and_then(|modified| SystemTime::now().duration_since(modified).ok())
289                .is_some_and(|age| age >= RUN_MARKER_STALE_AFTER);
290            if !stale {
291                return Ok(None);
292            }
293            match std::fs::remove_file(path.as_path()) {
294                Ok(()) => {}
295                Err(err) if err.kind() == io::ErrorKind::NotFound => {}
296                Err(err) => return Err(err),
297            }
298            match create_run_marker_file(path.as_path()) {
299                Ok(()) => Ok(Some(Self::new(path))),
300                Err(err) if err.kind() == io::ErrorKind::AlreadyExists => Ok(None),
301                Err(err) => Err(err),
302            }
303        }
304
305        fn new(path: PathBuf) -> Self {
306            Self {
307                path,
308                remove_on_drop: true,
309            }
310        }
311
312        pub(super) fn persist(mut self) {
313            self.remove_on_drop = false;
314        }
315    }
316
317    impl Drop for CompressionRunMarker {
318        fn drop(&mut self) {
319            if self.remove_on_drop {
320                let _ = std::fs::remove_file(self.path.as_path());
321            }
322        }
323    }
324
325    pub(super) fn spawn(codex_home: PathBuf) {
326        let Ok(handle) = tokio::runtime::Handle::try_current() else {
327            metrics::run("skipped_no_runtime");
328            warn!(
329                "failed to start rollout compression worker for {}: no Tokio runtime",
330                codex_home.display()
331            );
332            return;
333        };
334        handle.spawn(async move {
335            if let Err(err) = run(codex_home.clone()).await {
336                warn!(
337                    "rollout compression worker failed for {}: {err}",
338                    codex_home.display()
339                );
340            }
341        });
342    }
343
344    pub(super) async fn run(codex_home: PathBuf) -> io::Result<()> {
345        let marker = match CompressionRunMarker::try_claim(codex_home.as_path()) {
346            Ok(Some(marker)) => marker,
347            Ok(None) => {
348                metrics::run("skipped_already_running");
349                debug!(
350                    "rollout compression worker recently ran or is already running for {}",
351                    codex_home.display()
352                );
353                return Ok(());
354            }
355            Err(err) => {
356                metrics::run("failed");
357                return Err(err);
358            }
359        };
360
361        metrics::run("started");
362        let started_at = Instant::now();
363        let result = async {
364            cleanup_stale_temps(codex_home.as_path()).await?;
365            let Some(reference_index) = RolloutReferenceIndex::scan_until(
366                codex_home.as_path(),
367                started_at,
368                WORKER_MAX_RUNTIME,
369            )
370            .await?
371            else {
372                return Ok(CompressionStats::default());
373            };
374            let mut stats = CompressionStats::default();
375            for root in [
376                codex_home.join(ARCHIVED_SESSIONS_SUBDIR),
377                codex_home.join(SESSIONS_SUBDIR),
378            ] {
379                if started_at.elapsed() >= WORKER_MAX_RUNTIME {
380                    break;
381                }
382                compress_rollouts_in_root(root.as_path(), started_at, &reference_index, &mut stats)
383                    .await?;
384            }
385            Ok::<_, io::Error>(stats)
386        }
387        .await;
388        let stats = match result {
389            Ok(stats) => stats,
390            Err(err) => {
391                metrics::run("failed");
392                metrics::run_duration("failed", started_at.elapsed());
393                return Err(err);
394            }
395        };
396        info!(
397            "rollout compression worker finished: scanned={}, compressed={}, skipped={}, failed={}",
398            stats.scanned, stats.compressed, stats.skipped, stats.failed
399        );
400        metrics::run("completed");
401        metrics::run_duration("completed", started_at.elapsed());
402        marker.persist();
403        Ok(())
404    }
405
406    fn create_run_marker_file(path: &Path) -> io::Result<()> {
407        let mut file = std::fs::OpenOptions::new()
408            .write(true)
409            .create_new(true)
410            .open(path)?;
411        writeln!(
412            file,
413            "pid={} started_at={:?}",
414            std::process::id(),
415            SystemTime::now()
416        )?;
417        Ok(())
418    }
419
420    async fn compress_rollouts_in_root(
421        root: &Path,
422        started_at: Instant,
423        reference_index: &RolloutReferenceIndex,
424        stats: &mut CompressionStats,
425    ) -> io::Result<()> {
426        if !tokio::fs::try_exists(root).await.unwrap_or(false) {
427            return Ok(());
428        }
429        let mut stack = vec![root.to_path_buf()];
430        let mut jobs = JoinSet::new();
431        while let Some(dir) = stack.pop() {
432            if started_at.elapsed() >= WORKER_MAX_RUNTIME {
433                break;
434            }
435            let mut read_dir = match tokio::fs::read_dir(dir.as_path()).await {
436                Ok(read_dir) => read_dir,
437                Err(err) => {
438                    warn!(
439                        "failed to read rollout compression directory {}: {err}",
440                        dir.display()
441                    );
442                    continue;
443                }
444            };
445            loop {
446                let entry = match read_dir.next_entry().await {
447                    Ok(Some(entry)) => entry,
448                    Ok(None) => break,
449                    Err(err) => {
450                        drain_compression_jobs(&mut jobs, stats).await;
451                        return Err(err);
452                    }
453                };
454                if started_at.elapsed() >= WORKER_MAX_RUNTIME {
455                    break;
456                }
457                let path = entry.path();
458                let file_type = match entry.file_type().await {
459                    Ok(file_type) => file_type,
460                    Err(err) => {
461                        warn!(
462                            "failed to read rollout compression file type {}: {err}",
463                            path.display()
464                        );
465                        continue;
466                    }
467                };
468                if file_type.is_dir() {
469                    stack.push(path);
470                    continue;
471                }
472                if !file_type.is_file() {
473                    continue;
474                }
475                let Some(rollout_file) = RolloutFile::from_path(path) else {
476                    continue;
477                };
478                if rollout_file.is_compressed() {
479                    continue;
480                }
481                let path = rollout_file.into_path();
482                let Ok(meta) = crate::read_session_meta_line(path.as_path()).await else {
483                    stats.skipped = stats.skipped.saturating_add(1);
484                    metrics::file("skipped_unreadable_meta");
485                    continue;
486                };
487                if reference_index.reference_count(meta.meta.id) > 0 {
488                    stats.skipped = stats.skipped.saturating_add(1);
489                    metrics::file("skipped_referenced");
490                    continue;
491                }
492                if meta.meta.history_base.is_some() {
493                    stats.skipped = stats.skipped.saturating_add(1);
494                    metrics::file("skipped_fork_pointer");
495                    continue;
496                }
497                stats.scanned = stats.scanned.saturating_add(1);
498                metrics::file("scanned");
499                while jobs.len() >= MAX_CONCURRENT_COMPRESSION_JOBS {
500                    collect_next_compression_job(&mut jobs, stats).await;
501                }
502                jobs.spawn_blocking(move || {
503                    let started_at = Instant::now();
504                    let result = compress_rollout_if_cold_blocking(path.as_path());
505                    let duration = started_at.elapsed();
506                    (path, duration, result)
507                });
508            }
509        }
510        drain_compression_jobs(&mut jobs, stats).await;
511        Ok(())
512    }
513
514    type CompressionJobResult = (PathBuf, Duration, io::Result<CompressionMeasurement>);
515
516    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
517    enum CompressionOutcome {
518        Compressed,
519        SkippedNotCold,
520        SkippedChanged,
521        SkippedAlreadyCompressed,
522    }
523
524    impl CompressionOutcome {
525        fn tag(self) -> &'static str {
526            match self {
527                CompressionOutcome::Compressed => "compressed",
528                CompressionOutcome::SkippedNotCold => "skipped_not_cold",
529                CompressionOutcome::SkippedChanged => "skipped_changed",
530                CompressionOutcome::SkippedAlreadyCompressed => "skipped_already_compressed",
531            }
532        }
533    }
534
535    struct CompressionMeasurement {
536        outcome: CompressionOutcome,
537        source_bytes: Option<u64>,
538        compressed_bytes: Option<u64>,
539    }
540
541    impl CompressionMeasurement {
542        fn new(
543            outcome: CompressionOutcome,
544            source_bytes: Option<u64>,
545            compressed_bytes: Option<u64>,
546        ) -> Self {
547            Self {
548                outcome,
549                source_bytes,
550                compressed_bytes,
551            }
552        }
553    }
554
555    enum ColdFileState {
556        Cold(FileState),
557        NotCold(Option<FileState>),
558    }
559
560    async fn drain_compression_jobs(
561        jobs: &mut JoinSet<CompressionJobResult>,
562        stats: &mut CompressionStats,
563    ) {
564        while !jobs.is_empty() {
565            collect_next_compression_job(jobs, stats).await;
566        }
567    }
568
569    async fn collect_next_compression_job(
570        jobs: &mut JoinSet<CompressionJobResult>,
571        stats: &mut CompressionStats,
572    ) {
573        let Some(result) = jobs.join_next().await else {
574            return;
575        };
576        match result {
577            Ok((_, duration, Ok(measurement))) => {
578                let outcome = measurement.outcome;
579                match outcome {
580                    CompressionOutcome::Compressed => {
581                        stats.compressed = stats.compressed.saturating_add(1);
582                    }
583                    CompressionOutcome::SkippedNotCold
584                    | CompressionOutcome::SkippedChanged
585                    | CompressionOutcome::SkippedAlreadyCompressed => {
586                        stats.skipped = stats.skipped.saturating_add(1);
587                    }
588                }
589                metrics::file(outcome.tag());
590                metrics::file_duration(outcome.tag(), duration);
591                if let Some(source_bytes) = measurement.source_bytes {
592                    metrics::source_bytes(outcome.tag(), source_bytes);
593                }
594                if let Some(compressed_bytes) = measurement.compressed_bytes {
595                    metrics::compressed_bytes(outcome.tag(), compressed_bytes);
596                    if let Some(source_bytes) = measurement.source_bytes {
597                        metrics::compression_ratio(outcome.tag(), source_bytes, compressed_bytes);
598                    }
599                }
600            }
601            Ok((path, duration, Err(err))) => {
602                stats.failed = stats.failed.saturating_add(1);
603                metrics::file("failed");
604                metrics::file_duration("failed", duration);
605                warn!("failed to compress rollout {}: {err}", path.display());
606            }
607            Err(err) => {
608                stats.failed = stats.failed.saturating_add(1);
609                metrics::file("failed");
610                warn!("rollout compression task failed: {err}");
611            }
612        }
613    }
614
615    fn compress_rollout_if_cold_blocking(path: &Path) -> io::Result<CompressionMeasurement> {
616        let before = match cold_file_state(path)? {
617            ColdFileState::Cold(state) => state,
618            ColdFileState::NotCold(state) => {
619                return Ok(CompressionMeasurement::new(
620                    CompressionOutcome::SkippedNotCold,
621                    state.map(|state| state.len),
622                    /*compressed_bytes*/ None,
623                ));
624            }
625        };
626        let source_bytes = Some(before.len);
627        let compressed_path = path::compressed_rollout_path(path);
628        if compressed_path.exists() {
629            return Ok(CompressionMeasurement::new(
630                CompressionOutcome::SkippedAlreadyCompressed,
631                source_bytes,
632                /*compressed_bytes*/ None,
633            ));
634        }
635
636        let temp_dir = compressed_path
637            .parent()
638            .filter(|parent| !parent.as_os_str().is_empty())
639            .unwrap_or_else(|| Path::new("."));
640        std::fs::create_dir_all(temp_dir)?;
641        let mut temp_file = tempfile::Builder::new()
642            .prefix("rollout-compress-")
643            .suffix(TEMP_SUFFIX)
644            .tempfile_in(temp_dir)?;
645        encode_zstd_to_writer(path, temp_file.as_file_mut())?;
646        temp_file.as_file_mut().flush()?;
647        verify_zstd(temp_file.path())?;
648        if !same_file_state(path, &before)? {
649            return Ok(CompressionMeasurement::new(
650                CompressionOutcome::SkippedChanged,
651                source_bytes,
652                /*compressed_bytes*/ None,
653            ));
654        }
655        set_file_metadata(temp_file.as_file(), before.modified, &before.permissions)?;
656        temp_file.as_file().sync_all()?;
657        let compressed_bytes = temp_file.as_file().metadata()?.len();
658
659        match temp_file.persist_noclobber(compressed_path.as_path()) {
660            Ok(_) => {}
661            Err(err) if err.error.kind() == io::ErrorKind::AlreadyExists => {
662                return Ok(CompressionMeasurement::new(
663                    CompressionOutcome::SkippedAlreadyCompressed,
664                    source_bytes,
665                    /*compressed_bytes*/ None,
666                ));
667            }
668            Err(err) => return Err(err.error),
669        }
670        if !same_file_state(path, &before)? {
671            let _ = std::fs::remove_file(compressed_path.as_path());
672            return Ok(CompressionMeasurement::new(
673                CompressionOutcome::SkippedChanged,
674                source_bytes,
675                /*compressed_bytes*/ None,
676            ));
677        }
678        std::fs::remove_file(path)?;
679        Ok(CompressionMeasurement::new(
680            CompressionOutcome::Compressed,
681            source_bytes,
682            Some(compressed_bytes),
683        ))
684    }
685
686    struct FileState {
687        len: u64,
688        modified: SystemTime,
689        permissions: Permissions,
690    }
691
692    fn cold_file_state(path: &Path) -> io::Result<ColdFileState> {
693        let metadata = match std::fs::metadata(path) {
694            Ok(metadata) => metadata,
695            Err(err) if err.kind() == io::ErrorKind::NotFound => {
696                return Ok(ColdFileState::NotCold(None));
697            }
698            Err(err) => return Err(err),
699        };
700        if !metadata.is_file() {
701            return Ok(ColdFileState::NotCold(None));
702        }
703        let modified = metadata.modified()?;
704        let state = FileState {
705            len: metadata.len(),
706            modified,
707            permissions: metadata.permissions(),
708        };
709        let age = SystemTime::now()
710            .duration_since(modified)
711            .unwrap_or(Duration::ZERO);
712        if age < MIN_ROLLOUT_AGE {
713            return Ok(ColdFileState::NotCold(Some(state)));
714        }
715        Ok(ColdFileState::Cold(state))
716    }
717
718    fn same_file_state(path: &Path, expected: &FileState) -> io::Result<bool> {
719        match std::fs::metadata(path) {
720            Ok(metadata) => Ok(metadata.len() == expected.len
721                && metadata.modified()? == expected.modified
722                && metadata.permissions() == expected.permissions),
723            Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(false),
724            Err(err) => Err(err),
725        }
726    }
727
728    fn encode_zstd_to_writer(source: &Path, output: impl Write) -> io::Result<()> {
729        let mut input = File::open(source)?;
730        let mut encoder = zstd::stream::write::Encoder::new(output, COMPRESSION_LEVEL)?;
731        io::copy(&mut input, &mut encoder)?;
732        encoder.finish()?;
733        Ok(())
734    }
735
736    fn verify_zstd(path: &Path) -> io::Result<()> {
737        let input = File::open(path)?;
738        let mut decoder = zstd::stream::read::Decoder::new(input)?;
739        let mut sink = io::sink();
740        io::copy(&mut decoder, &mut sink)?;
741        Ok(())
742    }
743
744    fn set_file_metadata(
745        file: &File,
746        modified: SystemTime,
747        permissions: &Permissions,
748    ) -> io::Result<()> {
749        file.set_times(FileTimes::new().set_modified(modified))?;
750        file.set_permissions(permissions.clone())
751    }
752
753    async fn cleanup_stale_temps(codex_home: &Path) -> io::Result<()> {
754        for root in [
755            codex_home.join(SESSIONS_SUBDIR),
756            codex_home.join(ARCHIVED_SESSIONS_SUBDIR),
757        ] {
758            cleanup_stale_temps_in_root(root.as_path()).await?;
759        }
760        Ok(())
761    }
762
763    async fn cleanup_stale_temps_in_root(root: &Path) -> io::Result<()> {
764        if !tokio::fs::try_exists(root).await.unwrap_or(false) {
765            return Ok(());
766        }
767        let mut stack = vec![root.to_path_buf()];
768        while let Some(dir) = stack.pop() {
769            let mut read_dir = match tokio::fs::read_dir(dir.as_path()).await {
770                Ok(read_dir) => read_dir,
771                Err(err) => {
772                    warn!(
773                        "failed to read rollout temp cleanup directory {}: {err}",
774                        dir.display()
775                    );
776                    continue;
777                }
778            };
779            while let Some(entry) = read_dir.next_entry().await? {
780                let path = entry.path();
781                let file_type = match entry.file_type().await {
782                    Ok(file_type) => file_type,
783                    Err(err) => {
784                        warn!(
785                            "failed to read rollout temp cleanup file type {}: {err}",
786                            path.display()
787                        );
788                        continue;
789                    }
790                };
791                if file_type.is_dir() {
792                    stack.push(path);
793                    continue;
794                }
795                if file_type.is_file()
796                    && path
797                        .file_name()
798                        .and_then(OsStr::to_str)
799                        .is_some_and(|name| name.ends_with(TEMP_SUFFIX))
800                {
801                    let stale = entry
802                        .metadata()
803                        .await
804                        .ok()
805                        .and_then(|metadata| metadata.modified().ok())
806                        .and_then(|modified| SystemTime::now().duration_since(modified).ok())
807                        .is_some_and(|age| age >= TEMP_FILE_STALE_AFTER);
808                    if !stale {
809                        continue;
810                    }
811                    match tokio::fs::remove_file(path.as_path()).await {
812                        Ok(()) => metrics::temp_cleanup("removed"),
813                        Err(err) if err.kind() == io::ErrorKind::NotFound => {}
814                        Err(err) => {
815                            metrics::temp_cleanup("failed");
816                            warn!(
817                                "failed to remove stale rollout temp {}: {err}",
818                                path.display()
819                            );
820                        }
821                    }
822                }
823            }
824        }
825        Ok(())
826    }
827}
828
829mod metrics {
830    use std::time::Duration;
831
832    const FILE_COMPRESSED_BYTES_HISTOGRAM: &str = "codex.rollout_compression.file.compressed_bytes";
833    const FILE_COUNTER: &str = "codex.rollout_compression.file";
834    const FILE_DURATION_HISTOGRAM: &str = "codex.rollout_compression.file.duration_ms";
835    const FILE_SOURCE_BYTES_HISTOGRAM: &str = "codex.rollout_compression.file.source_bytes";
836    const FILE_COMPRESSION_RATIO_HISTOGRAM: &str =
837        "codex.rollout_compression.file.compression_ratio";
838    const MATERIALIZE_COUNTER: &str = "codex.rollout_compression.materialize";
839    const RUN_COUNTER: &str = "codex.rollout_compression.run";
840    const RUN_DURATION_HISTOGRAM: &str = "codex.rollout_compression.run.duration_ms";
841    const RATIO_BASIS_POINTS: u128 = 10_000;
842    const TEMP_CLEANUP_COUNTER: &str = "codex.rollout_compression.temp_cleanup";
843
844    pub(super) fn file(outcome: &'static str) {
845        counter(FILE_COUNTER, &[("outcome", outcome)]);
846    }
847
848    pub(super) fn file_duration(outcome: &'static str, duration: Duration) {
849        duration_histogram(FILE_DURATION_HISTOGRAM, duration, &[("outcome", outcome)]);
850    }
851
852    pub(super) fn source_bytes(outcome: &'static str, bytes: u64) {
853        histogram(
854            FILE_SOURCE_BYTES_HISTOGRAM,
855            saturating_i64(bytes),
856            &[("outcome", outcome)],
857        );
858    }
859
860    pub(super) fn compressed_bytes(outcome: &'static str, bytes: u64) {
861        histogram(
862            FILE_COMPRESSED_BYTES_HISTOGRAM,
863            saturating_i64(bytes),
864            &[("outcome", outcome)],
865        );
866    }
867
868    pub(super) fn compression_ratio(
869        outcome: &'static str,
870        source_bytes: u64,
871        compressed_bytes: u64,
872    ) {
873        if source_bytes == 0 {
874            return;
875        }
876        // Keep the ratio histogram integer-valued while preserving sub-percent precision.
877        let ratio = (u128::from(compressed_bytes) * RATIO_BASIS_POINTS) / u128::from(source_bytes);
878        histogram(
879            FILE_COMPRESSION_RATIO_HISTOGRAM,
880            saturating_i64(ratio),
881            &[("outcome", outcome)],
882        );
883    }
884
885    pub(super) fn materialize(outcome: &'static str) {
886        counter(MATERIALIZE_COUNTER, &[("outcome", outcome)]);
887    }
888
889    pub(super) fn run(status: &'static str) {
890        counter(RUN_COUNTER, &[("status", status)]);
891    }
892
893    pub(super) fn run_duration(status: &'static str, duration: Duration) {
894        duration_histogram(RUN_DURATION_HISTOGRAM, duration, &[("status", status)]);
895    }
896
897    pub(super) fn temp_cleanup(outcome: &'static str) {
898        counter(TEMP_CLEANUP_COUNTER, &[("outcome", outcome)]);
899    }
900
901    fn counter(name: &str, tags: &[(&str, &str)]) {
902        let Some(metrics) = codex_otel::global() else {
903            return;
904        };
905        let _ = metrics.counter(name, /*inc*/ 1, tags);
906    }
907
908    fn histogram(name: &str, value: i64, tags: &[(&str, &str)]) {
909        let Some(metrics) = codex_otel::global() else {
910            return;
911        };
912        let _ = metrics.histogram(name, value, tags);
913    }
914
915    fn duration_histogram(name: &str, duration: Duration, tags: &[(&str, &str)]) {
916        let Some(metrics) = codex_otel::global() else {
917            return;
918        };
919        let _ = metrics.record_duration(name, duration, tags);
920    }
921
922    fn saturating_i64(value: impl TryInto<i64>) -> i64 {
923        value.try_into().unwrap_or(i64::MAX)
924    }
925}
926
927/// Returns the existing rollout path, preferring the plain `.jsonl` file over
928/// its `.jsonl.zst` compressed sibling.
929pub async fn existing_rollout_path(path: &Path) -> Option<PathBuf> {
930    path::existing_rollout_path(path).await
931}
932
933mod path {
934    use std::ffi::OsStr;
935    use std::path::Path;
936    use std::path::PathBuf;
937
938    use super::COMPRESSED_SUFFIX;
939
940    pub(super) fn compressed_rollout_path(path: &Path) -> PathBuf {
941        if is_compressed_rollout_path(path) {
942            return path.to_path_buf();
943        }
944        let mut file_name = path
945            .file_name()
946            .map(OsStr::to_os_string)
947            .unwrap_or_else(|| OsStr::new("rollout.jsonl").to_os_string());
948        file_name.push(COMPRESSED_SUFFIX);
949        path.with_file_name(file_name)
950    }
951
952    pub(super) fn plain_rollout_path(path: &Path) -> PathBuf {
953        let Some(file_name) = path.file_name().and_then(OsStr::to_str) else {
954            return path.to_path_buf();
955        };
956        let Some(plain_file_name) = file_name.strip_suffix(COMPRESSED_SUFFIX) else {
957            return path.to_path_buf();
958        };
959        path.with_file_name(plain_file_name)
960    }
961
962    pub(super) fn is_compressed_rollout_path(path: &Path) -> bool {
963        path.file_name()
964            .and_then(OsStr::to_str)
965            .is_some_and(|name| name.ends_with(".jsonl.zst"))
966    }
967
968    pub(super) fn should_skip_compressed_sibling(path: &Path) -> bool {
969        is_compressed_rollout_path(path) && plain_rollout_path(path).exists()
970    }
971
972    pub(super) async fn existing_rollout_path(path: &Path) -> Option<PathBuf> {
973        let plain_path = plain_rollout_path(path);
974        if matches!(tokio::fs::metadata(plain_path.as_path()).await, Ok(metadata) if metadata.is_file())
975        {
976            return Some(plain_path);
977        }
978        let compressed_path = compressed_rollout_path(plain_path.as_path());
979        if matches!(tokio::fs::metadata(compressed_path.as_path()).await, Ok(metadata) if metadata.is_file())
980        {
981            return Some(compressed_path);
982        }
983        None
984    }
985}
986
987mod file_name {
988    use super::COMPRESSED_SUFFIX;
989
990    pub(super) fn parse_rollout_file_name(name: &str) -> Option<&str> {
991        let name = name.strip_suffix(COMPRESSED_SUFFIX).unwrap_or(name);
992        if name.starts_with("rollout-") && name.ends_with(".jsonl") {
993            Some(name)
994        } else {
995            None
996        }
997    }
998}
999
1000mod reader {
1001    use std::fs::File;
1002    use std::io;
1003    use std::io::BufRead;
1004    use std::io::Read;
1005    use std::path::Path;
1006
1007    use super::RolloutLineReader;
1008    use super::RolloutLineReaderInner;
1009    use super::path;
1010    use tokio::io::AsyncBufReadExt;
1011
1012    pub(super) async fn open_once(path: &Path) -> io::Result<RolloutLineReader> {
1013        let path = path::existing_rollout_path(path)
1014            .await
1015            .unwrap_or_else(|| path.to_path_buf());
1016        if path::is_compressed_rollout_path(path.as_path()) {
1017            let reader = tokio::task::spawn_blocking(move || {
1018                let input = File::open(path.as_path())?;
1019                let decoder = zstd::stream::read::Decoder::new(input)?;
1020                Ok::<_, io::Error>(
1021                    io::BufReader::new(Box::new(decoder) as Box<dyn Read + Send>).lines(),
1022                )
1023            })
1024            .await
1025            .map_err(io::Error::other)??;
1026            return Ok(RolloutLineReader {
1027                inner: RolloutLineReaderInner::Blocking(Some(reader)),
1028            });
1029        }
1030        let file = tokio::fs::File::open(path).await?;
1031        Ok(RolloutLineReader {
1032            inner: RolloutLineReaderInner::Plain(tokio::io::BufReader::new(file).lines()),
1033        })
1034    }
1035}
1036
1037#[cfg(unix)]
1038fn create_file_with_permissions(path: &Path, permissions: &Permissions) -> io::Result<File> {
1039    let file = std::fs::OpenOptions::new()
1040        .write(true)
1041        .create_new(true)
1042        .mode(permissions.mode() & 0o7777)
1043        .open(path)?;
1044    file.set_permissions(permissions.clone())?;
1045    Ok(file)
1046}
1047
1048#[cfg(not(unix))]
1049fn create_file_with_permissions(path: &Path, permissions: &Permissions) -> io::Result<File> {
1050    let file = std::fs::OpenOptions::new()
1051        .write(true)
1052        .create_new(true)
1053        .open(path)?;
1054    file.set_permissions(permissions.clone())?;
1055    Ok(file)
1056}
1057
1058fn temp_path_for(path: &Path, operation: &str) -> PathBuf {
1059    let mut file_name = path
1060        .file_name()
1061        .map(OsStr::to_os_string)
1062        .unwrap_or_else(|| OsStr::new("rollout").to_os_string());
1063    let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed);
1064    file_name.push(format!(
1065        ".{operation}.{}.{counter}{TEMP_SUFFIX}",
1066        std::process::id()
1067    ));
1068    path.with_file_name(file_name)
1069}
1070
1071#[cfg(test)]
1072#[path = "compression_tests.rs"]
1073mod tests;
codex_rollout/compression.rs

codex_rollout/
compression.rs
