diff --git a/CHANGELOG.md b/CHANGELOG.md index e3db8409..5368172f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -66,6 +66,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Exact state-cache hits now retain and reopen coordinate-keyed materialization + descriptors through the `git-cas` cache API. Repeated materialization, + including through a fresh runtime adapter, resumes the retained node/edge + trie roots without replaying writer patches already covered by the snapshot. +- Failed state-session reductions no longer flush partial trie roots into + unretained CAS objects; roots are flushed only after successful projection. - Git-backed state-cache payload trees are now anchored through a graph-scoped `git-cas` RootSet before their index record is published, then reconciled after publication so live cache entries remain reachable across Git garbage diff --git a/docs/topics/cas-first-memoized-materialization.md b/docs/topics/cas-first-memoized-materialization.md index 9dfb57a8..0472c273 100644 --- a/docs/topics/cas-first-memoized-materialization.md +++ b/docs/topics/cas-first-memoized-materialization.md @@ -4,11 +4,12 @@ Use this page when you need to understand how `git-warp` skips redundant materialization replay by memoizing WARP-owned state snapshots in `@git-stunts/git-cas`. -`git-cas` provides byte storage and generic Git-reachability primitives. It does -not know about WARP frontiers, optics, checkpoints, graph state, or -materialization rules. `git-warp` owns those semantics through -`WarpStateCachePort`; the Git-backed adapter stores snapshot payloads in -`git-cas` and declares the live payload trees through a `RootSet`. +`git-cas` provides byte storage, retained cache entries, and generic +Git-reachability primitives. It does not know about WARP frontiers, optics, +checkpoints, graph state, or materialization rules. `git-warp` owns those +semantics through `WarpStateCachePort` and `MaterializationStorePort`; the +Git-backed adapters store snapshot payloads and coordinate-keyed retained roots +through `git-cas`. ## The Live Materialization Lifecycle @@ -19,7 +20,7 @@ coordinate-first lifecycle: [current frontier] | v -[state-cache exact hit?] ---- yes ---> [return cached state] +[state-cache exact hit?] ---- yes ---> [reopen retained roots; zero patch replay] | no v @@ -44,8 +45,13 @@ This coordinate belongs to `git-warp`; it is not a `git-cas` concept. ### 2. Check the WARP state cache The runtime asks `WarpStateCachePort` for an exact snapshot at that coordinate. -On a hit, it returns the cached state without replaying writer patch streams and -without republishing the same snapshot. +On a hit, it asks `MaterializationStorePort` for the matching retained-root +descriptor. A descriptor hit reopens the node/edge trie roots and projects the +result without replaying writer patch streams or republishing the same snapshot. +The descriptor records every named materialization root as `retained`, `empty`, +or `unavailable`; only retained roots become bundle members. On the first exact +snapshot hit without a descriptor, the runtime seeds the trie roots from the +snapshot and retains the resulting descriptor for later runtime instances. The current payload records state but not the provenance index. A runtime may retain its resident provenance index when the cached state has the same hash and @@ -67,10 +73,10 @@ equivalent read can hit the cache. ## Memory Boundaries -State-cache hits avoid redundant CRDT replay and can remove repeated startup -costs for graph-sized materializations. They do not make legacy full -materialization an `O(1)` memory API: a caller that asks for a full -`SnapshotWarpState` still receives a full in-memory state object. +State-cache hits with retained roots avoid redundant CRDT patch replay across +runtime instances. They do not make legacy full materialization an `O(1)` time +or memory API: the current result contract still loads a full snapshot and scans +the retained node/edge tries to produce a full `WarpState` and adjacency map. The bounded-memory read path is optic/worldline/query work over a sharded or streamed basis. The state cache is the replay-skipping compatibility bridge for @@ -78,9 +84,13 @@ legacy materialization and checkpoint flows. ## `git-cas` Encapsulation -All state-cache payload storage routes through the formal `@git-stunts/git-cas` -library API. Raw Git plumbing remains an adapter concern for WARP refs and Git -object access; WARP state-cache payloads should not hand-roll a parallel CAS. +Materialization-root retention routes through the formal +`@git-stunts/git-cas` `CacheSet` API. The legacy state-cache adapter also routes +payload bytes through `git-cas`, but still owns its snapshot index and RootSet +reconciliation. Removing that compatibility cache lifecycle is required before +the one-cache boundary is complete. Raw Git plumbing remains an adapter concern +for WARP refs and Git object access; WARP code must not hand-roll a parallel +CAS. Routing state snapshots through `git-cas` allows content-addressed storage and chunk-level reuse where the underlying CAS representation can identify unchanged @@ -134,7 +144,14 @@ lost payload bytes, or run Git garbage collection. ## Current Limitations - Exact state-cache hits bypass replay, but full materialization still hydrates - a full `WarpState`. + a full `WarpState`, scans retained node/edge tries, and builds full adjacency. +- Retained materialization descriptors currently carry node/edge trie roots; + property, frontier, edge-birth, adjacency, provenance-support, and roaring + roots are explicitly marked unavailable until their paged representations + land. +- `WarpStateCachePort` remains a legacy full-snapshot compatibility cache with + a WARP-owned index. Ordinary bounded observers cannot rely on it as their + final storage contract. - The Git-backed state-cache adapter stores full-state snapshots today. A future sharded basis format should make optic reads avoid full-state hydration. - Cache coordinates must stay schema/version aware. A snapshot is reusable only diff --git a/src/domain/RuntimeHost.ts b/src/domain/RuntimeHost.ts index bd1057d9..9e9f7e32 100644 --- a/src/domain/RuntimeHost.ts +++ b/src/domain/RuntimeHost.ts @@ -288,6 +288,7 @@ export default class RuntimeHost { checkpointStore, indexStore, intentStore, + materializations, viewService, stateHashService, auditService, @@ -400,6 +401,7 @@ export default class RuntimeHost { crypto: this._crypto, persistence: this._persistence, checkpointStore, + materializations, getStateCache: () => this._stateCache ?? null, ...(openStateSession === undefined ? {} : { openStateSession }), patches: new RuntimePatchCollector(this), diff --git a/src/domain/materialization/MaterializationRoot.ts b/src/domain/materialization/MaterializationRoot.ts new file mode 100644 index 00000000..33e17d1e --- /dev/null +++ b/src/domain/materialization/MaterializationRoot.ts @@ -0,0 +1,35 @@ +import WarpError from '../errors/WarpError.ts'; +import BundleHandle from '../storage/BundleHandle.ts'; + +export type MaterializationRootStatus = 'retained' | 'empty' | 'unavailable'; + +/** Availability and optional retained handle for one materialization root. */ +export default class MaterializationRoot { + readonly status: MaterializationRootStatus; + readonly handle: BundleHandle | null; + + private constructor(status: MaterializationRootStatus, handle: BundleHandle | null) { + this.status = status; + this.handle = handle; + Object.freeze(this); + } + + static retained(handle: BundleHandle): MaterializationRoot { + if (!(handle instanceof BundleHandle)) { + throw rootError('retained root must carry a BundleHandle'); + } + return new MaterializationRoot('retained', handle); + } + + static empty(): MaterializationRoot { + return new MaterializationRoot('empty', null); + } + + static unavailable(): MaterializationRoot { + return new MaterializationRoot('unavailable', null); + } +} + +function rootError(message: string): WarpError { + return new WarpError(`Materialization root ${message}`, 'E_MATERIALIZATION_ROOT'); +} diff --git a/src/domain/materialization/MaterializationRoots.ts b/src/domain/materialization/MaterializationRoots.ts index 48bb9922..33986025 100644 --- a/src/domain/materialization/MaterializationRoots.ts +++ b/src/domain/materialization/MaterializationRoots.ts @@ -1,5 +1,5 @@ import WarpError from '../errors/WarpError.ts'; -import BundleHandle from '../storage/BundleHandle.ts'; +import MaterializationRoot from './MaterializationRoot.ts'; export const MATERIALIZATION_ROOT_NAMES = defineRootNames( 'adjacency', @@ -15,63 +15,63 @@ export const MATERIALIZATION_ROOT_NAMES = defineRootNames( export type MaterializationRootName = (typeof MATERIALIZATION_ROOT_NAMES)[number]; export type MaterializationRootsOptions = Readonly<{ - adjacency: BundleHandle; - edgeAlive: BundleHandle; - edgeBirths: BundleHandle; - frontier: BundleHandle; - nodeAlive: BundleHandle; - properties: BundleHandle; - provenanceSupport: BundleHandle; - roaringIndexes: BundleHandle; + adjacency: MaterializationRoot; + edgeAlive: MaterializationRoot; + edgeBirths: MaterializationRoot; + frontier: MaterializationRoot; + nodeAlive: MaterializationRoot; + properties: MaterializationRoot; + provenanceSupport: MaterializationRoot; + roaringIndexes: MaterializationRoot; }>; /** Independently addressable retained roots for one materialized causal chart. */ export default class MaterializationRoots { - private readonly handles: Readonly>; - readonly adjacency: BundleHandle; - readonly edgeAlive: BundleHandle; - readonly edgeBirths: BundleHandle; - readonly frontier: BundleHandle; - readonly nodeAlive: BundleHandle; - readonly properties: BundleHandle; - readonly provenanceSupport: BundleHandle; - readonly roaringIndexes: BundleHandle; + private readonly roots: Readonly>; + readonly adjacency: MaterializationRoot; + readonly edgeAlive: MaterializationRoot; + readonly edgeBirths: MaterializationRoot; + readonly frontier: MaterializationRoot; + readonly nodeAlive: MaterializationRoot; + readonly properties: MaterializationRoot; + readonly provenanceSupport: MaterializationRoot; + readonly roaringIndexes: MaterializationRoot; constructor(options: MaterializationRootsOptions) { requireOptions(options); - this.handles = Object.freeze({ - adjacency: requireBundle(options.adjacency, 'adjacency'), - 'edge-alive': requireBundle(options.edgeAlive, 'edgeAlive'), - 'edge-births': requireBundle(options.edgeBirths, 'edgeBirths'), - frontier: requireBundle(options.frontier, 'frontier'), - 'node-alive': requireBundle(options.nodeAlive, 'nodeAlive'), - properties: requireBundle(options.properties, 'properties'), - 'provenance-support': requireBundle(options.provenanceSupport, 'provenanceSupport'), - 'roaring-indexes': requireBundle(options.roaringIndexes, 'roaringIndexes'), - } satisfies Record); - this.adjacency = this.handles.adjacency; - this.edgeAlive = this.handles['edge-alive']; - this.edgeBirths = this.handles['edge-births']; - this.frontier = this.handles.frontier; - this.nodeAlive = this.handles['node-alive']; - this.properties = this.handles.properties; - this.provenanceSupport = this.handles['provenance-support']; - this.roaringIndexes = this.handles['roaring-indexes']; + this.roots = Object.freeze({ + adjacency: requireRoot(options.adjacency, 'adjacency'), + 'edge-alive': requireRoot(options.edgeAlive, 'edgeAlive'), + 'edge-births': requireRoot(options.edgeBirths, 'edgeBirths'), + frontier: requireRoot(options.frontier, 'frontier'), + 'node-alive': requireRoot(options.nodeAlive, 'nodeAlive'), + properties: requireRoot(options.properties, 'properties'), + 'provenance-support': requireRoot(options.provenanceSupport, 'provenanceSupport'), + 'roaring-indexes': requireRoot(options.roaringIndexes, 'roaringIndexes'), + } satisfies Record); + this.adjacency = this.roots.adjacency; + this.edgeAlive = this.roots['edge-alive']; + this.edgeBirths = this.roots['edge-births']; + this.frontier = this.roots.frontier; + this.nodeAlive = this.roots['node-alive']; + this.properties = this.roots.properties; + this.provenanceSupport = this.roots['provenance-support']; + this.roaringIndexes = this.roots['roaring-indexes']; Object.freeze(this); } - entries(): readonly (readonly [MaterializationRootName, BundleHandle])[] { + entries(): readonly (readonly [MaterializationRootName, MaterializationRoot])[] { return Object.freeze( - MATERIALIZATION_ROOT_NAMES.map((name) => rootEntry(name, this.handles[name])), + MATERIALIZATION_ROOT_NAMES.map((name) => rootEntry(name, this.roots[name])), ); } } function rootEntry( name: MaterializationRootName, - handle: BundleHandle, -): readonly [MaterializationRootName, BundleHandle] { - return Object.freeze([name, handle]); + root: MaterializationRoot, +): readonly [MaterializationRootName, MaterializationRoot] { + return Object.freeze([name, root]); } function defineRootNames(...names: Names): Names { @@ -79,11 +79,11 @@ function defineRootNames(...names: Names) return names; } -function requireBundle(handle: BundleHandle, field: string): BundleHandle { - if (!(handle instanceof BundleHandle)) { - throw rootsError(`${field} must be a BundleHandle`); +function requireRoot(root: MaterializationRoot, field: string): MaterializationRoot { + if (!(root instanceof MaterializationRoot)) { + throw rootsError(`${field} must be a MaterializationRoot`); } - return handle; + return root; } function requireOptions(options: object): void { diff --git a/src/domain/services/controllers/MaterializeController.ts b/src/domain/services/controllers/MaterializeController.ts index ce25972c..5b680f7d 100644 --- a/src/domain/services/controllers/MaterializeController.ts +++ b/src/domain/services/controllers/MaterializeController.ts @@ -21,6 +21,7 @@ import { type MaterializeAdjacency, } from './MaterializeHelpers.ts'; import { + materializationSessionOpen, reduceSessionBackedState, type MaterializeSessionOpener, } from './MaterializeSessionBridge.ts'; @@ -42,6 +43,7 @@ import type CodecPort from '../../../ports/CodecPort.ts'; import type CryptoPort from '../../../ports/CryptoPort.ts'; import type CheckpointStorePort from '../../../ports/CheckpointStorePort.ts'; import type WarpStateCachePort from '../../../ports/WarpStateCachePort.ts'; +import type MaterializationStorePort from '../../../ports/MaterializationStorePort.ts'; import type { WarpStateSnapshotProvenancePosture } from '../../../ports/WarpStateCachePort.ts'; import type PatchCollector from '../../capabilities/PatchCollector.ts'; import type { PatchWithSha } from '../../capabilities/PatchCollector.ts'; @@ -51,6 +53,11 @@ import type WarpState from '../state/WarpState.ts'; import type { TickReceipt } from '../../types/TickReceipt.ts'; import type { PatchDiff } from '../../types/PatchDiff.ts'; import AdjacencyMap from '../../capabilities/AdjacencyMap.ts'; +import MaterializationCoordinate from '../../materialization/MaterializationCoordinate.ts'; +import type MaterializationHandle from '../../materialization/MaterializationHandle.ts'; +import type MaterializationRoots from '../../materialization/MaterializationRoots.ts'; +import WarpError from '../../errors/WarpError.ts'; +import type { UsableSnapshotRecord } from './MaterializeSnapshotCacheResult.ts'; import type { MaterializeResultBuildInput, MaterializeStrategyRuntime, @@ -69,6 +76,7 @@ export type MaterializeDeps = { crypto: CryptoPort; persistence: MaterializePersistence; checkpointStore: CheckpointStorePort; + materializations: MaterializationStorePort; getStateCache?: () => WarpStateCachePort | null; openStateSession?: MaterializeSessionOpener; patches: PatchCollector; @@ -91,6 +99,7 @@ export type MaterializeResult = { provenanceDegraded: boolean; frontier: Map | null; ceiling: number | null; + materialization?: MaterializationHandle; }; // ── Reduce helpers ────────────────────────────────────────────────── @@ -104,6 +113,7 @@ function toReducerInput(patches: PatchWithSha[]): ReducerInput { export type MaterializeReduceOutput = { state: WarpState; adjacency?: MaterializeAdjacency; + roots?: MaterializationRoots; receipts?: TickReceipt[]; diff?: PatchDiff; }; @@ -255,7 +265,8 @@ export default class MaterializeController { private async _buildResult(params: MaterializeResultBuildInput): Promise { const stateHash = await computeHash(this._deps, params.reduced.state); const adjacency = params.reduced.adjacency ?? buildAdjacency(params.reduced.state); - if (params.reduced.receipts === undefined) { + const materialization = await this._resolveMaterialization(params, stateHash); + if (params.reduced.receipts === undefined && params.publishSnapshot !== false) { await this._publishSnapshot({ state: params.reduced.state, stateHash, @@ -278,9 +289,70 @@ export default class MaterializeController { provenanceDegraded: params.degraded, frontier: params.frontier, ceiling: params.ceiling, + ...(materialization === undefined ? {} : { materialization }), }; } + 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'); + } + return params.materialization; + } + if (params.reduced.roots === undefined || params.frontier === null) { + return undefined; + } + return await this._deps.materializations.retain({ + coordinate: new MaterializationCoordinate({ + frontier: params.frontier, + ceiling: params.ceiling, + }), + roots: params.reduced.roots, + stateHash, + }); + } + + private async _resumeExactMaterialization( + snapshot: UsableSnapshotRecord, + options: { wantDiff: boolean }, + ): Promise { + const { openStateSession } = this._deps; + if (openStateSession === undefined) { + return null; + } + const coordinate = new MaterializationCoordinate(snapshot.coordinate); + const retained = await this._deps.materializations.findExact(coordinate); + if (retained !== null && retained.stateHash !== snapshot.stateHash) { + throw materializationResumeError('retained handle and snapshot state hashes differ'); + } + const retainedRoots = retained === null + ? null + : materializationSessionOpen(retained.roots); + const reduced = await reduceSessionBackedState({ + openStateSession, + patches: [], + baseState: snapshot.state, + ...(retainedRoots === null ? {} : { roots: retainedRoots }), + receipts: false, + wantDiff: options.wantDiff, + }); + return await this._buildResult({ + reduced, + summary: new MaterializePatchSummaryAccumulator().toSummary(), + degraded: true, + ceiling: snapshot.coordinate.ceiling, + frontier: snapshot.coordinate.frontier, + publishSnapshot: false, + ...(retained === null || retainedRoots === null + ? {} + : { materialization: retained }), + }); + } + private async _reducePatches( patches: PatchWithSha[], base: WarpState | undefined, @@ -345,6 +417,8 @@ export default class MaterializeController { reducePatchStream: async (stream, base, opts, provenanceBase) => await this._reducePatchStream(stream, base, opts, provenanceBase), buildResult: async (params) => await this._buildResult(params), + resumeExactMaterialization: async (snapshot, options) => + await this._resumeExactMaterialization(snapshot, options), buildProvenance: (patches, base) => buildProvenance(patches, base), }; } @@ -374,3 +448,10 @@ export default class MaterializeController { }); } } + +function materializationResumeError(message: string): WarpError { + return new WarpError( + `Materialization resume ${message}`, + 'E_MATERIALIZATION_RESUME', + ); +} diff --git a/src/domain/services/controllers/MaterializeCoordinateStrategy.ts b/src/domain/services/controllers/MaterializeCoordinateStrategy.ts index e4c74f1f..b6b90c23 100644 --- a/src/domain/services/controllers/MaterializeCoordinateStrategy.ts +++ b/src/domain/services/controllers/MaterializeCoordinateStrategy.ts @@ -113,7 +113,9 @@ export default class MaterializeCoordinateStrategy { ): Promise { const exact = await stateCache.getExact(opts.coordinate); if (canUseSnapshot(exact, { receipts: opts.receipts })) { - return snapshotToMaterializeResult(exact); + return await this.runtime.resumeExactMaterialization(exact, { + wantDiff: false, + }) ?? snapshotToMaterializeResult(exact); } return null; } diff --git a/src/domain/services/controllers/MaterializeLiveStrategy.ts b/src/domain/services/controllers/MaterializeLiveStrategy.ts index 51a58a60..f2041c72 100644 --- a/src/domain/services/controllers/MaterializeLiveStrategy.ts +++ b/src/domain/services/controllers/MaterializeLiveStrategy.ts @@ -255,11 +255,13 @@ export default class MaterializeLiveStrategy { private async tryResolveExactSnapshot( stateCache: WarpStateCachePort, - opts: { coordinate: WarpStateCoordinate; receipts: boolean }, + opts: { coordinate: WarpStateCoordinate; receipts: boolean; wantDiff: boolean }, ): Promise { const exact = await stateCache.getExact(opts.coordinate); if (canUseSnapshot(exact, { receipts: opts.receipts })) { - return snapshotToMaterializeResult(exact); + return await this.runtime.resumeExactMaterialization(exact, { + wantDiff: opts.wantDiff, + }) ?? snapshotToMaterializeResult(exact); } return null; } diff --git a/src/domain/services/controllers/MaterializeSessionBridge.ts b/src/domain/services/controllers/MaterializeSessionBridge.ts index 0243beb2..68b8fd44 100644 --- a/src/domain/services/controllers/MaterializeSessionBridge.ts +++ b/src/domain/services/controllers/MaterializeSessionBridge.ts @@ -3,6 +3,9 @@ import VersionVector from "../../crdt/VersionVector.ts"; import { Dot } from "../../crdt/Dot.ts"; import type PatchEntry from "../../artifacts/PatchEntry.ts"; import type StateSession from "../../orset/session/StateSession.ts"; +import MaterializationRoot from "../../materialization/MaterializationRoot.ts"; +import MaterializationRoots from "../../materialization/MaterializationRoots.ts"; +import BundleHandle from "../../storage/BundleHandle.ts"; import type { PatchDiff } from "../../types/PatchDiff.ts"; import type { TickReceipt } from "../../types/TickReceipt.ts"; import WarpStateClass from "../state/WarpState.ts"; @@ -32,62 +35,72 @@ export async function reduceSessionBackedState(args: { readonly openStateSession: MaterializeSessionOpener; readonly patches: MaterializeSessionPatchSource; readonly baseState?: WarpStateClass; + readonly roots?: MaterializeSessionOpen; readonly receipts: boolean; readonly wantDiff: boolean; }): Promise<{ readonly state: WarpStateClass; readonly adjacency: MaterializeAdjacency; + readonly roots: MaterializationRoots; readonly receipts?: TickReceipt[]; readonly diff?: PatchDiff; }> { const frame = await openReducerSessionFrame( args.openStateSession, args.baseState, + args.roots, ); - try { - if (args.receipts) { - const result = await reducePatchesInSession(args.patches, frame, { - receipts: true, - }); - const adjacency = await buildAdjacencyFromSession(result.frame.session); - return { - state: await projectFrameToState(result.frame), - adjacency, - receipts: result.receipts, - }; - } - if (args.wantDiff) { - const result = await reducePatchesInSession(args.patches, frame, { - trackDiff: true, - }); - const adjacency = await buildAdjacencyFromSession(result.frame.session); - return { - state: await projectFrameToState(result.frame), - adjacency, - diff: result.diff, - }; - } + let reduced: { + readonly state: WarpStateClass; + readonly adjacency: MaterializeAdjacency; + readonly receipts?: TickReceipt[]; + readonly diff?: PatchDiff; + }; + if (args.receipts) { + const result = await reducePatchesInSession(args.patches, frame, { + receipts: true, + }); + const adjacency = await buildAdjacencyFromSession(result.frame.session); + reduced = { + state: await projectFrameToState(result.frame), + adjacency, + receipts: result.receipts, + }; + } else if (args.wantDiff) { + const result = await reducePatchesInSession(args.patches, frame, { + trackDiff: true, + }); + const adjacency = await buildAdjacencyFromSession(result.frame.session); + reduced = { + state: await projectFrameToState(result.frame), + adjacency, + diff: result.diff, + }; + } else { const result = await reducePatchesInSession(args.patches, frame); const adjacency = await buildAdjacencyFromSession(result.session); - return { + reduced = { state: await projectFrameToState(result), adjacency, }; - } finally { - await frame.session.close(); } + + // close() flushes trie pages, so failed reductions must leave no CAS artifacts. + const roots = materializationRootsFromSession(await frame.session.close()); + return { ...reduced, roots }; } async function openReducerSessionFrame( openStateSession: MaterializeSessionOpener, baseState?: WarpStateClass, + roots?: MaterializeSessionOpen, ): Promise { - const session = await openStateSession({ + const session = await openStateSession(roots ?? { nodeAliveRootOid: null, edgeAliveRootOid: null, }); - if (baseState !== undefined) { + if (baseState !== undefined && roots === undefined) { await seedSessionWithORSet({ session, kind: "node", @@ -108,6 +121,45 @@ async function openReducerSessionFrame( }); } +export function materializationSessionOpen( + roots: MaterializationRoots, +): MaterializeSessionOpen | null { + const nodeAliveRootOid = sessionRootToken(roots.nodeAlive); + const edgeAliveRootOid = sessionRootToken(roots.edgeAlive); + if (nodeAliveRootOid === undefined || edgeAliveRootOid === undefined) { + return null; + } + return Object.freeze({ nodeAliveRootOid, edgeAliveRootOid }); +} + +function materializationRootsFromSession( + roots: MaterializeSessionOpen, +): MaterializationRoots { + return new MaterializationRoots({ + adjacency: MaterializationRoot.unavailable(), + edgeAlive: sessionMaterializationRoot(roots.edgeAliveRootOid), + edgeBirths: MaterializationRoot.unavailable(), + frontier: MaterializationRoot.unavailable(), + nodeAlive: sessionMaterializationRoot(roots.nodeAliveRootOid), + properties: MaterializationRoot.unavailable(), + provenanceSupport: MaterializationRoot.unavailable(), + roaringIndexes: MaterializationRoot.unavailable(), + }); +} + +function sessionRootToken(root: MaterializationRoot): string | null | undefined { + if (root.status === "unavailable") { + return undefined; + } + return root.handle?.toString() ?? null; +} + +function sessionMaterializationRoot(token: string | null): MaterializationRoot { + return token === null + ? MaterializationRoot.empty() + : MaterializationRoot.retained(new BundleHandle(token)); +} + async function seedSessionWithORSet(args: { readonly session: StateSession; readonly kind: "node" | "edge"; diff --git a/src/domain/services/controllers/MaterializeStrategyRuntime.ts b/src/domain/services/controllers/MaterializeStrategyRuntime.ts index 81695cbb..e0d85e7f 100644 --- a/src/domain/services/controllers/MaterializeStrategyRuntime.ts +++ b/src/domain/services/controllers/MaterializeStrategyRuntime.ts @@ -8,6 +8,8 @@ import type { import type { MaterializePatchSummary } from './MaterializePatchSummary.ts'; import type { MaterializeSnapshotPublicationOptions } from './MaterializeSnapshotPublication.ts'; import type { WarpStateSnapshotProvenancePosture } from '../../../ports/WarpStateCachePort.ts'; +import type { UsableSnapshotRecord } from './MaterializeSnapshotCacheResult.ts'; +import type MaterializationHandle from '../../materialization/MaterializationHandle.ts'; import type { MaterializeDeps, MaterializeResult, @@ -36,6 +38,8 @@ export type MaterializeResultBuildInput = { degraded: boolean; ceiling: number | null; frontier: Map | null; + materialization?: MaterializationHandle; + publishSnapshot?: boolean; }; export type MaterializeStrategyRuntime = { @@ -64,5 +68,9 @@ export type MaterializeStrategyRuntime = { provenanceBase?: ProvenanceIndex, ): Promise; buildResult(params: MaterializeResultBuildInput): Promise; + resumeExactMaterialization( + snapshot: UsableSnapshotRecord, + options: { wantDiff: boolean }, + ): Promise; buildProvenance(patches: PatchWithSha[], base?: ProvenanceIndex): ProvenanceIndex; }; diff --git a/src/domain/warp/RuntimeHostBoot.ts b/src/domain/warp/RuntimeHostBoot.ts index 2437ce24..4b6de23c 100644 --- a/src/domain/warp/RuntimeHostBoot.ts +++ b/src/domain/warp/RuntimeHostBoot.ts @@ -33,6 +33,7 @@ import type CommitMessageCodecPort from '../../ports/CommitMessageCodecPort.ts'; import type CheckpointStorePort from '../../ports/CheckpointStorePort.ts'; import type IndexStorePort from '../../ports/IndexStorePort.ts'; import type IntentStorePort from '../../ports/IntentStorePort.ts'; +import type MaterializationStorePort from '../../ports/MaterializationStorePort.ts'; import type EffectSinkPort from '../../ports/EffectSinkPort.ts'; import type RuntimeStorageProviderPort from '../../ports/RuntimeStorageProviderPort.ts'; import type SchedulerPort from '../../ports/SchedulerPort.ts'; @@ -69,6 +70,7 @@ export type RuntimeHostConstructionOptions = { checkpointStore: CheckpointStorePort; indexStore: IndexStorePort; intentStore: IntentStorePort; + materializations: MaterializationStorePort; viewService: MaterializedViewService; stateHashService?: StateHashService; auditService?: AuditReceiptService; @@ -375,6 +377,7 @@ export async function resolveRuntimeHostConstructionOptions( checkpointStore: storageServices.checkpoints, indexStore: resolvedIndexStore, intentStore: storageServices.intents, + materializations: storageServices.materializations, viewService: resolvedViewService, stateHashService: resolvedStateHashService, ...(resolvedAuditService !== undefined ? { auditService: resolvedAuditService } : {}), diff --git a/src/infrastructure/adapters/GitCasMaterializationDescriptor.ts b/src/infrastructure/adapters/GitCasMaterializationDescriptor.ts new file mode 100644 index 00000000..274ad996 --- /dev/null +++ b/src/infrastructure/adapters/GitCasMaterializationDescriptor.ts @@ -0,0 +1,222 @@ +import MaterializationCoordinate from '../../domain/materialization/MaterializationCoordinate.ts'; +import MaterializationRoot, { + type MaterializationRootStatus, +} from '../../domain/materialization/MaterializationRoot.ts'; +import MaterializationRoots, { + MATERIALIZATION_ROOT_NAMES, + type MaterializationRootName, +} from '../../domain/materialization/MaterializationRoots.ts'; +import type BundleHandle from '../../domain/storage/BundleHandle.ts'; +import WarpError from '../../domain/errors/WarpError.ts'; + +export const MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION = 2; + +export type DecodedMaterializationDescriptor = Readonly<{ + coordinate: MaterializationCoordinate; + stateHash: string; + laneName: string; + rootStatuses: ReadonlyMap; +}>; + +export function materializationDescriptorData(input: { + coordinate: MaterializationCoordinate; + stateHash: string; + laneName: string; + roots: MaterializationRoots; +}): object { + return { + schemaVersion: MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION, + laneName: input.laneName, + stateHash: input.stateHash, + coordinate: materializationCoordinateData(input.coordinate), + roots: input.roots.entries().map(([name, root]) => [name, root.status]), + }; +} + +export function materializationCoordinateData( + coordinate: MaterializationCoordinate, +): object { + return { + ceiling: coordinate.ceiling, + frontier: coordinate.frontierEntries.map((entry) => [entry.writerId, entry.patchSha]), + }; +} + +export function decodeMaterializationDescriptor( + value: unknown, +): DecodedMaterializationDescriptor { + requireRecord(value, 'descriptor'); + if (value['schemaVersion'] !== MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION) { + throw descriptorError('materialization descriptor schema is unsupported'); + } + const coordinateValue = value['coordinate']; + requireRecord(coordinateValue, 'descriptor.coordinate'); + return Object.freeze({ + laneName: requireNonEmpty(value['laneName'], 'descriptor.laneName'), + stateHash: requireNonEmpty(value['stateHash'], 'descriptor.stateHash'), + rootStatuses: decodeRootStatuses(value['roots']), + coordinate: new MaterializationCoordinate({ + frontier: decodeFrontier(coordinateValue['frontier']), + ceiling: requireCeiling(coordinateValue['ceiling']), + }), + }); +} + +export function materializationRootsFromDescriptor( + descriptor: DecodedMaterializationDescriptor, + retainedRoots: ReadonlyMap, +): MaterializationRoots { + const statuses = descriptor.rootStatuses; + return new MaterializationRoots({ + adjacency: rootFromMaps(statuses, retainedRoots, 'adjacency'), + edgeAlive: rootFromMaps(statuses, retainedRoots, 'edge-alive'), + edgeBirths: rootFromMaps(statuses, retainedRoots, 'edge-births'), + frontier: rootFromMaps(statuses, retainedRoots, 'frontier'), + nodeAlive: rootFromMaps(statuses, retainedRoots, 'node-alive'), + properties: rootFromMaps(statuses, retainedRoots, 'properties'), + provenanceSupport: rootFromMaps(statuses, retainedRoots, 'provenance-support'), + roaringIndexes: rootFromMaps(statuses, retainedRoots, 'roaring-indexes'), + }); +} + +function decodeFrontier(value: unknown): Map { + if (!Array.isArray(value)) { + throw descriptorError('descriptor.coordinate.frontier must be an array'); + } + const frontier = new Map(); + for (const entry of value) { + const [writerId, patchSha] = decodeFrontierEntry(entry); + if (frontier.has(writerId)) { + throw descriptorError('descriptor coordinate contains a duplicate frontier writer'); + } + frontier.set(writerId, patchSha); + } + return frontier; +} + +function decodeFrontierEntry(value: unknown): readonly [string, string] { + if (!Array.isArray(value) || value.length !== 2) { + throw descriptorError('descriptor coordinate contains an invalid frontier entry'); + } + return Object.freeze([ + requireNonEmpty(arrayValue(value, 0), 'descriptor frontier writerId'), + requireNonEmpty(arrayValue(value, 1), 'descriptor frontier patchSha'), + ]); +} + +function decodeRootStatuses( + value: unknown, +): ReadonlyMap { + if (!Array.isArray(value)) { + throw descriptorError('descriptor.roots must be an array'); + } + const statuses = new Map(); + for (const entry of value) { + const [name, status] = decodeRootStatusEntry(entry); + if (statuses.has(name)) { + throw descriptorError(`descriptor has duplicate ${name} root status`); + } + statuses.set(name, status); + } + for (const name of MATERIALIZATION_ROOT_NAMES) { + requireRootStatus(statuses, name); + } + return statuses; +} + +function decodeRootStatusEntry( + value: unknown, +): readonly [MaterializationRootName, MaterializationRootStatus] { + if (!Array.isArray(value) || value.length !== 2) { + throw descriptorError('descriptor contains an invalid root status entry'); + } + const name = decodeRootStatusName(arrayValue(value, 0)); + return Object.freeze([name, decodeRootStatus(arrayValue(value, 1), name)]); +} + +function decodeRootStatusName(value: unknown): MaterializationRootName { + if (typeof value !== 'string') { + throw descriptorError('descriptor contains an unknown root status name'); + } + const name = MATERIALIZATION_ROOT_NAMES.find((candidate) => candidate === value); + if (name === undefined) { + throw descriptorError('descriptor contains an unknown root status name'); + } + return name; +} + +function decodeRootStatus( + value: unknown, + name: MaterializationRootName, +): MaterializationRootStatus { + if (value !== 'retained' && value !== 'empty' && value !== 'unavailable') { + throw descriptorError(`descriptor contains an invalid ${name} root status`); + } + return value; +} + +function rootFromMaps( + statuses: ReadonlyMap, + roots: ReadonlyMap, + name: MaterializationRootName, +): MaterializationRoot { + const status = requireRootStatus(statuses, name); + const retained = roots.get(name); + if (status === 'retained') { + if (retained === undefined) { + throw descriptorError(`materialization bundle has no ${name} root bundle`); + } + return MaterializationRoot.retained(retained); + } + if (retained !== undefined) { + throw descriptorError(`materialization bundle has an unexpected ${name} root bundle`); + } + return status === 'empty' + ? MaterializationRoot.empty() + : MaterializationRoot.unavailable(); +} + +function requireRootStatus( + statuses: ReadonlyMap, + name: MaterializationRootName, +): MaterializationRootStatus { + const status = statuses.get(name); + if (status === undefined) { + throw descriptorError(`descriptor has no ${name} root status`); + } + return status; +} + +function requireCeiling(value: unknown): number | null { + if (value === null) { + return null; + } + if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 0) { + throw descriptorError('descriptor coordinate ceiling is invalid'); + } + return value; +} + +function requireRecord( + value: unknown, + field: string, +): asserts value is Record { + if (value === null || typeof value !== 'object' || Array.isArray(value)) { + throw descriptorError(`${field} must be an object`); + } +} + +function requireNonEmpty(value: unknown, field: string): string { + if (typeof value !== 'string' || value.length === 0) { + throw descriptorError(`${field} must be a non-empty string`); + } + return value; +} + +function arrayValue(values: readonly unknown[], index: number): unknown { + return values[index]; +} + +function descriptorError(message: string): WarpError { + return new WarpError(`Materialization storage ${message}`, 'E_MATERIALIZATION_STORAGE'); +} diff --git a/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts b/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts index 669d6dab..73410f55 100644 --- a/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts +++ b/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts @@ -10,6 +10,7 @@ import type { } from '@git-stunts/git-cas'; import MaterializationCoordinate from '../../domain/materialization/MaterializationCoordinate.ts'; import MaterializationHandle from '../../domain/materialization/MaterializationHandle.ts'; +import type MaterializationRoot from '../../domain/materialization/MaterializationRoot.ts'; import MaterializationRoots, { MATERIALIZATION_ROOT_NAMES, type MaterializationRootName, @@ -23,11 +24,18 @@ import MaterializationStorePort, { type RetainMaterializationRequest, } from '../../ports/MaterializationStorePort.ts'; import { adaptGitCasRetentionWitness } from './GitCasRetentionWitnessAdapter.ts'; +import { + decodeMaterializationDescriptor, + MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION, + materializationCoordinateData, + materializationDescriptorData, + materializationRootsFromDescriptor, + type DecodedMaterializationDescriptor, +} from './GitCasMaterializationDescriptor.ts'; const CACHE_NAMESPACE = 'git-warp/materializations'; const DESCRIPTOR_PATH = 'meta/descriptor'; const MAX_DESCRIPTOR_BYTES = 1024 * 1024; -const SCHEMA_VERSION = 1; // A root-list change also requires a descriptor schema-version change. const MATERIALIZATION_MEMBER_COUNT = MATERIALIZATION_ROOT_NAMES.length + 1; @@ -41,15 +49,9 @@ export type GitCasMaterializationFacade = { readonly pages: Pick; }; -type DecodedDescriptor = Readonly<{ - coordinate: MaterializationCoordinate; - stateHash: string; - laneName: string; -}>; - type DecodedMaterializationMembers = Readonly<{ descriptor: PageHandle; - roots: MaterializationRoots; + retainedRoots: ReadonlyMap; }>; type MaterializationMemberAccumulator = { @@ -101,10 +103,11 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt request: RetainMaterializationRequest, stateHash: string, ): Promise { - const descriptorBytes = this.#codec.encode(descriptorData({ + const descriptorBytes = this.#codec.encode(materializationDescriptorData({ coordinate: request.coordinate, stateHash, laneName: this.#laneName, + roots: request.roots, })); requireDescriptorSize(descriptorBytes); @@ -167,7 +170,7 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt laneName: descriptor.laneName, bundle, coordinate: descriptor.coordinate, - roots: members.roots, + roots: materializationRootsFromDescriptor(descriptor, members.retainedRoots), stateHash: descriptor.stateHash, retention: adaptGitCasRetentionWitness(hit.evidence.toJSON()), }); @@ -175,23 +178,23 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt async #cacheKey(coordinate: MaterializationCoordinate): Promise { const encoded = this.#codec.encode({ - schemaVersion: SCHEMA_VERSION, + schemaVersion: MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION, laneName: this.#laneName, - coordinate: coordinateData(coordinate), + coordinate: materializationCoordinateData(coordinate), }); const digest = requireNonEmpty( await this.#crypto.hash('sha256', encoded), 'coordinate digest', ); - return `v${SCHEMA_VERSION}:${digest}`; + return `v${MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION}:${digest}`; } - async #readDescriptor(handle: PageHandle): Promise { + async #readDescriptor(handle: PageHandle): Promise { const bytes = await this.#cas.pages.get({ handle, maxBytes: MAX_DESCRIPTOR_BYTES, }); - return decodeDescriptor(this.#codec.decode(bytes)); + return decodeMaterializationDescriptor(this.#codec.decode(bytes)); } async #readMembers(bundle: BundleHandle): Promise { @@ -205,87 +208,16 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt } } -function descriptorData(descriptor: DecodedDescriptor): object { - return { - schemaVersion: SCHEMA_VERSION, - laneName: descriptor.laneName, - stateHash: descriptor.stateHash, - coordinate: coordinateData(descriptor.coordinate), - }; -} - -function coordinateData(coordinate: MaterializationCoordinate): object { - return { - ceiling: coordinate.ceiling, - frontier: coordinate.frontierEntries.map((entry) => [entry.writerId, entry.patchSha]), - }; -} - function* materializationMembers( descriptorHandle: string, roots: MaterializationRoots, ): Generator<[string, BundleMemberInput]> { yield [DESCRIPTOR_PATH, descriptorHandle]; - for (const [name, handle] of roots.entries()) { - yield [`roots/${name}`, handle.toString()]; - } -} - -function decodeDescriptor(value: unknown): DecodedDescriptor { - requireRecord(value, 'descriptor'); - const descriptor = value; - if (descriptor['schemaVersion'] !== SCHEMA_VERSION) { - throw storageError('materialization descriptor schema is unsupported'); - } - const coordinateValue = descriptor['coordinate']; - requireRecord(coordinateValue, 'descriptor.coordinate'); - const frontier = decodeFrontier(coordinateValue['frontier']); - return Object.freeze({ - laneName: requireNonEmpty(descriptor['laneName'], 'descriptor.laneName'), - stateHash: requireNonEmpty(descriptor['stateHash'], 'descriptor.stateHash'), - coordinate: new MaterializationCoordinate({ - frontier, - ceiling: requireCeiling(coordinateValue['ceiling']), - }), - }); -} - -function decodeFrontier(value: unknown): Map { - if (!Array.isArray(value)) { - throw storageError('descriptor.coordinate.frontier must be an array'); - } - const frontier = new Map(); - for (const entry of value) { - const [writerId, patchSha] = decodeFrontierEntry(entry); - if (frontier.has(writerId)) { - throw storageError('descriptor coordinate contains a duplicate frontier writer'); + for (const [name, root] of roots.entries()) { + if (root.status === 'retained') { + yield [`roots/${name}`, requireRetainedHandle(root, name).toString()]; } - frontier.set(writerId, patchSha); - } - return frontier; -} - -function decodeFrontierEntry(value: unknown): readonly [string, string] { - if (!Array.isArray(value) || value.length !== 2) { - throw storageError('descriptor coordinate contains an invalid frontier entry'); } - return Object.freeze([ - requireNonEmpty(value[0], 'descriptor frontier writerId'), - requireNonEmpty(value[1], 'descriptor frontier patchSha'), - ]); -} - -function rootsFromMap(roots: ReadonlyMap): MaterializationRoots { - return new MaterializationRoots({ - adjacency: requireRoot(roots, 'adjacency'), - edgeAlive: requireRoot(roots, 'edge-alive'), - edgeBirths: requireRoot(roots, 'edge-births'), - frontier: requireRoot(roots, 'frontier'), - nodeAlive: requireRoot(roots, 'node-alive'), - properties: requireRoot(roots, 'properties'), - provenanceSupport: requireRoot(roots, 'provenance-support'), - roaringIndexes: requireRoot(roots, 'roaring-indexes'), - }); } function createMemberAccumulator(): MaterializationMemberAccumulator { @@ -349,19 +281,18 @@ function finishMaterializationMembers( } return Object.freeze({ descriptor: accumulator.descriptor, - roots: rootsFromMap(accumulator.roots), + retainedRoots: new Map(accumulator.roots), }); } -function requireRoot( - roots: ReadonlyMap, +function requireRetainedHandle( + root: MaterializationRoot, name: MaterializationRootName, ): BundleHandle { - const root = roots.get(name); - if (root === undefined) { - throw storageError(`materialization bundle has no ${name} root bundle`); + if (root.handle === null) { + throw storageError(`${name} retained root has no bundle handle`); } - return root; + return root.handle; } function parseRootName(path: string): MaterializationRootName | null { @@ -373,25 +304,6 @@ function parseRootName(path: string): MaterializationRootName | null { return MATERIALIZATION_ROOT_NAMES.find((name) => name === candidate) ?? null; } -function requireCeiling(value: unknown): number | null { - if (value === null) { - return null; - } - if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 0) { - throw storageError('descriptor coordinate ceiling is invalid'); - } - return value; -} - -function requireRecord( - value: unknown, - field: string, -): asserts value is Record { - if (value === null || typeof value !== 'object' || Array.isArray(value)) { - throw storageError(`${field} must be an object`); - } -} - function requireRetainRequest(request: RetainMaterializationRequest): void { if (request === null || typeof request !== 'object' || Array.isArray(request)) { throw storageError('retain request must be an object'); diff --git a/test/helpers/InMemoryMaterializationStore.ts b/test/helpers/InMemoryMaterializationStore.ts new file mode 100644 index 00000000..0a55be41 --- /dev/null +++ b/test/helpers/InMemoryMaterializationStore.ts @@ -0,0 +1,63 @@ +import MaterializationHandle from '../../src/domain/materialization/MaterializationHandle.ts'; +import type MaterializationCoordinate from '../../src/domain/materialization/MaterializationCoordinate.ts'; +import BundleHandle from '../../src/domain/storage/BundleHandle.ts'; +import StorageRetentionWitness, { + StorageRetentionRoot, +} from '../../src/domain/storage/StorageRetentionWitness.ts'; +import MaterializationStorePort, { + type RetainMaterializationRequest, +} from '../../src/ports/MaterializationStorePort.ts'; + +/** Behavioral retained-materialization store for controller tests. */ +export default class InMemoryMaterializationStore extends MaterializationStorePort { + readonly exactLookups: MaterializationCoordinate[] = []; + readonly retainedRequests: RetainMaterializationRequest[] = []; + readonly #handles = new Map(); + #nextHandle = 1; + + override retain(request: RetainMaterializationRequest): Promise { + this.retainedRequests.push(request); + const bundle = new BundleHandle(`test:materialization:${this.#nextHandle}`); + this.#nextHandle += 1; + const handle = new MaterializationHandle({ + laneName: 'test-lane', + bundle, + coordinate: request.coordinate, + roots: request.roots, + stateHash: request.stateHash, + retention: retentionWitness(bundle), + }); + this.#handles.set(coordinateKey(request.coordinate), handle); + return Promise.resolve(handle); + } + + override findExact( + coordinate: MaterializationCoordinate, + ): Promise { + this.exactLookups.push(coordinate); + return Promise.resolve(this.#handles.get(coordinateKey(coordinate)) ?? null); + } +} + +function coordinateKey(coordinate: MaterializationCoordinate): string { + return JSON.stringify({ + ceiling: coordinate.ceiling, + frontier: coordinate.frontierEntries, + }); +} + +function retentionWitness(handle: BundleHandle): StorageRetentionWitness { + return new StorageRetentionWitness({ + handle, + policy: 'evictable', + reachability: 'anchored', + root: new StorageRetentionRoot({ + kind: 'cache-set', + namespace: 'test/materializations', + locator: 'test/materializations', + generation: 'test-generation', + path: handle.toString(), + }), + observedAt: '1970-01-01T00:00:00.000Z', + }); +} diff --git a/test/integration/api/materialization.retainedResume.test.ts b/test/integration/api/materialization.retainedResume.test.ts new file mode 100644 index 00000000..e4d32d4c --- /dev/null +++ b/test/integration/api/materialization.retainedResume.test.ts @@ -0,0 +1,147 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import RuntimeHost from '../../../src/domain/RuntimeHost.ts'; +import type MaterializationCoordinate from '../../../src/domain/materialization/MaterializationCoordinate.ts'; +import type MaterializationHandle from '../../../src/domain/materialization/MaterializationHandle.ts'; +import GitCasRepositoryAdapter from '../../../src/infrastructure/adapters/GitCasRepositoryAdapter.ts'; +import MaterializationStorePort, { + type RetainMaterializationRequest, +} from '../../../src/ports/MaterializationStorePort.ts'; +import type RuntimeStorageProviderPort from '../../../src/ports/RuntimeStorageProviderPort.ts'; +import type { + RuntimeStorageRequest, + RuntimeStorageServices, +} from '../../../src/ports/RuntimeStorageProviderPort.ts'; +import { createTestRepo } from './helpers/setup.ts'; + +describe('API: retained materialization resume', () => { + let repo: Awaited> | null = null; + + beforeEach(async () => { + repo = await createTestRepo('retained-materialization-resume'); + }); + + afterEach(async () => { + await repo?.cleanup(); + }); + + it('reopens exact roots without replay in-process and through a fresh runtime adapter', async () => { + if (repo === null) { + throw new Error('Test repository is not initialized'); + } + const firstProvider = recordingProvider(repo); + const firstRuntime = await openRuntime(repo, firstProvider); + await firstRuntime.patch((patch) => { + patch.addNode('node:retained'); + }); + + const cold = await firstRuntime.materialize(); + const firstStore = requireMaterializations(firstProvider); + expect(firstStore.retainRequests).toHaveLength(1); + expect(firstStore.retainedHandles).toHaveLength(1); + const coldHandle = firstStore.retainedHandles[0]; + expect(coldHandle?.roots.nodeAlive.status).toBe('retained'); + expect(coldHandle?.roots.edgeAlive.status).toBe('empty'); + + const sameRuntimeReplay = vi.spyOn(firstRuntime, '_loadPatchChainFromSha'); + const warm = await firstRuntime.materialize(); + + expect(sameRuntimeReplay).not.toHaveBeenCalled(); + expect(firstStore.exactLookups).toHaveLength(1); + expect(firstStore.retainedHandles).toHaveLength(1); + expect(firstStore.exactHits[0]?.bundle.equals(coldHandle?.bundle)).toBe(true); + expect(warm).toEqual(cold); + + const reopenedProvider = recordingProvider(repo); + const reopenedRuntime = await openRuntime(repo, reopenedProvider); + const reopenedReplay = vi.spyOn(reopenedRuntime, '_loadPatchChainFromSha'); + const reopened = await reopenedRuntime.materialize(); + const reopenedStore = requireMaterializations(reopenedProvider); + + expect(reopenedReplay).not.toHaveBeenCalled(); + expect(reopenedStore.exactLookups).toHaveLength(1); + expect(reopenedStore.retainRequests).toHaveLength(0); + expect(reopenedStore.exactHits[0]?.bundle.equals(coldHandle?.bundle)).toBe(true); + expect(reopened.nodeAlive.contains('node:retained')).toBe(true); + }); +}); + +class RecordingMaterializationStore extends MaterializationStorePort { + readonly retainRequests: RetainMaterializationRequest[] = []; + readonly retainedHandles: MaterializationHandle[] = []; + readonly exactLookups: MaterializationCoordinate[] = []; + readonly exactHits: MaterializationHandle[] = []; + readonly #delegate: MaterializationStorePort; + + constructor(delegate: MaterializationStorePort) { + super(); + this.#delegate = delegate; + } + + override async retain(request: RetainMaterializationRequest): Promise { + this.retainRequests.push(request); + const retained = await this.#delegate.retain(request); + this.retainedHandles.push(retained); + return retained; + } + + override async findExact( + coordinate: MaterializationCoordinate, + ): Promise { + this.exactLookups.push(coordinate); + const hit = await this.#delegate.findExact(coordinate); + if (hit !== null) { + this.exactHits.push(hit); + } + return hit; + } +} + +class RecordingRuntimeStorageProvider implements RuntimeStorageProviderPort { + readonly #delegate: RuntimeStorageProviderPort; + materializations: RecordingMaterializationStore | null = null; + + constructor(delegate: RuntimeStorageProviderPort) { + this.#delegate = delegate; + } + + async createRuntimeStorageServices( + request: RuntimeStorageRequest, + ): Promise { + const services = await this.#delegate.createRuntimeStorageServices(request); + const materializations = new RecordingMaterializationStore(services.materializations); + this.materializations = materializations; + return Object.freeze({ ...services, materializations }); + } +} + +function recordingProvider( + repo: NonNullable>>, +): RecordingRuntimeStorageProvider { + return new RecordingRuntimeStorageProvider(new GitCasRepositoryAdapter({ + plumbing: repo.plumbing, + history: repo.persistence, + })); +} + +async function openRuntime( + repo: NonNullable>>, + runtimeStorage: RuntimeStorageProviderPort, +): Promise { + return await RuntimeHost.open({ + persistence: repo.persistence, + runtimeStorage, + graphName: 'events', + writerId: 'writer-1', + codec: repo.codec, + crypto: repo.crypto, + }); +} + +function requireMaterializations( + provider: RecordingRuntimeStorageProvider, +): RecordingMaterializationStore { + if (provider.materializations === null) { + throw new Error('Runtime storage services were not created'); + } + return provider.materializations; +} diff --git a/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts b/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts index 885fe873..41f5aa58 100644 --- a/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts +++ b/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts @@ -9,6 +9,7 @@ import ContentAddressableStore, { import Plumbing from '@git-stunts/plumbing'; import { afterEach, beforeEach, 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 BundleHandle from '../../../../src/domain/storage/BundleHandle.ts'; import GitCasRepositoryAdapter from '../../../../src/infrastructure/adapters/GitCasRepositoryAdapter.ts'; @@ -54,8 +55,12 @@ describe('GitCasMaterializationStoreAdapter integration', () => { if (resolved === null) { throw new Error('Retained materialization was not reopened'); } + const nodeAliveRoot = resolved.roots.nodeAlive.handle; + if (nodeAliveRoot === null) { + throw new Error('Retained materialization did not expose its node root'); + } const reopenedTrie = new GitCasTrieStoreAdapter({ cas: reopenedCas }); - const children = await reopenedTrie.readBranch(resolved.roots.nodeAlive.toString()); + const children = await reopenedTrie.readBranch(nodeAliveRoot.toString()); const child = children.get(0); if (child === undefined) { throw new Error('Retained trie root did not contain its leaf child'); @@ -63,8 +68,8 @@ describe('GitCasMaterializationStoreAdapter integration', () => { const unreachable = await prunableOids(harness.path); expect(resolved.bundle.equals(retained.bundle)).toBe(true); - expect(resolved.roots.entries().map(([name, handle]) => [name, handle.toString()])) - .toEqual(rootFixture.roots.entries().map(([name, handle]) => [name, handle.toString()])); + expect(resolved.roots.entries().map(([name, root]) => rootSignature(name, root))) + .toEqual(rootFixture.roots.entries().map(([name, root]) => rootSignature(name, root))); expect(await reopenedTrie.readLeaf(child)).toEqual(trieFixture.bytes); expect(unreachable).not.toContain(GitCasBundleHandle.parse(retained.bundle.toString()).oid); for (const oid of rootFixture.retainedOids) { @@ -197,17 +202,24 @@ function rootsFromHandles(handles: readonly BundleHandle[]): MaterializationRoot throw new Error('Root integration fixture did not create every root'); } return new MaterializationRoots({ - adjacency, - edgeAlive, - edgeBirths, - frontier, - nodeAlive, - properties, - provenanceSupport, - roaringIndexes, + adjacency: MaterializationRoot.retained(adjacency), + edgeAlive: MaterializationRoot.retained(edgeAlive), + edgeBirths: MaterializationRoot.retained(edgeBirths), + frontier: MaterializationRoot.retained(frontier), + nodeAlive: MaterializationRoot.retained(nodeAlive), + properties: MaterializationRoot.retained(properties), + provenanceSupport: MaterializationRoot.retained(provenanceSupport), + roaringIndexes: MaterializationRoot.retained(roaringIndexes), }); } +function rootSignature( + name: string, + root: MaterializationRoot, +): readonly [string, string, string | null] { + return [name, root.status, root.handle?.toString() ?? null]; +} + async function prunableOids(path: string): Promise> { const { stdout } = await execFileAsync( 'git', diff --git a/test/unit/domain/materialization/MaterializationIdentity.test.ts b/test/unit/domain/materialization/MaterializationIdentity.test.ts index ede2ae63..f85e80f2 100644 --- a/test/unit/domain/materialization/MaterializationIdentity.test.ts +++ b/test/unit/domain/materialization/MaterializationIdentity.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from 'vitest'; import MaterializationCoordinate from '../../../../src/domain/materialization/MaterializationCoordinate.ts'; import MaterializationHandle from '../../../../src/domain/materialization/MaterializationHandle.ts'; +import MaterializationRoot from '../../../../src/domain/materialization/MaterializationRoot.ts'; import MaterializationRoots, { MATERIALIZATION_ROOT_NAMES, type MaterializationRootsOptions, @@ -123,7 +124,7 @@ describe('MaterializationRoots', () => { 'properties', 'provenanceSupport', 'roaringIndexes', - ])('rejects a non-bundle %s root', (field) => { + ])('rejects a non-materialization %s root', (field) => { const options = rootsOptions(); Reflect.set(options, field, new StorageHandle('not-a-bundle')); expect(() => construct(MaterializationRoots, options)).toThrowError( @@ -136,6 +137,30 @@ describe('MaterializationRoots', () => { }); }); +describe('MaterializationRoot', () => { + it('distinguishes retained, empty, and unavailable root posture', () => { + const handle = bundleHandle('retained'); + const retained = MaterializationRoot.retained(handle); + const empty = MaterializationRoot.empty(); + const unavailable = MaterializationRoot.unavailable(); + + expect(retained).toMatchObject({ status: 'retained', handle }); + expect(empty).toMatchObject({ status: 'empty', handle: null }); + expect(unavailable).toMatchObject({ status: 'unavailable', handle: null }); + expect(Object.isFrozen(retained)).toBe(true); + expect(Object.isFrozen(empty)).toBe(true); + expect(Object.isFrozen(unavailable)).toBe(true); + }); + + it('rejects a retained root without bundle identity', () => { + expect(() => Reflect.apply( + MaterializationRoot.retained, + MaterializationRoot, + [new StorageHandle('not-a-bundle')], + )).toThrowError(/Materialization root/u); + }); +}); + describe('MaterializationHandle', () => { it('binds an exact coordinate and independent roots to retained bundle evidence', () => { const bundle = bundleHandle('materialization'); @@ -195,17 +220,21 @@ function bundleHandle(name: string): BundleHandle { function rootsOptions(): MaterializationRootsOptions { return { - adjacency: bundleHandle('adjacency'), - edgeAlive: bundleHandle('edge-alive'), - edgeBirths: bundleHandle('edge-births'), - frontier: bundleHandle('frontier'), - nodeAlive: bundleHandle('node-alive'), - properties: bundleHandle('properties'), - provenanceSupport: bundleHandle('provenance-support'), - roaringIndexes: bundleHandle('roaring-indexes'), + adjacency: retainedRoot('adjacency'), + edgeAlive: retainedRoot('edge-alive'), + edgeBirths: retainedRoot('edge-births'), + frontier: retainedRoot('frontier'), + nodeAlive: retainedRoot('node-alive'), + properties: retainedRoot('properties'), + provenanceSupport: retainedRoot('provenance-support'), + roaringIndexes: retainedRoot('roaring-indexes'), }; } +function retainedRoot(name: string): MaterializationRoot { + return MaterializationRoot.retained(bundleHandle(name)); +} + function materializationRoots(): MaterializationRoots { return new MaterializationRoots(rootsOptions()); } diff --git a/test/unit/domain/services/controllers/MaterializeController.snapshotCache.test.ts b/test/unit/domain/services/controllers/MaterializeController.snapshotCache.test.ts index 59c4e86a..1aa959aa 100644 --- a/test/unit/domain/services/controllers/MaterializeController.snapshotCache.test.ts +++ b/test/unit/domain/services/controllers/MaterializeController.snapshotCache.test.ts @@ -5,6 +5,7 @@ import Patch from '../../../../../src/domain/types/Patch.ts'; import type { CheckpointData, PatchWithSha } from '../../../../../src/domain/capabilities/PatchCollector.ts'; import { DEFAULT_COMMIT_MESSAGE_CODEC } from '../../../../../src/infrastructure/adapters/TrailerCommitMessageCodecAdapter.ts'; import InMemoryCheckpointStore from '../../../../helpers/InMemoryCheckpointStore.ts'; +import InMemoryMaterializationStore from '../../../../helpers/InMemoryMaterializationStore.ts'; type Coordinate = { frontier: Map; @@ -132,6 +133,7 @@ function createControllerFixtures() { readBlob: vi.fn().mockResolvedValue(new Uint8Array([1])), }, checkpointStore: new InMemoryCheckpointStore(), + materializations: new InMemoryMaterializationStore(), commitMessageCodec: DEFAULT_COMMIT_MESSAGE_CODEC, getStateCache: () => stateCache, patches, diff --git a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts index 71607dc5..2e7cbc92 100644 --- a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts +++ b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts @@ -1,7 +1,9 @@ import { describe, expect, it, vi } from "vitest"; import { Dot } from "../../../../../src/domain/crdt/Dot.ts"; +import PatchEntry from "../../../../../src/domain/artifacts/PatchEntry.ts"; import MaterializeController from "../../../../../src/domain/services/controllers/MaterializeController.ts"; +import { reduceSessionBackedState } from "../../../../../src/domain/services/controllers/MaterializeSessionBridge.ts"; import NodeAdd from "../../../../../src/domain/types/ops/NodeAdd.ts"; import EdgeAdd from "../../../../../src/domain/types/ops/EdgeAdd.ts"; import StateSession from "../../../../../src/domain/orset/session/StateSession.ts"; @@ -15,6 +17,7 @@ import { createEmptyState } from "../../../../../src/domain/services/JoinReducer import Patch from "../../../../../src/domain/types/Patch.ts"; import type { CheckpointData, PatchWithSha } from "../../../../../src/domain/capabilities/PatchCollector.ts"; import InMemoryCheckpointStore from "../../../../helpers/InMemoryCheckpointStore.ts"; +import InMemoryMaterializationStore from "../../../../helpers/InMemoryMaterializationStore.ts"; const GEOMETRY = TrieGeometry.default16way(); @@ -134,6 +137,7 @@ function createControllerFixtures() { }; const store = new InMemoryTrieStore(); const pageCache = new PageCache({ maxResident: 32 }); + const materializations = new InMemoryMaterializationStore(); const openStateSession = vi.fn( async (roots: { readonly nodeAliveRootOid: string | null; @@ -170,6 +174,7 @@ function createControllerFixtures() { readBlob: vi.fn().mockResolvedValue(new Uint8Array([1])), }, checkpointStore: new InMemoryCheckpointStore(), + materializations, commitMessageCodec: DEFAULT_COMMIT_MESSAGE_CODEC, getStateCache: () => stateCache, patches, @@ -183,12 +188,44 @@ function createControllerFixtures() { patches, stateCache, openStateSession, + materializations, + deps, }; } describe("MaterializeController — state session integration", () => { + it("does not flush partial trie roots when session reduction fails", async () => { + const { openStateSession } = createControllerFixtures(); + const close = vi.spyOn(StateSession.prototype, "close"); + const failure = new Error("patch source failed"); + const partial = new PatchEntry(nodeAddPatchRecord({ + writer: "writer-1", + lamport: 1, + sha: "deadbeef", + node: "node:partial", + })); + const patches = { + async *[Symbol.asyncIterator]() { + yield partial; + throw failure; + }, + }; + + try { + await expect(reduceSessionBackedState({ + openStateSession, + patches, + receipts: false, + wantDiff: false, + })).rejects.toBe(failure); + expect(close).not.toHaveBeenCalled(); + } finally { + close.mockRestore(); + } + }); + it("replays live materialization through StateSession and returns an explicit WarpState projection bridge", async () => { - const { controller, patches, openStateSession } = createControllerFixtures(); + const { controller, patches, openStateSession, materializations } = createControllerFixtures(); patches.collectForFrontier.mockResolvedValue([ nodeAddPatchRecord({ writer: "writer-1", @@ -223,6 +260,54 @@ describe("MaterializeController — state session integration", () => { expect(result.adjacency.outgoing.get("node:session")).toEqual([ { neighborId: "node:peer", label: "follows" }, ]); + expect(materializations.retainedRequests).toHaveLength(1); + expect(materializations.retainedRequests[0]?.roots.nodeAlive.status).toBe("retained"); + expect(materializations.retainedRequests[0]?.roots.edgeAlive.status).toBe("retained"); + expect(materializations.retainedRequests[0]?.roots.properties.status).toBe("unavailable"); + expect(result.materialization?.roots.nodeAlive.status).toBe("retained"); + }); + + it("reopens exact retained roots with zero covered patch replay after controller restart", async () => { + const fixtures = createControllerFixtures(); + fixtures.patches.collectForFrontier.mockResolvedValue([ + nodeAddPatchRecord({ + writer: "writer-1", + lamport: 1, + sha: "a1b2", + node: "node:retained", + }), + ]); + + const cold = await fixtures.controller.materialize(); + const published = fixtures.stateCache.put.mock.calls[0]?.[0]; + if (published === undefined) { + throw new Error("Cold materialization did not publish its state snapshot"); + } + fixtures.stateCache.getExact.mockResolvedValue(published); + fixtures.patches.collectForFrontier.mockClear(); + fixtures.stateCache.put.mockClear(); + fixtures.openStateSession.mockClear(); + + const warm = await fixtures.controller.materialize(); + const reopened = await new MaterializeController(fixtures.deps).materialize(); + + const retained = fixtures.materializations.retainedRequests[0]; + expect(retained).toBeDefined(); + expect(fixtures.materializations.retainedRequests).toHaveLength(1); + expect(fixtures.materializations.exactLookups).toHaveLength(2); + expect(fixtures.patches.collectForFrontier).not.toHaveBeenCalled(); + expect(fixtures.stateCache.put).not.toHaveBeenCalled(); + expect(fixtures.openStateSession).toHaveBeenCalledTimes(2); + expect(fixtures.openStateSession).toHaveBeenNthCalledWith(1, { + nodeAliveRootOid: retained?.roots.nodeAlive.handle?.toString(), + edgeAliveRootOid: retained?.roots.edgeAlive.handle?.toString() ?? null, + }); + expect(warm.patchCount).toBe(0); + expect(reopened.patchCount).toBe(0); + expect(warm.state.nodeAlive.contains("node:retained")).toBe(true); + expect(reopened.state.nodeAlive.contains("node:retained")).toBe(true); + expect(warm.materialization?.bundle.equals(cold.materialization?.bundle)).toBe(true); + expect(reopened.materialization?.bundle.equals(cold.materialization?.bundle)).toBe(true); }); it("hydrates a predecessor snapshot into StateSession before replaying the suffix", async () => { diff --git a/test/unit/domain/services/controllers/MaterializePatchStreamReducer.test.ts b/test/unit/domain/services/controllers/MaterializePatchStreamReducer.test.ts index 09cb1109..c5c67b26 100644 --- a/test/unit/domain/services/controllers/MaterializePatchStreamReducer.test.ts +++ b/test/unit/domain/services/controllers/MaterializePatchStreamReducer.test.ts @@ -25,6 +25,7 @@ import type CodecValue from '../../../../../src/domain/types/codec/CodecValue.ts import type LogFields from '../../../../../src/domain/types/log/LogFields.ts'; import InMemoryCheckpointStore from '../../../../helpers/InMemoryCheckpointStore.ts'; import { InMemoryTrieStore } from '../../../../helpers/trieHelpers.ts'; +import InMemoryMaterializationStore from '../../../../helpers/InMemoryMaterializationStore.ts'; describe('MaterializePatchStreamReducer', () => { it('reduces each patch before requesting the next stream item', async () => { @@ -212,6 +213,7 @@ function materializeDeps(patches: PatchCollector): MaterializeDeps { crypto: new TestCrypto(), persistence: new TestPersistence(), checkpointStore: new InMemoryCheckpointStore(), + materializations: new InMemoryMaterializationStore(), patches, graphCloner: new UnusedDetachedGraphFactory(), graphName: 'stream-memory-witness', diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts index 22bed298..d3bb71fb 100644 --- a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts +++ b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from 'vitest'; import type { CacheStoreResult } from '@git-stunts/git-cas'; 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 BundleHandle from '../../../../src/domain/storage/BundleHandle.ts'; import GitCasMaterializationStoreAdapter, { @@ -56,14 +57,14 @@ describe('GitCasMaterializationStoreAdapter', () => { expect(resolved?.bundle.equals(retained.bundle)).toBe(true); expect(resolved?.coordinate.equals(coordinate)).toBe(true); expect(resolved?.stateHash).toBe('state-hash'); - expect(resolved?.roots.entries().map(([name, handle]) => [name, handle.toString()])) - .toEqual(roots.entries().map(([name, handle]) => [name, handle.toString()])); + expect(resolved?.roots.entries().map(([name, root]) => rootSignature(name, root))) + .toEqual(roots.entries().map(([name, root]) => rootSignature(name, root))); const members = harness.cas.readBundleMembers(retained.bundle.toString()); 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(/^v1:[0-9a-f]{64}$/u); + expect(cacheKeys[0]).toMatch(/^v2:[0-9a-f]{64}$/u); expect(cacheKeys[0]?.length).toBeLessThan(1024); }); @@ -72,6 +73,29 @@ describe('GitCasMaterializationStoreAdapter', () => { expect(await harness.adapter.findExact(exactCoordinate())).toBeNull(); }); + it('round-trips partial roots without inventing bundles for empty or unavailable state', async () => { + const harness = await createHarness(); + const page = await harness.cas.pages.put({ source: new Uint8Array([1]) }); + const nodeBundle = await harness.cas.bundles.putOrdered({ + members: [['root', page.handle]], + }); + const roots = partialRoots(new BundleHandle(nodeBundle.handle.toString())); + + const retained = await harness.adapter.retain({ + coordinate: exactCoordinate(), + roots, + stateHash: 'partial-state-hash', + }); + const resolved = await harness.adapter.findExact(exactCoordinate()); + + expect(harness.cas.readBundleMembers(retained.bundle.toString()).map(([path]) => path)) + .toEqual(['meta/descriptor', 'roots/node-alive']); + expect(resolved?.roots.nodeAlive.status).toBe('retained'); + expect(resolved?.roots.nodeAlive.handle?.toString()).toBe(nodeBundle.handle.toString()); + expect(resolved?.roots.edgeAlive.status).toBe('empty'); + expect(resolved?.roots.properties.status).toBe('unavailable'); + }); + it('round-trips an unbounded live coordinate with a null ceiling', async () => { const harness = await createHarness(); const coordinate = new MaterializationCoordinate({ frontier: new Map(), ceiling: null }); @@ -207,7 +231,7 @@ describe('GitCasMaterializationStoreAdapter', () => { it.each([ ['non-object descriptor', null, 'descriptor must be an object'], - ['schema', { schemaVersion: 2 }, 'schema is unsupported'], + ['schema', { schemaVersion: 3 }, 'schema is unsupported'], [ 'coordinate object', descriptor({ coordinate: null }), @@ -248,6 +272,32 @@ describe('GitCasMaterializationStoreAdapter', () => { descriptor({ coordinate: { ceiling: -1, frontier: [] } }), 'coordinate ceiling is invalid', ], + ['root status collection', descriptor({ roots: {} }), 'roots must be an array'], + [ + 'root status tuple', + descriptor({ roots: [['adjacency']] }), + 'invalid root status entry', + ], + [ + 'root status name', + descriptor({ roots: replaceRootStatusName(rootStatusFixture(), 'adjacency', 'unknown') }), + 'unknown root status name', + ], + [ + 'root status value', + descriptor({ roots: replaceRootStatus(rootStatusFixture(), 'adjacency', 'missing') }), + 'invalid adjacency root status', + ], + [ + 'duplicate root status', + descriptor({ roots: [...rootStatusFixture(), ['adjacency', 'retained']] }), + 'duplicate adjacency root status', + ], + [ + 'missing root status', + descriptor({ roots: rootStatusFixture().filter(([name]) => name !== 'adjacency') }), + 'no adjacency root status', + ], ['lane', descriptor({ laneName: '' }), 'laneName must be a non-empty string'], ['state hash', descriptor({ stateHash: '' }), 'stateHash must be a non-empty string'], ])('rejects an invalid %s', async (_case, value, message) => { @@ -279,6 +329,17 @@ describe('GitCasMaterializationStoreAdapter', () => { }); }); + it('rejects a bundle member whose descriptor marks that root empty', async () => { + const harness = await retainedHarness(); + replaceDescriptor(harness, descriptor({ + roots: replaceRootStatus(rootStatusFixture(), 'edge-alive', 'empty'), + })); + await expect(harness.adapter.findExact(harness.coordinate)).rejects.toMatchObject({ + code: 'E_MATERIALIZATION_STORAGE', + message: expect.stringContaining('unexpected edge-alive root bundle'), + }); + }); + it('enforces the descriptor page read bound', async () => { const harness = await retainedHarness(); const descriptorHandle = requireMember( @@ -455,14 +516,34 @@ async function createRoots(cas: InMemoryGitCasFacade): Promise = {}): object { return { - schemaVersion: 1, + schemaVersion: 2, laneName: 'events', stateHash: 'state-hash', + roots: rootStatusFixture(), coordinate: { ceiling: 12, frontier: [['writer-a', 'patch-a'], ['writer-b', 'patch-b']], @@ -489,6 +571,26 @@ function descriptor(overrides: Record = }; } +function rootStatusFixture(): string[][] { + return ROOT_PATHS.map((path) => [path.slice('roots/'.length), 'retained']); +} + +function replaceRootStatus( + roots: readonly string[][], + target: string, + status: string, +): string[][] { + return roots.map(([name, current]) => [name ?? '', name === target ? status : current ?? '']); +} + +function replaceRootStatusName( + roots: readonly string[][], + target: string, + replacement: string, +): string[][] { + return roots.map(([name, status]) => [name === target ? replacement : name ?? '', status ?? '']); +} + function replaceDescriptor(harness: RetainedHarness, value: object | null): void { const descriptorHandle = requireMember( harness.cas.readBundleMembers(harness.retainedBundle.toString()),