Chapter 9 — Streams and Async Iterators

Pulling data in pieces: iterators, backpressure, pipeline, Web Streams, and LLM output over SSE

Prerequisites: Chapter 4 — async/await Traps, Chapter 6 — Cancellation and Cleanup, Chapter 7 — Controlling Concurrency, Chapter 8 — The Node Process Under Failure (memory and the blocked loop). Practice: the gym, exercise 08-stream-parser.

How to read this chapter

A stream is data that arrives in pieces over time: a file read in 64 KB chunks, an HTTP body, an LLM's tokens. Code that handles streams has three jobs that a plain await res.json() never has: reassemble pieces that were split at arbitrary points, slow the producer down when the consumer is slower (backpressure), and tell the producer to stop when nobody wants the rest (cancellation). Every bug in this chapter is one of those three jobs not done.

Part 1 is the protocol under all of it (iterators and for await). Parts 2 and 3 are Node streams and Web Streams. Part 4 is the LLM case, with the parser you build. Part 5 is composing pipelines, with the second build.

Part 1: Iterators, async generators, and what break does

for (const x of obj) calls obj[Symbol.iterator]() and then next() until done. for await (const x of obj) does the same with Symbol.asyncIterator and awaits each next(). A generator function writes the iterator for you, and the important property is that it is lazy: the body runs only as far as the next yield, and only when the consumer calls next(). That laziness is backpressure for free: a generator cannot get ahead of its consumer.

async function* rows(fetchPage) {
  let cursor = null;
  const conn = await pool.connect();
  try {
    do {
      const page = await fetchPage(conn, cursor);
      for (const row of page.rows) yield row;
      cursor = page.next;
    } while (cursor);
  } finally {
    conn.release();
  }
}

for await (const row of rows(fetchPage)) {
  if (row.id === 3) break;
}

break calls the iterator's return() method. For a generator, return() resumes it at the paused yield as if that yield were a return statement, so its finally blocks run and the connection is released. The same happens on return from inside the loop and on a throw out of the loop body. In a test, finally ran on break before the next line after the loop.

Two ways to lose that cleanup:

• Calling next() by hand and stopping. const g = rows(...); await g.next(); and then dropping g never runs finally. The generator stays paused forever, holding its connection, until it is garbage collected (and GC does not run finally). If you drive an iterator manually, call g.return() when you stop.
• Wrapping it in something that does not forward return(). A hand-written iterator object with only next() cannot be cleaned up by break.

Spot the bug: for await over an array of promises

const calls = providers.map((p) => p.complete(prompt)); // all start now
try {
  for await (const reply of calls) {
    render(reply);
  }
} catch (err) {
  showError(err);
}

Provider B fails quickly while A is still running. What happens?

Part 2: Node streams

Node has four kinds: Readable (a file, an HTTP request body on the server, a response body on the client), Writable (a file, res on the server), Duplex (a socket) and Transform (gzip, a parser). Every Readable is also an async iterable, so for await (const chunk of readable) is the simplest way to consume one, and it handles backpressure: the stream does not read more than its buffer while your loop body is busy.

Backpressure on the write side

writable.write(chunk) always accepts the chunk. Its return value is the only signal: false means "my buffer is above highWaterMark; stop until I emit 'drain'". Ignore it and the chunks pile up in memory.

import { once } from 'node:events';

for (const row of rows) {
  if (!res.write(toCsvLine(row))) await once(res, 'drain');
}
res.end();

Measured against a writable with a 16 KB highWaterMark that takes one 1 KB chunk per millisecond (a slow mobile client): writing 50,000 chunks while ignoring the return value left 51.2 MB buffered in the process, with write() returning false 49,985 times. The loop that waits for 'drain' never had more than 15 KB buffered. Multiply the first number by a few hundred concurrent CSV exports and the process is OOM-killed.

pipeline vs .pipe

src.pipe(dest) handles backpressure, but not errors or cleanup:

