From c2b33237d07f60b3e5c1e4f75e29e9be3531e231 Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 19 Sep 2026 17:04:25 +0900 Subject: [PATCH] fix(test): raise the native passthrough reset from a requested pull (#5073) The reset fixture in tests/server/server-auth.test.ts leaked its stream controller out of start() and errored it from the test body. Whether that rejection had a consumer depended on where Bun's server-side response sink happened to be: between its reads there is no pending read request to reject, so on a loaded runner the fixture's own error escaped as an unhandled error and failed the whole file. It fired on four unrelated heads (#4989, #5024, dev at ecd3adae75, and #5085). Raise it from inside pull() on a stream whose high-water mark is zero instead. shouldCallPull is then true only while a read request is outstanding, so pull() runs if and only if a consumer is waiting for the next chunk, and throwing there rejects that read request. The reset now has a consumer no matter when the test calls it. What the code under test sees is unchanged: one SSE chunk, then a mid-stream body error. Closes #5073 --- tests/helpers/deferred-reset-sse-upstream.ts | 73 ++++++++++++++++++++ tests/server/server-auth.test.ts | 16 ++--- 2 files changed, 77 insertions(+), 12 deletions(-) create mode 100644 tests/helpers/deferred-reset-sse-upstream.ts diff --git a/tests/helpers/deferred-reset-sse-upstream.ts b/tests/helpers/deferred-reset-sse-upstream.ts new file mode 100644 index 0000000000..00a3bfa307 --- /dev/null +++ b/tests/helpers/deferred-reset-sse-upstream.ts @@ -0,0 +1,73 @@ +/** + * An SSE upstream body that resets mid-stream on demand, where the reset is raised somewhere + * the stream itself observes rather than somewhere nothing is listening (#5073). + * + * The direct way to write this fixture is to leak the stream's controller out of `start()` + * and call `controller.error()` from the test body. That is what + * `tests/server/server-auth.test.ts` did, and under load the fixture's own error escaped as + * an unhandled error instead of being delivered to the code under test, failing the whole + * file on four unrelated heads (#4989, #5024, `dev` at ecd3adae75, #5085). + * + * The reason is that a leaked controller can be errored at a moment of the test's choosing, + * which is not necessarily a moment when anything is reading. Between the server sink's + * reads there is no pending read request to reject, so the rejection's only subscriber is + * whatever the runtime attaches next — and on a busy runner the unhandled-rejection report + * can win that race. The test cannot see or control that window from outside the stream. + * + * So the reset is raised from inside `pull()` instead. Two properties come from that, and + * they are worth separating because only the first one is absolute: + * + * The pull algorithm's promise is always observed. `CallPullIfNeeded` attaches its own + * rejection handler and routes the failure into `ReadableStreamDefaultControllerError`, so a + * throw from `pull()` cannot be an orphaned rejection whatever the consumer is doing. That is + * the guarantee this fixture rests on. + * + * `highWaterMark: 0` then keeps the reset in a faithful place. The queue is never stocked + * ahead of demand, so `shouldCallPull` is true only while a read request is outstanding, and + * `pull()` runs if and only if a consumer has asked for the next chunk. The stream is never + * errored before anything has attached to it, which is the state where the error has nowhere + * to go. It is not a promise that a read request is still pending at the instant of the + * throw — a consumer that cancels in between removes its own request — only that the error is + * raised in response to demand and is consumed by the stream either way. + * + * The enqueued opening chunks are unaffected: `enqueue()` ignores the high-water mark, and the + * first read is still served from the queue. + * + * What the code under test sees is unchanged: one SSE chunk, then a mid-stream body error. + */ + +/** The reset every body from this helper raises, so assertions can name it. */ +export const DEFERRED_RESET_MESSAGE = "fixture upstream connection reset"; + +export type DeferredResetSseUpstream = { + /** A fresh body for one upstream attempt; a retried dispatch gets its own stream. */ + response: (init?: ResponseInit) => Response; + /** Reset every body this helper has handed out, and every later one. */ + reset: () => void; +}; + +/** + * Build an SSE upstream that emits `chunks` and then resets when `reset()` is called. + * + * `reset()` is idempotent and order-independent: calling it before the first attempt is + * dispatched arms the reset for that attempt rather than losing it. + */ +export function deferredResetSseUpstream(...chunks: string[]): DeferredResetSseUpstream { + const encoder = new TextEncoder(); + const requested = Promise.withResolvers(); + return { + reset: () => requested.resolve(), + response: (init = { headers: { "content-type": "text/event-stream" } }) => new Response( + new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(encoder.encode(chunk)); + }, + async pull() { + await requested.promise; + throw new Error(DEFERRED_RESET_MESSAGE); + }, + }, { highWaterMark: 0 }), + init, + ), + }; +} diff --git a/tests/server/server-auth.test.ts b/tests/server/server-auth.test.ts index 1e226dd0b8..ad8d5b48bf 100644 --- a/tests/server/server-auth.test.ts +++ b/tests/server/server-auth.test.ts @@ -52,6 +52,7 @@ import { getDebugLogEntries, resetDebugLogBufferForTests } from "../../src/lib/d import { resetDebugSettingsForTests, setDebugSettings } from "../../src/lib/debug-settings"; import { watchdogMs } from "../helpers/ci-watchdog"; import { removeTreeWithRetry } from "../helpers/remove-tree"; +import { deferredResetSseUpstream } from "../helpers/deferred-reset-sse-upstream"; const previousApiToken = process.env.OPENCODEX_API_AUTH_TOKEN; const previousOpencodexHome = process.env.OPENCODEX_HOME; const originalGlobalFetch = globalThis.fetch; @@ -4245,23 +4246,14 @@ describe("server local API auth", () => { }, { timeout: SERVER_BUDGET_MS }); test("native passthrough upstream reset still logs 502 and penalizes the pool", async () => { - const enc = new TextEncoder(); - const source = Promise.withResolvers>(); - const harness = await startPoolRetryHarness(() => new Response( - new ReadableStream({ - start(controller) { - source.resolve(controller); - controller.enqueue(enc.encode('data: {"type":"response.output_text.delta","delta":"hello"}\n\n')); - }, - }), - { headers: { "content-type": "text/event-stream" } }, - ), { secondAccount: false, streamMode: "legacy-tee" }); + const upstream = deferredResetSseUpstream('data: {"type":"response.output_text.delta","delta":"hello"}\n\n'); + const harness = await startPoolRetryHarness(() => upstream.response(), { secondAccount: false, streamMode: "legacy-tee" }); try { const response = await harness.request({ stream: true }); const requestId = response.headers.get("x-opencodex-request-id"); const reader = response.body!.getReader(); expect((await reader.read()).done).toBe(false); - (await source.promise).error(new Error("fixture upstream connection reset")); + upstream.reset(); while (!(await reader.read()).done) { /* drain the synthetic failed terminal */ } const deadline = Date.now() + INTERNAL_DEADLINE_MS; while (!getRequestLogEntries().some(entry => entry.requestId === requestId) && Date.now() < deadline) {