Skip to content

Validate recurrent proof at publication time #38

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
FujitsuPolycom wants to merge 6 commits into from
Closed

Validate recurrent proof at publication time #38

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
37bbc60
Validate recurrent proof at publication time
FujitsuPolycom Aug 30, 2026
99e101d
Latch every proven recurrent CoW target
FujitsuPolycom Aug 30, 2026
972b203
Accept the boundary-crossing producer postimage
FujitsuPolycom Aug 30, 2026
bf7174e
Retain the exact-stop recurrent producer contract
FujitsuPolycom Aug 30, 2026
bd3eec1
Ignore proven earlier recurrent checkpoints
FujitsuPolycom Aug 30, 2026
c56f77f
Log future recurrent proof identity
FujitsuPolycom Aug 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion deploy/deepseek_v4/tp4_profile.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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": "83853050f790b18af95d424fec837abeb1a9a33f0538b5e4b97c16fb9c681781"
"source_sha256": "788686e858ba4af01f535e95122c7650f412fddc40cd221a0924f4ce2b32ff98"
},
"model": {
"repository": "deepseek-ai/DeepSeek-V4-Flash-0731",
Expand Down
2 changes: 1 addition & 1 deletion deploy/glm52_35bpw/profile.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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": "83853050f790b18af95d424fec837abeb1a9a33f0538b5e4b97c16fb9c681781"
"source_sha256": "788686e858ba4af01f535e95122c7650f412fddc40cd221a0924f4ce2b32ff98"
},
"model": {
"repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",
Expand Down
21 changes: 14 additions & 7 deletions sparkcache/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,13 +225,20 @@ when placement completes and intentionally excludes that bookkeeping.
`sparkcache-hybrid-page-delta/v1` codec reuses only byte-identical page
prefixes and binds the base snapshot, layout, block counts, and semantic
token boundaries. A boundary inside an HMA page replaces that complete page
while retaining earlier byte-identical pages. For an aligned recurrent group,
vLLM may retain the replay-boundary page outside the advancing request block
table. Its `SchedulerOutput.recurrent_boundary_blocks` hand-off names the
pinned physical block by request, group, and token boundary. SparkCache uses
that block only after all three identities and the recurrent topology match;
missing or contradictory metadata skips publication rather than scanning
later running or speculative state. The
while retaining earlier byte-identical pages. At an exact recurrent-page
boundary, vLLM may retain the replay-boundary page outside the advancing
request block table. Its `SchedulerOutput.recurrent_boundary_blocks` hand-off
names the pinned physical block by request, group, and token boundary.
SparkCache defers a new recurrent request until a later cached scheduler step,
when the preceding forward's hand-off can be observed. It latches one matching
entry for every recurrent group, including a partial-tail CoW target when the
boundary lies inside a recurrent page. Valid entries for an earlier checkpoint
are ignored while the request advances; outputs with no target-boundary entry
leave publication pending. Incomplete, future, contradictory, or changed
target evidence cancels it. A store is emitted only after every recurrent
group has a proven pinned block at the exact publication boundary. SparkCache
never substitutes an accumulated request-table ID because vLLM may have
replaced that source block while producing the durable CoW target. The
`sparkcache-page-delta-manifest/v2` schema embeds its authenticated base graph
and groups delta bytes into immutable objects of at most 64 MiB. A
1,575,821,491-byte delta therefore uses at most 24 physical delta objects
Expand Down
181 changes: 110 additions & 71 deletions sparkcache/spark_context_cache_connector.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -570,6 +570,11 @@ def __init__(
self._storage_mode = config.storage_mode
self._publication_schema = config.publication_schema
self._group_topology = config.group_topology
self._recurrent_group_indexes = frozenset(
group_index
for group_index, topology in enumerate(self._group_topology)
if topology["reuse_policy"] == "recurrent_align"
)
self._chunk_tokens = config.chunk_tokens
self._root = config.root
self._store = ManifestStore(self._root)
Expand Down Expand Up @@ -1613,15 +1618,21 @@ def _validated_recurrent_boundary_blocks(
scheduler_output: "SchedulerOutput",
request_id: str,
boundary_tokens: int,
*,
latched: tuple[tuple[int, int], ...] = (),
) -> tuple[tuple[int, int], ...] | None:
"""Validate vLLM's exact recurrent replay-boundary block hand-off.

An empty tuple is valid only when the registered topology has no
aligned recurrent group. None means the metadata is absent,
incomplete, or contradictory, so publication must be skipped.
SparkCache never derives a replacement from another non-null table
entry because later entries can hold running or speculative state
beyond ``boundary_tokens``.
vLLM may expose an earlier aligned checkpoint while the request is
still advancing toward this store boundary, followed by a partial-tail
CoW target on a later scheduler output. Valid older entries and absent
per-request metadata therefore preserve ``latched`` and keep the store
pending. None means supplied metadata is incomplete, malformed, ahead
of the plan, or conflicts with an earlier same-boundary latch, so this
publication attempt must be poisoned. SparkCache never derives a
replacement from another request-table entry because it can name
overwritten running or speculative state instead of vLLM's pinned CoW
target.
"""

def reject(reason: str) -> None:
Expand All @@ -1635,26 +1646,24 @@ def reject(reason: str) -> None:
)
return None

required_groups = {
group_index
for group_index, topology in enumerate(self._group_topology)
if topology["reuse_policy"] == "recurrent_align"
}
required_groups = self._recurrent_group_indexes
if not required_groups:
return ()
raw = getattr(scheduler_output, "recurrent_boundary_blocks", None)
if raw is None:
return reject("vLLM supplied no recurrent boundary mapping")
return latched
if not isinstance(raw, Mapping):
return reject("top-level value is not a mapping")
entries = raw.get(request_id)
if entries is None:
return reject("request has no recurrent boundary entries")
return latched
if not isinstance(entries, (list, tuple)):
return reject("request value is not a sequence")
if not entries:
return reject("request has no recurrent boundary entries")

overrides: list[tuple[int, int]] = []
seen_groups: set[int] = set()
seen_target_groups: set[int] = set()
for entry in entries:
if not isinstance(entry, (list, tuple)) or len(entry) != 3:
return reject("entry is not a group, block, boundary triple")
Expand All @@ -1663,20 +1672,31 @@ def reject(reason: str) -> None:
return reject("entry values are not integers")
if not 0 <= group_index < len(self._group_topology):
return reject("group index is outside the registered topology")
if group_index in seen_groups:
return reject("multiple blocks claim the same recurrent group")
topology = self._group_topology[group_index]
if topology["reuse_policy"] != "recurrent_align":
return reject("group is not an aligned recurrent cache")
if block_id <= 0:
return reject("physical block is vLLM's null block")
if entry_boundary != boundary_tokens:
return reject("entry boundary differs from the store plan")
seen_groups.add(group_index)
if entry_boundary < boundary_tokens:
continue
if entry_boundary > boundary_tokens:
return reject(
"entry boundary is ahead of the store plan"
f" observed={entry_boundary} target={boundary_tokens}"
f" group={group_index} block={block_id}"
)
if group_index in seen_target_groups:
return reject("multiple blocks claim the same recurrent group")
seen_target_groups.add(group_index)
overrides.append((group_index, block_id))
if seen_groups != required_groups:
if not overrides:
return latched
if seen_target_groups != required_groups:
return reject("entries do not cover every aligned recurrent group")
return tuple(sorted(overrides))
validated = tuple(sorted(overrides))
if latched and validated != latched:
return reject("entries conflict with the latched recurrent boundary")
return validated

def build_connector_meta(
self, scheduler_output: "SchedulerOutput"
Expand Down Expand Up @@ -1741,15 +1761,6 @@ def build_connector_meta(
if self._has_full_quorum(digest):
self.counters["store_skipped_quorum"] += 1
continue
recurrent_boundary_blocks = (
self._validated_recurrent_boundary_blocks(
scheduler_output,
req_id,
span,
)
)
if recurrent_boundary_blocks is None:
continue
already = new_req.num_computed_tokens + scheduled
if self._streaming_snapshots_enabled:
self._append_streaming_snapshot_offer(
Expand All @@ -1771,10 +1782,34 @@ def build_connector_meta(
[list(group) for group in group_blocks],
)
self._store_token_ids[req_id] = exact_token_ids
if recurrent_boundary_blocks:
self._store_recurrent_boundaries[req_id] = (
recurrent_boundary_blocks
)
elif self._recurrent_group_indexes:
recurrent_boundary_blocks = (
self._validated_recurrent_boundary_blocks(
scheduler_output,
req_id,
span,
)
)
if recurrent_boundary_blocks is None:
continue
# Full-page proof and partial-tail CoW hand-offs can arrive
# after the prefill which began this store. Retain the
# complete request table and any early proof until a later
# cached step has both finished the span and proven every
# recurrent group.
self._store_progress[req_id] = (
digest,
span,
already,
[list(group) for group in group_blocks],
)
self._store_token_ids[req_id] = exact_token_ids
if recurrent_boundary_blocks:
self._store_recurrent_boundaries[req_id] = (
recurrent_boundary_blocks
)
if base_digest:
self._store_bases[req_id] = (base_digest, base_span)
elif already >= span:
meta.plans.append(
_ReqPlan(
Expand All @@ -1787,9 +1822,6 @@ def build_connector_meta(
token_ids=exact_token_ids,
base_context_digest=base_digest,
base_span_tokens=base_span,
recurrent_boundary_blocks=(
recurrent_boundary_blocks
),
)
)
else:
Expand All @@ -1803,10 +1835,6 @@ def build_connector_meta(
[list(group) for group in group_blocks],
)
self._store_token_ids[req_id] = exact_token_ids
if recurrent_boundary_blocks:
self._store_recurrent_boundaries[req_id] = (
recurrent_boundary_blocks
)
if base_digest:
self._store_bases[req_id] = (base_digest, base_span)
cached = scheduler_output.scheduled_cached_reqs
Expand All @@ -1816,10 +1844,18 @@ def build_connector_meta(
digest, span, done, blocks_by_group = self._store_progress[req_id]
exact_token_ids = self._store_token_ids.get(req_id, ())
base_digest, base_span = self._store_bases.get(req_id, ("", 0))
if self._has_full_quorum(digest):
del self._store_progress[req_id]
self._store_token_ids.pop(req_id, None)
self._store_bases.pop(req_id, None)
self._store_recurrent_boundaries.pop(req_id, None)
self.counters["store_skipped_quorum"] += 1
continue
recurrent_boundary_blocks = self._validated_recurrent_boundary_blocks(
scheduler_output,
req_id,
span,
latched=self._store_recurrent_boundaries.get(req_id, ()),
)
if recurrent_boundary_blocks is None:
del self._store_progress[req_id]
Expand All @@ -1829,30 +1865,31 @@ def build_connector_meta(
continue
if recurrent_boundary_blocks:
self._store_recurrent_boundaries[req_id] = recurrent_boundary_blocks
new_block_ids = cached.new_block_ids[index]
appended = (
[
list(group)
for group in self._normalize_group_blocks(
new_block_ids,
allow_empty_groups=True,
)
]
if new_block_ids is not None
else [[] for _ in blocks_by_group]
)
if len(appended) != len(blocks_by_group):
raise RuntimeError(
"spark-context-cache: KV-cache group count changed while"
" accumulating a store"
if done < span or req_id in cached.resumed_req_ids:
new_block_ids = cached.new_block_ids[index]
appended = (
[
list(group)
for group in self._normalize_group_blocks(
new_block_ids,
allow_empty_groups=True,
)
]
if new_block_ids is not None
else [[] for _ in blocks_by_group]
)
if req_id in cached.resumed_req_ids:
blocks_by_group = appended
else:
blocks_by_group = [
existing + added
for existing, added in zip(blocks_by_group, appended)
]
if len(appended) != len(blocks_by_group):
raise RuntimeError(
"spark-context-cache: KV-cache group count changed while"
" accumulating a store"
)
if req_id in cached.resumed_req_ids:
blocks_by_group = appended
else:
blocks_by_group = [
existing + added
for existing, added in zip(blocks_by_group, appended)
]
blocks = blocks_by_group[0]
done = cached.num_computed_tokens[index] + (
scheduler_output.num_scheduled_tokens.get(req_id, 0)
Expand Down Expand Up @@ -1882,16 +1919,18 @@ def build_connector_meta(
block_ids=blocks,
)
elif done >= span:
if self._recurrent_group_indexes and not recurrent_boundary_blocks:
self._store_progress[req_id] = (
digest,
span,
done,
blocks_by_group,
)
continue
del self._store_progress[req_id]
self._store_token_ids.pop(req_id, None)
self._store_bases.pop(req_id, None)
recurrent_boundary_blocks = self._store_recurrent_boundaries.pop(
req_id,
(),
)
if self._has_full_quorum(digest):
self.counters["store_skipped_quorum"] += 1
continue
self._store_recurrent_boundaries.pop(req_id, None)
normalized = tuple(tuple(group) for group in blocks_by_group)
meta.plans.append(
_ReqPlan(
Expand Down
Loading