子任务接受、取消与顺序提交

策略的具体实现如下。

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);
}