znippy-common 0.9.14

//! 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.3.0 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).";

// ─── 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(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"))?;
        let version = unsafe { ZL_getDefaultEncodingVersion() } as i32;
        // stickyParameters=1 keeps params across compress calls (reuse ctx)
        cctx.set_parameter(ZL_CParam_ZL_CParam_stickyParameters, 1)
            .map_err(|e| anyhow!(e))?;
        cctx.set_parameter(ZL_CParam_ZL_CParam_formatVersion, version)
            .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)
}

/// 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");
    }
}