CP HiCache L3 3.4: cold-start (restart durability) — load|clear via SGLANG_CP_L3_COLD_START

On restart the LMDB index + disk-slab blobs persist, but CpL3Store.from_config built the slot pools all-free and the GC LRU empty -> the next spill re-hands-out a slot the durable index still references (clobbers a live blob) and GC never reclaims the carried-over entries. Neither a clean start nor a durable reload was actually realized. connect() now applies a cold-start policy BEFORE the bg threads start (the write thread is the sole pool/GC owner, so the single-threaded reconcile must precede it): - clear (default): wipe the persisted index + reset the pools/GC -> genuinely empty start (disk blobs are inert, overwritten lazily on slot reuse). - load: rebuild this rank's slot free-list + GC LRU from the durable disk blobs. Drive the scan from the rank's OWN slab file (header-only reads) so it never inspects another rank's slots even when ranks share a disk; a slot is LIVE iff its blob header parses AND the shared index still maps that content hash back to this exact slot -> occupy + seed the GC LRU with the durable last_access; orphan/unwritten slots stay free. Header-only (no payload CRC); reload-time verify-on-read still fail-softs a torn payload. The L3 durable floor now survives a process restart. Primitives: CpL3SlotPool.rebuild_from_allocated (O(num_slots) bulk-occupy) + CpL3DiskSlab.read_header (one aligned block, not the multi-MB slot). Env SGLANG_CP_L3_COLD_START (default "clear"), read in _maybe_init_cp_l3 and passed to connect(). Tests: 4 cold-start e2e (load rebuilds the floor + no slot collision after a fresh spill; GC LRU rebuilt; clear starts empty; unknown mode fails loud) + 2 unit (rebuild_from_allocated, read_header). 23 L3 store/disk/posix tests green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-23 18:09:37 +00:00
parent d6aabace90
commit ea4ce31f95
8 changed files with 296 additions and 5 deletions
@@ -135,6 +135,25 @@ class TestSlotPool(unittest.TestCase):
        with self.assertRaises(ValueError):
            m.CpL3SlotPool(0)

+    def test_rebuild_from_allocated(self):
+        # cold-rebuild: occupy a subset in bulk, free list = the complement (deterministic lowest-first).
+        p = m.CpL3SlotPool(6)
+        p.rebuild_from_allocated([1, 3, 4])
+        self.assertEqual(p.num_allocated, 3)
+        self.assertEqual(p.num_free, 3)
+        self.assertEqual([p.alloc() for _ in range(3)], [0, 2, 5])  # complement, lowest-first
+        self.assertEqual(p.alloc(), -1)  # full
+        # the rebuilt-allocated slots are genuinely occupied (free() accepts them, double-alloc cannot reissue)
+        p.free(3)
+        self.assertEqual(p.alloc(), 3)
+
+    def test_rebuild_from_allocated_fail_loud(self):
+        p = m.CpL3SlotPool(4)
+        with self.assertRaises(ValueError):
+            p.rebuild_from_allocated([0, 4])  # out of range
+        with self.assertRaises(ValueError):
+            p.rebuild_from_allocated([2, 2])  # duplicate
+

 if __name__ == "__main__":
    unittest.main()
@@ -110,6 +110,19 @@ class TestDiskSlab(unittest.TestCase):
        self.slab.write_slot(0, _h(0), 1, b"\x33" * PAGE)
        self.slab.fdatasync()  # must not raise

+    def test_read_header_for_cold_rebuild(self):
+        # header-only read (cold-rebuild scan): a written slot returns its parsed header; an unwritten slot
+        # returns None (fresh fallocate = zeros = bad magic). Cheap: reads one aligned block, not the slot.
+        payload = bytes((i * 5) & 0xFF for i in range(PAGE))
+        self.slab.write_slot(2, _h(13), 78, payload)
+        hdr = self.slab.read_header(2)
+        self.assertIsNotNone(hdr)
+        self.assertEqual(hdr.payload_kind, "target_kv")
+        self.assertEqual(hdr.content_hash, _h(13))  # _h returns 32 raw bytes here
+        self.assertEqual(hdr.n_layers, 78)
+        self.assertEqual(hdr.page_bytes, PAGE)
+        self.assertIsNone(self.slab.read_header(5))  # unwritten -> None
+

 if __name__ == "__main__":
    unittest.main()
