WebSocketStream: receive backpressure is not implemented — a slow consumer buffers every received message in memory

[carlotestor]

Bug Description

WebSocketStream's readable side never applies backpressure. Every parsed message is enqueued into the readable stream unconditionally, whether or not the consumer is reading:

• The ReadableStream is set up without a pull algorithm (only start and cancel), so consumer demand is never observable:

  undici/lib/web/websocket/stream/websocketstream.js

  Lines 286 to 295 in cb4c2f1

  	// 5. Let pullAlgorithm be an action that pulls bytes from stream .
  	// 6. Let cancelAlgorithm be an action that cancels stream with reason , given reason .
  	// 7. Let readable be a new ReadableStream .
  	// 8. Set up readable with pullAlgorithm and cancelAlgorithm .
  	const readable = new ReadableStream({
  	start: (controller) => {
  	this.#readableStreamController = controller
  	},
  	cancel: (reason) => this.#cancel(reason)
  	})

• #onMessage enqueues unconditionally, and the spec's step 4 ("Apply backpressure to the WebSocket") is a comment with no code:

  undici/lib/web/websocket/stream/websocketstream.js

  Lines 350 to 352 in cb4c2f1

  	this.#readableStreamController.enqueue(chunk)
  	// 4. Apply backpressure to the WebSocket.

• Constructor step 7 ("Apply backpressure to the WebSocket") is a // TODO:

  undici/lib/web/websocket/stream/websocketstream.js

  Lines 119 to 120 in cb4c2f1

  	// 7. Apply backpressure to the WebSocket.
  	// TODO

controller.desiredSize is never consulted and the socket is never paused on behalf of the consumer. The existing onSocketData → parser.write() === false → socket.pause() / onParserDrain → socket.resume() pair only reflects the parser's internal buffer, which drains as fast as frames can be parsed — it is unrelated to consumer demand.

Consequence: with a consumer that is slower than the sender (or momentarily stalled), every received message piles up in the readable queue — unbounded memory growth — and no TCP backpressure ever reaches the server. Since backpressure is the headline feature of the WebSocketStream API, this defeats its main use case. Note that webSocket.maxPayloadSize / maxFragments offer no protection here: each individual message is small and valid.

Reproducible By

npm i undici ws && node repro.js — the server (child process) pushes 200k × 1 KiB messages; the client reads one message per second.

'use strict'
// undici WebSocketStream: receive backpressure is not implemented.
// Server (child process) pushes 200k x 1 KiB messages; client reads ONE message
// per second. Expected: client memory stays bounded and the server's sends stall.
// Actual: every message is buffered in the client's ReadableStream queue.
//
// npm i undici ws && node repro.js
const { fork } = require('node:child_process')

const MESSAGES = 200_000
const SIZE = 1024 // 1 KiB per message ~= 200 MiB total

if (process.argv[2] === 'server') {
  const { WebSocketServer } = require('ws')
  const wss = new WebSocketServer({ port: 0, perMessageDeflate: false })
  wss.on('listening', () => process.send(wss.address().port))
  wss.on('connection', (ws) => {
    const payload = Buffer.alloc(SIZE, 0x61)
    let sent = 0
    const pump = () => {
      while (sent < MESSAGES) {
        ws.send(payload)
        if (++sent % 100_000 === 0) console.log(`[server] queued ${sent} messages`)
        if (ws.bufferedAmount > 8 * 1024 * 1024) return void setTimeout(pump, 10)
      }
      const check = setInterval(() => {
        if (ws.bufferedAmount === 0) {
          clearInterval(check)
          console.log(`[server] all ${MESSAGES} messages (${MESSAGES * SIZE / 1048576 | 0} MiB) fully flushed to the socket, even though the client read almost nothing`)
        }
      }, 100)
    }
    pump()
  })
} else {
  const server = fork(__filename, ['server'])
  server.once('message', async (port) => {
    const { WebSocketStream } = require('undici')
    const rss = () => `${(process.memoryUsage.rss() / 1048576).toFixed(0)} MiB`
    console.log(`[client] node ${process.version}, undici ${require('undici/package.json').version}, rss before connect: ${rss()}`)

    const stream = new WebSocketStream(`ws://127.0.0.1:${port}`)
    const { readable } = await stream.opened
    const reader = readable.getReader()

    // Slow consumer: one read per second. The readable has the default HWM (1),
    // so a backpressuring implementation would stop reading from the socket
    // after ~1 unread message until read() is called.
    for (let t = 1; t <= 10; t++) {
      await new Promise((resolve) => setTimeout(resolve, 1000))
      await reader.read()
      console.log(`[client] t=${t}s read=${t} of ${MESSAGES} rss=${rss()}`)
    }
    server.kill()
    process.exit(0)
  })
}

Expected Behavior

The readable is created with the default high-water mark of 1, so once ~1 unread chunk is queued (desiredSize <= 0) the implementation should stop reading from the socket until read() pulls, propagating TCP backpressure to the server (this is how Chromium's implementation of the same API behaves). For the repro: client RSS stays roughly flat, and the server's bufferedAmount never fully drains because its sends stall against a full TCP window.

Actual Behavior

The server flushed all ~195 MiB into the socket within the first second while the client had read a single message. Client RSS grew from 75 MiB to 337 MiB (+262 MiB ≈ payload plus per-chunk queue overhead) and all 199,990 unread messages stayed buffered in the readable queue:

[client] node v22.22.1, undici 8.7.0, rss before connect: 75 MiB
(node:2680200) [UNDICI-WSS] Warning: WebSocketStream is experimental! Expect it to change at any time.
[server] queued 100000 messages
[server] queued 200000 messages
[server] all 200000 messages (195 MiB) fully flushed to the socket, even though the client read almost nothing
[client] t=1s read=1 of 200000 rss=337 MiB
[client] t=2s read=2 of 200000 rss=337 MiB
[client] t=3s read=3 of 200000 rss=337 MiB
[client] t=4s read=4 of 200000 rss=337 MiB
[client] t=5s read=5 of 200000 rss=337 MiB
[client] t=6s read=6 of 200000 rss=337 MiB
[client] t=7s read=7 of 200000 rss=337 MiB
[client] t=8s read=8 of 200000 rss=337 MiB
[client] t=9s read=9 of 200000 rss=337 MiB
[client] t=10s read=10 of 200000 rss=337 MiB

Environment

• Linux x64
• Node.js v22.22.1
• undici 8.7.0 (npm); code paths verified unchanged on main @ cb4c2f1
• ws 8.21.0 (repro server only)

Additional context

• The spec the implementation cites includes "Apply backpressure to the WebSocket" steps in both the constructor and the "receive a WebSocket message" algorithm, and the explainer names backpressure as the primary motivation for the API: https://github.com/ricea/websocketstream-explainer
• I searched existing issues/PRs (WebSocketStream backpressure, desiredSize, flow control, memory/leak variants) and found no report covering this. Nearest neighbors are the WebSocket umbrella WebSockets #1811 and the fragment-count DoS fixed by GHSA-vxpw-j846-p89q, which is a different unbounded-growth vector (attacker-controlled fragments rather than a slow consumer).
• A possible fix shape: resume the socket from a pull algorithm and, after the enqueue in #onMessage, socket.pause() when controller.desiredSize <= 0. Granularity is per socket chunk (one chunk can contain many frames), so the queue may overshoot the HWM by one chunk's worth of messages — consistent with other implementations.