From 5c8d5b2a8a25fa8cc4e9cf00f328a9c7bbe57330 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=B4=E4=BD=B3=E6=B5=A9Alben?= <35291372+XiaoBinGan@users.noreply.github.com> Date: Fri, 23 Oct 2026 02:23:16 +0800 Subject: [PATCH] fix(hopper): keep an observed launcher exit over a later cancellation (#5816) * fix(hopper): keep an observed launcher exit over a later cancellation `#connect` projected every interrupted wait as `#startupInterruption` without consulting the launcher the owner already held, so a launcher that exited 3 before the caller aborted was reported as `HopperCancelledError`. An exit or signal the owned process already holds is now projected as the physical startup failure first; a cancellation observed before any exit is unchanged, and successful exits keep their previous behaviour. Fixes #5429 * fix(hopper): arbitrate startup exits through process owner --------- Co-authored-by: XiaoBinGan Co-authored-by: morluto <76467478+morluto@users.noreply.github.com> --- src/hopper/HopperClient.ts | 124 ++++++++++++++---- .../hopperClient.startupWaiters.test.ts | 54 +++++++- .../hopperClientObservedExitPriority.test.ts | 121 +++++++++++++++++ 3 files changed, 273 insertions(+), 26 deletions(-) create mode 100644 tests/boundary/providers/hopper/hopperClientObservedExitPriority.test.ts diff --git a/src/hopper/HopperClient.ts b/src/hopper/HopperClient.ts index cbd99cf9b4..8159333b28 100644 --- a/src/hopper/HopperClient.ts +++ b/src/hopper/HopperClient.ts @@ -100,6 +100,13 @@ export interface HopperClientOptions { type HopperStartupResult = Result; type HopperStartupAttempt = SharedStartupAttempt; +type HopperStartupOutcome = + | { readonly kind: "interrupted" } + | { + readonly kind: "launcher-exit"; + readonly exitCode: number | null; + readonly ownsProviderLifetime: boolean; + }; type HopperClientCloseOptions = { readonly progress?: ProgressReporter; @@ -134,6 +141,8 @@ export class HopperClient { #closing = false; #launcherExitCode: number | null | undefined; #launcherFailureDiagnostic: HopperStartupFailureDiagnostic | undefined; + #startupOutcome: HopperStartupOutcome | undefined; + #startupSignal: AbortSignal | undefined; #operationFailure: OperationFailureRecord | undefined; #startupFailureResult: AnalysisError | undefined; #runId: string | undefined; @@ -394,12 +403,22 @@ export class HopperClient { this.#options.startupTimeoutMs, signal, ); + this.#startupOutcome = undefined; + this.#startupSignal = deadline.signal; + const onStartupAbort = (): void => { + if (this.#startupOutcome === undefined) + this.#startupOutcome = { kind: "interrupted" }; + }; + deadline.signal.addEventListener("abort", onStartupAbort, { once: true }); try { const result = await this.#startWithin(deadline); return result.ok && deadline.signal.aborted - ? this.#startupFailure(await this.#startupInterruption(deadline)) + ? this.#startupFailure(await this.#interruptionFailure(deadline)) : result; } finally { + deadline.signal.removeEventListener("abort", onStartupAbort); + if (this.#startupSignal === deadline.signal) + this.#startupSignal = undefined; deadline.dispose(); } } @@ -473,7 +492,7 @@ export class HopperClient { } const remainingMs = deadline.remainingMs(); if (remainingMs <= 0) { - return this.#startupFailure(await this.#startupInterruption(deadline)); + return this.#startupFailure(await this.#interruptionFailure(deadline)); } const health = await this.#request( "health", @@ -486,7 +505,7 @@ export class HopperClient { if (!health.ok) { return this.#startupFailure( deadline.signal.aborted || health.error instanceof HopperTimeoutError - ? await this.#startupInterruption(deadline) + ? await this.#interruptionFailure(deadline) : health.error, ); } @@ -497,12 +516,12 @@ export class HopperClient { } catch (cause: unknown) { return this.#startupFailure( deadline.signal.aborted - ? await this.#startupInterruption(deadline) + ? await this.#interruptionFailure(deadline) : new HopperStartError({ cause }), ); } if (deadline.signal.aborted) - return this.#startupFailure(await this.#startupInterruption(deadline)); + return this.#startupFailure(await this.#interruptionFailure(deadline)); return parsed; } @@ -788,31 +807,36 @@ export class HopperClient { deadline: ProviderStartupDeadline, ): Promise> { while (deadline.remainingMs() > 0) { - if (deadline.signal.aborted) - return err(await this.#startupInterruption(deadline)); - if (this.#closing) return err(new HopperProcessError(null)); - if ( - this.#launcherExitCode !== undefined && - this.#resources.launch !== undefined && - (this.#resources.launch.providerLifetime === "launcher-process" || - this.#launcherExitCode !== 0) - ) { - const failure = await this.#launcherStartupFailure( - this.#resources.launch.providerLifetime === "launcher-process", - this.#launcherExitCode, - deadline, + const observedExit = + this.#startupOutcome?.kind === "launcher-exit" + ? this.#startupOutcome + : this.#startupOutcome?.kind === "interrupted" + ? undefined + : this.#observedLauncherExit(); + // A launcher exit the owned process already holds is a physical startup + // failure; a cancellation that arrives afterwards must not replace it. + if (deadline.signal.aborted && observedExit === undefined) + return err( + this.#startupOutcome?.kind === "interrupted" + ? await this.#startupInterruption(deadline) + : await this.#interruptionFailure(deadline), + ); + if (this.#closing) return err(new HopperProcessError(null)); + if (observedExit !== undefined) + return err( + await this.#launcherStartupFailure( + observedExit.ownsProviderLifetime, + observedExit.exitCode, + deadline, + ), ); - return deadline.interruption === "cancelled" - ? err(await this.#startupInterruption(deadline)) - : err(failure); - } try { await access(socketPath); } catch (cause: unknown) { // best-effort cleanup: socket polling; absence means keep waiting. void cause; if ((await deadline.wait(50)) === "aborted") - return err(await this.#startupInterruption(deadline)); + return err(await this.#interruptionFailure(deadline)); continue; } const attempt = await connectHopperSocketOnce( @@ -825,9 +849,46 @@ export class HopperClient { return ok(undefined); } if ((await deadline.wait(50)) === "aborted") - return err(await this.#startupInterruption(deadline)); + return err(await this.#interruptionFailure(deadline)); } - return err(await this.#startupInterruption(deadline)); + return err(await this.#interruptionFailure(deadline)); + } + + /** A launcher exit the owned process already holds, when it is a failure. */ + #observedLauncherExit(): + | { + readonly exitCode: number | null; + readonly ownsProviderLifetime: boolean; + } + | undefined { + const launch = this.#resources.launch; + if (launch === undefined) return undefined; + const exit = this.#resources.processSupervisor?.observedExit(); + if (exit === undefined) return undefined; + const exitCode = exit.code; + const ownsProviderLifetime = launch.providerLifetime === "launcher-process"; + return ownsProviderLifetime || exitCode === null || exitCode !== 0 + ? { exitCode, ownsProviderLifetime } + : undefined; + } + + /** Project an interruption, keeping an exit the owner already observed. */ + async #interruptionFailure( + deadline: ProviderStartupDeadline, + ): Promise { + const observed = + this.#startupOutcome?.kind === "launcher-exit" + ? this.#startupOutcome + : this.#startupOutcome?.kind === "interrupted" + ? undefined + : this.#observedLauncherExit(); + return observed === undefined + ? this.#startupInterruption(deadline) + : this.#launcherStartupFailure( + observed.ownsProviderLifetime, + observed.exitCode, + deadline, + ); } async #startupInterruption( @@ -1070,6 +1131,19 @@ export class HopperClient { if (event.type !== "exit") return; this.#launcherExitCode = event.code; this.#launcherFailureDiagnostic = hopperLauncherFailureDiagnostic(event); + if ( + this.#startupSignal !== undefined && + this.#startupOutcome === undefined + ) { + const ownsProviderLifetime = + launch.providerLifetime === "launcher-process"; + if (ownsProviderLifetime || event.code === null || event.code !== 0) + this.#startupOutcome = { + kind: "launcher-exit", + exitCode: event.code, + ownsProviderLifetime, + }; + } this.#options.onDiagnostic?.({ type: "launcher-exit", code: event.code, diff --git a/tests/boundary/providers/hopper/hopperClient.startupWaiters.test.ts b/tests/boundary/providers/hopper/hopperClient.startupWaiters.test.ts index fd25fc14d8..4863cb5199 100644 --- a/tests/boundary/providers/hopper/hopperClient.startupWaiters.test.ts +++ b/tests/boundary/providers/hopper/hopperClient.startupWaiters.test.ts @@ -2,7 +2,10 @@ import { access } from "node:fs/promises"; import { expect, it, vi } from "vitest"; -import { HopperCancelledError } from "../../../../src/domain/hopperErrors.js"; +import { + HopperCancelledError, + HopperProcessError, +} from "../../../../src/domain/hopperErrors.js"; import { HopperClient } from "../../../../src/hopper/HopperClient.js"; import { ok } from "../../../../src/domain/result.js"; import { ProviderRunLineage } from "../../../../src/process/ProviderRunLineage.js"; @@ -227,6 +230,55 @@ it.each(["cancelled", "rejected"] as const)( }, ); +it("keeps a post-connect launcher exit ahead of cancellation during lineage finalization", 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"); + observe.mockImplementationOnce(async function ( + this: ProviderRunLineage, + launch, + ) { + entered?.(); + await gate; + await original.call(this, launch); + }); + const launcher = new HopperFixtureLauncher(); + const client = trackHopperClient(new HopperClient({ launcher })); + const controller = new AbortController(); + const starting = client.start(controller.signal); + try { + await pending; + const child = launcher.processes[0]; + if (child === undefined) throw new Error("Missing fixture process"); + child.kill("SIGKILL"); + expect(await waitForExit(child, 5_000)).toBe(true); + controller.abort({ source: "after launcher exit" }); + release?.(); + + const result = await starting; + expect(result).toMatchObject({ + ok: false, + error: { _tag: "HopperProcessError", exitCode: null }, + }); + if (result.ok) throw new Error("Expected launcher exit failure"); + expect(result.error).toBeInstanceOf(HopperProcessError); + expect((result.error as HopperProcessError).launcherFailure).toMatchObject({ + exit_code: null, + signal: "SIGKILL", + }); + } finally { + release?.(); + await starting; + } +}); + it("keeps an unstarted client reusable after an already cancelled startup wait", async () => { const launcher = new HopperFixtureLauncher(); const client = trackHopperClient( diff --git a/tests/boundary/providers/hopper/hopperClientObservedExitPriority.test.ts b/tests/boundary/providers/hopper/hopperClientObservedExitPriority.test.ts new file mode 100644 index 0000000000..bffe579bfc --- /dev/null +++ b/tests/boundary/providers/hopper/hopperClientObservedExitPriority.test.ts @@ -0,0 +1,121 @@ +import { spawn } from "node:child_process"; + +import { describe, expect, it } from "vitest"; + +import { + HopperCancelledError, + HopperProcessError, +} from "../../../../src/domain/hopperErrors.js"; +import { ok } from "../../../../src/domain/result.js"; +import type { + BridgeLauncher, + BridgeSession, +} from "../../../../src/hopper/BridgeLauncher.js"; +import { HopperClient } from "../../../../src/hopper/HopperClient.js"; + +/** A launcher process that exits 3 on its own, before any bridge readiness. */ +class ObservedExitLauncher implements BridgeLauncher { + #notified = false; + + constructor(readonly onExit: () => void) {} + + launch() { + const child = spawn(process.execPath, ["-e", "process.exit(3)"], { + stdio: ["ignore", "ignore", "pipe"], + }); + child.once("exit", () => { + if (this.#notified) return; + this.#notified = true; + setImmediate(() => this.onExit()); + }); + return Promise.resolve( + ok({ + process: child, + ownsProcessLifetime: true as const, + providerLifetime: "launcher-process" as const, + shutdownMode: "bridge-request" as const, + }), + ); + } +} + +class AbortBeforeExitLauncher implements BridgeLauncher { + constructor(readonly onSpawn: () => void) {} + + launch(_session: BridgeSession, options?: { readonly signal?: AbortSignal }) { + const child = spawn( + process.execPath, + ["-e", "setTimeout(() => process.exit(3), 5000)"], + { stdio: ["ignore", "ignore", "ignore"] }, + ); + options?.signal?.addEventListener("abort", () => child.kill("SIGTERM"), { + once: true, + }); + child.once("spawn", this.onSpawn); + return Promise.resolve( + ok({ + process: child, + ownsProcessLifetime: true as const, + providerLifetime: "launcher-process" as const, + shutdownMode: "bridge-request" as const, + }), + ); + } +} + +const start = async (abort: "never" | "after-exit") => { + const controller = new AbortController(); + const reason = { caller: "after-launcher-exit" }; + const client = new HopperClient({ + launcher: new ObservedExitLauncher(() => { + if (abort === "after-exit") controller.abort(reason); + }), + startupTimeoutMs: 5_000, + }); + try { + const started = await client.start( + abort === "after-exit" ? controller.signal : undefined, + ); + if (started.ok) + throw new Error("the launcher failure was reported as success"); + return started.error; + } finally { + await client.close().catch(() => undefined); + } +}; + +describe("Hopper startup failure priority", () => { + it("keeps the observed launcher exit when cancellation arrives after it", async () => { + const error = await start("after-exit"); + expect(error).toBeInstanceOf(HopperProcessError); + if (!(error instanceof HopperProcessError)) return; + expect(error.exitCode).toBe(3); + expect(error.stage).toBe("launch"); + }, 30_000); + + it("reports the same launcher exit without any cancellation", async () => { + const error = await start("never"); + expect(error._tag).toBe("HopperProcessError"); + }, 30_000); + + it("keeps cancellation first when its abort handler causes the later launcher exit", async () => { + const controller = new AbortController(); + const reason = { caller: "abort-before-exit" }; + const client = new HopperClient({ + launcher: new AbortBeforeExitLauncher(() => controller.abort(reason)), + startupTimeoutMs: 5_000, + }); + try { + const started = await client.start(controller.signal); + expect(started).toMatchObject({ + ok: false, + error: { _tag: "HopperCancelledError" }, + }); + if (started.ok) throw new Error("Expected startup cancellation"); + expect(started.error).toBeInstanceOf(HopperCancelledError); + expect(started.error.cause).toBe(reason); + } finally { + await client.close().catch(() => undefined); + } + }, 30_000); +});