• An error on src is not forwarded to dest, and an 'error' event with no listener throws. createReadStream('missing.csv').pipe(gzip).pipe(res) crashes the process with ENOENT (tested: exit code 1).
• If dest fails or the client disconnects, src is unpiped but not destroyed. In a test, after the destination errored, the file stream was still open with its file descriptor held. Under load that is an EMFILE incident.

pipeline from node:stream/promises fixes both: it forwards errors, destroys every stream in the chain when any one fails or closes early, and returns a promise.

import { pipeline } from 'node:stream/promises';
import { createReadStream } from 'node:fs';
import { createGzip } from 'node:zlib';

app.get('/exports/:id', async (req, res) => {
  res.setHeader('content-type', 'application/gzip');
  try {
    await pipeline(createReadStream(exportPath(req.params.id)), createGzip(), res);
  } catch (err) {
    if (!res.headersSent) res.status(500).end();
    // client disconnects show up here as ERR_STREAM_PREMATURE_CLOSE; nothing to do
  }
});

In the same test that left .pipe's source open, pipeline rejected with the destination's error and the source had destroyed === true.

pipeline also accepts async generator functions as transforms, which is usually simpler than subclassing Transform (Part 5 uses this).

Memory: buffering the whole thing

Hashing a 300 MB export file:

The same applies to await res.text() or res.json() on a big upstream response, Buffer.concat(chunks) "so we can parse it at the end", stream.toArray(), and JSON.parse of a whole NDJSON export (which also blocks the loop, Chapter 8). If the data is a sequence of records, process it as a sequence.

Part 3: Web Streams

fetch (in Node 18+ and browsers) returns its body as a Web ReadableStream, a different API from Node streams with the same ideas: pull-based reading, backpressure through a queue with a highWaterMark, and cancellation with cancel().

const res = await fetch(url, { signal });
const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
try {
  while (true) {
    const { value, done } = await reader.read();
    if (done) break;
    parser.push(value); // a string
    if (enough()) break;
  }
} finally {
  await reader.cancel(); // tell the source we are done; fetch closes the connection
}

• getReader() locks the stream to that reader. Nothing else can read it until releaseLock(). A second getReader() throws.
• reader.cancel() propagates upstream: through pipeThrough, to the fetch body, which closes the HTTP connection. In a test against a server streaming a token every 10 ms, the server saw the connection close 2 ms after cancel(). When the code read one chunk and then simply stopped, without cancel(), the server was still streaming 500 ms later.
• TextDecoderStream decodes UTF-8 across chunk boundaries correctly (Part 4 explains why that matters).
• In Node, a ReadableStream is async iterable, so for await (const chunk of res.body) works, and break cancels the stream through return() (tested: the server saw the close within 1 ms). Browser support for async iteration of ReadableStream arrived later and unevenly; check your targets, or use getReader() in browser code.
• Convert between the two worlds with Readable.fromWeb(webStream) and Readable.toWeb(nodeStream).

Part 4: LLM streaming over SSE

Provider streaming APIs send text/event-stream. The wire format is simple:

event: content_block_delta
data: {"type":"content_block_delta","delta":{"text":"Hel"}}

data: {"choices":[{"delta":{"content":"lo"}}]}

: this line is a comment, often a keep-alive

data: [DONE]

Each line is field: value (one optional space after the colon). Lines may end in \n, \r\n or \r. data lines accumulate (several data: lines in one event are joined with \n), event sets the type, id sets the last event id, and a blank line dispatches the event. Lines starting with : are comments.

The chunk-boundary bugs

The network does not know about lines. A chunk can end anywhere: in the middle of a data: line, in the middle of the JSON, between the \r and \n of a line ending, and in the middle of a multi-byte UTF-8 character.

Why { stream: true }: a TextDecoder called with decode(bytes, { stream: true }) keeps the trailing bytes of an incomplete character inside the decoder and emits them with the next call. One decoder must live for the whole stream; call decoder.decode() with no arguments at the end to flush. TextDecoderStream, and a Node stream with encoding: 'utf8' (which uses StringDecoder), do this for you.

