diff --git a/src/sqlite/SqliteDatabaseProvider.ts b/src/sqlite/SqliteDatabaseProvider.ts index a4ed9aeb1b..43a14636a7 100644 --- a/src/sqlite/SqliteDatabaseProvider.ts +++ b/src/sqlite/SqliteDatabaseProvider.ts @@ -48,6 +48,7 @@ import { SQLITE_DATABASE_LIMITS } from "./SqliteDatabaseLimits.js"; import { executeSqliteDatabaseWorker, + sqliteReplyFinalizationFailure, type SqliteWorkerLauncher, } from "./SqliteDatabaseWorkerExecution.js"; @@ -283,7 +284,7 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort { options?.signal, ); phase = "worker"; - const inspection = await executeSqliteDatabaseWorker({ + const completed = await executeSqliteDatabaseWorker({ input, root: root.path, snapshotPath: snapshot.snapshotPath, @@ -303,18 +304,29 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort { const validated = sqliteDatabaseSchema.safeParse({ artifact: snapshot.artifact, wal: snapshot.wal, - ...inspection, + ...completed.inspection, worker_timeout_ms: workerTimeoutMs, }); if (!validated.success) - throw new AnalysisOutputError( - operation, - "SQLite worker report dimensions or identity fields are invalid", - retainedOutput === undefined - ? undefined - : { capturedOutput: retainedOutput }, + throw sqliteReplyFinalizationFailure( + new AnalysisOutputError( + operation, + "SQLite worker report dimensions or identity fields are invalid", + retainedOutput === undefined + ? undefined + : { capturedOutput: retainedOutput }, + ), + completed.finalization, ); - result = ok(createSqliteObservation(validated.data, input.path)); + const execution = createSqliteObservation(validated.data, input.path); + result = + completed.finalization === undefined + ? ok(execution) + : err( + completed.finalization.retainPartialObservation( + completedSqliteEvidence(input, execution), + ), + ); } catch (cause: unknown) { // A failure rethrown after retainCleanup carries no owner of its own; // keep the retained worker instead of discarding it. diff --git a/src/sqlite/SqliteDatabaseWorkerExecution.ts b/src/sqlite/SqliteDatabaseWorkerExecution.ts index 48ea8593a0..092f109503 100644 --- a/src/sqlite/SqliteDatabaseWorkerExecution.ts +++ b/src/sqlite/SqliteDatabaseWorkerExecution.ts @@ -7,7 +7,12 @@ import { writeFile } from "node:fs/promises"; import { join } from "node:path"; import { fileURLToPath } from "node:url"; import { z } from "zod"; -import type { AnalysisCapturedOutput } from "../domain/analysisErrorBase.js"; +import type { + AnalysisCapturedOutput, + AnalysisError, +} from "../domain/analysisErrorBase.js"; +import { analysisErrorWithCleanupFailure } from "../domain/analysisErrorCleanup.js"; +import { ProviderCleanupError } from "../domain/providerCleanupError.js"; import { AnalysisCancelledError, AnalysisInputError, @@ -19,7 +24,10 @@ import { sqliteDatabaseSchema, type InspectSqliteDatabaseInput, } from "../domain/sqlite/sqliteDatabase.js"; -import { readStableArtifact } from "../artifacts/readStableArtifact.js"; +import { + completedStableArtifact, + readStableArtifact, +} from "../artifacts/readStableArtifact.js"; import { ArtifactReaderFailure } from "../artifacts/ArtifactReader.js"; import type { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js"; import { runOwnedCommand } from "../process/OwnedCommand.js"; @@ -73,6 +81,12 @@ export type SqliteWorkerLauncher = NonNullable< Parameters[2] >["launcher"]; +/** Validated worker facts and the exact reply descriptor's finalization outcome. */ +interface SqliteWorkerInspection { + readonly inspection: z.output; + readonly finalization: AnalysisError | undefined; +} + // Each input rejection names the caller argument that requires correction. const rejectedArgument = { format: "path", @@ -195,7 +209,7 @@ export const executeSqliteDatabaseWorker = async (request: { owner: ProviderProcessSupervisor, result: Extract, ) => void; -}): Promise> => { +}): Promise => { const requestPath = join(request.root, "request.json"); const replyPath = join(request.root, "reply.json"); await writeFile( @@ -237,42 +251,102 @@ export const executeSqliteDatabaseWorker = async (request: { const capturedOutput = capturedSqliteOutput(execution); request.captured(capturedOutput); let reply: z.output; + let finalization: AnalysisError | undefined; try { - const file = await readStableArtifact( - replyPath, - SQLITE_DATABASE_LIMITS.outputBytes, - request.signal, - undefined, - request.resourceScope, - ); + let file; + try { + file = await readStableArtifact( + replyPath, + SQLITE_DATABASE_LIMITS.outputBytes, + request.signal, + undefined, + request.resourceScope, + ); + } catch (cause: unknown) { + const completed = completedStableArtifact(cause); + if (!(cause instanceof ArtifactReaderFailure) || completed === undefined) + throw cause; + finalization = sqliteReplyCompletionFailure(cause, capturedOutput); + file = completed; + } reply = replySchema.parse(JSON.parse(file.bytes.toString("utf8"))); } catch (cause: unknown) { - const options = { - cause, - capturedOutput, - ...(cause instanceof ArtifactReaderFailure && cause.cleanup !== undefined - ? { cleanup: cause.cleanup } - : {}), - }; - if ( - cause instanceof ArtifactReaderFailure && - cause.reason === "cancelled" - ) { - const error = new AnalysisCancelledError(operation, { - ...options, - cause: artifactFileCancellationCause(cause), - }); - bindArtifactFileCleanup(cause, error); - throw error; - } - const error = new AnalysisOutputError( - operation, - `Owned SQLite worker reply is missing or malformed: ${cause instanceof Error ? cause.message : String(cause)}`, - options, + throw sqliteReplyFinalizationFailure( + sqliteReplyReadFailure(cause, capturedOutput), + finalization, ); - bindArtifactFileCleanup(cause, error); - throw error; } - if (!reply.ok) throw rejectedReply(reply, request.input.path, capturedOutput); - return reply.inspection; + if (!reply.ok) + throw sqliteReplyFinalizationFailure( + rejectedReply(reply, request.input.path, capturedOutput), + finalization, + ); + return { inspection: reply.inspection, finalization }; }; + +const sqliteReplyReadFailure = ( + cause: unknown, + capturedOutput: AnalysisCapturedOutput, +): AnalysisError => { + const options = { + cause, + capturedOutput, + ...(cause instanceof ArtifactReaderFailure && cause.cleanup !== undefined + ? { cleanup: cause.cleanup } + : {}), + }; + const error = + cause instanceof ArtifactReaderFailure && cause.reason === "cancelled" + ? new AnalysisCancelledError(operation, { + ...options, + cause: artifactFileCancellationCause(cause), + }) + : new AnalysisOutputError( + operation, + `Owned SQLite worker reply is missing or malformed: ${cause instanceof Error ? cause.message : String(cause)}`, + options, + ); + bindArtifactFileCleanup(cause, error); + return error; +}; + +/** Stable bytes remain usable when only their descriptor's final close failed. */ +const sqliteReplyCompletionFailure = ( + cause: ArtifactReaderFailure, + capturedOutput: AnalysisCapturedOutput, +): AnalysisError => { + if (cause.cleanup === undefined) + return sqliteReplyReadFailure(cause, capturedOutput); + const cleanup = new ProviderCleanupError( + "sqlite", + cause.cleanup.resources, + { phase: "reply_descriptor_close", reason: cause.cleanup.reason }, + { operation, cause, capturedOutput }, + ); + bindArtifactFileCleanup(cause, cleanup); + if (cause.reason !== "cancelled") return cleanup; + const cancellation = new AnalysisCancelledError(operation, { + cause: artifactFileCancellationCause(cause), + capturedOutput, + }); + return analysisErrorWithCleanupFailure( + cancellation, + cleanup, + operation, + ).bindCleanupDependencies([cancellation, cleanup]); +}; + +/** A malformed reply or producer refusal stays primary over its later close. */ +export const sqliteReplyFinalizationFailure = ( + primary: AnalysisError, + finalization: AnalysisError | undefined, +): AnalysisError => + finalization === undefined + ? primary + : finalization.cleanupIncomplete + ? analysisErrorWithCleanupFailure( + primary, + finalization, + operation, + ).bindCleanupDependencies([primary, finalization]) + : primary.retainSecondaryFailure(finalization); diff --git a/tests/boundary/sqlite/sqliteDatabaseDescriptorCleanup.test.ts b/tests/boundary/sqlite/sqliteDatabaseDescriptorCleanup.test.ts index 8b1c824240..a6a0c4c536 100644 --- a/tests/boundary/sqlite/sqliteDatabaseDescriptorCleanup.test.ts +++ b/tests/boundary/sqlite/sqliteDatabaseDescriptorCleanup.test.ts @@ -1,5 +1,6 @@ import { Cli } from "incur"; -import { access, writeFile } from "node:fs/promises"; +import { access, readFile, writeFile } from "node:fs/promises"; +import { createHash } from "node:crypto"; import type { FileHandle } from "node:fs/promises"; import { dirname, join } from "node:path"; import { fileURLToPath } from "node:url"; @@ -11,7 +12,9 @@ import { ArtifactResourceScope } from "../../../src/artifacts/ArtifactResourceSc import { registerSqliteCommands } from "../../../src/cli/sqliteCommands.js"; import { createSqliteDatabaseService } from "../../../src/composition/sqlite.js"; import { projectAnalysisError } from "../../../src/domain/analysisErrorProjection.js"; -import { AnalysisOutputError } from "../../../src/domain/analysisErrorCore.js"; +import { ProviderCleanupError } from "../../../src/domain/providerCleanupError.js"; +import { primaryAnalysisError } from "../../../src/domain/analysisErrorBase.js"; +import { parseEvidence } from "../../../src/domain/evidence.js"; import { silentLogger } from "../../../src/logger.js"; import { SqliteDatabaseProvider } from "../../../src/sqlite/SqliteDatabaseProvider.js"; import { PrivateRuntimeRoot } from "../../../src/process/PrivateRuntimeRoot.js"; @@ -30,6 +33,8 @@ const descriptorCleanup = vi.hoisted(() => ({ handles: [] as FileHandle[], readFailure: undefined as Error | undefined, onOpen: undefined as (() => void) | undefined, + onClose: undefined as (() => void) | undefined, + reply: undefined as string | undefined, })); vi.mock("../../../src/composition/sqlite.js", () => ({ @@ -54,9 +59,11 @@ vi.mock("node:fs/promises", async (importOriginal) => { descriptorCleanup.attempts += 1; if (descriptorCleanup.denied) { if (descriptorCleanup.invalidate) await close(); + descriptorCleanup.onClose?.(); throw new Error("injected SQLite descriptor close failure"); } await close(); + descriptorCleanup.onClose?.(); }; if (descriptorCleanup.readFailure !== undefined) vi.spyOn(handle, "stat").mockRejectedValue( @@ -90,6 +97,15 @@ const fixture = async ( (spawn) => { if (member !== "reply") throw new Error("SQLite worker deliberately unavailable"); + if (descriptorCleanup.reply !== undefined) + return spawnOwnedProviderProcess({ + ...spawn, + arguments: [ + "-e", + `const fs=require('node:fs');const request=JSON.parse(fs.readFileSync(process.argv[1],'utf8'));fs.writeFileSync(request.reply_path,${JSON.stringify(descriptorCleanup.reply)});`, + spawn.arguments[2] ?? "", + ], + }); return spawnOwnedProviderProcess({ ...spawn, arguments: spawn.arguments.map((argument, index) => @@ -119,6 +135,8 @@ const fixture = async ( descriptorCleanup.handles = []; descriptorCleanup.readFailure = undefined; descriptorCleanup.onOpen = undefined; + descriptorCleanup.onClose = undefined; + descriptorCleanup.reply = undefined; onTestFinished(async () => { descriptorCleanup.denied = false; try { @@ -128,6 +146,8 @@ const fixture = async ( } for (const handle of descriptorCleanup.handles) await handle.close(); descriptorCleanup.path = ""; + descriptorCleanup.onClose = undefined; + descriptorCleanup.reply = undefined; database.close(); await removeTestWorkspace(workspace.root); }); @@ -279,6 +299,7 @@ it("retains reply-read cancellation and retries the failed reply descriptor clos if (result.ok) throw new Error("Expected cancelled reply read"); expect(result.error).toMatchObject({ _tag: "AnalysisCancelledError" }); expect(result.error.cause).toBe(reason); + expect(result.error.partialObservation).toBeUndefined(); const cancelled = { code: "cleanup_incomplete", details: { @@ -330,6 +351,135 @@ it("retains reply-read cancellation and retries the failed reply descriptor clos await expect(access(root.path)).rejects.toMatchObject({ code: "ENOENT" }); }); +it.each([ + { cancel: false, denied: false }, + { cancel: true, denied: false }, + { cancel: false, denied: true }, + { cancel: true, denied: true }, +])( + "retains completed SQLite Evidence through reply close (cancel=$cancel, denied=$denied)", + async ({ cancel, denied }) => { + const selected = await fixture("reply"); + descriptorCleanup.denied = denied; + const controller = new AbortController(); + const reason = { requested: "cancel after complete SQLite reply read" }; + if (cancel) descriptorCleanup.onClose = () => controller.abort(reason); + const input = { + path: selected.database.path, + table: "records", + columns: ["id"], + row_limit: 1, + worker_timeout_ms: 4321, + }; + const databaseBytes = await readFile(input.path); + const walBytes = await readFile(selected.database.walPath); + const databaseDigest = createHash("sha256") + .update(databaseBytes) + .digest("hex"); + const walDigest = createHash("sha256").update(walBytes).digest("hex"); + const result = await new SqliteDatabaseService(selected.provider).inspect( + input, + { signal: controller.signal }, + ); + expect(result.ok).toBe(!cancel && !denied); + const evidence = parseEvidence( + result.ok ? result.value : result.error.partialObservation, + ); + expect(evidence).toMatchObject({ + operation: "inspect_sqlite_database", + parameters: input, + subject: { + local_path: input.path, + format: "file", + digest: { sha256: databaseDigest }, + }, + normalized_result: { + artifact: { + path: input.path, + sha256: databaseDigest, + bytes: databaseBytes.length, + }, + wal: { + path: selected.database.walPath, + sha256: walDigest, + bytes: walBytes.length, + }, + worker_timeout_ms: input.worker_timeout_ms, + schema: { scope: "selected_table", completeness: "complete" }, + rows: { + table: "records", + columns: ["id"], + values: [[{ type: "integer", value: "1" }]], + returned_rows: 1, + truncated: true, + }, + }, + }); + if (!result.ok) { + if (cancel) { + expect(primaryAnalysisError(result.error)._tag).toBe( + "AnalysisCancelledError", + ); + expect(primaryAnalysisError(result.error).cause).toBe(reason); + } + expect(projectAnalysisError(result.error)).toMatchObject({ + details: { partial_observation: evidence }, + }); + expect(result.error.cleanupIncomplete).toBe(denied); + if (denied) { + const handle = descriptorCleanup.handles[0]; + expect(handle?.fd).toBeGreaterThanOrEqual(0); + await expect(selected.provider.close()).rejects.toMatchObject({ + cleanupIncomplete: true, + }); + descriptorCleanup.denied = false; + await selected.provider.close(); + expect(result.error.cleanupIncomplete).toBe(false); + expect(result.error.cleanupResources).toEqual([]); + expect(result.error.partialObservation).toEqual(evidence); + } + } + await selected.provider.close(); + expect(descriptorCleanup.handles.every((handle) => handle.fd === -1)).toBe( + true, + ); + for (const root of selected.roots) + await expect(access(root.path)).rejects.toMatchObject({ code: "ENOENT" }); + }, +); + +it.each([ + { reply: "{ malformed reply", tag: "AnalysisOutputError" }, + { + reply: JSON.stringify({ + ok: false, + reason: "selection", + message: "Selected table does not exist", + }), + tag: "AnalysisInputError", + }, +])( + "preserves a completed $tag reply alongside denied close without inventing SQLite Evidence", + async ({ reply, tag }) => { + const selected = await fixture("reply"); + descriptorCleanup.reply = reply; + const result = await selected.provider.inspect({ + path: selected.database.path, + }); + if (result.ok) throw new Error("Expected rejected worker reply"); + expect(primaryAnalysisError(result.error)._tag).toBe(tag); + expect(result.error.partialObservation).toBeUndefined(); + expect(result.error.cleanupIncomplete).toBe(true); + descriptorCleanup.denied = false; + await selected.provider.close(); + expect(result.error.cleanupIncomplete).toBe(false); + expect(primaryAnalysisError(result.error)._tag).toBe(tag); + expect(descriptorCleanup.handles.every((handle) => handle.fd === -1)).toBe( + true, + ); + }, +); + it("keeps a completed reply close failure primary when abort follows lease retention", async () => { const selected = await fixture("reply"); const controller = new AbortController(); @@ -357,12 +507,9 @@ it("keeps a completed reply close failure primary when abort follows lease reten expect(result.ok).toBe(false); if (result.ok) return; - const primary = - result.error instanceof AnalysisOutputError - ? result.error - : result.error.cleanupPrimary; - expect(primary).toBeInstanceOf(AnalysisOutputError); - if (!(primary instanceof AnalysisOutputError)) return; + const primary = primaryAnalysisError(result.error); + expect(primary).toBeInstanceOf(ProviderCleanupError); + if (!(primary instanceof ProviderCleanupError)) return; expect(primary).not.toMatchObject({ _tag: "AnalysisCancelledError" }); expect(primary.cause).toBeInstanceOf(ArtifactFileCleanupFailure); expect(primary.cause).toMatchObject({ @@ -386,7 +533,7 @@ it("keeps a completed reply close failure primary when abort follows lease reten expect(descriptorCleanup.attempts).toBe(2); expect(handle.fd).toBe(-1); expect(primary.cleanupIncomplete).toBe(false); - expect(primary).toBeInstanceOf(AnalysisOutputError); + expect(primary).toBeInstanceOf(ProviderCleanupError); } finally { retainSpy.mockRestore(); }