znippy_common/
codec.rs

1//! Codec layer: OpenZL compression/decompression.
2//! OpenZL is znippy's codec — it wraps zstd+lz4 with improved framing.
3//!
4//! # The `openzl` feature — this file is the ONE gate
5//!
6//! `openzl-sys-rs` 0.3.0 vendors the OpenZL, zstd and lz4 C sources and compiles
7//! them with `cc` — no network, no shell, no `curl`/`tar`/`cmake`, no `stdc++`.
8//! What it still requires is a build script and a C compiler, which a consumer
9//! targeting pure Rust, or building with no C toolchain present, does not have.
10//!
11//! Rather than `#[cfg]` the six call sites that touch the codec — `archive.rs`,
12//! `views.rs`, `decompress.rs` (×2), `meta_sink_append.rs`, `plugins/wasm_loader.rs`
13//! — and fracture the public API into two shapes, the gate lives **here alone**
14//! (LAW 5: one writer both paths route through). Every item below keeps its exact
15//! signature with the feature off; the bodies return [`NO_CODEC`] instead. So:
16//!
17//! - the default build is byte-identical to before — same code, same behaviour;
18//! - `default-features = false` drops `openzl-sys-rs` from the graph outright,
19//!   leaving the archive format, the Arrow index, the searchable metadata
20//!   sub-index and the plugin trait fully usable with no build script at all;
21//! - a caller that *does* reach a compressed blob in such a build gets a named
22//!   error, never a wrong answer or a silent empty result.
23//!
24//! Reading **stored** (uncompressed) entries needs no codec and keeps working
25//! either way — that path never enters this module.
26
27use anyhow::{Result, anyhow};
28
29/// The error every entry point in this module returns when the crate was built
30/// without the `openzl` feature. Names the feature so the fix is in the message.
31#[cfg(not(feature = "openzl"))]
32const NO_CODEC: &str = "znippy-common was built without the `openzl` feature, so the OpenZL codec \
33     is not linked: compressed blobs cannot be read or written. Stored (uncompressed) \
34     entries, the Arrow index and the metadata sub-index are unaffected. Enable the \
35     `openzl` feature to link the codec (note: it builds the vendored OpenZL C sources \
36     and needs a C compiler at compile time).";
37
38// ─── Compression Context ────────────────────────────────────────────
39
40pub struct CompressCtx {
41    #[cfg(feature = "openzl")]
42    cctx: openzl_sys_rs::ZlCCtx,
43}
44
45unsafe impl Send for CompressCtx {}
46
47#[cfg(not(feature = "openzl"))]
48impl CompressCtx {
49    pub fn new(_compression_level: i32) -> Result<Self> {
50        Err(anyhow!(NO_CODEC))
51    }
52
53    pub fn compress(&mut self, _input: &[u8]) -> Result<Vec<u8>> {
54        Err(anyhow!(NO_CODEC))
55    }
56
57    pub fn compress_into(&mut self, _input: &[u8], _out: &mut Vec<u8>) -> Result<usize> {
58        Err(anyhow!(NO_CODEC))
59    }
60}
61
62#[cfg(not(feature = "openzl"))]
63pub fn decompress_frame(_compressed: &[u8]) -> Result<Vec<u8>> {
64    Err(anyhow!(NO_CODEC))
65}
66
67#[cfg(not(feature = "openzl"))]
68pub fn decompress_into(_compressed: &[u8], _out: &mut Vec<u8>) -> Result<usize> {
69    Err(anyhow!(NO_CODEC))
70}
71
72#[cfg(feature = "openzl")]
73impl CompressCtx {
74    pub fn new(compression_level: i32) -> Result<Self> {
75        use openzl_sys_rs::*;
76        let mut cctx = ZlCCtx::new().ok_or_else(|| anyhow!("ZL_CCtx_create failed"))?;
77        let version = unsafe { ZL_getDefaultEncodingVersion() } as i32;
78        // stickyParameters=1 keeps params across compress calls (reuse ctx)
79        cctx.set_parameter(ZL_CParam_ZL_CParam_stickyParameters, 1)
80            .map_err(|e| anyhow!(e))?;
81        cctx.set_parameter(ZL_CParam_ZL_CParam_formatVersion, version)
82            .map_err(|e| anyhow!(e))?;
83        cctx.set_parameter(ZL_CParam_ZL_CParam_compressionLevel, compression_level)
84            .map_err(|e| anyhow!(e))?;
85        Ok(Self { cctx })
86    }
87
88    pub fn compress(&mut self, input: &[u8]) -> Result<Vec<u8>> {
89        use openzl_sys_rs::zl_compress_bound;
90        let bound = zl_compress_bound(input.len());
91        let mut output = vec![0u8; bound];
92        let compressed_size = self.cctx.compress(&mut output, input)
93            .map_err(|e| anyhow!(e))?;
94        output.truncate(compressed_size);
95        Ok(output)
96    }
97
98    /// Compress into a reusable buffer; returns the number of bytes written
99    /// (`out` is truncated to that). Reuse the same `out` across slices to keep
100    /// the compress path allocation-free (the no-hot-path-alloc rule).
101    pub fn compress_into(&mut self, input: &[u8], out: &mut Vec<u8>) -> Result<usize> {
102        use openzl_sys_rs::zl_compress_bound;
103        let bound = zl_compress_bound(input.len());
104        if out.len() < bound {
105            out.resize(bound, 0);
106        }
107        let compressed_size = self
108            .cctx
109            .compress(out.as_mut_slice(), input)
110            .map_err(|e| anyhow!(e))?;
111        out.truncate(compressed_size);
112        Ok(compressed_size)
113    }
114}
115
116#[cfg(feature = "openzl")]
117pub fn decompress_frame(compressed: &[u8]) -> Result<Vec<u8>> {
118    let mut out = Vec::new();
119    decompress_into(compressed, &mut out)?;
120    Ok(out)
121}
122
123/// Decompress into a reusable buffer; returns bytes written (`out` is truncated
124/// to that). Reuse the same `out` across chunks to keep the decompress path
125/// allocation-free (the no-hot-path-alloc rule).
126#[cfg(feature = "openzl")]
127pub fn decompress_into(compressed: &[u8], out: &mut Vec<u8>) -> Result<usize> {
128    use openzl_sys_rs::*;
129    let decompressed_size = zl_get_decompressed_size(compressed)
130        .map_err(|e| anyhow!("OpenZL getDecompressedSize: {}", e))?;
131    if out.len() < decompressed_size {
132        out.resize(decompressed_size, 0);
133    }
134    let written = zl_decompress(&mut out[..decompressed_size], compressed)
135        .map_err(|e| anyhow!("OpenZL decompress: {}", e))?;
136    out.truncate(written);
137    Ok(written)
138}
139
140/// With the `openzl` feature OFF the codec is not linked. Prove that every entry
141/// point says so **out loud** — an `Err` naming the feature — instead of the two
142/// failure modes that would actually hurt: a silent `Ok` with empty/garbage bytes,
143/// or a panic. LAW 2: this is the red the default build can never show, so it is
144/// asserted from the build that can.
145#[cfg(all(test, not(feature = "openzl")))]
146mod no_codec_tests {
147    use super::*;
148
149    fn assert_names_the_feature(err: anyhow::Error) {
150        let msg = err.to_string();
151        assert!(
152            msg.contains("`openzl` feature"),
153            "error must name the feature that is missing, got: {msg}"
154        );
155    }
156
157    #[test]
158    fn compress_ctx_refuses_to_exist() {
159        assert_names_the_feature(
160            CompressCtx::new(3).err().expect("CompressCtx::new must fail with no codec linked"),
161        );
162    }
163
164    #[test]
165    fn decompress_refuses_and_does_not_touch_the_buffer() {
166        // A real OpenZL frame header; without the codec it must not be decoded,
167        // and `out` must be left exactly as the caller handed it over.
168        let frame = [0x5Bu8, 0x2A, 0x4D, 0x18, 0x00, 0x00, 0x00, 0x00];
169        let mut out = vec![0xAAu8; 4];
170        assert_names_the_feature(
171            decompress_into(&frame, &mut out).err().expect("decompress_into must fail"),
172        );
173        assert_eq!(out, vec![0xAAu8; 4], "buffer must be untouched on refusal");
174        assert_names_the_feature(
175            decompress_frame(&frame).err().expect("decompress_frame must fail"),
176        );
177    }
178}
179
180#[cfg(all(test, feature = "openzl"))]
181mod tests {
182    use super::*;
183
184    #[test]
185    fn test_roundtrip() {
186        let mut ctx = CompressCtx::new(3).unwrap();
187        let input = b"Hello world! This is a test of compression roundtrip. Repeated data helps compression. Repeated data helps compression. Repeated data helps compression.";
188        let compressed = ctx.compress(input).unwrap();
189        println!("Compressed {} -> {} bytes", input.len(), compressed.len());
190        let decompressed = decompress_frame(&compressed).unwrap();
191        assert_eq!(&decompressed[..], &input[..]);
192    }
193
194    #[test]
195    fn test_multi_compress_same_ctx() {
196        let mut ctx = CompressCtx::new(3).unwrap();
197        for i in 0..10 {
198            let input: Vec<u8> = (0..4096).map(|x| ((x + i) % 251) as u8).collect();
199            let compressed = ctx.compress(&input).unwrap();
200            let decompressed = decompress_frame(&compressed).unwrap();
201            assert_eq!(decompressed, input, "Failed at iteration {}", i);
202        }
203        println!("10 sequential compress calls OK");
204    }
205
206    #[test]
207    fn test_parallel_contexts() {
208        let handles: Vec<_> = (0..8).map(|t| {
209            std::thread::spawn(move || {
210                let mut ctx = CompressCtx::new(3).unwrap();
211                for i in 0..5 {
212                    let input: Vec<u8> = (0..8192).map(|x| ((x + i + t*100) % 251) as u8).collect();
213                    let compressed = ctx.compress(&input).unwrap();
214                    let decompressed = decompress_frame(&compressed).unwrap();
215                    assert_eq!(decompressed, input);
216                }
217            })
218        }).collect();
219        for h in handles {
220            h.join().unwrap();
221        }
222        println!("8 parallel contexts x 5 calls each OK");
223    }
224}
znippy_common/codec.rs

znippy_common/
codec.rs
