test(perf): extend peak-RSS suite to backend-inclusive File read paths (LAB-350)

tests/performance/README.md

@@ -88,7 +88,19 @@ Tests per-request wrapper creation overhead:
 
 **Key Insight**: Per-request pattern adds negligible overhead (<0.15% of network latency).
 
-### 7. `stats_utils.py` - **Statistical Utilities**
+### 7. `test_large_object_memory.py` - **Memory Regression Guards** (CI-gated: `performance and slow`)
+
+Deterministic peak-memory bounds for large-object caching, at two scopes (#169):
+**serializer-only** (`test_store_path...` / `test_load_path...`, tracemalloc in isolation)
+and **backend-inclusive** (`test_file_backend_*`, the read path end-to-end through the
+File backend, tracemalloc + subprocess peak RSS). Bounds, measured multipliers, and the
+metric-choice rationale (VmHWM vs `ru_maxrss`, tracemalloc/RSS blind spots) live in the
+module and per-test docstrings — the single source of truth.
+
+**Key Insight**: The headline low-read-memory claim maps to the backend-inclusive numbers;
+serializer-only numbers can stay green while a backend read path regrows a full-payload copy.
+
+### 8. `stats_utils.py` - **Statistical Utilities**
 
 Provides rigorous performance measurement tools:
 

tests/performance/test_large_object_memory.py

@@ -1,23 +1,30 @@
 """Memory regression guards for large-object (Arrow/DataFrame) caching.
 
 These lock in the fixes that removed the base64+JSON wrapper inflation and the
-Arrow serializer's copy chain. They assert DETERMINISTIC, environment-independent
-metrics (Python-tracked peak via tracemalloc + on-wire size), not process RSS — so
-they are stable in CI yet fail loudly if the regressions return:
+Arrow serializer's copy chain, asserting DETERMINISTIC metrics only: tracemalloc
+peaks and on-wire size in-process, and — for allocations tracemalloc cannot see
+(Rust, Arrow pool, mmap residency) — peak RSS via VmHWM in dedicated subprocesses.
 
 - base64+JSON wrap drove store tracemalloc peak to ~5.7x logical and the wire to 1.33x.
 - the read path's base64-decode + JSON-parse + full-body slice drove read peak to ~5.4x.
 
 Pre-fix these assertions fail; post-fix store peak is ~2x, read ~1.1x, wire ~1x.
+
+Two measurement scopes live here — serializer-only (`test_store_path...` /
+`test_load_path...`) and backend-inclusive (`test_file_backend_*`, cachekit-py#169);
+see the "Backend-inclusive read paths" section banner below for the split and why
+the headline low-read-memory claim maps to the backend-inclusive numbers.
 """
 
 from __future__ import annotations
 
 import gc
+import os
 import subprocess
 import sys
 import textwrap
 import tracemalloc
+from pathlib import Path
 
 import numpy as np
 import pandas as pd
@@ -29,6 +36,27 @@
 
 _MB = 1024 * 1024
 
+# Measurement subprocesses finish in seconds; a wedged child must fail the test
+# (TimeoutExpired), not hang the CI job to its ceiling. Generous for slow runners.
+_SUBPROCESS_TIMEOUT_S = 300
+
+# Subprocess peak-RSS reads use VmHWM from /proc/self/status, NOT resource.ru_maxrss:
+# on Linux ru_maxrss lives in the signal struct and SURVIVES fork+exec, so a child
+# spawned from a fat pytest process inherits the parent's watermark — the child's
+# whole measurement then hides under the inherited peak and a real regression false-
+# passes (observed: base 1997 MiB, read cost 0.00x). VmHWM is per-mm and starts fresh
+# for the exec'd image. Linux-only, hence the skipif on the tests that embed this.
+_VMHWM_SNIPPET = textwrap.dedent(
+    """
+    def vmhwm_kib():
+        with open("/proc/self/status") as f:
+            for line in f:
+                if line.startswith("VmHWM"):
+                    return int(line.split()[1])
+        raise RuntimeError("VmHWM not found in /proc/self/status")
+    """
+)
+
 
 def _numeric_df(mb: int) -> pd.DataFrame:
     """Incompressible float64 frame ~mb MiB (worst case for compression)."""
@@ -102,36 +130,240 @@ def test_full_roundtrip_through_cache_handler_is_correct_and_compact():
 
 @pytest.mark.slow
 @pytest.mark.performance
+@pytest.mark.skipif(sys.platform != "linux", reason="peak-RSS via /proc/self/status VmHWM is Linux-only")
 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.
+    tracemalloc cannot see Rust allocations, so this invariant uses peak RSS
+    (VmHWM) 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(
+    script = _VMHWM_SNIPPET + 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
+        print(vmhwm_kib())
         """
     )
-    result = subprocess.run([sys.executable, "-c", script], capture_output=True, text=True)  # noqa: S603 (trusted: sys.executable + literal code)
+    result = subprocess.run([sys.executable, "-c", script], capture_output=True, text=True, timeout=_SUBPROCESS_TIMEOUT_S)  # 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)"
     )
+
+
+# ---------------------------------------------------------------------------
+# Backend-inclusive read paths (cachekit-py#169)
+#
+# Everything above measures the serializer layer with the read input already in
+# memory, so backend read-side copies (File os.read + the file_data[14:] slice)
+# are structurally invisible. These tests run the read END-TO-END through the
+# real stack (FileBackend -> StandardCacheHandler -> CacheOperationHandler ->
+# unwrap -> deserialize) and bound what the whole path allocates.
+# ---------------------------------------------------------------------------
+
+
+def _file_read_stack(cache_dir: Path, serializer_name: str):
+    """The real L2 read wiring the decorator uses, on a File backend.
+
+    encryption=False is the explicit opt-out: it keeps supports_mmap_read() True for
+    "arrow" regardless of any CACHEKIT_MASTER_KEY leaking in from the environment.
+    """
+    from cachekit.backends.file import FileBackend
+    from cachekit.backends.file.config import FileBackendConfig
+    from cachekit.cache_handler import CacheOperationHandler, CacheSerializationHandler, StandardCacheHandler
+    from cachekit.key_generator import CacheKeyGenerator
+
+    backend = FileBackend(FileBackendConfig(cache_dir=cache_dir, max_size_mb=1024, max_value_mb=400))
+    operation = CacheOperationHandler(
+        CacheSerializationHandler(serializer_name=serializer_name, encryption=False),
+        CacheKeyGenerator(),
+        cache_handler=StandardCacheHandler(backend),
+    )
+    return backend, operation
+
+
+@pytest.mark.slow
+@pytest.mark.performance
+@pytest.mark.skipif(os.name != "posix", reason="mmap read path is POSIX-only (#171); non-POSIX falls back to os.read")
+def test_file_backend_read_python_allocations_bounded(tmp_path: Path) -> None:
+    """END-TO-END DataFrame read through the File backend (mmap fast path, #171).
+
+    The whole pipeline — get_buffer mmap -> envelope unwrap -> Arrow deserialize ->
+    pandas — allocates ~0.0x logical on the Python heap: the mapped payload and the
+    Arrow-pool buffers are not Python allocations, and ``to_pandas(self_destruct,
+    split_blocks)`` hands back numpy views over pool buffers rather than copies, so
+    even the output df is invisible to tracemalloc. Any stage rematerializing the
+    payload as Python bytes trips the 0.5x bound (fault-injected: a bytes() coercion
+    of the zero-copy view measures ~1.0x; get_buffer regressing to os.read measures
+    ~1.9x). If a pyarrow upgrade makes to_pandas copy again, this fails LOUD at
+    ~1.0x — recalibrate consciously; the guard never silently widens.
+    """
+    df = _numeric_df(50)
+    logical = _logical(df)
+    backend, operation = _file_read_stack(tmp_path / "cache", "arrow")
+    key = "perf:file-read:arrow"
+    backend.set(key, operation.serialization_handler.serialize_data(df, cache_key=key))
+
+    # Guard the measurement's precondition explicitly: if mmap doesn't apply here,
+    # the test would silently measure the fallback path and fail confusingly.
+    assert operation.serialization_handler.supports_mmap_read()
+    probe = backend.get_buffer(key)
+    assert probe is not None, "get_buffer returned None; end-to-end read would fall back to os.read"
+    probe.close()
+
+    gc.collect()
+    tracemalloc.start()
+    hit = operation.get_cached_value(key)
+    peak = tracemalloc.get_traced_memory()[1]
+    tracemalloc.stop()
+
+    assert hit is not None, "end-to-end File read missed (errors read as miss — check logs)"
+    pd.testing.assert_frame_equal(hit[1], df)
+    assert peak / logical < 0.5, (
+        f"File-backend end-to-end read peak {peak / logical:.2f}x logical — a full-payload copy is "
+        f"back in the read path (expected ~0.0x zero-copy; ~1x = bytes() coercion, ~2x = os.read "
+        f"fallback, ~1x with neither = pyarrow to_pandas stopped returning views — recalibrate)"
+    )
+
+
+@pytest.mark.slow
+@pytest.mark.performance
+def test_file_backend_bytes_read_python_allocations_bounded(tmp_path: Path) -> None:
+    """END-TO-END read through FileBackend.get() (default serializer, no mmap).
+
+    This is the path most cached functions take (anything that isn't a plaintext
+    Arrow DataFrame). Measured cost today: ~5x payload on the Python heap —
+    FileBackend.get's two full-payload copies (os.read + the file_data[14:] slice,
+    the exact copies #169 calls out), StandardSerializer.deserialize's ``bytes(data)``
+    re-coercion of the envelope's zero-copy memoryview (Rust retrieve needs bytes),
+    the decompressed msgpack document, and the unpacked output. The bound pins that:
+    one MORE full-payload copy (~6x) fails. Tightening below 5x means fixing those
+    copies (separate ticket per #169 — this test is the measurement).
+    """
+    payload = np.random.default_rng(0).bytes(50 * _MB)  # incompressible: envelope ~= payload size
+    backend, operation = _file_read_stack(tmp_path / "cache", "default")
+    key = "perf:file-read:bytes"
+    backend.set(key, operation.serialization_handler.serialize_data(payload, cache_key=key))
+
+    gc.collect()
+    tracemalloc.start()
+    hit = operation.get_cached_value(key)
+    peak = tracemalloc.get_traced_memory()[1]
+    tracemalloc.stop()
+
+    assert hit is not None, "end-to-end File read missed (errors read as miss — check logs)"
+    assert hit[1] == payload
+    assert peak / len(payload) < 5.7, (
+        f"File-backend bytes read peak {peak / len(payload):.2f}x payload — an additional full-payload "
+        f"read-side copy crept in (known cost ~5x: os.read + slice + bytes() coercion + decode + output)"
+    )
+
+
+@pytest.mark.slow
+@pytest.mark.performance
+@pytest.mark.skipif(sys.platform != "linux", reason="mmap read path is POSIX-only (#171); VmHWM measurement is Linux-only")
+def test_file_backend_end_to_end_read_peak_rss_bounded(tmp_path: Path) -> None:
+    """Peak-RSS bound for a large DataFrame read end-to-end through the File backend.
+
+    tracemalloc cannot see mmap page residency, Arrow-pool buffers, or Rust-side
+    allocations, so the headline low-read-RSS claim gets a real RSS measurement in a
+    dedicated subprocess (same pattern as the ByteStorage store test above). The
+    entry is written by a separate subprocess so neither payload creation nor the
+    write path pollutes the read process's high-water mark, and this pytest process
+    never holds payload-scale memory (which would skew later RSS-delta tests).
+
+    Expected read cost above the post-import baseline: ~2x payload — the checksum
+    pass faults every mapped page in (1x, file-backed) and to_pandas materializes
+    the df (1x). Any ADDED full-payload buffer on that floor — an owned-bytes
+    coercion of the zero-copy view, an Arrow-pool or Rust-side copy (invisible to
+    tracemalloc) — measures ~3x and blows the 2.6x bound (verified by fault
+    injection). The one class this metric cannot see: mmap falling back to os.read
+    swaps file-backed pages for anonymous heap at the same ~2x total — that one is
+    caught by the tracemalloc test above (~0.0x -> ~1.9x). Payload is uncompressed
+    Arrow IPC of incompressible float64 (compression="none", the mmap-friendly
+    on-disk format), so on-disk size ~= logical size and the ratios are meaningful.
+    """
+    payload_mb = 256
+    cols = 20
+    rows = payload_mb * _MB // (8 * cols)
+    cache_dir = str(tmp_path / "cache")
+    key = "perf:file-read:rss"
+
+    backend_setup = textwrap.dedent(
+        f"""
+        from cachekit.backends.file import FileBackend
+        from cachekit.backends.file.config import FileBackendConfig
+
+        backend = FileBackend(FileBackendConfig(cache_dir={cache_dir!r}, max_size_mb=1024, max_value_mb=400))
+        """
+    )
+
+    write_script = backend_setup + textwrap.dedent(
+        f"""
+        import numpy as np
+        import pandas as pd
+
+        from cachekit.serializers.arrow_serializer import ArrowSerializer
+        from cachekit.serializers.wrapper import SerializationWrapper
+
+        rng = np.random.default_rng(0)
+        df = pd.DataFrame({{f"c{{i}}": rng.standard_normal({rows}) for i in range({cols})}})
+        data, meta = ArrowSerializer(compression="none").serialize(df)
+        backend.set({key!r}, SerializationWrapper.wrap(data, meta.to_dict(), "arrow"))
+        """
+    )
+
+    read_script = (
+        backend_setup
+        + _VMHWM_SNIPPET
+        + textwrap.dedent(
+            f"""
+        from cachekit.cache_handler import CacheOperationHandler, CacheSerializationHandler, StandardCacheHandler
+        from cachekit.key_generator import CacheKeyGenerator
+
+        operation = CacheOperationHandler(
+            CacheSerializationHandler(serializer_name="arrow", encryption=False),
+            CacheKeyGenerator(),
+            cache_handler=StandardCacheHandler(backend),
+        )
+        assert operation.serialization_handler.supports_mmap_read()
+        probe = backend.get_buffer({key!r})
+        assert probe is not None, "get_buffer returned None; read would fall back to os.read"
+        probe.close()
+
+        base = vmhwm_kib()
+        hit = operation.get_cached_value({key!r})
+        peak = vmhwm_kib()
+        assert hit is not None, "end-to-end File read missed"
+        assert hit[1].shape == ({rows}, {cols}), f"wrong shape: {{hit[1].shape}}"
+        print(base, peak)
+        """
+        )
+    )
+
+    write = subprocess.run([sys.executable, "-c", write_script], capture_output=True, text=True, timeout=_SUBPROCESS_TIMEOUT_S)  # noqa: S603 (trusted: sys.executable + literal code)
+    assert write.returncode == 0, f"write subprocess failed: {write.stderr}"
+    read = subprocess.run([sys.executable, "-c", read_script], capture_output=True, text=True, timeout=_SUBPROCESS_TIMEOUT_S)  # noqa: S603 (trusted: sys.executable + literal code)
+    assert read.returncode == 0, f"read subprocess failed: {read.stderr}"
+
+    base_kib, peak_kib = (int(v) for v in read.stdout.split())
+    read_cost = (peak_kib - base_kib) * 1024
+    payload_bytes = payload_mb * _MB
+    assert read_cost < payload_bytes * 2.6, (
+        f"end-to-end File read peak RSS {read_cost / payload_bytes:.2f}x payload above baseline "
+        f"({base_kib / 1024:.0f} MiB) — a full-payload read-side copy is back "
+        f"(expected ~2x: mapped pages + df; ~3x with a heap copy)"
+    )