fix(agent): release replay-buffer backlog on terminal consumer/cancel paths - #116
Merged
Merged
Conversation
… paths Three memory/lifecycle gaps in ReusableReadableStream and ToolEventBroadcaster, found while profiling fusion OOM (exceededMemory) in openrouter-web: 1. Zero-consumer backlog wipe (active-consumers mode): trimConsumed early-returned when consumers.size === 0, so when the LAST consumer departed, the entire remaining backlog stayed pinned until the stream object itself was GC'd — O(total streamed events) per run, stacking across concurrent runs. Now dropped wholesale once at least one consumer has existed and none remain; guarded on nextConsumerId so a stream nobody has consumed yet keeps its catch-up history. 2. Pending next() hang: return()/throw() deleted the consumer without settling a registered waitingPromise — a concurrently pending next() hung until the next pump tick (or forever on push-driven broadcasters). return() now resolves the waiter; throw() rejects with a normalized error. 3. RRS cancel() now drops the buffered backlog (swept once after source reader cancellation, since the pump can land one in-flight chunk between the sync section and cancellation). 'full' replay mode is untouched: late consumers still replay everything. Tests pin the wipe, slowest-consumer retention, waiter wake, and cancel sweep for both classes.
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Why
Profiling fusion
exceededMemoryOOMs in openrouter-web (real workerd CDP heap snapshots + floor-vs-request-count probes, see OpenRouterTeam/openrouter-web#33901 / #34035 for the method) surfaced three retention/lifecycle gaps in the replay-buffer classes. In production the dominant one retains O(total streamed events) per run, stacking across concurrent fusion runs — the mid-flight peak mechanism behind the gateway isolate OOMs.What
All changes are symmetric across
ReusableReadableStreamandToolEventBroadcaster, and all are scoped toactive-consumersmode (defaultfullreplay is untouched):1. Zero-consumer backlog wipe
trimConsumed()early-returned whenconsumers.size === 0— precisely the moment the last consumer departs. Everything buffered past the watermark stayed pinned until the stream object itself became garbage. Now the backlog is dropped wholesale once at least one consumer has existed and none remain (nextConsumerId > 0guard: a stream nobody has consumed yet keeps its catch-up history). A late-attaching consumer starts at the new watermark — documented behavior of active-consumers mode.Note the interplay with drain-to-done: natural completion unregisters without a trailing trim, so a consumer that attaches before others finish still gets full history; only after the final departure does the next trim trigger wipe.
2. Pending next() hang on return()/throw()
Both iterators deleted the consumer without settling its registered waitingPromise. A concurrently pending next() then hung until the next pump tick — forever on push-driven broadcasters where no push may ever come. return() resolves the waiter (the re-run next() observes removal and reports done); throw() rejects it with a normalized error.
3. ReusableReadableStream.cancel() clears the buffer
Cancel removed consumers but kept every buffered chunk alive. It now drops the backlog, and sweeps a second time after awaiting source-reader cancellation because the pump can land one in-flight chunk between the synchronous section and cancellation.
Tests
New tests/unit/reusable-stream.test.ts + a lifecycle-compaction describe in the broadcaster suite pin: slowest-consumer retention, zero-consumer wipe + watermark start for late joiners, waiter wake on return/throw, cancel sweep under both microtask interleavings, and full-mode replay regression guards.
Full agent unit project: 102 files / 1217 tests green; biome + tsc clean.
Out of scope (flagged)
Even with this change, a producer that keeps pushing after all consumers departed re-accumulates from the new watermark until terminal event. The durable fix is producer-side cancellation on consumer disconnect (openrouter-web carries that as PR #27591's pattern); worth considering an onConsumerExhausted hook here later.