znippy-common 0.9.18

//! 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(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)
}

/// 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[..]);
    }
}