fix(websocket): apply WebSocketStream receive backpressure#5562
fix(websocket): apply WebSocketStream receive backpressure#5562mcollina wants to merge 2 commits into
Conversation
Signed-off-by: Matteo Collina <hello@matteocollina.com>
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #5562 +/- ##
==========================================
+ Coverage 93.44% 93.48% +0.04%
==========================================
Files 110 110
Lines 37434 37541 +107
==========================================
+ Hits 34979 35094 +115
+ Misses 2455 2447 -8 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
|
||
| // 7. Apply backpressure to the WebSocket. | ||
| // TODO | ||
| // This is done when the socket becomes available. |
There was a problem hiding this comment.
This is wrong and could possibly lead to dropped messages.
- user does
new WebSocketStream(...) - await ws.opened
- sets up the ReadableStream idk, 10 seconds later
any messages between the connection opening and reading messages will get "dropped" (stream will pull them even if the user isn't)
| start: (controller) => { | ||
| this.#readableStreamController = controller | ||
| }, | ||
| pull: () => this.#resumeSocketIfNeeded(), |
There was a problem hiding this comment.
" // 5. Let pullAlgorithm be an action that pulls bytes from stream ."
#resumeSocketIfNeeded does not do that
There was a problem hiding this comment.
doesn't it make sense to handle backpressure in the WHATWG stream instead of the node socket? This implementation doesn't look correct to me
There was a problem hiding this comment.
What do you mean? When the WebSocket is pulled, the stream restarts.
There was a problem hiding this comment.
If I understand you correctly, you want to switch this from using on('data') to using .read() and .on('readable'), essentially making this a pull rather than a push?
I didn't do that because you usually prefer to avoid large refactorings. I'll get it done.
|
|
||
| // 4. Apply backpressure to the WebSocket. | ||
| if (this.#readableStreamController.desiredSize <= 0) { | ||
| this.#handler.socket.pause() |
There was a problem hiding this comment.
ditto wrt node socket vs whatwg stream.
Signed-off-by: Matteo Collina <hello@matteocollina.com>
4 participants
This relates to...
Fixes #5503
Rationale
WebSocketStreamcontinued consuming transport data after its readable queue reached its high-water mark. Slow consumers therefore accumulated an unbounded number of messages instead of propagating backpressure to the peer.Changes
Features
N/A
Bug Fixes
.read()from the'readable'event instead of flowing'data'events.WebSocketStreamonly while the readable controller has demand and the frame parser is ready.Breaking Changes and Deprecations
N/A
Status
Validation
npm run lintnpm run test:websocketnode test/web-platform-tests/wpt-runner.mjs run '/websockets/stream/tentative/backpressure-receive.any.html?wss' '/websockets/stream/tentative/backpressure-send.any.html?wss'