Compare commits

...
Author SHA1 Message Date
morluto d81a0f02e5 fix(hopper): retain launcher exits observed before startup interruption
Record whether the current owned launcher exit was known when the startup deadline aborted. Reuse the existing launcher failure projection for exit-first ordering while preserving the exact caller cancellation reason when cancellation occurs first.

Validation: root focused run passed 38/40 tests including both new real-child ordering controls; two existing cleanup tests timed out and then passed unchanged in a targeted 13/13 rerun. Full check:ci passed. This uses inert Node children, not real Hopper. Keep this audit-branch candidate out of main pending required real-engine qualification.

Refs #5429
2026-10-21 11:34:42 +08:00
morluto bf79235da3 fix(hopper): retain caller cancellation causes across provider boundaries
Forward the exact abort reason through request, socket, startup and launcher conversion. Preserve argument omission separately from explicit undefined, remove the queue cancellation default that could silently discard a cause, and retain the caller cancellation as primary through cleanup failure.

Validation: the unchanged queue production boundary failed both new cause assertions; the compiled queue changed from same_reason:false to true. Full check:ci and build passed; 92 tests passed and 5 Linux-only cases skipped. Real Hopper qualification remains pending on this audit branch.

Closes #5714
2026-10-21 07:48:55 +08:00
morluto 82863af551 fix(hopper): bound orderly Linux demo helper shutdown
Send owner stdin EOF once and retain a single monotonic grace deadline across cleanup retries before verified process-group cleanup. Skip the grace when the helper has already exited, including signal exits. Adapted from PR #4966 with retry ownership and finalization coverage.

Validation: full check:ci and build passed; combined provider-boundary/source suite passed 92 tests, with 5 Linux-only cases skipped on macOS. Real Linux Hopper verification is pending; this candidate is held on an audit branch rather than integrated into main.

Refs #4921
2026-10-21 07:48:55 +08:00
12 changed files with 888 additions and 191 deletions

No files matched your search

