diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json index 7935530..a6a7f07 100644 --- a/deploy/deepseek_v4/tp4_profile.json +++ b/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", diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json index 4262360..0bbbfb9 100644 --- a/deploy/glm52_35bpw/profile.json +++ b/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", diff --git a/sparkcache/README.md b/sparkcache/README.md index 1c669eb..768f7de 100644 --- a/sparkcache/README.md +++ b/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 diff --git a/sparkcache/spark_context_cache_config.py b/sparkcache/spark_context_cache_config.py index 0727762..2d57eb4 100644 --- a/sparkcache/spark_context_cache_config.py +++ b/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, diff --git a/sparkcache/spark_context_cache_connector.py b/sparkcache/spark_context_cache_connector.py index 15e152d..ec00a7f 100644 --- a/sparkcache/spark_context_cache_connector.py +++ b/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,6 +759,15 @@ 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: @@ -747,9 +775,24 @@ def is_empty(self) -> bool: 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): diff --git a/sparkcache/test_periodic_page_snapshot.py b/sparkcache/test_periodic_page_snapshot.py new file mode 100644 index 0000000..0e6fc51 --- /dev/null +++ b/sparkcache/test_periodic_page_snapshot.py @@ -0,0 +1,138 @@ +"""Opt-in token-bucket selection of complete manager-page captures.""" + +from dataclasses import replace +from pathlib import Path + +import pytest + +from sparkcache import spark_context_cache_config as config +from sparkcache.test_spark_context_cache_config import _make_vllm_config +from sparkcache import test_publication_base_retention as retention_tests +from sparkcache.test_async_page_capture_connector import FakeSparseRing +from sparkcache.spark_context_cache_connector import SparkCacheConnectorMetadata +from sparkcache.streaming.manager_page_runtime import ManagerPageCaptureRuntime +from sparkcache.spark_context_cache_hybrid import PageGroup, PageLayer, PageLayout, encode_page_snapshot + + +KEY = "spark_cache_page_snapshot_interval_tokens" +ENV = "SPARK_CONTEXT_CACHE_PAGE_SNAPSHOT_INTERVAL_TOKENS" +retained_base = retention_tests.retained_base + + +def parse(extra=None): + vllm, transfer = _make_vllm_config(extra) + return config.parse_connector_config(vllm, transfer, None) + + +def test_interval_defaults_off_and_does_not_change_cache_identity(monkeypatch): + monkeypatch.delenv(ENV, raising=False) + disabled = parse() + enabled = parse({KEY: 16384}) + assert disabled.page_snapshot_interval_tokens == 0 + assert enabled.page_snapshot_interval_tokens == 16384 + assert disabled.build_identity(0, 0).to_wire() == enabled.build_identity(0, 0).to_wire() + + +def test_interval_extra_config_overrides_environment(monkeypatch): + monkeypatch.setenv(ENV, "16384") + assert parse().page_snapshot_interval_tokens == 16384 + assert parse({KEY: 8192}).page_snapshot_interval_tokens == 8192 + assert parse({KEY: 0}).page_snapshot_interval_tokens == 0 + + +@pytest.mark.parametrize("value", [-1, True, False, 1.5, "1.5", "bad", None]) +def test_interval_rejects_invalid_settings(value): + with pytest.raises(RuntimeError, match=KEY): + parse({KEY: value}) + + +@pytest.mark.parametrize("interval,base,result,full", [ + (0, 14336, 16384, False), + (16384, 12288, 14336, False), + (16384, 14336, 16384, True), + (16384, 16384, 18432, False), + (16384, 14336, 49152, True), +]) +def test_periodic_selection_crosses_buckets_without_history_reads( + retained_base, monkeypatch, interval, base, result, full +): + connector, runtime, original, *_ = retained_base + connector._page_snapshot_interval_tokens = interval + plan = replace(original, base_span_tokens=base, span_tokens=result) + connector.bind_connector_metadata(SparkCacheConnectorMetadata(plans=[plan])) + monkeypatch.setattr(Path, "read_bytes", lambda *args: pytest.fail("capture policy read the filesystem")) + connector.wait_for_save() + submitted = runtime.submitted[0][0] + assert submitted.base_context_digest == ("" if full else plan.base_context_digest) + assert submitted.base_span_tokens == (0 if full else base) + assert submitted.token_ids == plan.token_ids + assert submitted.digest == plan.digest + assert connector.counters.get("publication_periodic_full_capture_selected", 0) == int(full) + assert connector.counters.get("publication_base_full_capture_fallback", 0) == 0 + assert bool(connector._publication_base_pins) is (not full) + + +def test_periodic_capture_submits_complete_groups_and_respects_ring_rejection(retained_base): + connector, _runtime, plan, *_ = retained_base + plan = replace(plan, block_ids_by_group=((0, 1), (2, 3, 4))) + connector.bind_connector_metadata(SparkCacheConnectorMetadata(plans=[plan])) + connector._page_layout = PageLayout(( + PageGroup(256, (PageLayer("attention", "u8", (64,), 64),)), + PageGroup(1, (PageLayer("recurrent", "u8", (8,), 8),)), + )) + connector._page_snapshot_interval_tokens = 512 + connector._select_group_blocks_for_span = lambda groups, *args, **kw: groups + submitted = [] + + class BoundedRing: + active_ticket_count = 0 + + def submit(self, **kwargs): + submitted.append(kwargs) + return None + + def shutdown(self): + return None + + runtime = ManagerPageCaptureRuntime(connector, ring=BoundedRing(), + progress_thread_initializer=lambda: None) + connector._async_page_capture_runtime = runtime + connector.wait_for_save() + assert submitted[0]["logical_start"] == 0 + assert submitted[0]["physical_pages_by_group"] == plan.group_block_ids + assert connector._store_inflight == 0 + assert not connector._publication_base_pins + assert connector.counters["async_page_capture_aborted"] == 1 + assert connector.counters["publication_periodic_full_capture_selected"] == 1 + assert runtime.take_finished({plan.request_id}) == {plan.request_id} + + +def test_already_complete_snapshot_is_not_counted_as_periodic(retained_base): + connector, runtime, plan, *_ = retained_base + connector._page_snapshot_interval_tokens = 512 + plan = replace(plan, base_context_digest="", base_span_tokens=0) + connector.bind_connector_metadata(SparkCacheConnectorMetadata(plans=[plan])) + connector.wait_for_save() + assert runtime.submitted[0][0] == plan + assert connector.counters.get("publication_periodic_full_capture_selected", 0) == 0 + + +def test_periodic_full_capture_publishes_independent_existing_format(retained_base, monkeypatch): + connector, _runtime, plan, identity, layout, *_ = retained_base + connector._page_snapshot_interval_tokens = 512 + connector._select_group_blocks_for_span = lambda groups, *args, **kw: groups + ring = FakeSparseRing(b"a" * 128) + ring.ready.set() + connector._async_page_capture_runtime = ManagerPageCaptureRuntime( + connector, ring=ring, progress_poll_seconds=0.001, + progress_thread_initializer=lambda: None) + monkeypatch.setattr(connector._store, "commit_page_extension", + lambda **kw: pytest.fail("periodic capture used a dependent publication")) + connector.wait_for_save() + assert connector.wait_for_pending_stores(timeout=5) + lookup = connector._store.lookup(identity, plan.digest, verify_chunks=False) + assert lookup.is_hit and lookup.root_kind == "page_snapshot" + restored = connector._store.restore_page_snapshot(lookup, layout=layout, + result_block_counts=(2,), result_boundary_tokens=512) + assert restored == encode_page_snapshot(layout, (2,), {"page": b"a" * 128}) + assert connector.counters["publication_periodic_full_capture_selected"] == 1 diff --git a/sparkcache/test_publication_work_telemetry.py b/sparkcache/test_publication_work_telemetry.py new file mode 100644 index 0000000..40ec970 --- /dev/null +++ b/sparkcache/test_publication_work_telemetry.py @@ -0,0 +1,184 @@ +"""Publication age and maintenance activity remain observable without I/O.""" + +from pathlib import Path +from types import SimpleNamespace +import threading + +import pytest + +from sparkcache import test_publication_base_retention as retention_tests +from sparkcache import spark_context_cache_connector as connector_module +from sparkcache.persistent_context_cache.cache_manifest import MaintenanceReport + + +retained_base = retention_tests.retained_base + + +def work_report(connector): + return connector.get_kv_connector_stats().data["reports"][0]["publication_work"] + + +def test_admission_age_includes_capture_and_clears_after_commit(retained_base, monkeypatch): + connector, _runtime, plan, *_ = retained_base + now = [1_000_000_000] + monkeypatch.setattr(connector_module.time, "perf_counter_ns", lambda: now[0]) + connector.wait_for_save() + now[0] += 125_000_000 + monkeypatch.setattr(Path, "read_bytes", lambda *args: pytest.fail("metrics read the filesystem")) + assert work_report(connector) == {"pending": 1, "oldest_pending_ms": 125.0, "maintenance_active": False} + connector._finish_store(plan.digest, committed=True) + now[0] += 1_000_000_000 + assert work_report(connector) == {"pending": 0, "oldest_pending_ms": 0.0, "maintenance_active": False} + + +@pytest.mark.parametrize("terminal", ["failure", "abort", "shutdown"]) +def test_terminal_publication_clears_pending_age(retained_base, monkeypatch, terminal): + connector, runtime, plan, *_ = retained_base + now = [1_000_000_000] + monkeypatch.setattr(connector_module.time, "perf_counter_ns", lambda: now[0]) + connector.wait_for_save() + now[0] += 2_000_000_000 + assert work_report(connector)["oldest_pending_ms"] == 2000.0 + if terminal == "failure": + connector._finish_store(plan.digest, committed=False, error=ValueError("publication failed")) + elif terminal == "abort": + connector._abort_async_page_capture(plan.digest, "capture aborted") + else: + runtime.quiesce = lambda: connector._abort_async_page_capture(plan.digest, "shutdown drain") + connector.shutdown() + assert connector._store_pending_started_ns is None + assert work_report(connector)["pending"] == 0 + assert work_report(connector)["oldest_pending_ms"] == 0.0 + + +@pytest.mark.parametrize("failed", [False, True]) +def test_maintenance_activity_includes_reconciliation_and_clears(retained_base, monkeypatch, failed): + connector, *_ = retained_base + stages = [] + + def maintain(*args, **kwargs): + assert work_report(connector)["maintenance_active"] is True + stages.append("scan") + if failed: + raise OSError("scan failed") + return MaintenanceReport(capacity_satisfied=True) + + def reconcile(): + assert work_report(connector)["maintenance_active"] is True + stages.append("reconcile") + + monkeypatch.setattr(connector._store, "maintain", maintain) + monkeypatch.setattr(connector, "_reconcile_held_capacity", reconcile) + connector._maintain_capacity(force=True) + assert stages == ["scan", "reconcile"] + assert work_report(connector)["maintenance_active"] is False + + +def test_reconciliation_exception_clears_maintenance_activity(retained_base, monkeypatch): + connector, *_ = retained_base + monkeypatch.setattr(connector._store, "maintain", lambda *args, **kw: MaintenanceReport()) + monkeypatch.setattr(connector, "_reconcile_held_capacity", + lambda: (_ for _ in ()).throw(RuntimeError("reconciliation failed"))) + with pytest.raises(RuntimeError, match="reconciliation failed"): + connector._maintain_capacity(force=True) + assert work_report(connector)["maintenance_active"] is False + + +def test_report_does_not_wait_for_a_blocked_capacity_scan(retained_base, monkeypatch): + connector, *_ = retained_base + entered, release, reported = threading.Event(), threading.Event(), threading.Event() + observed = [] + + def maintain(*args, **kwargs): + entered.set() + assert release.wait(timeout=5) + return MaintenanceReport(capacity_satisfied=True) + + def report(): + observed.append(work_report(connector)) + reported.set() + + monkeypatch.setattr(connector._store, "maintain", maintain) + maintenance = threading.Thread(target=lambda: connector._maintain_capacity(force=True)) + reader = threading.Thread(target=report) + maintenance.start() + try: + assert entered.wait(timeout=2) + reader.start() + assert reported.wait(timeout=0.5), "metrics waited behind capacity maintenance" + assert observed[0]["maintenance_active"] is True + finally: + release.set() + maintenance.join(timeout=2) + if reader.ident is not None: + reader.join(timeout=2) + + +def test_shutdown_timeout_does_not_hide_pending_publication(retained_base, monkeypatch): + connector, _runtime, _plan, *_ = retained_base + now = [1_000_000_000] + monkeypatch.setattr(connector_module.time, "perf_counter_ns", lambda: now[0]) + connector.wait_for_save() + monkeypatch.setattr(connector, "wait_for_pending_stores", lambda timeout: False) + connector._store_thread = SimpleNamespace(is_alive=lambda: True, join=lambda **kw: None) + now[0] += 2_000_000_000 + connector.shutdown() + assert work_report(connector)["pending"] == 1 + assert work_report(connector)["oldest_pending_ms"] == 2000.0 + + +def test_existing_report_age_is_a_snapshot_until_worker_reports_again(retained_base, monkeypatch): + connector, _runtime, _plan, *_ = retained_base + now = [1_000_000_000] + monkeypatch.setattr(connector_module.time, "perf_counter_ns", lambda: now[0]) + connector.wait_for_save() + now[0] += 1_000_000_000 + snapshot = connector.get_kv_connector_stats() + now[0] += 2_000_000_000 + assert snapshot.reduce()["sparkcache_publication_oldest_pending_ms"] == 1000.0 + assert work_report(connector)["oldest_pending_ms"] == 3000.0 + + +def test_work_reduction_sums_rank_slots_takes_max_age_and_preserves_reports(): + incoming = {"rank": 1, "held_count": 2, + "publication_work": {"pending": 1, "oldest_pending_ms": 2500, "maintenance_active": True}} + stats = connector_module.SparkCacheStats(data={"reports": [ + {"rank": 0, "held_count": 3, + "publication_work": {"pending": 1, "oldest_pending_ms": 1250, "maintenance_active": False}}, + ]}) + stats.aggregate(connector_module.SparkCacheStats(data={"reports": [incoming]})) + incoming["publication_work"]["pending"] = 0 + reduced = stats.reduce() + assert reduced["sparkcache_entries"] == 5 + assert reduced["sparkcache_publication_pending_rank_slots"] == 2 + assert reduced["sparkcache_publication_oldest_pending_ms"] == 2500 + assert reduced["sparkcache_maintenance_active_ranks"] == 1 + assert any("publication_work" in line for line in stats.format_log_lines()) + legacy = connector_module.SparkCacheStats(data={"reports": [{"rank": 2, "held": []}]}) + assert legacy.reduce() == {"sparkcache_ranks": 1, "sparkcache_entries": 0} + + +def test_prometheus_work_gauges_convert_age_to_seconds_and_reset(): + gauges = {} + + class Gauge: + def __init__(self, *, name, **kwargs): + gauges[name] = self + + def labels(self, *args): + return self + + def set(self, value): + self.value = value + + metrics = connector_module.SparkCachePromMetrics(SimpleNamespace(), {object: Gauge}, [], {0: []}) + metrics.observe({"reports": [{"rank": 0, "held": [], "publication_work": { + "pending": 1, "oldest_pending_ms": 1250, "maintenance_active": True, + }}]}) + assert gauges["vllm:sparkcache_publication_pending_rank_slots"].value == 1 + assert gauges["vllm:sparkcache_publication_oldest_pending_seconds"].value == 1.25 + assert gauges["vllm:sparkcache_maintenance_active_ranks"].value == 1 + metrics.observe({"reports": [{"rank": 0, "held": []}]}) + assert gauges["vllm:sparkcache_publication_pending_rank_slots"].value == 0 + assert gauges["vllm:sparkcache_publication_oldest_pending_seconds"].value == 0 + assert gauges["vllm:sparkcache_maintenance_active_ranks"].value == 0