History

Rotate long-lived MCP SSE streams after a max age to stop stalled-client memory buildup (#1279)

* Rotate long-lived MCP SSE streams after a max age to stop stalled-client memory buildup Stalled SSE clients (socket open, reader gone) accumulate buffered output inside the runtime with no JS-observable backpressure signal: writes resolve promptly even when the peer stops reading, so no stall detector on the write path can ever fire. Connections in this state survive for over an hour in production and eventually push the isolate past its memory limit, killing every connection it hosts. Since detection is impossible, bound the damage instead: close every long-lived SSE stream (streamable-http GET channel and legacy SSE) once it exceeds MAX_SSE_AGE_MS (5 minutes), checked on the existing keepalive tick. Healthy clients transparently reconnect and resume via Last-Event-ID against the default DO-backed event store; stalled clients stop accumulating. Each close emits a structured sse_max_age_close log with session id and variant. The POST response stream is deliberately NOT rotated: those streams are finite (closed by the DO when the response completes) so they cannot leak unboundedly, and rotating one mid-response orphans the in-flight result -- the client SDK can only resume a stream that has already delivered an event id, which an in-progress response may not have. The existing 8 MB pending-bytes cap still bounds pathological POST readers. Validated end-to-end on a deployed worker with a real MCP SDK client: GET-channel rotation survived two full cycles with zero client-visible errors and push notifications delivered across the rotation boundary, while the load profile that previously produced repeated exceededMemory isolate deaths ran clean with rotation enabled. * Fix lint and add URLPattern polyfill for CI test environment

Rhys Sullivan committed 4 months ago

Commit 9bbf5d41 parent d74dc558 files · +484 −34Browse files

Unified · Split

8 files
  1. bun.lock +1 −0
  2. packages/hosts/cloudflare/package.json +1 −0
  3. packages/hosts/cloudflare/src/mcp/agents-sse-max-age.test.ts +344 −0
  4. packages/hosts/cloudflare/src/test-setup.ts +9 −0
  5. packages/hosts/cloudflare/src/test-stubs/cloudflare-email.ts +7 −0
  6. packages/hosts/cloudflare/src/test-stubs/cloudflare-workers.ts +9 −0
  7. packages/hosts/cloudflare/vitest.config.ts +16 −0
  8. patches/agents@0.17.3.patch +97 −34
bun.lock+1−0
@@ -664,6 +664,7 @@ · 663 unmodified lines
664 "@types/node": "catalog:",664 "@types/node": "catalog:",
665 "bun-types": "catalog:",665 "bun-types": "catalog:",
666 "typescript": "catalog:",666 "typescript": "catalog:",
667 "urlpattern-polyfill": "^10.1.0",
667 "vitest": "catalog:",668 "vitest": "catalog:",
668 },669 },
669 },670 },
packages/hosts/cloudflare/package.json+1−0
@@ -37,6 +37,7 @@ · 36 unmodified lines
37 "@types/node": "catalog:",37 "@types/node": "catalog:",
38 "bun-types": "catalog:",38 "bun-types": "catalog:",
39 "typescript": "catalog:",39 "typescript": "catalog:",
40 "urlpattern-polyfill": "^10.1.0",
40 "vitest": "catalog:"41 "vitest": "catalog:"
41 }42 }
42}43}
packages/hosts/cloudflare/src/mcp/agents-sse-max-age.test.ts+344−0
@@ -0,0 +1,344 @@
1import { afterEach, beforeEach, describe, expect, it, vi } from "@effect/vitest";
2import { MAX_SSE_AGE_MS, McpAgent } from "agents/mcp";
3import { Effect, Option, Schema } from "effect";
4
5const KEEPALIVE_INTERVAL_MS = 25_000;
6const MAX_PENDING_SSE_BYTES = 8 * 1024 * 1024;
7
8type FakeWebSocket = EventTarget & {
9 accepted: boolean;
10 closeCode: number | undefined;
11 closeReason: string | undefined;
12 accept: () => void;
13 close: (code?: number, reason?: string) => void;
14};
15
16type FakeAgentStub = {
17 readonly setName: (name: string, props?: unknown) => Promise<void>;
18 readonly getInitializeRequest: () => Promise<unknown>;
19 readonly fetch: (request: Request) => Promise<{ readonly webSocket: FakeWebSocket }>;
20};
21
22type RotationLog = {
23 readonly event: "sse_max_age_close";
24 readonly sessionId: string;
25 readonly variant: "streamable-get" | "streamable-post" | "legacy-sse";
26 readonly ageMs: number;
27 readonly pendingBytes: number;
28};
29
30const encoder = new TextEncoder();
31const RotationLogEvent = Schema.Struct({
32 ageMs: Schema.Number,
33 event: Schema.Literal("sse_max_age_close"),
34 pendingBytes: Schema.Number,
35 sessionId: Schema.String,
36 variant: Schema.Union([
37 Schema.Literal("streamable-get"),
38 Schema.Literal("streamable-post"),
39 Schema.Literal("legacy-sse"),
40 ]),
41});
42const decodeRotationLogEvent = Schema.decodeUnknownOption(Schema.fromJsonString(RotationLogEvent));
43
44const flushMicrotasks = async (): Promise<void> => {
45 await Promise.resolve();
46 await Promise.resolve();
47 await Promise.resolve();
48};
49
50const waitFor = async (predicate: () => boolean): Promise<void> => {
51 for (let attempt = 0; attempt < 20; attempt += 1) {
52 if (predicate()) return;
53 await flushMicrotasks();
54 }
55 expect(predicate()).toBe(true);
56};
57
58const drainResponse = (response: Response): Promise<void> => {
59 return Effect.runPromise(
60 Effect.ignore(
61 Effect.tryPromise({
62 try: () =>
63 response.body?.pipeTo(
64 new WritableStream<Uint8Array>({
65 write: () => {},
66 }),
67 ) ?? Promise.resolve(),
68 catch: () => undefined,
69 }),
70 ),
71 );
72};
73
74const installStallingTransformStream = () => {
75 let abortReason: unknown;
76 let writeCount = 0;
77 let stalledWriteStarted: (() => void) | undefined;
78 const stalledWrite = new Promise<void>((resolve) => {
79 stalledWriteStarted = resolve;
80 });
81 const writer = {
82 abort: (reason: unknown) => {
83 abortReason = reason;
84 return Promise.resolve();
85 },
86 close: () => Promise.resolve(),
87 write: () => {
88 writeCount += 1;
89 stalledWriteStarted?.();
90 return new Promise<void>(() => {});
91 },
92 };
93
94 vi.stubGlobal(
95 "TransformStream",
96 class {
97 readonly readable = new ReadableStream<Uint8Array>();
98 readonly writable = {
99 getWriter: () => writer,
100 };
101 },
102 );
103
104 return {
105 abortReason: () => abortReason,
106 stalledWrite,
107 writeCount: () => writeCount,
108 };
109};
110
111const makeExecutionContext = (): ExecutionContext => ({
112 passThroughOnException: () => {},
113 props: undefined,
114 waitUntil: () => {},
115});
116
117const makeWebSocket = (): FakeWebSocket => {
118 const ws = new EventTarget() as FakeWebSocket;
119 ws.accepted = false;
120 ws.closeCode = undefined;
121 ws.closeReason = undefined;
122 ws.accept = () => {
123 ws.accepted = true;
124 };
125 ws.close = (code?: number, reason?: string) => {
126 ws.closeCode = code;
127 ws.closeReason = reason;
128 };
129 return ws;
130};
131
132const makeAgentStub = (ws: FakeWebSocket): FakeAgentStub => ({
133 setName: async () => {},
134 getInitializeRequest: async () => ({}),
135 fetch: async () => ({ webSocket: ws }),
136});
137
138const makeNamespace = (agent: FakeAgentStub) => ({
139 newUniqueId: () => ({ toString: () => "generated-session" }),
140 idFromName: (name: string) => ({
141 equals: () => true,
142 name,
143 toString: () => name,
144 }),
145 get: () => agent,
146});
147
148const openSse = async () => {
149 const ws = makeWebSocket();
150 const agent = makeAgentStub(ws);
151 const namespace = makeNamespace(agent);
152 const handler = McpAgent.serve("/mcp", {
153 binding: "MCP_SESSION",
154 transport: "streamable-http",
155 });
156 const response = await handler.fetch(
157 new Request("https://executor.sh/mcp", {
158 headers: {
159 accept: "text/event-stream",
160 "mcp-session-id": "session-1",
161 },
162 method: "GET",
163 }),
164 { MCP_SESSION: namespace },
165 makeExecutionContext(),
166 );
167
168 expect(response.status).toBe(200);
169 expect(ws.accepted).toBe(true);
170 expect(response.body).toBeDefined();
171
172 return { response, ws };
173};
174
175const openPostSse = async () => {
176 const ws = makeWebSocket();
177 const agent = makeAgentStub(ws);
178 const namespace = makeNamespace(agent);
179 const handler = McpAgent.serve("/mcp", {
180 binding: "MCP_SESSION",
181 transport: "streamable-http",
182 });
183 const response = await handler.fetch(
184 new Request("https://executor.sh/mcp", {
185 body: JSON.stringify({
186 id: 1,
187 jsonrpc: "2.0",
188 method: "tools/call",
189 params: {
190 arguments: {},
191 name: "example",
192 },
193 }),
194 headers: {
195 accept: "application/json, text/event-stream",
196 "content-type": "application/json",
197 "mcp-session-id": "session-1",
198 },
199 method: "POST",
200 }),
201 { MCP_SESSION: namespace },
202 makeExecutionContext(),
203 );
204
205 expect(response.status).toBe(200);
206 expect(ws.accepted).toBe(true);
207 expect(response.body).toBeDefined();
208
209 return { response, ws };
210};
211
212const emitAgentEvent = (ws: FakeWebSocket, event: string, close?: true): void => {
213 ws.dispatchEvent(
214 new MessageEvent("message", {
215 data: JSON.stringify({
216 close,
217 event,
218 type: "cf_mcp_agent_event",
219 }),
220 }),
221 );
222};
223
224const emitClose = (ws: FakeWebSocket): void => {
225 ws.dispatchEvent(new Event("close"));
226};
227
228const rotationLogs = (logs: ReadonlyArray<string>): ReadonlyArray<RotationLog> =>
229 logs.flatMap((line) => {
230 const decoded = decodeRotationLogEvent(line);
231 return Option.isSome(decoded) ? [decoded.value] : [];
232 });
233
234describe("agents SSE max-age rotation", () => {
235 let errorLogs: string[] = [];
236
237 beforeEach(() => {
238 errorLogs = [];
239 vi.useFakeTimers();
240 vi.setSystemTime(0);
241 vi.spyOn(console, "error").mockImplementation((line) => {
242 errorLogs.push(String(line));
243 });
244 });
245
246 afterEach(() => {
247 vi.unstubAllGlobals();
248 vi.restoreAllMocks();
249 vi.useRealTimers();
250 });
251
252 it("closes a healthy draining SSE connection within one keepalive tick after max age", async () => {
253 const { response, ws } = await openSse();
254 const drained = drainResponse(response);
255
256 await vi.advanceTimersByTimeAsync(MAX_SSE_AGE_MS + KEEPALIVE_INTERVAL_MS);
257 await waitFor(() => ws.closeCode === 1013);
258
259 expect(ws.closeReason).toBe("SSE client not draining");
260 const [rotationLog] = rotationLogs(errorLogs);
261 expect(rotationLog?.event).toBe("sse_max_age_close");
262 expect(rotationLog?.ageMs).toBeGreaterThan(MAX_SSE_AGE_MS);
263 expect(rotationLog?.ageMs).toBeLessThanOrEqual(MAX_SSE_AGE_MS + KEEPALIVE_INTERVAL_MS);
264 expect(rotationLog?.pendingBytes).toBeGreaterThanOrEqual(0);
265 expect(vi.getTimerCount()).toBe(0);
266
267 await drained;
268 });
269
270 it("does not rotate an in-flight POST response past max age", async () => {
271 const { response, ws } = await openPostSse();
272 const drained = drainResponse(response);
273
274 emitAgentEvent(ws, `event: message\ndata: {"jsonrpc":"2.0","id":1,"result":{}}\n\n`);
275 await vi.advanceTimersByTimeAsync(MAX_SSE_AGE_MS + KEEPALIVE_INTERVAL_MS * 4);
276 await flushMicrotasks();
277
278 expect(ws.closeCode).toBeUndefined();
279 expect(ws.closeReason).toBeUndefined();
280 expect(rotationLogs(errorLogs)).toEqual([]);
281
282 emitAgentEvent(
283 ws,
284 `event: message\ndata: {"jsonrpc":"2.0","id":1,"result":{"ok":true}}\n\n`,
285 true,
286 );
287 await drained;
288
289 expect(ws.closeCode).toBeUndefined();
290 expect(ws.closeReason).toBeUndefined();
291 expect(vi.getTimerCount()).toBe(0);
292 });
293
294 it("leaves an SSE connection younger than max age untouched", async () => {
295 const { response, ws } = await openSse();
296 const drained = drainResponse(response);
297
298 await vi.advanceTimersByTimeAsync(MAX_SSE_AGE_MS - KEEPALIVE_INTERVAL_MS * 2);
299 await flushMicrotasks();
300
301 expect(ws.closeCode).toBeUndefined();
302 expect(rotationLogs(errorLogs)).toEqual([]);
303
304 emitClose(ws);
305 await drained;
306 expect(vi.getTimerCount()).toBe(0);
307 });
308
309 it("still closes a stalled SSE writer at the byte cap without logging rotation", async () => {
310 const stalledFrame = `event: message\ndata: ${"x".repeat(2 * 1024 * 1024)}\n\n`;
311 const transform = installStallingTransformStream();
312 const { ws } = await openSse();
313
314 emitAgentEvent(ws, stalledFrame);
315 await transform.stalledWrite;
316 expect(transform.writeCount()).toBe(1);
317
318 const stalledFrameBytes = encoder.encode(stalledFrame).byteLength;
319 expect(stalledFrameBytes).toBeLessThan(MAX_PENDING_SSE_BYTES);
320
321 for (let attempt = 0; attempt < 8 && ws.closeCode === undefined; attempt += 1) {
322 emitAgentEvent(ws, stalledFrame);
323 }
324
325 expect(ws.closeCode).toBe(1013);
326 expect(ws.closeReason).toBe("SSE client not draining");
327 expect(transform.abortReason()).toBeInstanceOf(Error);
328 expect(rotationLogs(errorLogs)).toEqual([]);
329 expect(vi.getTimerCount()).toBe(0);
330 });
331
332 it("cleans up timers when the client closes before max age", async () => {
333 const { response, ws } = await openSse();
334 const drained = drainResponse(response);
335
336 await vi.advanceTimersByTimeAsync(KEEPALIVE_INTERVAL_MS);
337 emitClose(ws);
338 await drained;
339
340 expect(ws.closeCode).toBeUndefined();
341 expect(rotationLogs(errorLogs)).toEqual([]);
342 expect(vi.getTimerCount()).toBe(0);
343 });
344});
packages/hosts/cloudflare/src/test-setup.ts+9−0
@@ -0,0 +1,9 @@
1import { URLPattern } from "urlpattern-polyfill";
2
3type UrlPatternPolyfillGlobal = Omit<typeof globalThis, "URLPattern"> & {
4 URLPattern?: typeof URLPattern;
5};
6
7const globals = globalThis as UrlPatternPolyfillGlobal;
8
9globals.URLPattern ??= URLPattern;
packages/hosts/cloudflare/src/test-stubs/cloudflare-email.ts+7−0
@@ -0,0 +1,7 @@
1export class EmailMessage {
2 constructor(
3 readonly from: string,
4 readonly to: string,
5 readonly raw: ReadableStream | string,
6 ) {}
7}
packages/hosts/cloudflare/src/test-stubs/cloudflare-workers.ts+9−0
@@ -0,0 +1,9 @@
1export const env: Record<string, unknown> = {};
2
3export class DurableObject {}
4
5export class RpcTarget {}
6
7export class WorkerEntrypoint {}
8
9export const exports: Record<string, unknown> = {};
packages/hosts/cloudflare/vitest.config.ts+16−0
@@ -1,8 +1,24 @@
1import { dirname, resolve } from "node:path";
2import { fileURLToPath } from "node:url";
1import { defineConfig } from "vitest/config";3import { defineConfig } from "vitest/config";
2 4
5const __dirname = dirname(fileURLToPath(import.meta.url));
6
3export default defineConfig({7export default defineConfig({
8 resolve: {
9 alias: {
10 "cloudflare:email": resolve(__dirname, "./src/test-stubs/cloudflare-email.ts"),
11 "cloudflare:workers": resolve(__dirname, "./src/test-stubs/cloudflare-workers.ts"),
12 },
13 },
4 test: {14 test: {
5 include: ["src/**/*.test.ts"],15 include: ["src/**/*.test.ts"],
6 passWithNoTests: true,16 passWithNoTests: true,
17 setupFiles: ["./src/test-setup.ts"],
18 server: {
19 deps: {
20 inline: ["agents", "partyserver"],
21 },
22 },
7 },23 },
8});24});
patches/agents@0.17.3.patch+97−34
@@ -1,8 +1,33 @@
1diff --git a/dist/mcp/index.d.ts b/dist/mcp/index.d.ts
2index c8fad44..439da01 100644
3--- a/dist/mcp/index.d.ts
4+++ b/dist/mcp/index.d.ts
5@@ -29,6 +29,7 @@ import {
6 xt as MCPClientOAuthCallbackConfig,
7 zt as ElicitResult
8 } from "../agent-tool-types-CNyE1iz_.js";
9+declare const MAX_SSE_AGE_MS = 300000;
10 export {
11 type ClearableEventStore,
12 type CreateMcpHandlerOptions,
13@@ -41,6 +42,7 @@ export {
14 type MCPConnectionResult,
15 type MCPDiscoverResult,
16 type MCPServerOptions,
17+ MAX_SSE_AGE_MS,
18 MCP_SERVER_ID_MAX_LENGTH,
19 McpAgent,
20 type McpAuthContext,
1diff --git a/dist/mcp/index.js b/dist/mcp/index.js21diff --git a/dist/mcp/index.js b/dist/mcp/index.js
2index 1edcf0c8..e65cd346 10064422index 1edcf0c..707f1cf 100644
3--- a/dist/mcp/index.js23--- a/dist/mcp/index.js
4+++ b/dist/mcp/index.js24+++ b/dist/mcp/index.js
5@@ -32,9 +32,9 @@ const KEEPALIVE_FRAME = ": keepalive\n\n";25@@ -28,13 +28,14 @@ import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/
26 const KEEPALIVE_INTERVAL_MS = 25e3;
27 /** SSE comment frame the parser drops before any event dispatch. */
28 const KEEPALIVE_FRAME = ": keepalive\n\n";
29+const MAX_SSE_AGE_MS = 300_000;
30 /**
6 * Start an SSE keepalive on `writer`. Returns a `clearInterval` handle31 * Start an SSE keepalive on `writer`. Returns a `clearInterval` handle
7 * that the stream cleanup must invoke when the stream closes.32 * that the stream cleanup must invoke when the stream closes.
8 */33 */
@@ -14,7 +39,7 @@ index 1edcf0c8..e65cd346 100644 · 5 unmodified lines
14 }, KEEPALIVE_INTERVAL_MS);39 }, KEEPALIVE_INTERVAL_MS);
15 return handle;40 return handle;
16 }41 }
17@@ -180,10 +180,15 @@ const createStreamingHttpHandler = (basePath, namespace, options = {}) => {42@@ -180,10 +181,15 @@ const createStreamingHttpHandler = (basePath, namespace, options = {}) => {
18 });43 });
19 return new Response(body, { status: 404 });44 return new Response(body, { status: 404 });
20 }45 }
@@ -34,7 +59,7 @@ index 1edcf0c8..e65cd346 100644 · 13 unmodified lines
34 request.headers.forEach((value, key) => {59 request.headers.forEach((value, key) => {
35 existingHeaders[key] = value;60 existingHeaders[key] = value;
36 });61 });
37@@ -206,45 +211,68 @@ const createStreamingHttpHandler = (basePath, namespace, options = {}) => {62@@ -206,45 +212,71 @@ const createStreamingHttpHandler = (basePath, namespace, options = {}) => {
38 jsonrpc: "2.0"63 jsonrpc: "2.0"
39 });64 });
40 return new Response(body, { status: 500 });65 return new Response(body, { status: 500 });
@@ -45,17 +70,20 @@ index 1edcf0c8..e65cd346 100644 · 4 unmodified lines
45- return new Response(null, {70- return new Response(null, {
46+ }71+ }
47+ ws.accept();72+ ws.accept();
73+ const __closeSse = () => {
74+ __sseClosed = true;
75+ try {
76+ clearInterval(keepAlive);
77+ } catch {}
78+ try {
79+ ws.close(1013, "SSE client not draining");
80+ } catch {}
81+ writer.abort(new Error("SSE client not draining")).catch(() => {});
82+ };
48+ const __forwardSse = (frame) => {83+ const __forwardSse = (frame) => {
49+ if (__sseClosed) return __writeChain;84+ if (__sseClosed) return __writeChain;
50+ if (__pendingBytes + frame.byteLength > MAX_PENDING_SSE_BYTES) {85+ if (__pendingBytes + frame.byteLength > MAX_PENDING_SSE_BYTES) {
51+ __sseClosed = true;86+ __closeSse();
52+ try {
53+ clearInterval(keepAlive);
54+ } catch {}
55+ try {
56+ ws.close(1013, "SSE client not draining");
57+ } catch {}
58+ writer.abort(new Error("SSE client not draining")).catch(() => {});
59+ return Promise.resolve();87+ return Promise.resolve();
60+ }88+ }
61+ __pendingBytes += frame.byteLength;89+ __pendingBytes += frame.byteLength;
@@ -129,7 +157,7 @@ index 1edcf0c8..e65cd346 100644 · 67 unmodified lines
129 }157 }
130 onClose().catch(console.error);158 onClose().catch(console.error);
131 });159 });
132@@ -279,10 +307,15 @@ const createStreamingHttpHandler = (basePath, namespace, options = {}) => {160@@ -279,10 +311,16 @@ const createStreamingHttpHandler = (basePath, namespace, options = {}) => {
133 id: null,161 id: null,
134 jsonrpc: "2.0"162 jsonrpc: "2.0"
135 }), { status: 400 });163 }), { status: 400 });
@@ -140,6 +168,7 @@ index 1edcf0c8..e65cd346 100644 · 4 unmodified lines
140+ const { readable, writable } = new TransformStream();168+ const { readable, writable } = new TransformStream();
141+ const writer = writable.getWriter();169+ const writer = writable.getWriter();
142+ const encoder = new TextEncoder();170+ const encoder = new TextEncoder();
171+ const __openedAt = Date.now();
143+ const MAX_PENDING_SSE_BYTES = 8 * 1024 * 1024;172+ const MAX_PENDING_SSE_BYTES = 8 * 1024 * 1024;
144+ let __pendingBytes = 0;173+ let __pendingBytes = 0;
145+ let __writeChain = Promise.resolve();174+ let __writeChain = Promise.resolve();
@@ -149,7 +178,7 @@ index 1edcf0c8..e65cd346 100644 · 3 unmodified lines
149 props: ctx.props,178 props: ctx.props,
150 jurisdiction: options.jurisdiction179 jurisdiction: options.jurisdiction
151 });180 });
152@@ -306,27 +339,51 @@ const createStreamingHttpHandler = (basePath, namespace, options = {}) => {181@@ -306,27 +344,66 @@ const createStreamingHttpHandler = (basePath, namespace, options = {}) => {
153 if (!ws) {182 if (!ws) {
154 await writer.close();183 await writer.close();
155 return new Response("Failed to establish WS to DO", { status: 500 });184 return new Response("Failed to establish WS to DO", { status: 500 });
@@ -158,17 +187,32 @@ index 1edcf0c8..e65cd346 100644 · 2 unmodified lines
158- ws.addEventListener("message", (event) => {187- ws.addEventListener("message", (event) => {
159+ }188+ }
160+ ws.accept();189+ ws.accept();
190+ const __closeSse = () => {
191+ __sseClosed = true;
192+ try {
193+ clearInterval(keepAlive);
194+ } catch {}
195+ try {
196+ ws.close(1013, "SSE client not draining");
197+ } catch {}
198+ writer.abort(new Error("SSE client not draining")).catch(() => {});
199+ };
161+ const __forwardSse = (frame) => {200+ const __forwardSse = (frame) => {
162+ if (__sseClosed) return __writeChain;201+ if (__sseClosed) return __writeChain;
202+ const ageMs = Date.now() - __openedAt;
203+ if (ageMs > MAX_SSE_AGE_MS) {
204+ console.error(JSON.stringify({
205+ event: "sse_max_age_close",
206+ sessionId,
207+ variant: "streamable-get",
208+ ageMs,
209+ pendingBytes: __pendingBytes
210+ }));
211+ __closeSse();
212+ return Promise.resolve();
213+ }
163+ if (__pendingBytes + frame.byteLength > MAX_PENDING_SSE_BYTES) {214+ if (__pendingBytes + frame.byteLength > MAX_PENDING_SSE_BYTES) {
164+ __sseClosed = true;215+ __closeSse();
165+ try {
166+ clearInterval(keepAlive);
167+ } catch {}
168+ try {
169+ ws.close(1013, "SSE client not draining");
170+ } catch {}
171+ writer.abort(new Error("SSE client not draining")).catch(() => {});
172+ return Promise.resolve();216+ return Promise.resolve();
173+ }217+ }
174+ __pendingBytes += frame.byteLength;218+ __pendingBytes += frame.byteLength;
@@ -216,7 +260,7 @@ index 1edcf0c8..e65cd346 100644 · 41 unmodified lines
216 return new Response(readable, {260 return new Response(readable, {
217 headers: {261 headers: {
218 "Cache-Control": "no-cache",262 "Cache-Control": "no-cache",
219@@ -389,10 +446,15 @@ const createLegacySseHandler = (basePath, namespace, options = {}) => {263@@ -389,10 +466,16 @@ const createLegacySseHandler = (basePath, namespace, options = {}) => {
220 const url = new URL(request.url);264 const url = new URL(request.url);
221 if (request.method === "GET" && basePattern.test(url)) {265 if (request.method === "GET" && basePattern.test(url)) {
222 const sessionId = url.searchParams.get("sessionId") || namespace.newUniqueId().toString();266 const sessionId = url.searchParams.get("sessionId") || namespace.newUniqueId().toString();
@@ -227,6 +271,7 @@ index 1edcf0c8..e65cd346 100644 · 4 unmodified lines
227+ const { readable, writable } = new TransformStream();271+ const { readable, writable } = new TransformStream();
228+ const writer = writable.getWriter();272+ const writer = writable.getWriter();
229+ const encoder = new TextEncoder();273+ const encoder = new TextEncoder();
274+ const __openedAt = Date.now();
230+ const MAX_PENDING_SSE_BYTES = 8 * 1024 * 1024;275+ const MAX_PENDING_SSE_BYTES = 8 * 1024 * 1024;
231+ let __pendingBytes = 0;276+ let __pendingBytes = 0;
232+ let __writeChain = Promise.resolve();277+ let __writeChain = Promise.resolve();
@@ -236,7 +281,7 @@ index 1edcf0c8..e65cd346 100644 · 3 unmodified lines
236 endpointUrl.pathname = encodeURI(`${basePath}/message`);281 endpointUrl.pathname = encodeURI(`${basePath}/message`);
237 endpointUrl.searchParams.set("sessionId", sessionId);282 endpointUrl.searchParams.set("sessionId", sessionId);
238 const endpointMessage = `event: endpoint\ndata: ${endpointUrl.pathname + endpointUrl.search + endpointUrl.hash}\n\n`;283 const endpointMessage = `event: endpoint\ndata: ${endpointUrl.pathname + endpointUrl.search + endpointUrl.hash}\n\n`;
239@@ -414,35 +476,59 @@ const createLegacySseHandler = (basePath, namespace, options = {}) => {284@@ -414,35 +497,74 @@ const createLegacySseHandler = (basePath, namespace, options = {}) => {
240 console.error("Failed to establish WebSocket connection");285 console.error("Failed to establish WebSocket connection");
241 await writer.close();286 await writer.close();
242 return new Response("Failed to establish WebSocket connection", { status: 500 });287 return new Response("Failed to establish WebSocket connection", { status: 500 });
@@ -245,17 +290,32 @@ index 1edcf0c8..e65cd346 100644 · 2 unmodified lines
245- ws.addEventListener("message", (event) => {290- ws.addEventListener("message", (event) => {
246+ }291+ }
247+ ws.accept();292+ ws.accept();
293+ const __closeSse = () => {
294+ __sseClosed = true;
295+ try {
296+ clearInterval(keepAlive);
297+ } catch {}
298+ try {
299+ ws.close(1013, "SSE client not draining");
300+ } catch {}
301+ writer.abort(new Error("SSE client not draining")).catch(() => {});
302+ };
248+ const __forwardSse = (frame) => {303+ const __forwardSse = (frame) => {
249+ if (__sseClosed) return __writeChain;304+ if (__sseClosed) return __writeChain;
305+ const ageMs = Date.now() - __openedAt;
306+ if (ageMs > MAX_SSE_AGE_MS) {
307+ console.error(JSON.stringify({
308+ event: "sse_max_age_close",
309+ sessionId,
310+ variant: "legacy-sse",
311+ ageMs,
312+ pendingBytes: __pendingBytes
313+ }));
314+ __closeSse();
315+ return Promise.resolve();
316+ }
250+ if (__pendingBytes + frame.byteLength > MAX_PENDING_SSE_BYTES) {317+ if (__pendingBytes + frame.byteLength > MAX_PENDING_SSE_BYTES) {
251+ __sseClosed = true;318+ __closeSse();
252+ try {
253+ clearInterval(keepAlive);
254+ } catch {}
255+ try {
256+ ws.close(1013, "SSE client not draining");
257+ } catch {}
258+ writer.abort(new Error("SSE client not draining")).catch(() => {});
259+ return Promise.resolve();319+ return Promise.resolve();
260+ }320+ }
261+ __pendingBytes += frame.byteLength;321+ __pendingBytes += frame.byteLength;
@@ -314,9 +374,12 @@ index 1edcf0c8..e65cd346 100644 · 52 unmodified lines
314 console.error("Error closing SSE connection:", error);374 console.error("Error closing SSE connection:", error);
315 }375 }
316 }376 }
317@@ -1700,4 +1786,4 @@ McpAgent.STREAM_REQS_KEY_PREFIX = "__mcp_stream_reqs__:";377@@ -1698,6 +1820,6 @@ var McpAgent = class McpAgent extends Agent {
378 };
379 McpAgent.STREAM_REQS_KEY_PREFIX = "__mcp_stream_reqs__:";
318 //#endregion380 //#endregion
319 export { DurableObjectEventStore, ElicitRequestSchema, MCP_SERVER_ID_MAX_LENGTH, McpAgent, RPCClientTransport, RPCServerTransport, RPC_DO_PREFIX, SSEEdgeClientTransport, StreamableHTTPEdgeClientTransport, WorkerTransport, createMcpHandler, experimental_createMcpHandler, getMcpAuthContext, normalizeServerId };381-export { DurableObjectEventStore, ElicitRequestSchema, MCP_SERVER_ID_MAX_LENGTH, McpAgent, RPCClientTransport, RPCServerTransport, RPC_DO_PREFIX, SSEEdgeClientTransport, StreamableHTTPEdgeClientTransport, WorkerTransport, createMcpHandler, experimental_createMcpHandler, getMcpAuthContext, normalizeServerId };
382+export { DurableObjectEventStore, ElicitRequestSchema, MAX_SSE_AGE_MS, MCP_SERVER_ID_MAX_LENGTH, McpAgent, RPCClientTransport, RPCServerTransport, RPC_DO_PREFIX, SSEEdgeClientTransport, StreamableHTTPEdgeClientTransport, WorkerTransport, createMcpHandler, experimental_createMcpHandler, getMcpAuthContext, normalizeServerId };
320 383
321-//# sourceMappingURL=index.js.map384-//# sourceMappingURL=index.js.map
322\ No newline at end of file385\ No newline at end of file