From 85e4911cbb95b1d89459eb62d5383b55b38ec89e Mon Sep 17 00:00:00 2001 From: kassoulet <1905+kassoulet@users.noreply.github.com> Date: Fri, 29 May 2026 20:12:32 +0000 Subject: [PATCH 1/3] =?UTF-8?q?=F0=9F=9B=A1=EF=B8=8F=20Sentinel:=20Improve?= =?UTF-8?q?=20robustness=20and=20validate=20zstd=20level?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add range validation (-7 to 22) for zstd compression level in CLI. - Pre-validate zstd level in main thread to prevent panics in workers. - Replace `unwrap()` in parallel worker threads with proper error handling when sending results, ensuring graceful shutdown if decoder is dropped. Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com> --- bz2zstd/Cargo.toml | 4 ++++ bz2zstd/src/main.rs | 20 ++++++++++++++++---- examples/check_zstd.rs | 6 ++++++ parallel_bzip2_decoder/src/decoder.rs | 8 ++++++-- 4 files changed, 32 insertions(+), 6 deletions(-) create mode 100644 examples/check_zstd.rs diff --git a/bz2zstd/Cargo.toml b/bz2zstd/Cargo.toml index 80ad0aa..273905e 100644 --- a/bz2zstd/Cargo.toml +++ b/bz2zstd/Cargo.toml @@ -22,3 +22,7 @@ crossbeam-channel = "0.5" zstd = { version = "0.13", features = ["zstdmt"] } indicatif = "0.17" parallel_bzip2_decoder = { path = "../parallel_bzip2_decoder", version = "0.2.0" } + +[[test]] +name = "e2e" +path = "../tests/e2e.rs" diff --git a/bz2zstd/src/main.rs b/bz2zstd/src/main.rs index c7a4ed7..97826fb 100644 --- a/bz2zstd/src/main.rs +++ b/bz2zstd/src/main.rs @@ -61,9 +61,10 @@ struct Args { #[arg(short, long)] output: Option, - /// Zstd compression level (1-22, default = 9) - /// Higher values provide better compression but are slower - #[arg(short = 'z', long, default_value_t = 9)] + /// Zstd compression level (-7 to 22, default = 9) + /// Higher values provide better compression but are slower. + /// Negative values provide faster but less efficient compression. + #[arg(short = 'z', long, default_value_t = 9, value_parser = clap::value_parser!(i32).range(-7..=22))] zstd_level: i32, /// Number of threads to use (default = number of logical cores) @@ -79,6 +80,11 @@ struct Args { fn main() -> Result<()> { let args = Args::parse(); + // Pre-validate zstd compression level to ensure it's supported + // This prevents a panic in the worker threads later + zstd::bulk::Compressor::new(args.zstd_level) + .context("Invalid zstd compression level")?; + // Configure global thread pool if user specified thread count // This affects all Rayon parallel iterators in the application if let Some(jobs) = args.jobs { @@ -309,7 +315,13 @@ fn main() -> Result<()> { .try_for_each_init( // Per-thread initialization: create buffers and compressor once per thread // This avoids lock contention and repeated allocations - || (Vec::new(), Compressor::new(args.zstd_level).unwrap()), + || { + ( + Vec::new(), + Compressor::new(args.zstd_level) + .expect("Compression level was pre-validated"), + ) + }, |(decomp_buf, compressor), (idx, (start_bit, end_bit))| -> Result<()> { // Security Check: Verify that the compressed block size is within limits. // This protects against resource exhaustion from maliciously large compressed blocks. diff --git a/examples/check_zstd.rs b/examples/check_zstd.rs new file mode 100644 index 0000000..2053286 --- /dev/null +++ b/examples/check_zstd.rs @@ -0,0 +1,6 @@ +fn main() { + for level in [-10, 0, 3, 9, 22, 23, 100] { + let c = zstd::bulk::Compressor::new(level); + println!("Level {}: {:?}", level, c.is_ok()); + } +} diff --git a/parallel_bzip2_decoder/src/decoder.rs b/parallel_bzip2_decoder/src/decoder.rs index b60b005..3288712 100644 --- a/parallel_bzip2_decoder/src/decoder.rs +++ b/parallel_bzip2_decoder/src/decoder.rs @@ -169,8 +169,12 @@ impl Bz2Decoder { let mut decomp_buf = Vec::new(); // Decompress this block decompress_block_into(slice, start_bit, end_bit, &mut decomp_buf, scratch)?; - // Send result with index for reordering - result_sender.send((idx, decomp_buf)).unwrap(); + // Send result with index for reordering. + // We use map_err to stop the parallel worker pool gracefully + // if the receiver has been dropped. + result_sender.send((idx, decomp_buf)).map_err(|_| { + Bz2Error::Io(io::Error::new(io::ErrorKind::Other, "Receiver dropped")) + })?; Ok(()) }, ); From 93e1e0cf125ab84098b1acf5147ac3e39b9e8dd2 Mon Sep 17 00:00:00 2001 From: kassoulet <1905+kassoulet@users.noreply.github.com> Date: Fri, 29 May 2026 20:18:37 +0000 Subject: [PATCH 2/3] =?UTF-8?q?=F0=9F=9B=A1=EF=B8=8F=20Sentinel:=20Improve?= =?UTF-8?q?=20robustness=20and=20validate=20zstd=20level=20(formatted)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add range validation (-7 to 22) for zstd compression level in CLI. - Pre-validate zstd level in main thread to prevent panics in workers. - Replace `unwrap()` in parallel worker threads with proper error handling when sending results, ensuring graceful shutdown if decoder is dropped. - Run `cargo fmt` to comply with CI formatting requirements. Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com> --- bz2zstd/src/main.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/bz2zstd/src/main.rs b/bz2zstd/src/main.rs index 97826fb..5a806df 100644 --- a/bz2zstd/src/main.rs +++ b/bz2zstd/src/main.rs @@ -82,8 +82,7 @@ fn main() -> Result<()> { // Pre-validate zstd compression level to ensure it's supported // This prevents a panic in the worker threads later - zstd::bulk::Compressor::new(args.zstd_level) - .context("Invalid zstd compression level")?; + zstd::bulk::Compressor::new(args.zstd_level).context("Invalid zstd compression level")?; // Configure global thread pool if user specified thread count // This affects all Rayon parallel iterators in the application From 70ed2f291e336211633631f94c954662452441ca Mon Sep 17 00:00:00 2001 From: kassoulet <1905+kassoulet@users.noreply.github.com> Date: Fri, 29 May 2026 20:25:27 +0000 Subject: [PATCH 3/3] =?UTF-8?q?=F0=9F=9B=A1=EF=B8=8F=20Sentinel:=20Improve?= =?UTF-8?q?=20robustness=20and=20validate=20zstd=20level?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add range validation (-7 to 22) for zstd compression level in CLI using clap value_parser. - Pre-validate zstd level in main thread to provide clean error messages and prevent panics in worker threads. - Replace `unwrap()` in parallel worker threads with proper error handling (Result propagation) when sending results, ensuring graceful shutdown if the decoder is dropped early. - Ensure all changes are properly formatted with `cargo fmt`. Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com> --- bz2zstd/Cargo.toml | 4 ---- examples/check_zstd.rs | 6 ------ 2 files changed, 10 deletions(-) delete mode 100644 examples/check_zstd.rs diff --git a/bz2zstd/Cargo.toml b/bz2zstd/Cargo.toml index 273905e..80ad0aa 100644 --- a/bz2zstd/Cargo.toml +++ b/bz2zstd/Cargo.toml @@ -22,7 +22,3 @@ crossbeam-channel = "0.5" zstd = { version = "0.13", features = ["zstdmt"] } indicatif = "0.17" parallel_bzip2_decoder = { path = "../parallel_bzip2_decoder", version = "0.2.0" } - -[[test]] -name = "e2e" -path = "../tests/e2e.rs" diff --git a/examples/check_zstd.rs b/examples/check_zstd.rs deleted file mode 100644 index 2053286..0000000 --- a/examples/check_zstd.rs +++ /dev/null @@ -1,6 +0,0 @@ -fn main() { - for level in [-10, 0, 3, 9, 22, 23, 100] { - let c = zstd::bulk::Compressor::new(level); - println!("Level {}: {:?}", level, c.is_ok()); - } -}