diff --git a/docs/topics/cas-first-memoized-materialization.md b/docs/topics/cas-first-memoized-materialization.md index 0472c273..aba2d170 100644 --- a/docs/topics/cas-first-memoized-materialization.md +++ b/docs/topics/cas-first-memoized-materialization.md @@ -13,8 +13,28 @@ through `git-cas`. ## The Live Materialization Lifecycle -When a Git-backed runtime has a state cache, live materialization follows this -coordinate-first lifecycle: +There are now two deliberately different controller contracts: + +- `resolveLiveMaterialization()` returns an operation-scoped retained-handle + resolution. An exact coordinate hit uses git-cas `CacheSet.acquire()` and + does not open the legacy state cache, patch streams, a state session, or the + whole-state projector. The acquisition pins the observed cache generation + until the caller invokes `release()`; replacement or eviction cannot collect + its roots during that scope. A miss does not publish a legacy full-state + snapshot and acquires the newly retained handle before returning it. +- `materialize()` remains the explicit compatibility and diagnostic operation + that returns a complete state projection. + +The split prevents callers that need only a durable basis from paying the +graph-sized cost required by the legacy result shape. On a retained-handle miss, +handle resolution still performs the cold materialization path until every +independently addressable root can be produced without a complete state. +Acquisition release is mandatory on both warm and cold non-empty resolutions; +release failures remain operational failures, while failure-path cleanup never +replaces the primary materialization error. + +When a Git-backed runtime has a state cache, the compatibility `materialize()` +operation follows this coordinate-first lifecycle: ```text [current frontier] @@ -143,6 +163,8 @@ lost payload bytes, or run Git garbage collection. ## Current Limitations +- RuntimeHost and checkpoint creation do not yet consume the handle-first + result, so their compatibility path still owns 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 materialization descriptors currently carry node/edge trie roots; diff --git a/package-lock.json b/package-lock.json index c1576074..26f5d3ae 100644 --- a/package-lock.json +++ b/package-lock.json @@ -13,7 +13,7 @@ ], "dependencies": { "@git-stunts/alfred": "^0.10.4", - "@git-stunts/git-cas": "^6.2.0", + "@git-stunts/git-cas": "^6.3.0", "@git-stunts/plumbing": "^3.1.0", "@git-stunts/trailer-codec": "^2.1.1", "@noble/hashes": "^2.2.0", @@ -489,9 +489,9 @@ "license": "Apache-2.0" }, "node_modules/@git-stunts/git-cas": { - "version": "6.2.0", - "resolved": "https://registry.npmjs.org/@git-stunts/git-cas/-/git-cas-6.2.0.tgz", - "integrity": "sha512-m8+ZzgNhKU6pVS9pjqJlwAnwYI/s+NMEnINC+Q0g3h6T6mNPdH8U0jb4nEoxU9N1TF+Ut5bjtRMRRaYT75dlew==", + "version": "6.3.0", + "resolved": "https://registry.npmjs.org/@git-stunts/git-cas/-/git-cas-6.3.0.tgz", + "integrity": "sha512-Cl/WPjj60LvjXl3BqSb1M3a0tx2xpx6KxGEC1TXKekNzgn5so/t43LG7Qz2XuXle+YmXWoCi8H94cJYvfgI8Yw==", "license": "Apache-2.0", "dependencies": { "@flyingrobots/bijou": "^5.0.0", diff --git a/package.json b/package.json index f0b6bb8f..4be54594 100644 --- a/package.json +++ b/package.json @@ -117,7 +117,7 @@ }, "dependencies": { "@git-stunts/alfred": "^0.10.4", - "@git-stunts/git-cas": "^6.2.0", + "@git-stunts/git-cas": "^6.3.0", "@git-stunts/plumbing": "^3.1.0", "@git-stunts/trailer-codec": "^2.1.1", "@noble/hashes": "^2.2.0", diff --git a/src/domain/materialization/LiveMaterializationResolution.ts b/src/domain/materialization/LiveMaterializationResolution.ts new file mode 100644 index 00000000..b394270a --- /dev/null +++ b/src/domain/materialization/LiveMaterializationResolution.ts @@ -0,0 +1,135 @@ +import WarpError from '../errors/WarpError.ts'; +import MaterializationHandle from './MaterializationHandle.ts'; + +export type LiveMaterializationSource = 'empty' | 'retained' | 'materialized'; + +export type LiveMaterializationResolutionOptions = + | Readonly<{ + materialization: null; + source: 'empty'; + replayedPatchCount: 0; + release: () => Promise; + }> + | Readonly<{ + materialization: MaterializationHandle; + source: 'retained'; + replayedPatchCount: 0; + release: () => Promise; + }> + | Readonly<{ + materialization: MaterializationHandle; + source: 'materialized'; + replayedPatchCount: number; + release: () => Promise; + }>; + +const LIVE_MATERIALIZATION_SOURCES = new Set([ + 'empty', + 'retained', + 'materialized', +]); + +/** Validated operation-scoped access to a live materialization handle. */ +export default class LiveMaterializationResolution { + readonly materialization: MaterializationHandle | null; + readonly source: LiveMaterializationSource; + readonly replayedPatchCount: number; + readonly #releaseOperation: () => Promise; + + constructor(options: LiveMaterializationResolutionOptions | null | undefined) { + const fields = requireOptions(options); + const materialization = requireMaterialization(fields.materialization); + const source = requireSource(fields.source); + const replayedPatchCount = requirePatchCount(fields.replayedPatchCount); + const releaseOperation = requireRelease(fields.release); + validateCombination({ materialization, source, replayedPatchCount }); + + this.materialization = materialization; + this.source = source; + this.replayedPatchCount = replayedPatchCount; + this.#releaseOperation = releaseOperation; + Object.freeze(this); + } + + release(): Promise { + return this.#releaseOperation(); + } +} + +function requireOptions( + options: LiveMaterializationResolutionOptions | null | undefined, +): LiveMaterializationResolutionOptions { + if (options === null || options === undefined) { + throw resolutionError('options are required'); + } + return options; +} + +function requireMaterialization( + materialization: MaterializationHandle | null, +): MaterializationHandle | null { + if (materialization !== null && !(materialization instanceof MaterializationHandle)) { + throw resolutionError('materialization has an invalid runtime identity'); + } + return materialization; +} + +function requireSource(source: LiveMaterializationSource): LiveMaterializationSource { + if (!LIVE_MATERIALIZATION_SOURCES.has(source)) { + throw resolutionError('source is invalid'); + } + return source; +} + +function requirePatchCount(replayedPatchCount: number): number { + if (!Number.isSafeInteger(replayedPatchCount) || replayedPatchCount < 0) { + throw resolutionError('replayedPatchCount must be a non-negative safe integer'); + } + return replayedPatchCount; +} + +function requireRelease(release: () => Promise): () => Promise { + if (typeof release !== 'function') { + throw resolutionError('release must be a function'); + } + return release; +} + +function validateCombination(fields: Readonly<{ + materialization: MaterializationHandle | null; + source: LiveMaterializationSource; + replayedPatchCount: number; +}>): void { + if (fields.source === 'empty') { + validateEmptyCombination(fields); + return; + } + if (fields.materialization === null) { + throw resolutionError(`${fields.source} source requires a materialization`); + } + if (fields.source === 'retained') { + validateRetainedCombination(fields.replayedPatchCount); + } +} + +function validateEmptyCombination(fields: Readonly<{ + materialization: MaterializationHandle | null; + replayedPatchCount: number; +}>): void { + if (fields.materialization !== null || fields.replayedPatchCount !== 0) { + throw resolutionError('empty source cannot carry a materialization or replayed patches'); + } +} + +function validateRetainedCombination(replayedPatchCount: number): void { + if (replayedPatchCount !== 0) { + throw resolutionError('retained source cannot report replayed patches'); + } +} + +function resolutionError(message: string): WarpError { + return new WarpError( + `Live materialization resolution ${message}`, + 'E_MATERIALIZATION_RESOLUTION', + ); +} diff --git a/src/domain/services/controllers/MaterializationWorkspaceCleanup.ts b/src/domain/services/controllers/MaterializationWorkspaceCleanup.ts index 7997a76a..cff025a8 100644 --- a/src/domain/services/controllers/MaterializationWorkspaceCleanup.ts +++ b/src/domain/services/controllers/MaterializationWorkspaceCleanup.ts @@ -1,4 +1,5 @@ import type LoggerPort from '../../../ports/LoggerPort.ts'; +import type { MaterializationAcquisition } from '../../../ports/MaterializationStorePort.ts'; import type MaterializationWorkspacePort from '../../../ports/MaterializationWorkspacePort.ts'; /** Release a workspace without allowing cleanup to replace an existing failure. */ @@ -13,9 +14,25 @@ export async function releaseWorkspaceAfterFailure( } } -function reportCleanupFailure(logger: LoggerPort | undefined, failure: Failure): void { +/** Release a retained materialization without replacing an active failure. */ +export async function releaseAcquisitionAfterFailure( + acquisition: Pick | null, + logger?: LoggerPort, +): Promise { + try { + await acquisition?.release(); + } catch (cleanupFailure) { + reportCleanupFailure(logger, cleanupFailure, 'acquisition'); + } +} + +function reportCleanupFailure( + logger: LoggerPort | undefined, + failure: Failure, + scope = 'workspace', +): void { try { - logger?.warn('[warp] materialization workspace release failed during error cleanup', { + logger?.warn(`[warp] materialization ${scope} release failed during error cleanup`, { error: failure instanceof Error ? failure.message : String(failure), }); } catch { diff --git a/src/domain/services/controllers/MaterializeController.ts b/src/domain/services/controllers/MaterializeController.ts index ffccb581..fef84bea 100644 --- a/src/domain/services/controllers/MaterializeController.ts +++ b/src/domain/services/controllers/MaterializeController.ts @@ -1,14 +1,4 @@ -/** - * MaterializeController — CRDT state replay from patches. - * - * Three materialization pipelines: - * 1. materialize() — full or checkpoint-incremental, live frontier - * 2. materializeCoordinate() — explicit frontier snapshot - * 3. materializeAt() — specific checkpoint SHA - * - * Dependencies are constructor-injected. No host bag. - * Side effects (notification, GC, auto-checkpoint) are the caller's job. - */ +/** CRDT replay and retained-handle resolution across supported coordinates. */ import { reducePatches as reduceJoinedPatches, createEmptyState } from '../JoinReducer.ts'; import { ProvenanceIndex } from '../provenance/ProvenanceIndex.ts'; import { computeStateHash } from '../state/StateSerializer.ts'; @@ -58,6 +48,7 @@ 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 LiveMaterializationResolution from '../../materialization/LiveMaterializationResolution.ts'; import type MaterializationRoots from '../../materialization/MaterializationRoots.ts'; import type StorageRetentionWitness from '../../storage/StorageRetentionWitness.ts'; import WarpError from '../../errors/WarpError.ts'; @@ -66,7 +57,10 @@ import type { MaterializeResultBuildInput, MaterializeStrategyRuntime, } from './MaterializeStrategyRuntime.ts'; -import { releaseWorkspaceAfterFailure } from './MaterializationWorkspaceCleanup.ts'; +import { + releaseAcquisitionAfterFailure, + releaseWorkspaceAfterFailure, +} from './MaterializationWorkspaceCleanup.ts'; export type MaterializePersistence = { readRef(ref: string): Promise; }; @@ -145,8 +139,6 @@ function reduceMaterializePatches( return reducePlain(patches, base); } -// ── Provenance ────────────────────────────────────────────────────── - function buildProvenance(patches: PatchWithSha[], base?: ProvenanceIndex): ProvenanceIndex { const index = base ? base.clone() : new ProvenanceIndex(); for (const { patch, sha } of patches) { @@ -155,14 +147,10 @@ function buildProvenance(patches: PatchWithSha[], base?: ProvenanceIndex): Prove return index; } -// ── State hash ────────────────────────────────────────────────────── - async function computeHash(deps: MaterializeDeps, state: WarpState): Promise { return await computeStateHash(state, { crypto: deps.crypto, codec: deps.codec }); } -// ── Controller ────────────────────────────────────────────────────── - export default class MaterializeController { private readonly _deps: MaterializeDeps; private readonly _liveStrategy: MaterializeLiveStrategy; @@ -192,6 +180,9 @@ export default class MaterializeController { wantDiff: opts.wantDiff === true, }); } + resolveLiveMaterialization(): Promise { + return this._liveStrategy.resolveMaterialization(); + } /** Coordinate materialization — explicit frontier. */ async materializeCoordinate( @@ -352,35 +343,44 @@ export default class MaterializeController { 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 acquisition = await this._deps.materializations.acquireExact(coordinate); + 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'); + } + const retainedRoots = retained === null + ? null + : materializationSessionOpen(retained.roots); + const reduced = await reduceSessionBackedState({ + openStateSession, + materializations: this._deps.materializations, + logger: this._deps.logger, + coordinate, + patches: [], + baseState: snapshot.state, + ...(retainedRoots === null ? {} : { roots: retainedRoots }), + receipts: false, + wantDiff: options.wantDiff, + }); + result = 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 }), + }); + } catch (raw) { + await releaseAcquisitionAfterFailure(acquisition, this._deps.logger); + throw raw; } - const retainedRoots = retained === null - ? null - : materializationSessionOpen(retained.roots); - const reduced = await reduceSessionBackedState({ - openStateSession, - materializations: this._deps.materializations, - logger: this._deps.logger, - coordinate, - 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 }), - }); + await acquisition?.release(); + return result; } private async _reducePatches( diff --git a/src/domain/services/controllers/MaterializeLiveStrategy.ts b/src/domain/services/controllers/MaterializeLiveStrategy.ts index 27fb4bd8..d754f80a 100644 --- a/src/domain/services/controllers/MaterializeLiveStrategy.ts +++ b/src/domain/services/controllers/MaterializeLiveStrategy.ts @@ -4,6 +4,10 @@ import type { MaterializeStrategyRuntime, } from './MaterializeStrategyRuntime.ts'; import type { MaterializeResult } from './MaterializeController.ts'; +import MaterializationCoordinate from '../../materialization/MaterializationCoordinate.ts'; +import LiveMaterializationResolution from '../../materialization/LiveMaterializationResolution.ts'; +import WarpError from '../../errors/WarpError.ts'; +import type { MaterializationAcquisition } from '../../../ports/MaterializationStorePort.ts'; import type { CheckpointData, PatchWithSha, @@ -17,6 +21,7 @@ import { snapshotToMaterializeResult, } from './MaterializeSnapshotCacheResult.ts'; import { snapshotPublicationForReceipts } from './MaterializeSnapshotPublication.ts'; +import { releaseAcquisitionAfterFailure } from './MaterializationWorkspaceCleanup.ts'; function nonEmptySha(value: string | undefined): value is string { return typeof value === 'string' && value.length > 0; @@ -32,7 +37,7 @@ export default class MaterializeLiveStrategy { async materialize(opts: MaterializeLiveOptions): Promise { const frontier = await this.runtime.deps.patches.getFrontier(); if (frontier.size === 0) { - return await this.runtime.emptyResult(null, frontier, snapshotPublicationForReceipts(opts)); + return await this.runtime.emptyResult(null, frontier, snapshotPublicationForLiveOptions(opts)); } const coordinate = this.snapshotCoordinate(frontier); const stateCache = this.runtime.deps.getStateCache?.() ?? null; @@ -47,6 +52,54 @@ export default class MaterializeLiveStrategy { return await this.replayCurrentCoordinate(coordinate, opts); } + async resolveMaterialization(): Promise { + const frontier = await this.runtime.deps.patches.getFrontier(); + if (frontier.size === 0) { + return emptyResolution(); + } + + const coordinate = this.snapshotCoordinate(frontier); + const materializationCoordinate = new MaterializationCoordinate(coordinate); + const retained = await this.runtime.deps.materializations.acquireExact(materializationCoordinate); + if (retained !== null) { + return await this.resolveRetainedMaterialization(retained, materializationCoordinate); + } + return await this.materializeAndAcquire(coordinate, materializationCoordinate); + } + + private async resolveRetainedMaterialization( + retained: MaterializationAcquisition, + coordinate: MaterializationCoordinate, + ): Promise { + try { + return retainedResolution(retained, coordinate); + } catch (raw) { + await releaseAcquisitionAfterFailure(retained, this.runtime.deps.logger); + throw raw; + } + } + + private async materializeAndAcquire( + coordinate: WarpStateCoordinate, + materializationCoordinate: MaterializationCoordinate, + ): Promise { + const result = await this.replayCurrentCoordinate(coordinate, { + receipts: false, + wantDiff: false, + publishSnapshot: false, + }); + const acquired = await this.runtime.deps.materializations.acquireExact(materializationCoordinate); + if (acquired === null) { + throw resolutionError('newly retained handle could not be acquired'); + } + try { + return materializedResolution(result, acquired); + } catch (raw) { + await releaseAcquisitionAfterFailure(acquired, this.runtime.deps.logger); + throw raw; + } + } + private async tryResolveConfiguredStateCache( stateCache: WarpStateCachePort | null, coordinate: WarpStateCoordinate, @@ -137,6 +190,9 @@ export default class MaterializeLiveStrategy { degraded: provenanceBase === undefined, ceiling: null, frontier, + ...(opts.publishSnapshot === undefined + ? {} + : { publishSnapshot: opts.publishSnapshot }), }); } @@ -187,7 +243,7 @@ export default class MaterializeLiveStrategy { return await this.runtime.emptyResult( coordinate.ceiling, coordinate.frontier, - snapshotPublicationForReceipts(opts), + snapshotPublicationForLiveOptions(opts), ); } return await this.runtime.buildResult({ @@ -196,6 +252,9 @@ export default class MaterializeLiveStrategy { degraded: false, ceiling: coordinate.ceiling, frontier: coordinate.frontier, + ...(opts.publishSnapshot === undefined + ? {} + : { publishSnapshot: opts.publishSnapshot }), }); } @@ -267,3 +326,65 @@ export default class MaterializeLiveStrategy { }); } } + +function emptyResolution(): LiveMaterializationResolution { + return new LiveMaterializationResolution({ + materialization: null, + source: 'empty', + replayedPatchCount: 0, + release: () => Promise.resolve(), + }); +} + +function retainedResolution( + acquired: MaterializationAcquisition, + coordinate: MaterializationCoordinate, +): LiveMaterializationResolution { + const retained = acquired.materialization; + if (!retained.coordinate.equals(coordinate)) { + throw resolutionError('retained handle coordinate does not match the requested coordinate'); + } + return new LiveMaterializationResolution({ + materialization: retained, + source: 'retained', + replayedPatchCount: 0, + release: async () => await acquired.release(), + }); +} + +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 { + return snapshotPublicationForReceipts({ + receipts: opts.receipts || opts.publishSnapshot === false, + }); +} + +function resolutionError(message: string): WarpError { + return new WarpError( + `Materialization resolution ${message}`, + 'E_MATERIALIZATION_RESUME', + ); +} diff --git a/src/domain/services/controllers/MaterializeStrategyRuntime.ts b/src/domain/services/controllers/MaterializeStrategyRuntime.ts index 14800ac2..9ffee379 100644 --- a/src/domain/services/controllers/MaterializeStrategyRuntime.ts +++ b/src/domain/services/controllers/MaterializeStrategyRuntime.ts @@ -22,6 +22,7 @@ import type { export type MaterializeLiveOptions = { receipts: boolean; wantDiff: boolean; + publishSnapshot?: boolean; }; export type MaterializeCeilingOptions = { diff --git a/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts b/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts index 55eb3df5..88ee8c1c 100644 --- a/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts +++ b/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts @@ -2,6 +2,7 @@ import type { BundleCapability, BundleMember, BundleMemberInput, + CacheAcquisition, CacheHit, CacheSet, PageHandle, @@ -22,6 +23,7 @@ import type CodecPort from '../../ports/CodecPort.ts'; import type CryptoPort from '../../ports/CryptoPort.ts'; import type MaterializationWorkspacePort from '../../ports/MaterializationWorkspacePort.ts'; import MaterializationStorePort, { + type MaterializationAcquisition, type RetainMaterializationRequest, } from '../../ports/MaterializationStorePort.ts'; import { adaptGitCasRetentionWitness } from './GitCasRetentionWitnessAdapter.ts'; @@ -42,7 +44,7 @@ const MAX_DESCRIPTOR_BYTES = 1024 * 1024; // A root-list change also requires a descriptor schema-version change. const MATERIALIZATION_MEMBER_COUNT = MATERIALIZATION_ROOT_NAMES.length + 1; -type MaterializationCacheSet = Pick; +type MaterializationCacheSet = Pick; export type GitCasMaterializationFacade = { readonly bundles: Pick; @@ -158,23 +160,34 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt return adaptGitCasRetentionWitness(stored.witness.toJSON()); } - override async findExact( + override async acquireExact( coordinate: MaterializationCoordinate, - ): Promise { + ): Promise { requireCoordinate(coordinate); const cache = await this.#cas.caches.open({ namespace: CACHE_NAMESPACE }); - const hit = await cache.get(await this.#cacheKey(coordinate)); - if (hit === null) { + const acquisition = await cache.acquire(await this.#cacheKey(coordinate)); + if (acquisition === null) { return null; } - if (hit.handle.kind !== 'bundle') { - throw storageError('cache entry does not reference a materialization bundle'); + try { + if (acquisition.hit.handle.kind !== 'bundle') { + throw storageError('cache entry does not reference a materialization bundle'); + } + const materialization = await this.#resolveHit( + acquisition.hit, + acquisition.evidence, + coordinate, + ); + return materializationAcquisition(acquisition, materialization); + } catch (raw) { + await releaseCacheAcquisitionAfterFailure(acquisition); + throw raw; } - return await this.#resolveHit(hit, coordinate); } async #resolveHit( hit: CacheHit, + retention: CacheAcquisition['evidence'], requestedCoordinate: MaterializationCoordinate, ): Promise { const bundle = new BundleHandle(hit.handle.toString()); @@ -193,7 +206,7 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt coordinate: descriptor.coordinate, roots: materializationRootsFromDescriptor(descriptor, members.retainedRoots), stateHash: descriptor.stateHash, - retention: adaptGitCasRetentionWitness(hit.evidence.toJSON()), + retention: adaptGitCasRetentionWitness(retention.toJSON()), }); } @@ -229,6 +242,29 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt } } +function materializationAcquisition( + acquisition: CacheAcquisition, + materialization: MaterializationHandle, +): MaterializationAcquisition { + return Object.freeze({ + materialization, + acquiredAt: acquisition.acquiredAt, + release: async () => { + await acquisition.release(); + }, + }); +} + +async function releaseCacheAcquisitionAfterFailure( + acquisition: CacheAcquisition, +): Promise { + try { + await acquisition.release(); + } catch { + // git-cas doctor owns abandoned-acquisition diagnostics; preserve the primary failure. + } +} + function* materializationMembers( descriptorHandle: string, roots: MaterializationRoots, diff --git a/src/ports/MaterializationStorePort.ts b/src/ports/MaterializationStorePort.ts index 4a06873e..76865faa 100644 --- a/src/ports/MaterializationStorePort.ts +++ b/src/ports/MaterializationStorePort.ts @@ -5,6 +5,12 @@ import type { PromoteMaterializationRequest } from './MaterializationWorkspacePo export type RetainMaterializationRequest = PromoteMaterializationRequest; +export type MaterializationAcquisition = Readonly<{ + materialization: MaterializationHandle; + acquiredAt: string; + release(): Promise; +}>; + /** Storage-neutral lifecycle for retained, independently addressable materializations. */ export default abstract class MaterializationStorePort { abstract openWorkspace( @@ -13,7 +19,7 @@ export default abstract class MaterializationStorePort { abstract retain(_request: RetainMaterializationRequest): Promise; - abstract findExact( + abstract acquireExact( _coordinate: MaterializationCoordinate, - ): Promise; + ): Promise; } diff --git a/test/helpers/InMemoryGitCasFacade.ts b/test/helpers/InMemoryGitCasFacade.ts index fb0ad91b..f1ae0418 100644 --- a/test/helpers/InMemoryGitCasFacade.ts +++ b/test/helpers/InMemoryGitCasFacade.ts @@ -14,6 +14,7 @@ import { type BundleHandleInput, type BundleCapability, type BundleMember, + type CacheAcquisition, type CacheSet, type PageHandleInput, type PageCapability, @@ -48,7 +49,9 @@ export default class InMemoryGitCasFacade { readonly assets: Pick; readonly bundles: Pick; readonly caches: { - open(options: { readonly namespace: string }): Promise>; + open(options: { + readonly namespace: string; + }): Promise>; }; readonly pages: Pick; readonly publications: Pick; @@ -57,10 +60,12 @@ export default class InMemoryGitCasFacade { readonly #storage: InMemoryBlobStorageAdapter; readonly #stagedAssetsByOid = new Map(); readonly #bundleMembers = new Map(); + readonly #cacheAcquisitions = new Set(); readonly #cacheEntries = new Map>(); readonly #pageBytes = new Map(); readonly #publicationRoots = new Map(); #cacheGeneration = 0; + #nextCacheAcquisition = 1; constructor(options: { history: PublicationHistory; @@ -106,6 +111,10 @@ export default class InMemoryGitCasFacade { return Object.freeze([...(this.#cacheEntries.get(namespace)?.values() ?? [])]); } + readActiveCacheAcquisitionCount(): number { + return this.#cacheAcquisitions.size; + } + replaceStoredPage(handle: string, bytes: Uint8Array): void { this.#pageBytes.set(handle, bytes.slice()); } @@ -292,10 +301,47 @@ export default class InMemoryGitCasFacade { return bytes.slice(); } - async #openCache(namespace: string): Promise> { + async #openCache( + namespace: string, + ): Promise> { const entries = this.#cacheEntries.get(namespace) ?? new Map(); this.#cacheEntries.set(namespace, entries); return Object.freeze({ + acquire: async (key): Promise => { + const hit = entries.get(key); + if (hit === undefined) { + return null; + } + const id = `test-acquisition-${String(this.#nextCacheAcquisition)}`; + this.#nextCacheAcquisition += 1; + this.#cacheAcquisitions.add(id); + const acquiredAt = new Date(0).toISOString(); + const evidence = new RetentionWitness({ + handle: hit.handle, + policy: 'pinned', + reachability: 'anchored', + root: { + kind: 'cache-set', + namespace, + ref: `refs/cas/cache-acquisitions/${namespace}/${id}`, + generation: hit.generation, + path: 'root-00000000', + }, + observedAt: acquiredAt, + }); + return Object.freeze({ + id, + hit, + evidence, + acquiredAt, + release: async () => Object.freeze({ + id, + generation: hit.generation, + changed: this.#cacheAcquisitions.delete(id), + releasedAt: new Date(0).toISOString(), + }), + }); + }, get: async (key) => entries.get(key) ?? null, put: async (key, handle, options) => { const previous = entries.get(key) ?? null; diff --git a/test/helpers/InMemoryMaterializationStore.ts b/test/helpers/InMemoryMaterializationStore.ts index b1f52c78..d9b19e13 100644 --- a/test/helpers/InMemoryMaterializationStore.ts +++ b/test/helpers/InMemoryMaterializationStore.ts @@ -5,6 +5,7 @@ import StorageRetentionWitness, { StorageRetentionRoot, } from '../../src/domain/storage/StorageRetentionWitness.ts'; import MaterializationStorePort, { + type MaterializationAcquisition, type RetainMaterializationRequest, } from '../../src/ports/MaterializationStorePort.ts'; import MaterializationWorkspacePort, { @@ -44,8 +45,26 @@ export class InMemoryMaterializationWorkspace extends MaterializationWorkspacePo } } +export class InMemoryMaterializationAcquisition implements MaterializationAcquisition { + readonly materialization: MaterializationHandle; + readonly acquiredAt = '1970-01-01T00:00:00.000Z'; + releaseCalls = 0; + released = false; + + constructor(materialization: MaterializationHandle) { + this.materialization = materialization; + } + + release(): Promise { + this.releaseCalls += 1; + this.released = true; + return Promise.resolve(); + } +} + /** Behavioral retained-materialization store for controller tests. */ export default class InMemoryMaterializationStore extends MaterializationStorePort { + readonly acquisitions: InMemoryMaterializationAcquisition[] = []; readonly exactLookups: MaterializationCoordinate[] = []; readonly retainedRequests: RetainMaterializationRequest[] = []; readonly workspaces: InMemoryMaterializationWorkspace[] = []; @@ -81,11 +100,17 @@ export default class InMemoryMaterializationStore extends MaterializationStorePo return Promise.resolve(handle); } - override findExact( + override acquireExact( coordinate: MaterializationCoordinate, - ): Promise { + ): Promise { this.exactLookups.push(coordinate); - return Promise.resolve(this.#handles.get(coordinateKey(coordinate)) ?? null); + const handle = this.#handles.get(coordinateKey(coordinate)); + if (handle === undefined) { + return Promise.resolve(null); + } + const acquisition = new InMemoryMaterializationAcquisition(handle); + this.acquisitions.push(acquisition); + return Promise.resolve(acquisition); } } diff --git a/test/integration/api/materialization.retainedResume.test.ts b/test/integration/api/materialization.retainedResume.test.ts index a4085d68..cf0bcf93 100644 --- a/test/integration/api/materialization.retainedResume.test.ts +++ b/test/integration/api/materialization.retainedResume.test.ts @@ -4,6 +4,7 @@ import type MaterializationCoordinate from '../../../src/domain/materialization/ import type MaterializationHandle from '../../../src/domain/materialization/MaterializationHandle.ts'; import GitCasRepositoryAdapter from '../../../src/infrastructure/adapters/GitCasRepositoryAdapter.ts'; import MaterializationStorePort, { + type MaterializationAcquisition, type RetainMaterializationRequest, } from '../../../src/ports/MaterializationStorePort.ts'; import MaterializationWorkspacePort, { @@ -100,15 +101,15 @@ class RecordingMaterializationStore extends MaterializationStorePort { return retained; } - override async findExact( + override async acquireExact( coordinate: MaterializationCoordinate, - ): Promise { + ): Promise { this.exactLookups.push(coordinate); - const hit = await this.#delegate.findExact(coordinate); - if (hit !== null) { - this.exactHits.push(hit); + const acquisition = await this.#delegate.acquireExact(coordinate); + if (acquisition !== null) { + this.exactHits.push(acquisition.materialization); } - return hit; + return acquisition; } } diff --git a/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts b/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts index 9130df6a..680585ce 100644 --- a/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts +++ b/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts @@ -56,10 +56,11 @@ describe('GitCasMaterializationStoreAdapter integration', () => { harness.plumbing, reopenedCas, ); - const resolved = await reopened.findExact(coordinate); - if (resolved === null) { + const acquisition = await reopened.acquireExact(coordinate); + if (acquisition === null) { throw new Error('Retained materialization was not reopened'); } + const resolved = acquisition.materialization; const nodeAliveRoot = resolved.roots.nodeAlive.handle; if (nodeAliveRoot === null) { throw new Error('Retained materialization did not expose its node root'); @@ -83,6 +84,52 @@ describe('GitCasMaterializationStoreAdapter integration', () => { expect(await harness.plumbing.execute({ args: ['show-ref', '--verify', '--hash', 'refs/cas/caches/git-warp/materializations'], })).toMatch(/^[0-9a-f]{40}\n?$/u); + await acquisition.release(); + }); + + it('keeps an acquired generation reachable across replacement until release', async () => { + const coordinate = workspaceCoordinate(); + 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', + }); + const acquisition = await harness.materializations.acquireExact(coordinate); + if (acquisition === null) { + throw new Error('Retained materialization could not be acquired'); + } + + const secondTrie = await createTrieRoot(harness.cas, 17); + const secondRoots = await createRoots(harness.cas, secondTrie, 16); + const second = await harness.materializations.retain({ + coordinate, + roots: secondRoots.roots, + stateHash: 'second-state-hash', + }); + + await expireAllReflogs(harness.path); + await execFileAsync('git', ['-C', harness.path, 'prune', '--expire=now']); + expect(acquisition.materialization.bundle.equals(first.bundle)).toBe(true); + expect(acquisition.materialization.retention).toMatchObject({ + policy: 'pinned', + reachability: 'anchored', + root: { locator: expect.stringContaining('cache-acquisitions') }, + }); + expect(await harness.cas.bundles.getMember({ + handle: first.bundle.toString(), + path: 'meta/descriptor', + })).not.toBeNull(); + expect(await prunableOids(harness.path)).not.toContain( + GitCasBundleHandle.parse(first.bundle.toString()).oid, + ); + + await acquisition.release(); + await expireAllReflogs(harness.path); + const prunable = await prunableOids(harness.path); + expect(prunable).toContain(GitCasBundleHandle.parse(first.bundle.toString()).oid); + expect(prunable).not.toContain(GitCasBundleHandle.parse(second.bundle.toString()).oid); }); it('keeps workspace roots readable across aggressive Git pruning', async () => { @@ -209,9 +256,12 @@ type TrieRootFixture = Readonly<{ root: BundleHandle; }>; -async function createTrieRoot(cas: ContentAddressableStore): Promise { +async function createTrieRoot( + cas: ContentAddressableStore, + seed = 7, +): Promise { const adapter = new GitCasTrieStoreAdapter({ cas }); - const bytes = new Uint8Array([7, 8, 9]); + const bytes = new Uint8Array([seed, seed + 1, seed + 2]); const leafRoot = await adapter.writeLeaf(bytes); const branchRoot = await adapter.writeBranch(new Map([[0, leafRoot]])); const leafMember = await cas.bundles.getMember({ @@ -235,6 +285,7 @@ async function createTrieRoot(cas: ContentAddressableStore): Promise> { .filter((oid): oid is string => oid !== undefined && oid.length > 0), ); } + +async function expireAllReflogs(path: string): Promise { + await execFileAsync('git', ['-C', path, 'reflog', 'expire', '--expire=now', '--all']); +} diff --git a/test/unit/domain/materialization/MaterializationIdentity.test.ts b/test/unit/domain/materialization/MaterializationIdentity.test.ts index f85e80f2..dc4b0f92 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 LiveMaterializationResolution from '../../../../src/domain/materialization/LiveMaterializationResolution.ts'; import MaterializationRoot from '../../../../src/domain/materialization/MaterializationRoot.ts'; import MaterializationRoots, { MATERIALIZATION_ROOT_NAMES, @@ -207,6 +208,52 @@ describe('MaterializationHandle', () => { }); }); +describe('LiveMaterializationResolution', () => { + it('freezes a valid retained resolution and delegates release', async () => { + let releaseCalls = 0; + const materialization = materializationHandle(); + const resolution = new LiveMaterializationResolution({ + materialization, + source: 'retained', + replayedPatchCount: 0, + release: () => { + releaseCalls += 1; + return Promise.resolve(); + }, + }); + + expect(resolution).toMatchObject({ + materialization, + source: 'retained', + replayedPatchCount: 0, + }); + expect(Object.isFrozen(resolution)).toBe(true); + await resolution.release(); + expect(releaseCalls).toBe(1); + }); + + it.each([ + ['empty materialization', { source: 'empty' }], + ['missing retained materialization', { materialization: null }], + ['retained replay count', { replayedPatchCount: 1 }], + ['invalid source', { source: 'other' }], + ['negative replay count', { replayedPatchCount: -1 }], + ['missing release', { release: null }], + ])('rejects an invalid %s combination', (_case, replacement) => { + const options = resolutionOptions(); + for (const [field, value] of Object.entries(replacement)) { + Reflect.set(options, field, value); + } + expect(() => construct(LiveMaterializationResolution, options)).toThrowError( + /Live materialization resolution/u, + ); + }); + + it('rejects a missing options object', () => { + expect(() => construct(LiveMaterializationResolution, null)).toThrowError(/options/u); + }); +}); + function exactCoordinate(): MaterializationCoordinate { return new MaterializationCoordinate({ frontier: new Map([['writer-a', 'patch-a']]), @@ -255,6 +302,27 @@ function retentionWitness(handle: BundleHandle): StorageRetentionWitness { }); } +function materializationHandle(): MaterializationHandle { + const bundle = bundleHandle('materialization-resolution'); + return new MaterializationHandle({ + laneName: 'events', + bundle, + coordinate: exactCoordinate(), + roots: materializationRoots(), + stateHash: 'state-hash', + retention: retentionWitness(bundle), + }); +} + +function resolutionOptions(): Record Promise)> { + return { + materialization: materializationHandle(), + source: 'retained', + replayedPatchCount: 0, + release: () => Promise.resolve(), + }; +} + function handleOptions(): Record { const bundle = bundleHandle('materialization'); return { diff --git a/test/unit/domain/services/controllers/MaterializationWorkspaceCleanup.test.ts b/test/unit/domain/services/controllers/MaterializationWorkspaceCleanup.test.ts index 16b90235..3eb45f0a 100644 --- a/test/unit/domain/services/controllers/MaterializationWorkspaceCleanup.test.ts +++ b/test/unit/domain/services/controllers/MaterializationWorkspaceCleanup.test.ts @@ -1,5 +1,8 @@ import { describe, expect, it } from 'vitest'; -import { releaseWorkspaceAfterFailure } from '../../../../../src/domain/services/controllers/MaterializationWorkspaceCleanup.ts'; +import { + releaseAcquisitionAfterFailure, + releaseWorkspaceAfterFailure, +} from '../../../../../src/domain/services/controllers/MaterializationWorkspaceCleanup.ts'; import MaterializationWorkspacePort from '../../../../../src/ports/MaterializationWorkspacePort.ts'; import { createMockLogger } from '../../../../helpers/WarpGraphMockLogger.ts'; @@ -33,6 +36,21 @@ describe('releaseWorkspaceAfterFailure', () => { }); }); +describe('releaseAcquisitionAfterFailure', () => { + it('preserves an active failure when acquisition release also fails', async () => { + const logger = createMockLogger(); + const acquisition = { + release: () => Promise.reject(new Error('acquisition release unavailable')), + }; + + await expect(releaseAcquisitionAfterFailure(acquisition, logger)).resolves.toBeUndefined(); + expect(logger.warn).toHaveBeenCalledWith( + '[warp] materialization acquisition release failed during error cleanup', + { error: 'acquisition release unavailable' }, + ); + }); +}); + class RejectingWorkspace extends MaterializationWorkspacePort { readonly #failure: Error | { readonly toString: () => never }; diff --git a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts index 78cb4b38..959727dc 100644 --- a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts +++ b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts @@ -18,10 +18,13 @@ 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, { + InMemoryMaterializationAcquisition, InMemoryMaterializationWorkspace, } from "../../../../helpers/InMemoryMaterializationStore.ts"; import type MaterializationWorkspacePort from "../../../../../src/ports/MaterializationWorkspacePort.ts"; import MaterializationCoordinate from "../../../../../src/domain/materialization/MaterializationCoordinate.ts"; +import MaterializationHandle from "../../../../../src/domain/materialization/MaterializationHandle.ts"; +import type { RetainMaterializationRequest } from "../../../../../src/ports/MaterializationStorePort.ts"; const GEOMETRY = TrieGeometry.default16way(); @@ -200,6 +203,40 @@ function createControllerFixtures() { }; } +type ChangedMaterializationIdentity = "coordinate" | "stateHash" | "bundle"; + +async function createMismatchedAcquisition( + retained: MaterializationHandle, + changedField: ChangedMaterializationIdentity, + retain: (request: RetainMaterializationRequest) => Promise, + acquisitions: InMemoryMaterializationAcquisition[], +): Promise { + const mismatched = changedField === "bundle" + ? await retain({ + coordinate: retained.coordinate, + roots: retained.roots, + stateHash: retained.stateHash, + }) + : new MaterializationHandle({ + laneName: retained.laneName, + bundle: retained.bundle, + coordinate: changedField === "coordinate" + ? new MaterializationCoordinate({ + frontier: new Map([["writer-1", "changed-tip"]]), + ceiling: retained.coordinate.ceiling, + }) + : retained.coordinate, + roots: retained.roots, + stateHash: changedField === "stateHash" + ? `${retained.stateHash}-changed` + : retained.stateHash, + retention: retained.retention, + }); + const acquisition = new InMemoryMaterializationAcquisition(mismatched); + acquisitions.push(acquisition); + return acquisition; +} + describe("MaterializeController — state session integration", () => { it("preserves a session reduction failure when workspace release also fails", async () => { const { openStateSession, materializations, deps } = createControllerFixtures(); @@ -468,6 +505,181 @@ describe("MaterializeController — state session integration", () => { expect(reopened.materialization?.bundle.equals(cold.materialization?.bundle)).toBe(true); }); + it("does not retry an exact acquisition release failure", 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 (cold.materialization === undefined || published === undefined) { + throw new Error("Cold materialization did not publish retained state"); + } + fixtures.stateCache.getExact.mockResolvedValue(published); + const acquisition = new InMemoryMaterializationAcquisition(cold.materialization); + const releaseFailure = new Error("acquisition release failed"); + const release = vi.spyOn(acquisition, "release").mockRejectedValue(releaseFailure); + vi.spyOn(fixtures.materializations, "acquireExact").mockResolvedValue(acquisition); + + await expect(fixtures.controller.materialize()).rejects.toBe(releaseFailure); + + expect(release).toHaveBeenCalledOnce(); + }); + + it("resolves an exact retained handle without touching whole-state dependencies", 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 workspaceCount = fixtures.materializations.workspaces.length; + + fixtures.patches.getFrontier.mockClear(); + fixtures.patches.collectForFrontier.mockClear(); + fixtures.patches.loadCheckpoint.mockClear(); + fixtures.stateCache.getExact.mockClear(); + fixtures.stateCache.getBestCompatiblePredecessor.mockClear(); + fixtures.stateCache.put.mockClear(); + fixtures.openStateSession.mockClear(); + fixtures.deps.crypto.hash.mockClear(); + + const resolution = await new MaterializeController(fixtures.deps) + .resolveLiveMaterialization(); + + expect(resolution.source).toBe("retained"); + expect(resolution.replayedPatchCount).toBe(0); + expect(resolution.materialization?.bundle.equals(cold.materialization?.bundle)).toBe(true); + expect(fixtures.patches.getFrontier).toHaveBeenCalledOnce(); + expect(fixtures.materializations.workspaces).toHaveLength(workspaceCount); + expect(fixtures.patches.collectForFrontier).not.toHaveBeenCalled(); + expect(fixtures.patches.loadCheckpoint).not.toHaveBeenCalled(); + expect(fixtures.stateCache.getExact).not.toHaveBeenCalled(); + expect(fixtures.stateCache.getBestCompatiblePredecessor).not.toHaveBeenCalled(); + expect(fixtures.stateCache.put).not.toHaveBeenCalled(); + expect(fixtures.openStateSession).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(); + expect(fixtures.materializations.acquisitions[0]?.released).toBe(true); + }); + + it("creates and returns a retained handle after a live-coordinate miss", async () => { + const fixtures = createControllerFixtures(); + fixtures.patches.collectForFrontier.mockResolvedValue([ + nodeAddPatchRecord({ + writer: "writer-1", + lamport: 1, + sha: "a1b2", + node: "node:created", + }), + ]); + + const resolution = await fixtures.controller.resolveLiveMaterialization(); + + expect(resolution.source).toBe("materialized"); + expect(resolution.replayedPatchCount).toBe(1); + expect(resolution.materialization).not.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.materializations.acquisitions).toHaveLength(1); + expect(fixtures.materializations.acquisitions[0]?.released).toBe(false); + await resolution.release(); + expect(fixtures.materializations.acquisitions[0]?.released).toBe(true); + }); + + it.each(["coordinate", "stateHash", "bundle"] as const)( + "rejects a newly acquired handle whose %s changed", + async (changedField) => { + const fixtures = createControllerFixtures(); + fixtures.patches.collectForFrontier.mockResolvedValue([ + nodeAddPatchRecord({ + writer: "writer-1", + lamport: 1, + sha: "a1b2", + node: "node:created", + }), + ]); + const retain = fixtures.materializations.retain.bind(fixtures.materializations); + let retained: MaterializationHandle | null = null; + const mismatchedAcquisitions: InMemoryMaterializationAcquisition[] = []; + vi.spyOn(fixtures.materializations, "retain").mockImplementation(async (request) => { + retained = await retain(request); + return retained; + }); + vi.spyOn(fixtures.materializations, "acquireExact").mockImplementation(() => { + if (retained === null) { + return Promise.resolve(null); + } + return createMismatchedAcquisition( + retained, + changedField, + retain, + mismatchedAcquisitions, + ); + }); + + await expect(fixtures.controller.resolveLiveMaterialization()).rejects.toMatchObject({ + code: "E_MATERIALIZATION_RESUME", + }); + + expect(mismatchedAcquisitions[0]?.released).toBe(true); + expect(mismatchedAcquisitions[0]?.releaseCalls).toBe(1); + }, + ); + + it("fails closed without publishing a snapshot when a non-empty coordinate has no patches", async () => { + const fixtures = createControllerFixtures(); + + await expect(fixtures.controller.resolveLiveMaterialization()).rejects.toMatchObject({ + code: "E_MATERIALIZATION_RESUME", + }); + + expect(fixtures.stateCache.put).not.toHaveBeenCalled(); + expect(fixtures.materializations.retainedRequests).toHaveLength(0); + expect(fixtures.materializations.workspaces[0]?.released).toBe(true); + }); + + it("rejects a retained handle for a different coordinate", 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(); + if (cold.materialization === undefined) { + throw new Error("Cold materialization did not retain a handle"); + } + fixtures.patches.getFrontier.mockResolvedValue(new Map([["writer-1", "tip-2"]])); + const acquisition = new InMemoryMaterializationAcquisition(cold.materialization); + vi.spyOn(fixtures.materializations, "acquireExact").mockResolvedValue(acquisition); + + await expect(fixtures.controller.resolveLiveMaterialization()).rejects.toMatchObject({ + code: "E_MATERIALIZATION_RESUME", + }); + + expect(fixtures.patches.collectForFrontier).toHaveBeenCalledTimes(1); + expect(fixtures.openStateSession).toHaveBeenCalledTimes(1); + expect(acquisition.released).toBe(true); + }); + it("hydrates a predecessor snapshot into StateSession before replaying the suffix", async () => { const { controller, stateCache, patches, openStateSession } = createControllerFixtures(); const target: Coordinate = { diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts index e3866349..1635e2a6 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 type MaterializationHandle from '../../../../src/domain/materialization/MaterializationHandle.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'; @@ -37,13 +38,14 @@ describe('GitCasMaterializationStoreAdapter', () => { roots, stateHash: 'state-hash', }); - const resolved = await harness.adapter.findExact(new MaterializationCoordinate({ + const acquisition = await harness.adapter.acquireExact(new MaterializationCoordinate({ frontier: new Map([ ['writer-b', 'patch-b'], ['writer-a', 'patch-a'], ]), ceiling: 12, })); + const resolved = acquisition?.materialization ?? null; expect(retained.laneName).toBe('events'); expect(retained.retention).toMatchObject({ @@ -60,6 +62,10 @@ describe('GitCasMaterializationStoreAdapter', () => { expect(resolved?.stateHash).toBe('state-hash'); expect(resolved?.roots.entries().map(([name, root]) => rootSignature(name, root))) .toEqual(roots.entries().map(([name, root]) => rootSignature(name, root))); + expect(resolved?.retention).toMatchObject({ + policy: 'pinned', + reachability: 'anchored', + }); const members = harness.cas.readBundleMembers(retained.bundle.toString()); expect(members.map(([path]) => path)).toEqual(['meta/descriptor', ...ROOT_PATHS]); @@ -67,11 +73,15 @@ describe('GitCasMaterializationStoreAdapter', () => { expect(cacheKeys).toHaveLength(1); expect(cacheKeys[0]).toMatch(/^v2:[0-9a-f]{64}$/u); expect(cacheKeys[0]?.length).toBeLessThan(1024); + expect(harness.cas.readActiveCacheAcquisitionCount()).toBe(1); + await acquisition?.release(); + await acquisition?.release(); + expect(harness.cas.readActiveCacheAcquisitionCount()).toBe(0); }); it('returns null for a coordinate with no retained materialization', async () => { const harness = await createHarness(); - expect(await harness.adapter.findExact(exactCoordinate())).toBeNull(); + expect(await harness.adapter.acquireExact(exactCoordinate())).toBeNull(); }); it('promotes terminal roots without installing an unnecessary workspace entry', async () => { @@ -91,7 +101,8 @@ describe('GitCasMaterializationStoreAdapter', () => { expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toEqual([]); expect(harness.cas.readCacheKeys(CACHE_NAMESPACE)).toHaveLength(1); - expect((await harness.adapter.findExact(coordinate))?.bundle.equals(promoted.bundle)).toBe(true); + expect((await acquireAndRelease(harness.adapter, coordinate))?.bundle.equals(promoted.bundle)) + .toBe(true); }); it('removes an in-progress coordinate after mismatched promotion fails', async () => { @@ -134,7 +145,7 @@ describe('GitCasMaterializationStoreAdapter', () => { roots, stateHash: 'partial-state-hash', }); - const resolved = await harness.adapter.findExact(exactCoordinate()); + const resolved = await acquireAndRelease(harness.adapter, exactCoordinate()); expect(harness.cas.readBundleMembers(retained.bundle.toString()).map(([path]) => path)) .toEqual(['meta/descriptor', 'roots/node-alive']); @@ -152,7 +163,7 @@ describe('GitCasMaterializationStoreAdapter', () => { roots: await createRoots(harness.cas), stateHash: 'empty-state-hash', }); - expect((await harness.adapter.findExact(coordinate))?.coordinate.ceiling).toBeNull(); + expect((await acquireAndRelease(harness.adapter, coordinate))?.coordinate.ceiling).toBeNull(); }); it('fails closed when git-cas declines materialization retention', async () => { @@ -213,10 +224,11 @@ describe('GitCasMaterializationStoreAdapter', () => { const cache = await harness.cas.caches.open({ namespace: CACHE_NAMESPACE }); await cache.put(cacheKey, page.handle); - await expect(harness.adapter.findExact(coordinate)).rejects.toMatchObject({ + await expect(harness.adapter.acquireExact(coordinate)).rejects.toMatchObject({ code: 'E_MATERIALIZATION_STORAGE', message: expect.stringContaining('does not reference a materialization bundle'), }); + expect(harness.cas.readActiveCacheAcquisitionCount()).toBe(0); }); it.each([ @@ -251,7 +263,7 @@ describe('GitCasMaterializationStoreAdapter', () => { const members = harness.cas.readBundleMembers(retained.bundle.toString()); harness.cas.replaceBundleMembers(retained.bundle.toString(), mutate(members)); - await expect(harness.adapter.findExact(coordinate)).rejects.toMatchObject({ + await expect(harness.adapter.acquireExact(coordinate)).rejects.toMatchObject({ code: 'E_MATERIALIZATION_STORAGE', }); }); @@ -271,7 +283,7 @@ describe('GitCasMaterializationStoreAdapter', () => { replaceMember(members, 'roots/edge-alive', page.handle.toString()), ); - await expect(harness.adapter.findExact(coordinate)).rejects.toMatchObject({ + await expect(harness.adapter.acquireExact(coordinate)).rejects.toMatchObject({ code: 'E_MATERIALIZATION_STORAGE', message: expect.stringContaining('edge-alive root bundle'), }); @@ -351,7 +363,7 @@ describe('GitCasMaterializationStoreAdapter', () => { ])('rejects an invalid %s', async (_case, value, message) => { const harness = await retainedHarness(); replaceDescriptor(harness, value); - await expect(harness.adapter.findExact(harness.coordinate)).rejects.toMatchObject({ + await expect(harness.adapter.acquireExact(harness.coordinate)).rejects.toMatchObject({ code: 'E_MATERIALIZATION_STORAGE', message: expect.stringContaining(message), }); @@ -360,7 +372,7 @@ describe('GitCasMaterializationStoreAdapter', () => { it('rejects a descriptor for another lane', async () => { const harness = await retainedHarness(); replaceDescriptor(harness, descriptor({ laneName: 'other' })); - await expect(harness.adapter.findExact(harness.coordinate)).rejects.toMatchObject({ + await expect(harness.adapter.acquireExact(harness.coordinate)).rejects.toMatchObject({ code: 'E_MATERIALIZATION_STORAGE', message: expect.stringContaining('belongs to another lane'), }); @@ -371,7 +383,7 @@ describe('GitCasMaterializationStoreAdapter', () => { replaceDescriptor(harness, descriptor({ coordinate: { ceiling: 99, frontier: [] }, })); - await expect(harness.adapter.findExact(harness.coordinate)).rejects.toMatchObject({ + await expect(harness.adapter.acquireExact(harness.coordinate)).rejects.toMatchObject({ code: 'E_MATERIALIZATION_STORAGE', message: expect.stringContaining('does not match its cache key'), }); @@ -382,7 +394,7 @@ describe('GitCasMaterializationStoreAdapter', () => { replaceDescriptor(harness, descriptor({ roots: replaceRootStatus(rootStatusFixture(), 'edge-alive', 'empty'), })); - await expect(harness.adapter.findExact(harness.coordinate)).rejects.toMatchObject({ + await expect(harness.adapter.acquireExact(harness.coordinate)).rejects.toMatchObject({ code: 'E_MATERIALIZATION_STORAGE', message: expect.stringContaining('unexpected edge-alive root bundle'), }); @@ -395,7 +407,7 @@ describe('GitCasMaterializationStoreAdapter', () => { 'meta/descriptor', ); harness.cas.replaceStoredPage(descriptorHandle, new Uint8Array(1024 * 1024 + 1)); - await expect(harness.adapter.findExact(harness.coordinate)).rejects.toMatchObject({ + await expect(harness.adapter.acquireExact(harness.coordinate)).rejects.toMatchObject({ code: 'PAGE_TOO_LARGE', }); }); @@ -434,7 +446,7 @@ describe('GitCasMaterializationStoreAdapter', () => { roots, stateHash: '', })).rejects.toMatchObject({ code: 'E_MATERIALIZATION_STORAGE' }); - await expect(Reflect.apply(harness.adapter.findExact, harness.adapter, [{ + await expect(Reflect.apply(harness.adapter.acquireExact, harness.adapter, [{ frontier: new Map(), ceiling: null, }])).rejects.toMatchObject({ code: 'E_MATERIALIZATION_STORAGE' }); @@ -463,6 +475,21 @@ type RetainedHarness = Harness & Readonly<{ retainedBundle: BundleHandle; }>; +async function acquireAndRelease( + adapter: GitCasMaterializationStoreAdapter, + coordinate: MaterializationCoordinate, +): Promise { + const acquisition = await adapter.acquireExact(coordinate); + if (acquisition === null) { + return null; + } + try { + return acquisition.materialization; + } finally { + await acquisition.release(); + } +} + async function createHarness(laneName = 'events'): Promise { const history = new InMemoryGraphAdapter(); const cas = new InMemoryGitCasFacade({ @@ -512,7 +539,7 @@ function withCacheResult( open: async (options) => { const cache = await cas.caches.open(options); return { - get: async (key) => await cache.get(key), + acquire: async (key) => await cache.acquire(key), put: async (key, handle, entryOptions) => rewrite( await cache.put(key, handle, entryOptions), ), diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationWorkspace.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationWorkspace.test.ts index 0b94afdc..662c1f19 100644 --- a/test/unit/infrastructure/adapters/GitCasMaterializationWorkspace.test.ts +++ b/test/unit/infrastructure/adapters/GitCasMaterializationWorkspace.test.ts @@ -436,7 +436,7 @@ function withCacheResult( open: async (options) => { const cache = await cas.caches.open(options); return { - get: async (key) => await cache.get(key), + acquire: async (key) => await cache.acquire(key), put: async (key, handle, entryOptions) => rewrite( await cache.put(key, handle, entryOptions), ), diff --git a/test/unit/scripts/v18-v17-public-read-legacy-reading-builder.test.ts b/test/unit/scripts/v18-v17-public-read-legacy-reading-builder.test.ts index 58edfa62..56d833f7 100644 --- a/test/unit/scripts/v18-v17-public-read-legacy-reading-builder.test.ts +++ b/test/unit/scripts/v18-v17-public-read-legacy-reading-builder.test.ts @@ -46,8 +46,10 @@ describe('v18 v17 public-read legacy reading builder', () => { 'alice', 'alice', ]); + // git-cas package versions are stamped into asset manifests, so dependency bumps + // intentionally advance this migration-reading golden handle. expect(reading.facts.find((fact) => fact.factKey === 'node:alpha:_content')?.value) - .toBe('git-cas:1:asset:manifest-tree:cbor:sha1:92c0f5ef3874549768bceef59319608437cfc926'); + .toBe('git-cas:1:asset:manifest-tree:cbor:sha1:5453687e3d0e74b5fbf43f347a4403824a52a697'); }); it('fails closed when a restored v17 writer ref drifts after restore', async () => {