From 6cc2fdf80ce2d0220fe996ddbed2f0378f5397cd Mon Sep 17 00:00:00 2001 From: N0zoM1z0 <1705004398@qq.com> Date: Tue, 14 Jul 2026 21:03:14 +0000 Subject: [PATCH] fix(hopper): cancel startup before closing --- src/hopper/HopperClient.ts | 68 ++++++++++++++++++++++++++++++++------ tests/hopperClient.test.ts | 41 +++++++++++++++++++++++ 2 files changed, 98 insertions(+), 11 deletions(-) diff --git a/src/hopper/HopperClient.ts b/src/hopper/HopperClient.ts index 8de3ad704..91c3fe0ef 100644 --- a/src/hopper/HopperClient.ts +++ b/src/hopper/HopperClient.ts @@ -72,6 +72,7 @@ export class HopperClient { #buffer = ""; #closing = false; #launcherExitCode: number | null | undefined; + #startupController: AbortController | undefined; #startPromise: Promise> | undefined; #closePromise: Promise | undefined; @@ -86,14 +87,34 @@ export class HopperClient { /** Launch the bridge once and complete its authenticated health handshake. */ start(signal?: AbortSignal): Promise> { - this.#startPromise ??= this.#start(signal); - return this.#startPromise; + if (this.#closePromise !== undefined) { + return this.#closePromise.then(() => this.start(signal)); + } + if (this.#startPromise !== undefined) return this.#startPromise; + + const controller = new AbortController(); + const startupSignal = + signal === undefined + ? controller.signal + : AbortSignal.any([signal, controller.signal]); + const started = this.#start(startupSignal); + this.#startupController = controller; + this.#startPromise = started; + void started.then((result) => { + if (!result.ok && this.#startPromise === started) { + this.#startPromise = undefined; + if (this.#startupController === controller) { + this.#startupController = undefined; + } + } + }); + return started; } async #start( signal?: AbortSignal, ): Promise> { - if (signal?.aborted === true) return err(new HopperCancelledError()); + if (isAborted(signal)) return err(new HopperCancelledError()); if (this.#socket !== undefined || this.#directory !== undefined) { return err(new HopperProtocolError("Hopper client is already started")); } @@ -102,6 +123,10 @@ export class HopperClient { } catch (cause: unknown) { return err(new HopperStartError({ cause })); } + if (isAborted(signal)) { + await this.#cleanup(); + return err(new HopperCancelledError()); + } const socketPath = join(this.#directory, "bridge.sock"); this.#token = randomBytes(32).toString("hex"); this.#launcherExitCode = undefined; @@ -116,14 +141,14 @@ export class HopperClient { signal === undefined ? {} : { signal }, ); if (!launched.ok) { - await this.close(); + await this.#cleanup(); return launched; } this.#launch = launched.value; this.#attachLauncher(launched.value); const connected = await this.#connect(socketPath, signal); if (!connected.ok) { - await this.close(); + await this.#cleanup(); return connected; } const health = await this.#request( @@ -135,11 +160,11 @@ export class HopperClient { }, ); if (!health.ok) { - await this.close(); + await this.#cleanup(); return health; } const parsed = parseServerInfo(health.value, runId); - if (!parsed.ok) await this.close(); + if (!parsed.ok) await this.#cleanup(); return parsed; } @@ -159,11 +184,32 @@ export class HopperClient { /** Stop the bridge, settle outstanding requests, and remove session artifacts. */ close(): Promise { - this.#closePromise ??= Promise.resolve().then(() => this.#close()); + const starting = this.#startPromise; + const controller = this.#startupController; + controller?.abort(); + this.#closePromise ??= Promise.resolve().then(() => + this.#close(starting, controller), + ); return this.#closePromise; } - async #close(): Promise { + async #close( + starting: Promise> | undefined, + controller: AbortController | undefined, + ): Promise { + try { + await starting?.catch(() => undefined); + await this.#cleanup(); + } finally { + if (this.#startPromise === starting) this.#startPromise = undefined; + if (this.#startupController === controller) { + this.#startupController = undefined; + } + this.#closePromise = undefined; + } + } + + async #cleanup(): Promise { this.#closing = true; try { const socket = this.#socket; @@ -216,10 +262,8 @@ export class HopperClient { this.#directory = undefined; } this.#token = undefined; - this.#startPromise = undefined; } finally { this.#closing = false; - this.#closePromise = undefined; } } @@ -411,6 +455,8 @@ export class HopperClient { } } +const isAborted = (signal?: AbortSignal): boolean => signal?.aborted === true; + const waitForDelay = ( milliseconds: number, signal?: AbortSignal, diff --git a/tests/hopperClient.test.ts b/tests/hopperClient.test.ts index 67169868f..d27765581 100644 --- a/tests/hopperClient.test.ts +++ b/tests/hopperClient.test.ts @@ -254,4 +254,45 @@ describe("HopperClient", () => { expect(firstSettled).toBe(true); await first; }); + + it("does not finish startup after an immediate close", async () => { + const client = new HopperClient({ + launcher: new FixtureLauncher(), + startupTimeoutMs: 1_000, + }); + clients.push(client); + + const starting = client.start(); + await client.close(); + + const result = await starting; + expect(result.ok).toBe(false); + if (!result.ok) expect(result.error._tag).toBe("HopperCancelledError"); + }); + + it("starts a fresh session after close completes", async () => { + const client = await startClient(); + await client.close(); + + await expect(client.start()).resolves.toEqual({ + ok: true, + value: { name: "REA Hopper bridge", version: "1.0.0" }, + }); + }); + + it("serializes a restart requested while close is in progress", async () => { + const client = await startClient(); + const closing = client.close(); + const restarting = client.start(); + + await closing; + await expect(restarting).resolves.toEqual({ + ok: true, + value: { name: "REA Hopper bridge", version: "1.0.0" }, + }); + await expect(client.callTool("echo", { value: "fresh" })).resolves.toEqual({ + ok: true, + value: { value: "fresh" }, + }); + }); });