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. Output buffers come from the caller's own memory provider, and so does any input view the crate has to build.

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 an engine 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, with_pool_capacity(n) sizes or disables it, and recycling is transparent: it applies to the engines that benefit and quietly skips the rest.

Building a compressor, then using it

Each format module's compress/decompress is the whole-buffer convenience. When a setting matters, build the compressor through its builder and hand it to the crate-level compress, which accepts any implementor of Compression, however it was constructed:

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 zstd, whose native library validates the parameters it is given, reports 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

An engine is a state machine, not a one-shot transform, so a stream of any length moves through it while the output it has buffered but not yet handed back stays bounded by the configured chunk size. Pending input and the engine's own window and tables are additional, and depend on the format. 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 an engine 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().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.
• zstd validates its configuration as it applies it, so its build returns a BuildError rather than deferring the failure to the first chunk. Brotli's construction is infallible: no value its builders can express is one the encoder rejects.
• compress / decompress at the crate root take any engine, statically dispatched.
• core::Compression is the contract the formats share, so an API can name an engine: impl Compression<Mode = Compress> accepts any compressor and no decompressor. How an engine is actually driven -- push, pull, end of input -- lives on a crate-private supertrait, and the trait is Sized so no dyn Compression can expose it. Neither is 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.
• Resources implements ThreadAware. Relocation moves the memory provider, since a NUMA-aware provider will want the destination's memory; the engine pool is a deliberate no-op, because every clone shares it and an idle engine is plain memory with no affinity to where it was built.

Documentation

docs/DESIGN.md records the user-visible policies that span several APIs -- format selection, what is uniform across formats and what is not, how decompression is bounded, stream framing, and why the public surface is sealed. docs/IMPLEMENTATION.md covers the mechanisms: the pump state machine, the unsafe initialized-output contract every backend adapter honours, engine pooling and its exclusions, and the async driving rules.

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 (efd8178) to head (276fee0).

Additional details and impacted files

@@            Coverage Diff            @@
##             main     #722     +/-   ##
=========================================
  Coverage   100.0%   100.0%             
=========================================
  Files         583      606     +23     
  Lines       62930    64533   +1603     
=========================================
+ Hits        62930    64533   +1603     

Flag	Coverage Δ
linux	99.9% <100.0%> (?)
linux-arm	100.0% <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]

🟡 Changes recommended

crates/compressors/src/core/output.rs matches on *self in several predicate/accessor methods, which attempts to move a non-Copy enum out of &self and should be rewritten to match on self by reference.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (4)

crates/compressors/src/core/output.rs:65

• as_data uses match *self / ref, which attempts to move the enum out of &self. Match on self directly so this method can borrow the contained BytesView without moving the enum.

    pub fn as_data(&self) -> Option<&BytesView> {
        match *self {
            Self::Data(ref data) => Some(data),
            _ => None,
        }

crates/compressors/src/core/output.rs:72

• is_need_input matches on *self, which would require moving the enum out of &self. Use matches!(self, ...) to match on the reference instead.

    #[must_use]
    pub fn is_need_input(&self) -> bool {
        matches!(*self, Self::NeedInput)
    }

crates/compressors/src/core/output.rs:78

• is_progress matches on *self, which would require moving the enum out of &self. Use matches!(self, ...) to match on the reference instead.

    /// Whether the engine made progress and should be pulled again.
    #[must_use]
    pub fn is_progress(&self) -> bool {
        matches!(*self, Self::Progress)
    }

crates/compressors/src/core/output.rs:84

• is_done matches on *self, which would require moving the enum out of &self. Use matches!(self, ...) to match on the reference instead.

    /// Whether the stream has ended.
    #[must_use]
    pub fn is_done(&self) -> bool {
        matches!(*self, Self::Done)
    }

• Files reviewed: 40/42 changed files
• Comments generated: 1
• Review effort level: Lite

[Copilot]

🟢 Approval recommended

The changes introduce a well-structured new crate with consistent API shape and strong test/benchmark/docs support, and the remaining feedback is limited to maintainability nits.

Review details

Suppressed comments (1)

crates/compressors/src/macros.rs:331

• These docs hard-code the buffering caps ("64 MiB" / "1024 streams") even though the authoritative values live in limits.rs and are already documented on DecompressorLimits. Duplicating the numbers here risks doc drift if the policy changes; consider referencing DecompressorLimits instead of repeating the literals.

• Files reviewed: 41/43 changed files
• Comments generated: 2
• Review effort level: Lite

[Copilot]

🔵 Needs a closer look

It introduces a large new crate with unsafe-adjacent buffer/codec boundaries and a broad public API surface that warrants final human review despite only minor fixable nits found.

Review details

Suppressed comments (2)

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

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

• Step cap is described as "within {MAX_STEPS} steps" but the assertion fails at exactly MAX_STEPS (it currently allows only MAX_STEPS - 1 iterations). This is a minor off-by-one that can make a non-spinning test fail earlier than the constant and docs imply.
  crates/compressors/src/pool.rs:237
• The doc comment mentions a zstd compressor built for a level, but take_zstd_compressor is intentionally unkeyed and takes no level parameter. Updating the comment avoids confusion about how pooling works for zstd.

• Files reviewed: 41/43 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Martin Taillefer (geeknoid)]

