diff --git a/docs/topics/cas-first-memoized-materialization.md b/docs/topics/cas-first-memoized-materialization.md index aba2d170..bb810b5d 100644 --- a/docs/topics/cas-first-memoized-materialization.md +++ b/docs/topics/cas-first-memoized-materialization.md @@ -102,6 +102,18 @@ The bounded-memory read path is optic/worldline/query work over a sharded or streamed basis. The state cache is the replay-skipping compatibility bridge for legacy materialization and checkpoint flows. +`RuntimeHost.hasNode()` is the first compatibility read to consume the +handle-first path directly when the runtime uses the built-in trie-backed state +session and matching materialization reader. On an exact retained-coordinate +hit, it acquires the materialization, opens only the node-liveness trie through +a bounded page cache, answers one membership question, and releases the +acquisition. It does not hydrate `WarpState`, build adjacency, publish a state +snapshot, or populate `_cachedState`. A cold handle miss still performs the +current materialization path before retaining and reading the new root. A +custom state-session opener owns its root storage and encoding, so git-warp does +not pair it with the default reader and instead preserves the compatibility +fallback. + ## `git-cas` Encapsulation Materialization-root retention routes through the formal @@ -163,8 +175,10 @@ lost payload bytes, or run Git garbage collection. ## Current Limitations -- RuntimeHost and checkpoint creation do not yet consume the handle-first - result, so their compatibility path still owns process-resident whole state. +- RuntimeHost exact node-liveness reads consume the handle-first result when + the built-in trie session and reader pair is active. Custom session openers, + properties, neighborhoods, list reads, checkpoint creation, and other + compatibility operations still own process-resident whole state. - Exact state-cache hits bypass replay, but full materialization still hydrates a full `WarpState`, scans retained node/edge tries, and builds full adjacency. - Retained materialization descriptors currently carry node/edge trie roots; diff --git a/src/domain/RuntimeHost.ts b/src/domain/RuntimeHost.ts index 9e9f7e32..f3a0aa5a 100644 --- a/src/domain/RuntimeHost.ts +++ b/src/domain/RuntimeHost.ts @@ -402,6 +402,9 @@ export default class RuntimeHost { persistence: this._persistence, checkpointStore, materializations, + ...(options.materializationRead === undefined + ? {} + : { materializationRead: options.materializationRead }), getStateCache: () => this._stateCache ?? null, ...(openStateSession === undefined ? {} : { openStateSession }), patches: new RuntimePatchCollector(this), @@ -422,6 +425,10 @@ export default class RuntimeHost { this._auditService = auditService || null; } + _readLiveNodePresence(nodeId: string): Promise { + return this._materializeController.readLiveNodePresence(nodeId); + } + /** * Advanced substrate replay primitive over the live frontier. */ diff --git a/src/domain/materialization/TrieMaterializationReader.ts b/src/domain/materialization/TrieMaterializationReader.ts new file mode 100644 index 00000000..ef93d96d --- /dev/null +++ b/src/domain/materialization/TrieMaterializationReader.ts @@ -0,0 +1,96 @@ +import type CodecPort from '../../ports/CodecPort.ts'; +import MaterializationReadPort from '../../ports/MaterializationReadPort.ts'; +import BundleHandle from '../storage/BundleHandle.ts'; +import WarpError from '../errors/WarpError.ts'; +import PageCache from '../orset/trie/PageCache.ts'; +import TrieCursor from '../orset/trie/TrieCursor.ts'; +import TrieGeometry from '../orset/trie/TrieGeometry.ts'; +import type TrieStorePort from '../orset/trie/TrieStorePort.ts'; + +const MAX_RESIDENT_READ_PAGES = 256; + +/** Reads retained liveness roots without reconstructing a complete WarpState. */ +export default class TrieMaterializationReader extends MaterializationReadPort { + readonly #store: TrieStorePort; + readonly #codec: CodecPort; + readonly #geometry: TrieGeometry; + + constructor(options: { + readonly store: TrieStorePort; + readonly codec: CodecPort; + readonly geometry?: TrieGeometry; + }) { + super(); + requireOptions(options); + this.#store = requireStore(options.store); + this.#codec = requireCodec(options.codec); + this.#geometry = options.geometry === undefined + ? TrieGeometry.default16way() + : requireGeometry(options.geometry); + Object.freeze(this); + } + + override async hasNode(nodeAliveRoot: BundleHandle, nodeId: string): Promise { + if (!(nodeAliveRoot instanceof BundleHandle)) { + throw new WarpError( + 'Materialization node-liveness root must be a BundleHandle', + 'E_MATERIALIZATION_RESUME' + ); + } + const cursor = new TrieCursor({ + rootOid: nodeAliveRoot.toString(), + store: this.#store, + geometry: this.#geometry, + codec: this.#codec, + pageCache: new PageCache({ maxResident: MAX_RESIDENT_READ_PAGES }), + }); + return await cursor.contains(nodeId); + } +} + +function requireOptions(options: object): void { + if (options === null || typeof options !== 'object' || Array.isArray(options)) { + throw readerError('options must be an object'); + } +} + +function requireStore(store: TrieStorePort): TrieStorePort { + if ( + store === null + || typeof store !== 'object' + || !hasTrieOperations(store) + ) { + throw readerError('store must provide trie read/write operations'); + } + return store; +} + +function hasTrieOperations(store: TrieStorePort): boolean { + return typeof store.readLeaf === 'function' + && typeof store.readBranch === 'function' + && typeof store.writeLeaf === 'function' + && typeof store.writeBranch === 'function'; +} + +function requireCodec(codec: CodecPort): CodecPort { + if ( + codec === null + || typeof codec !== 'object' + || typeof codec.encode !== 'function' + || typeof codec.decode !== 'function' + ) { + throw readerError('codec must provide encode/decode operations'); + } + return codec; +} + +function requireGeometry(geometry: TrieGeometry): TrieGeometry { + if (!(geometry instanceof TrieGeometry)) { + throw readerError('geometry must be a TrieGeometry instance'); + } + return geometry; +} + +function readerError(message: string): WarpError { + return new WarpError(`Materialization reader ${message}`, 'E_MATERIALIZATION_RESUME'); +} diff --git a/src/domain/services/controllers/MaterializeController.ts b/src/domain/services/controllers/MaterializeController.ts index fef84bea..f75d1dbd 100644 --- a/src/domain/services/controllers/MaterializeController.ts +++ b/src/domain/services/controllers/MaterializeController.ts @@ -12,7 +12,6 @@ import { import { materializationSessionOpen, reduceSessionBackedState, - type MaterializeSessionOpener, } from './MaterializeSessionBridge.ts'; import MaterializeLiveStrategy from './MaterializeLiveStrategy.ts'; import MaterializeCoordinateStrategy from './MaterializeCoordinateStrategy.ts'; @@ -27,20 +26,12 @@ import { shouldPublishMaterializeSnapshot, type MaterializeSnapshotPublicationOptions, } from './MaterializeSnapshotPublication.ts'; -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 { - default as WarpStateCachePort, WarpStateCoordinate, WarpStateSnapshotProvenancePosture, } from '../../../ports/WarpStateCachePort.ts'; -import type MaterializationStorePort from '../../../ports/MaterializationStorePort.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'; import PatchEntry from '../../artifacts/PatchEntry.ts'; import type WarpState from '../state/WarpState.ts'; import type { TickReceipt } from '../../types/TickReceipt.ts'; @@ -61,23 +52,9 @@ import { releaseAcquisitionAfterFailure, releaseWorkspaceAfterFailure, } from './MaterializationWorkspaceCleanup.ts'; -export type MaterializePersistence = { - readRef(ref: string): Promise; -}; -/** Constructor dependencies for MaterializeController. */ -export type MaterializeDeps = { - logger: LoggerPort; - codec: CodecPort; - crypto: CryptoPort; - persistence: MaterializePersistence; - checkpointStore: CheckpointStorePort; - materializations: MaterializationStorePort; - getStateCache?: () => WarpStateCachePort | null; - openStateSession?: MaterializeSessionOpener; - patches: PatchCollector; - graphCloner: DetachedGraphFactory; - graphName: string; -}; +import type { MaterializeDeps } from './MaterializeDeps.ts'; + +export type { MaterializeDeps, MaterializePersistence } from './MaterializeDeps.ts'; /** Full result of a materialization, returned to the caller. */ export type MaterializeResult = { @@ -99,6 +76,7 @@ type ReducerInput = Parameters[0]; function toReducerInput(patches: PatchWithSha[]): ReducerInput { return patches as ReducerInput; } + export type MaterializeReduceOutput = { state: WarpState; adjacency?: MaterializeAdjacency; @@ -183,6 +161,9 @@ export default class MaterializeController { resolveLiveMaterialization(): Promise { return this._liveStrategy.resolveMaterialization(); } + readLiveNodePresence(nodeId: string): Promise { + return this._liveStrategy.readNodePresence(nodeId); + } /** Coordinate materialization — explicit frontier. */ async materializeCoordinate( diff --git a/src/domain/services/controllers/MaterializeDeps.ts b/src/domain/services/controllers/MaterializeDeps.ts new file mode 100644 index 00000000..527df7dd --- /dev/null +++ b/src/domain/services/controllers/MaterializeDeps.ts @@ -0,0 +1,30 @@ +import type CheckpointStorePort from '../../../ports/CheckpointStorePort.ts'; +import type CodecPort from '../../../ports/CodecPort.ts'; +import type CryptoPort from '../../../ports/CryptoPort.ts'; +import type LoggerPort from '../../../ports/LoggerPort.ts'; +import type MaterializationReadPort from '../../../ports/MaterializationReadPort.ts'; +import type MaterializationStorePort from '../../../ports/MaterializationStorePort.ts'; +import type WarpStateCachePort from '../../../ports/WarpStateCachePort.ts'; +import type DetachedGraphFactory from '../../capabilities/DetachedGraphFactory.ts'; +import type PatchCollector from '../../capabilities/PatchCollector.ts'; +import type { MaterializeSessionOpener } from './MaterializeSessionBridge.ts'; + +export type MaterializePersistence = { + readRef(ref: string): Promise; +}; + +/** Constructor dependencies for retained-handle materialization operations. */ +export type MaterializeDeps = { + logger: LoggerPort; + codec: CodecPort; + crypto: CryptoPort; + persistence: MaterializePersistence; + checkpointStore: CheckpointStorePort; + materializations: MaterializationStorePort; + materializationRead?: MaterializationReadPort; + getStateCache?: () => WarpStateCachePort | null; + openStateSession?: MaterializeSessionOpener; + patches: PatchCollector; + graphCloner: DetachedGraphFactory; + graphName: string; +}; diff --git a/src/domain/services/controllers/MaterializeLiveStrategy.ts b/src/domain/services/controllers/MaterializeLiveStrategy.ts index d754f80a..ca04083c 100644 --- a/src/domain/services/controllers/MaterializeLiveStrategy.ts +++ b/src/domain/services/controllers/MaterializeLiveStrategy.ts @@ -13,6 +13,8 @@ import type { PatchWithSha, } from '../../capabilities/PatchCollector.ts'; import type WarpStateCachePort from '../../../ports/WarpStateCachePort.ts'; +import type MaterializationReadPort from '../../../ports/MaterializationReadPort.ts'; +import type MaterializationHandle from '../../materialization/MaterializationHandle.ts'; import type { WarpStateCoordinate, } from '../../../ports/WarpStateCachePort.ts'; @@ -67,6 +69,23 @@ export default class MaterializeLiveStrategy { return await this.materializeAndAcquire(coordinate, materializationCoordinate); } + async readNodePresence(nodeId: string): Promise { + const reader = this.runtime.deps.materializationRead; + if (reader === undefined) { + return null; + } + const resolution = await this.resolveMaterialization(); + let presence: boolean | null; + try { + presence = await readNodePresence(reader, resolution.materialization, nodeId); + } catch (raw) { + await releaseAcquisitionAfterFailure(resolution, this.runtime.deps.logger); + throw raw; + } + await resolution.release(); + return presence; + } + private async resolveRetainedMaterialization( retained: MaterializationAcquisition, coordinate: MaterializationCoordinate, @@ -327,6 +346,27 @@ export default class MaterializeLiveStrategy { } } +async function readNodePresence( + reader: MaterializationReadPort, + materialization: MaterializationHandle | null, + nodeId: string, +): Promise { + if (materialization === null) { + return false; + } + const root = materialization.roots.nodeAlive; + if (root.status === 'unavailable') { + return null; + } + if (root.status === 'empty') { + return false; + } + if (root.handle === null) { + throw resolutionError('retained node-liveness root has no handle'); + } + return await reader.hasNode(root.handle, nodeId); +} + function emptyResolution(): LiveMaterializationResolution { return new LiveMaterializationResolution({ materialization: null, diff --git a/src/domain/services/controllers/QueryReads.ts b/src/domain/services/controllers/QueryReads.ts index a608a4c3..31a890e5 100644 --- a/src/domain/services/controllers/QueryReads.ts +++ b/src/domain/services/controllers/QueryReads.ts @@ -122,6 +122,12 @@ async function ensureAndGetState(host: QueryReadHost): Promise { // ── Read implementations ──────────────────────────────────────────── export async function hasNodeImpl(host: QueryReadHost, nodeId: string): Promise { + if (host._readLiveNodePresence !== undefined) { + const retainedPresence = await host._readLiveNodePresence(nodeId); + if (retainedPresence !== null) { + return retainedPresence; + } + } const state = await ensureAndGetState(host); return state.nodeAlive.contains(nodeId); } diff --git a/src/domain/services/controllers/ReadGraphHost.ts b/src/domain/services/controllers/ReadGraphHost.ts index 8d9a66f5..9e8b9029 100644 --- a/src/domain/services/controllers/ReadGraphHost.ts +++ b/src/domain/services/controllers/ReadGraphHost.ts @@ -28,6 +28,7 @@ export type FreshStateHost = { }; export type QueryReadHost = FreshStateHost & { + _readLiveNodePresence?(nodeId: string): Promise; _propertyReader: PropertyIndexReader | null; _logicalIndex: LogicalIndex | null; _materializedGraph: MaterializedReadGraph | null; diff --git a/src/domain/warp/RuntimeHostBoot.ts b/src/domain/warp/RuntimeHostBoot.ts index fc1f5601..af06ea2c 100644 --- a/src/domain/warp/RuntimeHostBoot.ts +++ b/src/domain/warp/RuntimeHostBoot.ts @@ -5,6 +5,7 @@ import StateHashService from '../services/state/StateHashService.ts'; import StateSession from '../orset/session/StateSession.ts'; import PageCache from '../orset/trie/PageCache.ts'; import TrieGeometry from '../orset/trie/TrieGeometry.ts'; +import TrieMaterializationReader from '../materialization/TrieMaterializationReader.ts'; import WarpError from '../errors/WarpError.ts'; import { buildEffectPipeline, @@ -34,6 +35,7 @@ import type CheckpointStorePort from '../../ports/CheckpointStorePort.ts'; import type IndexStorePort from '../../ports/IndexStorePort.ts'; import type IntentStorePort from '../../ports/IntentStorePort.ts'; import type MaterializationStorePort from '../../ports/MaterializationStorePort.ts'; +import type MaterializationReadPort from '../../ports/MaterializationReadPort.ts'; import type EffectSinkPort from '../../ports/EffectSinkPort.ts'; import type RuntimeStorageProviderPort from '../../ports/RuntimeStorageProviderPort.ts'; import type SchedulerPort from '../../ports/SchedulerPort.ts'; @@ -71,6 +73,7 @@ export type RuntimeHostConstructionOptions = { indexStore: IndexStorePort; intentStore: IntentStorePort; materializations: MaterializationStorePort; + materializationRead?: MaterializationReadPort; viewService: MaterializedViewService; stateHashService?: StateHashService; auditService?: AuditReceiptService; @@ -334,6 +337,7 @@ export async function resolveRuntimeHostConstructionOptions( } let resolvedOpenStateSession: MaterializeSessionOpener | undefined; + let resolvedMaterializationRead: MaterializationReadPort | undefined; if (openStateSession !== undefined) { resolvedOpenStateSession = openStateSession; } else if (storageServices.trie !== undefined) { @@ -349,6 +353,13 @@ export async function resolveRuntimeHostConstructionOptions( pageCache: new PageCache({ maxResident: 256 }), workspace: sessionOptions.workspace, }); + // A custom session opener owns its root encoding; pair this reader only + // with the built-in session that shares its store and geometry. + resolvedMaterializationRead = new TrieMaterializationReader({ + store, + codec: resolvedCodec, + geometry, + }); } return { @@ -378,6 +389,9 @@ export async function resolveRuntimeHostConstructionOptions( indexStore: resolvedIndexStore, intentStore: storageServices.intents, materializations: storageServices.materializations, + ...(resolvedMaterializationRead === undefined + ? {} + : { materializationRead: resolvedMaterializationRead }), viewService: resolvedViewService, stateHashService: resolvedStateHashService, ...(resolvedAuditService !== undefined ? { auditService: resolvedAuditService } : {}), diff --git a/src/ports/MaterializationReadPort.ts b/src/ports/MaterializationReadPort.ts new file mode 100644 index 00000000..e80bfc76 --- /dev/null +++ b/src/ports/MaterializationReadPort.ts @@ -0,0 +1,6 @@ +import type BundleHandle from '../domain/storage/BundleHandle.ts'; + +/** Bounded reads over independently retained materialization roots. */ +export default abstract class MaterializationReadPort { + abstract hasNode(_nodeAliveRoot: BundleHandle, _nodeId: string): Promise; +} diff --git a/test/integration/api/materialization.retainedResume.test.ts b/test/integration/api/materialization.retainedResume.test.ts index cf0bcf93..1f67f3d1 100644 --- a/test/integration/api/materialization.retainedResume.test.ts +++ b/test/integration/api/materialization.retainedResume.test.ts @@ -1,3 +1,5 @@ +import { execFile } from 'node:child_process'; +import { promisify } from 'node:util'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import RuntimeHost from '../../../src/domain/RuntimeHost.ts'; import type MaterializationCoordinate from '../../../src/domain/materialization/MaterializationCoordinate.ts'; @@ -18,6 +20,8 @@ import type { } from '../../../src/ports/RuntimeStorageProviderPort.ts'; import { createTestRepo } from './helpers/setup.ts'; +const execFileAsync = promisify(execFile); + describe('API: retained materialization resume', () => { let repo: Awaited> | null = null; @@ -68,6 +72,43 @@ describe('API: retained materialization resume', () => { expect(reopenedStore.exactHits[0]?.bundle.equals(coldHandle?.bundle)).toBe(true); expect(reopened.nodeAlive.contains('node:retained')).toBe(true); }); + + it('answers exact node presence from retained roots after aggressive pruning', async () => { + if (repo === null) { + throw new Error('Test repository is not initialized'); + } + const firstProvider = recordingProvider(repo); + const firstRuntime = await openRuntime(repo, firstProvider); + await firstRuntime.patch((patch) => { + patch.addNode('node:retained'); + }); + await firstRuntime.materialize(); + + await execFileAsync('git', [ + '-C', + repo.tempDir, + 'reflog', + 'expire', + '--expire=now', + '--all', + ]); + await execFileAsync('git', ['-C', repo.tempDir, 'prune', '--expire=now']); + + const reopenedProvider = recordingProvider(repo); + const reopenedRuntime = await openRuntime(repo, reopenedProvider); + const replay = vi.spyOn(reopenedRuntime, '_loadPatchChainFromSha'); + const publishWholeState = vi.spyOn(reopenedRuntime, '_onMaterialized'); + + await expect(reopenedRuntime.hasNode('node:retained')).resolves.toBe(true); + + const reopenedStore = requireMaterializations(reopenedProvider); + expect(replay).not.toHaveBeenCalled(); + expect(publishWholeState).not.toHaveBeenCalled(); + expect(reopenedRuntime._cachedState).toBeNull(); + expect(reopenedStore.exactLookups).toHaveLength(1); + expect(reopenedStore.exactHits).toHaveLength(1); + expect(reopenedStore.exactReleaseCount).toBe(1); + }); }); class RecordingMaterializationStore extends MaterializationStorePort { @@ -75,6 +116,7 @@ class RecordingMaterializationStore extends MaterializationStorePort { readonly retainedHandles: MaterializationHandle[] = []; readonly exactLookups: MaterializationCoordinate[] = []; readonly exactHits: MaterializationHandle[] = []; + exactReleaseCount = 0; readonly #delegate: MaterializationStorePort; constructor(delegate: MaterializationStorePort) { @@ -108,8 +150,16 @@ class RecordingMaterializationStore extends MaterializationStorePort { const acquisition = await this.#delegate.acquireExact(coordinate); if (acquisition !== null) { this.exactHits.push(acquisition.materialization); + return Object.freeze({ + materialization: acquisition.materialization, + acquiredAt: acquisition.acquiredAt, + release: async () => { + await acquisition.release(); + this.exactReleaseCount += 1; + }, + }); } - return acquisition; + return null; } } diff --git a/test/unit/domain/WarpCore.stateSessionAutoConstruct.test.ts b/test/unit/domain/WarpCore.stateSessionAutoConstruct.test.ts index 2c1aa6f6..2cc4adfb 100644 --- a/test/unit/domain/WarpCore.stateSessionAutoConstruct.test.ts +++ b/test/unit/domain/WarpCore.stateSessionAutoConstruct.test.ts @@ -91,5 +91,6 @@ describe("WarpCore state-session auto-construction", () => { expect(resolved.options.stateCache).toBe(stateCache); expect(resolved.options.openStateSession).toBe(openStateSession); + expect(resolved.options.materializationRead).toBeUndefined(); }); }); diff --git a/test/unit/domain/materialization/TrieMaterializationReader.test.ts b/test/unit/domain/materialization/TrieMaterializationReader.test.ts new file mode 100644 index 00000000..317847e3 --- /dev/null +++ b/test/unit/domain/materialization/TrieMaterializationReader.test.ts @@ -0,0 +1,53 @@ +import { describe, expect, it } from 'vitest'; + +import { Dot } from '../../../../src/domain/crdt/Dot.ts'; +import WarpError from '../../../../src/domain/errors/WarpError.ts'; +import TrieMaterializationReader from '../../../../src/domain/materialization/TrieMaterializationReader.ts'; +import StateSession from '../../../../src/domain/orset/session/StateSession.ts'; +import PageCache from '../../../../src/domain/orset/trie/PageCache.ts'; +import TrieGeometry from '../../../../src/domain/orset/trie/TrieGeometry.ts'; +import BundleHandle from '../../../../src/domain/storage/BundleHandle.ts'; +import cborCodec from '../../../../src/infrastructure/codecs/CborCodec.ts'; +import { InMemoryTrieStore } from '../../../helpers/trieHelpers.ts'; + +describe('TrieMaterializationReader', () => { + it('reads exact node presence without writing or scanning the full trie', async () => { + const store = new InMemoryTrieStore(); + const session = await StateSession.open({ + nodeAliveRootOid: null, + edgeAliveRootOid: null, + store, + codec: cborCodec, + geometry: TrieGeometry.default16way(), + pageCache: new PageCache({ maxResident: 32 }), + }); + await session.addNode('node:present', Dot.create('writer-1', 1)); + const roots = await session.close(); + if (roots.nodeAliveRootOid === null) { + throw new Error('Seed session did not write a node root'); + } + const writesBeforeRead = store.writeCounts(); + const reader = new TrieMaterializationReader({ store, codec: cborCodec }); + const root = new BundleHandle(roots.nodeAliveRootOid); + + expect(Object.isFrozen(reader)).toBe(true); + await expect(reader.hasNode(root, 'node:present')).resolves.toBe(true); + await expect(reader.hasNode(root, 'node:missing')).resolves.toBe(false); + + expect(store.writeCounts()).toEqual(writesBeforeRead); + const reads = store.readCounts(); + expect(reads.leaf + reads.branch).toBeGreaterThan(0); + expect(reads.leaf + reads.branch).toBeLessThanOrEqual(4); + }); + + it.each([ + ['options', null], + ['store', { store: {}, codec: cborCodec }], + ['codec', { store: new InMemoryTrieStore(), codec: {} }], + ['geometry', { store: new InMemoryTrieStore(), codec: cborCodec, geometry: {} }], + ])('rejects malformed %s with a domain error', (_field, options) => { + expect(() => Reflect.construct(TrieMaterializationReader, [options])).toThrowError( + expect.objectContaining>({ code: 'E_MATERIALIZATION_RESUME' }) + ); + }); +}); diff --git a/test/unit/domain/services/controllers/MaterializeController.liveNodeRead.test.ts b/test/unit/domain/services/controllers/MaterializeController.liveNodeRead.test.ts new file mode 100644 index 00000000..ed9f59c2 --- /dev/null +++ b/test/unit/domain/services/controllers/MaterializeController.liveNodeRead.test.ts @@ -0,0 +1,260 @@ +import { describe, expect, it, vi } from 'vitest'; + +import PatchCollector, { + type CheckpointData, + type PatchWithSha, +} from '../../../../../src/domain/capabilities/PatchCollector.ts'; +import MaterializationCoordinate from '../../../../../src/domain/materialization/MaterializationCoordinate.ts'; +import MaterializationRoot, { + type MaterializationRootStatus, +} from '../../../../../src/domain/materialization/MaterializationRoot.ts'; +import MaterializationRoots from '../../../../../src/domain/materialization/MaterializationRoots.ts'; +import MaterializeController, { + type MaterializeDeps, +} from '../../../../../src/domain/services/controllers/MaterializeController.ts'; +import BundleHandle from '../../../../../src/domain/storage/BundleHandle.ts'; +import cborCodec from '../../../../../src/infrastructure/codecs/CborCodec.ts'; +import InMemoryCheckpointStore from '../../../../helpers/InMemoryCheckpointStore.ts'; +import InMemoryMaterializationStore, { + InMemoryMaterializationAcquisition, +} from '../../../../helpers/InMemoryMaterializationStore.ts'; + +const FRONTIER = new Map([['writer-1', 'tip-1']]); + +class RetainedOnlyPatchCollector extends PatchCollector { + frontier = new Map(FRONTIER); + + override discoverWriters(): Promise { + return Promise.resolve([]); + } + + override loadWriterPatches(_writerId: string): Promise { + return Promise.resolve([]); + } + + override loadCheckpoint(): Promise { + return Promise.resolve(null); + } + + override loadPatchesSince(_checkpoint: CheckpointData): Promise { + return Promise.resolve([]); + } + + override loadPatchChain(_toSha: string, _fromSha?: string | null): Promise { + return Promise.resolve([]); + } + + override getFrontier(): Promise> { + return Promise.resolve(new Map(this.frontier)); + } +} + +describe('MaterializeController live node reads', () => { + it.each([true, false])( + 'reads retained node presence %s without projecting whole state', + async (presence) => { + const fixture = await createFixture({ presence }); + + await expect(fixture.controller.readLiveNodePresence('node:retained')).resolves.toBe( + presence + ); + + expect(fixture.materializationRead.hasNode).toHaveBeenCalledWith( + fixture.nodeRoot, + 'node:retained' + ); + expect(fixture.materializations.exactLookups).toHaveLength(1); + expect(fixture.materializations.acquisitions).toHaveLength(1); + expect(fixture.materializations.acquisitions[0]?.releaseCalls).toBe(1); + expect(fixture.materializations.acquisitions[0]?.released).toBe(true); + expect(fixture.deps.crypto.hash).not.toHaveBeenCalled(); + expect(fixture.deps.persistence.readRef).not.toHaveBeenCalled(); + } + ); + + it('returns false for an empty live frontier without opening retained storage', async () => { + const fixture = await createFixture({ frontier: new Map(), retain: false }); + + await expect(fixture.controller.readLiveNodePresence('node:missing')).resolves.toBe(false); + + expect(fixture.materializationRead.hasNode).not.toHaveBeenCalled(); + expect(fixture.materializations.exactLookups).toHaveLength(0); + }); + + it('returns false from an empty retained node root without invoking the trie reader', async () => { + const fixture = await createFixture({ rootStatus: 'empty' }); + + await expect(fixture.controller.readLiveNodePresence('node:missing')).resolves.toBe(false); + + expect(fixture.materializationRead.hasNode).not.toHaveBeenCalled(); + expect(fixture.materializations.acquisitions[0]?.releaseCalls).toBe(1); + }); + + it('releases retained roots when the bounded node read fails', async () => { + const readFailure = new Error('node read failed'); + const fixture = await createFixture({ readFailure }); + + await expect(fixture.controller.readLiveNodePresence('node:retained')).rejects.toBe( + readFailure + ); + + expect(fixture.materializations.acquisitions).toHaveLength(1); + expect(fixture.materializations.acquisitions[0]?.releaseCalls).toBe(1); + expect(fixture.materializations.acquisitions[0]?.released).toBe(true); + }); + + it('preserves a node read failure when acquisition cleanup also fails', async () => { + const readFailure = new Error('node read failed'); + const releaseFailure = new Error('acquisition release failed'); + const fixture = await createFixture({ readFailure }); + const retained = await requireRetained(fixture.materializations); + const acquisition = new InMemoryMaterializationAcquisition(retained); + const release = vi.spyOn(acquisition, 'release').mockRejectedValue(releaseFailure); + vi.spyOn(fixture.materializations, 'acquireExact').mockResolvedValue(acquisition); + + await expect(fixture.controller.readLiveNodePresence('node:retained')).rejects.toBe( + readFailure + ); + + expect(release).toHaveBeenCalledOnce(); + expect(fixture.deps.logger.warn).toHaveBeenCalledOnce(); + }); + + it('surfaces a successful read release failure without retrying it', async () => { + const fixture = await createFixture({ presence: true }); + const retained = await requireRetained(fixture.materializations); + const acquisition = new InMemoryMaterializationAcquisition(retained); + const releaseFailure = new Error('acquisition release failed'); + const release = vi.spyOn(acquisition, 'release').mockRejectedValue(releaseFailure); + vi.spyOn(fixture.materializations, 'acquireExact').mockResolvedValue(acquisition); + + await expect(fixture.controller.readLiveNodePresence('node:retained')).rejects.toBe( + releaseFailure + ); + + expect(release).toHaveBeenCalledOnce(); + }); + + it('falls back when bounded materialization reads are not configured', async () => { + const fixture = await createFixture({ materializationRead: false }); + + await expect(fixture.controller.readLiveNodePresence('node:retained')).resolves.toBeNull(); + + expect(fixture.materializations.exactLookups).toHaveLength(0); + }); + + it('falls back and releases when the retained node root is unavailable', async () => { + const fixture = await createFixture({ rootStatus: 'unavailable' }); + + await expect(fixture.controller.readLiveNodePresence('node:retained')).resolves.toBeNull(); + + expect(fixture.materializationRead.hasNode).not.toHaveBeenCalled(); + expect(fixture.materializations.acquisitions[0]?.releaseCalls).toBe(1); + expect(fixture.materializations.acquisitions[0]?.released).toBe(true); + }); +}); + +async function createFixture( + options: { + readonly frontier?: Map; + readonly materializationRead?: boolean; + readonly presence?: boolean; + readonly readFailure?: Error; + readonly retain?: boolean; + readonly rootStatus?: MaterializationRootStatus; + } = {} +) { + const patches = new RetainedOnlyPatchCollector(); + patches.frontier = new Map(options.frontier ?? FRONTIER); + const materializations = new InMemoryMaterializationStore(); + const nodeRoot = new BundleHandle('test:node-root'); + if (options.retain !== false && patches.frontier.size > 0) { + await materializations.retain({ + coordinate: new MaterializationCoordinate({ frontier: patches.frontier, ceiling: null }), + roots: rootsWithNodeStatus(options.rootStatus ?? 'retained', nodeRoot), + stateHash: 'state-hash', + }); + } + const hasNode = vi.fn<(nodeAliveRoot: BundleHandle, nodeId: string) => Promise>(); + if (options.readFailure === undefined) { + hasNode.mockResolvedValue(options.presence ?? true); + } else { + hasNode.mockRejectedValue(options.readFailure); + } + const materializationRead = { hasNode }; + const deps = createDeps({ materializations, patches, materializationRead }); + const controller = + options.materializationRead === false + ? new MaterializeController(withoutMaterializationRead(deps)) + : new MaterializeController(deps); + return { controller, deps, materializationRead, materializations, nodeRoot }; +} + +function createDeps(options: { + readonly materializations: InMemoryMaterializationStore; + readonly patches: PatchCollector; + readonly materializationRead: { + hasNode(nodeAliveRoot: BundleHandle, nodeId: string): Promise; + }; +}): MaterializeDeps { + return { + logger: { + warn: vi.fn(), + info: vi.fn(), + debug: vi.fn(), + error: vi.fn(), + child: vi.fn(), + }, + codec: cborCodec, + crypto: { + hash: vi.fn().mockResolvedValue('unused-state-hash'), + hmac: vi.fn().mockResolvedValue(new Uint8Array([1])), + timingSafeEqual: vi.fn().mockReturnValue(false), + }, + persistence: { readRef: vi.fn().mockResolvedValue(null) }, + checkpointStore: new InMemoryCheckpointStore(), + materializations: options.materializations, + materializationRead: options.materializationRead, + patches: options.patches, + graphCloner: { openReadOnly: vi.fn() }, + graphName: 'test-graph', + }; +} + +function withoutMaterializationRead(deps: MaterializeDeps): MaterializeDeps { + const { materializationRead: _materializationRead, ...withoutReader } = deps; + return withoutReader; +} + +function rootsWithNodeStatus( + status: MaterializationRootStatus, + nodeRoot: BundleHandle +): MaterializationRoots { + const unavailable = MaterializationRoot.unavailable(); + return new MaterializationRoots({ + adjacency: unavailable, + edgeAlive: MaterializationRoot.empty(), + edgeBirths: unavailable, + frontier: unavailable, + nodeAlive: + status === 'retained' + ? MaterializationRoot.retained(nodeRoot) + : status === 'empty' + ? MaterializationRoot.empty() + : unavailable, + properties: unavailable, + provenanceSupport: unavailable, + roaringIndexes: unavailable, + }); +} + +async function requireRetained(materializations: InMemoryMaterializationStore) { + const coordinate = new MaterializationCoordinate({ frontier: FRONTIER, ceiling: null }); + const acquisition = await materializations.acquireExact(coordinate); + if (acquisition === null) { + throw new Error('Test materialization was not retained'); + } + await acquisition.release(); + materializations.acquisitions.splice(0); + return acquisition.materialization; +} diff --git a/test/unit/domain/services/controllers/QueryController.test.ts b/test/unit/domain/services/controllers/QueryController.test.ts index 3b1f38ce..ae86154b 100644 --- a/test/unit/domain/services/controllers/QueryController.test.ts +++ b/test/unit/domain/services/controllers/QueryController.test.ts @@ -118,6 +118,7 @@ function createHost(state, overrides = {}) { _stateDirty: false, getFrontier: vi.fn().mockResolvedValue(new Map(frontier)), _ensureFreshState: vi.fn().mockResolvedValue(undefined), + _readLiveNodePresence: vi.fn().mockResolvedValue(null), _assetStorage: { open: vi.fn(() => chunks(new Uint8Array([1, 2, 3]))), }, @@ -180,6 +181,19 @@ describe('QueryController', () => { await ctrl.hasNode('alice'); expect(host._ensureFreshState).toHaveBeenCalled(); }); + + it.each([true, false])( + 'returns retained node presence %s without requiring whole state', + async (presence) => { + host._cachedState = null; + host._readLiveNodePresence.mockResolvedValue(presence); + + await expect(ctrl.hasNode('alice')).resolves.toBe(presence); + + expect(host._readLiveNodePresence).toHaveBeenCalledWith('alice'); + expect(host._ensureFreshState).not.toHaveBeenCalled(); + }, + ); }); // ── getNodes ─────────────────────────────────────────────────────────────