mirror of
https://github.com/morluto/rea
synced 2026-10-25 22:08:11 -04:00
fix(providers): own shared startup and retained cleanup
Give concurrent Ghidra and Hopper callers independent waits over one owned startup. Join final-caller cleanup, preserve failed cleanup owners, and derive readiness from the current startup result and live transport. Distinguish provider closure from caller cancellation and clean up interrupted final lineage observation. Reuse the Ghidra socket fixture, replace stale cancellation health assertions with cold and warm reuse controls, and cover public MCP peer cancellation. Validation: 315 affected tests, build, typecheck, lint, docs, and 25 relocated startup controls passed. Controlled child bridges only; real Ghidra JVM and Hopper GUI verification remains unperformed. Refs #3983, #4227, #4274.
This commit is contained in:
morluto
@@ -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. |
|
||||
|
||||
@@ -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.
|
||||
|
||||
+113
-62
@@ -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<GhidraStartResult>;
|
||||
|
||||
/** Closed Java-bridge operation union callable after the exact handshake. */
|
||||
export type GhidraOperation =
|
||||
| GhidraInventoryOperation
|
||||
@@ -120,9 +127,7 @@ export class GhidraClient {
|
||||
#outputDrainPromise: Promise<boolean> | undefined;
|
||||
#outputDrainIncomplete = false;
|
||||
#lastDiagnostics: Readonly<Record<string, JsonValue>> = {};
|
||||
#startupController: AbortController | undefined;
|
||||
#startPromise: Promise<GhidraStartResult> | undefined;
|
||||
#ready = false;
|
||||
#startupAttempt: GhidraStartupAttempt | undefined;
|
||||
#closePromise: Promise<void> | undefined;
|
||||
readonly #failure = bindGhidraSessionFailure(
|
||||
() => this.#diagnostics(),
|
||||
@@ -220,9 +225,18 @@ export class GhidraClient {
|
||||
): Promise<GhidraStartResult> {
|
||||
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<GhidraStartResult> {
|
||||
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<GhidraStartResult> {
|
||||
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<void> {
|
||||
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<GhidraStartResult> {
|
||||
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<GhidraStartResult> {
|
||||
@@ -838,18 +886,15 @@ export class GhidraClient {
|
||||
}
|
||||
|
||||
async #close(
|
||||
starting: Promise<GhidraStartResult> | undefined,
|
||||
controller: AbortController | undefined,
|
||||
starting: GhidraStartupAttempt | undefined,
|
||||
forceStop: boolean,
|
||||
): Promise<void> {
|
||||
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, {
|
||||
|
||||
+89
-56
@@ -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<HopperServerInfo, AnalysisError>;
|
||||
type HopperStartupAttempt = SharedStartupAttempt<HopperStartupResult>;
|
||||
|
||||
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<Result<HopperServerInfo, AnalysisError>> | undefined;
|
||||
#startupAttempt: HopperStartupAttempt | undefined;
|
||||
#closePromise: Promise<Result<null, AnalysisError>> | 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<Result<HopperServerInfo, AnalysisError>> {
|
||||
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<HopperStartupResult> {
|
||||
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<HopperStartupResult> {
|
||||
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<HopperStartupResult> {
|
||||
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<Result<null, AnalysisError>> {
|
||||
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<Result<HopperServerInfo, AnalysisError>> | undefined,
|
||||
controller: AbortController | undefined,
|
||||
starting: HopperStartupAttempt | undefined,
|
||||
options: HopperClientCloseOptions,
|
||||
): Promise<Result<null, AnalysisError>> {
|
||||
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<Result<null, AnalysisError>> {
|
||||
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<HopperCancelledError | HopperTimeoutError> {
|
||||
): Promise<HopperCancelledError | HopperTimeoutError | HopperProtocolError> {
|
||||
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(
|
||||
|
||||
@@ -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<string>();
|
||||
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<string>();
|
||||
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<never>((_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);
|
||||
});
|
||||
@@ -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<Value> =
|
||||
| { readonly status: "completed"; readonly value: Value }
|
||||
| { readonly status: "cancelled" }
|
||||
| { readonly status: "stopped"; readonly value: Value };
|
||||
|
||||
type StartupSettlement<Value> =
|
||||
| { readonly status: "pending" }
|
||||
| { readonly status: "completed"; readonly value: Value }
|
||||
| { readonly status: "rejected" };
|
||||
|
||||
/** Own one startup operation while its callers wait independently. */
|
||||
export class SharedStartupAttempt<Value> {
|
||||
readonly #controller = new AbortController();
|
||||
readonly promise: Promise<Value>;
|
||||
#waiters = 0;
|
||||
#settlement: StartupSettlement<Value> = { status: "pending" };
|
||||
|
||||
constructor(start: (signal: AbortSignal) => Promise<Value>) {
|
||||
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<Value> {
|
||||
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<SharedStartupWait<Value>> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<void>();
|
||||
const admitted = [createDeferred<void>(), createDeferred<void>()];
|
||||
const gate = createDeferred<void>();
|
||||
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<ReturnType<Client["callTool"]>> }
|
||||
| { 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" });
|
||||
}
|
||||
},
|
||||
);
|
||||
@@ -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<ChildProcess>((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"),
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -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<ChildProcess>((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<void>((resolve) => {
|
||||
entered = resolve;
|
||||
});
|
||||
const gate = new Promise<void>((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<PrivateRuntimeRoot>();
|
||||
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<void>((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<void>((resolve) => {
|
||||
entered = resolve;
|
||||
});
|
||||
let releaseObservation: (() => void) | undefined;
|
||||
const observationGate = new Promise<void>((resolve) => {
|
||||
releaseObservation = resolve;
|
||||
const launchGate = new Promise<void>((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<ReturnType<GhidraClient["start"]>> | 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<void>((resolve) => {
|
||||
markObservationPending = resolve;
|
||||
});
|
||||
let releaseObservation: (() => void) | undefined;
|
||||
const observationGate = new Promise<void>((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", () => {
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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 });
|
||||
|
||||
@@ -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<void>((resolve) => {
|
||||
entered = resolve;
|
||||
});
|
||||
const launchGate = new Promise<void>((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<ReturnType<HopperClient["callTool"]>> | 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<void>((resolve) => {
|
||||
entered = resolve;
|
||||
});
|
||||
const gate = new Promise<void>((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);
|
||||
});
|
||||
@@ -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<void>();
|
||||
const gate = createDeferred<void>();
|
||||
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<typeof client.close> | 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<typeof retry> | 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();
|
||||
}
|
||||
},
|
||||
);
|
||||
Reference in New Issue
Block a user