From ac34e6a6563b710d46317090536b3edb1158a389 Mon Sep 17 00:00:00 2001 From: James Ross Date: Fri, 24 Jul 2026 09:28:46 -0700 Subject: [PATCH 1/3] feat: retain bounded live liveness roots --- CHANGELOG.md | 8 + .../cas-first-memoized-materialization.md | 72 +++--- .../materialization/MaterializationHandle.ts | 17 +- src/domain/services/JoinReducerSession.ts | 32 +++ .../controllers/BoundedLiveMaterialization.ts | 179 +++++++++++++++ .../controllers/MaterializationRetention.ts | 21 +- .../controllers/MaterializeController.ts | 6 +- .../controllers/MaterializeLiveStrategy.ts | 31 +-- .../controllers/MaterializedLiveResolution.ts | 41 ++++ .../GitCasMaterializationDescriptor.ts | 34 ++- .../GitCasMaterializationReplayBasis.ts | 3 + .../GitCasMaterializationStoreAdapter.ts | 43 +--- .../GitCasMaterializationStoreValidation.ts | 15 +- .../GitCasMaterializationStoreWitness.ts | 9 - src/ports/MaterializationWorkspacePort.ts | 2 +- .../MaterializationIdentity.test.ts | 14 ++ .../services/JoinReducer.stateSession.test.ts | 30 +++ .../BoundedLiveMaterialization.test.ts | 206 ++++++++++++++++++ ...MaterializeController.liveNodeRead.test.ts | 12 + ...MaterializeController.stateSession.test.ts | 3 +- .../MaterializedLiveResolution.test.ts | 105 +++++++++ ...erializationStoreAdapter.lifecycle.test.ts | 118 ---------- .../GitCasMaterializationStoreAdapter.test.ts | 28 ++- ...tCasMaterializationStoreValidation.test.ts | 59 +++++ 24 files changed, 834 insertions(+), 254 deletions(-) create mode 100644 src/domain/services/controllers/BoundedLiveMaterialization.ts create mode 100644 src/domain/services/controllers/MaterializedLiveResolution.ts create mode 100644 test/unit/domain/services/controllers/BoundedLiveMaterialization.test.ts create mode 100644 test/unit/domain/services/controllers/MaterializedLiveResolution.test.ts create mode 100644 test/unit/infrastructure/adapters/GitCasMaterializationStoreValidation.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 153d10d0..632ead1d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -66,6 +66,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 operation-local instead of shared across materializations. This bounds mutated-page residency only; removal of whole-state projection remains tracked by #738. +- Changed cold live-handle resolution to stream only node and edge liveness + into bounded git-cas pages. Partial materialization handles carry + `stateHash: null`, cannot masquerade as whole-state snapshots, and avoid + constructing `WarpState`, adjacency, properties, receipts, diffs, + provenance, or a state-cache snapshot. ### Removed @@ -73,6 +78,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 compatibility barrel, and their compatibility-only examples and tests. - Removed the v18 graph-first package exports rather than carrying a deprecated second application API through v19. +- Removed migration and cleanup support for unreleased materialization + descriptor schemas v2 through v4. Schema v5 is the first release contract; + stale derived-cache entries miss and rebuild from authoritative WARP history. ### Fixed diff --git a/docs/topics/cas-first-memoized-materialization.md b/docs/topics/cas-first-memoized-materialization.md index 25383b5d..838c447f 100644 --- a/docs/topics/cas-first-memoized-materialization.md +++ b/docs/topics/cas-first-memoized-materialization.md @@ -98,14 +98,17 @@ 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. +separate state-cache hit. A complete descriptor 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. +Descriptor schema v5 also admits explicitly partial handles: `stateHash: null` +means the roots can answer only the reads whose root status is retained or +empty, and the handle cannot resume a whole-state snapshot. 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 complete 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 @@ -140,23 +143,23 @@ Both exact paths release their operation borrow without hydrating `WarpState`, building adjacency, publishing a state snapshot, or populating `_cachedState`. The runtime storage adapter keeps one git-cas acquisition for the current coordinate, retires it after in-flight readers finish when the coordinate -changes, and releases it from `RuntimeHost.close()`. A cold handle miss still -performs the current materialization path before retaining and reading the new -roots. Once assembled, a newly built property root joins the operation's -expiring git-cas workspace before state hashing and final promotion, so the -completed root remains reachable throughout promotion. A custom -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 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. +changes, and releases it from `RuntimeHost.close()`. On a cold handle miss, the +built-in session now streams only node and edge OR-Set operations into bounded +git-cas pages and retains those roots as a partial handle. It does not construct +`WarpState`, adjacency, property registers, receipts, diffs, provenance, or a +state-cache snapshot. A later compatibility or property operation can replace +that partial entry with a complete descriptor. + +Once assembled, a newly built property root joins the operation's expiring +git-cas workspace before state hashing and final promotion, so the completed +root remains reachable throughout promotion. A custom 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. + +Schema v5 is the first release contract for retained-materialization +descriptors. Earlier v2, v3, and v4 profiles were unreleased derived-cache +formats; v19 does not decode or migrate them. A current-key miss rebuilds from +authoritative WARP history. ## `git-cas` Encapsulation @@ -221,20 +224,23 @@ lost payload bytes, or run Git garbage collection. - RuntimeHost exact node-liveness and node-property reads consume the handle-first result when the built-in trie session and reader pair is active. - Custom session openers, neighborhoods, list reads, checkpoint creation, and - other compatibility operations still own process-resident whole state. + Cold node liveness now produces a partial retained handle through bounded + node/edge replay. Cold properties, custom session openers, neighborhoods, + list reads, checkpoint creation, and 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 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. +- Complete retained materialization descriptors carry node/edge trie roots, a + per-node property-shard root, and the full-state replay basis. Partial cold + handles carry node/edge roots and mark the remaining roots unavailable. + 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 property reads avoid that state. - One node's complete encoded property bag must currently fit within the 16 MiB shard limit. Property-key pagination or a property trie is not yet available. - The first property-root profile stores one bundle member per property-bearing diff --git a/src/domain/materialization/MaterializationHandle.ts b/src/domain/materialization/MaterializationHandle.ts index 22ba6ddf..dcdb9509 100644 --- a/src/domain/materialization/MaterializationHandle.ts +++ b/src/domain/materialization/MaterializationHandle.ts @@ -10,7 +10,14 @@ export default class MaterializationHandle { readonly bundle: BundleHandle; readonly coordinate: MaterializationCoordinate; readonly roots: MaterializationRoots; - readonly stateHash: string; + /** + * Semantic hash of a complete WarpState replay basis. + * + * Partial materializations intentionally carry `null`: their retained roots + * can answer bounded reads, but must never masquerade as a whole-state + * snapshot. + */ + readonly stateHash: string | null; readonly retention: StorageRetentionWitness; constructor(options: { @@ -18,7 +25,7 @@ export default class MaterializationHandle { readonly bundle: BundleHandle; readonly coordinate: MaterializationCoordinate; readonly roots: MaterializationRoots; - readonly stateHash: string; + readonly stateHash: string | null; readonly retention: StorageRetentionWitness; }) { requireOptions(options); @@ -30,7 +37,7 @@ export default class MaterializationHandle { 'coordinate', ); this.roots = requireInstance(options.roots, MaterializationRoots, 'roots'); - this.stateHash = requireNonEmpty(options.stateHash, 'stateHash'); + this.stateHash = requireOptionalStateHash(options.stateHash); this.retention = requireInstance( options.retention, StorageRetentionWitness, @@ -59,6 +66,10 @@ function requireNonEmpty(value: string, field: string): string { return value; } +function requireOptionalStateHash(value: string | null): string | null { + return value === null ? null : requireNonEmpty(value, 'stateHash'); +} + function requireOptions(options: object): void { if (options === null || typeof options !== 'object' || Array.isArray(options)) { throw handleError('options must be an object'); diff --git a/src/domain/services/JoinReducerSession.ts b/src/domain/services/JoinReducerSession.ts index 4b67f1d6..22844c61 100644 --- a/src/domain/services/JoinReducerSession.ts +++ b/src/domain/services/JoinReducerSession.ts @@ -131,6 +131,38 @@ export async function applyWithReceiptInSession( return result.receipt; } +/** + * Applies only the page-backed OR-Set portion of a patch. + * + * This is the bounded cold-read reducer: it deliberately does not allocate + * property registers, edge-birth metadata, receipts, diffs, or a frontier + * projection. + */ +export async function applyLivenessInSession( + session: StateSession, + patch: PatchLike, // nosemgrep: ts-no-like-types -- 0025C +): Promise { + for (const rawOp of patch.ops) { + const op = normalizeRawOp(rawOp); + if (!(op instanceof Op)) { + continue; + } + op.validate(); + if (op instanceof NodeAdd) { + await session.addNode(op.node, op.dot); + } else if (op instanceof NodeRemove) { + await session.removeNode(op.node, new Set(op.observedDots)); + } else if (op instanceof EdgeAdd) { + await session.addEdge(encodeEdgeKey(op.from, op.to, op.label), op.dot); + } else if (op instanceof EdgeRemove) { + await session.removeEdge( + encodeEdgeKey(op.from, op.to, op.label), + new Set(op.observedDots), + ); + } + } +} + export function reducePatchesInSession( patches: SessionPatchSource, frame: ReducerSessionFrame, diff --git a/src/domain/services/controllers/BoundedLiveMaterialization.ts b/src/domain/services/controllers/BoundedLiveMaterialization.ts new file mode 100644 index 00000000..1f41cf53 --- /dev/null +++ b/src/domain/services/controllers/BoundedLiveMaterialization.ts @@ -0,0 +1,179 @@ +import type MaterializationCoordinate from '../../materialization/MaterializationCoordinate.ts'; +import LiveMaterializationResolution from '../../materialization/LiveMaterializationResolution.ts'; +import type MaterializationHandle from '../../materialization/MaterializationHandle.ts'; +import MaterializationRoot from '../../materialization/MaterializationRoot.ts'; +import MaterializationRoots from '../../materialization/MaterializationRoots.ts'; +import BundleHandle from '../../storage/BundleHandle.ts'; +import WarpError from '../../errors/WarpError.ts'; +import type StateSession from '../../orset/session/StateSession.ts'; +import type MaterializationWorkspacePort from '../../../ports/MaterializationWorkspacePort.ts'; +import { applyLivenessInSession } from '../JoinReducerSession.ts'; +import type { MaterializeDeps } from './MaterializeDeps.ts'; +import type { MaterializeSessionOpener } from './MaterializeSessionBridge.ts'; +import { + releaseAcquisitionAfterFailure, + releaseWorkspaceAfterFailure, +} from './MaterializationWorkspaceCleanup.ts'; + +export type BoundedLiveMaterializationResult = Readonly<{ + materialization: Awaited< + ReturnType + >; + patchCount: number; +}>; + +export async function resolveBoundedLiveMaterialization(args: { + readonly deps: MaterializeDeps; + readonly coordinate: MaterializationCoordinate; +}): Promise { + const bounded = await retainBoundedLiveMaterialization(args); + if (bounded === null) { + return null; + } + const acquired = await args.deps.materializations.acquireExact(args.coordinate); + if (acquired === null) { + throw resolutionError('newly retained bounded handle could not be acquired'); + } + try { + requireSameHandle(bounded.materialization, acquired.materialization); + return new LiveMaterializationResolution({ + materialization: acquired.materialization, + source: 'materialized', + replayedPatchCount: bounded.patchCount, + release: async () => await acquired.release(), + }); + } catch (raw) { + await releaseAcquisitionAfterFailure(acquired, args.deps.logger); + throw raw; + } +} + +/** + * Replays only node/edge OR-Set state into bounded CAS pages and retains those + * roots as a partial materialization. No WarpState or whole-graph adjacency is + * constructed. + */ +export async function retainBoundedLiveMaterialization(args: { + readonly deps: MaterializeDeps; + readonly coordinate: MaterializationCoordinate; +}): Promise { + const { deps, coordinate } = args; + const { openStateSession } = deps; + if (openStateSession === undefined) { + return null; + } + const workspace = await deps.materializations.openWorkspace(coordinate); + try { + return await buildBoundedMaterialization({ + deps, + coordinate, + workspace, + openStateSession, + }); + } catch (raw) { + await releaseWorkspaceAfterFailure(workspace, deps.logger); + throw raw; + } +} + +async function buildBoundedMaterialization(args: { + readonly deps: MaterializeDeps; + readonly coordinate: MaterializationCoordinate; + readonly workspace: MaterializationWorkspacePort; + readonly openStateSession: MaterializeSessionOpener; +}): Promise { + const { deps, coordinate, workspace, openStateSession } = args; + const session = await openStateSession( + { nodeAliveRootOid: null, edgeAliveRootOid: null }, + { workspace }, + ); + const patchCount = await replayLiveness(session, deps, coordinate); + if (patchCount === 0) { + await session.close(); + await workspace.release(); + return null; + } + return await retainPreparedLiveness({ + session, + coordinate, + workspace, + patchCount, + }); +} + +async function replayLiveness( + session: StateSession, + deps: MaterializeDeps, + coordinate: MaterializationCoordinate, +): Promise { + let patchCount = 0; + for await (const entry of deps.patches.streamForFrontier( + coordinate.frontier(), + coordinate.ceiling, + )) { + patchCount += 1; + await applyLivenessInSession(session, entry.patch); + } + return patchCount; +} + +async function retainPreparedLiveness(args: { + readonly session: StateSession; + readonly coordinate: MaterializationCoordinate; + readonly workspace: MaterializationWorkspacePort; + readonly patchCount: number; +}): Promise { + const { session, coordinate, workspace, patchCount } = args; + const prepared = await session.prepareClose(); + const materialization = await workspace.promote({ + coordinate, + roots: livenessRoots(prepared.roots), + stateHash: null, + }); + prepared.accept(materialization.retention); + await workspace.release(); + return Object.freeze({ materialization, patchCount }); +} + +function livenessRoots(roots: { + readonly nodeAliveRootOid: string | null; + readonly edgeAliveRootOid: string | null; +}): MaterializationRoots { + return new MaterializationRoots({ + adjacency: MaterializationRoot.unavailable(), + edgeAlive: sessionRoot(roots.edgeAliveRootOid), + edgeBirths: MaterializationRoot.unavailable(), + frontier: MaterializationRoot.unavailable(), + nodeAlive: sessionRoot(roots.nodeAliveRootOid), + properties: MaterializationRoot.unavailable(), + provenanceSupport: MaterializationRoot.unavailable(), + replayBasis: MaterializationRoot.unavailable(), + roaringIndexes: MaterializationRoot.unavailable(), + }); +} + +function sessionRoot(token: string | null): MaterializationRoot { + return token === null + ? MaterializationRoot.empty() + : MaterializationRoot.retained(new BundleHandle(token)); +} + +function requireSameHandle( + expected: MaterializationHandle, + acquired: MaterializationHandle, +): void { + if ( + !acquired.coordinate.equals(expected.coordinate) + || acquired.stateHash !== expected.stateHash + || !acquired.bundle.equals(expected.bundle) + ) { + throw resolutionError('newly retained handle changed before it could be acquired'); + } +} + +function resolutionError(message: string): WarpError { + return new WarpError( + `Materialization resolution ${message}`, + 'E_MATERIALIZATION_RESUME', + ); +} diff --git a/src/domain/services/controllers/MaterializationRetention.ts b/src/domain/services/controllers/MaterializationRetention.ts index 136cacb4..910c4e20 100644 --- a/src/domain/services/controllers/MaterializationRetention.ts +++ b/src/domain/services/controllers/MaterializationRetention.ts @@ -27,12 +27,10 @@ function resolveExistingMaterialization( stateHash: string, ): MaterializationHandle | undefined { const retained = params.materialization; - if (retained === undefined) { + if (!isWholeStateMaterialization(retained)) { return undefined; } - if (retained.stateHash !== stateHash) { - throw retentionError('retained handle state hash does not match resumed state'); - } + requireMatchingStateHash(retained, stateHash); if (!rootsMatch(params, retained)) { return undefined; } @@ -40,6 +38,21 @@ function resolveExistingMaterialization( return retained; } +function isWholeStateMaterialization( + retained: MaterializationHandle | undefined, +): retained is MaterializationHandle & { readonly stateHash: string } { + return retained !== undefined && retained.stateHash !== null; +} + +function requireMatchingStateHash( + retained: MaterializationHandle & { readonly stateHash: string }, + stateHash: string, +): void { + if (retained.stateHash !== stateHash) { + throw retentionError('retained handle state hash does not match resumed state'); + } +} + function rootsMatch( params: MaterializeResultBuildInput, retained: MaterializationHandle, diff --git a/src/domain/services/controllers/MaterializeController.ts b/src/domain/services/controllers/MaterializeController.ts index c8e9675c..787c610f 100644 --- a/src/domain/services/controllers/MaterializeController.ts +++ b/src/domain/services/controllers/MaterializeController.ts @@ -285,9 +285,13 @@ export default class MaterializeController { } const coordinate = new MaterializationCoordinate(snapshot.coordinate); const acquisition = await this._deps.materializations.acquireExact(coordinate); + const retained = acquisition?.materialization ?? null; + if (retained?.stateHash === null) { + await acquisition?.release(); + return null; + } let result: MaterializeResult; try { - const retained = acquisition?.materialization ?? null; if (retained !== null && retained.stateHash !== snapshot.stateHash) { throw materializationResumeError('retained handle and snapshot state hashes differ'); } diff --git a/src/domain/services/controllers/MaterializeLiveStrategy.ts b/src/domain/services/controllers/MaterializeLiveStrategy.ts index 0ac1b8d7..3941f2c8 100644 --- a/src/domain/services/controllers/MaterializeLiveStrategy.ts +++ b/src/domain/services/controllers/MaterializeLiveStrategy.ts @@ -27,6 +27,8 @@ import { snapshotPublicationForReceipts } from './MaterializeSnapshotPublication import { releaseAcquisitionAfterFailure } from './MaterializationWorkspaceCleanup.ts'; import RetainedMaterializationResumeStrategy from './RetainedMaterializationResumeStrategy.ts'; import { isNonEmptyPatchSha } from './MaterializeHelpers.ts'; +import { resolveBoundedLiveMaterialization } from './BoundedLiveMaterialization.ts'; +import { materializedResolution } from './MaterializedLiveResolution.ts'; export default class MaterializeLiveStrategy { private readonly runtime: MaterializeStrategyRuntime; @@ -118,6 +120,13 @@ export default class MaterializeLiveStrategy { coordinate: WarpStateCoordinate, materializationCoordinate: MaterializationCoordinate, ): Promise { + const bounded = await resolveBoundedLiveMaterialization({ + deps: this.runtime.deps, + coordinate: materializationCoordinate, + }); + if (bounded !== null) { + return bounded; + } const result = await this.replayCurrentCoordinate(coordinate, { receipts: false, wantDiff: false, @@ -456,28 +465,6 @@ function retainedResolution( }); } -function materializedResolution( - result: MaterializeResult, - acquired: MaterializationAcquisition, -): LiveMaterializationResolution { - if (result.materialization === undefined) { - throw resolutionError('non-empty coordinate did not produce a retained handle'); - } - if ( - !acquired.materialization.coordinate.equals(result.materialization.coordinate) - || acquired.materialization.stateHash !== result.materialization.stateHash - || !acquired.materialization.bundle.equals(result.materialization.bundle) - ) { - throw resolutionError('newly retained handle changed before it could be acquired'); - } - return new LiveMaterializationResolution({ - materialization: acquired.materialization, - source: 'materialized', - replayedPatchCount: result.patchCount, - release: async () => await acquired.release(), - }); -} - function snapshotPublicationForLiveOptions( opts: MaterializeLiveOptions, ): ReturnType { diff --git a/src/domain/services/controllers/MaterializedLiveResolution.ts b/src/domain/services/controllers/MaterializedLiveResolution.ts new file mode 100644 index 00000000..5c8c1057 --- /dev/null +++ b/src/domain/services/controllers/MaterializedLiveResolution.ts @@ -0,0 +1,41 @@ +import LiveMaterializationResolution from '../../materialization/LiveMaterializationResolution.ts'; +import type MaterializationHandle from '../../materialization/MaterializationHandle.ts'; +import WarpError from '../../errors/WarpError.ts'; +import type { MaterializationAcquisition } from '../../../ports/MaterializationStorePort.ts'; +import type { MaterializeResult } from './MaterializeController.ts'; + +export function materializedResolution( + result: MaterializeResult, + acquired: MaterializationAcquisition, +): LiveMaterializationResolution { + if (result.materialization === undefined) { + throw resolutionError('non-empty coordinate did not produce a retained handle'); + } + requireSameHandle(result.materialization, acquired.materialization); + return new LiveMaterializationResolution({ + materialization: acquired.materialization, + source: 'materialized', + replayedPatchCount: result.patchCount, + release: async () => await acquired.release(), + }); +} + +function requireSameHandle( + expected: MaterializationHandle, + acquired: MaterializationHandle, +): void { + if ( + !acquired.coordinate.equals(expected.coordinate) + || acquired.stateHash !== expected.stateHash + || !acquired.bundle.equals(expected.bundle) + ) { + throw resolutionError('newly retained handle changed before it could be acquired'); + } +} + +function resolutionError(message: string): WarpError { + return new WarpError( + `Materialization resolution ${message}`, + 'E_MATERIALIZATION_RESUME', + ); +} diff --git a/src/infrastructure/adapters/GitCasMaterializationDescriptor.ts b/src/infrastructure/adapters/GitCasMaterializationDescriptor.ts index c372638b..5e9aa537 100644 --- a/src/infrastructure/adapters/GitCasMaterializationDescriptor.ts +++ b/src/infrastructure/adapters/GitCasMaterializationDescriptor.ts @@ -9,18 +9,18 @@ import MaterializationRoots, { import type BundleHandle from '../../domain/storage/BundleHandle.ts'; import WarpError from '../../domain/errors/WarpError.ts'; -export const MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION = 4; +export const MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION = 5; export type DecodedMaterializationDescriptor = Readonly<{ coordinate: MaterializationCoordinate; - stateHash: string; + stateHash: string | null; laneName: string; rootStatuses: ReadonlyMap; }>; export function materializationDescriptorData(input: { coordinate: MaterializationCoordinate; - stateHash: string; + stateHash: string | null; laneName: string; roots: MaterializationRoots; }): object { @@ -51,10 +51,13 @@ export function decodeMaterializationDescriptor( } const coordinateValue = value['coordinate']; requireRecord(coordinateValue, 'descriptor.coordinate'); + const stateHash = requireOptionalStateHash(value['stateHash']); + const rootStatuses = decodeRootStatuses(value['roots']); + requireReplayBasisHash(stateHash, rootStatuses); return Object.freeze({ laneName: requireNonEmpty(value['laneName'], 'descriptor.laneName'), - stateHash: requireNonEmpty(value['stateHash'], 'descriptor.stateHash'), - rootStatuses: decodeRootStatuses(value['roots']), + stateHash, + rootStatuses, coordinate: new MaterializationCoordinate({ frontier: decodeFrontier(coordinateValue['frontier']), ceiling: requireCeiling(coordinateValue['ceiling']), @@ -122,15 +125,24 @@ function decodeRootStatuses( for (const name of MATERIALIZATION_ROOT_NAMES) { requireRootStatus(statuses, name); } - requireCurrentPropertyRoot(statuses); + requireAvailableRoot(statuses); return statuses; } -function requireCurrentPropertyRoot( +function requireAvailableRoot( statuses: ReadonlyMap, ): void { - if (statuses.get('properties') === 'unavailable') { - throw descriptorError('current materialization descriptor requires a property root'); + if ([...statuses.values()].every((status) => status === 'unavailable')) { + throw descriptorError('descriptor must provide at least one materialization root'); + } +} + +function requireReplayBasisHash( + stateHash: string | null, + statuses: ReadonlyMap, +): void { + if (stateHash === null && statuses.get('replay-basis') === 'retained') { + throw descriptorError('partial materialization cannot retain a replay basis'); } } @@ -223,6 +235,10 @@ function requireNonEmpty(value: unknown, field: string): string { return value; } +function requireOptionalStateHash(value: unknown): string | null { + return value === null ? null : requireNonEmpty(value, 'descriptor.stateHash'); +} + function arrayValue(values: readonly unknown[], index: number): unknown { return values[index]; } diff --git a/src/infrastructure/adapters/GitCasMaterializationReplayBasis.ts b/src/infrastructure/adapters/GitCasMaterializationReplayBasis.ts index e77818c0..2d1e8b6b 100644 --- a/src/infrastructure/adapters/GitCasMaterializationReplayBasis.ts +++ b/src/infrastructure/adapters/GitCasMaterializationReplayBasis.ts @@ -71,6 +71,9 @@ export default class GitCasMaterializationReplayBasis { if (root.status !== 'retained' || root.handle === null) { return null; } + if (materialization.stateHash === null) { + throw storageError('partial materialization cannot contain a replay basis'); + } const member = await this.#cas.bundles.getMemberReference({ handle: root.handle.toString(), path: REPLAY_BASIS_PATH, diff --git a/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts b/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts index 235b780d..a99cd27f 100644 --- a/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts +++ b/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts @@ -36,14 +36,10 @@ import GitCasMaterializationReplayBasis, { replaceReplayBasisRoot, } from './GitCasMaterializationReplayBasis.ts'; import GitCasMaterializationCacheKey from './GitCasMaterializationCacheKey.ts'; -import type { - GitCasMaterializationFacade, - MaterializationCacheSet, -} from './GitCasMaterializationStoreTypes.ts'; +import type { GitCasMaterializationFacade } from './GitCasMaterializationStoreTypes.ts'; import { releaseCacheAcquisitionAfterFailure, requireDescriptorSize, - requireExpectedAcquisition, requireStoredMaterialization, requireWorkspaceStage, } from './GitCasMaterializationStoreWitness.ts'; @@ -64,7 +60,6 @@ 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_VERSIONS = Object.freeze([2, 3]); export type { GitCasMaterializationFacade } from './GitCasMaterializationStoreTypes.ts'; @@ -170,7 +165,7 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt request: RetainMaterializationRequest, ): Promise { requireRetainRequest(request); - const stateHash = requireNonEmpty(request.stateHash, 'stateHash'); + const { stateHash } = request; const roots = request.replayBasis === undefined ? request.roots : replaceReplayBasisRoot( @@ -197,7 +192,7 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt async #stageWorkspaceBundle( workspace: GitCasStagingWorkspace, request: RetainMaterializationRequest, - stateHash: string, + stateHash: string | null, ): Promise { const descriptorBytes = this.#codec.encode(materializationDescriptorData({ coordinate: request.coordinate, @@ -234,32 +229,9 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt options: { retention: 'evictable' }, }); const retention = requireStoredMaterialization(promoted.destination, expectedHandle); - await this.#cleanLegacyAfterPromotion({ cache, cacheKey, expectedHandle, coordinate }); return adaptGitCasRetentionWitness(retention.toJSON()); } - async #cleanLegacyAfterPromotion(args: { - cache: MaterializationCacheSet; - cacheKey: string; - expectedHandle: string; - coordinate: MaterializationCoordinate; - }): Promise { - const acquisition = await args.cache.acquire(args.cacheKey); - if (acquisition === null) { - throw storageError('git-cas lost the retained materialization before legacy cleanup'); - } - await completeWithCleanup( - async () => { - requireExpectedAcquisition(acquisition, args.expectedHandle); - await this.#removeLegacyEntries(args.cache, args.coordinate); - }, - async () => { - await acquisition.release(); - }, - 'Legacy materialization cleanup and acquisition release both failed', - ); - } - override async acquireExact( coordinate: MaterializationCoordinate, ): Promise { @@ -457,15 +429,6 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt }); } - async #removeLegacyEntries( - cache: MaterializationCacheSet, - coordinate: MaterializationCoordinate, - ): Promise { - for (const schemaVersion of LEGACY_MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSIONS) { - await cache.remove(await this.#cacheKeys.forCoordinate(coordinate, schemaVersion)); - } - } - async #readDescriptor(handle: PageHandle): Promise { const bytes = await this.#cas.pages.get({ handle, diff --git a/src/infrastructure/adapters/GitCasMaterializationStoreValidation.ts b/src/infrastructure/adapters/GitCasMaterializationStoreValidation.ts index acbd39dc..0717aba1 100644 --- a/src/infrastructure/adapters/GitCasMaterializationStoreValidation.ts +++ b/src/infrastructure/adapters/GitCasMaterializationStoreValidation.ts @@ -9,14 +9,25 @@ export function requireRetainRequest(request: RetainMaterializationRequest): voi } requireCoordinate(request.coordinate); requireRetainRoots(request.roots); + requireRetainStateHash(request); } function requireRetainRoots(roots: MaterializationRoots): void { if (!(roots instanceof MaterializationRoots)) { throw storageError('retain request roots have an invalid runtime identity'); } - if (roots.properties.status === 'unavailable') { - throw storageError('current materialization profile requires a property root'); + if (roots.entries().every(([, root]) => root.status === 'unavailable')) { + throw storageError('retain request must provide at least one materialization root'); + } +} + +function requireRetainStateHash(request: RetainMaterializationRequest): void { + if (request.stateHash !== null) { + requireNonEmpty(request.stateHash, 'stateHash'); + return; + } + if (request.replayBasis !== undefined || request.roots.replayBasis.status === 'retained') { + throw storageError('partial materialization cannot retain a whole-state replay basis'); } } diff --git a/src/infrastructure/adapters/GitCasMaterializationStoreWitness.ts b/src/infrastructure/adapters/GitCasMaterializationStoreWitness.ts index 93a99093..2bb25eb1 100644 --- a/src/infrastructure/adapters/GitCasMaterializationStoreWitness.ts +++ b/src/infrastructure/adapters/GitCasMaterializationStoreWitness.ts @@ -47,15 +47,6 @@ export function requireStoredMaterialization( 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/ports/MaterializationWorkspacePort.ts b/src/ports/MaterializationWorkspacePort.ts index 08297da6..800a0f67 100644 --- a/src/ports/MaterializationWorkspacePort.ts +++ b/src/ports/MaterializationWorkspacePort.ts @@ -14,7 +14,7 @@ export type MaterializationWorkspaceRoots = Readonly<{ export type PromoteMaterializationRequest = Readonly<{ coordinate: MaterializationCoordinate; roots: MaterializationRoots; - stateHash: string; + stateHash: string | null; replayBasis?: WarpState; }>; diff --git a/test/unit/domain/materialization/MaterializationIdentity.test.ts b/test/unit/domain/materialization/MaterializationIdentity.test.ts index 6d78372d..6be14584 100644 --- a/test/unit/domain/materialization/MaterializationIdentity.test.ts +++ b/test/unit/domain/materialization/MaterializationIdentity.test.ts @@ -199,6 +199,20 @@ describe('MaterializationHandle', () => { expect(Object.isFrozen(handle)).toBe(true); }); + it('marks a partial materialization by omitting a whole-state hash', () => { + const bundle = bundleHandle('partial-materialization'); + const handle = new MaterializationHandle({ + laneName: 'events', + bundle, + coordinate: exactCoordinate(), + roots: materializationRoots(), + stateHash: null, + retention: retentionWitness(bundle), + }); + + expect(handle.stateHash).toBeNull(); + }); + it.each([ ['laneName', ''], ['bundle', new StorageHandle('not-a-bundle')], diff --git a/test/unit/domain/services/JoinReducer.stateSession.test.ts b/test/unit/domain/services/JoinReducer.stateSession.test.ts index 24bcf88e..0b03a655 100644 --- a/test/unit/domain/services/JoinReducer.stateSession.test.ts +++ b/test/unit/domain/services/JoinReducer.stateSession.test.ts @@ -118,6 +118,36 @@ function expectReduceV5State(result: ReturnType) { describe("JoinReducer session-backed path", () => { describe("golden path", () => { + it("replays only page-backed liveness for bounded materialization", async () => { + const { applyLivenessInSession } = await loadJoinReducerSessionModule(); + const { session } = await openSession(); + const removedNodeDot = new Dot("alice", 1); + const liveNodeDot = new Dot("alice", 2); + const removedEdgeDot = new Dot("alice", 3); + + await applyLivenessInSession(session, makePatch("alice", 1, [ + nodeAdd("node:removed", removedNodeDot), + nodeAdd("node:live", liveNodeDot), + edgeAdd("node:removed", "node:live", "knows", removedEdgeDot), + propSet("node:live", "ignored", "no register allocation"), + ])); + await applyLivenessInSession(session, makePatch("alice", 2, [ + nodeRemove("node:removed", [Dot.encode(removedNodeDot)]), + edgeRemove( + "node:removed", + "node:live", + "knows", + [Dot.encode(removedEdgeDot)], + ), + ])); + + await expect(session.nodeContains("node:removed")).resolves.toBe(false); + await expect(session.nodeContains("node:live")).resolves.toBe(true); + await expect(session.edgeContains("node:removed\0node:live\0knows")) + .resolves.toBe(false); + await session.close(); + }); + it("replays mixed patches through one session-backed reducer frame", async () => { const { ReducerSessionFrame, reducePatchesInSession } = await loadJoinReducerSessionModule(); diff --git a/test/unit/domain/services/controllers/BoundedLiveMaterialization.test.ts b/test/unit/domain/services/controllers/BoundedLiveMaterialization.test.ts new file mode 100644 index 00000000..fb8335cb --- /dev/null +++ b/test/unit/domain/services/controllers/BoundedLiveMaterialization.test.ts @@ -0,0 +1,206 @@ +import { describe, expect, it, vi } from 'vitest'; + +import PatchCollector, { + type CheckpointData, + type PatchWithSha, +} from '../../../../../src/domain/capabilities/PatchCollector.ts'; +import { Dot } from '../../../../../src/domain/crdt/Dot.ts'; +import MaterializationCoordinate from '../../../../../src/domain/materialization/MaterializationCoordinate.ts'; +import StateSession from '../../../../../src/domain/orset/session/StateSession.ts'; +import PageCache from '../../../../../src/domain/orset/trie/PageCache.ts'; +import TrieGeometry from '../../../../../src/domain/orset/trie/TrieGeometry.ts'; +import { + resolveBoundedLiveMaterialization, + retainBoundedLiveMaterialization, +} from '../../../../../src/domain/services/controllers/BoundedLiveMaterialization.ts'; +import type { + MaterializeDeps, +} from '../../../../../src/domain/services/controllers/MaterializeController.ts'; +import MaterializeController from '../../../../../src/domain/services/controllers/MaterializeController.ts'; +import { createEmptyState } from '../../../../../src/domain/services/JoinReducer.ts'; +import Patch from '../../../../../src/domain/types/Patch.ts'; +import NodeAdd from '../../../../../src/domain/types/ops/NodeAdd.ts'; +import cborCodec from '../../../../../src/infrastructure/codecs/CborCodec.ts'; +import InMemoryCheckpointStore from '../../../../helpers/InMemoryCheckpointStore.ts'; +import InMemoryMaterializationStore from '../../../../helpers/InMemoryMaterializationStore.ts'; +import { InMemoryTrieStore } from '../../../../helpers/trieHelpers.ts'; +import type MaterializationWorkspacePort from '../../../../../src/ports/MaterializationWorkspacePort.ts'; +import WarpStateCachePort, { + type WarpStateCoordinate, + type WarpStateSnapshotRecord, +} from '../../../../../src/ports/WarpStateCachePort.ts'; + +const coordinate = new MaterializationCoordinate({ + frontier: new Map([['writer', 'tip']]), + ceiling: null, +}); + +describe('BoundedLiveMaterialization', () => { + it('fails closed when a newly retained handle cannot be acquired', async () => { + const materializations = new InMemoryMaterializationStore(); + vi.spyOn(materializations, 'acquireExact').mockResolvedValue(null); + + await expect(resolveBoundedLiveMaterialization({ + deps: createDeps(materializations), + coordinate, + })).rejects.toMatchObject({ + code: 'E_MATERIALIZATION_RESUME', + message: expect.stringContaining('could not be acquired'), + }); + }); + + it('releases the staging workspace when session opening fails', async () => { + const materializations = new InMemoryMaterializationStore(); + const failure = new Error('session open failed'); + + await expect(retainBoundedLiveMaterialization({ + deps: createDeps( + materializations, + vi.fn().mockRejectedValue(failure), + ), + coordinate, + })).rejects.toBe(failure); + + expect(materializations.workspaces[0]?.released).toBe(true); + }); + + it('releases a partial acquisition before using an exact whole-state snapshot', async () => { + const materializations = new InMemoryMaterializationStore(); + const deps = createDeps(materializations); + const bounded = await resolveBoundedLiveMaterialization({ deps, coordinate }); + await bounded?.release(); + deps.getStateCache = () => new ExactSnapshotCache(); + + const result = await new MaterializeController(deps).materialize(); + + expect(result.state.nodeAlive.contains('snapshot-node')).toBe(true); + expect(materializations.acquisitions.at(-1)?.released).toBe(true); + }); +}); + +function createDeps( + materializations: InMemoryMaterializationStore, + openStateSession = defaultSessionOpener(), +): MaterializeDeps { + return { + logger: { + warn: vi.fn(), + info: vi.fn(), + debug: vi.fn(), + error: vi.fn(), + child: vi.fn(), + }, + codec: cborCodec, + crypto: { + hash: vi.fn().mockResolvedValue('hash'), + hmac: vi.fn().mockResolvedValue(new Uint8Array([1])), + timingSafeEqual: vi.fn().mockReturnValue(false), + }, + persistence: { readRef: vi.fn().mockResolvedValue(null) }, + checkpointStore: new InMemoryCheckpointStore(), + materializations, + patches: new SinglePatchCollector(), + graphCloner: { openReadOnly: vi.fn() }, + graphName: 'test-graph', + openStateSession, + }; +} + +function defaultSessionOpener() { + const store = new InMemoryTrieStore(); + const pageCache = new PageCache({ maxResident: 8 }); + return async ( + roots: { + readonly nodeAliveRootOid: string | null; + readonly edgeAliveRootOid: string | null; + }, + options: { readonly workspace: MaterializationWorkspacePort }, + ): Promise => await StateSession.open({ + ...roots, + store, + codec: cborCodec, + geometry: TrieGeometry.default16way(), + pageCache, + maxDirtyPages: 1, + workspace: options.workspace, + }); +} + +class SinglePatchCollector extends PatchCollector { + readonly #entry: PatchWithSha = { + patch: new Patch({ + writer: 'writer', + lamport: 1, + context: {}, + ops: [new NodeAdd('node', Dot.create('writer', 1))], + reads: [], + writes: ['node'], + }), + sha: 'tip', + }; + + override discoverWriters(): Promise { + return Promise.resolve(['writer']); + } + + override loadWriterPatches(_writerId: string): Promise { + return Promise.resolve([this.#entry]); + } + + override loadCheckpoint(): Promise { + return Promise.resolve(null); + } + + override loadPatchesSince(_checkpoint: CheckpointData): Promise { + return Promise.resolve([]); + } + + override loadPatchChain(_toSha: string, _fromSha?: string | null): Promise { + return Promise.resolve([this.#entry]); + } + + override getFrontier(): Promise> { + return Promise.resolve(coordinate.frontier()); + } +} + +class ExactSnapshotCache extends WarpStateCachePort { + override getExact(_coordinate: WarpStateCoordinate): Promise { + const state = createEmptyState(); + state.nodeAlive.add('snapshot-node', Dot.create('snapshot', 1)); + return Promise.resolve({ + snapshotId: 'snapshot', + coordinate: { frontier: coordinate.frontier(), ceiling: null }, + retention: 'evictable', + provenancePosture: 'degraded', + stateHash: 'snapshot-hash', + payloadRef: 'snapshot-payload', + createdAt: '1970-01-01T00:00:00.000Z', + state, + }); + } + + override getBestCompatiblePredecessor(): Promise { + return Promise.resolve(null); + } + + override put(snapshot: WarpStateSnapshotRecord): Promise { + return Promise.resolve(snapshot); + } + + override pin(_snapshotId: string): Promise { + throw new Error('not used'); + } + + override publishCheckpointHead(): Promise { + return Promise.resolve(); + } + + override resolveCheckpointHead(): Promise { + return Promise.resolve(null); + } + + override pruneEvictable(): Promise { + return Promise.resolve(); + } +} diff --git a/test/unit/domain/services/controllers/MaterializeController.liveNodeRead.test.ts b/test/unit/domain/services/controllers/MaterializeController.liveNodeRead.test.ts index 1e78298c..ec818d1e 100644 --- a/test/unit/domain/services/controllers/MaterializeController.liveNodeRead.test.ts +++ b/test/unit/domain/services/controllers/MaterializeController.liveNodeRead.test.ts @@ -12,6 +12,7 @@ import MaterializationRoots from '../../../../../src/domain/materialization/Mate import MaterializeController, { type MaterializeDeps, } from '../../../../../src/domain/services/controllers/MaterializeController.ts'; +import { retainBoundedLiveMaterialization } from '../../../../../src/domain/services/controllers/BoundedLiveMaterialization.ts'; import BundleHandle from '../../../../../src/domain/storage/BundleHandle.ts'; import cborCodec from '../../../../../src/infrastructure/codecs/CborCodec.ts'; import InMemoryCheckpointStore from '../../../../helpers/InMemoryCheckpointStore.ts'; @@ -143,6 +144,17 @@ describe('MaterializeController live node reads', () => { expect(fixture.materializations.exactLookups).toHaveLength(0); }); + it('declines bounded cold replay when no state session is configured', async () => { + const fixture = await createFixture({ retain: false }); + + await expect(retainBoundedLiveMaterialization({ + deps: fixture.deps, + coordinate: new MaterializationCoordinate({ frontier: FRONTIER, ceiling: null }), + })).resolves.toBeNull(); + + expect(fixture.materializations.workspaces).toHaveLength(0); + }); + it('falls back and releases when the retained node root is unavailable', async () => { const fixture = await createFixture({ rootStatus: 'unavailable' }); diff --git a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts index 0725cdac..c3ff6ce8 100644 --- a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts +++ b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts @@ -653,12 +653,13 @@ describe("MaterializeController — state session integration", () => { expect(resolution.source).toBe("materialized"); expect(resolution.replayedPatchCount).toBe(1); - expect(resolution.materialization).not.toBeNull(); + expect(resolution.materialization?.stateHash).toBeNull(); expect(fixtures.materializations.retainedRequests).toHaveLength(1); expect(fixtures.patches.collectForFrontier).toHaveBeenCalledOnce(); expect(fixtures.stateCache.getExact).not.toHaveBeenCalled(); expect(fixtures.stateCache.getBestCompatiblePredecessor).not.toHaveBeenCalled(); expect(fixtures.stateCache.put).not.toHaveBeenCalled(); + expect(fixtures.deps.crypto.hash).not.toHaveBeenCalled(); expect(fixtures.materializations.acquisitions).toHaveLength(1); expect(fixtures.materializations.acquisitions[0]?.released).toBe(false); await resolution.release(); diff --git a/test/unit/domain/services/controllers/MaterializedLiveResolution.test.ts b/test/unit/domain/services/controllers/MaterializedLiveResolution.test.ts new file mode 100644 index 00000000..f3933536 --- /dev/null +++ b/test/unit/domain/services/controllers/MaterializedLiveResolution.test.ts @@ -0,0 +1,105 @@ +import { describe, expect, it, vi } 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 from '../../../../../src/domain/materialization/MaterializationRoots.ts'; +import { materializedResolution } from '../../../../../src/domain/services/controllers/MaterializedLiveResolution.ts'; +import type { MaterializeResult } from '../../../../../src/domain/services/controllers/MaterializeController.ts'; +import BundleHandle from '../../../../../src/domain/storage/BundleHandle.ts'; +import StorageRetentionWitness, { + StorageRetentionRoot, +} from '../../../../../src/domain/storage/StorageRetentionWitness.ts'; +import type { MaterializationAcquisition } from '../../../../../src/ports/MaterializationStorePort.ts'; + +describe('materializedResolution', () => { + it('returns and releases an acquisition that matches the published result', async () => { + const handle = materializationHandle('same'); + const release = vi.fn().mockResolvedValue(undefined); + const resolution = materializedResolution( + materializeResult(handle), + acquisition(handle, release), + ); + + expect(resolution.materialization).toBe(handle); + expect(resolution.source).toBe('materialized'); + expect(resolution.replayedPatchCount).toBe(3); + await resolution.release(); + expect(release).toHaveBeenCalledOnce(); + }); + + it('rejects a non-empty result without a retained handle', () => { + expect(() => materializedResolution( + materializeResult(undefined), + acquisition(materializationHandle('acquired'), vi.fn()), + )).toThrowError(/did not produce a retained handle/u); + }); + + it('rejects an acquisition whose retained identity changed', () => { + expect(() => materializedResolution( + materializeResult(materializationHandle('expected')), + acquisition(materializationHandle('changed'), vi.fn()), + )).toThrowError(/changed before it could be acquired/u); + }); +}); + +function materializeResult( + materialization: MaterializationHandle | undefined, +): MaterializeResult { + return { + materialization, + patchCount: 3, + } as MaterializeResult; +} + +function acquisition( + materialization: MaterializationHandle, + release: () => Promise, +): MaterializationAcquisition { + return Object.freeze({ + materialization, + acquiredAt: '1970-01-01T00:00:00.000Z', + release, + }); +} + +function materializationHandle(suffix: string): MaterializationHandle { + const bundle = new BundleHandle(`materialization:${suffix}`); + return new MaterializationHandle({ + laneName: 'events', + bundle, + coordinate: new MaterializationCoordinate({ + frontier: new Map([['writer', 'patch']]), + ceiling: null, + }), + roots: partialRoots(), + stateHash: null, + retention: new StorageRetentionWitness({ + handle: bundle, + policy: 'evictable', + reachability: 'anchored', + root: new StorageRetentionRoot({ + kind: 'cache-set', + namespace: 'git-warp/materializations', + locator: 'refs/cas/caches/git-warp/materializations', + generation: `generation:${suffix}`, + path: 'root', + }), + observedAt: '1970-01-01T00:00:00.000Z', + }), + }); +} + +function partialRoots(): MaterializationRoots { + return new MaterializationRoots({ + adjacency: MaterializationRoot.unavailable(), + edgeAlive: MaterializationRoot.empty(), + edgeBirths: MaterializationRoot.unavailable(), + frontier: MaterializationRoot.unavailable(), + nodeAlive: MaterializationRoot.empty(), + properties: MaterializationRoot.unavailable(), + provenanceSupport: MaterializationRoot.unavailable(), + replayBasis: MaterializationRoot.unavailable(), + roaringIndexes: MaterializationRoot.unavailable(), + }); +} diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.lifecycle.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.lifecycle.test.ts index fd6a4691..9c2203df 100644 --- a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.lifecycle.test.ts +++ b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.lifecycle.test.ts @@ -9,15 +9,12 @@ import GitCasMaterializationStoreAdapter, { import type { GitCasStagingWorkspace, } from '../../../../src/infrastructure/adapters/GitCasMaterializationWorkspace.ts'; -import { materializationCoordinateData } from '../../../../src/infrastructure/adapters/GitCasMaterializationDescriptor.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'; -const CACHE_NAMESPACE = 'git-warp/materializations'; -const CURRENT_CACHE_KEY_PATTERN = /^v4:[0-9a-f]{64}:[0-9a-f]{64}$/u; const ROOT_COUNT = 9; describe('GitCasMaterializationStoreAdapter lifecycle', () => { @@ -131,79 +128,6 @@ describe('GitCasMaterializationStoreAdapter lifecycle', () => { }); }); - 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 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.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, v3Key]); - - await harness.adapter.retain({ coordinate, roots, stateHash: 'replacement-state-hash' }); - expect(requireSingleCacheKey(harness.cas)).toMatch(CURRENT_CACHE_KEY_PATTERN); - }); - - 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 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.put(v3Key, retained.bundle.toString()); - await cache.remove(v4Key); - - const replacement = await harness.adapter.retain({ - coordinate, - roots, - stateHash: 'replacement-state-hash', - }); - const acquisition = await harness.adapter.acquireExact(coordinate); - - expect(harness.cas.readCacheKeys(CACHE_NAMESPACE)).toEqual([ - expect.stringMatching(CURRENT_CACHE_KEY_PATTERN), - ]); - expect(replacement.retention.root.generation) - .toBe(acquisition?.materialization.retention.root.generation); - await acquisition?.release(); - }); - - 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 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.put(v3Key, original.bundle.toString()); - await cache.remove(v4Key); - - await expect(adapterFor(withoutCacheAcquisition(harness.cas)).retain({ - coordinate, - roots, - stateHash: 'replacement-state-hash', - })).rejects.toMatchObject({ - code: 'E_MATERIALIZATION_STORAGE', - message: expect.stringContaining('before legacy cleanup'), - }); - - expect(harness.cas.readCacheKeys(CACHE_NAMESPACE)).toContain(v2Key); - expect(harness.cas.readCacheKeys(CACHE_NAMESPACE)).toContain(v3Key); - }); }); async function createHarness(): Promise { - const cache = await cas.caches.open(options); - 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), - }; - }, - }, - workspaces: cas.workspaces, - }; -} - async function createRoots(cas: InMemoryGitCasFacade): Promise { const handles: BundleHandle[] = []; for (let index = 0; index < ROOT_COUNT; index += 1) { @@ -287,24 +190,3 @@ function exactCoordinate(): MaterializationCoordinate { ceiling: 12, }); } - -async function cacheKeyForSchema( - coordinate: MaterializationCoordinate, - schemaVersion: number, -): Promise { - const encoded = defaultCodec.encode({ - schemaVersion, - laneName: 'events', - coordinate: materializationCoordinateData(coordinate), - }); - return `v${String(schemaVersion)}:${await new NodeCryptoAdapter().hash('sha256', encoded)}`; -} - -function requireSingleCacheKey(cas: InMemoryGitCasFacade): string { - const keys = cas.readCacheKeys(CACHE_NAMESPACE); - const key = keys[0]; - if (keys.length !== 1 || key === undefined) { - throw new Error('Expected exactly one materialization cache key'); - } - return key; -} diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts index dbb3bb25..a6ed5706 100644 --- a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts +++ b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts @@ -71,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(/^v4:[0-9a-f]{64}:[0-9a-f]{64}$/u); + expect(cacheKeys[0]).toMatch(/^v5:[0-9a-f]{64}:[0-9a-f]{64}$/u); expect(cacheKeys[0]?.length).toBeLessThan(1024); expect(harness.cas.readActiveCacheAcquisitionCount()).toBe(1); await acquisition?.release(); @@ -176,13 +176,14 @@ describe('GitCasMaterializationStoreAdapter', () => { const retained = await harness.adapter.retain({ coordinate: exactCoordinate(), roots, - stateHash: 'partial-state-hash', + stateHash: null, }); const resolved = await acquireReleaseAndClose(harness.adapter, 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?.stateHash).toBeNull(); expect(resolved?.roots.nodeAlive.handle?.toString()).toBe(nodeBundle.handle.toString()); expect(resolved?.roots.edgeAlive.status).toBe('empty'); expect(resolved?.roots.properties.status).toBe('empty'); @@ -376,6 +377,11 @@ describe('GitCasMaterializationStoreAdapter', () => { descriptor({ roots: replaceRootStatusName(rootStatusFixture(), 'adjacency', 'unknown') }), 'unknown root status name', ], + [ + 'non-string root status name', + descriptor({ roots: [[1, 'retained']] }), + 'unknown root status name', + ], [ 'root status value', descriptor({ roots: replaceRootStatus(rootStatusFixture(), 'adjacency', 'missing') }), @@ -392,12 +398,13 @@ describe('GitCasMaterializationStoreAdapter', () => { 'no adjacency root status', ], [ - 'unavailable current property root', - descriptor({ roots: replaceRootStatus(rootStatusFixture(), 'properties', 'unavailable') }), - 'requires a property root', + 'entirely unavailable root profile', + descriptor({ roots: rootStatusFixture().map(([name]) => [name ?? '', 'unavailable']) }), + 'at least one materialization root', ], ['lane', descriptor({ laneName: '' }), 'laneName must be a non-empty string'], ['state hash', descriptor({ stateHash: '' }), 'stateHash must be a non-empty string'], + ['partial replay basis', descriptor({ stateHash: null }), 'cannot retain a replay basis'], ])('rejects an invalid %s', async (_case, value, message) => { const harness = await retainedHarness(); replaceDescriptor(harness, value); @@ -488,14 +495,13 @@ describe('GitCasMaterializationStoreAdapter', () => { if (nodeAlive === null) { throw new Error('Expected retained node-alive test root'); } - await expect(harness.adapter.retain({ + const partial = await harness.adapter.retain({ coordinate: exactCoordinate(), roots: unavailablePropertyRoots(nodeAlive), - stateHash: 'state-hash', - })).rejects.toMatchObject({ - code: 'E_MATERIALIZATION_STORAGE', - message: expect.stringContaining('requires a property root'), + stateHash: null, }); + expect(partial.stateHash).toBeNull(); + expect(partial.roots.properties.status).toBe('unavailable'); await expect(Reflect.apply(harness.adapter.acquireExact, harness.adapter, [{ frontier: new Map(), ceiling: null, @@ -709,7 +715,7 @@ function exactCoordinate(): MaterializationCoordinate { function descriptor(overrides: Record = {}): object { return { - schemaVersion: 4, + schemaVersion: 5, laneName: 'events', stateHash: 'state-hash', roots: rootStatusFixture(), diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationStoreValidation.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationStoreValidation.test.ts new file mode 100644 index 00000000..4e55b69c --- /dev/null +++ b/test/unit/infrastructure/adapters/GitCasMaterializationStoreValidation.test.ts @@ -0,0 +1,59 @@ +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 BundleHandle from '../../../../src/domain/storage/BundleHandle.ts'; +import { requireRetainRequest } from '../../../../src/infrastructure/adapters/GitCasMaterializationStoreValidation.ts'; + +const coordinate = new MaterializationCoordinate({ + frontier: new Map([['writer', 'patch']]), + ceiling: null, +}); + +describe('GitCasMaterializationStoreValidation', () => { + it('accepts a partial liveness materialization without a state hash', () => { + expect(() => requireRetainRequest({ + coordinate, + roots: rootsWith({ + nodeAlive: MaterializationRoot.retained(new BundleHandle('node-root')), + }), + stateHash: null, + })).not.toThrow(); + }); + + it('rejects a materialization that cannot answer any read', () => { + expect(() => requireRetainRequest({ + coordinate, + roots: rootsWith({}), + stateHash: null, + })).toThrowError(/at least one materialization root/u); + }); + + it('rejects a partial handle that claims a whole-state replay basis', () => { + expect(() => requireRetainRequest({ + coordinate, + roots: rootsWith({ + replayBasis: MaterializationRoot.retained(new BundleHandle('basis-root')), + }), + stateHash: null, + })).toThrowError(/cannot retain a whole-state replay basis/u); + }); +}); + +function rootsWith(overrides: { + readonly nodeAlive?: MaterializationRoot; + readonly replayBasis?: MaterializationRoot; +}): MaterializationRoots { + return new MaterializationRoots({ + adjacency: MaterializationRoot.unavailable(), + edgeAlive: MaterializationRoot.unavailable(), + edgeBirths: MaterializationRoot.unavailable(), + frontier: MaterializationRoot.unavailable(), + nodeAlive: overrides.nodeAlive ?? MaterializationRoot.unavailable(), + properties: MaterializationRoot.unavailable(), + provenanceSupport: MaterializationRoot.unavailable(), + replayBasis: overrides.replayBasis ?? MaterializationRoot.unavailable(), + roaringIndexes: MaterializationRoot.unavailable(), + }); +} From acdbc5df9cdb6c60684a2f7c33136d2da5c38b05 Mon Sep 17 00:00:00 2001 From: James Ross Date: Fri, 24 Jul 2026 09:53:42 -0700 Subject: [PATCH 2/3] fix: close bounded materialization failure paths --- src/domain/orset/session/StateSession.ts | 10 +++++ .../controllers/BoundedLiveMaterialization.ts | 27 +++++++------ .../GitCasMaterializationStoreValidation.ts | 10 +++++ .../domain/orset/session/StateSession.test.ts | 13 +++++++ .../BoundedLiveMaterialization.test.ts | 39 ++++++++++++++++++- ...MaterializeController.stateSession.test.ts | 3 +- ...MaterializationStoreAdapter.resume.test.ts | 4 +- ...tCasMaterializationStoreValidation.test.ts | 32 +++++++++++++++ 8 files changed, 123 insertions(+), 15 deletions(-) diff --git a/src/domain/orset/session/StateSession.ts b/src/domain/orset/session/StateSession.ts index ce695772..91f5c546 100644 --- a/src/domain/orset/session/StateSession.ts +++ b/src/domain/orset/session/StateSession.ts @@ -222,6 +222,16 @@ export default class StateSession { return roots; } + /** + * Terminates a failed session without flushing or retaining its pending pages. + * + * The workspace owns cleanup for any pages already staged before the failure. + */ + abort(): void { + this.#closed = true; + this.#closePrepared = false; + } + async prepareClose(): Promise { this.#assertOpen(); this.#closePrepared = true; diff --git a/src/domain/services/controllers/BoundedLiveMaterialization.ts b/src/domain/services/controllers/BoundedLiveMaterialization.ts index 1f41cf53..b6b5bf98 100644 --- a/src/domain/services/controllers/BoundedLiveMaterialization.ts +++ b/src/domain/services/controllers/BoundedLiveMaterialization.ts @@ -87,18 +87,23 @@ async function buildBoundedMaterialization(args: { { nodeAliveRootOid: null, edgeAliveRootOid: null }, { workspace }, ); - const patchCount = await replayLiveness(session, deps, coordinate); - if (patchCount === 0) { - await session.close(); - await workspace.release(); - return null; + try { + const patchCount = await replayLiveness(session, deps, coordinate); + if (patchCount === 0) { + await session.close(); + await workspace.release(); + return null; + } + return await retainPreparedLiveness({ + session, + coordinate, + workspace, + patchCount, + }); + } catch (raw) { + session.abort(); + throw raw; } - return await retainPreparedLiveness({ - session, - coordinate, - workspace, - patchCount, - }); } async function replayLiveness( diff --git a/src/infrastructure/adapters/GitCasMaterializationStoreValidation.ts b/src/infrastructure/adapters/GitCasMaterializationStoreValidation.ts index 0717aba1..79262f1a 100644 --- a/src/infrastructure/adapters/GitCasMaterializationStoreValidation.ts +++ b/src/infrastructure/adapters/GitCasMaterializationStoreValidation.ts @@ -24,6 +24,7 @@ function requireRetainRoots(roots: MaterializationRoots): void { function requireRetainStateHash(request: RetainMaterializationRequest): void { if (request.stateHash !== null) { requireNonEmpty(request.stateHash, 'stateHash'); + requireCompleteReplayBasis(request); return; } if (request.replayBasis !== undefined || request.roots.replayBasis.status === 'retained') { @@ -31,6 +32,15 @@ function requireRetainStateHash(request: RetainMaterializationRequest): void { } } +function requireCompleteReplayBasis(request: RetainMaterializationRequest): void { + if ( + request.replayBasis === undefined + && request.roots.replayBasis.status !== 'retained' + ) { + throw storageError('complete materialization requires a whole-state replay basis'); + } +} + export function requireCoordinate(coordinate: MaterializationCoordinate): void { if (!(coordinate instanceof MaterializationCoordinate)) { throw storageError('coordinate has an invalid runtime identity'); diff --git a/test/unit/domain/orset/session/StateSession.test.ts b/test/unit/domain/orset/session/StateSession.test.ts index ec9d841b..4703db1b 100644 --- a/test/unit/domain/orset/session/StateSession.test.ts +++ b/test/unit/domain/orset/session/StateSession.test.ts @@ -294,6 +294,19 @@ describe("StateSession", () => { expect(session.dirtyPageCount()).toBeGreaterThan(0); expect(await session.nodeContains("node:retryable")).toBe(true); }); + + it("aborts a failed session without retaining its pending pages", async () => { + const workspace = new RecordingWorkspace(); + const { session } = await openSession({ workspace, maxDirtyPages: 1_000 }); + await session.addNode("node:aborted", new Dot("alice", 1)); + + session.abort(); + + expect(workspace.checkpoints).toEqual([]); + await expect(session.nodeContains("node:aborted")) + .rejects.toBeInstanceOf(StateSessionError); + await expect(session.close()).rejects.toBeInstanceOf(StateSessionError); + }); }); describe("edge cases", () => { diff --git a/test/unit/domain/services/controllers/BoundedLiveMaterialization.test.ts b/test/unit/domain/services/controllers/BoundedLiveMaterialization.test.ts index fb8335cb..93148b57 100644 --- a/test/unit/domain/services/controllers/BoundedLiveMaterialization.test.ts +++ b/test/unit/domain/services/controllers/BoundedLiveMaterialization.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it, vi } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import PatchCollector, { type CheckpointData, @@ -36,6 +36,10 @@ const coordinate = new MaterializationCoordinate({ }); describe('BoundedLiveMaterialization', () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + it('fails closed when a newly retained handle cannot be acquired', async () => { const materializations = new InMemoryMaterializationStore(); vi.spyOn(materializations, 'acquireExact').mockResolvedValue(null); @@ -64,6 +68,39 @@ describe('BoundedLiveMaterialization', () => { expect(materializations.workspaces[0]?.released).toBe(true); }); + it('aborts the session and releases its workspace when liveness replay fails', async () => { + const materializations = new InMemoryMaterializationStore(); + const failure = new Error('patch replay failed'); + const abort = vi.spyOn(StateSession.prototype, 'abort'); + const deps = createDeps(materializations); + vi.spyOn(deps.patches, 'loadPatchChain').mockRejectedValue(failure); + + await expect(retainBoundedLiveMaterialization({ deps, coordinate })) + .rejects.toBe(failure); + + expect(abort).toHaveBeenCalledOnce(); + expect(materializations.workspaces[0]?.released).toBe(true); + }); + + it('aborts the session and preserves a workspace promotion failure', async () => { + const materializations = new InMemoryMaterializationStore(); + const failure = new Error('workspace promotion failed'); + const openWorkspace = materializations.openWorkspace.bind(materializations); + vi.spyOn(materializations, 'openWorkspace').mockImplementation(async (requested) => { + const workspace = await openWorkspace(requested); + vi.spyOn(workspace, 'promote').mockRejectedValue(failure); + return workspace; + }); + const abort = vi.spyOn(StateSession.prototype, 'abort'); + const deps = createDeps(materializations); + + await expect(retainBoundedLiveMaterialization({ deps, coordinate })) + .rejects.toBe(failure); + + expect(abort).toHaveBeenCalledOnce(); + expect(materializations.workspaces[0]?.released).toBe(true); + }); + it('releases a partial acquisition before using an exact whole-state snapshot', async () => { const materializations = new InMemoryMaterializationStore(); const deps = createDeps(materializations); diff --git a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts index c3ff6ce8..ebc33c4a 100644 --- a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts +++ b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts @@ -653,7 +653,8 @@ describe("MaterializeController — state session integration", () => { expect(resolution.source).toBe("materialized"); expect(resolution.replayedPatchCount).toBe(1); - expect(resolution.materialization?.stateHash).toBeNull(); + expect(resolution.materialization).toBeInstanceOf(MaterializationHandle); + expect(resolution.materialization).toMatchObject({ stateHash: null }); expect(fixtures.materializations.retainedRequests).toHaveLength(1); expect(fixtures.patches.collectForFrontier).toHaveBeenCalledOnce(); expect(fixtures.stateCache.getExact).not.toHaveBeenCalled(); diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.resume.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.resume.test.ts index 5f3652ec..438170bd 100644 --- a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.resume.test.ts +++ b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.resume.test.ts @@ -80,12 +80,12 @@ describe('GitCasMaterializationStoreAdapter retained resume', () => { await harness.adapter.close(); }); - it('excludes compatible predecessors without a retained replay basis', async () => { + it('excludes partial 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', + stateHash: null, }); const acquisition = await harness.adapter.acquireBestCompatiblePredecessor( diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationStoreValidation.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationStoreValidation.test.ts index 4e55b69c..602790ad 100644 --- a/test/unit/infrastructure/adapters/GitCasMaterializationStoreValidation.test.ts +++ b/test/unit/infrastructure/adapters/GitCasMaterializationStoreValidation.test.ts @@ -3,6 +3,7 @@ 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 BundleHandle from '../../../../src/domain/storage/BundleHandle.ts'; import { requireRetainRequest } from '../../../../src/infrastructure/adapters/GitCasMaterializationStoreValidation.ts'; @@ -39,6 +40,37 @@ describe('GitCasMaterializationStoreValidation', () => { stateHash: null, })).toThrowError(/cannot retain a whole-state replay basis/u); }); + + it('accepts a complete materialization with a replay basis to stage', () => { + expect(() => requireRetainRequest({ + coordinate, + roots: rootsWith({ + nodeAlive: MaterializationRoot.retained(new BundleHandle('node-root')), + }), + replayBasis: createEmptyState(), + stateHash: 'state-hash', + })).not.toThrow(); + }); + + it('accepts a complete materialization with an already retained replay basis', () => { + expect(() => requireRetainRequest({ + coordinate, + roots: rootsWith({ + replayBasis: MaterializationRoot.retained(new BundleHandle('basis-root')), + }), + stateHash: 'state-hash', + })).not.toThrow(); + }); + + it('rejects a complete materialization without any replay basis', () => { + expect(() => requireRetainRequest({ + coordinate, + roots: rootsWith({ + nodeAlive: MaterializationRoot.retained(new BundleHandle('node-root')), + }), + stateHash: 'state-hash', + })).toThrowError(/requires a whole-state replay basis/u); + }); }); function rootsWith(overrides: { From f5eae416602152a661785577bcc4263de46c786a Mon Sep 17 00:00:00 2001 From: James Ross Date: Fri, 24 Jul 2026 10:05:58 -0700 Subject: [PATCH 3/3] test: retain valid complete materialization fixtures --- ...ializationStoreAdapter.integration.test.ts | 23 +++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts b/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts index cbf35e69..2d1d760e 100644 --- a/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts +++ b/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts @@ -11,6 +11,8 @@ 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 { 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 GitCasRepositoryAdapter from '../../../../src/infrastructure/adapters/GitCasRepositoryAdapter.ts'; import GitCasTrieStoreAdapter from '../../../../src/infrastructure/adapters/GitCasTrieStoreAdapter.ts'; @@ -55,10 +57,12 @@ describe('GitCasMaterializationStoreAdapter integration', () => { }); const trieFixture = await createTrieRoot(harness.cas); const rootFixture = await createRoots(harness.cas, trieFixture); + const replayBasis = createEmptyState(); const retained = await harness.materializations.retain({ coordinate, roots: rootFixture.roots, - stateHash: 'state-hash', + replayBasis, + stateHash: await stateHash(replayBasis), }); const reopenedCas = createCas(harness.plumbing); @@ -88,7 +92,7 @@ describe('GitCasMaterializationStoreAdapter integration', () => { expect(resolved.bundle.equals(retained.bundle)).toBe(true); expect(resolved.roots.entries().map(([name, root]) => rootSignature(name, root))) - .toEqual(rootFixture.roots.entries().map(([name, root]) => rootSignature(name, root))); + .toEqual(retained.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) { @@ -116,12 +120,15 @@ describe('GitCasMaterializationStoreAdapter integration', () => { it('keeps a replaced generation reachable until its runtime lease closes', async () => { const coordinate = workspaceCoordinate(); + const replayBasis = createEmptyState(); + const retainedStateHash = await stateHash(replayBasis); const firstTrie = await createTrieRoot(harness.cas, 7); const firstRoots = await createRoots(harness.cas, firstTrie, 0); const first = await harness.materializations.retain({ coordinate, roots: firstRoots.roots, - stateHash: 'first-state-hash', + replayBasis, + stateHash: retainedStateHash, }); const acquisition = await harness.materializations.acquireExact(coordinate); if (acquisition === null) { @@ -133,7 +140,8 @@ describe('GitCasMaterializationStoreAdapter integration', () => { const second = await harness.materializations.retain({ coordinate, roots: secondRoots.roots, - stateHash: 'second-state-hash', + replayBasis, + stateHash: retainedStateHash, }); await expireAllReflogs(harness.path); @@ -400,6 +408,13 @@ function workspaceCoordinate(): MaterializationCoordinate { }); } +async function stateHash(state: ReturnType): Promise { + return await computeStateHash(state, { + codec: defaultCodec, + crypto: new NodeCryptoAdapter(), + }); +} + class MutableClock { #current: number;