Async iterators and generators: pull-based streams without the buffer

The async iteration protocol: next() returns a promise

Why does this matter to you as an engineer? Because the protocol is exactly what gives for await its bounded-memory guarantee — understanding it lets you build data pipelines that never OOM regardless of result set size, and explain to a reviewer why your code is safe where a naive await db.query(...) is not.

A normal iterable exposes Symbol.iterator; an async iterable exposes Symbol.asyncIterator, a method that returns an iterator whose next() returns a Promise<{ value, done }> instead of a plain { value, done }. That single difference — the result is a promise — is what lets each step await something (a network round-trip, a disk read) before producing the next value. for await (const x of it) is pure sugar: it calls it[Symbol.asyncIterator]() once, then loops const { value, done } = await iter.next() until done is true, running the body for each value. It is sequential by construction — the loop awaits one next() fully before asking for the following one.

You almost never write the protocol by hand. An async generator — async function* with yield — is the ergonomic author’s tool: it builds a conforming async iterator for you, and every yield suspends the function until the consumer calls next() again. The function literally stops at yield and does not advance until pulled.

// Paginate a cursor-based API, yielding one record at a time.
async function* fetchAllUsers(client) {
  let cursor = null;
  do {
    const page = await client.get("/users", { cursor, limit: 100 });
    for (const user of page.items) {
      yield user;        // suspends here until the consumer pulls the next value
    }
    cursor = page.nextCursor;
  } while (cursor);       // only fetches the next page when the buffer is drained
}

// Consume it: one user at a time, never more than one page in memory.
for await (const user of fetchAllUsers(client)) {
  await index(user);     // each await fully completes before the next user is pulled
}

The page loop cannot run ahead of the consumer: after yielding the last user of page N, the generator is frozen mid-for; it only resumes — and only then issues the request for page N+1 — when index(user) resolves and the for await calls next() again. Memory is bounded to one page plus one in-flight record, whether the API has 100 users or 100 million.

Pull beats push: backpressure you get for free

This is the senior insight. Callbacks, EventEmitters, and readable.on('data', …) are push: the producer decides when data arrives and fires it at you. If you can’t keep up — a slow DB write behind a fast socket read — the events keep coming and you must implement backpressure by hand: pause the source, buffer, track the buffer size, resume when drained. Forget any of that and the unbounded buffer becomes the OOM in the hook, just relocated.

Async iterators are pull: nothing is produced until the consumer calls next(). Backpressure is therefore structural, not a feature you bolt on — the producer is suspended at yield and is physically incapable of running ahead, because the only thing that resumes it is your next pull. A slow consumer simply pulls slower; the producer waits. And because a Node Readable implements Symbol.asyncIterator, you get this for free over real I/O:

import { createReadStream } from "node:fs";
import { createInterface } from "node:readline";

// readline over a Readable is itself async-iterable.
const rl = createInterface({
  input: createReadStream("huge.log"),    // a 40 GB file is fine
  crlfDelay: Infinity,
});

for await (const line of rl) {
  await ship(line);   // if ship() is slow, the file read naturally throttles
}

While ship(line) is pending, the for await does not call next(), so readline does not request more, so the underlying file stream is not read past its highWaterMark. The OS read pauses. You wrote no .pause(), no .resume(), no buffer accounting — the pull model enforced it.

Cleanup on early exit: finally is where cursors close

A pull producer holding a resource — a DB cursor, an open file handle, a pooled connection — has a real risk: the consumer might break, return, or throw out of the for await loop before the source is exhausted. The runtime handles this. When you leave a for await early, it calls the iterator’s return() method. Inside an async generator, return() makes the suspended yield act as if a return ran there: the code after the current yield does not execute, but any enclosing finally block does. That finally is the only reliable place to release the resource.

async function* streamRows(pool, sql) {
  const conn = await pool.connect();
  const cursor = conn.cursor(sql);
  try {
    for (let batch; (batch = await cursor.read(500)).length; ) {
      for (const row of batch) yield row;   // consumer may break out mid-stream
    }
  } finally {
    // Runs on normal exhaustion AND on break/return/throw, via the generator's return().
    await cursor.close();
    conn.release();        // the leak you'd get without finally: a held connection forever
  }
}

for await (const row of streamRows(pool, "SELECT * FROM events")) {
  if (await isDuplicate(row)) break;   // triggers return() → the finally above runs
  await upsert(row);
}

The failure mode is concrete: put the cursor.close() and conn.release() after the loop instead of in finally, and the first break skips them. The connection is never returned to the pool; do that a few thousand times and the pool is exhausted and every subsequent query hangs. One subtlety to keep on your radar: return() only fires if iteration actually starts and then stops via break/return/throw or runs to completion. If you grab the iterator manually (const it = gen[Symbol.asyncIterator]()), pull a few values, and then just drop the reference without calling it.return() or exhausting it, the generator stays suspended at its last yield forever and the finally never runs — the resource leaks. for await always does the right thing; manual iteration makes cleanup your job.

Composition: lazy pipelines at constant memory

Because a generator both is an async iterable and can consume one, you can chain them into a pipeline where each stage transforms one item at a time. Compare the eager array style — (await getAll()).filter(...).map(...) — which materializes every intermediate array, with a lazy generator pipeline that holds one item per stage:

async function* mapGen(src, fn)    { for await (const x of src) yield fn(x); }
async function* filterGen(src, ok) { for await (const x of src) if (ok(x)) yield x; }

// source → filter → map → consumer, one row flowing through at a time
const pipeline =
  mapGen(
    filterGen(streamRows(pool, "SELECT * FROM events"), (r) => r.amount > 0),
    (r) => ({ id: r.id, cents: Math.round(r.amount * 100) }),
  );

for await (const record of pipeline) {
  await sink.write(record);   // backpressure propagates all the way to the cursor
}

The memory math is the whole point. Materializing 10M rows averaging ~150 bytes is ~1.5 GB of live array — exactly the OOM in the hook, and it gets worse with each intermediate .map/.filter that allocates another full array. The generator pipeline is O(1) in the number of rows: at any instant there is one row in streamRows, one in filterGen, one in mapGen, one being written — four objects, not ten million. You can stream a result set larger than RAM. And because pull-backpressure threads through every stage, a slow sink.write throttles the DB cursor at the far end with no extra code.