Testing Backpressure Handling on a Slow Consumer of a Stream

A model can produce tokens faster than a phone on a train can read them. If your proxy does not propagate that back to the provider connection, the difference accumulates in your process’s heap, once per concurrent request, and the failure arrives as an out-of-memory kill under load rather than as a wrong answer.

What backpressure is, in one paragraph

Every buffered pipe needs a way for the slow end to tell the fast end to wait. In Node streams that signal is the return value of writable.write(): it returns false once the internal buffer has passed highWaterMark, and emits drain when it is ready again. In Web Streams the same idea is expressed as desiredSize on the controller — the source’s pull is only called while the queue has room, so a source that only produces inside pull is automatically well behaved.

Both signals are advisory. Nothing stops you writing anyway, and nothing warns you when you do. That is what makes this worth a test: the incorrect code is not merely correct-looking, it produces byte-identical output.

It is worth being clear about which direction the pressure travels in a chat proxy. The provider is the producer, your process is the middle, and the browser is the consumer. The chain only holds if every link passes the signal along: the browser’s TCP receive window slows your socket writes, which makes write() return false, which must stop you reading the upstream body, which eventually slows the provider connection. Break the chain anywhere and everything upstream of the break keeps running at full speed into a buffer that nobody is bounding.

The bug, in one line

// Wrong: the return value is the backpressure signal, and it is discarded.
for await (const chunk of upstream) res.write(chunk);

// Right: wait for drain when the sink says it is full.
import { once } from "node:events";
for await (const chunk of upstream) {
  if (!res.write(chunk)) await once(res, "drain");
}

// Better: let the stream machinery do it, including error and close propagation.
import { pipeline } from "node:stream/promises";
await pipeline(upstream, res);

The first form is what almost everybody writes, because it reads naturally and works perfectly whenever the consumer is at least as fast as the producer — which is every local test and every fast connection. The middle form is correct but easy to get subtly wrong when errors or aborts interleave; pipeline is the version to prefer, because it also destroys the source when the destination fails, which is exactly what you want when a client disconnects.

If your upstream is a Web ReadableStream (which is what fetch gives you) and your downstream is a Node response, bridge them with Readable.fromWeb rather than manually looping, for the same reason.

A slow sink you control

To test this you need a destination that is deliberately slow and instrumented. A Writable with a small highWaterMark whose _write defers its callback is exactly that, and unlike a real socket it is completely deterministic.

import { Readable, Writable } from "node:stream";
import { pipeline } from "node:stream/promises";
import { expect, test } from "vitest";

function instrumentedPipe(total: number, highWaterMark: number) {
  let produced = 0;
  let delivered = 0;
  let maxLag = 0;

  const source = Readable.from(
    (async function* () {
      for (let i = 0; i < total; i++) {
        produced++;
        yield "chunk " + i;
      }
    })(),
    { objectMode: true },
  );

  const sink = new Writable({
    objectMode: true,
    highWaterMark,
    write(_chunk, _enc, cb) {
      delivered++;
      maxLag = Math.max(maxLag, produced - delivered);
      setImmediate(cb);              // a consumer that is always one tick behind
    },
  });

  return { source, sink, stats: () => ({ produced, delivered, maxLag }) };
}

maxLag is the number that matters: the largest gap that ever opened between what the source had generated and what the sink had accepted. It is a direct measure of how much data was in flight, and it does not depend on wall-clock time, CPU speed or how loaded the CI machine is.

The assertion that does not flake

test("a slow consumer stops the producer rather than filling memory", async () => {
  const { source, sink, stats } = instrumentedPipe(5_000, 2);
  await pipeline(source, sink);
  const { produced, delivered, maxLag } = stats();

  expect(delivered).toBe(5_000);   // nothing was dropped
  expect(produced).toBe(5_000);    // and nothing was left unproduced
  expect(maxLag).toBeLessThanOrEqual(16);
});

The bound is the judgement call. It is not one: each stream in the chain has its own buffer, an async generator source has a small internal queue, and object mode counts objects rather than bytes, so the steady-state lag is a small multiple of the high-water marks involved rather than exactly highWaterMark. Pick a bound generously above what a correct implementation produces and far below the total — sixteen against five thousand is a factor of three hundred, so the test cannot pass by accident and cannot fail for being one buffer out.

Run the same test against the naive for await loop and maxLag approaches total, because the source runs to completion while the sink is still on chunk three. That is the whole signal, and it is a hard failure rather than a slow test.

For a byte stream rather than object mode, assert on sink.writableLength instead — sample it inside _write and assert its peak stayed within a small multiple of highWaterMark. Do not assert on process.memoryUsage().heapUsed: garbage collection timing makes it non-deterministic, and it is the single most common reason a backpressure test gets marked flaky and then deleted.

The same test shape works for the Web Streams side of the boundary, with one substitution. Instead of a slow Writable, build the source as a ReadableStream that only produces inside pull, and pipe it to a WritableStream whose write returns a promise that resolves late. Count invocations of pull rather than yields, and assert the same bounded lag. The reason to prefer that construction in production code is that it makes the bug structurally impossible: a source that produces only when asked cannot outrun its consumer, whereas a source that produces in a loop and enqueues has to check desiredSize by hand, and nothing reminds it to.

Things that silently disable it

• An unbounded queue in the middle. A Transform constructed with a large highWaterMark — or an array you push into and drain elsewhere — absorbs the entire response and reports no pressure. Any handwritten queue between the two ends needs its own bound and its own test.
• Compression with a forced flush. Calling res.flush() after every write, which people add to fix buffering, can defeat the flow control the compressor was providing. If you need both, test them together rather than separately.
• Reading the upstream eagerly. Code that consumes the provider stream into a variable and then writes it out has already lost: backpressure has to reach the upstream socket, not just the local pipe. The test above catches this only if the source in your production code is genuinely the upstream response body, so wire the real handler into an end-to-end variant of this test as well.
• Fire-and-forget writes on cancellation. When the client disappears, an unpropagated backpressure signal becomes an unpropagated close, and you keep pulling tokens you will pay for and throw away. That is the same defect seen from the other side; see testing cancellation of an in-flight streaming request.