From 5da235b3e61565e02b0f4dd3e6a81a9d91487482 Mon Sep 17 00:00:00 2001 From: ROOOO Date: Mon, 17 Aug 2026 04:48:35 -0400 Subject: [PATCH] fix(responses-ws): reset client socket after HTTP fallback --- server.js | 73 +++++++++++----- ...server-response-write-backpressure.test.ts | 42 ++++++++- tests/unit/server-ws-close-handshake.test.ts | 85 +++++++++++++------ 3 files changed, 152 insertions(+), 48 deletions(-) diff --git a/server.js b/server.js index 8c4469a34..5a8c71908 100644 --- a/server.js +++ b/server.js @@ -45,6 +45,7 @@ const INTERNAL_TUNNEL_HOST = const WS_PATH = "/v1/responses"; const CLIENT_TRANSPORT_HEADER = "x-cch-client-transport"; +const UPSTREAM_TRANSPORT_HEADER = "x-cch-upstream-transport"; const WS_FORWARD_FLAG_HEADER = "x-cch-responses-ws-forward"; const WS_SESSION_HEADER = "x-cch-responses-ws-session"; const INTERNAL_SECRET_HEADER = "x-cch-internal-secret"; @@ -624,6 +625,12 @@ async function forwardToInternalHttp( (res) => { const contentType = (res.headers["content-type"] || "").toLowerCase(); const isSse = contentType.includes("text/event-stream"); + const upstreamTransportValue = res.headers[UPSTREAM_TRANSPORT_HEADER]; + const upstreamTransport = ( + Array.isArray(upstreamTransportValue) + ? upstreamTransportValue[0] + : upstreamTransportValue || "" + ).toLowerCase(); let responseSettled = false; let responseBodyEnded = false; let terminalSendAcknowledged = false; @@ -636,10 +643,6 @@ async function forwardToInternalHttp( resolve(); return true; }; - const acknowledgeTerminalSend = () => { - terminalSendAcknowledged = true; - settleResponse(); - }; const forceSettleResponse = () => { if (responseSettled) return false; responseSettled = true; @@ -656,6 +659,24 @@ async function forwardToInternalHttp( initiateClose(1011, reason); forceSettleResponse(); }; + const settleAndResetTransport = () => { + initiateClose(1000, "upstream_transport_reset"); + forceSettleResponse(); + }; + const canReuseClientWebSocket = (terminalEvent) => { + if (upstreamTransport !== "websocket") return false; + if (!terminalEvent || terminalEvent.type !== "response.completed") return false; + const errorCode = terminalEvent.error?.code || terminalEvent.response?.error?.code; + return errorCode !== "websocket_connection_limit_reached"; + }; + const acknowledgeTerminalSend = (terminalEvent) => { + terminalSendAcknowledged = true; + if (canReuseClientWebSocket(terminalEvent)) { + settleResponse(); + return; + } + settleAndResetTransport(); + }; const sendFatalError = (code, message, closeReason) => { const sent = safeSend( ws, @@ -686,17 +707,22 @@ async function forwardToInternalHttp( } const isHttpError = !!(res.statusCode && res.statusCode >= 400); if (isHttpError) { + const terminalEvent = { + type: "error", + status: res.statusCode, + error: + typeof parsed === "object" && parsed && parsed.error + ? parsed.error + : { code: `http_${res.statusCode}`, message: text.slice(0, 512) }, + }; safeSend( ws, + terminalEvent, { - type: "error", - status: res.statusCode, - error: - typeof parsed === "object" && parsed && parsed.error - ? parsed.error - : { code: `http_${res.statusCode}`, message: text.slice(0, 512) }, - }, - { response: res, onSuccess: acknowledgeTerminalSend, onFailure: settleAndClose } + response: res, + onSuccess: () => acknowledgeTerminalSend(terminalEvent), + onFailure: settleAndClose, + } ); log("info", "ws_terminal_event_sent", { type: "error", @@ -704,10 +730,15 @@ async function forwardToInternalHttp( status: res.statusCode, }); } else { + const terminalEvent = { type: "response.completed", response: parsed }; safeSend( ws, - { type: "response.completed", response: parsed }, - { response: res, onSuccess: acknowledgeTerminalSend, onFailure: settleAndClose } + terminalEvent, + { + response: res, + onSuccess: () => acknowledgeTerminalSend(terminalEvent), + onFailure: settleAndClose, + } ); log("info", "ws_terminal_event_sent", { type: "response.completed", source: "json" }); } @@ -778,9 +809,10 @@ async function forwardToInternalHttp( // Some upstreams close SSE with [DONE] without a preceding // response.completed. Synthesize one so the client sees a // clean terminal event. - safeSend(ws, { type: "response.completed", response: null }, { + const terminalEvent = { type: "response.completed", response: null }; + safeSend(ws, terminalEvent, { response: res, - onSuccess: acknowledgeTerminalSend, + onSuccess: () => acknowledgeTerminalSend(terminalEvent), onFailure: settleAndClose, }); sawTerminal = true; @@ -802,7 +834,7 @@ async function forwardToInternalHttp( event && typeof event.type === "string" && TERMINAL_EVENT_TYPES.has(event.type); safeSend(ws, event, { response: res, - onSuccess: isTerminalEvent ? acknowledgeTerminalSend : undefined, + onSuccess: isTerminalEvent ? () => acknowledgeTerminalSend(event) : undefined, onFailure: settleAndClose, }); if (isTerminalEvent) { @@ -834,10 +866,9 @@ async function forwardToInternalHttp( "stream_ended_without_terminal" ); } else { - // OpenAI Responses WebSocket mode is persistent: after a terminal - // event, the same client connection can send the next - // response.create. Do not close here; only fatal transport/protocol - // errors initiate a close handshake. + // Only a successful upstream WebSocket turn is reusable. HTTP + // fallback and terminal error paths reset the client transport in + // acknowledgeTerminalSend after the terminal frame is delivered. log("info", "ws_turn_completed", { terminalEventType }); settleResponse(); } diff --git a/tests/unit/server-response-write-backpressure.test.ts b/tests/unit/server-response-write-backpressure.test.ts index f8c02a456..68e264f15 100644 --- a/tests/unit/server-response-write-backpressure.test.ts +++ b/tests/unit/server-response-write-backpressure.test.ts @@ -393,6 +393,7 @@ describe("server response write backpressure", () => { const events: string[] = []; const request = createClientRequest(true, events); const response = createIncomingResponse(); + response.headers["x-cch-upstream-transport"] = "websocket"; response.complete = false; let respond: ((response: http.IncomingMessage) => void) | undefined; vi.spyOn(http, "request").mockImplementation((_options, callback) => { @@ -420,7 +421,44 @@ describe("server response write backpressure", () => { expect(close).not.toHaveBeenCalled(); }); - it("waits for the JSON terminal send acknowledgement across end and close", async () => { + it("waits for an HTTP SSE terminal acknowledgement before resetting the client", async () => { + const events: string[] = []; + const request = createClientRequest(true, events); + const response = createIncomingResponse(); + let respond: ((response: http.IncomingMessage) => void) | undefined; + vi.spyOn(http, "request").mockImplementation((_options, callback) => { + if (callback) respond = callback; + return request; + }); + let sendCallback: ((error?: Error) => void) | undefined; + const close = vi.fn(); + const input = requestInput(); + input.ws.send = (_payload, callback) => { + sendCallback = callback; + }; + + const forwarding = serverModule.forwardToInternalHttp( + input.ws, + input.request, + input.body, + "http-sse-session", + undefined, + close + ); + respond?.(response); + response.emit( + "data", + `data: ${JSON.stringify({ type: "response.completed", response: { id: "r1" } })}\n\n` + ); + response.emit("end"); + + expect(close).not.toHaveBeenCalled(); + sendCallback?.(); + await forwarding; + expect(close).toHaveBeenCalledWith(1000, "upstream_transport_reset"); + }); + + it("waits for the JSON terminal send acknowledgement before resetting the client", async () => { const events: string[] = []; const request = createClientRequest(true, events); const response = new http.IncomingMessage(new Socket()); @@ -463,7 +501,7 @@ describe("server response write backpressure", () => { sendCallback?.(); await forwarding; - expect(close).not.toHaveBeenCalled(); + expect(close).toHaveBeenCalledWith(1000, "upstream_transport_reset"); }); it("does not start a request while a fatal protocol frame awaits acknowledgement", async () => { diff --git a/tests/unit/server-ws-close-handshake.test.ts b/tests/unit/server-ws-close-handshake.test.ts index 806b18824..9df6e60b5 100644 --- a/tests/unit/server-ws-close-handshake.test.ts +++ b/tests/unit/server-ws-close-handshake.test.ts @@ -273,6 +273,7 @@ describe("server.js WebSocket close-handshake (issue #1150)", () => { harness.setSseHandler((_req, res) => { res.statusCode = 200; res.setHeader("content-type", "text/event-stream"); + res.setHeader("x-cch-upstream-transport", "websocket"); res.write( `data: ${JSON.stringify({ type: "response.created", response: { id: "r_1" } })}\n\n` ); @@ -306,6 +307,36 @@ describe("server.js WebSocket close-handshake (issue #1150)", () => { expect(types).toContain("response.completed"); }); + it("resets the client socket after an HTTP fallback and drops queued turns", async () => { + if (!harness) throw new Error("harness not initialized"); + let upstreamCalls = 0; + harness.setSseHandler((_req, res) => { + upstreamCalls += 1; + res.statusCode = 200; + res.setHeader("content-type", "text/event-stream"); + res.write( + `data: ${JSON.stringify({ + type: "response.completed", + response: { id: `r_http_${upstreamCalls}` }, + })}\n\n` + ); + res.end(); + }); + + const client = connectClient(harness.port); + await client.opened; + client.ws.send(JSON.stringify({ type: "response.create", model: "gpt-5.5", input: "first" })); + client.ws.send(JSON.stringify({ type: "response.create", model: "gpt-5.5", input: "queued" })); + + const close = await client.closeEvent; + expect(close).toEqual({ code: 1000, reason: "upstream_transport_reset" }); + expect(client.messages).toContainEqual({ + type: "response.completed", + response: { id: "r_http_1" }, + }); + expect(upstreamCalls).toBe(1); + }); + it("sends close(1011) when the upstream stream ends without a terminal event", async () => { if (!harness) throw new Error("harness not initialized"); harness.setSseHandler((_req, res) => { @@ -363,7 +394,7 @@ describe("server.js WebSocket close-handshake (issue #1150)", () => { ); }); - it("forwards a non-stream HTTP error without closing the persistent client socket", async () => { + it("forwards a non-stream HTTP error before resetting the client socket", async () => { if (!harness) throw new Error("harness not initialized"); harness.setSseHandler((_req, res) => { res.statusCode = 502; @@ -376,46 +407,49 @@ describe("server.js WebSocket close-handshake (issue #1150)", () => { client.ws.send(JSON.stringify({ type: "response.create", model: "gpt-5.5", input: "hi" })); await waitForMessageCount(client.messages, 1, 3000, "HTTP error was not forwarded"); - expect(client.ws.readyState).toBe(WebSocket.OPEN); const errorEvent = client.messages.find( (m): m is { type: string; status: number; error: { code: string } } => typeof m === "object" && m !== null && (m as { type?: unknown }).type === "error" ); expect(errorEvent?.status).toBe(502); expect(errorEvent?.error.code).toBe("bad_gateway"); - client.ws.close(1000, "test_done"); - await client.closeEvent; + const close = await client.closeEvent; + expect(close).toEqual({ code: 1000, reason: "upstream_transport_reset" }); }); - it("forwards terminal type 'error' without closing the persistent client socket", async () => { - if (!harness) throw new Error("harness not initialized"); - harness.setSseHandler((_req, res) => { - res.statusCode = 200; - res.setHeader("content-type", "text/event-stream"); - res.write( - `data: ${JSON.stringify({ - type: "error", - error: { code: "upstream_failure", message: "boom" }, - })}\n\n` - ); - res.end(); - }); + it.each(["upstream_failure", "websocket_connection_limit_reached"])( + "resets the client socket after upstream WebSocket terminal error %s", + async (errorCode) => { + if (!harness) throw new Error("harness not initialized"); + harness.setSseHandler((_req, res) => { + res.statusCode = 200; + res.setHeader("content-type", "text/event-stream"); + res.setHeader("x-cch-upstream-transport", "websocket"); + res.write( + `data: ${JSON.stringify({ + type: "error", + error: { code: errorCode, message: "upstream WebSocket error" }, + })}\n\n` + ); + res.end(); + }); - const client = connectClient(harness.port); - await client.opened; - client.ws.send(JSON.stringify({ type: "response.create", model: "gpt-5.5", input: "hi" })); + const client = connectClient(harness.port); + await client.opened; + client.ws.send(JSON.stringify({ type: "response.create", model: "gpt-5.5", input: "hi" })); - await waitForMessageCount(client.messages, 1, 3000, "terminal error was not forwarded"); - expect(client.ws.readyState).toBe(WebSocket.OPEN); - client.ws.close(1000, "test_done"); - await client.closeEvent; - }); + await waitForMessageCount(client.messages, 1, 3000, "terminal error was not forwarded"); + const close = await client.closeEvent; + expect(close).toEqual({ code: 1000, reason: "upstream_transport_reset" }); + } + ); it("accepts response.create bodies up to 4 MiB without a maxPayload teardown", async () => { if (!harness) throw new Error("harness not initialized"); harness.setSseHandler((_req, res) => { res.statusCode = 200; res.setHeader("content-type", "text/event-stream"); + res.setHeader("x-cch-upstream-transport", "websocket"); res.write( `data: ${JSON.stringify({ type: "response.completed", @@ -448,6 +482,7 @@ describe("server.js WebSocket close-handshake (issue #1150)", () => { const callNo = upstreamCalls; res.statusCode = 200; res.setHeader("content-type", "text/event-stream"); + res.setHeader("x-cch-upstream-transport", "websocket"); // Stagger the response so the second frame remains queued until the // first turn's terminal event has been fully forwarded. setTimeout(() => {