Repository navigation
fix(peer): release client request state once a streamed response finishes - #157
Conversation
…shes ClientPeer kept a streamed response's request entry, its abort listener, and every buffered stream message until the caller consumed or cancelled the body. A caller that only read the status leaked all of it, even after the server had finished the stream and released its own state. Release the request as soon as the terminal stream message arrives (an event-stream `close`/`error` event, or an octet-stream `close` chunk). The queue is closed, not aborted, so a later resolveBody() still yields every buffered item, and cancelling that body sends no cancel for a stream the server already finished. A stream that never finishes still needs the caller to cancel the body or abort the signal; document that in the README and on ClientPeer.request. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AEr5Gq656PLdpCLEAw2zht
- drop the late-abort and exact-listener checks from the new unit tests; the existing abort-listener tests already cover them - cancel a single endless stream in the integration test instead of ten - shorten the ClientPeer.request JSDoc so it no longer repeats the README, and stop claiming the server always stops streaming on cancel (a hibernated stream is no longer tracked by ServerPeer) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AEr5Gq656PLdpCLEAw2zht
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AEr5Gq656PLdpCLEAw2zht
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AEr5Gq656PLdpCLEAw2zht
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AEr5Gq656PLdpCLEAw2zht
@standard-server/aws-lambda
@standard-server/core
@standard-server/fastify
@standard-server/fetch
@standard-server/node
@standard-server/peer
@standard-server/shared
commit: |
Codecov Report✅ All modified and coverable lines are covered by tests. 📢 Thoughts on this report? Let us know! |
Merging this PR will not alter performance
|
There was a problem hiding this comment.
✅ No new issues found.
Reviewed changes
This run reviewed the full PR (all 5 commits) against main.
- Eager release on terminal stream message (
packages/peer/src/client.ts) —ClientPeer.message()now callscloseById(id)when anevent-streammessage is'close'/'error'or anoctet-streammessage hasclose: true, instead of waiting for the consumer to resolve/consume the body. - Unit coverage (
packages/peer/src/client.test.ts) — release without resolving the body for event-stream close and error, nocancelsent whenreturn()/reader.cancel()runs on an already-finished stream, and abort-listener removal asserted via a spy. - Integration coverage (
packages/peer/tests/peer.test.ts) — 100 concurrent streamed responses left unresolved drain both peers'requeststo 0 with no residual abort listeners, plus an endless stream cancelled without reading finalizes the server generator.
The ordering is sound: the terminal message is pushed before closeById, so Queue.close keeps it buffered and pullable (only pending pulls are rejected), and closeById is idempotent against the later AsyncIteratorClass/toOctetStream cleanup. Peer suite (273 tests) passes locally.
deepseek-v4.1-flash (free via Pullfrog for OSS) | 𝕏

Summary
This PR fixes a resource leak where streaming response bodies (event-stream and octet-stream) were not properly released when the stream completed, even if the body was never resolved by the client. This caused request state to persist indefinitely and event listeners to accumulate.
Key Changes
Client peer streaming message handling: Modified
ClientPeerto automatically close requests when receiving final stream messages:closeflag is set to trueTest coverage: Added comprehensive tests to verify:
Implementation Details
The fix ensures that the server's stream completion message (which indicates the server has already released its resources) triggers an immediate cleanup on the client side via
closeById(id). This prevents the accumulation of request state and event listeners that would otherwise only be cleaned up when the body is eventually resolved or garbage collected.https://claude.ai/code/session_01AEr5Gq656PLdpCLEAw2zht