How to add backpressure to a WebSocket consumer#

The feed sends ten thousand messages a second and the database takes two thousand writes a second. Every message the consumer cannot write yet waits in memory, and after four minutes the process is killed for using all of it. The producer never noticed, because nothing on the wire told it to slow down.

What you get

You will end up with a consumer whose memory stays flat while a slow sink drains at its own pace. A test pushes a hundred thousand messages through it to prove that. This is for you if a WebSocket feed can outrun whatever you write each message to.

Short answer

Put a byte-counted queue between the socket and the sink. In ws, call pause() when the queue passes a high-water mark and resume() when the sink has drained it, so the producer’s send buffer fills and TCP slows the wire. WebSocketStream does the same through its ReadableStream. In a browser or through socket.io the socket cannot be paused, so keep a window of unacknowledged messages and let the producer stop when it is full.

You will need

Node 22 or later, and a feed that can outpace your consumer; the sample runs its own producer on 127.0.0.1. Verified 2026-09-25 against Node 22.22.2, ws 8.21.3, socket.io 4.8.4 and socket.io-client 4.8.4. WebSocketStream is compared from its documentation, since neither Node 22 nor the CI runner has it.

Approaches compared

Only the first two rows slow the wire. Pausing the socket stops reading it, the kernel’s receive buffer fills, and the sender’s send buffer fills after it, which is TCP doing what it was built for. Acknowledgements never reach the socket: the producer keeps sending until it has chosen to wait, so they work in every runtime and only where the producer agrees.

Count bytes, not messages, and pause the socket

The consumer keeps one queue and two numbers. Messages go in from the 'message' event and come out one at a time into the sink. The socket is paused when the queue holds more than highWater bytes and resumed once the sink has drained it below lowWater.

    ws.on('message', (data) => {
      if (settled) return
      const text = data.toString()
      if (text === '{"done":true}') {
        done = true
        if (!draining && queue.length === 0) finish()
        return
      }
      let id
      try {
        id = JSON.parse(text)?.id
      } catch (err) {
        return finish(new Error(`the feed sent a message that is not JSON: ${err.message}`))
      }
      stats.received++
      stats.lastId = id
      queue.push(data)
      stats.queuedBytes += data.length
      if (stats.queuedBytes > stats.peakQueuedBytes) stats.peakQueuedBytes = stats.queuedBytes
      if (backpressure && !ws.isPaused && stats.queuedBytes >= highWater) {
        stats.pauses++
        ws.pause()
      }
      drain()
    })

pause() calls socket.pause() on the TCP socket underneath, which stops reading from it. The ws documentation adds that some events can still be emitted after the call, until buffered data is consumed, and that is the detail the high-water mark has to be sized for. Node reads the socket in chunks of up to 64 KiB, and ws parses every frame in a chunk before the pause takes effect. So the queue overshoots the mark by up to one chunk. Counting bytes keeps that overshoot at 64 KiB whatever the message size. Counting messages does not: with 64 messages as the mark and messages of a dozen bytes, the same chunk delivers five thousand of them after the pause.¹

The drain loop is the other half, and it is short: shift, await sink.write(message), and resume the socket when the queue is small enough. await is what makes the sink’s pace the consumer’s pace.

Watch the queue stay flat

A hundred thousand messages of 256 bytes into a file sink that yields every hundred writes. The demo runs it with the pause and then runs it without. The peak is printed as a band, because the exact figure moves with the machine; the band is what stays put.

node demo.mjs

with pause() and resume()
  received 100000, written to the sink 100000, last id 99999
  peak bytes waiting in the consumer: under 256 KiB
  socket paused: yes
  producer waited on its send buffer: yes
without backpressure
  received 100000, written to the sink 100000, last id 99999
  peak bytes waiting in the consumer: over 8 MiB
  socket paused: never
  producer waited on its send buffer: yes

Both runs deliver every message, and that is the point: backpressure does not drop anything, it moves the waiting from the consumer’s heap to the producer’s send buffer. With the pause the queue peaks at a few tens of kilobytes over the 64 KiB mark. Without it the whole feed sits in the consumer, 25 MiB of it, and the sink catches up after the producer has finished.

The producer waited in both runs, and that line deserves a second look. TCP flow control reflects how fast the consumer reads the socket, and a consumer that reads everything into a queue reads fast. The kernel pushed back a little because parsing a hundred thousand messages kept the event loop busy. Only pause() makes the reading speed equal the sink’s speed, which is what the producer needs to feel.

Window acknowledgements where you cannot pause

