fix(sqlite): retain primary outcomes through cleanup retries

This commit is contained in:
morluto committed 2026-10-20 17:17:15 +08:00
1 parent e367750b1b
commit 1006714271
7 files changed
+739 -127

No files matched your search

+30 -11
View File
@@ -59,14 +59,20 @@ export const sqliteDatabaseFailure = (
cause instanceof ArtifactFileCleanupFailure
? cause.fileCleanupFailure.outcome
: undefined;
const previous =
physicalRead?.kind === "failed"
? sqliteDatabaseFailure(physicalRead.cause, path, phase)
: signal?.aborted
? new AnalysisCancelledError(operation, { cause: signal.reason })
: physicalRead?.kind === "completed" || cause.cause === undefined
? undefined
: sqliteDatabaseFailure(cause.cause, path, phase);
let previous: AnalysisError | undefined;
if (physicalRead?.kind === "failed")
previous = sqliteDatabaseFailure(physicalRead.cause, path, phase);
else if (cause.reason === "cancelled")
previous = new AnalysisCancelledError(operation, {
cause: artifactFileCancellationCause(cause),
});
else if (physicalRead?.kind === "completed") previous = undefined;
else if (cause.cause !== undefined)
previous = sqliteDatabaseFailure(cause.cause, path, phase);
else if (signal?.aborted)
previous = new AnalysisCancelledError(operation, {
cause: signal.reason,
});
const error = new ProviderCleanupError(
SQLITE_PROVIDER_IDENTITY.id,
cause.cleanup.resources,
@@ -75,7 +81,11 @@ export const sqliteDatabaseFailure = (
previous_error:
previous === undefined ? null : projectAnalysisError(previous),
},
{ operation, cause },
{
operation,
cause,
...(previous === undefined ? {} : { cleanupPrimary: previous }),
},
);
bindArtifactFileCleanup(cause, error);
return error;
@@ -99,7 +109,12 @@ export const sqliteDatabaseFailure = (
]);
}
if (cause instanceof Error && cause.name === "AbortError")
return new AnalysisCancelledError(operation, { cause });
return new AnalysisCancelledError(operation, {
cause:
signal?.aborted === true && cause.cause === signal.reason
? signal.reason
: cause,
});
if (cause instanceof OwnedCommandFailure) {
const projected = ownedCommandFailure(cause);
if (projected !== undefined) return projected;
@@ -183,7 +198,11 @@ const ownedCommandFailure = (
{ operation, cause },
);
if (cause.reason === "cancelled")
return new AnalysisCancelledError(operation, options);
return new AnalysisCancelledError(operation, {
...options,
cause:
cause.cancellation === undefined ? cause : cause.cancellation.reason,
});
if (cause.reason === "timeout")
return new AnalysisTimeoutError(
operation,
+155 -90
View File
@@ -6,6 +6,7 @@ import {
import type { SqliteDatabasePort } from "../application/sqlite/SqliteDatabasePort.js";
import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js";
import { ArtifactReaderFailure } from "../artifacts/ArtifactReader.js";
import { bindArtifactFileCleanup } from "../artifacts/ArtifactFileOperation.js";
import type {
AnalysisCapturedOutput,
AnalysisError,
@@ -54,6 +55,8 @@ interface PendingSnapshotCleanup {
readonly input: InspectSqliteDatabaseInput;
readonly result: SqliteOutcome;
readonly output: AnalysisCapturedOutput | undefined;
readonly rootFailure: AnalysisError;
readonly processFailure: AnalysisError | undefined;
process: ProviderProcessSupervisor | undefined;
}
/** Inspect a provider-owned copy through a cancellable SQLite child, without source writes. */
@@ -80,8 +83,13 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort {
input: InspectSqliteDatabaseInput,
options?: ExecutionOptions,
): Promise<SqliteOutcome> {
if (this.#closed || options?.signal?.aborted)
return err(new AnalysisCancelledError(operation));
if (options?.signal?.aborted)
return err(
new AnalysisCancelledError(operation, {
cause: options.signal.reason,
}),
);
if (this.#closed) return err(new AnalysisCancelledError(operation));
if (this.#active.size >= SQLITE_DATABASE_LIMITS.concurrentInspections)
return err(
new AnalysisResourceConstraintError(
@@ -104,13 +112,10 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort {
.run(() => this.#inspect(input, options))
.catch((cause: unknown) =>
err(
selectedFailure(
sqliteDatabaseFailure(
cause,
input.path,
"artifact-read",
options?.signal,
),
sqliteDatabaseFailure(
cause,
input.path,
"artifact-read",
options?.signal,
),
),
@@ -142,10 +147,11 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort {
{ reason: cleanup.reason },
{ operation, cause },
);
bindArtifactFileCleanup(cause, descriptorFailure);
}
const cleaned = await this.#retryCleanup();
if (descriptorFailure !== undefined && !cleaned.ok)
throw new ProviderCleanupError(
if (descriptorFailure !== undefined && !cleaned.ok) {
const failure = new ProviderCleanupError(
SQLITE_PROVIDER_IDENTITY.id,
[
...descriptorFailure.cleanupResources,
@@ -159,6 +165,9 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort {
},
{ operation },
);
failure.bindCleanupDependencies([descriptorFailure, cleaned.error]);
throw failure;
}
if (descriptorFailure !== undefined) throw descriptorFailure;
if (!cleaned.ok) throw cleaned.error;
})
@@ -181,30 +190,32 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort {
for (const [root, pending] of [...this.#pendingCleanup]) {
if (pending.process !== undefined) {
const stopped = await pending.process.stop();
if (stopped.status === "incomplete") {
failures.push(
new ProviderCleanupError(
SQLITE_PROVIDER_IDENTITY.id,
[
pending.process.launch.ownership?.runId ??
"owned-sqlite-worker",
root.path,
],
{
reason: stopped.reason,
previous_error: pending.result.ok
? null
: projectAnalysisError(pending.result.error),
...(pending.output === undefined
? {}
: { captured_output: { ...pending.output } }),
},
{ operation },
if (stopped.status === "incomplete" || stopped.status === "not-owned") {
const reason =
stopped.status === "incomplete"
? stopped.reason
: "SQLite worker ownership remains unresolved.";
const failure = sqliteCleanupFailure(
[
pending.process.launch.ownership?.runId ?? "owned-sqlite-worker",
root.path,
],
reason,
pending.result,
pending.input,
pending.output,
);
failure.bindCleanupDependencies(
[pending.processFailure, pending.rootFailure].filter(
(dependency): dependency is AnalysisError =>
dependency !== undefined,
),
);
failures.push(failure);
continue;
}
pending.process = undefined;
pending.processFailure?.resolveCleanup();
}
const cleaned = await cleanupSqliteRoot(
root,
@@ -212,8 +223,13 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort {
pending.input,
pending.output,
);
if (cleaned.ok) this.#pendingCleanup.delete(root);
else failures.push(cleaned.error);
if (cleaned.ok) {
pending.rootFailure.resolveCleanup();
this.#pendingCleanup.delete(root);
} else {
cleaned.error.bindCleanupDependencies([pending.rootFailure]);
failures.push(cleaned.error);
}
}
if (failures.length === 1 && failures[0] !== undefined)
return err(failures[0]);
@@ -225,7 +241,7 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort {
failures.flatMap((failure) => failure.cleanupResources),
{ failures: failures.map(projectAnalysisError) },
{ operation },
),
).bindCleanupDependencies(failures),
);
}
@@ -242,12 +258,18 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort {
let phase = "configuration";
let result: Result<AnalysisExecution, AnalysisError>;
try {
if (options?.signal?.aborted) throw new AnalysisCancelledError(operation);
if (options?.signal?.aborted)
throw new AnalysisCancelledError(operation, {
cause: options.signal.reason,
});
validateSqliteInput(input);
const retired = await this.#retryCleanup();
if (!retired.ok) return retired;
if (this.#closed || options?.signal?.aborted)
throw new AnalysisCancelledError(operation);
if (options?.signal?.aborted)
throw new AnalysisCancelledError(operation, {
cause: options.signal.reason,
});
if (this.#closed) throw new AnalysisCancelledError(operation);
root = await this.createRuntime();
phase = "artifact-read";
const snapshot = await captureSqliteDatabaseSnapshot(
@@ -302,50 +324,84 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort {
options?.signal,
);
retainedOutput ??= failure.capturedOutput;
result = err(selectedFailure(failure, options?.signal, retainedOutput));
result = err(failure);
}
if (root !== undefined) {
if (processOwner !== undefined) {
const ownership = processOwner.launch.ownership;
const rootFailure = new ProviderCleanupError(
SQLITE_PROVIDER_IDENTITY.id,
[root.path],
{ reason: "SQLite runtime root cleanup remains unresolved." },
{ operation },
);
const processReason =
processCleanup?.reason ?? "SQLite worker cleanup remains unresolved.";
const processFailure = new ProviderCleanupError(
SQLITE_PROVIDER_IDENTITY.id,
ownership === undefined
? ["owned-sqlite-worker"]
: ownedCommandResources(ownership),
{ reason: processReason },
{ operation },
);
this.#pendingCleanup.set(root, {
input,
result,
output: retainedOutput,
rootFailure,
processFailure,
process: processOwner,
});
if (processCleanup !== undefined) {
const ownership = processOwner.launch.ownership;
return err(
sqliteCleanupFailure(
ownership === undefined
? ["owned-sqlite-worker", root.path]
: ownedCommandResources(ownership, root.path),
processCleanup.reason,
result,
input,
retainedOutput,
processCleanup.cleanupResult === undefined
? undefined
: {
...(processCleanup.cleanupResult.failures === undefined
? {}
: {
failures: processCleanup.cleanupResult.failures.map(
(failure) => ({ ...failure }),
),
}),
...(processCleanup.cleanupResult.unverified === undefined
? {}
: {
unverified:
processCleanup.cleanupResult.unverified.map(
(process) => ({ ...process }),
),
}),
},
),
const failure = sqliteCleanupFailure(
ownership === undefined
? ["owned-sqlite-worker", root.path]
: ownedCommandResources(ownership, root.path),
processCleanup.reason,
result,
input,
retainedOutput,
processCleanup.cleanupResult === undefined
? undefined
: {
...(processCleanup.cleanupResult.failures === undefined
? {}
: {
failures: processCleanup.cleanupResult.failures.map(
(failure) => ({ ...failure }),
),
}),
...(processCleanup.cleanupResult.unverified === undefined
? {}
: {
unverified: processCleanup.cleanupResult.unverified.map(
(process) => ({ ...process }),
),
}),
},
);
failure.bindCleanupDependencies([processFailure, rootFailure]);
return err(failure);
}
return result;
if (!result.ok) {
result.error.bindCleanupDependencies([processFailure, rootFailure]);
return result;
}
const failure = sqliteCleanupFailure(
[
...(ownership === undefined
? ["owned-sqlite-worker"]
: ownedCommandResources(ownership)),
root.path,
],
processReason,
result,
input,
retainedOutput,
);
failure.bindCleanupDependencies([processFailure, rootFailure]);
return err(failure);
}
const cleaned = await cleanupSqliteRoot(
root,
@@ -354,33 +410,35 @@ export class SqliteDatabaseProvider implements SqliteDatabasePort {
retainedOutput,
);
if (!cleaned.ok) {
const rootFailure = new ProviderCleanupError(
SQLITE_PROVIDER_IDENTITY.id,
[root.path],
{ reason: "SQLite runtime root cleanup remains unresolved." },
{ operation },
);
cleaned.error.bindCleanupDependencies([rootFailure]);
this.#pendingCleanup.set(root, {
input,
result,
output: retainedOutput,
rootFailure,
processFailure: undefined,
process: undefined,
});
return cleaned;
}
}
// Completion was acknowledged before cleanup. A later abort does not erase it.
if (result.ok && options?.signal?.aborted)
return err(
new AnalysisCancelledError(operation, {
cause: options.signal.reason,
partialObservation: completedSqliteEvidence(input, result.value),
}),
);
return result;
}
}
const selectedFailure = (
failure: AnalysisError,
signal?: AbortSignal,
output?: AnalysisCapturedOutput,
): AnalysisError =>
signal?.aborted &&
failure._tag !== "AnalysisCancelledError" &&
!failure.cleanupIncomplete
? new AnalysisCancelledError(
operation,
output === undefined ? undefined : { capturedOutput: output },
)
: failure;
const createSqliteObservation = (
report: ReturnType<typeof sqliteDatabaseSchema.parse>,
path: string,
@@ -435,14 +493,7 @@ const sqliteCleanupFailure = (
cause?: unknown,
): ProviderCleanupError => {
const partialObservation = result.ok
? createEvidence(result.value.subject ?? undefined, result.value.provider, {
operation,
parameters: sqliteInspectionParameters(input),
result: result.value.result,
confidence: "observed",
locations: result.value.locations,
limitations: result.value.limitations,
})
? completedSqliteEvidence(input, result.value)
: result.error.partialObservation;
return new ProviderCleanupError(
SQLITE_PROVIDER_IDENTITY.id,
@@ -455,12 +506,26 @@ const sqliteCleanupFailure = (
},
{
operation,
...(result.ok ? {} : { cleanupPrimary: result.error }),
...(cause === undefined ? {} : { cause }),
...(partialObservation === undefined ? {} : { partialObservation }),
},
);
};
const completedSqliteEvidence = (
input: InspectSqliteDatabaseInput,
execution: AnalysisExecution,
) =>
createEvidence(execution.subject ?? undefined, execution.provider, {
operation,
parameters: sqliteInspectionParameters(input),
result: execution.result,
confidence: "observed",
locations: execution.locations,
limitations: execution.limitations,
});
const validateSqliteInput = (input: InspectSqliteDatabaseInput): void => {
const parsed = inspectSqliteDatabaseInputSchema.safeParse(input);
if (!parsed.success || !isAbsolute(input.path))
+2 -2
View File
@@ -248,7 +248,7 @@ export const captureSqliteDatabaseSnapshot = async (
signal?: AbortSignal,
selectedFileSystem: StableArtifactFileSystem = fileSystem,
) => {
signal?.throwIfAborted();
assertArtifactReadActive(signal);
const walPath = `${path}-wal`;
const journalPath = `${path}-journal`;
const paths = [path, walPath, journalPath] as const;
@@ -355,6 +355,6 @@ export const captureSqliteDatabaseSnapshot = async (
"integrity",
`SQLite snapshot file set changed during acquisition: ${paths[index] ?? path}`,
);
signal?.throwIfAborted();
assertArtifactReadActive(signal);
return { snapshotPath, artifact, wal: walArtifact };
};
+101 -1
View File
@@ -1,4 +1,12 @@
import { lstat, mkdtemp, readdir, rm, writeFile } from "node:fs/promises";
import {
lstat,
mkdir,
mkdtemp,
open,
readdir,
rm,
writeFile,
} from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
@@ -8,12 +16,104 @@ import { ArtifactReaderFailure } from "../artifacts/ArtifactReader.js";
import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js";
import type { StableArtifactFileSystem } from "../artifacts/readStableArtifact.js";
import { projectAnalysisError } from "../domain/analysisErrorProjection.js";
import { AnalysisCancelledError } from "../domain/analysisErrorCore.js";
import { ProviderCleanupError } from "../domain/providerCleanupError.js";
import type { AnalysisPartialObservation } from "../domain/analysisErrorBase.js";
import { captureSqliteDatabaseSnapshot } from "./SqliteDatabaseSnapshot.js";
import { SQLITE_DATABASE_LIMITS } from "./SqliteDatabaseLimits.js";
import { sqliteDatabaseFailure } from "./SqliteDatabaseFailures.js";
it("tags an exact caller reason at snapshot admission", async () => {
const workspace = await mkdtemp(join(tmpdir(), "rea-sqlite-snapshot-entry-"));
const reason = { requested: "cancel at snapshot admission" };
const controller = new AbortController();
controller.abort(reason);
const resourceScope = new ArtifactResourceScope();
try {
const failure = await captureSqliteDatabaseSnapshot(
join(workspace, "selected.db"),
join(workspace, "private"),
resourceScope,
controller.signal,
).then(
() => {
throw new Error("Expected cancellation at snapshot admission");
},
(cause: unknown) => cause,
);
expect(failure).toBeInstanceOf(ArtifactReaderFailure);
expect(failure).toMatchObject({
reason: "cancelled",
cancellation: { reason },
});
const projected = sqliteDatabaseFailure(
failure,
join(workspace, "selected.db"),
"artifact-read",
controller.signal,
);
expect(projected).toBeInstanceOf(AnalysisCancelledError);
expect(projected.cause).toBe(reason);
} finally {
await resourceScope.close();
await rm(workspace, { force: true, recursive: true });
}
});
it("keeps the exact caller reason when cancellation lands at final snapshot verification", async () => {
const workspace = await mkdtemp(
join(tmpdir(), "rea-sqlite-snapshot-cancel-"),
);
const databasePath = join(workspace, "selected.db");
const privateRoot = join(workspace, "private");
const journalPath = `${databasePath}-journal`;
const reason = { requested: "cancel after snapshot copy" };
const controller = new AbortController();
await writeFile(databasePath, "tiny fixture");
await mkdir(privateRoot);
let journalStats = 0;
const fileSystem: StableArtifactFileSystem = {
lstat: async (path) => {
if (path === journalPath) {
journalStats += 1;
if (journalStats === 2) controller.abort(reason);
throw Object.assign(new Error("journal absent"), { code: "ENOENT" });
}
return lstat(path, { bigint: true });
},
open: async (path, flags) => open(path, flags),
};
const resourceScope = new ArtifactResourceScope();
try {
const failure = await captureSqliteDatabaseSnapshot(
databasePath,
privateRoot,
resourceScope,
controller.signal,
fileSystem,
).then(
() => {
throw new Error("Expected cancellation during final verification");
},
(cause: unknown) => cause,
);
const projected = sqliteDatabaseFailure(
failure,
databasePath,
"artifact-read",
controller.signal,
);
expect(journalStats).toBe(2);
expect(projected).toBeInstanceOf(AnalysisCancelledError);
expect(projected.cause).toBe(reason);
} finally {
await resourceScope.close();
await rm(workspace, { force: true, recursive: true });
}
});
it("reports filesystem sizes above the safe-integer range exactly before opening files", async () => {
const databaseBytes = BigInt(Number.MAX_SAFE_INTEGER) + 1n;
const walBytes = 17n;
@@ -1,6 +1,7 @@
import { access } from "node:fs/promises";
import { expect, it, onTestFinished } from "vitest";
import { SqliteDatabaseProvider } from "../../../src/sqlite/SqliteDatabaseProvider.js";
import { ProviderCleanupError } from "../../../src/domain/providerCleanupError.js";
import { PrivateRuntimeRoot } from "../../../src/process/PrivateRuntimeRoot.js";
import { OwnedCommandFailure } from "../../../src/process/OwnedCommand.js";
import {
@@ -68,6 +69,7 @@ const cleanupFixture = async () => {
it("retains failed snapshot owners and blocks new acquisition until retry succeeds", async () => {
const fixture = await cleanupFixture();
const first = await fixture.provider.inspect({ path: fixture.path });
if (first.ok) throw new Error("Expected retained snapshot cleanup");
expect(first).toMatchObject({
ok: false,
error: { cleanupIncomplete: true },
@@ -82,13 +84,182 @@ it("retains failed snapshot owners and blocks new acquisition until retry succee
});
expect(fixture.roots).toHaveLength(1);
expect(fixture.attempts()).toBe(2);
expect(first.error.cleanupIncomplete).toBe(true);
expect(first.error.cleanupResources).toEqual([root.path]);
fixture.allowRemoval();
await fixture.provider.inspect({ path: fixture.path });
expect(first.error.cleanupIncomplete).toBe(false);
expect(first.error.cleanupResources).toEqual([]);
expect(fixture.roots).toHaveLength(2);
await expect(access(root.path)).rejects.toMatchObject({ code: "ENOENT" });
await fixture.provider.close();
});
it("shrinks inspect and close errors as each retained root is released", async () => {
const workspace = await createTestWorkspace(
"rea-sqlite-multiple-root-cleanup-",
);
const database = createSqliteDatabaseFixture(workspace.root);
database.close();
const roots: Array<{
readonly root: PrivateRuntimeRoot;
allowed: boolean;
attempts: number;
}> = [];
const provider = createUnavailableProvider(async () => {
const root = await PrivateRuntimeRoot.create({ parent: workspace.root });
const owner = { root, allowed: false, attempts: 0 };
roots.push(owner);
return {
path: root.path,
close: async () => {
owner.attempts += 1;
if (!owner.allowed) throw new Error("SQLite root removal denied");
await root.close();
},
};
});
onTestFinished(async () => {
for (const owner of roots) owner.allowed = true;
await provider.close();
await Promise.all(roots.map(({ root }) => root.close()));
await removeTestWorkspace(workspace.root);
});
const outcomes = await Promise.all([
provider.inspect({ path: database.path }),
provider.inspect({ path: database.path }),
]);
const inspectionErrors = outcomes.map((outcome) => {
if (outcome.ok) throw new Error("Expected retained root cleanup");
expect(outcome.error.cleanupIncomplete).toBe(true);
return outcome.error;
});
expect(roots).toHaveLength(2);
for (const error of inspectionErrors)
expect(error.cleanupResources).toHaveLength(1);
expect(
new Set(inspectionErrors.flatMap((error) => error.cleanupResources)),
).toEqual(new Set(roots.map(({ root }) => root.path)));
const firstOwner = roots[0];
const secondOwner = roots[1];
if (firstOwner === undefined || secondOwner === undefined)
throw new Error("Missing retained runtime root");
const firstError = inspectionErrors.find((error) =>
error.cleanupResources.includes(firstOwner.root.path),
);
const secondError = inspectionErrors.find((error) =>
error.cleanupResources.includes(secondOwner.root.path),
);
if (firstError === undefined || secondError === undefined)
throw new Error("Missing cleanup error for its retained runtime root");
const initialCloseFailure = await provider.close().then(
() => {
throw new Error("Expected both roots to remain owned");
},
(cause: unknown) => cause,
);
if (!(initialCloseFailure instanceof ProviderCleanupError))
throw new Error("Expected aggregate provider cleanup failure");
expect(initialCloseFailure.cleanupResources).toHaveLength(2);
expect(new Set(initialCloseFailure.cleanupResources)).toEqual(
new Set(roots.map(({ root }) => root.path)),
);
firstOwner.allowed = true;
const secondCloseFailure = await provider.close().then(
() => {
throw new Error("Expected the second root to remain owned");
},
(cause: unknown) => cause,
);
expect(secondCloseFailure).toMatchObject({
cleanupIncomplete: true,
cleanupResources: [secondOwner.root.path],
});
expect(firstError.cleanupIncomplete).toBe(false);
expect(firstError.cleanupResources).toEqual([]);
expect(secondError.cleanupResources).toEqual([secondOwner.root.path]);
expect(initialCloseFailure.cleanupIncomplete).toBe(true);
expect(initialCloseFailure.cleanupResources).toEqual([secondOwner.root.path]);
await expect(access(firstOwner.root.path)).rejects.toMatchObject({
code: "ENOENT",
});
await expect(access(secondOwner.root.path)).resolves.toBeUndefined();
secondOwner.allowed = true;
await provider.close();
expect(inspectionErrors.every((error) => !error.cleanupIncomplete)).toBe(
true,
);
expect(
inspectionErrors.every((error) => error.cleanupResources.length === 0),
).toBe(true);
expect(initialCloseFailure.cleanupIncomplete).toBe(false);
expect(initialCloseFailure.cleanupResources).toEqual([]);
await expect(access(secondOwner.root.path)).rejects.toMatchObject({
code: "ENOENT",
});
});
it("preserves the caller reason when cancellation arrives during awaited cleanup retry", async () => {
const workspace = await createTestWorkspace("rea-sqlite-retry-cancel-");
const database = createSqliteDatabaseFixture(workspace.root);
database.close();
const retryStarted = createDeferred<void>();
const releaseRetry = createDeferred<void>();
let closeAttempts = 0;
const provider = new SqliteDatabaseProvider(
{},
() => {
throw new Error("SQLite worker deliberately unavailable");
},
async () => {
const root = await PrivateRuntimeRoot.create({ parent: workspace.root });
return {
path: root.path,
close: async () => {
closeAttempts += 1;
if (closeAttempts === 1)
throw new Error("snapshot removal denied once");
if (closeAttempts === 2) {
retryStarted.resolve();
await releaseRetry.promise;
}
await root.close();
},
};
},
);
onTestFinished(async () => {
releaseRetry.resolve();
await provider.close();
await removeTestWorkspace(workspace.root);
});
const first = await provider.inspect({ path: database.path });
expect(first).toMatchObject({
ok: false,
error: { cleanupIncomplete: true },
});
const controller = new AbortController();
const reason = { requested: "cancel after retry cleanup" };
const inspection = provider.inspect(
{ path: database.path },
{ signal: controller.signal },
);
await retryStarted.promise;
controller.abort(reason);
releaseRetry.resolve();
const result = await inspection;
if (result.ok || result.error._tag !== "AnalysisCancelledError")
throw new Error("Expected cancellation after cleanup retry");
expect(result.error.cause).toBe(reason);
expect(closeAttempts).toBe(2);
});
it("awaits shared shutdown cleanup and retains failures for a later close", async () => {
const fixture = await cleanupFixture();
await fixture.provider.inspect({ path: fixture.path });
@@ -246,14 +417,20 @@ it("retains the snapshot until a failed worker cleanup owner can be stopped", as
await removeTestWorkspace(workspace.root);
});
await waitForProviderProcessReady(child);
expect(await provider.inspect({ path: database.path })).toMatchObject({
const original = await provider.inspect({ path: database.path });
expect(original).toMatchObject({
ok: false,
error: { cleanupIncomplete: true },
});
if (original.ok) throw new Error("Expected unresolved worker cleanup");
// The snapshot cannot be removed while its worker may still use it.
expect(rootAttempts).toBe(0);
const root = roots[0];
if (root === undefined) throw new Error("Missing retained snapshot");
expect(original.error.cleanupResources).toEqual([
"owned-sqlite-worker",
root.path,
]);
await expect(access(root.path)).resolves.toBeUndefined();
expect(await provider.inspect({ path: database.path })).toMatchObject({
ok: false,
@@ -261,19 +438,33 @@ it("retains the snapshot until a failed worker cleanup owner can be stopped", as
});
expect(roots).toHaveLength(1);
expect(workerAttempts).toBe(2);
await expect(provider.close()).rejects.toMatchObject({
cleanupIncomplete: true,
});
const workerCloseFailure = await provider.close().then(
() => {
throw new Error("Expected worker cleanup failure");
},
(cause: unknown) => cause,
);
if (!(workerCloseFailure instanceof ProviderCleanupError))
throw new Error("Expected provider cleanup failure");
expect(workerCloseFailure.cleanupIncomplete).toBe(true);
expect(rootAttempts).toBe(0);
workerRemovalAllowed = true;
await expect(provider.close()).rejects.toMatchObject({
cleanupResources: [root.path],
});
expect(original.error.cleanupIncomplete).toBe(true);
expect(original.error.cleanupResources).toEqual([root.path]);
expect(workerCloseFailure.cleanupIncomplete).toBe(true);
expect(workerCloseFailure.cleanupResources).toEqual([root.path]);
expect(child.signalCode).toBe("SIGKILL");
expect(workerAttempts).toBe(4);
expect(rootAttempts).toBe(1);
rootRemovalAllowed = true;
await provider.close();
expect(original.error.cleanupIncomplete).toBe(false);
expect(original.error.cleanupResources).toEqual([]);
expect(workerCloseFailure.cleanupIncomplete).toBe(false);
expect(workerCloseFailure.cleanupResources).toEqual([]);
expect(workerAttempts).toBe(4);
await expect(access(root.path)).rejects.toMatchObject({ code: "ENOENT" });
});
@@ -267,13 +267,15 @@ it("retains cancellation diagnostics and the interrupt exit code through CLI shu
it("retains reply-read cancellation and retries the failed reply descriptor close", async () => {
const selected = await fixture("reply");
const controller = new AbortController();
descriptorCleanup.onOpen = () =>
controller.abort(new Error("requested cancellation during reply read"));
const reason = { requested: "cancel during reply read" };
descriptorCleanup.onOpen = () => controller.abort(reason);
const result = await selected.provider.inspect(
{ path: selected.database.path },
{ signal: controller.signal },
);
if (result.ok) throw new Error("Expected cancelled reply read");
expect(result.error).toMatchObject({ _tag: "AnalysisCancelledError" });
expect(result.error.cause).toBe(reason);
const cancelled = {
code: "cleanup_incomplete",
details: {
@@ -378,12 +380,13 @@ it.each(["database", "WAL", "journal"] as const)(
async (member) => {
const selected = await fixture(member);
const controller = new AbortController();
descriptorCleanup.onOpen = () =>
controller.abort(new Error("requested caller cancellation"));
const reason = { requested: `cancel while reading ${member}` };
descriptorCleanup.onOpen = () => controller.abort(reason);
const result = await selected.provider.inspect(
{ path: selected.database.path },
{ signal: controller.signal },
);
if (result.ok) throw new Error("Expected caller cancellation");
expect(result).toMatchObject({
ok: false,
error: {
@@ -391,5 +394,17 @@ it.each(["database", "WAL", "journal"] as const)(
diagnostics: { previous_error: { code: "cancelled" } },
},
});
descriptorCleanup.denied = false;
await selected.provider.close();
expect(result.error.cleanupIncomplete).toBe(false);
expect(projectAnalysisError(result.error)).toMatchObject({
code: "cancelled",
});
expect(result.error.cleanupPrimary).toMatchObject({
_tag: "AnalysisCancelledError",
});
expect(result.error.cleanupPrimary?.cause).toBe(reason);
for (const root of selected.roots)
await expect(access(root.path)).rejects.toMatchObject({ code: "ENOENT" });
},
);
@@ -8,6 +8,7 @@ import { expect, it as nativeTest, onTestFinished } from "vitest";
import { projectAnalysisError } from "../../../src/domain/analysisErrorProjection.js";
import { SqliteDatabaseProvider } from "../../../src/sqlite/SqliteDatabaseProvider.js";
import { inspectSqliteDatabaseSnapshot } from "../../../src/sqlite/SqliteDatabaseInspection.js";
import { cleanupOwnedProcessGroup } from "../../../src/process/ProcessOwnership.js";
import { spawnOwnedProviderProcess } from "../../../src/process/ProviderProcess.js";
import { PrivateRuntimeRoot } from "../../../src/process/PrivateRuntimeRoot.js";
import { waitForProviderProcessReady } from "../../fixtures/providerProcess.js";
@@ -55,6 +56,7 @@ it.each(["completed reply", "malformed reply"] as const)(
async (replyKind) => {
const { workspace, database } = await createFixture();
let cleanupAttempts = 0;
let cleanupAllowed = false;
let ownedRoot = "";
const provider = new SqliteDatabaseProvider(process.env, async (spawn) => {
ownedRoot = spawn.cwd ?? "";
@@ -85,13 +87,14 @@ it.each(["completed reply", "malformed reply"] as const)(
...launched,
cleanup: async () => {
cleanupAttempts += 1;
return cleanupAttempts === 1
? { cleaned: false, reason: "fixture cleanup denial" }
: { cleaned: true, signaled: false };
return cleanupAllowed
? { cleaned: true, signaled: false }
: { cleaned: false, reason: "fixture cleanup denial" };
},
};
});
onTestFinished(async () => {
cleanupAllowed = true;
await provider.close();
await removeTestWorkspace(workspace.root);
});
@@ -128,18 +131,135 @@ it.each(["completed reply", "malformed reply"] as const)(
});
expect(cleanupAttempts).toBe(1);
await expect(access(`${ownedRoot}/reply.json`)).resolves.toBeUndefined();
let closeFailure: unknown;
try {
await provider.close();
} catch (cause) {
closeFailure = cause;
}
expect(closeFailure).toMatchObject({
cleanupIncomplete: true,
cleanupResources: expect.arrayContaining([ownedRoot]),
...(replyKind === "completed reply"
? {
partialObservation: {
operation: "inspect_sqlite_database",
confidence: "observed",
normalized_result: {
schema: { tables: expect.any(Array) },
rows: { returned_rows: 1 },
},
},
}
: {}),
});
expect(result).toMatchObject({
ok: false,
error: {
cleanupIncomplete: true,
cleanupResources: expect.arrayContaining([ownedRoot]),
...(replyKind === "completed reply"
? { partialObservation: { operation: "inspect_sqlite_database" } }
: {}),
},
});
cleanupAllowed = true;
await provider.close();
expect(cleanupAttempts).toBe(2);
expect(cleanupAttempts).toBe(3);
expect(result).toMatchObject({
ok: false,
error: { cleanupIncomplete: false, cleanupResources: [] },
});
expect(closeFailure).toMatchObject({
cleanupIncomplete: false,
cleanupResources: [],
});
await expectPrivateRootRemoved(ownedRoot);
},
);
nativeTest(
"retains a typed worker output failure after exact runtime-root cleanup retry",
async () => {
const { workspace, database } = await createFixture();
let root: PrivateRuntimeRoot | undefined;
let rootCloseAttempts = 0;
const provider = new SqliteDatabaseProvider(
process.env,
async (spawn) => {
const requestPath = spawn.arguments[2];
if (requestPath === undefined)
throw new Error("SQLite request path missing");
const request = JSON.parse(await readFile(requestPath, "utf8")) as {
reply_path: string;
};
await writeFile(
request.reply_path,
JSON.stringify({ ok: true, inspection: {} }),
{ flag: "wx", mode: 0o600 },
);
return spawnOwnedProviderProcess({
...spawn,
arguments: ["-e", "process.exit(0)"],
});
},
async () => {
root = await PrivateRuntimeRoot.create({ parent: workspace.root });
const acquired = root;
return {
path: acquired.path,
close: async () => {
rootCloseAttempts += 1;
if (rootCloseAttempts === 1)
throw new Error("runtime root removal denied once");
await acquired.close();
},
};
},
);
onTestFinished(async () => {
await provider.close();
await root?.close();
await removeTestWorkspace(workspace.root);
});
const result = await provider.inspect({ path: database.path });
if (result.ok) throw new Error("Expected malformed worker output");
expect(result.error).toMatchObject({
_tag: "ProviderAdapterError",
cleanupIncomplete: true,
diagnostics: {
previous_error: {
code: "unreadable_output",
},
},
});
await provider.close();
expect(rootCloseAttempts).toBe(2);
expect(result.error.cleanupIncomplete).toBe(false);
expect(result.error.cleanupResources).toEqual([]);
if (root === undefined) throw new Error("Runtime root was not acquired");
await expectPrivateRootRemoved(root.path);
expect(projectAnalysisError(result.error)).toMatchObject({
code: "unreadable_output",
});
expect(result.error.cleanupPrimary).toMatchObject({
_tag: "AnalysisOutputError",
capturedOutput: { stdout: "", stderr: "", truncated: false },
});
},
);
it.each(["late-cancellation", "cleanup-failure"])(
"preserves completed native-worker observations across %s",
async (scenario) => {
const { workspace, database } = await createFixture();
const original = await readFile(database.path);
const controller = new AbortController();
const cancellationReason = {
requested: "cancel during final root cleanup",
};
let cleanupAllowed = false;
let root: PrivateRuntimeRoot | undefined;
const provider = new SqliteDatabaseProvider(
@@ -157,7 +277,8 @@ it.each(["late-cancellation", "cleanup-failure"])(
return {
path: acquired.path,
close: async () => {
if (scenario === "late-cancellation") controller.abort();
if (scenario === "late-cancellation")
controller.abort(cancellationReason);
else if (!cleanupAllowed)
throw new Error("Completed snapshot cleanup denied");
await acquired.close();
@@ -176,9 +297,30 @@ it.each(["late-cancellation", "cleanup-failure"])(
if (scenario === "late-cancellation") {
expect(controller.signal.aborted).toBe(true);
expect(result).toMatchObject({
ok: true,
value: { result: { rows: { columns: ["id"], returned_rows: 1 } } },
ok: false,
error: {
_tag: "AnalysisCancelledError",
cause: cancellationReason,
partialObservation: {
operation: "inspect_sqlite_database",
confidence: "observed",
subject: {
local_path: database.path,
digest: { sha256: digestForTest(original) },
},
parameters: {
path: database.path,
table: "records",
columns: ["id"],
row_limit: 1,
},
normalized_result: { rows: { columns: ["id"], returned_rows: 1 } },
},
},
});
if (result.ok || result.error._tag !== "AnalysisCancelledError")
throw new Error("Expected cancellation after final root cleanup");
expect(result.error.cause).toBe(cancellationReason);
} else {
expect(result).toMatchObject({
ok: false,
@@ -587,13 +729,90 @@ nativeTest(
async () => {
const { database } = await createFixture();
const controller = new AbortController();
controller.abort();
expect(
await createProvider().inspect(
{ path: database.path },
{ signal: controller.signal },
),
).toMatchObject({ ok: false, error: { _tag: "AnalysisCancelledError" } });
const reason = { requested: "cancel before admission" };
controller.abort(reason);
const result = await createProvider().inspect(
{ path: database.path },
{ signal: controller.signal },
);
if (result.ok || result.error._tag !== "AnalysisCancelledError")
throw new Error("Expected pre-admission cancellation");
expect(result.error.cause).toBe(reason);
},
);
nativeTest(
"preserves the exact caller reason when cancellation lands before snapshot work",
async () => {
const { database } = await createFixture();
const controller = new AbortController();
const reason = { requested: "cancel after admission" };
const provider = new SqliteDatabaseProvider({}, () => {
throw new Error("SQLite worker must not launch before snapshot work");
});
onTestFinished(() => provider.close());
const inspection = provider.inspect(
{ path: database.path },
{ signal: controller.signal },
);
controller.abort(reason);
const result = await inspection;
if (result.ok || result.error._tag !== "AnalysisCancelledError")
throw new Error("Expected cancellation before snapshot work");
expect(result.error.cause).toBe(reason);
},
);
nativeTest(
"keeps an owned worker failure primary when cancellation arrives during cleanup",
async () => {
const { database } = await createFixture();
const controller = new AbortController();
const reason = { requested: "stop after worker exit" };
const provider = new SqliteDatabaseProvider(process.env, async (spawn) => {
const launched = await spawnOwnedProviderProcess({
...spawn,
arguments: [
"-e",
'process.stderr.write("authored SQLite worker failure"); process.exitCode = 7',
],
});
const cleanup =
launched.cleanup ??
(() => cleanupOwnedProcessGroup(launched.ownership));
return {
...launched,
cleanup: async () => {
const outcome = await cleanup();
controller.abort(reason);
return outcome;
},
};
});
onTestFinished(async () => {
await provider.close();
});
const result = await provider.inspect(
{ path: database.path, table: "records" },
{ signal: controller.signal },
);
expect(controller.signal.reason).toBe(reason);
expect(result).toMatchObject({
ok: false,
error: {
_tag: "ProviderAdapterError",
diagnostics: {
failure_kind: "process",
exit_code: 7,
captured_output: {
stderr: "authored SQLite worker failure",
},
},
},
});
},
);
@@ -709,6 +928,7 @@ nativeTest(
async () => {
const { database } = await createFixture();
const controller = new AbortController();
const reason = { requested: "stop owned SQLite worker" };
let ownedRoot = "";
let pid: number | undefined;
const provider = new SqliteDatabaseProvider(
@@ -733,7 +953,7 @@ nativeTest(
pid = launched.process.pid;
await waitForProviderProcessReady(launched.process);
await access(`${ownedRoot}/reply.json`);
setImmediate(() => controller.abort());
setImmediate(() => controller.abort(reason));
return launched;
},
);
@@ -745,6 +965,8 @@ nativeTest(
ok: false,
error: { _tag: "AnalysisCancelledError" },
});
if (!result.ok && result.error._tag === "AnalysisCancelledError")
expect(result.error.cause).toBe(reason);
if (pid === undefined) throw new Error("SQLite worker not acquired");
const acquiredPid = pid;
expect(() => process.kill(acquiredPid, 0)).toThrow();