task-queue.ts in TypeScript

A typed task queue running jobs with bounded concurrency.

#!/usr/bin/env node
// A typed task queue that runs jobs with bounded concurrency.

type Job<T> = () => Promise<T>;

interface Outcome<T> {
  index: number;
  value?: T;
  error?: string;
}

async function runPool<T>(
  jobs: Job<T>[],
  limit: number,
): Promise<Outcome<T>[]> {
  const results: Outcome<T>[] = [];
  let next = 0;

  async function worker(): Promise<void> {
    while (next < jobs.length) {
      const index = next++;
      try {
        results.push({ index, value: await jobs[index]() });
      } catch (error) {
        const message = error instanceof Error ? error.message : "unknown";
        results.push({ index, error: message });
      }
    }
  }

  const size = Math.min(limit, jobs.length);
  await Promise.all(Array.from({ length: size }, worker));
  return results.sort((a, b) => a.index - b.index);
}

const jobs: Job<string>[] = [1, 2, 3, 4, 5].map((n) => async () => {
  await new Promise((r) => setTimeout(r, n * 20));
  if (n === 3) throw new Error("job 3 failed");
  return `job ${n} done`;
});

const outcomes = await runPool(jobs, 2);
for (const outcome of outcomes) {
  console.log(outcome.error ? `x ${outcome.error}` : `. ${outcome.value}`);
}

How it works

1. Job<T> and Outcome<T> type the work and results.
2. Workers pull the next index until the queue drains.
3. Failures are captured per job, not thrown away.

Keywords and builtins used here

• Array
• Job
• Math
• Outcome
• Promise
• T
• async
• await
• catch
• const
• for
• from
• function
• if
• interface
• let
• message
• number
• of
• return
• size
• string
• throw
• try
• type
• while

The run, in numbers

Lines

45

Characters to type

1100

Tokens

344

Three-star pace

115 tpm

At the three-star pace of 115 tokens a minute, this run takes about 179 seconds.

Type this snippet

Step 1 of 1 in Encore, step 8 of 8 in Async & modules.

← Previous