diff --git a/packages/gateway-client/src/pending-request.test.ts b/packages/gateway-client/src/pending-request.test.ts index d5f3810ee72f..76727b901017 100644 --- a/packages/gateway-client/src/pending-request.test.ts +++ b/packages/gateway-client/src/pending-request.test.ts @@ -21,4 +21,78 @@ describe("GatewayPendingRequests", () => { const retained = (requests as unknown as { retiredIds?: ReadonlySet }).retiredIds; expect(retained?.size ?? 0).toBe(0); }); + + it("preserves replacement-generation requests created by a close timing observer", async () => { + const sent: Array<{ id: string; method: string }> = []; + let replacement: Promise<{ healthy: boolean }> | undefined; + const sender = { + send: (frame: string) => { + const { id, method } = JSON.parse(frame) as { id: string; method: string }; + sent.push({ id, method }); + }, + }; + const requests = new GatewayPendingRequests({ + createRequestId: () => "stable", + nowMs: () => 0, + onTiming: ({ method }) => { + if (method === "retired") { + replacement = requests.request(sender, "replacement", {}, { timeoutMs: null }); + void replacement.catch(() => undefined); + } + }, + }); + const retired = requests.request(sender, "retired", {}, { timeoutMs: null }); + const alsoRetired = requests.request(sender, "also-retired", {}, { timeoutMs: null }); + void retired.catch(() => undefined); + void alsoRetired.catch(() => undefined); + + requests.flush(new Error("old socket closed")); + + expect(sent).toEqual([ + { id: "1:stable", method: "retired" }, + { id: "2:stable", method: "also-retired" }, + { id: "1:stable", method: "replacement" }, + ]); + expect(requests.hasPending).toBe(true); + requests.handleResponse({ + type: "res", + id: "1:stable", + ok: true, + payload: { healthy: true }, + }); + + await expect(retired).rejects.toThrow("old socket closed"); + await expect(alsoRetired).rejects.toThrow("old socket closed"); + await expect(replacement).resolves.toEqual({ healthy: true }); + expect(requests.hasPending).toBe(false); + }); + + it("settles each retired request once when its timing observer shuts down again", async () => { + const timings: string[] = []; + const requests = new GatewayPendingRequests({ + createRequestId: () => "stable", + nowMs: () => 0, + onTiming: ({ method }) => { + timings.push(method); + if (timings.length === 1) { + requests.flush(new Error("nested shutdown")); + } + }, + }); + const retired = requests.request( + { send: () => {} }, + "session.observe", + {}, + { + timeoutMs: null, + }, + ); + void retired.catch(() => undefined); + + requests.flush(new Error("transport closed")); + + expect(timings).toEqual(["session.observe"]); + await expect(retired).rejects.toThrow("transport closed"); + expect(requests.hasPending).toBe(false); + }); }); diff --git a/packages/gateway-client/src/pending-request.ts b/packages/gateway-client/src/pending-request.ts index 080d43864a31..fc5d5132c4b7 100644 --- a/packages/gateway-client/src/pending-request.ts +++ b/packages/gateway-client/src/pending-request.ts @@ -178,15 +178,16 @@ export class GatewayPendingRequests { } flush(error: Error): void { - for (const [id, pending] of this.pending) { - this.finishTiming(id, pending, false, "CLIENT_CLOSED"); + const retired = [...this.pending]; + this.pending.clear(); + // Timing observers can reconnect synchronously, so detach the entire old + // generation and reset its sequence before running any caller-owned code. + this.requestSequence = 0; + for (const [id, pending] of retired) { pending.cleanup?.(); + this.finishTiming(id, pending, false, "CLIENT_CLOSED"); pending.reject(error); } - this.pending.clear(); - // Request sequences belong to one socket generation. Retired socket frames - // are fenced by GatewayProtocolClient before the sequence restarts. - this.requestSequence = 0; } private allocateRequestId(): string { diff --git a/packages/gateway-client/src/protocol-client.request.test.ts b/packages/gateway-client/src/protocol-client.request.test.ts index 711b042abe80..3b990ad8c3ce 100644 --- a/packages/gateway-client/src/protocol-client.request.test.ts +++ b/packages/gateway-client/src/protocol-client.request.test.ts @@ -512,4 +512,43 @@ describe("GatewayProtocolClient requests", () => { await expect(replacement).resolves.toEqual({ ok: true }); client.stop(); }); + + it("preserves requests started on a replacement socket by a close timing observer", async () => { + let recoveredRequest: Promise<{ healthy: boolean }> | undefined; + const { client, connections } = createRequestHarness({ + createRequestId: () => "same-id", + onRequestTiming: ({ method }) => { + if (method === "retired") { + client.start(); + recoveredRequest = client.request("replacement", {}, { timeoutMs: null }); + void recoveredRequest.catch(() => undefined); + } + }, + }); + const firstConnection = connections[0]; + if (!firstConnection) { + throw new Error("expected initial request connection"); + } + const retired = client.request("retired", {}, { timeoutMs: null }); + void retired.catch(() => undefined); + + firstConnection.close(1012, "service restart"); + + const replacementConnection = connections[1]; + if (!replacementConnection) { + throw new Error("expected replacement request connection"); + } + expect(latestFrame(replacementConnection)).toMatchObject({ + id: "1:same-id", + method: "replacement", + }); + expect(client.connected).toBe(true); + expect(client.hasPendingRequests).toBe(true); + respond(replacementConnection, "1:same-id", { healthy: true }); + + await expect(retired).rejects.toThrow("gateway closed (1012): service restart"); + await expect(recoveredRequest).resolves.toEqual({ healthy: true }); + expect(client.hasPendingRequests).toBe(false); + client.stop(); + }); });