L3 3.0: on-disk blob format + disk-slot free-list allocator

Phase 3.0 (L3 disk durable floor) foundational primitives, pure + unit-tested:
- 64B self-describing blob header (magic/version/payload-kind/content-hash/CRC),
  verified on every read (verify_blob) -> torn/bit-rot slots become a MISS, not garbage.
- slot_bytes_for_page: header+payload rounded to the 4K O_DIRECT boundary.
- CpL3SlotPool: O(1) free-list over fixed-size per-payload disk page-slots (the L3
  analog of CpSharedL2PageAllocator's free list), fail-loud on double-free/OOB, with
  reset() for the flush_cache hook. NOT a ring buffer (eviction frees arbitrary slots).

Rank-local (a rank stores only its owned pages; L3 eviction selection rides the shared
LMDB replicated clock). 12/12 unit tests. Design: docs_internal/cp_hicache_l3_phase3_impl_design.md.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

python/sglang/srt/mem_cache/cp_l3_disk.py

@@ -0,0 +1,218 @@
+# Copyright 2023-2024 SGLang Team
+# Licensed under the Apache License, Version 2.0
+"""CP HiCache L3 (disk durable floor) — on-disk primitives.
+
+Pure, backend-agnostic building blocks for the L3 disk tier (Phase 3.0):
+  * the self-describing per-page **blob header** (CRC + content hash, verified on every read), and
+  * the **disk-slot free-list allocator** (``CpL3SlotPool``) — O(1) alloc/free of fixed-size page slots,
+    the L3 analog of ``CpSharedL2PageAllocator``'s free list.
+
+A page is stored on disk as one *contiguous* blob (header + the gathered, layer-major page payload), even
+though the host slab is ``layer_page_first`` (layer-strided) — the spill writer gathers the per-layer slices
+into a contiguous buffer (measured faster than vectored I/O under O_DIRECT). Slots are fixed-size **per
+payload kind** so allocation is O(1) and fragmentation-free. Each ``(payload_kind, drive)`` owns one
+``CpL3SlotPool`` + one preallocated disk-slab file; pools are **local to each CP rank** (a rank stores only its
+owned pages — ``owner=(page-1)%cp_size``), so slot indices need no cross-rank agreement (L3 *eviction
+selection*, which must be replicated, rides the shared LMDB index's replicated clock — see the impl design).
+"""
+
+from __future__ import annotations
+
+import struct
+import zlib
+from typing import Tuple
+
+CP_L3_FAILFAST_PREFIX = "[CP_L3_FAILFAST]"
+
+# Disk/O_DIRECT alignment: slot byte sizes + offsets are rounded up to this so a single O_DIRECT
+# read/write of a whole slot uses one aligned buffer (the header lives at offset 0, payload at
+# HEADER_BYTES, trailing pad to the boundary). 4 KiB covers ext4 on 4K-sector NVMe.
+CP_L3_DIO_ALIGN = 4096
+
+# --- blob header: 64 bytes, self-describing, CRC verified on read ---
+# magic u32 | version u16 | payload_kind_id u8 | flags u8 | content_hash[32] | n_layers u16 |
+# page_bytes u32 | crc32 u32 | 14x pad  ==  4+2+1+1+32+2+4+4 = 50, padded to 64.
+CP_L3_BLOB_MAGIC = 0x4C33424C  # "L3BL"
+CP_L3_BLOB_VERSION = 1
+_HEADER_STRUCT = struct.Struct("<IHBB32sHII14x")
+CP_L3_HEADER_BYTES = _HEADER_STRUCT.size  # 64
+assert CP_L3_HEADER_BYTES == 64
+
+# payload_kind <-> stable on-disk id (frozen; never renumber — cold-rebuild reads old blobs)
+_PAYLOAD_KIND_TO_ID = {"target_kv": 0, "draft_kv": 1, "index_k": 2}
+_PAYLOAD_ID_TO_KIND = {v: k for k, v in _PAYLOAD_KIND_TO_ID.items()}
+
+CONTENT_HASH_BYTES = 32  # sha256 digest (node.hash_value entries are 64-hex == 32 bytes)
+
+
+def _l3_failfast(message: str) -> None:
+    raise ValueError(f"{CP_L3_FAILFAST_PREFIX} {message}")
+
+
+def align_up(nbytes: int, align: int = CP_L3_DIO_ALIGN) -> int:
+    if nbytes < 0 or align <= 0:
+        _l3_failfast(f"align_up requires nbytes>=0, align>0; got nbytes={nbytes} align={align}")
+    return (nbytes + align - 1) // align * align
+
+
+def slot_bytes_for_page(page_bytes: int) -> int:
+    """On-disk slot size = header + payload, rounded up to the O_DIRECT boundary."""
+    if page_bytes <= 0:
+        _l3_failfast(f"page_bytes must be positive; got {page_bytes}")
+    return align_up(CP_L3_HEADER_BYTES + int(page_bytes))
+
+
+def content_hash_to_bytes(hash_str: str) -> bytes:
+    """node.hash_value entry (64-char sha256 hex) -> 32 raw bytes for the blob header / index key."""
+    try:
+        raw = bytes.fromhex(hash_str)
+    except (ValueError, TypeError) as exc:
+        _l3_failfast(f"content hash is not valid hex: {hash_str!r} ({exc})")
+    if len(raw) != CONTENT_HASH_BYTES:
+        _l3_failfast(
+            f"content hash must be {CONTENT_HASH_BYTES} bytes ({2*CONTENT_HASH_BYTES} hex chars); "
+            f"got {len(raw)} bytes from {hash_str!r}"
+        )
+    return raw
+
+
+def compute_crc(payload) -> int:
+    """CRC-32 (zlib, ISO-HDLC) over the page payload bytes. Integrity check, not crypto."""
+    return zlib.crc32(payload) & 0xFFFFFFFF
+
+
+def pack_blob_header(
+    *, payload_kind: str, content_hash: bytes, n_layers: int, page_bytes: int, crc: int, flags: int = 0
+) -> bytes:
+    if payload_kind not in _PAYLOAD_KIND_TO_ID:
+        _l3_failfast(f"unknown payload_kind: {payload_kind!r}")
+    if len(content_hash) != CONTENT_HASH_BYTES:
+        _l3_failfast(f"content_hash must be {CONTENT_HASH_BYTES} bytes; got {len(content_hash)}")
+    if n_layers <= 0 or n_layers > 0xFFFF:
+        _l3_failfast(f"n_layers out of range: {n_layers}")
+    if page_bytes <= 0 or page_bytes > 0xFFFFFFFF:
+        _l3_failfast(f"page_bytes out of range: {page_bytes}")
+    return _HEADER_STRUCT.pack(
+        CP_L3_BLOB_MAGIC,
+        CP_L3_BLOB_VERSION,
+        _PAYLOAD_KIND_TO_ID[payload_kind],
+        int(flags) & 0xFF,
+        bytes(content_hash),
+        int(n_layers),
+        int(page_bytes),
+        int(crc) & 0xFFFFFFFF,
+    )
+
+
+class CpL3BlobHeader:
+    """Parsed blob header. Use ``unpack_blob_header`` to construct from bytes."""
+
+    __slots__ = ("payload_kind", "content_hash", "n_layers", "page_bytes", "crc", "flags", "version")
+
+    def __init__(self, payload_kind, content_hash, n_layers, page_bytes, crc, flags, version):
+        self.payload_kind = payload_kind
+        self.content_hash = content_hash
+        self.n_layers = n_layers
+        self.page_bytes = page_bytes
+        self.crc = crc
+        self.flags = flags
+        self.version = version
+
+
+def unpack_blob_header(raw) -> CpL3BlobHeader:
+    """Parse + validate a 64-byte header. Fail loud on magic/version/kind corruption."""
+    if len(raw) < CP_L3_HEADER_BYTES:
+        _l3_failfast(f"blob header needs {CP_L3_HEADER_BYTES} bytes; got {len(raw)}")
+    magic, version, kind_id, flags, content_hash, n_layers, page_bytes, crc = _HEADER_STRUCT.unpack(
+        bytes(raw[:CP_L3_HEADER_BYTES])
+    )
+    if magic != CP_L3_BLOB_MAGIC:
+        _l3_failfast(f"bad blob magic 0x{magic:08X} (expected 0x{CP_L3_BLOB_MAGIC:08X}) — torn/foreign slot")
+    if version != CP_L3_BLOB_VERSION:
+        _l3_failfast(f"unsupported blob version {version} (expected {CP_L3_BLOB_VERSION})")
+    if kind_id not in _PAYLOAD_ID_TO_KIND:
+        _l3_failfast(f"unknown payload_kind_id {kind_id} in blob header")
+    return CpL3BlobHeader(
+        payload_kind=_PAYLOAD_ID_TO_KIND[kind_id],
+        content_hash=content_hash,
+        n_layers=n_layers,
+        page_bytes=page_bytes,
+        crc=crc,
+        flags=flags,
+        version=version,
+    )
+
+
+def verify_blob(header: CpL3BlobHeader, payload, *, expect_hash: bytes = None) -> bool:
+    """Verify a read-back blob: payload length, CRC, and (optionally) the expected content hash.
+
+    Returns True iff intact. A False result means treat the slot as a MISS (recompute) + reclaim it —
+    this is the verify-on-read guard against torn same-length writes / bit-rot (HiCacheFile has none).
+    """
+    if len(payload) != header.page_bytes:
+        return False
+    if compute_crc(payload) != header.crc:
+        return False
+    if expect_hash is not None and bytes(expect_hash) != bytes(header.content_hash):
+        return False
+    return True
+
+
+class CpL3SlotPool:
+    """O(1) free-list over fixed-size disk page-slots for one ``(payload_kind, drive)`` (rank-local).
+
+    Mirrors ``CpSharedL2PageAllocator``'s free-list discipline: explicit free stack + allocated set, with
+    double-free / out-of-range / leak fail-loud, and a ``reset()`` for the ``flush_cache`` hook. Deterministic
+    allocation order (lowest free index first) so tests/observability are reproducible. NOT a ring buffer (a
+    ring = FIFO eviction = drops hot objects); eviction frees arbitrary slots back here.
+    """
+
+    def __init__(self, num_slots: int):
+        if num_slots <= 0:
+            _l3_failfast(f"CpL3SlotPool num_slots must be positive; got {num_slots}")
+        self._num_slots = int(num_slots)
+        # stack ordered so pop() yields 0,1,2,... (deterministic, reproducible)
+        self._free = list(range(self._num_slots - 1, -1, -1))
+        self._allocated: set[int] = set()
+
+    @property
+    def num_slots(self) -> int:
+        return self._num_slots
+
+    @property
+    def num_free(self) -> int:
+        return len(self._free)
+
+    @property
+    def num_allocated(self) -> int:
+        return len(self._allocated)
+
+    @property
+    def occupancy(self) -> float:
+        return self._num_slots and (len(self._allocated) / self._num_slots) or 0.0
+
+    def has_free(self) -> bool:
+        return bool(self._free)
+
+    def alloc(self) -> int:
+        """Reserve one slot; returns its index. Caller handles a full pool (returns -1) by evicting first."""
+        if not self._free:
+            return -1
+        idx = self._free.pop()
+        self._allocated.add(idx)
+        return idx
+
+    def free(self, idx: int) -> None:
+        idx = int(idx)
+        if idx < 0 or idx >= self._num_slots:
+            _l3_failfast(f"free slot index {idx} out of range [0,{self._num_slots})")
+        if idx not in self._allocated:
+            _l3_failfast(f"double-free / never-allocated L3 slot {idx}")
+        self._allocated.discard(idx)
+        self._free.append(idx)
+
+    def reset(self) -> None:
+        """Return to the freshly-constructed all-free state (the ``flush_cache``/``clear`` hook)."""
+        self._free = list(range(self._num_slots - 1, -1, -1))
+        self._allocated = set()
+