LAB-347: perf: release the GIL during ByteStorage compress/hash

rust/src/python_bindings.rs

@@ -39,9 +39,11 @@ impl PyByteStorage {
     /// Returns:
     ///     Bytes: Serialized StorageEnvelope
     pub fn store(&self, py: Python, data: &[u8], format: Option<String>) -> PyResult<Py<PyBytes>> {
-        let envelope_bytes = self
-            .inner
-            .store(data, format)
+        // Detach from the GIL: LZ4 + xxh3 on a large payload otherwise blocks every
+        // Python thread for the full compression duration (cachekit-core#45).
+        // Sound: `data` borrows an immutable `bytes` buffer kept alive by this call.
+        let envelope_bytes = py
+            .detach(|| self.inner.store(data, format))
             .map_err(|e| PyValueError::new_err(format!("Storage failed: {}", e)))?;
 
         Ok(PyBytes::new(py, &envelope_bytes).into())
@@ -54,22 +56,23 @@ impl PyByteStorage {
     ///
     /// Returns:
     ///     Tuple[bytes, str]: (original_data, format_identifier)
-    pub fn retrieve(&self, envelope_bytes: &[u8]) -> PyResult<(Vec<u8>, String)> {
-        self.inner
-            .retrieve(envelope_bytes)
+    pub fn retrieve(&self, py: Python, envelope_bytes: &[u8]) -> PyResult<(Vec<u8>, String)> {
+        // Detach from the GIL for decompression + checksum (see store()).
+        py.detach(|| self.inner.retrieve(envelope_bytes))
             .map_err(|e| PyValueError::new_err(format!("Retrieval failed: {}", e)))
     }
 
     /// Get compression ratio for given data
-    pub fn estimate_compression(&self, data: &[u8]) -> PyResult<f64> {
-        self.inner
-            .estimate_compression(data)
+    pub fn estimate_compression(&self, py: Python, data: &[u8]) -> PyResult<f64> {
+        // Full-payload LZ4 pass — same GIL-blocking profile as store().
+        py.detach(|| self.inner.estimate_compression(data))
             .map_err(|e| PyValueError::new_err(format!("Compression estimation failed: {}", e)))
     }
 
     /// Validate envelope without extracting data
-    pub fn validate(&self, envelope_bytes: &[u8]) -> PyResult<bool> {
-        Ok(self.inner.validate(envelope_bytes))
+    pub fn validate(&self, py: Python, envelope_bytes: &[u8]) -> PyResult<bool> {
+        // Full decompression + checksum under the hood — same GIL-blocking profile.
+        Ok(py.detach(|| self.inner.validate(envelope_bytes)))
     }
 
     /// Get security limits for clients

tests/critical/conftest.py

@@ -13,6 +13,7 @@ def pytest_runtest_setup(item):
         or "secure_env_fallback" in item.nodeid
         or "local_cache_works" in item.nodeid
         or "backpressure_load_control" in item.nodeid
+        or "byte_storage_gil" in item.nodeid
     )
     if skip_redis:
         # Remove autouse redis fixtures for tests that don't need Redis

tests/critical/test_byte_storage_gil.py

@@ -0,0 +1,113 @@
+"""ByteStorage store()/retrieve() must release the GIL (cachekit-core#45).
+
+The Rust FFI detaches from the GIL (PyO3 ``Python::detach``) for the whole
+compress/hash/serialize core, so a large store() no longer freezes every other
+Python thread for the full compression duration.
+
+Proof technique: a ticker thread timestamps every ~1ms. If the GIL were held
+for the whole FFI call, the ticker could only run at the call's bytecode
+boundaries — never in the *interior* of the call window. We therefore assert
+ticker stamps strictly inside the middle 50% of the call window; the 25%
+margins on both sides dwarf any boundary-slice scheduling (GIL switch interval
+is ~5ms, margins are >=25ms). Deterministic in both directions: GIL held ->
+zero interior stamps possible; GIL released -> hundreds.
+"""
+
+from __future__ import annotations
+
+import os
+import threading
+import time
+
+import pytest
+
+from cachekit._rust_serializer import ByteStorage
+
+# Incompressible payload, large enough that store/retrieve take >=100ms even on
+# fast hardware (msgpack envelope encoding dominates, ~150MB/s). 64MB keeps the
+# critical suite fast and peak test memory under ~300MB.
+_PAYLOAD_BYTES = 64 * 1024 * 1024
+_MIN_CALL_SECONDS = 0.05  # below this the interior-window proof loses its margin
+
+
+@pytest.fixture(scope="module")
+def payload() -> bytes:
+    return os.urandom(_PAYLOAD_BYTES)
+
+
+def _stamps_during(call) -> tuple[list[float], float, float]:
+    """Run *call* while a ticker thread timestamps; return (stamps, t0, t1)."""
+    stamps: list[float] = []
+    stop = threading.Event()
+
+    def ticker() -> None:
+        while not stop.is_set():
+            stamps.append(time.monotonic())
+            time.sleep(0.001)
+
+    thread = threading.Thread(target=ticker, daemon=True)
+    thread.start()
+    time.sleep(0.02)  # let the ticker reach steady state
+    t0 = time.monotonic()
+    call()
+    t1 = time.monotonic()
+    stop.set()
+    thread.join(timeout=5)
+    return stamps, t0, t1
+
+
+def _assert_gil_released(stamps: list[float], t0: float, t1: float, op: str) -> None:
+    duration = t1 - t0
+    assert duration >= _MIN_CALL_SECONDS, (
+        f"{op} finished in {duration * 1000:.0f}ms — too fast for the interior-window "
+        f"proof; bump _PAYLOAD_BYTES so the call takes >={_MIN_CALL_SECONDS * 1000:.0f}ms"
+    )
+    lo = t0 + duration * 0.25
+    hi = t1 - duration * 0.25
+    interior = [s for s in stamps if lo < s < hi]
+    assert len(interior) >= 3, (
+        f"{op} held the GIL: ticker made no progress inside the middle 50% of a "
+        f"{duration * 1000:.0f}ms call window ({len(interior)} interior stamps)"
+    )
+
+
+def test_store_releases_gil(payload: bytes) -> None:
+    storage = ByteStorage(None)
+    stamps, t0, t1 = _stamps_during(lambda: storage.store(payload, None))
+    _assert_gil_released(stamps, t0, t1, "store()")
+
+
+def test_retrieve_releases_gil(payload: bytes) -> None:
+    storage = ByteStorage(None)
+    envelope = storage.store(payload, None)
+    stamps, t0, t1 = _stamps_during(lambda: storage.retrieve(envelope))
+    _assert_gil_released(stamps, t0, t1, "retrieve()")
+
+
+def test_roundtrip_unchanged_by_gil_release(payload: bytes) -> None:
+    """GIL release is a threading change only — bytes must round-trip identically."""
+    storage = ByteStorage(None)
+    data, fmt = storage.retrieve(storage.store(payload, None))
+    assert data == payload
+    assert fmt == "msgpack"
+
+
+def test_concurrent_stores_are_correct(payload: bytes) -> None:
+    """Two threads storing through one ByteStorage while detached from the GIL
+    must not corrupt each other (inner metrics state is Mutex-guarded)."""
+    storage = ByteStorage(None)
+    chunks = [payload[: 8 * 1024 * 1024], payload[8 * 1024 * 1024 : 16 * 1024 * 1024]]
+    results: dict[int, bytes] = {}
+
+    def worker(idx: int) -> None:
+        envelope = storage.store(chunks[idx], None)
+        data, _ = storage.retrieve(envelope)
+        results[idx] = data
+
+    threads = [threading.Thread(target=worker, args=(i,)) for i in range(2)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(timeout=60)
+    assert results[0] == chunks[0]
+    assert results[1] == chunks[1]

tests/performance/test_large_object_memory.py

@@ -14,6 +14,9 @@
 from __future__ import annotations
 
 import gc
+import subprocess
+import sys
+import textwrap
 import tracemalloc
 
 import numpy as np
@@ -95,3 +98,40 @@ def test_full_roundtrip_through_cache_handler_is_correct_and_compact():
     assert len(blob) / _logical(df) < 1.1
     out = handler.deserialize_data(blob, cache_key="k")
     pd.testing.assert_frame_equal(out, df)
+
+
+@pytest.mark.slow
+@pytest.mark.performance
+def test_byte_storage_store_has_no_full_payload_copy():
+    """Rust-side allocation guard for ByteStorage.store() (cachekit-core#45).
+
+    tracemalloc cannot see Rust allocations, so this invariant uses peak RSS in
+    a dedicated subprocess. Determinism comes from the payload: 512MB of a
+    repeating 8-byte pattern LZ4-compresses to ~2MB, so every Rust-side buffer
+    downstream of the input (compressed data, msgpack envelope, returned bytes)
+    is negligible and peak RSS ~= interpreter + payload (~1.1x). The eliminated
+    ``data.to_vec()`` full-payload copy (cachekit-core < 0.3.0) re-adds ~1.0x
+    payload and trips the 1.7x bound with margin on both sides.
+    """
+    payload_mb = 512
+    script = textwrap.dedent(
+        f"""
+        import resource
+
+        from cachekit._rust_serializer import ByteStorage
+
+        payload = b"cachekit" * ({payload_mb} * 1024 * 1024 // 8)
+        envelope = ByteStorage(None).store(payload, None)
+        # Sanity: compressible payload => tiny envelope, or the RSS bound is meaningless.
+        assert len(envelope) < 32 * 1024 * 1024, f"envelope unexpectedly large: {{len(envelope)}}"
+        print(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss)  # KiB on Linux
+        """
+    )
+    result = subprocess.run([sys.executable, "-c", script], capture_output=True, text=True)  # noqa: S603 (trusted: sys.executable + literal code)
+    assert result.returncode == 0, f"store subprocess failed: {result.stderr}"
+    peak = int(result.stdout.strip()) * 1024
+    payload_bytes = payload_mb * 1024 * 1024
+    assert peak < payload_bytes * 1.7, (
+        f"store() peak RSS {peak / payload_bytes:.2f}x payload — a full-payload copy is back "
+        f"on the write path (expected ~1.1x without the to_vec copy, ~2.1x with it)"
+    )