From 988494abdbb27fb91ca6bd80362c9f1a332f86c3 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Sat, 17 Oct 2026 14:22:46 +0800 Subject: [PATCH] fix(kuna): own cancellable installation startup Reuse the shared startup owner for independent caller waits, final departure cleanup, and session closure. Carry the producer signal through version probing and fingerprint finalization. Retry cleanly cancelled acquisition after settlement while retaining authored failures and exact descriptor cleanup dependencies. Refs: #4698 --- docs/kuna-provider.md | 4 + src/kuna/KunaSessionClient.ts | 162 ++++++++++++------ tests/boundary/providers/kuna/fakeKuna.ts | 10 +- .../kuna/kunaInstallationCancellation.test.ts | 132 ++++++++++++++ .../kuna/kunaInstallationLifetime.test.ts | 69 ++++++++ 5 files changed, 321 insertions(+), 56 deletions(-) create mode 100644 tests/boundary/providers/kuna/kunaInstallationCancellation.test.ts 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); + } + }, +);