znippy-common/src/codec.rs
//! Codec layer: OpenZL compression/decompression.
//! OpenZL is znippy's codec — it wraps zstd+lz4 with improved framing.
//!
//! # The `openzl` feature — this file is the ONE gate
//!
//! `openzl-sys-rs` 0.4 vendors the OpenZL, zstd and lz4 C sources and compiles
//! them with `cc` — no network, no shell, no `curl`/`tar`/`cmake`, no `stdc++`.
//! What it still requires is a build script and a C compiler, which a consumer
//! targeting pure Rust, or building with no C toolchain present, does not have.
//!
//! Rather than `#[cfg]` the six call sites that touch the codec — `archive.rs`,
//! `views.rs`, `decompress.rs` (×2), `meta_sink_append.rs`, `plugins/wasm_loader.rs`
//! — and fracture the public API into two shapes, the gate lives **here alone**
//! (LAW 5: one writer both paths route through). Every item below keeps its exact
//! signature with the feature off; the bodies return [`NO_CODEC`] instead. So:
//!
//! - the default build is byte-identical to before — same code, same behaviour;
//! - `default-features = false` drops `openzl-sys-rs` from the graph outright,
//! leaving the archive format, the Arrow index, the searchable metadata
//! sub-index and the plugin trait fully usable with no build script at all;
//! - a caller that *does* reach a compressed blob in such a build gets a named
//! error, never a wrong answer or a silent empty result.
//!
//! Reading **stored** (uncompressed) entries needs no codec and keeps working
//! either way — that path never enters this module.
use anyhow::{Result, anyhow};
/// The error every entry point in this module returns when the crate was built
/// without the `openzl` feature. Names the feature so the fix is in the message.
#[cfg(not(feature = "openzl"))]
const NO_CODEC: &str = "znippy-common was built without the `openzl` feature, so the OpenZL codec \
is not linked: compressed blobs cannot be read or written. Stored (uncompressed) \
entries, the Arrow index and the metadata sub-index are unaffected. Enable the \
`openzl` feature to link the codec (note: it builds the vendored OpenZL C sources \
and needs a C compiler at compile time).";
// ─── Frame format written ───────────────────────────────────────────
/// The OpenZL frame format every archive blob is **written** in. Reading is not
/// limited by it: the linked library decodes every format it knows (8..=27 with
/// openzl-sys-rs 0.4 / OpenZL 0.3.0).
///
/// Pinned to 25 on 2026-10-02 so archives stay readable by readers on
/// openzl-sys-rs 0.3.x (OpenZL 0.2.1, max format 25) — the estates, holger and
/// gunnar binaries already deployed. OpenZL 0.3.0's own default
/// (`ZL_getDefaultEncodingVersion`) is 27, which those readers refuse.
///
/// The flip is two releases, on purpose:
/// 1. this release: every reader links openzl-sys-rs 0.4 (understands 27),
/// writers keep producing 25;
/// 2. the next release, once test2 and the gunnar.rs estates run (1): change
/// this one line to 27.
///
/// Raising it before every reader runs 0.4 makes new archives unreadable to
/// old binaries. `writes_the_pinned_format` checks the value reaches the frame.
pub const OPENZL_WRITE_FORMAT: i32 = 25;
// ─── Compression Context ────────────────────────────────────────────
pub struct CompressCtx {
#[cfg(feature = "openzl")]
cctx: openzl_sys_rs::ZlCCtx,
}
unsafe impl Send for CompressCtx {}
#[cfg(not(feature = "openzl"))]
impl CompressCtx {
pub fn new(_compression_level: i32) -> Result<Self> {
Err(anyhow!(NO_CODEC))
}
pub fn compress(&mut self, _input: &[u8]) -> Result<Vec<u8>> {
Err(anyhow!(NO_CODEC))
}
pub fn compress_into(&mut self, _input: &[u8], _out: &mut Vec<u8>) -> Result<usize> {
Err(anyhow!(NO_CODEC))
}
}
#[cfg(not(feature = "openzl"))]
pub fn decompress_frame(_compressed: &[u8]) -> Result<Vec<u8>> {
Err(anyhow!(NO_CODEC))
}
#[cfg(not(feature = "openzl"))]
pub fn decompress_into(_compressed: &[u8], _out: &mut Vec<u8>) -> Result<usize> {
Err(anyhow!(NO_CODEC))
}
#[cfg(not(feature = "openzl"))]
pub fn decompress_into_budgeted(_compressed: &[u8], _out: &mut Vec<u8>, _max: usize) -> Result<usize> {
Err(anyhow!(NO_CODEC))
}
#[cfg(not(feature = "openzl"))]
pub fn decompress_into_max(_compressed: &[u8], _out: &mut Vec<u8>, _max: usize) -> Result<usize> {
Err(anyhow!(NO_CODEC))
}
#[cfg(feature = "openzl")]
impl CompressCtx {
pub fn new(compression_level: i32) -> Result<Self> {
use openzl_sys_rs::*;
let mut cctx = ZlCCtx::new().ok_or_else(|| anyhow!("ZL_CCtx_create failed"))?;
// stickyParameters=1 keeps params across compress calls (reuse ctx)
cctx.set_parameter(ZL_CParam_ZL_CParam_stickyParameters, 1)
.map_err(|e| anyhow!(e))?;
// NOT ZL_getDefaultEncodingVersion(): that is 27 with OpenZL 0.3.0, which
// readers on openzl-sys-rs 0.3.x cannot decode. See OPENZL_WRITE_FORMAT.
cctx.set_parameter(ZL_CParam_ZL_CParam_formatVersion, OPENZL_WRITE_FORMAT)
.map_err(|e| anyhow!(e))?;
cctx.set_parameter(ZL_CParam_ZL_CParam_compressionLevel, compression_level)
.map_err(|e| anyhow!(e))?;
Ok(Self { cctx })
}
pub fn compress(&mut self, input: &[u8]) -> Result<Vec<u8>> {
use openzl_sys_rs::zl_compress_bound;
let bound = zl_compress_bound(input.len());
let mut output = vec![0u8; bound];
let compressed_size = self.cctx.compress(&mut output, input)
.map_err(|e| anyhow!(e))?;
output.truncate(compressed_size);
Ok(output)
}
/// Compress into a reusable buffer; returns the number of bytes written
/// (`out` is truncated to that). Reuse the same `out` across slices to keep
/// the compress path allocation-free (the no-hot-path-alloc rule).
pub fn compress_into(&mut self, input: &[u8], out: &mut Vec<u8>) -> Result<usize> {
use openzl_sys_rs::zl_compress_bound;
let bound = zl_compress_bound(input.len());
if out.len() < bound {
out.resize(bound, 0);
}
let compressed_size = self
.cctx
.compress(out.as_mut_slice(), input)
.map_err(|e| anyhow!(e))?;
out.truncate(compressed_size);
Ok(compressed_size)
}
}
#[cfg(feature = "openzl")]
pub fn decompress_frame(compressed: &[u8]) -> Result<Vec<u8>> {
let mut out = Vec::new();
decompress_into(compressed, &mut out)?;
Ok(out)
}
/// Decompress into a reusable buffer; returns bytes written (`out` is truncated
/// to that). Reuse the same `out` across chunks to keep the decompress path
/// allocation-free (the no-hot-path-alloc rule).
#[cfg(feature = "openzl")]
pub fn decompress_into(compressed: &[u8], out: &mut Vec<u8>) -> Result<usize> {
use openzl_sys_rs::*;
let decompressed_size = zl_get_decompressed_size(compressed)
.map_err(|e| anyhow!("OpenZL getDecompressedSize: {}", e))?;
if out.len() < decompressed_size {
out.resize(decompressed_size, 0);
}
let written = zl_decompress(&mut out[..decompressed_size], compressed)
.map_err(|e| anyhow!("OpenZL decompress: {}", e))?;
out.truncate(written);
Ok(written)
}
// ─── Untrusted frames: the allocation budget ────────────────────────
/// How many times a chunk's declared size OpenZL may allocate while decoding
/// it, on top of [`OPENZL_BUDGET_MARGIN`].
///
/// Measured on oden (2026-10-31, lane gatwasm) with openzl-sys-rs 0.4.2: an
/// honest frame charges a FIXED 7,612 to 11,988 bytes whatever its size, from
/// 0 bytes to 64 MiB, for text, zeros, random, mixed and integer data, frame
/// formats 25 and 27, levels 1 to 19, the 0.3.1-written fixture and a delta
/// instruction stream: the default graph decodes straight into the output and
/// zstd's own context is not charged. So the margin carries every honest frame
/// today; the factor 2 leaves room for a writer whose graph keeps intermediate
/// streams (each at most the output's size) without letting a frame allocate
/// more than twice what its index row declares.
pub const OPENZL_BUDGET_FACTOR: usize = 2;
/// The fixed part of the budget: 1 MiB, about 85 times the 11,988 bytes the
/// largest honest frame was charged.
pub const OPENZL_BUDGET_MARGIN: usize = 1 << 20;
/// The allocation budget for decoding a frame of `max` declared bytes.
pub fn openzl_budget(max: usize) -> usize {
max.saturating_mul(OPENZL_BUDGET_FACTOR).saturating_add(OPENZL_BUDGET_MARGIN)
}
/// [`decompress_into_max`] for an UNTRUSTED frame: OpenZL's own allocations
/// while decoding it are charged to a per-thread budget of
/// [`openzl_budget`]`(max)` (openzl-sys-rs `with_alloc_budget`). OpenZL sizes
/// its internal streams from the frame's graph, not from the header, so without
/// this a 36-byte frame can make it allocate and fill a gigabyte before it fails
/// (the heimdall fuzz find). A frame over its budget is refused by name.
///
/// The output buffer itself is Rust's and bounded by `max` before it grows.
/// The fast (trusted) read does not use this; it costs one thread-local read
/// per OpenZL allocation where it is used.
#[cfg(feature = "openzl")]
pub fn decompress_into_budgeted(compressed: &[u8], out: &mut Vec<u8>, max: usize) -> Result<usize> {
let budget = openzl_budget(max);
let (got, spent) = openzl_sys_rs::with_alloc_budget(budget, || decompress_into_max(compressed, out, max));
if spent.exceeded {
return Err(anyhow!(
"OpenZL frame asked for more memory than its {budget}-byte allocation budget \
({OPENZL_BUDGET_FACTOR} x the {max} bytes its chunk declares + {OPENZL_BUDGET_MARGIN}); refused"
));
}
got
}
/// [`decompress_into`] for a frame the caller has a size for: a frame whose
/// header declares more than `max` bytes is refused BEFORE `out` is grown, so
/// the header alone can never size an allocation. The reader passes the index
/// row's `uncompressed_size` (see `archive::frame_limit`).
///
/// OpenZL's own internal buffers are not bounded by this; they are sized from
/// the frame's graph, which OpenZL 0.3.0 gives no allocator hook for.
#[cfg(feature = "openzl")]
pub fn decompress_into_max(compressed: &[u8], out: &mut Vec<u8>, max: usize) -> Result<usize> {
let declared = openzl_sys_rs::zl_get_decompressed_size(compressed)
.map_err(|e| anyhow!("OpenZL getDecompressedSize: {}", e))?;
if declared > max {
return Err(anyhow!(
"OpenZL frame declares {declared} decompressed bytes, more than the {max} its chunk allows"
));
}
decompress_into(compressed, out)
}
/// With the `openzl` feature OFF the codec is not linked. Prove that every entry
/// point says so **out loud** — an `Err` naming the feature — instead of the two
/// failure modes that would actually hurt: a silent `Ok` with empty/garbage bytes,
/// or a panic. LAW 2: this is the red the default build can never show, so it is
/// asserted from the build that can.
#[cfg(all(test, not(feature = "openzl")))]
mod no_codec_tests {
use super::*;
fn assert_names_the_feature(err: anyhow::Error) {
let msg = err.to_string();
assert!(
msg.contains("`openzl` feature"),
"error must name the feature that is missing, got: {msg}"
);
}
#[test]
fn compress_ctx_refuses_to_exist() {
assert_names_the_feature(
CompressCtx::new(3).err().expect("CompressCtx::new must fail with no codec linked"),
);
}
#[test]
fn decompress_refuses_and_does_not_touch_the_buffer() {
// A real OpenZL frame header; without the codec it must not be decoded,
// and `out` must be left exactly as the caller handed it over.
let frame = [0x5Bu8, 0x2A, 0x4D, 0x18, 0x00, 0x00, 0x00, 0x00];
let mut out = vec![0xAAu8; 4];
assert_names_the_feature(
decompress_into(&frame, &mut out).err().expect("decompress_into must fail"),
);
assert_eq!(out, vec![0xAAu8; 4], "buffer must be untouched on refusal");
assert_names_the_feature(
decompress_frame(&frame).err().expect("decompress_frame must fail"),
);
}
}
#[cfg(all(test, feature = "openzl"))]
mod tests {
use super::*;
#[test]
fn test_roundtrip() {
let mut ctx = CompressCtx::new(3).unwrap();
let input = b"Hello world! This is a test of compression roundtrip. Repeated data helps compression. Repeated data helps compression. Repeated data helps compression.";
let compressed = ctx.compress(input).unwrap();
println!("Compressed {} -> {} bytes", input.len(), compressed.len());
let decompressed = decompress_frame(&compressed).unwrap();
assert_eq!(&decompressed[..], &input[..]);
}
#[test]
fn test_multi_compress_same_ctx() {
let mut ctx = CompressCtx::new(3).unwrap();
for i in 0..10 {
let input: Vec<u8> = (0..4096).map(|x| ((x + i) % 251) as u8).collect();
let compressed = ctx.compress(&input).unwrap();
let decompressed = decompress_frame(&compressed).unwrap();
assert_eq!(decompressed, input, "Failed at iteration {}", i);
}
println!("10 sequential compress calls OK");
}
#[test]
fn test_parallel_contexts() {
let handles: Vec<_> = (0..8).map(|t| {
std::thread::spawn(move || {
let mut ctx = CompressCtx::new(3).unwrap();
for i in 0..5 {
let input: Vec<u8> = (0..8192).map(|x| ((x + i + t*100) % 251) as u8).collect();
let compressed = ctx.compress(&input).unwrap();
let decompressed = decompress_frame(&compressed).unwrap();
assert_eq!(decompressed, input);
}
})
}).collect();
for h in handles {
h.join().unwrap();
}
println!("8 parallel contexts x 5 calls each OK");
}
/// The frame format a blob was written in, read from the frame header.
fn frame_format(frame: &[u8]) -> usize {
use openzl_sys_rs::*;
let r = unsafe { ZL_getFormatVersionFromFrame(frame.as_ptr() as *const _, frame.len()) };
assert!(!zl_is_error(r), "not a readable OpenZL frame");
zl_report_value(r)
}
fn sample() -> Vec<u8> {
(0..65_536u32).map(|x| ((x * 7 + x / 13) % 251) as u8).collect()
}
/// Writes stay on the pinned format even though the linked OpenZL's own
/// default is newer — the guarantee old readers depend on.
#[test]
fn writes_the_pinned_format() {
let default = unsafe { openzl_sys_rs::ZL_getDefaultEncodingVersion() } as i32;
assert!(
default >= OPENZL_WRITE_FORMAT,
"linked OpenZL defaults to {default}, older than the pinned write format"
);
let frame = CompressCtx::new(3).unwrap().compress(&sample()).unwrap();
assert_eq!(frame_format(&frame), OPENZL_WRITE_FORMAT as usize);
assert_eq!(decompress_frame(&frame).unwrap(), sample());
}
/// The reader decodes the newest format this OpenZL writes (27 with 0.3.0),
/// which is what writers will switch to in the release after the pin.
#[test]
fn reads_the_latest_format() {
use openzl_sys_rs::*;
let latest = unsafe { ZL_getDefaultEncodingVersion() } as i32;
let mut cctx = ZlCCtx::new().unwrap();
cctx.set_parameter(ZL_CParam_ZL_CParam_formatVersion, latest).unwrap();
cctx.set_parameter(ZL_CParam_ZL_CParam_compressionLevel, 3).unwrap();
let input = sample();
let mut frame = vec![0u8; zl_compress_bound(input.len())];
let n = cctx.compress(&mut frame, &input).unwrap();
frame.truncate(n);
assert_eq!(frame_format(&frame), latest as usize);
assert_eq!(decompress_frame(&frame).unwrap(), input);
}
/// A frame written by the OLD wrapper (openzl-sys-rs 0.3.1 = OpenZL 0.2.1,
/// format 25) — every archive on disk today — still decodes. The fixture and
/// its plaintext were produced by openzl-sys-rs 0.3.1 from crates.io; see
/// tests/fixtures/openzl/README.md.
#[test]
fn reads_frames_from_wrapper_0_3_1() {
let frame = include_bytes!("../tests/fixtures/openzl/wrapper-0.3.1-format25.zl");
let plain = include_bytes!("../tests/fixtures/openzl/plain.bin");
assert_eq!(frame_format(frame), 25);
assert_eq!(decompress_frame(frame).unwrap(), &plain[..]);
}
}