feat(compressors): add streaming compression crate

[martintmk]

Adds compressors, a streaming compression crate for bytesbuf byte sequences.

Five formats, each behind a cargo feature of its own: deflate, zlib, gzip, brotli and zstd. None is enabled by default, so a build that speaks only brotli never compiles flate2.

let resources = Resources::global();

let compressed = gzip::compress(b"hello", resources)?;
assert_eq!(gzip::decompress(compressed, resources)?.to_vec(), b"hello".to_vec());

Native bytesbuf integration

Input is read segment by segment straight out of a BytesView, and output is written into the uninitialized spare capacity of a BytesBuf. A view is a chain of segments, so nothing is flattened into a contiguous buffer on the way in and nothing is copied out of a scratch buffer on the way back. Every allocation comes from the caller's own memory provider.

The whole-buffer conveniences take anything implementing the sealed InputData trait, so a caller with a plain slice does not have to build a view first -- gzip::compress(b"hello", resources) and gzip::compress(view, resources) are both accepted, and an existing view is forwarded without a copy.

Resource pooling

Resources carries what a codec draws on -- a memory provider and recycled engine state -- and is what every API takes instead of the two separately.

Building a compressor allocates and initializes a substantial amount of state; on a small message that setup can cost as much as the compression itself. Recycling it is therefore on by default, so a service compressing many small bodies spends its budget compressing rather than getting ready to. Resources::global() shares one set process-wide, enable_pooling(n) sizes or disables it, and recycling is transparent: it applies to the engines that benefit and quietly skips the rest.

Building a codec, then using it

Each format module's compress/decompress is the whole-buffer convenience. When a setting matters, build the codec through its builder and hand it to the crate-level compress, which takes any compressor whatever built it:

let compressor = gzip::Compressor::builder()
    .level(Level::HIGH)
    .output_chunk_size(chunk(16 * 1024))
    .build(resources);

let body = BytesView::copied_from_slice(b"a response body", resources.memory());
let compressed = compressors::compress(body, compressor)?;

The same compressor can instead be driven incrementally, or handed to CompressionStream; building it is the same either way. Committing to a format also unlocks that format's own settings, and the formats whose engines validate their configuration -- brotli and zstd -- report that from build rather than deferring it to the first chunk:

let compressor = brotli::Compressor::builder()
    .quality(brotli::Quality::new(4).expect("in range"))
    .window_size(brotli::WindowSize::new(18).expect("in range"))
    .mode(brotli::Mode::Text)
    .build(resources)?;

Streamed compression and decompression

A codec is a state machine, not a one-shot transform, so a stream of any length moves through it with a bounded working set -- one pending input view and one output chunk, however many gigabytes pass through. Behind the futures-stream feature, CompressionStream presents that as a futures_core::Stream, turning any stream of byte sequences into its compressed or decompressed counterpart.

Runtime format selection

The format module is where a format that is only known at runtime lives, and it has the same shape as every compile-time format module: a Compressor, a Decompressor, and compress / decompress / decompress_with_limits, with the Format threaded through.

let format = Format::from_content_encoding(encoding).expect("a supported encoding");

let compressed = format::compress(format, b"negotiated body", resources)?;
let plain = format::decompress(format, compressed, resources)?;

CompressorBuilder::build_format produces one when the level or the chunk size matters. It returns the module's own Compressor -- a concrete type holding the chosen format internally -- rather than a boxed trait object, so the runtime-format path is not a second-class citizen and the mechanics that drive a codec stay out of the public API.

let compressor = CompressorBuilder::new()
    .level(Level::FAST)
    .build_format(format, resources)?;

Bounded decompression

Every one of these formats can expand its input by orders of magnitude. Nothing in the crate accumulates, so the exposure is in what a caller buffers: DecompressorLimits documents what each format bounds by default, why a ratio alone is not protection, and what to set for untrusted input.

let decompressor = gzip::Decompressor::builder()
    .limits(DecompressorLimits::new().with_max_output_len(NonZeroU64::new(8 * 1024 * 1024).unwrap()))
    .build(resources);

