Chapter 26: Concurrency and Streams

Concurrency in Node comes from cooperative asynchronous execution, worker threads for CPU bound tasks, and streaming primitives that move data in small chunks. TypeScript adds confidence through typed iterators, typed messages, and safe resource handling. This chapter ties these together and shows practical patterns that avoid contention while maintaining backpressure and predictable flow.

Async iteration and typed streams

Async iteration lets you consume data chunk by chunk with for await. When you pair it with Node streams, you get a natural way to process files, sockets, and generated values without buffering everything. TypeScript types document each chunk and reduce accidental mixing of object and byte modes.

AsyncIterator fundamentals

An async iterable exposes [Symbol.asyncIterator]() and yields values over time. This meshes well with I/O and timers because each await cooperates with the event loop; nothing blocks between chunks.

async function* lines(src: AsyncIterable<Buffer>): AsyncIterable<string> {
  let acc = "";
  for await (const buf of src) {
    acc += buf.toString("utf8");
    let idx;
    while ((idx = acc.indexOf("\n")) >= 0) {
      yield acc.slice(0, idx);
      acc = acc.slice(idx + 1);
    }
  }
  if (acc) yield acc;
}

Bridging Node stream.Readable to async iteration

Every modern Node Readable is also an async iterable, so you can loop with for await. This avoids manual data handlers and keeps control flow linear.

import { createReadStream } from "node:fs";
import { once } from "node:events";

async function* readUtf8(path: string): AsyncIterable<string> {
  const rs = createReadStream(path, { encoding: "utf8" });
  for await (const chunk of rs) yield chunk as string;
  await once(rs, "close");
}

Typed object mode streams

When you set objectMode: true, a stream passes arbitrary JavaScript values instead of bytes. Define a type for the chunk shape and keep it consistent across your pipeline.

type Row = { id: string; name: string; email?: string };
import { Readable, Transform, Writable } from "node:stream";

const rows = Readable.from<Row>([{ id: "1", name: "Ada" }, { id: "2", name: "Linus", email: "l@…" }], { objectMode: true });

const toCsv = new Transform({
  objectMode: true,
  transform(chunk: Row, _enc, cb) {
    const line = [chunk.id, chunk.name, chunk.email ?? ""].map(v => JSON.stringify(v)).join(",") + "\n";
    cb(null, line);
  }
});

const out = new Writable({
  write(chunk, _enc, cb) {
    process.stdout.write(chunk);
    cb();
  }
});

rows.pipe(toCsv).pipe(out);

Worker threads with typed messages

Worker threads run JavaScript in parallel on separate threads. Use them for CPU heavy tasks such as parsing, compression, or cryptography. Structure your messages with a discriminated union so both sides get exhaustive checks and safe narrowing.

Message protocol as a discriminated union

A small schema goes a long way. Include a type field for narrowing and a reqId so you can correlate responses to requests when concurrency increases.

type Req =
  | { type: "hash"; reqId: string; input: string; algo: "sha256" | "sha512" }
  | { type: "compress"; reqId: string; input: Buffer };

type Res =
  | { type: "ok"; reqId: string; result: Buffer | string }
  | { type: "error"; reqId: string; message: string };

Main thread creating a worker

The main thread posts typed messages and awaits responses. Keep a pending map keyed by reqId. This pattern supports many in flight jobs while staying type safe.

import { Worker } from "node:worker_threads";
import { randomUUID } from "node:crypto";

const worker = new Worker(new URL("./worker.js", import.meta.url), { execArgv: ["--experimental-import-meta-resolve"] });

const pending = new Map<string, (res: Res) => void>();

worker.on("message", (res: Res) => {
  const cb = pending.get(res.reqId);
  if (cb) { pending.delete(res.reqId); cb(res); }
});

function call(req: Omit<Req, "reqId">): Promise<Res> {
  const reqId = randomUUID();
  const full = { ...req, reqId } as Req;
  return new Promise(resolve => {
    pending.set(reqId, resolve);
    worker.postMessage(full);
  });
}

Worker thread implementation

The worker listens for messages, narrows by type, and posts a response. Avoid synchronous CPU bound loops on the main thread; move them here.

// worker.ts
import { parentPort } from "node:worker_threads";
import { createHash } from "node:crypto";
import { brotliCompress } from "node:zlib";

parentPort!.on("message", async (req: Req) => {
  try {
    if (req.type === "hash") {
      const h = createHash(req.algo).update(req.input, "utf8").digest("hex");
      parentPort!.postMessage({ type: "ok", reqId: req.reqId, result: h } satisfies Res);
    } else if (req.type === "compress") {
      const out = await new Promise<Buffer>((res, rej) => brotliCompress(req.input, (e, b) => e ? rej(e) : res(b)));
      parentPort!.postMessage({ type: "ok", reqId: req.reqId, result: out } satisfies Res);
    }
  } catch (e) {
    parentPort!.postMessage({ type: "error", reqId: req.reqId, message: (e as Error).message } satisfies Res);
  }
});