@@ -318,6 +318,143 @@ class TestCpL3Store(unittest.TestCase):
        self.assertFalse(self.store.has_inflight())


+class TestCpL3StoreColdStart(unittest.TestCase):
+    """Restart durability (cold-start): a NEW store reopened on the SAME disk + LMDB index must reconcile its
+    in-memory state with the persisted blobs. ``load`` rebuilds the slot free-list + GC LRU from the durable
+    disk blobs (the floor survives a restart, and the pool never re-hands-out a live slot). ``clear`` wipes the
+    index + resets the pools (a clean empty start). The bug being guarded: from_config builds the pools all-free
+    while the durable index still references occupied slots, so doing NEITHER corrupts the floor on restart."""
+
+    def setUp(self):
+        self._td = tempfile.TemporaryDirectory()
+        self.mm, self.acc = _make_slab_and_accessor()
+        self._cfg_dict = {
+            "backend": "posix", "require_plp": False, "index_map_gb": 0.05,
+            "disks": [{"path": os.path.join(self._td.name, "disk0"), "budget_gb": 0.02}],
+        }
+
+    def tearDown(self):
+        self._td.cleanup()
+
+    def _new_store(self):
+        cfg = cfg_mod.CpL3Config.from_dict(self._cfg_dict)
+        store = store_mod.CpL3Store.from_config(
+            cfg, cp_rank=0, cp_size=1, accessors={"target_kv": self.acc})
+        return store, cfg
+
+    def _spill(self, store, object_key, pages, last_access):
+        store.submit_spill(object_key, pages, last_access=last_access)
+        self.assertTrue(_wait_ack(store.ack_durable_qsize, 1))
+        store.drain_gather_acks(store.ack_gather_qsize())
+        acks = store.drain_durable_acks(store.ack_durable_qsize())
+        self.assertEqual(len(acks), 1)
+        self.assertTrue(acks[0][1])  # durable ok
+
+    def _zero_host_page(self, p):
+        for layer in range(N_LAYERS):
+            off = layer * (PAGE_NUM * SLICE) + p * SLICE
+            self.mm[off:off + SLICE] = bytes(SLICE)
+
+    def test_load_rebuilds_floor_and_no_slot_collision(self):
+        # boot 1: spill 3 pages of an object, drain durable, close (disk blobs + LMDB index persist on disk).
+        s1, cfg1 = self._new_store()
+        s1.connect(cfg1, cold_start="clear")
+        pages = {"target_kv": [(1, _h(1)), (2, _h(2)), (5, _h(5))]}
+        orig = {p: self.acc.gather(p) for p in (1, 2, 5)}
+        self._spill(s1, "obj", pages, last_access=10)
+        used = s1.pools["target_kv"].num_allocated
+        self.assertEqual(used, 3)
+        s1.close()
+
+        # boot 2: reopen on the SAME dirs with cold_start="load" -> the slot pool occupancy is rebuilt from
+        # the durable blobs (so the 3 live slots are marked allocated, not free) and the index still hits.
+        s2, cfg2 = self._new_store()
+        s2.connect(cfg2, cold_start="load")
+        self.addCleanup(s2.close)
+        self.assertEqual(s2.pools["target_kv"].num_allocated, used)
+        self.assertEqual(s2.exists_prefix([_h(1), _h(2)], ["target_kv"]), 2)
+        self.assertEqual(s2.exists_prefix([_h(5)], ["target_kv"]), 1)
+
+        # the airtight check: a NEW spill must consume a FREE slot, NOT re-hand-out one of the 3 restored
+        # live slots. Without the rebuild the pool is all-free -> this new page would alloc slot 0 (a live
+        # slot) and clobber _h(1)'s blob -> the reload below would return garbage. With the rebuild it lands
+        # on a fresh slot and the restored floor is intact.
+        self._spill(s2, "obj-new", {"target_kv": [(0, _h(7))]}, last_access=20)
+        self.assertEqual(s2.pools["target_kv"].num_allocated, used + 1)
+
+        # reload the 3 restored pages after a simulated L2 eviction -> byte-exact (proves the floor survived
+        # the restart AND was not clobbered by the new spill).
+        for p in (1, 2, 5):
+            self._zero_host_page(p)
+        s2.submit_reload("obj", pages)
+        self.assertTrue(_wait_ack(s2.ack_reload_qsize, 1))
+        racks = s2.drain_reload_acks(s2.ack_reload_qsize())
+        self.assertEqual(len(racks), 1)
+        self.assertTrue(racks[0][1])  # reload ok (op_id is store-local; a fresh store restarts the counter)
+        for p in (1, 2, 5):
+            self.assertEqual(self.acc.gather(p), orig[p])
+
+    def test_load_rebuilds_gc_lru(self):
+        # the rebuilt GC LRU must be able to reclaim RESTORED pages (proves _gc_current/_gc_heap repopulated,
+        # not just the slot pool): fill near the start watermark in boot 1, restart with load, push one page
+        # over the watermark -> the write-thread GC reclaims the coldest RESTORED page.
+        td = tempfile.TemporaryDirectory()
+        self.addCleanup(td.cleanup)
+        cfg_dict = {
+            "backend": "posix", "require_plp": False, "index_map_gb": 0.05,
+            "disks": [{"path": os.path.join(td.name, "d"), "budget_gb": 0.0001}],
+        }
+
+        def new_store():
+            c = cfg_mod.CpL3Config.from_dict(cfg_dict)
+            return store_mod.CpL3Store.from_config(
+                c, cp_rank=0, cp_size=1, accessors={"target_kv": self.acc}), c
+
+        s1, c1 = new_store()
+        s1.connect(c1, cold_start="clear")
+        n = s1.pools["target_kv"].num_slots
+        self.assertGreaterEqual(n, 8)
+        start = int(s1._gc_start_frac * n)
+        # fill to just below the start watermark, page i with last_access=i (smaller == colder)
+        for i in range(start):
+            self._spill(s1, f"o{i}", {"target_kv": [(i % PAGE_NUM, _h(5000 + i))]}, last_access=i)
+        self.assertEqual(s1.pools["target_kv"].num_allocated, start)
+        s1.close()
+
+        s2, c2 = new_store()
+        s2.connect(c2, cold_start="load")
+        self.addCleanup(s2.close)
+        self.assertEqual(s2.pools["target_kv"].num_allocated, start)  # floor restored
+        # push over the start watermark with WARM pages -> GC must fire on the rebuilt LRU and reclaim the
+        # coldest RESTORED page (last_access=0). If the heap were empty (no rebuild) GC would see nothing.
+        for i in range(start, n + 1):
+            self._spill(s2, f"o{i}", {"target_kv": [(i % PAGE_NUM, _h(5000 + i))]}, last_access=10_000 + i)
+        t0 = time.time()
+        while s2.pools["target_kv"].num_allocated > start and time.time() - t0 < 5.0:
+            time.sleep(0.01)
+        self.assertLessEqual(s2.pools["target_kv"].num_allocated, start)
+        self.assertEqual(s2.exists_prefix([_h(5000 + 0)], ["target_kv"]), 0)  # coldest restored -> reclaimed
+
+    def test_clear_starts_empty(self):
+        s1, c1 = self._new_store()
+        s1.connect(c1, cold_start="clear")
+        self._spill(s1, "o", {"target_kv": [(0, _h(0))]}, last_access=1)
+        self.assertEqual(s1.exists_prefix([_h(0)], ["target_kv"]), 1)
+        s1.close()
+        # reopen with clear -> the persisted index is wiped + the pool is empty (clean fresh start)
+        s2, c2 = self._new_store()
+        s2.connect(c2, cold_start="clear")
+        self.addCleanup(s2.close)
+        self.assertEqual(s2.exists_prefix([_h(0)], ["target_kv"]), 0)
+        self.assertEqual(s2.pools["target_kv"].num_free, s2.pools["target_kv"].num_slots)
+
+    def test_unknown_cold_start_mode_fails_loud(self):
+        s1, c1 = self._new_store()
+        self.addCleanup(s1.close)
+        with self.assertRaises(ValueError):
+            s1.connect(c1, cold_start="bogus")
+
+
 class TestCpL3StoreMultiSlab(unittest.TestCase):
    """The store must work when the host cache is split across multiple physical slabs:
    the accessor dispatches each global page to its owning slab (per-slab layer stride)."""