Five static passes — security, correctness, testing, performance, conformance — scoped to the diff. Nothing was executed: no build, test, coverage, mutation, or Miri run, so everything below is by argument from the source.

Two findings, both inline. The crate-level decompress retrofits the 64 MiB output ceiling but not the 1024-stream cap, so it is not at the parity with the per-format conveniences that its own Security table implies. And the two Level tests assert best <= fast, which equality satisfies — a codec that ignored the configured level would pass them.

What I attacked and could not break. The unsafe initialized-output contract holds at all four production sites: brotli bulk-fills before the MaybeUninit → &mut [u8] cast, zstd's filled_until now asserts in every build, and Pump independently rejects produced > provided_output before advance (engine.rs:472) — defence in depth rather than trust in the codec. The one-probe-byte overshoot is exactly one byte and never reaches the caller; remaining is recomputed per step so it cannot go stale; chunk_size - output.len() cannot underflow, since a full buffer always returns early. Per-format ratio defaults match the documented table. Pooling rests on one uniform decision — reset on checkout, not on return — which is what makes a failed or mid-stream engine harmless to recycle, and gzip decompressors are correctly excluded from it. The truncation-reports-Unknown claim has oracles at both layers. Optional deps are all mirrored as non-optional dev-dependencies, so default = [] does not mean CI tests nothing.

Open questions, each with the experiment rather than a verdict:

• The Miri exclusion is documented and I am not objecting to it, but it leaves the uninit-output contract as the crate's least-checked invariant. Brotli is pure Rust and would run — a brotli-only cfg(miri) pass over the pump would cover the one cast that most needs it.
• limits.rs:315 never exercises output == input * ratio, so a > → >= mutant there may survive. A two-line boundary test settles it.
• Only gzip has an independent decode vector. Decoding one crate-produced zstd or brotli stream with an external tool would retire the doubt for the rest.

[Martin Taillefer (geeknoid)]

I think this name is not appropriate as this crate in fact contains no compressors.

Consider bytesbuf_compression instead

[martintmk]

I think this name is not appropriate as this crate in fact contains no compressors.

Can you elaborate? We have a bunch of processors each under crate::<format>::Processor. While I am not opposed to bytesbuf_compression I would prefer one-word crate name if possible. (compressors is free)

[Copilot]

🔵 Needs a closer look

There are small but concrete correctness/maintainability issues in new test/helper code (stored in PR comments) that should be addressed before approval.

Review details

Suppressed comments (2)

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

crates/compressors/src/pool.rs:442

• Inside this #[cfg(test)] module, the #[cfg(all(not(test), ...))] wrapper() definition can never be compiled, so the test always uses Wrapper::Raw and never exercises the zlib reset path in a zlib-only feature build. Consider making wrapper selection depend on feature flags instead of not(test) so zlib-only configurations actually cover Wrapper::Zlib.
  crates/compressors/src/tests/round_trip.rs:50
• StepGuard::step increments the counter and then asserts self.0 < MAX_STEPS, which means the maximum permitted steps is actually MAX_STEPS - 1 while the panic message claims {MAX_STEPS}. This off-by-one makes the guard stricter than documented and can cause a failure exactly at the configured limit.

• Files reviewed: 45/47 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🔵 Needs a closer look

