fix(gateway): preserve requests across reentrant connection cleanup (#128393)

packages/gateway-client/src/pending-request.test.ts

@@ -21,4 +21,78 @@ describe("GatewayPendingRequests", () => {
     const retained = (requests as unknown as { retiredIds?: ReadonlySet<string> }).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);
+  });
 });

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 {

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();
+  });
 });