Add opt-in full-capture cadence and publication backlog metrics

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": "4c629645b49012969295dc3942821228e4aca887c994be749cd0375ac860ee24"
+    "source_sha256": "72f33311d1ee5811c2b3e4eefda98c9d17b5afe8d9d015a2e3c26c893f25aabc"
   },
   "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": "4c629645b49012969295dc3942821228e4aca887c994be749cd0375ac860ee24"
+    "source_sha256": "72f33311d1ee5811c2b3e4eefda98c9d17b5afe8d9d015a2e3c26c893f25aabc"
   },
   "model": {
     "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",

sparkcache/README.md

@@ -255,6 +255,44 @@ pinned unified memory.
 Saturation always skips the optional publication instead of waiting for a
 slot.
 
+`spark_cache_page_snapshot_interval_tokens` optionally selects complete
+asynchronous page captures at token boundaries. It accepts a non-negative
+integer and defaults to `0`, which disables the policy.
+
+The environment fallback is
+`SPARK_CONTEXT_CACHE_PAGE_SNAPSHOT_INTERVAL_TOKENS`; an explicit connector
+setting takes precedence. For example:
+
+```json
+"spark_cache_page_snapshot_interval_tokens": 16384
+```
+
+A dependent publication captures complete state when its result span and
+selected base span fall into different interval buckets, measured from token
+zero. The choice happens before sparse capture, without reading history.
+
+For interval 16,384, base 14,336 to result 16,384 selects full capture;
+base 16,384 to result 18,432 remains sparse. This is a token cadence, not a
+universal bound on history depth for arbitrary prompt increments.
+
+Full captures keep the same cache identity and format. Existing ring size and
+admission limits still apply.
+
+The counter
+`publication_periodic_full_capture_selected` counts selections, including
+attempts later rejected by a busy or undersized ring.
+
+Performance status: **research-only**. A CPU fixture used 25 extensions of
+2,048 tokens, eight appended attention layers, and four overwritten recurrent
+layers. Payload widths and macro-object size were scaled down by eight.
+
+With a full snapshot every eight extensions, staged writes increased from
+160.36 to 238.21 MiB, about **49%**, with no deduplication savings. Source
+preparation was excluded from commit timing; no GPU work was executed.
+
+This CPU measurement does not establish GPU capture interference, eviction
+behavior, or a causal end-to-end serving improvement from the interval policy.
+
 The delayed-store limit reserves at most 16 request lifetimes by default.
 When the limit is full, SparkCache omits another optional store plan before
 worker capture begins, so vLLM can release that request's pages normally.
@@ -265,6 +303,14 @@ worker capture begins, so vLLM can release that request's pages normally.
 evicts least-recently-used manifests down to
 `spark_cache_low_watermark_bytes`, which defaults to 90% of the high watermark.
 
+Choose capacity and the high-to-low watermark gap from the reusable working
+set, largest admitted publication, and measured publication and reclamation
+rates.
+
+A larger gap amortizes maintenance across more writes, but each pass evicts
+more data and can increase future misses. Compare those costs with observed
+publication age and maintenance activity before changing the gap.
+
 `spark_cache_ttl_seconds` expires manifests by recency; zero disables TTL.
 Maintenance preserves shared objects referenced by surviving manifests.
 
@@ -322,6 +368,21 @@ pages, and the oldest ownership age.
 
 The line disappears after every rank reports its terminal completion.
 
+`sparkcache: publication_work` reports pending saver admissions, their oldest
+age, and ranks performing capacity maintenance. Admission age includes capture,
+queue time, commit, and post-commit reconciliation.
+
+Each worker admits at most one saver publication. The pending rank-slot gauge
+sums these admissions across physical ranks; it is not a count of unique user
+requests. Age is the maximum reported across ranks.
+
+The maintenance flag covers the scan and survivor reconciliation, including
+failure cleanup. Metrics sample it without taking the capacity lock or reading
+the filesystem.
+
+Completed, failed, and aborted publications clear their age. A timed-out
+shutdown with a live saver remains pending instead of falsely reporting idle.
+
 The same ownership state is available from the vLLM Prometheus endpoint:
 
 | Gauge | Meaning |
@@ -331,6 +392,17 @@ The same ownership state is available from the vLLM Prometheus endpoint:
 | `vllm:sparkcache_capture_retained_manager_pages` | Physical manager pages retained across ranks. |
 | `vllm:sparkcache_capture_oldest_delayed_seconds` | Age of the oldest retained request ownership. |
 | `vllm:sparkcache_capture_ownership_uncertain_ranks` | Ranks that cannot prove whether capture still owns source pages. |
+| `vllm:sparkcache_publication_pending_rank_slots` | Pending saver admissions summed across physical ranks. |
+| `vllm:sparkcache_publication_oldest_pending_seconds` | Maximum admission age at the last worker reports. |
+| `vllm:sparkcache_maintenance_active_ranks` | Ranks reporting an active scan or survivor reconciliation. |
+
+These gauges describe the last worker reports received through the existing
+statistics channel. Reports may stop refreshing while the engine is idle;
+scraping Prometheus again does not make a cached age a live clock.
+
+Use report freshness when correlating idle-probe slowdowns with pending work.
+The existing streaming-publication handoff count remains separate from saver
+admissions and capture ownership.
 
 Exact process-local totals are available from
 `ManifestStore.publication_telemetry_snapshot()` using schema

sparkcache/spark_context_cache_config.py

@@ -408,6 +408,7 @@ class ConnectorConfig:
     restore_enabled: bool
     streaming_snapshots_enabled: bool
     async_page_capture_enabled: bool
+    page_snapshot_interval_tokens: int
     cuda_restore_enabled: bool
     cuda_placement_library_path: str
     cuda_placement_library_sha256: str
@@ -642,6 +643,13 @@ def parse_connector_config(
             os.environ.get("SPARK_CONTEXT_CACHE_MIN_SPAN", "1024"),
         )
     )