Spot the bug: the token reader

export async function streamCompletion(prompt, onToken) {
  const res = await fetch(PROVIDER_URL, { method: 'POST', headers, body: JSON.stringify({ prompt, stream: true }) });
  const reader = res.body.getReader();

  while (true) {
    const { value, done } = await reader.read();
    if (done) break;
    const text = Buffer.from(value).toString('utf8');
    for (const line of text.split('\n')) {
      if (line.startsWith('data: ')) {
        const payload = line.slice(6);
        if (payload === '[DONE]') return;
        onToken(JSON.parse(payload).choices[0].delta.content);
      }
    }
  }
}

It works in every local test. In production, about one response in fifty throws SyntaxError: Unexpected end of JSON input, and users with non-English text see � now and then. What's wrong?

export async function* streamCompletion(prompt, { signal } = {}) {
  const res = await fetch(PROVIDER_URL, { method: 'POST', headers, body: JSON.stringify({ prompt, stream: true }), signal });
  if (!res.ok) throw Object.assign(new Error(`provider ${res.status}`), { status: res.status });
  for await (const ev of sseEvents(res.body)) { // the build below; break cancels res.body
    if (ev.data === '[DONE]') return;
    yield JSON.parse(ev.data).choices[0].delta.content ?? '';
  }
}

Build: an SSE parser that handles any chunking

Spec:

• createSSEParser(onEvent) returns { push(text), end() }. push accepts strings split at any point. For each complete event, call onEvent({ event, data, id, retry }), with event defaulting to 'message'.
• Handle \n, \r\n and \r line endings, including a \r\n split across two pushes. Ignore comment lines and unknown fields. Remove one leading space from the value. Join multiple data lines with \n. An event with no data lines is not dispatched. At end(), drop an unfinished event (the spec says so).
• sseEvents(byteStream) is an async generator over any async iterable of Uint8Array (a fetch body, a Node stream) that decodes UTF-8 correctly across chunks and yields events.
• Test it by feeding every possible split point.

export function createSSEParser(onEvent) {
  let buf = ''; // the incomplete last line
  let pendingCR = false; // last chunk ended in '\r'; a '\n' next is part of the same break
  let data = [];
  let event = '';
  let lastEventId = '';
  let retry;

  function processLine(line) {
    if (line === '') return dispatch(); // a blank line ends the event
    if (line[0] === ':') return; // comment / keep-alive
    const colon = line.indexOf(':');
    const field = colon === -1 ? line : line.slice(0, colon);
    let value = colon === -1 ? '' : line.slice(colon + 1);
    if (value[0] === ' ') value = value.slice(1); // one optional space
    if (field === 'data') data.push(value);
    else if (field === 'event') event = value;
    else if (field === 'id') { if (!value.includes('\0')) lastEventId = value; }
    else if (field === 'retry') { if (/^\d+$/.test(value)) retry = Number(value); }
    // unknown fields are ignored
  }

  function dispatch() {
    if (data.length > 0) {
      onEvent({ event: event || 'message', data: data.join('\n'), id: lastEventId, retry });
    }
    data = [];
    event = '';
  }

  return {
    push(chunk) {
      if (chunk === '') return;
      let i = 0;
      if (pendingCR) {
        pendingCR = false;
        if (chunk[0] === '\n') i = 1; // the '\n' of a '\r\n' split across chunks
      }
      let start = i;
      for (; i < chunk.length; i++) {
        const c = chunk[i];
        if (c !== '\n' && c !== '\r') continue;
        processLine(buf + chunk.slice(start, i));
        buf = '';
        if (c === '\r') {
          if (i + 1 === chunk.length) pendingCR = true;
          else if (chunk[i + 1] === '\n') i++;
        }
        start = i + 1;
      }
      buf += chunk.slice(start);
    },
    // End of stream: an event without its closing blank line is dropped (per the spec).
    end() {
      buf = '';
      data = [];
      event = '';
    },
  };
}

