diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index e7fd44f8d..500371dc4 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -120,6 +120,9 @@ import { GitStorage } from '@git-stunts/git-warp/storage'; const storage = await GitStorage.open({ cwd: '.' }); const warp = await openWarp({ storage, writer: 'agent-1' }); const team = await warp.timeline('team'); + +// After the final lane operation: +await storage.close(); ``` Application code writes with `timeline.write(intent)` and reads with diff --git a/CHANGELOG.md b/CHANGELOG.md index cc11ec9f0..db522520b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `Timeline.draft(name)`, draft writes, `Timeline.previewJoin(draft)`, and `Timeline.join(draft)` with join receipts for first-use speculative workflows. +- Added idempotent `GitStorage.close()` and async-disposal support to release + local Git and git-cas processes without changing history or retention. - Added the `lint:test-law` gate to reject conditional bare `return;` statements in test bodies so skipped assertions cannot masquerade as passing tests. @@ -50,8 +52,9 @@ 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.5.0` so Git-backed materializations can - use managed `CacheSet` retention and opaque page and bundle handles. +- Upgraded `@git-stunts/git-cas` to `^6.5.3` so Git-backed materializations can + use managed `CacheSet` retention, opaque page and bundle handles, and + persistent Git object sessions. - 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 @@ -72,6 +75,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Reused repository-scoped Git object sessions across immutable history and + CAS operations, then closed history, materialization, and CAS owners at + application, CLI, migration, and test boundaries. This removes repeated + `cat-file` and `mktree` process startup without leaking subprocesses. - Live exact node-property reads now retain collision-safe property shards as an independently addressed materialization root. Each node uses its full BLAKE3 route key and a deterministic schema-v2 entry-bag envelope; writes and diff --git a/README.md b/README.md index 3547772c3..72dff5a90 100644 --- a/README.md +++ b/README.md @@ -104,8 +104,13 @@ if (role.receipt.outcome === 'accepted') { } else { console.error(role.receipt.reason, role.receipt.repairHints); } + +await storage.close(); ``` +`GitStorage.close()` releases local Git and git-cas processes only. It does not +delete timelines, rewrite history, or change retention anchors. + The v18 graph-first API is not exported in v19. Migrate uses of `openWarpWorldline()`, `openWarpGraph()`, `WarpApp`, `WarpCore`, `GitGraphAdapter`, patch builders, and graph operation creators before diff --git a/bin/cli/commands/mcp.ts b/bin/cli/commands/mcp.ts index 560bbded7..70a3e22da 100644 --- a/bin/cli/commands/mcp.ts +++ b/bin/cli/commands/mcp.ts @@ -1,9 +1,8 @@ -import fs from 'node:fs'; import process from 'node:process'; import readline from 'node:readline'; import { usageError } from '../infrastructure.ts'; -import { openGraph } from '../shared.ts'; +import { openGraph, readCliPackageVersion } from '../shared.ts'; import { handleMcpMessage, mcpParseError, @@ -14,17 +13,9 @@ import type { CliOptions } from '../types.ts'; type McpCommandResult = { readonly payload: undefined; readonly close: () => Promise; + readonly completion: Promise; }; -function readPackageVersion(): string { - const packageUrl = new URL('../../../package.json', import.meta.url); - const packageText = fs.readFileSync(packageUrl, 'utf8'); - const packageJson = JSON.parse(packageText) as { readonly version?: string }; - return typeof packageJson.version === 'string' && packageJson.version.length > 0 - ? packageJson.version - : '0.0.0'; -} - function writeResponse(response: McpResponse): void { process.stdout.write(`${JSON.stringify(response)}\n`); } @@ -42,25 +33,74 @@ export default async function handleMcp({ } const { graph } = await openGraph(options); - const serverVersion = readPackageVersion(); + return createMcpCommandResult(graph, readCliPackageVersion()); +} + +function createMcpCommandResult( + graph: Parameters[0], + serverVersion: string, +): McpCommandResult { const lines = readline.createInterface({ input: process.stdin, terminal: false, }); - - lines.on('line', (line) => { - void dispatchLine(graph, serverVersion, line); - }); + const completion = trackMcpLines(lines, graph, serverVersion); return { payload: undefined, - close: () => { + completion, + close: async () => { lines.close(); - return Promise.resolve(); + await completion; }, }; } +function trackMcpLines( + lines: readline.Interface, + graph: Parameters[0], + serverVersion: string, +): Promise { + const pending = new Set>(); + const completedFailures: unknown[] = []; + const completion = Promise.withResolvers(); + + lines.on('line', (line) => { + const operation = dispatchLine(graph, serverVersion, line); + pending.add(operation); + void operation.then( + () => pending.delete(operation), + (error: unknown) => { + pending.delete(operation); + completedFailures.push(error); + }, + ); + }); + lines.on('error', (error: unknown) => { + completedFailures.push(error); + lines.close(); + }); + lines.once('close', () => { + void settlePendingDispatches(pending, completedFailures) + .then(completion.resolve, completion.reject); + }); + return completion.promise; +} + +async function settlePendingDispatches( + pending: ReadonlySet>, + completedFailures: readonly unknown[], +): Promise { + const failures = [...completedFailures]; + const results = await Promise.allSettled([...pending]); + failures.push(...results + .filter((result): result is PromiseRejectedResult => result.status === 'rejected') + .map((result) => result.reason as unknown)); + if (failures.length > 0) { + throw new AggregateError(failures, 'MCP requests failed while stdin was closing'); + } +} + async function dispatchLine( graph: Parameters[0], serverVersion: string, diff --git a/bin/cli/lifecycle.ts b/bin/cli/lifecycle.ts new file mode 100644 index 000000000..e8ceaf8bc --- /dev/null +++ b/bin/cli/lifecycle.ts @@ -0,0 +1,22 @@ +type CloseResource = () => Promise; + +/** Drains a long-running command before releasing the storage it may still use. */ +export async function closeCommandResources( + closeCommand: CloseResource, + closeStorage: CloseResource, +): Promise { + const failures: unknown[] = []; + for (const close of [closeCommand, closeStorage]) { + try { + await close(); + } catch (error) { + failures.push(error); + } + } + if (failures.length === 1) { + throw failures[0]; + } + if (failures.length > 1) { + throw new AggregateError(failures, 'CLI command and storage failed to close cleanly'); + } +} diff --git a/bin/cli/shared.ts b/bin/cli/shared.ts index f4e230848..fbb3ea77a 100644 --- a/bin/cli/shared.ts +++ b/bin/cli/shared.ts @@ -30,23 +30,51 @@ export type CliStorageBinding = { readonly hookPaths: HookPathPort; }; +const activeCliStorages = new Set(); + +/** Releases every storage composition opened by the current CLI invocation. */ +export async function closeCliStorages(): Promise { + const storages = [...activeCliStorages]; + activeCliStorages.clear(); + const results = await Promise.allSettled(storages.map(async (storage) => await storage.close())); + const failures = results + .filter((result): result is PromiseRejectedResult => result.status === 'rejected') + .map((result) => result.reason as unknown); + if (failures.length > 0) { + throw new AggregateError(failures, 'CLI storage failed to close cleanly'); + } +} + /** * Creates a persistence adapter for the given repository path. */ export async function createPersistence(repoPath: string): Promise { const storage = await GitStorage.open({ cwd: repoPath }); - const binding = resolveWarpStorage(storage); - if (!(binding.history instanceof GitTimelineHistoryAdapter) - || binding.createTrustChain === undefined - || binding.hookPaths === undefined) { - throw usageError('GitStorage returned an incomplete CLI storage binding'); + try { + const binding = resolveWarpStorage(storage); + if (!(binding.history instanceof GitTimelineHistoryAdapter) + || binding.createTrustChain === undefined + || binding.hookPaths === undefined) { + throw usageError('GitStorage returned an incomplete CLI storage binding'); + } + activeCliStorages.add(storage); + return { + persistence: binding.history, + runtimeStorage: binding.runtimeStorage, + createTrustChain: binding.createTrustChain, + hookPaths: binding.hookPaths, + }; + } catch (error) { + try { + await storage.close(); + } catch (closeError) { + throw new AggregateError( + [error, closeError], + 'CLI storage binding failed and local resources did not close cleanly', + ); + } + throw error; } - return { - persistence: binding.history, - runtimeStorage: binding.runtimeStorage, - createTrustChain: binding.createTrustChain, - hookPaths: binding.hookPaths, - }; } /** @@ -188,6 +216,13 @@ export function createHookInstaller(hookPathPort: HookPathPort): HookInstaller { }); } +/** Reads the package version from either source or built CLI layouts. */ +export function readCliPackageVersion(): string { + const packageRoot = findPackageRoot(fileURLToPath(new URL('.', import.meta.url))); + const rawJson = fs.readFileSync(path.join(packageRoot, 'package.json'), 'utf8'); + return readPackageVersion(rawJson); +} + /** * Finds the repository/package root from either source or built CLI paths. */ @@ -199,7 +234,7 @@ function findPackageRoot(startDir: string): string { } const parent = path.dirname(current); if (parent === current) { - throw usageError('Unable to locate package.json for hook installation'); + throw usageError('Unable to locate the git-warp package root'); } current = parent; } diff --git a/bin/warp-graph.ts b/bin/warp-graph.ts index bede0386c..2b9c184aa 100755 --- a/bin/warp-graph.ts +++ b/bin/warp-graph.ts @@ -5,6 +5,8 @@ import { installDefaultRuntimeHostNodePorts } from '../src/application/RuntimeHo import { EXIT_CODES, HELP_TEXT, CliError, parseArgs, usageError } from './cli/infrastructure.ts'; import { stableStringify, compactStringify } from './presenters/json.ts'; import { COMMANDS } from './cli/commands/registry.ts'; +import { closeCliStorages } from './cli/shared.ts'; +import { closeCommandResources } from './cli/lifecycle.ts'; installDefaultRuntimeHostNodePorts(); @@ -26,12 +28,22 @@ function hasPayload(value: unknown): value is { payload: unknown; exitCode?: num } /** Runtime guard: does this value carry an async `close` function? */ -function hasCloseFn(value: unknown): value is { close: () => Promise } { +function hasCloseFn(value: unknown): value is { + close: () => Promise; + completion?: Promise; +} { if (typeof value !== 'object' || value === null) { return false; } const rec = value as Record; return typeof rec['close'] === 'function'; } +function hasCompletion(value: { readonly completion?: Promise }): value is { + readonly completion: Promise; +} { + return value.completion !== undefined + && typeof value.completion.then === 'function'; +} + /** Normalizes any handler return shape into { payload, exitCode }. */ function normalizeResult(result: unknown): NormalizedCommandResult { if (hasPayload(result)) { @@ -71,16 +83,41 @@ function handleEarlyExits(parsed: ParsedInvocation): boolean { /** Registers SIGINT/SIGTERM handlers that shut down a long-running * command gracefully. */ -function installShutdownHandlers(close: () => Promise): void { - let closing = false; - const shutdown = async (): Promise => { - if (closing) { return; } - closing = true; - await close(); - process.exit(EXIT_CODES.OK); +function installShutdownHandlers(close: () => Promise): () => Promise { + let shutdownPromise: Promise | null = null; + const shutdown = (): Promise => { + shutdownPromise ??= closeCommandResources(close, closeCliStorages); + return shutdownPromise; + }; + const exitAfterShutdown = (): void => { + void shutdown().then( + () => process.exit(EXIT_CODES.OK), + () => process.exit(EXIT_CODES.INTERNAL), + ); }; - process.on('SIGINT', () => { shutdown().catch(() => process.exit(1)); }); - process.on('SIGTERM', () => { shutdown().catch(() => process.exit(1)); }); + process.once('SIGINT', exitAfterShutdown); + process.once('SIGTERM', exitAfterShutdown); + return shutdown; +} + +async function finishCompletedCommand( + completion: Promise, + shutdown: () => Promise, +): Promise { + const [operation] = await Promise.allSettled([completion]); + const [cleanup] = await Promise.allSettled([shutdown()]); + if (operation.status === 'rejected' && cleanup.status === 'rejected') { + throw new AggregateError( + [operation.reason, cleanup.reason], + 'CLI command completion and shutdown both failed', + ); + } + if (operation.status === 'rejected') { + throw operation.reason; + } + if (cleanup.status === 'rejected') { + throw cleanup.reason; + } } /** Writes the payload (when present) in the requested stringify @@ -115,16 +152,27 @@ async function main(): Promise { emitPayload(normalized.payload, options.ndjson); // Long-running commands may return a `close` function. - // Wait for SIGINT/SIGTERM instead of exiting immediately. + // Wait for normal completion or SIGINT/SIGTERM instead of exiting immediately. if (hasCloseFn(result)) { - installShutdownHandlers(result.close); - return; // Keep the process alive + const shutdown = installShutdownHandlers(result.close); + if (hasCompletion(result)) { + await finishCompletedCommand(result.completion, shutdown); + process.exit(normalized.exitCode); + } + return; } + await closeCliStorages(); process.exit(normalized.exitCode); } -main().catch((error: unknown) => { +main().catch(async (caught: unknown) => { + let error = caught; + try { + await closeCliStorages(); + } catch (closeError) { + error = new AggregateError([caught, closeError], 'CLI command and storage cleanup failed'); + } const exitCode = error instanceof CliError ? error.exitCode : EXIT_CODES.INTERNAL; const code = error instanceof CliError ? error.code : 'E_INTERNAL'; const message = error instanceof Error ? error.message : 'Unknown error'; diff --git a/docs/TECHNICAL_TEARDOWN.md b/docs/TECHNICAL_TEARDOWN.md index 1afd430a5..c2e8f5777 100644 --- a/docs/TECHNICAL_TEARDOWN.md +++ b/docs/TECHNICAL_TEARDOWN.md @@ -185,7 +185,9 @@ flowchart TB ``` The root import has no storage bootstrap side effect. `GitStorage.open()` owns -the Node Git and git-cas composition used by first-use applications. +the Node Git and git-cas composition used by first-use applications; +`GitStorage.close()` releases that composition's local processes without +changing history or retention. ## System Architecture diff --git a/docs/migrations/v19/README.md b/docs/migrations/v19/README.md index 1e5b6803a..0056e09c6 100644 --- a/docs/migrations/v19/README.md +++ b/docs/migrations/v19/README.md @@ -85,6 +85,9 @@ const warp = await openWarp({ }); const events = await warp.timeline('events'); + +// After the final lane operation: +await storage.close(); ``` The old names are accurate implementation names, but they leak substrate diff --git a/docs/topics/getting-started.md b/docs/topics/getting-started.md index 5ecc9ddfe..bf7a1f861 100644 --- a/docs/topics/getting-started.md +++ b/docs/topics/getting-started.md @@ -155,7 +155,13 @@ methods so there is no boolean dry-run mode. `GitStorage` is one opaque repository-scoped handle. It composes timeline history and git-cas services internally; application code does not construct -plumbing, CAS, cache, or retention adapters. +plumbing, CAS, cache, or retention adapters. Close it when the application is +finished to release local Git and git-cas processes. Closing storage does not +delete timelines, rewrite history, or change retention anchors. + +```typescript +await storage.close(); +``` WARP history lives under `refs/warp/**`, separate from source branches such as `refs/heads/main`. Writing a timeline does not create a source-tree commit on diff --git a/docs/topics/git-perf.md b/docs/topics/git-perf.md index c100e6883..eff47795e 100644 --- a/docs/topics/git-perf.md +++ b/docs/topics/git-perf.md @@ -35,8 +35,8 @@ only as a process-startup control; the retained harness invokes it once after timed scenarios to record the exact Git build in result metadata. No production recommendation invokes it. -The current git-cas adapter starts a fresh process for each blob write, tree -write, blob stream, tree lookup, and uncached object-info lookup: +Before git-cas v6.5.2, the adapter started a fresh process for each blob write, +tree write, blob stream, tree lookup, and uncached object-info lookup: - `hash-object -w --stdin` is invoked per blob. [cite: `git-cas/src/infrastructure/adapters/GitPersistenceAdapter.js#57-65@49b7d5cb9d589d73fa17d393e48d40bd6f139e57`] - `mktree` is invoked per tree. [cite: `git-cas/src/infrastructure/adapters/GitPersistenceAdapter.js#68-79@49b7d5cb9d589d73fa17d393e48d40bd6f139e57`] @@ -49,6 +49,13 @@ On the measured Apple M1 Pro, one Git process had an approximately 18-20 ms startup floor. The problem is therefore both process overhead and our use of the process boundary: paying that floor once per object is self-inflicted. +git-cas v6.5.2 shipped bounded persistent object sessions, and v6.5.3 corrected +their visibility/retirement policy: `cat-file` survives successful immutable +writes, `mktree` survives loose writes, and `mktree` is retired after a bounded +`fast-import` pack write. The original process-per-object measurements below +remain the diagnosis and baseline, not a description of the current adapter. +[cite: `git-cas/src/infrastructure/adapters/GitPersistenceAdapter.js#93-175@7bdcbf1f9eccd16acd324c94d576e1ecd2e11d98`] + The 128-operation profile makes the distinction concrete: | Packed workload | One process per operation | Persistent Git | NodeGit | @@ -64,6 +71,41 @@ Persistent Git matches or beats libgit2 for object metadata and packed blob reads. Parsing one bounded tree page and caching its immutable entry index beats re-entering either Git implementation for every path lookup. +### Published v6.5.3 consumer checkpoint + +git-warp consumed the published npm artifact `@git-stunts/git-cas@6.5.3` with +`@git-stunts/plumbing@3.2.0` and repeated the temporary 128-node cold retained +materialization diagnostic three times. The injected plumbing counter began +after repository initialization, so it observed 398 workload Git children per +run; adding the excluded `git init` and two `git config` setup commands yields +the same 401 total recorded for the release candidate. + +Every run produced the same major command counts: + +| Command | Processes | +| -------------- | --------: | +| `hash-object` | 140 | +| `rev-parse` | 87 | +| `symbolic-ref` | 43 | +| `update-ref` | 43 | +| `commit-tree` | 39 | +| `rev-list` | 37 | +| `cat-file` | 4 | +| `mktree` | 1 | + +The three local cold-materialization timings were 7.02 s, 7.18 s, and 6.67 s. +They are diagnostics only: this rerun used Node 26, did not capture process-tree +CPU/RSS, and intentionally did not reproduce the old timing harness. The +structural process counts are the comparable result. Focused real-Git retained +materialization, materialization-store, and trie integration tests passed 11/11 +against the registry artifact. + +This checkpoint proves that npm v6.5.3 matches the audited session topology. It +does not satisfy the versioned cold/warm/incremental benchmark contract in +[#759](https://github.com/git-stunts/git-warp/issues/759): that issue still +requires a committed corpus, process-tree CPU/RSS, repeated base/head CI runs, +and blocking regression thresholds. + ## Bounded page reads The resource workload used random 4 KiB objects so compression could not hide diff --git a/docs/topics/querying.md b/docs/topics/querying.md index 7f26b22cb..411ef2792 100644 --- a/docs/topics/querying.md +++ b/docs/topics/querying.md @@ -114,6 +114,16 @@ executable `Optic`, and the type-only `Witness`. Capture a coordinate with `captureCoordinate(timeline)`; ordinary application reads should continue to use `reading.*`. +## Close Storage + +After the final reading or diagnostic inspection, close the storage handle to +release its local Git and git-cas processes. This does not change admitted +history or retention anchors. + +```typescript +await storage.close(); +``` + ## Removed Compatibility Surface The graph-first `openWarpWorldline()` and `openWarpGraph()` package exports are diff --git a/docs/topics/reference.md b/docs/topics/reference.md index 64913fdea..f188d5053 100644 --- a/docs/topics/reference.md +++ b/docs/topics/reference.md @@ -94,7 +94,7 @@ Git-backed storage for first-use applications. Source: `storage.ts`. Count: 1. ```text -GitStorage @ storage.ts#L16 +GitStorage @ storage.ts#L38 ``` ### Type exports @@ -192,7 +192,7 @@ ReceiptSubstrateInspection @ diagnostics.ts#L15 Structured CLI errors for `--json` and `--ndjson` use the payload shape `{ error: { code, message, cause? } }` from the CLI entry point. -Source: `bin/warp-graph.ts#L131`. +Source: `bin/warp-graph.ts#L179`. ## Public error classes diff --git a/package-lock.json b/package-lock.json index 194c1d89d..c7e61e26e 100644 --- a/package-lock.json +++ b/package-lock.json @@ -13,8 +13,8 @@ ], "dependencies": { "@git-stunts/alfred": "^0.10.4", - "@git-stunts/git-cas": "^6.5.0", - "@git-stunts/plumbing": "^3.1.0", + "@git-stunts/git-cas": "^6.5.3", + "@git-stunts/plumbing": "^3.2.0", "@git-stunts/trailer-codec": "^2.1.1", "@noble/hashes": "^2.2.0", "boxen": "^7.1.1", @@ -489,9 +489,9 @@ "license": "Apache-2.0" }, "node_modules/@git-stunts/git-cas": { - "version": "6.5.0", - "resolved": "https://registry.npmjs.org/@git-stunts/git-cas/-/git-cas-6.5.0.tgz", - "integrity": "sha512-KfKperNdXu3xWw07tpo1yYpLTynhwAP60PhYiZ5MRsSydPdNspQzJmi6Pv0Jz+6WULD883/NJCR0V1IUhBwOBw==", + "version": "6.5.3", + "resolved": "https://registry.npmjs.org/@git-stunts/git-cas/-/git-cas-6.5.3.tgz", + "integrity": "sha512-to7bk0BCcp0He5rSwViI7ZD0gb5CL0fFrIPbKpuIwXpuc9MBW0y5AzqvZFGicEncN4iwccEaIZm87paOfpEDrg==", "license": "Apache-2.0", "dependencies": { "@flyingrobots/bijou": "^5.0.0", @@ -499,7 +499,7 @@ "@flyingrobots/bijou-tui": "^5.0.0", "@flyingrobots/bijou-tui-app": "^5.0.0", "@git-stunts/alfred": "^0.10.0", - "@git-stunts/plumbing": "^3.1.0", + "@git-stunts/plumbing": "^3.2.0", "@git-stunts/vault": "^1.0.1", "cbor-x": "^1.6.0", "commander": "14.0.3", @@ -513,9 +513,9 @@ } }, "node_modules/@git-stunts/plumbing": { - "version": "3.1.0", - "resolved": "https://registry.npmjs.org/@git-stunts/plumbing/-/plumbing-3.1.0.tgz", - "integrity": "sha512-SLRySIS8FyTOy3mV43GY0xp/1QPu+Q/5CAbDE1HmL+NfnMhuMe6stiycjCQW10WLzbI6ihP+WKqrCgoMfAdRTw==", + "version": "3.2.0", + "resolved": "https://registry.npmjs.org/@git-stunts/plumbing/-/plumbing-3.2.0.tgz", + "integrity": "sha512-wW3Rzq6KNX0iygU9LpF8odQpcAyylH0KrUxNTDeRJHnwxfIamm7DWBL8lDkP1TqltE+4Bi0H0rjyPtvSkAJJeg==", "license": "Apache-2.0", "dependencies": { "zod": "^3.24.1" diff --git a/package.json b/package.json index 6150cca44..e2b0ba3fb 100644 --- a/package.json +++ b/package.json @@ -117,8 +117,8 @@ }, "dependencies": { "@git-stunts/alfred": "^0.10.4", - "@git-stunts/git-cas": "^6.5.0", - "@git-stunts/plumbing": "^3.1.0", + "@git-stunts/git-cas": "^6.5.3", + "@git-stunts/plumbing": "^3.2.0", "@git-stunts/trailer-codec": "^2.1.1", "@noble/hashes": "^2.2.0", "boxen": "^7.1.1", diff --git a/scripts/v18.0.0/migrations/graph-model/GraphModelMigrationScratchRuntimeReplayer.ts b/scripts/v18.0.0/migrations/graph-model/GraphModelMigrationScratchRuntimeReplayer.ts index 49176ddd4..c4d2ef1a5 100644 --- a/scripts/v18.0.0/migrations/graph-model/GraphModelMigrationScratchRuntimeReplayer.ts +++ b/scripts/v18.0.0/migrations/graph-model/GraphModelMigrationScratchRuntimeReplayer.ts @@ -106,7 +106,9 @@ export async function replayGraphModelMigrationScratchIntoRuntime( runtimeRepositoryPath = await mkdtemp(join(tmpdir(), 'git-warp-v18-runtime-replay-')); shouldCleanup = true; } - try { + let persistence: GitTimelineHistoryAdapter | null = null; + let runtimeStorage: GitCasRepositoryAdapter | null = null; + const replay = (async (): Promise => { const operations = await readGraphModelMigrationScratchOperationRecords({ repositoryPath: sourceRepositoryPath, scratchRefName: request.scratchRef.refName, @@ -115,8 +117,8 @@ export async function replayGraphModelMigrationScratchIntoRuntime( await plumbing.execute({ args: ['init', '-q'] }); await plumbing.execute({ args: ['config', 'user.email', 'git-warp@example.invalid'] }); await plumbing.execute({ args: ['config', 'user.name', 'git-warp migration replay'] }); - const persistence = new GitTimelineHistoryAdapter({ plumbing }); - const runtimeStorage = new GitCasRepositoryAdapter({ plumbing, history: persistence }); + persistence = new GitTimelineHistoryAdapter({ plumbing }); + runtimeStorage = new GitCasRepositoryAdapter({ plumbing, history: persistence }); const graph = await openRuntimeHostProduct({ persistence, runtimeStorage, @@ -132,11 +134,35 @@ export async function replayGraphModelMigrationScratchIntoRuntime( operationCount: operations.length, state, }); - } finally { - if (shouldCleanup && runtimeRepositoryPath !== null) { - await rm(runtimeRepositoryPath, { recursive: true, force: true }); + })(); + const [result] = await Promise.allSettled([replay]); + const failures: unknown[] = result.status === 'rejected' ? [result.reason] : []; + const cleanups = [ + async (): Promise => await runtimeStorage?.close(), + async (): Promise => await persistence?.close(), + async (): Promise => { + if (shouldCleanup && runtimeRepositoryPath !== null) { + await rm(runtimeRepositoryPath, { recursive: true, force: true }); + } + }, + ]; + for (const cleanup of cleanups) { + try { + await cleanup(); + } catch (error) { + failures.push(error); } } + if (failures.length === 1) { + throw failures[0]; + } + if (failures.length > 1) { + throw new AggregateError(failures, 'Runtime replay and cleanup failed'); + } + if (result.status === 'rejected') { + throw result.reason; + } + return result.value; } async function applyOperations( diff --git a/scripts/v18.0.0/migrations/graph-model/V17RestoredPublicReadLegacyReadingBuilder.ts b/scripts/v18.0.0/migrations/graph-model/V17RestoredPublicReadLegacyReadingBuilder.ts index 3df60cd47..e703f1add 100644 --- a/scripts/v18.0.0/migrations/graph-model/V17RestoredPublicReadLegacyReadingBuilder.ts +++ b/scripts/v18.0.0/migrations/graph-model/V17RestoredPublicReadLegacyReadingBuilder.ts @@ -79,7 +79,7 @@ async function readingWithRuntimeContentOids( repositoryPath: string | null, ): Promise { const resolver = await RuntimeContentOidResolver.open(repositoryPath); - try { + const [result] = await Promise.allSettled([(async () => { const facts: GenesisEquivalenceReadingFact[] = []; for (const fact of reading.facts) { facts.push(await factWithRuntimeContentOid(fact, graphId, resolver)); @@ -88,9 +88,21 @@ async function readingWithRuntimeContentOids( readingId: reading.readingId, facts, }); - } finally { - await resolver.close(); + })()]); + const [cleanup] = await Promise.allSettled([resolver.close()]); + if (result.status === 'rejected' && cleanup.status === 'rejected') { + throw new AggregateError( + [result.reason, cleanup.reason], + 'Legacy reading construction and runtime cleanup both failed', + ); + } + if (result.status === 'rejected') { + throw result.reason; } + if (cleanup.status === 'rejected') { + throw cleanup.reason; + } + return result.value; } async function factWithRuntimeContentOid( @@ -124,38 +136,59 @@ function nodeIdFromContentFactKey(factKey: string): string { } class RuntimeContentOidResolver { + private closePromise: Promise | null = null; + private constructor( private readonly repositoryPath: string, private readonly shouldCleanup: boolean, private readonly storage: AssetStoragePort, + private readonly runtimeStorage: GitCasRepositoryAdapter, + private readonly history: GitTimelineHistoryAdapter, ) { } static async open(repositoryPath: string | null): Promise { - let runtimeRepositoryPath = repositoryPath; - let shouldCleanup = false; - if (runtimeRepositoryPath === null) { - runtimeRepositoryPath = await mkdtemp(join(tmpdir(), 'git-warp-v18-content-oid-')); - shouldCleanup = true; + const shouldCleanup = repositoryPath === null; + const runtimeRepositoryPath = repositoryPath + ?? await mkdtemp(join(tmpdir(), 'git-warp-v18-content-oid-')); + let history: GitTimelineHistoryAdapter | null = null; + let runtimeStorage: GitCasRepositoryAdapter | null = null; + try { + const plumbing = await Plumbing.createDefault({ cwd: runtimeRepositoryPath }); + await plumbing.execute({ args: ['init', '-q'] }); + history = new GitTimelineHistoryAdapter({ plumbing }); + runtimeStorage = new GitCasRepositoryAdapter({ + plumbing, + history, + }); + const services = await runtimeStorage.createRuntimeStorageServices({ + timelineName: 'migration-content', + codec: defaultCodec, + crypto: defaultCrypto, + commitMessageCodec: DEFAULT_COMMIT_MESSAGE_CODEC, + }); + return new RuntimeContentOidResolver( + runtimeRepositoryPath, + shouldCleanup, + services.content, + runtimeStorage, + history, + ); + } catch (error) { + try { + await closeRuntimeContentOidResources( + runtimeStorage, + history, + shouldCleanup ? runtimeRepositoryPath : null, + ); + } catch (closeError) { + throw new AggregateError( + [error, closeError], + 'Runtime content resolver failed to open and clean up', + ); + } + throw error; } - const plumbing = await Plumbing.createDefault({ cwd: runtimeRepositoryPath }); - await plumbing.execute({ args: ['init', '-q'] }); - const adapter = new GitTimelineHistoryAdapter({ plumbing }); - const runtimeStorage = new GitCasRepositoryAdapter({ - plumbing, - history: adapter, - }); - const services = await runtimeStorage.createRuntimeStorageServices({ - timelineName: 'migration-content', - codec: defaultCodec, - crypto: defaultCrypto, - commitMessageCodec: DEFAULT_COMMIT_MESSAGE_CODEC, - }); - return new RuntimeContentOidResolver( - runtimeRepositoryPath, - shouldCleanup, - services.content, - ); } async oidFor(options: { @@ -173,11 +206,44 @@ class RuntimeContentOidResolver { return staged.handle.toString(); } - async close(): Promise { - if (this.shouldCleanup) { - await rm(this.repositoryPath, { recursive: true, force: true }); + close(): Promise { + this.closePromise ??= closeRuntimeContentOidResources( + this.runtimeStorage, + this.history, + this.shouldCleanup ? this.repositoryPath : null, + ); + return this.closePromise; + } +} + +async function closeRuntimeContentOidResources( + runtimeStorage: GitCasRepositoryAdapter | null, + history: GitTimelineHistoryAdapter | null, + temporaryRepositoryPath: string | null, +): Promise { + const failures: unknown[] = []; + const cleanups = [ + async (): Promise => await runtimeStorage?.close(), + async (): Promise => await history?.close(), + async (): Promise => { + if (temporaryRepositoryPath !== null) { + await rm(temporaryRepositoryPath, { recursive: true, force: true }); + } + }, + ]; + for (const cleanup of cleanups) { + try { + await cleanup(); + } catch (error) { + failures.push(error); } } + if (failures.length === 1) { + throw failures[0]; + } + if (failures.length > 1) { + throw new AggregateError(failures, 'Runtime content resolver failed to close cleanly'); + } } async function* singleChunk(bytes: Uint8Array): AsyncGenerator { diff --git a/src/application/WarpStorage.ts b/src/application/WarpStorage.ts index 0b246a14b..c8b02a49f 100644 --- a/src/application/WarpStorage.ts +++ b/src/application/WarpStorage.ts @@ -1,9 +1,26 @@ +type CloseStorage = () => Promise; + +const CLOSE_NOOP: CloseStorage = () => Promise.resolve(); + /** Opaque, runtime-backed handle for one WARP storage composition. */ export default abstract class WarpStorage { readonly #identity = 'warp-storage'; + readonly #closeStorage: CloseStorage; + #closePromise: Promise | null = null; - protected constructor() { + protected constructor(closeStorage: CloseStorage = CLOSE_NOOP) { void this.#identity; + this.#closeStorage = closeStorage; Object.freeze(this); } + + /** Releases local storage resources without changing admitted history. */ + close(): Promise { + this.#closePromise ??= Promise.resolve().then(this.#closeStorage); + return this.#closePromise; + } + + async [Symbol.asyncDispose](): Promise { + await this.close(); + } } diff --git a/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts b/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts index d519c29ec..1c199e8a0 100644 --- a/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts +++ b/src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts @@ -8,12 +8,10 @@ import type { WorkspaceRetainedBundle, WorkspaceRetainedPage, } from '@git-stunts/git-cas'; -import MaterializationCoordinate from '../../domain/materialization/MaterializationCoordinate.ts'; +import type MaterializationCoordinate from '../../domain/materialization/MaterializationCoordinate.ts'; import MaterializationHandle from '../../domain/materialization/MaterializationHandle.ts'; -import MaterializationRoots from '../../domain/materialization/MaterializationRoots.ts'; import BundleHandle from '../../domain/storage/BundleHandle.ts'; import type StorageRetentionWitness from '../../domain/storage/StorageRetentionWitness.ts'; -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'; @@ -25,7 +23,17 @@ import { adaptGitCasRetentionWitness } from './GitCasRetentionWitnessAdapter.ts' import GitCasMaterializationWorkspace, { type GitCasStagingWorkspace, } from './GitCasMaterializationWorkspace.ts'; +import GitCasMaterializationWorkspaceOwner from './GitCasMaterializationWorkspaceOwner.ts'; +import { + requireAdapterOptions, + requireCoordinate, + requireDependency, + requireNonEmpty, + requireRetainRequest, + storageError, +} from './GitCasMaterializationStoreValidation.ts'; import GitCasMaterializationLease from './GitCasMaterializationLease.ts'; +import { completeWithCleanup } from './OperationCleanup.ts'; import { decodeMaterializationDescriptor, MATERIALIZATION_DESCRIPTOR_SCHEMA_VERSION, @@ -69,9 +77,13 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt readonly #codec: CodecPort; readonly #crypto: CryptoPort; readonly #laneName: string; + readonly #onClose: () => void; #currentLease: GitCasMaterializationLease | null = null; #leaseMutation: Promise = Promise.resolve(); readonly #retirements = new Set>(); + readonly #workspaceOwner = new GitCasMaterializationWorkspaceOwner( + () => storageError('adapter is closed'), + ); #retirementFailure: Readonly<{ cause: unknown }> | null = null; #closed = false; #closePromise: Promise | null = null; @@ -81,6 +93,7 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt readonly codec: CodecPort; readonly crypto: CryptoPort; readonly laneName: string; + readonly onClose?: () => void; }) { super(); requireAdapterOptions(options); @@ -91,35 +104,41 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt this.#codec = options.codec; this.#crypto = options.crypto; this.#laneName = requireNonEmpty(options.laneName, 'laneName'); + this.#onClose = options.onClose ?? (() => undefined); } override async openWorkspace( coordinate: MaterializationCoordinate, ): Promise { + this.#assertOpen(); requireCoordinate(coordinate); - const workspace = await this.#cas.workspaces.open({ - namespace: WORKSPACE_NAMESPACE, - ttlMs: WORKSPACE_TTL_MS, - }); - return new GitCasMaterializationWorkspace({ - workspace, - promote: async (activeWorkspace, request) => { - if (!request.coordinate.equals(coordinate)) { - throw storageError('workspace promotion coordinate does not match its open coordinate'); - } - return await this.#promoteWorkspace(activeWorkspace, request); - }, + return await this.#workspaceOwner.open({ + open: async () => await this.#cas.workspaces.open({ + namespace: WORKSPACE_NAMESPACE, + ttlMs: WORKSPACE_TTL_MS, + }), + create: (workspace, onRelease) => new GitCasMaterializationWorkspace({ + workspace, + promote: async (activeWorkspace, request) => { + if (!request.coordinate.equals(coordinate)) { + throw storageError('workspace promotion coordinate does not match its open coordinate'); + } + return await this.#promoteWorkspace(activeWorkspace, request); + }, + onRelease, + }), }); } override async retain(request: RetainMaterializationRequest): Promise { + this.#assertOpen(); requireRetainRequest(request); const workspace = await this.openWorkspace(request.coordinate); - try { - return await workspace.promote(request); - } finally { - await workspace.release(); - } + return await completeWithCleanup( + async () => await workspace.promote(request), + async () => await workspace.release(), + 'Materialization promotion and workspace release both failed', + ); } async #promoteWorkspace( @@ -198,17 +217,22 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt if (acquisition === null) { throw storageError('git-cas lost the retained materialization before legacy cleanup'); } - try { - requireExpectedAcquisition(acquisition, args.expectedHandle); - await this.#removeLegacyEntry(args.cache, args.coordinate); - } finally { - await acquisition.release(); - } + await completeWithCleanup( + async () => { + requireExpectedAcquisition(acquisition, args.expectedHandle); + await this.#removeLegacyEntry(args.cache, args.coordinate); + }, + async () => { + await acquisition.release(); + }, + 'Legacy materialization cleanup and acquisition release both failed', + ); } override async acquireExact( coordinate: MaterializationCoordinate, ): Promise { + this.#assertOpen(); requireCoordinate(coordinate); return await this.#withLeaseMutation( async () => await this.#acquireExactLocked(coordinate), @@ -216,7 +240,10 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt } override close(): Promise { - this.#closePromise ??= this.#close(); + if (this.#closePromise === null) { + this.#closed = true; + this.#closePromise = this.#close().finally(this.#onClose); + } return this.#closePromise; } @@ -276,8 +303,14 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt } async #close(): Promise { + const failures: unknown[] = []; + try { + await this.#workspaceOwner.close(); + } catch (error) { + failures.push(error); + } + await this.#withLeaseMutation(() => { - this.#closed = true; if (this.#currentLease !== null) { this.#retireLease(this.#currentLease); this.#currentLease = null; @@ -286,7 +319,13 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt }); await Promise.allSettled([...this.#retirements]); if (this.#retirementFailure !== null) { - throw this.#retirementFailure.cause; + failures.push(this.#retirementFailure.cause); + } + if (failures.length === 1) { + throw failures[0]; + } + if (failures.length > 1) { + throw new AggregateError(failures, 'Materialization storage failed to close cleanly'); } } @@ -316,6 +355,12 @@ export default class GitCasMaterializationStoreAdapter extends MaterializationSt } } + #assertOpen(): void { + if (this.#closed) { + throw storageError('adapter is closed'); + } + } + async #resolveHit( hit: CacheHit, retention: CacheAcquisition['evidence'], @@ -431,54 +476,8 @@ function requireExpectedAcquisition( } } -function requireRetainRequest(request: RetainMaterializationRequest): void { - if (request === null || typeof request !== 'object' || Array.isArray(request)) { - throw storageError('retain request must be an object'); - } - requireCoordinate(request.coordinate); - if (!(request.roots instanceof MaterializationRoots)) { - throw storageError('retain request roots have an invalid runtime identity'); - } - requireCurrentPropertyRoot(request.roots); -} - -function requireCurrentPropertyRoot(roots: MaterializationRoots): void { - if (roots.properties.status === 'unavailable') { - throw storageError('current materialization profile requires a property root'); - } -} - -function requireCoordinate(coordinate: MaterializationCoordinate): void { - if (!(coordinate instanceof MaterializationCoordinate)) { - throw storageError('coordinate has an invalid runtime identity'); - } -} - function requireDescriptorSize(bytes: Uint8Array): void { if (bytes.byteLength > MAX_DESCRIPTOR_BYTES) { throw storageError('materialization descriptor exceeds its byte limit'); } } - -function requireDependency(value: object, field: string): void { - if (value === null || typeof value !== 'object') { - throw storageError(`${field} dependency is required`); - } -} - -function requireAdapterOptions(options: object): void { - if (options === null || typeof options !== 'object' || Array.isArray(options)) { - throw storageError('adapter options must be an object'); - } -} - -function requireNonEmpty(value: unknown, field: string): string { - if (typeof value !== 'string' || value.length === 0) { - throw storageError(`${field} must be a non-empty string`); - } - return value; -} - -function storageError(message: string): WarpError { - return new WarpError(`Materialization storage ${message}`, 'E_MATERIALIZATION_STORAGE'); -} diff --git a/src/infrastructure/adapters/GitCasMaterializationStoreValidation.ts b/src/infrastructure/adapters/GitCasMaterializationStoreValidation.ts new file mode 100644 index 000000000..acbd39dc4 --- /dev/null +++ b/src/infrastructure/adapters/GitCasMaterializationStoreValidation.ts @@ -0,0 +1,50 @@ +import MaterializationCoordinate from '../../domain/materialization/MaterializationCoordinate.ts'; +import MaterializationRoots from '../../domain/materialization/MaterializationRoots.ts'; +import WarpError from '../../domain/errors/WarpError.ts'; +import type { RetainMaterializationRequest } from '../../ports/MaterializationStorePort.ts'; + +export function requireRetainRequest(request: RetainMaterializationRequest): void { + if (request === null || typeof request !== 'object' || Array.isArray(request)) { + throw storageError('retain request must be an object'); + } + requireCoordinate(request.coordinate); + requireRetainRoots(request.roots); +} + +function requireRetainRoots(roots: MaterializationRoots): void { + if (!(roots instanceof MaterializationRoots)) { + throw storageError('retain request roots have an invalid runtime identity'); + } + if (roots.properties.status === 'unavailable') { + throw storageError('current materialization profile requires a property root'); + } +} + +export function requireCoordinate(coordinate: MaterializationCoordinate): void { + if (!(coordinate instanceof MaterializationCoordinate)) { + throw storageError('coordinate has an invalid runtime identity'); + } +} + +export function requireDependency(value: object, field: string): void { + if (value === null || typeof value !== 'object') { + throw storageError(`${field} dependency is required`); + } +} + +export function requireAdapterOptions(options: object): void { + if (options === null || typeof options !== 'object' || Array.isArray(options)) { + throw storageError('adapter options must be an object'); + } +} + +export function requireNonEmpty(value: unknown, field: string): string { + if (typeof value !== 'string' || value.length === 0) { + throw storageError(`${field} must be a non-empty string`); + } + return value; +} + +export function storageError(message: string): WarpError { + return new WarpError(`Materialization storage ${message}`, 'E_MATERIALIZATION_STORAGE'); +} diff --git a/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts b/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts index f9a072165..a03d7b50b 100644 --- a/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts +++ b/src/infrastructure/adapters/GitCasMaterializationWorkspace.ts @@ -41,12 +41,14 @@ export type GitCasMaterializationWorkspaceOptions = Readonly<{ workspace: GitCasStagingWorkspace, request: PromoteMaterializationRequest, ) => Promise; + onRelease?: () => void; }>; /** git-cas-owned retention scope for one in-progress materialization. */ export default class GitCasMaterializationWorkspace extends MaterializationWorkspacePort { readonly #workspace: GitCasStagingWorkspace; readonly #promoteMaterialization: GitCasMaterializationWorkspaceOptions['promote']; + readonly #onRelease: () => void; #promoting = false; #promoted = false; #releaseRequested = false; @@ -59,6 +61,7 @@ export default class GitCasMaterializationWorkspace extends MaterializationWorks requireWorkspaceOptions(options); this.#workspace = options.workspace; this.#promoteMaterialization = options.promote; + this.#onRelease = options.onRelease ?? (() => undefined); } override stagePage( @@ -131,6 +134,7 @@ export default class GitCasMaterializationWorkspace extends MaterializationWorks if (!this.#released) { await this.#workspace.release(); this.#released = true; + this.#onRelease(); } }); return this.#releasePromise; diff --git a/src/infrastructure/adapters/GitCasMaterializationWorkspaceOwner.ts b/src/infrastructure/adapters/GitCasMaterializationWorkspaceOwner.ts new file mode 100644 index 000000000..8e50379b8 --- /dev/null +++ b/src/infrastructure/adapters/GitCasMaterializationWorkspaceOwner.ts @@ -0,0 +1,92 @@ +import type GitCasMaterializationWorkspace from './GitCasMaterializationWorkspace.ts'; +import type { GitCasStagingWorkspace } from './GitCasMaterializationWorkspace.ts'; + +type WorkspaceOpenRequest = Readonly<{ + open(): Promise; + create( + workspace: GitCasStagingWorkspace, + onRelease: () => void, + ): GitCasMaterializationWorkspace; +}>; + +/** Owns materialization workspaces across open/close races. */ +export default class GitCasMaterializationWorkspaceOwner { + readonly #closedError: () => Error; + readonly #openings = new Set>(); + readonly #workspaces = new Set(); + #closed = false; + #closePromise: Promise | null = null; + + constructor(closedError: () => Error) { + this.#closedError = closedError; + } + + async open(request: WorkspaceOpenRequest): Promise { + if (this.#closed) { + throw this.#closedError(); + } + const opening = this.#open(request); + this.#openings.add(opening); + let workspace: GitCasMaterializationWorkspace; + try { + workspace = await opening; + } finally { + this.#openings.delete(opening); + } + if (this.#closed) { + const closed = this.#closedError(); + try { + await workspace.release(); + } catch (releaseError) { + throw new AggregateError( + [closed, releaseError], + 'Materialization workspace opened during owner closure and failed to release', + ); + } + throw closed; + } + return workspace; + } + + close(): Promise { + this.#closed = true; + this.#closePromise ??= this.#close(); + return this.#closePromise; + } + + async #open(request: WorkspaceOpenRequest): Promise { + const staging = await request.open(); + try { + const workspace = request.create(staging, () => this.#workspaces.delete(workspace)); + this.#workspaces.add(workspace); + return workspace; + } catch (error) { + try { + await staging.release(); + } catch (releaseError) { + throw new AggregateError( + [error, releaseError], + 'Materialization workspace failed to initialize and release', + ); + } + throw error; + } + } + + async #close(): Promise { + await Promise.allSettled([...this.#openings]); + const results = await Promise.allSettled( + [...this.#workspaces].map(async (workspace) => await workspace.release()), + ); + this.#workspaces.clear(); + const failures = results + .filter((result): result is PromiseRejectedResult => result.status === 'rejected') + .map((result) => result.reason as unknown); + if (failures.length === 1) { + throw failures[0]; + } + if (failures.length > 1) { + throw new AggregateError(failures, 'Materialization workspaces failed to close cleanly'); + } + } +} diff --git a/src/infrastructure/adapters/GitCasRepositoryAdapter.ts b/src/infrastructure/adapters/GitCasRepositoryAdapter.ts index fbc219630..71bcfa452 100644 --- a/src/infrastructure/adapters/GitCasRepositoryAdapter.ts +++ b/src/infrastructure/adapters/GitCasRepositoryAdapter.ts @@ -30,6 +30,7 @@ import GitTrustChainAdapter from './GitTrustChainAdapter.ts'; import type { GitPlumbing } from './gitErrorClassification.ts'; import LoggerObservabilityBridge from './LoggerObservabilityBridge.ts'; import type GitTimelineHistoryAdapter from './GitTimelineHistoryAdapter.ts'; +import AdapterValidationError from '../../domain/errors/AdapterValidationError.ts'; type GitCasPolicy = { execute(operation: () => Promise): Promise; @@ -71,15 +72,18 @@ export default class GitCasRepositoryAdapter implements RuntimeStorageProviderPo private readonly _plumbing: GitPlumbing; private readonly _history: GitTimelineHistoryAdapter; private readonly _cas: GitCasFacade; + private readonly _closeCas: (() => Promise) | null; private readonly _cbor: InstanceType; private readonly _contentEncryption: CasContentEncryptionPolicy | undefined; + private readonly _materializations = new Set(); + private _closePromise: Promise | null = null; + private _closed = false; constructor(options: GitCasRepositoryAdapterOptions) { this._plumbing = options.plumbing; this._history = options.history; - this._cas = - options.cas ?? - ContentAddressableStore.createCbor({ + if (options.cas === undefined) { + const cas = ContentAddressableStore.createCbor({ plumbing: options.plumbing, chunking: { strategy: 'cdc' }, applicationRefPrefixes: ['refs/warp/'], @@ -88,12 +92,21 @@ export default class GitCasRepositoryAdapter implements RuntimeStorageProviderPo ? {} : { observability: new LoggerObservabilityBridge(options.logger) }), }); + this._cas = cas; + this._closeCas = async () => await cas.close(); + } else { + this._cas = options.cas; + this._closeCas = null; + } this._cbor = new CborCodec(); this._contentEncryption = options.contentEncryption; } createRuntimeStorageServices(request: RuntimeStorageRequest): Promise { + this._assertOpen(); const content = this._createContentStorage(); + const materializations = this._createMaterializationStore(request); + this._materializations.add(materializations); return Promise.resolve( Object.freeze({ content, @@ -103,7 +116,7 @@ export default class GitCasRepositoryAdapter implements RuntimeStorageProviderPo patchJournal: this._createPatchJournal(request, content), checkpoints: this._createCheckpointStore(request, content), indexes: this._createIndexStore(request, content), - materializations: this._createMaterializationStore(request), + materializations, stateSnapshots: this._createStateSnapshots(request), trie: new GitCasTrieStoreAdapter({ cas: this._cas }), }) @@ -191,15 +204,18 @@ export default class GitCasRepositoryAdapter implements RuntimeStorageProviderPo private _createMaterializationStore( request: RuntimeStorageRequest, ): GitCasMaterializationStoreAdapter { - return new GitCasMaterializationStoreAdapter({ + const materializations = new GitCasMaterializationStoreAdapter({ cas: this._cas, codec: request.codec, crypto: request.crypto, laneName: request.timelineName, + onClose: () => this._materializations.delete(materializations), }); + return materializations; } createTrustChain(crypto: CryptoPort): GitTrustChainAdapter { + this._assertOpen(); return new GitTrustChainAdapter({ cas: this._cas, cbor: this._cbor, @@ -217,4 +233,43 @@ export default class GitCasRepositoryAdapter implements RuntimeStorageProviderPo : { contentEncryption: this._contentEncryption }), }); } + + /** Releases repository-scoped local resources without changing retained data. */ + close(): Promise { + this._closed = true; + this._closePromise ??= this._close(); + return this._closePromise; + } + + async [Symbol.asyncDispose](): Promise { + await this.close(); + } + + private async _close(): Promise { + const materializationResults = await Promise.allSettled( + [...this._materializations].map(async (materializations) => await materializations.close()), + ); + this._materializations.clear(); + const failures = materializationResults + .filter((result): result is PromiseRejectedResult => result.status === 'rejected') + .map((result) => result.reason as unknown); + + if (this._closeCas !== null) { + try { + await this._closeCas(); + } catch (error) { + failures.push(error); + } + } + + if (failures.length > 0) { + throw new AggregateError(failures, 'Git CAS repository storage failed to close cleanly'); + } + } + + private _assertOpen(): void { + if (this._closed) { + throw new AdapterValidationError('Git CAS repository storage is closed'); + } + } } diff --git a/src/infrastructure/adapters/GitTimelineHistoryAdapter.ts b/src/infrastructure/adapters/GitTimelineHistoryAdapter.ts index 56eb5a823..c468e02e5 100644 --- a/src/infrastructure/adapters/GitTimelineHistoryAdapter.ts +++ b/src/infrastructure/adapters/GitTimelineHistoryAdapter.ts @@ -101,6 +101,7 @@ export default class GitTimelineHistoryAdapter extends GraphPersistencePort { private readonly _gitCasPersistence: GitPersistenceAdapter; private readonly _gitCasGraphReader: GitCasGraphReaderAdapter; private readonly _recursiveTreeOidReader: GitRecursiveTreeOidReaderAdapter; + private _closePromise: Promise | null = null; constructor({ plumbing, retryOptions = {}, policy }: GitTimelineHistoryAdapterOptions) { super(); @@ -130,6 +131,16 @@ export default class GitTimelineHistoryAdapter extends GraphPersistencePort { }); } + /** Releases local Git object sessions without changing timeline history. */ + close(): Promise { + this._closePromise ??= this._gitCasPersistence.close(); + return this._closePromise; + } + + async [Symbol.asyncDispose](): Promise { + await this.close(); + } + private async _executeWithRetry(options: { args: string[]; input?: string | Buffer; diff --git a/src/infrastructure/adapters/OperationCleanup.ts b/src/infrastructure/adapters/OperationCleanup.ts new file mode 100644 index 000000000..2dcd66821 --- /dev/null +++ b/src/infrastructure/adapters/OperationCleanup.ts @@ -0,0 +1,19 @@ +/** Runs cleanup after an operation while preserving every failure. */ +export async function completeWithCleanup( + operation: () => Promise, + cleanup: () => Promise, + aggregateMessage: string, +): Promise { + const [result] = await Promise.allSettled([Promise.resolve().then(operation)]); + const [cleanupResult] = await Promise.allSettled([Promise.resolve().then(cleanup)]); + if (result.status === 'rejected' && cleanupResult.status === 'rejected') { + throw new AggregateError([result.reason, cleanupResult.reason], aggregateMessage); + } + if (result.status === 'rejected') { + throw result.reason; + } + if (cleanupResult.status === 'rejected') { + throw cleanupResult.reason; + } + return result.value; +} diff --git a/storage.ts b/storage.ts index 9e94cd02e..a49f4a662 100644 --- a/storage.ts +++ b/storage.ts @@ -13,26 +13,64 @@ export type GitStorageOptions = { readonly cwd: string; }; +async function closeGitStorageResources( + repository: GitCasRepositoryAdapter | null, + history: GitTimelineHistoryAdapter, +): Promise { + const failures: unknown[] = []; + if (repository !== null) { + try { + await repository.close(); + } catch (error) { + failures.push(error); + } + } + try { + await history.close(); + } catch (error) { + failures.push(error); + } + if (failures.length > 0) { + throw new AggregateError(failures, 'Git storage failed to close cleanly'); + } +} + export class GitStorage extends WarpStorage { - private constructor() { - super(); + private constructor(closeStorage: () => Promise) { + super(closeStorage); } static async open(options: GitStorageOptions): Promise { const plumbing = await GitPlumbing.createDefault({ cwd: options.cwd }); const history = new GitTimelineHistoryAdapter({ plumbing }); - const available = await history.ping(); - if (!available.ok) { - throw new AdapterValidationError(`Repository is not accessible: ${options.cwd}`); + let repository: GitCasRepositoryAdapter | null = null; + try { + const available = await history.ping(); + if (!available.ok) { + throw new AdapterValidationError(`Repository is not accessible: ${options.cwd}`); + } + const openedRepository = new GitCasRepositoryAdapter({ plumbing, history }); + repository = openedRepository; + const storage = new GitStorage( + async () => await closeGitStorageResources(openedRepository, history), + ); + bindWarpStorage(storage, { + history, + runtimeStorage: openedRepository, + createTrustChain: (crypto) => openedRepository.createTrustChain(crypto), + hookPaths: new PlumbingHookPathAdapter({ plumbing, path }), + }); + return storage; + } catch (error) { + try { + await closeGitStorageResources(repository, history); + } catch (closeError) { + throw new AggregateError( + [error, closeError], + 'Git storage failed to open and release local resources', + ); + } + throw error; } - const repository = new GitCasRepositoryAdapter({ plumbing, history }); - const storage = new GitStorage(); - bindWarpStorage(storage, { - history, - runtimeStorage: repository, - createTrustChain: (crypto) => repository.createTrustChain(crypto), - hookPaths: new PlumbingHookPathAdapter({ plumbing, path }), - }); - return storage; } } diff --git a/test/bats/cli-mcp.bats b/test/bats/cli-mcp.bats new file mode 100644 index 000000000..74b437c11 --- /dev/null +++ b/test/bats/cli-mcp.bats @@ -0,0 +1,56 @@ +#!/usr/bin/env bats + +load helpers/setup.bash + +setup() { + setup_test_repo + seed_graph "seed-graph.js" +} + +teardown() { + teardown_test_repo +} + +@test "mcp reports its package version and exits after stdin EOF" { + run python3 - "${PROJECT_ROOT}" "${TEST_REPO}" <<'PY' +import json +import os +import subprocess +import sys + +project_root = sys.argv[1] +dist_cli = os.path.join(project_root, "dist", "bin", "warp-graph.js") +source_cli = os.path.join(project_root, "bin", "warp-graph.ts") +cli = dist_cli if os.path.exists(dist_cli) else source_cli +request = json.dumps({ + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": {"protocolVersion": "2025-06-18"}, +}) + "\n" +result = subprocess.run( + ["node", cli, "--repo", sys.argv[2], "--graph", "demo", "mcp"], + input=request, + text=True, + capture_output=True, + timeout=10, + check=False, +) +if result.returncode != 0: + sys.stderr.write(result.stderr) + raise SystemExit(result.returncode) +sys.stdout.write(result.stdout) +PY + assert_success + + JSON="$output" PACKAGE_JSON="${PROJECT_ROOT}/package.json" python3 - <<'PY' +import json +import os + +response = json.loads(os.environ["JSON"]) +with open(os.environ["PACKAGE_JSON"], encoding="utf-8") as package_file: + expected_version = json.load(package_file)["version"] +assert response["result"]["serverInfo"]["name"] == "git-warp" +assert response["result"]["serverInfo"]["version"] == expected_version +PY +} diff --git a/test/bats/helpers/seed-setup.ts b/test/bats/helpers/seed-setup.ts index 4f343c597..d70b1b41e 100644 --- a/test/bats/helpers/seed-setup.ts +++ b/test/bats/helpers/seed-setup.ts @@ -7,7 +7,7 @@ import { existsSync } from 'node:fs'; import { resolve } from 'node:path'; import { pathToFileURL } from 'node:url'; -import GitPlumbing, { ShellRunnerFactory } from '@git-stunts/plumbing'; +import GitPlumbing from '@git-stunts/plumbing'; const projectRoot = process.env['PROJECT_ROOT'] || resolve(import.meta.dirname, '../../..'); const repoPath = process.env['REPO_PATH']; @@ -43,8 +43,7 @@ const { installDefaultRuntimeHostNodePorts } = await import(runtimeNodeDefaultsU installDefaultRuntimeHostNodePorts(); -const runner = ShellRunnerFactory.create(); -const plumbing = new GitPlumbing({ cwd: repoPath, runner }); +const plumbing = await GitPlumbing.createDefault({ cwd: repoPath }); const persistence = new GitTimelineHistoryAdapter({ plumbing }); const crypto = new NodeCryptoAdapter(); const runtimeStorage = new GitCasRepositoryAdapter({ plumbing, history: persistence }); @@ -64,4 +63,21 @@ function createTrustChain() { return runtimeStorage.createTrustChain(crypto); } -export { openGraph, persistence, crypto, createTrustChain }; +async function closeSeedRuntime(): Promise { + const failures: unknown[] = []; + for (const close of [ + async (): Promise => await runtimeStorage.close(), + async (): Promise => await persistence.close(), + ]) { + try { + await close(); + } catch (error) { + failures.push(error); + } + } + if (failures.length > 0) { + throw new AggregateError(failures, 'Seed runtime failed to close cleanly'); + } +} + +export { openGraph, persistence, crypto, createTrustChain, closeSeedRuntime }; diff --git a/test/bats/helpers/setup.bash b/test/bats/helpers/setup.bash index 66a933330..c06bb4c37 100644 --- a/test/bats/helpers/setup.bash +++ b/test/bats/helpers/setup.bash @@ -60,7 +60,27 @@ seed_graph() { cd "${PROJECT_ROOT}" || return 1 NODE_NO_WARNINGS=1 REPO_PATH="${TEST_REPO}" node --experimental-strip-types -e ' import("node:url") - .then(({ pathToFileURL }) => import(pathToFileURL(process.argv[1]).href)) + .then(async ({ pathToFileURL }) => { + const seedUrl = pathToFileURL(process.argv[1]).href; + const setupUrl = pathToFileURL(process.argv[2]).href; + let failure; + try { + await import(seedUrl); + } catch (error) { + failure = error; + } + try { + const { closeSeedRuntime } = await import(setupUrl); + await closeSeedRuntime(); + } catch (error) { + failure = failure === undefined + ? error + : new AggregateError([failure, error], "Seed and cleanup failed"); + } + if (failure !== undefined) { + throw failure; + } + }) .then( () => process.exit(0), (error) => { @@ -68,6 +88,6 @@ seed_graph() { process.exit(1); }, ); - ' "${script}" + ' "${script}" "${BATS_TEST_DIRNAME}/helpers/seed-setup.ts" cd "${TEST_REPO}" || return 1 } diff --git a/test/helpers/WarpGraphTestRepositories.ts b/test/helpers/WarpGraphTestRepositories.ts index 773a75394..c63b90374 100644 --- a/test/helpers/WarpGraphTestRepositories.ts +++ b/test/helpers/WarpGraphTestRepositories.ts @@ -15,7 +15,11 @@ export class GitRepoFixture { ) {} readonly cleanup = async (): Promise => { - await rm(this.tempDir, { recursive: true, force: true }); + try { + await this.persistence.close(); + } finally { + await rm(this.tempDir, { recursive: true, force: true }); + } }; } diff --git a/test/integration/WarpGraph.integration.test.ts b/test/integration/WarpGraph.integration.test.ts index 9c801c705..380a55b2c 100644 --- a/test/integration/WarpGraph.integration.test.ts +++ b/test/integration/WarpGraph.integration.test.ts @@ -36,7 +36,15 @@ describe('WarpCore Integration', () => { }); afterEach(async () => { - await rm(tempDir, { recursive: true, force: true }); + try { + await runtimeStorage.close(); + } finally { + try { + await persistence.close(); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } + } }); describe('Single Writer Workflow', () => { diff --git a/test/integration/api/checkpoint.test.ts b/test/integration/api/checkpoint.test.ts index 81c860a42..242a9ac51 100644 --- a/test/integration/api/checkpoint.test.ts +++ b/test/integration/api/checkpoint.test.ts @@ -20,16 +20,20 @@ async function readCheckpointArtifacts(repo, checkpointSha) { chunking: { strategy: 'cdc' }, applicationRefPrefixes: ['refs/warp/'], }); - const members: BundleMember[] = []; - for await (const member of cas.bundles.iterateMembers({ - handle: decoded.bundleHandle.toString(), - })) { - members.push(member); + try { + const members: BundleMember[] = []; + for await (const member of cas.bundles.iterateMembers({ + handle: decoded.bundleHandle.toString(), + })) { + members.push(member); + } + const memberHandles = Object.fromEntries( + members.map((member) => [member.path, member.handle.toString()]), + ); + return { decoded, members, memberHandles }; + } finally { + await cas.close(); } - const memberHandles = Object.fromEntries( - members.map((member) => [member.path, member.handle.toString()]), - ); - return { decoded, members, memberHandles }; } describe('API: Checkpoint', () => { diff --git a/test/integration/api/helpers/setup.ts b/test/integration/api/helpers/setup.ts index 9ce9a2b5d..7de11e564 100644 --- a/test/integration/api/helpers/setup.ts +++ b/test/integration/api/helpers/setup.ts @@ -66,7 +66,15 @@ export async function createTestRepo(label = 'api-test') { crypto, openGraph, async cleanup() { - await rm(tempDir, { recursive: true, force: true }); + try { + await runtimeStorage.close(); + } finally { + try { + await persistence.close(); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } + } }, }; } catch (err) { diff --git a/test/integration/api/materialization.retainedResume.test.ts b/test/integration/api/materialization.retainedResume.test.ts index 9a52ced58..3ae52673d 100644 --- a/test/integration/api/materialization.retainedResume.test.ts +++ b/test/integration/api/materialization.retainedResume.test.ts @@ -24,20 +24,35 @@ const execFileAsync = promisify(execFile); describe('API: retained materialization resume', () => { let repo: Awaited> | null = null; + let providers: RecordingRuntimeStorageProvider[] = []; beforeEach(async () => { repo = await createTestRepo('retained-materialization-resume'); + providers = []; }); afterEach(async () => { - await repo?.cleanup(); + const results = await Promise.allSettled( + providers.map(async (provider) => await provider.close()), + ); + const failures = results + .filter((result): result is PromiseRejectedResult => result.status === 'rejected') + .map((result) => result.reason as unknown); + try { + await repo?.cleanup(); + } catch (error) { + failures.push(error); + } + if (failures.length > 0) { + throw new AggregateError(failures, 'Retained materialization test cleanup failed'); + } }); it('reopens exact roots without replay in-process and through a fresh runtime adapter', async () => { if (repo === null) { throw new Error('Test repository is not initialized'); } - const firstProvider = recordingProvider(repo); + const firstProvider = recordingProvider(repo, providers); const firstRuntime = await openRuntime(repo, firstProvider); await firstRuntime.patch((patch) => { patch.addNode('node:retained'); @@ -61,7 +76,7 @@ describe('API: retained materialization resume', () => { expect(firstStore.exactHits[0]?.bundle.equals(coldHandle?.bundle)).toBe(true); expect(warm).toEqual(cold); - const reopenedProvider = recordingProvider(repo); + const reopenedProvider = recordingProvider(repo, providers); const reopenedRuntime = await openRuntime(repo, reopenedProvider); const reopenedReplay = vi.spyOn(reopenedRuntime, '_loadPatchChainFromSha'); const reopened = await reopenedRuntime.materialize(); @@ -78,7 +93,7 @@ describe('API: retained materialization resume', () => { if (repo === null) { throw new Error('Test repository is not initialized'); } - const firstProvider = recordingProvider(repo); + const firstProvider = recordingProvider(repo, providers); const firstRuntime = await openRuntime(repo, firstProvider); await firstRuntime.patch((patch) => { patch.addNode('node:retained'); @@ -95,7 +110,7 @@ describe('API: retained materialization resume', () => { ]); await execFileAsync('git', ['-C', repo.tempDir, 'prune', '--expire=now']); - const reopenedProvider = recordingProvider(repo); + const reopenedProvider = recordingProvider(repo, providers); const reopenedRuntime = await openRuntime(repo, reopenedProvider); const replay = vi.spyOn(reopenedRuntime, '_loadPatchChainFromSha'); const publishWholeState = vi.spyOn(reopenedRuntime, '_onMaterialized'); @@ -115,7 +130,7 @@ describe('API: retained materialization resume', () => { if (repo === null) { throw new Error('Test repository is not initialized'); } - const firstProvider = recordingProvider(repo); + const firstProvider = recordingProvider(repo, providers); const firstRuntime = await openRuntime(repo, firstProvider); await firstRuntime.patch((patch) => { patch @@ -139,7 +154,7 @@ describe('API: retained materialization resume', () => { ]); await execFileAsync('git', ['-C', repo.tempDir, 'prune', '--expire=now']); - const reopenedProvider = recordingProvider(repo); + const reopenedProvider = recordingProvider(repo, providers); const reopenedRuntime = await openRuntime(repo, reopenedProvider); const replay = vi.spyOn(reopenedRuntime, '_loadPatchChainFromSha'); const publishWholeState = vi.spyOn(reopenedRuntime, '_onMaterialized'); @@ -255,10 +270,10 @@ class RecordingMaterializationWorkspace extends MaterializationWorkspacePort { } class RecordingRuntimeStorageProvider implements RuntimeStorageProviderPort { - readonly #delegate: RuntimeStorageProviderPort; + readonly #delegate: GitCasRepositoryAdapter; materializations: RecordingMaterializationStore | null = null; - constructor(delegate: RuntimeStorageProviderPort) { + constructor(delegate: GitCasRepositoryAdapter) { this.#delegate = delegate; } @@ -270,15 +285,22 @@ class RecordingRuntimeStorageProvider implements RuntimeStorageProviderPort { this.materializations = materializations; return Object.freeze({ ...services, materializations }); } + + close(): Promise { + return this.#delegate.close(); + } } function recordingProvider( repo: NonNullable>>, + providers: RecordingRuntimeStorageProvider[], ): RecordingRuntimeStorageProvider { - return new RecordingRuntimeStorageProvider(new GitCasRepositoryAdapter({ + const provider = new RecordingRuntimeStorageProvider(new GitCasRepositoryAdapter({ plumbing: repo.plumbing, history: repo.persistence, })); + providers.push(provider); + return provider; } async function openRuntime( diff --git a/test/integration/api/querybuilder.test.ts b/test/integration/api/querybuilder.test.ts index 37895b2c8..a16ada21d 100644 --- a/test/integration/api/querybuilder.test.ts +++ b/test/integration/api/querybuilder.test.ts @@ -1,11 +1,23 @@ -import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; import { createTestRepo } from './helpers/setup.ts'; +import type { AggregateResult } from '../../../src/domain/services/query/QueryAggregation.ts'; +import type { QueryResult } from '../../../src/domain/services/query/QueryRunner.ts'; + +type TestRepo = Awaited>; +type TestGraph = Awaited>; + +function requireQueryResult(result: AggregateResult | QueryResult): QueryResult { + if (!('nodes' in result)) { + throw new Error('QueryBuilder fixture expected a node result'); + } + return result; +} describe('API: QueryBuilder', () => { - let repo; - let graph; + let repo: TestRepo; + let graph: TestGraph; - beforeEach(async () => { + beforeAll(async () => { repo = await createTestRepo('query'); graph = await repo.openGraph('test', 'alice'); @@ -31,15 +43,17 @@ describe('API: QueryBuilder', () => { .commit(); await graph.materialize(); - }); + }, 30_000); - afterEach(async () => { - await repo?.cleanup(); + afterAll(async () => { + await repo.cleanup(); }); it('match glob returns matching nodes', async () => { - const result = await graph.query().match('user:*').select(['id']).run(); - const ids = result.nodes.map((/** @type {any} */ n) => n.id); + const result = requireQueryResult( + await graph.query().match('user:*').select(['id']).run(), + ); + const ids = result.nodes.map((node) => node.id); expect(ids).toContain('user:alice'); expect(ids).toContain('user:bob'); expect(ids).toContain('user:carol'); @@ -47,69 +61,82 @@ describe('API: QueryBuilder', () => { }); it('where filters by property', async () => { - const result = await graph - .query() - .match('user:*') - .where({ role: 'engineering' }) - .select(['id']) - .run(); - const ids = result.nodes.map((/** @type {any} */ n) => n.id); + const result = requireQueryResult( + await graph + .query() + .match('user:*') + .where({ role: 'engineering' }) + .select(['id']) + .run(), + ); + const ids = result.nodes.map((node) => node.id); expect(ids).toContain('user:alice'); expect(ids).toContain('user:bob'); expect(ids).not.toContain('user:carol'); }); it('outgoing traversal follows edges', async () => { - const result = await graph - .query() - .match('user:alice') - .outgoing('manages') - .select(['id']) - .run(); - const ids = result.nodes.map((/** @type {any} */ n) => n.id); + const result = requireQueryResult( + await graph + .query() + .match('user:alice') + .outgoing('manages') + .select(['id']) + .run(), + ); + const ids = result.nodes.map((node) => node.id); expect(ids).toEqual(['user:bob']); }); it('incoming traversal follows reverse edges', async () => { - const result = await graph - .query() - .match('user:bob') - .incoming('manages') - .select(['id']) - .run(); - const ids = result.nodes.map((/** @type {any} */ n) => n.id); + const result = requireQueryResult( + await graph + .query() + .match('user:bob') + .incoming('manages') + .select(['id']) + .run(), + ); + const ids = result.nodes.map((node) => node.id); expect(ids).toEqual(['user:alice']); }); it('chained traversal works', async () => { - const result = await graph - .query() - .match('user:alice') - .outgoing('manages') - .outgoing('knows') - .select(['id']) - .run(); - const ids = result.nodes.map((/** @type {any} */ n) => n.id); + const result = requireQueryResult( + await graph + .query() + .match('user:alice') + .outgoing('manages') + .outgoing('knows') + .select(['id']) + .run(), + ); + const ids = result.nodes.map((node) => node.id); expect(ids).toEqual(['user:carol']); }); it('select with props returns properties', async () => { - const result = await graph - .query() - .match('user:alice') - .select(['id', 'props']) - .run(); + const result = requireQueryResult( + await graph + .query() + .match('user:alice') + .select(['id', 'props']) + .run(), + ); + const [alice] = result.nodes; expect(result.nodes).toHaveLength(1); - expect(result.nodes[0].id).toBe('user:alice'); - expect(result.nodes[0].props.role).toBe('engineering'); + expect(alice?.id).toBe('user:alice'); + expect(alice?.props?.['role']).toBe('engineering'); }); it('empty result set when no matches', async () => { - const result = await graph - .query() - .match('nonexistent:*') - .select(['id']) - .run(); + const result = requireQueryResult( + await graph + .query() + .match('nonexistent:*') + .select(['id']) + .run(), + ); expect(result.nodes).toHaveLength(0); }); }); diff --git a/test/integration/application/GitStorage.integration.test.ts b/test/integration/application/GitStorage.integration.test.ts index b42ac3d31..f304861f1 100644 --- a/test/integration/application/GitStorage.integration.test.ts +++ b/test/integration/application/GitStorage.integration.test.ts @@ -17,15 +17,19 @@ describe('GitStorage public composition', () => { it('opens a real repository and writes through the storage-neutral API', async () => { const storage = await GitStorage.open({ cwd: repository.tempDir }); - const warp = await openWarp({ storage, writer: 'agent-1' }); - const timeline = await warp.timeline('events'); + try { + const warp = await openWarp({ storage, writer: 'agent-1' }); + const timeline = await warp.timeline('events'); - const receipt = await timeline.write(intent.node.add({ subject: 'user:alice' })); + const receipt = await timeline.write(intent.node.add({ subject: 'user:alice' })); - expect(receipt.outcome).toBe('accepted'); - expect(receipt.timeline).toBe('events'); - expect(await repository.persistence.listRefs('refs/warp/events/writers/')).toEqual([ - 'refs/warp/events/writers/agent-1', - ]); + expect(receipt.outcome).toBe('accepted'); + expect(receipt.timeline).toBe('events'); + expect(await repository.persistence.listRefs('refs/warp/events/writers/')).toEqual([ + 'refs/warp/events/writers/agent-1', + ]); + } finally { + await storage.close(); + } }); }); diff --git a/test/integration/domain/orset/trie/TrieCursor.flush.integration.test.ts b/test/integration/domain/orset/trie/TrieCursor.flush.integration.test.ts index 7cbf161c7..96b4c8ff4 100644 --- a/test/integration/domain/orset/trie/TrieCursor.flush.integration.test.ts +++ b/test/integration/domain/orset/trie/TrieCursor.flush.integration.test.ts @@ -49,7 +49,11 @@ async function createHarness(): Promise { tempDir, adapter, async cleanup(): Promise { - await rm(tempDir, { recursive: true, force: true }); + try { + await cas.close(); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } }, }; } catch (err) { diff --git a/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts b/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts index f896588bc..9df70892b 100644 --- a/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts +++ b/test/integration/infrastructure/adapters/GitCasMaterializationStoreAdapter.integration.test.ts @@ -19,6 +19,9 @@ import NodeCryptoAdapter from '../../../../src/infrastructure/adapters/NodeCrypt import { DEFAULT_COMMIT_MESSAGE_CODEC } from '../../../../src/infrastructure/adapters/TrailerCommitMessageCodecAdapter.ts'; import defaultCodec from '../../../../src/infrastructure/codecs/CborCodec.ts'; import type MaterializationStorePort from '../../../../src/ports/MaterializationStorePort.ts'; +import type { + MaterializationAcquisition, +} from '../../../../src/ports/MaterializationStorePort.ts'; const execFileAsync = promisify(execFile); @@ -30,7 +33,19 @@ describe('GitCasMaterializationStoreAdapter integration', () => { }); afterEach(async () => { - await rm(harness.path, { recursive: true, force: true }); + try { + await harness.materializations.close(); + } finally { + try { + await harness.cas.close(); + } finally { + try { + await harness.history.close(); + } finally { + await rm(harness.path, { recursive: true, force: true }); + } + } + } }); it('retains the materialization graph and resumes from a fresh repository adapter', async () => { @@ -47,39 +62,56 @@ describe('GitCasMaterializationStoreAdapter integration', () => { }); const reopenedCas = createCas(harness.plumbing); - const reopened = await createMaterializations( + const reopenedFixture = await createMaterializations( harness.plumbing, reopenedCas, ); - const acquisition = await reopened.acquireExact(coordinate); - if (acquisition === null) { - throw new Error('Retained materialization was not reopened'); - } - const resolved = acquisition.materialization; - const nodeAliveRoot = resolved.roots.nodeAlive.handle; - if (nodeAliveRoot === null) { - throw new Error('Retained materialization did not expose its node root'); - } - const reopenedTrie = new GitCasTrieStoreAdapter({ cas: reopenedCas }); - const children = await reopenedTrie.readBranch(nodeAliveRoot.toString()); - const child = children.get(0); - if (child === undefined) { - throw new Error('Retained trie root did not contain its leaf child'); - } - const unreachable = await prunableOids(harness.path); - - expect(resolved.bundle.equals(retained.bundle)).toBe(true); - expect(resolved.roots.entries().map(([name, root]) => rootSignature(name, root))) - .toEqual(rootFixture.roots.entries().map(([name, root]) => rootSignature(name, root))); - expect(await reopenedTrie.readLeaf(child)).toEqual(trieFixture.bytes); - expect(unreachable).not.toContain(GitCasBundleHandle.parse(retained.bundle.toString()).oid); - for (const oid of rootFixture.retainedOids) { - expect(unreachable).not.toContain(oid); + const reopened = reopenedFixture.materializations; + let acquisition: MaterializationAcquisition | null = null; + try { + acquisition = await reopened.acquireExact(coordinate); + if (acquisition === null) { + throw new Error('Retained materialization was not reopened'); + } + const resolved = acquisition.materialization; + const nodeAliveRoot = resolved.roots.nodeAlive.handle; + if (nodeAliveRoot === null) { + throw new Error('Retained materialization did not expose its node root'); + } + const reopenedTrie = new GitCasTrieStoreAdapter({ cas: reopenedCas }); + const children = await reopenedTrie.readBranch(nodeAliveRoot.toString()); + const child = children.get(0); + if (child === undefined) { + throw new Error('Retained trie root did not contain its leaf child'); + } + const unreachable = await prunableOids(harness.path); + + expect(resolved.bundle.equals(retained.bundle)).toBe(true); + expect(resolved.roots.entries().map(([name, root]) => rootSignature(name, root))) + .toEqual(rootFixture.roots.entries().map(([name, root]) => rootSignature(name, root))); + expect(await reopenedTrie.readLeaf(child)).toEqual(trieFixture.bytes); + expect(unreachable).not.toContain(GitCasBundleHandle.parse(retained.bundle.toString()).oid); + for (const oid of rootFixture.retainedOids) { + expect(unreachable).not.toContain(oid); + } + expect(await harness.plumbing.execute({ + args: ['show-ref', '--verify', '--hash', 'refs/cas/caches/git-warp/materializations'], + })).toMatch(/^[0-9a-f]{40}\n?$/u); + } finally { + try { + await acquisition?.release(); + } finally { + try { + await reopened.close(); + } finally { + try { + await reopenedCas.close(); + } finally { + await reopenedFixture.history.close(); + } + } + } } - expect(await harness.plumbing.execute({ - args: ['show-ref', '--verify', '--hash', 'refs/cas/caches/git-warp/materializations'], - })).toMatch(/^[0-9a-f]{40}\n?$/u); - await acquisition.release(); }); it('keeps a replaced generation reachable until its runtime lease closes', async () => { @@ -192,7 +224,19 @@ describe('GitCasMaterializationStoreAdapter integration', () => { expect((await trie.readBranch(branchRoot)).size).toBe(1); await workspace.release(); } finally { - await rm(leaseHarness.path, { recursive: true, force: true }); + try { + await leaseHarness.materializations.close(); + } finally { + try { + await leaseHarness.cas.close(); + } finally { + try { + await leaseHarness.history.close(); + } finally { + await rm(leaseHarness.path, { recursive: true, force: true }); + } + } + } } }); }); @@ -200,6 +244,7 @@ describe('GitCasMaterializationStoreAdapter integration', () => { type Harness = Readonly<{ cas: ContentAddressableStore; materializations: MaterializationStorePort; + history: GitTimelineHistoryAdapter; path: string; plumbing: Awaited>; }>; @@ -211,11 +256,13 @@ async function createHarness(clock?: { readonly now: () => Date }): Promise>, cas: ContentAddressableStore, -): Promise { +): Promise> { const history = new GitTimelineHistoryAdapter({ plumbing }); const repository = new GitCasRepositoryAdapter({ plumbing, history, cas }); const services = await repository.createRuntimeStorageServices({ @@ -243,7 +293,7 @@ async function createMaterializations( crypto: new NodeCryptoAdapter(), commitMessageCodec: DEFAULT_COMMIT_MESSAGE_CODEC, }); - return services.materializations; + return Object.freeze({ history, materializations: services.materializations }); } type TrieRootFixture = Readonly<{ diff --git a/test/integration/infrastructure/adapters/GitCasTrieStoreAdapter.integration.test.ts b/test/integration/infrastructure/adapters/GitCasTrieStoreAdapter.integration.test.ts index 560592053..e19c8ef4d 100644 --- a/test/integration/infrastructure/adapters/GitCasTrieStoreAdapter.integration.test.ts +++ b/test/integration/infrastructure/adapters/GitCasTrieStoreAdapter.integration.test.ts @@ -17,7 +17,11 @@ describe('GitCasTrieStoreAdapter integration', () => { }); afterEach(async () => { - await rm(harness.path, { recursive: true, force: true }); + try { + await harness.cas.close(); + } finally { + await rm(harness.path, { recursive: true, force: true }); + } }); it('stores leaves as pages wrapped by bundle roots', async () => { diff --git a/test/integration/infrastructure/adapters/GitCasWarpStateCacheAdapter.retention.integration.test.ts b/test/integration/infrastructure/adapters/GitCasWarpStateCacheAdapter.retention.integration.test.ts index d88d799fa..1ab172e9d 100644 --- a/test/integration/infrastructure/adapters/GitCasWarpStateCacheAdapter.retention.integration.test.ts +++ b/test/integration/infrastructure/adapters/GitCasWarpStateCacheAdapter.retention.integration.test.ts @@ -108,7 +108,15 @@ async function createHarness(): Promise { plumbing, cache, async cleanup(): Promise { - await rm(tempDir, { recursive: true, force: true }); + try { + await runtimeStorage.close(); + } finally { + try { + await persistence.close(); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } + } }, }; } catch (error) { diff --git a/test/runtime/deno/helpers.ts b/test/runtime/deno/helpers.ts index db4e180c4..0fb2f314c 100644 --- a/test/runtime/deno/helpers.ts +++ b/test/runtime/deno/helpers.ts @@ -77,7 +77,21 @@ export async function createTestRepo(label = "deno-test") { crypto, openGraph, async cleanup() { - await rm(tempDir, { recursive: true, force: true }); + const failures: unknown[] = []; + for (const cleanup of [ + async (): Promise => await runtimeStorage.close(), + async (): Promise => await persistence.close(), + async (): Promise => await rm(tempDir, { recursive: true, force: true }), + ]) { + try { + await cleanup(); + } catch (error) { + failures.push(error); + } + } + if (failures.length > 0) { + throw new AggregateError(failures, "Deno test repository failed to clean up"); + } }, }; } diff --git a/test/type-check/v19-consumer.ts b/test/type-check/v19-consumer.ts index e5ccee35a..fdcaf1d30 100644 --- a/test/type-check/v19-consumer.ts +++ b/test/type-check/v19-consumer.ts @@ -139,3 +139,5 @@ void draftReceipt; void preview; void joinReceipt; void joinOutcome; + +await storage.close(); diff --git a/test/type-check/v19-first-use.ts b/test/type-check/v19-first-use.ts index 20957ba78..ec6937643 100644 --- a/test/type-check/v19-first-use.ts +++ b/test/type-check/v19-first-use.ts @@ -8,8 +8,9 @@ import { intent, openWarp, reading } from '../../index.ts'; import { GitStorage } from '../../storage.ts'; +const storage = await GitStorage.open({ cwd: '.' }); const warp = await openWarp({ - storage: await GitStorage.open({ cwd: '.' }), + storage, writer: 'agent-1', }); const timeline = await warp.timeline('events'); @@ -33,3 +34,5 @@ void role.value; void role.receipt; void userExists.value; void userExists.receipt; + +await storage.close(); diff --git a/test/type-check/v19-subpaths.ts b/test/type-check/v19-subpaths.ts index 63d2bc36b..fc8aa8087 100644 --- a/test/type-check/v19-subpaths.ts +++ b/test/type-check/v19-subpaths.ts @@ -31,7 +31,7 @@ const substrate: ReceiptSubstrateInspection = inspection.substrate; // @ts-expect-error diagnostics require explicit storage context. inspectReceipt(receipt); -void gitStorage; +await gitStorage.close(); void optic; void witness; void inspection; diff --git a/test/unit/application/WarpStorageRegistry.test.ts b/test/unit/application/WarpStorageRegistry.test.ts index a59800c5b..5148dd47d 100644 --- a/test/unit/application/WarpStorageRegistry.test.ts +++ b/test/unit/application/WarpStorageRegistry.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import { openWarp } from '../../../index.ts'; import MemoryStorage from '../../helpers/MemoryStorage.ts'; @@ -14,6 +14,12 @@ class UnsupportedStorage extends WarpStorage { } } +class ClosableStorage extends WarpStorage { + constructor(closeStorage: () => Promise) { + super(closeStorage); + } +} + describe('WarpStorageRegistry', () => { it('freezes supported storage handles and their internal binding', () => { const storage = MemoryStorage.create(); @@ -37,4 +43,26 @@ describe('WarpStorageRegistry', () => { openWarp({ storage: new UnsupportedStorage(), writer: 'agent-1' }) ).rejects.toMatchObject({ code: 'E_WARP_STORAGE_UNBOUND' }); }); + + it('releases local resources idempotently', async () => { + const closeStorage = vi.fn(() => Promise.resolve()); + const storage = new ClosableStorage(closeStorage); + + const first = storage.close(); + const second = storage.close(); + await Promise.all([first, second]); + + expect(first).toBe(second); + expect(closeStorage).toHaveBeenCalledTimes(1); + }); + + it('supports asynchronous disposal', async () => { + const closeStorage = vi.fn(() => Promise.resolve()); + const storage = new ClosableStorage(closeStorage); + + await storage[Symbol.asyncDispose](); + await storage.close(); + + expect(closeStorage).toHaveBeenCalledTimes(1); + }); }); diff --git a/test/unit/cli/commands/mcp-lifecycle.test.ts b/test/unit/cli/commands/mcp-lifecycle.test.ts new file mode 100644 index 000000000..57370e60f --- /dev/null +++ b/test/unit/cli/commands/mcp-lifecycle.test.ts @@ -0,0 +1,49 @@ +import { EventEmitter } from 'node:events'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +const mocks = vi.hoisted(() => ({ + createInterface: vi.fn(), + openGraph: vi.fn(), + readCliPackageVersion: vi.fn(), +})); + +vi.mock('node:readline', () => ({ + default: { createInterface: mocks.createInterface }, +})); + +vi.mock('../../../../bin/cli/shared.ts', () => ({ + openGraph: mocks.openGraph, + readCliPackageVersion: mocks.readCliPackageVersion, +})); + +const handleMcp = (await import('../../../../bin/cli/commands/mcp.ts')).default; + +describe('MCP command lifecycle', () => { + beforeEach(() => { + vi.clearAllMocks(); + mocks.openGraph.mockResolvedValue({ graph: {} }); + mocks.readCliPackageVersion.mockReturnValue('19.0.0-test'); + }); + + it('propagates readline errors through completion after closing input', async () => { + const lines = new EventEmitter() as EventEmitter & { + close: ReturnType; + }; + lines.close = vi.fn(() => lines.emit('close')); + mocks.createInterface.mockReturnValue(lines); + const inputFailure = new Error('stdin failed'); + + const result = await handleMcp({ + options: { repo: '.', graph: 'demo', writer: 'cli' } as Parameters< + typeof handleMcp + >[0]['options'], + args: [], + }); + lines.emit('error', inputFailure); + + await expect(result.completion).rejects.toMatchObject({ + errors: [inputFailure], + }); + expect(lines.close).toHaveBeenCalledOnce(); + }); +}); diff --git a/test/unit/cli/lifecycle.test.ts b/test/unit/cli/lifecycle.test.ts new file mode 100644 index 000000000..93af918f2 --- /dev/null +++ b/test/unit/cli/lifecycle.test.ts @@ -0,0 +1,41 @@ +import { describe, expect, it, vi } from 'vitest'; +import { closeCommandResources } from '../../../bin/cli/lifecycle.ts'; + +describe('CLI long-running command lifecycle', () => { + it('drains the command before closing its storage', async () => { + const commandClosed = Promise.withResolvers(); + const events: string[] = []; + const closeCommand = vi.fn(async () => { + events.push('command:start'); + await commandClosed.promise; + events.push('command:end'); + }); + const closeStorage = vi.fn(async () => { + events.push('storage'); + }); + + const closing = closeCommandResources(closeCommand, closeStorage); + await vi.waitFor(() => expect(closeCommand).toHaveBeenCalledTimes(1)); + expect(closeStorage).not.toHaveBeenCalled(); + commandClosed.resolve(); + await closing; + + expect(events).toEqual(['command:start', 'command:end', 'storage']); + }); + + it('attempts storage closure and preserves both shutdown failures', async () => { + const commandFailure = new Error('command close failed'); + const storageFailure = new Error('storage close failed'); + const closeStorage = vi.fn().mockRejectedValue(storageFailure); + + const closing = closeCommandResources( + vi.fn().mockRejectedValue(commandFailure), + closeStorage, + ); + + await expect(closing).rejects.toMatchObject({ + errors: [commandFailure, storageFailure], + }); + expect(closeStorage).toHaveBeenCalledTimes(1); + }); +}); diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.lifecycle.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.lifecycle.test.ts index d34556c96..c6bac9a7f 100644 --- a/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.lifecycle.test.ts +++ b/test/unit/infrastructure/adapters/GitCasMaterializationStoreAdapter.lifecycle.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import MaterializationCoordinate from '../../../../src/domain/materialization/MaterializationCoordinate.ts'; import MaterializationRoot from '../../../../src/domain/materialization/MaterializationRoot.ts'; import MaterializationRoots from '../../../../src/domain/materialization/MaterializationRoots.ts'; @@ -6,6 +6,9 @@ import BundleHandle from '../../../../src/domain/storage/BundleHandle.ts'; import GitCasMaterializationStoreAdapter, { type GitCasMaterializationFacade, } from '../../../../src/infrastructure/adapters/GitCasMaterializationStoreAdapter.ts'; +import type { + GitCasStagingWorkspace, +} from '../../../../src/infrastructure/adapters/GitCasMaterializationWorkspace.ts'; import { materializationCoordinateData } from '../../../../src/infrastructure/adapters/GitCasMaterializationDescriptor.ts'; import NodeCryptoAdapter from '../../../../src/infrastructure/adapters/NodeCryptoAdapter.ts'; import defaultCodec from '../../../../src/infrastructure/codecs/CborCodec.ts'; @@ -16,7 +19,113 @@ import InMemoryGraphAdapter from '../../../helpers/InMemoryGraphAdapter.ts'; const CACHE_NAMESPACE = 'git-warp/materializations'; const ROOT_COUNT = 8; -describe('GitCasMaterializationStoreAdapter legacy lifecycle', () => { +describe('GitCasMaterializationStoreAdapter lifecycle', () => { + it('rejects new materialization operations as soon as closure starts', async () => { + const harness = await createHarness(); + const coordinate = exactCoordinate(); + const roots = await createRoots(harness.cas); + + const closing = harness.adapter.close(); + + await expect(harness.adapter.openWorkspace(coordinate)).rejects.toMatchObject({ + code: 'E_MATERIALIZATION_STORAGE', + message: expect.stringContaining('adapter is closed'), + }); + await expect(harness.adapter.retain({ coordinate, roots, stateHash: 'closed' })) + .rejects.toMatchObject({ + code: 'E_MATERIALIZATION_STORAGE', + message: expect.stringContaining('adapter is closed'), + }); + await expect(harness.adapter.acquireExact(coordinate)).rejects.toMatchObject({ + code: 'E_MATERIALIZATION_STORAGE', + message: expect.stringContaining('adapter is closed'), + }); + await closing; + + expect(harness.cas.readActiveWorkspaceCount()).toBe(0); + expect(harness.cas.readActiveCacheAcquisitionCount()).toBe(0); + }); + + it('releases active workspaces when the adapter closes', async () => { + const harness = await createHarness(); + const workspace = await harness.adapter.openWorkspace(exactCoordinate()); + const release = vi.spyOn(workspace, 'release'); + + await harness.adapter.close(); + + expect(release).toHaveBeenCalledTimes(1); + expect(() => workspace.stagePage(new Uint8Array([1]), { maxBytes: 1 })) + .toThrow('closed workspace'); + }); + + it('waits for and releases a workspace whose open races with closure', async () => { + const harness = await createHarness(); + const staging = await harness.cas.workspaces.open({ namespace: 'pending-close' }); + const release = vi.fn(async () => await staging.release()); + const controlled: GitCasStagingWorkspace = { + pages: staging.pages, + bundles: staging.bundles, + checkpoint: async (options) => await staging.checkpoint(options), + promoteToCache: async (options) => await staging.promoteToCache(options), + release, + }; + const deferred = Promise.withResolvers(); + const adapter = adapterFor({ + bundles: harness.cas.bundles, + caches: harness.cas.caches, + pages: harness.cas.pages, + workspaces: { open: async () => await deferred.promise }, + }); + + const opening = adapter.openWorkspace(exactCoordinate()); + const opened = expect(opening).rejects.toMatchObject({ + code: 'E_MATERIALIZATION_STORAGE', + message: expect.stringContaining('adapter is closed'), + }); + const closing = adapter.close(); + deferred.resolve(controlled); + + await opened; + await closing; + expect(release).toHaveBeenCalledTimes(1); + }); + + it('preserves promotion and workspace release failures', async () => { + const harness = await createHarness(); + const promotionFailure = new Error('promotion failed'); + const releaseFailure = new Error('release failed'); + const staging = await harness.cas.workspaces.open({ namespace: 'failed-retain' }); + const controlled: GitCasStagingWorkspace = { + pages: { + put: async () => { + throw promotionFailure; + }, + }, + bundles: staging.bundles, + checkpoint: async (options) => await staging.checkpoint(options), + promoteToCache: async (options) => await staging.promoteToCache(options), + release: async () => { + await staging.release(); + throw releaseFailure; + }, + }; + const adapter = adapterFor({ + bundles: harness.cas.bundles, + caches: harness.cas.caches, + pages: harness.cas.pages, + workspaces: { open: async () => controlled }, + }); + const roots = await createRoots(harness.cas); + + await expect(adapter.retain({ + coordinate: exactCoordinate(), + roots, + stateHash: 'state-hash', + })).rejects.toMatchObject({ + errors: [promotionFailure, releaseFailure], + }); + }); + it('keeps the matching v2 cache anchor until the v3 profile is retained', async () => { const harness = await createHarness(); const coordinate = exactCoordinate(); diff --git a/test/unit/infrastructure/adapters/GitCasMaterializationWorkspaceOwner.test.ts b/test/unit/infrastructure/adapters/GitCasMaterializationWorkspaceOwner.test.ts new file mode 100644 index 000000000..938b268e6 --- /dev/null +++ b/test/unit/infrastructure/adapters/GitCasMaterializationWorkspaceOwner.test.ts @@ -0,0 +1,108 @@ +import { describe, expect, it, vi } from 'vitest'; +import GitCasMaterializationWorkspace, { + type GitCasStagingWorkspace, +} from '../../../../src/infrastructure/adapters/GitCasMaterializationWorkspace.ts'; +import GitCasMaterializationWorkspaceOwner from '../../../../src/infrastructure/adapters/GitCasMaterializationWorkspaceOwner.ts'; + +describe('GitCasMaterializationWorkspaceOwner', () => { + it('rejects workspace opens after closure', async () => { + const closed = new Error('owner closed'); + const owner = new GitCasMaterializationWorkspaceOwner(() => closed); + await owner.close(); + + await expect(owner.open(openRequest(vi.fn()))).rejects.toBe(closed); + }); + + it('preserves a release failure when an open races with closure', async () => { + const closed = new Error('owner closed'); + const releaseFailure = new Error('release failed'); + const owner = new GitCasMaterializationWorkspaceOwner(() => closed); + const staging = Promise.withResolvers(); + const opening = owner.open({ + open: async () => await staging.promise, + create: createWorkspace, + }); + const closing = owner.close(); + staging.resolve(stagingWorkspace(vi.fn().mockRejectedValue(releaseFailure))); + + const [openResult, closeResult] = await Promise.allSettled([opening, closing]); + + expect(openResult).toMatchObject({ + status: 'rejected', + reason: { errors: [closed, releaseFailure] }, + }); + expect(closeResult).toEqual({ status: 'rejected', reason: releaseFailure }); + }); + + it('aggregates failures while releasing multiple active workspaces', async () => { + const firstFailure = new Error('first release failed'); + const secondFailure = new Error('second release failed'); + const owner = new GitCasMaterializationWorkspaceOwner(() => new Error('owner closed')); + await owner.open(openRequest(vi.fn().mockRejectedValue(firstFailure))); + await owner.open(openRequest(vi.fn().mockRejectedValue(secondFailure))); + + await expect(owner.close()).rejects.toMatchObject({ + errors: [firstFailure, secondFailure], + }); + }); + + it('releases the staging workspace when wrapper creation fails', async () => { + const creationFailure = new Error('workspace creation failed'); + const release = vi.fn().mockResolvedValue(undefined); + const owner = new GitCasMaterializationWorkspaceOwner(() => new Error('owner closed')); + + await expect(owner.open({ + open: async () => stagingWorkspace(release), + create: () => { + throw creationFailure; + }, + })).rejects.toBe(creationFailure); + + expect(release).toHaveBeenCalledOnce(); + }); + + it('preserves creation and release failures when wrapper creation fails', async () => { + const creationFailure = new Error('workspace creation failed'); + const releaseFailure = new Error('release failed'); + const owner = new GitCasMaterializationWorkspaceOwner(() => new Error('owner closed')); + + await expect(owner.open({ + open: async () => stagingWorkspace(vi.fn().mockRejectedValue(releaseFailure)), + create: () => { + throw creationFailure; + }, + })).rejects.toMatchObject({ + errors: [creationFailure, releaseFailure], + }); + }); +}); + +function openRequest(release: () => Promise) { + return { + open: async (): Promise => stagingWorkspace(release), + create: createWorkspace, + }; +} + +function createWorkspace( + workspace: GitCasStagingWorkspace, + onRelease: () => void, +): GitCasMaterializationWorkspace { + return new GitCasMaterializationWorkspace({ + workspace, + promote: async () => { + throw new Error('promotion is outside this ownership test'); + }, + onRelease, + }); +} + +function stagingWorkspace(release: () => Promise): GitCasStagingWorkspace { + return { + pages: { put: vi.fn() }, + bundles: { putOrdered: vi.fn() }, + checkpoint: vi.fn(), + promoteToCache: vi.fn(), + release, + } as unknown as GitCasStagingWorkspace; +} diff --git a/test/unit/infrastructure/adapters/GitCasRepositoryAdapter.test.ts b/test/unit/infrastructure/adapters/GitCasRepositoryAdapter.test.ts index 8f025c0bf..d3fb401ea 100644 --- a/test/unit/infrastructure/adapters/GitCasRepositoryAdapter.test.ts +++ b/test/unit/infrastructure/adapters/GitCasRepositoryAdapter.test.ts @@ -116,6 +116,7 @@ describe('GitCasRepositoryAdapter', () => { const putAsset = vi.fn(highLevelCas.assets.put); const rootSet = createRootSet(); const store = vi.fn().mockResolvedValue({ slug: 'manifest', chunks: [] }); + const closeCas = vi.fn().mockResolvedValue(undefined); const createTree = vi.fn() .mockResolvedValueOnce('1'.repeat(40)) .mockResolvedValueOnce('2'.repeat(40)) @@ -138,6 +139,7 @@ describe('GitCasRepositoryAdapter', () => { restoreStream: vi.fn(), store, createTree, + close: closeCas, }; const repository = new GitCasRepositoryAdapter({ plumbing, history, cas }); const services = await repository.createRuntimeStorageServices({ @@ -190,6 +192,30 @@ describe('GitCasRepositoryAdapter', () => { expect(putAsset).toHaveBeenCalledWith(expect.objectContaining({ slug: 'trust-record-hash' })); expect(store).toHaveBeenCalledTimes(1); expect(store).toHaveBeenCalledWith(expect.objectContaining({ slug: 'snapshot-1' })); + + const activeServices = await repository.createRuntimeStorageServices({ + timelineName: 'active-at-close', + codec: defaultCodec, + crypto: new TestCrypto(), + commitMessageCodec: DEFAULT_COMMIT_MESSAGE_CODEC, + }); + const closeMaterializations = vi.spyOn(services.materializations, 'close'); + const closeActiveMaterializations = vi.spyOn(activeServices.materializations, 'close'); + await services.materializations.close(); + await repository[Symbol.asyncDispose](); + await repository.close(); + + expect(closeMaterializations).toHaveBeenCalledTimes(1); + expect(closeActiveMaterializations).toHaveBeenCalledTimes(1); + expect(closeCas).not.toHaveBeenCalled(); + expect(() => repository.createRuntimeStorageServices({ + timelineName: 'closed', + codec: defaultCodec, + crypto: new TestCrypto(), + commitMessageCodec: DEFAULT_COMMIT_MESSAGE_CODEC, + })).toThrow('Git CAS repository storage is closed'); + expect(() => repository.createTrustChain(new TestCrypto())) + .toThrow('Git CAS repository storage is closed'); }); }); diff --git a/test/unit/infrastructure/adapters/GitGraphAdapter.gitCasPersistence.test.ts b/test/unit/infrastructure/adapters/GitGraphAdapter.gitCasPersistence.test.ts index 340d32a43..e01ff7457 100644 --- a/test/unit/infrastructure/adapters/GitGraphAdapter.gitCasPersistence.test.ts +++ b/test/unit/infrastructure/adapters/GitGraphAdapter.gitCasPersistence.test.ts @@ -244,4 +244,25 @@ describe('GitTimelineHistoryAdapter git-cas persistence bridge', () => { await expect(adapter.readBlob(oid)).resolves.toEqual(payload); expect(plumbing.streamCalls).toEqual([{ args: ['cat-file', 'blob', oid] }]); }); + + it('closes the delegated git-cas persistence adapter idempotently', async () => { + const adapter = new GitTimelineHistoryAdapter({ + plumbing: new RecordingPlumbing('f'.repeat(40)), + }); + + const first = adapter.close(); + const second = adapter.close(); + + expect(first).toBe(second); + await expect(first).resolves.toBeUndefined(); + }); + + it('supports asynchronous disposal of delegated git-cas persistence', async () => { + const adapter = new GitTimelineHistoryAdapter({ + plumbing: new RecordingPlumbing('f'.repeat(40)), + }); + + await adapter[Symbol.asyncDispose](); + await expect(adapter.close()).resolves.toBeUndefined(); + }); }); diff --git a/test/unit/infrastructure/adapters/OperationCleanup.test.ts b/test/unit/infrastructure/adapters/OperationCleanup.test.ts new file mode 100644 index 000000000..25033f3c2 --- /dev/null +++ b/test/unit/infrastructure/adapters/OperationCleanup.test.ts @@ -0,0 +1,58 @@ +import { describe, expect, it, vi } from 'vitest'; +import { completeWithCleanup } from '../../../../src/infrastructure/adapters/OperationCleanup.ts'; + +describe('completeWithCleanup', () => { + it('returns the operation value after cleanup', async () => { + const events: string[] = []; + + const result = await completeWithCleanup( + async () => { + events.push('operation'); + return 'value'; + }, + async () => { + events.push('cleanup'); + }, + 'both failed', + ); + + expect(result).toBe('value'); + expect(events).toEqual(['operation', 'cleanup']); + }); + + it('preserves an operation failure after cleanup succeeds', async () => { + const operationFailure = new Error('operation failed'); + const cleanup = vi.fn().mockResolvedValue(undefined); + + await expect(completeWithCleanup( + vi.fn().mockRejectedValue(operationFailure), + cleanup, + 'both failed', + )).rejects.toBe(operationFailure); + expect(cleanup).toHaveBeenCalledOnce(); + }); + + it('reports a cleanup failure after the operation succeeds', async () => { + const cleanupFailure = new Error('cleanup failed'); + + await expect(completeWithCleanup( + vi.fn().mockResolvedValue('value'), + vi.fn().mockRejectedValue(cleanupFailure), + 'both failed', + )).rejects.toBe(cleanupFailure); + }); + + it('aggregates operation and cleanup failures', async () => { + const operationFailure = new Error('operation failed'); + const cleanupFailure = new Error('cleanup failed'); + + await expect(completeWithCleanup( + vi.fn().mockRejectedValue(operationFailure), + vi.fn().mockRejectedValue(cleanupFailure), + 'both failed', + )).rejects.toMatchObject({ + message: 'both failed', + errors: [operationFailure, cleanupFailure], + }); + }); +}); diff --git a/test/unit/scripts/v18-v17-public-read-legacy-reading-builder.test.ts b/test/unit/scripts/v18-v17-public-read-legacy-reading-builder.test.ts index bf26ddc18..9a17a3bc3 100644 --- a/test/unit/scripts/v18-v17-public-read-legacy-reading-builder.test.ts +++ b/test/unit/scripts/v18-v17-public-read-legacy-reading-builder.test.ts @@ -46,10 +46,10 @@ describe('v18 v17 public-read legacy reading builder', () => { 'alice', 'alice', ]); - // git-cas package versions are stamped into asset manifests, so dependency bumps - // intentionally advance this migration-reading golden handle. + // git-cas stamps its package version into the manifest formatVersion, so dependency + // bumps intentionally advance this migration-reading golden handle. expect(reading.facts.find((fact) => fact.factKey === 'node:alpha:_content')?.value) - .toBe('git-cas:1:asset:manifest-tree:cbor:sha1:8851a151d100f2faad3a93892dfb83fa6f1345f7'); + .toBe('git-cas:1:asset:manifest-tree:cbor:sha1:88b6deeb1a9e0e364f80d244731170f29fbe09a4'); }); it('fails closed when a restored v17 writer ref drifts after restore', async () => {