Concurrency

Concurrency#

Concurrency in TypeScript is JavaScript’s (see Concurrency): a single event loop, Promise<T>, async / await, AbortController, and worker threads. The type layer carries promise types through (Promise<User>), strips them with Awaited<T>, and types worker message structures.

This page covers the TypeScript-specific patterns. For the runtime mechanics, see the JavaScript page.

Typed promises#

Promise<T> carries the resolved type; async functions return Promise<T> where T is the function’s return type.

async function getUser(id: string): Promise<User> {
  const r = await fetch(`/users/${id}`);
  if (!r.ok) throw new Error(`HTTP ${r.status}`);
  return r.json() as Promise<User>;
}

const user: User = await getUser("1");

For unwrapping nested promises, Awaited<T> (the deep-unwrap helper) drops every layer.

type Inner = Awaited<Promise<Promise<User>>>;   // User

Promise combinators (typed)#

The combinators carry types automatically when the array elements have known types.

// Promise.all: tuple in, tuple out
const [a, b]: [User, Org] = await Promise.all([loadUser(id), loadOrg(id)]);

// Promise.allSettled: per-element result shape
const results = await Promise.allSettled(urls.map(fetch));
const ok = results.flatMap(r => r.status === "fulfilled" ? [r.value] : []);
//         ^? Response[]

// Promise.race: first to settle
const winner = await Promise.race([primary(), backup()]);
//              ^? T (the common type)

// Promise.any: AggregateError if all reject
try { const v = await Promise.any(candidates); }
catch (e) { if (e instanceof AggregateError) /* … */; }

Cancellation with AbortSignal#

AbortController is typed; signal is AbortSignal. Most async APIs accept it.

const ctrl = new AbortController();
const t = setTimeout(() => ctrl.abort(), 5000);

try {
  const r: Response = await fetch(url, {signal: ctrl.signal});
} catch (e) {
  if (e instanceof Error && e.name === "AbortError") log.warn("timed out");
  else throw e;
} finally {
  clearTimeout(t);
}

// shortcut
const r2 = await fetch(url, {signal: AbortSignal.timeout(3000)});

Concurrency cap#

A typed pool function keeps N requests in flight at a time.

async function pool<T, U>(items: T[], n: number, fn: (item: T) => Promise<U>): Promise<U[]> {
  const out: U[] = [];
  const queue = [...items];
  await Promise.all(Array.from({length: n}, async () => {
    while (queue.length) {
      const item = queue.shift()!;
      out.push(await fn(item));
    }
  }));
  return out;
}

const responses: Response[] = await pool(urls, 5, u => fetch(u));

Worker threads with typed messages#

Node’s worker_threads is untyped at the message boundary; the operator wraps it with a shared message-type definition.

// protocol.ts
export type WorkerMessage =
  | {kind: "hash"; data: Uint8Array}
  | {kind: "exit"};

export type WorkerResponse =
  | {kind: "hash"; sha256: string}
  | {kind: "error"; message: string};

// main.ts
import {Worker} from "node:worker_threads";
import type {WorkerMessage, WorkerResponse} from "./protocol.js";

const w = new Worker("./worker.js");
w.on("message", (m: WorkerResponse) => {
  if (m.kind === "hash") console.log(m.sha256);
});
const req: WorkerMessage = {kind: "hash", data};
w.postMessage(req);

// worker.ts
import {parentPort} from "node:worker_threads";
import {createHash}  from "node:crypto";
import type {WorkerMessage, WorkerResponse} from "./protocol.js";

parentPort!.on("message", (m: WorkerMessage) => {
  if (m.kind === "hash") {
    const sha256 = createHash("sha256").update(m.data).digest("hex");
    const reply: WorkerResponse = {kind: "hash", sha256};
    parentPort!.postMessage(reply);
  }
});

The discriminant (kind) drives both the parsing inside the worker and the type narrowing on the main side.

Async generators#

async function* yields typed values; for await...of consumes them.

async function* paginate(url: string): AsyncGenerator<User, void, undefined> {
  let next: string | null = url;
  while (next) {
    const r: Response = await fetch(next);
    const page: {users: User[]; next: string | null} = await r.json();
    for (const u of page.users) yield u;
    next = page.next;
  }
}

for await (const u of paginate("/users")) {
  handle(u);                       // u: User
}

Effect (advanced)#

For programs where you want typed concurrency, cancellation, retries, and dependency injection together, the effect library models computation as Effect<R, E, A> (requirements, error, success). It replaces Promise / try / catch with a typed pipeline.

This is heavyweight; reach for it on long-lived services with complex error flow, not for one-off scripts.

References#

Contents