export async function* sseEvents(byteStream) {
  const decoder = new TextDecoder(); // UTF-8; also strips a leading BOM
  const queue = [];
  const parser = createSSEParser((e) => queue.push(e));
  for await (const bytes of byteStream) {
    parser.push(decoder.decode(bytes, { stream: true })); // keep a split character for later
    while (queue.length) yield queue.shift();
  }
  parser.push(decoder.decode()); // flush any bytes still held
  parser.end();
  while (queue.length) yield queue.shift();
}

import { test } from 'node:test';
import assert from 'node:assert/strict';

const BYTES = new TextEncoder().encode(STREAM); // the fixture

async function parseBytes(chunks) {
  const out = [];
  async function* source() { for (const c of chunks) yield c; }
  for await (const e of sseEvents(source())) out.push(e);
  return out;
}

test('every pair of split points (three chunks)', async () => {
  for (let i = 0; i <= BYTES.length; i++) {
    for (let j = i; j <= BYTES.length; j++) {
      const got = await parseBytes([BYTES.subarray(0, i), BYTES.subarray(i, j), BYTES.subarray(j)]);
      assert.deepEqual(got, EXPECTED, `split at ${i},${j}`);
    }
  }
});

test('one byte at a time', async () => {
  assert.deepEqual(await parseBytes(Array.from(BYTES, (b) => Uint8Array.of(b))), EXPECTED);
});

Cancelling when the client disconnects

Your server sits between the user and the provider: the browser calls /chat, your handler calls the provider with stream: true, and forwards the events. If the user closes the tab, the provider does not know unless you tell it. It keeps generating, and you keep paying for output tokens nobody will read.

The handler, with cancellation and backpressure:

app.post('/chat', async (req, res) => {
  const ac = new AbortController();
  res.on('close', () => {
    if (!res.writableFinished) ac.abort(); // closed before we finished: the client left
  });

  try {
    const upstream = await fetch(PROVIDER_URL, {
      method: 'POST',
      headers: providerHeaders,
      body: JSON.stringify({ ...req.body, stream: true }),
      signal: ac.signal,
    });
    res.writeHead(upstream.status, { 'content-type': 'text/event-stream', 'cache-control': 'no-cache' });
    for await (const chunk of upstream.body) {
      if (!res.write(chunk)) await once(res, 'drain', { signal: ac.signal });
    }
    res.end();
  } catch (err) {
    if (!ac.signal.aborted) res.destroy(err); // a real error, not a disconnect
  }
});

Or let pipeline do the forwarding: await pipeline(Readable.fromWeb(upstream.body), res). It applies backpressure and, when res closes early, destroys the upstream stream, which cancels the fetch body. Both versions were tested against a fake provider that streams a token every 10 ms: the provider saw its connection close 3 ms after the client disconnected and had sent 5 tokens, the same number the client read. Without cancellation, it had sent 30 by 300 ms later and was still going.

Details that matter:

• once(res, 'drain', { signal }): without the signal, a client that disconnects while you wait for 'drain' leaves the handler waiting forever, because a closed response never emits 'drain'.
• Listen on res, not req. See the drill.
• If you parse events on the way through (to count tokens, store the reply, or charge credits), use sseEvents on the upstream body and re-serialise; record what was actually delivered, and charge for what the provider generated up to the abort.

Spot the bug: the abort that fires too soon

app.post('/chat', async (req, res) => {
  const ac = new AbortController();
  req.on('close', () => ac.abort()); // stop the provider if the user leaves

  const upstream = await fetch(PROVIDER_URL, { method: 'POST', body: JSON.stringify(req.body), signal: ac.signal });
  res.writeHead(200, { 'content-type': 'text/event-stream' });
  for await (const chunk of upstream.body) res.write(chunk);
  res.end();
});

Every request fails at once with AbortError, and the client gets nothing. What's wrong?

res.on('close', () => {
  if (!res.writableFinished) ac.abort();
});

Part 5: Async iterator pipelines

