From 7ac639bd31937e5d4d72183f11a6ee46d11ff65b Mon Sep 17 00:00:00 2001 From: ditadi Date: Wed, 2 Sep 2026 14:44:28 +0200 Subject: [PATCH] fix(stream): raise SSE event limit to 5 MiB and honour per-call maxEventSize Signed-off-by: ditadi --- packages/appkit-ui/src/js/sse/connect-sse.ts | 8 +- packages/appkit/src/stream/defaults.ts | 9 +- packages/appkit/src/stream/stream-manager.ts | 10 ++- .../src/stream/tests/stream-registry.test.ts | 2 + .../appkit/src/stream/tests/stream.test.ts | 90 +++++++++++++++++++ packages/appkit/src/stream/types.ts | 6 ++ 6 files changed, 112 insertions(+), 13 deletions(-) diff --git a/packages/appkit-ui/src/js/sse/connect-sse.ts b/packages/appkit-ui/src/js/sse/connect-sse.ts index 010ba3bf5..0e9eff9e0 100644 --- a/packages/appkit-ui/src/js/sse/connect-sse.ts +++ b/packages/appkit-ui/src/js/sse/connect-sse.ts @@ -18,11 +18,9 @@ export async function connectSSE( lastEventId: initialLastEventId = null, retryDelay = 2000, maxRetries = 3, - // 1 MiB — matches the server's `streamDefaults.maxEventSize`. SSE - // carries only short JSON control messages; bulk Arrow payloads stream - // back as the raw HTTP response body of the query request, so this - // buffer never needs to hold multi-MiB attachments. - maxBufferSize = 1 * 1024 * 1024, + // Matches the server's `streamDefaults.maxEventSize`: a smaller buffer here + // turns a legal server event into "Buffer size exceeded" plus retries. + maxBufferSize = 5 * 1024 * 1024, timeout = 300000, // 5 minutes onError, } = options; diff --git a/packages/appkit/src/stream/defaults.ts b/packages/appkit/src/stream/defaults.ts index fcbdb68f1..d34bafe15 100644 --- a/packages/appkit/src/stream/defaults.ts +++ b/packages/appkit/src/stream/defaults.ts @@ -1,10 +1,9 @@ export const streamDefaults = { bufferSize: 100, - // 1 MiB. SSE carries only short JSON control messages — JSON_ARRAY result - // rows (already row-size-bounded by the warehouse) plus warehouse-readiness - // and error events. ARROW_STREAM never uses SSE: the raw Arrow IPC bytes - // stream back on the query response body (`_handleArrowStreamQuery`). - maxEventSize: 1 * 1024 * 1024, + // Bounds the JSON_ARRAY path only: ARROW_STREAM bytes stream back on the + // query response body (`_handleArrowStreamQuery`), never over SSE. + // Keep in sync with `connectSSE`'s `maxBufferSize` default. + maxEventSize: 5 * 1024 * 1024, bufferTTL: 10 * 60 * 1000, // 10 minutes heartbeatInterval: 10 * 1000, // 10 seconds maxActiveStreams: 1000, // 1000 streams diff --git a/packages/appkit/src/stream/stream-manager.ts b/packages/appkit/src/stream/stream-manager.ts index c68cccbab..9d4840141 100644 --- a/packages/appkit/src/stream/stream-manager.ts +++ b/packages/appkit/src/stream/stream-manager.ts @@ -197,6 +197,7 @@ export class StreamManager { const eventBuffer = new EventRingBuffer( options?.bufferSize ?? streamDefaults.bufferSize, ); + const maxEventSize = options?.maxEventSize ?? this.maxEventSize; // setup signals and heartbeat const combinedSignal = this._combineSignals( @@ -225,6 +226,7 @@ export class StreamManager { lastAccess: Date.now(), abortController, traceContext, + maxEventSize, }; this.streamRegistry.add(streamEntry); @@ -270,10 +272,12 @@ export class StreamManager { if (streamEntry.abortController.signal.aborted) break; const eventId = randomUUID(); const eventData = JSON.stringify(event); + const { maxEventSize } = streamEntry; - // validate event size - if (eventData.length > this.maxEventSize) { - const errorMsg = `Event exceeds max size of ${this.maxEventSize} bytes`; + // UTF-8 bytes, not `String.length`: non-ASCII payloads used to slip + // past a limit they exceeded on the wire. + if (Buffer.byteLength(eventData, "utf8") > maxEventSize) { + const errorMsg = `Event exceeds max size of ${maxEventSize} bytes`; const errorCode = SSEErrorCode.INVALID_REQUEST; // broadcast error to all connected clients this._broadcastErrorToClients( diff --git a/packages/appkit/src/stream/tests/stream-registry.test.ts b/packages/appkit/src/stream/tests/stream-registry.test.ts index a0dfc185d..f52f4cea4 100644 --- a/packages/appkit/src/stream/tests/stream-registry.test.ts +++ b/packages/appkit/src/stream/tests/stream-registry.test.ts @@ -2,6 +2,7 @@ import type { Context } from "@opentelemetry/api"; import { beforeEach, describe, expect, test, vi } from "vitest"; import { EventRingBuffer } from "../buffers"; +import { streamDefaults } from "../defaults"; import { StreamRegistry } from "../stream-registry"; import type { StreamEntry } from "../types"; import { SSEErrorCode } from "../types"; @@ -20,6 +21,7 @@ function createMockStreamEntry( lastAccess: Date.now(), abortController: new AbortController(), traceContext: {} as Context, + maxEventSize: streamDefaults.maxEventSize, ...overrides, }; } diff --git a/packages/appkit/src/stream/tests/stream.test.ts b/packages/appkit/src/stream/tests/stream.test.ts index 06917e52b..15b3b63d6 100644 --- a/packages/appkit/src/stream/tests/stream.test.ts +++ b/packages/appkit/src/stream/tests/stream.test.ts @@ -1,5 +1,6 @@ import { beforeEach, describe, expect, test, vi } from "vitest"; +import { streamDefaults } from "../defaults"; import { StreamManager } from "../index"; function createMockResponse(headers: Record = {}) { @@ -1319,3 +1320,92 @@ describe("StreamManager", () => { }); }); }); + +/** + * `maxEventSize` was read off the StreamManager instance (constructor-only), so + * per-call values passed via `executeStream({ stream })` type-checked and were + * then silently ignored. The constructor path had coverage; the per-call path + * did not, which is how it survived. + */ +describe("maxEventSize", () => { + /** Tracks the default, so raising it can't quietly neuter these tests. */ + function oversizedEvent() { + return { + type: "result", + data: "x".repeat(streamDefaults.maxEventSize + 64), + }; + } + + function errorEvents(events: string[]) { + // SSEWriter emits one res.write per line, so rejoin before parsing + return events + .join("") + .split("\n\n") + .filter((frame) => frame.includes("event: error")) + .map((frame) => { + const line = frame.split("\n").find((l) => l.startsWith("data: ")); + return line ? JSON.parse(line.slice("data: ".length)) : undefined; + }); + } + + test("drops an oversized event under the default limit", async () => { + const manager = new StreamManager(); + const { mockRes, events } = createMockResponse(); + + async function* generator() { + yield oversizedEvent(); + } + + await manager.stream(mockRes as any, generator); + + const errors = errorEvents(events); + expect(errors).toHaveLength(1); + expect(errors[0].code).toBe("INVALID_REQUEST"); + expect(events.some((e) => e.includes("event: result"))).toBe(false); + }); + + test("honours a per-call override", async () => { + const manager = new StreamManager(); + const { mockRes, events } = createMockResponse(); + + async function* generator() { + yield oversizedEvent(); + } + + await manager.stream(mockRes as any, generator, { + maxEventSize: streamDefaults.maxEventSize * 2, + }); + + expect(errorEvents(events)).toHaveLength(0); + expect(events.some((e) => e.includes("event: result"))).toBe(true); + }); + + test("a per-call override beats the constructor value", async () => { + const manager = new StreamManager({ maxEventSize: 10_000 }); + const { mockRes, events } = createMockResponse(); + + async function* generator() { + yield { type: "result", data: "x".repeat(1000) }; + } + + await manager.stream(mockRes as any, generator, { maxEventSize: 50 }); + + expect(errorEvents(events)).toHaveLength(1); + expect(errorEvents(events)[0].error).toContain("50 bytes"); + }); + + test("measures UTF-8 bytes, not UTF-16 code units", async () => { + // ~730 UTF-16 units (under the limit, so String.length accepted it) but + // ~1430 UTF-8 bytes, which is what crosses the wire. + const manager = new StreamManager({ maxEventSize: 1000 }); + const { mockRes, events } = createMockResponse(); + + async function* generator() { + yield { type: "result", data: "\u00e9".repeat(700) }; + } + + await manager.stream(mockRes as any, generator); + + expect(errorEvents(events)).toHaveLength(1); + }); +}); diff --git a/packages/appkit/src/stream/types.ts b/packages/appkit/src/stream/types.ts index 07c771106..74a982003 100644 --- a/packages/appkit/src/stream/types.ts +++ b/packages/appkit/src/stream/types.ts @@ -56,6 +56,12 @@ export interface StreamEntry { lastAccess: number; abortController: AbortController; traceContext: Context; + /** + * UTF-8 bytes. Lives on the entry, not on `StreamManager`: one manager is + * shared by every stream a plugin opens, so reading it off the instance made + * per-call overrides silently ineffective. + */ + maxEventSize: number; // pending grace-window abort, set while the last client is disconnected disconnectGraceTimer?: NodeJS.Timeout; /**