| @@ -1148,14 +1148,36 @@ export abstract class McpAgentSessionDOBase< · 1147 unmodified lines |
| 1148 | 1148 | }, |
| 1149 | 1149 | onClose: (reason) => { |
| 1150 | 1150 | this.activeLegacyStreamCount = Math.max(0, this.activeLegacyStreamCount - 1); |
| 1151 | | − if (reason === "complete" && options.acknowledge && options.acknowledge.length > 0) { |
| 1151 | + // "complete": the source body drained to its natural end. "rotate": |
| 1152 | + // the client stayed attached for the whole max-age window, so the |
| 1153 | + // initial replay frames were drained long before. Both count as |
| 1154 | + // delivered. "cancel"/"error" leave the streams replayable for the |
| 1155 | + // client's reconnect GET. |
| 1156 | + if ( |
| 1157 | + (reason === "complete" || reason === "rotate") && |
| 1158 | + options.acknowledge && |
| 1159 | + options.acknowledge.length > 0 |
| 1160 | + ) { |
| 1152 | 1161 | this.ctx.waitUntil(this.eventStore.acknowledgeUndeliveredStreams(options.acknowledge)); |
| 1153 | 1162 | } |
| 1154 | 1163 | }, |
| 1155 | 1164 | }); |
| 1156 | 1165 | |
| 1157 | | − private async replayUndeliveredOnStandaloneGet(request: Request): Promise<Response | null> { |
| 1158 | | − if (request.method !== "GET" || request.headers.has("last-event-id")) return null; |
| 1166 | + /** |
| 1167 | + * Collect every undelivered stream's events as one SSE frame block, for |
| 1168 | + * prepending onto the standalone GET stream. Returning a FINITE response |
| 1169 | + * here instead (the pre-2026-08-17 behavior) was catastrophic: the client's |
| 1170 | + * standalone listener closed the moment the replay flushed, and because |
| 1171 | + * every POST marks its stream undelivered until acknowledged, an active |
| 1172 | + * session ALWAYS had something to replay — so every listener GET became a |
| 1173 | + * ~3s reconnect short-poll. Fleet-wide, that reconnect storm multiplied |
| 1174 | + * /mcp volume ~15x, drove the worker into memory-limit kills, and (by |
| 1175 | + * recycling isolates) pushed homepage TTFB from ~15ms to ~3s. |
| 1176 | + */ |
| 1177 | + private async collectUndeliveredReplay(): Promise<{ |
| 1178 | + readonly frame: Uint8Array; |
| 1179 | + readonly streamIds: readonly string[]; |
| 1180 | + } | null> { |
| 1159 | 1181 | const frames: Uint8Array[] = []; |
| 1160 | 1182 | const streamIds = await this.eventStore.replayUndeliveredStreams({ |
| 1161 | 1183 | send: (eventId, message) => { |
| @@ -1164,18 +1186,7 @@ export abstract class McpAgentSessionDOBase< · 2 unmodified lines |
| 1164 | 1186 | }, |
| 1165 | 1187 | }); |
| 1166 | 1188 | if (frames.length === 0) return null; |
| 1167 | | − return this.trackedLegacyResponse( |
| 1168 | | − new Response(combineFrames(frames), { |
| 1169 | | − headers: { |
| 1170 | | − "content-type": "text/event-stream", |
| 1171 | | − "cache-control": "no-cache, no-transform", |
| 1172 | | − connection: "keep-alive", |
| 1173 | | − "x-accel-buffering": "no", |
| 1174 | | − "mcp-session-id": this.sessionId, |
| 1175 | | − }, |
| 1176 | | − }), |
| 1177 | | − { acknowledge: streamIds }, |
| 1178 | | − ); |
| 1189 | + return { frame: new Uint8Array(combineFrames(frames)), streamIds }; |
| 1179 | 1190 | } |
| 1180 | 1191 | |
| 1181 | 1192 | private async serializedTransportRequest<A>(run: () => Promise<A>): Promise<A> { |
| @@ -1211,8 +1222,19 @@ export abstract class McpAgentSessionDOBase< · 29 unmodified lines |
| 1211 | 1222 | // through every workerd/Vite streaming hop, so explicitly retire a |
| 1212 | 1223 | // stale standalone mapping before opening its replacement. |
| 1213 | 1224 | transport.closeStandaloneSSEStream(); |
| 1214 | | − const replay = await this.replayUndeliveredOnStandaloneGet(request); |
| 1215 | | − if (replay) return replay; |
| 1225 | + const replay = await this.collectUndeliveredReplay(); |
| 1226 | + if (replay) { |
| 1227 | + // Prepend the replay onto the transport's own long-lived |
| 1228 | + // standalone stream — the stream MUST stay open afterwards (see |
| 1229 | + // collectUndeliveredReplay). Acknowledgement moves to stream end: |
| 1230 | + // "complete"/"rotate" imply the client stayed attached long |
| 1231 | + // enough to have drained the prepended frames. |
| 1232 | + const response = await transport.handleRequest(request); |
| 1233 | + return this.trackedLegacyResponse(response, { |
| 1234 | + initialFrame: replay.frame, |
| 1235 | + acknowledge: replay.streamIds, |
| 1236 | + }); |
| 1237 | + } |
| 1216 | 1238 | } |
| 1217 | 1239 | } |
| 1218 | 1240 | const toolCallIds = legacyToolCallRequestIds(parsedBody); |