+1 -1
View File
@@ -66,7 +66,7 @@ export class HopperCancelledError extends HopperError {
) {
super(
"Hopper operation was cancelled",
cause === undefined ? options : { ...options, cause },
arguments.length === 0 ? options : { ...options, cause },
);
}
}
+22 -7
View File
@@ -18,6 +18,10 @@ import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js";
import { AnalysisCleanupLeaseGroup } from "../artifacts/AnalysisCleanupLease.js";
import { AnalysisError } from "../domain/analysisErrorBase.js";
import { cleanupOwnedProcessGroup } from "../process/ProcessOwnership.js";
import {
createLinuxDemoCleanup,
LINUX_DEMO_ORDERLY_SHUTDOWN_GRACE_MS,
} from "./LinuxDemoCleanup.js";
import { execFileOutput } from "../process/ExecFileOutput.js";
import {
type ProviderProcessLaunch,
@@ -147,6 +151,7 @@ export interface HopperApplicationLauncherDependencies {
readonly helperPath: string;
readonly signal?: AbortSignal;
}) => Promise<LinuxPrivateDisplaySelection>;
readonly demoOrderlyShutdownGraceMs?: number;
}
/**
@@ -331,7 +336,8 @@ export class HopperApplicationLauncher implements BridgeLauncher {
];
let launched = false;
try {
if (signal?.aborted === true) return err(new HopperCancelledError());
if (signal?.aborted === true)
return err(new HopperCancelledError(signal.reason));
let expectedLinuxSha256: string | undefined;
if (usesLinuxDemo(this.options)) {
const inspected = await (
@@ -339,7 +345,7 @@ export class HopperApplicationLauncher implements BridgeLauncher {
)(this.options.launcherPath, this.options.environment, signal);
if (!inspected.ok)
return inspected.error.reason === "cancelled"
? err(new HopperCancelledError())
? err(new HopperCancelledError(signal?.reason))
: err(
new HopperStartError({
userMessage: inspected.error.message,
@@ -366,7 +372,7 @@ export class HopperApplicationLauncher implements BridgeLauncher {
this.options.environment,
signal,
);
if (!prepared) return err(new HopperCancelledError());
if (!prepared) return err(new HopperCancelledError(signal?.reason));
const linuxDemo = linuxDemoLaunch(
this.options,
session,
@@ -383,9 +389,18 @@ export class HopperApplicationLauncher implements BridgeLauncher {
linuxDemo,
signal,
});
if (started === null) return err(new HopperCancelledError());
if (started === null)
return err(new HopperCancelledError(signal?.reason));
launched = true;
const cleanup = () => cleanupOwnedProcessGroup(started.ownership);
const cleanup =
linuxDemo === undefined
? () => cleanupOwnedProcessGroup(started.ownership)
: createLinuxDemoCleanup(
started.process,
started.ownership,
this.dependencies.demoOrderlyShutdownGraceMs ??
LINUX_DEMO_ORDERLY_SHUTDOWN_GRACE_MS,
);
const ownedLauncher = {
...(this.options.preparedImage === undefined
? {}
@@ -518,7 +533,7 @@ export class HopperApplicationLauncher implements BridgeLauncher {
...(signal === undefined ? {} : { signal }),
});
if (signal?.aborted === true)
return { ok: false, error: new HopperCancelledError() };
return { ok: false, error: new HopperCancelledError(signal.reason) };
return selection.ok
? { ok: true, strategy: selection.strategy }
: {
@@ -642,7 +657,7 @@ const hopperLaunchFailure = (
signal: AbortSignal | undefined,
): HopperCancelledError | HopperStartError =>
signal?.aborted === true
? new HopperCancelledError()
? new HopperCancelledError(signal.reason)
: new HopperStartError({ cause });
/** Select the publisher-verified demo adapter from the caller's explicit launch mode. */
+33 -8
View File
@@ -132,6 +132,7 @@ export class HopperClient {
#nextId = 1;
#closing = false;
#launcherExitCode: number | null | undefined;
#launcherExitPrecededInterruption = false;
#launcherFailureDiagnostic: HopperStartupFailureDiagnostic | undefined;
#operationFailure: OperationFailureRecord | undefined;
#startupFailureResult: AnalysisError | undefined;
@@ -313,11 +314,11 @@ export class HopperClient {
/** 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()));
return Promise.resolve(err(new HopperCancelledError(signal?.reason)));
if (this.#closePromise !== undefined)
return waitForAbortable(this.#closePromise, signal).then((closed) =>
closed === ABORTED
? err(new HopperCancelledError())
? err(new HopperCancelledError(signal?.reason))
: closed.ok
? this.start(signal)
: closed,
@@ -355,7 +356,7 @@ export class HopperClient {
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());
return err(new HopperCancelledError(signal?.reason));
}
/** Cleanup belongs to the abandoned attempt, never to a prospective retry. */
@@ -365,7 +366,7 @@ export class HopperClient {
): Promise<HopperStartupResult> {
const waited = await waitForAbortable(attempt.promise, signal);
if (waited === ABORTED || (waited.ok && isAborted(signal)))
return err(new HopperCancelledError());
return err(new HopperCancelledError(signal?.reason));
if (waited.ok || waited.error.cleanupIncomplete) return waited;
return this.start(signal);
}
@@ -373,7 +374,7 @@ export class HopperClient {
async #start(
signal?: AbortSignal,
): Promise<Result<HopperServerInfo, AnalysisError>> {
if (isAborted(signal)) return err(new HopperCancelledError());
if (isAborted(signal)) return err(new HopperCancelledError(signal?.reason));
if (
this.#socket !== undefined ||
this.#resources.launch !== undefined ||
@@ -393,12 +394,23 @@ export class HopperClient {
this.#options.startupTimeoutMs,
signal,
);
this.#launcherExitPrecededInterruption = false;
const recordObservedExitOrder = (): void => {
this.#launcherExitPrecededInterruption =
this.#resources.launch !== undefined &&
this.#launcherExitCode !== undefined;
};
deadline.signal.addEventListener("abort", recordObservedExitOrder, {
once: true,
});
if (deadline.signal.aborted) recordObservedExitOrder();
try {
const result = await this.#startWithin(deadline);
return result.ok && deadline.signal.aborted
? this.#startupFailure(await this.#startupInterruption(deadline))
: result;
} finally {
deadline.signal.removeEventListener("abort", recordObservedExitOrder);
deadline.dispose();
}
}
@@ -831,11 +843,24 @@ export class HopperClient {
async #startupInterruption(
deadline: ProviderStartupDeadline,
): Promise<HopperCancelledError | HopperTimeoutError | HopperProtocolError> {
): Promise<HopperError> {
const launch = this.#resources.launch;
const exitCode = this.#launcherExitCode;
if (
this.#launcherExitPrecededInterruption &&
launch !== undefined &&
exitCode !== undefined &&
(launch.providerLifetime === "launcher-process" || exitCode !== 0)
)
return this.#launcherStartupFailure(
launch.providerLifetime === "launcher-process",
exitCode,
deadline,
);
if (deadline.signal.reason === STARTUP_OWNER_CLOSED)
return new HopperProtocolError("Hopper session closed during startup");
if (deadline.interruption === "cancelled")
return new HopperCancelledError();
return new HopperCancelledError(deadline.signal.reason);
return new HopperTimeoutError(
deadline.timeoutMs,
undefined,
@@ -969,7 +994,7 @@ export class HopperClient {
);
}
if (options.signal?.aborted === true)
return err(new HopperCancelledError());
return err(new HopperCancelledError(options.signal.reason));
const startedAt = performance.now();
const result = await this.#requests.run(id, method, params, {
...(options.signal === undefined ? {} : { signal: options.signal }),
+157 -142
View File
@@ -12,159 +12,174 @@ afterEach(() => {
vi.restoreAllMocks();
});
describe("HopperRequestQueue deadlines", () => {
it("times out active callers while retaining the wire until the late reply", async () => {
vi.useFakeTimers();
const sent: number[] = [];
const queue = new HopperRequestQueue(({ id }) => sent.push(id));
const result = vi.fn();
const pending = queue
.run(1, "analyze_function", {}, { timeoutMs: 25 })
.then(result);
const next = queue.run(2, "echo", {}, {});
await vi.advanceTimersByTimeAsync(25);
expect(result).toHaveBeenCalledWith({
it("preserves explicit undefined as a cancellation cause", async () => {
const noCause = new HopperCancelledError();
const explicitUndefined = new HopperCancelledError(undefined);
expect(Object.hasOwn(noCause, "cause")).toBe(false);
expect(Object.hasOwn(explicitUndefined, "cause")).toBe(true);
expect(explicitUndefined.cause).toBeUndefined();
const controller = new AbortController();
const queue = new HopperRequestQueue(() => {});
controller.abort(undefined);
const result = await queue.run(1, "echo", {}, { signal: controller.signal });
expect(result.ok).toBe(false);
if (!result.ok) expect(result.error.cause).toBe(controller.signal.reason);
});
it("times out active callers while retaining the wire until the late reply", async () => {
vi.useFakeTimers();
const sent: number[] = [];
const queue = new HopperRequestQueue(({ id }) => sent.push(id));
const result = vi.fn();
const pending = queue
.run(1, "analyze_function", {}, { timeoutMs: 25 })
.then(result);
const next = queue.run(2, "echo", {}, {});
await vi.advanceTimersByTimeAsync(25);
expect(result).toHaveBeenCalledWith({
ok: false,
error: expect.objectContaining({
_tag: "HopperTimeoutError",
timeoutMs: 25,
operation: "analyze_function",
requestId: 1,
providerState: "busy",
}),
});
expect(sent).toEqual([1]);
expect(queue.activity()).toMatchObject({
callerState: "cancelled",
queuedRequests: 1,
});
expect(vi.getTimerCount()).toBe(0);
expect(queue.accept(1, ok("late"))).toBe(true);
expect(sent).toEqual([1, 2]);
queue.accept(2, ok("next"));
await pending;
await expect(next).resolves.toEqual(ok("next"));
expect(result).toHaveBeenCalledTimes(1);
expect(queue.activity()).toBeNull();
expect(vi.getTimerCount()).toBe(0);
});
it("expires a queued request without transmitting it", async () => {
vi.useFakeTimers();
const sent: number[] = [];
const queue = new HopperRequestQueue(({ id }) => sent.push(id));
const active = queue.run(1, "active", {}, {});
const settled = vi.fn();
const queued = queue.run(2, "queued", {}, { timeoutMs: 10 }).then(settled);
await vi.advanceTimersByTimeAsync(10);
expect(settled).toHaveBeenCalledWith(
expect.objectContaining({
ok: false,
error: expect.objectContaining({
_tag: "HopperTimeoutError",
timeoutMs: 25,
operation: "analyze_function",
requestId: 1,
providerState: "busy",
requestId: 2,
providerState: "not_started",
}),
});
expect(sent).toEqual([1]);
expect(queue.activity()).toMatchObject({
callerState: "cancelled",
queuedRequests: 1,
});
expect(vi.getTimerCount()).toBe(0);
expect(queue.accept(1, ok("late"))).toBe(true);
expect(sent).toEqual([1, 2]);
queue.accept(2, ok("next"));
await pending;
await expect(next).resolves.toEqual(ok("next"));
expect(result).toHaveBeenCalledTimes(1);
expect(queue.activity()).toBeNull();
expect(vi.getTimerCount()).toBe(0);
});
}),
);
expect(queue.hasQueued(2)).toBe(false);
queue.accept(1, ok(null));
await Promise.all([active, queued]);
expect(sent).toEqual([1]);
expect(vi.getTimerCount()).toBe(0);
});
it("expires a queued request without transmitting it", async () => {
vi.useFakeTimers();
const sent: number[] = [];
const queue = new HopperRequestQueue(({ id }) => sent.push(id));
const active = queue.run(1, "active", {}, {});
const settled = vi.fn();
const queued = queue.run(2, "queued", {}, { timeoutMs: 10 }).then(settled);
await vi.advanceTimersByTimeAsync(10);
expect(settled).toHaveBeenCalledWith(
expect.objectContaining({
ok: false,
error: expect.objectContaining({
_tag: "HopperTimeoutError",
requestId: 2,
providerState: "not_started",
}),
}),
);
expect(queue.hasQueued(2)).toBe(false);
queue.accept(1, ok(null));
await Promise.all([active, queued]);
expect(sent).toEqual([1]);
expect(vi.getTimerCount()).toBe(0);
});
it("cleans deadline and abort observers on success before the deadline", async () => {
vi.useFakeTimers();
const queue = new HopperRequestQueue(() => {});
const controller = new AbortController();
const remove = vi.spyOn(controller.signal, "removeEventListener");
const settled = vi.fn();
const pending = queue
.run(1, "echo", {}, { timeoutMs: 25, signal: controller.signal })
.then(settled);
queue.accept(1, ok(42));
controller.abort();
await vi.advanceTimersByTimeAsync(100);
await pending;
expect(settled).toHaveBeenCalledExactlyOnceWith(ok(42));
expect(remove).toHaveBeenCalledWith("abort", expect.any(Function));
expect(vi.getTimerCount()).toBe(0);
});
it("cleans deadline and abort observers on success before the deadline", async () => {
vi.useFakeTimers();
const queue = new HopperRequestQueue(() => {});
const controller = new AbortController();
const remove = vi.spyOn(controller.signal, "removeEventListener");
const settled = vi.fn();
const pending = queue
.run(1, "echo", {}, { timeoutMs: 25, signal: controller.signal })
.then(settled);
queue.accept(1, ok(42));
controller.abort();
await vi.advanceTimersByTimeAsync(100);
await pending;
expect(settled).toHaveBeenCalledExactlyOnceWith(ok(42));
expect(remove).toHaveBeenCalledWith("abort", expect.any(Function));
expect(vi.getTimerCount()).toBe(0);
});
it("keeps explicit cancellation distinct from a later timeout", async () => {
vi.useFakeTimers();
const queue = new HopperRequestQueue(() => {});
const controller = new AbortController();
const reason = { source: "caller", opaque: 7 };
const pending = queue.run(
1,
"echo",
{},
{ timeoutMs: 25, signal: controller.signal },
);
controller.abort(reason);
const result = await pending;
expect(result.ok).toBe(false);
if (result.ok) throw new Error("expected cancellation");
expect(result.error).toBeInstanceOf(HopperCancelledError);
expect(result.error.cause).toBe(reason);
expect(vi.getTimerCount()).toBe(0);
queue.accept(1, ok(null));
await vi.advanceTimersByTimeAsync(100);
expect(queue.activity()).toBeNull();
});
it("keeps explicit cancellation distinct from a later timeout", async () => {
vi.useFakeTimers();
const queue = new HopperRequestQueue(() => {});
const controller = new AbortController();
const pending = queue.run(
1,
"echo",
{},
{ timeoutMs: 25, signal: controller.signal },
);
controller.abort();
await expect(pending).resolves.toEqual({
ok: false,
error: new HopperCancelledError(),
});
expect(vi.getTimerCount()).toBe(0);
queue.accept(1, ok(null));
await vi.advanceTimersByTimeAsync(100);
expect(queue.activity()).toBeNull();
});
it("does not recreate timers when a progress observer cancels synchronously", async () => {
vi.useFakeTimers();
const send = vi.fn();
const queue = new HopperRequestQueue(send);
const controller = new AbortController();
const pending = queue.run(
1,
"echo",
{},
{
timeoutMs: 100,
signal: controller.signal,
progress: {
report: () => {
controller.abort();
return Promise.resolve();
},
it("does not recreate timers when a progress observer cancels synchronously", async () => {
vi.useFakeTimers();
const send = vi.fn();
const queue = new HopperRequestQueue(send);
const controller = new AbortController();
const pending = queue.run(
1,
"echo",
{},
{
timeoutMs: 100,
signal: controller.signal,
progress: {
report: () => {
controller.abort();
return Promise.resolve();
},
},
);
await expect(pending).resolves.toMatchObject({
ok: false,
error: { _tag: "HopperCancelledError" },
});
expect(vi.getTimerCount()).toBe(0);
expect(send).not.toHaveBeenCalled();
expect(queue.activity()).toBeNull();
const next = queue.run(2, "next", {}, {});
expect(send).toHaveBeenCalledExactlyOnceWith(
{ id: 2, method: "next", params: {} },
expect.any(Function),
);
queue.accept(2, ok(null));
await next;
},
);
await expect(pending).resolves.toMatchObject({
ok: false,
error: { _tag: "HopperCancelledError" },
});
expect(vi.getTimerCount()).toBe(0);
expect(send).not.toHaveBeenCalled();
expect(queue.activity()).toBeNull();
const next = queue.run(2, "next", {}, {});
expect(send).toHaveBeenCalledExactlyOnceWith(
{ id: 2, method: "next", params: {} },
expect.any(Function),
);
queue.accept(2, ok(null));
await next;
});
it("cleans all deadlines on teardown, including after an active timeout", async () => {
vi.useFakeTimers();
const queue = new HopperRequestQueue(() => {});
const result = vi.fn();
const active = queue.run(1, "active", {}, { timeoutMs: 10 }).then(result);
const queued = queue.run(2, "queued", {}, { timeoutMs: 100 });
await vi.advanceTimersByTimeAsync(10);
const failure = new HopperProtocolError("closed");
queue.failAll(failure);
await active;
await expect(queued).resolves.toEqual({ ok: false, error: failure });
expect(result).toHaveBeenCalledTimes(1);
expect(queue.activity()).toBeNull();
expect(queue.queuedCount()).toBe(0);
expect(vi.getTimerCount()).toBe(0);
});
it("cleans all deadlines on teardown, including after an active timeout", async () => {
vi.useFakeTimers();
const queue = new HopperRequestQueue(() => {});
const result = vi.fn();
const active = queue.run(1, "active", {}, { timeoutMs: 10 }).then(result);
const queued = queue.run(2, "queued", {}, { timeoutMs: 100 });
await vi.advanceTimersByTimeAsync(10);
const failure = new HopperProtocolError("closed");
queue.failAll(failure);
await active;
await expect(queued).resolves.toEqual({ ok: false, error: failure });
expect(result).toHaveBeenCalledTimes(1);
expect(queue.activity()).toBeNull();
expect(queue.queuedCount()).toBe(0);
expect(vi.getTimerCount()).toBe(0);
});
describe("HopperRequestQueue deadline admission and races", () => {
+16 -7
View File
@@ -77,7 +77,9 @@ export class HopperRequestQueue {
options: HopperRequestQueueOptions,
): Promise<HopperRequestResult> {
if (options.signal?.aborted === true)
return Promise.resolve(err(new HopperCancelledError()));
return Promise.resolve(
err(new HopperCancelledError(options.signal.reason)),
);
const timeoutMs = options.timeoutMs;
if (
timeoutMs !== undefined &&
@@ -107,7 +109,13 @@ export class HopperRequestQueue {
return new Promise((resolve) => {
let entry: QueuedRequest;
const onAbort =
options.signal === undefined ? undefined : () => this.#cancel(entry);
options.signal === undefined
? undefined
: () =>
this.#cancel(
entry,
new HopperCancelledError(options.signal?.reason),
);
entry = {
id,
method,
@@ -239,7 +247,11 @@ export class HopperRequestQueue {
return;
}
if (entry.signal?.aborted === true) {
this.#settleCaller(entry, err(new HopperCancelledError()), "cancelled");
this.#settleCaller(
entry,
err(new HopperCancelledError(entry.signal.reason)),
"cancelled",
);
this.#drain();
return;
}
@@ -314,10 +326,7 @@ export class HopperRequestQueue {
return entry.callerSettled;
}
#cancel(
entry: QueuedRequest,
error: HopperError = new HopperCancelledError(),
): void {
#cancel(entry: QueuedRequest, error: HopperError): void {
if (entry.callerSettled) return;
if (this.#active === entry) {
this.#settleCaller(entry, err(error), "cancelled");
+1 -1
View File
@@ -30,7 +30,7 @@ export const connectHopperSocketOnce = (
const onAbort = (): void => {
detach();
socket.destroy();
resolve(err(new HopperCancelledError()));
resolve(err(new HopperCancelledError(signal.reason)));
};
socket.once("connect", onConnect);
socket.once("error", onError);
+70
View File
@@ -0,0 +1,70 @@
import { EventEmitter } from "node:events";
import { PassThrough } from "node:stream";
import { expect, it, vi } from "vitest";
import type { OwnedProcessGroup } from "../process/ProcessOwnership.js";
import type { ProviderProcessHandle } from "../process/ProviderProcess.js";
import { createLinuxDemoCleanup } from "./LinuxDemoCleanup.js";
class FixtureHelper extends EventEmitter implements ProviderProcessHandle {
readonly stdout = new PassThrough();
readonly stderr = new PassThrough();
readonly stdin = new PassThrough();
readonly pid = 901;
exitCode: number | null = null;
signalCode: NodeJS.Signals | null = null;
kill(): boolean {
return true;
}
}
const fixtureOwnership = {
runId: "linux-demo-cleanup-test",
leaderPid: 901,
processGroupId: 901,
} satisfies OwnedProcessGroup;
it("sends stdin EOF once and does not restart the grace period on cleanup retry", async () => {
const helper = new FixtureHelper();
const end = vi.spyOn(helper.stdin, "end");
let now = 0;
const waits: number[] = [];
const groupCleanup = vi
.fn()
.mockResolvedValueOnce({ cleaned: false, reason: "retry required" })
.mockResolvedValueOnce({ cleaned: true, signaled: true });
const cleanup = createLinuxDemoCleanup(helper, fixtureOwnership, 100, {
cleanupGroup: groupCleanup,
now: () => now,
waitForExit: async (_helper, timeoutMs) => {
waits.push(timeoutMs);
},
});
await expect(cleanup()).resolves.toMatchObject({ cleaned: false });
now = 100;
await expect(cleanup()).resolves.toMatchObject({ cleaned: true });
expect(waits).toEqual([100, 0]);
expect(end).toHaveBeenCalledOnce();
expect(groupCleanup).toHaveBeenCalledTimes(2);
});
it("does not wait for orderly shutdown after the helper has already exited", async () => {
const helper = new FixtureHelper();
helper.exitCode = 0;
helper.emit("exit", 0, null);
const end = vi.spyOn(helper.stdin, "end");
const groupCleanup = vi
.fn()
.mockResolvedValue({ cleaned: true, signaled: false });
const cleanup = createLinuxDemoCleanup(helper, fixtureOwnership, 10_000, {
cleanupGroup: groupCleanup,
});
await expect(cleanup()).resolves.toMatchObject({ cleaned: true });
expect(end).not.toHaveBeenCalled();
expect(groupCleanup).toHaveBeenCalledOnce();
});
+77
View File
@@ -0,0 +1,77 @@
import type {
OwnedProcessGroup,
ProcessCleanupResult,
} from "../process/ProcessOwnership.js";
import type { ProviderProcessHandle } from "../process/ProviderProcess.js";
import { cleanupOwnedProcessGroup } from "../process/ProcessOwnership.js";
export const LINUX_DEMO_ORDERLY_SHUTDOWN_GRACE_MS = 10_000;
interface LinuxDemoCleanupOptions {
readonly cleanupGroup?: (
ownership: OwnedProcessGroup,
) => Promise<ProcessCleanupResult>;
readonly now?: () => number;
readonly waitForExit?: typeof waitForHelperExit;
}
/** Keep one orderly-shutdown request and deadline across supervisor retries. */
export const createLinuxDemoCleanup = (
helper: ProviderProcessHandle,
ownership: OwnedProcessGroup,
orderlyShutdownGraceMs: number,
options: LinuxDemoCleanupOptions = {},
): (() => Promise<ProcessCleanupResult>) => {
const cleanupGroup = options.cleanupGroup ?? cleanupOwnedProcessGroup;
const now = options.now ?? (() => performance.now());
const waitForExit = options.waitForExit ?? waitForHelperExit;
let deadline: number | undefined;
let orderlyShutdownRequested = false;
return async () => {
if (deadline === undefined) {
deadline = now() + Math.max(0, orderlyShutdownGraceMs);
const stdin = helper.stdin;
if (
helper.exitCode === null &&
helper.signalCode === null &&
stdin !== undefined &&
stdin !== null
) {
orderlyShutdownRequested = true;
try {
stdin.end();
} catch {
// Process-group cleanup below remains the verified fallback.
}
}
}
if (
orderlyShutdownRequested &&
helper.exitCode === null &&
helper.signalCode === null
)
await waitForExit(helper, Math.max(0, deadline - now()));
return cleanupGroup(ownership);
};
};
const waitForHelperExit = (
helper: ProviderProcessHandle,
timeoutMs: number,
): Promise<void> =>
new Promise((resolve) => {
if (helper.exitCode !== null || helper.signalCode !== null)
return resolve();
const timer = setTimeout(() => {
helper.off("exit", onExit);
resolve();
}, timeoutMs);
const onExit = () => {
clearTimeout(timer);
resolve();
};
helper.once("exit", onExit);
});
@@ -66,6 +66,94 @@ const options = (launcherPath: string): HopperApplicationLauncherOptions => ({
demoHelperPath: "/rea/hopper-demo-x11.py",
});
const launchFixtureProcess = async (
launcher: HopperApplicationLauncher,
directory: string,
) => {
const result = await launcher.launch({
directory,
socketPath: join(directory, "bridge.sock"),
token: "fixture-token",
runId: randomUUID(),
});
if (!result.ok) throw result.error;
return {
launched: result.value,
supervisor: new ProviderProcessSupervisor(result.value),
};
};
const verifyFixtureLinuxLauncher = async () =>
ok({
sha256: verifiedSha256,
version: "fixture",
primaryFingerprint: "0".repeat(40),
signingFingerprint: "1".repeat(40),
source: "signed_package_cache" as const,
});
const releaseFixtureProcess = async (
supervisor: ProviderProcessSupervisor,
launched: Awaited<ReturnType<typeof launchFixtureProcess>>["launched"],
) => {
await supervisor.stop();
await launched.releaseLease?.();
};
const stopWithinFixtureDeadline = async (
supervisor: ProviderProcessSupervisor,
) => {
const started = Date.now();
expect(await supervisor.stop()).toMatchObject({ status: "verified-cleanup" });
expect(Date.now() - started).toBeLessThan(5_000);
};
type OrderlyHelperMode = "orderly" | "ignore-shutdown" | "signal-death";
const orderlyHelperLauncher = (directory: string, mode: OrderlyHelperMode) =>
new HopperApplicationLauncher(
{
...options(process.execPath),
launchMode: "verified_linux_demo",
environment: {
REA_FIXTURE_LOCK_DIR: directory,
...(mode === "ignore-shutdown"
? { REA_FIXTURE_IGNORE_SHUTDOWN: "1" }
: {}),
...(mode === "signal-death" ? { REA_FIXTURE_SIGNAL_DEATH: "1" } : {}),
},
demoHelperPath: fileURLToPath(
new URL("../../../fixtures/hopperOrderlyShutdown.py", import.meta.url),
),
},
{
verifyLinuxLauncher: verifyFixtureLinuxLauncher,
selectPrivateDisplay: readyDirectDisplay,
...(mode === "ignore-shutdown"
? { demoOrderlyShutdownGraceMs: 500 }
: {}),
},
);
const waitForFixtureFile = async (path: string, missingMessage: string) => {
const deadline = Date.now() + 5_000;
for (;;) {
try {
await access(path);
return;
} catch (cause) {
if (
!(cause instanceof Error) ||
!("code" in cause) ||
cause.code !== "ENOENT"
)
throw cause;
if (Date.now() >= deadline) throw new Error(missingMessage);
await new Promise((resolve) => setTimeout(resolve, 50));
}
}
};
it("resolves mixed image and publisher cleanup only after each owner releases", async () => {
const directory = await createTestTempDirectory("rea-hopper-mixed-cleanup-");
const targetPath = join(directory, "target.macho");
@@ -436,14 +524,10 @@ it("launches Hopper helpers with the caller-selected environment without restori
bridgeScriptPath: "/rea/unused-bridge.py",
launchMode: "native",
});
const launched = await launcher.launch({
const { launched, supervisor } = await launchFixtureProcess(
launcher,
directory,
socketPath: join(directory, "bridge.sock"),
token: "fixture-token",
runId: randomUUID(),
});
if (!launched.ok) throw launched.error;
const supervisor = new ProviderProcessSupervisor(launched.value);
);
try {
expect(await supervisor.waitForOutputClose(5_000)).toBe(true);
expect(JSON.parse(supervisor.snapshot().stdout.text)).toEqual({
@@ -455,7 +539,7 @@ it("launches Hopper helpers with the caller-selected environment without restori
});
} finally {
await supervisor.stop();
await launched.value.releaseLease?.();
await launched.releaseLease?.();
}
});
@@ -473,26 +557,15 @@ it.skipIf(process.platform !== "linux")(
),
},
{
verifyLinuxLauncher: async () =>
ok({
sha256: verifiedSha256,
version: "fixture",
primaryFingerprint: "0".repeat(40),
signingFingerprint: "1".repeat(40),
source: "signed_package_cache",
}),
verifyLinuxLauncher: verifyFixtureLinuxLauncher,
selectPrivateDisplay,
},
);
const launched = await launcher.launch({
const { launched, supervisor } = await launchFixtureProcess(
launcher,
directory,
socketPath: join(directory, "bridge.sock"),
token: "fixture-token",
runId: randomUUID(),
});
if (!launched.ok) throw launched.error;
);
expect(selectPrivateDisplay).toHaveBeenCalledOnce();
const supervisor = new ProviderProcessSupervisor(launched.value);
try {
expect(await supervisor.waitForOutputClose(5_000)).toBe(true);
expect(supervisor.snapshot().exitCode).toBe(0);
@@ -500,8 +573,7 @@ it.skipIf(process.platform !== "linux")(
owner_alive: true,
});
} finally {
await supervisor.stop();
await launched.value.releaseLease?.();
await releaseFixtureProcess(supervisor, launched);
}
},
);
@@ -547,3 +619,88 @@ it.skipIf(process.platform !== "linux")(
expect(selectPrivateDisplay).not.toHaveBeenCalled();
},
);
it.skipIf(process.platform !== "linux")(
"asks the demo helper to shut down orderly so its X lock is removed",
async () => {
const directory = await createTestTempDirectory("rea-hopper-orderly-");
const lockPath = join(directory, ".X199-lock");
const markerPath = join(directory, "orderly-shutdown");
const launcher = orderlyHelperLauncher(directory, "orderly");
const { launched, supervisor } = await launchFixtureProcess(
launcher,
directory,
);
try {
await waitForFixtureFile(lockPath, "lock file never appeared");
expect(await supervisor.stop()).toMatchObject({
status: "verified-cleanup",
});
// The helper shut down orderly: it removed its own lock before exiting.
await expect(access(lockPath)).rejects.toMatchObject({ code: "ENOENT" });
await expect(access(markerPath)).resolves.toBeUndefined();
} finally {
await releaseFixtureProcess(supervisor, launched);
}
},
);
it.skipIf(process.platform !== "linux")(
"still cleans up when the demo helper ignores the orderly shutdown request",
async () => {
const directory = await createTestTempDirectory(
"rea-hopper-orderly-ignored-",
);
const lockPath = join(directory, ".X199-lock");
const launcher = orderlyHelperLauncher(directory, "ignore-shutdown");
const { launched, supervisor } = await launchFixtureProcess(
launcher,
directory,
);
try {
// The fixture really started: its lock is on disk before we ask it to
// shut down, so an early fixture failure cannot fake a passing test.
await waitForFixtureFile(lockPath, "lock file never appeared");
await stopWithinFixtureDeadline(supervisor);
// Bounded by the configured 500 ms grace, not a hardcoded 10 s wait:
// the group kill must follow the ignored shutdown request promptly.
// The fallback path ran: the helper was reaped by the group SIGKILL,
// and its lock survives because the orderly path never ran.
expect(launched.process.signalCode).toBe("SIGKILL");
await expect(access(lockPath)).resolves.toBeUndefined();
} finally {
await releaseFixtureProcess(supervisor, launched);
}
},
);
it.skipIf(process.platform !== "linux")(
"does not wait out the orderly grace when the demo helper already died from a signal",
async () => {
const directory = await createTestTempDirectory("rea-hopper-signal-dead-");
const lockPath = join(directory, ".X199-lock");
const launcher = orderlyHelperLauncher(directory, "signal-death");
const { launched, supervisor } = await launchFixtureProcess(
launcher,
directory,
);
try {
// The fixture wrote its lock, then killed itself with SIGKILL before
// the launcher asked for cleanup.
await waitForFixtureFile(lockPath, "lock file never appeared");
const exitDeadline = Date.now() + 5_000;
for (;;) {
if (launched.process.signalCode !== null) break;
if (Date.now() >= exitDeadline)
throw new Error("signal exit never observed");
await new Promise((resolve) => setTimeout(resolve, 50));
}
await stopWithinFixtureDeadline(supervisor);
// No grace wait: the helper was already gone, so cleanup must not sit
// out the 10 s orderly shutdown window.
expect(launched.process.signalCode).toBe("SIGKILL");
} finally {
await releaseFixtureProcess(supervisor, launched);
}
},
);
@@ -1,8 +1,14 @@
import { access } from "node:fs/promises";
import { spawn, type ChildProcess } from "node:child_process";
import { expect, it, vi } from "vitest";
import { HopperClient } from "../../../../src/hopper/HopperClient.js";
import type { BridgeLauncher } from "../../../../src/hopper/BridgeLauncher.js";
import {
HopperCancelledError,
HopperProcessError,
} from "../../../../src/domain/hopperErrors.js";
import { ok } from "../../../../src/domain/result.js";
import { ProviderRunLineage } from "../../../../src/process/ProviderRunLineage.js";
import { waitForExit } from "../../../support/process/processFixture.js";
@@ -11,6 +17,151 @@ import {
trackHopperClient,
} from "./hopperClient.fixture.js";
const ownedChildLauncher = (
source: string,
onSpawn?: (child: ChildProcess, sessionDirectory: string) => void,
): BridgeLauncher => ({
async launch(session) {
const child = spawn(process.execPath, ["-e", source], {
stdio: ["ignore", "pipe", "pipe"],
});
await new Promise<void>((resolve, reject) => {
child.once("spawn", resolve);
child.once("error", reject);
});
onSpawn?.(child, session.directory);
return ok({
process: child,
ownsProcessLifetime: true,
providerLifetime: "launcher-process",
shutdownMode: "process-cleanup",
cleanup: async () => {
const signaled = child.exitCode === null && child.signalCode === null;
if (signaled) child.kill("SIGTERM");
return (await waitForExit(child, 5_000))
? { cleaned: true as const, signaled }
: {
cleaned: false as const,
reason: "Fixture child did not exit",
};
},
});
},
});
it("keeps an owned launcher exit observed before abort as the startup failure", async () => {
const controller = new AbortController();
const abortReason = new Error("caller aborted after observed launcher exit");
let ownedChild: ChildProcess | undefined;
let runtimeDirectory: string | undefined;
let observeExit!: () => void;
const exitObserved = new Promise<void>((resolve) => {
observeExit = resolve;
});
const client = trackHopperClient(
new HopperClient({
startupTimeoutMs: 10_000,
launcher: ownedChildLauncher("process.exit(3)", (child, directory) => {
ownedChild = child;
runtimeDirectory = directory;
}),
onDiagnostic: (event) => {
if (event.type !== "launcher-exit") return;
observeExit();
setImmediate(() => controller.abort(abortReason));
},
}),
);
try {
const started = await client.start(controller.signal);
await exitObserved;
expect(controller.signal.reason).toBe(abortReason);
expect(started).toMatchObject({
ok: false,
error: {
_tag: "HopperProcessError",
exitCode: 3,
stage: "launch",
},
});
if (started.ok) throw new Error("Expected the owned launcher failure");
if (!(started.error instanceof HopperProcessError))
throw new Error("Expected HopperProcessError");
expect(started.error.cleanupIncomplete).toBe(false);
expect(started.error.launcherFailure).toMatchObject({ exit_code: 3 });
if (ownedChild === undefined) throw new Error("Missing owned launcher");
expect(await waitForExit(ownedChild, 5_000)).toBe(true);
expect(ownedChild.exitCode).toBe(3);
if (runtimeDirectory === undefined)
throw new Error("Missing owned runtime directory");
await expect(access(runtimeDirectory)).rejects.toMatchObject({
code: "ENOENT",
});
} finally {
await client.close();
}
const control = trackHopperClient(
new HopperClient({
startupTimeoutMs: 10_000,
launcher: ownedChildLauncher("process.exit(3)"),
}),
);
try {
const started = await control.start();
expect(started).toMatchObject({
ok: false,
error: { _tag: "HopperProcessError", exitCode: 3, stage: "launch" },
});
} finally {
await control.close();
}
});
it("preserves the exact caller reason when cancellation precedes launcher exit", async () => {
const controller = new AbortController();
const abortReason = new Error("caller cancelled before launcher exit");
let launchChild!: () => void;
let runtimeDirectory: string | undefined;
const childLaunched = new Promise<void>((resolve) => {
launchChild = resolve;
});
const client = trackHopperClient(
new HopperClient({
startupTimeoutMs: 10_000,
launcher: ownedChildLauncher(
'process.on("SIGTERM", () => process.exit(0)); setInterval(() => {}, 1000);',
(_child, directory) => {
runtimeDirectory = directory;
launchChild();
},
),
}),
);
try {
const starting = client.start(controller.signal);
await childLaunched;
controller.abort(abortReason);
const started = await starting;
expect(started).toMatchObject({
ok: false,
error: { _tag: "HopperCancelledError", cause: abortReason },
});
if (started.ok) throw new Error("Expected caller cancellation");
expect(started.error).toBeInstanceOf(HopperCancelledError);
expect(started.error.cause).toBe(abortReason);
if (runtimeDirectory === undefined)
throw new Error("Missing owned runtime directory");
await expect(access(runtimeDirectory)).rejects.toMatchObject({
code: "ENOENT",
});
} finally {
await client.close();
}
});
it("preserves a warm provider when an already cancelled caller joins", async () => {
const launcher = new HopperFixtureLauncher();
const client = trackHopperClient(new HopperClient({ launcher }));
@@ -0,0 +1,132 @@
import { EventEmitter } from "node:events";
import { PassThrough } from "node:stream";
import { expect, it, onTestFinished, vi } from "vitest";
import { ok } from "../../../../src/domain/result.js";
import type {
BridgeLaunch,
BridgeLauncher,
BridgeSession,
} from "../../../../src/hopper/BridgeLauncher.js";
import { HopperClient } from "../../../../src/hopper/HopperClient.js";
import { createLinuxDemoCleanup } from "../../../../src/hopper/LinuxDemoCleanup.js";
import type { OwnedProcessGroup } from "../../../../src/process/ProcessOwnership.js";
import type { ProviderProcessHandle } from "../../../../src/process/ProviderProcess.js";
class FixtureHelper extends EventEmitter implements ProviderProcessHandle {
readonly stdout = new PassThrough();
readonly stderr = new PassThrough();
readonly stdin = new PassThrough();
readonly pid = 902;
exitCode: number | null = null;
signalCode: NodeJS.Signals | null = null;
kill(): boolean {
return true;
}
exit(): void {
this.exitCode = 0;
this.emit("exit", 0, null);
this.stdout.end();
this.stderr.end();
this.emit("close", 0, null);
}
}
const fixtureOwnership = {
runId: "linux-demo-client-cleanup-test",
leaderPid: 902,
processGroupId: 902,
} satisfies OwnedProcessGroup;
class CancelCleanupLauncher implements BridgeLauncher {
readonly helper = new FixtureHelper();
#launchReady: (() => void) | undefined;
readonly launched: Promise<void>;
readonly cleanupGroup = vi
.fn()
.mockResolvedValueOnce({
cleaned: false as const,
reason: "fixture cleanup needs retry",
})
.mockResolvedValueOnce({ cleaned: true as const, signaled: false });
constructor() {
this.launched = new Promise<void>((resolve) => {
this.#launchReady = resolve;
});
}
async launch(_session: BridgeSession) {
const launch: BridgeLaunch = {
process: this.helper,
ownsProcessLifetime: true,
providerLifetime: "launcher-process",
shutdownMode: "process-cleanup",
ownership: fixtureOwnership,
cleanup: createLinuxDemoCleanup(this.helper, fixtureOwnership, 10_000, {
cleanupGroup: async () => this.cleanupGroup(),
}),
};
this.#launchReady?.();
return Promise.resolve(ok(launch));
}
}
it("awaits cancellation-triggered final cleanup and retries the same retained owner", async () => {
const launcher = new CancelCleanupLauncher();
const client = new HopperClient({ launcher, startupTimeoutMs: 30_000 });
onTestFinished(async () => {
launcher.helper.exit();
await client.close();
});
const controller = new AbortController();
const starting = client.start(controller.signal);
await launcher.launched;
// The client has attached its production supervisor before reaching connect.
await new Promise<void>((resolve) => setImmediate(resolve));
expect(launcher.helper.listenerCount("exit")).toBeGreaterThan(0);
const callerReason = { reason: "fixture caller cancellation", token: 17 };
controller.abort(callerReason);
await new Promise<void>((resolve, reject) => {
const timer = setTimeout(
() => reject(new Error("stdin EOF was not sent")),
2_000,
);
launcher.helper.stdin.once("finish", () => {
clearTimeout(timer);
resolve();
});
});
let settled = false;
void starting.finally(() => {
settled = true;
});
await new Promise<void>((resolve) => setImmediate(resolve));
expect(settled).toBe(false);
// The orderly helper exit completes the first cleanup attempt, which is
// deliberately unresolved at the group boundary. Startup must retain the
// cancellation as primary while keeping this owner for close() to retry.
launcher.helper.exit();
const failed = await starting;
expect(failed).toMatchObject({
ok: false,
error: {
_tag: "ProviderAdapterError",
cleanupIncomplete: true,
cause: { cause: { _tag: "HopperCancelledError" } },
},
});
if (failed.ok) throw new Error("expected startup cancellation");
const cancelled = failed.error.cleanupPrimary;
expect(cancelled?._tag).toBe("HopperCancelledError");
expect(cancelled?.cause).toBe(callerReason);
await expect(client.close()).resolves.toMatchObject({ ok: true });
expect(launcher.cleanupGroup).toHaveBeenCalledTimes(2);
});
+46
View File
@@ -0,0 +1,46 @@
#!/usr/bin/env python3
"""Demo-helper stand-in: shuts down orderly when the owner pipe closes.
Mirrors the production helper contract: the launcher owns this process group
and asks for an orderly shutdown by closing stdin (OwnerLifetime). The
production run_launch finally block then stops Hopper and Xvfb with SIGTERM;
here the fixture removes its fake X lock file instead, which is what Xvfb
does for /tmp/.X<n>-lock on a clean exit.
"""
import importlib.util
import os
import signal
import sys
import time
from pathlib import Path
spec = importlib.util.spec_from_file_location(
"hopper_demo_x11", Path(__file__).resolve().parents[2] / "scripts/hopper-demo-x11.py"
)
assert spec is not None and spec.loader is not None
helper = importlib.util.module_from_spec(spec)
spec.loader.exec_module(helper)
assert "--owner-stdin" in sys.argv
lock_dir = Path(os.environ["REA_FIXTURE_LOCK_DIR"])
lock = lock_dir / ".X199-lock"
# Xvfb writes its PID right-justified in a 10-character field.
lock.write_text(f"{os.getpid():10d}\n")
if os.environ.get("REA_FIXTURE_IGNORE_SHUTDOWN") == "1":
while True:
time.sleep(60)
if os.environ.get("REA_FIXTURE_SIGNAL_DEATH") == "1":
# Die from a signal before the launcher asks for cleanup: teardown must
# notice the signal exit instead of waiting out the orderly grace period.
os.kill(os.getpid(), signal.SIGKILL)
owner = helper.OwnerLifetime(True)
while owner.alive():
time.sleep(0.05)
# Orderly shutdown path: remove our own lock, like Xvfb does on SIGTERM.
lock.unlink(missing_ok=True)
(lock_dir / "orderly-shutdown").write_text("ok")
sys.exit(0)