let plain = compressors::decompress(untrusted, decompressor)?;

Each bound takes a non-zero type, so "allow nothing" is not expressible by accident.

Shape of the API

• CompressorBuilder<T> / DecompressorBuilder<T> carry every setting that means the same thing in every format. The type parameter names the format: <()> has not chosen one and gains a build_gzip-style method per enabled format plus build_format(Format, ..); <Brotli> gains brotli's quality, window and content mode.
• Brotli and zstd validate their configuration as they apply it, so their build returns a BuildError rather than deferring the failure to the first chunk.
• compress / decompress at the crate root take any operation, statically dispatched.
• core::Compression is the contract the formats share, so an API can name an operation: impl Compression<Mode = Compress> accepts any compressor and no decompressor. How a codec is actually driven -- push, pull, end of input -- lives on a crate-private supertrait, so it is not public API.
• Error and BuildError both implement recoverable::Recovery, so a caller with a uniform retry policy can classify either. A truncated stream reports Unknown rather than Retry: re-running the same decode is deterministic, so whether asking again helps belongs to whoever owns the byte source.
• Error::other wraps a foreign failure and detects its recovery from an io::Error anywhere in the cause chain; Error::other_with_recovery takes the classification when the caller knows better.
• A build with no format enabled still gets the shared contract, the builders, Resources, and the format module's types -- there is simply no Format variant to hand them.

Testing

• One contract suite runs every format through the same scenarios, so a format that behaves differently from its siblings fails there rather than surprising a consumer.
• 100% line coverage, no surviving mutants. Test modules are excluded from the coverage gate, so the figure is production code.
• Clippy clean across the feature matrix, including a build with no format at all, and doctests pass on a single-format build rather than only under --all-features.

[github-actions]

✅ Version increments look sufficient

cargo semver-checks compared the 1 crate(s) this PR publishes against their previous version-bump commit in git history. Every version increment is sufficient for the detected API changes.

Crate	Baseline	Baseline commit	This PR	Minimum required	Status
compressors	new crate	—	0.1.0	0.1.0	✅ ok

This check is informational and does not block the merge.

View the check run

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 100.0%. Comparing base (1b0f3fe) to head (a3f1881).
⚠️ Report is 2 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff            @@
##             main     #722     +/-   ##
=========================================
  Coverage   100.0%   100.0%             
=========================================
  Files         587      605     +18     
  Lines       63007    64398   +1391     
=========================================
+ Hits        63007    64398   +1391     

Flag	Coverage Δ
linux	69.3% <100.0%> (?)
linux-arm	68.5% <100.0%> (?)
windows	69.3% <100.0%> (?)

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[martintmk]

🔄 [AspBot] ## Automated multi-facet review — PR #722 (feat(compressors): add streaming compression crate)

This PR was reviewed across build, correctness, complexity, consolidation, idiomaticity, documentation, security, and performance facets. Overall this is a well-engineered, defensively-written, and unusually well-documented crate. The typestate builders, sealed traits, canonical error type, MaybeUninit-based zero-copy output path, and saturating limit arithmetic are all high quality.

Overall assessment: REQUEST CHANGES — one High-severity, safe-by-default hardening item; everything else is Medium/Low polish.

---

🔴 High

H1 — Decompression is effectively UNBOUNDED BY DEFAULT for every format (CWE-409/770/400)
FormatLimits::new(max_ratio, max_output) has no stream-count parameter, and per-format decompressor defaults ship no absolute output cap — brotli defaults to new(None, None) (zero bounds), zstd to ratio-only 250,000×, deflate/gzip to ratio-only 1,100×. decompress() accumulates all output into one BytesBuf, so a small crafted body can expand to multi-GB → OOM DoS on the crate's stated use case (decompressing untrusted Content-Encoding bodies). Ratio-only caps are not real protection (250,000× lets ~1 MiB → ~244 GiB). The enforcement mechanism itself is sound — only the default policy is permissive.

