Child acceptance, cancellation, and ordered commit
The policy implementation follows.
ts
export async function runChildren<Value>(
parent: Pick<ModelRunner, 'createChild' | 'childLimits' | 'signal'>,
tasks: readonly ModelChildTask<Value>[],
options: ModelChildrenOptions<Value> = {},
): Promise<readonly Value[]> {
const concurrency = options.concurrency ?? parent.childLimits.maxConcurrentCalls;
breakWhile(
!Number.isSafeInteger(concurrency) ||
concurrency < 1 ||
concurrency > parent.childLimits.maxConcurrentCalls,
'Child group concurrency exceeds the parent policy',
);
const accept = options.accept?.bind(options),
commit = options.commit?.bind(options);
const definitions = tasks.map((task) => {
const { id, maxCalls, reserveCalls, timeoutMs, transports, diagnostics, signal, run } = task;
breakWhile(
typeof id !== 'string' || !id.trim() || typeof run !== 'function',
'Invalid child task',
);
return {
id,
maxCalls,
reserveCalls,
timeoutMs,
transports: transports === undefined ? undefined : [...transports],
diagnostics,
signal,
run: run.bind(task),
};
});
breakWhile(
new Set(definitions.map((task) => task.id)).size !== definitions.length,
'Duplicate child task id',
);
// One group controller so the first failure cancels every sibling worker at once.
const cancellation = new AbortController();
const signal = combineSignals(parent.signal, cancellation.signal, options.signal);
signal.throwIfAborted();
const values = new Array<Value>(definitions.length);
let next = 0,
failure: { error: unknown } | undefined;
const worker = async () => {
while (!failure && next < definitions.length) {
const index = next++,
definition = definitions[index];
try {
signal.throwIfAborted();
const child = parent.createChild({
...definition,
signal: combineSignals(signal, definition.signal),
});
const value = await awaitWithSignal(definition.run(child, child.signal), child.signal);
child.signal.throwIfAborted();
if (accept)
await awaitAssertion(
() => accept(value, definition.id),
child.signal,
'Child acceptance must return void',
);
child.signal.throwIfAborted();
// Index by declaration position so completion order never changes the result order.
values[index] = value;
} catch (error) {
if (!failure) {
failure = { error };
cancellation.abort(error);
}
}
}
};
await Promise.all(Array.from({ length: Math.min(concurrency, definitions.length) }, worker));
if (failure) throw failure.error;
signal.throwIfAborted();
if (commit)
for (let index = 0; index < definitions.length; index++) {
signal.throwIfAborted();
// Do not abandon an in-flight commit: cancellation never means it was rolled back.
const result = await commit(values[index], definitions[index].id);
signal.throwIfAborted();
assertion(result, 'Child commit must return void');
}
return Object.freeze(values);
}