Overview

All

Run keyed behaviors and cancel active siblings on first failure.

parallelcancellation

Install

$
Terminal window
pnpm dlx shadcn@latest add https://archyn.net/r/all.json

Copy the files below into the target paths shown in each tab.

ts@lib/flow/all.ts
import { type Result } from "./core/result";
import {
type BehaviorRecord,
type ErrorUnion,
type OutputRecord,
type RecordContext,
type RecordInput,
abortChildren,
normalizedConcurrency,
runChild,
} from "./core/record";
import {
type Behavior,
type RuntimeContext,
type RuntimeError,
resultFromSignal,
thrown,
} from "./core/runtime";
import { err, ok } from "./core/result";
/**
* Runs keyed behaviors in parallel and fails fast on the first child failure.
*
* Successful output preserves the input object keys. When a child fails, active siblings are
* aborted and the first domain or runtime error is returned.
*
* @typeParam B - Behavior record type.
* @param behaviors - Keyed behaviors to run with the same input.
* @param options - Optional concurrency limit.
* @returns A behavior producing a keyed output record or the first child error.
*
* @example
* ```ts
* const loadPage = all({
* profile: loadProfile,
* settings: loadSettings,
* });
* ```
*/
export function all<const B extends BehaviorRecord>(
behaviors: B,
options?: { concurrency?: number },
): Behavior<RecordInput<B>, OutputRecord<B>, ErrorUnion<B>, RecordContext<B>> {
return async (input, context) => {
if (context.signal.aborted) {
return resultFromSignal(context.signal);
}
return runAll(behaviors, input, context, options?.concurrency);
};
}
function runAll<const B extends BehaviorRecord>(
behaviors: B,
input: RecordInput<B>,
context: RuntimeContext<RecordContext<B>>,
concurrency: number | undefined,
): Promise<Result<OutputRecord<B>, ErrorUnion<B> | RuntimeError>> {
const keys = Object.keys(behaviors) as Array<keyof B>;
const output = {} as OutputRecord<B>;
const childControllers: AbortController[] = [];
if (keys.length === 0) {
return Promise.resolve(ok(output));
}
return new Promise((resolve) => {
let nextIndex = 0;
let active = 0;
let completed = 0;
let settled = false;
const limit = normalizedConcurrency(concurrency, keys.length);
const finish = (result: Result<OutputRecord<B>, ErrorUnion<B> | RuntimeError>) => {
if (settled) {
return;
}
settled = true;
context.signal.removeEventListener("abort", onAbort);
abortChildren(childControllers, result.ok ? undefined : result.error);
resolve(result);
};
const onAbort = () => finish(resultFromSignal(context.signal));
context.signal.addEventListener("abort", onAbort, { once: true });
const startNext = () => {
if (settled) {
return;
}
while (active < limit && nextIndex < keys.length) {
const key = keys[nextIndex];
if (key === undefined) {
break;
}
const behavior = behaviors[key];
if (behavior === undefined) {
break;
}
nextIndex += 1;
active += 1;
runChild(
behavior as unknown as Behavior<RecordInput<B>, unknown, unknown, RecordContext<B>>,
input,
context,
childControllers,
)
.then((result) => {
active -= 1;
if (settled) {
return;
}
if (!result.ok) {
finish(err(result.error as ErrorUnion<B> | RuntimeError));
return;
}
output[key] = result.value as OutputRecord<B>[typeof key];
completed += 1;
if (completed === keys.length) {
finish(ok(output));
return;
}
startNext();
})
.catch((error: unknown) => {
active -= 1;
finish(thrown(error));
});
}
};
startNext();
});
}

Update the import paths to match your project setup.

Run keyed Behavior values concurrently and fail fast on the first failed child.

Usage

all is for fan-out workflows where every child must succeed. It preserves object keys in the output type and cancels active siblings when one child fails.

Use it when:

  • Load independent resources for one screen or operation.
  • Preserve named outputs instead of array positions.
  • Stop unnecessary work as soon as one required child fails.