• Fix (prepared): add a max_streams param to FormatLimits::new and ship conservative safe-by-default caps for every decompressor — 64 MiB absolute output + 1024 streams — with explicit opt-out via with_max_output_len / with_max_streams / UNLIMITED. Values chosen to sit above the largest legitimate default-path test payload (~22.5 MiB) and concatenation count (2–3), so existing tests remain green. Also update the two doc lines that now contradict the bounded defaults (limits.rs:67-68, flate/mod.rs:21).

🟠 Medium

• Security M1 — zstd decompressor window-log defaults to 128 MiB for untrusted input (zstd/codec.rs:190-194); consider a stricter max_window_log default or document the per-stream cost.
• Security/Correctness M2/M3 — gzip multi-member: unbounded member count + a fresh new_gzip inflate-state allocation per member (raw/zlib recycle via reset(), only gzip reallocates) → ~1000× alloc/CPU amplification from many tiny empty members. Truncated subsequent member is mislabeled corrupt_data vs unexpected_end_of_stream. Fix via the default max_streams above + reuse inflate state + EOF label.
• Security M4 — unsafe { output.advance(produced) } (engine.rs:348) is OOB-sound, but delegates the initialized invariant to the safe Codec::step; mark Codec::step unsafe with a # Safety clause or zero-fill in the driver.
• Perf H1 (Medium impact) — brotli & zstd zero-fill the whole output chunk (up to 64 KiB) before every engine step even though both backends are write-only; flate proves it's avoidable via *_uninit. The MaybeUninit abstraction is defeated on the hottest path.
• Perf H2/H3 — empty source chunks enter the full hot codec path (reserve + zero-fill + native step + self-wake); whole-buffer compress/decompress pays streaming-chunking overhead. Perf H4 — the flagship CompressionStream incremental path has zero benchmark coverage.
• Complexity — Pump::pull is a single ~185-line function and the StreamEnd match has 12 guard-heavy arms recomputing max_streams() up to 3×/iteration; extract cohesive helpers.
• Consolidation — the decompressor limit-delegation trio and stream_ended() logic are byte-identical across all three codecs; the unsafe initialize() helper is copy-pasted verbatim in two files (duplicated unsafe is the riskiest kind).

🟡 Low (defense-in-depth / polish)

• stream.rs source is not fused (possible re-poll-after-None panic on caller-supplied Compression); self-wake spin can peg CPU on a perpetually-ready empty source.
• Pooled engine buffers are not zeroized between uses (in-memory remanence; reset-on-checkout already prevents any functional cross-message leak).
• zstd is the only C parser on the untrusted path — add a cargo audit/cargo deny CI gate.
• output_chunk_size has no upper bound; dead done_reported field; brotli collapses NeedsMoreInput/NeedsMoreOutput into one state; crate-root module named core forces ::core:: disambiguation and inconsistent Result spellings; a handful of near-duplicate per-format unit tests could share a harness.

✅ Verified clean

No memory-safety bug, no exploitable panic, no integer-overflow bug. All 4 unsafe blocks proven sound (the task-flagged flate/codec.rs:233 from_raw_parts is test-only and sound). FFI return codes checked; no content-size trust / no huge-alloc bomb pre-allocation; saturating integer math; no PII in error messages; fail-closed config validation; zero-copy input/output paths; no O(n²) append; dependencies current and advisory-clean (flate2 → zlib-rs only, avoiding the C-zlib CVE class). Idiomaticity is exemplary.

Build/clippy/test status

⚠️ Not verified by compiler. The review environment's egress to index.crates.io / static.crates.io was blocked, so cargo build/clippy/test could not fetch dependencies. Source was reviewed at head 17f82b1 via the GitHub API; the H1 fix was implemented on the real tree and parse-checked with rustfmt (clean), and verified complete/non-regressing by static inspection (all 12 FormatLimits::new sites accounted for, enforcement plumbing confirmed end-to-end, existing test payloads confirmed under the new caps). Please run cargo build/test/clippy -p compressors in an environment with crates.io access to confirm.

---

Review performed by an automated multi-agent review team. Line numbers reference head 17f82b1.

[Copilot]

