mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-24 03:15:46 -06:00
fix(provider-transport-fetch): bound SSE buffer to prevent OOM (#96989)
* fix(provider-transport-fetch): bound SSE buffer to prevent OOM
* fix(provider-transport-fetch): appease oxlint curly rule in test
* fix(provider-transport-fetch): drain events before cap + cancel reader on overflow
* fix(provider-transport-fetch): remove unused encoder from coalesced chunk test
Co-Authored-By: Claude <noreply@anthropic.com>
* fix(transport): tighten SSE buffer guards
---------
Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Vincent Koc <vincentkoc@ieee.org>
(cherry picked from commit 1bccd29304)
This commit is contained in:
wangmiao0668000666
committed by
Dallin Romney
Dallin Romney
parent
f63e6e1752
commit
7e95c02be5
@@ -1130,6 +1130,37 @@ describe("buildGuardedModelFetch", () => {
|
||||
expect(items).toEqual([{ ok: true }]);
|
||||
});
|
||||
|
||||
it("handles a large transport chunk containing many valid small SSE events", async () => {
|
||||
// Regression: one TCP read can deliver >64 KiB of already-delimited SSE
|
||||
// events; the cap must apply only to the unterminated tail, not the full chunk.
|
||||
const eventCount = 5_000;
|
||||
const manyEvents = `data: ${JSON.stringify({ ok: true })}\n\n`.repeat(eventCount);
|
||||
fetchWithSsrFGuardMock.mockResolvedValue({
|
||||
response: new Response(manyEvents, {
|
||||
headers: { "content-type": "text/event-stream" },
|
||||
}),
|
||||
finalUrl: "https://openrouter.ai/api/v1/chat/completions",
|
||||
release: vi.fn(async () => undefined),
|
||||
});
|
||||
const model = {
|
||||
id: "gpt-5.4",
|
||||
provider: "openrouter",
|
||||
api: "openai-completions",
|
||||
baseUrl: "https://openrouter.ai/api/v1",
|
||||
} as unknown as Model<"openai-completions">;
|
||||
|
||||
const response = await buildGuardedModelFetch(model)(
|
||||
"https://openrouter.ai/api/v1/chat/completions",
|
||||
{ method: "POST" },
|
||||
);
|
||||
const items: unknown[] = [];
|
||||
for await (const item of Stream.fromSSEResponse(response, new AbortController())) {
|
||||
items.push(item);
|
||||
}
|
||||
expect(items.length).toBe(eventCount);
|
||||
expect(items[0]).toEqual({ ok: true });
|
||||
});
|
||||
|
||||
it("synthesizes SSE frames for JSON bodies returned to streaming OpenAI SDK requests", async () => {
|
||||
fetchWithSsrFGuardMock.mockResolvedValue({
|
||||
response: new Response(' {"ok": true} ', {
|
||||
@@ -1338,6 +1369,102 @@ describe("buildGuardedModelFetch", () => {
|
||||
expect(refreshTimeout).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("errors on oversized SSE body without event boundary in sanitizer", async () => {
|
||||
const oversized = "x".repeat(65 * 1024);
|
||||
const encoder = new TextEncoder();
|
||||
fetchWithSsrFGuardMock.mockResolvedValue({
|
||||
response: new Response(
|
||||
new ReadableStream({
|
||||
start(controller) {
|
||||
controller.enqueue(encoder.encode(oversized));
|
||||
controller.close();
|
||||
},
|
||||
}),
|
||||
{ headers: { "content-type": "text/event-stream" } },
|
||||
),
|
||||
finalUrl: "https://openrouter.ai/api/v1/chat/completions",
|
||||
release: vi.fn(async () => undefined),
|
||||
});
|
||||
const model = {
|
||||
id: "gpt-5.4",
|
||||
provider: "openrouter",
|
||||
api: "openai-completions",
|
||||
baseUrl: "https://openrouter.ai/api/v1",
|
||||
} as unknown as Model<"openai-completions">;
|
||||
|
||||
const response = await buildGuardedModelFetch(model)(
|
||||
"https://openrouter.ai/api/v1/chat/completions",
|
||||
{ method: "POST" },
|
||||
);
|
||||
|
||||
const reader = response.body?.getReader();
|
||||
let caught: unknown = null;
|
||||
try {
|
||||
while (true) {
|
||||
const { done } = await reader!.read();
|
||||
if (done) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
} catch (e) {
|
||||
caught = e;
|
||||
}
|
||||
expect(caught).toBeTruthy();
|
||||
expect(String(caught)).toMatch(/exceeded max buffer size/i);
|
||||
});
|
||||
|
||||
it("errors on oversized streaming JSON body without content-length in SSE synthesis", async () => {
|
||||
const CHUNK = 1024 * 1024;
|
||||
let sends = 0;
|
||||
fetchWithSsrFGuardMock.mockResolvedValue({
|
||||
response: new Response(
|
||||
new ReadableStream({
|
||||
pull(controller) {
|
||||
if (sends < 17) {
|
||||
sends++;
|
||||
controller.enqueue(new Uint8Array(CHUNK));
|
||||
} else {
|
||||
controller.close();
|
||||
}
|
||||
},
|
||||
}),
|
||||
{ headers: { "content-type": "application/json" } },
|
||||
),
|
||||
finalUrl: "https://openrouter.ai/api/v1/chat/completions",
|
||||
release: vi.fn(async () => undefined),
|
||||
});
|
||||
const model = {
|
||||
id: "moonshotai/kimi-k2.6",
|
||||
provider: "openrouter",
|
||||
api: "openai-completions",
|
||||
baseUrl: "https://openrouter.ai/api/v1",
|
||||
} as unknown as Model<"openai-completions">;
|
||||
|
||||
const response = await buildGuardedModelFetch(model)(
|
||||
"https://openrouter.ai/api/v1/chat/completions",
|
||||
{
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ model: "moonshotai/kimi-k2.6", stream: true }),
|
||||
},
|
||||
);
|
||||
|
||||
const reader = response.body?.getReader();
|
||||
let caught: unknown = null;
|
||||
try {
|
||||
while (true) {
|
||||
const { done } = await reader!.read();
|
||||
if (done) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
} catch (e) {
|
||||
caught = e;
|
||||
}
|
||||
expect(caught).toBeTruthy();
|
||||
expect(String(caught)).toMatch(/exceeded.*bytes while synthesizing SSE/i);
|
||||
});
|
||||
|
||||
describe("long retry-after handling", () => {
|
||||
const anthropicModel = {
|
||||
id: "sonnet-4.6",
|
||||
|
||||
@@ -45,6 +45,17 @@ import {
|
||||
const DEFAULT_MAX_SDK_RETRY_WAIT_SECONDS = 60;
|
||||
const OPENAI_SDK_STREAM_CONTENT_SNIFF_BYTES = 2 * 1024;
|
||||
const log = createSubsystemLogger("provider-transport-fetch");
|
||||
|
||||
/** Max bytes for an entire JSON body synthesized into SSE frames. Prevents OOM
|
||||
* when a hostile streaming endpoint returns a never-ending JSON response
|
||||
* without Content-Length. */
|
||||
const SSE_SYNTHESIZE_JSON_MAX_BYTES = 16 * 1024 * 1024;
|
||||
|
||||
/** Max bytes for the internal SSE sanitization buffer between event boundaries.
|
||||
* A response that cannot find a \n\n boundary within this many characters is
|
||||
* almost certainly hostile or broken — cap the buffer rather than let it grow. */
|
||||
const SSE_SANITIZE_BUFFER_MAX_BYTES = 64 * 1024;
|
||||
|
||||
const BLOCKED_EXACT_ORIGIN_TRUST_HOSTNAME_LABELS = new Set(["instance-data"]);
|
||||
const PLAIN_DECIMAL_NUMBER_RE = /^\d+(?:\.\d+)?$/;
|
||||
const RETRY_AFTER_HTTP_DATE_RE =
|
||||
@@ -102,6 +113,7 @@ function sanitizeOpenAISdkSseResponse(
|
||||
const encoder = new TextEncoder();
|
||||
let reader: ReadableStreamDefaultReader<Uint8Array> | undefined;
|
||||
let buffer = "";
|
||||
let totalBytes = 0;
|
||||
const sseBody = new ReadableStream<Uint8Array>({
|
||||
start() {
|
||||
reader = source.getReader();
|
||||
@@ -120,9 +132,17 @@ function sanitizeOpenAISdkSseResponse(
|
||||
controller.close();
|
||||
return;
|
||||
}
|
||||
const nextTotalBytes = totalBytes + chunk.value.byteLength;
|
||||
if (nextTotalBytes > SSE_SYNTHESIZE_JSON_MAX_BYTES) {
|
||||
throw new Error(
|
||||
`Streaming JSON body exceeded ${SSE_SYNTHESIZE_JSON_MAX_BYTES} bytes while synthesizing SSE frames`,
|
||||
);
|
||||
}
|
||||
totalBytes = nextTotalBytes;
|
||||
buffer += decoder.decode(chunk.value, { stream: true });
|
||||
}
|
||||
} catch (error) {
|
||||
await reader?.cancel(error).catch(() => {});
|
||||
controller.error(error);
|
||||
}
|
||||
},
|
||||
@@ -157,6 +177,11 @@ function sanitizeOpenAISdkSseResponse(
|
||||
for (;;) {
|
||||
const boundary = findSseEventBoundary(buffer);
|
||||
if (!boundary) {
|
||||
if (buffer.length > SSE_SANITIZE_BUFFER_MAX_BYTES) {
|
||||
throw new Error(
|
||||
`SSE response exceeded max buffer size (${SSE_SANITIZE_BUFFER_MAX_BYTES} bytes) without event boundary`,
|
||||
);
|
||||
}
|
||||
return enqueued;
|
||||
}
|
||||
const block = buffer.slice(0, boundary.index);
|
||||
@@ -167,6 +192,7 @@ function sanitizeOpenAISdkSseResponse(
|
||||
if (hasReadableSseData(block)) {
|
||||
controller.enqueue(encoder.encode(`${block}${separator}`));
|
||||
enqueued += 1;
|
||||
return enqueued;
|
||||
}
|
||||
}
|
||||
};
|
||||
@@ -178,6 +204,10 @@ function sanitizeOpenAISdkSseResponse(
|
||||
async pull(controller) {
|
||||
try {
|
||||
for (;;) {
|
||||
const pending = enqueueSanitized(controller, "");
|
||||
if (pending > 0) {
|
||||
return;
|
||||
}
|
||||
const chunk = await reader?.read();
|
||||
if (!chunk || chunk.done) {
|
||||
const tail = decoder.decode();
|
||||
@@ -200,6 +230,7 @@ function sanitizeOpenAISdkSseResponse(
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
await reader?.cancel(error).catch(() => {});
|
||||
controller.error(error);
|
||||
}
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user