fix(websocket): apply WebSocketStream receive backpressure#5562
fix(websocket): apply WebSocketStream receive backpressure#5562mcollina wants to merge 1 commit into
Conversation
Signed-off-by: Matteo Collina <hello@matteocollina.com>
mcollina
requested review from
KhafraDev and
tsctx
and removed request for
tsctx
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #5562 +/- ##
==========================================
+ Coverage 93.44% 93.45% +0.01%
==========================================
Files 110 110
Lines 37434 37459 +25
==========================================
+ Hits 34979 35007 +28
+ Misses 2455 2452 -3 ☔ 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
|
|
||
| // 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.
4 participants
This relates to...
Fixes #5503
Rationale
WebSocketStreamcontinued reading from the transport 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
desiredSizereaches zero.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'