🔵 Needs a closer look

There are correctness issues in the new crate’s fallible-format compress convenience (panic path) and a test that doesn’t actually exercise the custom-memory-provider path it claims to verify.

Review details

Suppressed comments (2)

crates/compressors/src/tests/round_trip.rs:306

• This test intends to verify that a caller-supplied memory provider is used, but it never uses the memory it constructs: it calls gzip::compress/...decompress with Resources::default(), which uses the crate's global resources.

Create a Resources from memory and pass it through so the test actually exercises the custom-provider path.
crates/compressors/src/macros.rs:149

• In the fallible-compressor macro branch, compress(...) calls Compressor::new(resources), which uses an expect(...) internally. That makes this convenience function capable of panicking on a build-time rejection (e.g., if an upstream engine version changes what it accepts), even though the signature advertises error reporting via Result.

Prefer building via the builder and propagating BuildError (which already converts into Error) so this function never panics for configuration rejection.

        pub fn compress(input: impl $crate::InputData, resources: &$crate::Resources) -> Result<BytesView> {
            let input = $crate::InputData::into_view(input, resources);

            $crate::compress(input, Compressor::new(resources))
        }

• Files reviewed: 37/40 changed files
• Comments generated: 0 new
• Review effort level: Lite

[martin-kolinek]

🤖: The no-format build no longer exports Level, even though its public builders still expose methods that require it. Restore the unconditional pub use level::Level, or gate every public API that names Level consistently.

When pub use format::Format was removed, its #[cfg(...)] remained and now applies to the next item, Level. A downstream crate using the advertised featureless contract can therefore name CompressorBuilder, but cannot supply the public type required by .level(...).

[martin-kolinek]

🤖: Non-blocking — The compress docs still promise a boxed runtime-selected compressor, but that type no longer implements Compression. Name the concrete format::Compressor returned by build_format instead, and make the same correction in the generated compressor docs in src/macros.rs.

The runtime-format redesign removed impl Compression for Box<dyn Compression> and made Compression: Sized, so readers following the boxed guidance reach a value that cannot be passed to this helper.

[Copilot]

🔵 Needs a closer look

A newly added test claims to verify use of a custom memory provider but currently uses Resources::default() (ignoring the custom provider), so it does not validate its stated behavior.

Review details

Suppressed comments (1)

crates/compressors/src/tests/round_trip.rs:307

• This test claims to validate that a custom MemoryShared provider is used for output, but it still calls gzip::compress/gzip::decompress with &Resources::default(), so the memory created here is unused (and the view(..) input is also allocated from an unrelated throwaway pool). Build a Resources from memory and use it for both calls (and pass a slice so input is allocated from that same provider too).

• Files reviewed: 37/40 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🔵 Needs a closer look