+    page_snapshot_interval_tokens = _nonnegative_config_int(
+        extra(
+            "spark_cache_page_snapshot_interval_tokens",
+            os.environ.get("SPARK_CONTEXT_CACHE_PAGE_SNAPSHOT_INTERVAL_TOKENS", "0"),
+        ),
+        "spark_cache_page_snapshot_interval_tokens",
+    )
     model_max = int(getattr(vllm_config.model_config, "max_model_len", 0) or 0)
     default_max_span = str(model_max if model_max > 0 else 1 << 30)
     max_span = int(
@@ -983,6 +991,7 @@ def parse_connector_config(
         restore_enabled=restore_enabled,
         streaming_snapshots_enabled=streaming_snapshots_enabled,
         async_page_capture_enabled=async_page_capture_enabled,
+        page_snapshot_interval_tokens=page_snapshot_interval_tokens,
         cuda_restore_enabled=cuda_restore_enabled,
         cuda_placement_library_path=cuda_placement_library_path,
         cuda_placement_library_sha256=cuda_placement_library_sha256,

sparkcache/spark_context_cache_connector.py

@@ -103,6 +103,7 @@
 )
 from sparkcache.spark_context_cache_store import (
     CacheIdentity,
+    CapacityPolicy,
     ContextChunk,
     EntryKey,
     LookupResult,
@@ -527,6 +528,7 @@ def aggregate(self, other: "KVConnectorStats") -> "KVConnectorStats":
                     "capacity",
                     "async_capture",
                     "publication",
+                    "publication_work",
                 ):
                     if isinstance(report.get(field), dict):
                         normalized[field] = dict(report[field])
@@ -654,6 +656,23 @@ def reduce(self) -> dict[str, int | float]:
                     for status in async_capture
                 ),
             )
