Bound restore memory and reduce cache publication work

deploy/deepseek_v4/tp4_profile.json

@@ -4,7 +4,7 @@
   "cache_model_profile": "deepseek-v4-fp8-hma",
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "sparkcache": {
-    "source_sha256": "b727ff2e6f5d50966a42a5811ab23136e5160754286cd0c33c16b4f811082ef4"
+    "source_sha256": "4c629645b49012969295dc3942821228e4aca887c994be749cd0375ac860ee24"
   },
   "model": {
     "repository": "deepseek-ai/DeepSeek-V4-Flash-0731",

deploy/glm52_35bpw/profile.json

@@ -5,7 +5,7 @@
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "base_image_requirement": "exact GLM-5.2 3.5-bpw R7 image recorded by the source container inspection",
   "sparkcache": {
-    "source_sha256": "b727ff2e6f5d50966a42a5811ab23136e5160754286cd0c33c16b4f811082ef4"
+    "source_sha256": "4c629645b49012969295dc3942821228e4aca887c994be749cd0375ac860ee24"
   },
   "model": {
     "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",

sparkcache/README.md

@@ -171,6 +171,28 @@ SparkCache authenticates objects, checks logical positions, places bytes into
 request-owned GPU blocks, and resumes the request only after CUDA completion.
 Any error discards those private blocks and recomputes the prompt.
 
+The optional `spark_cache_cuda_restore_arena_budget_bytes` setting bounds
+restore arena allocation per worker rank. Its environment equivalent is
+`SPARK_CONTEXT_CACHE_CUDA_RESTORE_ARENA_BUDGET_BYTES`.
+
+The default, `0`, keeps the configured lane count. A positive budget must fit
+at least one lane's two arenas.
+
+The connector caps page restore lanes at the smaller of
+`spark_cache_load_threads` (maximum eight) and the number of complete arena
+pairs the budget permits. Row restore uses one lane.
+
+For example, 256 MiB arenas and a 1 GiB budget permit two page restore lanes,
+allocating 1 GiB instead of the 4 GiB required by eight lanes. Arenas are
+allocated at startup.
+
+This budget covers restore arenas; capture rings,
+authenticated host objects, GPU KV blocks, and placement metadata have
+separate allocations.
+
+Adjusting the budget preserves cache identities and on-disk compatibility.
+Concurrent restore throughput requires deployment testing.
+
 See [`native/README.md`](native/README.md) for the ABI and memory-ordering
 rules. Deployment profiles record the model layouts tested with this path.
 
@@ -246,6 +268,22 @@ evicts least-recently-used manifests down to
 `spark_cache_ttl_seconds` expires manifests by recency; zero disables TTL.
 Maintenance preserves shared objects referenced by surviving manifests.
 
+An admitted asynchronous publication can protect its base and result roots
+until post-commit reconciliation. Only one such publication runs per rank.
+Protected bytes still count against capacity.
+
+Other roots remain eligible for eviction. Capacity may temporarily remain
+unsatisfied while a protected publication finishes; the single inflight
+admission prevents another protected publication from accumulating.
+
+Success releases that protection before post-commit cleanup. Failure,
+preemption, and shutdown completion release it too. An unsatisfied capacity
+budget prevents another base reservation.
+
+If the worker no longer offers the selected base, or maintenance is busy,
+capture switches to a complete snapshot without waiting. The existing ring
+size and admission limits still apply; an oversized snapshot is skipped.
+
 To clear one cache root once, set `spark_cache_clear_once` to a deliberate
 token:
 

sparkcache/held_inventory.py

@@ -0,0 +1,104 @@
+"""Mutation-tracked digest inventory for bounded worker status reporting."""
+
+from collections.abc import Iterable, Iterator, MutableSet
+
+
+class HeldInventory(MutableSet[str]):
+    """Set of offered digests with a revision for content changes.
+
+    Callers serialize reads and mutations with the connector's worker-state
+    lock. The revision belongs to this object: replacing an inventory requires
+    comparing its identity as well as its revision. Input collections are
+    copied so mutations cannot bypass revision tracking through another owner.
+    """
+
+    def __init__(self, values: Iterable[str] = ()) -> None:
+        self._values = set(values)
+        self._revision = 0
+
+    @property
+    def revision(self) -> int:
+        return self._revision
+
+    def __contains__(self, value: object) -> bool:
+        return value in self._values
+
+    def __iter__(self) -> Iterator[str]:
+        return iter(self._values)
+
+    def __len__(self) -> int:
+        return len(self._values)
+
+    def add(self, value: str) -> None:
+        before = len(self._values)
+        self._values.add(value)
+        self._revision += len(self._values) != before
+
+    def discard(self, value: str) -> None:
+        before = len(self._values)
+        self._values.discard(value)
+        self._revision += len(self._values) != before
+
+    def remove(self, value: str) -> None:
+        self._values.remove(value)
+        self._revision += 1
+
+    def pop(self) -> str:
+        value = self._values.pop()
+        self._revision += 1
+        return value
+
+    def clear(self) -> None:
+        if self._values:
+            self._values.clear()
+            self._revision += 1
+
+    def update(self, *values: Iterable[str]) -> None:
+        before = len(self._values)
+        try:
+            self._values.update(*values)
+        finally:
+            # Iterators can fail after yielding elements. A partial mutation
+            # must still invalidate the report's inventory snapshot.
+            self._revision += len(self._values) != before
+
+    def difference_update(self, *values: Iterable[str]) -> None:
+        before = len(self._values)
+        try:
+            self._values.difference_update(
+                *(self._values if other is self else other for other in values)
+            )
+        finally:
+            self._revision += len(self._values) != before
+
+    def intersection_update(self, *values: Iterable[str]) -> None:
+        before = len(self._values)
+        try:
+            self._values.intersection_update(*values)
+        finally:
+            self._revision += len(self._values) != before
+
+    def symmetric_difference_update(self, values: Iterable[str]) -> None:
+        other = set(values)
+        if other:
+            self._values.symmetric_difference_update(other)
+            self._revision += 1
+
+    def __ior__(self, values: Iterable[str]) -> "HeldInventory":
+        self.update(values)
+        return self
+
+    def __isub__(self, values: Iterable[str]) -> "HeldInventory":
+        self.difference_update(values)
+        return self
+
+    def __iand__(self, values: Iterable[str]) -> "HeldInventory":
+        self.intersection_update(values)
+        return self
+
+    def __ixor__(self, values: Iterable[str]) -> "HeldInventory":
+        self.symmetric_difference_update(values)
+        return self
+
+    def copy(self) -> set[str]:
+        return self._values.copy()

sparkcache/native/CMakeLists.txt

@@ -63,6 +63,10 @@ option(
     SPARK_CACHE_PLACEMENT_ENABLE_CUDA
     "Build the mapped/managed/staged CUDA placement engine"
     ON)
+option(
+    SPARK_CACHE_PLACEMENT_GPU_TESTS
+    "Register GPU page-copy correctness tests for all placement arena modes"
+    OFF)
 option(
     SPARK_CACHE_SNAPSHOT_TEST_FORCE_EVENT_RECORD_FAILURE
     "Test seam: force every snapshot completion-event record to fail"
@@ -111,6 +115,27 @@ if(SPARK_CACHE_PLACEMENT_ENABLE_CUDA)
         target_link_libraries(spark_cache_hybrid_page_probe
             PRIVATE spark_cache_placement CUDA::cudart)
 
+        if(UNIX)
+            # Resolve the tested placement library dynamically so one binary
+            # can compare independently built libraries without interposition.
+            add_executable(spark_cache_page_copy_benchmark
+                app/spark_cache_page_copy_benchmark.cu)
+            target_include_directories(spark_cache_page_copy_benchmark PRIVATE include)
+            target_link_libraries(spark_cache_page_copy_benchmark
+                PRIVATE CUDA::cudart ${CMAKE_DL_LIBS})
+            add_dependencies(spark_cache_page_copy_benchmark spark_cache_placement)
+            if(BUILD_TESTING AND SPARK_CACHE_PLACEMENT_GPU_TESTS)
+                foreach(arena_mode RANGE 1 3)
+                    add_test(
+                        NAME spark_cache_page_copy_mode_${arena_mode}
+                        COMMAND spark_cache_page_copy_benchmark
+                            $<TARGET_FILE:spark_cache_placement> 3 ${arena_mode})
+                    set_tests_properties(spark_cache_page_copy_mode_${arena_mode}
+                        PROPERTIES LABELS "gpu;page-copy" RUN_SERIAL TRUE TIMEOUT 180)
+                endforeach()
+            endif()
+        endif()
+
         add_library(spark_cache_snapshot SHARED
             src/spark_cache_snapshot.cu)
         target_compile_definitions(spark_cache_snapshot

sparkcache/native/README.md

@@ -27,6 +27,63 @@ read objects into a mapped host arena
 If any step fails, SparkCache discards the request's private blocks and lets
 vLLM compute the prompt normally.
 
+### Page-copy correctness and timing
+
+Status: **implemented** benchmark and opt-in GPU tests. Page scatter divides
+slabs into 64 KiB tiles and locates their extents in the validated span table.
+
+Aligned page fragments use 16-byte or 4-byte copies; unaligned fragments use
+byte copies. Blocks handle additional tiles when the grid reaches its limit.
+No expanded span table is allocated.
+
+[`app/spark_cache_page_copy_benchmark.cu`](app/spark_cache_page_copy_benchmark.cu)
+loads a selected library through its public ABI.
+
+It compares destination bytes with the independent CPU reference, including
+padding, unused physical pages, and guard bytes.
+
+Fixtures cover 64 MiB spans, a 257 MiB capped-grid continuation, large pages, 1,024 spans across eight layers,
+irregular framing, odd page widths, shuffled slots, and split submissions.
+
+Each iteration clears the destination outside the timed region to expose
+missing writes.
+
+On a CUDA host, build `spark_cache_page_copy_benchmark` with CMake. Enable
+`-DSPARK_CACHE_PLACEMENT_GPU_TESTS=ON` to register serial CTest checks for all
+three arena modes.
+
+GPU tests are disabled by default: CPU test hosts may have a CUDA compiler
+without a GPU. Run `ctest --test-dir BUILD -L page-copy --output-on-failure`,
+replacing `BUILD` with the build directory.
+
+For comparison runs, pass a library path, iteration count, and arena mode:
+
+```bash
+BUILD/spark_cache_page_copy_benchmark /absolute/path/libspark_cache_placement.so 15 1
+BUILD/spark_cache_page_copy_benchmark /absolute/path/libspark_cache_placement.so 15 2
+BUILD/spark_cache_page_copy_benchmark /absolute/path/libspark_cache_placement.so 15 3
+```
+
+Modes 1, 2, and 3 use mapped host, managed, and staged device memory,
+respectively. Three warmups precede timed iterations.
+
+The measured interval includes begin, acquire, submission, and completion
+fences; it is not an isolated kernel measurement. Source filling and byte
+comparisons are excluded.
+
+Managed mode reuses the filled source after warmup, without optional prefetch
+or CPU refills between iterations. Its timing does not measure migration for
+fresh CPU-produced slabs.
+
+Explicit data allocations stay below 400 MiB, excluding CUDA context and
+library overhead.
+
+Compare libraries in separate processes, then reverse their order. Retain
+compiler settings, library SHA-256 values, GPU identity, and serving activity.
+
+A microbenchmark improvement does not establish an end-to-end restore or
+serving improvement.
+
 ## Python interface
 
 [`../spark_cache_cuda.py`](../spark_cache_cuda.py) defines the Python ABI.