Async generators compose. Each stage takes an async iterable and returns one, and because each pulls from the stage before it only when asked, the whole chain runs at the speed of the slowest stage with a bounded amount in memory.

async function* lines(source) {
  let rest = '';
  for await (const chunk of source) {
    rest += chunk;
    const parts = rest.split('\n');
    rest = parts.pop(); // the last piece may be half a line
    yield* parts;
  }
  if (rest) yield rest;
}

async function* filter(source, pred) {
  for await (const x of source) if (pred(x)) yield x;
}

await pipeline(
  createReadStream('docs.ndjson', { encoding: 'utf8' }), // StringDecoder: split characters are safe
  lines,
  async function* (src) { for await (const line of src) yield JSON.parse(line); },
  (src) => filter(src, (doc) => !doc.deleted),
  (src) => mapConcurrent(src, 8, embed), // the build below
  async function* (src) { for await (const row of src) yield JSON.stringify(row) + '\n'; },
  createWriteStream('embedded.ndjson'),
);

lines is the same chunk-boundary fix as the SSE parser: keep the incomplete tail. The pipeline was run with a 7-byte highWaterMark to force splits everywhere, on 1,000 records containing "é" and "€"; all 500 kept records came out intact and in order.

What exists already, so you know when not to write your own:

Build: mapConcurrent(iterable, limit, fn)

Spec: an async generator. It pulls from iterable (sync or async), runs fn(item, index, signal) on up to limit items at once, and yields results in input order. It pulls lazily: never more than limit items ahead of what has been yielded. A rejection is thrown to the consumer at that item's turn, without crashing the process if it happens early. When the consumer breaks or an error is thrown, it closes the source (calls its return()) and aborts signal for in-flight calls.

export async function* mapConcurrent(iterable, limit, fn) {
  if (!Number.isInteger(limit) || limit < 1) throw new RangeError('limit must be an integer >= 1');
  const it = iterable[Symbol.asyncIterator]?.() ?? iterable[Symbol.iterator]();
  const ac = new AbortController();
  const window = []; // promises of started items, oldest first
  let sourceDone = false;
  let index = 0;

  try {
    while (true) {
      // Top up: pull from the source until `limit` items are started but not yet yielded.
      while (!sourceDone && window.length < limit) {
        const { value, done } = await it.next();
        if (done) {
          sourceDone = true;
          break;
        }
        const p = Promise.resolve().then(() => fn(value, index++, ac.signal));
        p.catch(() => {}); // handled later, when its turn comes; stops an early unhandled rejection
        window.push(p);
      }
      if (window.length === 0) return;
      yield await window.shift(); // in order; a rejection is thrown here
    }
  } finally {
    ac.abort(); // tell in-flight calls to stop (error, break, or return)
    if (!sourceDone) await it.return?.(); // close the source
  }
}

The Causal Chains

Review checklist

1. Whole-body reads of large or unbounded data: res.text(), res.json(), readFile, Buffer.concat, toArray, Array.fromAsync.
2. .pipe( anywhere in server code: replace with pipeline and handle its rejection.
3. write( in a loop without checking the return value and waiting for 'drain' (with a signal).
4. Manual stream parsing: is there a buffer for the incomplete tail? Is there one TextDecoder with { stream: true } per stream? Does it handle \r\n?
5. SSE consumers parse per event and check res.ok before streaming.
6. Early exits from a reader loop call reader.cancel() (or use for await with break); manually driven iterators call return().
7. Proxy routes abort the upstream call on res 'close' when !res.writableFinished, and not on req 'close'.
8. for await over arrays of promises: replace with Promise.allSettled or attach handlers up front.
9. Hand-written concurrent iterators attach a handler to each started promise and close the source in finally.

What's next

You now have the parts: ordering (Chapters 1–4), races and cancellation (5–6), limits and failure policy (7), the process (8) and streams (9). Chapter 10 — Reviewing a Codebase and Defending Your Designs puts them together the way the interview does: a codebase you did not write, 30 minutes, and the questions to ask out loud.