It introduces a large new crate with a broad public API surface and soundness-sensitive codec boundary, which warrants final human review.

Review details

• Files reviewed: 47/49 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🔵 Needs a closer look

It introduces a large new, security-sensitive crate (including unsafe codec boundary code and many new APIs), so it warrants final human review despite only minor actionable feedback.

Review details

Suppressed comments (1)

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

crates/compressors/src/engine.rs:499

• The invalid byte-counts check is shared by both compression and decompression codecs, but the error message says "compression engine". If a decompressor backend misreports counts, this will be misleading during diagnostics; consider using a direction-neutral message (e.g., "codec" or "backend engine").

• Files reviewed: 48/50 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🟢 Approval recommended

The changes introduce the new crate with consistent feature gating, documented security/limits model, and comprehensive tests/benchmarks, and no objective issues were found in the reviewed diffs.

Review details

• Files reviewed: 48/50 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🔵 Needs a closer look

It introduces a large new crate with multiple backends and soundness-sensitive codec boundaries, so it warrants final human review despite only minor nits found.

Review details

Suppressed comments (1)

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

.spelling:885

• .spelling contains a duplicate entry for zstd's (it already appears earlier in the file). Duplicates add noise and make future dictionary edits harder to review.

• Files reviewed: 48/50 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🔵 Needs a closer look

It introduces a large new crate with multiple backends and unsafe codec adapters, so it needs final human review despite only minor specific issues found here.

Review details

Suppressed comments (2)

crates/compressors/src/core/output.rs:57

• matches!(*self, ...) forces dereferencing (and, if Output is Copy, copying) the entire Output value just to test the variant. Since Output::Data carries a large BytesView (noted above as ~272 bytes), these predicates should match on &self to avoid an unnecessary copy.

    pub fn is_data(&self) -> bool {
        matches!(*self, Self::Data(_))
    }

crates/compressors/src/core/output.rs:78

• Same as above: matches!(*self, ...) needlessly dereferences (and potentially copies) the full Output value. Matching on self avoids copying a large enum payload when checking Progress/Done.

    pub fn is_progress(&self) -> bool {
        matches!(*self, Self::Progress)
    }

• Files reviewed: 48/50 changed files
• Comments generated: 0 new
• Review effort level: Lite

[martin-kolinek]

🤖: The Linux fuzz job compiles this target with no compressor-format features, so every generated case returns at Format::ALL.get(...) and the advertised campaign performs no decompression. Pass --all-features (or explicit format features) to both Linux cargo bolero list and cargo bolero test invocations.

compressors declares default = []; unlike the non-Linux fallback, the Linux commands in justfiles/extended.just do not enable any features. This is the wiring gap in the earlier fuzz-target thread: with Format::ALL empty, line 123 returns for every scenario, so the required 60-second job can remain green without exercising a backend.

[Copilot]

🔵 Needs a closer look

It introduces a large, security-relevant new crate (compression/decompression + unsafe codec boundary) and warrants final human review despite no specific blocking issues found in the provided diffs.

Review details

• Files reviewed: 49/51 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🔵 Needs a closer look

crates/compressors/src/core/output.rs uses matches!(*self, ...) / match *self on a non-Copy enum, which needs to be corrected to inspect &self without moving.

Review details

Suppressed comments (3)

crates/compressors/src/core/output.rs:57

• Output contains BytesView (which is not Copy), so using matches!(*self, ...) attempts to move out of &self. Match on self instead so the enum is inspected by reference.

    pub fn is_data(&self) -> bool {
        matches!(*self, Self::Data(_))
    }

crates/compressors/src/core/output.rs:65

• as_data matches on *self, which would move the Output out of &self. Match on self and bind the payload by reference.

    pub fn as_data(&self) -> Option<&BytesView> {
        match *self {
            Self::Data(ref data) => Some(data),
            _ => None,
        }

crates/compressors/src/core/output.rs:72

• These predicate helpers also use matches!(*self, ...), which tries to move out of &self for a non-Copy enum. Use matches!(self, ...) consistently.

    /// Whether the engine needs more input before it can produce more output.
    #[must_use]
    pub fn is_need_input(&self) -> bool {
        matches!(*self, Self::NeedInput)
    }

• Files reviewed: 49/51 changed files
• Comments generated: 0 new
• Review effort level: Lite