diff --git a/docs/kuna-provider.md b/docs/kuna-provider.md index 9b53c31f82..50e81368ba 100644 --- a/docs/kuna-provider.md +++ b/docs/kuna-provider.md @@ -87,6 +87,10 @@ caller requested. `diagnostics.producer_operation` retains the internal producer name when it differs. Caller-specific failures keep the producer's typed outcome, captured output, partial facts and live cleanup dependencies; cleanup retry updates every projection rather than preserving an independent stale marker. +Installation acquisition is shared, with independently cancellable caller waits. +The final departing caller stops the probe and joins its cleanup; session close +also stops and joins acquisition. A cleanly cancelled acquisition can be retried, +while a retained cleanup failure blocks replacement until its owner is closed. ## Provenance and limitations diff --git a/src/kuna/KunaSessionClient.ts b/src/kuna/KunaSessionClient.ts index 22d398c3dc..6b57315fc0 100644 --- a/src/kuna/KunaSessionClient.ts +++ b/src/kuna/KunaSessionClient.ts @@ -33,6 +33,7 @@ import { ProviderAdapterError } from "../domain/providerAdapterError.js"; import { ProviderCleanupError } from "../domain/providerCleanupError.js"; import { err, ok, type Result } from "../domain/result.js"; import { ABORTED, waitForAbortable } from "../process/AbortablePromise.js"; +import { SharedStartupAttempt } from "../process/SharedStartupAttempt.js"; import { KUNA_INVENTORY_OUTPUT_BYTES, kunaFileCleanupError, @@ -249,7 +250,9 @@ export class KunaSessionClient implements AnalysisClient { readonly #queryWork: KunaQueryWork; readonly #references: KunaReferenceQueries; readonly #strings: KunaStringQueries; - #installation: Promise> | undefined; + #installation: + | SharedStartupAttempt> + | undefined; #inventory: | Promise> | undefined; @@ -362,9 +365,11 @@ export class KunaSessionClient implements AnalysisClient { ], ), ); - const installation = await this.#ensureInstallation(); - if (!installation.ok) - return err(kunaQueryFailureForCaller(installation.error, operation)); + const installation = await this.#ensureInstallation( + operation, + options.signal, + ); + if (!installation.ok) return installation; try { if (operation === "health") return ok( @@ -1306,58 +1311,107 @@ export class KunaSessionClient implements AnalysisClient { return this.#inventoryIndex; } - async #ensureInstallation(): Promise< - Result - > { - this.#installation ??= (async () => { - const parentFileCloseResources: FileCloseResourceOwner = { - retain: (owner, resource) => - this.#fileCloseResources.retain({ - kind: "lease", - resource, - lease: { release: () => owner.close() }, - }), - run: (operation: () => Promise) => - this.#fileCloseResources.run(operation), - }; - const fileCloseScope = createKunaFileCloseScope(parentFileCloseResources); - try { - const inspected = await inspectKunaInstallation(this.#environment, { - fileCloseResources: fileCloseScope.owner, - }); - if (!inspected.ok) - return err(kunaUnavailable("health", inspected.error.reason)); - const committed = this.#profile?.parameters["kuna_sha256"]; - if (committed !== undefined && committed !== inspected.value.sha256) - return err( - new ProviderAdapterError(KUNA_PROVIDER_ID, "health", { - userMessage: - "The Kuna executable changed after this target was opened; reopen the target to analyze it with the current executable.", - diagnostics: { - committed_sha256: committed, - observed_sha256: inspected.value.sha256, - }, - }), - ); - return ok(inspected.value); - } catch (cause: unknown) { - const cancellation = regularFileCancellation(cause); + /** One installation producer; only the final departing caller stops it. */ + async #ensureInstallation( + operation: string, + signal: AbortSignal | undefined, + ): Promise> { + let attempt = this.#installation; + if (attempt === undefined) { + attempt = new SharedStartupAttempt( + (ownerSignal) => this.#own(this.#inspectInstallation(ownerSignal)), + this.#session.signal, + ); + const owned = attempt; + this.#installation = owned; + void owned.promise.then((result) => { + // A proven cancellation has no reusable installation result. Its + // complete cleanup precedes replacement; physical failures stay cached. if ( - !(cause instanceof RegularFileCleanupFailure) && - cancellation !== undefined + this.#installation === owned && + owned.signal.aborted && + (result.ok || + (result.error instanceof AnalysisCancelledError && + !result.error.cleanupIncomplete)) ) - return err( - new AnalysisCancelledError("health", { - cause: cancellation.reason, - }), - ); - if (!(cause instanceof RegularFileCleanupFailure)) throw cause; - const error = kunaFileCleanupError(cause, "health"); - fileCloseScope.bindFailure(cause, error); - return err(error); - } - })(); - return this.#installation; + this.#installation = undefined; + }); + } + const waited = await attempt.wait(signal); + if (waited.status === "cancelled") + return err( + new AnalysisCancelledError(operation, { cause: signal?.reason }), + ); + if (!waited.value.ok) { + // A fresh caller can arrive while the previous final waiter joins stop. + // Retry only proven cancellation, never an authored physical failure. + if ( + !signal?.aborted && + attempt.signal.aborted && + waited.value.error instanceof AnalysisCancelledError && + !waited.value.error.cleanupIncomplete + ) + return this.#ensureInstallation(operation, signal); + return err(kunaQueryFailureForCaller(waited.value.error, operation)); + } + if (signal?.aborted) + return err( + new AnalysisCancelledError(operation, { cause: signal.reason }), + ); + return waited.value; + } + + async #inspectInstallation( + signal: AbortSignal, + ): Promise> { + const parentFileCloseResources: FileCloseResourceOwner = { + retain: (owner, resource) => + this.#fileCloseResources.retain({ + kind: "lease", + resource, + lease: { release: () => owner.close() }, + }), + run: (operation: () => Promise) => + this.#fileCloseResources.run(operation), + }; + const fileCloseScope = createKunaFileCloseScope(parentFileCloseResources); + try { + const inspected = await inspectKunaInstallation(this.#environment, { + fileCloseResources: fileCloseScope.owner, + signal, + }); + if (!inspected.ok) + return err(kunaUnavailable("health", inspected.error.reason)); + const committed = this.#profile?.parameters["kuna_sha256"]; + if (committed !== undefined && committed !== inspected.value.sha256) + return err( + new ProviderAdapterError(KUNA_PROVIDER_ID, "health", { + userMessage: + "The Kuna executable changed after this target was opened; reopen the target to analyze it with the current executable.", + diagnostics: { + committed_sha256: committed, + observed_sha256: inspected.value.sha256, + }, + }), + ); + return ok(inspected.value); + } catch (cause: unknown) { + const cancellation = regularFileCancellation(cause); + if ( + !(cause instanceof RegularFileCleanupFailure) && + cancellation !== undefined + ) + return err( + new AnalysisCancelledError("health", { + cause: cancellation.reason, + }), + ); + if (!(cause instanceof RegularFileCleanupFailure)) + return err(failure("health", cause)); + const error = kunaFileCleanupError(cause, "health"); + fileCloseScope.bindFailure(cause, error); + return err(error); + } } /** Reuse a verified export, or start building one in the background. */ diff --git a/tests/boundary/providers/kuna/fakeKuna.ts b/tests/boundary/providers/kuna/fakeKuna.ts index 0965bc2f58..133da86357 100644 --- a/tests/boundary/providers/kuna/fakeKuna.ts +++ b/tests/boundary/providers/kuna/fakeKuna.ts @@ -19,7 +19,7 @@ export interface FakeKunaString { export interface FakeKunaBehavior { /** Text printed by `--version`; `exit` fails the probe instead. */ readonly version?: - | { readonly stdout: string } + | { readonly stdout: string; readonly gate?: string } | { readonly exit: number } | { readonly ignoreTerm: true }; readonly inventory?: "ok" | "not-json" | "hang"; @@ -95,7 +95,13 @@ if (args[0] === "--version") { const version = behavior.version ?? { stdout: "kuna 1.833\n" }; if ("exit" in version) { process.stderr.write("version probe failed\n"); process.exit(version.exit); } if ("ignoreTerm" in version) { process.on("SIGTERM", () => {}); hang(); } - else process.stdout.write(version.stdout); + else if (version.gate) { + const waiting = setInterval(() => { + if (!fs.existsSync(version.gate)) return; + clearInterval(waiting); + process.stdout.write(version.stdout); + }, 10); + } else process.stdout.write(version.stdout); } else if (args[0] === "functions") { const mode = behavior.inventory ?? "ok"; if (mode === "hang") hang(); diff --git a/tests/boundary/providers/kuna/kunaInstallationCancellation.test.ts b/tests/boundary/providers/kuna/kunaInstallationCancellation.test.ts new file mode 100644 index 0000000000..a6642322f1 --- /dev/null +++ b/tests/boundary/providers/kuna/kunaInstallationCancellation.test.ts @@ -0,0 +1,132 @@ +import { writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { expect, it as test } from "vitest"; +import { ProviderCleanupError } from "../../../../src/domain/providerCleanupError.js"; +import { fakeKunaCalls, setFakeKunaBehavior } from "./fakeKuna.js"; +import { session } from "./kunaSessionFixture.js"; + +const it = test.skipIf(process.platform === "win32"); + +const gatedInstallation = async () => { + const fixture = await session("rea-kuna-installation-cancel-"); + const gate = join(fixture.root, "release-version"); + await setFakeKunaBehavior(fixture.kuna, { + version: { stdout: "kuna 1.833\n", gate }, + }); + const started = async () => { + await expect + .poll(async () => (await fakeKunaCalls(fixture.kuna)).length) + .toBe(2); + const call = (await fakeKunaCalls(fixture.kuna))[1]; + if (call === undefined) throw new Error("Missing version probe"); + return call; + }; + return { + ...fixture, + started, + release: () => writeFile(gate, "release"), + }; +}; + +const gone = (pid: number) => { + try { + process.kill(pid, 0); + return false; + } catch (cause) { + if ((cause as NodeJS.ErrnoException).code !== "ESRCH") throw cause; + return true; + } +}; + +it("lets one caller leave acquisition with its exact reason while a peer keeps the probe", async () => { + const fixture = await gatedInstallation(); + const client = fixture.open(); + const controller = new AbortController(); + const reason = new ProviderCleanupError("foreign", ["foreign-owner"], {}); + const leaving = client.execute("health", {}, { signal: controller.signal }); + const peer = client.execute("health", {}); + let settled = false; + void leaving.then(() => { + settled = true; + }); + try { + const probe = await fixture.started(); + controller.abort(reason); + await expect.poll(() => settled, { timeout: 1500 }).toBe(true); + const result = await leaving; + if (result.ok) throw new Error("Expected cancellation"); + expect(result.error).toMatchObject({ + _tag: "AnalysisCancelledError", + cleanupIncomplete: false, + }); + expect(result.error.cause).toBe(reason); + expect(gone(probe.pid)).toBe(false); + await fixture.release(); + expect((await peer).ok).toBe(true); + expect((await fakeKunaCalls(fixture.kuna)).length).toBe(2); + expect(reason.cleanupIncomplete).toBe(true); + } finally { + await fixture.release(); + await Promise.all([leaving, peer]); + expect((await client.close()).ok).toBe(true); + } +}); + +it.each([null, "caller stopped"])( + "joins the final caller's probe stop before retrying: %j", + async (reason) => { + const fixture = await gatedInstallation(); + const client = fixture.open(); + const controller = new AbortController(); + const pending = client.execute("health", {}, { signal: controller.signal }); + let settled = false; + void pending.then(() => { + settled = true; + }); + try { + const probe = await fixture.started(); + controller.abort(reason); + await expect.poll(() => settled, { timeout: 1500 }).toBe(true); + const result = await pending; + if (result.ok) throw new Error("Expected cancellation"); + expect(result.error).toMatchObject({ + _tag: "AnalysisCancelledError", + cleanupIncomplete: false, + }); + expect(result.error.cause).toBe(reason); + expect(gone(probe.pid)).toBe(true); + await fixture.release(); + expect((await client.execute("health", {})).ok).toBe(true); + expect((await fakeKunaCalls(fixture.kuna)).length).toBe(3); + } finally { + await fixture.release(); + await pending; + expect((await client.close()).ok).toBe(true); + } + }, +); + +it("close cancels and joins an installation probe while callers wait", async () => { + const fixture = await gatedInstallation(); + const client = fixture.open(); + const pending = client.execute("health", {}); + let closed = false; + let closing: ReturnType | undefined; + try { + const probe = await fixture.started(); + closing = client.close(); + void closing.then(() => { + closed = true; + }); + await expect.poll(() => closed, { timeout: 1500 }).toBe(true); + expect((await closing).ok).toBe(true); + const result = await pending; + if (result.ok) throw new Error("Expected cancellation"); + expect(result.error._tag).toBe("AnalysisCancelledError"); + expect(gone(probe.pid)).toBe(true); + } finally { + await fixture.release(); + await Promise.all([pending, closing]); + expect((await client.close()).ok).toBe(true); + } +}); diff --git a/tests/boundary/providers/kuna/kunaInstallationLifetime.test.ts b/tests/boundary/providers/kuna/kunaInstallationLifetime.test.ts index 2b123ddea7..ed70d6d0ff 100644 --- a/tests/boundary/providers/kuna/kunaInstallationLifetime.test.ts +++ b/tests/boundary/providers/kuna/kunaInstallationLifetime.test.ts @@ -18,6 +18,7 @@ const gates = vi.hoisted(() => { hashClose: createGate(), shebangRead: createGate(), opens: 0, + failClose: false, }; }); @@ -50,6 +51,8 @@ vi.mock("../../../../src/filesystem/RegularFile.js", async (importOriginal) => { gates.hashClose.enter(); await gates.hashClose.hold; } + if (gates.failClose) + throw new Error("controlled final close refusal"); await handle.close(); }, } as unknown as typeof handle; @@ -63,14 +66,17 @@ import type { FileCloseResourceOwner } from "../../../../src/application/Analysi import { inspectKunaInstallation } from "../../../../src/kuna/KunaInstallation.js"; import { createTestTempDirectory } from "../../../fixtures/temporaryDirectory.js"; import { writeFakeKuna } from "./fakeKuna.js"; +import { session } from "./kunaSessionFixture.js"; beforeEach(() => { gates.opens = 0; + gates.failClose = false; gates.hashClose = gates.createGate(); gates.shebangRead = gates.createGate(); }); afterEach(() => { + gates.failClose = false; gates.hashClose.release(); gates.shebangRead.release(); }); @@ -147,3 +153,66 @@ it } }, ); + +it + .skipIf(process.platform === "win32") + .each(["hash-close", "shebang-read", "close-refusal"] as const)( + "joins session acquisition finalization and retains cleanup retry during %s cancellation", + async (phase) => { + gates.hashClose.release(); + gates.shebangRead.release(); + const fixture = await session("rea-kuna-session-finalization-"); + gates.opens = 0; + gates.hashClose = gates.createGate(); + gates.shebangRead = gates.createGate(); + const client = fixture.open(); + const controller = new AbortController(); + const reason = { selected: "caller reason" }; + const pending = client.execute("health", {}, { signal: controller.signal }); + let finished = false; + let replacement: ReturnType | undefined; + void pending.then(() => { + finished = true; + }); + try { + await gates.hashClose.started; + if (phase === "shebang-read") { + gates.hashClose.release(); + await gates.shebangRead.started; + } + controller.abort(reason); + await new Promise((resolve) => setImmediate(resolve)); + expect(finished).toBe(false); + if (phase === "hash-close") replacement = client.execute("health", {}); + gates.failClose = phase === "close-refusal"; + gates.hashClose.release(); + gates.shebangRead.release(); + const result = await pending; + if (result.ok) throw new Error("Expected interrupted installation"); + if (phase === "close-refusal") { + expect(result.error.cleanupIncomplete).toBe(true); + const retained = await client.execute("health", {}); + if (retained.ok) throw new Error("Expected retained cleanup failure"); + expect(retained.error).toBe(result.error); + expect((await client.close()).ok).toBe(false); + gates.failClose = false; + expect((await client.close()).ok).toBe(true); + expect(result.error.cleanupIncomplete).toBe(false); + } else { + expect(result.error._tag).toBe("AnalysisCancelledError"); + expect(result.error.cause).toBe(reason); + expect(result.error.cleanupIncomplete).toBe(false); + if (replacement !== undefined) + expect((await replacement).ok).toBe(true); + expect((await client.close()).ok).toBe(true); + } + } finally { + gates.failClose = false; + gates.hashClose.release(); + gates.shebangRead.release(); + await pending; + await replacement; + expect((await client.close()).ok).toBe(true); + } + }, +);