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 (8624f90) to head (fa5ef10).

Additional details and impacted files

@@            Coverage Diff            @@
##             main     #722     +/-   ##
=========================================
  Coverage   100.0%   100.0%             
=========================================
  Files         611      634     +23     
  Lines       83142    84733   +1591     
=========================================
+ Hits        83142    84733   +1591     

Flag	Coverage Δ
linux	100.0% <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]

🔵 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

[Copilot]

🟢 Approval recommended

The changes are consistent with the workspace’s packaging/LFS policy and include substantial tests and documentation for the new crate’s safety and API contracts.

Review details

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

[Copilot]

🟢 Approval recommended

The changes appear internally consistent with the stated design/security goals and include substantial documentation plus deterministic and property-based coverage for the new API surface.

Review details

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

[Copilot]

🔵 Needs a closer look

It introduces a large new compression/decompression crate with unsafe codec boundaries and multiple backend integrations that warrant final human review.

Review details

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

[martin-kolinek]

🤖: max_output_len(NonZeroU64::new(configured_limit)) now silently disables the decompression ceiling when configured_limit is zero. Keep bound-setting value-only and expose bound removal through an explicitly named operation; apply the same separation to the ratio and stream setters.

NonZeroU64::new(0) returns None, and this setter interprets None as Limit::Unlimited, overriding the buffering helper's 64 MiB fallback. An invalid external limit can therefore turn bounded decompression into unbounded memory growth instead of being rejected.

[Copilot]

🟢 Approval recommended

The changes are coherent and well-tested overall, with only a small maintainability nit called out in the review comments.

Review details

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

[Copilot]

🔵 Needs a closer look

It introduces a large new crate with multiple backends and unsafe codec/pump invariants, which warrants final human review despite only minor review notes found.

Review details

Suppressed comments (1)

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

crates/compressors/src/stream.rs:112

• The doc comment implies a source of plain BytesView values will compile and then only fail at runtime when polled, but CompressionStream only implements Stream when the source item type is Result<BytesView, E>. As written, a plain-view source won’t type-check as a Stream consumer at all, so this should be described as a trait-bound/impl-availability constraint rather than a runtime failure mode.

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

[Copilot]

🔵 Needs a closer look

It introduces a large new crate with format backends and unsafe codec boundaries, so it warrants final human review even though only minor issues were identified.

Review details

Suppressed comments (1)

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

crates/compressors/tests/format.rs:11

• The test name says it verifies an unknown content-encoding is rejected, but it actually asserts that the well-known "identity" token maps to None. Either rename the test (recommended) or change the asserted token to something genuinely unknown to keep the intent aligned with the assertion.

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