@kylecodes/concurrency-tools

Small async concurrency primitives for TypeScript, composed from two pieces — a bounded pool and an async channel — into the two helpers you actually reach for:

| Export | What it does | | ------------------------------------------------ | ------------------------------------------------------------------------------------------------------ | | runPool(items, size, fn) | Bounded-concurrency Promise.allSettled(items.map(fn)) — results in input order | | boundedConcurrencyPoolStream(source, size, fn) | Stream fn over a (possibly lazy/infinite) source with backpressure — results in completion order | | createConcurrencyPool({ size }) | The underlying pool: submit, whenCapacityAvailable, onIdle | | createChannel() | The underlying unbounded async queue: push / close / fail / for await |

Zero dependencies. Plain ESM. No timers, no Node-specific APIs — runs on Node ≥18, Bun, Deno, and browsers.

Install

npm install @kylecodes/concurrency-tools
# or
bun add @kylecodes/concurrency-tools

runPool — bounded allSettled over a fixed array

You have N tasks and want at most size in flight, every outcome reported, and results lined up with inputs:

import { runPool } from "@kylecodes/concurrency-tools";

const settled = await runPool(job.tasks, 4, (task) => runTask(task));

for (const [i, outcome] of settled.entries()) {
  if (outcome.status === "fulfilled") {
    console.log(job.tasks[i].id, "→", outcome.value);
  } else {
    console.error(job.tasks[i].id, "failed:", outcome.reason);
  }
}

Like Promise.allSettled, it never rejects — one task's failure is isolated to its own entry and never affects siblings.

boundedConcurrencyPoolStream — backpressured streaming map

The workhorse for pipelines where the input is lazy (paginated API, DB cursor, message queue) and you want results as soon as each one finishes:

import { boundedConcurrencyPoolStream } from "@kylecodes/concurrency-tools";

// A lazy source: pages are only fetched as the pool has room.
async function* messageIds(): AsyncGenerator<string> {
  let pageToken: string | undefined;
  do {
    const page = await listMessages({ pageToken });
    yield* page.ids;
    pageToken = page.nextPageToken;
  } while (pageToken);
}

// At most 8 fetches in flight; each result yields the moment it completes.
for await (const message of boundedConcurrencyPoolStream(
  messageIds(),
  8,
  (id) => fetchMessage(id),
)) {
  await store(message);
}

Properties that make this safe for long pipelines:

• Backpressure. The source is pulled one item per free slot — a million-row cursor never piles up in memory because the producer suspends while the pool is saturated.
• Completion order. Fast items don't wait behind slow ones.
• Fail fast. The first rejection (from fn or the source) fails the stream; in-flight siblings settle quietly and are dropped.
• Early exit. break-ing out of the loop stops the producer from pulling the source further.

It also makes a tidy worker loop — stream over an infinite source of claimed jobs and let the pool pace how fast you claim:

for await (const _ of boundedConcurrencyPoolStream(
  claimedJobs(),
  concurrency,
  process,
)) {
  // drained for its side effects
}

createConcurrencyPool — the primitive underneath

When the two helpers don't fit, use the pool directly:

import { createConcurrencyPool } from "@kylecodes/concurrency-tools";

const pool = createConcurrencyPool({ size: 3 });

// Run a thunk under the bound. Resolves/rejects with the thunk's own outcome.
const result = await pool.submit(() => doWork());

// Pace a producer: resolves once active + queued < size.
await pool.whenCapacityAvailable();

// Wait for full drain (nothing active, nothing queued). Re-arms after idle.
await pool.onIdle();

pool.activeCount; // thunks currently running
pool.queuedCount; // thunks waiting for a slot

Guarantees: at most size thunks in flight (size: 1 is strictly sequential), one slot refills per completion (no batching), and a throwing thunk always frees its slot — a failure can never stall the pool.

createChannel — a minimal async queue

A single-consumer, unbounded channel connecting a pushing producer to a for await consumer:

import { createChannel } from "@kylecodes/concurrency-tools";

const ch = createChannel<string>();

ch.push("a"); // never blocks
ch.push("b");
ch.close(); // or ch.fail(err) — iteration throws after the buffer drains

for await (const value of ch) {
  console.log(value); // 'a', 'b'
}

Buffered values always drain before a close/fail takes effect, and the first failure wins.

Development

bun install
bun run build   # tsc → dist/ (ESM + .d.ts + source maps)
bun test

License

MIT