From b37a2e1a38c4941da3b4d1cb72d0931eb48948e9 Mon Sep 17 00:00:00 2001 From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com> Date: Sun, 30 Aug 2026 00:54:57 -0500 Subject: [PATCH] Group page deltas into authenticated macro objects Per-file overhead dominated the 256K restore because physical delta objects followed the 256-token logical boundary. Page-delta manifest v2 stores ordered 64-MiB authenticated extents and reads or publishes them in bounded batches. Version 1 remains readable; cache identity, digest salts, logical chunk geometry, page sharing, and the page-tail namespace are unchanged. GPU-free tests cover exact restore, corruption, bounded memory, file count, compatibility, and capacity behavior. --- README.md | 11 + deploy/deepseek_v4/tp4_profile.json | 2 +- deploy/glm52_35bpw/profile.json | 2 +- sparkcache/README.md | 19 +- .../cache_manifest.py | 284 ++++++++++---- .../test_cache_manifest.py | 4 +- .../test_page_delta_macro_objects.py | 346 ++++++++++++++++++ sparkcache/test_defect_regressions.py | 11 +- .../test_spark_context_cache_connector.py | 2 +- 9 files changed, 586 insertions(+), 95 deletions(-) create mode 100644 sparkcache/persistent_context_cache/test_page_delta_macro_objects.py diff --git a/README.md b/README.md index b947429..004b12c 100644 --- a/README.md +++ b/README.md @@ -298,6 +298,17 @@ byte-identical opaque pages. Restore reconstructs and verifies the complete snapshot before Python or native page placement. Arbitrary earlier-prefix aliases cannot be derived from opaque page snapshots. +Page-delta publication writes `sparkcache-page-delta-manifest/v2` metadata over +authenticated byte extents of at most 64 MiB. This physical grouping reduces +the 1,024 delta files implied by a 262,144-token logical boundary to at most 24 +objects for a 1,575,821,491-byte delta. Reads retain at most four extent +payloads in addition to one assembled delta buffer. The logical admission and +digest boundary remains 256 tokens. Version 1 page-delta manifests remain +readable; cache identity, digest salts, and the `page-tail-cow-v1` namespace do +not change. Restore still materializes one authenticated delta buffer and the +verified reconstructed snapshot before placement. Direct placement from base +and delta extents is unsupported by this schema. + Opaque HMA snapshots cannot be shortened by truncating chunk lists. SparkCache therefore uses the page-semantic format and distinct namespace described above. At most two page deltas may form one graph; the following publication compacts diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json index 7377823..45a2848 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": "bc238f96e550c7ec27d4081dd1f2e741d404aaf5c8572d89ccc5e76812be4d63" + "source_sha256": "9b8b2a6863d91f07354dab67d608cc15f551f1a5a7682b89873c7ae6ba468ee5" }, "model": { "repository": "deepseek-ai/DeepSeek-V4-Flash-0731", diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json index 14dc752..30fa0a5 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": "bc238f96e550c7ec27d4081dd1f2e741d404aaf5c8572d89ccc5e76812be4d63" + "source_sha256": "9b8b2a6863d91f07354dab67d608cc15f551f1a5a7682b89873c7ae6ba468ee5" }, "model": { "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78", diff --git a/sparkcache/README.md b/sparkcache/README.md index b48f193..b41fdc4 100644 --- a/sparkcache/README.md +++ b/sparkcache/README.md @@ -226,12 +226,19 @@ when placement completes and intentionally excludes that bookkeeping. 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. The - `sparkcache-page-delta-manifest/v1` schema embeds its authenticated base - graph, allowing capacity maintenance to retain shared objects after - predecessor roots are removed. Restore reconstructs the verified full - snapshot before Python or native page placement. GPU-free regression coverage - exists; live model-serving qualification does not. A graph contains at most - two deltas. The following extension publishes a fresh flat snapshot, bounding + `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 + instead of 1,024 objects derived from logical token chunks. Ordered restore + batches retain at most four object payloads in addition to one assembled + delta buffer. Version 1 manifests remain readable. Cache identity, digest + salts, the 256-token logical boundary, and the `page-tail-cow-v1` namespace + are unchanged. Capacity maintenance retains shared objects after predecessor + roots are removed. Restore reconstructs the verified full snapshot before + Python or native page placement. GPU-free regression coverage exists; live + model-serving qualification does not. Direct placement from base and delta + extents is unsupported by this schema. A graph contains at most two deltas. + The following extension publishes a fresh flat snapshot, bounding reconstruction work and metadata ancestry. - **Concurrent shared GPU prefix — implemented.** One leader restores a persistent digest. After every rank succeeds, up to sixteen waiting followers diff --git a/sparkcache/persistent_context_cache/cache_manifest.py b/sparkcache/persistent_context_cache/cache_manifest.py index 764aa63..9946de4 100644 --- a/sparkcache/persistent_context_cache/cache_manifest.py +++ b/sparkcache/persistent_context_cache/cache_manifest.py @@ -42,6 +42,18 @@ _PREFIX_ALIAS_SCHEMA = "sparkcache-prefix-alias/v1" _TAIL_MANIFEST_SCHEMA = "sparkcache-tail-manifest/v1" _PAGE_DELTA_MANIFEST_SCHEMA = "sparkcache-page-delta-manifest/v1" +_PAGE_DELTA_MANIFEST_SCHEMA_V2 = "sparkcache-page-delta-manifest/v2" +_PAGE_DELTA_MANIFEST_SCHEMAS = frozenset( + (_PAGE_DELTA_MANIFEST_SCHEMA, _PAGE_DELTA_MANIFEST_SCHEMA_V2) +) +# The v2 physical geometry is independent of the 256-token digest and +# admission boundary. A 64-MiB extent reduces a 1.58-GB delta to 24 objects; +# bounded batches cap temporary payload bytes at 128 MiB while publishing and +# 256 MiB while reading, in addition to the assembled authenticated delta. +_PAGE_DELTA_OBJECT_BYTES = 64 * 1024 * 1024 +_MAX_PAGE_DELTA_OBJECT_BYTES = 64 * 1024 * 1024 +_PAGE_DELTA_WRITE_BATCH_SIZE = 2 +_PAGE_DELTA_READ_BATCH_SIZE = 4 # Two delta roots cap reconstruction at two full-snapshot applications. A # following extension is compacted by the connector into a fresh flat root. _MAX_PAGE_DELTA_DEPTH = 2 @@ -140,6 +152,13 @@ class PageDeltaDepthExceeded(ValueError): """Another page delta would exceed the bounded reconstruction depth.""" +def _is_page_delta_root(value: Any) -> bool: + return ( + isinstance(value, Mapping) + and value.get("schema") in _PAGE_DELTA_MANIFEST_SCHEMAS + ) + + @dataclass(frozen=True) class CacheIdentity: target_checkpoint: str @@ -1139,6 +1158,19 @@ def _validate_page_delta_root( ) -> tuple[Mapping[str, Any], tuple[Mapping[str, Any], ...]]: if not isinstance(manifest, dict): raise CacheFormatError("page delta manifest is not an object") + schema = manifest.get("schema") + if schema == _PAGE_DELTA_MANIFEST_SCHEMA: + delta_keys = {"delta_chunks"} + elif schema == _PAGE_DELTA_MANIFEST_SCHEMA_V2: + delta_keys = { + "delta_encoded_bytes", + "delta_object_bytes", + "delta_objects", + "delta_sha256", + "logical_chunk_tokens", + } + else: + raise _IncompatibleManifestError("page delta manifest schema differs") _strict_keys( manifest, { @@ -1154,8 +1186,8 @@ def _validate_page_delta_root( "layout_sha256", "base_block_counts", "result_block_counts", - "delta_chunks", "metadata_sha256", + *delta_keys, }, "page delta manifest", ) @@ -1171,8 +1203,7 @@ def _validate_page_delta_root( if _sha256(_canonical_json(authenticated)) != metadata_digest: raise CacheFormatError("page delta metadata checksum mismatch") if ( - manifest["schema"] != _PAGE_DELTA_MANIFEST_SCHEMA - or manifest["format_abi"] != FORMAT_ABI + manifest["format_abi"] != FORMAT_ABI or identity.publication_schema != "page-tail-cow-v1" or manifest["identity"] != identity.to_wire() or manifest["context_digest"] != context_digest @@ -1191,20 +1222,71 @@ def _validate_page_delta_root( or len(manifest["base_block_counts"]) != len(manifest["result_block_counts"]) ): raise CacheFormatError("page delta manifest geometry differs") - delta_chunks = manifest["delta_chunks"] - synthetic = { - "format_abi": FORMAT_ABI, - "identity": identity.to_wire(), - "context_digest": context_digest, - "committed_tokens": manifest["committed_tokens"], - "chunks": delta_chunks, - } - descriptors = _validate_manifest_metadata( - synthetic, - EntryKey(identity.storage_key, context_digest), - expected_identity=identity, - ) - return manifest["base_root"], descriptors + if schema == _PAGE_DELTA_MANIFEST_SCHEMA: + synthetic = { + "format_abi": FORMAT_ABI, + "identity": identity.to_wire(), + "context_digest": context_digest, + "committed_tokens": manifest["committed_tokens"], + "chunks": manifest["delta_chunks"], + } + descriptors = _validate_manifest_metadata( + synthetic, + EntryKey(identity.storage_key, context_digest), + expected_identity=identity, + ) + return manifest["base_root"], descriptors + + try: + _validate_digest(manifest["delta_sha256"], "page delta payload sha256") + except ValueError as error: + raise CacheFormatError(str(error)) from error + encoded_bytes = manifest["delta_encoded_bytes"] + object_bytes = manifest["delta_object_bytes"] + if ( + type(encoded_bytes) is not int + or encoded_bytes <= 0 + or type(object_bytes) is not int + or not 0 < object_bytes <= _MAX_PAGE_DELTA_OBJECT_BYTES + or manifest["logical_chunk_tokens"] != identity.chunk_tokens + ): + raise CacheFormatError("page delta object geometry differs") + objects = manifest["delta_objects"] + if not isinstance(objects, list) or not objects: + raise CacheFormatError("page delta object descriptors are invalid") + expected_start = 0 + descriptors: list[Mapping[str, Any]] = [] + for index, descriptor in enumerate(objects): + if not isinstance(descriptor, dict): + raise CacheFormatError("page delta object descriptor is not an object") + _strict_keys( + descriptor, + {"sha256", "bytes", "encoded_start", "encoded_end"}, + "page delta object descriptor", + ) + try: + _validate_digest(descriptor["sha256"], "page delta object sha256") + except ValueError as error: + raise CacheFormatError(str(error)) from error + size = descriptor["bytes"] + start = descriptor["encoded_start"] + end = descriptor["encoded_end"] + if ( + type(size) is not int + or type(start) is not int + or type(end) is not int + or size <= 0 + or start != expected_start + or end != start + size + or (index < len(objects) - 1 and size != object_bytes) + or size > object_bytes + ): + raise CacheFormatError("page delta object descriptor geometry differs") + descriptors.append(descriptor) + expected_start = end + if expected_start != encoded_bytes: + raise CacheFormatError("page delta object coverage differs") + return manifest["base_root"], tuple(descriptors) def _decode_chunk( @@ -1270,6 +1352,29 @@ def _decode_chunk( raise CacheFormatError(str(error)) from error +def _read_page_delta_object_batch( + object_root: Path, + descriptors: Sequence[Mapping[str, Any]], +) -> tuple[bytes, ...]: + """Read one bounded batch of authenticated page-delta byte extents.""" + + def read_one(descriptor: Mapping[str, Any]) -> bytes: + encoded = (object_root / f"{descriptor['sha256']}.spcc").read_bytes() + if ( + len(encoded) != descriptor["bytes"] + or _sha256(encoded) != descriptor["sha256"] + ): + raise CacheFormatError("page delta object checksum mismatch") + return encoded + + if not descriptors: + return () + with ThreadPoolExecutor( + max_workers=min(len(descriptors), _PAGE_DELTA_READ_BATCH_SIZE) + ) as pool: + return tuple(pool.map(read_one, descriptors)) + + class ManifestTransaction: """Incrementally publish chunks, then expose them with one final manifest. @@ -1551,7 +1656,9 @@ def _capacity_entry(self, path: Path) -> _CapacityEntry: manifest = json.loads(path.read_bytes()) segments: tuple[str, ...] = () schema_name = manifest.get("schema") if isinstance(manifest, dict) else None - if schema_name in (_TAIL_MANIFEST_SCHEMA, _PAGE_DELTA_MANIFEST_SCHEMA): + if schema_name == _TAIL_MANIFEST_SCHEMA or schema_name in ( + _PAGE_DELTA_MANIFEST_SCHEMAS + ): identity_wire = dict(manifest.get("identity", {})) if "record_schema" in identity_wire: schema = identity_wire["record_schema"] @@ -2229,7 +2336,7 @@ def _page_graph_descriptors( context_digest=context_digest, ) base_digest = manifest["base_context_digest"] - if base_root.get("schema") == _PAGE_DELTA_MANIFEST_SCHEMA: + if _is_page_delta_root(base_root): base_chunks = self._page_graph_descriptors( base_root, identity=identity, @@ -2251,8 +2358,7 @@ def _page_delta_root_count(manifest: Mapping[str, Any]) -> int: count = 0 root: Any = manifest while ( - isinstance(root, Mapping) - and root.get("schema") == _PAGE_DELTA_MANIFEST_SCHEMA + _is_page_delta_root(root) ): count += 1 root = root.get("base_root") @@ -2286,6 +2392,38 @@ def _read_one(descriptor: Mapping[str, Any]) -> ContextChunk: with ThreadPoolExecutor(max_workers=min(8, len(descriptors))) as pool: return tuple(pool.map(_read_one, descriptors)) + def _read_page_delta_objects( + self, + descriptors: Sequence[Mapping[str, Any]], + *, + encoded_bytes: int, + encoded_sha256: str, + ) -> bytearray: + """Read ordered macro objects with bounded transient payload memory.""" + + if encoded_bytes <= 0 or not descriptors: + raise CacheFormatError("page delta object coverage is empty") + result = bytearray(encoded_bytes) + digest = hashlib.sha256() + expected_start = 0 + object_root = self.root / "chunks" + for first in range(0, len(descriptors), _PAGE_DELTA_READ_BATCH_SIZE): + batch = tuple( + descriptors[first : first + _PAGE_DELTA_READ_BATCH_SIZE] + ) + payloads = _read_page_delta_object_batch(object_root, batch) + for descriptor, payload in zip(batch, payloads, strict=True): + start = int(descriptor["encoded_start"]) + end = int(descriptor["encoded_end"]) + if start != expected_start or end != start + len(payload): + raise CacheFormatError("page delta object coverage differs") + result[start:end] = payload + digest.update(payload) + expected_start = end + if expected_start != encoded_bytes or digest.hexdigest() != encoded_sha256: + raise CacheFormatError("page delta payload checksum mismatch") + return result + def publish_prefix_aliases( self, *, @@ -2620,11 +2758,9 @@ def commit_page_extension( ) from sparkcache.spark_context_cache_codec import ( context_prefix_digest, - pack_positions, ) from sparkcache.spark_context_cache_hybrid import ( encode_page_delta, - split_snapshot, ) with _RootGuard(self.root, shared=True, blocking=True): @@ -2668,50 +2804,35 @@ def commit_page_extension( base_boundary_tokens=base_boundary_tokens, result_boundary_tokens=result_boundary_tokens, ) - part_count = ( - result_boundary_tokens + identity.chunk_tokens - 1 - ) // identity.chunk_tokens - parts = split_snapshot(delta, part_count) - chunks = tuple( - ContextChunk( - index * identity.chunk_tokens, - min( - result_boundary_tokens, - (index + 1) * identity.chunk_tokens, - ), - { - StateRecord.LOGICAL_POSITIONS: pack_positions( - range( - index * identity.chunk_tokens, - min( - result_boundary_tokens, - (index + 1) * identity.chunk_tokens, - ), - ) - ), - StateRecord.TARGET_CKV: part, - }, - ) - for index, part in enumerate(parts) - ) descriptors: list[dict[str, Any]] = [] objects: list[tuple[Path, bytes]] = [] - for chunk in chunks: - encoded = _encode_chunk(chunk) - digest = _sha256(encoded) + delta_view = memoryview(delta) + for start in range(0, len(delta), _PAGE_DELTA_OBJECT_BYTES): + end = min(len(delta), start + _PAGE_DELTA_OBJECT_BYTES) + encoded = delta_view[start:end].tobytes() + object_digest = _sha256(encoded) descriptors.append( { - "sha256": digest, + "sha256": object_digest, "bytes": len(encoded), - "logical_start": chunk.logical_start, - "logical_end": chunk.logical_end, + "encoded_start": start, + "encoded_end": end, } ) - objects.append((self.root / "chunks" / f"{digest}.spcc", encoded)) - _publish_immutable_batch(objects) + objects.append( + ( + self.root / "chunks" / f"{object_digest}.spcc", + encoded, + ) + ) + if len(objects) == _PAGE_DELTA_WRITE_BATCH_SIZE: + _publish_immutable_batch(objects) + objects.clear() + if objects: + _publish_immutable_batch(objects) base_root = dict(base._manifest) root = { - "schema": _PAGE_DELTA_MANIFEST_SCHEMA, + "schema": _PAGE_DELTA_MANIFEST_SCHEMA_V2, "format_abi": FORMAT_ABI, "identity": identity.to_wire(), "context_digest": result_context_digest, @@ -2723,7 +2844,11 @@ def commit_page_extension( "layout_sha256": layout.digest, "base_block_counts": list(base_block_counts), "result_block_counts": list(result_block_counts), - "delta_chunks": descriptors, + "delta_encoded_bytes": len(delta), + "delta_object_bytes": _PAGE_DELTA_OBJECT_BYTES, + "delta_objects": descriptors, + "delta_sha256": _sha256(delta), + "logical_chunk_tokens": identity.chunk_tokens, } root["metadata_sha256"] = _sha256(_canonical_json(root)) encoded_root = _canonical_json(root) @@ -2759,7 +2884,7 @@ def restore_page_snapshot( if not lookup.is_hit or lookup._manifest is None: raise ValueError("cannot restore a cache miss") manifest = lookup._manifest - if manifest.get("schema") != _PAGE_DELTA_MANIFEST_SCHEMA: + if not _is_page_delta_root(manifest): chunks = self.restore(lookup) if chunks is None: raise CacheFormatError("page snapshot restore failed") @@ -2788,7 +2913,7 @@ def restore_page_snapshot( _manifest=base_root, root_kind=( "page_delta" - if base_root.get("schema") == _PAGE_DELTA_MANIFEST_SCHEMA + if _is_page_delta_root(base_root) else "manifest" ), ) @@ -2799,13 +2924,20 @@ def restore_page_snapshot( result_boundary_tokens=manifest["base_committed_tokens"], _depth=_depth + 1, ) - delta_chunks = self._read_context_chunks( - delta_descriptors, - identity.required_records, - ) - encoded_delta = b"".join( - chunk.records[StateRecord.TARGET_CKV] for chunk in delta_chunks - ) + if manifest["schema"] == _PAGE_DELTA_MANIFEST_SCHEMA_V2: + encoded_delta = self._read_page_delta_objects( + delta_descriptors, + encoded_bytes=manifest["delta_encoded_bytes"], + encoded_sha256=manifest["delta_sha256"], + ) + else: + delta_chunks = self._read_context_chunks( + delta_descriptors, + identity.required_records, + ) + encoded_delta = b"".join( + chunk.records[StateRecord.TARGET_CKV] for chunk in delta_chunks + ) from sparkcache.spark_context_cache_hybrid import apply_page_delta return apply_page_delta( @@ -2930,10 +3062,7 @@ def lookup( identity=identity, context_digest=context_digest, ) - is_page_delta = ( - isinstance(manifest, dict) - and manifest.get("schema") == _PAGE_DELTA_MANIFEST_SCHEMA - ) + is_page_delta = _is_page_delta_root(manifest) if is_page_delta: chunks = self._page_graph_descriptors( manifest, @@ -2949,8 +3078,6 @@ def lookup( for descriptor in chunks: digest = descriptor["sha256"] encoded_bytes = descriptor["bytes"] - logical_start = descriptor["logical_start"] - logical_end = descriptor["logical_end"] if verify_chunks: encoded_chunk = ( self.root / "chunks" / f"{digest}.spcc" @@ -2960,6 +3087,10 @@ def lookup( or _sha256(encoded_chunk) != digest ): raise CacheFormatError("chunk checksum mismatch") + if "encoded_start" in descriptor: + continue + logical_start = descriptor["logical_start"] + logical_end = descriptor["logical_end"] # The descriptor digest authenticates the complete encoded # chunk: prefix, header (including record digests and # offsets), and every payload byte. Re-hashing each record @@ -3154,9 +3285,8 @@ def invalidate( try: manifest = json.loads(raw) schema_name = manifest.get("schema") - if schema_name in ( - _TAIL_MANIFEST_SCHEMA, - _PAGE_DELTA_MANIFEST_SCHEMA, + if schema_name == _TAIL_MANIFEST_SCHEMA or schema_name in ( + _PAGE_DELTA_MANIFEST_SCHEMAS ): identity_wire = dict(manifest["identity"]) if "record_schema" in identity_wire: diff --git a/sparkcache/persistent_context_cache/test_cache_manifest.py b/sparkcache/persistent_context_cache/test_cache_manifest.py index cf7dfac..b3309c1 100644 --- a/sparkcache/persistent_context_cache/test_cache_manifest.py +++ b/sparkcache/persistent_context_cache/test_cache_manifest.py @@ -220,7 +220,7 @@ def test_page_extension_materializes_full_snapshot_after_base_root_removal( page_root_path, *( root / "chunks" / f"{item['sha256']}.spcc" - for item in page_root["delta_chunks"] + for item in page_root["delta_objects"] ), ] self.assertGreaterEqual( @@ -250,7 +250,7 @@ def test_page_extension_materializes_full_snapshot_after_base_root_removal( chained_root_path, *( root / "chunks" / f"{item['sha256']}.spcc" - for item in chained_root["delta_chunks"] + for item in chained_root["delta_objects"] ), ] self.assertGreaterEqual( diff --git a/sparkcache/persistent_context_cache/test_page_delta_macro_objects.py b/sparkcache/persistent_context_cache/test_page_delta_macro_objects.py new file mode 100644 index 0000000..f77ff54 --- /dev/null +++ b/sparkcache/persistent_context_cache/test_page_delta_macro_objects.py @@ -0,0 +1,346 @@ +from __future__ import annotations + +import dataclasses +import hashlib +import json +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +import sparkcache.persistent_context_cache.cache_manifest as cache_manifest +from sparkcache.persistent_context_cache.cache_manifest import ( + CacheIdentity, + ContextChunk, + ManifestStore, + StateRecord, +) +from sparkcache.spark_context_cache_codec import context_prefix_digest, pack_positions +from sparkcache.spark_context_cache_hybrid import ( + PageGroup, + PageLayer, + PageLayout, + encode_page_snapshot, + split_snapshot, +) + + +def _identity() -> CacheIdentity: + return CacheIdentity( + target_checkpoint="1" * 64, + draft_checkpoint="2" * 64, + quantization_layout="nvfp4-ds-mla-v1", + rope_layout="glm53-hybrid-v1", + tp_degree=4, + dcp_degree=1, + chunk_tokens=256, + record_schema=("target_ckv", "logical_positions"), + publication_schema="page-tail-cow-v1", + ) + + +@dataclasses.dataclass(frozen=True) +class _Fixture: + identity: CacheIdentity + layout: PageLayout + tokens: tuple[int, ...] + salt: str + base_digest: str + result_digest: str + result_snapshot: bytes + result_tokens: int + result_blocks: int + + +def _commit_fixture(store: ManifestStore, *, result_blocks: int = 16) -> _Fixture: + identity = _identity() + layout = PageLayout( + (PageGroup(256, (PageLayer("page", "u8", (1024,), 1024),)),) + ) + result_tokens = result_blocks * 256 + tokens = tuple(range(result_tokens)) + salt = "page-delta-macro-object-test" + base_digest = context_prefix_digest(tokens, salt, token_count=256) + result_digest = context_prefix_digest(tokens, salt, token_count=result_tokens) + base_snapshot = encode_page_snapshot(layout, (1,), {"page": b"A" * 1024}) + result_payload = b"A" * 1024 + b"".join( + bytes((index % 251,)) * 1024 for index in range(1, result_blocks) + ) + result_snapshot = encode_page_snapshot( + layout, + (result_blocks,), + {"page": result_payload}, + ) + store.commit( + identity=identity, + context_digest=base_digest, + chunks=( + ContextChunk( + 0, + 256, + { + StateRecord.LOGICAL_POSITIONS: pack_positions(range(256)), + StateRecord.TARGET_CKV: base_snapshot, + }, + ), + ), + span_tokens=256, + ) + store.commit_page_extension( + identity=identity, + base_context_digest=base_digest, + token_ids=tokens, + identity_salt=salt, + layout=layout, + base_block_counts=(1,), + result_block_counts=(result_blocks,), + base_boundary_tokens=256, + result_boundary_tokens=result_tokens, + result_snapshot=result_snapshot, + ) + return _Fixture( + identity=identity, + layout=layout, + tokens=tokens, + salt=salt, + base_digest=base_digest, + result_digest=result_digest, + result_snapshot=result_snapshot, + result_tokens=result_tokens, + result_blocks=result_blocks, + ) + + +def _manifest(root: Path, fixture: _Fixture) -> tuple[Path, dict[str, object]]: + path = ( + root + / "manifests" + / fixture.identity.storage_key + / f"{fixture.result_digest}.json" + ) + return path, json.loads(path.read_bytes()) + + +class PageDeltaMacroObjectTests(unittest.TestCase): + def test_v2_uses_fewer_physical_objects_than_logical_token_chunks(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + store = ManifestStore(root) + published_batch_sizes: list[int] = [] + original_publish = cache_manifest._publish_immutable_batch + + def record_publish(objects: list[tuple[Path, bytes]]) -> None: + published_batch_sizes.append(len(objects)) + original_publish(objects) + + with ( + mock.patch.object(cache_manifest, "_PAGE_DELTA_OBJECT_BYTES", 2048), + mock.patch.object( + cache_manifest, + "_publish_immutable_batch", + side_effect=record_publish, + ), + ): + fixture = _commit_fixture(store) + + _path, manifest = _manifest(root, fixture) + objects = manifest["delta_objects"] + self.assertEqual(manifest["schema"], "sparkcache-page-delta-manifest/v2") + self.assertEqual(manifest["logical_chunk_tokens"], 256) + self.assertLess(len(objects), fixture.result_tokens // 256) + self.assertLess( + manifest["delta_encoded_bytes"], + len(fixture.result_snapshot), + ) + base_descriptor = manifest["base_root"]["chunks"][0] + self.assertTrue( + (root / "chunks" / f"{base_descriptor['sha256']}.spcc").is_file() + ) + self.assertLessEqual( + max(published_batch_sizes), + cache_manifest._PAGE_DELTA_WRITE_BATCH_SIZE, + ) + self.assertEqual( + [item["encoded_start"] for item in objects], + [0, *[item["encoded_end"] for item in objects[:-1]]], + ) + lookup = store.lookup(fixture.identity, fixture.result_digest) + self.assertTrue(lookup.is_hit, lookup.reason) + self.assertEqual( + store.restore_page_snapshot( + lookup, + layout=fixture.layout, + result_block_counts=(fixture.result_blocks,), + result_boundary_tokens=fixture.result_tokens, + ), + fixture.result_snapshot, + ) + + def test_live_scale_payload_needs_at_most_24_macro_objects(self) -> None: + encoded_bytes = 1_575_821_491 + + object_count = ( + encoded_bytes + cache_manifest._PAGE_DELTA_OBJECT_BYTES - 1 + ) // cache_manifest._PAGE_DELTA_OBJECT_BYTES + + self.assertEqual(object_count, 24) + self.assertLess(object_count, 1_024) + + def test_corrupt_macro_object_fails_closed(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + store = ManifestStore(root) + with mock.patch.object(cache_manifest, "_PAGE_DELTA_OBJECT_BYTES", 2048): + fixture = _commit_fixture(store) + _path, manifest = _manifest(root, fixture) + descriptor = manifest["delta_objects"][0] + object_path = root / "chunks" / f"{descriptor['sha256']}.spcc" + encoded = object_path.read_bytes() + object_path.write_bytes(encoded[:-1] + bytes((encoded[-1] ^ 0xFF,))) + + lookup = store.lookup(fixture.identity, fixture.result_digest) + + self.assertFalse(lookup.is_hit) + self.assertEqual(lookup.reason, "corrupt") + + def test_corrupt_macro_descriptor_fails_closed(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + store = ManifestStore(root) + with mock.patch.object(cache_manifest, "_PAGE_DELTA_OBJECT_BYTES", 2048): + fixture = _commit_fixture(store) + manifest_path, manifest = _manifest(root, fixture) + manifest["delta_objects"][0]["encoded_end"] += 1 + manifest_path.write_bytes(cache_manifest._canonical_json(manifest)) + + lookup = store.lookup(fixture.identity, fixture.result_digest) + + self.assertFalse(lookup.is_hit) + self.assertEqual(lookup.reason, "corrupt") + + def test_restore_reads_only_one_bounded_object_batch_at_a_time(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + store = ManifestStore(root) + with mock.patch.object(cache_manifest, "_PAGE_DELTA_OBJECT_BYTES", 2048): + fixture = _commit_fixture(store, result_blocks=24) + _path, manifest = _manifest(root, fixture) + self.assertGreater( + len(manifest["delta_objects"]), + cache_manifest._PAGE_DELTA_READ_BATCH_SIZE, + ) + observed_batch_sizes: list[int] = [] + original = cache_manifest._read_page_delta_object_batch + + def record_batch( + object_root: Path, + descriptors: tuple[dict[str, object], ...], + ) -> tuple[bytes, ...]: + observed_batch_sizes.append(len(descriptors)) + return original(object_root, descriptors) + + lookup = store.lookup( + fixture.identity, + fixture.result_digest, + verify_chunks=False, + ) + with mock.patch.object( + cache_manifest, + "_read_page_delta_object_batch", + side_effect=record_batch, + ): + restored = store.restore_page_snapshot( + lookup, + layout=fixture.layout, + result_block_counts=(fixture.result_blocks,), + result_boundary_tokens=fixture.result_tokens, + ) + + self.assertEqual(restored, fixture.result_snapshot) + self.assertGreater(len(observed_batch_sizes), 1) + self.assertLessEqual( + max(observed_batch_sizes), + cache_manifest._PAGE_DELTA_READ_BATCH_SIZE, + ) + + def test_v1_page_delta_remains_readable(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + store = ManifestStore(root) + with mock.patch.object(cache_manifest, "_PAGE_DELTA_OBJECT_BYTES", 2048): + fixture = _commit_fixture(store) + manifest_path, manifest = _manifest(root, fixture) + encoded_delta = store._read_page_delta_objects( + manifest["delta_objects"], + encoded_bytes=manifest["delta_encoded_bytes"], + encoded_sha256=manifest["delta_sha256"], + ) + parts = split_snapshot( + encoded_delta, + fixture.result_tokens // fixture.identity.chunk_tokens, + ) + descriptors = [] + for index, part in enumerate(parts): + chunk = ContextChunk( + index * fixture.identity.chunk_tokens, + (index + 1) * fixture.identity.chunk_tokens, + { + StateRecord.LOGICAL_POSITIONS: pack_positions( + range( + index * fixture.identity.chunk_tokens, + (index + 1) * fixture.identity.chunk_tokens, + ) + ), + StateRecord.TARGET_CKV: bytes(part), + }, + ) + encoded = cache_manifest._encode_chunk(chunk) + digest = hashlib.sha256(encoded).hexdigest() + cache_manifest._publish_immutable( + root / "chunks" / f"{digest}.spcc", + encoded, + ) + descriptors.append( + { + "sha256": digest, + "bytes": len(encoded), + "logical_start": chunk.logical_start, + "logical_end": chunk.logical_end, + } + ) + legacy = { + key: value + for key, value in manifest.items() + if key + not in { + "delta_objects", + "delta_encoded_bytes", + "delta_object_bytes", + "delta_sha256", + "logical_chunk_tokens", + "metadata_sha256", + } + } + legacy["schema"] = "sparkcache-page-delta-manifest/v1" + legacy["delta_chunks"] = descriptors + legacy["metadata_sha256"] = hashlib.sha256( + cache_manifest._canonical_json(legacy) + ).hexdigest() + manifest_path.write_bytes(cache_manifest._canonical_json(legacy)) + + lookup = store.lookup(fixture.identity, fixture.result_digest) + + self.assertTrue(lookup.is_hit, lookup.reason) + self.assertEqual( + store.restore_page_snapshot( + lookup, + layout=fixture.layout, + result_block_counts=(fixture.result_blocks,), + result_boundary_tokens=fixture.result_tokens, + ), + fixture.result_snapshot, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/sparkcache/test_defect_regressions.py b/sparkcache/test_defect_regressions.py index e2f68b7..0fb626a 100644 --- a/sparkcache/test_defect_regressions.py +++ b/sparkcache/test_defect_regressions.py @@ -979,13 +979,10 @@ class MambaSpec: assert manifest is not None self.assertEqual(manifest["base_block_counts"], [28, 1]) self.assertEqual(manifest["result_block_counts"], [47, 1]) - delta_chunks = connector._store._read_context_chunks( - manifest["delta_chunks"], - connector._identity(0).required_records, - ) - encoded_delta = b"".join( - chunk.records[package_store.StateRecord.TARGET_CKV] - for chunk in delta_chunks + encoded_delta = connector._store._read_page_delta_objects( + manifest["delta_objects"], + encoded_bytes=manifest["delta_encoded_bytes"], + encoded_sha256=manifest["delta_sha256"], ) self.assertLess( len(encoded_delta), diff --git a/sparkcache/test_spark_context_cache_connector.py b/sparkcache/test_spark_context_cache_connector.py index 43b7f03..343a1b4 100644 --- a/sparkcache/test_spark_context_cache_connector.py +++ b/sparkcache/test_spark_context_cache_connector.py @@ -606,7 +606,7 @@ def native_placement(**kwargs): ).read_bytes() ) delta_path = ( - root / "chunks" / f"{manifest['delta_chunks'][0]['sha256']}.spcc" + root / "chunks" / f"{manifest['delta_objects'][0]['sha256']}.spcc" ) damaged = bytearray(delta_path.read_bytes()) damaged[-1] ^= 1