diff --git a/Cargo.toml b/Cargo.toml index 321ab20..fe8be8c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,6 +28,7 @@ clap = { version = "4.6.6", features = ["derive"] } memmap2 = "0.9.11" pyo3 = { version = ">=0.29.2", optional = true } rayon = "1.12.0" +serde_json = "1.0.151" tempfile = "3.27.0" [dev-dependencies] diff --git a/DEV.md b/DEV.md index 38a32ba..47f769b 100644 --- a/DEV.md +++ b/DEV.md @@ -26,6 +26,8 @@ The current scanner deliberately does not treat 48-bit marker matches or later ` The decoder remains independent of files, threads, Python, and the CLI. Parallel scanning/decoding and indexed seeking are layered over it. Native workers never call Python. Large offsets use explicit 64-bit bit/byte types, and speculative block-marker hits are accepted only when they form an exact stream chain with valid block and combined stream CRCs. +The core decode APIs can report completed compressed and decoded byte counts without knowing anything about terminals. The CLI layers delayed, rate-limited TTY progress rendering over those callbacks; redirected stderr and `--quiet` produce no progress output. Decoded files use same-directory temporary files and atomic persistence, then inherit the compressed input's modification time and permissions. `--rm` removes an input only after decode, persistence, and metadata copying all succeed. Output-size limits are enforced by a writer wrapper, so the decoder has one code path for files, stdout, validation, and indexing. + Parallel decoding uses a rolling candidate queue rather than stopping at stream boundaries or waiting for fixed batches. Workers reserve the maximum possible decoded block size before starting; once a block finishes, that conservative reservation shrinks to its actual output size and is released when ordered validation consumes or rejects it. Thus the `memory_limit` bounds speculative decoded output while short multistream inputs can keep the worker pool busy. The 1 GiB default admits one worst-case block per worker on the primary 18-core machine. The production decoder is safe scalar Rust designed for LLVM auto-vectorisation. Huffman decoding uses a 4096-entry direct table for codes up to 12 bits and canonical fallback for longer codes. Add narrowly scoped unsafe or architecture-specific SIMD only after profiling; `libbz2-rs-sys` remains the dev-only differential oracle. @@ -117,7 +119,7 @@ Line 1000 is the start of stream 1001 because byte zero is stream 1 and is absen To create the separately useful well-formed parser fixture, append only the XML root close after decoding; those 13 bytes are deliberately excluded from `ENWIKI_1000_LEN`: ```bash -fastbz2 decode "$wiki/data/enwiki-first-1000-streams.xml.bz2" -o "$wiki/data/enwiki-first-1000-streams.xml" +fastbz2 "$wiki/data/enwiki-first-1000-streams.xml.bz2" -o "$wiki/data/enwiki-first-1000-streams.xml" printf '\n' >> "$wiki/data/enwiki-first-1000-streams.xml" xmllint --stream --noout "$wiki/data/enwiki-first-1000-streams.xml" ``` diff --git a/README.md b/README.md index 73cb826..f6c92e3 100644 --- a/README.md +++ b/README.md @@ -2,11 +2,128 @@ Fast parallel and indexed bzip2 decompression for Rust and Python. -`fastbz2` is initially focused solely on bzip2: a portable Rust core, a native CLI, and a thin PyO3 seekable-file API. The primary targets are Linux on x86-64 and ARM64, and macOS on ARM64; macOS Intel is best-effort. Correct output, block and stream CRC validation, bounded memory, and deterministic behaviour across thread counts are hard requirements. +`fastbz2` provides a native CLI, a Rust library, and a Python module. It handles ordinary and concatenated bzip2 streams, validates every block and stream CRC, and keeps speculative parallel output within a configurable memory bound. Persistent indexes support efficient random access from Python without first expanding the whole file. -The performance floor is end-to-end decompression within 20% of the maintained pure-Rust `libbz2-rs-sys` decoder on a representative corpus. Portable, SIMD-friendly Rust comes first; architecture-specific SIMD is added only when profiles justify it. The much larger Simple English Wikipedia dump is used for local throughput measurements. +This project decompresses bzip2; it does not compress it. -The implementation includes a safe structural scanner, an in-repo decoder with a tuned 12-bit Huffman lookup table, CRC-validated block decoding, memory-bounded rolling parallel scheduling, persistent indexes, a native CLI, and a seekable Python file API. Marker scans remain speculative until decoding establishes an exact stream chain and validates block and combined-stream CRCs. +## Install + +PyPI wheels contain both the Python module and the native `fastbz2` executable—there is no Python CLI wrapper: + +```bash +pip install fastbz2 +``` + +Python 3.10 and later are supported. Prebuilt wheels target Linux on x86-64 and ARM64, and macOS on ARM64. macOS Intel is best-effort and can build from source. + +The Rust crate is not yet published separately on crates.io. Install the CLI from the repository, or add the library as a Git dependency: + +```bash +cargo install --git https://github.com/AnswerDotAI/fastbz2 +cargo add fastbz2 --git https://github.com/AnswerDotAI/fastbz2 +``` + +## CLI + +Decoding is the default operation. A `.bz2` suffix is removed for the output name; other input names gain `.out`. + +```bash +fastbz2 dump.xml.bz2 # write dump.xml +fastbz2 dump.xml.bz2 -o result.xml # choose the output path +fastbz2 dump.xml.bz2 -o - # write plaintext to stdout +fastbz2 - # read compressed data from stdin +``` + +Multiple inputs are decoded in order, with parallelism applied inside each file. `-C/--output-dir` collects their outputs in one directory: + +```bash +fastbz2 *.bz2 -C decoded +fastbz2 *.bz2 -C decoded --skip-existing +``` + +The alternative modes are flags rather than subcommands: + +```bash +fastbz2 --test dump.xml.bz2 # fully decode and validate, writing nothing +fastbz2 --index dump.xml.bz2 # write dump.xml.bz2.fbz2i +fastbz2 --list dump.xml.bz2 # print the validated stream/block layout +fastbz2 --list --json dump.xml.bz2 # emit the complete layout as JSON +``` + +`--test`, `--index`, and `--list` are mutually exclusive. Human-readable `--list` output labels each input when given multiple files; JSON output is one object for one input and an array for multiple inputs. + +### Output safety + +- Existing outputs are rejected by default. Use `--force` to replace them or `--skip-existing` to leave them untouched. +- File outputs are written to a temporary file in the destination directory and persisted atomically only after successful CRC validation. +- Extracted files inherit the compressed input's permissions and modification time. +- `--rm` removes each compressed input only after its output has been persisted and its metadata copied successfully. +- `--max-output SIZE` limits decoded bytes per input. Sizes accept binary suffixes such as `K`, `MiB`, and `G`. + +Long interactive operations report completion, decoded throughput, compression ratio, and ETA on stderr. Progress is disabled automatically when stderr is redirected; `-q/--quiet` also suppresses progress and skip notices. + +`-P/--threads 0`, the default, uses all available CPUs. `--memory-limit` bounds speculative decoded output and defaults to `1G`. + +## Python + +### One-shot decompression and validation + +```python +import fastbz2 + +plain = fastbz2.decompress(compressed_bytes) +fastbz2.test("dump.xml.bz2") # returns None after successful validation +``` + +`decompress` accepts a bytes-like object and returns `bytes`. `test` accepts either compressed bytes or a path and avoids retaining the decoded result. + +### Seekable reads and persistent indexes + +`fastbz2.open` returns a seekable binary `io.RawIOBase`. Opening without an index performs a complete validation pass and builds an in-memory block index; `build_index` can persist that work for later processes: + +```python +import fastbz2 + +fastbz2.build_index("dump.xml.bz2", "dump.xml.bz2.fbz2i") + +with fastbz2.open("dump.xml.bz2", index="dump.xml.bz2.fbz2i") as f: + f.seek(1_000_000_000) + chunk = f.read(64 * 1024) + print(f.tell(), f.size) +``` + +Building an index fully decodes into a sink but does not write or retain the plaintext. Indexes contain compressed and decoded block offsets and are bound to the exact compressed source by its length and BLAKE3 hash. Loading one verifies that identity without decoding the whole payload; subsequent reads decode only the blocks needed for the requested range and cache recent blocks. `cache_limit` controls that cache. Path sources are memory-mapped, while bytes-like sources stay in memory. + +### Structural scanning + +`scan` cheaply finds candidate stream headers and bit-level block markers without decoding: + +```python +import bz2 +from fastbz2 import scan + +result = scan(bz2.compress(b"hello")) +assert result.blocks[0].bit_offset == 32 +``` + +Scan results are deliberately untrusted candidates. Use `test`, `decompress`, `build_index`, or `open` when validation is required. + +## Rust + +The streaming API accepts any `Write` destination and uses the serial fast path when `threads` is one: + +```rust +use fastbz2::{DecodeOptions, Source, decompress_to_writer}; + +fn main() -> fastbz2::Result<()> { + let source = Source::open("dump.xml.bz2")?; + let mut output = std::io::stdout().lock(); + decompress_to_writer(source.as_slice(), &mut output, DecodeOptions::default())?; + Ok(()) +} +``` + +`decompress` returns a `Vec`. `decode_to_writer` returns a validated `Index` while streaming output, `build_index` validates into a sink, and their `*_with_progress` variants report completed compressed and decoded byte counts. `IndexedReader` implements `Read` and `Seek`; it can build an index itself or load a persisted one with `open_with_index`. ## Performance @@ -15,7 +132,7 @@ These are single local release-mode runs on the primary Apple Silicon developmen Full Simple English Wikipedia (`338 MB` compressed, `1,688,460,257` bytes decoded): | Decoder | Mode | Seconds | -|---|---:|---:| +|---|---|---:| | fastbz2 | parallel, 18 threads, streaming sink | 2.244 | | crabz2 0.4.0 | parallel | 4.460 | | bzip2 | serial CLI | 20.310 | @@ -26,7 +143,7 @@ Full Simple English Wikipedia (`338 MB` compressed, `1,688,460,257` bytes decode The first 1,000 streams of English Wikipedia (`654,362,682` bytes compressed, `2,715,335,085` bytes decoded, 99,853 pages) exercise scheduling across many short concatenated streams: | Decoder | Mode | Seconds | -|---|---:|---:| +|---|---|---:| | crabz2 0.4.0 | parallel, in process | 3.815 | | fastbz2 | parallel, 18 threads, in process | 3.881 | | fastbz2 | serial, in process | 37.198 | @@ -34,46 +151,20 @@ The first 1,000 streams of English Wikipedia (`654,362,682` bytes compressed, `2 | pbzip2 1.1.13 | 18-thread CLI + byte comparison | 88.080 | | bzip2 | serial CLI + byte comparison | 92.960 | -The CLI rows in the second table stream 2.5 GB through `cmp` against the validated XML, so their absolute times are not directly comparable with the in-process rows. DEV “Local Wikipedia benchmarks” documents exact fixture generation and commands. +The CLI rows in the second table stream 2.5 GB through `cmp` against the validated XML, so their absolute times are not directly comparable with the in-process rows. [DEV.md](DEV.md#local-wikipedia-benchmarks) documents exact fixture generation and commands. Homebrew `pbzip2` 1.1.13 could not safely decompress the complete 26,668,484,995-byte English Wikipedia multistream dump on this machine. It segfaulted, and repeated attempts produced divergent and truncated plaintext. Its successful 1,000-stream result above does not establish full-file reliability. -```python -import bz2 -from fastbz2 import scan +## Implementation and compatibility -result = scan(bz2.compress(b"hello")) -result.blocks[0].bit_offset -# 32 -``` +The decoder is safe, portable Rust with a tuned 4096-entry Huffman lookup table for codes up to 12 bits and canonical fallback for longer codes. A structural scan finds possible non-byte-aligned block markers; these remain speculative until ordered decoding establishes the exact stream chain and validates all block and combined-stream CRCs. A rolling scheduler keeps workers busy across concatenated streams while bounding decoded results awaiting validation. -## Inspiration and credit +Legacy randomized blocks generated by bzip2 releases before 0.9.5 are intentionally unsupported. Normal `BZh1` through `BZh9` streams and concatenated streams are supported. The architecture was inspired by Maximilian Knespel's [`librapidarchive`](https://github.com/mxmlnkn/librapidarchive) and [`indexed_bzip2`](https://github.com/mxmlnkn/indexed_bzip2): in particular, scanning for non-byte-aligned bzip2 block markers, independently decoding blocks, ordered prefetch, and indexed seeking. That project's specialised decoder is itself derived from Rob Landley's 0BSD [`bzcat` implementation in Toybox](https://github.com/landley/toybox). ## Development -```bash -pip install -e .[dev] -cargo build --release --bins && python tools/stage_binaries.py -cargo test --release -maturin develop --release && pytest -q -``` - -Python wheels also install the native `fastbz2` executable directly into the environment's scripts directory; it is not a Python entry point or wrapper. - -## Build - -```bash -ship-rs-build -``` - -## Release - -```bash -cargo build --release --bins && python tools/stage_binaries.py -maturin develop --release && pytest -q -ship-release -``` +[DEV.md](DEV.md) documents the architecture, test strategy, benchmark fixture generation, build commands, and release process. -`ship-release` tags the Cargo version, leaves wheel publication to GitHub Actions, then bumps the project. +`fastbz2` is licensed under the [Apache License 2.0](LICENSE). diff --git a/src/bin/fastbz2.rs b/src/bin/fastbz2.rs index f1908af..1e39f72 100644 --- a/src/bin/fastbz2.rs +++ b/src/bin/fastbz2.rs @@ -1,65 +1,67 @@ use std::{ fs, - io::{self, Read, Write}, + io::{self, IsTerminal, Read, Write}, path::{Path, PathBuf}, process::ExitCode, + time::{Duration, Instant}, }; -use clap::{Parser, Subcommand}; -use fastbz2::{DecodeOptions, Error, Index, Source, build_index, decompress_to_writer}; +use clap::{ArgGroup, Parser}; +use fastbz2::{ + DecodeOptions, DecodeProgress, Error, Index, Source, build_index_with_progress, decode_to_writer_with_progress, decompress_to_writer_with_progress, +}; +use serde_json::{Value, json}; use tempfile::NamedTempFile; #[derive(Parser)] -#[command(version, about = "Fast parallel and indexed bzip2 decompression")] +#[command( + version, + about = "Fast parallel and indexed bzip2 decompression", + group(ArgGroup::new("mode").args(["test", "index", "list"])) +)] struct Cli { - #[command(subcommand)] - command: Command, -} - -#[derive(Subcommand)] -enum Command { - /// Decompress INPUT, using - for stdin. - Decode { - input: String, - #[arg(short, long)] - output: Option, - #[arg(short = 'c', long, conflicts_with = "output")] - stdout: bool, - #[arg(short = 'P', long, default_value_t = 0)] - threads: usize, - #[arg(long, default_value = "1G", value_parser = parse_size)] - memory_limit: usize, - #[arg(short, long)] - force: bool, - }, - /// Fully decode and validate INPUT without writing plaintext. - Test { - input: String, - #[arg(short = 'P', long, default_value_t = 0)] - threads: usize, - #[arg(long, default_value = "1G", value_parser = parse_size)] - memory_limit: usize, - }, - /// Build a validated, source-bound block index. - Index { - input: PathBuf, - #[arg(short, long)] - output: Option, - #[arg(short = 'P', long, default_value_t = 0)] - threads: usize, - #[arg(long, default_value = "1G", value_parser = parse_size)] - memory_limit: usize, - #[arg(short, long)] - force: bool, - }, - /// Validate INPUT and show its stream/block layout. - List { - input: PathBuf, - #[arg(short = 'P', long, default_value_t = 0)] - threads: usize, - #[arg(long, default_value = "1G", value_parser = parse_size)] - memory_limit: usize, - }, + /// Input files, or - for stdin when decoding or testing. + #[arg(required = true, num_args = 1..)] + inputs: Vec, + /// Fully decode and validate without writing plaintext. + #[arg(long)] + test: bool, + /// Build validated, source-bound block indexes. + #[arg(long)] + index: bool, + /// Validate and show stream/block layouts. + #[arg(long)] + list: bool, + /// Output path, or - for stdout; requires one input. + #[arg(short, long, conflicts_with_all = ["test", "list", "output_dir"])] + output: Option, + /// Put decoded files in DIRECTORY. + #[arg(short = 'C', long = "output-dir", conflicts_with_all = ["test", "index", "list", "output"])] + output_dir: Option, + /// Worker threads; 0 uses all available CPUs. + #[arg(short = 'P', long, default_value_t = 0)] + threads: usize, + /// Maximum speculative decoded output. + #[arg(long, default_value = "1G", value_parser = parse_size)] + memory_limit: usize, + /// Refuse to decode more than SIZE bytes per input. + #[arg(long, value_parser = parse_size)] + max_output: Option, + /// Replace existing output files. + #[arg(short, long, conflicts_with_all = ["test", "list", "skip_existing"])] + force: bool, + /// Skip existing output files. + #[arg(long, conflicts_with_all = ["test", "list", "force"])] + skip_existing: bool, + /// Remove compressed inputs after successful extraction. + #[arg(long = "rm", conflicts_with_all = ["test", "index", "list"])] + remove_input: bool, + /// Suppress progress and skip notices. + #[arg(short, long)] + quiet: bool, + /// Emit list output as JSON. + #[arg(long, requires = "list")] + json: bool, } fn main() -> ExitCode { @@ -73,51 +75,240 @@ fn main() -> ExitCode { } fn run(cli: Cli) -> fastbz2::Result<()> { - match cli.command { - Command::Decode { input, output, stdout, threads, memory_limit, force } => { - let options = DecodeOptions { threads, memory_limit }; - if input == "-" { - if stdout || output.is_none() { - return decode_stdin(&mut io::stdout().lock()); - } - return atomic_write(output.as_ref().unwrap(), force, decode_stdin); - } - let input_path = Path::new(&input); - let source = Source::open(input_path)?; - if stdout { - decompress_to_writer(source.as_slice(), &mut io::stdout().lock(), options)?; - } else { - let output = output.unwrap_or_else(|| default_output(input_path)); - atomic_write(&output, force, |writer| decompress_to_writer(source.as_slice(), writer, options))?; - } + validate_cli(&cli)?; + let options = DecodeOptions { threads: cli.threads, memory_limit: cli.memory_limit }; + if cli.test { + for input in &cli.inputs { + decode_input(input, &mut io::sink(), options, cli.max_output, cli.quiet)?; } - Command::Test { input, threads, memory_limit } => { - if input == "-" { - decode_stdin(&mut io::sink())?; - } else { - let source = Source::open(input)?; - decompress_to_writer(source.as_slice(), &mut io::sink(), DecodeOptions { threads, memory_limit })?; + return Ok(()); + } + if cli.index { + return run_index(&cli, options); + } + if cli.list { + return run_list(&cli, options); + } + run_decode(&cli, options) +} + +fn validate_cli(cli: &Cli) -> fastbz2::Result<()> { + if cli.output.is_some() && cli.inputs.len() != 1 { + return Err(invalid("--output requires exactly one input")); + } + if cli.inputs.iter().any(|input| input == "-") && cli.inputs.len() != 1 { + return Err(invalid("stdin must be the only input")); + } + if (cli.index || cli.list) && cli.inputs.iter().any(|input| input == "-") { + return Err(invalid("stdin is supported only for decoding and --test")); + } + Ok(()) +} + +fn run_decode(cli: &Cli, options: DecodeOptions) -> fastbz2::Result<()> { + if let Some(directory) = &cli.output_dir { + fs::create_dir_all(directory)?; + } + for input in &cli.inputs { + if input == "-" || cli.output.as_deref() == Some(Path::new("-")) { + decode_input(input, &mut io::stdout().lock(), options, cli.max_output, cli.quiet)?; + if cli.remove_input && input != "-" { + fs::remove_file(input)?; } + continue; + } + let input_path = Path::new(input); + let output = cli.output.clone().unwrap_or_else(|| output_in(input_path, cli.output_dir.as_deref())); + if input_path == output { + return Err(invalid(format!("input and output are both {}", input_path.display()))); + } + if should_skip(&output, cli.skip_existing, cli.quiet) { + continue; + } + let source = Source::open(input_path)?; + atomic_write(&output, cli.force, |writer| decode_data(source.as_slice(), input, writer, options, cli.max_output, cli.quiet))?; + preserve_metadata(input_path, &output)?; + if cli.remove_input { + fs::remove_file(input_path)?; + } + } + Ok(()) +} + +fn run_index(cli: &Cli, options: DecodeOptions) -> fastbz2::Result<()> { + for input in &cli.inputs { + let input_path = Path::new(input); + let output = cli.output.clone().unwrap_or_else(|| PathBuf::from(format!("{}.fbz2i", input_path.display()))); + if output != Path::new("-") && should_skip(&output, cli.skip_existing, cli.quiet) { + continue; } - Command::Index { input, output, threads, memory_limit, force } => { - let source = Source::open(&input)?; - let index = build_index(source.as_slice(), DecodeOptions { threads, memory_limit })?; - let output = output.unwrap_or_else(|| PathBuf::from(format!("{}.fbz2i", input.display()))); - atomic_write(&output, force, |writer| writer.write_all(&index.to_bytes()).map_err(Error::from))?; + let source = Source::open(input_path)?; + let index = build_index_data(source.as_slice(), input, options, cli.max_output, cli.quiet)?; + let encoded = index.to_bytes(); + if output == Path::new("-") { + io::stdout().lock().write_all(&encoded)?; + } else { + atomic_write(&output, cli.force, |writer| writer.write_all(&encoded).map_err(Error::from))?; } - Command::List { input, threads, memory_limit } => { - let source = Source::open(input)?; - let index = build_index(source.as_slice(), DecodeOptions { threads, memory_limit })?; - print_index(&index); + } + Ok(()) +} + +fn run_list(cli: &Cli, options: DecodeOptions) -> fastbz2::Result<()> { + let mut values = Vec::new(); + for input in &cli.inputs { + let source = Source::open(input)?; + let index = build_index_data(source.as_slice(), input, options, cli.max_output, cli.quiet)?; + if cli.json { + values.push(index_json(input, &index)); + } else { + print_index((cli.inputs.len() > 1).then_some(input), &index); } } + if cli.json { + let value = if values.len() == 1 { values.pop().unwrap() } else { Value::Array(values) }; + serde_json::to_writer_pretty(io::stdout().lock(), &value).map_err(|error| Error::Io(io::Error::other(error)))?; + println!(); + } Ok(()) } -fn decode_stdin(output: &mut impl Write) -> fastbz2::Result<()> { - let mut input = Vec::new(); - io::stdin().lock().read_to_end(&mut input)?; - decompress_to_writer(&input, output, DecodeOptions::default()) +fn decode_input(input: &str, output: &mut impl Write, options: DecodeOptions, max_output: Option, quiet: bool) -> fastbz2::Result<()> { + if input == "-" { + let mut data = Vec::new(); + io::stdin().lock().read_to_end(&mut data)?; + decode_data(&data, "stdin", output, options, max_output, quiet) + } else { + let source = Source::open(input)?; + decode_data(source.as_slice(), input, output, options, max_output, quiet) + } +} + +fn decode_data(data: &[u8], label: &str, output: &mut impl Write, options: DecodeOptions, max_output: Option, quiet: bool) -> fastbz2::Result<()> { + let mut output = LimitedWriter::new(output, max_output); + let mut display = ProgressDisplay::new(label, data.len() as u64, quiet); + let result = decompress_to_writer_with_progress(data, &mut output, options, |progress| display.update(progress)); + display.finish(); + result +} + +fn build_index_data(data: &[u8], label: &str, options: DecodeOptions, max_output: Option, quiet: bool) -> fastbz2::Result { + let mut display = ProgressDisplay::new(label, data.len() as u64, quiet); + let result = if let Some(limit) = max_output { + let mut sink = LimitedWriter::new(io::sink(), Some(limit)); + decode_to_writer_with_progress(data, &mut sink, options, |progress| display.update(progress)) + } else { + build_index_with_progress(data, options, |progress| display.update(progress)) + }; + display.finish(); + result +} + +struct LimitedWriter { + inner: W, + written: usize, + limit: Option, +} + +impl LimitedWriter { + fn new(inner: W, limit: Option) -> Self { + Self { inner, written: 0, limit } + } +} + +impl Write for LimitedWriter { + fn write(&mut self, buffer: &[u8]) -> io::Result { + if self.limit.is_some_and(|limit| self.written.saturating_add(buffer.len()) > limit) { + return Err(io::Error::new(io::ErrorKind::InvalidData, format!("decoded output exceeds {}", format_bytes(self.limit.unwrap() as u64)))); + } + let written = self.inner.write(buffer)?; + self.written += written; + Ok(written) + } + + fn flush(&mut self) -> io::Result<()> { + self.inner.flush() + } +} + +struct ProgressDisplay { + stderr: io::Stderr, + label: String, + total: u64, + started: Instant, + last_draw: Instant, + enabled: bool, + drawn: bool, +} + +impl ProgressDisplay { + fn new(label: &str, total: u64, quiet: bool) -> Self { + let now = Instant::now(); + Self { stderr: io::stderr(), label: label.into(), total, started: now, last_draw: now, enabled: !quiet && io::stderr().is_terminal(), drawn: false } + } + + fn update(&mut self, progress: DecodeProgress) { + if !self.enabled { + return; + } + let elapsed = self.started.elapsed(); + let finished = progress.compressed_bytes >= self.total; + if (!self.drawn && elapsed < Duration::from_millis(200)) || (!finished && self.last_draw.elapsed() < Duration::from_millis(100)) { + return; + } + let compressed = progress.compressed_bytes.min(self.total); + let percent = if self.total == 0 { 100.0 } else { compressed as f64 * 100.0 / self.total as f64 }; + let seconds = elapsed.as_secs_f64().max(0.001); + let rate = progress.decoded_bytes as f64 / seconds; + let ratio = if compressed == 0 { 0.0 } else { progress.decoded_bytes as f64 / compressed as f64 }; + let eta = if compressed == 0 || finished { 0.0 } else { seconds * (self.total - compressed) as f64 / compressed as f64 }; + let _ = write!( + self.stderr, + "\r\x1b[2K{}: {:5.1}% {} → {} {}/s {:.1}× ETA {}", + self.label, + percent, + format_bytes(compressed), + format_bytes(progress.decoded_bytes), + format_bytes(rate as u64), + ratio, + format_duration(eta), + ); + let _ = self.stderr.flush(); + self.last_draw = Instant::now(); + self.drawn = true; + } + + fn finish(&mut self) { + if self.drawn { + let _ = writeln!(self.stderr); + self.drawn = false; + } + } +} + +impl Drop for ProgressDisplay { + fn drop(&mut self) { + self.finish(); + } +} + +fn should_skip(path: &Path, skip_existing: bool, quiet: bool) -> bool { + if !path.exists() || !skip_existing { + return false; + } + if !quiet { + eprintln!("fastbz2: skipping existing {}", path.display()); + } + true +} + +fn preserve_metadata(input: &Path, output: &Path) -> fastbz2::Result<()> { + let metadata = fs::metadata(input)?; + if let Ok(modified) = metadata.modified() { + fs::OpenOptions::new().write(true).open(output)?.set_times(fs::FileTimes::new().set_modified(modified))?; + } + fs::set_permissions(output, metadata.permissions())?; + Ok(()) } fn atomic_write(path: &Path, force: bool, write: impl FnOnce(&mut fs::File) -> fastbz2::Result<()>) -> fastbz2::Result<()> { @@ -136,6 +327,11 @@ fn atomic_write(path: &Path, force: bool, write: impl FnOnce(&mut fs::File) -> f Ok(()) } +fn output_in(input: &Path, directory: Option<&Path>) -> PathBuf { + let output = default_output(input.file_name().map(Path::new).unwrap_or(input)); + directory.map_or_else(|| default_output(input), |directory| directory.join(output)) +} + fn default_output(input: &Path) -> PathBuf { match input.extension().and_then(|extension| extension.to_str()) { Some("bz2") => input.with_extension(""), @@ -143,7 +339,10 @@ fn default_output(input: &Path) -> PathBuf { } } -fn print_index(index: &Index) { +fn print_index(input: Option<&String>, index: &Index) { + if let Some(input) = input { + println!("input\t{input}"); + } println!("compressed_bytes\t{}", index.source_len); println!("decoded_bytes\t{}", index.decoded_len); println!("streams\t{}", index.streams.len()); @@ -156,6 +355,61 @@ fn print_index(index: &Index) { } } +fn index_json(input: &str, index: &Index) -> Value { + json!({ + "input": input, + "source_bytes": index.source_len, + "source_hash": hex(&index.source_hash), + "decoded_bytes": index.decoded_len, + "streams": index.streams.iter().enumerate().map(|(number, stream)| json!({ + "number": number, + "header_byte": stream.compressed_header_byte, + "block_size_100k": stream.block_size_100k, + "first_block": stream.first_block, + "block_count": stream.block_count, + "decoded_start": stream.decoded_start, + "decoded_bytes": stream.decoded_len, + "eos_bit": stream.eos_bit, + "expected_crc": stream.expected_stream_crc, + })).collect::>(), + "blocks": index.blocks.iter().enumerate().map(|(number, block)| json!({ + "number": number, + "compressed_start_bit": block.compressed_start_bit, + "compressed_end_bit": block.compressed_end_bit, + "decoded_start": block.decoded_start, + "decoded_bytes": block.decoded_len, + "expected_crc": block.expected_crc, + "stream": block.stream, + })).collect::>(), + }) +} + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|byte| format!("{byte:02x}")).collect() +} + +fn format_bytes(bytes: u64) -> String { + const UNITS: [&str; 5] = ["B", "KiB", "MiB", "GiB", "TiB"]; + let mut value = bytes as f64; + let mut unit = 0; + while value >= 1024.0 && unit + 1 < UNITS.len() { + value /= 1024.0; + unit += 1; + } + if unit == 0 { format!("{bytes} B") } else { format!("{value:.1} {}", UNITS[unit]) } +} + +fn format_duration(seconds: f64) -> String { + let seconds = seconds.max(0.0).round() as u64; + if seconds >= 3600 { + format!("{}h{:02}m", seconds / 3600, seconds / 60 % 60) + } else if seconds >= 60 { + format!("{}m{:02}s", seconds / 60, seconds % 60) + } else { + format!("{seconds}s") + } +} + fn parse_size(value: &str) -> Result { let split = value.find(|character: char| !character.is_ascii_digit()).unwrap_or(value.len()); let number: usize = value[..split].parse().map_err(|_| format!("invalid size {value:?}"))?; @@ -164,14 +418,20 @@ fn parse_size(value: &str) -> Result { "k" | "kb" | "kib" => 1024, "m" | "mb" | "mib" => 1024 * 1024, "g" | "gb" | "gib" => 1024 * 1024 * 1024, + "t" | "tb" | "tib" => 1024_usize.pow(4), _ => return Err(format!("invalid size suffix in {value:?}")), }; number.checked_mul(multiplier).ok_or_else(|| format!("size {value:?} overflows this platform")) } +fn invalid(message: impl Into) -> Error { + Error::InvalidConfiguration(message.into()) +} + fn exit_status(error: &Error) -> u8 { match error { - Error::Io(source) if source.kind() != io::ErrorKind::InvalidData => 1, + Error::Io(source) if source.kind() == io::ErrorKind::InvalidData => 3, + Error::Io(_) => 1, Error::InvalidConfiguration(_) => 2, Error::InvalidStreamHeader | Error::Decode { .. } | Error::InvalidIndex(_) => 3, _ => 4, diff --git a/src/decode.rs b/src/decode.rs index 6983006..f6df7d4 100644 --- a/src/decode.rs +++ b/src/decode.rs @@ -12,6 +12,12 @@ use crate::{BlockCandidate, BlockIndex, DecodeError, EndCandidate, Error, Index, pub const DEFAULT_MEMORY_LIMIT: usize = 1024 * 1024 * 1024; +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct DecodeProgress { + pub compressed_bytes: u64, + pub decoded_bytes: u64, +} + #[derive(Clone, Copy, Debug)] pub struct DecodeOptions { /// Zero selects the process's available parallelism. @@ -69,19 +75,43 @@ pub fn decompress(data: &[u8], options: DecodeOptions) -> Result> { /// Decode to a streaming output. A one-thread request uses the same pure-Rust /// block codec without the indexing/speculation overhead. pub fn decompress_to_writer(data: &[u8], output: &mut impl Write, options: DecodeOptions) -> Result<()> { + decompress_to_writer_with_progress(data, output, options, |_| {}) +} + +pub fn decompress_to_writer_with_progress( + data: &[u8], + output: &mut impl Write, + options: DecodeOptions, + mut progress: impl FnMut(DecodeProgress), +) -> Result<()> { let options = options.validate()?; if options.resolved_threads() == 1 { - return decoder::decode_serial(data, output); + return decoder::decode_serial_with_progress(data, output, &mut |compressed_bytes, decoded_bytes| { + progress(DecodeProgress { compressed_bytes, decoded_bytes }); + }); } - decode_to_writer(data, output, options).map(|_| ()) + decode_to_writer_impl(data, output, options, &mut progress).map(|_| ()) } pub fn build_index(data: &[u8], options: DecodeOptions) -> Result { - decode_to_writer(data, &mut std::io::sink(), options) + build_index_with_progress(data, options, |_| {}) +} + +pub fn build_index_with_progress(data: &[u8], options: DecodeOptions, mut progress: impl FnMut(DecodeProgress)) -> Result { + decode_to_writer_impl(data, &mut std::io::sink(), options, &mut progress) } /// Decode a complete bzip2 input, validate every CRC, and build its block index. pub fn decode_to_writer(data: &[u8], output: &mut impl Write, options: DecodeOptions) -> Result { + decode_to_writer_impl(data, output, options, &mut |_| {}) +} + +/// Decode a complete bzip2 input, return its index, and report completed work. +pub fn decode_to_writer_with_progress(data: &[u8], output: &mut impl Write, options: DecodeOptions, mut progress: impl FnMut(DecodeProgress)) -> Result { + decode_to_writer_impl(data, output, options, &mut progress) +} + +fn decode_to_writer_impl(data: &[u8], output: &mut impl Write, options: DecodeOptions, progress: &mut impl FnMut(DecodeProgress)) -> Result { let options = options.validate()?; let threads = options.resolved_threads(); let pool = thread_pool(threads)?; @@ -97,7 +127,7 @@ pub fn decode_to_writer(data: &[u8], output: &mut impl Write, options: DecodeOpt let Some(pool) = pool else { let mut candidates = SerialCandidates { data, markers: &markers }; - return assemble(data, output, &markers, &mut candidates); + return assemble(data, output, &markers, &mut candidates, progress); }; let jobs: Vec<_> = markers .iter() @@ -116,7 +146,7 @@ pub fn decode_to_writer(data: &[u8], output: &mut impl Write, options: DecodeOpt }); let result = { let mut candidates = ParallelCandidates { receiver, ready: HashMap::new(), work: &work }; - assemble(data, output, &markers, &mut candidates) + assemble(data, output, &markers, &mut candidates, progress) }; work.cancel(); worker.join().map_err(|_| Error::InvalidConfiguration("parallel decoder worker panicked".into()))?; @@ -124,7 +154,13 @@ pub fn decode_to_writer(data: &[u8], output: &mut impl Write, options: DecodeOpt }) } -fn assemble(data: &[u8], output: &mut impl Write, markers: &[Marker], candidates: &mut impl Candidates) -> Result { +fn assemble( + data: &[u8], + output: &mut impl Write, + markers: &[Marker], + candidates: &mut impl Candidates, + progress: &mut impl FnMut(DecodeProgress), +) -> Result { let mut blocks = Vec::new(); let mut streams = Vec::new(); let mut decoded_offset = 0_u64; @@ -176,6 +212,7 @@ fn assemble(data: &[u8], output: &mut impl Write, markers: &[Marker], candidates stream: stream_number, }); decoded_offset = decoded_offset.checked_add(decoded_len).ok_or_else(offset_overflow)?; + progress(DecodeProgress { compressed_bytes: decoded.end_bit.div_ceil(8), decoded_bytes: decoded_offset }); combined_crc = combine_stream_crc(combined_crc, block.expected_crc); current_bit = markers[end_index].bit_offset(); marker_index = end_index; @@ -191,6 +228,7 @@ fn assemble(data: &[u8], output: &mut impl Write, markers: &[Marker], candidates } output.flush()?; + progress(DecodeProgress { compressed_bytes: data.len() as u64, decoded_bytes: decoded_offset }); Ok(Index::new(data, decoded_offset, streams, blocks)) } @@ -388,6 +426,21 @@ mod tests { } } + #[test] + fn progress_reaches_exact_input_and_output_lengths() { + let plain = patterned(350_000); + let compressed = compress(&plain, Level::FASTEST); + for threads in [1, 4] { + let options = DecodeOptions { threads, ..DecodeOptions::default() }; + let mut output = Vec::new(); + let mut reports = Vec::new(); + decompress_to_writer_with_progress(&compressed, &mut output, options, |progress| reports.push(progress)).unwrap(); + assert_eq!(output, plain); + assert!(reports.windows(2).all(|pair| { pair[0].compressed_bytes <= pair[1].compressed_bytes && pair[0].decoded_bytes <= pair[1].decoded_bytes })); + assert_eq!(reports.last(), Some(&DecodeProgress { compressed_bytes: compressed.len() as u64, decoded_bytes: plain.len() as u64 })); + } + } + #[test] fn validates_concatenated_streams_and_indexes_them() { let first = patterned(180_000); diff --git a/src/decoder.rs b/src/decoder.rs index 1a28de2..8c95a0a 100644 --- a/src/decoder.rs +++ b/src/decoder.rs @@ -401,8 +401,13 @@ pub(crate) fn decode_block(data: &[u8], start_bit: u64, end_bit: u64, level: u8, } pub(crate) fn decode_serial(data: &[u8], output: &mut impl Write) -> Result<()> { + decode_serial_with_progress(data, output, &mut |_, _| {}) +} + +pub(crate) fn decode_serial_with_progress(data: &[u8], output: &mut impl Write, progress: &mut impl FnMut(u64, u64)) -> Result<()> { let mut bits = Bits::at(data, 0)?; let mut decoder = Decoder::new(); + let mut decoded_bytes = 0_u64; while bits.remaining() != 0 { bits.align_byte(); if bits.remaining() < 32 { @@ -423,6 +428,9 @@ pub(crate) fn decode_serial(data: &[u8], output: &mut impl Write) -> Result<()> BLOCK_MAGIC => { let (block, crc, _) = decoder.block(&mut bits, level, None)?; output.write_all(&block)?; + decoded_bytes = + decoded_bytes.checked_add(block.len() as u64).ok_or_else(|| Error::InvalidConfiguration("decoded offset overflow".into()))?; + progress(bits.position().div_ceil(8), decoded_bytes); combined_crc = combine_stream_crc(combined_crc, crc); } END_MAGIC => { @@ -437,6 +445,7 @@ pub(crate) fn decode_serial(data: &[u8], output: &mut impl Write) -> Result<()> bits.align_byte(); } output.flush()?; + progress(data.len() as u64, decoded_bytes); Ok(()) } diff --git a/src/lib.rs b/src/lib.rs index eaa07c3..15ef4e6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -14,7 +14,10 @@ mod source; pub use bitreader::BitReader; pub use block::{MAX_DECODED_BLOCK, MAX_ENCODED_BLOCK, decode_block}; pub use crc::{bz2_crc32, combine_stream_crc}; -pub use decode::{DEFAULT_MEMORY_LIMIT, DecodeOptions, build_index, decode_to_writer, decompress, decompress_to_writer}; +pub use decode::{ + DEFAULT_MEMORY_LIMIT, DecodeOptions, DecodeProgress, build_index, build_index_with_progress, decode_to_writer, decode_to_writer_with_progress, decompress, + decompress_to_writer, decompress_to_writer_with_progress, +}; pub use error::{DecodeError, Error, Result}; pub use format::{BLOCK_MAGIC, BlockCandidate, END_MAGIC, EndCandidate, ScanResult, StreamHeaderCandidate, scan}; pub use index::{BlockIndex, Index, StreamIndex}; diff --git a/tests/cli.rs b/tests/cli.rs index 7fd9e62..2f38adb 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -1,4 +1,10 @@ -use std::{fs, process::Command}; +use std::{ + fs, + io::Write, + path::Path, + process::{Command, Stdio}, + time::{Duration, UNIX_EPOCH}, +}; use crabz2::{Level, compress}; @@ -6,29 +12,51 @@ fn binary() -> Command { Command::new(env!("CARGO_BIN_EXE_fastbz2")) } +fn write_compressed(path: &Path, plain: &[u8]) { + fs::write(path, compress(plain, Level::FASTEST)).unwrap(); +} + #[test] fn decode_test_index_and_list() { let directory = tempfile::tempdir().unwrap(); let input = directory.path().join("sample.bz2"); - let output = directory.path().join("sample.txt"); + let output = directory.path().join("sample"); let index = directory.path().join("sample.fbz2i"); let plain: Vec<_> = (0..250_000).map(|i| ((i * 31 + i / 97) & 255) as u8).collect(); fs::write(&input, compress(&plain, Level::FASTEST)).unwrap(); - let decoded = binary().args(["decode", input.to_str().unwrap(), "-o", output.to_str().unwrap(), "-P", "2"]).status().unwrap(); + let decoded = binary().args([input.to_str().unwrap(), "-P", "2"]).status().unwrap(); assert!(decoded.success()); assert_eq!(fs::read(&output).unwrap(), plain); - let tested = binary().args(["test", input.to_str().unwrap()]).status().unwrap(); + let stdout = binary().args([input.to_str().unwrap(), "-o", "-"]).output().unwrap(); + assert!(stdout.status.success()); + assert_eq!(stdout.stdout, plain); + + let tested = binary().args(["--test", input.to_str().unwrap()]).status().unwrap(); assert!(tested.success()); - let indexed = binary().args(["index", input.to_str().unwrap(), "-o", index.to_str().unwrap()]).status().unwrap(); + let indexed = binary().args(["--index", input.to_str().unwrap(), "-o", index.to_str().unwrap()]).status().unwrap(); assert!(indexed.success()); assert!(fs::metadata(index).unwrap().len() > 100); - let listed = binary().args(["list", input.to_str().unwrap()]).output().unwrap(); + let listed = binary().args(["--list", input.to_str().unwrap()]).output().unwrap(); assert!(listed.status.success()); assert!(String::from_utf8(listed.stdout).unwrap().contains("blocks\t")); + + let conflicting = binary().args(["--test", "--list", input.to_str().unwrap()]).output().unwrap(); + assert_eq!(conflicting.status.code(), Some(2)); +} + +#[test] +fn stdin_decodes_to_stdout() { + let plain = b"stdin uses the same decode options"; + let compressed = compress(plain, Level::BEST); + let mut child = binary().args(["-", "-P", "2", "--memory-limit", "128M"]).stdin(Stdio::piped()).stdout(Stdio::piped()).spawn().unwrap(); + child.stdin.take().unwrap().write_all(&compressed).unwrap(); + let output = child.wait_with_output().unwrap(); + assert!(output.status.success()); + assert_eq!(output.stdout, plain); } #[test] @@ -41,7 +69,130 @@ fn corruption_has_distinct_exit_status_and_atomic_output() { compressed[last] ^= 1; fs::write(&input, compressed).unwrap(); - let result = binary().args(["decode", input.to_str().unwrap(), "-o", output.to_str().unwrap()]).status().unwrap(); + let result = binary().args([input.to_str().unwrap(), "-o", output.to_str().unwrap()]).status().unwrap(); assert_eq!(result.code(), Some(3)); assert!(!output.exists()); } + +#[test] +fn multiple_inputs_support_output_directory_and_skip_existing() { + let directory = tempfile::tempdir().unwrap(); + let first = directory.path().join("first.bz2"); + let second = directory.path().join("second.bz2"); + let output_dir = directory.path().join("decoded"); + write_compressed(&first, b"first contents"); + write_compressed(&second, b"second contents"); + + let decoded = binary().args(["-C", output_dir.to_str().unwrap(), first.to_str().unwrap(), second.to_str().unwrap()]).output().unwrap(); + assert!(decoded.status.success(), "{}", String::from_utf8_lossy(&decoded.stderr)); + assert_eq!(fs::read(output_dir.join("first")).unwrap(), b"first contents"); + assert_eq!(fs::read(output_dir.join("second")).unwrap(), b"second contents"); + + fs::write(output_dir.join("first"), b"keep me").unwrap(); + let skipped = binary().args(["--skip-existing", "-C", output_dir.to_str().unwrap(), first.to_str().unwrap(), second.to_str().unwrap()]).output().unwrap(); + assert!(skipped.status.success()); + assert_eq!(fs::read(output_dir.join("first")).unwrap(), b"keep me"); + assert!(String::from_utf8(skipped.stderr).unwrap().contains("skipping existing")); + + let quiet = + binary().args(["--quiet", "--skip-existing", "-C", output_dir.to_str().unwrap(), first.to_str().unwrap(), second.to_str().unwrap()]).output().unwrap(); + assert!(quiet.status.success()); + assert!(quiet.stderr.is_empty()); + + let replaced = binary().args(["--force", "-C", output_dir.to_str().unwrap(), first.to_str().unwrap(), second.to_str().unwrap()]).output().unwrap(); + assert!(replaced.status.success()); + assert_eq!(fs::read(output_dir.join("first")).unwrap(), b"first contents"); + + let rejected = binary().args([first.to_str().unwrap(), second.to_str().unwrap(), "-o", output_dir.join("one").to_str().unwrap()]).output().unwrap(); + assert_eq!(rejected.status.code(), Some(2)); +} + +#[test] +fn remove_input_happens_only_after_success() { + let directory = tempfile::tempdir().unwrap(); + let valid = directory.path().join("valid.bz2"); + let corrupt = directory.path().join("corrupt.bz2"); + write_compressed(&valid, b"remove after success"); + fs::write(&corrupt, b"not bzip2").unwrap(); + + let decoded = binary().args(["--rm", valid.to_str().unwrap()]).output().unwrap(); + assert!(decoded.status.success()); + assert!(!valid.exists()); + assert_eq!(fs::read(directory.path().join("valid")).unwrap(), b"remove after success"); + + let failed = binary().args(["--rm", corrupt.to_str().unwrap()]).output().unwrap(); + assert!(!failed.status.success()); + assert!(corrupt.exists()); +} + +#[test] +fn decode_preserves_modified_time_and_permissions() { + let directory = tempfile::tempdir().unwrap(); + let input = directory.path().join("metadata.bz2"); + let output = directory.path().join("metadata"); + write_compressed(&input, b"metadata"); + let modified = UNIX_EPOCH + Duration::from_secs(1_700_000_123); + fs::OpenOptions::new().write(true).open(&input).unwrap().set_times(fs::FileTimes::new().set_modified(modified)).unwrap(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + fs::set_permissions(&input, fs::Permissions::from_mode(0o640)).unwrap(); + } + + let decoded = binary().arg(input.to_str().unwrap()).output().unwrap(); + assert!(decoded.status.success()); + let input_metadata = fs::metadata(input).unwrap(); + let output_metadata = fs::metadata(output).unwrap(); + assert_eq!(output_metadata.modified().unwrap(), input_metadata.modified().unwrap()); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + assert_eq!(output_metadata.permissions().mode() & 0o777, 0o640); + } +} + +#[test] +fn max_output_is_enforced_before_persisting() { + let directory = tempfile::tempdir().unwrap(); + let input = directory.path().join("limited.bz2"); + let output = directory.path().join("limited"); + let plain = vec![b'x'; 20_000]; + write_compressed(&input, &plain); + + let rejected = binary().args(["--max-output", "19999", input.to_str().unwrap()]).output().unwrap(); + assert_eq!(rejected.status.code(), Some(3)); + assert!(!output.exists()); + assert!(String::from_utf8(rejected.stderr).unwrap().contains("decoded output exceeds")); + + let accepted = binary().args(["--max-output", "20K", input.to_str().unwrap()]).output().unwrap(); + assert!(accepted.status.success()); + assert_eq!(fs::read(output).unwrap(), plain); +} + +#[test] +fn list_json_describes_one_or_many_inputs() { + let directory = tempfile::tempdir().unwrap(); + let first = directory.path().join("first.bz2"); + let second = directory.path().join("second.bz2"); + write_compressed(&first, b"first"); + write_compressed(&second, b"second"); + + let single = binary().args(["--list", "--json", first.to_str().unwrap()]).output().unwrap(); + assert!(single.status.success()); + let value: serde_json::Value = serde_json::from_slice(&single.stdout).unwrap(); + assert_eq!(value["input"], first.to_str().unwrap()); + assert_eq!(value["decoded_bytes"], 5); + assert_eq!(value["streams"].as_array().unwrap().len(), 1); + assert_eq!(value["blocks"].as_array().unwrap().len(), 1); + + let multiple = binary().args(["--list", "--json", first.to_str().unwrap(), second.to_str().unwrap()]).output().unwrap(); + assert!(multiple.status.success()); + let value: serde_json::Value = serde_json::from_slice(&multiple.stdout).unwrap(); + assert_eq!(value.as_array().unwrap().len(), 2); + + let limited = binary().args(["--list", "--max-output", "4", first.to_str().unwrap()]).output().unwrap(); + assert_eq!(limited.status.code(), Some(3)); + + let missing_mode = binary().args(["--json", first.to_str().unwrap()]).output().unwrap(); + assert_eq!(missing_mode.status.code(), Some(2)); +}