diff --git a/docs-site/src/content/docs/reference/proxy-formats.md b/docs-site/src/content/docs/reference/proxy-formats.md index 77a67147ac..19049e87f5 100644 --- a/docs-site/src/content/docs/reference/proxy-formats.md +++ b/docs-site/src/content/docs/reference/proxy-formats.md @@ -83,6 +83,11 @@ This applies to both tee inspection and eager relay, including Windows rewrite t even when the upstream read rejects before the response-body cancellation hook runs. A terminal captured during the bounded post-disconnect drain retains its actual outcome. +If native passthrough rewriting fails, including when it exceeds the translation +buffer budget, the relay reports the failure without waiting for upstream inspection +to finish. It cancels the upstream work and emits `response.failed` followed by +`data: [DONE]`; a budget overflow uses the `translation_buffer_limit` error code. + Client-facing Responses SSE frames are limited to 4 MiB per frame, measured in raw bytes before the SSE block delimiter. On HTTP, an unterminated upstream frame that exceeds the limit fails closed with a synthetic `response.failed` event followed by `data: [DONE]`. On the Responses WebSocket diff --git a/src/server/sse-payload-rewrite.ts b/src/server/sse-payload-rewrite.ts index 3c6d825e6c..f9fb62065d 100644 --- a/src/server/sse-payload-rewrite.ts +++ b/src/server/sse-payload-rewrite.ts @@ -249,7 +249,9 @@ export function relaySseWithBlockRewrite( } catch (error) { releaseBuffer(); disposeRewrite(); - try { await reader.cancel(error); } catch { /* already closed */ } + // Cancelling one tee branch waits for its sibling. Surface the failure + // now so downstream can abort upstream and release the inspection branch. + void reader.cancel(error).catch(() => {}); controller.error(error); } }, diff --git a/tests/responses/sse-payload-rewrite.test.ts b/tests/responses/sse-payload-rewrite.test.ts index 34dae59e07..773665a054 100644 --- a/tests/responses/sse-payload-rewrite.test.ts +++ b/tests/responses/sse-payload-rewrite.test.ts @@ -153,4 +153,82 @@ describe("SSE payload rewrite composition", () => { expect(budget.snapshot().currentBytes).toBe(0); budget.dispose(); }); + + test.each(["resolve", "reject"] as const)( + "surfaces a rewrite failure before tee cancellation can %s", + async cancellationOutcome => { + const budget = createTestTranslatorBudget({ maxTurnBytes: 64 }); + const upstream = new AbortController(); + const cancellation = Promise.withResolvers(); + const cancellationError = new Error("upstream cancellation failed"); + let cancelCalls = 0; + let disposeCalls = 0; + const source = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("data: partial")); + controller.enqueue(new TextEncoder().encode("x".repeat(80))); + // Keep the source open after exhausting the rewrite budget. + }, + cancel() { + cancelCalls += 1; + return cancellation.promise; + }, + }); + const [native, inspection] = source.tee(); + const inspectionReader = inspection.getReader(); + await inspectionReader.read(); + await inspectionReader.read(); + let inspectionSettled = false; + const pendingInspection = inspectionReader.read().then(() => { inspectionSettled = true; }); + const rewrite = Object.assign((block: string) => [block], { + dispose() { disposeCalls += 1; }, + }); + const rewritten = relaySseWithBlockRewrite(native, rewrite, budget); + const client = relaySseWithFailedTail(rewritten, upstream); + const completion = readAll(client); + let deadline: ReturnType | undefined; + + try { + const out = await Promise.race([ + completion, + new Promise((_, reject) => { + deadline = setTimeout(() => reject(new Error("rewrite failure waited for the inspection tee")), 1_000); + }), + ]); + expect(out.match(/event: response.failed/g)).toHaveLength(1); + expect(out).toContain('"code":"translation_buffer_limit"'); + expect(out).toEndWith("data: [DONE]\n\n"); + expect(upstream.signal.aborted).toBe(true); + expect(inspectionSettled).toBe(false); + expect(cancelCalls).toBe(0); + expect(disposeCalls).toBe(1); + expect(budget.snapshot().currentBytes).toBe(0); + expect(budget.snapshot().overflows).toBe(1); + + // Releasing inspection settles both tee cancellation promises. A late + // rejection must be handled by the rewriter as well as this reader. + const siblingCancellation = inspectionReader.cancel("inspection cleanup"); + expect(cancelCalls).toBe(1); + if (cancellationOutcome === "reject") { + cancellation.reject(cancellationError); + await expect(siblingCancellation).rejects.toBe(cancellationError); + } else { + cancellation.resolve(); + await siblingCancellation; + } + await pendingInspection; + await Bun.sleep(0); // Let the runner observe any unhandled cancellation rejection. + expect(disposeCalls).toBe(1); + } finally { + clearTimeout(deadline); + const cleanup = inspectionReader.cancel().catch(() => {}); + cancellation.resolve(); + await cleanup; + await pendingInspection; + await completion.catch(() => {}); + inspectionReader.releaseLock(); + budget.dispose(); + } + }, + ); });