diff --git a/docs/topics/cas-first-memoized-materialization.md b/docs/topics/cas-first-memoized-materialization.md index a796352e..25383b5d 100644 --- a/docs/topics/cas-first-memoized-materialization.md +++ b/docs/topics/cas-first-memoized-materialization.md @@ -40,11 +40,19 @@ operation follows this coordinate-first lifecycle: [current frontier] | v -[state-cache exact hit?] ---- yes ---> [reopen retained roots; zero patch replay] +[retained exact hit?] ------- yes ---> [load basis; reuse roots; zero patch replay] | no v -[compatible predecessor?] --- yes ---> [replay suffix, publish snapshot] +[retained predecessor?] ----- yes ---> [load basis; replay suffix] + | + no + v +[state-cache exact hit?] ---- yes ---> [reopen roots; zero patch replay] + | + no + v +[state-cache predecessor?] -- yes ---> [replay suffix, publish snapshot] | no v @@ -62,7 +70,12 @@ WARP state coordinate: This coordinate belongs to `git-warp`; it is not a `git-cas` concept. -### 2. Check the WARP state cache +### 2. Check retained materializations, then the WARP state cache + +The compatibility path first asks `MaterializationStorePort` for an exact +retained materialization and then for a causally compatible retained +predecessor. Only when neither retained resume applies does it consult the +legacy `WarpStateCachePort`. The runtime asks `WarpStateCachePort` for an exact snapshot at that coordinate. On a hit, it asks `MaterializationStorePort` for the matching retained-root @@ -84,6 +97,16 @@ that cached coordinate, then publish a fresh snapshot for the current frontier. Until cache payloads carry provenance indexes, that derived snapshot retains a degraded provenance posture rather than claiming support for the cached prefix. +Retained materializations can now satisfy the same two resume cases without a +separate state-cache hit. Descriptor schema v4 retains a canonical replay basis +beside the node, edge, and property roots. An exact retained hit validates and +loads that basis, reuses the retained roots, and performs no patch replay. When +there is no exact hit, the adapter inspects at most 1,024 current-schema cache +entries in pages of 100. It validates their descriptors and coordinates, checks +causal ancestry, and resumes the newest compatible predecessor by replaying only +the suffix. Receipt-producing reads still use the ordinary replay path, as do +diff-producing predecessor reads. + ### 3. Fall back to replay and publish When there is no usable cached snapshot, the runtime falls back to the existing @@ -126,13 +149,14 @@ state-session opener owns its root storage and encoding, so git-warp does not pair it with the default reader and instead preserves the compatibility fallback. -The property-root contract advances the retained-materialization descriptor and -coordinate cache key to schema v3. A v3 exact miss leaves any corresponding v2 -entry anchored until replacement succeeds. Successful v3 retention then removes -the incompatible v2 anchor through the git-cas cache API. A v2 descriptor may -still be structurally valid, but it cannot satisfy the v3 root profile because -its property root may be unavailable. New v3 descriptors reject an unavailable -property root; an empty graph records the root as explicitly empty. +The replay-basis contract advances the retained-materialization descriptor and +coordinate cache key to schema v4. A v4 exact miss leaves corresponding legacy +entries anchored until replacement succeeds. Successful v4 retention then +removes incompatible v2 and v3 anchors through the git-cas cache API. Legacy +descriptors may still be structurally valid, but they cannot satisfy the v4 root +profile because their property or replay-basis root may be unavailable. New v4 +descriptors reject either root as unavailable; an empty graph still records the +property root as explicitly empty. ## `git-cas` Encapsulation @@ -201,8 +225,13 @@ lost payload bytes, or run Git garbage collection. other compatibility operations still own process-resident whole state. - Exact state-cache hits bypass replay, but full materialization still hydrates a full `WarpState`, scans retained node/edge tries, and builds full adjacency. +- Retained exact and predecessor resume load a complete canonical `WarpState` + replay basis before they reuse roots or replay a suffix. This eliminates + redundant prefix replay but is still a whole-state compatibility bridge, not + the bounded-memory observer representation. - Retained materialization descriptors currently carry node/edge trie roots and - a per-node property-shard root. Frontier, edge-birth, adjacency, + a per-node property-shard root plus the full-state replay basis. Frontier, + edge-birth, adjacency, provenance-support, and roaring roots remain explicitly unavailable until their paged representations land. Cold property-root construction still projects a complete `WarpState`; only exact retained reads avoid that state. @@ -225,6 +254,9 @@ lost payload bytes, or run Git garbage collection. 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. +- Compatible-predecessor lookup is deliberately bounded to 1,024 inspected + materialization entries. Exceeding that bound fails closed instead of silently + selecting from an incomplete cache scan. - Retention repair cannot restore payload objects that Git has already pruned; those entries remain visible as doctor findings until normal cache lifecycle replacement or explicit operator cleanup. diff --git a/src/domain/materialization/MaterializationRoots.ts b/src/domain/materialization/MaterializationRoots.ts index 1d3ba2a5..38607fda 100644 --- a/src/domain/materialization/MaterializationRoots.ts +++ b/src/domain/materialization/MaterializationRoots.ts @@ -9,6 +9,7 @@ export const MATERIALIZATION_ROOT_NAMES = defineRootNames( 'node-alive', 'properties', 'provenance-support', + 'replay-basis', 'roaring-indexes', ); @@ -22,6 +23,7 @@ export type MaterializationRootsOptions = Readonly<{ nodeAlive: MaterializationRoot; properties: MaterializationRoot; provenanceSupport: MaterializationRoot; + replayBasis: MaterializationRoot; roaringIndexes: MaterializationRoot; }>; @@ -35,6 +37,7 @@ export default class MaterializationRoots { readonly nodeAlive: MaterializationRoot; readonly properties: MaterializationRoot; readonly provenanceSupport: MaterializationRoot; + readonly replayBasis: MaterializationRoot; readonly roaringIndexes: MaterializationRoot; constructor(options: MaterializationRootsOptions) { @@ -47,6 +50,7 @@ export default class MaterializationRoots { 'node-alive': requireRoot(options.nodeAlive, 'nodeAlive'), properties: requireRoot(options.properties, 'properties'), 'provenance-support': requireRoot(options.provenanceSupport, 'provenanceSupport'), + 'replay-basis': requireRoot(options.replayBasis, 'replayBasis'), 'roaring-indexes': requireRoot(options.roaringIndexes, 'roaringIndexes'), } satisfies Record); this.adjacency = this.roots.adjacency; @@ -56,6 +60,7 @@ export default class MaterializationRoots { this.nodeAlive = this.roots['node-alive']; this.properties = this.roots.properties; this.provenanceSupport = this.roots['provenance-support']; + this.replayBasis = this.roots['replay-basis']; this.roaringIndexes = this.roots['roaring-indexes']; Object.freeze(this); } diff --git a/src/domain/materialization/TrieMaterializationReader.ts b/src/domain/materialization/TrieMaterializationReader.ts index cbbdbb55..36579dcb 100644 --- a/src/domain/materialization/TrieMaterializationReader.ts +++ b/src/domain/materialization/TrieMaterializationReader.ts @@ -23,12 +23,14 @@ export default class TrieMaterializationReader extends MaterializationReadPort { readonly #codec: CodecPort; readonly #geometry: TrieGeometry; readonly #indexStore: IndexStorePort | null; + readonly #pageCache: PageCache; constructor(options: { readonly store: TrieStorePort; readonly codec: CodecPort; readonly geometry?: TrieGeometry; readonly indexStore?: IndexStorePort; + readonly pageCache?: PageCache; }) { super(); requireOptions(options); @@ -40,6 +42,9 @@ export default class TrieMaterializationReader extends MaterializationReadPort { this.#indexStore = options.indexStore === undefined ? null : requireIndexStore(options.indexStore); + this.#pageCache = options.pageCache === undefined + ? new PageCache({ maxResident: MAX_RESIDENT_READ_PAGES }) + : requirePageCache(options.pageCache); Object.freeze(this); } @@ -55,7 +60,7 @@ export default class TrieMaterializationReader extends MaterializationReadPort { store: this.#store, geometry: this.#geometry, codec: this.#codec, - pageCache: new PageCache({ maxResident: MAX_RESIDENT_READ_PAGES }), + pageCache: this.#pageCache, }); return await cursor.contains(nodeId); } @@ -142,6 +147,13 @@ function requireIndexStore(indexStore: IndexStorePort): IndexStorePort { return indexStore; } +function requirePageCache(pageCache: PageCache): PageCache { + if (!(pageCache instanceof PageCache)) { + throw readerError('pageCache must be a PageCache instance'); + } + return pageCache; +} + function readerError(message: string): WarpError { return new WarpError(`Materialization reader ${message}`, 'E_MATERIALIZATION_RESUME'); } diff --git a/src/domain/services/controllers/MaterializationRetention.ts b/src/domain/services/controllers/MaterializationRetention.ts new file mode 100644 index 00000000..136cacb4 --- /dev/null +++ b/src/domain/services/controllers/MaterializationRetention.ts @@ -0,0 +1,99 @@ +import MaterializationCoordinate from '../../materialization/MaterializationCoordinate.ts'; +import type MaterializationHandle from '../../materialization/MaterializationHandle.ts'; +import WarpError from '../../errors/WarpError.ts'; +import { + materializationSessionOpen, +} from './MaterializeSessionBridge.ts'; +import type { + MaterializeReduceOutput, +} from './MaterializeController.ts'; +import type { MaterializeDeps } from './MaterializeDeps.ts'; +import type { + MaterializeResultBuildInput, +} from './MaterializeStrategyRuntime.ts'; + +/** Publishes session roots through their git-cas workspace retention scope. */ +export async function resolveMaterializationRetention(input: { + readonly deps: MaterializeDeps; + readonly params: MaterializeResultBuildInput; + readonly stateHash: string; +}): Promise { + const retained = resolveExistingMaterialization(input.params, input.stateHash); + return retained ?? await publishMaterialization(input); +} + +function resolveExistingMaterialization( + params: MaterializeResultBuildInput, + stateHash: string, +): MaterializationHandle | undefined { + const retained = params.materialization; + if (retained === undefined) { + return undefined; + } + if (retained.stateHash !== stateHash) { + throw retentionError('retained handle state hash does not match resumed state'); + } + if (!rootsMatch(params, retained)) { + return undefined; + } + params.reduced.acceptMaterialization?.(retained.retention); + return retained; +} + +function rootsMatch( + params: MaterializeResultBuildInput, + retained: MaterializationHandle, +): boolean { + return params.reduced.roots === undefined + || retained.roots.equals(params.reduced.roots); +} + +async function publishMaterialization(input: { + readonly deps: MaterializeDeps; + readonly params: MaterializeResultBuildInput; + readonly stateHash: string; +}): Promise { + const { params } = input; + if (params.reduced.roots === undefined || params.frontier === null) { + await acceptSessionWithoutMaterialization(params.reduced); + return undefined; + } + const request = { + coordinate: new MaterializationCoordinate({ + frontier: params.frontier, + ceiling: params.ceiling, + }), + roots: params.reduced.roots, + stateHash: input.stateHash, + replayBasis: params.reduced.state, + }; + const materialization = params.reduced.workspace === undefined + ? await input.deps.materializations.retain(request) + : await params.reduced.workspace.promote(request); + params.reduced.acceptMaterialization?.(materialization.retention); + return materialization; +} + +async function acceptSessionWithoutMaterialization( + reduced: MaterializeReduceOutput, +): Promise { + if (reduced.acceptMaterialization === undefined) { + return; + } + if (reduced.roots === undefined || reduced.workspace === undefined) { + throw retentionError('prepared session is missing roots or workspace retention'); + } + const roots = materializationSessionOpen(reduced.roots); + if (roots === null) { + throw retentionError('prepared session roots cannot be checkpointed'); + } + const witness = await reduced.workspace.checkpoint({ + nodeAliveRoot: roots.nodeAliveRootOid, + edgeAliveRoot: roots.edgeAliveRootOid, + }); + reduced.acceptMaterialization(witness); +} + +function retentionError(message: string): WarpError { + return new WarpError(message, 'E_MATERIALIZATION_RESUME'); +} diff --git a/src/domain/services/controllers/MaterializeController.ts b/src/domain/services/controllers/MaterializeController.ts index 0bc2065f..c8e9675c 100644 --- a/src/domain/services/controllers/MaterializeController.ts +++ b/src/domain/services/controllers/MaterializeController.ts @@ -53,6 +53,9 @@ import { releaseAcquisitionAfterFailure, releaseWorkspaceAfterFailure, } from './MaterializationWorkspaceCleanup.ts'; +import { + resolveMaterializationRetention, +} from './MaterializationRetention.ts'; import type { MaterializeDeps } from './MaterializeDeps.ts'; export type { MaterializeDeps, MaterializePersistence } from './MaterializeDeps.ts'; @@ -235,7 +238,11 @@ export default class MaterializeController { try { const stateHash = await computeHash(this._deps, params.reduced.state); const adjacency = params.reduced.adjacency ?? buildAdjacency(params.reduced.state); - const materialization = await this._resolveMaterialization(params, stateHash); + const materialization = await resolveMaterializationRetention({ + deps: this._deps, + params, + stateHash, + }); if (params.reduced.receipts === undefined && params.publishSnapshot !== false) { await this._publishSnapshot({ state: params.reduced.state, @@ -268,61 +275,6 @@ export default class MaterializeController { await params.reduced.workspace?.release(); return result; } - private async _resolveMaterialization( - params: MaterializeResultBuildInput, - stateHash: string, - ): Promise { - if (params.materialization !== undefined) { - if (params.materialization.stateHash !== stateHash) { - throw materializationResumeError('retained handle state hash does not match resumed state'); - } - if ( - params.reduced.roots === undefined - || params.materialization.roots.equals(params.reduced.roots) - ) { - params.reduced.acceptMaterialization?.(params.materialization.retention); - return params.materialization; - } - } - if (params.reduced.roots === undefined || params.frontier === null) { - await this._acceptSessionWithoutMaterialization(params.reduced); - return undefined; - } - const request = { - coordinate: new MaterializationCoordinate({ - frontier: params.frontier, - ceiling: params.ceiling, - }), - roots: params.reduced.roots, - stateHash, - }; - const materialization = params.reduced.workspace === undefined - ? await this._deps.materializations.retain(request) - : await params.reduced.workspace.promote(request); - params.reduced.acceptMaterialization?.(materialization.retention); - return materialization; - } - - private async _acceptSessionWithoutMaterialization( - reduced: MaterializeReduceOutput, - ): Promise { - if (reduced.acceptMaterialization === undefined) { - return; - } - if (reduced.roots === undefined || reduced.workspace === undefined) { - throw materializationResumeError('prepared session is missing roots or workspace retention'); - } - const roots = materializationSessionOpen(reduced.roots); - if (roots === null) { - throw materializationResumeError('prepared session roots cannot be checkpointed'); - } - const witness = await reduced.workspace.checkpoint({ - nodeAliveRoot: roots.nodeAliveRootOid, - edgeAliveRoot: roots.edgeAliveRootOid, - }); - reduced.acceptMaterialization(witness); - } - private async _resumeExactMaterialization( snapshot: UsableSnapshotRecord, options: { wantDiff: boolean }, @@ -356,6 +308,9 @@ export default class MaterializeController { ...(retainedRoots === null || retained === null ? {} : { propertyRoot: retained.roots.properties }), + ...(retainedRoots === null || retained === null + ? {} + : { replayBasisRoot: retained.roots.replayBasis }), receipts: false, wantDiff: options.wantDiff, }); @@ -410,6 +365,7 @@ export default class MaterializeController { opts: MaterializePatchStreamOptions, coordinate: WarpStateCoordinate, provenanceBase?: ProvenanceIndex, + resumeFrom?: MaterializationHandle, ): Promise { if (this._deps.openStateSession === undefined) { return await MaterializePatchStreamReducer.reduce({ @@ -420,6 +376,9 @@ export default class MaterializeController { }); } const summary = new MaterializePatchSummaryAccumulator(provenanceBase); + const retainedRoots = resumeFrom === undefined + ? null + : materializationSessionOpen(resumeFrom.roots); const recordingStream = async function* (): AsyncIterable { for await (const entry of stream) { summary.record(entry); @@ -438,6 +397,13 @@ export default class MaterializeController { receipts: opts.receipts, wantDiff: opts.wantDiff, ...(base === undefined ? {} : { baseState: base }), + ...(retainedRoots === null ? {} : { roots: retainedRoots }), + ...(retainedRoots === null || resumeFrom === undefined + ? {} + : { + propertyRoot: resumeFrom.roots.properties, + replayBasisRoot: resumeFrom.roots.replayBasis, + }), }); return { reduced, @@ -454,8 +420,21 @@ export default class MaterializeController { await this._wrapState(state, ceiling, frontier, provenance, options), reducePatches: async (patches, base, opts, coordinate) => await this._reducePatches(patches, base, opts, coordinate), - reducePatchStream: async (stream, base, opts, coordinate, provenanceBase) => - await this._reducePatchStream(stream, base, opts, coordinate, provenanceBase), + reducePatchStream: async ( + stream, + base, + opts, + coordinate, + provenanceBase, + resumeFrom, + ) => await this._reducePatchStream( + stream, + base, + opts, + coordinate, + provenanceBase, + resumeFrom, + ), buildResult: async (params) => await this._buildResult(params), resumeExactMaterialization: async (snapshot, options) => await this._resumeExactMaterialization(snapshot, options), diff --git a/src/domain/services/controllers/MaterializeHelpers.ts b/src/domain/services/controllers/MaterializeHelpers.ts index b3ea4cc0..a90c8d63 100644 --- a/src/domain/services/controllers/MaterializeHelpers.ts +++ b/src/domain/services/controllers/MaterializeHelpers.ts @@ -19,6 +19,10 @@ import { collectVisibleEdgesFromSession, } from '../state/SessionVisibleGraph.ts'; +export function isNonEmptyPatchSha(value: string | undefined): value is string { + return typeof value === 'string' && value.length > 0; +} + // ── Public state freezing ─────────────────────────────────────────── /** Wraps materialized state in a frozen defensive copy. */ diff --git a/src/domain/services/controllers/MaterializeLiveStrategy.ts b/src/domain/services/controllers/MaterializeLiveStrategy.ts index 1a65c6f1..0ac1b8d7 100644 --- a/src/domain/services/controllers/MaterializeLiveStrategy.ts +++ b/src/domain/services/controllers/MaterializeLiveStrategy.ts @@ -25,16 +25,16 @@ import { } from './MaterializeSnapshotCacheResult.ts'; import { snapshotPublicationForReceipts } from './MaterializeSnapshotPublication.ts'; import { releaseAcquisitionAfterFailure } from './MaterializationWorkspaceCleanup.ts'; - -function nonEmptySha(value: string | undefined): value is string { - return typeof value === 'string' && value.length > 0; -} +import RetainedMaterializationResumeStrategy from './RetainedMaterializationResumeStrategy.ts'; +import { isNonEmptyPatchSha } from './MaterializeHelpers.ts'; export default class MaterializeLiveStrategy { private readonly runtime: MaterializeStrategyRuntime; + private readonly retainedResume: RetainedMaterializationResumeStrategy; constructor(runtime: MaterializeStrategyRuntime) { this.runtime = runtime; + this.retainedResume = new RetainedMaterializationResumeStrategy(runtime); } async materialize(opts: MaterializeLiveOptions): Promise { @@ -43,16 +43,12 @@ export default class MaterializeLiveStrategy { return await this.runtime.emptyResult(null, frontier, snapshotPublicationForLiveOptions(opts)); } const coordinate = this.snapshotCoordinate(frontier); - const stateCache = this.runtime.deps.getStateCache?.() ?? null; - const cacheResolved = await this.tryResolveConfiguredStateCache( - stateCache, - coordinate, - opts, - ); - if (cacheResolved !== null) { - return cacheResolved; + const retainedResolved = await this.retainedResume.tryResume(coordinate, opts); + if (retainedResolved !== null) { + return retainedResolved; } - return await this.replayCurrentCoordinate(coordinate, opts); + const stateCache = this.runtime.deps.getStateCache?.() ?? null; + return await this.resolveConfiguredStateCache(stateCache, coordinate, opts); } async resolveMaterialization(): Promise { @@ -139,19 +135,22 @@ export default class MaterializeLiveStrategy { } } - private async tryResolveConfiguredStateCache( + private async resolveConfiguredStateCache( stateCache: WarpStateCachePort | null, coordinate: WarpStateCoordinate, opts: MaterializeLiveOptions, - ): Promise { - if (stateCache === null) { - return null; + ): Promise { + if (stateCache !== null) { + const resolved = await this.tryResolveSnapshotCache(stateCache, { + coordinate, + receipts: opts.receipts, + wantDiff: opts.wantDiff, + }); + if (resolved !== null) { + return resolved; + } } - return await this.tryResolveSnapshotCache(stateCache, { - coordinate, - receipts: opts.receipts, - wantDiff: opts.wantDiff, - }); + return await this.replayCurrentCoordinate(coordinate, opts); } private async replayCurrentCoordinate( @@ -193,7 +192,7 @@ export default class MaterializeLiveStrategy { checkpointTip: string, targetTip: string | undefined, ): Promise { - if (!nonEmptySha(checkpointTip) || !nonEmptySha(targetTip)) { + if (!isNonEmptyPatchSha(checkpointTip) || !isNonEmptyPatchSha(targetTip)) { return false; } return checkpointTip === targetTip || await this.checkpointTipPrecedesTarget(checkpointTip, targetTip); diff --git a/src/domain/services/controllers/MaterializeSessionBridge.ts b/src/domain/services/controllers/MaterializeSessionBridge.ts index 7443df39..edf4d3ec 100644 --- a/src/domain/services/controllers/MaterializeSessionBridge.ts +++ b/src/domain/services/controllers/MaterializeSessionBridge.ts @@ -54,6 +54,7 @@ export async function reduceSessionBackedState(args: { readonly logger?: LoggerPort; readonly propertyStore?: IndexStorePort; readonly propertyRoot?: MaterializationRoot; + readonly replayBasisRoot?: MaterializationRoot; readonly coordinate: MaterializationCoordinate; readonly patches: MaterializeSessionPatchSource; readonly baseState?: WarpStateClass; @@ -127,7 +128,10 @@ export async function reduceSessionBackedState(args: { reducedPatchCount === 0 ? args.propertyRoot : undefined, ); await retainPreparedPropertyRoot(workspace, close.roots, properties); - const roots = materializationRootsFromSession(close.roots, properties); + const replayBasis = reducedPatchCount === 0 && args.replayBasisRoot !== undefined + ? args.replayBasisRoot + : MaterializationRoot.unavailable(); + const roots = materializationRootsFromSession(close.roots, properties, replayBasis); return { ...reduced, roots, @@ -189,6 +193,7 @@ export function materializationSessionOpen( function materializationRootsFromSession( roots: MaterializeSessionOpen, properties: MaterializationRoot, + replayBasis: MaterializationRoot, ): MaterializationRoots { return new MaterializationRoots({ adjacency: MaterializationRoot.unavailable(), @@ -198,6 +203,7 @@ function materializationRootsFromSession( nodeAlive: sessionMaterializationRoot(roots.nodeAliveRootOid), properties, provenanceSupport: MaterializationRoot.unavailable(), + replayBasis, roaringIndexes: MaterializationRoot.unavailable(), }); } diff --git a/src/domain/services/controllers/MaterializeStrategyRuntime.ts b/src/domain/services/controllers/MaterializeStrategyRuntime.ts index 9ffee379..484c45d4 100644 --- a/src/domain/services/controllers/MaterializeStrategyRuntime.ts +++ b/src/domain/services/controllers/MaterializeStrategyRuntime.ts @@ -72,6 +72,7 @@ export type MaterializeStrategyRuntime = { opts: MaterializePatchStreamOptions, coordinate: WarpStateCoordinate, provenanceBase?: ProvenanceIndex, + resumeFrom?: MaterializationHandle, ): Promise; buildResult(params: MaterializeResultBuildInput): Promise; resumeExactMaterialization( diff --git a/src/domain/services/controllers/RetainedMaterializationResumeStrategy.ts b/src/domain/services/controllers/RetainedMaterializationResumeStrategy.ts new file mode 100644 index 00000000..ea1922fd --- /dev/null +++ b/src/domain/services/controllers/RetainedMaterializationResumeStrategy.ts @@ -0,0 +1,206 @@ +import MaterializationCoordinate from '../../materialization/MaterializationCoordinate.ts'; +import type MaterializationHandle from '../../materialization/MaterializationHandle.ts'; +import type { MaterializationAcquisition } from '../../../ports/MaterializationStorePort.ts'; +import type { + WarpStateCoordinate, +} from '../../../ports/WarpStateCachePort.ts'; +import type { PatchWithSha } from '../../capabilities/PatchCollector.ts'; +import WarpStream from '../../stream/WarpStream.ts'; +import type { MaterializeResult } from './MaterializeController.ts'; +import type { + MaterializeLiveOptions, + MaterializeStrategyRuntime, +} from './MaterializeStrategyRuntime.ts'; +import { releaseAcquisitionAfterFailure } from './MaterializationWorkspaceCleanup.ts'; + +/** Resolves exact or causally compatible retained materialization state. */ +export default class RetainedMaterializationResumeStrategy { + readonly #runtime: MaterializeStrategyRuntime; + + constructor(runtime: MaterializeStrategyRuntime) { + this.#runtime = runtime; + } + + async tryResume( + coordinate: WarpStateCoordinate, + options: MaterializeLiveOptions, + ): Promise { + const exact = await this.#tryExact(coordinate, options); + return exact ?? await this.#tryPredecessor(coordinate, options); + } + + async #tryExact( + coordinate: WarpStateCoordinate, + options: MaterializeLiveOptions, + ): Promise { + if (options.receipts) { + return null; + } + const acquisition = await this.#runtime.deps.materializations.acquireExact( + new MaterializationCoordinate(coordinate), + ); + if (acquisition === null) { + return null; + } + return await this.#completeAcquisition( + acquisition, + async () => await this.#resumeExact(acquisition, coordinate, options), + ); + } + + async #resumeExact( + acquisition: MaterializationAcquisition, + coordinate: WarpStateCoordinate, + options: MaterializeLiveOptions, + ): Promise { + const { materialization } = acquisition; + const basis = await this.#runtime.deps.materializations.loadReplayBasis(materialization); + if (basis === null) { + return null; + } + const reduction = await this.#runtime.reducePatchStream( + emptyPatchStream(), + basis, + { receipts: false, wantDiff: options.wantDiff }, + coordinate, + undefined, + materialization, + ); + return await this.#runtime.buildResult({ + reduced: reduction.reduced, + summary: reduction.summary, + degraded: true, + ceiling: coordinate.ceiling, + frontier: coordinate.frontier, + materialization, + publishSnapshot: false, + }); + } + + async #tryPredecessor( + coordinate: WarpStateCoordinate, + options: MaterializeLiveOptions, + ): Promise { + if (options.receipts || options.wantDiff) { + return null; + } + const target = new MaterializationCoordinate(coordinate); + const acquisition = await this.#runtime.deps.materializations + .acquireBestCompatiblePredecessor( + target, + async (candidate) => await this.#coordinatePrecedes(candidate, target), + ); + if (acquisition === null) { + return null; + } + return await this.#completeAcquisition( + acquisition, + async () => await this.#resumePredecessor(acquisition, coordinate, options), + ); + } + + async #resumePredecessor( + acquisition: MaterializationAcquisition, + coordinate: WarpStateCoordinate, + options: MaterializeLiveOptions, + ): Promise { + const { materialization } = acquisition; + const basis = await this.#runtime.deps.materializations.loadReplayBasis(materialization); + if (basis === null) { + return null; + } + const reduction = await this.#runtime.reducePatchStream( + this.#suffixStream(materialization, coordinate), + basis, + { receipts: false, wantDiff: false }, + coordinate, + undefined, + materialization, + ); + return await this.#runtime.buildResult({ + reduced: reduction.reduced, + summary: reduction.summary, + degraded: true, + ceiling: coordinate.ceiling, + frontier: coordinate.frontier, + ...(options.publishSnapshot === undefined + ? {} + : { publishSnapshot: options.publishSnapshot }), + }); + } + + #suffixStream( + materialization: MaterializationHandle, + coordinate: WarpStateCoordinate, + ): AsyncIterable { + return this.#runtime.deps.patches.streamForFrontierSinceCoordinate( + coordinate.frontier, + coordinate.ceiling, + { + frontier: materialization.coordinate.frontier(), + ceiling: materialization.coordinate.ceiling, + }, + ); + } + + async #coordinatePrecedes( + candidate: MaterializationCoordinate, + target: MaterializationCoordinate, + ): Promise { + if (!ceilingPrecedes(candidate.ceiling, target.ceiling)) { + return false; + } + const targetFrontier = target.frontier(); + for (const entry of candidate.frontierEntries) { + if (!await this.#frontierEntryPrecedes( + entry.writerId, + entry.patchSha, + targetFrontier, + )) { + return false; + } + } + return true; + } + + async #frontierEntryPrecedes( + writerId: string, + candidateTip: string, + targetFrontier: ReadonlyMap, + ): Promise { + const targetTip = targetFrontier.get(writerId); + if (targetTip === undefined) { + return false; + } + return candidateTip === targetTip || await this.#isAncestor(candidateTip, targetTip); + } + + async #isAncestor(candidate: string, target: string): Promise { + const { patches } = this.#runtime.deps; + return typeof patches.isAncestor === 'function' + && await patches.isAncestor(candidate, target); + } + + async #completeAcquisition( + acquisition: MaterializationAcquisition, + operation: () => Promise, + ): Promise { + let result: MaterializeResult | null; + try { + result = await operation(); + } catch (raw) { + await releaseAcquisitionAfterFailure(acquisition, this.#runtime.deps.logger); + throw raw; + } + await acquisition.release(); + return result; + } +} + +function ceilingPrecedes(candidate: number | null, target: number | null): boolean { + return target === null || (candidate !== null && candidate <= target); +} + +function emptyPatchStream(): AsyncIterable { + return WarpStream.from([]); +} diff --git a/src/domain/warp/RuntimeHostBoot.ts b/src/domain/warp/RuntimeHostBoot.ts index 109d2488..f3deaa88 100644 --- a/src/domain/warp/RuntimeHostBoot.ts +++ b/src/domain/warp/RuntimeHostBoot.ts @@ -343,6 +343,7 @@ export async function resolveRuntimeHostConstructionOptions( } else if (storageServices.trie !== undefined) { const store = storageServices.trie; const geometry = TrieGeometry.default16way(); + const pageCache = new PageCache({ maxResident: 256 }); resolvedOpenStateSession = async (roots, sessionOptions) => await StateSession.open({ nodeAliveRootOid: roots.nodeAliveRootOid, @@ -350,7 +351,7 @@ export async function resolveRuntimeHostConstructionOptions( store, codec: resolvedCodec, geometry, - pageCache: new PageCache({ maxResident: 256 }), + pageCache, workspace: sessionOptions.workspace, }); // A custom session opener owns its root encoding; pair this reader only @@ -359,6 +360,7 @@ export async function resolveRuntimeHostConstructionOptions( store, codec: resolvedCodec, geometry, + pageCache, indexStore: resolvedIndexStore, }); } diff --git a/src/infrastructure/adapters/GitCasMaterializationCacheKey.ts b/src/infrastructure/adapters/GitCasMaterializationCacheKey.ts new file mode 100644 index 00000000..0e8bb7e7 --- /dev/null +++ b/src/infrastructure/adapters/GitCasMaterializationCacheKey.ts @@ -0,0 +1,55 @@ +import type MaterializationCoordinate from '../../domain/materialization/MaterializationCoordinate.ts'; +import type CodecPort from '../../ports/CodecPort.ts'; +import type CryptoPort from '../../ports/CryptoPort.ts'; +import { + MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION, + materializationCoordinateData, +} from './GitCasMaterializationDescriptor.ts'; +import { requireNonEmpty } from './GitCasMaterializationStoreValidation.ts'; + +/** Lane-scoped identity for current materialization cache entries. */ +export default class GitCasMaterializationCacheKey { + readonly #codec: CodecPort; + readonly #crypto: CryptoPort; + readonly #laneName: string; + + constructor(options: { + readonly codec: CodecPort; + readonly crypto: CryptoPort; + readonly laneName: string; + }) { + this.#codec = options.codec; + this.#crypto = options.crypto; + this.#laneName = options.laneName; + } + + async forCoordinate( + coordinate: MaterializationCoordinate, + schemaVersion = MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION, + ): Promise { + const digest = await this.#digest({ + schemaVersion, + laneName: this.#laneName, + coordinate: materializationCoordinateData(coordinate), + }, 'coordinate digest'); + if (schemaVersion === MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION) { + return `${await this.currentPrefix()}${digest}`; + } + return `v${String(schemaVersion)}:${digest}`; + } + + async currentPrefix(): Promise { + const digest = await this.#digest({ + schemaVersion: MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION, + laneName: this.#laneName, + }, 'lane digest'); + return `v${String(MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION)}:${digest}:`; + } + + async #digest(value: object, field: string): Promise { + return requireNonEmpty( + await this.#crypto.hash('sha256', this.#codec.encode(value)), + field, + ); + } +} diff --git a/src/infrastructure/adapters/GitCasMaterializationDescriptor.ts b/src/infrastructure/adapters/GitCasMaterializationDescriptor.ts index 946a0db6..c372638b 100644 --- a/src/infrastructure/adapters/GitCasMaterializationDescriptor.ts +++ b/src/infrastructure/adapters/GitCasMaterializationDescriptor.ts @@ -9,7 +9,7 @@ import MaterializationRoots, { import type BundleHandle from '../../domain/storage/BundleHandle.ts'; import WarpError from '../../domain/errors/WarpError.ts'; -export const MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION = 3; +export const MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION = 4; export type DecodedMaterializationDescriptor = Readonly<{ coordinate: MaterializationCoordinate; @@ -75,6 +75,7 @@ export function materializationRootsFromDescriptor( nodeAlive: rootFromMaps(statuses, retainedRoots, 'node-alive'), properties: rootFromMaps(statuses, retainedRoots, 'properties'), provenanceSupport: rootFromMaps(statuses, retainedRoots, 'provenance-support'), + replayBasis: rootFromMaps(statuses, retainedRoots, 'replay-basis'), roaringIndexes: rootFromMaps(statuses, retainedRoots, 'roaring-indexes'), }); } diff --git a/src/infrastructure/adapters/GitCasMaterializationPredecessorResolver.ts b/src/infrastructure/adapters/GitCasMaterializationPredecessorResolver.ts new file mode 100644 index 00000000..f49635da --- /dev/null +++ b/src/infrastructure/adapters/GitCasMaterializationPredecessorResolver.ts @@ -0,0 +1,145 @@ +import type { + CacheEntryMetadata, + CacheInspection, + CacheSet, +} from '@git-stunts/git-cas'; +import type MaterializationCoordinate from '../../domain/materialization/MaterializationCoordinate.ts'; +import BundleHandle from '../../domain/storage/BundleHandle.ts'; +import type { + MaterializationPredecessorPredicate, +} from '../../ports/MaterializationStorePort.ts'; +import type { + DecodedMaterializationDescriptor, +} from './GitCasMaterializationDescriptor.ts'; + +const MAX_CACHE_INSPECTION_PAGE = 100; +const MAX_MATERIALIZATION_CANDIDATES = 1024; + +type MaterializationCandidate = Readonly<{ + coordinate: MaterializationCoordinate; + createdAt: string; + key: string; +}>; + +type InspectionCache = Pick; + +/** Finds the newest bounded, validated retained predecessor in the git-cas cache. */ +export default class GitCasMaterializationPredecessorResolver { + readonly #openCache: () => Promise; + readonly #laneName: string; + readonly #readDescriptor: ( + bundle: BundleHandle, + ) => Promise; + readonly #cacheKey: (coordinate: MaterializationCoordinate) => Promise; + readonly #cacheKeyPrefix: () => Promise; + + constructor(options: { + readonly openCache: () => Promise; + readonly laneName: string; + readonly readDescriptor: ( + bundle: BundleHandle, + ) => Promise; + readonly cacheKey: (coordinate: MaterializationCoordinate) => Promise; + readonly cacheKeyPrefix: () => Promise; + }) { + this.#openCache = options.openCache; + this.#laneName = options.laneName; + this.#readDescriptor = options.readDescriptor; + this.#cacheKey = options.cacheKey; + this.#cacheKeyPrefix = options.cacheKeyPrefix; + } + + async find( + target: MaterializationCoordinate, + isCompatible: MaterializationPredecessorPredicate, + ): Promise { + const cache = await this.#openCache(); + const entries = await inspectEntries(cache, await this.#cacheKeyPrefix()); + if (entries === null) { + return null; + } + let best: MaterializationCandidate | null = null; + for (const entry of entries) { + const candidate = await this.#candidateFromEntry(entry, target, isCompatible); + best = selectBetterCandidate(candidate, best); + } + return best?.coordinate ?? null; + } + + async #candidateFromEntry( + entry: CacheEntryMetadata, + target: MaterializationCoordinate, + isCompatible: MaterializationPredecessorPredicate, + ): Promise { + const descriptor = await this.#readDescriptor(new BundleHandle(entry.handle)); + if (!descriptorCanResume(descriptor, target, this.#laneName)) { + return null; + } + if (entry.key !== await this.#cacheKey(descriptor.coordinate)) { + return null; + } + if (!await isCompatible(descriptor.coordinate)) { + return null; + } + return Object.freeze({ + coordinate: descriptor.coordinate, + createdAt: entry.createdAt, + key: entry.key, + }); + } +} + +async function inspectEntries( + cache: InspectionCache, + cacheKeyPrefix: string, +): Promise { + const entries: CacheEntryMetadata[] = []; + let cursor: string | null = null; + do { + const page: CacheInspection = await cache.inspect({ + limit: MAX_CACHE_INSPECTION_PAGE, + cursor, + }); + for (const entry of page.entries) { + if (!entry.key.startsWith(cacheKeyPrefix)) { + continue; + } + if (entries.length === MAX_MATERIALIZATION_CANDIDATES) { + return null; + } + entries.push(entry); + } + cursor = page.nextCursor; + } while (cursor !== null); + return Object.freeze(entries); +} + +function descriptorCanResume( + descriptor: DecodedMaterializationDescriptor, + target: MaterializationCoordinate, + laneName: string, +): boolean { + return descriptor.laneName === laneName + && !descriptor.coordinate.equals(target) + && descriptor.rootStatuses.get('replay-basis') === 'retained'; +} + +function candidateIsBetter( + candidate: MaterializationCandidate, + current: MaterializationCandidate | null, +): boolean { + if (current === null || candidate.createdAt > current.createdAt) { + return true; + } + return candidate.createdAt === current.createdAt && candidate.key > current.key; +} + +function selectBetterCandidate( + candidate: MaterializationCandidate | null, + current: MaterializationCandidate | null, +): MaterializationCandidate | null { + if (candidate === null) { + return current; + } + return candidateIsBetter(candidate, current) ? candidate : current; +} diff --git a/src/infrastructure/adapters/GitCasMaterializationReplayBasis.ts b/src/infrastructure/adapters/GitCasMaterializationReplayBasis.ts new file mode 100644 index 00000000..e77818c0 --- /dev/null +++ b/src/infrastructure/adapters/GitCasMaterializationReplayBasis.ts @@ -0,0 +1,118 @@ +import type { + AssetCapability, + AssetHandle, + BundleCapability, + BundleMemberReference, +} from '@git-stunts/git-cas'; +import MaterializationHandle from '../../domain/materialization/MaterializationHandle.ts'; +import MaterializationRoot from '../../domain/materialization/MaterializationRoot.ts'; +import MaterializationRoots from '../../domain/materialization/MaterializationRoots.ts'; +import type WarpState from '../../domain/services/state/WarpState.ts'; +import { computeStateHash } from '../../domain/services/state/StateSerializer.ts'; +import WarpStream from '../../domain/stream/WarpStream.ts'; +import BundleHandle from '../../domain/storage/BundleHandle.ts'; +import { collectAsyncIterable } from '../../domain/utils/streamUtils.ts'; +import type CodecPort from '../../ports/CodecPort.ts'; +import type CryptoPort from '../../ports/CryptoPort.ts'; +import { + decodeCanonicalWarpFullState, + encodeWarpFullState, +} from '../codecs/WarpStateCborCodec.ts'; +import type { + GitCasStagingWorkspace, +} from './GitCasMaterializationWorkspace.ts'; +import { storageError } from './GitCasMaterializationStoreValidation.ts'; + +const REPLAY_BASIS_PATH = 'state.cbor'; + +type ReplayBasisFacade = Readonly<{ + assets: Pick; + bundles: Pick; +}>; + +/** Stages and verifies the compatibility state attached to retained materializations. */ +export default class GitCasMaterializationReplayBasis { + readonly #cas: ReplayBasisFacade; + readonly #codec: CodecPort; + readonly #crypto: CryptoPort; + + constructor(options: { + readonly cas: ReplayBasisFacade; + readonly codec: CodecPort; + readonly crypto: CryptoPort; + }) { + this.#cas = options.cas; + this.#codec = options.codec; + this.#crypto = options.crypto; + } + + async stage( + workspace: GitCasStagingWorkspace, + state: WarpState, + ): Promise { + const bytes = encodeWarpFullState(state, this.#codec); + const asset = await workspace.assets.put({ + source: WarpStream.from([bytes]), + slug: 'git-warp-materialization-replay-basis', + filename: REPLAY_BASIS_PATH, + }); + const bundle = await workspace.bundles.putOrdered({ + members: [[REPLAY_BASIS_PATH, asset.handle]], + limits: { maxMembers: 1 }, + }); + return MaterializationRoot.retained(new BundleHandle(bundle.handle.toString())); + } + + async load(materialization: MaterializationHandle): Promise { + if (!(materialization instanceof MaterializationHandle)) { + throw storageError('replay basis requires a MaterializationHandle'); + } + const root = materialization.roots.replayBasis; + if (root.status !== 'retained' || root.handle === null) { + return null; + } + const member = await this.#cas.bundles.getMemberReference({ + handle: root.handle.toString(), + path: REPLAY_BASIS_PATH, + }); + const asset = requireReplayAsset(member); + const bytes = await collectAsyncIterable(this.#cas.assets.open({ handle: asset })); + const state = decodeCanonicalWarpFullState(bytes, this.#codec); + await this.#requireMatchingHash(state, materialization.stateHash); + return state; + } + + async #requireMatchingHash(state: WarpState, expected: string): Promise { + const actual = await computeStateHash(state, { + codec: this.#codec, + crypto: this.#crypto, + }); + if (actual !== expected) { + throw storageError('replay basis state hash does not match its descriptor'); + } + } +} + +export function replaceReplayBasisRoot( + roots: MaterializationRoots, + replayBasis: MaterializationRoot, +): MaterializationRoots { + return new MaterializationRoots({ + adjacency: roots.adjacency, + edgeAlive: roots.edgeAlive, + edgeBirths: roots.edgeBirths, + frontier: roots.frontier, + nodeAlive: roots.nodeAlive, + properties: roots.properties, + provenanceSupport: roots.provenanceSupport, + replayBasis, + roaringIndexes: roots.roaringIndexes, + }); +} + +function requireReplayAsset(member: BundleMemberReference | null): AssetHandle { + if (member === null || member.handle.kind !== 'asset') { + throw storageError('replay basis root has no state asset'); + } + return member.handle; +} diff --git a/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts b/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts index 1c199e8a..235b780d 100644 --- a/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts +++ b/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts @@ -1,15 +1,12 @@ import type { - BundleCapability, CacheAcquisition, CacheHit, - CacheSet, PageHandle, - PageCapability, WorkspaceRetainedBundle, - WorkspaceRetainedPage, } from '@git-stunts/git-cas'; import type MaterializationCoordinate from '../../domain/materialization/MaterializationCoordinate.ts'; import MaterializationHandle from '../../domain/materialization/MaterializationHandle.ts'; +import type WarpState from '../../domain/services/state/WarpState.ts'; import BundleHandle from '../../domain/storage/BundleHandle.ts'; import type StorageRetentionWitness from '../../domain/storage/StorageRetentionWitness.ts'; import type CodecPort from '../../ports/CodecPort.ts'; @@ -17,6 +14,7 @@ import type CryptoPort from '../../ports/CryptoPort.ts'; import type MaterializationWorkspacePort from '../../ports/MaterializationWorkspacePort.ts'; import MaterializationStorePort, { type MaterializationAcquisition, + type MaterializationPredecessorPredicate, type RetainMaterializationRequest, } from '../../ports/MaterializationStorePort.ts'; import { adaptGitCasRetentionWitness } from './GitCasRetentionWitnessAdapter.ts'; @@ -33,11 +31,25 @@ import { storageError, } from './GitCasMaterializationStoreValidation.ts'; import GitCasMaterializationLease from './GitCasMaterializationLease.ts'; +import GitCasMaterializationPredecessorResolver from './GitCasMaterializationPredecessorResolver.ts'; +import GitCasMaterializationReplayBasis, { + replaceReplayBasisRoot, +} from './GitCasMaterializationReplayBasis.ts'; +import GitCasMaterializationCacheKey from './GitCasMaterializationCacheKey.ts'; +import type { + GitCasMaterializationFacade, + MaterializationCacheSet, +} from './GitCasMaterializationStoreTypes.ts'; +import { + releaseCacheAcquisitionAfterFailure, + requireDescriptorSize, + requireExpectedAcquisition, + requireStoredMaterialization, + requireWorkspaceStage, +} from './GitCasMaterializationStoreWitness.ts'; import { completeWithCleanup } from './OperationCleanup.ts'; import { decodeMaterializationDescriptor, - MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION, - materializationCoordinateData, materializationDescriptorData, materializationRootsFromDescriptor, type DecodedMaterializationDescriptor, @@ -52,32 +64,20 @@ const CACHE_NAMESPACE = 'git-warp/materializations'; const WORKSPACE_NAMESPACE = 'git-warp/materializations'; const WORKSPACE_TTL_MS = 2 * 60 * 60 * 1000; const MAX_DESCRIPTOR_BYTES = 1024 * 1024; -const LEGACY_MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION = 2; +const LEGACY_MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSIONS = Object.freeze([2, 3]); -type MaterializationCacheSet = Pick; -type MaterializationCachePut = Awaited>; - -export type GitCasMaterializationFacade = { - readonly bundles: Pick; - readonly caches: { - open(options: { readonly namespace: string }): Promise; - }; - readonly pages: Pick; - readonly workspaces: { - open(options: { - readonly namespace: string; - readonly ttlMs?: number; - }): Promise; - }; -}; +export type { GitCasMaterializationFacade } from './GitCasMaterializationStoreTypes.ts'; /** git-cas-backed retained materialization lifecycle. */ export default class GitCasMaterializationStoreAdapter extends MaterializationStorePort { readonly #cas: GitCasMaterializationFacade; readonly #codec: CodecPort; readonly #crypto: CryptoPort; + readonly #cacheKeys: GitCasMaterializationCacheKey; readonly #laneName: string; readonly #onClose: () => void; + readonly #predecessorResolver: GitCasMaterializationPredecessorResolver; + readonly #replayBasis: GitCasMaterializationReplayBasis; #currentLease: GitCasMaterializationLease | null = null; #leaseMutation: Promise = Promise.resolve(); readonly #retirements = new Set>(); @@ -104,7 +104,31 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt this.#codec = options.codec; this.#crypto = options.crypto; this.#laneName = requireNonEmpty(options.laneName, 'laneName'); + this.#cacheKeys = new GitCasMaterializationCacheKey({ + codec: this.#codec, + crypto: this.#crypto, + laneName: this.#laneName, + }); this.#onClose = options.onClose ?? (() => undefined); + this.#replayBasis = new GitCasMaterializationReplayBasis({ + cas: this.#cas, + codec: this.#codec, + crypto: this.#crypto, + }); + this.#predecessorResolver = this.#createPredecessorResolver(); + } + + #createPredecessorResolver(): GitCasMaterializationPredecessorResolver { + return new GitCasMaterializationPredecessorResolver({ + openCache: async () => await this.#cas.caches.open({ namespace: CACHE_NAMESPACE }), + laneName: this.#laneName, + readDescriptor: async (bundle) => { + const members = await this.#readMembers(bundle); + return await this.#readDescriptor(members.descriptor); + }, + cacheKey: async (coordinate) => await this.#cacheKeys.forCoordinate(coordinate), + cacheKeyPrefix: async () => await this.#cacheKeys.currentPrefix(), + }); } override async openWorkspace( @@ -147,7 +171,14 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt ): Promise { requireRetainRequest(request); const stateHash = requireNonEmpty(request.stateHash, 'stateHash'); - const bundle = await this.#stageWorkspaceBundle(workspace, request, stateHash); + const roots = request.replayBasis === undefined + ? request.roots + : replaceReplayBasisRoot( + request.roots, + await this.#replayBasis.stage(workspace, request.replayBasis), + ); + const retainedRequest = { ...request, roots }; + const bundle = await this.#stageWorkspaceBundle(workspace, retainedRequest, stateHash); const retention = await this.#promoteWorkspaceBundle( workspace, bundle, @@ -157,7 +188,7 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt laneName: this.#laneName, bundle: new BundleHandle(bundle.handle.toString()), coordinate: request.coordinate, - roots: request.roots, + roots, stateHash, retention, }); @@ -194,7 +225,7 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt coordinate: MaterializationCoordinate, ): Promise { const cache = await this.#cas.caches.open({ namespace: CACHE_NAMESPACE }); - const cacheKey = await this.#cacheKey(coordinate); + const cacheKey = await this.#cacheKeys.forCoordinate(coordinate); const expectedHandle = bundle.handle.toString(); const promoted = await workspace.promoteToCache({ cache, @@ -220,7 +251,7 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt await completeWithCleanup( async () => { requireExpectedAcquisition(acquisition, args.expectedHandle); - await this.#removeLegacyEntry(args.cache, args.coordinate); + await this.#removeLegacyEntries(args.cache, args.coordinate); }, async () => { await acquisition.release(); @@ -239,6 +270,28 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt ); } + override async acquireBestCompatiblePredecessor( + coordinate: MaterializationCoordinate, + isCompatible: MaterializationPredecessorPredicate, + ): Promise { + requireCoordinate(coordinate); + if (typeof isCompatible !== 'function') { + throw storageError('predecessor compatibility predicate must be a function'); + } + return await this.#withLeaseMutation( + async () => await this.#acquireBestCompatiblePredecessorLocked( + coordinate, + isCompatible, + ), + ); + } + + override async loadReplayBasis( + materialization: MaterializationHandle, + ): Promise { + return await this.#replayBasis.load(materialization); + } + override close(): Promise { if (this.#closePromise === null) { this.#closed = true; @@ -264,6 +317,24 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt return this.#replaceCurrentLease(next); } + async #acquireBestCompatiblePredecessorLocked( + coordinate: MaterializationCoordinate, + isCompatible: MaterializationPredecessorPredicate, + ): Promise { + if (this.#closed) { + throw storageError('adapter is closed'); + } + const candidate = await this.#predecessorResolver.find( + coordinate, + isCompatible, + ); + if (candidate === null) { + return null; + } + const next = await this.#openLease(candidate); + return next === null ? null : this.#replaceCurrentLease(next); + } + #replaceCurrentLease(next: GitCasMaterializationLease): MaterializationAcquisition { const previous = this.#currentLease; this.#currentLease = next; @@ -278,7 +349,7 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt coordinate: MaterializationCoordinate, ): Promise { const cache = await this.#cas.caches.open({ namespace: CACHE_NAMESPACE }); - const acquisition = await cache.acquire(await this.#cacheKey(coordinate)); + const acquisition = await cache.acquire(await this.#cacheKeys.forCoordinate(coordinate)); if (acquisition === null) { return null; } @@ -386,31 +457,13 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt }); } - async #cacheKey( - coordinate: MaterializationCoordinate, - schemaVersion = MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION, - ): Promise { - const encoded = this.#codec.encode({ - schemaVersion, - laneName: this.#laneName, - coordinate: materializationCoordinateData(coordinate), - }); - const digest = requireNonEmpty( - await this.#crypto.hash('sha256', encoded), - 'coordinate digest', - ); - return `v${String(schemaVersion)}:${digest}`; - } - - async #removeLegacyEntry( + async #removeLegacyEntries( cache: MaterializationCacheSet, coordinate: MaterializationCoordinate, ): Promise { - const key = await this.#cacheKey( - coordinate, - LEGACY_MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION, - ); - await cache.remove(key); + for (const schemaVersion of LEGACY_MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSIONS) { + await cache.remove(await this.#cacheKeys.forCoordinate(coordinate, schemaVersion)); + } } async #readDescriptor(handle: PageHandle): Promise { @@ -427,57 +480,3 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt })); } } - -async function releaseCacheAcquisitionAfterFailure( - acquisition: CacheAcquisition, -): Promise { - try { - await acquisition.release(); - } catch { - // git-cas doctor owns abandoned-acquisition diagnostics; preserve the primary failure. - } -} - -function requireWorkspaceStage( - staged: WorkspaceRetainedPage | WorkspaceRetainedBundle, -): void { - const valid = [ - staged.state === 'retained', - staged.retention.policy === 'evictable', - staged.retention.reachability === 'anchored', - staged.retention.protection === 'workspace', - staged.witness.handle.toString() === staged.handle.toString(), - staged.witness.root.kind === 'root-set', - ]; - if (valid.includes(false)) { - throw storageError('git-cas did not retain a staged materialization artifact'); - } -} - -function requireStoredMaterialization( - stored: MaterializationCachePut, - expectedHandle: string, -): Exclude { - if (!stored.accepted || stored.hit === null || stored.witness === null) { - throw storageError('git-cas did not retain the materialization bundle'); - } - if (stored.hit.handle.toString() !== expectedHandle) { - throw storageError('git-cas retained an unexpected materialization handle'); - } - return stored.witness; -} - -function requireExpectedAcquisition( - acquisition: CacheAcquisition, - expectedHandle: string, -): void { - if (acquisition.hit.handle.toString() !== expectedHandle) { - throw storageError('git-cas acquired an unexpected materialization before legacy cleanup'); - } -} - -function requireDescriptorSize(bytes: Uint8Array): void { - if (bytes.byteLength > MAX_DESCRIPTOR_BYTES) { - throw storageError('materialization descriptor exceeds its byte limit'); - } -} diff --git a/src/infrastructure/adapters/GitCasMaterializationStoreTypes.ts b/src/infrastructure/adapters/GitCasMaterializationStoreTypes.ts new file mode 100644 index 00000000..a67fa350 --- /dev/null +++ b/src/infrastructure/adapters/GitCasMaterializationStoreTypes.ts @@ -0,0 +1,36 @@ +import type { + AssetCapability, + BundleCapability, + CacheSet, + PageCapability, +} from '@git-stunts/git-cas'; +import type { + GitCasStagingWorkspace, +} from './GitCasMaterializationWorkspace.ts'; + +export type MaterializationCacheSet = Pick< + CacheSet, + 'acquire' | 'inspect' | 'put' | 'remove' | 'ref' +>; + +export type MaterializationCachePut = Awaited< + ReturnType +>; + +export type GitCasMaterializationFacade = { + readonly assets: Pick; + readonly bundles: Pick< + BundleCapability, + 'getMemberReference' | 'iterateMemberReferences' + >; + readonly caches: { + open(options: { readonly namespace: string }): Promise; + }; + readonly pages: Pick; + readonly workspaces: { + open(options: { + readonly namespace: string; + readonly ttlMs?: number; + }): Promise; + }; +}; diff --git a/src/infrastructure/adapters/GitCasMaterializationStoreWitness.ts b/src/infrastructure/adapters/GitCasMaterializationStoreWitness.ts new file mode 100644 index 00000000..93a99093 --- /dev/null +++ b/src/infrastructure/adapters/GitCasMaterializationStoreWitness.ts @@ -0,0 +1,63 @@ +import type { + CacheAcquisition, + WorkspaceRetainedBundle, + WorkspaceRetainedPage, +} from '@git-stunts/git-cas'; +import { storageError } from './GitCasMaterializationStoreValidation.ts'; +import type { + MaterializationCachePut, +} from './GitCasMaterializationStoreTypes.ts'; + +export async function releaseCacheAcquisitionAfterFailure( + acquisition: CacheAcquisition, +): Promise { + try { + await acquisition.release(); + } catch { + // git-cas doctor owns abandoned-acquisition diagnostics; preserve the primary failure. + } +} + +export function requireWorkspaceStage( + staged: WorkspaceRetainedPage | WorkspaceRetainedBundle, +): void { + const valid = [ + staged.state === 'retained', + staged.retention.policy === 'evictable', + staged.retention.reachability === 'anchored', + staged.retention.protection === 'workspace', + staged.witness.handle.toString() === staged.handle.toString(), + staged.witness.root.kind === 'root-set', + ]; + if (valid.includes(false)) { + throw storageError('git-cas did not retain a staged materialization artifact'); + } +} + +export function requireStoredMaterialization( + stored: MaterializationCachePut, + expectedHandle: string, +): Exclude { + if (!stored.accepted || stored.hit === null || stored.witness === null) { + throw storageError('git-cas did not retain the materialization bundle'); + } + if (stored.hit.handle.toString() !== expectedHandle) { + throw storageError('git-cas retained an unexpected materialization handle'); + } + return stored.witness; +} + +export function requireExpectedAcquisition( + acquisition: CacheAcquisition, + expectedHandle: string, +): void { + if (acquisition.hit.handle.toString() !== expectedHandle) { + throw storageError('git-cas acquired an unexpected materialization before legacy cleanup'); + } +} + +export function requireDescriptorSize(bytes: Uint8Array): void { + if (bytes.byteLength > 1024 * 1024) { + throw storageError('materialization descriptor exceeds its byte limit'); + } +} diff --git a/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts b/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts index a03d7b50..e9bcc551 100644 --- a/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts +++ b/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts @@ -25,7 +25,7 @@ import { adaptGitCasRetentionWitness } from './GitCasRetentionWitnessAdapter.ts' export type GitCasStagingWorkspace = Pick< StagingWorkspace, - 'pages' | 'bundles' | 'checkpoint' | 'release' + 'assets' | 'pages' | 'bundles' | 'checkpoint' | 'release' > & Readonly<{ promoteToCache(options: { cache: Pick; diff --git a/src/infrastructure/codecs/WarpStateCborCodec.ts b/src/infrastructure/codecs/WarpStateCborCodec.ts index 0dd16ff3..2a56aa91 100644 --- a/src/infrastructure/codecs/WarpStateCborCodec.ts +++ b/src/infrastructure/codecs/WarpStateCborCodec.ts @@ -52,6 +52,19 @@ export function decodeWarpFullState(buffer: Uint8Array, codec: CodecPort): WarpS return hydrateWarpState(obj); } +/** Decode only the canonical full-state envelope emitted by encodeWarpFullState. */ +export function decodeCanonicalWarpFullState(buffer: Uint8Array, codec: CodecPort): WarpState { + const decoded: unknown = codec.decode(buffer); + if (!isRecord(decoded) || decoded.version !== FULL_STATE_VERSION) { + throw invalidCanonicalFullState(); + } + const state = hydrateWarpState(decoded); + if (!equalBytes(buffer, encodeWarpFullState(state, codec))) { + throw invalidCanonicalFullState(); + } + return state; +} + function decodeFullStatePayload(buffer: Uint8Array | null | undefined, codec: CodecPort): DecodedFullState | null { if (buffer === null || buffer === undefined) { return null; @@ -214,3 +227,18 @@ function numberOrZero(value: unknown): number { function stringOr(value: unknown, fallback: string): string { return typeof value === 'string' ? value : fallback; } + +function isRecord(value: unknown): value is DecodedFullState & Record { + return value !== null && typeof value === 'object' && !Array.isArray(value); +} + +function equalBytes(left: Uint8Array, right: Uint8Array): boolean { + if (left.byteLength !== right.byteLength) { + return false; + } + return left.every((value, index) => value === right[index]); +} + +function invalidCanonicalFullState(): WarpError { + return new WarpError('Full state payload is not canonical', 'E_FULL_STATE_INVALID'); +} diff --git a/src/ports/MaterializationStorePort.ts b/src/ports/MaterializationStorePort.ts index 9852d4c5..5b329537 100644 --- a/src/ports/MaterializationStorePort.ts +++ b/src/ports/MaterializationStorePort.ts @@ -2,6 +2,7 @@ import type MaterializationCoordinate from '../domain/materialization/Materializ import type MaterializationHandle from '../domain/materialization/MaterializationHandle.ts'; import type MaterializationWorkspacePort from './MaterializationWorkspacePort.ts'; import type { PromoteMaterializationRequest } from './MaterializationWorkspacePort.ts'; +import type WarpState from '../domain/services/state/WarpState.ts'; export type RetainMaterializationRequest = PromoteMaterializationRequest; @@ -11,6 +12,10 @@ export type MaterializationAcquisition = Readonly<{ release(): Promise; }>; +export type MaterializationPredecessorPredicate = ( + _candidate: MaterializationCoordinate, +) => Promise; + /** Storage-neutral lifecycle for retained, independently addressable materializations. */ export default abstract class MaterializationStorePort { abstract openWorkspace( @@ -23,6 +28,17 @@ export default abstract class MaterializationStorePort { _coordinate: MaterializationCoordinate, ): Promise; + acquireBestCompatiblePredecessor( + _coordinate: MaterializationCoordinate, + _isCompatible: MaterializationPredecessorPredicate, + ): Promise { + return Promise.resolve(null); + } + + loadReplayBasis(_materialization: MaterializationHandle): Promise { + return Promise.resolve(null); + } + /** Releases runtime-local materialization resources without changing retained storage. */ close(): Promise { return Promise.resolve(); diff --git a/src/ports/MaterializationWorkspacePort.ts b/src/ports/MaterializationWorkspacePort.ts index cde0e455..08297da6 100644 --- a/src/ports/MaterializationWorkspacePort.ts +++ b/src/ports/MaterializationWorkspacePort.ts @@ -2,6 +2,7 @@ import type MaterializationCoordinate from '../domain/materialization/Materializ import type MaterializationHandle from '../domain/materialization/MaterializationHandle.ts'; import type MaterializationRoots from '../domain/materialization/MaterializationRoots.ts'; import type StorageRetentionWitness from '../domain/storage/StorageRetentionWitness.ts'; +import type WarpState from '../domain/services/state/WarpState.ts'; import ArtifactStagingPort from './ArtifactStagingPort.ts'; export type MaterializationWorkspaceRoots = Readonly<{ @@ -14,6 +15,7 @@ export type PromoteMaterializationRequest = Readonly<{ coordinate: MaterializationCoordinate; roots: MaterializationRoots; stateHash: string; + replayBasis?: WarpState; }>; /** diff --git a/test/helpers/InMemoryGitCasFacade.ts b/test/helpers/InMemoryGitCasFacade.ts index d21478ff..758a6cc7 100644 --- a/test/helpers/InMemoryGitCasFacade.ts +++ b/test/helpers/InMemoryGitCasFacade.ts @@ -15,6 +15,7 @@ import { type BundleCapability, type BundleMember, type CacheAcquisition, + type CacheInspection, type CacheSet, type CacheStoreResult, type PageHandleInput, @@ -22,6 +23,7 @@ import { type PublicationCapability, type WorkspaceCheckpointResult, type WorkspaceReleaseResult, + type WorkspaceRetainedAsset, type WorkspaceRetainedBundle, type WorkspaceRetainedPage, } from '@git-stunts/git-cas'; @@ -64,7 +66,7 @@ export default class InMemoryGitCasFacade { readonly caches: { open(options: { readonly namespace: string; - }): Promise>; + }): Promise>; }; readonly pages: Pick; readonly publications: Pick; @@ -342,7 +344,7 @@ export default class InMemoryGitCasFacade { async #openCache( namespace: string, - ): Promise> { + ): Promise> { const entries = this.#cacheEntries.get(namespace) ?? new Map(); this.#cacheEntries.set(namespace, entries); return Object.freeze({ @@ -383,6 +385,37 @@ export default class InMemoryGitCasFacade { }); }, get: async (key) => entries.get(key) ?? null, + inspect: async ( + { limit = 100, cursor = null }: { limit?: number; cursor?: string | null } = {}, + ): Promise => { + const candidates = [...entries.values()] + .sort((left, right) => left.key.localeCompare(right.key)) + .filter((hit) => cursor === null || hit.key > cursor); + const selected = candidates.slice(0, limit); + return Object.freeze({ + namespace, + ref: `refs/cas/caches/${namespace}`, + generation: null, + state: null, + observed: null, + policy: null, + entries: Object.freeze(selected.map((hit) => Object.freeze({ + version: 1 as const, + accountingVersion: 1 as const, + key: hit.key, + keyDigest: hit.key, + handle: hit.handle.toString(), + policy: hit.policy, + expiresAt: hit.expiresAt, + logicalBytes: hit.logicalBytes, + createdAt: hit.createdAt, + accessedAt: hit.accessedAt, + }))), + nextCursor: candidates.length > limit + ? selected.at(-1)?.key ?? null + : null, + }); + }, put: async (key, handle, options) => { const previous = entries.get(key) ?? null; const target = parseApplicationHandle(handle); @@ -512,6 +545,16 @@ export default class InMemoryGitCasFacade { }; return Object.freeze({ + assets: Object.freeze({ + put: async (request): Promise => { + const staged = await this.#putAsset(request); + return retainedAsset(staged, retain(staged.handle)); + }, + adopt: async ({ treeOid }): Promise => { + const staged = await this.#adoptAsset(treeOid); + return retainedAsset(staged, retain(staged.handle)); + }, + }), pages: Object.freeze({ put: async (request): Promise => { const staged = await this.#putPage(request); @@ -584,6 +627,32 @@ export default class InMemoryGitCasFacade { } } +function retainedAsset( + staged: StagedAsset, + witness: RetentionWitness, +): WorkspaceRetainedAsset { + const retention = Object.freeze({ + policy: 'evictable' as const, + reachability: 'anchored' as const, + protection: 'workspace' as const, + }); + return Object.freeze({ + version: staged.version, + state: 'retained', + handle: staged.handle, + asset: staged.asset, + retention, + witness, + observedAt: staged.observedAt, + toJSON: () => Object.freeze({ + ...staged.toJSON(), + state: 'retained' as const, + retention, + witness: witness.toJSON(), + }), + }); +} + function retainedPage( staged: StagedPage, witness: RetentionWitness, diff --git a/test/helpers/InMemoryMaterializationStore.ts b/test/helpers/InMemoryMaterializationStore.ts index 2b078760..c3230212 100644 --- a/test/helpers/InMemoryMaterializationStore.ts +++ b/test/helpers/InMemoryMaterializationStore.ts @@ -12,6 +12,12 @@ import MaterializationWorkspacePort, { type MaterializationWorkspaceRoots, type PromoteMaterializationRequest, } from '../../src/ports/MaterializationWorkspacePort.ts'; +import type WarpState from '../../src/domain/services/state/WarpState.ts'; +import { + decodeCanonicalWarpFullState, + encodeWarpFullState, +} from '../../src/infrastructure/codecs/WarpStateCborCodec.ts'; +import defaultCodec from '../../src/infrastructure/codecs/CborCodec.ts'; export class InMemoryMaterializationWorkspace extends MaterializationWorkspacePort { readonly checkpoints: MaterializationWorkspaceRoots[] = []; @@ -92,6 +98,7 @@ export default class InMemoryMaterializationStore extends MaterializationStorePo readonly retainedRequests: RetainMaterializationRequest[] = []; readonly workspaces: InMemoryMaterializationWorkspace[] = []; readonly #handles = new Map(); + readonly #replayBases = new Map(); #nextHandle = 1; override openWorkspace( @@ -120,6 +127,12 @@ export default class InMemoryMaterializationStore extends MaterializationStorePo retention: retentionWitness(bundle), }); this.#handles.set(coordinateKey(request.coordinate), handle); + if (request.replayBasis !== undefined) { + this.#replayBases.set( + bundle.toString(), + encodeWarpFullState(request.replayBasis, defaultCodec), + ); + } return Promise.resolve(handle); } @@ -135,6 +148,13 @@ export default class InMemoryMaterializationStore extends MaterializationStorePo this.acquisitions.push(acquisition); return Promise.resolve(acquisition); } + + override loadReplayBasis(materialization: MaterializationHandle): Promise { + const bytes = this.#replayBases.get(materialization.bundle.toString()); + return Promise.resolve(bytes === undefined + ? null + : decodeCanonicalWarpFullState(bytes, defaultCodec)); + } } function coordinateKey(coordinate: MaterializationCoordinate): string { diff --git a/test/integration/api/materialization.retainedResume.test.ts b/test/integration/api/materialization.retainedResume.test.ts index 3ae52673..d10df1f8 100644 --- a/test/integration/api/materialization.retainedResume.test.ts +++ b/test/integration/api/materialization.retainedResume.test.ts @@ -7,8 +7,11 @@ import type MaterializationHandle from '../../../src/domain/materialization/Mate import GitCasRepositoryAdapter from '../../../src/infrastructure/adapters/GitCasRepositoryAdapter.ts'; import MaterializationStorePort, { type MaterializationAcquisition, + type MaterializationPredecessorPredicate, type RetainMaterializationRequest, } from '../../../src/ports/MaterializationStorePort.ts'; +import type WarpState from '../../../src/domain/services/state/WarpState.ts'; +import { encodePropKey } from '../../../src/domain/services/KeyCodec.ts'; import MaterializationWorkspacePort, { type MaterializationWorkspaceRoots, type PromoteMaterializationRequest, @@ -71,7 +74,8 @@ describe('API: retained materialization resume', () => { const warm = await firstRuntime.materialize(); expect(sameRuntimeReplay).not.toHaveBeenCalled(); - expect(firstStore.exactLookups).toHaveLength(1); + expect(firstStore.exactLookups).toHaveLength(2); + expect(firstStore.exactHits).toHaveLength(1); expect(firstStore.retainedHandles).toHaveLength(1); expect(firstStore.exactHits[0]?.bundle.equals(coldHandle?.bundle)).toBe(true); expect(warm).toEqual(cold); @@ -173,6 +177,50 @@ describe('API: retained materialization resume', () => { expect(reopenedStore.exactHits).toHaveLength(1); expect(reopenedStore.exactReleaseCount).toBe(1); }); + + it('resumes a fresh runtime from a retained predecessor and applies only its suffix', async () => { + if (repo === null) { + throw new Error('Test repository is not initialized'); + } + const firstProvider = recordingProvider(repo, providers); + const firstRuntime = await openRuntime(repo, firstProvider); + await firstRuntime.patch((patch) => { + patch + .addNode('node:base') + .setProperty('node:base', 'status', 'created'); + }); + await firstRuntime.materialize(); + const coldHandle = requireMaterializations(firstProvider).retainedHandles[0]; + expect(coldHandle?.roots.replayBasis.status).toBe('retained'); + + await firstRuntime.patch((patch) => { + patch + .setProperty('node:base', 'status', 'updated') + .addNode('node:suffix') + .setProperty('node:suffix', 'status', 'added'); + }); + await firstRuntime.close(); + + const reopenedProvider = recordingProvider(repo, providers); + const reopenedRuntime = await openRuntime(repo, reopenedProvider); + const replay = vi.spyOn(reopenedRuntime, '_loadPatchChainFromSha'); + const resumed = await reopenedRuntime.materialize(); + const reopenedStore = requireMaterializations(reopenedProvider); + const coldTip = coldHandle?.coordinate.frontier().get('writer-1'); + + expect(replay).toHaveBeenCalledTimes(1); + expect(coldTip).toBeDefined(); + expect(replay.mock.calls[0]?.[1]).toBe(coldTip); + await expect(replay.mock.results[0]?.value).resolves.toHaveLength(1); + expect(resumed.nodeAlive.contains('node:base')).toBe(true); + expect(resumed.nodeAlive.contains('node:suffix')).toBe(true); + expect(resumed.prop.get(encodePropKey('node:base', 'status'))?.value).toBe('updated'); + expect(resumed.prop.get(encodePropKey('node:suffix', 'status'))?.value).toBe('added'); + expect(reopenedStore.predecessorLookups).toHaveLength(1); + expect(reopenedStore.predecessorHits[0]?.bundle.equals(coldHandle?.bundle)).toBe(true); + expect(reopenedStore.retainRequests).toHaveLength(1); + await reopenedRuntime.close(); + }); }); class RecordingMaterializationStore extends MaterializationStorePort { @@ -180,6 +228,8 @@ class RecordingMaterializationStore extends MaterializationStorePort { readonly retainedHandles: MaterializationHandle[] = []; readonly exactLookups: MaterializationCoordinate[] = []; readonly exactHits: MaterializationHandle[] = []; + readonly predecessorLookups: MaterializationCoordinate[] = []; + readonly predecessorHits: MaterializationHandle[] = []; exactReleaseCount = 0; readonly #delegate: MaterializationStorePort; @@ -225,6 +275,29 @@ class RecordingMaterializationStore extends MaterializationStorePort { } return null; } + + override async acquireBestCompatiblePredecessor( + coordinate: MaterializationCoordinate, + isCompatible: MaterializationPredecessorPredicate, + ): Promise { + this.predecessorLookups.push(coordinate); + const acquisition = await this.#delegate.acquireBestCompatiblePredecessor( + coordinate, + isCompatible, + ); + if (acquisition !== null) { + this.predecessorHits.push(acquisition.materialization); + } + return acquisition; + } + + override async loadReplayBasis(materialization: MaterializationHandle): Promise { + return await this.#delegate.loadReplayBasis(materialization); + } + + override async close(): Promise { + await this.#delegate.close(); + } } class RecordingMaterializationWorkspace extends MaterializationWorkspacePort { diff --git a/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts b/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts index 9df70892..cbf35e69 100644 --- a/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts +++ b/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts @@ -381,6 +381,7 @@ function rootsFromHandles(handles: readonly BundleHandle[]): MaterializationRoot nodeAlive: MaterializationRoot.retained(nodeAlive), properties: MaterializationRoot.retained(properties), provenanceSupport: MaterializationRoot.retained(provenanceSupport), + replayBasis: MaterializationRoot.unavailable(), roaringIndexes: MaterializationRoot.retained(roaringIndexes), }); } diff --git a/test/unit/domain/materialization/MaterializationIdentity.test.ts b/test/unit/domain/materialization/MaterializationIdentity.test.ts index df491804..6d78372d 100644 --- a/test/unit/domain/materialization/MaterializationIdentity.test.ts +++ b/test/unit/domain/materialization/MaterializationIdentity.test.ts @@ -107,6 +107,7 @@ describe('MaterializationRoots', () => { 'node-alive', 'properties', 'provenance-support', + 'replay-basis', 'roaring-indexes', ]); expect(MATERIALIZATION_ROOT_NAMES).toEqual(roots.entries().map(([name]) => name)); @@ -292,6 +293,7 @@ function rootsOptions(): MaterializationRootsOptions { nodeAlive: retainedRoot('node-alive'), properties: retainedRoot('properties'), provenanceSupport: retainedRoot('provenance-support'), + replayBasis: retainedRoot('replay-basis'), roaringIndexes: retainedRoot('roaring-indexes'), }; } diff --git a/test/unit/domain/materialization/TrieMaterializationReader.test.ts b/test/unit/domain/materialization/TrieMaterializationReader.test.ts index 672280e2..d7671cf3 100644 --- a/test/unit/domain/materialization/TrieMaterializationReader.test.ts +++ b/test/unit/domain/materialization/TrieMaterializationReader.test.ts @@ -39,10 +39,12 @@ describe('TrieMaterializationReader', () => { expect(Object.isFrozen(reader)).toBe(true); await expect(reader.hasNode(root, 'node:present')).resolves.toBe(true); + const readsAfterFirstLookup = store.readCounts(); await expect(reader.hasNode(root, 'node:missing')).resolves.toBe(false); expect(store.writeCounts()).toEqual(writesBeforeRead); const reads = store.readCounts(); + expect(reads).toEqual(readsAfterFirstLookup); expect(reads.leaf + reads.branch).toBeGreaterThan(0); expect(reads.leaf + reads.branch).toBeLessThanOrEqual(4); }); @@ -140,6 +142,7 @@ describe('TrieMaterializationReader', () => { ['codec', { store: new InMemoryTrieStore(), codec: {} }], ['geometry', { store: new InMemoryTrieStore(), codec: cborCodec, geometry: {} }], ['indexStore', { store: new InMemoryTrieStore(), codec: cborCodec, indexStore: {} }], + ['pageCache', { store: new InMemoryTrieStore(), codec: cborCodec, pageCache: {} }], ])('rejects malformed %s with a domain error', (_field, options) => { expect(() => Reflect.construct(TrieMaterializationReader, [options])).toThrowError( expect.objectContaining>({ code: 'E_MATERIALIZATION_RESUME' }) diff --git a/test/unit/domain/services/controllers/MaterializeController.liveNodeRead.test.ts b/test/unit/domain/services/controllers/MaterializeController.liveNodeRead.test.ts index 1be517b9..1e78298c 100644 --- a/test/unit/domain/services/controllers/MaterializeController.liveNodeRead.test.ts +++ b/test/unit/domain/services/controllers/MaterializeController.liveNodeRead.test.ts @@ -363,6 +363,7 @@ function rootsWithStatus(options: { ? MaterializationRoot.empty() : unavailable, provenanceSupport: unavailable, + replayBasis: unavailable, roaringIndexes: unavailable, }); } diff --git a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts index f430536b..0725cdac 100644 --- a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts +++ b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts @@ -515,7 +515,8 @@ describe("MaterializeController — state session integration", () => { const retained = fixtures.materializations.retainedRequests[0]; expect(retained).toBeDefined(); expect(fixtures.materializations.retainedRequests).toHaveLength(1); - expect(fixtures.materializations.exactLookups).toHaveLength(2); + expect(fixtures.materializations.exactLookups).toHaveLength(3); + expect(fixtures.stateCache.getExact).toHaveBeenCalledTimes(1); expect(fixtures.patches.collectForFrontier).not.toHaveBeenCalled(); expect(fixtures.stateCache.put).not.toHaveBeenCalled(); expect(fixtures.openStateSession).toHaveBeenCalledTimes(2); diff --git a/test/unit/domain/services/controllers/MaterializeController.test.ts b/test/unit/domain/services/controllers/MaterializeController.test.ts index 811ddec5..8026b908 100644 --- a/test/unit/domain/services/controllers/MaterializeController.test.ts +++ b/test/unit/domain/services/controllers/MaterializeController.test.ts @@ -162,6 +162,10 @@ function makeDeps({ patchesOverrides = {}, persistenceOverrides = {}, depsOverri hmac: vi.fn().mockResolvedValue(new Uint8Array([1, 2, 3])), }, persistence, + materializations: { + acquireExact: vi.fn().mockResolvedValue(null), + acquireBestCompatiblePredecessor: vi.fn().mockResolvedValue(null), + }, patches, graphCloner: { openReadOnly: vi.fn() }, graphName: 'test', diff --git a/test/unit/domain/services/controllers/RetainedMaterializationResumeStrategy.test.ts b/test/unit/domain/services/controllers/RetainedMaterializationResumeStrategy.test.ts new file mode 100644 index 00000000..c6dd568f --- /dev/null +++ b/test/unit/domain/services/controllers/RetainedMaterializationResumeStrategy.test.ts @@ -0,0 +1,351 @@ +import { describe, expect, it, vi } from 'vitest'; +import AdjacencyMap from '../../../../../src/domain/capabilities/AdjacencyMap.ts'; +import MaterializationCoordinate from '../../../../../src/domain/materialization/MaterializationCoordinate.ts'; +import MaterializationRoot from '../../../../../src/domain/materialization/MaterializationRoot.ts'; +import MaterializationRoots from '../../../../../src/domain/materialization/MaterializationRoots.ts'; +import { + resolveMaterializationRetention, +} from '../../../../../src/domain/services/controllers/MaterializationRetention.ts'; +import RetainedMaterializationResumeStrategy from '../../../../../src/domain/services/controllers/RetainedMaterializationResumeStrategy.ts'; +import type { MaterializeDeps } from '../../../../../src/domain/services/controllers/MaterializeDeps.ts'; +import type { + MaterializeResult, +} from '../../../../../src/domain/services/controllers/MaterializeController.ts'; +import { MaterializePatchSummary } from '../../../../../src/domain/services/controllers/MaterializePatchSummary.ts'; +import type { + MaterializeStrategyRuntime, +} from '../../../../../src/domain/services/controllers/MaterializeStrategyRuntime.ts'; +import type MaterializationWorkspacePort from '../../../../../src/ports/MaterializationWorkspacePort.ts'; +import { createEmptyState } from '../../../../../src/domain/services/JoinReducer.ts'; +import { ProvenanceIndex } from '../../../../../src/domain/services/provenance/ProvenanceIndex.ts'; +import type WarpState from '../../../../../src/domain/services/state/WarpState.ts'; +import type { + MaterializationAcquisition, +} from '../../../../../src/ports/MaterializationStorePort.ts'; +import type { + WarpStateCoordinate, +} from '../../../../../src/ports/WarpStateCachePort.ts'; +import InMemoryMaterializationStore from '../../../../helpers/InMemoryMaterializationStore.ts'; + +const TARGET_TIP = 'b'.repeat(40); +const PREDECESSOR_TIP = 'a'.repeat(40); + +describe('RetainedMaterializationResumeStrategy', () => { + it('reuses an exact retained basis without replaying patches', async () => { + const coordinate = warpCoordinate(TARGET_TIP, null); + const basis = createEmptyState(); + const store = new InMemoryMaterializationStore(); + const handle = await store.retain({ + coordinate: new MaterializationCoordinate(coordinate), + roots: emptyRoots(), + stateHash: 'state-hash', + replayBasis: basis, + }); + const harness = strategyHarness(store); + + const result = await harness.strategy.tryResume( + coordinate, + { receipts: false, wantDiff: true }, + ); + + expect(result).toBe(harness.result); + expect(harness.reducePatchStream).toHaveBeenCalledWith( + expect.anything(), + basis, + { receipts: false, wantDiff: true }, + coordinate, + undefined, + handle, + ); + const source = harness.reducePatchStream.mock.calls[0]?.[0]; + expect(source).toBeDefined(); + expect(await collect(source ?? emptyStream())).toEqual([]); + expect(harness.buildResult).toHaveBeenCalledWith(expect.objectContaining({ + degraded: true, + materialization: handle, + publishSnapshot: false, + })); + expect(store.acquisitions[0]?.released).toBe(true); + }); + + it('falls through when an exact handle has no replay basis', async () => { + const coordinate = warpCoordinate(TARGET_TIP, null); + const store = new InMemoryMaterializationStore(); + await store.retain({ + coordinate: new MaterializationCoordinate(coordinate), + roots: emptyRoots(), + stateHash: 'state-hash', + }); + const harness = strategyHarness(store); + + await expect(harness.strategy.tryResume( + coordinate, + { receipts: false, wantDiff: false }, + )).resolves.toBeNull(); + + expect(harness.reducePatchStream).not.toHaveBeenCalled(); + expect(store.acquisitions[0]?.released).toBe(true); + }); + + it('resumes a causally compatible predecessor and replays only its suffix', async () => { + const target = warpCoordinate(TARGET_TIP, 8); + const predecessor = new MaterializationCoordinate(warpCoordinate(PREDECESSOR_TIP, 4)); + const basis = createEmptyState(); + const store = new InMemoryMaterializationStore(); + const handle = await store.retain({ + coordinate: predecessor, + roots: emptyRoots(), + stateHash: 'state-hash', + replayBasis: basis, + }); + const acquisition = await requireAcquisition(store, predecessor); + const compatible = vi.spyOn(store, 'acquireBestCompatiblePredecessor') + .mockImplementation(async (_coordinate, isCompatible) => + await isCompatible(predecessor) ? acquisition : null); + const harness = strategyHarness(store, { isAncestor: true }); + + const result = await harness.strategy.tryResume( + target, + { receipts: false, wantDiff: false, publishSnapshot: false }, + ); + + expect(result).toBe(harness.result); + expect(compatible).toHaveBeenCalledOnce(); + expect(harness.isAncestor).toHaveBeenCalledWith(PREDECESSOR_TIP, TARGET_TIP); + expect(harness.streamForFrontierSinceCoordinate).toHaveBeenCalledWith( + target.frontier, + target.ceiling, + { frontier: predecessor.frontier(), ceiling: predecessor.ceiling }, + ); + expect(harness.reducePatchStream).toHaveBeenCalledWith( + expect.anything(), + basis, + { receipts: false, wantDiff: false }, + target, + undefined, + handle, + ); + expect(harness.buildResult).toHaveBeenCalledWith(expect.objectContaining({ + degraded: true, + publishSnapshot: false, + })); + expect(store.acquisitions[0]?.released).toBe(true); + }); + + it.each([ + { + name: 'a ceiling beyond the target', + candidate: warpCoordinate(PREDECESSOR_TIP, 9), + target: warpCoordinate(TARGET_TIP, 8), + ancestor: true, + }, + { + name: 'a writer absent from the target', + candidate: warpCoordinate(PREDECESSOR_TIP, null, 'other-writer'), + target: warpCoordinate(TARGET_TIP, null), + ancestor: true, + }, + { + name: 'a non-ancestor writer tip', + candidate: warpCoordinate(PREDECESSOR_TIP, null), + target: warpCoordinate(TARGET_TIP, null), + ancestor: false, + }, + ])('rejects $name', async ({ candidate, target, ancestor }) => { + const store = new InMemoryMaterializationStore(); + const candidateCoordinate = new MaterializationCoordinate(candidate); + await store.retain({ + coordinate: candidateCoordinate, + roots: emptyRoots(), + stateHash: 'state-hash', + replayBasis: createEmptyState(), + }); + const acquisition = await requireAcquisition(store, candidateCoordinate); + vi.spyOn(store, 'acquireBestCompatiblePredecessor') + .mockImplementation(async (_coordinate, isCompatible) => + await isCompatible(candidateCoordinate) ? acquisition : null); + const harness = strategyHarness(store, { isAncestor: ancestor }); + + await expect(harness.strategy.tryResume( + target, + { receipts: false, wantDiff: false }, + )).resolves.toBeNull(); + + expect(harness.reducePatchStream).not.toHaveBeenCalled(); + }); + + it('releases an acquisition when retained resume fails', async () => { + const coordinate = warpCoordinate(TARGET_TIP, null); + const store = new InMemoryMaterializationStore(); + await store.retain({ + coordinate: new MaterializationCoordinate(coordinate), + roots: emptyRoots(), + stateHash: 'state-hash', + replayBasis: createEmptyState(), + }); + const harness = strategyHarness(store); + harness.reducePatchStream.mockRejectedValueOnce(new Error('resume failed')); + + await expect(harness.strategy.tryResume( + coordinate, + { receipts: false, wantDiff: false }, + )).rejects.toThrow('resume failed'); + + expect(store.acquisitions[0]?.released).toBe(true); + }); +}); + +describe('resolveMaterializationRetention', () => { + it('rejects a prepared session without roots and workspace retention', async () => { + await expect(resolveMaterializationRetention({ + deps: {} as MaterializeDeps, // nosemgrep: ts-no-unsafe-type-assertion -- validation fails before dependencies are read + params: { + reduced: { state: createEmptyState(), acceptMaterialization: vi.fn() }, + summary: MaterializePatchSummary.empty(), + degraded: false, + ceiling: null, + frontier: null, + }, + stateHash: 'state-hash', + })).rejects.toMatchObject({ code: 'E_MATERIALIZATION_RESUME' }); + }); + + it('rejects a prepared session whose roots cannot be reopened', async () => { + await expect(resolveMaterializationRetention({ + deps: {} as MaterializeDeps, // nosemgrep: ts-no-unsafe-type-assertion -- validation fails before dependencies are read + params: { + reduced: { + state: createEmptyState(), + roots: unavailableSessionRoots(), + workspace: {} as MaterializationWorkspacePort, // nosemgrep: ts-no-unsafe-type-assertion -- invalid roots fail before workspace use + acceptMaterialization: vi.fn(), + }, + summary: MaterializePatchSummary.empty(), + degraded: false, + ceiling: null, + frontier: null, + }, + stateHash: 'state-hash', + })).rejects.toMatchObject({ code: 'E_MATERIALIZATION_RESUME' }); + }); +}); + +function strategyHarness( + materializations: InMemoryMaterializationStore, + options: { readonly isAncestor?: boolean } = {}, +) { + const state = createEmptyState(); + const result = materializeResult(state); + const streamForFrontierSinceCoordinate = vi.fn(() => emptyStream()); + const isAncestor = vi.fn().mockResolvedValue(options.isAncestor ?? false); + const reducePatchStream = vi.fn(async ( + source: AsyncIterable, + basis: WarpState | undefined, + ) => { + await collect(source); + return { + reduced: { state: basis ?? createEmptyState() }, + summary: MaterializePatchSummary.empty(), + }; + }); + const buildResult = vi.fn().mockResolvedValue(result); + const deps = { + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, + materializations, + patches: { isAncestor, streamForFrontierSinceCoordinate }, + } as unknown as MaterializeDeps; // nosemgrep: ts-no-unsafe-type-assertion -- focused strategy test supplies every dependency the strategy reads + const unavailable = async (): Promise => { + throw new Error('unused retained resume dependency'); + }; + const runtime: MaterializeStrategyRuntime = { + deps, + emptyResult: unavailable, + wrapState: unavailable, + reducePatches: unavailable, + reducePatchStream, + buildResult, + resumeExactMaterialization: unavailable, + buildProvenance: () => new ProvenanceIndex(), + }; + return { + buildResult, + isAncestor, + reducePatchStream, + result, + strategy: new RetainedMaterializationResumeStrategy(runtime), + streamForFrontierSinceCoordinate, + }; +} + +async function requireAcquisition( + store: InMemoryMaterializationStore, + coordinate: MaterializationCoordinate, +): Promise { + const acquisition = await store.acquireExact(coordinate); + if (acquisition === null) { + throw new Error('Expected retained materialization acquisition'); + } + return acquisition; +} + +function materializeResult(state: WarpState): MaterializeResult { + return { + state, + stateHash: 'state-hash', + adjacency: new AdjacencyMap({ incoming: new Map(), outgoing: new Map() }), + patchCount: 0, + maxObservedLamport: 0, + provenanceIndex: new ProvenanceIndex(), + provenanceDegraded: true, + frontier: null, + ceiling: null, + }; +} + +function warpCoordinate( + tip: string, + ceiling: number | null, + writer = 'writer-1', +): WarpStateCoordinate { + return { frontier: new Map([[writer, tip]]), ceiling }; +} + +function emptyRoots(): MaterializationRoots { + const empty = () => MaterializationRoot.empty(); + return new MaterializationRoots({ + adjacency: empty(), + edgeAlive: empty(), + edgeBirths: empty(), + frontier: empty(), + nodeAlive: empty(), + properties: empty(), + provenanceSupport: empty(), + replayBasis: empty(), + roaringIndexes: empty(), + }); +} + +function unavailableSessionRoots(): MaterializationRoots { + const roots = emptyRoots(); + return new MaterializationRoots({ + adjacency: roots.adjacency, + edgeAlive: MaterializationRoot.unavailable(), + edgeBirths: roots.edgeBirths, + frontier: roots.frontier, + nodeAlive: MaterializationRoot.unavailable(), + properties: roots.properties, + provenanceSupport: roots.provenanceSupport, + replayBasis: roots.replayBasis, + roaringIndexes: roots.roaringIndexes, + }); +} + +async function* emptyStream(): AsyncGenerator {} + +async function collect(source: AsyncIterable): Promise { + const values: T[] = []; + for await (const value of source) { + values.push(value); + } + return values; +} diff --git a/test/unit/domain/warp/hydrateCheckpointIndex.regression.test.ts b/test/unit/domain/warp/hydrateCheckpointIndex.regression.test.ts index bf6a390f..fdd4a42e 100644 --- a/test/unit/domain/warp/hydrateCheckpointIndex.regression.test.ts +++ b/test/unit/domain/warp/hydrateCheckpointIndex.regression.test.ts @@ -54,6 +54,10 @@ describe('materialize stale-checkpoint regression', () => { persistence: {}, graphName: 'test', graphCloner: {}, + materializations: { + acquireExact: async () => null, + acquireBestCompatiblePredecessor: async () => null, + }, patches: { loadCheckpoint: async () => ({ schema: 5, diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationPredecessorResolver.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationPredecessorResolver.test.ts new file mode 100644 index 00000000..d2759cce --- /dev/null +++ b/test/unit/infrastructure/adapters/GitCasMaterializationPredecessorResolver.test.ts @@ -0,0 +1,125 @@ +import type { + CacheEntryMetadata, + CacheInspection, + CacheSet, +} from '@git-stunts/git-cas'; +import { describe, expect, it, vi } from 'vitest'; +import MaterializationCoordinate from '../../../../src/domain/materialization/MaterializationCoordinate.ts'; +import type { + MaterializationRootName, +} from '../../../../src/domain/materialization/MaterializationRoots.ts'; +import type { + MaterializationRootStatus, +} from '../../../../src/domain/materialization/MaterializationRoot.ts'; +import GitCasMaterializationPredecessorResolver from '../../../../src/infrastructure/adapters/GitCasMaterializationPredecessorResolver.ts'; +import type { + DecodedMaterializationDescriptor, +} from '../../../../src/infrastructure/adapters/GitCasMaterializationDescriptor.ts'; + +const LANE_PREFIX = 'v4:lane-digest:'; +const TARGET = coordinate('b'.repeat(40)); +const PREDECESSOR = coordinate('a'.repeat(40)); + +describe('GitCasMaterializationPredecessorResolver', () => { + it('excludes other lanes before descriptor reads and budget accounting', async () => { + const unrelated = Array.from( + { length: 1_500 }, + (_, index) => cacheEntry(`v4:other-lane:${String(index)}`, `other:${String(index)}`), + ); + const matching = cacheEntry(`${LANE_PREFIX}predecessor`, 'matching'); + const readDescriptor = vi.fn().mockResolvedValue(descriptor(PREDECESSOR)); + const resolver = resolverFor([...unrelated, matching], readDescriptor); + + await expect(resolver.find(TARGET, () => Promise.resolve(true))).resolves.toEqual( + PREDECESSOR, + ); + + expect(readDescriptor).toHaveBeenCalledOnce(); + }); + + it('returns no predecessor when the lane-local candidate budget is exhausted', async () => { + const entries = Array.from( + { length: 1_025 }, + (_, index) => cacheEntry(`${LANE_PREFIX}${String(index)}`, `matching:${String(index)}`), + ); + const readDescriptor = vi.fn().mockResolvedValue(descriptor(PREDECESSOR)); + const resolver = resolverFor(entries, readDescriptor); + + await expect(resolver.find(TARGET, () => Promise.resolve(true))).resolves.toBeNull(); + + expect(readDescriptor).not.toHaveBeenCalled(); + }); +}); + +function resolverFor( + entries: readonly CacheEntryMetadata[], + readDescriptor: (handle: object) => Promise, +): GitCasMaterializationPredecessorResolver { + const cache = { + inspect: async (options: { readonly limit: number; readonly cursor?: string | null }) => { + const start = Number(options.cursor ?? '0'); + const page = entries.slice(start, start + options.limit); + const next = start + page.length; + return inspection(page, next < entries.length ? String(next) : null); + }, + } satisfies Pick; + return new GitCasMaterializationPredecessorResolver({ + openCache: () => Promise.resolve(cache), + laneName: 'events', + readDescriptor, + cacheKey: () => Promise.resolve(`${LANE_PREFIX}predecessor`), + cacheKeyPrefix: () => Promise.resolve(LANE_PREFIX), + }); +} + +function cacheEntry(key: string, handle: string): CacheEntryMetadata { + return Object.freeze({ + version: 1, + accountingVersion: 1, + key, + keyDigest: `digest:${key}`, + handle, + policy: 'evictable', + expiresAt: null, + logicalBytes: 1, + createdAt: '2026-07-24T00:00:00.000Z', + accessedAt: '2026-07-24T00:00:00.000Z', + }); +} + +function inspection( + entries: readonly CacheEntryMetadata[], + nextCursor: string | null, +): CacheInspection { + return Object.freeze({ + namespace: 'git-warp/materializations', + ref: 'refs/cas/cache/test', + generation: 'generation-1', + state: null, + observed: null, + policy: null, + entries, + nextCursor, + }); +} + +function descriptor( + materializationCoordinate: MaterializationCoordinate, +): DecodedMaterializationDescriptor { + const rootStatuses = new Map([ + ['replay-basis', 'retained'], + ]); + return Object.freeze({ + coordinate: materializationCoordinate, + stateHash: 'state-hash', + laneName: 'events', + rootStatuses, + }); +} + +function coordinate(tip: string): MaterializationCoordinate { + return new MaterializationCoordinate({ + frontier: new Map([['writer-1', tip]]), + ceiling: null, + }); +} diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.lifecycle.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.lifecycle.test.ts index c6bac9a7..fd6a4691 100644 --- a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.lifecycle.test.ts +++ b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.lifecycle.test.ts @@ -17,7 +17,8 @@ import InMemoryGitCasFacade from '../../../helpers/InMemoryGitCasFacade.ts'; import InMemoryGraphAdapter from '../../../helpers/InMemoryGraphAdapter.ts'; const CACHE_NAMESPACE = 'git-warp/materializations'; -const ROOT_COUNT = 8; +const CURRENT_CACHE_KEY_PATTERN = /^v4:[0-9a-f]{64}:[0-9a-f]{64}$/u; +const ROOT_COUNT = 9; describe('GitCasMaterializationStoreAdapter lifecycle', () => { it('rejects new materialization operations as soon as closure starts', async () => { @@ -63,6 +64,7 @@ describe('GitCasMaterializationStoreAdapter lifecycle', () => { const staging = await harness.cas.workspaces.open({ namespace: 'pending-close' }); const release = vi.fn(async () => await staging.release()); const controlled: GitCasStagingWorkspace = { + assets: staging.assets, pages: staging.pages, bundles: staging.bundles, checkpoint: async (options) => await staging.checkpoint(options), @@ -71,6 +73,7 @@ describe('GitCasMaterializationStoreAdapter lifecycle', () => { }; const deferred = Promise.withResolvers(); const adapter = adapterFor({ + assets: harness.cas.assets, bundles: harness.cas.bundles, caches: harness.cas.caches, pages: harness.cas.pages, @@ -96,6 +99,7 @@ describe('GitCasMaterializationStoreAdapter lifecycle', () => { const releaseFailure = new Error('release failed'); const staging = await harness.cas.workspaces.open({ namespace: 'failed-retain' }); const controlled: GitCasStagingWorkspace = { + assets: staging.assets, pages: { put: async () => { throw promotionFailure; @@ -110,6 +114,7 @@ describe('GitCasMaterializationStoreAdapter lifecycle', () => { }, }; const adapter = adapterFor({ + assets: harness.cas.assets, bundles: harness.cas.bundles, caches: harness.cas.caches, pages: harness.cas.pages, @@ -126,34 +131,38 @@ describe('GitCasMaterializationStoreAdapter lifecycle', () => { }); }); - it('keeps the matching v2 cache anchor until the v3 profile is retained', async () => { + it('keeps matching legacy cache anchors until the v4 profile is retained', async () => { const harness = await createHarness(); const coordinate = exactCoordinate(); const roots = await createRoots(harness.cas); const retained = await harness.adapter.retain({ coordinate, roots, stateHash: 'state-hash' }); const cache = await harness.cas.caches.open({ namespace: CACHE_NAMESPACE }); - const v3Key = requireSingleCacheKey(harness.cas); + const v4Key = requireSingleCacheKey(harness.cas); const v2Key = await cacheKeyForSchema(coordinate, 2); + const v3Key = await cacheKeyForSchema(coordinate, 3); await cache.put(v2Key, retained.bundle.toString()); - await cache.remove(v3Key); + await cache.put(v3Key, retained.bundle.toString()); + await cache.remove(v4Key); await expect(harness.adapter.acquireExact(coordinate)).resolves.toBeNull(); - expect(harness.cas.readCacheKeys(CACHE_NAMESPACE)).toEqual([v2Key]); + expect(harness.cas.readCacheKeys(CACHE_NAMESPACE)).toEqual([v2Key, v3Key]); await harness.adapter.retain({ coordinate, roots, stateHash: 'replacement-state-hash' }); - expect(requireSingleCacheKey(harness.cas)).toMatch(/^v3:[0-9a-f]{64}$/u); + expect(requireSingleCacheKey(harness.cas)).toMatch(CURRENT_CACHE_KEY_PATTERN); }); - it('removes the matching v2 cache anchor after direct v3 retention', async () => { + it('removes matching legacy cache anchors after direct v4 retention', async () => { const harness = await createHarness(); const coordinate = exactCoordinate(); const roots = await createRoots(harness.cas); const retained = await harness.adapter.retain({ coordinate, roots, stateHash: 'state-hash' }); const cache = await harness.cas.caches.open({ namespace: CACHE_NAMESPACE }); - const v3Key = requireSingleCacheKey(harness.cas); + const v4Key = requireSingleCacheKey(harness.cas); const v2Key = await cacheKeyForSchema(coordinate, 2); + const v3Key = await cacheKeyForSchema(coordinate, 3); await cache.put(v2Key, retained.bundle.toString()); - await cache.remove(v3Key); + await cache.put(v3Key, retained.bundle.toString()); + await cache.remove(v4Key); const replacement = await harness.adapter.retain({ coordinate, @@ -163,23 +172,25 @@ describe('GitCasMaterializationStoreAdapter lifecycle', () => { const acquisition = await harness.adapter.acquireExact(coordinate); expect(harness.cas.readCacheKeys(CACHE_NAMESPACE)).toEqual([ - expect.stringMatching(/^v3:[0-9a-f]{64}$/u), + expect.stringMatching(CURRENT_CACHE_KEY_PATTERN), ]); expect(replacement.retention.root.generation) .toBe(acquisition?.materialization.retention.root.generation); await acquisition?.release(); }); - it('keeps the v2 anchor when the exact v3 target cannot be acquired', async () => { + it('keeps legacy anchors when the exact v4 target cannot be acquired', async () => { const harness = await createHarness(); const coordinate = exactCoordinate(); const roots = await createRoots(harness.cas); const original = await harness.adapter.retain({ coordinate, roots, stateHash: 'state-hash' }); const cache = await harness.cas.caches.open({ namespace: CACHE_NAMESPACE }); - const v3Key = requireSingleCacheKey(harness.cas); + const v4Key = requireSingleCacheKey(harness.cas); const v2Key = await cacheKeyForSchema(coordinate, 2); + const v3Key = await cacheKeyForSchema(coordinate, 3); await cache.put(v2Key, original.bundle.toString()); - await cache.remove(v3Key); + await cache.put(v3Key, original.bundle.toString()); + await cache.remove(v4Key); await expect(adapterFor(withoutCacheAcquisition(harness.cas)).retain({ coordinate, @@ -191,6 +202,7 @@ describe('GitCasMaterializationStoreAdapter lifecycle', () => { }); expect(harness.cas.readCacheKeys(CACHE_NAMESPACE)).toContain(v2Key); + expect(harness.cas.readCacheKeys(CACHE_NAMESPACE)).toContain(v3Key); }); }); @@ -216,6 +228,7 @@ function adapterFor(cas: GitCasMaterializationFacade): GitCasMaterializationStor function withoutCacheAcquisition(cas: InMemoryGitCasFacade): GitCasMaterializationFacade { return { + assets: cas.assets, bundles: cas.bundles, pages: cas.pages, caches: { @@ -224,6 +237,7 @@ function withoutCacheAcquisition(cas: InMemoryGitCasFacade): GitCasMaterializati return { ref: cache.ref, acquire: async () => null, + inspect: async (inspectOptions) => await cache.inspect(inspectOptions), put: async (key, handle, entryOptions) => await cache.put(key, handle, entryOptions), remove: async (key) => await cache.remove(key), }; @@ -259,7 +273,8 @@ function rootsFromHandles(handles: readonly BundleHandle[]): MaterializationRoot nodeAlive: root(4), properties: root(5), provenanceSupport: root(6), - roaringIndexes: root(7), + replayBasis: root(7), + roaringIndexes: root(8), }); } diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.resume.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.resume.test.ts new file mode 100644 index 00000000..5f3652ec --- /dev/null +++ b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.resume.test.ts @@ -0,0 +1,201 @@ +import { describe, expect, it } from 'vitest'; +import MaterializationCoordinate from '../../../../src/domain/materialization/MaterializationCoordinate.ts'; +import MaterializationRoot from '../../../../src/domain/materialization/MaterializationRoot.ts'; +import MaterializationRoots from '../../../../src/domain/materialization/MaterializationRoots.ts'; +import { createEmptyState } from '../../../../src/domain/services/JoinReducer.ts'; +import { computeStateHash } from '../../../../src/domain/services/state/StateSerializer.ts'; +import BundleHandle from '../../../../src/domain/storage/BundleHandle.ts'; +import GitCasMaterializationStoreAdapter from '../../../../src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts'; +import NodeCryptoAdapter from '../../../../src/infrastructure/adapters/NodeCryptoAdapter.ts'; +import defaultCodec from '../../../../src/infrastructure/codecs/CborCodec.ts'; +import InMemoryBlobStorageAdapter from '../../../helpers/InMemoryBlobStorageAdapter.ts'; +import InMemoryGitCasFacade from '../../../helpers/InMemoryGitCasFacade.ts'; +import InMemoryGraphAdapter from '../../../helpers/InMemoryGraphAdapter.ts'; + +describe('GitCasMaterializationStoreAdapter retained resume', () => { + it('retains and restores a replay basis through the materialization bundle', async () => { + const harness = await createHarness(); + const state = createEmptyState(); + const retained = await harness.adapter.retain({ + coordinate: exactCoordinate(), + roots: await createRoots(harness.cas), + stateHash: await hashState(state), + replayBasis: state, + }); + + expect(retained.roots.replayBasis.status).toBe('retained'); + await expect(harness.adapter.loadReplayBasis(retained)).resolves.toEqual(state); + }); + + it('fails closed when a retained replay basis is corrupt', async () => { + const harness = await createHarness(); + const retained = await harness.adapter.retain({ + coordinate: exactCoordinate(), + roots: await createRoots(harness.cas), + stateHash: await hashState(createEmptyState()), + replayBasis: createEmptyState(), + }); + const replayRoot = retained.roots.replayBasis.handle; + if (replayRoot === null) { + throw new Error('Expected a retained replay basis root'); + } + const asset = requireMember( + harness.cas.readBundleMembers(replayRoot.toString()), + 'state.cbor', + ); + harness.cas.replaceStoredAsset(asset, new Uint8Array([0xff])); + + await expect(harness.adapter.loadReplayBasis(retained)).rejects.toBeDefined(); + }); + + it('acquires a compatible retained predecessor and excludes the target', async () => { + const harness = await createHarness(); + const predecessor = exactCoordinate(); + const target = targetCoordinate(); + const retained = await harness.adapter.retain({ + coordinate: predecessor, + roots: await createRoots(harness.cas), + stateHash: 'predecessor-state-hash', + replayBasis: createEmptyState(), + }); + await harness.adapter.retain({ + coordinate: target, + roots: await createRoots(harness.cas), + stateHash: 'target-state-hash', + replayBasis: createEmptyState(), + }); + const observed: MaterializationCoordinate[] = []; + + const acquisition = await harness.adapter.acquireBestCompatiblePredecessor( + target, + (candidate) => { + observed.push(candidate); + return Promise.resolve(candidate.equals(predecessor)); + }, + ); + + expect(observed).toHaveLength(1); + expect(acquisition?.materialization.bundle.equals(retained.bundle)).toBe(true); + await acquisition?.release(); + await harness.adapter.close(); + }); + + it('excludes compatible predecessors without a retained replay basis', async () => { + const harness = await createHarness(); + await harness.adapter.retain({ + coordinate: exactCoordinate(), + roots: rootsWithoutReplayBasis(await createRoots(harness.cas)), + stateHash: 'predecessor-state-hash', + }); + + const acquisition = await harness.adapter.acquireBestCompatiblePredecessor( + targetCoordinate(), + () => Promise.resolve(true), + ); + + expect(acquisition).toBeNull(); + }); +}); + +async function createHarness(): Promise> { + const cas = new InMemoryGitCasFacade({ + history: new InMemoryGraphAdapter(), + storage: new InMemoryBlobStorageAdapter(), + }); + return Object.freeze({ + cas, + adapter: new GitCasMaterializationStoreAdapter({ + cas, + codec: defaultCodec, + crypto: new NodeCryptoAdapter(), + laneName: 'events', + }), + }); +} + +async function createRoots(cas: InMemoryGitCasFacade): Promise { + const handles: BundleHandle[] = []; + for (let index = 0; index < 9; index += 1) { + const staged = await cas.bundles.putOrdered({ members: [] }); + handles.push(new BundleHandle(staged.handle.toString())); + } + return rootsFromHandles(handles); +} + +function rootsFromHandles(handles: readonly BundleHandle[]): MaterializationRoots { + const roots = handles.map((handle) => MaterializationRoot.retained(handle)); + return new MaterializationRoots({ + adjacency: requireRoot(roots, 0), + edgeAlive: requireRoot(roots, 1), + edgeBirths: requireRoot(roots, 2), + frontier: requireRoot(roots, 3), + nodeAlive: requireRoot(roots, 4), + properties: requireRoot(roots, 5), + provenanceSupport: requireRoot(roots, 6), + replayBasis: requireRoot(roots, 7), + roaringIndexes: requireRoot(roots, 8), + }); +} + +function requireRoot(roots: readonly MaterializationRoot[], index: number): MaterializationRoot { + const root = roots[index]; + if (root === undefined) { + throw new Error('Root fixture did not create every materialization root'); + } + return root; +} + +function rootsWithoutReplayBasis(roots: MaterializationRoots): MaterializationRoots { + return new MaterializationRoots({ + adjacency: roots.adjacency, + edgeAlive: roots.edgeAlive, + edgeBirths: roots.edgeBirths, + frontier: roots.frontier, + nodeAlive: roots.nodeAlive, + properties: roots.properties, + provenanceSupport: roots.provenanceSupport, + replayBasis: MaterializationRoot.unavailable(), + roaringIndexes: roots.roaringIndexes, + }); +} + +async function hashState(state: ReturnType): Promise { + return await computeStateHash(state, { + codec: defaultCodec, + crypto: new NodeCryptoAdapter(), + }); +} + +function exactCoordinate(): MaterializationCoordinate { + return new MaterializationCoordinate({ + frontier: new Map([ + ['writer-a', 'patch-a'], + ['writer-b', 'patch-b'], + ]), + ceiling: 12, + }); +} + +function targetCoordinate(): MaterializationCoordinate { + return new MaterializationCoordinate({ + frontier: new Map([ + ['writer-a', 'patch-next'], + ['writer-b', 'patch-b'], + ]), + ceiling: 13, + }); +} + +function requireMember( + members: readonly (readonly [string, string])[], + path: string, +): string { + const member = members.find(([candidate]) => candidate === path); + if (member === undefined) { + throw new Error(`Missing bundle member: ${path}`); + } + return member[1]; +} diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts index 5b9763cb..dbb3bb25 100644 --- a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts +++ b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts @@ -23,6 +23,7 @@ const ROOT_PATHS = Object.freeze([ 'roots/node-alive', 'roots/properties', 'roots/provenance-support', + 'roots/replay-basis', 'roots/roaring-indexes', ]); @@ -70,7 +71,7 @@ describe('GitCasMaterializationStoreAdapter', () => { expect(members.map(([path]) => path)).toEqual(['meta/descriptor', ...ROOT_PATHS]); const cacheKeys = harness.cas.readCacheKeys(CACHE_NAMESPACE); expect(cacheKeys).toHaveLength(1); - expect(cacheKeys[0]).toMatch(/^v3:[0-9a-f]{64}$/u); + expect(cacheKeys[0]).toMatch(/^v4:[0-9a-f]{64}:[0-9a-f]{64}$/u); expect(cacheKeys[0]?.length).toBeLessThan(1024); expect(harness.cas.readActiveCacheAcquisitionCount()).toBe(1); await acquisition?.release(); @@ -584,6 +585,7 @@ function withCacheResult( rewrite: (stored: CacheStoreResult) => CacheStoreResult, ): GitCasMaterializationFacade { return { + assets: cas.assets, bundles: cas.bundles, pages: cas.pages, caches: { @@ -592,6 +594,7 @@ function withCacheResult( return { ref: cache.ref, acquire: async (key) => await cache.acquire(key), + inspect: async (inspectOptions) => await cache.inspect(inspectOptions), put: async (key, handle, entryOptions) => rewrite( await cache.put(key, handle, entryOptions), ), @@ -635,12 +638,13 @@ async function createRoots(cas: InMemoryGitCasFacade): Promise = {}): object { return { - schemaVersion: 3, + schemaVersion: 4, laneName: 'events', stateHash: 'state-hash', roots: rootStatusFixture(),