Capstone: Ordered Task Pool

The complete library connects the week’s contracts. It obtains one synchronous iterator, starts a bounded number of worker loops, assigns each item an index before awaiting, and writes results at that index. It stops pulling once a failure is observed.

Each invocation owns an internal AbortController. External cancellation or the first mapper failure aborts peers, but the library awaits active callbacks before settling. It therefore depends on callbacks eventually settling. TaskError preserves the mapper index and cause; source errors and external abort reasons follow their documented policies.

Read the failed flag independently of failure’s value. Notice cleanup in finally: remove the external listener and close an unfinished iterator. A cleanup failure does not overwrite a primary error. This is a finite-input, collected-output API, not an async-iterable or streaming-output system.

Download the project to inspect the real modules and controlled-schedule tests. The long example below is the same implementation combined into one editor; it needs no account or network service.

Predict, run, and explain

Write the output and trace the important state changes before checking it. Then run the program. Each editor is independent. Async examples end with main() so this editor waits for their work.

Predict, then run JavaScript

async function main() {
  /** A synchronous iterable, async mapper, bounded active calls, ordered results.
   * Stop pulling on the first observed failure; cooperatively abort peers, then
   * wait for active callbacks before rejecting. Callbacks must settle themselves.
   */
  class TaskError extends Error {
    constructor(index, cause) {
      super(`Task ${index} failed`, { cause });
      this.name = "TaskError";
      this.index = index;
    }
  }

  async function mapLimit(items, mapper, { limit = 2, signal } = {}) {
    if (!Number.isInteger(limit) || limit < 1 || limit > 64) {
      throw new RangeError("limit must be an integer from 1 to 64");
    }

    if (typeof mapper !== "function") {
      throw new TypeError("mapper must be a function");
    }

    if (typeof items?.[Symbol.iterator] !== "function") {
      throw new TypeError("Expected an iterable");
    }

    if (signal !== undefined && !(signal instanceof AbortSignal)) {
      throw new TypeError("Expected an AbortSignal");
    }

    signal?.throwIfAborted();

    const iterator = items[Symbol.iterator]();
    const controller = new AbortController();
    const results = [];
    let nextIndex = 0;
    let exhausted = false;
    let failed = false;
    let failure;
    const fail = error => {
      if (failed) {
        return;
      }

      failed = true;
      failure = error;
      controller.abort(error);
    };
    const onAbort = () => fail(signal.reason);

    signal?.addEventListener("abort", onAbort, { once: true });

    // A custom iterable may have triggered an abort during iterator creation.
    if (signal?.aborted) {
      onAbort();
    }

    async function worker() {
      while (!failed && !exhausted) {
        let step;

        try {
          step = iterator.next();

          if (
            step === null ||
            (typeof step !== "object" && typeof step !== "function")
          ) {
            throw new TypeError("Iterator result must be an object");
          }
        } catch (error) {
          fail(error);
          return;
        }

        if (step.done) {
          exhausted = true;
          return;
        }

        if (failed) {
          return;
        }

        const index = nextIndex++;

        try {
          results[index] = await mapper(step.value, index, controller.signal);
        } catch (error) {
          fail(new TaskError(index, error));
        }
      }
    }

    try {
      await Promise.all(Array.from({ length: limit }, () => worker()));
    } finally {
      signal?.removeEventListener("abort", onAbort);

      if (!exhausted) {
        try {
          iterator.return?.();
        } catch (error) {
          if (!failed) {
            fail(error);
          }
        }
      }
    }

    if (failed) {
      throw failure;
    }

    return results;
  }

  console.log(
    (await mapLimit([3, 1, 2], async value => value * 10, { limit: 2 })).join(
      ","
    )
  );
}

main();

Check your prediction

30,10,20

Read the pool as four collaborating phases

Trace one input through validation, source consumption, callback execution and cleanup instead of memorizing the listing. Keep input index, active-call count and failure state in separate columns. The worker reserves an index before awaiting and writes the result there. After failure, it stops new pulls while the outer operation still waits for active callbacks to settle.

Boundary or case	What to inspect
Invocation	Validate limit, mapper, source and optional signal
Worker	Pull with capacity and reserve the result position before await
Settlement	Preserve primary failure and release listener and unfinished iterator

Use three inputs with visibly different outputs. Follow ordinary completion, then the controlled reverse-completion test. Identify the exact point new work stops after failure and which already active callback must still settle before rejection.

Your turn

Exercise

Capstone: Ordered Task Pool: practise

Without copying the pool, implement a sequential async map over an iterable. Preserve input order, await each callback, and return a fresh array. Print 6,2,4 for [3,1,2] with an async doubler.

Your workspace JavaScript

// This is the limit-one baseline. It does not need a worker pool.

Hint 1: the idea

Establish a correct baseline before optimizing concurrency.

Hint 2: a first step

Use for/of with await inside the loop and push each accepted result.

Show one solution

async function main() {
  async function sequentialMap(values, mapper) {
    const output = [];

    for (const value of values) {
      output.push(await mapper(value));
    }

    return output;
  }

  console.log(
    (await sequentialMap([3, 1, 2], async value => value * 2)).join(",")
  );
}

main();

6,2,4

Explain why the implementation meets the stated contract. Which part is enforced by the language or checker, and which part needs behavioral evidence?

Change it and test it

Compare the baseline with the full pool at limit 1 and 2 using controlled gates. Test an empty source. Explain why the full implementation needs more lifecycle logic than the baseline, and list the callback assumptions that remain.