Skip to main content

s4_codec/
lib.rs

1// PyO3 0.22 ? on PyResult triggers `useless_conversion` because `From<PyErr>
2// for PyErr` is identity. The clippy warning is intrinsic to the binding
3// idiom; suppress at file scope.
4#![allow(clippy::useless_conversion)]
5//! Python bindings for `s4-codec`.
6//!
7//! Exposes the CPU codecs (`CpuZstd`, `CpuGzip`) and a `gpu_available()`
8//! helper. GPU codec classes are intentionally NOT exposed in v1.0 — the
9//! `nvcomp-gpu` cargo feature on this crate forwards to the underlying
10//! `s4-codec` GPU paths for the server build, but the Python module's
11//! runtime classes remain CPU-only. See `crates/s4-codec-py/README.md`
12//! for the rationale; GPU-Python exposure is a v1.x roadmap candidate.
13//!
14//! # Async bridge
15//!
16//! `s4_codec_rs::Codec` is async. Python callers expect blocking calls. We
17//! resolve this by running each call on a process-wide multi-thread tokio
18//! runtime stashed in a `OnceLock`. `Python::detach` (pyo3 0.26+ name for allow_threads) releases the
19//! GIL across the await so other Python threads can progress while the
20//! blocking compression worker churns.
21
22use std::sync::{Arc, OnceLock};
23
24use bytes::{Bytes, BytesMut};
25use pyo3::create_exception;
26use pyo3::exceptions::{PyIOError, PyNotImplementedError, PyRuntimeError, PyValueError};
27use pyo3::prelude::*;
28use pyo3::types::{PyBytes, PyDict, PyList};
29use s4_codec_rs::index::{FrameIndex, IndexError};
30use s4_codec_rs::multipart::{write_frame, FrameError, FrameHeader, FrameIter};
31use s4_codec_rs::{cpu_gzip, cpu_zstd, cpu_zstd_dict, ChunkManifest, Codec, CodecError, CodecKind};
32use tokio::runtime::{Builder, Runtime};
33
34// v0.8.5 #85 M-5: surface CodecError variants as discriminable Python
35// exception classes so callers can `except S4CrcMismatchError:` instead of
36// string-matching on a flattened `PyValueError`. Hierarchy:
37//
38//   S4Error (base, ⊂ ValueError for backward-compat with code that catches
39//            ValueError from the previous flat mapping)
40//     ├─ S4CrcMismatchError              (CodecError::CrcMismatch)
41//     ├─ S4SizeMismatchError             (CodecError::SizeMismatch)
42//     ├─ S4CodecMismatchError            (CodecError::CodecMismatch)
43//     ├─ S4UnregisteredCodecError        (CodecError::UnregisteredCodec)
44//     ├─ S4ManifestSizeExceedsLimitError (CodecError::ManifestSizeExceedsLimit)
45//     └─ S4ManifestSizeMismatchError     (CodecError::ManifestSizeMismatch)
46//   S4BackendError (⊂ RuntimeError) — wraps anyhow / nvCOMP backend faults
47//   S4IoError      (⊂ IOError)      — wraps std::io::Error
48//
49// `Backend` and `Io` deliberately do NOT inherit S4Error: they map onto
50// stdlib semantics (RuntimeError / IOError) so frameworks already wired to
51// retry-on-IOError continue to do the right thing. `TruncatedStream` is rare
52// enough on the binding surface (server-side streaming) that we leave it on
53// the S4Error base rather than minting another class.
54create_exception!(s4_codec, S4Error, PyValueError);
55create_exception!(s4_codec, S4CrcMismatchError, S4Error);
56create_exception!(s4_codec, S4SizeMismatchError, S4Error);
57create_exception!(s4_codec, S4CodecMismatchError, S4Error);
58create_exception!(s4_codec, S4UnregisteredCodecError, S4Error);
59create_exception!(s4_codec, S4ManifestSizeExceedsLimitError, S4Error);
60create_exception!(s4_codec, S4ManifestSizeMismatchError, S4Error);
61create_exception!(s4_codec, S4BackendError, PyRuntimeError);
62create_exception!(s4_codec, S4IoError, PyIOError);
63// v1.1 s4fs: frame-parse / sidecar-decode failures. Both sit under the
64// S4Error base (⊂ ValueError) so existing `except S4Error:` handlers
65// catch them without changes. Messages carry the upstream Display text
66// (`FrameError` / `IndexError` are `#[non_exhaustive]`, so we map by
67// message rather than per-variant subclasses).
68create_exception!(s4_codec, S4FrameError, S4Error);
69create_exception!(s4_codec, S4IndexError, S4Error);
70
71fn runtime() -> &'static Runtime {
72    static RT: OnceLock<Runtime> = OnceLock::new();
73    RT.get_or_init(|| {
74        Builder::new_multi_thread()
75            .enable_all()
76            .thread_name("s4-codec-py")
77            .build()
78            .expect("failed to start tokio runtime for s4_codec python binding")
79    })
80}
81
82fn codec_err_to_py(e: CodecError) -> PyErr {
83    use s4_codec_rs::CodecError::*;
84    match e {
85        SizeMismatch { expected, got } => {
86            S4SizeMismatchError::new_err(format!("size mismatch: expected {expected}, got {got}"))
87        }
88        CrcMismatch { expected, got } => S4CrcMismatchError::new_err(format!(
89            "crc32c mismatch: expected {expected:#010x}, got {got:#010x}"
90        )),
91        CodecMismatch { expected, got } => S4CodecMismatchError::new_err(format!(
92            "codec mismatch: expected {expected:?}, got {got:?}"
93        )),
94        UnregisteredCodec(k) => {
95            S4UnregisteredCodecError::new_err(format!("codec {k:?} not registered"))
96        }
97        ManifestSizeExceedsLimit { requested, limit } => S4ManifestSizeExceedsLimitError::new_err(
98            format!("manifest claims {requested} bytes but limit is {limit}"),
99        ),
100        ManifestSizeMismatch { manifest, actual } => S4ManifestSizeMismatchError::new_err(format!(
101            "manifest claims {manifest} bytes but body is {actual}"
102        )),
103        Backend(msg) => S4BackendError::new_err(format!("backend: {msg}")),
104        Io(e) => S4IoError::new_err(format!("io: {e}")),
105        TruncatedStream { expected, got } => S4Error::new_err(format!(
106            "stream truncated: expected {expected} input bytes, got {got}"
107        )),
108        // v0.8.15 M-4: AWS S3 over-length analogue (declared
109        // Content-Length smaller than the wire body). Same shape as
110        // TruncatedStream — surface to Python callers via the
111        // generic `S4Error` since the in-process codec doesn't
112        // emit HTTP status codes.
113        OverlengthStream { expected, got } => S4Error::new_err(format!(
114            "stream over-length: expected {expected} input bytes, got at least {got}"
115        )),
116        // `Join` is a tokio internal that surfaces only if a blocking worker
117        // panics — surface as backend so retries hit the same class as
118        // anyhow-wrapped backend faults.
119        Join(e) => S4BackendError::new_err(format!("backend (worker join): {e}")),
120        // v1.0 F1: `CodecError` is `#[non_exhaustive]`, so newly-added
121        // variants in a future minor release must have a fallback here.
122        // Map to the generic `S4Error` carrying the upstream Display
123        // text — Python callers see the message but cannot pattern-
124        // match on the specific subclass until this wrapper is updated.
125        other => S4Error::new_err(format!("codec error: {other}")),
126    }
127}
128
129fn manifest_from_parts(
130    kind: CodecKind,
131    payload_len: u64,
132    original_size: u64,
133    crc32c: u32,
134) -> ChunkManifest {
135    ChunkManifest {
136        codec: kind,
137        original_size,
138        compressed_size: payload_len,
139        crc32c,
140    }
141}
142
143/// Run the supplied future on the shared multi-thread runtime, releasing
144/// the GIL while it runs so other Python threads aren't starved by a
145/// long-running compression worker.
146fn block_on<F, T>(py: Python<'_>, fut: F) -> T
147where
148    F: std::future::Future<Output = T> + Send,
149    T: Send,
150{
151    py.detach(|| runtime().block_on(fut))
152}
153
154/// CPU zstd codec. Level is clamped to 1..=22 by the underlying crate;
155/// default 3 matches `zstd(1)`'s out-of-the-box level.
156#[pyclass(name = "CpuZstd", module = "s4_codec")]
157struct PyCpuZstd {
158    inner: cpu_zstd::CpuZstd,
159}
160
161#[pymethods]
162impl PyCpuZstd {
163    #[new]
164    #[pyo3(signature = (level = 3))]
165    fn new(level: i32) -> Self {
166        Self {
167            inner: cpu_zstd::CpuZstd::new(level),
168        }
169    }
170
171    /// Compress `data`. Returns `(compressed: bytes, original_size: int, crc32c: int)`.
172    /// The original size and crc32c are the manifest fields decompress needs;
173    /// the caller is expected to round-trip them alongside the payload.
174    fn compress<'py>(
175        &self,
176        py: Python<'py>,
177        data: &Bound<'py, PyBytes>,
178    ) -> PyResult<(Bound<'py, PyBytes>, u64, u32)> {
179        let input = Bytes::copy_from_slice(data.as_bytes());
180        let codec = self.inner.clone();
181        let (out, manifest) =
182            block_on(py, async move { codec.compress(input).await }).map_err(codec_err_to_py)?;
183        Ok((
184            PyBytes::new(py, &out),
185            manifest.original_size,
186            manifest.crc32c,
187        ))
188    }
189
190    /// Decompress `data`. `original_size` and `crc32c` are the matching
191    /// manifest fields returned by `compress`.
192    fn decompress<'py>(
193        &self,
194        py: Python<'py>,
195        data: &Bound<'py, PyBytes>,
196        original_size: u64,
197        crc32c: u32,
198    ) -> PyResult<Bound<'py, PyBytes>> {
199        let input = Bytes::copy_from_slice(data.as_bytes());
200        let manifest = manifest_from_parts(
201            CodecKind::CpuZstd,
202            input.len() as u64,
203            original_size,
204            crc32c,
205        );
206        let codec = self.inner.clone();
207        let out = block_on(py, async move { codec.decompress(input, &manifest).await })
208            .map_err(codec_err_to_py)?;
209        Ok(PyBytes::new(py, &out))
210    }
211
212    fn __repr__(&self) -> String {
213        format!("CpuZstd(level={})", cpu_zstd::CpuZstd::DEFAULT_LEVEL)
214    }
215}
216
217/// CPU gzip codec (RFC 1952). Level 0..=9, default 6 (matches `gzip(1)`).
218/// Output is a real gzip stream — any standard `gunzip`-aware decoder
219/// (browser, curl, Python's `gzip` module) decodes the payload bytes.
220#[pyclass(name = "CpuGzip", module = "s4_codec")]
221struct PyCpuGzip {
222    inner: cpu_gzip::CpuGzip,
223}
224
225#[pymethods]
226impl PyCpuGzip {
227    #[new]
228    #[pyo3(signature = (level = 6))]
229    fn new(level: u32) -> Self {
230        Self {
231            inner: cpu_gzip::CpuGzip::new(level),
232        }
233    }
234
235    fn compress<'py>(
236        &self,
237        py: Python<'py>,
238        data: &Bound<'py, PyBytes>,
239    ) -> PyResult<(Bound<'py, PyBytes>, u64, u32)> {
240        let input = Bytes::copy_from_slice(data.as_bytes());
241        let codec = self.inner.clone();
242        let (out, manifest) =
243            block_on(py, async move { codec.compress(input).await }).map_err(codec_err_to_py)?;
244        Ok((
245            PyBytes::new(py, &out),
246            manifest.original_size,
247            manifest.crc32c,
248        ))
249    }
250
251    fn decompress<'py>(
252        &self,
253        py: Python<'py>,
254        data: &Bound<'py, PyBytes>,
255        original_size: u64,
256        crc32c: u32,
257    ) -> PyResult<Bound<'py, PyBytes>> {
258        let input = Bytes::copy_from_slice(data.as_bytes());
259        let manifest = manifest_from_parts(
260            CodecKind::CpuGzip,
261            input.len() as u64,
262            original_size,
263            crc32c,
264        );
265        let codec = self.inner.clone();
266        let out = block_on(py, async move { codec.decompress(input, &manifest).await })
267            .map_err(codec_err_to_py)?;
268        Ok(PyBytes::new(py, &out))
269    }
270
271    fn __repr__(&self) -> String {
272        format!("CpuGzip(level={})", cpu_gzip::CpuGzip::DEFAULT_LEVEL)
273    }
274}
275
276/// CPU zstd codec bound to a shared trained dictionary (`cpu-zstd-dict`,
277/// v1.1 `--zstd-dict`). The dictionary is a stock zstd dictionary
278/// (`zstd --train` / `ZDICT_trainFromBuffer` output); the compressed
279/// payload is a stock zstd frame referencing it. Level clamped to 1..=22.
280#[pyclass(name = "CpuZstdDict", module = "s4_codec")]
281struct PyCpuZstdDict {
282    inner: cpu_zstd_dict::CpuZstdDict,
283    level: i32,
284}
285
286#[pymethods]
287impl PyCpuZstdDict {
288    #[new]
289    #[pyo3(signature = (dict_bytes, level = 3))]
290    fn new(dict_bytes: &Bound<'_, PyBytes>, level: i32) -> PyResult<Self> {
291        let dict: Arc<[u8]> = Arc::from(dict_bytes.as_bytes().to_vec().into_boxed_slice());
292        let inner = cpu_zstd_dict::CpuZstdDict::new(dict, level).map_err(codec_err_to_py)?;
293        Ok(Self {
294            inner,
295            level: level.clamp(1, 22),
296        })
297    }
298
299    /// Compress `data` against the bound dictionary. Returns
300    /// `(compressed: bytes, original_size: int, crc32c: int)` — same shape
301    /// as `CpuZstd.compress`.
302    fn compress<'py>(
303        &self,
304        py: Python<'py>,
305        data: &Bound<'py, PyBytes>,
306    ) -> PyResult<(Bound<'py, PyBytes>, u64, u32)> {
307        let input = Bytes::copy_from_slice(data.as_bytes());
308        let codec = self.inner.clone();
309        let (out, manifest) =
310            block_on(py, async move { codec.compress(input).await }).map_err(codec_err_to_py)?;
311        Ok((
312            PyBytes::new(py, &out),
313            manifest.original_size,
314            manifest.crc32c,
315        ))
316    }
317
318    /// Decompress `data` against the bound dictionary. `original_size`
319    /// and `crc32c` are the matching manifest fields from `compress` (or
320    /// the S4F2 frame header for gateway-written objects).
321    fn decompress<'py>(
322        &self,
323        py: Python<'py>,
324        data: &Bound<'py, PyBytes>,
325        original_size: u64,
326        crc32c: u32,
327    ) -> PyResult<Bound<'py, PyBytes>> {
328        let input = Bytes::copy_from_slice(data.as_bytes());
329        let manifest = manifest_from_parts(
330            CodecKind::CpuZstdDict,
331            input.len() as u64,
332            original_size,
333            crc32c,
334        );
335        let codec = self.inner.clone();
336        let out = block_on(py, async move { codec.decompress(input, &manifest).await })
337            .map_err(codec_err_to_py)?;
338        Ok(PyBytes::new(py, &out))
339    }
340
341    fn __repr__(&self) -> String {
342        format!(
343            "CpuZstdDict(dict_len={}, level={})",
344            self.inner.dict().len(),
345            self.level
346        )
347    }
348}
349
350/// True iff the wheel was built with `--features nvcomp-gpu` AND a
351/// CUDA-capable GPU is reachable at runtime. Default wheels return False.
352#[pyfunction]
353fn gpu_available() -> bool {
354    s4_codec_rs::nvcomp::is_gpu_available()
355}
356
357fn frame_err_to_py(e: FrameError) -> PyErr {
358    S4FrameError::new_err(format!("frame error: {e}"))
359}
360
361fn index_err_to_py(e: IndexError) -> PyErr {
362    S4IndexError::new_err(format!("index error: {e}"))
363}
364
365/// S4F2 frame header → Python dict
366/// `{codec: str, original_size: int, compressed_size: int, crc32c: int}`.
367/// `codec` is the stable `CodecKind::as_str()` name (`"cpu-zstd"`, …).
368fn frame_header_dict<'py>(py: Python<'py>, h: &FrameHeader) -> PyResult<Bound<'py, PyDict>> {
369    let d = PyDict::new(py);
370    d.set_item("codec", h.codec.as_str())?;
371    d.set_item("original_size", h.original_size)?;
372    d.set_item("compressed_size", h.compressed_size)?;
373    d.set_item("crc32c", h.crc32c)?;
374    Ok(d)
375}
376
377/// Parse one S4F2 frame off the front of `data`.
378///
379/// Returns `(header: dict, payload: bytes, rest: bytes)` where `header`
380/// is `{codec, original_size, compressed_size, crc32c}`, `payload` is the
381/// (still-compressed) frame payload, and `rest` is everything after the
382/// frame. Raises `S4FrameError` on truncated / bad-magic / unknown-codec
383/// input. Thin wrapper over `s4_codec::multipart::read_frame` — the wire
384/// layout is the frozen v1.0 S4F2 format.
385#[pyfunction]
386fn read_frame<'py>(
387    py: Python<'py>,
388    data: &Bound<'py, PyBytes>,
389) -> PyResult<(Bound<'py, PyDict>, Bound<'py, PyBytes>, Bound<'py, PyBytes>)> {
390    let input = Bytes::copy_from_slice(data.as_bytes());
391    let (header, payload, rest) =
392        s4_codec_rs::multipart::read_frame(input).map_err(frame_err_to_py)?;
393    Ok((
394        frame_header_dict(py, &header)?,
395        PyBytes::new(py, &payload),
396        PyBytes::new(py, &rest),
397    ))
398}
399
400/// Parse `data` as a sequence of S4F2 frames (S4P1 padding frames are
401/// skipped, matching the gateway's GET path). Returns a list of
402/// `(header: dict, payload: bytes)` tuples. Raises `S4FrameError` on the
403/// first malformed frame. Thin wrapper over `s4_codec::multipart::FrameIter`.
404#[pyfunction]
405fn frame_iter<'py>(py: Python<'py>, data: &Bound<'py, PyBytes>) -> PyResult<Bound<'py, PyList>> {
406    let input = Bytes::copy_from_slice(data.as_bytes());
407    let out = PyList::empty(py);
408    for item in FrameIter::new(input) {
409        let (header, payload) = item.map_err(frame_err_to_py)?;
410        out.append((frame_header_dict(py, &header)?, PyBytes::new(py, &payload)))?;
411    }
412    Ok(out)
413}
414
415/// Decode a `<key>.s4index` sidecar (v1 / v2 / v3 layouts).
416///
417/// Returns a dict:
418///
419/// ```text
420/// {
421///   "total_padded_size": int,      # backend object size incl. padding
422///   "total_original_size": int,    # decompressed size (sum of entries)
423///   "source_etag": str | None,     # v2+: backend ETag binding
424///   "source_compressed_size": int | None,
425///   "entries": [ {original_offset, original_size,
426///                 compressed_offset, compressed_size}, ... ],
427///   "sse": None | {enc_chunk_size, enc_chunk_count, enc_key_id,
428///                  enc_salt: bytes, enc_plaintext_len, enc_header_bytes},
429/// }
430/// ```
431///
432/// `entries[i].compressed_size` includes the 28-byte frame header.
433/// Raises `S4IndexError` on malformed input. Thin wrapper over
434/// `s4_codec::index::decode_index`.
435#[pyfunction]
436fn decode_index<'py>(py: Python<'py>, data: &Bound<'py, PyBytes>) -> PyResult<Bound<'py, PyDict>> {
437    let input = Bytes::copy_from_slice(data.as_bytes());
438    let idx: FrameIndex = s4_codec_rs::index::decode_index(input).map_err(index_err_to_py)?;
439    let d = PyDict::new(py);
440    d.set_item("total_padded_size", idx.total_padded_size)?;
441    d.set_item("total_original_size", idx.total_original_size())?;
442    d.set_item("source_etag", idx.source_etag.as_deref())?;
443    d.set_item("source_compressed_size", idx.source_compressed_size)?;
444    let entries = PyList::empty(py);
445    for e in &idx.entries {
446        let ed = PyDict::new(py);
447        ed.set_item("original_offset", e.original_offset)?;
448        ed.set_item("original_size", e.original_size)?;
449        ed.set_item("compressed_offset", e.compressed_offset)?;
450        ed.set_item("compressed_size", e.compressed_size)?;
451        entries.append(ed)?;
452    }
453    d.set_item("entries", entries)?;
454    match idx.sse_v3 {
455        Some(sse) => {
456            let sd = PyDict::new(py);
457            sd.set_item("enc_chunk_size", sse.enc_chunk_size)?;
458            sd.set_item("enc_chunk_count", sse.enc_chunk_count)?;
459            sd.set_item("enc_key_id", sse.enc_key_id)?;
460            sd.set_item("enc_salt", PyBytes::new(py, &sse.enc_salt))?;
461            sd.set_item("enc_plaintext_len", sse.enc_plaintext_len)?;
462            sd.set_item("enc_header_bytes", sse.enc_header_bytes)?;
463            d.set_item("sse", sd)?;
464        }
465        None => d.set_item("sse", py.None())?,
466    }
467    Ok(d)
468}
469
470/// CRC32C (Castagnoli) of `data` — the checksum the S4F2 frame header
471/// carries. Exposed so pure-Python readers (s4fs) can verify
472/// `passthrough` frame payloads without an extra dependency.
473#[pyfunction]
474#[pyo3(name = "crc32c")]
475fn crc32c_py(data: &Bound<'_, PyBytes>) -> u32 {
476    ::crc32c::crc32c(data.as_bytes())
477}
478
479// ---------------------------------------------------------------------------
480// v1.2 s4fs write support: gateway-equivalent client-side object encoder.
481// ---------------------------------------------------------------------------
482
483/// Default S4F2 chunk size for client-side framed writes.
484///
485/// KEEP IN SYNC with `crates/s4-server/src/streaming.rs`
486/// (`DEFAULT_S4F2_CHUNK_SIZE` + `pick_chunk_size`). This crate cannot
487/// depend on `s4-server` (the server depends on this codec layer, and the
488/// Python wheel must not pull the full gateway), so the gateway's
489/// single-PUT chunk policy is replicated here verbatim. A drift between
490/// the two would NOT corrupt data (each S4F2 frame is self-describing and
491/// the GET path parses whatever frame sizes it finds) but it would change
492/// the Range-GET partial-fetch granularity vs gateway-written objects.
493const DEFAULT_S4F2_CHUNK_SIZE: usize = 4 * 1024 * 1024;
494
495/// Replicates `s4_server::streaming::pick_chunk_size(Some(len))` — the
496/// chunk size the gateway uses for a single PUT whose `Content-Length`
497/// is known (a client-side writer always knows the buffer length, so the
498/// `None`/chunked-transfer arm is intentionally not replicated).
499///
500/// KEEP IN SYNC with `crates/s4-server/src/streaming.rs::pick_chunk_size`:
501/// `<= 1 MiB → 1 MiB`, `<= 100 MiB → 4 MiB`, `> 100 MiB → 16 MiB`.
502fn pick_chunk_size_impl(content_length: u64) -> usize {
503    if content_length <= 1024 * 1024 {
504        1024 * 1024
505    } else if content_length <= 100 * 1024 * 1024 {
506        DEFAULT_S4F2_CHUNK_SIZE
507    } else {
508        16 * 1024 * 1024
509    }
510}
511
512/// Chunk size (bytes) the S4 gateway — and [`encode_s4_object`] — uses
513/// for an object of `content_length` bytes. Exposed so tests can pin the
514/// threshold table without allocating 100 MiB+ bodies.
515#[pyfunction]
516fn pick_chunk_size(content_length: u64) -> usize {
517    pick_chunk_size_impl(content_length)
518}
519
520/// Compress `input` into a gateway-identical S4F2 framed body with
521/// `cpu-zstd` frames. Mirrors the server's
522/// `streaming_compress_to_frames` output for a fully-buffered input:
523/// per-chunk frame header carries the *chunk* original size / crc32c,
524/// the aggregate manifest carries the rolling crc32c of the whole
525/// original body and the framed body length as `compressed_size`.
526fn encode_framed_zstd(
527    py: Python<'_>,
528    input: Bytes,
529    level: i32,
530) -> PyResult<(Bytes, ChunkManifest)> {
531    let chunk_size = pick_chunk_size_impl(input.len() as u64);
532    let codec = cpu_zstd::CpuZstd::new(level);
533    let res: Result<(Bytes, ChunkManifest), CodecError> = block_on(py, async move {
534        let mut framed = BytesMut::new();
535        let mut rolling_crc: u32 = 0;
536        let total = input.len() as u64;
537        let mut off = 0usize;
538        while off < input.len() {
539            let end = usize::min(off + chunk_size, input.len());
540            let chunk = input.slice(off..end);
541            let chunk_crc = ::crc32c::crc32c(&chunk);
542            rolling_crc = ::crc32c::crc32c_append(rolling_crc, &chunk);
543            let chunk_len = chunk.len() as u64;
544            let (compressed, _manifest) = codec.compress(chunk).await?;
545            let header = FrameHeader {
546                codec: CodecKind::CpuZstd,
547                original_size: chunk_len,
548                compressed_size: compressed.len() as u64,
549                crc32c: chunk_crc,
550            };
551            write_frame(&mut framed, header, &compressed);
552            off = end;
553        }
554        let body = framed.freeze();
555        let compressed_size = body.len() as u64;
556        Ok((
557            body,
558            ChunkManifest {
559                codec: CodecKind::CpuZstd,
560                original_size: total,
561                compressed_size,
562                crc32c: rolling_crc,
563            },
564        ))
565    });
566    res.map_err(codec_err_to_py)
567}
568
569/// Encode `data` exactly the way the S4 gateway's single-PUT path would
570/// store it, so a client (s4fs) can write S4-format objects **without**
571/// the gateway and the gateway / `s4 verify-sidecar` / s4fs itself can
572/// read them back.
573///
574/// Returns a dict:
575///
576/// ```text
577/// {
578///   "body":     bytes,         # what to PUT as the object body
579///   "sidecar":  bytes | None,  # `<key>.s4index` payload (multi-frame only,
580///                              # UNBOUND — pass through `bind_index` with the
581///                              # backend ETag/size after the body PUT)
582///   "metadata": {str: str},    # S3 user metadata to stamp on the body PUT
583/// }
584/// ```
585///
586/// Codec support: `"cpu-zstd"` (S4F2 framed, the gateway default) and
587/// `"passthrough"` (raw body, manifest metadata only — matching the
588/// gateway, which never frames passthrough). Anything else (cpu-gzip /
589/// cpu-zstd-dict / nvcomp-* / dietgpu-ans) raises `NotImplementedError`
590/// pointing at the gateway, which owns those codec paths.
591///
592/// Metadata keys/values restate the gateway's PUT-path stamps.
593/// KEEP IN SYNC with `crates/s4-server/src/service.rs` (`write_manifest`
594/// + the `META_*` constants and the `META_FRAMED` stamp) and
595/// `crates/s4-server/src/migrate.rs` (same five keys):
596///
597/// - `s4-codec`:           `CodecKind::as_str()` of the manifest codec
598/// - `s4-original-size`:   decimal length of the original body
599/// - `s4-compressed-size`: decimal length of the stored body (framed
600///   bytes incl. the 28-byte headers for cpu-zstd; == original for
601///   passthrough)
602/// - `s4-crc32c`:          decimal CRC32C of the **original** body
603/// - `s4-framed`:          `"true"` — framed bodies only (absent on
604///   passthrough, exactly like the gateway)
605#[pyfunction]
606#[pyo3(signature = (data, codec = "cpu-zstd", level = 3))]
607fn encode_s4_object<'py>(
608    py: Python<'py>,
609    data: &Bound<'py, PyBytes>,
610    codec: &str,
611    level: i32,
612) -> PyResult<Bound<'py, PyDict>> {
613    let input = Bytes::copy_from_slice(data.as_bytes());
614    let (body, manifest, framed) = match codec {
615        "cpu-zstd" => {
616            let (body, manifest) = encode_framed_zstd(py, input, level)?;
617            (body, manifest, true)
618        }
619        // Passthrough buys nothing from S4F2 wrapping; the gateway stores
620        // it as a raw blob with manifest metadata and NO `s4-framed` flag
621        // (see service.rs "Passthrough buys nothing…" comment). Mirror that.
622        "passthrough" => {
623            // CRC32C is the only real work on this path, but it still
624            // scans the whole body — release the GIL for it like the
625            // cpu-zstd branch does (audit v1.2-R1 P3).
626            let crc = py.detach(|| ::crc32c::crc32c(&input));
627            let len = input.len() as u64;
628            let manifest = ChunkManifest {
629                codec: CodecKind::Passthrough,
630                original_size: len,
631                compressed_size: len,
632                crc32c: crc,
633            };
634            (input, manifest, false)
635        }
636        other => {
637            return Err(PyNotImplementedError::new_err(format!(
638                "encode_s4_object supports codecs 'cpu-zstd' and 'passthrough' only; \
639                 {other:?} (cpu-gzip / cpu-zstd-dict / nvcomp-* / dietgpu-ans) must be \
640                 written through the S4 gateway, which owns those codec paths"
641            )));
642        }
643    };
644    // Sidecar policy mirrors the gateway PUT path (service.rs): emit only
645    // for framed bodies with MORE than one frame — a single-frame sidecar
646    // buys nothing (partial fetch == full body).
647    let sidecar = if framed {
648        let idx = s4_codec_rs::index::build_index_from_body(&body).map_err(frame_err_to_py)?;
649        if idx.entries.len() > 1 {
650            Some(s4_codec_rs::index::encode_index(&idx))
651        } else {
652            None
653        }
654    } else {
655        None
656    };
657    let out = PyDict::new(py);
658    out.set_item("body", PyBytes::new(py, &body))?;
659    match sidecar {
660        Some(s) => out.set_item("sidecar", PyBytes::new(py, &s))?,
661        None => out.set_item("sidecar", py.None())?,
662    }
663    let meta = PyDict::new(py);
664    // KEEP IN SYNC: literal metadata key names restate s4-server/src/
665    // service.rs META_CODEC / META_ORIGINAL_SIZE / META_COMPRESSED_SIZE /
666    // META_CRC32C / META_FRAMED (pub(crate) there, so not importable).
667    meta.set_item("s4-codec", manifest.codec.as_str())?;
668    meta.set_item("s4-original-size", manifest.original_size.to_string())?;
669    meta.set_item("s4-compressed-size", manifest.compressed_size.to_string())?;
670    meta.set_item("s4-crc32c", manifest.crc32c.to_string())?;
671    if framed {
672        meta.set_item("s4-framed", "true")?;
673    }
674    out.set_item("metadata", meta)?;
675    Ok(out)
676}
677
678/// Stamp the v2 source binding (`source_etag` + `source_compressed_size`)
679/// into a `.s4index` sidecar produced by [`encode_s4_object`].
680///
681/// The gateway stamps these AFTER its backend PUT returns the
682/// authoritative ETag (service.rs `put_object` → `write_sidecar`); a
683/// client-side writer must do the same: PUT the body, read the backend
684/// ETag, then `bind_index` and PUT the sidecar. Without the binding the
685/// sidecar decodes as "legacy v1 / unbound" and neither the gateway's
686/// Range-GET fast-path nor s4fs's partial fetch will trust it
687/// (`s4 verify-sidecar` reports `LegacyV1` instead of `OK`).
688///
689/// `source_etag` must be the **quote-stripped** entity-tag (the form s3s
690/// `ETag::value()` yields and the gateway writes; see
691/// `s4-server/src/repair.rs::normalize_etag`). Pass `None` on
692/// ETag-less backends — the size-only binding is still a valid v2
693/// sidecar the verifier accepts.
694#[pyfunction]
695#[pyo3(signature = (sidecar, source_compressed_size, source_etag = None))]
696fn bind_index<'py>(
697    py: Python<'py>,
698    sidecar: &Bound<'py, PyBytes>,
699    source_compressed_size: u64,
700    source_etag: Option<String>,
701) -> PyResult<Bound<'py, PyBytes>> {
702    let raw = Bytes::copy_from_slice(sidecar.as_bytes());
703    let mut idx: FrameIndex = s4_codec_rs::index::decode_index(raw).map_err(index_err_to_py)?;
704    idx.source_compressed_size = Some(source_compressed_size);
705    idx.source_etag = source_etag;
706    Ok(PyBytes::new(py, &s4_codec_rs::index::encode_index(&idx)))
707}
708
709#[pymodule]
710fn s4_codec(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
711    m.add_class::<PyCpuZstd>()?;
712    m.add_class::<PyCpuGzip>()?;
713    m.add_class::<PyCpuZstdDict>()?;
714    m.add_function(wrap_pyfunction!(gpu_available, m)?)?;
715    // v1.1 s4fs: read-side wire-format helpers (S4F2 frames + .s4index
716    // sidecars) so pure-Python clients can decode gateway-written
717    // objects without the gateway.
718    m.add_function(wrap_pyfunction!(read_frame, m)?)?;
719    m.add_function(wrap_pyfunction!(frame_iter, m)?)?;
720    m.add_function(wrap_pyfunction!(decode_index, m)?)?;
721    m.add_function(wrap_pyfunction!(crc32c_py, m)?)?;
722    // v1.2 s4fs write support: gateway-equivalent client-side encoder +
723    // sidecar version binding + the gateway's chunk-size policy table.
724    m.add_function(wrap_pyfunction!(encode_s4_object, m)?)?;
725    m.add_function(wrap_pyfunction!(bind_index, m)?)?;
726    m.add_function(wrap_pyfunction!(pick_chunk_size, m)?)?;
727    m.add(
728        "FRAME_MAGIC",
729        PyBytes::new(py, s4_codec_rs::multipart::FRAME_MAGIC),
730    )?;
731    m.add(
732        "PADDING_MAGIC",
733        PyBytes::new(py, s4_codec_rs::multipart::PADDING_MAGIC),
734    )?;
735    m.add(
736        "FRAME_HEADER_BYTES",
737        s4_codec_rs::multipart::FRAME_HEADER_BYTES,
738    )?;
739    m.add("SIDECAR_SUFFIX", s4_codec_rs::index::SIDECAR_SUFFIX)?;
740    m.add("__version__", env!("CARGO_PKG_VERSION"))?;
741    // v0.8.5 #85 M-5: export per-CodecError exception classes so Python
742    // callers can branch on error kind. See module-level doc comments above
743    // `create_exception!` for the inheritance hierarchy.
744    m.add("S4Error", py.get_type::<S4Error>())?;
745    m.add("S4CrcMismatchError", py.get_type::<S4CrcMismatchError>())?;
746    m.add("S4SizeMismatchError", py.get_type::<S4SizeMismatchError>())?;
747    m.add(
748        "S4CodecMismatchError",
749        py.get_type::<S4CodecMismatchError>(),
750    )?;
751    m.add(
752        "S4UnregisteredCodecError",
753        py.get_type::<S4UnregisteredCodecError>(),
754    )?;
755    m.add(
756        "S4ManifestSizeExceedsLimitError",
757        py.get_type::<S4ManifestSizeExceedsLimitError>(),
758    )?;
759    m.add(
760        "S4ManifestSizeMismatchError",
761        py.get_type::<S4ManifestSizeMismatchError>(),
762    )?;
763    m.add("S4BackendError", py.get_type::<S4BackendError>())?;
764    m.add("S4IoError", py.get_type::<S4IoError>())?;
765    m.add("S4FrameError", py.get_type::<S4FrameError>())?;
766    m.add("S4IndexError", py.get_type::<S4IndexError>())?;
767    Ok(())
768}