mirror of
https://github.com/morluto/rea
synced 2026-10-26 06:16:17 -04:00
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 <XiaoBinGan@users.noreply.github.com> Co-authored-by: morluto <76467478+morluto@users.noreply.github.com>
This commit is contained in:
+99
-25
@@ -100,6 +100,13 @@ export interface HopperClientOptions {
|
||||
|
||||
type HopperStartupResult = Result<HopperServerInfo, AnalysisError>;
|
||||
type HopperStartupAttempt = SharedStartupAttempt<HopperStartupResult>;
|
||||
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<Result<undefined, HopperError>> {
|
||||
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<HopperError> {
|
||||
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,
|
||||
|
||||
@@ -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<void>((resolve) => {
|
||||
entered = resolve;
|
||||
});
|
||||
const gate = new Promise<void>((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(
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
Reference in New Issue
Block a user