diff --git a/CHANGELOG.md b/CHANGELOG.md index 54d7636bf..6dc556f09 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,31 +7,49 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Live materialization now derives the current writer frontier, checks + `WarpStateCachePort` for exact and compatible predecessor snapshots before + replay, and publishes live replay results with their real WARP coordinate. +- Exact state-cache hits now return cached materialization results without + republishing the same full snapshot back into the cache. +- Live `wantDiff` materialization now bypasses state-cache hits so callers keep + receiving diff data from replay-backed materialization paths. +- Receipt-producing materialization, including zero-patch empty results, now + bypasses state-cache reads and writes so callers keep receiving + replay-derived receipt arrays. +- Live checkpoint fallback replay is now constrained to the requested frontier + coordinate before publishing a cache snapshot for that coordinate. +- Live checkpoint reuse now verifies the checkpoint frontier is a predecessor + of the captured live coordinate before seeding state-cache publication. + +### Changed + +- Corrected CAS-first materialization documentation to describe the actual + WARP-owned state-cache lifecycle and its full-materialization memory limits. + ## [18.2.0] - 2026-06-28 ### Release notes -`v18.2.0` introduces the streaming CAS-First Memoization Engine for graph -materialization. By interrogating `git-cas` before initiating projection replay -and utilizing a constant-memory pass-through `teeStream` during materialization, -`git-warp` guarantees `O(1)` runtime memory footprints and zero-latency -snapshot hydration. +`v18.2.0` introduces the first state-cache memoization components for graph +materialization. The release adds the initial cache port and `git-cas`-backed +adapter groundwork for avoiding redundant projection replay, while keeping WARP +frontier and snapshot semantics inside `git-warp`. -It also enforces strict `@git-stunts/git-cas` boundary encapsulation across the -repository, utilizing Buzhash Content-Defined Chunking (CDC) for automated -rolling hash deduplication of unchanged sub-trees. +It also starts enforcing `@git-stunts/git-cas` boundary encapsulation across the +repository so WARP-owned state-cache payloads route through the storage library +instead of hand-rolled CAS plumbing. ### Added -- `CasFirstMemoizationEngine` now enforces the 3-step CAS memoization rule: - interrogate `git-cas` for existing snapshots, materialize via lazy streaming - upon a cache miss, and simultaneously pipe materialized buffers to `git-cas` - via `storeStream`. +- Added initial CAS-first memoization and WARP state-cache components for + runtime integration work. - Added `has()` check capability to `BlobStoragePort` and `CasBlobAdapter` to support zero-latency CAS interrogations before buffering events. - Added `docs/topics/cas-first-memoized-materialization.md` to document the - constant-memory streaming materialization pipeline and Buzhash CDC rolling - hash deduplication mechanics. + state-cache materialization lifecycle and its current memory boundaries. ## [18.1.2] - 2026-06-25 diff --git a/docs/topics/README.md b/docs/topics/README.md index eee5a0446..958f110ec 100644 --- a/docs/topics/README.md +++ b/docs/topics/README.md @@ -33,9 +33,9 @@ inventories. Operator workflows live outside the topic shelf in checkpoints, replay, and provenance. - [Content and CAS](content-and-cas.md): handle content attachments, content-addressed storage, and encrypted CAS payloads. -- [CAS-First memoized materialization](cas-first-memoized-materialization.md): - guarantee constant-memory streaming, Buzhash CDC deduplication, and - zero-latency snapshot hydration. +- [WARP state-cache materialization](cas-first-memoized-materialization.md): + skip redundant live materialization replay through coordinate-addressed + state-cache snapshots backed by `git-cas`. - [Continuum boundary](continuum-boundary.md): understand what git-warp owns locally and what Continuum owns as boundary vocabulary. diff --git a/docs/topics/cas-first-memoized-materialization.md b/docs/topics/cas-first-memoized-materialization.md index 5c2115d06..567cdfc96 100644 --- a/docs/topics/cas-first-memoized-materialization.md +++ b/docs/topics/cas-first-memoized-materialization.md @@ -1,71 +1,92 @@ -# CAS-First Memoized Materialization +# WARP State-Cache Materialization -Use this page when you need to understand `git-warp`'s constant-memory streaming -materialization pipeline, `@git-stunts/git-cas` boundary encapsulation, and -rolling hash deduplication mechanics. +Use this page when you need to understand how `git-warp` skips redundant +materialization replay by memoizing WARP-owned state snapshots in +`@git-stunts/git-cas`. -## The Materialization Lifecycle +`git-cas` is only the byte storage substrate. It does not know about WARP +frontiers, optics, checkpoints, graph state, or materialization rules. `git-warp` +owns those semantics through `WarpStateCachePort`; the Git-backed adapter stores +snapshot payloads in `git-cas`. -In `git-warp`, materialization is the bounded projection of raw CRDT graph -events into structural checkpoints, working set views, or specialized hologram -slices. +## The Live Materialization Lifecycle -To guarantee constant-memory `O(1)` runtime footprints and eliminate redundant -CPU/memory computation across stigmergic peers, `git-warp` enforces a -**CAS-First memoization pipeline**. +When a Git-backed runtime has a state cache, live materialization follows this +coordinate-first lifecycle: ```text -+++++++> [git-cas] ---------> (materialization) ------> * (object) - ^ | - | | - +----------------------------+ +[current frontier] + | + v +[state-cache exact hit?] ---- yes ---> [return cached state] + | + no + v +[compatible predecessor?] --- yes ---> [replay suffix, publish snapshot] + | + no + v +[checkpoint/frontier replay] --------> [publish snapshot] ``` -## CAS-First Memoization Rules +### 1. Derive a WARP coordinate -Every materialization request must execute the following strict lifecycle: +Before replay, the live path reads the current writer frontier and builds a +WARP state coordinate: -### 2.1. Is object already in git-cas? +```text +{ frontier: Map, ceiling: null } +``` + +This coordinate belongs to `git-warp`; it is not a `git-cas` concept. + +### 2. Check the WARP state cache + +The runtime asks `WarpStateCachePort` for an exact snapshot at that coordinate. +On a hit, it returns the cached state without replaying writer patch streams and +without republishing the same snapshot. -Before executing any projection logic or buffering events into V8 heap memory, -`git-warp` derives a deterministic materialization coordinate key: -`key = sha256(baseFrontierSha + opticLensSha + queryParams)`. +If no exact snapshot exists, the runtime asks for the best compatible +predecessor. A predecessor hit lets materialization replay only the suffix after +that cached coordinate, then publish a fresh snapshot for the current frontier. -The runtime immediately interrogates `git-cas` (`await cas.has(key)`). If the -object exists in storage, `git-warp` bypasses the entire projection calculation -and streams the pre-calculated object directly to the caller. +### 3. Fall back to replay and publish -### 2.2. No? Materialize via streaming +When there is no usable cached snapshot, the runtime falls back to the existing +checkpoint/frontier replay path. Successful live and coordinate materializations +publish an evictable state-cache snapshot with the actual coordinate so the next +equivalent read can hit the cache. -If the CAS interrogation returns a miss, `git-warp` initializes a lazy, chunked -streaming materialization pipeline. Events are pulled from the underlying CRDT -log in bounded batches, processed through the projection kernel, and immediately -piped out to avoid accumulating unbounded memory buffers. +## Memory Boundaries -### 2.3. Write materialized git-object to git-cas always +State-cache hits avoid redundant CRDT replay and can remove repeated startup +costs for graph-sized materializations. They do not make legacy full +materialization an `O(1)` memory API: a caller that asks for a full +`SnapshotWarpState` still receives a full in-memory state object. -As the object is materialized, the resulting buffer is simultaneously piped -directly into `git-cas` (`cas.writeStream(key)`). This permanently memoizes the -structural reality for all future causal code paths, background daemons, and -remote peers. +The bounded-memory read path is optic/worldline/query work over a sharded or +streamed basis. The state cache is the replay-skipping compatibility bridge for +legacy materialization and checkpoint flows. -## Strict @git-stunts/git-cas Encapsulation +## `git-cas` Encapsulation -All CAS operations must route through the formal `@git-stunts/git-cas` library -API. Direct invocation of raw git storage commands (`git hash-object`, -`git cat-file`, `git mktree`) is strictly banned within `git-warp`. +All state-cache payload storage routes through the formal `@git-stunts/git-cas` +library API. Raw Git plumbing remains an adapter concern for WARP refs and Git +object access; WARP state-cache payloads should not hand-roll a parallel CAS. -### Buzhash Content-Defined Chunking (CDC) +Routing state snapshots through `git-cas` allows content-addressed storage and +chunk-level reuse where the underlying CAS representation can identify unchanged +byte ranges. The WARP cache index remains responsible for determining whether a +snapshot is semantically usable for a materialization coordinate. -Routing through `@git-stunts/git-cas` unlocks advanced rolling hash capabilities: +## Current Limitations -- **Dynamic Chunking**: `@git-stunts/git-cas` employs a Buzhash rolling hash - algorithm to dynamically split streaming data into variable-length chunks - based on actual data content rather than fixed byte boundaries. -- **Structural Deduplication**: If 99% of a materialized graph snapshot remains - unchanged between two consecutive frontiers, Buzhash CDC produces the exact - same block OIDs for the unchanged sub-trees. `@git-stunts/git-cas` instantly - deduplicates these blocks in memory before anything touches disk storage. +- Exact state-cache hits bypass replay, but full materialization still hydrates + a full `WarpState`. +- The Git-backed state-cache adapter stores full-state snapshots today. A future + sharded basis format should make optic reads avoid full-state hydration. +- Cache coordinates must stay schema/version aware. A snapshot is reusable only + when WARP semantics say the coordinate is compatible. ## See also diff --git a/src/domain/capabilities/PatchCollector.ts b/src/domain/capabilities/PatchCollector.ts index f5d7671e5..3d463c2d2 100644 --- a/src/domain/capabilities/PatchCollector.ts +++ b/src/domain/capabilities/PatchCollector.ts @@ -27,10 +27,56 @@ function patchWithinCeiling(entry: PatchWithSha, ceiling: number | null): boolea return ceiling === null || entry.patch.lamport <= ceiling; } +function patchAfterBaseCeiling(entry: PatchWithSha, ceiling: number | null): boolean { + return ceiling === null || entry.patch.lamport > ceiling; +} + function validTipSha(tipSha: string | undefined): tipSha is string { return typeof tipSha === 'string' && tipSha.length > 0; } +function patchInCoordinateSuffix( + entry: PatchWithSha, + ceiling: number | null, + baseCeiling: number | null, +): boolean { + return patchWithinCeiling(entry, ceiling) && patchAfterBaseCeiling(entry, baseCeiling); +} + +function coordinateStopAtSha( + writerId: string, + baseCoordinate: { frontier: Map; ceiling: number | null }, +): string | null { + if (baseCoordinate.ceiling !== null) { + return null; + } + const baseTipSha = baseCoordinate.frontier.get(writerId); + return validTipSha(baseTipSha) ? baseTipSha : null; +} + +type PatchChainLoader = { + loadPatchChain(toSha: string, fromSha?: string | null): Promise; +}; + +type CoordinateSuffixInput = { + writerId: string; + tipSha: string; + ceiling: number | null; + baseCoordinate: { frontier: Map; ceiling: number | null }; +}; + +async function* streamWriterCoordinateSuffix( + loader: PatchChainLoader, + input: CoordinateSuffixInput, +): AsyncIterable { + const stopAtSha = coordinateStopAtSha(input.writerId, input.baseCoordinate); + for (const entry of await loader.loadPatchChain(input.tipSha, stopAtSha)) { + if (patchInCoordinateSuffix(entry, input.ceiling, input.baseCoordinate.ceiling)) { + yield entry; + } + } +} + /** * Collects patches for materialization. * @@ -84,9 +130,18 @@ export default abstract class PatchCollector { async *streamForFrontierSinceCoordinate( frontier: Map, ceiling: number | null, - _baseCoordinate: { frontier: Map; ceiling: number | null }, + baseCoordinate: { frontier: Map; ceiling: number | null }, ): AsyncIterable { - yield* this.streamForFrontier(frontier, ceiling); + for (const writerId of frontier.keys()) { + const tipSha = frontier.get(writerId); + if (!validTipSha(tipSha)) { continue; } + yield* streamWriterCoordinateSuffix(this, { + writerId, + tipSha, + ceiling, + baseCoordinate, + }); + } } /** Load the latest checkpoint, or null if none. */ @@ -105,6 +160,9 @@ export default abstract class PatchCollector { /** Load a patch chain between two SHAs. */ abstract loadPatchChain(_toSha: string, _fromSha?: string | null): Promise; + /** Return whether one patch commit is an ancestor of another, when available. */ + isAncestor?(_ancestorSha: string, _descendantSha: string): Promise; + /** Get the current writer frontier. */ abstract getFrontier(): Promise>; } diff --git a/src/domain/services/controllers/MaterializeController.ts b/src/domain/services/controllers/MaterializeController.ts index 915a0db2d..7f4868189 100644 --- a/src/domain/services/controllers/MaterializeController.ts +++ b/src/domain/services/controllers/MaterializeController.ts @@ -33,6 +33,10 @@ import MaterializePatchStreamReducer, { type MaterializePatchStreamReduction, } from './MaterializePatchStreamReducer.ts'; import { summarizeMaterializePatches } from './MaterializePatchSummary.ts'; +import { + shouldPublishMaterializeSnapshot, + type MaterializeSnapshotPublicationOptions, +} from './MaterializeSnapshotPublication.ts'; import type LoggerPort from '../../../ports/LoggerPort.ts'; import type CodecPort from '../../../ports/CodecPort.ts'; import type CryptoPort from '../../../ports/CryptoPort.ts'; @@ -220,24 +224,28 @@ export default class MaterializeController { private async _emptyResult( ceiling?: number | null, frontier?: Map | null, + options?: MaterializeSnapshotPublicationOptions, ): Promise { - return await this._wrapState(createEmptyState(), ceiling ?? null, frontier ?? null); + return await this._wrapState(createEmptyState(), ceiling ?? null, frontier ?? null, options); } private async _wrapState( state: WarpState, ceiling: number | null, frontier: Map | null, + options?: MaterializeSnapshotPublicationOptions, ): Promise { const stateHash = await computeHash(this._deps, state); const adjacency = buildAdjacency(state); - await this._publishSnapshot({ - state, - stateHash, - degraded: false, - ceiling, - frontier, - }); + if (shouldPublishMaterializeSnapshot(options)) { + await this._publishSnapshot({ + state, + stateHash, + degraded: false, + ceiling, + frontier, + }); + } return { state, stateHash, @@ -254,13 +262,15 @@ export default class MaterializeController { private async _buildResult(params: MaterializeResultBuildInput): Promise { const stateHash = await computeHash(this._deps, params.reduced.state); const adjacency = params.reduced.adjacency ?? buildAdjacency(params.reduced.state); - await this._publishSnapshot({ - state: params.reduced.state, - stateHash, - degraded: params.degraded, - ceiling: params.ceiling, - frontier: params.frontier, - }); + if (params.reduced.receipts === undefined) { + await this._publishSnapshot({ + state: params.reduced.state, + stateHash, + degraded: params.degraded, + ceiling: params.ceiling, + frontier: params.frontier, + }); + } return { state: params.reduced.state, stateHash, @@ -323,8 +333,10 @@ export default class MaterializeController { private _createStrategyRuntime(): MaterializeStrategyRuntime { return { deps: this._deps, - emptyResult: async (ceiling, frontier) => await this._emptyResult(ceiling, frontier), - wrapState: async (state, ceiling, frontier) => await this._wrapState(state, ceiling, frontier), + emptyResult: async (ceiling, frontier, options) => + await this._emptyResult(ceiling, frontier, options), + wrapState: async (state, ceiling, frontier, options) => + await this._wrapState(state, ceiling, frontier, options), reducePatches: async (patches, base, opts) => await this._reducePatches(patches, base, opts), reducePatchStream: async (stream, base, opts, provenanceBase) => await this._reducePatchStream(stream, base, opts, provenanceBase), diff --git a/src/domain/services/controllers/MaterializeCoordinateStrategy.ts b/src/domain/services/controllers/MaterializeCoordinateStrategy.ts index 7a8987309..1160b352a 100644 --- a/src/domain/services/controllers/MaterializeCoordinateStrategy.ts +++ b/src/domain/services/controllers/MaterializeCoordinateStrategy.ts @@ -1,4 +1,3 @@ -import { ProvenanceIndex } from '../provenance/ProvenanceIndex.ts'; import type { MaterializeCoordinateOptions, MaterializeStrategyRuntime, @@ -7,14 +6,12 @@ import type { MaterializeResult } from './MaterializeController.ts'; import type WarpStateCachePort from '../../../ports/WarpStateCachePort.ts'; import type { WarpStateCoordinate, - WarpStateSnapshotRecord, } from '../../../ports/WarpStateCachePort.ts'; -import type WarpState from '../state/WarpState.ts'; -import { MaterializePatchSummary } from './MaterializePatchSummary.ts'; - -type UsableSnapshotRecord = WarpStateSnapshotRecord & { - state: WarpState; -}; +import { + canUseSnapshot, + snapshotToMaterializeResult, +} from './MaterializeSnapshotCacheResult.ts'; +import { snapshotPublicationForReceipts } from './MaterializeSnapshotPublication.ts'; export default class MaterializeCoordinateStrategy { private readonly runtime: MaterializeStrategyRuntime; @@ -25,7 +22,7 @@ export default class MaterializeCoordinateStrategy { async materialize(opts: MaterializeCoordinateOptions): Promise { if (this.canReturnEmpty(opts)) { - return await this.runtime.emptyResult(opts.ceiling, opts.frontier); + return await this.emptyResult(opts); } const coordinate = this.snapshotCoordinate(opts.frontier, opts.ceiling); const cacheResolved = await this.tryResolveSnapshotCache({ @@ -37,7 +34,7 @@ export default class MaterializeCoordinateStrategy { } const reduction = await this.reduceFrontierPatches(opts); if (reduction.summary.patchCount === 0) { - return await this.runtime.emptyResult(opts.ceiling, opts.frontier); + return await this.emptyResult(opts); } return await this.runtime.buildResult({ reduced: reduction.reduced, @@ -48,6 +45,14 @@ export default class MaterializeCoordinateStrategy { }); } + private async emptyResult(opts: MaterializeCoordinateOptions): Promise { + return await this.runtime.emptyResult( + opts.ceiling, + opts.frontier, + snapshotPublicationForReceipts(opts), + ); + } + private async reduceFrontierPatches(opts: MaterializeCoordinateOptions) { return await this.runtime.reducePatchStream( this.runtime.deps.patches.streamForFrontier(opts.frontier, opts.ceiling), @@ -81,6 +86,9 @@ export default class MaterializeCoordinateStrategy { coordinate: WarpStateCoordinate; receipts: boolean; }): Promise { + if (opts.receipts) { + return null; + } const stateCache = this.runtime.deps.getStateCache?.() ?? null; if (stateCache === null) { return null; @@ -104,8 +112,8 @@ export default class MaterializeCoordinateStrategy { opts: { coordinate: WarpStateCoordinate; receipts: boolean }, ): Promise { const exact = await stateCache.getExact(opts.coordinate); - if (this.canUseSnapshot(exact, opts.receipts)) { - return await this.snapshotToResult(exact); + if (canUseSnapshot(exact, { receipts: opts.receipts })) { + return snapshotToMaterializeResult(exact); } return null; } @@ -115,7 +123,7 @@ export default class MaterializeCoordinateStrategy { opts: { coordinate: WarpStateCoordinate; receipts: boolean }, ): Promise { const predecessor = await stateCache.getBestCompatiblePredecessor(opts.coordinate); - if (!this.canUseSnapshot(predecessor, opts.receipts)) { + if (!canUseSnapshot(predecessor, { receipts: opts.receipts })) { return null; } @@ -139,27 +147,4 @@ export default class MaterializeCoordinateStrategy { frontier: opts.coordinate.frontier, }); } - - private canUseSnapshot( - snapshot: WarpStateSnapshotRecord | null, - receipts: boolean, - ): snapshot is UsableSnapshotRecord { - if (snapshot === null || snapshot.state === undefined) { - return false; - } - if (receipts && snapshot.provenancePosture === 'degraded') { - return false; - } - return true; - } - - private async snapshotToResult(snapshot: UsableSnapshotRecord): Promise { - return await this.runtime.buildResult({ - reduced: { state: snapshot.state }, - summary: MaterializePatchSummary.empty(new ProvenanceIndex()), - degraded: snapshot.provenancePosture === 'degraded', - ceiling: snapshot.coordinate.ceiling, - frontier: snapshot.coordinate.frontier, - }); - } } diff --git a/src/domain/services/controllers/MaterializeLiveStrategy.ts b/src/domain/services/controllers/MaterializeLiveStrategy.ts index 8212b2485..72f81279b 100644 --- a/src/domain/services/controllers/MaterializeLiveStrategy.ts +++ b/src/domain/services/controllers/MaterializeLiveStrategy.ts @@ -8,6 +8,19 @@ import type { CheckpointData, PatchWithSha, } from '../../capabilities/PatchCollector.ts'; +import type WarpStateCachePort from '../../../ports/WarpStateCachePort.ts'; +import type { + WarpStateCoordinate, +} from '../../../ports/WarpStateCachePort.ts'; +import { + canUseSnapshot, + snapshotToMaterializeResult, +} from './MaterializeSnapshotCacheResult.ts'; +import { snapshotPublicationForReceipts } from './MaterializeSnapshotPublication.ts'; + +function nonEmptySha(value: string | undefined): value is string { + return typeof value === 'string' && value.length > 0; +} export default class MaterializeLiveStrategy { private readonly runtime: MaterializeStrategyRuntime; @@ -17,19 +30,104 @@ export default class MaterializeLiveStrategy { } async materialize(opts: MaterializeLiveOptions): Promise { + const stateCache = this.runtime.deps.getStateCache?.() ?? null; + if (stateCache !== null) { + return await this.materializeWithStateCache(stateCache, opts); + } + return await this.materializeWithoutStateCache(opts); + } + + private async materializeWithoutStateCache(opts: MaterializeLiveOptions): Promise { const checkpoint = await this.runtime.deps.patches.loadCheckpoint(); if (checkpoint !== null && checkpoint !== undefined && isCurrentCheckpointSchema(checkpoint.schema)) { - return await this.fromCheckpoint(checkpoint, opts); + return await this.fromCheckpoint(checkpoint, opts, null); } return await this.fromScratch(opts); } + private async materializeWithStateCache( + stateCache: WarpStateCachePort, + opts: MaterializeLiveOptions, + ): Promise { + const frontier = await this.runtime.deps.patches.getFrontier(); + if (frontier.size === 0) { + return await this.runtime.emptyResult(null, frontier, snapshotPublicationForReceipts(opts)); + } + const coordinate = this.snapshotCoordinate(frontier); + const cacheResolved = await this.tryResolveSnapshotCache(stateCache, { + coordinate, + receipts: opts.receipts, + wantDiff: opts.wantDiff, + }); + if (cacheResolved !== null) { + return cacheResolved; + } + return await this.replayCurrentCoordinate(coordinate, opts); + } + + private async replayCurrentCoordinate( + coordinate: WarpStateCoordinate, + opts: MaterializeLiveOptions, + ): Promise { + const checkpoint = await this.runtime.deps.patches.loadCheckpoint(); + if ( + this.isCurrentCheckpoint(checkpoint) + && await this.checkpointSupportsCoordinate(checkpoint, coordinate) + ) { + return await this.fromCheckpoint(checkpoint, opts, coordinate.frontier); + } + return await this.fromFrontier(coordinate, opts); + } + + private isCurrentCheckpoint( + checkpoint: CheckpointData | null | undefined, + ): checkpoint is CheckpointData { + return checkpoint !== null && checkpoint !== undefined && isCurrentCheckpointSchema(checkpoint.schema); + } + + private async checkpointSupportsCoordinate( + checkpoint: CheckpointData, + coordinate: WarpStateCoordinate, + ): Promise { + for (const [writerId, checkpointTip] of checkpoint.frontier) { + if (!await this.checkpointWriterTipIsCompatible( + checkpointTip, + coordinate.frontier.get(writerId), + )) { + return false; + } + } + return true; + } + + private async checkpointWriterTipIsCompatible( + checkpointTip: string, + targetTip: string | undefined, + ): Promise { + if (!nonEmptySha(checkpointTip) || !nonEmptySha(targetTip)) { + return false; + } + return checkpointTip === targetTip || await this.checkpointTipPrecedesTarget(checkpointTip, targetTip); + } + + private async checkpointTipPrecedesTarget( + checkpointTip: string, + targetTip: string, + ): Promise { + const { patches } = this.runtime.deps; + if (typeof patches.isAncestor !== 'function') { + return false; + } + return await patches.isAncestor(checkpointTip, targetTip); + } + private async fromCheckpoint( checkpoint: CheckpointData, opts: MaterializeLiveOptions, + frontier: Map | null, ): Promise { const reduction = await this.runtime.reducePatchStream( - this.streamPatchesSince(checkpoint), + this.streamPatchesSinceCheckpoint(checkpoint, frontier), checkpoint.state, opts, checkpoint.provenanceIndex, @@ -39,10 +137,32 @@ export default class MaterializeLiveStrategy { summary: reduction.summary, degraded: false, ceiling: null, - frontier: null, + frontier, }); } + private async *streamPatchesSinceCheckpoint( + checkpoint: CheckpointData, + frontier: Map | null, + ): AsyncIterable { + if (frontier !== null) { + yield* this.runtime.deps.patches.streamForFrontierSinceCoordinate( + frontier, + null, + this.checkpointCoordinate(checkpoint), + ); + return; + } + yield* this.streamPatchesSince(checkpoint); + } + + private checkpointCoordinate(checkpoint: CheckpointData): WarpStateCoordinate { + return { + frontier: checkpoint.frontier, + ceiling: null, + }; + } + private async *streamPatchesSince(checkpoint: CheckpointData): AsyncIterable { if (typeof this.runtime.deps.patches.streamPatchesSince === 'function') { yield* this.runtime.deps.patches.streamPatchesSince(checkpoint); @@ -56,7 +176,11 @@ export default class MaterializeLiveStrategy { private async fromScratch(opts: MaterializeLiveOptions): Promise { const writers = await this.runtime.deps.patches.discoverWriters(); if (writers.length === 0) { - return await this.runtime.emptyResult(); + return await this.runtime.emptyResult( + undefined, + undefined, + snapshotPublicationForReceipts(opts), + ); } const reduction = await this.runtime.reducePatchStream( this.streamAllPatches(writers), @@ -64,7 +188,11 @@ export default class MaterializeLiveStrategy { opts, ); if (reduction.summary.patchCount === 0) { - return await this.runtime.emptyResult(); + return await this.runtime.emptyResult( + undefined, + undefined, + snapshotPublicationForReceipts(opts), + ); } return await this.runtime.buildResult({ reduced: reduction.reduced, @@ -75,6 +203,93 @@ export default class MaterializeLiveStrategy { }); } + private async fromFrontier( + coordinate: WarpStateCoordinate, + opts: MaterializeLiveOptions, + ): Promise { + const reduction = await this.runtime.reducePatchStream( + this.runtime.deps.patches.streamForFrontier(coordinate.frontier, coordinate.ceiling), + undefined, + opts, + ); + if (reduction.summary.patchCount === 0) { + return await this.runtime.emptyResult( + coordinate.ceiling, + coordinate.frontier, + snapshotPublicationForReceipts(opts), + ); + } + return await this.runtime.buildResult({ + reduced: reduction.reduced, + summary: reduction.summary, + degraded: false, + ceiling: coordinate.ceiling, + frontier: coordinate.frontier, + }); + } + + private snapshotCoordinate(frontier: Map): WarpStateCoordinate { + return { + frontier, + ceiling: null, + }; + } + + private async tryResolveSnapshotCache( + stateCache: WarpStateCachePort, + opts: { coordinate: WarpStateCoordinate; receipts: boolean; wantDiff: boolean }, + ): Promise { + if (opts.receipts || opts.wantDiff) { + return null; + } + const exactResult = await this.tryResolveExactSnapshot(stateCache, opts); + if (exactResult !== null) { + return exactResult; + } + return await this.tryResolvePredecessorSnapshot(stateCache, opts); + } + + private async tryResolveExactSnapshot( + stateCache: WarpStateCachePort, + opts: { coordinate: WarpStateCoordinate; receipts: boolean }, + ): Promise { + const exact = await stateCache.getExact(opts.coordinate); + if (canUseSnapshot(exact, { receipts: opts.receipts })) { + return snapshotToMaterializeResult(exact); + } + return null; + } + + private async tryResolvePredecessorSnapshot( + stateCache: WarpStateCachePort, + opts: { coordinate: WarpStateCoordinate; receipts: boolean }, + ): Promise { + const predecessor = await stateCache.getBestCompatiblePredecessor(opts.coordinate); + if (!canUseSnapshot(predecessor, { receipts: opts.receipts })) { + return null; + } + + const reduction = await this.runtime.reducePatchStream( + this.runtime.deps.patches.streamForFrontierSinceCoordinate( + opts.coordinate.frontier, + opts.coordinate.ceiling, + predecessor.coordinate, + ), + predecessor.state, + { + receipts: false, + wantDiff: false, + }, + ); + return await this.runtime.buildResult({ + reduced: reduction.reduced, + summary: reduction.summary, + degraded: predecessor.provenancePosture === 'degraded', + ceiling: opts.coordinate.ceiling, + frontier: opts.coordinate.frontier, + }); + } + private async *streamAllPatches(writers: string[]): AsyncIterable { for (const writerId of writers) { yield* this.runtime.deps.patches.streamWriterPatches(writerId); diff --git a/src/domain/services/controllers/MaterializeSnapshotCacheResult.ts b/src/domain/services/controllers/MaterializeSnapshotCacheResult.ts new file mode 100644 index 000000000..36c43434c --- /dev/null +++ b/src/domain/services/controllers/MaterializeSnapshotCacheResult.ts @@ -0,0 +1,54 @@ +import AdjacencyMap from '../../capabilities/AdjacencyMap.ts'; +import { ProvenanceIndex } from '../provenance/ProvenanceIndex.ts'; +import { buildAdjacency } from './MaterializeHelpers.ts'; +import type WarpState from '../state/WarpState.ts'; +import type { MaterializeResult } from './MaterializeController.ts'; +import type { + WarpStateSnapshotRecord, +} from '../../../ports/WarpStateCachePort.ts'; + +export type UsableSnapshotRecord = WarpStateSnapshotRecord & { + state: WarpState; +}; + +type SnapshotUseOptions = Readonly<{ + receipts: boolean; +}>; + +function snapshotHasState( + snapshot: WarpStateSnapshotRecord | null | undefined, +): snapshot is UsableSnapshotRecord { + return snapshot !== null && snapshot !== undefined && snapshot.state !== undefined; +} + +function receiptsAllowSnapshot( + snapshot: UsableSnapshotRecord, + options: SnapshotUseOptions, +): boolean { + return !options.receipts || snapshot.provenancePosture !== 'degraded'; +} + +export function canUseSnapshot( + snapshot: WarpStateSnapshotRecord | null | undefined, + options: SnapshotUseOptions, +): snapshot is UsableSnapshotRecord { + if (!snapshotHasState(snapshot)) { + return false; + } + return receiptsAllowSnapshot(snapshot, options); +} + +export function snapshotToMaterializeResult(snapshot: UsableSnapshotRecord): MaterializeResult { + const adjacency = buildAdjacency(snapshot.state); + return { + state: snapshot.state, + stateHash: snapshot.stateHash, + adjacency: new AdjacencyMap({ outgoing: adjacency.outgoing, incoming: adjacency.incoming }), + patchCount: 0, + maxObservedLamport: 0, + provenanceIndex: new ProvenanceIndex(), + provenanceDegraded: snapshot.provenancePosture === 'degraded', + frontier: snapshot.coordinate.frontier, + ceiling: snapshot.coordinate.ceiling, + }; +} diff --git a/src/domain/services/controllers/MaterializeSnapshotPublication.ts b/src/domain/services/controllers/MaterializeSnapshotPublication.ts new file mode 100644 index 000000000..6383ed486 --- /dev/null +++ b/src/domain/services/controllers/MaterializeSnapshotPublication.ts @@ -0,0 +1,28 @@ +export type MaterializeSnapshotPublication = 'publish' | 'skip'; + +export type MaterializeSnapshotPublicationOptions = Readonly<{ + snapshotPublication: MaterializeSnapshotPublication; +}>; + +const PUBLISH_SNAPSHOT_OPTIONS: MaterializeSnapshotPublicationOptions = Object.freeze({ + snapshotPublication: 'publish', +}); + +const SKIP_SNAPSHOT_OPTIONS: MaterializeSnapshotPublicationOptions = Object.freeze({ + snapshotPublication: 'skip', +}); + +export function snapshotPublicationForReceipts( + opts: { receipts: boolean }, +): MaterializeSnapshotPublicationOptions { + if (opts.receipts) { + return SKIP_SNAPSHOT_OPTIONS; + } + return PUBLISH_SNAPSHOT_OPTIONS; +} + +export function shouldPublishMaterializeSnapshot( + options?: MaterializeSnapshotPublicationOptions, +): boolean { + return options === undefined || options.snapshotPublication === 'publish'; +} diff --git a/src/domain/services/controllers/MaterializeStrategyRuntime.ts b/src/domain/services/controllers/MaterializeStrategyRuntime.ts index f4c967cb7..c3fe484ba 100644 --- a/src/domain/services/controllers/MaterializeStrategyRuntime.ts +++ b/src/domain/services/controllers/MaterializeStrategyRuntime.ts @@ -7,6 +7,7 @@ import type { MaterializePatchStreamReduction, } from './MaterializePatchStreamReducer.ts'; import type { MaterializePatchSummary } from './MaterializePatchSummary.ts'; +import type { MaterializeSnapshotPublicationOptions } from './MaterializeSnapshotPublication.ts'; import type { MaterializeDeps, MaterializePersistence, @@ -40,8 +41,17 @@ export type MaterializeResultBuildInput = { export type MaterializeStrategyRuntime = { deps: MaterializeDeps; - emptyResult(ceiling?: number | null, frontier?: Map | null): Promise; - wrapState(state: WarpState, ceiling: number | null, frontier: Map | null): Promise; + emptyResult( + ceiling?: number | null, + frontier?: Map | null, + options?: MaterializeSnapshotPublicationOptions, + ): Promise; + wrapState( + state: WarpState, + ceiling: number | null, + frontier: Map | null, + options?: MaterializeSnapshotPublicationOptions, + ): Promise; reducePatches( patches: PatchWithSha[], base: WarpState | undefined, diff --git a/src/domain/warp/RuntimePatchCollector.ts b/src/domain/warp/RuntimePatchCollector.ts index f04903a30..be586b138 100644 --- a/src/domain/warp/RuntimePatchCollector.ts +++ b/src/domain/warp/RuntimePatchCollector.ts @@ -24,6 +24,7 @@ type RuntimePatchCollectorHost = { _loadLatestCheckpoint(): Promise; _loadPatchesSince(checkpoint: RuntimeCheckpointData): Promise; getFrontier(): Promise>; + _isAncestor?(ancestorSha: string, descendantSha: string): Promise; }; function isProvenanceIndexShape(value: object | null | undefined): value is NonNullable { @@ -108,6 +109,13 @@ export default class RuntimePatchCollector extends PatchCollector { return await this._runtime._loadPatchChainFromSha(toSha, fromSha); } + override async isAncestor(ancestorSha: string, descendantSha: string): Promise { + if (typeof this._runtime._isAncestor !== 'function') { + return false; + } + return await this._runtime._isAncestor(ancestorSha, descendantSha); + } + async getFrontier(): Promise> { return await this._runtime.getFrontier(); } diff --git a/test/unit/domain/RuntimePatchCollector.stream.test.ts b/test/unit/domain/RuntimePatchCollector.stream.test.ts index f10658a45..000794cc8 100644 --- a/test/unit/domain/RuntimePatchCollector.stream.test.ts +++ b/test/unit/domain/RuntimePatchCollector.stream.test.ts @@ -48,4 +48,61 @@ describe('RuntimePatchCollector streams', () => { expect(collected).toEqual(streamed); expect(host._loadPatchChainFromSha).toHaveBeenCalledWith('tip-sha'); }); + + it('streams only patches after the base coordinate writer tip', async () => { + const target = new Map([['agent-1', 'tip-sha']]); + const base = { + frontier: new Map([['agent-1', 'base-sha']]), + ceiling: null, + }; + const entries = [ + patchEntry(2, 'sha-2'), + patchEntry(3, 'tip-sha'), + ]; + const host = { + discoverWriters: vi.fn(async () => ['agent-1']), + _loadWriterPatches: vi.fn(async () => []), + _loadPatchChainFromSha: vi.fn(async () => entries), + _loadLatestCheckpoint: vi.fn(async () => null), + _loadPatchesSince: vi.fn(async () => []), + getFrontier: vi.fn(async () => target), + }; + const collector = new RuntimePatchCollector(host); + + const streamed = await collect( + collector.streamForFrontierSinceCoordinate(target, null, base), + ); + + expect(streamed.map((entry) => entry.sha)).toEqual(['sha-2', 'tip-sha']); + expect(host._loadPatchChainFromSha).toHaveBeenCalledWith('tip-sha', 'base-sha'); + }); + + it('streams same-tip patches above a base ceiling', async () => { + const target = new Map([['agent-1', 'tip-sha']]); + const base = { + frontier: new Map([['agent-1', 'tip-sha']]), + ceiling: 1, + }; + const entries = [ + patchEntry(1, 'sha-1'), + patchEntry(2, 'sha-2'), + patchEntry(3, 'tip-sha'), + ]; + const host = { + discoverWriters: vi.fn(async () => ['agent-1']), + _loadWriterPatches: vi.fn(async () => []), + _loadPatchChainFromSha: vi.fn(async () => entries), + _loadLatestCheckpoint: vi.fn(async () => null), + _loadPatchesSince: vi.fn(async () => []), + getFrontier: vi.fn(async () => target), + }; + const collector = new RuntimePatchCollector(host); + + const streamed = await collect( + collector.streamForFrontierSinceCoordinate(target, 3, base), + ); + + expect(streamed.map((entry) => entry.sha)).toEqual(['sha-2', 'tip-sha']); + expect(host._loadPatchChainFromSha).toHaveBeenCalledWith('tip-sha', null); + }); }); diff --git a/test/unit/domain/services/controllers/MaterializeController.snapshotCache.test.ts b/test/unit/domain/services/controllers/MaterializeController.snapshotCache.test.ts index 56b4b8296..c9406db2c 100644 --- a/test/unit/domain/services/controllers/MaterializeController.snapshotCache.test.ts +++ b/test/unit/domain/services/controllers/MaterializeController.snapshotCache.test.ts @@ -62,8 +62,9 @@ async function* streamFromPromise(items: Promise): AsyncIterable { function createControllerFixtures() { const stateCache = { - getExact: vi.fn<(_coordinate: Coordinate) => Promise>(), - getBestCompatiblePredecessor: vi.fn<(_coordinate: Coordinate) => Promise>(), + getExact: vi.fn<(_coordinate: Coordinate) => Promise>().mockResolvedValue(null), + getBestCompatiblePredecessor: vi.fn<(_coordinate: Coordinate) => Promise>() + .mockResolvedValue(null), put: vi.fn(), pin: vi.fn(), publishCheckpointHead: vi.fn(), @@ -83,6 +84,7 @@ function createControllerFixtures() { loadPatchesSince: vi.fn<(_checkpoint: CheckpointData) => Promise>().mockResolvedValue([]), loadPatchChain: vi.fn<(_toSha: string, _fromSha?: string | null) => Promise>().mockResolvedValue([]), getFrontier: vi.fn().mockResolvedValue(new Map([['writer-1', 'tip-7']])), + isAncestor: vi.fn<(_ancestorSha: string, _descendantSha: string) => Promise>().mockResolvedValue(true), streamWriterPatches: vi.fn((writerId: string) => streamFromPromise(patches.loadWriterPatches(writerId))), streamForFrontier: vi.fn((frontier: Map, ceiling: number | null) => streamFromPromise(patches.collectForFrontier(frontier, ceiling))), @@ -141,6 +143,248 @@ describe('MaterializeController — unified snapshot cache', () => { expect(patches.loadCheckpoint).toHaveBeenCalled(); }); + it('uses an exact snapshot hit for live materialization before replay', async () => { + const { controller, stateCache, patches } = createControllerFixtures(); + const coordinate: Coordinate = { + frontier: new Map([['writer-1', 'tip-7']]), + ceiling: null, + }; + + stateCache.getExact.mockResolvedValue( + snapshotRecord('snapshot-live-exact', coordinate, 'full'), + ); + + const result = await controller.materialize(); + + expect(patches.getFrontier).toHaveBeenCalled(); + expect(stateCache.getExact).toHaveBeenCalledWith(coordinate); + expect(stateCache.getBestCompatiblePredecessor).not.toHaveBeenCalled(); + expect(stateCache.put).not.toHaveBeenCalled(); + expect(patches.loadCheckpoint).not.toHaveBeenCalled(); + expect(patches.loadWriterPatches).not.toHaveBeenCalled(); + expect(result.patchCount).toBe(0); + expect(result.frontier).toEqual(coordinate.frontier); + expect(result.ceiling).toBe(null); + }); + + it('replays only the live suffix after the best compatible predecessor snapshot', async () => { + const { controller, stateCache, patches } = createControllerFixtures(); + const target: Coordinate = { + frontier: new Map([['writer-1', 'tip-7']]), + ceiling: null, + }; + const predecessor = snapshotRecord( + 'snapshot-live-predecessor', + { + frontier: new Map([['writer-1', 'tip-5']]), + ceiling: null, + }, + 'full', + ); + + stateCache.getExact.mockResolvedValue(null); + stateCache.getBestCompatiblePredecessor.mockResolvedValue(predecessor); + patches.collectForFrontierSinceCoordinate.mockResolvedValue([ + patchRecord(6, 'sha-6'), + patchRecord(7, 'sha-7'), + ]); + + const result = await controller.materialize(); + + expect(patches.getFrontier).toHaveBeenCalled(); + expect(stateCache.getBestCompatiblePredecessor).toHaveBeenCalledWith(target); + expect(patches.collectForFrontierSinceCoordinate).toHaveBeenCalledWith( + target.frontier, + target.ceiling, + predecessor.coordinate, + ); + expect(patches.loadCheckpoint).not.toHaveBeenCalled(); + expect(patches.loadWriterPatches).not.toHaveBeenCalled(); + expect(result.patchCount).toBe(2); + expect(result.frontier).toEqual(target.frontier); + }); + + it('publishes a live snapshot with the current frontier after replay', async () => { + const { controller, stateCache, patches } = createControllerFixtures(); + const target: Coordinate = { + frontier: new Map([['writer-1', 'tip-7']]), + ceiling: null, + }; + + stateCache.getExact.mockResolvedValue(null); + stateCache.getBestCompatiblePredecessor.mockResolvedValue(null); + patches.collectForFrontier.mockResolvedValue([ + patchRecord(1, 'sha-1'), + patchRecord(2, 'sha-2'), + ]); + + const result = await controller.materialize(); + + expect(patches.getFrontier).toHaveBeenCalled(); + expect(patches.collectForFrontier).toHaveBeenCalledWith(target.frontier, null); + expect(stateCache.put).toHaveBeenCalledWith( + expect.objectContaining({ + snapshotId: 'snapshot:state-hash-1', + coordinate: target, + retention: 'evictable', + provenancePosture: 'full', + stateHash: 'state-hash-1', + state: result.state, + }), + ); + expect(result.patchCount).toBe(2); + expect(result.frontier).toEqual(target.frontier); + }); + + it('bypasses live snapshot hits when diff materialization is requested', async () => { + const { controller, stateCache, patches } = createControllerFixtures(); + const target: Coordinate = { + frontier: new Map([['writer-1', 'tip-7']]), + ceiling: null, + }; + + stateCache.getExact.mockResolvedValue( + snapshotRecord('snapshot-live-exact', target, 'full'), + ); + patches.collectForFrontier.mockResolvedValue([ + patchRecord(7, 'sha-7'), + ]); + + const result = await controller.materialize({ wantDiff: true }); + + expect(stateCache.getExact).not.toHaveBeenCalled(); + expect(stateCache.getBestCompatiblePredecessor).not.toHaveBeenCalled(); + expect(patches.collectForFrontier).toHaveBeenCalledWith( + target.frontier, + target.ceiling, + ); + expect(result.diff).toBeDefined(); + expect(result.frontier).toEqual(target.frontier); + }); + + it('bypasses live snapshot hits when receipts are requested', async () => { + const { controller, stateCache, patches } = createControllerFixtures(); + const target: Coordinate = { + frontier: new Map([['writer-1', 'tip-7']]), + ceiling: null, + }; + + stateCache.getExact.mockResolvedValue( + snapshotRecord('snapshot-live-exact', target, 'full'), + ); + patches.collectForFrontier.mockResolvedValue([ + patchRecord(7, 'sha-7'), + ]); + + const result = await controller.materialize({ receipts: true }); + + expect(stateCache.getExact).not.toHaveBeenCalled(); + expect(stateCache.getBestCompatiblePredecessor).not.toHaveBeenCalled(); + expect(stateCache.put).not.toHaveBeenCalled(); + expect(patches.collectForFrontier).toHaveBeenCalledWith( + target.frontier, + target.ceiling, + ); + expect(result.receipts).toBeDefined(); + expect(result.frontier).toEqual(target.frontier); + }); + + it('does not cache empty live receipt materialization results', async () => { + const { controller, stateCache, patches } = createControllerFixtures(); + const emptyFrontier = new Map(); + + patches.getFrontier.mockResolvedValue(emptyFrontier); + + const result = await controller.materialize({ receipts: true }); + + expect(stateCache.getExact).not.toHaveBeenCalled(); + expect(stateCache.getBestCompatiblePredecessor).not.toHaveBeenCalled(); + expect(stateCache.put).not.toHaveBeenCalled(); + expect(result.frontier).toEqual(emptyFrontier); + }); + + it('binds checkpoint fallback replay to the live frontier coordinate', async () => { + const { controller, stateCache, patches } = createControllerFixtures(); + const target: Coordinate = { + frontier: new Map([['writer-1', 'tip-7']]), + ceiling: null, + }; + const checkpoint: CheckpointData = { + state: createEmptyState(), + frontier: new Map([['writer-1', 'tip-5']]), + stateHash: 'checkpoint-hash', + schema: 5, + }; + + stateCache.getExact.mockResolvedValue(null); + stateCache.getBestCompatiblePredecessor.mockResolvedValue(null); + patches.loadCheckpoint.mockResolvedValue(checkpoint); + patches.collectForFrontierSinceCoordinate.mockResolvedValue([ + patchRecord(6, 'sha-6'), + patchRecord(7, 'sha-7'), + ]); + + const result = await controller.materialize(); + + expect(patches.collectForFrontierSinceCoordinate).toHaveBeenCalledWith( + target.frontier, + null, + { + frontier: checkpoint.frontier, + ceiling: null, + }, + ); + expect(patches.loadPatchesSince).not.toHaveBeenCalled(); + expect(stateCache.put).toHaveBeenCalledWith( + expect.objectContaining({ + coordinate: target, + state: result.state, + }), + ); + expect(result.patchCount).toBe(2); + expect(result.frontier).toEqual(target.frontier); + }); + + it('falls back to live frontier replay when the checkpoint is ahead of the captured coordinate', async () => { + const { controller, stateCache, patches } = createControllerFixtures(); + const target: Coordinate = { + frontier: new Map([['writer-1', 'tip-7']]), + ceiling: null, + }; + const checkpoint: CheckpointData = { + state: createEmptyState(), + frontier: new Map([['writer-1', 'tip-9']]), + stateHash: 'checkpoint-hash', + schema: 5, + }; + + stateCache.getExact.mockResolvedValue(null); + stateCache.getBestCompatiblePredecessor.mockResolvedValue(null); + patches.loadCheckpoint.mockResolvedValue(checkpoint); + patches.isAncestor.mockResolvedValue(false); + patches.collectForFrontier.mockResolvedValue([ + patchRecord(6, 'sha-6'), + patchRecord(7, 'sha-7'), + ]); + + const result = await controller.materialize(); + + expect(patches.isAncestor).toHaveBeenCalledWith('tip-9', 'tip-7'); + expect(patches.collectForFrontier).toHaveBeenCalledWith( + target.frontier, + target.ceiling, + ); + expect(patches.collectForFrontierSinceCoordinate).not.toHaveBeenCalled(); + expect(stateCache.put).toHaveBeenCalledWith( + expect.objectContaining({ + coordinate: target, + state: result.state, + }), + ); + expect(result.patchCount).toBe(2); + expect(result.frontier).toEqual(target.frontier); + }); + it('uses an exact snapshot hit for coordinate materialization before replay', async () => { const { controller, stateCache, patches } = createControllerFixtures(); const coordinate: Coordinate = { @@ -155,6 +399,7 @@ describe('MaterializeController — unified snapshot cache', () => { const result = await controller.materializeCoordinate(coordinate); expect(stateCache.getExact).toHaveBeenCalledWith(coordinate); + expect(stateCache.put).not.toHaveBeenCalled(); expect(patches.collectForFrontier).not.toHaveBeenCalled(); expect(result.patchCount).toBe(0); expect(result.provenanceDegraded).toBe(false); @@ -193,6 +438,52 @@ describe('MaterializeController — unified snapshot cache', () => { expect(result.patchCount).toBe(1); }); + it('bypasses coordinate snapshot hits when receipts are requested', async () => { + const { controller, stateCache, patches } = createControllerFixtures(); + const coordinate: Coordinate = { + frontier: new Map([['writer-1', 'tip-7']]), + ceiling: 7, + }; + + stateCache.getExact.mockResolvedValue( + snapshotRecord('snapshot-exact', coordinate, 'full'), + ); + patches.collectForFrontier.mockResolvedValue([ + patchRecord(7, 'sha-7'), + ]); + + const result = await controller.materializeCoordinate({ + frontier: coordinate.frontier, + ceiling: coordinate.ceiling, + receipts: true, + }); + + expect(stateCache.getExact).not.toHaveBeenCalled(); + expect(stateCache.getBestCompatiblePredecessor).not.toHaveBeenCalled(); + expect(stateCache.put).not.toHaveBeenCalled(); + expect(patches.collectForFrontier).toHaveBeenCalledWith( + coordinate.frontier, + coordinate.ceiling, + ); + expect(result.receipts).toBeDefined(); + }); + + it('does not cache empty coordinate receipt materialization results', async () => { + const { controller, stateCache, patches } = createControllerFixtures(); + const frontier = new Map(); + + const result = await controller.materializeCoordinate({ + frontier, + receipts: true, + }); + + expect(stateCache.getExact).not.toHaveBeenCalled(); + expect(stateCache.getBestCompatiblePredecessor).not.toHaveBeenCalled(); + expect(stateCache.put).not.toHaveBeenCalled(); + expect(patches.collectForFrontier).not.toHaveBeenCalled(); + expect(result.frontier).toEqual(frontier); + }); + it('refuses a degraded predecessor snapshot for provenance-rich materialization', async () => { const { controller, stateCache, patches } = createControllerFixtures(); const target: Coordinate = { @@ -218,7 +509,8 @@ describe('MaterializeController — unified snapshot cache', () => { receipts: true, }); - expect(stateCache.getBestCompatiblePredecessor).toHaveBeenCalledWith(target); + expect(stateCache.getExact).not.toHaveBeenCalled(); + expect(stateCache.getBestCompatiblePredecessor).not.toHaveBeenCalled(); expect(patches.collectForFrontierSinceCoordinate).not.toHaveBeenCalled(); expect(patches.collectForFrontier).toHaveBeenCalledWith( target.frontier, diff --git a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts index 8317b3564..ba7fe2aad 100644 --- a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts +++ b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts @@ -181,8 +181,7 @@ function createControllerFixtures() { describe("MaterializeController — state session integration", () => { it("replays live materialization through StateSession and returns an explicit WarpState projection bridge", async () => { const { controller, patches, openStateSession } = createControllerFixtures(); - patches.discoverWriters.mockResolvedValue(["writer-1"]); - patches.loadWriterPatches.mockResolvedValue([ + patches.collectForFrontier.mockResolvedValue([ nodeAddPatchRecord({ writer: "writer-1", lamport: 1,