fix(mcp): replay every event after POST stream interruption (#1583 follow-up)#1607
Conversation
🦋 Changeset detectedLatest commit: 22b9e0f The changes in this PR will be included in the next version bump. This PR includes changesets to release 1 package
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
fff89fd to
943188d
Compare
agents
@cloudflare/ai-chat
@cloudflare/codemode
hono-agents
@cloudflare/shell
@cloudflare/think
@cloudflare/voice
@cloudflare/worker-bundler
commit: |
…e#1583 followup) Reworking the resumability cleanup after a series of review rounds. Net result: -233 lines from the previous followup, no functional regression on the actual fixes. The Important Fixes (unchanged from earlier rounds) 1. Drop clearStream-on-shouldClose. Final tool response stays in the event store so a client that lost the connection right as it was in flight can reconnect with Last-Event-ID and recover it. 2. Unconditional storage on the request-bound send() path, matching the SDK. Falls back to McpAgent.getStreamIdForRequestId (a reverse lookup against __mcp_stream_reqs__:*) when the originating WS has dropped, so events are still recorded for replay. 3. Resumed GET registers under the source streamId, matching the SDK exactly. Active POST resumption inherits requestIds; standalone resumption takes over the listen role; completed POST resumption is a one-shot replay channel. The Storage Bounding (simplified) Earlier rounds layered cap-based eviction + a complex per-event-or- per-stream timestamped sweep + a recurring cron. After working through it: - Per-stream cap removed. Real streams never approach 256 events, and truncating replay history silently was arguably worse than letting the stream grow. DO storage is bounded structurally anyway. - Sweep is alarm-driven, not cron. storeEvent fires onStoreEvent; McpAgent uses it to idempotent-schedule a single cleanup at now + maxAgeMs. The callback sweeps streams whose lastWriteAt is past cutoff, then either reschedules at the next-earliest expiry or doesn't reschedule at all. Quiescent DOs do no periodic work. - Sweep cost is O(active streams), not O(total events): a per-stream metadata key (__mcp_stream_evt_meta__) tracks lastWriteAt, the sweep scans only that index and bulk-deletes events for streams whose meta is expired. - Stream-reqs cleanup folded into the same sweep. When sweep deletes a stream, the agent also deletes the matching __mcp_stream_reqs__ entry. No separate sweepStreamReqs, no timestamp on reqs entries. - All the knobs removed: getEventStoreSweepCron, getEventStoreSweepMaxIterations, batchSize, the cursor-pagination question. Just getEventStoreMaxAgeMs (default 24h). DurableObjectEventStore is exported so callers embedding WorkerTransport inside an Agent (e.g. the elicitation example) can opt into resumability with new DurableObjectEventStore(this.ctx.storage). Tests: 455/455 MCP tests pass. npm run check exit 0 across all 88 projects. event-store.ts is now 198 lines (down from 323); McpAgent additions for cleanup are ~50 lines instead of ~200.
943188d to
10844dc
Compare
|
clanker review, some of these look legit: High SeverityStale POST WebSockets can win routing after GET resume
const liveConnection = Array.from(
agent.getConnections<TransportConnState>()
).find((conn) => conn.state?.requestIds?.includes(requestId as RequestId));If the stale POST bridge is iterated first, subsequent tool progress or final Suggested fix: when a GET resumes a stream, supersede prior connections for that Bulk deletes can exceed Durable Object storage limits
A stream with more than 128 events, or a cleanup sweep covering many streams, Suggested fix: delete in chunks of at most 128 keys, and add tests with storage Sweep can delete a stream that receives a concurrent write
Suggested fix: before deleting a stream, re-read its metadata or use another Medium SeverityReplay and cleanup are not paginated
Suggested fix: use explicit limits and pagination loops for replay, per-stream Prefix scanning can collide if stream ids contain colonsEvent keys are stored as: Operations build prefixes with Default stream ids appear safe, but exported Suggested fix: encode Completed-POST resume test is mislabeled as standalone resume
Suggested fix: add a true standalone test: open standalone GET, receive a No test covers true mid-flight POST disconnect and live continuationThe POST replay test reads the final result, cancels the response, then resumes Suggested fix: add an end-to-end test with a deferred tool: start POST, read an Lower Severity
|
|
actually gonna drop the store sweep and just clean up on successful stream. only DOs with un finished streams will have events left and I think thats fine. this is way too much complexity since I am gonna rip it all out in a month anyway. |
Rip the alarm-driven sweep added in the previous push of this PR. Storage cost is now bounded by in-flight POST streams (cleared the moment their close frame is written) plus the standalone GET stream. - event-store: drop sweep(), drop the __mcp_stream_evt_meta__ index, drop the onStoreEvent option. Assert streamId contains no ':' to rule out prefix-scan collisions. Chunk multi-key deletes at the DO 128-key cap. Cap replayEventsAfter at 1000 events to bound memory. - mcp/index: drop _cf_armEventStoreCleanup, _cf_runEventStoreCleanup, and getEventStoreMaxAgeMs. getEventStore() is now plain 'new DurableObjectEventStore(this.ctx.storage)'. - transport: writeSSEEvent FIRST, then deleteStreamRequestIds + clearStream. Trade-off: if the WS pipe is enqueued but the client TCP dies before the bytes arrive, that one final message is lost. Every earlier event is still replayable while the stream is open. - transport: when a GET resumes an active POST stream, strip requestIds off any other connection still claiming them, so stale POST bridges can't win the routing race against the resumed GET. - transport: fan standalone notifications out to every _standaloneSse connection instead of last-writer-wins. Matches the spec. - tests: replace the completed-POST 'replay' test with a real mid-flight POST disconnect test using a new deferredGreet tool that emits a progress notification then sleeps. Drop the misnamed 'resumed GET is registered as standalone' test (it was actually a POST-id resume). Drop event-store sweep tests; add chunked-delete and replay-cap tests.
|
/bonk review this pls |
![ask-bonk[bot]](https://avatars.githubusercontent.com/in/2454580?s=60&v=4){kind=small}
There was a problem hiding this comment.
Solid PR overall — this is a meaningful correctness improvement to the resumability path. The decision to drop the alarm-based sweep in favor of immediate cleanup on shouldClose is the right call, and the changeset is well-written. A few things I'd flag before merging:
Key issues
-
clearStreamdoesn't paginate itslist()call — While deletes are correctly chunked at 128, thestorage.list({ prefix })call at the top ofclearStreamhas nolimitand loads every key for the stream into memory in one shot. For typical streams (a handful of progress events + a result) this is fine, but the class is now exported for embedders. Consider either paginating the list or at least documenting the memory ceiling. -
getStreamIdForRequestIdis a full scan — This scans every__mcp_stream_reqs__:*key on everysend()when the live connection has dropped. During a burst of tool progress messages on a flaky connection this could be hot. It's acceptable for now since the number of in-flight POST streams per DO is small, but worth a// TODOor a note in the docstring. -
clearStreamraces withreplayEventsAfter— AfterwriteSSEEventinsend(), the codeawaitsdeleteStreamRequestIdsand thenclearStream. Meanwhile, a concurrent GET reconnect could be mid-replayEventsAfteron the same stream. Thelist()inclearStreamwould then delete events thatreplayEventsAfteris still iterating. In practice the replaysendcallback writes to a WS synchronously, so the window is tiny, but it's worth a comment acknowledging the race. -
Standalone GET events are never cleaned up — POST stream events get
clearStream'd on close, butSTANDALONE_STREAM_IDevents accumulate for the lifetime of the DO. The doc says "the DO itself dies with the session" which is true, but for long-lived sessions with chatty server-initiated notifications this could grow. Fine for v1, but should be called out as a known limitation.
See inline comments for specifics.
|
Review submitted. Here's a summary of what I flagged: Top-level findings:
Inline comments on specific lines covering: the unpaginated list in |
Four structural cleanups, no behaviour change. All 453 mcp tests still pass. - Add ClearableEventStore interface (extends EventStore with clearStream). Replaces an inline cast + optional chaining dance with a real named contract and proper 'in' narrowing. - Supersede stale POST bridges by closing them (conn.close(1000, 'Superseded by resumed stream')) instead of reaching in and mutating their state. Matches the SDK's last-writer-wins _streamMapping mental model. Removes a 'spooky action at a distance' pattern. - Split send() into sendStandalone() and sendForRequest(). The two paths share nothing but the storeEvent call and were fused for no good reason. Removes the disjoint shouldClose-cleanup-after-write block as a side effect. - Drop the now-unused 'eventStore' getter on the transport. Its only caller was _cf_runEventStoreCleanup which was deleted in the previous commit.
|
/bonk look again pls. |
There was a problem hiding this comment.
Second pass review. The shift from alarm-based sweep to immediate cleanup on shouldClose is a good simplification — much less surface area, no billing for quiescent DOs, and the trade-off (final message lost if TCP dies after WS enqueue) is clearly documented. The sendStandalone/sendForRequest split and the stale-connection supersession fix are both solid.
A few remaining items, mostly around edge cases in clearStream and the reverse-lookup fallback path. Nothing blocking, but worth addressing before merge.
| async clearStream(streamId: StreamId): Promise<void> { | ||
| const prefix = `${DurableObjectEventStore.KEY_PREFIX}${streamId}:`; | ||
| const prefix = `${DurableObjectEventStore.EVENT_KEY_PREFIX}${streamId}:`; | ||
| const rows = await this.storage.list({ prefix }); |
There was a problem hiding this comment.
Previous review flagged this and it still applies: storage.list({ prefix }) with no limit loads every key for the stream into memory before chunking the deletes. The delete side is correctly chunked at 128, but the list isn't bounded. For typical streams this is fine, but since DurableObjectEventStore is now exported for embedders, a pathological stream (or a standalone stream that accumulates notifications) could pull a lot of keys.
You could fold the list into the same chunked loop:
| const rows = await this.storage.list({ prefix }); | |
| const prefix = `${DurableObjectEventStore.EVENT_KEY_PREFIX}${streamId}:`; | |
| let cursor: string | undefined; | |
| for (;;) { | |
| const rows = await this.storage.list({ | |
| prefix, | |
| ...(cursor ? { start: cursor } : {}), | |
| limit: DurableObjectEventStore.DELETE_CHUNK | |
| }); | |
| if (rows.size === 0) break; | |
| const keys = [...rows.keys()]; | |
| await this.storage.delete(keys); | |
| if (rows.size < DurableObjectEventStore.DELETE_CHUNK) break; | |
| cursor = keys[keys.length - 1] + "\x00"; | |
| } |
This eliminates deleteChunked entirely and caps memory at 128 keys.
| const streamId = | ||
| liveConnection?.state?.streamId ?? | ||
| (await agent.getStreamIdForRequestId(requestId)); | ||
| if (!streamId) { |
There was a problem hiding this comment.
Previous review noted the error message is misleading — when you land here it's because the stream-reqs mapping is gone (deleted by a concurrent shouldClose or never existed), not because there's no connection. Consider:
| if (!streamId) { | |
| if (!streamId) { | |
| throw new Error( | |
| `No active stream found for request ID: ${String(requestId)}` | |
| ); | |
| } |
| this._requestResponseMap.set(requestId, message); | ||
| const relatedIds = connection.state?.requestIds ?? []; | ||
| // Check if we have responses for all requests using this connection | ||
| const relatedIds = |
There was a problem hiding this comment.
When liveConnection is null (client dropped mid-tool-call), this falls through to agent.getStreamRequestIds(streamId) — which hits storage for the same __mcp_stream_reqs__:* data that getStreamIdForRequestId just scanned on line 483. Two storage reads for the disconnected-client path on every send(). Not a blocker since the set of in-flight POST streams is small, but worth a note.
One way to collapse it: have getStreamIdForRequestId return both the streamId and the requestIds in one pass, or cache the scan result for the duration of this sendForRequest call.
| if (shouldClose) { | ||
| await agent.deleteStreamRequestIds(streamId); | ||
| if (this._eventStore && "clearStream" in this._eventStore) { | ||
| await this._eventStore.clearStream(streamId); | ||
| } |
There was a problem hiding this comment.
After writeSSEEvent on line 508, the await deleteStreamRequestIds + await clearStream sequence is correct, but there's a subtle ordering window: a concurrent GET resume arriving between lines 512 and 514 would find no persisted requestIds (deleted), treat the stream as a completed-POST one-shot replay, and start iterating events that clearStream is about to delete. The replay send callback writes synchronously to the WS, so in practice the race is benign — but a brief comment here would save future readers from having to reason through it.
Also: "clearStream" in this._eventStore is a duck-type check. Since you've defined the ClearableEventStore interface, you could use a type guard or just check at construction time and store a narrowed reference. Minor style nit.
| async getStreamIdForRequestId( | ||
| requestId: RequestId | ||
| ): Promise<string | undefined> { | ||
| const rows = await this.ctx.storage.list<RequestId[]>({ |
There was a problem hiding this comment.
This is O(n) over all __mcp_stream_reqs__:* keys, called on every send() when the live connection has dropped. For the normal case (few in-flight POST streams per DO) this is fine, but it's worth a // O(n) in active POST streams comment so future readers know the cost.
Also, the list() call has no limit — a defensive cap (say 1000) would prevent a theoretical unbounded memory load if stream-reqs mappings accumulate from abandoned POSTs where deleteStreamRequestIds was never called.
| params: { name: "greet", arguments: { name: "resume" } } | ||
| params: { | ||
| name: "deferredGreet", | ||
| arguments: { name: "midflight", delayMs: 600 } |
There was a problem hiding this comment.
This is the key test for the PR and it's well-structured. One thing to consider: the delayMs: 600 makes this timing-dependent. If the test runner is slow and the tool completes before reader.cancel() on line 266, the POST stream will have already delivered the result and shouldClose will have fired (deleting the stream events), so the resumed GET will have nothing to replay.
The bounded loop on line 252 helps, but you might want an assertion that the POST was actually cancelled before the result arrived — e.g., assert that postBuf does not contain "result" at the point of cancel. That way a timing regression turns into a clear test failure rather than a false pass.
| return; | ||
| for (const conn of agent.getConnections<TransportConnState>()) { | ||
| if (!conn.state?._standaloneSse) continue; | ||
| this.writeSSEEvent(conn, message, eventId); |
There was a problem hiding this comment.
Good fix — fanning out to all _standaloneSse connections rather than last-writer-wins matches the spec's allowance for multiple concurrent SSE streams. Note that writeSSEEvent can throw (the WS might have been torn down), and this loop doesn't catch per-connection errors — a dead connection would abort delivery to all subsequent connections in the iteration. Consider wrapping each writeSSEEvent call in a try/catch so one dead WS doesn't block the rest:
| this.writeSSEEvent(conn, message, eventId); | |
| for (const conn of agent.getConnections<TransportConnState>()) { | |
| if (!conn.state?._standaloneSse) continue; | |
| try { | |
| this.writeSSEEvent(conn, message, eventId); | |
| } catch (error) { | |
| this.onerror?.(error as Error); | |
| } | |
| } |
| for (const other of agent.getConnections<TransportConnState>()) { | ||
| if (other.id === connection.id) continue; | ||
| if (other.state?.streamId !== resumedStreamId) continue; | ||
| other.close(1000, "Superseded by resumed stream"); |
There was a problem hiding this comment.
Good addition — this directly addresses the stale-connection routing race from the earlier review. close(1000, ...) is the right call since the old WS is dead anyway.
One thing: this fires for every connection whose streamId matches, which is correct for the POST-resume case. But if a client somehow opens two GET resumes for the same POST stream in quick succession, the second resume would close the first resumed GET. That's arguably correct (last-writer-wins on resume), but it's worth a comment noting the semantics.
|
Review submitted. Here's a summary of what I flagged on the second pass: Overall: The PR is in good shape. Dropping the alarm-based sweep for immediate cleanup on Inline comments (8 total):
|
Three legit findings, two no-ops: - clearStream: replace 'list-all-then-chunk-delete' with a paginated list+delete loop bounded by DELETE_CHUNK. Removes the deleteChunked helper as a side effect \u2014 the chunk size is now enforced by the list limit itself, not a separate pass. Net simpler. - Clearer error message in sendForRequest: 'No active stream found' instead of 'No connection established', since the failure is a missing stream mapping, not a missing connection. - resumability test: assert the progress notification is NOT re-delivered on the resumed GET. The Last-Event-ID *was* the progress event, so replayEventsAfter must skip it. Not acting on: - O(n) full scan in getStreamIdForRequestId: working set is in-flight POST streams per DO, which is small. - clearStream/replayEventsAfter race: window is between two awaits in the same microtask chain and the consequence (replaying about-to-be- deleted events) is harmless. - Double storage read in disconnected-client path: only on the slow path; threading data through would complicate the API. - Standalone GET events accumulate: already documented; bounded by session lifetime by design.
Five small wins from the second-pass review: - Replace the 'clearStream' in this._eventStore duck-type check with a proper isClearableEventStore type guard. Same runtime check, better readability, narrows the type through the inferred call. - Add an ordering note on the deleteStreamRequestIds -> clearStream sequence: the tiny window where a concurrent GET resume sees no requestIds and starts replaying soon-to-be-deleted events is benign, but the comment saves future readers from re-deriving it. - getStreamIdForRequestId: add an O(n) cost comment and a defensive limit: 1000 on the list call so an abandoned-POST leak can't unbounded-load the scan. - Standalone fan-out: wrap each writeSSEEvent in try/catch so one dead WS in the middle of the loop can't block delivery to the remaining standalone connections. - Mid-flight resumability test: assert postBuf does NOT contain '"result"' at the point of cancel, so a timing regression turns into a clear failure rather than a false pass. - Comment on the supersede-by-close loop noting the last-writer-wins semantic for rapid back-to-back GET resumes on the same stream. Skipped: - Collapsing the double storage read on the disconnected path (getStreamIdForRequestId + getStreamRequestIds): would require widening the agent API to return both in one pass. Slow-path-only, working set is single-digit streams. Not worth the API change.
|
/bonk look again |
|
UnknownError: ProviderInitError |
|
@mattzcarey Bonk workflow failed. Check the logs for details. View workflow run · To retry, trigger Bonk again. |
|
/bonk look again |
There was a problem hiding this comment.
Third-pass review. The PR is in solid shape after incorporating feedback from the first two rounds. The shift from alarm-based sweep to immediate cleanup on shouldClose is the right call, the clearStream pagination is correct, the stale-connection supersession logic is clean, the standalone fan-out try/catch is in place, and the mid-flight test has the "result" guard. A few remaining items below, mostly around the double storage read on the disconnected path and a type export gap.
|
Review submitted. Here's a summary of what I flagged on the third pass: Overall: The PR is in good shape. Most feedback from the first two rounds has been addressed — Inline comments (7 total):
|
Six fixes:
- Collapse the double storage read on the disconnected-client send
path. getStreamIdForRequestId is now getStreamForRequestId and
returns { streamId, requestIds } in one pass; sendForRequest no
longer issues a second getStreamRequestIds for the same key.
- Re-export ClearableEventStore from mcp/index.ts so embedders who
want to implement their own clearable store can import the type.
- Fix the replayEventsAfter off-by-one: use start: <key>+'\x00' so
the list result strictly excludes lastEventId rather than including
it and post-filtering. Effective cap is now exactly 1000, not 999.
Drops the (key <= lastKey) guard as a side effect.
- Tighten the replay-cap test from <=1000 to ==1000 \u2014 catches the
off-by-one if anyone reverts.
- Comment the no-live-connection branch in sendForRequest: the close
frame is intentionally dropped when there's nowhere to write it.
The client's reconnect with Last-Event-ID gets the final event
(until cleanup runs immediately after) and the spec lets them
treat it as final.
- Log a warning if getStreamForRequestId hits its 1000-key scan cap.
Hitting it means abandoned __mcp_stream_reqs__ entries are
accumulating, which would silently turn into 'No active stream
found' errors otherwise.
The merge of main into this branch left two fused copies of
sendForRequest \u2014 main's collision-routing version and this PR's
no-live-connection fallback version \u2014 plus dangling references to
the old _requestResponseMap (renamed to _streamResponseIds in main).
The file built but the logic was incoherent.
This commit restructures sendForRequest so both intents coexist
cleanly:
1. Pick the live connection using main's collision-safe rule:
prefer the originating connection; fall back to the unique
matching connection; otherwise null.
2. If multiple live connections claim the request id and none is
the originating, route an Internal Error to each \u2014 main's
cloudflare#1639 protocol-safety behaviour.
3. If a single live connection owns the request, delegate to
sendOnStream (main's canonical store + close-detect + cleanup
path).
4. Otherwise fall back to this PR's getStreamForRequestId reverse
lookup so the event is still stored for replay when the
originating WS has dropped, and run cleanup on shouldClose.
The close frame is intentionally dropped when there's no live
connection \u2014 documented trade-off in send()'s docstring.
No behaviour change for the live-connection path (that's main's
sendOnStream verbatim). The dropped-WS replay path now exists again,
which is what this PR was originally about.
The previous commit's repair of the botched merge left sendOnStream
and sendForRequest's no-live-connection fallback as two parallel
implementations of the same logical function. They had already drifted
in this branch (live path used a cast, fallback used the
isClearableEventStore type guard), and a 'void eventId;' silencing
betrayed the fact that the structure wasn't right.
sendOnStream now takes (streamId, relatedIds, liveConnection,
message, requestId) directly. The caller resolves where the message
goes; sendOnStream handles store + close-detect + cleanup + write.
writeSSEEvent is gated on liveConnection being non-null \u2014 same
behaviour, expressed as one branch instead of two.
sendForRequest is now four phases readable top-to-bottom:
1. find matching live connections
2. ambiguous-multi \u2192 error route to each
3. resolve streamId + relatedIds (from connection state, or
persisted fallback if the WS dropped)
4. one sendOnStream call
No behaviour change. 457/457 mcp tests still pass. Net -10 lines,
two cleanup paths collapsed to one, type-guard drift fixed, 'void
eventId;' smell removed.
|
/bonk do a last review. does it all look good? |
There was a problem hiding this comment.
Final review
This is a well-structured PR overall — the shift from alarm-based sweep to immediate cleanup, the sendStandalone/sendForRequest split, the stale-connection supersession logic, the standalone fan-out try/catch, the streamId colon guard, and the clearStream chunking are all good improvements. The test coverage for the mid-flight disconnect scenario is exactly the right test to have.
However, there is one critical ordering bug that undermines the main fix claimed by the PR, plus a changeset inaccuracy. Everything else looks good to ship.
Critical: sendOnStream cleanup runs before wire write — contradicts the stated fix
The PR description, the changeset, and the sendOnStream docstring all claim that writeSSEEvent runs last — after cleanup. The changeset explicitly says:
Fixed by flipping the order: write the SSE event to the wire first, then drop the persisted
streamId -> requestIdsmapping and clear the stored events.
The docstring on sendOnStream says:
Ordering note: writeSSEEvent runs LAST.
But the code does the opposite: deleteStreamRequestIds and clearStream both run before writeSSEEvent. This means the original bug described in the changeset (events deleted from store before the wire write completes) still exists when shouldClose is true. A client whose TCP dies between the clearStream await and the writeSSEEvent call cannot resume with Last-Event-ID because the events are already gone.
To match the stated intent, move the cleanup block after the wire write.
Minor: changeset references wrong method name
The changeset at line 26 says McpAgent.getStreamIdForRequestId but the actual method is McpAgent.getStreamForRequestId.
In the collapse-the-duplicate refactor (6b5990c), the cleanup sequence in sendOnStream ended up running BEFORE writeSSEEvent \u2014 the exact ordering the docstring and changeset warn against, and the exact bug Sunil's clanker review and the original cloudflare#1583 ticket are about. Failure mode: tool finishes, sendOnStream is called with a still-live liveConnection (e.g. resumed GET), storeEvent writes the final event, shouldClose=true triggers clearStream which wipes the entire stream INCLUDING the event we just stored, then writeSSEEvent emits a frame with an event id that no longer resolves. A client losing the WS pipe at that exact moment can reconnect with Last-Event-ID and find nothing to replay. Fix: write the SSE frame first, then run cleanup. Matches the docstring already in place. Also fixes bonk cloudflare#25: changeset referenced McpAgent.getStreamIdForRequestId; renamed to getStreamForRequestId in 880c013.
Connection state arrays are ImmutableArray<RequestId>; sendOnStream only reads via .every() so the parameter type can accept readonly RequestId[]. Caught by CI typecheck.
|
/bonk go fo it. rip me to shreds |
There was a problem hiding this comment.
Review
This is a well-structured PR. The core correctness fixes (unconditional event storage, write-before-cleanup ordering, stale-connection supersession, standalone fan-out) are sound and the test coverage for the mid-flight disconnect scenario is the right test to have. The simplification from alarm-based sweep to immediate cleanup on shouldClose is a clear win — less code, no idle DO billing, and the trade-off is well-documented.
Addressing the last bonk review's "critical" finding
The final bonk review flagged a critical ordering bug claiming deleteStreamRequestIds + clearStream run before writeSSEEvent. That's incorrect. Reading sendOnStream at lines 439-487:
- Line 471:
writeSSEEvent(wire write) runs first - Lines 474-486:
deleteStreamRequestIds+clearStream(cleanup) runs second
The code matches the docstring and changeset. The bonk review misidentified which lines correspond to which operations. No bug here.
Similarly, the bonk review said the changeset references getStreamIdForRequestId when the actual method is getStreamForRequestId. The changeset at line 26 already correctly says McpAgent.getStreamForRequestId. Also not a real issue.
Remaining items worth considering
See inline comments. Nothing blocking — two minor correctness hardening suggestions and one edge case note.
|
Review posted. Here's the summary: The PR is in good shape. The core fixes are correct and well-tested. I verified both "critical" findings from the last bonk review and they were both wrong:
I left 7 inline comments covering:
|
- Wrap supersede-by-close in try/catch so a dead WS can't abort the loop (bonk). - Trim verbose comments in sendOnStream + send dispatcher. No PR / ticket references in code.
|
/bonk review this |
There was a problem hiding this comment.
Review
This is a well-considered rewrite of the resumability path. The core fixes are correct and the simplification from alarm-based sweep to immediate cleanup on shouldClose is a clear improvement — less code surface, no idle DO billing, no background state machine to reason about. The changeset is thorough, the tests cover the primary failure mode, and the prior bonk reviews have been largely addressed.
I have a few findings below — one real correctness concern around standalone event accumulation, one gap in the sendForRequest null-connection path that's worth hardening, and a handful of smaller items.
Standalone GET events accumulate without bound
STANDALONE_STREAM_ID events are stored via sendStandalone but never cleaned up — clearStream is only called when shouldClose fires on a POST stream. For long-lived DO sessions with chatty server-initiated notifications, this storage grows without bound for the lifetime of the DO. The doc says "the DO itself dies with the session" which is true for short sessions, but for agents that stay warm for hours/days this could quietly accumulate. Worth either documenting as a known limitation or adding a lightweight cap (e.g. evict oldest standalone events when count exceeds some threshold).
No-live-connection + shouldClose clears events immediately after storing them
When liveConnection is null and shouldClose is true (tool finishes after the client dropped), sendOnStream stores the final event, skips the wire write, then immediately runs deleteStreamRequestIds + clearStream. The stored event exists for the duration of those two awaits. A GET resume arriving in that window would see it; otherwise it's gone. The comment at line 462 acknowledges the race as benign, and that's fair for now — but it means the PR description's claim that "every event up to and including the final response is replayable while the in-flight stream is open" isn't strictly true in this path. The changeset should probably note this edge case explicitly.
Test coverage is good but could be tighter on the standalone resume path
The mid-flight POST test is the right test. The existing "reconnecting GET with a valid Last-Event-ID returns 200" test resumes from a completed POST event id (stream already cleaned up), which is the one-shot replay channel path. There's no test that exercises a true standalone GET disconnect and resume: open standalone GET, receive a server-initiated notification, disconnect, reconnect with Last-Event-ID, trigger another notification, and verify it arrives live. This would exercise the STANDALONE_STREAM_ID branch in handleGetRequest.
| // leave a mid-flight client with a wiped stream on reconnect. | ||
| // `writeSSEEvent` is sync (enqueues, doesn't await), so the bytes | ||
| // are committed before any cleanup await can interleave. | ||
| if (liveConnection) { |
There was a problem hiding this comment.
The comment correctly notes that writeSSEEvent is sync, which is the key property that makes the write-before-cleanup ordering guarantee real. However connection.send() (which writeSSEEvent delegates to) can throw if the WS is dead — and that throw would skip the cleanup block entirely, leaving the stream-reqs mapping and stored events permanently orphaned.
Consider wrapping the write in try/catch so cleanup always runs:
| if (liveConnection) { | |
| if (liveConnection) { | |
| try { | |
| this.writeSSEEvent(liveConnection, message, eventId, shouldClose); | |
| } catch (error) { | |
| this.onerror?.(error as Error); | |
| } | |
| } |
This way a dead WS doesn't prevent clearStream from running.
| await Promise.all( | ||
| matchingConnections.map((candidate) => | ||
| this.sendOnStream(agent, candidate, routingError, requestId) | ||
| ) ?? (matchingConnections.length === 1 ? matchingConnections[0] : null); |
There was a problem hiding this comment.
When matchingConnections is empty and liveConnection is null, the code falls through to getStreamForRequestId. If that also returns undefined (e.g. the stream-reqs were already deleted by a concurrent shouldClose), we throw No active stream found. This is correct.
But there's a subtler case: the getStreamForRequestId lookup succeeds but by the time we reach sendOnStream, a concurrent shouldClose on another call has already run deleteStreamRequestIds + clearStream. The storeEvent in sendOnStream would then write an event to an already-cleared stream that nobody will ever read. Not a bug per se (it's a race between two legitimate paths), but worth noting that the orphaned event persists in storage until the DO dies.
| standaloneConnection.state?.streamId ?? standaloneConnection.id; | ||
| eventId = await this._eventStore.storeEvent(streamId, message); | ||
| } | ||
| const eventId = await this._eventStore?.storeEvent( |
There was a problem hiding this comment.
Standalone events under STANDALONE_STREAM_ID are stored here but never cleared — clearStream only fires on POST streams via shouldClose. For long-lived DOs with chatty server notifications, this grows without bound.
If you don't want to add eviction logic now, at minimum document this as a known limitation in the class docstring or the changeset so future readers know it's intentional.
| for (const other of agent.getConnections<TransportConnState>()) { | ||
| if (other.id === connection.id) continue; | ||
| if (other.state?.streamId !== resumedStreamId) continue; | ||
| try { |
There was a problem hiding this comment.
Good fix — this directly addresses the stale-connection routing race. The per-connection try/catch is important.
One note: other.close(1000, ...) will trigger onClose on the partyserver side, which may have downstream effects depending on what the agent's onClose handler does. In the typical MCP case this is fine (no onClose override), but it's worth noting that closing stale connections is a side-effectful operation that could interact with custom agent lifecycle hooks.
| streamId: StreamId, | ||
| message: JSONRPCMessage | ||
| ): Promise<EventId> { | ||
| if (streamId.includes(":")) { |
There was a problem hiding this comment.
Good defensive check. This closes the prefix-collision attack surface that the earlier review flagged. The error message with JSON.stringify is helpful for debugging empty strings and special characters.
| // semantics. Appending `\x00` (the smallest byte) to `lastKey` | ||
| // produces a key that sorts immediately after it, so the list | ||
| // result excludes `lastKey` itself without needing a post-filter. | ||
| const startKey = `${DurableObjectEventStore.EVENT_KEY_PREFIX}${lastEventId}\x00`; |
There was a problem hiding this comment.
The \x00 approach is cleaner than the previous key <= lastKey filter — it pushes the exclusive boundary into the query itself rather than post-filtering. This also fixes the off-by-one that the third bonk review flagged: the REPLAY_LIMIT now means exactly 1000 replayable events, not 999. The test at event-store.test.ts:278 correctly asserts toBe(1000).
| prefix: McpAgent.STREAM_REQS_KEY_PREFIX, | ||
| limit: STREAM_REQS_SCAN_LIMIT | ||
| }); | ||
| if (rows.size === STREAM_REQS_SCAN_LIMIT) { |
There was a problem hiding this comment.
The console.warn when hitting the scan limit is a good idea — it turns a silent miss into an observable signal. Worth noting that this warning fires on every send() call in the affected DO, so if there is a leak, logs will be noisy. That's arguably the right trade-off for a situation that "shouldn't happen."
| // enough that the tool already completed, the resumed GET would | ||
| // have nothing to deliver live and the test would be a false pass. | ||
| // We want this to fail loudly instead. | ||
| expect( |
There was a problem hiding this comment.
This timing guard is exactly right — without it, a slow test runner could produce a false pass by completing the tool before the cancel. delayMs: 600 gives a comfortable margin.
One concern: in CI with constrained resources, 600ms might not always be enough if the event loop is blocked. If this test becomes flaky, consider bumping to 1000ms+ or using a coordination mechanism (e.g. the tool signals readiness via a second notification before sleeping).
|
Review posted. Here's a summary of what I flagged: Overall: The PR is in good shape. The core fixes (write-before-cleanup ordering, unconditional event storage, stale-connection supersession, standalone fan-out) are correct. The simplification from alarm-based sweep to immediate cleanup is a clear win. The changeset is thorough and the mid-flight POST test covers the primary failure mode well. Top-level findings (3):
Inline comments (8):
|
…cleanup If connection.send throws inside writeSSEEvent (WS torn down between iteration and write), the exception would skip the cleanup block below and leave stream-reqs + stored events permanently orphaned. Also document standalone events accumulating for the DO lifetime in the event-store class docstring.
Reverting the try/catch I added around the supersede-by-close loop. Closing a WS in the Workers runtime is fire-and-forget and doesn't throw — no other close() call in the codebase is wrapped (including the transport's own close() method). Added it reactively to a review question; the answer was 'no, it doesn't throw'. Kept the two writeSSEEvent wraps: those guard connection.send (which the codebase consistently wraps) — one protects the cleanup block from being skipped, the other keeps one dead WS from aborting standalone fan-out to the rest.
Two behavioural changes in this PR had no direct coverage: - sendStandalone fans out to ALL _standaloneSse connections (was last-writer-wins). Two new tests: fan-out reaches every standalone stream and skips POST bridges; one throwing send doesn't block the rest of the loop. - handleGetRequest closes a stale POST connection when a GET resumes its stream. New test asserts the stale bridge is close()d with 1000/'Superseded by resumed stream', the resuming conn isn't closed, and it claims the persisted requestIds. Mutation-checked: commenting out the supersede close() fails the new test. 460/460 mcp tests pass.
MCP 2025-06-18 'Multiple Connections': the server MUST send each JSON-RPC message on only one of the connected streams and MUST NOT broadcast the same message across multiple streams. The prior commit fanned standalone notifications out to every _standaloneSse connection — a direct violation. (The SDK reference enforces the stronger 'only one standalone GET per session' with a 409.) Fix: - handleGetRequest now supersedes prior connections for the resumed streamId in ALL resumable cases (POST resume, standalone resume, and fresh standalone GET) via a shared supersedePriorStreamConnections helper. At most one live connection per stream. - sendStandalone reverts to single-send: find the one standalone connection and write to it, else store-only for replay. No more broadcast loop. - Tests updated: 'sends on exactly one standalone stream', 'stores but does not write when none live', and a new 'supersedes a prior standalone GET when a fresh GET opens'. 461/461 mcp tests pass. (5 unrelated typecheck failures in ai-chat/think predate this branch — from main's recovery work.)
MCP resumability rule: the SSE event id MUST be globally unique across all streams within a session. We had monotonic-within-stream and ignore-other-streams coverage but nothing pinning cross-stream uniqueness. A future move to a global seq counter could silently break it. New test stores the same seq across three streams and asserts no id collides. Mutation-checked: dropping streamId from the id format fails it.
- 'stripped requestIds' -> stale connections are closed (supersede). - Replace the fan-out bullet (reverted) with the one-stream-per-message spec rule. - Note standalone GET events accumulate for the DO lifetime.
The variable was renamed to startKey (built inline); the comment still said 'appending \x00 to lastKey'. No code change.
2 participants
Follow-up to #1602. Fixes the correctness bugs on the streamable-HTTP resumability path and replaces the cleanup strategy with immediate clear-on-close — no background sweep, no alarms, quiescent when idle.
Fixes
Final tool response is now replayable
The previous code stored the final response in the event store and then immediately called
clearStream(streamId)onshouldClose, deleting every event for that stream — including the one just written. A client that lost the connection mid-flight could reconnect withLast-Event-IDand find nothing to replay.Fixed by ordering: write the SSE frame to the wire first, then clear. Every event up to and including the final response stays replayable while the in-flight stream is open. Trade-off: if the WS message is enqueued but the client TCP dies before the bytes arrive, that one final message is lost — accepted in exchange for not running any background cleanup.
Request-bound
send()writes are unconditionalThe standalone branch already stored events regardless of whether a live connection was attached. The request-bound branch only stored when a live connection existed. Long-running tool calls on flaky networks routinely lose the POST connection mid-stream, and those messages were silently disappearing.
McpAgent.getStreamForRequestId()does a reverse lookup against__mcp_stream_reqs__:*, returning both thestreamIdand the stream'srequestIdsin one pass, so the transport can resolve a stream even after the originating WS has died. The event still gets stored; only the wire write is skipped when there's no live connection.Resumed GET registers under the source streamId, and supersedes stale connections
When a client reconnects with
Last-Event-ID:requestIdsare restored so future tool messages route to the new WS.STANDALONE_STREAM_ID.In every resumable case (and when a fresh standalone GET opens) any prior connection bound to the same streamId is closed, so there's at most one live connection per stream. This kills the routing race where
send()could pick a stale POST bridge over the resumed GET, and keeps us within the MCP rule of one stream per message.Spec: one stream per message
MCP 2025-06-18 "Multiple Connections": the server MUST send each JSON-RPC message on only one connected stream and MUST NOT broadcast across streams. Server-initiated notifications go to the single standalone GET; POST responses go to their own stream. Combined with the supersede-on-resume behaviour above, there is never more than one standalone GET to choose from.
Cleanup architecture
Immediate, not background. Each POST stream's events are cleared the moment the close frame is written — no alarms, no metadata index, no sweep. Idle DOs do zero periodic work.
Storage cost is bounded by the in-flight POST streams plus the standalone GET stream. Defensive bounds:
clearStreamlists + deletes in chunks of 128 (the DO multi-key delete cap), so a large stream never loads the whole event log into memory.replayEventsAftercaps each call at 1000 events (start: <lastEventId>\x00for strict-after, exclusive replay).getStreamForRequestIdcaps its scan at 1000 keys and warns if hit (signals leaked__mcp_stream_reqs__entries).storeEventrejectsstreamIds containing:so custom embedder ids can't cause prefix-scan collisions.Standalone GET events are not cleared automatically; they accumulate for the lifetime of the session's Durable Object (bounded by session length).
Exports
DurableObjectEventStoreand theClearableEventStoreinterface are exported fromagents/mcpso callers embeddingWorkerTransportinside an Agent or Durable Object can opt into resumability withnew DurableObjectEventStore(this.ctx.storage).Verified against the MCP spec + SDK reference
Last-Event-ID; never cross-stream. ✅_requestToStreamMapping). ✅streamId(matches SDKreplayEvents). ✅Testing
:rejection. Two behaviours mutation-checked (supersede close, id uniqueness).Compatibility
Patch-level, additive:
DurableObjectEventStore+ClearableEventStoreexported.McpAgent.getStreamForRequestId()is@internal.