perf(arrow): stream serialize-to-backend writes via BufferWritableBackend (LAB-766)

.secrets.baseline

@@ -213,7 +213,7 @@
         "filename": "src/cachekit/cache_handler.py",
         "hashed_secret": "5baa61e4c9b93f3f0682250b6cf8331b7ee68fd8",
         "is_verified": false,
-        "line_number": 411
+        "line_number": 427
       }
     ],
     "src/cachekit/config/decorator.py": [
@@ -425,5 +425,5 @@
       }
     ]
   },
-  "generated_at": "2026-07-23T09:21:58Z"
+  "generated_at": "2026-07-25T12:35:03Z"
 }

docs/backends/file.md

@@ -85,6 +85,27 @@ backend = FileBackend(config)
 - Cross-process: No (single-process only)
 - Platform support: Full on Linux/macOS, limited on Windows (no O_NOFOLLOW)
 
+## Bounded-Memory Large Values (Arrow)
+
+FileBackend is the only backend that implements **both** large-value capability protocols,
+making it the right choice for caching DataFrames bigger than your RAM headroom:
+
+- **Streaming writes** (`BufferWritableBackend.set_streaming`): plaintext Arrow values are
+  serialized record batch by record batch straight into the cache file (temp file + atomic
+  rename), so the full serialized payload never exists in memory. Write peak RSS is ~2.3x the
+  DataFrame's logical size, versus ~5.6x on the buffered path every other backend uses.
+- **Zero-copy reads** (`BufferReadableBackend.get_buffer`): with `compression="none"`, cached
+  entries are served back through a POSIX mmap without materializing the payload on the heap.
+
+Both engage automatically — no API change — whenever the serializer is plaintext Arrow
+(`serializer_name="arrow"`, no encryption). **Encrypted values are excluded from both paths**:
+AES-256-GCM needs the whole ciphertext to produce/verify its auth tag, so the secure path
+stays on buffered `set()`/`get()`. Streamed values also intentionally skip the L1 in-memory
+cache — holding a multi-GB envelope in-process would defeat the point.
+
+Other backends (Redis, CachekitIO, Memcached) fall back transparently to buffered
+`set(bytes)` with no behaviour change.
+
 ## TTL Inspection & Sliding Expiration
 
 FileBackend implements the `TTLInspectableBackend` protocol (`get_ttl` / `refresh_ttl`) by

docs/configuration.md