import { all } from "@/lib/flow/all";
import { task } from "@/lib/flow/task";
const loadDashboard = all(
{
profile: task(() => fetch("/api/profile").then((r) => r.json())),
alerts: task(() => fetch("/api/alerts").then((r) => r.json())),
},
{ concurrency: 2 },
);

Behavior

all starts children up to the concurrency limit. It completes only after every child succeeds, or stops early when one child fails.

The first domain or runtime failure is returned. Active sibling controllers are aborted with the failure reason. Parent cancellation aborts children and returns Cancelled.

Types

Output preserves behavior keys and domain errors are unioned.

Import from @/lib/flow/all.

Composition Notes

Use all for independent work where one failure makes the combined result unusable.

Do not use all when every child result is needed after failures; use allSettled.

Source

ts@lib/flow/all.ts
import { type Result } from "./core/result";
import {
type BehaviorRecord,
type ErrorUnion,
type OutputRecord,
type RecordContext,
type RecordInput,
abortChildren,
normalizedConcurrency,
runChild,
} from "./core/record";
import {
type Behavior,
type RuntimeContext,
type RuntimeError,
resultFromSignal,
thrown,
} from "./core/runtime";
import { err, ok } from "./core/result";
/**
* Runs keyed behaviors in parallel and fails fast on the first child failure.
*
* Successful output preserves the input object keys. When a child fails, active siblings are
* aborted and the first domain or runtime error is returned.
*
* @typeParam B - Behavior record type.
* @param behaviors - Keyed behaviors to run with the same input.
* @param options - Optional concurrency limit.
* @returns A behavior producing a keyed output record or the first child error.
*
* @example
* ```ts
* const loadPage = all({
* profile: loadProfile,
* settings: loadSettings,
* });
* ```
*/
export function all<const B extends BehaviorRecord>(
behaviors: B,
options?: { concurrency?: number },
): Behavior<RecordInput<B>, OutputRecord<B>, ErrorUnion<B>, RecordContext<B>> {
return async (input, context) => {
if (context.signal.aborted) {
return resultFromSignal(context.signal);
}
return runAll(behaviors, input, context, options?.concurrency);
};
}
function runAll<const B extends BehaviorRecord>(
behaviors: B,
input: RecordInput<B>,
context: RuntimeContext<RecordContext<B>>,
concurrency: number | undefined,
): Promise<Result<OutputRecord<B>, ErrorUnion<B> | RuntimeError>> {
const keys = Object.keys(behaviors) as Array<keyof B>;
const output = {} as OutputRecord<B>;
const childControllers: AbortController[] = [];
if (keys.length === 0) {
return Promise.resolve(ok(output));
}
return new Promise((resolve) => {
let nextIndex = 0;
let active = 0;
let completed = 0;
let settled = false;
const limit = normalizedConcurrency(concurrency, keys.length);
const finish = (result: Result<OutputRecord<B>, ErrorUnion<B> | RuntimeError>) => {
if (settled) {
return;
}
settled = true;
context.signal.removeEventListener("abort", onAbort);
abortChildren(childControllers, result.ok ? undefined : result.error);
resolve(result);
};
const onAbort = () => finish(resultFromSignal(context.signal));
context.signal.addEventListener("abort", onAbort, { once: true });
const startNext = () => {
if (settled) {
return;
}
while (active < limit && nextIndex < keys.length) {
const key = keys[nextIndex];
if (key === undefined) {
break;
}
const behavior = behaviors[key];
if (behavior === undefined) {
break;
}
nextIndex += 1;
active += 1;
runChild(
behavior as unknown as Behavior<RecordInput<B>, unknown, unknown, RecordContext<B>>,
input,
context,
childControllers,
)
.then((result) => {
active -= 1;
if (settled) {
return;
}
if (!result.ok) {
finish(err(result.error as ErrorUnion<B> | RuntimeError));
return;
}
output[key] = result.value as OutputRecord<B>[typeof key];
completed += 1;
if (completed === keys.length) {
finish(ok(output));
return;
}
startNext();
})
.catch((error: unknown) => {
active -= 1;
finish(thrown(error));
});
}
};
startNext();
});
}