nextest_runner/record/
reader.rs

1// Copyright (c) The nextest Contributors
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! Reading logic for recorded test runs.
5//!
6//! The [`RecordReader`] reads a recorded test run from disk, providing access
7//! to metadata and events stored during the run.
8//!
9//! The [`StoreReader`] trait provides a unified interface for reading from
10//! either on-disk stores or portable archives.
11
12use super::{
13    format::{
14        BUILD_SCOPE_JSON_PATH, CARGO_METADATA_JSON_PATH, OutputDict, RECORD_OPTS_JSON_PATH,
15        RUN_LOG_FILE_NAME, STDERR_DICT_PATH, STDOUT_DICT_PATH, STORE_ZIP_FILE_NAME,
16        TEST_LIST_JSON_PATH,
17    },
18    summary::{RecordOpts, TestEventSummary},
19};
20use crate::{
21    errors::RecordReadError,
22    output_spec::RecordingSpec,
23    record::format::{RERUN_INFO_JSON_PATH, RecordedBuildScope, RerunInfo},
24    user_config::elements::MAX_MAX_OUTPUT_SIZE,
25};
26use camino::{Utf8Path, Utf8PathBuf};
27use debug_ignore::DebugIgnore;
28use eazip::Archive;
29use nextest_metadata::TestListSummary;
30use std::{
31    fs::File,
32    io::{self, BufRead, BufReader, Read},
33};
34
35/// Trait for reading from a recorded run's store.
36///
37/// This trait abstracts over reading from either an on-disk store directory
38/// (via [`RecordReader`]) or from an inner store.zip within a portable archive
39/// (via [`PortableStoreReader`](super::portable::PortableStoreReader)).
40pub trait StoreReader {
41    /// Returns the cargo metadata JSON from the store.
42    fn read_cargo_metadata(&mut self) -> Result<String, RecordReadError>;
43
44    /// Returns the test list summary from the store.
45    fn read_test_list(&mut self) -> Result<TestListSummary, RecordReadError>;
46
47    /// Returns the record options from the store.
48    fn read_record_opts(&mut self) -> Result<RecordOpts, RecordReadError>;
49
50    /// Returns the build scope for this run from the store.
51    fn read_build_scope(&mut self) -> Result<RecordedBuildScope, RecordReadError>;
52
53    /// Returns the rerun info from the store, if this is a rerun.
54    ///
55    /// Returns `Ok(None)` if this run is not a rerun (the file doesn't exist).
56    fn read_rerun_info(&mut self) -> Result<Option<RerunInfo>, RecordReadError>;
57
58    /// Loads the dictionaries from the store.
59    ///
60    /// This must be called before reading output files.
61    fn load_dictionaries(&mut self) -> Result<(), RecordReadError>;
62
63    /// Reads output for a specific file from the store.
64    ///
65    /// The `file_name` should be the value from `ZipStoreOutput::file_name`,
66    /// e.g., "test-abc123-1-stdout".
67    ///
68    /// # Panics
69    ///
70    /// Panics if [`load_dictionaries`](Self::load_dictionaries) has not been called first.
71    fn read_output(&mut self, file_name: &str) -> Result<Vec<u8>, RecordReadError>;
72
73    /// Extracts a file from the store to a path, streaming directly.
74    ///
75    /// The `store_path` is relative to the store root (e.g., `meta/test-list.json`).
76    /// Returns the number of bytes written.
77    fn extract_file_to_path(
78        &mut self,
79        store_path: &str,
80        output_path: &Utf8Path,
81    ) -> Result<u64, RecordReadError>;
82}
83
84/// Reader for a recorded test run.
85///
86/// Provides access to the metadata and events stored during a test run.
87/// The archive is opened lazily when methods are called.
88#[derive(Debug)]
89pub struct RecordReader {
90    run_dir: Utf8PathBuf,
91    archive: Option<DebugIgnore<Archive<BufReader<File>>>>,
92    /// Cached stdout dictionary loaded from the archive.
93    stdout_dict: Option<Vec<u8>>,
94    /// Cached stderr dictionary loaded from the archive.
95    stderr_dict: Option<Vec<u8>>,
96}
97
98impl RecordReader {
99    /// Opens a recorded run from its directory.
100    ///
101    /// The directory should contain `store.zip` and `run.log.zst`.
102    pub fn open(run_dir: &Utf8Path) -> Result<Self, RecordReadError> {
103        if !run_dir.exists() {
104            return Err(RecordReadError::RunNotFound {
105                path: run_dir.to_owned(),
106            });
107        }
108
109        Ok(Self {
110            run_dir: run_dir.to_owned(),
111            archive: None,
112            stdout_dict: None,
113            stderr_dict: None,
114        })
115    }
116
117    /// Returns the path to the run directory.
118    pub fn run_dir(&self) -> &Utf8Path {
119        &self.run_dir
120    }
121
122    /// Opens the zip archive if not already open.
123    fn ensure_archive(&mut self) -> Result<&mut Archive<BufReader<File>>, RecordReadError> {
124        if self.archive.is_none() {
125            let store_path = self.run_dir.join(STORE_ZIP_FILE_NAME);
126            let file = File::open(&store_path).map_err(|error| RecordReadError::OpenArchive {
127                path: store_path.clone(),
128                error,
129            })?;
130            let archive = Archive::new(BufReader::new(file)).map_err(|error| {
131                RecordReadError::ParseArchive {
132                    path: store_path,
133                    error,
134                }
135            })?;
136            self.archive = Some(DebugIgnore(archive));
137        }
138        Ok(self.archive.as_mut().expect("archive was just set"))
139    }
140
141    /// Reads a file from the archive as bytes, with size limit.
142    ///
143    /// The size limit prevents malicious archives from causing OOM by
144    /// specifying a huge decompressed size. The limit is checked against the
145    /// claimed size in the ZIP header, and `take()` is used during decompression
146    /// to guard against spoofed headers.
147    ///
148    /// Since nextest controls archive creation, any mismatch between the header
149    /// size and actual size indicates corruption or tampering.
150    fn read_archive_file(&mut self, file_name: &str) -> Result<Vec<u8>, RecordReadError> {
151        let limit = MAX_MAX_OUTPUT_SIZE.as_u64();
152        let archive = self.ensure_archive()?;
153        let mut file =
154            archive
155                .get_by_name(file_name)
156                .ok_or_else(|| RecordReadError::FileNotFound {
157                    file_name: file_name.to_string(),
158                })?;
159
160        let claimed_size = file.metadata().uncompressed_size;
161        if claimed_size > limit {
162            return Err(RecordReadError::FileTooLarge {
163                file_name: file_name.to_string(),
164                size: claimed_size,
165                limit,
166            });
167        }
168
169        let capacity = usize::try_from(claimed_size).unwrap_or(usize::MAX);
170        let mut contents = Vec::with_capacity(capacity);
171
172        // file.read() returns a reader that decompresses and verifies CRC32 +
173        // size. The take(limit) is a safety belt against spoofed headers.
174        file.read()
175            .and_then(|reader| reader.take(limit).read_to_end(&mut contents))
176            .map_err(|error| RecordReadError::Decompress {
177                file_name: file_name.to_string(),
178                error,
179            })?;
180
181        let actual_size = contents.len() as u64;
182        if actual_size != claimed_size {
183            return Err(RecordReadError::SizeMismatch {
184                file_name: file_name.to_string(),
185                claimed_size,
186                actual_size,
187            });
188        }
189
190        Ok(contents)
191    }
192
193    /// Returns the cargo metadata JSON from the archive.
194    pub fn read_cargo_metadata(&mut self) -> Result<String, RecordReadError> {
195        let bytes = self.read_archive_file(CARGO_METADATA_JSON_PATH)?;
196        String::from_utf8(bytes).map_err(|e| RecordReadError::Decompress {
197            file_name: CARGO_METADATA_JSON_PATH.to_string(),
198            error: std::io::Error::new(std::io::ErrorKind::InvalidData, e),
199        })
200    }
201
202    /// Returns the test list from the archive.
203    pub fn read_test_list(&mut self) -> Result<TestListSummary, RecordReadError> {
204        let bytes = self.read_archive_file(TEST_LIST_JSON_PATH)?;
205        serde_json::from_slice(&bytes).map_err(|error| RecordReadError::DeserializeMetadata {
206            file_name: TEST_LIST_JSON_PATH.to_string(),
207            error,
208        })
209    }
210
211    /// Returns the record options from the archive.
212    pub fn read_record_opts(&mut self) -> Result<RecordOpts, RecordReadError> {
213        let bytes = self.read_archive_file(RECORD_OPTS_JSON_PATH)?;
214        serde_json::from_slice(&bytes).map_err(|error| RecordReadError::DeserializeMetadata {
215            file_name: RECORD_OPTS_JSON_PATH.to_string(),
216            error,
217        })
218    }
219
220    /// Returns the build scope from the archive.
221    pub fn read_build_scope(&mut self) -> Result<RecordedBuildScope, RecordReadError> {
222        let bytes = self.read_archive_file(BUILD_SCOPE_JSON_PATH)?;
223        serde_json::from_slice(&bytes).map_err(|error| RecordReadError::DeserializeMetadata {
224            file_name: BUILD_SCOPE_JSON_PATH.to_string(),
225            error,
226        })
227    }
228
229    /// Returns the rerun info from the archive, if this is a rerun.
230    ///
231    /// Returns `Ok(None)` if this run is not a rerun (the file doesn't exist).
232    /// Returns `Err` if the file exists but cannot be read or parsed.
233    pub fn read_rerun_info(&mut self) -> Result<Option<RerunInfo>, RecordReadError> {
234        match self.read_archive_file(RERUN_INFO_JSON_PATH) {
235            Ok(bytes) => {
236                let info = serde_json::from_slice(&bytes).map_err(|error| {
237                    RecordReadError::DeserializeMetadata {
238                        file_name: RERUN_INFO_JSON_PATH.to_string(),
239                        error,
240                    }
241                })?;
242                Ok(Some(info))
243            }
244            Err(RecordReadError::FileNotFound { .. }) => {
245                // File doesn't exist; this is not a rerun.
246                Ok(None)
247            }
248            Err(e) => Err(e),
249        }
250    }
251
252    /// Loads the dictionaries from the archive.
253    ///
254    /// This must be called before reading output files. The dictionaries are
255    /// used for decompressing test output.
256    ///
257    /// Note: The store format version is checked before opening the archive,
258    /// using the `store_format_version` field in runs.json.zst. This method
259    /// assumes the version has already been validated.
260    pub fn load_dictionaries(&mut self) -> Result<(), RecordReadError> {
261        self.stdout_dict = Some(self.read_archive_file(STDOUT_DICT_PATH)?);
262        self.stderr_dict = Some(self.read_archive_file(STDERR_DICT_PATH)?);
263        Ok(())
264    }
265
266    /// Returns an iterator over events in the run log.
267    ///
268    /// Events are read one at a time from the zstd-compressed JSON Lines file.
269    pub fn events(&self) -> Result<RecordEventIter, RecordReadError> {
270        let log_path = self.run_dir.join(RUN_LOG_FILE_NAME);
271        let file = File::open(&log_path).map_err(|error| RecordReadError::OpenRunLog {
272            path: log_path.clone(),
273            error,
274        })?;
275        let decoder =
276            zstd::stream::Decoder::new(file).map_err(|error| RecordReadError::OpenRunLog {
277                path: log_path,
278                error,
279            })?;
280        Ok(RecordEventIter {
281            reader: DebugIgnore(BufReader::new(decoder)),
282            line_buf: String::new(),
283            line_number: 0,
284        })
285    }
286
287    /// Reads output for a specific file from the archive.
288    ///
289    /// The `file_name` should be the value from `ZipStoreOutput::file_name`,
290    /// e.g., "test-abc123-1-stdout".
291    ///
292    /// The [`OutputFileName`](crate::record::OutputFileName) type ensures that
293    /// file names are validated during deserialization, preventing path traversal.
294    ///
295    /// # Panics
296    ///
297    /// Panics if [`load_dictionaries`](Self::load_dictionaries) has not been called first.
298    pub fn read_output(&mut self, file_name: &str) -> Result<Vec<u8>, RecordReadError> {
299        let path = format!("out/{file_name}");
300        let compressed = self.read_archive_file(&path)?;
301        let limit = MAX_MAX_OUTPUT_SIZE.as_u64();
302
303        // Output files are stored pre-compressed with zstd dictionaries.
304        // Unknown file types indicate a format revision that should have been
305        // rejected during version validation.
306        let dict_bytes = self.get_dict_for_output(file_name).ok_or_else(|| {
307            RecordReadError::UnknownOutputType {
308                file_name: file_name.to_owned(),
309            }
310        })?;
311
312        decompress_with_dict(&compressed, dict_bytes, limit).map_err(|error| {
313            RecordReadError::Decompress {
314                file_name: path,
315                error,
316            }
317        })
318    }
319
320    /// Returns the dictionary bytes for the given output file name, if known.
321    ///
322    /// Returns `None` for unknown file types, which indicates a format revision
323    /// that should have been rejected during version validation.
324    ///
325    /// # Panics
326    ///
327    /// Panics if [`load_dictionaries`](Self::load_dictionaries) has not been called first.
328    fn get_dict_for_output(&self, file_name: &str) -> Option<&[u8]> {
329        match OutputDict::for_output_file_name(file_name) {
330            OutputDict::Stdout => Some(
331                self.stdout_dict
332                    .as_ref()
333                    .expect("load_dictionaries must be called first"),
334            ),
335            OutputDict::Stderr => Some(
336                self.stderr_dict
337                    .as_ref()
338                    .expect("load_dictionaries must be called first"),
339            ),
340            OutputDict::None => None,
341        }
342    }
343}
344
345impl StoreReader for RecordReader {
346    fn read_cargo_metadata(&mut self) -> Result<String, RecordReadError> {
347        RecordReader::read_cargo_metadata(self)
348    }
349
350    fn read_test_list(&mut self) -> Result<TestListSummary, RecordReadError> {
351        RecordReader::read_test_list(self)
352    }
353
354    fn read_record_opts(&mut self) -> Result<RecordOpts, RecordReadError> {
355        RecordReader::read_record_opts(self)
356    }
357
358    fn read_build_scope(&mut self) -> Result<RecordedBuildScope, RecordReadError> {
359        RecordReader::read_build_scope(self)
360    }
361
362    fn read_rerun_info(&mut self) -> Result<Option<RerunInfo>, RecordReadError> {
363        RecordReader::read_rerun_info(self)
364    }
365
366    fn load_dictionaries(&mut self) -> Result<(), RecordReadError> {
367        RecordReader::load_dictionaries(self)
368    }
369
370    fn read_output(&mut self, file_name: &str) -> Result<Vec<u8>, RecordReadError> {
371        RecordReader::read_output(self, file_name)
372    }
373
374    fn extract_file_to_path(
375        &mut self,
376        store_path: &str,
377        output_path: &Utf8Path,
378    ) -> Result<u64, RecordReadError> {
379        let archive = self.ensure_archive()?;
380        let mut file =
381            archive
382                .get_by_name(store_path)
383                .ok_or_else(|| RecordReadError::FileNotFound {
384                    file_name: store_path.to_owned(),
385                })?;
386
387        let mut output_file =
388            File::create(output_path).map_err(|error| RecordReadError::ExtractFile {
389                store_path: store_path.to_owned(),
390                output_path: output_path.to_owned(),
391                error,
392            })?;
393
394        // file.read() decompresses and verifies CRC32 + size.
395        let mut reader = file
396            .read()
397            .map_err(|error| RecordReadError::ReadArchiveFile {
398                file_name: store_path.to_owned(),
399                error,
400            })?;
401
402        io::copy(&mut reader, &mut output_file).map_err(|error| RecordReadError::ExtractFile {
403            store_path: store_path.to_owned(),
404            output_path: output_path.to_owned(),
405            error,
406        })
407    }
408}
409
410/// Decompresses data using a pre-trained zstd dictionary, with a size limit.
411///
412/// The limit prevents compression bombs where a small compressed payload
413/// expands to an extremely large decompressed output.
414pub(super) fn decompress_with_dict(
415    compressed: &[u8],
416    dict_bytes: &[u8],
417    limit: u64,
418) -> std::io::Result<Vec<u8>> {
419    let dict = zstd::dict::DecoderDictionary::copy(dict_bytes);
420    let decoder = zstd::stream::Decoder::with_prepared_dictionary(compressed, &dict)?;
421    let mut decompressed = Vec::new();
422    decoder.take(limit).read_to_end(&mut decompressed)?;
423    Ok(decompressed)
424}
425
426/// Zstd decoder reading from a file.
427type LogDecoder = zstd::stream::Decoder<'static, BufReader<File>>;
428
429/// Iterator over recorded events.
430///
431/// Reads events one at a time from the zstd-compressed JSON Lines run log.
432#[derive(Debug)]
433pub struct RecordEventIter {
434    reader: DebugIgnore<BufReader<LogDecoder>>,
435    line_buf: String,
436    line_number: usize,
437}
438
439impl Iterator for RecordEventIter {
440    type Item = Result<TestEventSummary<RecordingSpec>, RecordReadError>;
441
442    fn next(&mut self) -> Option<Self::Item> {
443        loop {
444            self.line_buf.clear();
445            self.line_number += 1;
446
447            match self.reader.read_line(&mut self.line_buf) {
448                Ok(0) => return None,
449                Ok(_) => {
450                    let trimmed = self.line_buf.trim();
451                    if trimmed.is_empty() {
452                        continue;
453                    }
454                    return Some(serde_json::from_str(trimmed).map_err(|error| {
455                        RecordReadError::ParseEvent {
456                            line_number: self.line_number,
457                            error,
458                        }
459                    }));
460                }
461                Err(error) => {
462                    return Some(Err(RecordReadError::ReadRunLog {
463                        line_number: self.line_number,
464                        error,
465                    }));
466                }
467            }
468        }
469    }
470}
471
472#[cfg(test)]
473mod tests {
474    use super::*;
475
476    #[test]
477    fn test_record_reader_nonexistent_dir() {
478        let result = RecordReader::open(Utf8Path::new("/nonexistent/path"));
479        assert!(matches!(result, Err(RecordReadError::RunNotFound { .. })));
480    }
481}
nextest_runner/record/reader.rs

nextest_runner/record/
reader.rs