@@ -587,7 +587,7 @@ export CACHEKIT_ARROW_COMPRESSION=zstd
 ```
 
 > [!TIP]
-> Compression saves network bandwidth for large values. `none` can enable zero-copy mmap reads on the File backend — eligibility also requires a plaintext (unencrypted) Arrow payload read back as pandas (`return_format="pandas"`) and a backend that supports buffer reads.
+> Compression saves network bandwidth for large values. `none` can enable zero-copy mmap reads on the File backend — eligibility also requires a plaintext (unencrypted) Arrow payload read back as pandas (`return_format="pandas"`) and a backend that supports buffer reads. Writes are independent of this setting: plaintext Arrow values stream to the File backend record batch by record batch (compressed or not), never materializing the full payload in memory; encrypted values and other backends use the buffered write path.
 
 ### Connection Pooling
 

docs/serializers/arrow.md

@@ -158,6 +158,13 @@ ArrowSerializer uses memory-mapped deserialization, which means:
 - Default deserialization: +15 MB memory allocation
 - Arrow deserialization: +2 MB memory allocation
 
+**Writes stream on the File backend.** When the cache backend supports streaming writes
+(File backend only today) and the value is plaintext (no encryption), serialization streams
+~8 MiB record batches directly into the cache file instead of materializing the whole Arrow
+IPC payload in memory first — cutting write peak RSS from ~5.6x to ~2.3x the DataFrame's
+logical size. Wire backends (Redis, CachekitIO, Memcached) and encrypted values use the
+buffered path unchanged. See [File Backend](../backends/file.md#bounded-memory-large-values-arrow).
+
 ## Polars Support
 
 Polars DataFrames are supported via the `__arrow_c_stream__` interface (Arrow C Data Interface). This means zero-copy interchange between polars and Arrow — no intermediate conversion.

src/cachekit/backends/base.py

@@ -4,13 +4,14 @@
 All L2 backends (Redis, HTTP, etc.) must implement BaseBackend protocol.
 
 Optional capability protocols (TTLInspectableBackend, LockableBackend,
-TimeoutConfigurableBackend) enable advanced features with graceful degradation.
+TimeoutConfigurableBackend, BufferReadableBackend, BufferWritableBackend)
+enable advanced features with graceful degradation.
 """
 
 from __future__ import annotations
 
-from collections.abc import AsyncIterator
-from typing import Any, Optional, Protocol, runtime_checkable
+from collections.abc import AsyncIterator, Callable
+from typing import Any, BinaryIO, Optional, Protocol, runtime_checkable
 
 # Re-export BackendError for convenience (public API)
 from cachekit.backends.errors import BackendError  # noqa: F401
@@ -212,6 +213,53 @@ def get_buffer(self, key: str) -> Optional[BufferHandle]:
         ...
 
 
+@runtime_checkable
+class BufferWritableBackend(Protocol):
+    """Optional protocol for backends that can accept a value written incrementally through a
+    seekable sink, instead of one materialized ``bytes`` blob (LAB-766, write-side twin of
+    ``BufferReadableBackend``).
+
+    Lets large values (e.g. multi-GB Arrow IPC) reach storage without the whole serialized
+    payload ever existing in memory. Only the File backend implements this today; backends that
+    don't are simply written via ``set`` as usual (the caller falls back transparently).
+
+    Why a writer callback and not an ``Iterable[bytes]``: the Arrow envelope stores an
+    xxHash3-64 checksum BEFORE the payload it covers, so the producer must patch bytes it has
+    already emitted once the stream is complete. A forward-only chunk iterator cannot express
+    that without serializing twice; a seekable sink can. This deliberately scopes the protocol
+    to random-access backends (File) — a forward-only transport (chunked HTTP upload) needs a
+    different contract and is out of scope until a backend demands it.
+
+    Contract:
+    - ``write_payload`` receives a writable, seekable binary sink positioned at the START of
+      the payload region. Producers MUST anchor seeks with ``tell()`` (never absolute offset 0):
+      backends may keep private framing before the payload.
+    - The value becomes visible under ``key`` only if ``write_payload`` returns successfully;
+      on any exception the backend discards the partial write, leaves any previous value for
+      the key intact, and re-raises the producer's exception unwrapped (so serialization errors
+      keep their type for the caller's error handling).
+    - ``ttl`` semantics are identical to ``set``.
+    - A successful ``set_streaming(key, w, ttl)`` is observably identical to
+      ``set(key, full_bytes, ttl)`` for the same payload bytes — same readback via
+      ``get``/``get_buffer``, same expiry, same size-limit enforcement (violations raise
+      ``BackendError`` and nothing is stored).
+    """
+
+    def set_streaming(self, key: str, write_payload: Callable[[BinaryIO], None], ttl: Optional[int] = None) -> None:
+        """Store the payload produced by ``write_payload`` under ``key``, never holding it whole.
+
+        Args:
+            key: Cache key to store
+            write_payload: Callable that writes the complete payload to the provided sink
+            ttl: Time-to-live in seconds (None = no expiry), same semantics as ``set``
+
+        Raises:
+            BackendError: If the backend operation fails (I/O, size limits, TTL bounds)
+            Exception: Whatever ``write_payload`` raises, unwrapped (partial write discarded)
+        """
+        ...
+
+
 @runtime_checkable
 class LockableBackend(Protocol):
     """Optional protocol for backends supporting distributed locking.

src/cachekit/backends/file/backend.py

@@ -20,8 +20,9 @@
 import struct
 import threading
 import time
+from collections.abc import Callable
 from pathlib import Path
-from typing import Any
+from typing import Any, BinaryIO
 
 from cachekit.backends.errors import BackendError, BackendErrorType
 
@@ -330,45 +331,16 @@ def set(self, key: str, value: bytes, ttl: int | None = None) -> None:
 
         file_path = self._key_to_path(key)
 
-        # Calculate expiry timestamp (0 = never expire)
-        if ttl is None or ttl == 0:
-            expiry_timestamp = 0
-        else:
-            # Validate TTL bounds (security: prevent integer overflow/underflow)
-            if ttl < 0 or ttl > MAX_TTL_SECONDS:
-                raise BackendError(
-                    f"TTL {ttl} out of range [0, {MAX_TTL_SECONDS}] (max 10 years)",
-                    BackendErrorType.PERMANENT,
-                )
-            expiry_timestamp = int(time.time() + ttl)
-
-        # Build header (14 bytes)
-        header = (
-            MAGIC  # [0:2] Magic bytes
-            + bytes([FORMAT_VERSION])  # [2:3] Version
-            + bytes([RESERVED])  # [3:4] Reserved
-            + struct.pack(">H", 0)  # [4:6] Flags (no compression/encryption yet)
-            + struct.pack(">Q", expiry_timestamp)  # [6:14] Expiry timestamp
-        )
-
-        # Combine header + payload
-        file_data = header + value
+        # Combine header (14 bytes) + payload
+        file_data = self._build_header(self._expiry_from_ttl(ttl)) + value
 
         # Generate temp file name
         temp_path = self._generate_temp_path(file_path)
 
         with self._lock:
             try:
                 # Check entry count BEFORE write (security: prevent file persisting on error)
-                # Allow overwrites (existing key doesn't increase count)
-                if self.config.max_entry_count > 0:
-                    _, entry_count = self._calculate_cache_size()
-                    # Only check if this is a NEW entry (not overwriting existing)
-                    if not os.path.exists(file_path) and entry_count >= self.config.max_entry_count:
-                        raise BackendError(
-                            f"Entry count {entry_count} would exceed max_entry_count ({self.config.max_entry_count})",
-                            BackendErrorType.PERMANENT,
-                        )
+                self._check_entry_capacity(file_path)
 
                 # Write to temp file with O_NOFOLLOW for security
                 fd = os.open(
@@ -410,6 +382,105 @@ def set(self, key: str, value: bytes, ttl: int | None = None) -> None:
                     key=key,
                 ) from exc
 
+    def set_streaming(self, key: str, write_payload: Callable[[BinaryIO], None], ttl: int | None = None) -> None:
+        """Store a streamed value with the same atomicity, locking, and limits as ``set`` (LAB-766).
+
+        ``write_payload`` receives a writable, seekable buffered binary file positioned at the
+        start of the payload region (just past this backend's 14-byte header) and must write the
+        complete payload — record batch by record batch, so the full value never exists in
+        memory. The temp-file + atomic-rename pattern from ``set`` is preserved: a failed stream
+        never leaves a partial value visible under ``key``, and any previous value for the key
+        survives. Producer exceptions propagate unwrapped (after the temp file is discarded) so
+        serialization errors keep their type for the caller.
+
+        ``max_value_mb`` is enforced AFTER the stream completes (``fstat`` on the temp file,
+        before the rename): a streaming write cannot know its size upfront. The caller-side byte
+        budget (CACHEKIT_MAX_VALUE_SIZE, enforced incrementally by the serializer) bounds the
+        transient disk usage of an over-limit attempt.
+
+        Args:
+            key: Cache key to store
+            write_payload: Callable that writes the complete payload to the provided file
+            ttl: Time-to-live in seconds (None or 0 = never expire), same semantics as ``set``
+
+        Raises:
+            BackendError: If the write fails (disk full, permissions, size/TTL limits, etc.)
+        """
+        header = self._build_header(self._expiry_from_ttl(ttl))
+        file_path = self._key_to_path(key)
+        temp_path = self._generate_temp_path(file_path)
+
+        committed = False
+        try:
+            # Check entry count BEFORE write (security: prevent file persisting on error)
+            with self._lock:
+                self._check_entry_capacity(file_path)
+
+            # Stream OUTSIDE self._lock: unlike set()'s bounded os.write, the producer runs
+            # the whole serialization here (minutes for multi-GB frames), and holding the
+            # process-wide lock would block every other FileBackend op for that window. The
+            # in-process lock isn't what makes this safe anyway — the temp path is unique
+            # per pid+ns, the flock gives cross-process exclusion, and the rename commit
+            # below is atomic. Write to temp file with O_NOFOLLOW for security.
+            fd = os.open(
+                temp_path,
+                os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
+                self.config.permissions,
+            )
+            f: BinaryIO | None = None
+            try:
+                # Acquire exclusive write lock
+                self._acquire_file_lock(fd, exclusive=True)
+                try:
+                    f = os.fdopen(fd, "wb")  # buffered writer; takes ownership of fd
+                    f.write(header)
+                    write_payload(f)
+                    f.flush()
+
+                    # Enforce max_value_mb on the real on-disk size (fstat, not tell():
+                    # the producer may have seeked back to patch bytes it already wrote).
+                    payload_size = os.fstat(fd).st_size - HEADER_SIZE
+                    max_bytes = self.config.max_value_mb * 1024 * 1024
+                    if payload_size > max_bytes:
+                        raise BackendError(
+                            f"Value size {payload_size} exceeds max_value_mb ({self.config.max_value_mb}MB)",
+                            BackendErrorType.PERMANENT,
+                        )
+
+                    # fsync to ensure data is on disk
+                    os.fsync(fd)
+                finally:
+                    self._release_file_lock(fd)
+            finally:
+                if f is not None:
+                    f.close()
+                else:
+                    os.close(fd)
+
+            # Commit under the process lock: atomic rename + eviction bookkeeping only.
+            with self._lock:
+                os.rename(temp_path, file_path)
+                committed = True
+
+                # Trigger eviction if over threshold
+                self._maybe_evict()
+
+        except OSError as exc:
+            raise BackendError(
+                f"Failed to write cache file: {exc}",
+                error_type=self._classify_os_error(exc, is_directory=False),
+                original_exception=exc,
+                operation="set_streaming",
+                key=key,
+            ) from exc
+        finally:
+            # Any failure — backend I/O, size limit, or a producer exception (which
+            # propagates unwrapped per the BufferWritableBackend contract, so callers keep
+            # its type and log it once) — discards the partial write. finally (not a
+            # BaseException catch) so even KeyboardInterrupt can't leak a temp file.
+            if not committed:
+                self._safe_unlink(temp_path)
+
     def delete(self, key: str) -> bool:
         """Delete key from file storage.
 
@@ -640,16 +711,8 @@ async def refresh_ttl(self, key: str, ttl: int) -> bool:
         only bytes [6:14] are rewritten, so the payload and all other header fields are
         untouched (and cross-SDK File readers stay compatible).
         """
-        # Same TTL bounds as set() (security: prevent integer overflow/underflow).
-        if ttl == 0:
-            new_expiry = 0
-        elif ttl < 0 or ttl > MAX_TTL_SECONDS:
-            raise BackendError(
-                f"TTL {ttl} out of range [0, {MAX_TTL_SECONDS}] (max 10 years)",
-                BackendErrorType.PERMANENT,
-            )
-        else:
-            new_expiry = int(time.time() + ttl)
+        # Same TTL bounds as set()/set_streaming (security: prevent integer overflow/underflow).
+        new_expiry = self._expiry_from_ttl(ttl)
 
         file_path = self._key_to_path(key)
 
@@ -704,6 +767,50 @@ async def refresh_ttl(self, key: str, ttl: int) -> bool:
 
     # Private helper methods
 
+    @staticmethod
+    def _expiry_from_ttl(ttl: int | None) -> int:
+        """Expiry timestamp for a TTL (0 = never expire), with TTL bounds validation.
+
+        Raises:
+            BackendError: If TTL is negative or exceeds MAX_TTL_SECONDS (security:
+                prevent integer overflow/underflow in the packed uint64 timestamp)
+        """
+        if ttl is None or ttl == 0:
+            return 0
+        if ttl < 0 or ttl > MAX_TTL_SECONDS:
+            raise BackendError(
+                f"TTL {ttl} out of range [0, {MAX_TTL_SECONDS}] (max 10 years)",
+                BackendErrorType.PERMANENT,
+            )
+        return int(time.time() + ttl)
+
+    @staticmethod
+    def _build_header(expiry_timestamp: int) -> bytes:
+        """Build the 14-byte on-disk entry header."""
+        return (
+            MAGIC  # [0:2] Magic bytes
+            + bytes([FORMAT_VERSION])  # [2:3] Version
+            + bytes([RESERVED])  # [3:4] Reserved
+            + struct.pack(">H", 0)  # [4:6] Flags (no compression/encryption yet)
+            + struct.pack(">Q", expiry_timestamp)  # [6:14] Expiry timestamp
+        )
+
+    def _check_entry_capacity(self, file_path: str) -> None:
+        """Reject a NEW entry when max_entry_count is reached (overwrites always pass).
+
+        Raises:
+            BackendError: If storing a new entry would exceed max_entry_count
+        """
+        if self.config.max_entry_count <= 0:
+            return
+        _, entry_count = self._calculate_cache_size()
+        # Only check if this is a NEW entry (not overwriting existing)
+        if not os.path.exists(file_path) and entry_count >= self.config.max_entry_count:
+            raise BackendError(
+                f"Entry count {entry_count} would exceed max_entry_count ({self.config.max_entry_count})",
+                BackendErrorType.PERMANENT,
+            )
+
     def _key_to_path(self, key: str) -> str:
         """Convert cache key to file path using blake2b hash.