The standard browser WebSocket has no way to stop reading: messages arrive as events, and the only number it exposes, bufferedAmount, counts bytes queued by your send() calls, not bytes waiting to be read.² A browser consumer, or any consumer whose producer is willing to cooperate, needs flow control in the protocol instead. A socket.io acknowledgement is a request and a reply, so a producer can keep a fixed number of messages unanswered and wait before sending more.

  for (let id = 0; id < rows; id++) {
    if (state.inFlight >= window) await slot()
    if (failure) break
    state.inFlight++
    if (state.inFlight > state.peakInFlight) state.peakInFlight = state.inFlight
    const p = socket.timeout(timeoutMs).emitWithAck('row', { id }).then((reply) => {
      if (reply.id !== id) throw new Error(`acknowledged ${reply.id}, expected ${id}`)
      state.acked++
    }).catch((err) => {
      // Caught here, not left for Promise.all: a rejection with no handler
      // while the loop waits on a slot is an unhandled rejection.
      failure ??= err
    }).finally(() => {
      state.inFlight--
      if (release) { const r = release; release = null; r() }
    })
    pending.push(p)
  }
  await Promise.all(pending)
  if (failure) throw failure
  return state

The consumer acknowledges a row only after the sink has taken it, so the window is a promise about the sink and not about the network. timeout() on each emit stops a dead consumer from stalling the producer forever. The catch inside the loop matters as much: a timeout fired while the loop waits for a slot has no handler yet, and Node exits on an unhandled rejection.

node sio-demo.mjs

rows sent 10000, acknowledged 10000, written by the sink 10000
peak rows in flight: 8 (window 8)
a consumer that never acknowledges: operation has timed out after 200 ms

WebSocketStream is the third row, and it is the one the platform should have shipped first. Its opened promise resolves to a ReadableStream, and reading from a stream slower than it fills is backpressure with no code of yours. Chrome has it, and Deno has it behind an unstable flag. MDN marks it experimental and non-standard, so it is a code path for those two runtimes and the queue earlier for everyone else.

Check it worked

The test that matters pushes fifty thousand messages into a sink that accepts nothing until released, and asserts where they waited.

  const sink = stalledSink()
  const consuming = consume(producer.url, { sink, backpressure: true, highWater: 64 * KiB, lowWater: 16 * KiB })
  await new Promise((r) => setTimeout(r, 300))
  assert.ok(producer.state.sent < 50_000, `the producer is blocked with ${producer.state.sent} sent`)
  sink.release()
  const stats = await consuming
  assert.equal(stats.written, 50_000)
  assert.ok(stats.peakQueuedBytes <= 64 * KiB + 64 * KiB + 300)

node --test backpressure.test.mjs

# tests 9
# suites 0
# pass 9
# fail 0
# cancelled 0
# skipped 0
# todo 0

With the sink stalled and the pause in place, the producer stops with most of the feed unsent. Without the pause the same stalled sink leaves every message in the consumer, which the second test asserts. The others pin order, the socket.io window, and its timeout, and three failures that must reject rather than crash or hang.

When it goes wrong

Memory still climbs with the pause in place. The high-water mark counts messages, and the messages are small. One socket read holds thousands of them, and every one is delivered before the pause lands. Count bytes, as consumer.mjs does, or accept an overrun of one read at the message size you have.

allowSynchronousEvents: false does not help. It defers each message to its own tick, which is the behavior the WHATWG standard asks for, and the frames already read are still delivered before reading stops. Measured here against the stalled sink, the peak was the same to the byte with the option and without it.³

The producer runs out of memory instead. It ignores its own bufferedAmount and keeps calling send(), so the bytes the consumer is not reading pile up on the producer’s side. That is where they belong, and it is still a pile: a producer should stop generating when its buffer passes a threshold, as producer.mjs does, or drop what it can afford to lose.

The process stays up thirty seconds after the sink fails. A paused socket cannot read the close frame, so ws.close() waits out its timeout. finish() resumes first, and terminates on an error.

The connection drops while paused. A ping that arrives while the socket is paused is not read, so it is not answered until resume(). A server that closes clients after a missed pong will close a consumer that pauses for longer than its heartbeat. Keep pauses shorter than the heartbeat by fixing the sink, or raise the heartbeat.

When not to do this

Do not add a queue and a pause to a consumer whose sink is faster than the feed. The queue never fills, the pause never fires, and you have written code that only a change in load will exercise. Measure the sink first.

Do not use acknowledgements as flow control on a feed that must run at wire speed. A window of eight and a round trip of forty milliseconds caps the consumer at two hundred messages a second, whatever the sink can do. Widen the window or pause the socket.

Do not read bufferedAmount on the consumer and expect it to mean anything about what is waiting to be read. It counts what you have sent and not transmitted.

Do not pause a socket that carries messages you must answer promptly, such as a heartbeat the server enforces, unless the pauses are shorter than its timeout. The wire slows because nothing on it is read, control frames included.

Related how-tos

• Consume Server-Sent Events with fetch and custom headers

• Limit concurrency in a message worker

• Make queue consumer writes safe under redelivery

Last verified

Verified 2026-09-25 against Node 22.22.2, ws 8.21.3, socket.io 4.8.4 and socket.io-client 4.8.4. Every output block is what the command preceding it printed, against a producer on 127.0.0.1 in the code directory. WebSocketStream did not run; its behavior is read from its documentation.