From 960159a02404fae2dfffaa13f4ac34ab72e76f09 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 16 Jul 2026 13:22:59 -0700 Subject: [PATCH 1/2] fix(materialization): retain bounded trie roots --- CHANGELOG.md | 26 +- src/domain/orset/session/StateSession.ts | 227 +++++++- src/domain/orset/shadow/ShadowTrieORSet.ts | 16 +- src/domain/orset/trie/TrieCursor.ts | 23 +- .../controllers/MaterializeController.ts | 113 ++-- .../MaterializeCoordinateStrategy.ts | 3 + .../controllers/MaterializeLiveStrategy.ts | 83 +-- .../controllers/MaterializeSessionBridge.ts | 113 ++-- .../controllers/MaterializeStrategyRuntime.ts | 7 +- src/domain/warp/RuntimeHostBoot.ts | 6 +- .../GitCasMaterializationStoreAdapter.ts | 23 +- .../GitCasMaterializationWorkspace.ts | 371 +++++++++++++ src/ports/MaterializationStorePort.ts | 13 +- src/ports/MaterializationWorkspacePort.ts | 31 ++ test/helpers/InMemoryGitCasFacade.ts | 22 +- test/helpers/InMemoryMaterializationStore.ts | 66 +++ .../materialization.retainedResume.test.ts | 46 ++ ...ializationStoreAdapter.integration.test.ts | 130 ++++- .../domain/WarpGraph.autoCheckpoint.test.ts | 21 +- test/unit/domain/WarpGraph.patchCount.test.ts | 14 +- test/unit/domain/WarpGraph.test.ts | 10 +- .../domain/WarpGraph.versionVector.test.ts | 16 +- .../domain/orset/session/StateSession.test.ts | 210 +++++++- .../unit/domain/orset/trie/TrieCursor.test.ts | 6 + ...MaterializeController.stateSession.test.ts | 112 +++- .../controllers/MaterializeController.test.ts | 37 +- .../MaterializePatchStreamReducer.test.ts | 23 +- .../hydrateCheckpointIndex.regression.test.ts | 7 +- .../GitCasMaterializationStoreAdapter.test.ts | 49 ++ .../GitCasMaterializationWorkspace.test.ts | 492 ++++++++++++++++++ ...-graph-model-migration-command-cli.test.ts | 40 +- 31 files changed, 2122 insertions(+), 234 deletions(-) create mode 100644 src/infrastructure/adapters/GitCasMaterializationWorkspace.ts create mode 100644 src/ports/MaterializationWorkspacePort.ts create mode 100644 test/unit/infrastructure/adapters/GitCasMaterializationWorkspace.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 5368172ff..17b3bfd36 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -50,12 +50,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Raised the coverage ratchet from `92.10%` to `92.62%` after adding targeted coverage for bounded query node paging and memory-budget rejection paths and removing retired compatibility surfaces. -- Upgraded `@git-stunts/git-cas` to `^6.1.0` so Git-backed state caches can use - the library's crash-safe `RootSet` retention API. +- Upgraded `@git-stunts/git-cas` to `^6.2.0` so Git-backed materializations can + use managed `CacheSet` retention and opaque page and bundle handles. - Moved shadow-trie leaf and branch storage from unretained raw Git blobs and trees to bounded git-cas pages and composable bundle handles. Production storage now shares one git-cas facade with the trie adapter, and the storage ownership gate rejects direct raw Git object writers. +- Bounded state-session dirty-page residency with periodic immutable root + checkpoints. In-progress roots are retained only through a renewable, + expiring git-cas `CacheSet` workspace, while the page memo is + operation-local instead of shared across materializations. This bounds + mutated-page residency only; removal of whole-state projection remains + tracked by #738. ### Removed @@ -70,8 +76,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 descriptors through the `git-cas` cache API. Repeated materialization, including through a fresh runtime adapter, resumes the retained node/edge trie roots without replaying writer patches already covered by the snapshot. -- Failed state-session reductions no longer flush partial trie roots into - unretained CAS objects; roots are flushed only after successful projection. +- Failed state-session reductions release their git-cas-managed workspace; + successful reductions retain the final materialization before releasing the + workspace. No in-progress root is left as an abandoned WARP-owned cache. +- Intermediate state-session flushes require a pinned workspace witness before + accepting and forgetting dirty pages. Terminal pages remain dirty until the + final materialization returns an anchored retention witness, avoiding a + redundant terminal workspace publication and cleanup mutation for ordinary + reads. Workspace leases continue renewing during long projection and final + retention. Live materialization resolves its coordinate independently of the + legacy snapshot cache, so disabling that cache cannot bypass final root + retention. +- Reopened trie inserts now propagate a changed descendant handle through + every ancestor. Repeated bounded checkpoints no longer lose untouched + sibling branches after a deep leaf update or split. - Git-backed state-cache payload trees are now anchored through a graph-scoped `git-cas` RootSet before their index record is published, then reconciled after publication so live cache entries remain reachable across Git garbage diff --git a/src/domain/orset/session/StateSession.ts b/src/domain/orset/session/StateSession.ts index fa9aaae1e..213fd9afe 100644 --- a/src/domain/orset/session/StateSession.ts +++ b/src/domain/orset/session/StateSession.ts @@ -3,16 +3,19 @@ import type VersionVector from "../../crdt/VersionVector.ts"; import StateSessionError from "../../errors/StateSessionError.ts"; import type ORSetElementState from "../ORSetElementState.ts"; import type CodecPort from "../../../ports/CodecPort.ts"; +import type MaterializationWorkspacePort from "../../../ports/MaterializationWorkspacePort.ts"; +import StorageRetentionWitness from "../../storage/StorageRetentionWitness.ts"; import type TrieStorePort from "../trie/TrieStorePort.ts"; import PageCache from "../trie/PageCache.ts"; import TrieCursor from "../trie/TrieCursor.ts"; import TrieFlusher from "../trie/TrieFlusher.ts"; import TrieGeometry from "../trie/TrieGeometry.ts"; +import type FlushResult from "../trie/FlushResult.ts"; import ShadowTrieORSet from "../shadow/ShadowTrieORSet.ts"; import StateSessionCloseResult from "./StateSessionCloseResult.ts"; -export type StateSessionOpen = { +type StateSessionDependencies = { readonly nodeAliveRootOid: string | null; readonly edgeAliveRootOid: string | null; readonly store: TrieStorePort; @@ -21,17 +24,53 @@ export type StateSessionOpen = { readonly pageCache: PageCache; }; +type BoundedStateSessionOpen = Readonly<{ + workspace: MaterializationWorkspacePort; + maxDirtyPages?: number; +}>; + +type DiagnosticStateSessionOpen = Readonly<{ + workspace?: undefined; + maxDirtyPages?: undefined; +}>; + +export type StateSessionOpen = StateSessionDependencies & ( + | BoundedStateSessionOpen + | DiagnosticStateSessionOpen +); + +export type StateSessionPreparedClose = Readonly<{ + roots: StateSessionCloseResult; + accept(witness: StorageRetentionWitness | null): void; +}>; + +type PreparedSessionFlush = Readonly<{ + node: FlushResult; + edge: FlushResult; + roots: StateSessionCloseResult; +}>; + +const DEFAULT_MAX_DIRTY_PAGES = 1024; + export default class StateSession { readonly #nodeAlive: ShadowTrieORSet; readonly #edgeAlive: ShadowTrieORSet; + readonly #workspace: MaterializationWorkspacePort | null; + readonly #maxDirtyPages: number; + #workspaceCheckpointPending = false; + #closePrepared = false; #closed = false; constructor(fields: { readonly nodeAlive: ShadowTrieORSet; readonly edgeAlive: ShadowTrieORSet; + readonly workspace?: MaterializationWorkspacePort; + readonly maxDirtyPages: number; }) { this.#nodeAlive = fields.nodeAlive; this.#edgeAlive = fields.edgeAlive; + this.#workspace = fields.workspace ?? null; + this.#maxDirtyPages = fields.maxDirtyPages; Object.freeze(this); } @@ -42,6 +81,8 @@ export default class StateSession { validateCodec(init.codec); validateGeometry(init.geometry); validatePageCache(init.pageCache); + validateWorkspace(init.workspace); + const maxDirtyPages = normalizeMaxDirtyPages(init.maxDirtyPages, init.workspace); const nodeCursor = new TrieCursor({ rootOid: init.nodeAliveRootOid, @@ -75,6 +116,8 @@ export default class StateSession { cursor: edgeCursor, flusher: edgeFlusher, }), + ...(init.workspace === undefined ? {} : { workspace: init.workspace }), + maxDirtyPages, }); } @@ -111,21 +154,25 @@ export default class StateSession { async addNode(id: string, dot: Dot): Promise { this.#assertOpen(); await this.#nodeAlive.add(id, dot); + await this.#checkpointIfNeeded(); } async addEdge(key: string, dot: Dot): Promise { this.#assertOpen(); await this.#edgeAlive.add(key, dot); + await this.#checkpointIfNeeded(); } async removeNode(id: string, observedDots: ReadonlySet): Promise { this.#assertOpen(); await this.#nodeAlive.removeElement(id, observedDots); + await this.#checkpointIfNeeded(); } async removeEdge(key: string, observedDots: ReadonlySet): Promise { this.#assertOpen(); await this.#edgeAlive.removeElement(key, observedDots); + await this.#checkpointIfNeeded(); } scanNodes(): AsyncIterable { @@ -150,23 +197,103 @@ export default class StateSession { async compact(includedVV: VersionVector): Promise { this.#assertOpen(); + if (this.#workspace !== null) { + throw new StateSessionError( + "Bounded StateSession compaction requires resumable subtree checkpoints", + { code: "E_STATE_SESSION_INPUT" }, + ); + } await this.#nodeAlive.compact(includedVV); await this.#edgeAlive.compact(includedVV); + await this.#checkpointIfNeeded(); + } + + /** Mutated pages awaiting flush; this is not a total resident-memory metric. */ + dirtyPageCount(): number { + return this.#nodeAlive.dirtyPageCount() + this.#edgeAlive.dirtyPageCount(); } async close(): Promise { this.#assertOpen(); - const nodeFlush = await this.#nodeAlive.flush(); - const edgeFlush = await this.#edgeAlive.flush(); + const roots = await this.#flushAndRetain(); this.#closed = true; - return new StateSessionCloseResult({ + return roots; + } + + async prepareClose(): Promise { + this.#assertOpen(); + this.#closePrepared = true; + try { + const prepared = await this.#prepareFlush(); + let accepted = false; + return Object.freeze({ + roots: prepared.roots, + accept: (witness: StorageRetentionWitness | null) => { + if (accepted) { + throw new StateSessionError( + "StateSession prepared close was already accepted", + { code: "E_STATE_SESSION_CLOSED" }, + ); + } + requireFinalRetentionWitness(prepared.roots, witness); + this.#acceptFlush(prepared); + accepted = true; + this.#closed = true; + }, + }); + } catch (raw) { + this.#closePrepared = false; + throw raw; + } + } + + async #checkpointIfNeeded(): Promise { + if ( + !this.#workspaceCheckpointPending && + this.dirtyPageCount() < this.#maxDirtyPages + ) { + return; + } + await this.#flushAndRetain(); + } + + async #flushAndRetain(): Promise { + const prepared = await this.#prepareFlush(); + if (this.#workspaceCheckpointPending && this.#workspace !== null) { + const witness = await this.#workspace.checkpoint({ + nodeAliveRoot: prepared.roots.nodeAliveRootOid, + edgeAliveRoot: prepared.roots.edgeAliveRootOid, + }); + requireWorkspaceWitness(prepared.roots, witness); + } + this.#acceptFlush(prepared); + return prepared.roots; + } + + async #prepareFlush(): Promise { + const nodeFlush = await this.#nodeAlive.prepareFlush(); + if (!nodeFlush.isClean()) { + this.#workspaceCheckpointPending = true; + } + const edgeFlush = await this.#edgeAlive.prepareFlush(); + if (!edgeFlush.isClean()) { + this.#workspaceCheckpointPending = true; + } + const roots = new StateSessionCloseResult({ nodeAliveRootOid: nodeFlush.rootOid, edgeAliveRootOid: edgeFlush.rootOid, }); + return Object.freeze({ node: nodeFlush, edge: edgeFlush, roots }); + } + + #acceptFlush(prepared: PreparedSessionFlush): void { + this.#nodeAlive.acceptFlush(prepared.node); + this.#edgeAlive.acceptFlush(prepared.edge); + this.#workspaceCheckpointPending = false; } #assertOpen(): void { - if (this.#closed) { + if (this.#closed || this.#closePrepared) { throw new StateSessionError( "StateSession is closed", { code: "E_STATE_SESSION_CLOSED" }, @@ -245,3 +372,93 @@ function validatePageCache(pageCache: PageCache): void { ); } } + +function validateWorkspace(workspace: MaterializationWorkspacePort | undefined): void { + if (workspace === undefined) { + return; + } + if ( + typeof workspace.checkpoint !== "function" || + typeof workspace.promote !== "function" || + typeof workspace.release !== "function" + ) { + throw new StateSessionError( + "StateSession workspace must provide checkpoint/promote/release methods", + { + code: "E_STATE_SESSION_INPUT", + context: { field: "workspace" }, + }, + ); + } +} + +function normalizeMaxDirtyPages( + value: number | undefined, + workspace: MaterializationWorkspacePort | undefined, +): number { + if (workspace === undefined && value !== undefined) { + throw new StateSessionError( + "StateSession maxDirtyPages requires a materialization workspace", + { + code: "E_STATE_SESSION_INPUT", + context: { field: "maxDirtyPages" }, + }, + ); + } + const normalized = value ?? (workspace === undefined + ? Number.MAX_SAFE_INTEGER + : DEFAULT_MAX_DIRTY_PAGES); + if (!Number.isSafeInteger(normalized) || normalized <= 0) { + throw new StateSessionError( + `StateSession maxDirtyPages must be a positive safe integer; received ${String(normalized)}`, + { + code: "E_STATE_SESSION_INPUT", + context: { field: "maxDirtyPages", value: normalized }, + }, + ); + } + return normalized; +} + +function requireWorkspaceWitness( + roots: StateSessionCloseResult, + witness: Awaited>, +): void { + if ( + (roots.nodeAliveRootOid !== null || roots.edgeAliveRootOid !== null) && + !isPinnedWorkspaceWitness(witness) + ) { + throw new StateSessionError( + "StateSession workspace did not witness retained non-empty roots", + { code: "E_STATE_SESSION_STRUCTURE" }, + ); + } +} + +function isPinnedWorkspaceWitness( + witness: Awaited>, +): witness is StorageRetentionWitness { + return witness instanceof StorageRetentionWitness && + witness.policy === "pinned" && + witness.reachability === "anchored" && + witness.root.kind === "cache-set"; +} + +function requireFinalRetentionWitness( + roots: StateSessionCloseResult, + witness: StorageRetentionWitness | null, +): void { + if (roots.nodeAliveRootOid === null && roots.edgeAliveRootOid === null) { + return; + } + if ( + !(witness instanceof StorageRetentionWitness) || + witness.reachability !== "anchored" || + witness.root.kind !== "cache-set" + ) { + throw new StateSessionError( + "StateSession final materialization did not witness retained non-empty roots", + { code: "E_STATE_SESSION_STRUCTURE" }, + ); + } +} diff --git a/src/domain/orset/shadow/ShadowTrieORSet.ts b/src/domain/orset/shadow/ShadowTrieORSet.ts index ce61cb198..24fcdf6eb 100644 --- a/src/domain/orset/shadow/ShadowTrieORSet.ts +++ b/src/domain/orset/shadow/ShadowTrieORSet.ts @@ -76,7 +76,21 @@ export default class ShadowTrieORSet { yield* this.#cursor.scanElementStates(); } - async flush(): Promise { + dirtyPageCount(): number { + return this.#cursor.dirtyPageCount(); + } + + async prepareFlush(): Promise { return await this.#flusher.flush(this.#cursor.snapshot()); } + + acceptFlush(result: FlushResult): void { + this.#cursor.rebase(result.rootOid); + } + + async flush(): Promise { + const result = await this.prepareFlush(); + this.acceptFlush(result); + return result; + } } diff --git a/src/domain/orset/trie/TrieCursor.ts b/src/domain/orset/trie/TrieCursor.ts index 8ee2a18f7..ad9b2270d 100644 --- a/src/domain/orset/trie/TrieCursor.ts +++ b/src/domain/orset/trie/TrieCursor.ts @@ -67,7 +67,7 @@ interface SplitContext { * contract, split semantics, and error codes. */ export default class TrieCursor { - readonly #initialRootOid: string | null; + #initialRootOid: string | null; readonly #store: TrieStorePort; readonly #geometry: TrieGeometry; readonly #codec: CodecPort; @@ -194,6 +194,26 @@ export default class TrieCursor { }); } + dirtyPageCount(): number { + return this.#dirtyLeaves.size + this.#dirtyBranches.size; + } + + rebase(rootOid: string | null): void { + if (rootOid !== null && (typeof rootOid !== "string" || rootOid.length === 0)) { + throw new TrieCursorError( + `TrieCursor rebase root must be null or a non-empty string; received ${String(rootOid)}`, + { code: "E_TRIE_CURSOR_STRUCTURE" }, + ); + } + this.#initialRootOid = rootOid; + this.#dirtyLeaves.clear(); + this.#dirtyBranches.clear(); + this.#cleanChildren.clear(); + this.#workingLeaves.clear(); + this.#workingBranches.clear(); + this.#rootLoaded = false; + } + // -- construction helpers -------------------------------------------------- #makeInsertContext(element: string, dot: Dot): InsertContext { @@ -571,6 +591,7 @@ export default class TrieCursor { } const nextCtx = advanceCtx(args.ctx); await this.#insertBelowBranch(childPath, nextCtx); + this.#rebindParentBranch(args.parentPath, args.nibble); } #requireBranchAt(path: readonly number[]): TrieBranch { diff --git a/src/domain/services/controllers/MaterializeController.ts b/src/domain/services/controllers/MaterializeController.ts index 5b680f7d4..7d9800325 100644 --- a/src/domain/services/controllers/MaterializeController.ts +++ b/src/domain/services/controllers/MaterializeController.ts @@ -9,7 +9,6 @@ * Dependencies are constructor-injected. No host bag. * Side effects (notification, GC, auto-checkpoint) are the caller's job. */ - import { reducePatches as reduceJoinedPatches, createEmptyState } from '../JoinReducer.ts'; import { ProvenanceIndex } from '../provenance/ProvenanceIndex.ts'; import { computeStateHash } from '../state/StateSerializer.ts'; @@ -42,9 +41,13 @@ import type LoggerPort from '../../../ports/LoggerPort.ts'; import type CodecPort from '../../../ports/CodecPort.ts'; import type CryptoPort from '../../../ports/CryptoPort.ts'; import type CheckpointStorePort from '../../../ports/CheckpointStorePort.ts'; -import type WarpStateCachePort from '../../../ports/WarpStateCachePort.ts'; +import type { + default as WarpStateCachePort, + WarpStateCoordinate, + WarpStateSnapshotProvenancePosture, +} from '../../../ports/WarpStateCachePort.ts'; import type MaterializationStorePort from '../../../ports/MaterializationStorePort.ts'; -import type { WarpStateSnapshotProvenancePosture } from '../../../ports/WarpStateCachePort.ts'; +import type MaterializationWorkspacePort from '../../../ports/MaterializationWorkspacePort.ts'; import type PatchCollector from '../../capabilities/PatchCollector.ts'; import type { PatchWithSha } from '../../capabilities/PatchCollector.ts'; import type DetachedGraphFactory from '../../capabilities/DetachedGraphFactory.ts'; @@ -56,13 +59,13 @@ import AdjacencyMap from '../../capabilities/AdjacencyMap.ts'; import MaterializationCoordinate from '../../materialization/MaterializationCoordinate.ts'; import type MaterializationHandle from '../../materialization/MaterializationHandle.ts'; import type MaterializationRoots from '../../materialization/MaterializationRoots.ts'; +import type StorageRetentionWitness from '../../storage/StorageRetentionWitness.ts'; import WarpError from '../../errors/WarpError.ts'; import type { UsableSnapshotRecord } from './MaterializeSnapshotCacheResult.ts'; import type { MaterializeResultBuildInput, MaterializeStrategyRuntime, } from './MaterializeStrategyRuntime.ts'; - export type MaterializePersistence = { readRef(ref: string): Promise; }; @@ -105,15 +108,15 @@ export type MaterializeResult = { // ── Reduce helpers ────────────────────────────────────────────────── type ReducerInput = Parameters[0]; - function toReducerInput(patches: PatchWithSha[]): ReducerInput { return patches as ReducerInput; } - export type MaterializeReduceOutput = { state: WarpState; adjacency?: MaterializeAdjacency; roots?: MaterializationRoots; + workspace?: MaterializationWorkspacePort; + acceptMaterialization?: (witness: StorageRetentionWitness | null) => void; receipts?: TickReceipt[]; diff?: PatchDiff; }; @@ -126,7 +129,6 @@ function reduceWithReceipts(patches: PatchWithSha[], base?: WarpState): Material ) as { state: WarpState; receipts: TickReceipt[] }; return { state: r.state, receipts: r.receipts }; } - function reduceWithDiff(patches: PatchWithSha[], base?: WarpState): MaterializeReduceOutput { const r = reduceJoinedPatches( toReducerInput(patches), @@ -135,7 +137,6 @@ function reduceWithDiff(patches: PatchWithSha[], base?: WarpState): MaterializeR ) as { state: WarpState; diff: PatchDiff }; return { state: r.state, diff: r.diff }; } - function reducePlain(patches: PatchWithSha[], base?: WarpState): MaterializeReduceOutput { return { state: reduceJoinedPatches(toReducerInput(patches), base) }; } @@ -263,34 +264,38 @@ export default class MaterializeController { } private async _buildResult(params: MaterializeResultBuildInput): Promise { - const stateHash = await computeHash(this._deps, params.reduced.state); - const adjacency = params.reduced.adjacency ?? buildAdjacency(params.reduced.state); - const materialization = await this._resolveMaterialization(params, stateHash); - if (params.reduced.receipts === undefined && params.publishSnapshot !== false) { - await this._publishSnapshot({ + try { + const stateHash = await computeHash(this._deps, params.reduced.state); + const adjacency = params.reduced.adjacency ?? buildAdjacency(params.reduced.state); + const materialization = await this._resolveMaterialization(params, stateHash); + if (params.reduced.receipts === undefined && params.publishSnapshot !== false) { + await this._publishSnapshot({ + state: params.reduced.state, + stateHash, + ceiling: params.ceiling, + frontier: params.frontier, + }); + } + return { state: params.reduced.state, stateHash, - ceiling: params.ceiling, + adjacency: new AdjacencyMap({ outgoing: adjacency.outgoing, incoming: adjacency.incoming }), + receipts: params.reduced.receipts, + diff: params.reduced.diff, + patchCount: params.summary.patchCount, + maxObservedLamport: Math.max( + params.summary.maxObservedLamport, + maxObservedLamportInState(params.reduced.state), + ), + provenanceIndex: params.summary.provenance, + provenanceDegraded: params.degraded, frontier: params.frontier, - }); + ceiling: params.ceiling, + ...(materialization === undefined ? {} : { materialization }), + }; + } finally { + await params.reduced.workspace?.release(); } - return { - state: params.reduced.state, - stateHash, - adjacency: new AdjacencyMap({ outgoing: adjacency.outgoing, incoming: adjacency.incoming }), - receipts: params.reduced.receipts, - diff: params.reduced.diff, - patchCount: params.summary.patchCount, - maxObservedLamport: Math.max( - params.summary.maxObservedLamport, - maxObservedLamportInState(params.reduced.state), - ), - provenanceIndex: params.summary.provenance, - provenanceDegraded: params.degraded, - frontier: params.frontier, - ceiling: params.ceiling, - ...(materialization === undefined ? {} : { materialization }), - }; } private async _resolveMaterialization( @@ -301,19 +306,46 @@ export default class MaterializeController { if (params.materialization.stateHash !== stateHash) { throw materializationResumeError('retained handle state hash does not match resumed state'); } + params.reduced.acceptMaterialization?.(params.materialization.retention); return params.materialization; } if (params.reduced.roots === undefined || params.frontier === null) { + await this._acceptSessionWithoutMaterialization(params.reduced); return undefined; } - return await this._deps.materializations.retain({ + const request = { coordinate: new MaterializationCoordinate({ frontier: params.frontier, ceiling: params.ceiling, }), roots: params.reduced.roots, stateHash, + }; + const materialization = params.reduced.workspace === undefined + ? await this._deps.materializations.retain(request) + : await params.reduced.workspace.promote(request); + params.reduced.acceptMaterialization?.(materialization.retention); + return materialization; + } + + private async _acceptSessionWithoutMaterialization( + reduced: MaterializeReduceOutput, + ): Promise { + if (reduced.acceptMaterialization === undefined) { + return; + } + if (reduced.roots === undefined || reduced.workspace === undefined) { + throw materializationResumeError('prepared session is missing roots or workspace retention'); + } + const roots = materializationSessionOpen(reduced.roots); + if (roots === null) { + throw materializationResumeError('prepared session roots cannot be checkpointed'); + } + const witness = await reduced.workspace.checkpoint({ + nodeAliveRoot: roots.nodeAliveRootOid, + edgeAliveRoot: roots.edgeAliveRootOid, }); + reduced.acceptMaterialization(witness); } private async _resumeExactMaterialization( @@ -334,6 +366,8 @@ export default class MaterializeController { : materializationSessionOpen(retained.roots); const reduced = await reduceSessionBackedState({ openStateSession, + materializations: this._deps.materializations, + coordinate, patches: [], baseState: snapshot.state, ...(retainedRoots === null ? {} : { roots: retainedRoots }), @@ -357,6 +391,7 @@ export default class MaterializeController { patches: PatchWithSha[], base: WarpState | undefined, opts: { receipts: boolean; wantDiff: boolean }, + coordinate: WarpStateCoordinate, ): Promise { const {openStateSession} = this._deps; if (openStateSession === undefined) { @@ -364,6 +399,8 @@ export default class MaterializeController { } const sessionArgs = { openStateSession, + materializations: this._deps.materializations, + coordinate: new MaterializationCoordinate(coordinate), patches: patches.map((entry) => new PatchEntry(entry)), receipts: opts.receipts, wantDiff: opts.wantDiff, @@ -376,6 +413,7 @@ export default class MaterializeController { stream: AsyncIterable, base: WarpState | undefined, opts: MaterializePatchStreamOptions, + coordinate: WarpStateCoordinate, provenanceBase?: ProvenanceIndex, ): Promise { if (this._deps.openStateSession === undefined) { @@ -395,6 +433,8 @@ export default class MaterializeController { }; const reduced = await reduceSessionBackedState({ openStateSession: this._deps.openStateSession, + materializations: this._deps.materializations, + coordinate: new MaterializationCoordinate(coordinate), patches: recordingStream(), receipts: opts.receipts, wantDiff: opts.wantDiff, @@ -413,9 +453,10 @@ export default class MaterializeController { await this._emptyResult(ceiling, frontier, options), wrapState: async (state, ceiling, frontier, provenance, options) => await this._wrapState(state, ceiling, frontier, provenance, options), - reducePatches: async (patches, base, opts) => await this._reducePatches(patches, base, opts), - reducePatchStream: async (stream, base, opts, provenanceBase) => - await this._reducePatchStream(stream, base, opts, provenanceBase), + reducePatches: async (patches, base, opts, coordinate) => + await this._reducePatches(patches, base, opts, coordinate), + reducePatchStream: async (stream, base, opts, coordinate, provenanceBase) => + await this._reducePatchStream(stream, base, opts, coordinate, provenanceBase), buildResult: async (params) => await this._buildResult(params), resumeExactMaterialization: async (snapshot, options) => await this._resumeExactMaterialization(snapshot, options), diff --git a/src/domain/services/controllers/MaterializeCoordinateStrategy.ts b/src/domain/services/controllers/MaterializeCoordinateStrategy.ts index b6b90c231..89744bb1e 100644 --- a/src/domain/services/controllers/MaterializeCoordinateStrategy.ts +++ b/src/domain/services/controllers/MaterializeCoordinateStrategy.ts @@ -34,6 +34,7 @@ export default class MaterializeCoordinateStrategy { } const reduction = await this.reduceFrontierPatches(opts); if (reduction.summary.patchCount === 0) { + await reduction.reduced.workspace?.release(); return await this.emptyResult(opts); } return await this.runtime.buildResult({ @@ -61,6 +62,7 @@ export default class MaterializeCoordinateStrategy { receipts: opts.receipts, wantDiff: false, }, + { frontier: opts.frontier, ceiling: opts.ceiling }, ); } @@ -140,6 +142,7 @@ export default class MaterializeCoordinateStrategy { receipts: false, wantDiff: false, }, + opts.coordinate, ); return await this.runtime.buildResult({ reduced: reduction.reduced, diff --git a/src/domain/services/controllers/MaterializeLiveStrategy.ts b/src/domain/services/controllers/MaterializeLiveStrategy.ts index f2041c72b..27fb4bd88 100644 --- a/src/domain/services/controllers/MaterializeLiveStrategy.ts +++ b/src/domain/services/controllers/MaterializeLiveStrategy.ts @@ -30,41 +30,38 @@ export default class MaterializeLiveStrategy { } async materialize(opts: MaterializeLiveOptions): Promise { - const stateCache = this.runtime.deps.getStateCache?.() ?? null; - if (stateCache !== null) { - return await this.materializeWithStateCache(stateCache, opts); - } - return await this.materializeWithoutStateCache(opts); - } - - private async materializeWithoutStateCache(opts: MaterializeLiveOptions): Promise { - const checkpoint = await this.runtime.deps.patches.loadCheckpoint(); - if (checkpoint !== null && checkpoint !== undefined && isCurrentCheckpointSchema(checkpoint.schema)) { - return await this.fromCheckpoint(checkpoint, opts, null); - } - return await this.fromScratch(opts); - } - - private async materializeWithStateCache( - stateCache: WarpStateCachePort, - opts: MaterializeLiveOptions, - ): Promise { const frontier = await this.runtime.deps.patches.getFrontier(); if (frontier.size === 0) { return await this.runtime.emptyResult(null, frontier, snapshotPublicationForReceipts(opts)); } const coordinate = this.snapshotCoordinate(frontier); - const cacheResolved = await this.tryResolveSnapshotCache(stateCache, { + const stateCache = this.runtime.deps.getStateCache?.() ?? null; + const cacheResolved = await this.tryResolveConfiguredStateCache( + stateCache, coordinate, - receipts: opts.receipts, - wantDiff: opts.wantDiff, - }); + opts, + ); if (cacheResolved !== null) { return cacheResolved; } return await this.replayCurrentCoordinate(coordinate, opts); } + private async tryResolveConfiguredStateCache( + stateCache: WarpStateCachePort | null, + coordinate: WarpStateCoordinate, + opts: MaterializeLiveOptions, + ): Promise { + if (stateCache === null) { + return null; + } + return await this.tryResolveSnapshotCache(stateCache, { + coordinate, + receipts: opts.receipts, + wantDiff: opts.wantDiff, + }); + } + private async replayCurrentCoordinate( coordinate: WarpStateCoordinate, opts: MaterializeLiveOptions, @@ -131,6 +128,7 @@ export default class MaterializeLiveStrategy { this.streamPatchesSinceCheckpoint(checkpoint, frontier), checkpoint.state, opts, + this.snapshotCoordinate(frontier ?? checkpoint.frontier), provenanceBase, ); return await this.runtime.buildResult({ @@ -174,36 +172,6 @@ export default class MaterializeLiveStrategy { } } - private async fromScratch(opts: MaterializeLiveOptions): Promise { - const writers = await this.runtime.deps.patches.discoverWriters(); - if (writers.length === 0) { - return await this.runtime.emptyResult( - undefined, - undefined, - snapshotPublicationForReceipts(opts), - ); - } - const reduction = await this.runtime.reducePatchStream( - this.streamAllPatches(writers), - undefined, - opts, - ); - if (reduction.summary.patchCount === 0) { - return await this.runtime.emptyResult( - undefined, - undefined, - snapshotPublicationForReceipts(opts), - ); - } - return await this.runtime.buildResult({ - reduced: reduction.reduced, - summary: reduction.summary, - degraded: false, - ceiling: null, - frontier: null, - }); - } - private async fromFrontier( coordinate: WarpStateCoordinate, opts: MaterializeLiveOptions, @@ -212,8 +180,10 @@ export default class MaterializeLiveStrategy { this.runtime.deps.patches.streamForFrontier(coordinate.frontier, coordinate.ceiling), undefined, opts, + coordinate, ); if (reduction.summary.patchCount === 0) { + await reduction.reduced.workspace?.release(); return await this.runtime.emptyResult( coordinate.ceiling, coordinate.frontier, @@ -286,6 +256,7 @@ export default class MaterializeLiveStrategy { receipts: false, wantDiff: false, }, + opts.coordinate, ); return await this.runtime.buildResult({ reduced: reduction.reduced, @@ -295,10 +266,4 @@ export default class MaterializeLiveStrategy { frontier: opts.coordinate.frontier, }); } - - private async *streamAllPatches(writers: string[]): AsyncIterable { - for (const writerId of writers) { - yield* this.runtime.deps.patches.streamWriterPatches(writerId); - } - } } diff --git a/src/domain/services/controllers/MaterializeSessionBridge.ts b/src/domain/services/controllers/MaterializeSessionBridge.ts index 68b8fd447..ae8829d9f 100644 --- a/src/domain/services/controllers/MaterializeSessionBridge.ts +++ b/src/domain/services/controllers/MaterializeSessionBridge.ts @@ -3,6 +3,10 @@ import VersionVector from "../../crdt/VersionVector.ts"; import { Dot } from "../../crdt/Dot.ts"; import type PatchEntry from "../../artifacts/PatchEntry.ts"; import type StateSession from "../../orset/session/StateSession.ts"; +import type MaterializationStorePort from "../../../ports/MaterializationStorePort.ts"; +import type MaterializationWorkspacePort from "../../../ports/MaterializationWorkspacePort.ts"; +import type MaterializationCoordinate from "../../materialization/MaterializationCoordinate.ts"; +import type StorageRetentionWitness from "../../storage/StorageRetentionWitness.ts"; import MaterializationRoot from "../../materialization/MaterializationRoot.ts"; import MaterializationRoots from "../../materialization/MaterializationRoots.ts"; import BundleHandle from "../../storage/BundleHandle.ts"; @@ -25,6 +29,7 @@ export type MaterializeSessionOpen = { export type MaterializeSessionOpener = ( init: MaterializeSessionOpen, + options: { readonly workspace: MaterializationWorkspacePort }, ) => Promise; type MaterializeSessionPatchSource = @@ -33,6 +38,8 @@ type MaterializeSessionPatchSource = export async function reduceSessionBackedState(args: { readonly openStateSession: MaterializeSessionOpener; + readonly materializations: MaterializationStorePort; + readonly coordinate: MaterializationCoordinate; readonly patches: MaterializeSessionPatchSource; readonly baseState?: WarpStateClass; readonly roots?: MaterializeSessionOpen; @@ -42,63 +49,81 @@ export async function reduceSessionBackedState(args: { readonly state: WarpStateClass; readonly adjacency: MaterializeAdjacency; readonly roots: MaterializationRoots; + readonly workspace: MaterializationWorkspacePort; + readonly acceptMaterialization: (witness: StorageRetentionWitness | null) => void; readonly receipts?: TickReceipt[]; readonly diff?: PatchDiff; }> { - const frame = await openReducerSessionFrame( - args.openStateSession, - args.baseState, - args.roots, - ); - let reduced: { - readonly state: WarpStateClass; - readonly adjacency: MaterializeAdjacency; - readonly receipts?: TickReceipt[]; - readonly diff?: PatchDiff; - }; - if (args.receipts) { - const result = await reducePatchesInSession(args.patches, frame, { - receipts: true, - }); - const adjacency = await buildAdjacencyFromSession(result.frame.session); - reduced = { - state: await projectFrameToState(result.frame), - adjacency, - receipts: result.receipts, - }; - } else if (args.wantDiff) { - const result = await reducePatchesInSession(args.patches, frame, { - trackDiff: true, - }); - const adjacency = await buildAdjacencyFromSession(result.frame.session); - reduced = { - state: await projectFrameToState(result.frame), - adjacency, - diff: result.diff, + const workspace = await args.materializations.openWorkspace(args.coordinate); + try { + const frame = await openReducerSessionFrame( + args.openStateSession, + workspace, + args.baseState, + args.roots, + ); + let reduced: { + readonly state: WarpStateClass; + readonly adjacency: MaterializeAdjacency; + readonly receipts?: TickReceipt[]; + readonly diff?: PatchDiff; }; - } else { - const result = await reducePatchesInSession(args.patches, frame); - const adjacency = await buildAdjacencyFromSession(result.session); - reduced = { - state: await projectFrameToState(result), - adjacency, + if (args.receipts) { + const result = await reducePatchesInSession(args.patches, frame, { + receipts: true, + }); + const adjacency = await buildAdjacencyFromSession(result.frame.session); + reduced = { + state: await projectFrameToState(result.frame), + adjacency, + receipts: result.receipts, + }; + } else if (args.wantDiff) { + const result = await reducePatchesInSession(args.patches, frame, { + trackDiff: true, + }); + const adjacency = await buildAdjacencyFromSession(result.frame.session); + reduced = { + state: await projectFrameToState(result.frame), + adjacency, + diff: result.diff, + }; + } else { + const result = await reducePatchesInSession(args.patches, frame); + const adjacency = await buildAdjacencyFromSession(result.session); + reduced = { + state: await projectFrameToState(result), + adjacency, + }; + } + + const close = await frame.session.prepareClose(); + const roots = materializationRootsFromSession(close.roots); + return { + ...reduced, + roots, + workspace, + acceptMaterialization: close.accept, }; + } catch (raw) { + await workspace.release(); + throw raw; } - - // close() flushes trie pages, so failed reductions must leave no CAS artifacts. - const roots = materializationRootsFromSession(await frame.session.close()); - return { ...reduced, roots }; } async function openReducerSessionFrame( openStateSession: MaterializeSessionOpener, + workspace: MaterializationWorkspacePort, baseState?: WarpStateClass, roots?: MaterializeSessionOpen, ): Promise { - const session = await openStateSession(roots ?? { - nodeAliveRootOid: null, - edgeAliveRootOid: null, - }); + const session = await openStateSession( + roots ?? { + nodeAliveRootOid: null, + edgeAliveRootOid: null, + }, + { workspace }, + ); if (baseState !== undefined && roots === undefined) { await seedSessionWithORSet({ diff --git a/src/domain/services/controllers/MaterializeStrategyRuntime.ts b/src/domain/services/controllers/MaterializeStrategyRuntime.ts index e0d85e7f0..14800ac2e 100644 --- a/src/domain/services/controllers/MaterializeStrategyRuntime.ts +++ b/src/domain/services/controllers/MaterializeStrategyRuntime.ts @@ -7,7 +7,10 @@ import type { } from './MaterializePatchStreamReducer.ts'; import type { MaterializePatchSummary } from './MaterializePatchSummary.ts'; import type { MaterializeSnapshotPublicationOptions } from './MaterializeSnapshotPublication.ts'; -import type { WarpStateSnapshotProvenancePosture } from '../../../ports/WarpStateCachePort.ts'; +import type { + WarpStateCoordinate, + WarpStateSnapshotProvenancePosture, +} from '../../../ports/WarpStateCachePort.ts'; import type { UsableSnapshotRecord } from './MaterializeSnapshotCacheResult.ts'; import type MaterializationHandle from '../../materialization/MaterializationHandle.ts'; import type { @@ -60,11 +63,13 @@ export type MaterializeStrategyRuntime = { patches: PatchWithSha[], base: WarpState | undefined, opts: { receipts: boolean; wantDiff: boolean }, + coordinate: WarpStateCoordinate, ): Promise; reducePatchStream( stream: AsyncIterable, base: WarpState | undefined, opts: MaterializePatchStreamOptions, + coordinate: WarpStateCoordinate, provenanceBase?: ProvenanceIndex, ): Promise; buildResult(params: MaterializeResultBuildInput): Promise; diff --git a/src/domain/warp/RuntimeHostBoot.ts b/src/domain/warp/RuntimeHostBoot.ts index 4b6de23ca..fc1f56012 100644 --- a/src/domain/warp/RuntimeHostBoot.ts +++ b/src/domain/warp/RuntimeHostBoot.ts @@ -338,16 +338,16 @@ export async function resolveRuntimeHostConstructionOptions( resolvedOpenStateSession = openStateSession; } else if (storageServices.trie !== undefined) { const store = storageServices.trie; - const pageCache = new PageCache({ maxResident: 256 }); const geometry = TrieGeometry.default16way(); - resolvedOpenStateSession = async (roots) => + resolvedOpenStateSession = async (roots, sessionOptions) => await StateSession.open({ nodeAliveRootOid: roots.nodeAliveRootOid, edgeAliveRootOid: roots.edgeAliveRootOid, store, codec: resolvedCodec, geometry, - pageCache, + pageCache: new PageCache({ maxResident: 256 }), + workspace: sessionOptions.workspace, }); } diff --git a/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts b/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts index 73410f55d..55eb3df5a 100644 --- a/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts +++ b/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts @@ -20,10 +20,12 @@ import type StorageRetentionWitness from '../../domain/storage/StorageRetentionW import WarpError from '../../domain/errors/WarpError.ts'; import type CodecPort from '../../ports/CodecPort.ts'; import type CryptoPort from '../../ports/CryptoPort.ts'; +import type MaterializationWorkspacePort from '../../ports/MaterializationWorkspacePort.ts'; import MaterializationStorePort, { type RetainMaterializationRequest, } from '../../ports/MaterializationStorePort.ts'; import { adaptGitCasRetentionWitness } from './GitCasRetentionWitnessAdapter.ts'; +import GitCasMaterializationWorkspace from './GitCasMaterializationWorkspace.ts'; import { decodeMaterializationDescriptor, MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION, @@ -34,12 +36,13 @@ import { } from './GitCasMaterializationDescriptor.ts'; const CACHE_NAMESPACE = 'git-warp/materializations'; +const WORKSPACE_CACHE_NAMESPACE = 'git-warp/materialization-workspaces'; const DESCRIPTOR_PATH = 'meta/descriptor'; 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; @@ -84,6 +87,24 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt this.#laneName = requireNonEmpty(options.laneName, 'laneName'); } + override async openWorkspace( + coordinate: MaterializationCoordinate, + ): Promise { + requireCoordinate(coordinate); + const cache = await this.#cas.caches.open({ namespace: WORKSPACE_CACHE_NAMESPACE }); + return new GitCasMaterializationWorkspace({ + bundles: this.#cas.bundles, + cache, + key: `workspace:${this.#laneName}:${globalThis.crypto.randomUUID()}`, + promote: async (request) => { + if (!request.coordinate.equals(coordinate)) { + throw storageError('workspace promotion coordinate does not match its open coordinate'); + } + return await this.retain(request); + }, + }); + } + override async retain(request: RetainMaterializationRequest): Promise { requireRetainRequest(request); const stateHash = requireNonEmpty(request.stateHash, 'stateHash'); diff --git a/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts b/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts new file mode 100644 index 000000000..3935f4cdf --- /dev/null +++ b/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts @@ -0,0 +1,371 @@ +import { + BundleHandle as GitCasBundleHandle, + type BundleCapability, + type BundleMemberInput, + type CacheSet, + type RetentionWitness, +} from '@git-stunts/git-cas'; +import type MaterializationHandle from '../../domain/materialization/MaterializationHandle.ts'; +import type StorageRetentionWitness from '../../domain/storage/StorageRetentionWitness.ts'; +import WarpError from '../../domain/errors/WarpError.ts'; +import MaterializationWorkspacePort, { + type MaterializationWorkspaceRoots, + type PromoteMaterializationRequest, +} from '../../ports/MaterializationWorkspacePort.ts'; +import { adaptGitCasRetentionWitness } from './GitCasRetentionWitnessAdapter.ts'; + +const WORKSPACE_TTL_MS = 2 * 60 * 60 * 1000; +const WORKSPACE_RENEWAL_MS = WORKSPACE_TTL_MS / 2; + +type WorkspaceCache = Pick; +type WorkspaceTarget = Parameters[1]; +type ActiveLeaseTarget = Readonly<{ + input: WorkspaceTarget; + token: string; +}>; + +export type MaterializationWorkspaceLease = Readonly<{ + cancel(): void; +}>; + +export type MaterializationWorkspaceLeaseScheduler = Readonly<{ + schedule( + task: () => Promise, + delayMs: number, + ): MaterializationWorkspaceLease; +}>; + +export type GitCasMaterializationWorkspaceOptions = Readonly<{ + bundles: Pick; + cache: WorkspaceCache; + key: string; + clock?: { readonly now: () => Date }; + leaseTtlMs?: number; + leaseRenewalMs?: number; + leaseScheduler?: MaterializationWorkspaceLeaseScheduler; + promote: (request: PromoteMaterializationRequest) => Promise; +}>; + +const SYSTEM_LEASE_SCHEDULER: MaterializationWorkspaceLeaseScheduler = Object.freeze({ + schedule(task: () => Promise, delayMs: number): MaterializationWorkspaceLease { + const timer = setTimeout(() => { + void task().catch(() => undefined); + }, delayMs); + timer.unref(); + return Object.freeze({ cancel: () => clearTimeout(timer) }); + }, +}); + +/** CacheSet-backed reachability for one in-progress materialization. */ +export default class GitCasMaterializationWorkspace extends MaterializationWorkspacePort { + readonly #bundles: Pick; + readonly #cache: WorkspaceCache; + readonly #clock: { readonly now: () => Date }; + readonly #key: string; + readonly #leaseRenewalMs: number; + readonly #leaseScheduler: MaterializationWorkspaceLeaseScheduler; + readonly #leaseTtlMs: number; + readonly #promoteMaterialization: ( + request: PromoteMaterializationRequest, + ) => Promise; + #installationAttempted = false; + #leaseTarget: ActiveLeaseTarget | null = null; + #leaseFailure: WarpError | null = null; + #leaseTimer: MaterializationWorkspaceLease | null = null; + #promoting = false; + #promotionDone: Promise | null = null; + #releasePending = false; + #releaseRequested = false; + #released = false; + #tail: Promise = Promise.resolve(); + + constructor(options: GitCasMaterializationWorkspaceOptions) { + super(); + requireWorkspaceOptions(options); + this.#bundles = options.bundles; + this.#cache = options.cache; + this.#key = requireNonEmpty(options.key, 'key'); + this.#clock = options.clock ?? { now: () => new Date() }; + this.#leaseTtlMs = options.leaseTtlMs ?? WORKSPACE_TTL_MS; + this.#leaseRenewalMs = options.leaseRenewalMs ?? WORKSPACE_RENEWAL_MS; + this.#leaseScheduler = options.leaseScheduler ?? SYSTEM_LEASE_SCHEDULER; + this.#promoteMaterialization = options.promote; + } + + override checkpoint( + roots: MaterializationWorkspaceRoots, + ): Promise { + if (this.#releasePending || this.#releaseRequested || this.#promoting) { + return Promise.reject(workspaceError('cannot checkpoint a releasing workspace')); + } + return this.#serialize(async () => await this.#checkpoint(roots)); + } + + override async release(): Promise { + this.#releasePending = true; + await this.#promotionDone; + this.#releaseRequested = true; + this.#cancelLeaseTimer(); + await this.#serialize(async () => await this.#release()); + } + + override promote( + request: PromoteMaterializationRequest, + ): Promise { + if (this.#releasePending || this.#releaseRequested || this.#promoting) { + return Promise.reject(workspaceError('cannot promote a releasing workspace')); + } + this.#promoting = true; + const operation = this.#serialize(() => this.#assertLeaseHealthy()) + .then(async () => await this.#promoteMaterialization(request)) + .then(async (materialization) => await this.#finishPromotion(materialization)); + const done = operation.then( + () => undefined, + () => undefined, + ); + this.#promotionDone = done; + return operation.finally(() => { + this.#promoting = false; + if (this.#promotionDone === done) { + this.#promotionDone = null; + } + }); + } + + async #checkpoint( + roots: MaterializationWorkspaceRoots, + ): Promise { + this.#assertLeaseHealthy(); + const members = workspaceMembers(roots); + if (members.length === 0) { + return null; + } + const bundle = await this.#bundles.putOrdered({ members }); + const targetToken = bundle.handle.toString(); + const witness = await this.#retain(bundle.handle, targetToken); + const target = Object.freeze({ input: bundle.handle, token: targetToken }); + this.#leaseTarget = target; + this.#scheduleLeaseRenewal(target); + return adaptGitCasRetentionWitness(witness.toJSON()); + } + + async #retain( + target: WorkspaceTarget, + expectedHandle: string, + ): Promise { + this.#installationAttempted = true; + const retained = await this.#cache.put(this.#key, target, { + retention: 'pinned', + expiresAt: this.#expiresAt(), + }); + return requireAcceptedCheckpoint(retained, expectedHandle); + } + + async #finishPromotion( + materialization: MaterializationHandle, + ): Promise { + return await this.#serialize(() => { + this.#leaseFailure = null; + this.#cancelLeaseTimer(); + return materialization; + }); + } + + async #release(): Promise { + if (this.#released) { + return; + } + if (this.#installationAttempted) { + await this.#cache.remove(this.#key); + } + this.#released = true; + this.#assertLeaseHealthy(); + } + + #expiresAt(): string { + const now = this.#clock.now(); + if (!(now instanceof Date) || Number.isNaN(now.getTime())) { + throw workspaceError('clock returned an invalid Date'); + } + return new Date(now.getTime() + this.#leaseTtlMs).toISOString(); + } + + #scheduleLeaseRenewal(target: ActiveLeaseTarget): void { + this.#cancelLeaseTimer(); + if (!this.#leaseIsActive()) { + return; + } + this.#leaseTimer = this.#leaseScheduler.schedule( + async () => await this.#renewLease(target), + this.#leaseRenewalMs, + ); + } + + async #renewLease(target: ActiveLeaseTarget): Promise { + this.#leaseTimer = null; + await this.#serialize(async () => { + if (!this.#leaseIsActive()) { + return; + } + if (this.#leaseTarget !== target) { + return; + } + try { + await this.#retain(target.input, target.token); + this.#scheduleLeaseRenewal(target); + } catch (raw) { + this.#leaseFailure = workspaceError(`lease renewal failed: ${errorMessage(raw)}`); + this.#cancelLeaseTimer(); + } + }); + } + + #cancelLeaseTimer(): void { + this.#leaseTimer?.cancel(); + this.#leaseTimer = null; + } + + #leaseIsActive(): boolean { + return !this.#releaseRequested && !this.#released; + } + + #assertLeaseHealthy(): void { + if (this.#leaseFailure !== null) { + throw this.#leaseFailure; + } + } + + #serialize(operation: () => T | Promise): Promise { + const result = this.#tail.then(operation); + this.#tail = result.then( + () => undefined, + () => undefined, + ); + return result; + } +} + +function workspaceMembers( + roots: MaterializationWorkspaceRoots, +): Array<[string, BundleMemberInput]> { + requireRoots(roots); + const members: Array<[string, BundleMemberInput]> = []; + if (roots.edgeAliveRoot !== null) { + members.push(['roots/edge-alive', parseRoot(roots.edgeAliveRoot)]); + } + if (roots.nodeAliveRoot !== null) { + members.push(['roots/node-alive', parseRoot(roots.nodeAliveRoot)]); + } + return members; +} + +function parseRoot(token: string): string { + try { + return GitCasBundleHandle.parse(token).toString(); + } catch (raw) { + throw workspaceError(`root is not a bundle handle: ${errorMessage(raw)}`); + } +} + +function requireRoots(value: MaterializationWorkspaceRoots): void { + if (value === null || typeof value !== 'object' || Array.isArray(value)) { + throw workspaceError('checkpoint roots must be an object'); + } +} + +function requireWorkspaceOptions(options: GitCasMaterializationWorkspaceOptions): void { + requireOptionsObject(options); + requireBundles(options.bundles); + requireCache(options.cache); + requireClock(options.clock); + requireLeaseTiming(options.leaseTtlMs, options.leaseRenewalMs); + requireLeaseScheduler(options.leaseScheduler); + requirePromote(options.promote); +} + +function requireAcceptedCheckpoint( + retained: Awaited>, + expectedHandle: string, +): RetentionWitness { + if (!retained.accepted || retained.hit === null || retained.witness === null) { + throw workspaceError('git-cas did not retain the workspace checkpoint'); + } + if (retained.hit.handle.toString() !== expectedHandle) { + throw workspaceError('git-cas retained an unexpected workspace handle'); + } + return retained.witness; +} + +function requireOptionsObject(options: GitCasMaterializationWorkspaceOptions): void { + if (options === null || typeof options !== 'object' || Array.isArray(options)) { + throw workspaceError('options must be an object'); + } +} + +function requireBundles(bundles: Pick): void { + if (typeof bundles?.putOrdered !== 'function') { + throw workspaceError('bundles dependency is required'); + } +} + +function requireCache(cache: WorkspaceCache): void { + if (typeof cache?.put !== 'function' || typeof cache?.remove !== 'function') { + throw workspaceError('cache dependency is required'); + } +} + +function requireClock(clock: { readonly now: () => Date } | undefined): void { + if (clock !== undefined && typeof clock.now !== 'function') { + throw workspaceError('clock must provide now()'); + } +} + +function requireLeaseTiming(ttlMs: number | undefined, renewalMs: number | undefined): void { + const ttl = ttlMs ?? WORKSPACE_TTL_MS; + const renewal = renewalMs ?? WORKSPACE_RENEWAL_MS; + requirePositiveLeaseTtl(ttl); + requireLeaseRenewalBelowTtl(renewal, ttl); +} + +function requirePositiveLeaseTtl(ttl: number): void { + if (!Number.isSafeInteger(ttl) || ttl <= 0) { + throw workspaceError('leaseTtlMs must be a positive safe integer'); + } +} + +function requireLeaseRenewalBelowTtl(renewal: number, ttl: number): void { + if (!Number.isSafeInteger(renewal) || renewal <= 0 || renewal >= ttl) { + throw workspaceError('leaseRenewalMs must be a positive safe integer below leaseTtlMs'); + } +} + +function requireLeaseScheduler( + scheduler: MaterializationWorkspaceLeaseScheduler | undefined, +): void { + if (scheduler !== undefined && typeof scheduler.schedule !== 'function') { + throw workspaceError('leaseScheduler must provide schedule()'); + } +} + +function requirePromote( + promote: ((request: PromoteMaterializationRequest) => Promise) | undefined, +): void { + if (typeof promote !== 'function') { + throw workspaceError('promote dependency is required'); + } +} + +function requireNonEmpty(value: string, field: string): string { + if (typeof value !== 'string' || value.length === 0) { + throw workspaceError(`${field} must be a non-empty string`); + } + return value; +} + +function errorMessage(raw: unknown): string { + return raw instanceof Error ? raw.message : String(raw); +} + +function workspaceError(message: string): WarpError { + return new WarpError(`Materialization workspace ${message}`, 'E_MATERIALIZATION_STORAGE'); +} diff --git a/src/ports/MaterializationStorePort.ts b/src/ports/MaterializationStorePort.ts index 1b53ce87e..4a06873ee 100644 --- a/src/ports/MaterializationStorePort.ts +++ b/src/ports/MaterializationStorePort.ts @@ -1,15 +1,16 @@ import type MaterializationCoordinate from '../domain/materialization/MaterializationCoordinate.ts'; import type MaterializationHandle from '../domain/materialization/MaterializationHandle.ts'; -import type MaterializationRoots from '../domain/materialization/MaterializationRoots.ts'; +import type MaterializationWorkspacePort from './MaterializationWorkspacePort.ts'; +import type { PromoteMaterializationRequest } from './MaterializationWorkspacePort.ts'; -export type RetainMaterializationRequest = Readonly<{ - coordinate: MaterializationCoordinate; - roots: MaterializationRoots; - stateHash: string; -}>; +export type RetainMaterializationRequest = PromoteMaterializationRequest; /** Storage-neutral lifecycle for retained, independently addressable materializations. */ export default abstract class MaterializationStorePort { + abstract openWorkspace( + _coordinate: MaterializationCoordinate, + ): Promise; + abstract retain(_request: RetainMaterializationRequest): Promise; abstract findExact( diff --git a/src/ports/MaterializationWorkspacePort.ts b/src/ports/MaterializationWorkspacePort.ts new file mode 100644 index 000000000..fced8d34c --- /dev/null +++ b/src/ports/MaterializationWorkspacePort.ts @@ -0,0 +1,31 @@ +import type MaterializationCoordinate from '../domain/materialization/MaterializationCoordinate.ts'; +import type MaterializationHandle from '../domain/materialization/MaterializationHandle.ts'; +import type MaterializationRoots from '../domain/materialization/MaterializationRoots.ts'; +import type StorageRetentionWitness from '../domain/storage/StorageRetentionWitness.ts'; + +export type MaterializationWorkspaceRoots = Readonly<{ + nodeAliveRoot: string | null; + edgeAliveRoot: string | null; +}>; + +export type PromoteMaterializationRequest = Readonly<{ + coordinate: MaterializationCoordinate; + roots: MaterializationRoots; + stateHash: string; +}>; + +/** + * Operation-scoped retention for materialization roots that are not yet + * reachable from the final retained materialization handle. + */ +export default abstract class MaterializationWorkspacePort { + abstract checkpoint( + _roots: MaterializationWorkspaceRoots, + ): Promise; + + abstract promote( + _request: PromoteMaterializationRequest, + ): Promise; + + abstract release(): Promise; +} diff --git a/test/helpers/InMemoryGitCasFacade.ts b/test/helpers/InMemoryGitCasFacade.ts index 818f741cc..fb0ad91b0 100644 --- a/test/helpers/InMemoryGitCasFacade.ts +++ b/test/helpers/InMemoryGitCasFacade.ts @@ -48,7 +48,7 @@ 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; @@ -102,6 +102,10 @@ export default class InMemoryGitCasFacade { return Object.freeze([...(this.#cacheEntries.get(namespace)?.keys() ?? [])]); } + readCacheHits(namespace: string): readonly CacheHit[] { + return Object.freeze([...(this.#cacheEntries.get(namespace)?.values() ?? [])]); + } + replaceStoredPage(handle: string, bytes: Uint8Array): void { this.#pageBytes.set(handle, bytes.slice()); } @@ -288,7 +292,7 @@ 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({ @@ -316,7 +320,7 @@ export default class InMemoryGitCasFacade { key, handle: target, policy: options?.retention ?? 'evictable', - expiresAt: null, + expiresAt: options?.expiresAt ?? null, logicalBytes: 0, createdAt: observedAt, accessedAt: observedAt, @@ -334,6 +338,18 @@ export default class InMemoryGitCasFacade { witness: evidence, }); }, + remove: async (key) => { + const removed = entries.get(key) ?? null; + const changed = entries.delete(key); + this.#cacheGeneration += 1; + return Object.freeze({ + changed, + removed, + generation: this.#cacheGeneration.toString(16).padStart(40, '0'), + policy: null, + witness: null, + }); + }, }); } diff --git a/test/helpers/InMemoryMaterializationStore.ts b/test/helpers/InMemoryMaterializationStore.ts index 0a55be410..25602231c 100644 --- a/test/helpers/InMemoryMaterializationStore.ts +++ b/test/helpers/InMemoryMaterializationStore.ts @@ -7,14 +7,64 @@ import StorageRetentionWitness, { import MaterializationStorePort, { type RetainMaterializationRequest, } from '../../src/ports/MaterializationStorePort.ts'; +import MaterializationWorkspacePort, { + type MaterializationWorkspaceRoots, + type PromoteMaterializationRequest, +} from '../../src/ports/MaterializationWorkspacePort.ts'; + +export class InMemoryMaterializationWorkspace extends MaterializationWorkspacePort { + readonly checkpoints: MaterializationWorkspaceRoots[] = []; + readonly #promoteMaterialization: ( + request: PromoteMaterializationRequest, + ) => Promise; + released = false; + + constructor( + promoteMaterialization: ( + request: PromoteMaterializationRequest, + ) => Promise, + ) { + super(); + this.#promoteMaterialization = promoteMaterialization; + } + + override checkpoint(roots: MaterializationWorkspaceRoots): Promise { + this.checkpoints.push(roots); + const bundle = new BundleHandle(`test:workspace:${String(this.checkpoints.length)}`); + return Promise.resolve(workspaceRetentionWitness(bundle)); + } + + override release(): Promise { + this.released = true; + return Promise.resolve(); + } + + override promote(request: PromoteMaterializationRequest): Promise { + return this.#promoteMaterialization(request); + } +} /** Behavioral retained-materialization store for controller tests. */ export default class InMemoryMaterializationStore extends MaterializationStorePort { readonly exactLookups: MaterializationCoordinate[] = []; readonly retainedRequests: RetainMaterializationRequest[] = []; + readonly workspaces: InMemoryMaterializationWorkspace[] = []; readonly #handles = new Map(); #nextHandle = 1; + override openWorkspace( + coordinate: MaterializationCoordinate, + ): Promise { + const workspace = new InMemoryMaterializationWorkspace(async (request) => { + if (!request.coordinate.equals(coordinate)) { + throw new Error('Workspace promotion coordinate mismatch'); + } + return await this.retain(request); + }); + this.workspaces.push(workspace); + return Promise.resolve(workspace); + } + override retain(request: RetainMaterializationRequest): Promise { this.retainedRequests.push(request); const bundle = new BundleHandle(`test:materialization:${this.#nextHandle}`); @@ -61,3 +111,19 @@ function retentionWitness(handle: BundleHandle): StorageRetentionWitness { observedAt: '1970-01-01T00:00:00.000Z', }); } + +function workspaceRetentionWitness(handle: BundleHandle): StorageRetentionWitness { + return new StorageRetentionWitness({ + handle, + policy: 'pinned', + reachability: 'anchored', + root: new StorageRetentionRoot({ + kind: 'cache-set', + namespace: 'test/materialization-workspaces', + locator: 'test/materialization-workspaces', + generation: 'test-workspace-generation', + path: handle.toString(), + }), + observedAt: '1970-01-01T00:00:00.000Z', + }); +} diff --git a/test/integration/api/materialization.retainedResume.test.ts b/test/integration/api/materialization.retainedResume.test.ts index e4d32d4c5..a4085d686 100644 --- a/test/integration/api/materialization.retainedResume.test.ts +++ b/test/integration/api/materialization.retainedResume.test.ts @@ -6,6 +6,10 @@ import GitCasRepositoryAdapter from '../../../src/infrastructure/adapters/GitCas import MaterializationStorePort, { type RetainMaterializationRequest, } from '../../../src/ports/MaterializationStorePort.ts'; +import MaterializationWorkspacePort, { + type MaterializationWorkspaceRoots, + type PromoteMaterializationRequest, +} from '../../../src/ports/MaterializationWorkspacePort.ts'; import type RuntimeStorageProviderPort from '../../../src/ports/RuntimeStorageProviderPort.ts'; import type { RuntimeStorageRequest, @@ -77,6 +81,18 @@ class RecordingMaterializationStore extends MaterializationStorePort { this.#delegate = delegate; } + override async openWorkspace( + coordinate: MaterializationCoordinate, + ): Promise { + const workspace = await this.#delegate.openWorkspace(coordinate); + return new RecordingMaterializationWorkspace(workspace, async (request) => { + this.retainRequests.push(request); + const retained = await workspace.promote(request); + this.retainedHandles.push(retained); + return retained; + }); + } + override async retain(request: RetainMaterializationRequest): Promise { this.retainRequests.push(request); const retained = await this.#delegate.retain(request); @@ -96,6 +112,36 @@ class RecordingMaterializationStore extends MaterializationStorePort { } } +class RecordingMaterializationWorkspace extends MaterializationWorkspacePort { + readonly #delegate: MaterializationWorkspacePort; + readonly #promote: ( + request: PromoteMaterializationRequest, + ) => Promise; + + constructor( + delegate: MaterializationWorkspacePort, + promote: ( + request: PromoteMaterializationRequest, + ) => Promise, + ) { + super(); + this.#delegate = delegate; + this.#promote = promote; + } + + override checkpoint(roots: MaterializationWorkspaceRoots) { + return this.#delegate.checkpoint(roots); + } + + override promote(request: PromoteMaterializationRequest): Promise { + return this.#promote(request); + } + + override release(): Promise { + return this.#delegate.release(); + } +} + class RecordingRuntimeStorageProvider implements RuntimeStorageProviderPort { readonly #delegate: RuntimeStorageProviderPort; materializations: RecordingMaterializationStore | null = null; diff --git a/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts b/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts index 41f5aa58d..9130df6a8 100644 --- a/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts +++ b/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts @@ -13,6 +13,11 @@ import MaterializationRoot from '../../../../src/domain/materialization/Material import MaterializationRoots from '../../../../src/domain/materialization/MaterializationRoots.ts'; import BundleHandle from '../../../../src/domain/storage/BundleHandle.ts'; import GitCasRepositoryAdapter from '../../../../src/infrastructure/adapters/GitCasRepositoryAdapter.ts'; +import GitCasMaterializationWorkspace from '../../../../src/infrastructure/adapters/GitCasMaterializationWorkspace.ts'; +import type { + MaterializationWorkspaceLease, + MaterializationWorkspaceLeaseScheduler, +} from '../../../../src/infrastructure/adapters/GitCasMaterializationWorkspace.ts'; import GitCasTrieStoreAdapter from '../../../../src/infrastructure/adapters/GitCasTrieStoreAdapter.ts'; import GitTimelineHistoryAdapter from '../../../../src/infrastructure/adapters/GitTimelineHistoryAdapter.ts'; import NodeCryptoAdapter from '../../../../src/infrastructure/adapters/NodeCryptoAdapter.ts'; @@ -79,6 +84,74 @@ describe('GitCasMaterializationStoreAdapter integration', () => { args: ['show-ref', '--verify', '--hash', 'refs/cas/caches/git-warp/materializations'], })).toMatch(/^[0-9a-f]{40}\n?$/u); }); + + it('keeps workspace roots readable across aggressive Git pruning', async () => { + const trieFixture = await createTrieRoot(harness.cas); + const workspace = await harness.materializations.openWorkspace(workspaceCoordinate()); + + const witness = await workspace.checkpoint({ + nodeAliveRoot: trieFixture.root.toString(), + edgeAliveRoot: null, + }); + if (witness === null) { + throw new Error('Workspace did not witness its non-empty trie root'); + } + expect(witness).toMatchObject({ + policy: 'pinned', + reachability: 'anchored', + }); + expect(await prunableOids(harness.path)).not.toContain( + GitCasBundleHandle.parse(trieFixture.root.toString()).oid, + ); + + await execFileAsync('git', ['-C', harness.path, 'prune', '--expire=now']); + const trie = new GitCasTrieStoreAdapter({ cas: harness.cas }); + const children = await trie.readBranch(trieFixture.root.toString()); + expect(children.size).toBe(1); + + await workspace.release(); + }); + + it('renews an active workspace across expiry, sweep, and aggressive pruning', async () => { + const clock = new MutableClock('2026-07-16T00:00:00.000Z'); + const leaseHarness = await createHarness(clock); + try { + const trieFixture = await createTrieRoot(leaseHarness.cas); + const cache = await leaseHarness.cas.caches.open({ + namespace: 'git-warp/materialization-workspaces', + }); + const scheduler = new ManualLeaseScheduler(); + const workspace = new GitCasMaterializationWorkspace({ + bundles: leaseHarness.cas.bundles, + cache, + key: 'active-lease', + clock, + leaseTtlMs: 1_000, + leaseRenewalMs: 500, + leaseScheduler: scheduler, + promote: rejectPromotion, + }); + + await workspace.checkpoint({ + nodeAliveRoot: trieFixture.root.toString(), + edgeAliveRoot: null, + }); + clock.advance(600); + await scheduler.runNext(); + clock.advance(500); + + const sweep = await cache.sweep(); + expect(sweep.removed).toBe(0); + expect(await cache.get('active-lease')).not.toBeNull(); + await execFileAsync('git', ['-C', leaseHarness.path, 'prune', '--expire=now']); + + const trie = new GitCasTrieStoreAdapter({ cas: leaseHarness.cas }); + expect((await trie.readBranch(trieFixture.root.toString())).size).toBe(1); + await workspace.release(); + } finally { + await rm(leaseHarness.path, { recursive: true, force: true }); + } + }); }); type Harness = Readonly<{ @@ -88,13 +161,13 @@ type Harness = Readonly<{ plumbing: Awaited>; }>; -async function createHarness(): Promise { +async function createHarness(clock?: { readonly now: () => Date }): Promise { const path = await mkdtemp(join(tmpdir(), 'git-warp-materializations-')); const plumbing = await Plumbing.createDefault({ cwd: path }); await plumbing.execute({ args: ['init', '-q'] }); await plumbing.execute({ args: ['config', 'user.email', 'test@example.com'] }); await plumbing.execute({ args: ['config', 'user.name', 'Test'] }); - const cas = createCas(plumbing); + const cas = createCas(plumbing, clock); return Object.freeze({ cas, path, @@ -105,11 +178,13 @@ async function createHarness(): Promise { function createCas( plumbing: Awaited>, + clock?: { readonly now: () => Date }, ): ContentAddressableStore { return ContentAddressableStore.createCbor({ plumbing, chunking: { strategy: 'cdc' }, applicationRefPrefixes: ['refs/warp/'], + ...(clock === undefined ? {} : { clock }), }); } @@ -220,6 +295,57 @@ function rootSignature( return [name, root.status, root.handle?.toString() ?? null]; } +function workspaceCoordinate(): MaterializationCoordinate { + return new MaterializationCoordinate({ + frontier: new Map([['writer-a', 'a'.repeat(40)]]), + ceiling: null, + }); +} + +function rejectPromotion(): Promise { + return Promise.reject(new Error('Promotion is not used by the lease integration test')); +} + +class MutableClock { + #current: number; + + constructor(iso: string) { + this.#current = Date.parse(iso); + } + + now(): Date { + return new Date(this.#current); + } + + advance(milliseconds: number): void { + this.#current += milliseconds; + } +} + +class ManualLeaseScheduler implements MaterializationWorkspaceLeaseScheduler { + #task: (() => Promise) | null = null; + + schedule(task: () => Promise, _delayMs: number): MaterializationWorkspaceLease { + this.#task = task; + return Object.freeze({ + cancel: () => { + if (this.#task === task) { + this.#task = null; + } + }, + }); + } + + async runNext(): Promise { + const task = this.#task; + if (task === null) { + throw new Error('Expected a scheduled lease renewal'); + } + this.#task = null; + await task(); + } +} + async function prunableOids(path: string): Promise> { const { stdout } = await execFileAsync( 'git', diff --git a/test/unit/domain/WarpGraph.autoCheckpoint.test.ts b/test/unit/domain/WarpGraph.autoCheckpoint.test.ts index 945e870d3..b1d1db6e0 100644 --- a/test/unit/domain/WarpGraph.autoCheckpoint.test.ts +++ b/test/unit/domain/WarpGraph.autoCheckpoint.test.ts @@ -372,7 +372,7 @@ describe('AP/CKPT/3: auto-checkpoint in materialize() path', () => { const checkpointState = createEmptyState(); - // Build 4 fake patch objects for _loadPatchesSince + // Build 4 fake patch objects for the checkpoint suffix stream. const patches: any[] = []; for (let i = 1; i <= 4; i++) { patches.push({ @@ -384,9 +384,12 @@ describe('AP/CKPT/3: auto-checkpoint in materialize() path', () => { vi.spyOn(graph, ('_loadLatestCheckpoint' as any)).mockResolvedValue({ schema: 5, state: checkpointState, - frontier: {}, + frontier: new Map(), }); - vi.spyOn(graph, ('_loadPatchesSince' as any)).mockResolvedValue(patches); + vi.spyOn(graph, 'getFrontier').mockResolvedValue(new Map([ + ['w1', patches[patches.length - 1].sha], + ])); + vi.spyOn(graph, ('_loadPatchChainFromSha' as any)).mockResolvedValue(patches); const spy = vi .spyOn(graph, 'createCheckpoint') @@ -419,9 +422,12 @@ describe('AP/CKPT/3: auto-checkpoint in materialize() path', () => { vi.spyOn(graph, ('_loadLatestCheckpoint' as any)).mockResolvedValue({ schema: 5, state: checkpointState, - frontier: {}, + frontier: new Map(), }); - vi.spyOn(graph, ('_loadPatchesSince' as any)).mockResolvedValue(patches); + vi.spyOn(graph, 'getFrontier').mockResolvedValue(new Map([ + ['w1', patches[patches.length - 1].sha], + ])); + vi.spyOn(graph, ('_loadPatchChainFromSha' as any)).mockResolvedValue(patches); const spy = vi .spyOn(graph, 'createCheckpoint') @@ -517,7 +523,10 @@ describe('AP/CKPT/3: auto-checkpoint in materialize() path', () => { frontier: new Map(), indexShardOids: null, }); - vi.spyOn(graph, ('_loadPatchesSince' as any)).mockResolvedValue(patches); + vi.spyOn(graph, 'getFrontier').mockResolvedValue(new Map([ + ['w1', patches[patches.length - 1].sha], + ])); + vi.spyOn(graph, ('_loadPatchChainFromSha' as any)).mockResolvedValue(patches); const state = await graph.materialize(); diff --git a/test/unit/domain/WarpGraph.patchCount.test.ts b/test/unit/domain/WarpGraph.patchCount.test.ts index 6b0b34f90..b1fe3834d 100644 --- a/test/unit/domain/WarpGraph.patchCount.test.ts +++ b/test/unit/domain/WarpGraph.patchCount.test.ts @@ -135,15 +135,14 @@ describe('AP/CKPT/2: _patchesSinceCheckpoint tracking', () => { it('golden path: checkpoint then 10 patches yields count = 10', async () => { const patchCount = 10; - // We simulate materialization with a checkpoint by mocking _loadLatestCheckpoint - // and _loadPatchesSince. The simplest approach: spy on private methods. + // Simulate materialization from a checkpoint and an explicit live frontier. const { createEmptyState } = await import( '../../../src/domain/services/JoinReducer.ts' ); const checkpointState = createEmptyState(); - // Build 10 fake patch objects for _loadPatchesSince to return + // Build 10 fake patch objects for the checkpoint suffix stream. const patches: any[] = []; for (let i = 1; i <= patchCount; i++) { patches.push({ @@ -156,11 +155,12 @@ describe('AP/CKPT/2: _patchesSinceCheckpoint tracking', () => { vi.spyOn(graph, '_loadLatestCheckpoint').mockResolvedValue({ schema: 5, state: checkpointState, - frontier: {}, + frontier: new Map(), }); - - // Mock _loadPatchesSince to return the 10 patches - vi.spyOn(graph, '_loadPatchesSince').mockResolvedValue(patches); + vi.spyOn(graph, 'getFrontier').mockResolvedValue(new Map([ + ['w1', patches[patches.length - 1].sha], + ])); + vi.spyOn(graph, '_loadPatchChainFromSha').mockResolvedValue(patches); await graph.materialize(); diff --git a/test/unit/domain/WarpGraph.test.ts b/test/unit/domain/WarpGraph.test.ts index 9775b1195..e0d5ae1d6 100644 --- a/test/unit/domain/WarpGraph.test.ts +++ b/test/unit/domain/WarpGraph.test.ts @@ -631,11 +631,11 @@ describe('WarpCore', () => { 'refs/warp/events/writers/writer-2', ]); - // materialize() now checks for checkpoint first, then reads writer tips - persistence.readRef - .mockResolvedValueOnce(null) // checkpoint ref (none) - .mockResolvedValueOnce(commitSha1) // writer-1 tip - .mockResolvedValueOnce(commitSha2); // writer-2 tip + persistence.readRef.mockImplementation((ref: string) => Promise.resolve( + ref.includes('/checkpoints/') ? null + : ref.endsWith('/writer-1') ? commitSha1 + : ref.endsWith('/writer-2') ? commitSha2 : null, + )); persistence.getNodeInfo .mockResolvedValueOnce(mockPatch1.nodeInfo) diff --git a/test/unit/domain/WarpGraph.versionVector.test.ts b/test/unit/domain/WarpGraph.versionVector.test.ts index 56e79a808..281c4dd8f 100644 --- a/test/unit/domain/WarpGraph.versionVector.test.ts +++ b/test/unit/domain/WarpGraph.versionVector.test.ts @@ -134,10 +134,18 @@ describe('WarpCore', () => { 'refs/warp/events/writers/writer-b', ]); - persistence.readRef - .mockResolvedValueOnce(null) // checkpoint ref (none) - .mockResolvedValueOnce(commitShaA) // writer-a tip - .mockResolvedValueOnce(commitShaB); // writer-b tip + persistence.readRef.mockImplementation((ref: string) => { + if (ref.includes('/checkpoints/')) { + return Promise.resolve(null); + } + if (ref.endsWith('/writer-a')) { + return Promise.resolve(commitShaA); + } + if (ref.endsWith('/writer-b')) { + return Promise.resolve(commitShaB); + } + return Promise.resolve(null); + }); persistence.getNodeInfo .mockResolvedValueOnce({ diff --git a/test/unit/domain/orset/session/StateSession.test.ts b/test/unit/domain/orset/session/StateSession.test.ts index 3f3b90321..ec9bc1c00 100644 --- a/test/unit/domain/orset/session/StateSession.test.ts +++ b/test/unit/domain/orset/session/StateSession.test.ts @@ -3,11 +3,19 @@ import { describe, expect, it } from "vitest"; import { Dot } from "../../../../../src/domain/crdt/Dot.ts"; import VersionVector from "../../../../../src/domain/crdt/VersionVector.ts"; import StateSession from "../../../../../src/domain/orset/session/StateSession.ts"; +import type StateSessionCloseResult from "../../../../../src/domain/orset/session/StateSessionCloseResult.ts"; import StateSessionError from "../../../../../src/domain/errors/StateSessionError.ts"; import PageCache from "../../../../../src/domain/orset/trie/PageCache.ts"; import TrieGeometry from "../../../../../src/domain/orset/trie/TrieGeometry.ts"; import cborCodec from "../../../../../src/infrastructure/codecs/CborCodec.ts"; import { InMemoryTrieStore } from "../../../../helpers/trieHelpers.ts"; +import MaterializationWorkspacePort, { + type MaterializationWorkspaceRoots, +} from "../../../../../src/ports/MaterializationWorkspacePort.ts"; +import BundleHandle from "../../../../../src/domain/storage/BundleHandle.ts"; +import StorageRetentionWitness, { + StorageRetentionRoot, +} from "../../../../../src/domain/storage/StorageRetentionWitness.ts"; const GEOMETRY = TrieGeometry.default16way(); @@ -17,6 +25,8 @@ async function openSession(args?: { readonly store?: InMemoryTrieStore; readonly pageCache?: PageCache; readonly geometry?: TrieGeometry; + readonly workspace?: MaterializationWorkspacePort; + readonly maxDirtyPages?: number; }) { const store = args?.store ?? new InMemoryTrieStore(); const pageCache = args?.pageCache ?? new PageCache({ maxResident: 32 }); @@ -27,6 +37,8 @@ async function openSession(args?: { codec: cborCodec, geometry: args?.geometry ?? GEOMETRY, pageCache, + ...(args?.workspace === undefined ? {} : { workspace: args.workspace }), + ...(args?.maxDirtyPages === undefined ? {} : { maxDirtyPages: args.maxDirtyPages }), }); return { session, store, pageCache }; } @@ -117,6 +129,22 @@ describe("StateSession", () => { }), ).rejects.toBeInstanceOf(StateSessionError); }); + + it("rejects invalid dirty-page bounds and workspace contracts", async () => { + await expect(openSession({ + workspace: new RecordingWorkspace(), + maxDirtyPages: 0, + })).rejects.toBeInstanceOf( + StateSessionError, + ); + await expect(openSession({ maxDirtyPages: 8 })).rejects.toBeInstanceOf( + StateSessionError, + ); + await expect(openSession({ + // @ts-expect-error intentional runtime validation test + workspace: {}, + })).rejects.toBeInstanceOf(StateSessionError); + }); }); describe("golden path", () => { @@ -178,17 +206,106 @@ describe("StateSession", () => { expect(await reopened.edgeContains("shared:1")).toBe(true); expect(store.readCounts()).toEqual(afterNodeRead); }); + + it("checkpoints and rebases before dirty page residency can grow with the graph", async () => { + const workspace = new RecordingWorkspace(); + const geometry = new TrieGeometry({ + fanout: 16, + nibbleBits: 4, + leafCapacity: 2, + leafFloor: 0, + }); + const { session, store } = await openSession({ + workspace, + maxDirtyPages: 8, + geometry, + }); + + for (let index = 0; index < 100; index += 1) { + await session.addNode(`node:${String(index).padStart(3, "0")}`, new Dot("alice", index + 1)); + expect(session.dirtyPageCount()).toBeLessThan(8); + } + + const roots = await session.close(); + expect(workspace.checkpoints.length).toBeGreaterThan(1); + const reopened = await StateSession.open({ + nodeAliveRootOid: roots.nodeAliveRootOid, + edgeAliveRootOid: roots.edgeAliveRootOid, + store, + codec: cborCodec, + geometry, + pageCache: new PageCache({ maxResident: 32 }), + }); + expect(await scanAll(reopened.scanNodes())).toHaveLength(100); + }); + + it("retains terminal roots before accepting the flush", async () => { + const workspace = new RecordingWorkspace(); + const { session } = await openSession({ workspace }); + + await session.addNode("node:small", new Dot("alice", 1)); + const roots = await session.close(); + + expect(roots.nodeAliveRootOid).not.toBeNull(); + expect(workspace.checkpoints).toEqual([{ + nodeAliveRoot: roots.nodeAliveRootOid, + edgeAliveRoot: roots.edgeAliveRootOid, + }]); + }); + + it("keeps dirty state when a workspace cannot witness the staged root", async () => { + const { session } = await openSession({ + workspace: new NullWitnessWorkspace(), + maxDirtyPages: 1, + }); + + await expect( + session.addNode("node:unwitnessed", new Dot("alice", 1)), + ).rejects.toBeInstanceOf(StateSessionError); + expect(session.dirtyPageCount()).toBeGreaterThan(0); + }); + + it("accepts a prepared terminal flush only after final retention is witnessed", async () => { + const workspace = new RecordingWorkspace(); + const { session } = await openSession({ workspace, maxDirtyPages: 1_000 }); + await session.addNode("node:terminal", new Dot("alice", 1)); + + const prepared = await session.prepareClose(); + + expect(workspace.checkpoints).toEqual([]); + expect(session.dirtyPageCount()).toBeGreaterThan(0); + await expect(session.nodeContains("node:terminal")) + .rejects.toBeInstanceOf(StateSessionError); + expect(() => prepared.accept(null)).toThrow(StateSessionError); + + prepared.accept(finalRetentionWitness(prepared.roots)); + expect(session.dirtyPageCount()).toBe(0); + expect(() => prepared.accept(finalRetentionWitness(prepared.roots))) + .toThrow(StateSessionError); + }); + + it("reopens the session posture when preparing a terminal flush fails", async () => { + const { session } = await openSession({ store: new FailingWriteStore() }); + await session.addNode("node:retryable", new Dot("alice", 1)); + + await expect(session.prepareClose()).rejects.toThrow("terminal flush unavailable"); + + expect(session.dirtyPageCount()).toBeGreaterThan(0); + expect(await session.nodeContains("node:retryable")).toBe(true); + }); }); describe("edge cases", () => { it("keeps nodeAlive and edgeAlive independent inside one session", async () => { const { session } = await openSession(); + const edgeDot = new Dot("alice", 2); await session.addNode("node:only", new Dot("alice", 1)); - await session.addEdge("edge:only", new Dot("alice", 2)); + await session.addEdge("edge:only", edgeDot); expect(await session.nodeContains("node:only")).toBe(true); expect(await session.edgeContains("edge:only")).toBe(true); + expect([...await session.edgeDots("edge:only")]).toEqual([Dot.encode(edgeDot)]); expect(await session.nodeContains("edge:only")).toBe(false); expect(await session.edgeContains("node:only")).toBe(false); }); @@ -220,6 +337,28 @@ describe("StateSession", () => { expect(await scanAll(reopened.scanEdges())).toEqual([]); }); + it("rejects compaction before traversing a bounded session", async () => { + const workspace = new RecordingWorkspace(); + const { session } = await openSession({ workspace, maxDirtyPages: 2 }); + for (let index = 0; index < 32; index += 1) { + const dot = new Dot("alice", index + 1); + await session.addNode(`node:${String(index)}`, dot); + await session.removeNode(`node:${String(index)}`, new Set([Dot.encode(dot)])); + } + const checkpointsBeforeCompaction = workspace.checkpoints.length; + const dirtyPagesBeforeCompaction = session.dirtyPageCount(); + + await expect( + session.compact(VersionVector.from({ alice: 32 })), + ).rejects.toMatchObject({ + code: "E_STATE_SESSION_INPUT", + message: expect.stringContaining("resumable subtree checkpoints"), + }); + + expect(workspace.checkpoints).toHaveLength(checkpointsBeforeCompaction); + expect(session.dirtyPageCount()).toBe(dirtyPagesBeforeCompaction); + }); + it("surfaces tombstoned element state without pretending removed entries vanished", async () => { const { session } = await openSession(); const nodeDot = new Dot("alice", 1); @@ -305,3 +444,72 @@ describe("StateSession", () => { }); }); }); + +class FailingWriteStore extends InMemoryTrieStore { + override writeLeaf(_data: Uint8Array): Promise { + return Promise.reject(new Error("terminal flush unavailable")); + } +} + +class RecordingWorkspace extends MaterializationWorkspacePort { + readonly checkpoints: MaterializationWorkspaceRoots[] = []; + + override checkpoint(roots: MaterializationWorkspaceRoots): Promise { + this.checkpoints.push(roots); + const handle = new BundleHandle(`test:workspace:${String(this.checkpoints.length)}`); + return Promise.resolve(new StorageRetentionWitness({ + handle, + policy: "pinned", + reachability: "anchored", + root: new StorageRetentionRoot({ + kind: "cache-set", + namespace: "test/materialization-workspaces", + locator: "test/materialization-workspaces", + generation: `generation-${String(this.checkpoints.length)}`, + path: handle.toString(), + }), + observedAt: "1970-01-01T00:00:00.000Z", + })); + } + + override release(): Promise { + return Promise.resolve(); + } + + override promote(): Promise { + return Promise.reject(new Error("StateSession tests do not promote materializations")); + } +} + +class NullWitnessWorkspace extends MaterializationWorkspacePort { + override checkpoint(): Promise { + return Promise.resolve(null); + } + + override release(): Promise { + return Promise.resolve(); + } + + override promote(): Promise { + return Promise.reject(new Error("StateSession tests do not promote materializations")); + } +} + +function finalRetentionWitness(roots: StateSessionCloseResult): StorageRetentionWitness { + const handle = new BundleHandle( + roots.nodeAliveRootOid ?? roots.edgeAliveRootOid ?? "test:empty-materialization", + ); + return new StorageRetentionWitness({ + handle, + policy: "evictable", + reachability: "anchored", + root: new StorageRetentionRoot({ + kind: "cache-set", + namespace: "test/materializations", + locator: "test/materializations", + generation: "final-generation", + path: handle.toString(), + }), + observedAt: "1970-01-01T00:00:00.000Z", + }); +} diff --git a/test/unit/domain/orset/trie/TrieCursor.test.ts b/test/unit/domain/orset/trie/TrieCursor.test.ts index adff34a2a..99fcd6fbb 100644 --- a/test/unit/domain/orset/trie/TrieCursor.test.ts +++ b/test/unit/domain/orset/trie/TrieCursor.test.ts @@ -117,6 +117,12 @@ describe("TrieCursor", () => { TrieCursorError, ); }); + + it("rejects an empty rebase root", () => { + const { cursor } = makeCursor(); + + expect(() => cursor.rebase("")).toThrow(TrieCursorError); + }); }); describe("single-add insertion", () => { diff --git a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts index 2e7cbc92c..b2411e332 100644 --- a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts +++ b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts @@ -18,6 +18,8 @@ import Patch from "../../../../../src/domain/types/Patch.ts"; import type { CheckpointData, PatchWithSha } from "../../../../../src/domain/capabilities/PatchCollector.ts"; import InMemoryCheckpointStore from "../../../../helpers/InMemoryCheckpointStore.ts"; import InMemoryMaterializationStore from "../../../../helpers/InMemoryMaterializationStore.ts"; +import type MaterializationWorkspacePort from "../../../../../src/ports/MaterializationWorkspacePort.ts"; +import MaterializationCoordinate from "../../../../../src/domain/materialization/MaterializationCoordinate.ts"; const GEOMETRY = TrieGeometry.default16way(); @@ -123,6 +125,7 @@ function createControllerFixtures() { loadCheckpoint: vi.fn().mockResolvedValue(null), loadPatchesSince: vi.fn<(_checkpoint: CheckpointData) => Promise>().mockResolvedValue([]), loadPatchChain: vi.fn<(_toSha: string, _fromSha?: string | null) => Promise>().mockResolvedValue([]), + isAncestor: vi.fn().mockResolvedValue(false), getFrontier: vi.fn().mockResolvedValue(new Map([["writer-1", "tip-1"]])), streamWriterPatches: vi.fn((writerId: string) => streamFromPromise(patches.loadWriterPatches(writerId))), streamForFrontier: vi.fn((frontier: Map, ceiling: number | null) => @@ -142,7 +145,7 @@ function createControllerFixtures() { async (roots: { readonly nodeAliveRootOid: string | null; readonly edgeAliveRootOid: string | null; - }): Promise => + }, options: { readonly workspace: MaterializationWorkspacePort }): Promise => await StateSession.open({ nodeAliveRootOid: roots.nodeAliveRootOid, edgeAliveRootOid: roots.edgeAliveRootOid, @@ -150,6 +153,8 @@ function createControllerFixtures() { codec: cborCodec, geometry: GEOMETRY, pageCache, + maxDirtyPages: 1, + workspace: options.workspace, }), ); @@ -194,8 +199,8 @@ function createControllerFixtures() { } describe("MaterializeController — state session integration", () => { - it("does not flush partial trie roots when session reduction fails", async () => { - const { openStateSession } = createControllerFixtures(); + it("releases checkpointed workspace roots when session reduction fails", async () => { + const { openStateSession, materializations } = createControllerFixtures(); const close = vi.spyOn(StateSession.prototype, "close"); const failure = new Error("patch source failed"); const partial = new PatchEntry(nodeAddPatchRecord({ @@ -214,11 +219,18 @@ describe("MaterializeController — state session integration", () => { try { await expect(reduceSessionBackedState({ openStateSession, + materializations, + coordinate: new MaterializationCoordinate({ + frontier: new Map([["writer-1", "deadbeef"]]), + ceiling: null, + }), patches, receipts: false, wantDiff: false, })).rejects.toBe(failure); expect(close).not.toHaveBeenCalled(); + expect(materializations.workspaces[0]?.checkpoints).toHaveLength(1); + expect(materializations.workspaces[0]?.released).toBe(true); } finally { close.mockRestore(); } @@ -251,10 +263,11 @@ describe("MaterializeController — state session integration", () => { const result = await controller.materialize({}); - expect(openStateSession).toHaveBeenCalledWith({ + expect(openStateSession.mock.calls[0]?.[0]).toEqual({ nodeAliveRootOid: null, edgeAliveRootOid: null, }); + expect(openStateSession.mock.calls[0]?.[1]?.workspace).toBeDefined(); expect(result.state.nodeAlive.contains("node:session")).toBe(true); expect(result.patchCount).toBe(3); expect(result.adjacency.outgoing.get("node:session")).toEqual([ @@ -265,6 +278,93 @@ describe("MaterializeController — state session integration", () => { expect(materializations.retainedRequests[0]?.roots.edgeAlive.status).toBe("retained"); expect(materializations.retainedRequests[0]?.roots.properties.status).toBe("unavailable"); expect(result.materialization?.roots.nodeAlive.status).toBe("retained"); + expect(materializations.workspaces[0]?.released).toBe(true); + }); + + it("releases checkpointed roots when final materialization retention fails", async () => { + const { controller, patches, materializations } = createControllerFixtures(); + patches.collectForFrontier.mockResolvedValue([ + nodeAddPatchRecord({ + writer: "writer-1", + lamport: 1, + sha: "a1b2", + node: "node:session", + }), + ]); + const failure = new Error("final retention failed"); + vi.spyOn(materializations, "retain").mockRejectedValue(failure); + + await expect(controller.materialize()).rejects.toBe(failure); + expect(materializations.workspaces[0]?.checkpoints).toHaveLength(1); + expect(materializations.workspaces[0]?.released).toBe(true); + }); + + it("releases an unused workspace before returning an empty result", async () => { + const { controller, materializations } = createControllerFixtures(); + + const result = await controller.materialize(); + + expect(result.patchCount).toBe(0); + expect(materializations.workspaces[0]?.checkpoints).toEqual([]); + expect(materializations.workspaces[0]?.released).toBe(true); + }); + + it("retains coordinate-keyed roots when the legacy snapshot cache is disabled", async () => { + const fixtures = createControllerFixtures(); + fixtures.patches.collectForFrontier.mockResolvedValue([ + nodeAddPatchRecord({ + writer: "writer-1", + lamport: 1, + sha: "a1b2", + node: "node:no-snapshot-cache", + }), + ]); + const controller = new MaterializeController({ + ...fixtures.deps, + getStateCache: () => null, + }); + + const result = await controller.materialize(); + + expect(result.materialization).toBeDefined(); + expect(fixtures.materializations.retainedRequests).toHaveLength(1); + expect(fixtures.materializations.retainedRequests[0]?.coordinate.frontierEntries) + .toEqual([{ writerId: "writer-1", patchSha: "tip-1" }]); + expect(fixtures.materializations.workspaces[0]?.released).toBe(true); + }); + + it("retains checkpoint-suffix roots when the legacy snapshot cache is disabled", async () => { + const fixtures = createControllerFixtures(); + const checkpoint = { + schema: 5, + state: snapshotRecord({ + frontier: new Map([["writer-1", "tip-0"]]), + ceiling: null, + }).state, + frontier: new Map([["writer-1", "tip-0"]]), + stateHash: "checkpoint-hash", + }; + fixtures.patches.loadCheckpoint.mockResolvedValue(checkpoint); + fixtures.patches.collectForFrontierSinceCoordinate.mockResolvedValue([ + nodeAddPatchRecord({ + writer: "writer-1", + lamport: 2, + sha: "b2c3", + node: "node:checkpoint-suffix", + }), + ]); + fixtures.patches.isAncestor.mockResolvedValue(true); + const controller = new MaterializeController({ + ...fixtures.deps, + getStateCache: () => null, + }); + + const result = await controller.materialize(); + + expect(result.state.nodeAlive.contains("node:base")).toBe(true); + expect(result.state.nodeAlive.contains("node:checkpoint-suffix")).toBe(true); + expect(fixtures.materializations.retainedRequests).toHaveLength(1); + expect(fixtures.materializations.workspaces[0]?.released).toBe(true); }); it("reopens exact retained roots with zero covered patch replay after controller restart", async () => { @@ -301,7 +401,7 @@ describe("MaterializeController — state session integration", () => { expect(fixtures.openStateSession).toHaveBeenNthCalledWith(1, { nodeAliveRootOid: retained?.roots.nodeAlive.handle?.toString(), edgeAliveRootOid: retained?.roots.edgeAlive.handle?.toString() ?? null, - }); + }, expect.objectContaining({ workspace: expect.anything() })); expect(warm.patchCount).toBe(0); expect(reopened.patchCount).toBe(0); expect(warm.state.nodeAlive.contains("node:retained")).toBe(true); @@ -337,7 +437,7 @@ describe("MaterializeController — state session integration", () => { expect(openStateSession).toHaveBeenCalledWith({ nodeAliveRootOid: null, edgeAliveRootOid: null, - }); + }, expect.objectContaining({ workspace: expect.anything() })); expect(stateCache.put).toHaveBeenCalledWith( expect.objectContaining({ coordinate: target, diff --git a/test/unit/domain/services/controllers/MaterializeController.test.ts b/test/unit/domain/services/controllers/MaterializeController.test.ts index a096a96ce..811ddec5f 100644 --- a/test/unit/domain/services/controllers/MaterializeController.test.ts +++ b/test/unit/domain/services/controllers/MaterializeController.test.ts @@ -78,6 +78,39 @@ function makeMockPatches(overrides = {}) { getFrontier: vi.fn().mockResolvedValue(new Map()), ...overrides, }; + if (!Object.hasOwn(overrides, 'getFrontier')) { + patches.getFrontier.mockImplementation(async () => { + const checkpoint = await patches.loadCheckpoint(); + const frontier = new Map(checkpoint?.frontier ?? []); + if (checkpoint !== null && checkpoint !== undefined) { + for (const entry of await patches.loadPatchesSince(checkpoint)) { + frontier.set(entry.patch.writer, entry.sha); + } + return frontier; + } + for (const writerId of await patches.discoverWriters()) { + frontier.set(writerId, `tip-${writerId}`); + } + return frontier; + }); + } + if (!Object.hasOwn(overrides, 'collectForFrontier')) { + patches.collectForFrontier.mockImplementation(async (frontier: Map) => { + const entries: ReturnType[] = []; + for (const writerId of frontier.keys()) { + entries.push(...await patches.loadWriterPatches(writerId)); + } + return entries; + }); + } + if (!Object.hasOwn(overrides, 'collectForFrontierSinceCoordinate')) { + patches.collectForFrontierSinceCoordinate.mockImplementation(async () => { + const checkpoint = await patches.loadCheckpoint(); + return checkpoint === null || checkpoint === undefined + ? [] + : await patches.loadPatchesSince(checkpoint); + }); + } return { ...patches, streamWriterPatches: vi.fn((writerId: string) => streamFromPromise(patches.loadWriterPatches(writerId))), @@ -209,13 +242,13 @@ describe('MaterializeController', () => { expect(result.adjacency).toBeInstanceOf(AdjacencyMap); }); - it('returns null frontier and null ceiling for live materialization', async () => { + it('returns the observed empty frontier and null ceiling for live materialization', async () => { const { ctrl, patches } = setup(); patches.discoverWriters.mockResolvedValue([]); const result = await ctrl.materialize({}); - expect(result.frontier).toBeNull(); + expect(result.frontier).toEqual(new Map()); expect(result.ceiling).toBeNull(); }); }); diff --git a/test/unit/domain/services/controllers/MaterializePatchStreamReducer.test.ts b/test/unit/domain/services/controllers/MaterializePatchStreamReducer.test.ts index c5c67b268..dbb62d2b7 100644 --- a/test/unit/domain/services/controllers/MaterializePatchStreamReducer.test.ts +++ b/test/unit/domain/services/controllers/MaterializePatchStreamReducer.test.ts @@ -46,7 +46,7 @@ describe('MaterializePatchStreamReducer', () => { }); describe('MaterializeController patch streams', () => { - it('materializes live graphs from streamWriterPatches without loading writer arrays', async () => { + it('materializes live graphs from the declared frontier stream without loading arrays', async () => { const collector = new StreamingOnlyPatchCollector([ patchEntry(1), patchEntry(2), @@ -171,6 +171,27 @@ class StreamingOnlyPatchCollector extends PatchCollector { yield* ephemeralEntries(this.#entries); } + override async *streamForFrontier( + frontier: Map, + ceiling: number | null, + ): AsyncIterable { + for (const [writerId, tipSha] of frontier) { + let reachedTip = false; + for await (const entry of this.streamWriterPatches(writerId)) { + if (ceiling === null || entry.patch.lamport <= ceiling) { + yield entry; + } + if (entry.sha === tipSha) { + reachedTip = true; + break; + } + } + if (!reachedTip) { + throw new Error(`frontier tip ${tipSha} was not present in the writer stream`); + } + } + } + async loadCheckpoint(): Promise { return null; } diff --git a/test/unit/domain/warp/hydrateCheckpointIndex.regression.test.ts b/test/unit/domain/warp/hydrateCheckpointIndex.regression.test.ts index 2f5235209..bf6a390f9 100644 --- a/test/unit/domain/warp/hydrateCheckpointIndex.regression.test.ts +++ b/test/unit/domain/warp/hydrateCheckpointIndex.regression.test.ts @@ -39,6 +39,7 @@ function buildState(nodes, edges) { describe('materialize stale-checkpoint regression', () => { it('does not overwrite freshly built index with stale checkpoint index data', async () => { + const tipSha = 'a'.repeat(40); const latestState = buildState( ['A', 'B', 'C'], [['A', 'B', 'knows'], ['A', 'C', 'manages']], @@ -57,14 +58,16 @@ describe('materialize stale-checkpoint regression', () => { loadCheckpoint: async () => ({ schema: 5, state: latestState, - frontier: new Map(), + frontier: new Map([['w1', tipSha]]), provenanceIndex: null, }), loadPatchesSince: async () => [], discoverWriters: async () => [], loadWriterPatches: async () => [], collectForFrontier: async () => [], - getFrontier: async () => new Map(), + streamForFrontier: async function* () {}, + streamForFrontierSinceCoordinate: async function* () {}, + getFrontier: async () => new Map([['w1', tipSha]]), loadPatchChain: async () => [], }, }) as any); diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts index d3bb71fba..e3866349d 100644 --- a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts +++ b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.test.ts @@ -14,6 +14,7 @@ import InMemoryGitCasFacade from '../../../helpers/InMemoryGitCasFacade.ts'; import InMemoryGraphAdapter from '../../../helpers/InMemoryGraphAdapter.ts'; const CACHE_NAMESPACE = 'git-warp/materializations'; +const WORKSPACE_CACHE_NAMESPACE = 'git-warp/materialization-workspaces'; const ROOT_PATHS = Object.freeze([ 'roots/adjacency', 'roots/edge-alive', @@ -73,6 +74,53 @@ describe('GitCasMaterializationStoreAdapter', () => { expect(await harness.adapter.findExact(exactCoordinate())).toBeNull(); }); + it('promotes terminal roots without installing an unnecessary workspace entry', async () => { + const harness = await createHarness(); + const coordinate = exactCoordinate(); + const roots = await createRoots(harness.cas); + const workspace = await harness.adapter.openWorkspace(coordinate); + + expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toEqual([]); + + const promoted = await workspace.promote({ + coordinate, + roots, + stateHash: 'promoted-state-hash', + }); + await workspace.release(); + + 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); + }); + + it('removes an in-progress coordinate after mismatched promotion fails', async () => { + const harness = await createHarness(); + const roots = await createRoots(harness.cas); + const workspace = await harness.adapter.openWorkspace(exactCoordinate()); + await workspace.checkpoint({ + nodeAliveRoot: roots.nodeAlive.handle?.toString() ?? null, + edgeAliveRoot: null, + }); + expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toHaveLength(1); + + await expect(workspace.promote({ + coordinate: new MaterializationCoordinate({ + frontier: new Map([['writer-c', 'patch-c']]), + ceiling: 13, + }), + roots, + stateHash: 'wrong-coordinate', + })).rejects.toMatchObject({ + code: 'E_MATERIALIZATION_STORAGE', + message: expect.stringContaining('does not match'), + }); + await workspace.release(); + + expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toEqual([]); + expect(harness.cas.readCacheKeys(CACHE_NAMESPACE)).toEqual([]); + }); + it('round-trips partial roots without inventing bundles for empty or unavailable state', async () => { const harness = await createHarness(); const page = await harness.cas.pages.put({ source: new Uint8Array([1]) }); @@ -468,6 +516,7 @@ function withCacheResult( put: async (key, handle, entryOptions) => rewrite( await cache.put(key, handle, entryOptions), ), + remove: async (key) => await cache.remove(key), }; }, }, diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationWorkspace.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationWorkspace.test.ts new file mode 100644 index 000000000..a8102a3c7 --- /dev/null +++ b/test/unit/infrastructure/adapters/GitCasMaterializationWorkspace.test.ts @@ -0,0 +1,492 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { CacheStoreResult } from '@git-stunts/git-cas'; +import MaterializationCoordinate from '../../../../src/domain/materialization/MaterializationCoordinate.ts'; +import GitCasMaterializationStoreAdapter, { + type GitCasMaterializationFacade, +} from '../../../../src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts'; +import GitCasMaterializationWorkspace from '../../../../src/infrastructure/adapters/GitCasMaterializationWorkspace.ts'; +import type { + MaterializationWorkspaceLease, + MaterializationWorkspaceLeaseScheduler, +} from '../../../../src/infrastructure/adapters/GitCasMaterializationWorkspace.ts'; +import type { PromoteMaterializationRequest } from '../../../../src/ports/MaterializationWorkspacePort.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 WORKSPACE_CACHE_NAMESPACE = 'git-warp/materialization-workspaces'; + +describe('GitCasMaterializationWorkspace', () => { + it('retains the latest in-progress roots in a separate expiring workspace', async () => { + const harness = await createHarness(); + const roots = await createRoots(harness.cas); + const workspace = await harness.adapter.openWorkspace(workspaceCoordinate()); + + const witness = await workspace.checkpoint(roots); + + expect(witness).toMatchObject({ + policy: 'pinned', + reachability: 'anchored', + root: { + kind: 'cache-set', + namespace: WORKSPACE_CACHE_NAMESPACE, + }, + }); + expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toHaveLength(1); + expect(harness.cas.readCacheHits(WORKSPACE_CACHE_NAMESPACE)[0]?.expiresAt).not.toBeNull(); + + await workspace.release(); + await workspace.release(); + expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toEqual([]); + }); + + it('keeps empty and released workspaces side-effect free', async () => { + const harness = await createHarness(); + const workspace = await harness.adapter.openWorkspace(workspaceCoordinate()); + + expect(await workspace.checkpoint({ + nodeAliveRoot: null, + edgeAliveRoot: null, + })).toBeNull(); + await workspace.release(); + await workspace.release(); + + expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toEqual([]); + await expect(workspace.checkpoint({ + nodeAliveRoot: null, + edgeAliveRoot: null, + })).rejects.toMatchObject({ code: 'E_MATERIALIZATION_STORAGE' }); + }); + + it('fails closed when git-cas declines workspace retention', async () => { + const harness = await createHarness(); + const roots = await createRoots(harness.cas); + const facade = withCacheResult(harness.cas, (stored) => Object.freeze({ + ...stored, + accepted: false, + hit: null, + witness: null, + })); + const workspace = await adapterFor(facade).openWorkspace(workspaceCoordinate()); + + await expect(workspace.checkpoint(roots)).rejects.toMatchObject({ + code: 'E_MATERIALIZATION_STORAGE', + message: expect.stringContaining('did not retain'), + }); + await workspace.release(); + expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toEqual([]); + }); + + it('cleans up when git-cas reports an unexpected workspace target', async () => { + const harness = await createHarness(); + const roots = await createRoots(harness.cas); + const page = await harness.cas.pages.put({ source: new Uint8Array([9]) }); + const cache = await harness.cas.caches.open({ namespace: WORKSPACE_CACHE_NAMESPACE }); + const unexpected = await cache.put('unexpected-workspace-target', page.handle); + if (unexpected.hit === null) { + throw new Error('Expected an unexpected workspace cache hit'); + } + const facade = withCacheResult(harness.cas, (stored) => Object.freeze({ + ...stored, + hit: unexpected.hit, + })); + const workspace = await adapterFor(facade).openWorkspace(workspaceCoordinate()); + + await expect(workspace.checkpoint(roots)).rejects.toMatchObject({ + code: 'E_MATERIALIZATION_STORAGE', + message: expect.stringContaining('unexpected workspace handle'), + }); + await workspace.release(); + expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toEqual([ + 'unexpected-workspace-target', + ]); + }); + + it('serializes release behind an in-flight workspace checkpoint', async () => { + const harness = await createHarness(); + const roots = await createRoots(harness.cas); + const cache = await harness.cas.caches.open({ namespace: WORKSPACE_CACHE_NAMESPACE }); + let admitPut: () => void = () => undefined; + const putGate = new Promise((resolve) => { + admitPut = resolve; + }); + const workspace = new GitCasMaterializationWorkspace({ + bundles: harness.cas.bundles, + cache: { + put: async (...args) => { + await putGate; + return await cache.put(...args); + }, + remove: async (key) => await cache.remove(key), + }, + key: 'concurrent-release', + promote: rejectPromotion, + }); + + const checkpoint = workspace.checkpoint(roots); + const release = workspace.release(); + admitPut(); + + await checkpoint; + await release; + expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toEqual([]); + await expect(workspace.checkpoint({ + nodeAliveRoot: null, + edgeAliveRoot: null, + })).rejects.toMatchObject({ code: 'E_MATERIALIZATION_STORAGE' }); + }); + + it('renews an active lease without waiting for another root checkpoint', async () => { + const harness = await createHarness(); + const roots = await createRoots(harness.cas); + const cache = await harness.cas.caches.open({ namespace: WORKSPACE_CACHE_NAMESPACE }); + const scheduler = new ManualLeaseScheduler(); + let now = Date.parse('2026-07-16T00:00:00.000Z'); + const workspace = new GitCasMaterializationWorkspace({ + bundles: harness.cas.bundles, + cache, + key: 'renewing-workspace', + clock: { now: () => new Date(now) }, + leaseTtlMs: 100, + leaseRenewalMs: 40, + leaseScheduler: scheduler, + promote: rejectPromotion, + }); + + await workspace.checkpoint(roots); + const firstExpiry = requireWorkspaceExpiry(harness.cas); + now += 40; + await scheduler.runNext(); + const renewedExpiry = requireWorkspaceExpiry(harness.cas); + + expect(renewedExpiry).toBeGreaterThan(firstExpiry); + expect(scheduler.pending).toBe(true); + await workspace.release(); + expect(scheduler.pending).toBe(false); + }); + + it('runs lease renewal through the production scheduler callback', async () => { + vi.useFakeTimers(); + try { + const harness = await createHarness(); + const roots = await createRoots(harness.cas); + const cache = await harness.cas.caches.open({ namespace: WORKSPACE_CACHE_NAMESPACE }); + const workspace = new GitCasMaterializationWorkspace({ + bundles: harness.cas.bundles, + cache, + key: 'system-scheduler', + leaseTtlMs: 100, + leaseRenewalMs: 40, + promote: rejectPromotion, + }); + + await workspace.checkpoint(roots); + const firstExpiry = requireWorkspaceExpiry(harness.cas); + await vi.advanceTimersByTimeAsync(40); + + expect(requireWorkspaceExpiry(harness.cas)).toBeGreaterThan(firstExpiry); + await workspace.release(); + } finally { + vi.useRealTimers(); + } + }); + + it('ignores lease timers that fire after release or target replacement', async () => { + const harness = await createHarness(); + const roots = await createRoots(harness.cas); + const cache = await harness.cas.caches.open({ namespace: WORKSPACE_CACHE_NAMESPACE }); + const scheduler = new ManualLeaseScheduler(); + const workspace = new GitCasMaterializationWorkspace({ + bundles: harness.cas.bundles, + cache, + key: 'stale-renewal', + leaseTtlMs: 100, + leaseRenewalMs: 40, + leaseScheduler: scheduler, + promote: rejectPromotion, + }); + + await workspace.checkpoint(roots); + const replacedTargetTimer = scheduler.takeNext(); + const replacement = await createRoots(harness.cas); + await workspace.checkpoint(replacement); + await replacedTargetTimer(); + const releasedTargetTimer = scheduler.takeNext(); + await workspace.release(); + await releasedTargetTimer(); + + expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toEqual([]); + }); + + it('fails closed after a lease renewal error and still removes the workspace', async () => { + const harness = await createHarness(); + const roots = await createRoots(harness.cas); + const cache = await harness.cas.caches.open({ namespace: WORKSPACE_CACHE_NAMESPACE }); + const scheduler = new ManualLeaseScheduler(); + let puts = 0; + const workspace = new GitCasMaterializationWorkspace({ + bundles: harness.cas.bundles, + cache: { + put: async (...args) => { + puts += 1; + if (puts === 2) { + throw new Error('renewal unavailable'); + } + return await cache.put(...args); + }, + remove: async (key) => await cache.remove(key), + }, + key: 'failing-renewal', + leaseTtlMs: 100, + leaseRenewalMs: 40, + leaseScheduler: scheduler, + promote: rejectPromotion, + }); + + await workspace.checkpoint(roots); + await scheduler.runNext(); + + await expect(workspace.release()).rejects.toMatchObject({ + code: 'E_MATERIALIZATION_STORAGE', + message: expect.stringContaining('lease renewal failed'), + }); + expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toEqual([]); + }); + + it('continues renewing while final materialization promotion is in flight', async () => { + const harness = await createHarness(); + const roots = await createRoots(harness.cas); + const cache = await harness.cas.caches.open({ namespace: WORKSPACE_CACHE_NAMESPACE }); + const scheduler = new ManualLeaseScheduler(); + let now = Date.parse('2026-07-16T00:00:00.000Z'); + let rejectPromotionGate: (reason: Error) => void = () => undefined; + const promotionFailure = new Error('final retention unavailable'); + const promotionGate = new Promise((_resolve, reject) => { + rejectPromotionGate = reject; + }); + let signalPromotionStarted: () => void = () => undefined; + const promotionStarted = new Promise((resolve) => { + signalPromotionStarted = resolve; + }); + const workspace = new GitCasMaterializationWorkspace({ + bundles: harness.cas.bundles, + cache, + key: 'long-promotion', + clock: { now: () => new Date(now) }, + leaseTtlMs: 100, + leaseRenewalMs: 40, + leaseScheduler: scheduler, + promote: async (_request) => { + signalPromotionStarted(); + return await promotionGate; + }, + }); + + await workspace.checkpoint(roots); + const firstExpiry = requireWorkspaceExpiry(harness.cas); + const promotion = workspace.promote({} as PromoteMaterializationRequest); + const promotionAssertion = expect(promotion).rejects.toBe(promotionFailure); + await promotionStarted; + const release = workspace.release(); + await expect(workspace.promote({} as PromoteMaterializationRequest)).rejects.toMatchObject({ + code: 'E_MATERIALIZATION_STORAGE', + message: expect.stringContaining('cannot promote'), + }); + now += 40; + await scheduler.runNext(); + + expect(requireWorkspaceExpiry(harness.cas)).toBeGreaterThan(firstExpiry); + rejectPromotionGate(promotionFailure); + await promotionAssertion; + await release; + expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toEqual([]); + }); + + it('rejects malformed roots and invalid clocks before retention', async () => { + const harness = await createHarness(); + const cache = await harness.cas.caches.open({ namespace: WORKSPACE_CACHE_NAMESPACE }); + const workspace = new GitCasMaterializationWorkspace({ + bundles: harness.cas.bundles, + cache, + key: 'invalid-clock', + clock: { now: () => new Date(Number.NaN) }, + promote: rejectPromotion, + }); + + await expect(workspace.checkpoint({ + nodeAliveRoot: 'not-a-bundle', + edgeAliveRoot: null, + })).rejects.toMatchObject({ + code: 'E_MATERIALIZATION_STORAGE', + message: expect.stringContaining('not a bundle handle'), + }); + await expect(Reflect.apply(workspace.checkpoint, workspace, [null])) + .rejects.toMatchObject({ + code: 'E_MATERIALIZATION_STORAGE', + message: expect.stringContaining('roots must be an object'), + }); + + const roots = await createRoots(harness.cas); + await expect(workspace.checkpoint(roots)).rejects.toMatchObject({ + code: 'E_MATERIALIZATION_STORAGE', + message: expect.stringContaining('invalid Date'), + }); + }); + + it('validates workspace dependencies and options', async () => { + const harness = await createHarness(); + const cache = await harness.cas.caches.open({ namespace: WORKSPACE_CACHE_NAMESPACE }); + const valid = { + bundles: harness.cas.bundles, + cache, + key: 'workspace-options', + promote: rejectPromotion, + }; + + expect(() => Reflect.construct(GitCasMaterializationWorkspace, [null])) + .toThrowError(/options/u); + for (const field of ['bundles', 'cache', 'promote']) { + const options = { ...valid }; + Reflect.set(options, field, null); + expect(() => Reflect.construct(GitCasMaterializationWorkspace, [options])) + .toThrowError(new RegExp(`${field} dependency`, 'u')); + } + expect(() => new GitCasMaterializationWorkspace({ + ...valid, + key: '', + })).toThrowError(/key/u); + expect(() => Reflect.construct(GitCasMaterializationWorkspace, [{ + ...valid, + clock: {}, + }])).toThrowError(/clock/u); + expect(() => new GitCasMaterializationWorkspace({ + ...valid, + leaseTtlMs: 40, + leaseRenewalMs: 40, + })).toThrowError(/leaseRenewalMs/u); + expect(() => new GitCasMaterializationWorkspace({ + ...valid, + leaseTtlMs: 0, + })).toThrowError(/leaseTtlMs/u); + expect(() => Reflect.construct(GitCasMaterializationWorkspace, [{ + ...valid, + leaseScheduler: {}, + }])).toThrowError(/leaseScheduler/u); + }); +}); + +type WorkspaceRoots = Readonly<{ + nodeAliveRoot: string; + edgeAliveRoot: string; +}>; + +type Harness = Readonly<{ + adapter: GitCasMaterializationStoreAdapter; + cas: InMemoryGitCasFacade; +}>; + +async function createHarness(): Promise { + const cas = new InMemoryGitCasFacade({ + history: new InMemoryGraphAdapter(), + storage: new InMemoryBlobStorageAdapter(), + }); + return Object.freeze({ + cas, + adapter: adapterFor(cas), + }); +} + +async function createRoots(cas: InMemoryGitCasFacade): Promise { + const nodePage = await cas.pages.put({ source: new Uint8Array([1]) }); + const edgePage = await cas.pages.put({ source: new Uint8Array([2]) }); + const nodeBundle = await cas.bundles.putOrdered({ members: [['root', nodePage.handle]] }); + const edgeBundle = await cas.bundles.putOrdered({ members: [['root', edgePage.handle]] }); + return Object.freeze({ + nodeAliveRoot: nodeBundle.handle.toString(), + edgeAliveRoot: edgeBundle.handle.toString(), + }); +} + +function adapterFor(cas: GitCasMaterializationFacade): GitCasMaterializationStoreAdapter { + return new GitCasMaterializationStoreAdapter({ + cas, + codec: defaultCodec, + crypto: new NodeCryptoAdapter(), + laneName: 'events', + }); +} + +function withCacheResult( + cas: InMemoryGitCasFacade, + rewrite: (stored: CacheStoreResult) => CacheStoreResult, +): GitCasMaterializationFacade { + return { + bundles: cas.bundles, + pages: cas.pages, + caches: { + open: async (options) => { + const cache = await cas.caches.open(options); + return { + get: async (key) => await cache.get(key), + put: async (key, handle, entryOptions) => rewrite( + await cache.put(key, handle, entryOptions), + ), + remove: async (key) => await cache.remove(key), + }; + }, + }, + }; +} + +function requireWorkspaceExpiry(cas: InMemoryGitCasFacade): number { + const expiresAt = cas.readCacheHits(WORKSPACE_CACHE_NAMESPACE)[0]?.expiresAt; + if (expiresAt === undefined || expiresAt === null) { + throw new Error('Expected the workspace to have an expiry'); + } + return Date.parse(expiresAt); +} + +function workspaceCoordinate(): MaterializationCoordinate { + return new MaterializationCoordinate({ + frontier: new Map([['writer-a', 'patch-a']]), + ceiling: null, + }); +} + +function rejectPromotion(): Promise { + return Promise.reject(new Error('Promotion is not used by this workspace lifecycle test')); +} + +class ManualLeaseScheduler implements MaterializationWorkspaceLeaseScheduler { + #task: (() => Promise) | null = null; + + get pending(): boolean { + return this.#task !== null; + } + + schedule(task: () => Promise, _delayMs: number): MaterializationWorkspaceLease { + this.#task = task; + return Object.freeze({ + cancel: () => { + if (this.#task === task) { + this.#task = null; + } + }, + }); + } + + async runNext(): Promise { + await this.takeNext()(); + } + + takeNext(): () => Promise { + const task = this.#task; + if (task === null) { + throw new Error('Expected a scheduled lease renewal'); + } + this.#task = null; + return task; + } +} diff --git a/test/unit/scripts/v18-graph-model-migration-command-cli.test.ts b/test/unit/scripts/v18-graph-model-migration-command-cli.test.ts index 78eb22a02..311ae7bba 100644 --- a/test/unit/scripts/v18-graph-model-migration-command-cli.test.ts +++ b/test/unit/scripts/v18-graph-model-migration-command-cli.test.ts @@ -25,6 +25,7 @@ const ALICE_HEAD = '417fe95095a6feae3042c36505065bbd7b3d2a67'; const BOB_HEAD = 'd7c3a05b3894d5c3c151e03dd972b6bd6c341b0c'; const REVIEWED_LIVE_REF = 'refs/warp/v17-golden-graph/live'; const REVIEWED_ARCHIVE_REF = 'refs/warp-migration-archive/v17-golden-graph/cli/live'; +const SCRATCH_HEAD = 'eb75f3e966f5240f35106952fc42a46872df1300'; describe('v18 graph-model migration command CLI', () => { it('prints usage when help is requested', async () => { @@ -71,12 +72,13 @@ describe('v18 graph-model migration command CLI', () => { expect(result.stdout).toBe(report); expect(report).toContain('scratch: written'); expect(report).toContain(`scratchRef: ${SCRATCH_REF}`); + expect(report).toContain(`scratchHead: ${SCRATCH_HEAD}`); expect(report).toContain('equivalence: passed'); expect(report).toContain('finalization: skipped'); }); it('finalizes live refs only when the reviewed request matches command evidence', async () => { - const scratchHead = await previewScratchHead(); + const scratchHead = SCRATCH_HEAD; const directory = await mkdtemp(join(tmpdir(), 'git-warp-v18-command-cli-finalize-')); const restoreResult = await restoreV17GoldenGraphFixture({ @@ -109,7 +111,7 @@ describe('v18 graph-model migration command CLI', () => { }); it('blocks finalization when the reviewed live ref head drifts', async () => { - const scratchHead = await previewScratchHead(); + const scratchHead = SCRATCH_HEAD; const directory = await mkdtemp(join(tmpdir(), 'git-warp-v18-command-cli-drift-')); const restoreResult = await restoreV17GoldenGraphFixture({ manifestPath: FIXTURE_MANIFEST, @@ -146,7 +148,7 @@ describe('v18 graph-model migration command CLI', () => { }); it('blocks finalization when the archive ref already exists', async () => { - const scratchHead = await previewScratchHead(); + const scratchHead = SCRATCH_HEAD; const directory = await mkdtemp(join(tmpdir(), 'git-warp-v18-command-cli-archive-')); const restoreResult = await restoreV17GoldenGraphFixture({ manifestPath: FIXTURE_MANIFEST, @@ -182,7 +184,7 @@ describe('v18 graph-model migration command CLI', () => { }); it('blocks finalization when the reviewed runtime witness differs from observed replay', async () => { - const scratchHead = await previewScratchHead(); + const scratchHead = SCRATCH_HEAD; const directory = await mkdtemp(join(tmpdir(), 'git-warp-v18-command-cli-witness-')); const restoreResult = await restoreV17GoldenGraphFixture({ manifestPath: FIXTURE_MANIFEST, @@ -307,36 +309,6 @@ function finalizationRequestJson( }); } -async function previewScratchHead(): Promise { - const previewDirectory = await mkdtemp(join(tmpdir(), 'git-warp-v18-command-cli-preview-')); - const preview = await restoreV17GoldenGraphFixture({ - manifestPath: FIXTURE_MANIFEST, - targetDirectory: join(previewDirectory, 'repo'), - }); - const previewRequestPath = join(previewDirectory, 'request.json'); - await writeFile(previewRequestPath, canonicalRequestJson(), 'utf8'); - const previewResult = await runGraphModelMigrationCommandCli([ - '--repo', - preview.repositoryPath, - '--request', - previewRequestPath, - '--legacy-fixture-manifest', - FIXTURE_MANIFEST, - '--scratch-ref', - SCRATCH_REF, - ]); - expect(previewResult.exitCode).toBe(0); - return reportValue(previewResult.stdout, 'scratchHead'); -} - -function reportValue(report: string, label: string): string { - const line = report.split('\n').find((candidate) => candidate.startsWith(`${label}: `)); - if (line === undefined) { - throw new Error(`report line ${label} is missing`); - } - return line.slice(`${label}: `.length); -} - async function refExists(repositoryPath: string, refName: string): Promise { const result = await runMigrationGit( repositoryPath, From 3ecb818dba0060c61ff6bbd9fe24a3e0f7625644 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 16 Jul 2026 14:27:01 -0700 Subject: [PATCH 2/2] fix(materialization): preserve primary failures --- .../MaterializationWorkspaceCleanup.ts | 24 ++++++ .../controllers/MaterializeController.ts | 24 +++--- .../controllers/MaterializeSessionBridge.ts | 5 +- .../GitCasMaterializationWorkspace.ts | 1 - test/helpers/InMemoryMaterializationStore.ts | 15 +++- .../domain/orset/session/StateSession.test.ts | 15 +--- .../MaterializationWorkspaceCleanup.test.ts | 55 +++++++++++++ ...MaterializeController.stateSession.test.ts | 78 ++++++++++++++++--- .../GitCasMaterializationWorkspace.test.ts | 13 +++- 9 files changed, 186 insertions(+), 44 deletions(-) create mode 100644 src/domain/services/controllers/MaterializationWorkspaceCleanup.ts create mode 100644 test/unit/domain/services/controllers/MaterializationWorkspaceCleanup.test.ts diff --git a/src/domain/services/controllers/MaterializationWorkspaceCleanup.ts b/src/domain/services/controllers/MaterializationWorkspaceCleanup.ts new file mode 100644 index 000000000..7997a76a9 --- /dev/null +++ b/src/domain/services/controllers/MaterializationWorkspaceCleanup.ts @@ -0,0 +1,24 @@ +import type LoggerPort from '../../../ports/LoggerPort.ts'; +import type MaterializationWorkspacePort from '../../../ports/MaterializationWorkspacePort.ts'; + +/** Release a workspace without allowing cleanup to replace an existing failure. */ +export async function releaseWorkspaceAfterFailure( + workspace: MaterializationWorkspacePort | undefined, + logger?: LoggerPort +): Promise { + try { + await workspace?.release(); + } catch (cleanupFailure) { + reportCleanupFailure(logger, cleanupFailure); + } +} + +function reportCleanupFailure(logger: LoggerPort | undefined, failure: Failure): void { + try { + logger?.warn('[warp] materialization workspace release failed during error cleanup', { + error: failure instanceof Error ? failure.message : String(failure), + }); + } catch { + // Diagnostics are best-effort while preserving the active operation failure. + } +} diff --git a/src/domain/services/controllers/MaterializeController.ts b/src/domain/services/controllers/MaterializeController.ts index 7d9800325..ffccb581f 100644 --- a/src/domain/services/controllers/MaterializeController.ts +++ b/src/domain/services/controllers/MaterializeController.ts @@ -66,12 +66,10 @@ import type { MaterializeResultBuildInput, MaterializeStrategyRuntime, } from './MaterializeStrategyRuntime.ts'; +import { releaseWorkspaceAfterFailure } from './MaterializationWorkspaceCleanup.ts'; export type MaterializePersistence = { readRef(ref: string): Promise; }; - -// ── Deps ──────────────────────────────────────────────────────────── - /** Constructor dependencies for MaterializeController. */ export type MaterializeDeps = { logger: LoggerPort; @@ -87,8 +85,6 @@ export type MaterializeDeps = { graphName: string; }; -// ── Result types ──────────────────────────────────────────────────── - /** Full result of a materialization, returned to the caller. */ export type MaterializeResult = { state: WarpState; @@ -105,8 +101,6 @@ export type MaterializeResult = { materialization?: MaterializationHandle; }; -// ── Reduce helpers ────────────────────────────────────────────────── - type ReducerInput = Parameters[0]; function toReducerInput(patches: PatchWithSha[]): ReducerInput { return patches as ReducerInput; @@ -217,8 +211,6 @@ export default class MaterializeController { return await this._checkpointStrategy.materializeAt(checkpointSha); } - // ── Result building ─────────────────────────────────────────────── - private async _emptyResult( ceiling?: number | null, frontier?: Map | null, @@ -264,6 +256,7 @@ export default class MaterializeController { } private async _buildResult(params: MaterializeResultBuildInput): Promise { + let result: MaterializeResult; try { const stateHash = await computeHash(this._deps, params.reduced.state); const adjacency = params.reduced.adjacency ?? buildAdjacency(params.reduced.state); @@ -276,7 +269,7 @@ export default class MaterializeController { frontier: params.frontier, }); } - return { + result = { state: params.reduced.state, stateHash, adjacency: new AdjacencyMap({ outgoing: adjacency.outgoing, incoming: adjacency.incoming }), @@ -293,11 +286,13 @@ export default class MaterializeController { ceiling: params.ceiling, ...(materialization === undefined ? {} : { materialization }), }; - } finally { - await params.reduced.workspace?.release(); + } catch (raw) { + await releaseWorkspaceAfterFailure(params.reduced.workspace, this._deps.logger); + throw raw; } + await params.reduced.workspace?.release(); + return result; } - private async _resolveMaterialization( params: MaterializeResultBuildInput, stateHash: string, @@ -367,6 +362,7 @@ export default class MaterializeController { const reduced = await reduceSessionBackedState({ openStateSession, materializations: this._deps.materializations, + logger: this._deps.logger, coordinate, patches: [], baseState: snapshot.state, @@ -400,6 +396,7 @@ export default class MaterializeController { const sessionArgs = { openStateSession, materializations: this._deps.materializations, + logger: this._deps.logger, coordinate: new MaterializationCoordinate(coordinate), patches: patches.map((entry) => new PatchEntry(entry)), receipts: opts.receipts, @@ -434,6 +431,7 @@ export default class MaterializeController { const reduced = await reduceSessionBackedState({ openStateSession: this._deps.openStateSession, materializations: this._deps.materializations, + logger: this._deps.logger, coordinate: new MaterializationCoordinate(coordinate), patches: recordingStream(), receipts: opts.receipts, diff --git a/src/domain/services/controllers/MaterializeSessionBridge.ts b/src/domain/services/controllers/MaterializeSessionBridge.ts index ae8829d9f..edd2d1ba2 100644 --- a/src/domain/services/controllers/MaterializeSessionBridge.ts +++ b/src/domain/services/controllers/MaterializeSessionBridge.ts @@ -5,6 +5,7 @@ import type PatchEntry from "../../artifacts/PatchEntry.ts"; import type StateSession from "../../orset/session/StateSession.ts"; import type MaterializationStorePort from "../../../ports/MaterializationStorePort.ts"; import type MaterializationWorkspacePort from "../../../ports/MaterializationWorkspacePort.ts"; +import type LoggerPort from "../../../ports/LoggerPort.ts"; import type MaterializationCoordinate from "../../materialization/MaterializationCoordinate.ts"; import type StorageRetentionWitness from "../../storage/StorageRetentionWitness.ts"; import MaterializationRoot from "../../materialization/MaterializationRoot.ts"; @@ -21,6 +22,7 @@ import { buildAdjacencyFromSession, type MaterializeAdjacency, } from "./MaterializeHelpers.ts"; +import { releaseWorkspaceAfterFailure } from "./MaterializationWorkspaceCleanup.ts"; export type MaterializeSessionOpen = { readonly nodeAliveRootOid: string | null; @@ -39,6 +41,7 @@ type MaterializeSessionPatchSource = export async function reduceSessionBackedState(args: { readonly openStateSession: MaterializeSessionOpener; readonly materializations: MaterializationStorePort; + readonly logger?: LoggerPort; readonly coordinate: MaterializationCoordinate; readonly patches: MaterializeSessionPatchSource; readonly baseState?: WarpStateClass; @@ -106,7 +109,7 @@ export async function reduceSessionBackedState(args: { acceptMaterialization: close.accept, }; } catch (raw) { - await workspace.release(); + await releaseWorkspaceAfterFailure(workspace, args.logger); throw raw; } } diff --git a/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts b/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts index 3935f4cdf..008f529b9 100644 --- a/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts +++ b/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts @@ -179,7 +179,6 @@ export default class GitCasMaterializationWorkspace extends MaterializationWorks await this.#cache.remove(this.#key); } this.#released = true; - this.#assertLeaseHealthy(); } #expiresAt(): string { diff --git a/test/helpers/InMemoryMaterializationStore.ts b/test/helpers/InMemoryMaterializationStore.ts index 25602231c..b1f52c78f 100644 --- a/test/helpers/InMemoryMaterializationStore.ts +++ b/test/helpers/InMemoryMaterializationStore.ts @@ -112,16 +112,23 @@ function retentionWitness(handle: BundleHandle): StorageRetentionWitness { }); } -function workspaceRetentionWitness(handle: BundleHandle): StorageRetentionWitness { +export function workspaceRetentionWitness( + handle: BundleHandle, + options: { + readonly namespace?: string; + readonly generation?: string; + } = {}, +): StorageRetentionWitness { + const namespace = options.namespace ?? 'test/materialization-workspaces'; return new StorageRetentionWitness({ handle, policy: 'pinned', reachability: 'anchored', root: new StorageRetentionRoot({ kind: 'cache-set', - namespace: 'test/materialization-workspaces', - locator: 'test/materialization-workspaces', - generation: 'test-workspace-generation', + namespace, + locator: namespace, + generation: options.generation ?? 'test-workspace-generation', path: handle.toString(), }), observedAt: '1970-01-01T00:00:00.000Z', diff --git a/test/unit/domain/orset/session/StateSession.test.ts b/test/unit/domain/orset/session/StateSession.test.ts index ec9bc1c00..28a8313e5 100644 --- a/test/unit/domain/orset/session/StateSession.test.ts +++ b/test/unit/domain/orset/session/StateSession.test.ts @@ -16,6 +16,7 @@ import BundleHandle from "../../../../../src/domain/storage/BundleHandle.ts"; import StorageRetentionWitness, { StorageRetentionRoot, } from "../../../../../src/domain/storage/StorageRetentionWitness.ts"; +import { workspaceRetentionWitness } from "../../../../helpers/InMemoryMaterializationStore.ts"; const GEOMETRY = TrieGeometry.default16way(); @@ -457,18 +458,8 @@ class RecordingWorkspace extends MaterializationWorkspacePort { override checkpoint(roots: MaterializationWorkspaceRoots): Promise { this.checkpoints.push(roots); const handle = new BundleHandle(`test:workspace:${String(this.checkpoints.length)}`); - return Promise.resolve(new StorageRetentionWitness({ - handle, - policy: "pinned", - reachability: "anchored", - root: new StorageRetentionRoot({ - kind: "cache-set", - namespace: "test/materialization-workspaces", - locator: "test/materialization-workspaces", - generation: `generation-${String(this.checkpoints.length)}`, - path: handle.toString(), - }), - observedAt: "1970-01-01T00:00:00.000Z", + return Promise.resolve(workspaceRetentionWitness(handle, { + generation: `generation-${String(this.checkpoints.length)}`, })); } diff --git a/test/unit/domain/services/controllers/MaterializationWorkspaceCleanup.test.ts b/test/unit/domain/services/controllers/MaterializationWorkspaceCleanup.test.ts new file mode 100644 index 000000000..16b90235c --- /dev/null +++ b/test/unit/domain/services/controllers/MaterializationWorkspaceCleanup.test.ts @@ -0,0 +1,55 @@ +import { describe, expect, it } from 'vitest'; +import { releaseWorkspaceAfterFailure } from '../../../../../src/domain/services/controllers/MaterializationWorkspaceCleanup.ts'; +import MaterializationWorkspacePort from '../../../../../src/ports/MaterializationWorkspacePort.ts'; +import { createMockLogger } from '../../../../helpers/WarpGraphMockLogger.ts'; + +describe('releaseWorkspaceAfterFailure', () => { + it('contains logger failures while reporting cleanup errors', async () => { + const logger = createMockLogger(); + logger.warn.mockImplementation(() => { + throw new Error('logger unavailable'); + }); + + await expect(releaseWorkspaceAfterFailure( + new RejectingWorkspace(new Error('release unavailable')), + logger, + )).resolves.toBeUndefined(); + expect(logger.warn).toHaveBeenCalledOnce(); + }); + + it('contains cleanup failures that cannot be rendered', async () => { + const logger = createMockLogger(); + const cleanupFailure = { + toString(): never { + throw new Error('cleanup failure cannot be rendered'); + }, + }; + + await expect(releaseWorkspaceAfterFailure( + new RejectingWorkspace(cleanupFailure), + logger, + )).resolves.toBeUndefined(); + expect(logger.warn).not.toHaveBeenCalled(); + }); +}); + +class RejectingWorkspace extends MaterializationWorkspacePort { + readonly #failure: Error | { readonly toString: () => never }; + + constructor(failure: Error | { readonly toString: () => never }) { + super(); + this.#failure = failure; + } + + override checkpoint(): Promise { + return Promise.resolve(null); + } + + override promote(): Promise { + return Promise.reject(new Error('promotion is not used by this test')); + } + + override release(): Promise { + return Promise.reject(this.#failure); + } +} diff --git a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts index b2411e332..78cb4b380 100644 --- a/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts +++ b/test/unit/domain/services/controllers/MaterializeController.stateSession.test.ts @@ -17,7 +17,9 @@ import { createEmptyState } from "../../../../../src/domain/services/JoinReducer import Patch from "../../../../../src/domain/types/Patch.ts"; import type { CheckpointData, PatchWithSha } from "../../../../../src/domain/capabilities/PatchCollector.ts"; import InMemoryCheckpointStore from "../../../../helpers/InMemoryCheckpointStore.ts"; -import InMemoryMaterializationStore from "../../../../helpers/InMemoryMaterializationStore.ts"; +import InMemoryMaterializationStore, { + InMemoryMaterializationWorkspace, +} from "../../../../helpers/InMemoryMaterializationStore.ts"; import type MaterializationWorkspacePort from "../../../../../src/ports/MaterializationWorkspacePort.ts"; import MaterializationCoordinate from "../../../../../src/domain/materialization/MaterializationCoordinate.ts"; @@ -199,10 +201,24 @@ function createControllerFixtures() { } describe("MaterializeController — state session integration", () => { - it("releases checkpointed workspace roots when session reduction fails", async () => { - const { openStateSession, materializations } = createControllerFixtures(); + it("preserves a session reduction failure when workspace release also fails", async () => { + const { openStateSession, materializations, deps } = createControllerFixtures(); const close = vi.spyOn(StateSession.prototype, "close"); - const failure = new Error("patch source failed"); + const coercionFailure = new Error("primary coercion must not run"); + const failure = { + toString(): never { + throw coercionFailure; + }, + }; + const releaseFailure = new Error("workspace release failed"); + deps.logger.warn.mockImplementation(() => { + throw new Error("cleanup logging failed"); + }); + const release = vi.spyOn(InMemoryMaterializationWorkspace.prototype, "release") + .mockImplementation(function (this: InMemoryMaterializationWorkspace): Promise { + this.released = true; + return Promise.reject(releaseFailure); + }); const partial = new PatchEntry(nodeAddPatchRecord({ writer: "writer-1", lamport: 1, @@ -212,7 +228,7 @@ describe("MaterializeController — state session integration", () => { const patches = { async *[Symbol.asyncIterator]() { yield partial; - throw failure; + return await Promise.reject(failure); }, }; @@ -220,6 +236,7 @@ describe("MaterializeController — state session integration", () => { await expect(reduceSessionBackedState({ openStateSession, materializations, + logger: deps.logger, coordinate: new MaterializationCoordinate({ frontier: new Map([["writer-1", "deadbeef"]]), ceiling: null, @@ -231,8 +248,13 @@ describe("MaterializeController — state session integration", () => { expect(close).not.toHaveBeenCalled(); expect(materializations.workspaces[0]?.checkpoints).toHaveLength(1); expect(materializations.workspaces[0]?.released).toBe(true); + expect(deps.logger.warn).toHaveBeenCalledWith( + expect.stringContaining("workspace release failed"), + { error: releaseFailure.message }, + ); } finally { close.mockRestore(); + release.mockRestore(); } }); @@ -281,8 +303,8 @@ describe("MaterializeController — state session integration", () => { expect(materializations.workspaces[0]?.released).toBe(true); }); - it("releases checkpointed roots when final materialization retention fails", async () => { - const { controller, patches, materializations } = createControllerFixtures(); + it("preserves final retention failure when workspace release also fails", async () => { + const { controller, patches, materializations, deps } = createControllerFixtures(); patches.collectForFrontier.mockResolvedValue([ nodeAddPatchRecord({ writer: "writer-1", @@ -293,10 +315,46 @@ describe("MaterializeController — state session integration", () => { ]); const failure = new Error("final retention failed"); vi.spyOn(materializations, "retain").mockRejectedValue(failure); + const releaseFailure = new Error("workspace release failed"); + const release = vi.spyOn(InMemoryMaterializationWorkspace.prototype, "release") + .mockImplementation(function (this: InMemoryMaterializationWorkspace): Promise { + this.released = true; + return Promise.reject(releaseFailure); + }); - await expect(controller.materialize()).rejects.toBe(failure); - expect(materializations.workspaces[0]?.checkpoints).toHaveLength(1); - expect(materializations.workspaces[0]?.released).toBe(true); + try { + await expect(controller.materialize()).rejects.toBe(failure); + expect(materializations.workspaces[0]?.checkpoints).toHaveLength(1); + expect(materializations.workspaces[0]?.released).toBe(true); + expect(deps.logger.warn).toHaveBeenCalledWith( + expect.stringContaining("workspace release failed"), + { error: releaseFailure.message }, + ); + } finally { + release.mockRestore(); + } + }); + + it("surfaces workspace release failure after successful retention", async () => { + const { controller, patches, materializations } = createControllerFixtures(); + patches.collectForFrontier.mockResolvedValue([ + nodeAddPatchRecord({ + writer: "writer-1", + lamport: 1, + sha: "a1b2", + node: "node:session", + }), + ]); + const releaseFailure = new Error("workspace release failed"); + const release = vi.spyOn(InMemoryMaterializationWorkspace.prototype, "release") + .mockRejectedValue(releaseFailure); + + try { + await expect(controller.materialize()).rejects.toBe(releaseFailure); + expect(materializations.retainedRequests).toHaveLength(1); + } finally { + release.mockRestore(); + } }); it("releases an unused workspace before returning an empty result", async () => { diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationWorkspace.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationWorkspace.test.ts index a8102a3c7..0b94afdca 100644 --- a/test/unit/infrastructure/adapters/GitCasMaterializationWorkspace.test.ts +++ b/test/unit/infrastructure/adapters/GitCasMaterializationWorkspace.test.ts @@ -220,11 +220,12 @@ describe('GitCasMaterializationWorkspace', () => { expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toEqual([]); }); - it('fails closed after a lease renewal error and still removes the workspace', async () => { + it('fails closed after a lease renewal error and releases without masking cleanup', async () => { const harness = await createHarness(); const roots = await createRoots(harness.cas); const cache = await harness.cas.caches.open({ namespace: WORKSPACE_CACHE_NAMESPACE }); const scheduler = new ManualLeaseScheduler(); + const promote = vi.fn(rejectPromotion); let puts = 0; const workspace = new GitCasMaterializationWorkspace({ bundles: harness.cas.bundles, @@ -242,16 +243,22 @@ describe('GitCasMaterializationWorkspace', () => { leaseTtlMs: 100, leaseRenewalMs: 40, leaseScheduler: scheduler, - promote: rejectPromotion, + promote, }); await workspace.checkpoint(roots); await scheduler.runNext(); - await expect(workspace.release()).rejects.toMatchObject({ + await expect(workspace.checkpoint(roots)).rejects.toMatchObject({ + code: 'E_MATERIALIZATION_STORAGE', + message: expect.stringContaining('lease renewal failed'), + }); + await expect(workspace.promote({} as PromoteMaterializationRequest)).rejects.toMatchObject({ code: 'E_MATERIALIZATION_STORAGE', message: expect.stringContaining('lease renewal failed'), }); + expect(promote).not.toHaveBeenCalled(); + await expect(workspace.release()).resolves.toBeUndefined(); expect(harness.cas.readCacheKeys(WORKSPACE_CACHE_NAMESPACE)).toEqual([]); });