Queues and background jobs

Background jobs decouple request handling from work execution. A queue accepts items fast and processes them at a stable rate. You can keep it in process for simplicity or delegate to a shared broker when you scale beyond one instance.

A minimal in process queue

This queue uses an async generator to pull jobs and a concurrency limiter to run a few at a time. The type parameter ensures producers and consumers agree on the job shape.

type Job = { id: string; payload: { userId: string; email: string } };

class AsyncQueue<T> {
  private buf: T[] = [];
  private resolvers: Array<(v: IteratorResult<T>)> = [];
  enqueue(item: T) {
    const r = this.resolvers.shift();
    if (r) r({ value: item, done: false });
    else this.buf.push(item);
  }
  async *[Symbol.asyncIterator](): AsyncIterator<T> {
    while (true) {
      if (this.buf.length) yield this.buf.shift()!;
      else yield await new Promise<T>(res => this.resolvers.push(v => res(v.value)));
    }
  }
}

async function runQueue(q: AsyncQueue<Job>, concurrency = 4) {
  const running = new Set<Promise<unknown>>();
  for await (const job of q) {
    const p = handle(job).finally(() => running.delete(p));
    running.add(p);
    if (running.size >= concurrency) await Promise.race(running);
  }
}

async function handle(job: Job) {
  await sendEmail(job.payload); // pretend to send; returns a promise
}

Scheduling and retries with jitter

Retries should back off to avoid thundering herds. Add jitter so many jobs do not wake at once.

function backoff(attempt: number, baseMs = 250, capMs = 10_000) {
  const exp = Math.min(capMs, baseMs * 2 ** attempt);
  const jitter = Math.floor(Math.random() * baseMs);
  return exp + jitter;
}

Persisting jobs safely

In memory queues lose jobs on process exit. When durability matters, persist to a database table or a broker. A simple durable table uses a status field and a lease to keep one worker per job.

-- SQL shape for durability
CREATE TABLE job_queue (
  id TEXT PRIMARY KEY,
  kind TEXT NOT NULL,
  payload JSON NOT NULL,
  status TEXT NOT NULL DEFAULT 'queued',
  run_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  lease_until TIMESTAMP NULL
);

Backpressure and flow control

Backpressure keeps producers from outrunning consumers. Streams implement this with internal buffers, highWaterMark, and return values from write(). Flow control protects memory, reduces latency variance, and improves throughput by keeping queues short.

Understanding write() return values

When writable.write(chunk) returns false, the buffer is full. Pause the producer and wait for drain before continuing. This pattern respects downstream limits.

function writeMany(w, chunks: Buffer[]) {
  for (const c of chunks) {
    if (!w.write(c)) {
      w.once("drain", () => writeMany(w, chunks.slice(chunks.indexOf(c) + 1)));
      return;
    }
  }
  w.end();
}

Using stream.pipeline for safety

pipeline wires streams with proper error forwarding and cleanup. It also pairs well with async generators and transforms.

import { pipeline } from "node:stream/promises";
import { createReadStream, createWriteStream } from "node:fs";
import { createGzip } from "node:zlib";

await pipeline(
  createReadStream("input.txt"),
  createGzip(),
  createWriteStream("output.txt.gz")
);

Tuning highWaterMark and object sizes

Right size your buffers. Large chunks reduce overhead but increase latency for the last byte and risk memory pressure. Small chunks improve interactivity but raise syscall and scheduling cost. Measure with realistic inputs before fixing values.

import { createReadStream } from "node:fs";

// bytes in byte mode; items in object mode
const rs = createReadStream("big.bin", { highWaterMark: 1024 * 64 }); // 64 KiB

Building an async transform with backpressure

You can implement transforms as async generators and still participate in flow control. Each yield waits for the downstream consumer to pull the next value.

async function* mapAsync<A, B>(src: AsyncIterable<A>, fn: (a: A) => Promise<B> | B) {
  for await (const a of src) {
    yield await fn(a);
  }
}

const upper = mapAsync(readUtf8("input.txt"), s => s.toUpperCase());
for await (const chunk of upper) {
  process.stdout.write(chunk);
}

With these tools you can keep I/O flowing, keep CPU heavy work off the event loop, and keep your types aligned from edges to core. Concurrency and streams become predictable once you respect backpressure, define message protocols, and measure buffer sizes under load.