There are a couple of concrete correctness/documentation issues in new code (a test that doesn't exercise what it claims, and a broken intra-doc link) that should be fixed before approval.

Review details

Suppressed comments (3)

Previously missed (1) — in code that hasn't changed since the last review.

crates/compressors/src/zstd/codec.rs:28

• This intra-doc link points to crate::zstd::CompressorBuilder, but the zstd module doesn't define a CompressorBuilder type alias. This will render as a broken rustdoc link; link directly to the specialized crate::CompressorBuilder<crate::zstd::Zstd> method instead.

crates/compressors/src/tests/round_trip.rs:306

• This test claims to validate that a custom MemoryShared provider is used, but the compression/decompression calls use &Resources::default() instead of resources built from the GlobalPool created in the test. As written, it doesn't exercise the custom-provider path it describes.
  crates/compressors/src/zstd/codec.rs:350
• The test hard-codes zstd's level range as 1..=22, which can become inaccurate as the bundled zstd library changes. Using zstd_safe::min_c_level()/max_c_level() makes this assertion validate the real engine contract.

• Files reviewed: 39/42 changed files
• Comments generated: 0 new
• Review effort level: Lite

[martin-kolinek]

🤖: Non-blocking — The fallibility summary still says Brotli builders return BuildError, but this push made Brotli compressor construction infallible. Update this section and the matching summary in src/tests/format_contract.rs to identify zstd as the only fallible format.

BrotliCompress::new now returns Self, the Brotli macro invocation selects compressor_build = infallible, and the public BuildError documentation already reflects the new contract. Leaving the generator documentation unchanged gives maintainers the wrong rule for choosing the build macro.

[martin-kolinek]

🤖: Non-blocking — This soundness guide attributes the crate-wide over-report defense to the zstd-only UninitOutput::filled_until helper, and says its clamp rejects an engine report. Describe Pump::pull's produced > provided_output check as the guard that rejects over-reports before BytesBuf::advance.

filled_until only bounds the initialized prefix that zstd can read back; it neither exists for the flate and Brotli adapters nor rejects a count. The engine-independent check in Pump is what keeps an excessive produced value from marking uninitialized output as initialized for every backend.

[martin-kolinek]

🤖: Non-blocking — This guide presents direct compressor driving as a public consumption model, but downstream callers cannot invoke push, pull, or end_input because those methods live on the crate-private CompressionInternal trait. Describe CompressionStream as the public incremental model instead, including its futures-stream feature requirement.

As written, readers are directed toward an API the crate intentionally sealed in this same push; only the whole-buffer helpers and CompressionStream are available to callers.

[Copilot]

🔵 Needs a closer look

There are a couple of correctness/API-consistency issues in the new code (notably a test that doesn’t actually exercise custom resources and a fallible-format compress convenience that can still panic) that should be fixed before approval.

Review details

Suppressed comments (2)

crates/compressors/src/tests/round_trip.rs:307

• This test claims to validate that a custom MemoryShared provider is used for output, but it builds/uses Resources::default() (the global resources) for both compress and decompress, so the memory created above is not actually exercised by the codec output path.
  crates/compressors/src/macros.rs:149
• In the fallible format case, this compress convenience returns Result and documents errors, but it calls Compressor::new(resources) which can panic if the engine rejects the default configuration. Building via the fallible builder keeps the API consistent (no panic) and lets BuildError convert into Error.

        pub fn compress(input: impl $crate::InputData, resources: &$crate::Resources) -> Result<BytesView> {
            let input = $crate::InputData::into_view(input, resources);

            $crate::compress(input, Compressor::new(resources))
        }

• Files reviewed: 39/42 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🔵 Needs a closer look

A newly added test (a_custom_memory_provider_is_used_for_output) does not actually use the custom memory provider it claims to validate, so it needs to be corrected before approval.

Review details

Suppressed comments (1)

crates/compressors/src/tests/round_trip.rs:307

• This test claims to validate use of a custom MemoryShared provider, but it never uses the memory it creates: both compress and decompress are run with Resources::default() (the global resources). As written, it doesn't actually exercise the intended behavior.

• Files reviewed: 39/42 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🔵 Needs a closer look

It contains at least one behavior-test that does not actually exercise the intended custom-memory path and a macro-generated API that can panic on a fallible default build instead of returning an error.

Review details

Suppressed comments (2)

crates/compressors/src/macros.rs:149

• For fallible formats (e.g. zstd), this whole-buffer compress convenience can panic because it calls Compressor::new(resources), which uses .expect(...) if the engine rejects even the default configuration. Since this function already returns Result<BytesView>, it should propagate a build rejection as an error (leveraging the existing From<BuildError> for Error) instead of panicking.

        pub fn compress(input: impl $crate::InputData, resources: &$crate::Resources) -> Result<BytesView> {
            let input = $crate::InputData::into_view(input, resources);

            $crate::compress(input, Compressor::new(resources))
        }

crates/compressors/src/tests/round_trip.rs:306

• This test claims to validate that a custom MemoryShared provider is used for output, but it builds compressors with Resources::default() (global resources) and also constructs input via view(...) which allocates from its own throwaway pool. As written, the custom memory provider isn’t actually exercised by the compressor/decompressor under test.

• Files reviewed: 39/42 changed files
• Comments generated: 0 new
• Review effort level: Lite