mirror/rea
1
0
Fork 0
mirror of https://github.com/morluto/rea synced 2026-10-26 14:25:49 -04:00
Code Issues Packages Projects Releases Wiki Activity

fix(sqlite): retain completed evidence through reply finalization

Browse Source
This commit is contained in:
morluto
2026-10-26 17:56:47 +08:00
parent f2b4468183
commit e596a48251
3 changed files with 287 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/sqlite/SqliteDatabaseProvider.ts
+21 -9
View File
@@ -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.
src/sqlite/SqliteDatabaseWorkerExecution.ts
+110 -36
View File
@@ -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<typeof runOwnedCommand>[2]
>["launcher"];
/** Validated worker facts and the exact reply descriptor's finalization outcome. */
interface SqliteWorkerInspection {
readonly inspection: z.output<typeof inspectionSchema>;
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<ProviderProcessStopResult, { status: "incomplete" }>,
) => void;
}): Promise<z.output<typeof inspectionSchema>> => {
}): Promise<SqliteWorkerInspection> => {
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<typeof replySchema>;
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);
tests/boundary/sqlite/sqliteDatabaseDescriptorCleanup.test.ts
+156 -9
View File
@@ -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();
}