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    redact::SizeDisplay,
25    user_config::elements::MAX_MAX_OUTPUT_SIZE,
26};
27use bytesize::ByteSize;
28use camino::{Utf8Path, Utf8PathBuf};
29use debug_ignore::DebugIgnore;
30use eazip::Archive;
31use nextest_metadata::TestListSummary;
32use std::{
33    fs::File,
34    io::{self, BufRead, BufReader, Read},
35};
36
37/// Trait for reading from a recorded run's store.
38///
39/// This trait abstracts over reading from either an on-disk store directory
40/// (via [`RecordReader`]) or from an inner store.zip within a portable archive
41/// (via [`PortableStoreReader`](super::portable::PortableStoreReader)).
42pub trait StoreReader {
43    /// Returns the cargo metadata JSON from the store.
44    fn read_cargo_metadata(&mut self) -> Result<String, RecordReadError>;
45
46    /// Returns the test list summary from the store.
47    fn read_test_list(&mut self) -> Result<TestListSummary, RecordReadError>;
48
49    /// Returns the record options from the store.
50    fn read_record_opts(&mut self) -> Result<RecordOpts, RecordReadError>;
51
52    /// Returns the build scope for this run from the store.
53    fn read_build_scope(&mut self) -> Result<RecordedBuildScope, RecordReadError>;
54
55    /// Returns the rerun info from the store, if this is a rerun.
56    ///
57    /// Returns `Ok(None)` if this run is not a rerun (the file doesn't exist).
58    fn read_rerun_info(&mut self) -> Result<Option<RerunInfo>, RecordReadError>;
59
60    /// Loads the dictionaries from the store.
61    ///
62    /// This must be called before reading output files.
63    fn load_dictionaries(&mut self) -> Result<(), RecordReadError>;
64
65    /// Reads output for a specific file from the store.
66    ///
67    /// The `file_name` should be the value from `ZipStoreOutput::file_name`,
68    /// e.g., "test-abc123-1-stdout".
69    ///
70    /// # Panics
71    ///
72    /// Panics if [`load_dictionaries`](Self::load_dictionaries) has not been called first.
73    fn read_output(&mut self, file_name: &str) -> Result<Vec<u8>, RecordReadError>;
74
75    /// Extracts a file from the store to a path, streaming directly.
76    ///
77    /// The `store_path` is relative to the store root (e.g., `meta/test-list.json`).
78    /// Returns the number of bytes written.
79    fn extract_file_to_path(
80        &mut self,
81        store_path: &str,
82        output_path: &Utf8Path,
83    ) -> Result<u64, RecordReadError>;
84}
85
86/// Reader for a recorded test run.
87///
88/// Provides access to the metadata and events stored during a test run.
89/// The archive is opened lazily when methods are called.
90#[derive(Debug)]
91pub struct RecordReader {
92    run_dir: Utf8PathBuf,
93    archive: Option<DebugIgnore<Archive<BufReader<File>>>>,
94    /// Cached stdout dictionary loaded from the archive.
95    stdout_dict: Option<Vec<u8>>,
96    /// Cached stderr dictionary loaded from the archive.
97    stderr_dict: Option<Vec<u8>>,
98}
99
100impl RecordReader {
101    /// Opens a recorded run from its directory.
102    ///
103    /// The directory should contain `store.zip` and `run.log.zst`.
104    pub fn open(run_dir: &Utf8Path) -> Result<Self, RecordReadError> {
105        if !run_dir.exists() {
106            return Err(RecordReadError::RunNotFound {
107                path: run_dir.to_owned(),
108            });
109        }
110
111        Ok(Self {
112            run_dir: run_dir.to_owned(),
113            archive: None,
114            stdout_dict: None,
115            stderr_dict: None,
116        })
117    }
118
119    /// Returns the path to the run directory.
120    pub fn run_dir(&self) -> &Utf8Path {
121        &self.run_dir
122    }
123
124    /// Opens the zip archive if not already open.
125    fn ensure_archive(&mut self) -> Result<&mut Archive<BufReader<File>>, RecordReadError> {
126        if self.archive.is_none() {
127            let store_path = self.run_dir.join(STORE_ZIP_FILE_NAME);
128            let file = File::open(&store_path).map_err(|error| RecordReadError::OpenArchive {
129                path: store_path.clone(),
130                error,
131            })?;
132            let archive = Archive::new(BufReader::new(file)).map_err(|error| {
133                RecordReadError::ParseArchive {
134                    path: store_path,
135                    error,
136                }
137            })?;
138            self.archive = Some(DebugIgnore(archive));
139        }
140        Ok(self.archive.as_mut().expect("archive was just set"))
141    }
142
143    /// Reads a file from the archive as bytes, with size limit.
144    ///
145    /// The size limit prevents malicious archives from causing OOM by
146    /// specifying a huge decompressed size. The limit is checked against the
147    /// claimed size in the ZIP header, and `take()` is used during decompression
148    /// to guard against spoofed headers.
149    ///
150    /// Since nextest controls archive creation, any mismatch between the header
151    /// size and actual size indicates corruption or tampering.
152    fn read_archive_file(&mut self, file_name: &str) -> Result<Vec<u8>, RecordReadError> {
153        let limit = MAX_MAX_OUTPUT_SIZE.as_u64();
154        let archive = self.ensure_archive()?;
155        let mut file =
156            archive
157                .get_by_name(file_name)
158                .ok_or_else(|| RecordReadError::FileNotFound {
159                    file_name: file_name.to_string(),
160                })?;
161
162        let claimed_size = file.metadata().uncompressed_size;
163        if claimed_size > limit {
164            return Err(RecordReadError::FileTooLarge {
165                file_name: file_name.to_string(),
166                size: claimed_size,
167                limit,
168            });
169        }
170
171        let capacity = usize::try_from(claimed_size).unwrap_or(usize::MAX);
172        let mut contents = Vec::with_capacity(capacity);
173
174        // file.read() returns a reader that decompresses and verifies CRC32 +
175        // size. The take(limit) is a safety belt against spoofed headers.
176        file.read()
177            .and_then(|reader| reader.take(limit).read_to_end(&mut contents))
178            .map_err(|error| RecordReadError::Decompress {
179                file_name: file_name.to_string(),
180                error,
181            })?;
182
183        let actual_size = contents.len() as u64;
184        if actual_size != claimed_size {
185            return Err(RecordReadError::SizeMismatch {
186                file_name: file_name.to_string(),
187                claimed_size,
188                actual_size,
189            });
190        }
191
192        Ok(contents)
193    }
194
195    /// Returns the cargo metadata JSON from the archive.
196    pub fn read_cargo_metadata(&mut self) -> Result<String, RecordReadError> {
197        let bytes = self.read_archive_file(CARGO_METADATA_JSON_PATH)?;
198        String::from_utf8(bytes).map_err(|e| RecordReadError::Decompress {
199            file_name: CARGO_METADATA_JSON_PATH.to_string(),
200            error: std::io::Error::new(std::io::ErrorKind::InvalidData, e),
201        })
202    }
203
204    /// Returns the test list from the archive.
205    pub fn read_test_list(&mut self) -> Result<TestListSummary, RecordReadError> {
206        let bytes = self.read_archive_file(TEST_LIST_JSON_PATH)?;
207        serde_json::from_slice(&bytes).map_err(|error| RecordReadError::DeserializeMetadata {
208            file_name: TEST_LIST_JSON_PATH.to_string(),
209            error,
210        })
211    }
212
213    /// Returns the record options from the archive.
214    pub fn read_record_opts(&mut self) -> Result<RecordOpts, RecordReadError> {
215        let bytes = self.read_archive_file(RECORD_OPTS_JSON_PATH)?;
216        serde_json::from_slice(&bytes).map_err(|error| RecordReadError::DeserializeMetadata {
217            file_name: RECORD_OPTS_JSON_PATH.to_string(),
218            error,
219        })
220    }
221
222    /// Returns the build scope from the archive.
223    pub fn read_build_scope(&mut self) -> Result<RecordedBuildScope, RecordReadError> {
224        let bytes = self.read_archive_file(BUILD_SCOPE_JSON_PATH)?;
225        serde_json::from_slice(&bytes).map_err(|error| RecordReadError::DeserializeMetadata {
226            file_name: BUILD_SCOPE_JSON_PATH.to_string(),
227            error,
228        })
229    }
230
231    /// Returns the rerun info from the archive, if this is a rerun.
232    ///
233    /// Returns `Ok(None)` if this run is not a rerun (the file doesn't exist).
234    /// Returns `Err` if the file exists but cannot be read or parsed.
235    pub fn read_rerun_info(&mut self) -> Result<Option<RerunInfo>, RecordReadError> {
236        match self.read_archive_file(RERUN_INFO_JSON_PATH) {
237            Ok(bytes) => {
238                let info = serde_json::from_slice(&bytes).map_err(|error| {
239                    RecordReadError::DeserializeMetadata {
240                        file_name: RERUN_INFO_JSON_PATH.to_string(),
241                        error,
242                    }
243                })?;
244                Ok(Some(info))
245            }
246            Err(RecordReadError::FileNotFound { .. }) => {
247                // File doesn't exist; this is not a rerun.
248                Ok(None)
249            }
250            Err(e) => Err(e),
251        }
252    }
253
254    /// Loads the dictionaries from the archive.
255    ///
256    /// This must be called before reading output files. The dictionaries are
257    /// used for decompressing test output.
258    ///
259    /// Note: The store format version is checked before opening the archive,
260    /// using the `store_format_version` field in runs.json.zst. This method
261    /// assumes the version has already been validated.
262    pub fn load_dictionaries(&mut self) -> Result<(), RecordReadError> {
263        self.stdout_dict = Some(self.read_archive_file(STDOUT_DICT_PATH)?);
264        self.stderr_dict = Some(self.read_archive_file(STDERR_DICT_PATH)?);
265        Ok(())
266    }
267
268    /// Returns an iterator over events in the run log.
269    ///
270    /// Events are read one at a time from the zstd-compressed JSON Lines file.
271    pub fn events(&self) -> Result<RunLogEventIter<BufReader<LogDecoder>>, RecordReadError> {
272        let log_path = self.run_dir.join(RUN_LOG_FILE_NAME);
273        let file = File::open(&log_path).map_err(|error| RecordReadError::OpenRunLog {
274            path: log_path.clone(),
275            error,
276        })?;
277        let decoder =
278            zstd::stream::Decoder::new(file).map_err(|error| RecordReadError::OpenRunLog {
279                path: log_path,
280                error,
281            })?;
282        Ok(RunLogEventIter::new(BufReader::new(decoder)))
283    }
284
285    /// Reads output for a specific file from the archive.
286    ///
287    /// The `file_name` should be the value from `ZipStoreOutput::file_name`,
288    /// e.g., "test-abc123-1-stdout".
289    ///
290    /// The [`OutputFileName`](crate::record::OutputFileName) type ensures that
291    /// file names are validated during deserialization, preventing path traversal.
292    ///
293    /// # Panics
294    ///
295    /// Panics if [`load_dictionaries`](Self::load_dictionaries) has not been called first.
296    pub fn read_output(&mut self, file_name: &str) -> Result<Vec<u8>, RecordReadError> {
297        let path = format!("out/{file_name}");
298        let compressed = self.read_archive_file(&path)?;
299        let limit = MAX_MAX_OUTPUT_SIZE.as_u64();
300
301        // Output files are stored pre-compressed with zstd dictionaries.
302        // Unknown file types indicate a format revision that should have been
303        // rejected during version validation.
304        let dict_bytes = self.get_dict_for_output(file_name).ok_or_else(|| {
305            RecordReadError::UnknownOutputType {
306                file_name: file_name.to_owned(),
307            }
308        })?;
309
310        decompress_with_dict(&compressed, dict_bytes, limit).map_err(|error| {
311            RecordReadError::Decompress {
312                file_name: path,
313                error,
314            }
315        })
316    }
317
318    /// Returns the dictionary bytes for the given output file name, if known.
319    ///
320    /// Returns `None` for unknown file types, which indicates a format revision
321    /// that should have been rejected during version validation.
322    ///
323    /// # Panics
324    ///
325    /// Panics if [`load_dictionaries`](Self::load_dictionaries) has not been called first.
326    fn get_dict_for_output(&self, file_name: &str) -> Option<&[u8]> {
327        match OutputDict::for_output_file_name(file_name) {
328            OutputDict::Stdout => Some(
329                self.stdout_dict
330                    .as_ref()
331                    .expect("load_dictionaries must be called first"),
332            ),
333            OutputDict::Stderr => Some(
334                self.stderr_dict
335                    .as_ref()
336                    .expect("load_dictionaries must be called first"),
337            ),
338            OutputDict::None => None,
339        }
340    }
341}
342
343impl StoreReader for RecordReader {
344    fn read_cargo_metadata(&mut self) -> Result<String, RecordReadError> {
345        RecordReader::read_cargo_metadata(self)
346    }
347
348    fn read_test_list(&mut self) -> Result<TestListSummary, RecordReadError> {
349        RecordReader::read_test_list(self)
350    }
351
352    fn read_record_opts(&mut self) -> Result<RecordOpts, RecordReadError> {
353        RecordReader::read_record_opts(self)
354    }
355
356    fn read_build_scope(&mut self) -> Result<RecordedBuildScope, RecordReadError> {
357        RecordReader::read_build_scope(self)
358    }
359
360    fn read_rerun_info(&mut self) -> Result<Option<RerunInfo>, RecordReadError> {
361        RecordReader::read_rerun_info(self)
362    }
363
364    fn load_dictionaries(&mut self) -> Result<(), RecordReadError> {
365        RecordReader::load_dictionaries(self)
366    }
367
368    fn read_output(&mut self, file_name: &str) -> Result<Vec<u8>, RecordReadError> {
369        RecordReader::read_output(self, file_name)
370    }
371
372    fn extract_file_to_path(
373        &mut self,
374        store_path: &str,
375        output_path: &Utf8Path,
376    ) -> Result<u64, RecordReadError> {
377        let archive = self.ensure_archive()?;
378        let mut file =
379            archive
380                .get_by_name(store_path)
381                .ok_or_else(|| RecordReadError::FileNotFound {
382                    file_name: store_path.to_owned(),
383                })?;
384
385        let mut output_file =
386            File::create(output_path).map_err(|error| RecordReadError::ExtractFile {
387                store_path: store_path.to_owned(),
388                output_path: output_path.to_owned(),
389                error,
390            })?;
391
392        // file.read() decompresses and verifies CRC32 + size.
393        let mut reader = file
394            .read()
395            .map_err(|error| RecordReadError::ReadArchiveFile {
396                file_name: store_path.to_owned(),
397                error,
398            })?;
399
400        io::copy(&mut reader, &mut output_file).map_err(|error| RecordReadError::ExtractFile {
401            store_path: store_path.to_owned(),
402            output_path: output_path.to_owned(),
403            error,
404        })
405    }
406}
407
408/// Decompresses data using a pre-trained zstd dictionary, with a size limit.
409///
410/// The limit prevents compression bombs where a small compressed payload
411/// expands to an extremely large decompressed output.
412pub(super) fn decompress_with_dict(
413    compressed: &[u8],
414    dict_bytes: &[u8],
415    limit: u64,
416) -> std::io::Result<Vec<u8>> {
417    let dict = zstd::dict::DecoderDictionary::copy(dict_bytes);
418    let decoder = zstd::stream::Decoder::with_prepared_dictionary(compressed, &dict)?;
419    let mut decompressed = Vec::new();
420    decoder
421        .take(limit.saturating_add(1))
422        .read_to_end(&mut decompressed)?;
423    if decompressed.len() as u64 > limit {
424        return Err(io::Error::new(
425            io::ErrorKind::InvalidData,
426            format!(
427                "decompressed size exceeds the limit of {}",
428                SizeDisplay(limit)
429            ),
430        ));
431    }
432    Ok(decompressed)
433}
434
435/// Zstd decoder reading from a file.
436type LogDecoder = zstd::stream::Decoder<'static, BufReader<File>>;
437
438/// The maximum length of a run log line.
439///
440/// This is at least as large as any lin nextest writes today. In particular,
441/// event lines embed failure text, so this matches the archive size limit.
442const MAX_RUN_LOG_LINE_SIZE: ByteSize = MAX_MAX_OUTPUT_SIZE;
443
444/// An iterator over events in a zstd-compressed JSON Lines run log.
445///
446/// This is shared by the store and portable readers.
447#[derive(Debug)]
448pub struct RunLogEventIter<R> {
449    reader: DebugIgnore<R>,
450    line_buf: Vec<u8>,
451    line_number: usize,
452    limit: ByteSize,
453}
454
455impl<R: BufRead> RunLogEventIter<R> {
456    pub(super) fn new(reader: R) -> Self {
457        Self {
458            reader: DebugIgnore(reader),
459            line_buf: Vec::new(),
460            line_number: 0,
461            limit: MAX_RUN_LOG_LINE_SIZE,
462        }
463    }
464}
465
466impl<R: BufRead> Iterator for RunLogEventIter<R> {
467    type Item = Result<TestEventSummary<RecordingSpec>, RecordReadError>;
468
469    fn next(&mut self) -> Option<Self::Item> {
470        let limit = self.limit.as_u64();
471        loop {
472            self.line_buf.clear();
473            self.line_number += 1;
474
475            // Reading at most limit + 1 bytes bounds memory per line.
476            match self
477                .reader
478                .by_ref()
479                .take(limit.saturating_add(1))
480                .read_until(b'\n', &mut self.line_buf)
481            {
482                Ok(0) => return None,
483                Ok(_) => {
484                    let line = self.line_buf.strip_suffix(b"\n").unwrap_or(&self.line_buf);
485                    if line.len() as u64 > limit {
486                        return Some(Err(RecordReadError::RunLogLineTooLong {
487                            line_number: self.line_number,
488                            limit: self.limit,
489                        }));
490                    }
491                    let line = line.trim_ascii();
492                    if line.is_empty() {
493                        continue;
494                    }
495                    return Some(serde_json::from_slice(line).map_err(|error| {
496                        RecordReadError::ParseEvent {
497                            line_number: self.line_number,
498                            error,
499                        }
500                    }));
501                }
502                Err(error) => {
503                    return Some(Err(RecordReadError::ReadRunLog {
504                        line_number: self.line_number,
505                        error,
506                    }));
507                }
508            }
509        }
510    }
511}
512
513#[cfg(test)]
514mod tests {
515    use super::*;
516    use crate::record::recorder::compress_with_dict;
517
518    #[test]
519    fn test_run_log_line_limit() {
520        #[derive(Debug, PartialEq)]
521        enum LineResult {
522            End,
523            TooLong { line_number: usize },
524            ParseError { line_number: usize },
525        }
526
527        let limit = ByteSize::b(8);
528        let cases: [(&str, &[u8], LineResult); 8] = [
529            ("empty input", b"", LineResult::End),
530            ("only blank lines", b"\n\n  \n", LineResult::End),
531            (
532                "line at limit",
533                b"12345678\n",
534                LineResult::ParseError { line_number: 1 },
535            ),
536            (
537                "line at limit without newline",
538                b"12345678",
539                LineResult::ParseError { line_number: 1 },
540            ),
541            (
542                "line over limit",
543                b"123456789\n",
544                LineResult::TooLong { line_number: 1 },
545            ),
546            (
547                "line over limit without newline",
548                b"123456789",
549                LineResult::TooLong { line_number: 1 },
550            ),
551            (
552                "line over limit after a blank line",
553                b"\n123456789\n",
554                LineResult::TooLong { line_number: 2 },
555            ),
556            (
557                "multibyte character crossing the limit",
558                "1234567\u{e9}\n".as_bytes(),
559                LineResult::TooLong { line_number: 1 },
560            ),
561        ];
562
563        for (description, input, expected) in cases {
564            let mut iter = RunLogEventIter {
565                reader: DebugIgnore(io::Cursor::new(input)),
566                line_buf: Vec::new(),
567                line_number: 0,
568                limit,
569            };
570            let actual = match iter.next() {
571                None => LineResult::End,
572                Some(Err(RecordReadError::RunLogLineTooLong {
573                    line_number,
574                    limit: error_limit,
575                })) => {
576                    assert_eq!(error_limit, limit, "{description}: error limit");
577                    LineResult::TooLong { line_number }
578                }
579                Some(Err(RecordReadError::ParseEvent { line_number, .. })) => {
580                    LineResult::ParseError { line_number }
581                }
582                Some(other) => panic!("{description}: unexpected result {other:?}"),
583            };
584            assert_eq!(actual, expected, "{description}");
585            assert!(
586                iter.line_buf.len() as u64 <= limit.as_u64() + 1,
587                "{description}: buffered {} bytes, more than limit + 1",
588                iter.line_buf.len()
589            );
590        }
591    }
592
593    #[test]
594    fn test_decompress_with_dict_limit() {
595        let limit = 64;
596        let dict = OutputDict::Stdout
597            .dict_bytes()
598            .expect("stdout dictionary exists");
599        for (len, within_limit) in [
600            (0, true),
601            (63, true),
602            (64, true),
603            (65, false),
604            (4096, false),
605        ] {
606            let data = vec![b'x'; len];
607            let compressed = compress_with_dict(&data, dict).expect("compressed data");
608            let result = decompress_with_dict(&compressed, dict, limit);
609            if within_limit {
610                assert_eq!(
611                    result.expect("decompression within the limit succeeds"),
612                    data,
613                    "len {len}: round-trips"
614                );
615            } else {
616                let error = result.expect_err("decompression over the limit fails");
617                assert_eq!(
618                    error.kind(),
619                    io::ErrorKind::InvalidData,
620                    "len {len}: error kind"
621                );
622            }
623        }
624    }
625
626    #[test]
627    fn test_record_reader_nonexistent_dir() {
628        let result = RecordReader::open(Utf8Path::new("/nonexistent/path"));
629        assert!(matches!(result, Err(RecordReadError::RunNotFound { .. })));
630    }
631}
nextest_runner/record/reader.rs

nextest_runner/record/
reader.rs