+        publication_work = [
+            report["publication_work"]
+            for report in reports
+            if isinstance(report.get("publication_work"), dict)
+        ]
+        if publication_work:
+            reduced.update(
+                sparkcache_publication_pending_rank_slots=sum(
+                    int(status.get("pending", 0)) for status in publication_work
+                ),
+                sparkcache_publication_oldest_pending_ms=max(
+                    float(status.get("oldest_pending_ms", 0.0)) for status in publication_work
+                ),
+                sparkcache_maintenance_active_ranks=sum(
+                    int(bool(status.get("maintenance_active", False))) for status in publication_work
+                ),
+            )
         publication = [
             report.get("publication")
             for report in reports
@@ -740,16 +759,40 @@ def byte_count(name: str) -> str:
             if uncertain_ranks:
                 line += f" uncertain_ranks={uncertain_ranks}"
             lines.append(line)
+        pending_publications = int(reduced.get("sparkcache_publication_pending_rank_slots", 0))
+        maintenance_ranks = int(reduced.get("sparkcache_maintenance_active_ranks", 0))
+        if pending_publications or maintenance_ranks:
+            lines.append(
+                "sparkcache: publication_work"
+                f" pending_rank_slots={pending_publications}"
+                f" oldest={float(reduced.get('sparkcache_publication_oldest_pending_ms', 0.0)):.0f}ms"
+                f" maintenance_ranks={maintenance_ranks}"
+            )
         return tuple(lines)
 
     def is_empty(self) -> bool:
         return not self.data.get("reports")
 
 
 class SparkCachePromMetrics(KVConnectorPromMetrics):
-    """Prometheus gauges for asynchronous capture-page ownership."""
+    """Prometheus gauges for reported capture, publication, and maintenance work."""
 
     _GAUGES = {
+        "sparkcache_publication_pending_rank_slots": (
+            "vllm:sparkcache_publication_pending_rank_slots",
+            "Pending saver admissions summed across physical ranks at their last worker reports.",
+            1.0,
+        ),
+        "sparkcache_publication_oldest_pending_ms": (
+            "vllm:sparkcache_publication_oldest_pending_seconds",
+            "Oldest saver admission age in seconds at the last worker reports.",
+            0.001,
+        ),
+        "sparkcache_maintenance_active_ranks": (
+            "vllm:sparkcache_maintenance_active_ranks",
+            "Physical ranks reporting an active capacity scan or reconciliation at their last worker reports.",
+            1.0,
+        ),
         "sparkcache_capture_delayed_requests": (
             "vllm:sparkcache_capture_delayed_requests",
             "Maximum delayed SparkCache capture requests on any physical rank.",
@@ -971,6 +1014,7 @@ def __init__(
                 " start while persistent cache initialization is unavailable"
             )
         self._async_page_capture_enabled = config.async_page_capture_enabled
+        self._page_snapshot_interval_tokens = config.page_snapshot_interval_tokens
         self._async_page_capture_runtime: Any = None
         self._async_page_capture_settings: Any = None
         self._async_page_capture_eligible: set[str] = set()
@@ -1037,6 +1081,7 @@ def __init__(
         # capacity operation. This lock is never taken by inference callbacks;
         # streaming callbacks enqueue a receipt and wake the janitor instead.
         self._capacity_lock = threading.RLock()
+        self._capacity_maintenance_depth = 0
         self._capacity_commit_queue: "queue.SimpleQueue[tuple[str, Any]]" = (
             queue.SimpleQueue()
         )
@@ -1072,6 +1117,7 @@ def __init__(
         self._store_queue: "queue.SimpleQueue[_StoreSnapshot | _HybridStoreSnapshot | None]" = queue.SimpleQueue()
         self._store_thread: threading.Thread | None = None
         self._store_inflight = 0
+        self._store_pending_started_ns: int | None = None
         self._publication_base_pins: dict[str, EntryKey] = {}
         self._store_accepting = True
         self._load_queue: "queue.SimpleQueue[_QueuedLoad | _QueuedLoadBatch | None]" = queue.SimpleQueue()
@@ -3215,6 +3261,22 @@ def _maintain_capacity_locked(
             policy.max_bytes == 0 or self._capacity_estimated_bytes <= policy.max_bytes
         ):
             return None
+        self._capacity_maintenance_depth = getattr(self, "_capacity_maintenance_depth", 0) + 1
+        try:
+            return self._perform_capacity_maintenance_locked(
+                policy, force=force, wake_worker_on_unsatisfied=wake_worker_on_unsatisfied
+            )
+        finally:
+            self._capacity_maintenance_depth -= 1
+
+    def _perform_capacity_maintenance_locked(
+        self,
+        policy: CapacityPolicy,
+        *,
+        force: bool,
+        wake_worker_on_unsatisfied: bool,
+    ) -> MaintenanceReport | None:
+        """Keep scans and survivor reconciliation within the maintenance activity gauge."""
         try:
             with self._store_cv:
                 protected = tuple(
@@ -5132,6 +5194,12 @@ def _protect_capture_publication_base(self, plan: _ReqPlan) -> _ReqPlan:
         """
         if not plan.base_context_digest:
             return plan
+        interval = getattr(self, "_page_snapshot_interval_tokens", 0)
+        if interval and plan.span_tokens // interval > plan.base_span_tokens // interval:
+            self.counters["publication_periodic_full_capture_selected"] = (
+                self.counters.get("publication_periodic_full_capture_selected", 0) + 1
+            )
+            return replace(plan, base_context_digest="", base_span_tokens=0)
         if self._capacity_lock.acquire(blocking=False):
             try:
                 with self._store_cv:
@@ -5221,6 +5289,7 @@ def wait_for_save(self) -> None:
                     skipped_before_submit = True
                 else:
                     self._store_inflight = 1
+                    self._store_pending_started_ns = time.perf_counter_ns()
             if skipped_before_submit:
                 if self._async_page_capture_enabled:
                     runtime = self._async_page_capture_runtime
@@ -5750,6 +5819,7 @@ def _finish_store(
                 self._held.difference_update(additional_digests)
                 self.counters["store_evicted" if evicted else "store_failed"] += 1
             self._store_inflight = 0
+            self._store_pending_started_ns = None
             self._store_cv.notify_all()
         if error is not None:
             logger.warning(
@@ -6354,6 +6424,18 @@ def get_kv_connector_stats(self):
             return None
         with self._load_lock:
             report = self._build_quorum_report_locked()
+            pending = int(bool(self._store_inflight))
+            started_ns = getattr(self, "_store_pending_started_ns", None)
+            report["publication_work"] = {
+                "pending": pending,
+                "oldest_pending_ms": (
+                    max(0, time.perf_counter_ns() - started_ns) / 1_000_000
+                    if pending and started_ns is not None else 0.0
+                ),
+                # Reading the activity counter must not wait for the capacity
+                # lock held by the scan being measured.
+                "maintenance_active": bool(getattr(self, "_capacity_maintenance_depth", 0)),
+            }
         runtime = self._streaming_runtime
         status = getattr(runtime, "status", None)
         if self._streaming_snapshots_enabled and callable(status):