diff --git a/docs/ghidra-provider.md b/docs/ghidra-provider.md index 55ceadd3d9..434d822309 100644 --- a/docs/ghidra-provider.md +++ b/docs/ghidra-provider.md @@ -139,6 +139,12 @@ overview merely because its source artifact is unchanged. ## Admitted function-analysis semantics +Concurrent callers share one startup while retaining independent cancellation. +Only the final departing caller abandons startup and waits for its cleanup. +New attempts wait for that cleanup; failed cleanup retains its original error +and owned directories until a successful close. Closing the provider reports a +provider-state failure to live callers rather than cancelling their signals. + | Concern | Ghidra contract | | ----------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Decompiler lifetime | One persistent `DecompInterface` is opened for the imported Program and disposed during bridge shutdown. Native decompilation has no fixed per-function deadline. External functions or functions without bodies return `null`; cancellation and native failure remain distinct. | diff --git a/docs/tool-design.md b/docs/tool-design.md index e28e6e6886..dd923af342 100644 --- a/docs/tool-design.md +++ b/docs/tool-design.md @@ -83,6 +83,11 @@ providers need not expose identical state machines. recovery handshakes. Bound abandoned unauthenticated listeners where the protocol permits it; that deadline must not expire authenticated idle sessions or long-running native work. Listener expiry does not prove target closure. +- Shared startup has one producer and independently cancellable caller waits. + Keep it running while any caller still needs it. The final departing caller + waits for owned cleanup and retains its failure. Provider closure is distinct + from caller cancellation. Retained cleanup capabilities block replacement; + derive readiness from the current startup's completion and live connection. - For cancellable observations, carry cancellation through result consumption and awaited cleanup. Before returning success, check the signal after the last awaited finalization step. diff --git a/src/ghidra/GhidraClient.ts b/src/ghidra/GhidraClient.ts index 214ed73a8b..4f3676c21d 100644 --- a/src/ghidra/GhidraClient.ts +++ b/src/ghidra/GhidraClient.ts @@ -15,6 +15,11 @@ import { err, ok, type Result } from "../domain/result.js"; import { silentLogger } from "../logger.js"; import type { Logger } from "pino"; import { PendingOperations } from "../process/PendingOperations.js"; +import { ABORTED, waitForAbortable } from "../process/AbortablePromise.js"; +import { + SharedStartupAttempt, + STARTUP_OWNER_CLOSED, +} from "../process/SharedStartupAttempt.js"; import { PrivateRuntimeRoot } from "../process/PrivateRuntimeRoot.js"; import { ProviderStartupDeadline } from "../process/ProviderDeadline.js"; import { ProviderRunLineage } from "../process/ProviderRunLineage.js"; @@ -81,6 +86,8 @@ import { ghidraSessionRoot } from "./GhidraSessionRoot.js"; const SHUTDOWN_TIMEOUT_MS = 10_000; const SESSION_ROOT = ghidraSessionRoot(); +type GhidraStartupAttempt = SharedStartupAttempt; + /** Closed Java-bridge operation union callable after the exact handshake. */ export type GhidraOperation = | GhidraInventoryOperation @@ -120,9 +127,7 @@ export class GhidraClient { #outputDrainPromise: Promise | undefined; #outputDrainIncomplete = false; #lastDiagnostics: Readonly> = {}; - #startupController: AbortController | undefined; - #startPromise: Promise | undefined; - #ready = false; + #startupAttempt: GhidraStartupAttempt | undefined; #closePromise: Promise | undefined; readonly #failure = bindGhidraSessionFailure( () => this.#diagnostics(), @@ -220,9 +225,18 @@ export class GhidraClient { ): Promise { if (progress !== undefined) return this.#reportStartupWait(this.start(signal), signal, progress); + if (signal?.aborted === true) + return Promise.resolve( + err(this.#failure("cancelled", "Ghidra startup wait was cancelled")), + ); if (this.#closePromise !== undefined) - return this.#closePromise.then( - () => this.start(signal), + return waitForAbortable(this.#closePromise, signal).then( + (closed) => + closed === ABORTED + ? err( + this.#failure("cancelled", "Ghidra startup wait was cancelled"), + ) + : this.start(signal), (cause: unknown) => err( new GhidraSessionError( @@ -233,45 +247,63 @@ export class GhidraClient { ), ), ); - if (this.#startPromise !== undefined) { - if (this.#startupController?.signal.aborted && !signal?.aborted) - return this.#startPromise.then((result) => - result.ok || result.error.cleanupFailure !== undefined - ? result - : this.start(signal), - ); - return this.#startPromise; + const current = this.#startupAttempt; + if (current !== undefined) { + if (current.signal.aborted) + return this.#retryAfterAbortedStartup(current, signal); + return this.#waitForStartup(current, signal); } - const controller = new AbortController(); - const onAbort = (): void => controller.abort(signal?.reason); - if (signal?.aborted === true) onAbort(); - else signal?.addEventListener("abort", onAbort, { once: true }); - const started = this.#start(controller.signal); - this.#startupController = controller; - this.#startPromise = started; - const reset = (): void => { - signal?.removeEventListener("abort", onAbort); - if (this.#startPromise === started) { - this.#startPromise = undefined; - if (this.#startupController === controller) - this.#startupController = undefined; - } - }; - void started.then( + const attempt = new SharedStartupAttempt((signal) => this.#start(signal)); + const { promise } = attempt; + this.#startupAttempt = attempt; + void promise.then( (result) => { - signal?.removeEventListener("abort", onAbort); - if (result.ok) this.#ready = true; - else reset(); + if ( + !result.ok && + this.#startupAttempt === attempt && + result.error.cleanupFailure === undefined + ) + this.#startupAttempt = undefined; }, (cause: unknown) => { this.#logger.debug( { error: cause instanceof Error ? cause.message : String(cause) }, "Ghidra startup promise rejected during bookkeeping", ); - reset(); + if (this.#startupAttempt === attempt) this.#startupAttempt = undefined; }, ); - return started; + return this.#waitForStartup(attempt, signal); + } + + /** Preserve cleanup failures for the last caller leaving shared startup. */ + async #waitForStartup( + attempt: GhidraStartupAttempt, + signal: AbortSignal | undefined, + ): Promise { + const waited = await attempt.wait(signal); + if (waited.status === "completed") return waited.value; + if (waited.status === "stopped" && !waited.value.ok) return waited.value; + return err(this.#failure("cancelled", "Ghidra startup wait was cancelled")); + } + + /** Wait for abandoned startup cleanup before retrying under the new caller. */ + async #retryAfterAbortedStartup( + attempt: GhidraStartupAttempt, + signal: AbortSignal | undefined, + ): Promise { + const signalAborted = (): boolean => signal?.aborted === true; + if (signalAborted()) + return err( + this.#failure("cancelled", "Ghidra startup wait was cancelled"), + ); + const waited = await waitForAbortable(attempt.promise, signal); + if (waited === ABORTED || (waited.ok && signalAborted())) + return err( + this.#failure("cancelled", "Ghidra startup wait was cancelled"), + ); + if (waited.ok || waited.error.cleanupFailure !== undefined) return waited; + return this.start(signal); } async #reportStartupWait( @@ -475,12 +507,19 @@ export class GhidraClient { } } + /** Read completion from the current attempt; no independent readiness flag. */ + #startupSucceeded(): boolean { + const startup = this.#startupAttempt?.settlement; + return startup?.status === "completed" && startup.value.ok; + } + /** True only after the authenticated startup handshake has completed. */ isReady(): boolean { return ( - this.#ready && + this.#startupSucceeded() && this.#socket !== undefined && !this.#socket.destroyed && + !this.#closing && this.#closePromise === undefined ); } @@ -543,12 +582,10 @@ export class GhidraClient { } #requestClose(forceStop: boolean): Promise { - this.#ready = false; - const starting = this.#startPromise; - const controller = this.#startupController; - controller?.abort(); + const starting = this.#startupAttempt; + starting?.close(); this.#closePromise ??= Promise.resolve().then(() => - this.#close(starting, controller, forceStop), + this.#close(starting, forceStop), ); return this.#closePromise; } @@ -574,12 +611,10 @@ export class GhidraClient { ? "exited" : this.#socket?.destroyed === true ? "unreachable" - : this.#startPromise === undefined && - (this.#process !== undefined || - this.#socket !== undefined || - this.#runtimeRoot !== undefined || - this.#socketRoot !== undefined || - this.#snapshotOwners.size > 0) + : !this.#startupSucceeded() && + (this.#startupAttempt === undefined || + this.#startupAttempt.settled) && + this.#hasOwnedResources() ? "unknown" : undefined; if (state === undefined) return undefined; @@ -603,24 +638,25 @@ export class GhidraClient { } async #start(signal: AbortSignal): Promise { - if (this.#socket === undefined && this.#runtimeRoot === undefined) { - this.#lastDiagnostics = {}; - this.#processSnapshot = undefined; - this.#startupOutputCheckpoint = undefined; - this.#activeOutputCheckpoint = undefined; - this.#outputDrainPromise = undefined; - this.#outputDrainIncomplete = false; - } if (signal.aborted) return err(this.#failure("cancelled", "Ghidra startup was cancelled")); - if (this.#socket !== undefined || this.#runtimeRoot !== undefined) + if (this.#hasOwnedResources()) return err(this.#failure("protocol", "Ghidra client is already started")); + this.#lastDiagnostics = {}; + this.#processSnapshot = undefined; + this.#startupOutputCheckpoint = undefined; + this.#activeOutputCheckpoint = undefined; + this.#outputDrainPromise = undefined; + this.#outputDrainIncomplete = false; const deadline = new ProviderStartupDeadline( this.#options.startupTimeoutMs, signal, ); try { - return await this.#startWithin(deadline); + const result = await this.#startWithin(deadline); + return result.ok && deadline.signal.aborted + ? this.#startupInterrupted(deadline) + : result; } catch (cause: unknown) { if (cause instanceof GhidraSessionError) return err(cause); throw cause; @@ -629,6 +665,18 @@ export class GhidraClient { } } + /** Every retained cleanup capability blocks replacement by a new startup. */ + #hasOwnedResources(): boolean { + return ( + this.#process !== undefined || + this.#launch !== undefined || + this.#socket !== undefined || + this.#runtimeRoot !== undefined || + this.#socketRoot !== undefined || + this.#snapshotOwners.size > 0 + ); + } + async #startWithin( deadline: ProviderStartupDeadline, ): Promise { @@ -838,18 +886,15 @@ export class GhidraClient { } async #close( - starting: Promise | undefined, - controller: AbortController | undefined, + starting: GhidraStartupAttempt | undefined, forceStop: boolean, ): Promise { try { // best-effort cleanup: a rejected startup must not mask close/cleanup. - await starting?.catch(() => undefined); + await starting?.promise.catch(() => undefined); await this.#cleanup(forceStop); } finally { - if (this.#startPromise === starting) this.#startPromise = undefined; - if (this.#startupController === controller) - this.#startupController = undefined; + if (this.#startupAttempt === starting) this.#startupAttempt = undefined; this.#closePromise = undefined; } } @@ -1036,6 +1081,12 @@ export class GhidraClient { deadline: ProviderStartupDeadline, cause?: unknown, ): GhidraSessionError { + if (deadline.signal.reason === STARTUP_OWNER_CLOSED) + return this.#failure( + "process", + "Ghidra session closed during startup", + cause, + ); return deadline.interruption === "cancelled" ? this.#failure("cancelled", "Ghidra startup was cancelled", cause) : this.#failure("timeout", "Ghidra startup deadline elapsed", cause, { diff --git a/src/hopper/HopperClient.ts b/src/hopper/HopperClient.ts index 4f319207e6..2edf8eee6e 100644 --- a/src/hopper/HopperClient.ts +++ b/src/hopper/HopperClient.ts @@ -41,6 +41,11 @@ import { } from "../process/ProviderDeadline.js"; import { redactCapturedTransportCredential } from "../process/ProviderDiagnosticRedaction.js"; import { ProviderRunLineage } from "../process/ProviderRunLineage.js"; +import { ABORTED, waitForAbortable } from "../process/AbortablePromise.js"; +import { + SharedStartupAttempt, + STARTUP_OWNER_CLOSED, +} from "../process/SharedStartupAttempt.js"; import { type ProviderProcessDiagnostic, ProviderProcessSupervisor, @@ -91,6 +96,9 @@ export interface HopperClientOptions { readonly logger?: Logger; } +type HopperStartupResult = Result; +type HopperStartupAttempt = SharedStartupAttempt; + type HopperClientCloseOptions = { readonly progress?: ProgressReporter; readonly retainDocument?: boolean; @@ -125,11 +133,9 @@ export class HopperClient { #launcherExitCode: number | null | undefined; #launcherFailureDiagnostic: HopperStartupFailureDiagnostic | undefined; #operationFailure: OperationFailureRecord | undefined; - #startupController: AbortController | undefined; - #ready = false; #startupFailureResult: AnalysisError | undefined; #runId: string | undefined; - #startPromise: Promise> | undefined; + #startupAttempt: HopperStartupAttempt | undefined; #closePromise: Promise> | undefined; readonly #onSocketData = (chunk: string): void => { this.#responses.push(chunk); @@ -223,8 +229,8 @@ export class HopperClient { if ( this.#closing || this.#closePromise !== undefined || - (!this.#ready && - this.#startupController !== undefined && + (!this.#startupSucceeded() && + this.#startupAttempt !== undefined && this.#operationFailure === undefined) ) { return { @@ -286,63 +292,81 @@ export class HopperClient { }; } + /** Read completion from the current attempt; no independent readiness flag. */ + #startupSucceeded(): boolean { + const startup = this.#startupAttempt?.settlement; + return startup?.status === "completed" && startup.value.ok; + } + /** True only after the authenticated health handshake has completed. */ isReady(): boolean { return ( - this.#ready && + this.#startupSucceeded() && this.#socket !== undefined && !this.#socket.destroyed && + !this.#closing && this.#closePromise === undefined ); } - /** Launch the bridge once and complete its authenticated health handshake. */ - start( - signal?: AbortSignal, - ): Promise> { - if (this.#closePromise !== undefined) { - return this.#closePromise.then((closed) => - closed.ok ? this.start(signal) : closed, + /** Launch the bridge once while each caller retains its own cancellation. */ + start(signal?: AbortSignal): Promise { + if (isAborted(signal)) + return Promise.resolve(err(new HopperCancelledError())); + if (this.#closePromise !== undefined) + return waitForAbortable(this.#closePromise, signal).then((closed) => + closed === ABORTED + ? err(new HopperCancelledError()) + : closed.ok + ? this.start(signal) + : closed, ); - } - if (this.#startPromise !== undefined) return this.#startPromise; - - const controller = new AbortController(); - const onAbort = (): void => controller.abort(signal?.reason); - if (signal?.aborted === true) onAbort(); - else signal?.addEventListener("abort", onAbort, { once: true }); - const started = this.#start(controller.signal); - this.#startupController = controller; - this.#startPromise = started; - const release = (): void => { - signal?.removeEventListener("abort", onAbort); - }; - const reset = (): void => { - if (this.#startPromise === started) { - this.#startPromise = undefined; - if (this.#startupController === controller) { - this.#startupController = undefined; - } - } - }; - void started.then( + const current = this.#startupAttempt; + if (current !== undefined) + return current.signal.aborted + ? this.#retryAfterAbortedStartup(current, signal) + : this.#waitForStartup(current, signal); + const attempt = new SharedStartupAttempt((signal) => this.#start(signal)); + this.#startupAttempt = attempt; + void attempt.promise.then( (result) => { - release(); if (!result.ok) { this.#rememberStartupFailure(result.error); - reset(); - } else this.#ready = true; + if (this.#startupAttempt === attempt) + this.#startupAttempt = undefined; + } }, (cause: unknown) => { this.#logger.debug( { error: cause instanceof Error ? cause.message : String(cause) }, "Hopper startup promise rejected during bookkeeping", ); - release(); - reset(); + if (this.#startupAttempt === attempt) this.#startupAttempt = undefined; }, ); - return started; + return this.#waitForStartup(attempt, signal); + } + + async #waitForStartup( + attempt: HopperStartupAttempt, + signal?: AbortSignal, + ): Promise { + const waited = await attempt.wait(signal); + if (waited.status === "completed") return waited.value; + if (waited.status === "stopped" && !waited.value.ok) return waited.value; + return err(new HopperCancelledError()); + } + + /** Cleanup belongs to the abandoned attempt, never to a prospective retry. */ + async #retryAfterAbortedStartup( + attempt: HopperStartupAttempt, + signal?: AbortSignal, + ): Promise { + const waited = await waitForAbortable(attempt.promise, signal); + if (waited === ABORTED || (waited.ok && isAborted(signal))) + return err(new HopperCancelledError()); + if (waited.ok || waited.error.cleanupIncomplete) return waited; + return this.start(signal); } async #start( @@ -369,7 +393,10 @@ export class HopperClient { signal, ); try { - return await this.#startWithin(deadline); + const result = await this.#startWithin(deadline); + return result.ok && deadline.signal.aborted + ? this.#startupFailure(await this.#startupInterruption(deadline)) + : result; } finally { deadline.dispose(); } @@ -463,7 +490,17 @@ export class HopperClient { } const parsed = await this.#authenticateServerInfo(health.value, runId); if (!parsed.ok) return this.#startupFailure(parsed.error); - await this.#lineage.observe(this.#resources.launch); + try { + await this.#lineage.observe(this.#resources.launch); + } catch (cause: unknown) { + return this.#startupFailure( + deadline.signal.aborted + ? await this.#startupInterruption(deadline) + : new HopperStartError({ cause }), + ); + } + if (deadline.signal.aborted) + return this.#startupFailure(await this.#startupInterruption(deadline)); return parsed; } @@ -588,24 +625,22 @@ export class HopperClient { close( options: HopperClientCloseOptions = {}, ): Promise> { - const starting = this.#startPromise; - const controller = this.#startupController; - controller?.abort(); + const starting = this.#startupAttempt; + starting?.close(); this.#closePromise ??= Promise.resolve().then(() => - this.#close(starting, controller, options), + this.#close(starting, options), ); return this.#closePromise; } async #close( - starting: Promise> | undefined, - controller: AbortController | undefined, + starting: HopperStartupAttempt | undefined, options: HopperClientCloseOptions, ): Promise> { try { // best-effort cleanup: a rejected startup must not mask close/cleanup; // #cleanup reports the authoritative close result. - await starting?.catch(() => undefined); + await starting?.promise.catch(() => undefined); await confirmExternalApplicationExit(this.#resources, this.#socket, () => this.#releaseTransport(), ); @@ -614,10 +649,7 @@ export class HopperClient { if (closed.ok) this.#startupFailureResult = undefined; return closed; } finally { - if (this.#startPromise === starting) this.#startPromise = undefined; - if (this.#startupController === controller) { - this.#startupController = undefined; - } + if (this.#startupAttempt === starting) this.#startupAttempt = undefined; this.#closePromise = undefined; } } @@ -692,7 +724,6 @@ export class HopperClient { options: HopperClientCloseOptions = {}, ): Promise> { this.#closing = true; - this.#ready = false; try { const closed = await cleanupHopperSession({ socket: this.#socket, @@ -792,7 +823,9 @@ export class HopperClient { async #startupInterruption( deadline: ProviderStartupDeadline, - ): Promise { + ): Promise { + if (deadline.signal.reason === STARTUP_OWNER_CLOSED) + return new HopperProtocolError("Hopper session closed during startup"); if (deadline.interruption === "cancelled") return new HopperCancelledError(); return new HopperTimeoutError( diff --git a/src/process/SharedStartupAttempt.test.ts b/src/process/SharedStartupAttempt.test.ts new file mode 100644 index 0000000000..1ed9e3f2dd --- /dev/null +++ b/src/process/SharedStartupAttempt.test.ts @@ -0,0 +1,72 @@ +import { expect, it } from "vitest"; + +import { createDeferred } from "../../tests/fixtures/binarySession.js"; +import { + SharedStartupAttempt, + STARTUP_OWNER_CLOSED, +} from "./SharedStartupAttempt.js"; + +it("keeps shared startup running for an interested peer", async () => { + const completed = createDeferred(); + const attempt = new SharedStartupAttempt(() => completed.promise); + const controller = new AbortController(); + const departing = attempt.wait(controller.signal); + const staying = attempt.wait(); + controller.abort(); + await expect(departing).resolves.toEqual({ status: "cancelled" }); + expect(attempt.signal.aborted).toBe(false); + completed.resolve("ready"); + await expect(staying).resolves.toEqual({ + status: "completed", + value: "ready", + }); +}); + +it("retains the final cleanup result when all callers cancel together", async () => { + const cleanup = createDeferred<{ cleanupIncomplete: boolean }>(); + const attempt = new SharedStartupAttempt(() => cleanup.promise); + const first = new AbortController(); + const last = new AbortController(); + const waits = [attempt.wait(first.signal), attempt.wait(last.signal)]; + first.abort(); + last.abort(); + await expect(waits[0]).resolves.toEqual({ status: "cancelled" }); + expect(attempt.signal.aborted).toBe(true); + expect(attempt.settled).toBe(false); + const failure = { cleanupIncomplete: true }; + cleanup.resolve(failure); + await expect(waits[1]).resolves.toEqual({ + status: "stopped", + value: failure, + }); + expect(attempt.settled).toBe(true); +}); + +it("distinguishes provider closure from a caller's cancellation", async () => { + const cleanup = createDeferred(); + const attempt = new SharedStartupAttempt(() => cleanup.promise); + const controller = new AbortController(); + const waiting = attempt.wait(controller.signal); + attempt.close(); + expect(attempt.signal.reason).toBe(STARTUP_OWNER_CLOSED); + expect(controller.signal.aborted).toBe(false); + cleanup.resolve("provider stopped"); + await expect(waiting).resolves.toEqual({ + status: "completed", + value: "provider stopped", + }); +}); + +it("does not lose an unexpected producer rejection after its final caller cancels", async () => { + let reject: ((cause: unknown) => void) | undefined; + const promise = new Promise((_resolve, fail) => { + reject = fail; + }); + const attempt = new SharedStartupAttempt(() => promise); + const controller = new AbortController(); + const waiting = attempt.wait(controller.signal); + controller.abort(); + const cause = new TypeError("Cleanup rejected"); + reject?.(cause); + await expect(waiting).rejects.toBe(cause); +}); diff --git a/src/process/SharedStartupAttempt.ts b/src/process/SharedStartupAttempt.ts new file mode 100644 index 0000000000..16bf3b8d8b --- /dev/null +++ b/src/process/SharedStartupAttempt.ts @@ -0,0 +1,69 @@ +import { ABORTED, waitForAbortable } from "./AbortablePromise.js"; + +/** An owner closing a provider is distinct from a caller cancelling its wait. */ +export const STARTUP_OWNER_CLOSED = Symbol("provider startup owner closed"); + +/** The final departing caller retains the producer's complete cleanup result. */ +export type SharedStartupWait = + | { readonly status: "completed"; readonly value: Value } + | { readonly status: "cancelled" } + | { readonly status: "stopped"; readonly value: Value }; + +type StartupSettlement = + | { readonly status: "pending" } + | { readonly status: "completed"; readonly value: Value } + | { readonly status: "rejected" }; + +/** Own one startup operation while its callers wait independently. */ +export class SharedStartupAttempt { + readonly #controller = new AbortController(); + readonly promise: Promise; + #waiters = 0; + #settlement: StartupSettlement = { status: "pending" }; + + constructor(start: (signal: AbortSignal) => Promise) { + this.promise = start(this.#controller.signal); + void this.promise.then( + (value) => { + this.#settlement = { status: "completed", value }; + }, + () => { + this.#settlement = { status: "rejected" }; + }, + ); + } + + /** Signal consumed by the startup producer, never by an individual waiter. */ + get signal(): AbortSignal { + return this.#controller.signal; + } + + /** Completion includes the producer's final cleanup, not just its abort. */ + get settled(): boolean { + return this.#settlement.status !== "pending"; + } + + /** Parsed completion facts are owned here, rather than copied into adapters. */ + get settlement(): StartupSettlement { + return this.#settlement; + } + + /** Stop startup when its owning provider closes, regardless of active waits. */ + close(): void { + if (!this.settled) this.#controller.abort(STARTUP_OWNER_CLOSED); + } + + /** A peer's departure leaves the producer running; final departure joins cleanup. */ + async wait(signal?: AbortSignal): Promise> { + this.#waiters += 1; + try { + const waited = await waitForAbortable(this.promise, signal); + if (waited !== ABORTED) return { status: "completed", value: waited }; + if (this.#waiters > 1) return { status: "cancelled" }; + if (!this.settled) this.#controller.abort(signal?.reason); + return { status: "stopped", value: await this.promise }; + } finally { + this.#waiters -= 1; + } + } +} diff --git a/tests/boundary/mcp/ghidraStartupCancellationMcp.test.ts b/tests/boundary/mcp/ghidraStartupCancellationMcp.test.ts new file mode 100644 index 0000000000..f2f64b9670 --- /dev/null +++ b/tests/boundary/mcp/ghidraStartupCancellationMcp.test.ts @@ -0,0 +1,140 @@ +import { access } from "node:fs/promises"; + +import { Client, InMemoryTransport } from "@modelcontextprotocol/client"; +import { expect, it, vi } from "vitest"; + +import { AnalysisProviderRegistry } from "../../../src/application/binary/AnalysisProviderRegistry.js"; +import { composeBinarySession } from "../../../src/application/binary/BinarySessionComposition.js"; +import { parseConfig } from "../../../src/config/parseConfig.js"; +import { GhidraClient } from "../../../src/ghidra/GhidraClient.js"; +import type { GhidraQueuedRequestResult } from "../../../src/ghidra/GhidraRequestQueue.js"; +import { ghidraProfileInstallationHost } from "../../../src/ghidra/GhidraInstallation.fixture.js"; +import { GhidraProvider } from "../../../src/ghidra/GhidraProvider.js"; +import type { GhidraProviderClientFactory } from "../../../src/ghidra/GhidraProviderClient.js"; +import { silentLogger } from "../../../src/logger.js"; +import { createServer } from "../../../src/server/createServer.js"; +import { createDeferred } from "../../fixtures/binarySession.js"; + +import { GhidraFixtureLauncher } from "../providers/ghidra/ghidraClient.fixture.js"; + +it.each([0, 1])( + "releases cancelled MCP startup caller %s while its peer completes and stays warm", + async (cancelledIndex) => { + const entered = createDeferred(); + const admitted = [createDeferred(), createDeferred()]; + const gate = createDeferred(); + const fixture = new GhidraFixtureLauncher(); + const roots = fixture.runtimeRoots; + const children = fixture.processes; + const results: (GhidraQueuedRequestResult | undefined)[] = []; + let producerSignal: AbortSignal | undefined; + let callCount = 0; + const factory: GhidraProviderClientFactory = (options) => { + const raw = new GhidraClient({ + ...options, + launcher: { + async launch(launch, launchOptions) { + producerSignal = launchOptions?.signal; + entered.resolve(); + await gate.promise; + return fixture.launch(launch); + }, + }, + }); + return { + start: raw.start.bind(raw), + close: raw.close.bind(raw), + callTool: (operation, parameters, requestOptions) => { + const index = callCount++; + const pending = raw.callTool(operation, parameters, requestOptions); + admitted[index]?.resolve(); + return pending.then((result) => { + results[index] = result; + return result; + }); + }, + }; + }; + const config = parseConfig({ + GHIDRA_INSTALL_DIR: "/fixture/ghidra", + REA_GHIDRA_STARTUP_TIMEOUT_MS: "10000", + }); + if (!config.ok) throw config.error; + const provider = new GhidraProvider( + config.value, + silentLogger, + {}, + ghidraProfileInstallationHost, + factory, + ); + const session = composeBinarySession( + new AnalysisProviderRegistry([provider]), + ); + const server = createServer( + { kind: "session", session }, + { logger: silentLogger }, + ); + const mcp = new Client({ name: "ghidra-shared-startup", version: "1.0.0" }); + const [clientTransport, serverTransport] = + InMemoryTransport.createLinkedPair(); + const controllers = [new AbortController(), new AbortController()]; + const calls: Promise< + | { ok: true; value: Awaited> } + | { ok: false; cause: unknown } + >[] = []; + try { + await server.connect(serverTransport); + await mcp.connect(clientTransport); + const opened = await mcp.callTool({ + name: "open_binary", + arguments: { path: process.execPath, provider_id: "ghidra" }, + }); + expect(opened.isError).not.toBe(true); + for (const [index, controller] of controllers.entries()) { + calls.push( + mcp + .callTool( + { name: "list_strings", arguments: {} }, + { signal: controller.signal }, + ) + .then( + (value) => ({ ok: true as const, value }), + (cause: unknown) => ({ ok: false as const, cause }), + ), + ); + await admitted[index]?.promise; + } + await entered.promise; + controllers[cancelledIndex]?.abort(); + await vi.waitFor(() => + expect(results[cancelledIndex]).toMatchObject({ + ok: false, + error: { kind: "cancelled" }, + }), + ); + expect(await calls[cancelledIndex]).toMatchObject({ ok: false }); + expect(producerSignal?.aborted).toBe(false); + expect(children).toHaveLength(0); + gate.resolve(); + const staying = await calls[1 - cancelledIndex]; + if (staying === undefined || !staying.ok) + throw new Error("Live MCP peer did not complete", { cause: staying }); + expect(staying.value.isError).not.toBe(true); + expect(results[1 - cancelledIndex]).toMatchObject({ ok: true }); + expect(controllers[1 - cancelledIndex]?.signal.aborted).toBe(false); + const warm = await mcp.callTool({ name: "list_strings", arguments: {} }); + expect(warm.isError).not.toBe(true); + expect(callCount).toBe(3); + expect(children).toHaveLength(1); + } finally { + for (const controller of controllers) controller.abort(); + gate.resolve(); + await Promise.allSettled(calls); + await mcp.close(); + await server.close(); + await expect(session.close()).resolves.toMatchObject({ ok: true }); + for (const root of roots) + await expect(access(root)).rejects.toMatchObject({ code: "ENOENT" }); + } + }, +); diff --git a/tests/boundary/providers/ghidra/ghidraClient.fixture.ts b/tests/boundary/providers/ghidra/ghidraClient.fixture.ts new file mode 100644 index 0000000000..dc623abbf1 --- /dev/null +++ b/tests/boundary/providers/ghidra/ghidraClient.fixture.ts @@ -0,0 +1,93 @@ +import { spawn, type ChildProcess } from "node:child_process"; +import { mkdir } from "node:fs/promises"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; + +import { ok } from "../../../../src/domain/result.js"; +import type { + GhidraLaunchSession, + GhidraLauncher, +} from "../../../../src/ghidra/GhidraLauncher.js"; + +/** Shared real-socket producer for Ghidra client and public MCP regressions. */ +export const ghidraFixturePath = fileURLToPath( + new URL("../../../fixtures/fakeGhidra.mjs", import.meta.url), +); + +type FixtureMode = + | "reject_first_connection" + | "reject_all_connections" + | "success" + | "fragmented" + | "oversized_result" + | "wrong_identity" + | "malformed" + | "contradictory" + | "future_id" + | "analysis_timeout" + | "remote_error" + | "shutdown_error" + | "invalid_shutdown_ack" + | "hang_after_start" + | "hang_tools" + | "exit_tools" + | "silent" + | "exit" + | "old_oom_then_malformed" + | "diagnostic_output"; + +export class GhidraFixtureLauncher implements GhidraLauncher { + #notifyStarted: ((child: ChildProcess) => void) | undefined; + readonly started = new Promise((resolve) => { + this.#notifyStarted = resolve; + }); + readonly runtimeRoots: string[] = []; + readonly endpointPaths: string[] = []; + readonly tokens: string[] = []; + readonly processes: ChildProcess[] = []; + readonly requests: string[] = []; + + constructor(readonly mode: FixtureMode = "success") {} + + async launch(session: GhidraLaunchSession) { + this.runtimeRoots.push(session.runtimeRoot); + this.endpointPaths.push(session.endpointPath); + this.tokens.push(session.token); + const projectRoot = join(session.runtimeRoot, "project"); + await mkdir(projectRoot, { recursive: true }); + const process_ = spawn(process.execPath, [ghidraFixturePath], { + stdio: ["ignore", "pipe", "pipe", "ipc"], + env: { + ...process.env, + REA_GHIDRA_FIXTURE_CONFIG: JSON.stringify({ + endpointPath: session.endpointPath, + token: session.token, + runId: session.runId, + providerVersion: session.providerVersion, + profileDigest: session.profileDigest, + targetSha256: session.targetSha256, + transport: session.transport, + mode: this.mode, + }), + }, + }); + process_.on("message", (value: unknown) => { + if ( + typeof value === "object" && + value !== null && + "method" in value && + typeof value.method === "string" + ) + this.requests.push(value.method); + }); + this.processes.push(process_); + process_.once("spawn", () => this.#notifyStarted?.(process_)); + return ok({ + process: process_, + ownsProcessLifetime: true as const, + projectRoot, + ghidraLogPath: join(session.runtimeRoot, "ghidra.log"), + scriptLogPath: join(session.runtimeRoot, "script.log"), + }); + } +} diff --git a/tests/boundary/providers/ghidra/ghidraClient.test.ts b/tests/boundary/providers/ghidra/ghidraClient.test.ts index 5cc3d20756..a210add19e 100644 --- a/tests/boundary/providers/ghidra/ghidraClient.test.ts +++ b/tests/boundary/providers/ghidra/ghidraClient.test.ts @@ -3,9 +3,8 @@ import { createHash } from "node:crypto"; import { EventEmitter } from "node:events"; import { readFileSync } from "node:fs"; import { access, chmod, mkdir, writeFile } from "node:fs/promises"; -import { join } from "node:path"; +import { basename, join } from "node:path"; import { PassThrough } from "node:stream"; -import { fileURLToPath } from "node:url"; import { afterEach, describe, expect, it, onTestFinished, vi } from "vitest"; import pino from "pino"; @@ -15,6 +14,7 @@ import type { ProgressUpdate } from "../../../../src/application/ProgressReporte import { ok } from "../../../../src/domain/result.js"; import { GhidraClient } from "../../../../src/ghidra/GhidraClient.js"; import { ProviderRunLineage } from "../../../../src/process/ProviderRunLineage.js"; +import { PrivateRuntimeRoot } from "../../../../src/process/PrivateRuntimeRoot.js"; import { OwnedFileHandle } from "../../../../src/filesystem/OwnedFileHandle.js"; import { GHIDRA_PROCESS_DIAGNOSTIC_BYTES } from "../../../../src/ghidra/GhidraDiagnostics.js"; import { GhidraHeadlessLauncher } from "../../../../src/ghidra/GhidraLauncher.js"; @@ -31,10 +31,10 @@ import { } from "../../../../src/ghidra/GhidraSessionValues.js"; import { waitForExit as waitForChildExit } from "../../../support/process/processFixture.js"; import { createTestTempDirectory } from "../../../fixtures/temporaryDirectory.js"; - -const fixturePath = fileURLToPath( - new URL("../../../fixtures/fakeGhidra.mjs", import.meta.url), -); +import { + GhidraFixtureLauncher as FixtureLauncher, + ghidraFixturePath as fixturePath, +} from "./ghidraClient.fixture.js"; const PROFILE_DIGEST = "a".repeat(64); const PROVIDER_VERSION = "12.1.4"; @@ -44,84 +44,6 @@ const TARGET_SHA256 = createHash("sha256") const HOST_TRANSPORT: GhidraTransportKind = process.platform === "win32" ? "authenticated-loopback-tcp" : "unix-socket"; -type FixtureMode = - | "reject_first_connection" - | "reject_all_connections" - | "success" - | "fragmented" - | "oversized_result" - | "wrong_identity" - | "malformed" - | "contradictory" - | "future_id" - | "analysis_timeout" - | "remote_error" - | "shutdown_error" - | "invalid_shutdown_ack" - | "hang_after_start" - | "hang_tools" - | "exit_tools" - | "silent" - | "exit" - | "old_oom_then_malformed" - | "diagnostic_output"; - -class FixtureLauncher implements GhidraLauncher { - #notifyStarted: ((child: ChildProcess) => void) | undefined; - readonly started = new Promise((resolve) => { - this.#notifyStarted = resolve; - }); - readonly runtimeRoots: string[] = []; - readonly endpointPaths: string[] = []; - readonly tokens: string[] = []; - readonly processes: ChildProcess[] = []; - readonly requests: string[] = []; - - constructor(readonly mode: FixtureMode = "success") {} - - async launch(session: GhidraLaunchSession) { - this.runtimeRoots.push(session.runtimeRoot); - this.endpointPaths.push(session.endpointPath); - this.tokens.push(session.token); - const projectRoot = join(session.runtimeRoot, "project"); - await mkdir(projectRoot, { recursive: true }); - const process_ = spawn(process.execPath, [fixturePath], { - stdio: ["ignore", "pipe", "pipe", "ipc"], - env: { - ...process.env, - REA_GHIDRA_FIXTURE_CONFIG: JSON.stringify({ - endpointPath: session.endpointPath, - token: session.token, - runId: session.runId, - providerVersion: session.providerVersion, - profileDigest: session.profileDigest, - targetSha256: session.targetSha256, - transport: session.transport, - mode: this.mode, - }), - }, - }); - process_.on("message", (value: unknown) => { - if ( - typeof value === "object" && - value !== null && - "method" in value && - typeof value.method === "string" - ) - this.requests.push(value.method); - }); - this.processes.push(process_); - process_.once("spawn", () => this.#notifyStarted?.(process_)); - return ok({ - process: process_, - ownsProcessLifetime: true as const, - projectRoot, - ghidraLogPath: join(session.runtimeRoot, "ghidra.log"), - scriptLogPath: join(session.runtimeRoot, "script.log"), - }); - } -} - const preExitedProcessLauncher = (options: { readonly output: string; readonly closeOutput: boolean; @@ -561,123 +483,304 @@ describe("GhidraClient", () => { }); }); -describe("GhidraClient startup lifecycle", () => { - it("retains startup failure and cleans up when lineage observation rejects", async () => { - const cause = new TypeError("Lineage reader rejected"); - const observe = vi.spyOn(ProviderRunLineage.prototype, "observe"); - onTestFinished(() => observe.mockRestore()); - observe.mockRejectedValueOnce(cause); - const launcher = new FixtureLauncher(); - const client = clientFor(launcher); +it("reports owner closure to live startup callers without cancelling their signals", async () => { + let entered: (() => void) | undefined; + let release: (() => void) | undefined; + const pending = new Promise((resolve) => { + entered = resolve; + }); + const gate = new Promise((resolve) => { + release = resolve; + }); + const original = ProviderRunLineage.prototype.observe; + const observe = vi + .spyOn(ProviderRunLineage.prototype, "observe") + .mockImplementationOnce(async function (this: ProviderRunLineage, launch) { + entered?.(); + await gate; + await original.call(this, launch); + }); + const client = clientFor(new FixtureLauncher(), { + startupTimeoutMs: 10_000, + }); + const controllers = [new AbortController(), new AbortController()]; + const waits = controllers.map(({ signal }) => client.start(signal)); + try { + await pending; + const closing = client.close(); + release?.(); + for (const wait of waits) + await expect(wait).resolves.toMatchObject({ + ok: false, + error: { + kind: "process", + message: "Ghidra session closed during startup", + }, + }); + await expect(closing).resolves.toEqual({ ok: true, value: null }); + expect(controllers.every(({ signal }) => !signal.aborted)).toBe(true); + expect(client.isReady()).toBe(false); + } finally { + release?.(); + await client.close(); + observe.mockRestore(); + } +}); - const result = await client.start(); - expect(result).toMatchObject({ - ok: false, - error: { - kind: "start", - message: "Ghidra process lineage observation failed", +it.skipIf(process.platform === "win32")( + "retains a failed socket-root cleanup owner and the original startup error", + async () => { + const parent = join( + await createTestTempDirectory("rea-ghidra-socket-retry-"), + "long-".repeat(20), + ); + await mkdir(parent); + const roots = new Set(); + const originalCreate = PrivateRuntimeRoot.create; + const originalClose = PrivateRuntimeRoot.prototype.close; + let allowCleanup = false; + const create = vi + .spyOn(PrivateRuntimeRoot, "create") + .mockImplementation(async (options) => { + const root = await originalCreate.call( + PrivateRuntimeRoot, + options?.prefix === "rea-ghidra-" ? { ...options, parent } : options, + ); + roots.add(root); + return root; + }); + const close = vi + .spyOn(PrivateRuntimeRoot.prototype, "close") + .mockImplementation(async function (this: PrivateRuntimeRoot) { + if ( + !allowCleanup && + basename(this.path).startsWith("rea-ghidra-socket-") + ) + throw new Error("Socket root cleanup denied"); + await originalClose.call(this); + }); + const launcher = new FixtureLauncher(); + const controller = new AbortController(); + const client = clientFor({ + async launch(session) { + const launched = await launcher.launch(session); + if (launcher.processes.length === 1) controller.abort(); + return launched; }, }); - if (result.ok) throw new Error("Expected lineage observation failure"); - expect(result.error.cause).toBe(cause); - const child = launcher.processes[0]; - if (child === undefined) throw new Error("fixture child was not launched"); - await expect(waitForChildExit(child, 5_000)).resolves.toBe(true); - await expect(access(launcher.runtimeRoots[0] ?? "")).rejects.toMatchObject({ - code: "ENOENT", - }); - expect(client.isReady()).toBe(false); - await expect(client.start()).resolves.toMatchObject({ ok: true }); - expect(launcher.processes).toHaveLength(2); - }); + try { + const first = await client.start(controller.signal); + expect(first).toMatchObject({ + ok: false, + error: { + kind: "cancelled", + cleanupFailure: { cleanupIncomplete: true }, + }, + }); + if (first.ok) throw new Error("Expected retained cleanup"); + const socketRoot = [...roots].find(({ path }) => + basename(path).startsWith("rea-ghidra-socket-"), + ); + if (socketRoot === undefined) + throw new Error("Expected a separate socket root"); + await expect( + access(launcher.runtimeRoots[0] ?? ""), + ).rejects.toMatchObject({ code: "ENOENT" }); + await expect(access(socketRoot.path)).resolves.toBeUndefined(); + const retry = await client.start(); + if (retry.ok) throw new Error("Restart lost the retained socket root"); + expect(retry.error).toBe(first.error); + expect(launcher.processes).toHaveLength(1); + expect(client.operationHealthSnapshot()).toMatchObject({ + state: "unknown", + retryAction: "unknown", + }); + allowCleanup = true; + await expect(client.close()).resolves.toEqual({ + ok: true, + value: null, + }); + await expect(access(socketRoot.path)).rejects.toMatchObject({ + code: "ENOENT", + }); + await expect(client.start()).resolves.toMatchObject({ ok: true }); + expect(launcher.processes).toHaveLength(2); + } finally { + allowCleanup = true; + await client.close(); + for (const root of roots) await originalClose.call(root); + create.mockRestore(); + close.mockRestore(); + } + }, +); - it("cleans up when the final waiter cancels during lineage observation", async () => { - let markObservationPending: (() => void) | undefined; - const observationPending = new Promise((resolve) => { - markObservationPending = resolve; +it.each([0, 1])( + "cancels startup waiter %s without interrupting its peer", + async (cancelledIndex) => { + let entered: (() => void) | undefined; + let release: (() => void) | undefined; + const launchPending = new Promise((resolve) => { + entered = resolve; }); - let releaseObservation: (() => void) | undefined; - const observationGate = new Promise((resolve) => { - releaseObservation = resolve; + const launchGate = new Promise((resolve) => { + release = resolve; }); - const originalObserve = ProviderRunLineage.prototype.observe; - const observe = vi.spyOn(ProviderRunLineage.prototype, "observe"); - onTestFinished(() => observe.mockRestore()); - observe.mockImplementationOnce(async function ( - this: ProviderRunLineage, - launch, - ) { - markObservationPending?.(); - await observationGate; - return await originalObserve.call(this, launch); - }); - const launcher = new FixtureLauncher(); - const client = clientFor(launcher); - const controller = new AbortController(); - const starting = client.start(controller.signal); - await observationPending; - const child = launcher.processes[0]; - if (child === undefined) throw new Error("fixture child was not launched"); - - controller.abort(); - releaseObservation?.(); - - await expect(starting).resolves.toMatchObject({ - ok: false, - error: { kind: "cancelled" }, - }); - await expect(waitForChildExit(child, 5_000)).resolves.toBe(true); - await expect(access(launcher.runtimeRoots[0] ?? "")).rejects.toMatchObject({ - code: "ENOENT", - }); - await expect(client.callToolIfReady("list_strings", {})).resolves.toBe( - undefined, + const client = clientFor( + { + async launch(session) { + entered?.(); + await launchGate; + return launcher.launch(session); + }, + }, + { startupTimeoutMs: 10_000 }, ); + const controllers = [new AbortController(), new AbortController()]; + const waits = controllers.map(({ signal }) => client.start(signal)); + try { + await launchPending; + let cancelled: Awaited> | undefined; + void waits[cancelledIndex]?.then((result) => { + cancelled = result; + }); + controllers[cancelledIndex]?.abort(); + await vi.waitFor(() => + expect(cancelled).toMatchObject({ + ok: false, + error: { kind: "cancelled" }, + }), + ); + expect(launcher.processes).toHaveLength(0); + release?.(); + await expect(waits[1 - cancelledIndex]).resolves.toMatchObject({ + ok: true, + }); + expect(launcher.processes).toHaveLength(1); + expect(client.isReady()).toBe(true); + } finally { + release?.(); + await Promise.all(waits); + } + }, +); - await expect(client.start()).resolves.toMatchObject({ ok: true }); - expect(launcher.processes).toHaveLength(2); +it("retains startup failure and cleans up when lineage observation rejects", async () => { + const cause = new TypeError("Lineage reader rejected"); + const observe = vi.spyOn(ProviderRunLineage.prototype, "observe"); + onTestFinished(() => observe.mockRestore()); + observe.mockRejectedValueOnce(cause); + const launcher = new FixtureLauncher(); + const client = clientFor(launcher); + + const result = await client.start(); + expect(result).toMatchObject({ + ok: false, + error: { + kind: "start", + message: "Ghidra process lineage observation failed", + }, + }); + if (result.ok) throw new Error("Expected lineage observation failure"); + expect(result.error.cause).toBe(cause); + const child = launcher.processes[0]; + if (child === undefined) throw new Error("fixture child was not launched"); + await expect(waitForChildExit(child, 5_000)).resolves.toBe(true); + await expect(access(launcher.runtimeRoots[0] ?? "")).rejects.toMatchObject({ + code: "ENOENT", + }); + expect(client.isReady()).toBe(false); + await expect(client.start()).resolves.toMatchObject({ ok: true }); + expect(launcher.processes).toHaveLength(2); +}); + +it("cleans up when the final waiter cancels during lineage observation", async () => { + let markObservationPending: (() => void) | undefined; + const observationPending = new Promise((resolve) => { + markObservationPending = resolve; + }); + let releaseObservation: (() => void) | undefined; + const observationGate = new Promise((resolve) => { + releaseObservation = resolve; + }); + const originalObserve = ProviderRunLineage.prototype.observe; + const observe = vi.spyOn(ProviderRunLineage.prototype, "observe"); + onTestFinished(() => observe.mockRestore()); + observe.mockImplementationOnce(async function ( + this: ProviderRunLineage, + launch, + ) { + markObservationPending?.(); + await observationGate; + return await originalObserve.call(this, launch); }); - it("applies one startup deadline and removes the private runtime", async () => { - vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); - const launcher = new FixtureLauncher("silent"); - const client = clientFor(launcher, { startupTimeoutMs: 30 }); + const launcher = new FixtureLauncher(); + const client = clientFor(launcher); + const controller = new AbortController(); + const starting = client.start(controller.signal); + await observationPending; + const child = launcher.processes[0]; + if (child === undefined) throw new Error("fixture child was not launched"); - const pending = client.start(); - const child = await launcher.started; - await vi.advanceTimersByTimeAsync(30); - const result = await vi.waitFor(() => pending, { timeout: 10_000 }); - vi.useRealTimers(); + controller.abort(); + releaseObservation?.(); - expect(result).toMatchObject({ - ok: false, - error: { kind: "timeout", timeoutMs: 30 }, - }); - await expect(access(launcher.runtimeRoots[0] ?? "")).rejects.toMatchObject({ - code: "ENOENT", - }); - await client.close(); - await expect(waitForExit(child)).resolves.toBe(true); + await expect(starting).resolves.toMatchObject({ + ok: false, + error: { kind: "cancelled" }, }); - - it("cancels startup promptly and leaves no project or process", async () => { - const launcher = new FixtureLauncher("silent"); - const client = clientFor(launcher, { startupTimeoutMs: 10_000 }); - const controller = new AbortController(); - const pending = client.start(controller.signal); - const process_ = await launcher.started; - controller.abort(); - - const result = await pending; - - expect(result).toMatchObject({ ok: false, error: { kind: "cancelled" } }); - await expect(access(launcher.runtimeRoots[0] ?? "")).rejects.toMatchObject({ - code: "ENOENT", - }); - await client.close(); - await expect(waitForExit(process_)).resolves.toBe(true); + await expect(waitForChildExit(child, 5_000)).resolves.toBe(true); + await expect(access(launcher.runtimeRoots[0] ?? "")).rejects.toMatchObject({ + code: "ENOENT", }); + await expect(client.callToolIfReady("list_strings", {})).resolves.toBe( + undefined, + ); + + await expect(client.start()).resolves.toMatchObject({ ok: true }); + expect(launcher.processes).toHaveLength(2); +}); + +it("applies one startup deadline and removes the private runtime", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const launcher = new FixtureLauncher("silent"); + const client = clientFor(launcher, { startupTimeoutMs: 30 }); + + const pending = client.start(); + const child = await launcher.started; + await vi.advanceTimersByTimeAsync(30); + const result = await vi.waitFor(() => pending, { timeout: 10_000 }); + vi.useRealTimers(); + + expect(result).toMatchObject({ + ok: false, + error: { kind: "timeout", timeoutMs: 30 }, + }); + await expect(access(launcher.runtimeRoots[0] ?? "")).rejects.toMatchObject({ + code: "ENOENT", + }); + await client.close(); + await expect(waitForExit(child)).resolves.toBe(true); +}); + +it("cancels startup promptly and leaves no project or process", async () => { + const launcher = new FixtureLauncher("silent"); + const client = clientFor(launcher, { startupTimeoutMs: 10_000 }); + const controller = new AbortController(); + const pending = client.start(controller.signal); + const process_ = await launcher.started; + controller.abort(); + + const result = await pending; + + expect(result).toMatchObject({ ok: false, error: { kind: "cancelled" } }); + await expect(access(launcher.runtimeRoots[0] ?? "")).rejects.toMatchObject({ + code: "ENOENT", + }); + await client.close(); + await expect(waitForExit(process_)).resolves.toBe(true); }); describe("GhidraClient established requests", () => { diff --git a/tests/boundary/providers/hopper/hopperClient.restart.test.ts b/tests/boundary/providers/hopper/hopperClient.restart.test.ts index ef22cf0f4a..8e5ed8038d 100644 --- a/tests/boundary/providers/hopper/hopperClient.restart.test.ts +++ b/tests/boundary/providers/hopper/hopperClient.restart.test.ts @@ -21,7 +21,7 @@ describe("HopperClient restart", () => { const result = await starting; expect(result.ok).toBe(false); - if (!result.ok) expect(result.error._tag).toBe("HopperCancelledError"); + if (!result.ok) expect(result.error._tag).toBe("HopperProtocolError"); }); it("starts a fresh session after close completes", async () => { diff --git a/tests/boundary/providers/hopper/hopperClient.startupFailures.test.ts b/tests/boundary/providers/hopper/hopperClient.startupFailures.test.ts index 391a7a47bd..46f9733f12 100644 --- a/tests/boundary/providers/hopper/hopperClient.startupFailures.test.ts +++ b/tests/boundary/providers/hopper/hopperClient.startupFailures.test.ts @@ -310,23 +310,6 @@ describe("HopperClient startup failures", () => { }); }); - it("records unknown when startup is cancelled before launch", async () => { - const launcher = new FixtureLauncher(); - const client = new HopperClient({ launcher, startupTimeoutMs: 30_000 }); - clients.push(client); - const controller = new AbortController(); - controller.abort(); - - const result = await client.start(controller.signal); - - expect(result.ok).toBe(false); - if (!result.ok) expect(result.error._tag).toBe("HopperCancelledError"); - expect(launcher.processes).toHaveLength(0); - // A cancelled startup must not report idle: the same-target fast path - // would otherwise treat the client as live and skip the retire path. - expect(client.operationHealth()).toMatchObject({ state: "unknown" }); - }); - it("records unknown when the health handshake is invalid", async () => { const launcher = new BadHealthLauncher(); const client = new HopperClient({ launcher, startupTimeoutMs: 30_000 }); diff --git a/tests/boundary/providers/hopper/hopperClient.startupWaiters.test.ts b/tests/boundary/providers/hopper/hopperClient.startupWaiters.test.ts new file mode 100644 index 0000000000..6c2fdb7c23 --- /dev/null +++ b/tests/boundary/providers/hopper/hopperClient.startupWaiters.test.ts @@ -0,0 +1,245 @@ +import { access } from "node:fs/promises"; + +import { expect, it, vi } from "vitest"; + +import { HopperClient } from "../../../../src/hopper/HopperClient.js"; +import { ok } from "../../../../src/domain/result.js"; +import { ProviderRunLineage } from "../../../../src/process/ProviderRunLineage.js"; +import { waitForExit } from "../../../support/process/processFixture.js"; +import { + HopperFixtureLauncher, + trackHopperClient, +} from "./hopperClient.fixture.js"; + +it("preserves a warm provider when an already cancelled caller joins", async () => { + const launcher = new HopperFixtureLauncher(); + const client = trackHopperClient(new HopperClient({ launcher })); + await expect(client.start()).resolves.toMatchObject({ ok: true }); + const controller = new AbortController(); + controller.abort(); + await expect(client.start(controller.signal)).resolves.toMatchObject({ + ok: false, + error: { _tag: "HopperCancelledError" }, + }); + expect(client.isReady()).toBe(true); + expect(client.operationHealth()).toMatchObject({ state: "idle" }); + await expect( + client.callTool("echo", { value: "live caller" }), + ).resolves.toEqual({ + ok: true, + value: { value: "live caller" }, + }); + expect(launcher.processes).toHaveLength(1); +}); + +it("reports provider closure to live startup callers without cancelling their signals", async () => { + const launcher = new HopperFixtureLauncher(undefined, "quiet", "gated"); + const client = trackHopperClient( + new HopperClient({ launcher, startupTimeoutMs: 10_000 }), + ); + const first = new AbortController(); + const second = new AbortController(); + const waits = [client.start(first.signal), client.start(second.signal)]; + await launcher.waitForRequest("health"); + await expect(client.close()).resolves.toEqual({ ok: true, value: null }); + for (const waiting of waits) + await expect(waiting).resolves.toMatchObject({ + ok: false, + error: { + _tag: "HopperProtocolError", + message: "Hopper session closed during startup", + }, + }); + expect(first.signal.aborted).toBe(false); + expect(second.signal.aborted).toBe(false); + expect(client.isReady()).toBe(false); +}); + +it("retains the final caller's cleanup failure and blocks replacement until close succeeds", async () => { + const fixture = new HopperFixtureLauncher(); + const controller = new AbortController(); + let allowCleanup = false; + const client = trackHopperClient( + new HopperClient({ + startupTimeoutMs: 10_000, + launcher: { + async launch(session) { + const launched = await fixture.launch(session); + if (!launched.ok) return launched; + if (fixture.processes.length === 1) controller.abort(); + return ok({ + ...launched.value, + cleanup: async () => + allowCleanup + ? launched.value.cleanup() + : { cleaned: false as const, reason: "Fixture cleanup denied" }, + }); + }, + }, + }), + ); + try { + const first = await client.start(controller.signal); + expect(first).toMatchObject({ + ok: false, + error: { cleanupIncomplete: true }, + }); + if (first.ok) throw new Error("Expected retained cleanup"); + const retry = await client.start(); + if (retry.ok) throw new Error("Unexpected replacement startup"); + expect(retry.error).toBe(first.error); + expect(fixture.processes).toHaveLength(1); + await expect(access(fixture.directories[0] ?? "")).resolves.toBeUndefined(); + allowCleanup = true; + await expect(client.close()).resolves.toEqual({ ok: true, value: null }); + await expect(access(fixture.directories[0] ?? "")).rejects.toMatchObject({ + code: "ENOENT", + }); + await expect(client.start()).resolves.toMatchObject({ ok: true }); + expect(fixture.processes).toHaveLength(2); + } finally { + allowCleanup = true; + await client.close(); + } +}); + +it.each([0, 1])( + "cancels Hopper startup caller %s independently of its peer", + async (cancelledIndex) => { + let entered: (() => void) | undefined; + let release: (() => void) | undefined; + const launchPending = new Promise((resolve) => { + entered = resolve; + }); + const launchGate = new Promise((resolve) => { + release = resolve; + }); + const launcher = new HopperFixtureLauncher(); + const client = trackHopperClient( + new HopperClient({ + startupTimeoutMs: 10_000, + launcher: { + async launch(session) { + entered?.(); + await launchGate; + return launcher.launch(session); + }, + }, + }), + ); + const controllers = [new AbortController(), new AbortController()]; + const waits = controllers.map(({ signal }) => + client.callTool("echo", { value: "live peer" }, { signal }), + ); + try { + await launchPending; + let cancelled: Awaited> | undefined; + void waits[cancelledIndex]?.then((result) => { + cancelled = result; + }); + controllers[cancelledIndex]?.abort(); + await vi.waitFor(() => + expect(cancelled).toMatchObject({ + ok: false, + error: { _tag: "HopperCancelledError" }, + }), + ); + expect(launcher.processes).toHaveLength(0); + release?.(); + await expect(waits[1 - cancelledIndex]).resolves.toMatchObject({ + ok: true, + value: { value: "live peer" }, + }); + expect(launcher.processes).toHaveLength(1); + expect(client.isReady()).toBe(true); + } finally { + release?.(); + await Promise.all(waits); + } + }, +); + +it.each(["cancelled", "rejected"] as const)( + "owns Hopper final startup cleanup when lineage is %s", + async (stage) => { + let entered: (() => void) | undefined; + let release: (() => void) | undefined; + const pending = new Promise((resolve) => { + entered = resolve; + }); + const gate = new Promise((resolve) => { + release = resolve; + }); + const cause = new TypeError("Lineage reader rejected"); + const original = ProviderRunLineage.prototype.observe; + const observe = vi.spyOn(ProviderRunLineage.prototype, "observe"); + observe.mockImplementationOnce(async function ( + this: ProviderRunLineage, + launch, + ) { + entered?.(); + await gate; + if (stage === "rejected") throw cause; + await original.call(this, launch); + }); + const launcher = new HopperFixtureLauncher(); + const client = trackHopperClient( + new HopperClient({ launcher, startupTimeoutMs: 10_000 }), + ); + const controller = new AbortController(); + const starting = client.start(controller.signal); + try { + await pending; + if (stage === "cancelled") controller.abort(); + release?.(); + const result = await starting; + expect(result).toMatchObject({ + ok: false, + error: { + _tag: + stage === "cancelled" ? "HopperCancelledError" : "HopperStartError", + }, + }); + if (result.ok) throw new Error("Expected interrupted startup"); + if (stage === "rejected") expect(result.error.cause).toBe(cause); + const child = launcher.processes[0]; + if (child === undefined) throw new Error("Missing fixture process"); + expect(await waitForExit(child, 5_000)).toBe(true); + await expect(access(launcher.directories[0] ?? "")).rejects.toMatchObject( + { code: "ENOENT" }, + ); + expect(client.isReady()).toBe(false); + await expect(client.start()).resolves.toMatchObject({ ok: true }); + expect(launcher.processes).toHaveLength(2); + } finally { + release?.(); + observe.mockRestore(); + await starting.catch(() => undefined); + } + }, +); + +it("keeps an unstarted client reusable after an already cancelled startup wait", async () => { + const launcher = new HopperFixtureLauncher(); + const client = trackHopperClient( + new HopperClient({ launcher, startupTimeoutMs: 30_000 }), + ); + const controller = new AbortController(); + controller.abort(); + + const result = await client.start(controller.signal); + + expect(result.ok).toBe(false); + if (!result.ok) expect(result.error._tag).toBe("HopperCancelledError"); + expect(launcher.processes).toHaveLength(0); + expect(client.isReady()).toBe(false); + expect(client.operationHealth()).toMatchObject({ state: "idle" }); + await expect( + client.callTool("echo", { value: "live caller" }), + ).resolves.toEqual({ + ok: true, + value: { value: "live caller" }, + }); + expect(launcher.processes).toHaveLength(1); + expect(client.isReady()).toBe(true); +}); diff --git a/tests/boundary/providers/startupCloseWaiters.test.ts b/tests/boundary/providers/startupCloseWaiters.test.ts new file mode 100644 index 0000000000..bf54b7fee9 --- /dev/null +++ b/tests/boundary/providers/startupCloseWaiters.test.ts @@ -0,0 +1,106 @@ +import { createHash } from "node:crypto"; +import { readFileSync } from "node:fs"; +import { access } from "node:fs/promises"; + +import { expect, it, vi } from "vitest"; + +import { GhidraClient } from "../../../src/ghidra/GhidraClient.js"; +import { HopperClient } from "../../../src/hopper/HopperClient.js"; +import { PrivateRuntimeRoot } from "../../../src/process/PrivateRuntimeRoot.js"; +import { createDeferred } from "../../fixtures/binarySession.js"; +import { + GhidraFixtureLauncher, + ghidraFixturePath, +} from "./ghidra/ghidraClient.fixture.js"; +import { HopperFixtureLauncher } from "./hopper/hopperClient.fixture.js"; + +const providers = [ + { + name: "Ghidra", + cancelled: { kind: "cancelled" }, + create() { + const fixture = new GhidraFixtureLauncher(); + const client = new GhidraClient({ + launcher: fixture, + targetPath: ghidraFixturePath, + targetSha256: createHash("sha256") + .update(readFileSync(ghidraFixturePath)) + .digest("hex"), + providerVersion: "12.1.4", + profileDigest: "a".repeat(64), + startupTimeoutMs: 10_000, + }); + return { + client, + roots: fixture.runtimeRoots, + children: fixture.processes, + }; + }, + }, + { + name: "Hopper", + cancelled: { _tag: "HopperCancelledError" }, + create() { + const fixture = new HopperFixtureLauncher(); + const client = new HopperClient({ + launcher: fixture, + startupTimeoutMs: 10_000, + }); + return { + client, + roots: fixture.directories, + children: fixture.processes, + }; + }, + }, +]; + +it.each(providers)( + "cancels a $name restart wait without taking ownership of pending close", + async ({ create, cancelled }) => { + const { client, roots, children } = create(); + const entered = createDeferred(); + const gate = createDeferred(); + const originalClose = PrivateRuntimeRoot.prototype.close; + const close = vi + .spyOn(PrivateRuntimeRoot.prototype, "close") + .mockImplementation(async function (this: PrivateRuntimeRoot) { + if (this.path === roots[0]) { + entered.resolve(); + await gate.promise; + } + await originalClose.call(this); + }); + let closing: ReturnType | undefined; + try { + await expect(client.start()).resolves.toMatchObject({ ok: true }); + closing = client.close(); + await entered.promise; + const controller = new AbortController(); + const retry = client.start(controller.signal); + let settled: Awaited | undefined; + void retry.then((result) => { + settled = result; + }); + controller.abort(); + await vi.waitFor(() => + expect(settled).toMatchObject({ ok: false, error: cancelled }), + ); + await expect(access(roots[0] ?? "")).resolves.toBeUndefined(); + expect(children).toHaveLength(1); + gate.resolve(); + await expect(closing).resolves.toMatchObject({ ok: true }); + await retry; + await expect(access(roots[0] ?? "")).rejects.toMatchObject({ + code: "ENOENT", + }); + await expect(client.start()).resolves.toMatchObject({ ok: true }); + expect(children).toHaveLength(2); + } finally { + gate.resolve(); + await closing; + await client.close(); + close.mockRestore(); + } + }, +);