API reference › @evolu/common › Task › each

function each<TTasks>(
  tasks: TTasks,
  onResult: EachCallback<TTasks>,
  options?: TaskCollectionOptions,
): Task<
  void,
  never,
  ParameterIntersection<
    TTasks[number] extends TTask
      ? TTask extends AnyTask
        ? (deps: InferTaskDeps<TTask>) => void
        : never
      : never
  >
>;

Defined in: packages/common/src/Task.ts:4668

Runs Tasks under a concurrency limit and calls onResult for each Task Result as it settles.

onResult receives the Result and the original input index; call order is settlement order, not input order. Returning continue lets queued Tasks start when capacity is available. Returning stop prevents queued Tasks from starting and aborts already-running Tasks through structured Run disposal — each still waits for them to settle before returning.

each is the scheduling primitive under the collection helpers. Each one is a small onResult policy:

Use each directly to build a collection policy the helpers don't cover. For example, keep the first successful value together with its original input index:

Example

import {
  assertEqual,
  assertFalse,
  assertOk,
  createRun,
  each,
  err,
  ok,
  sleep,
  type Task,
  type Typed,
} from "@evolu/common";

let slowCompleted = false;
const slow: Task<string> = async (run) => {
  await run.ok(sleep("10ms"));
  slowCompleted = true;
  return ok("slow");
};
const unavailable: Task<never, ServiceUnavailableError> = () =>
  err({ type: "ServiceUnavailable" });

interface ServiceUnavailableError extends Typed<"ServiceUnavailable"> {}

const tasks = [slow, unavailable, () => ok("fast")] as const;
let first: readonly [string, number] | undefined;
await using run = createRun();
const result = await run(
  each(
    tasks,
    (result, index) => {
      if (!result.ok) return "continue";
      first = [result.value, index];
      return "stop";
    },
    { concurrency: 2 },
  ),
);

assertOk(result, undefined);
assertEqual(first, ["fast", 2]);
assertFalse(slowCompleted);

onResult is a synchronous scheduling decision, not a place to do work. It runs in the scheduler's own continuation, bracketed by abort checks, and its return value gates whether queued Tasks may start. For async work per result, put it inside the Task itself — the Task is the async slot — or start a supervised side effect with void run(task) from inside the callback and keep the decision synchronous. Like RetryOptions.shouldRetry and RetryOptions.onRetry, onResult must not throw: a thrown exception is a defect that panics the Run tree.

Sequential by default; pass a concurrency option to run more than one Task at a time. Defects from child Tasks keep caller-linked async stack traces; building on each preserves diagnostics that a hand-rolled scheduling loop typically loses.