複数実行でリクエストの制限を共有する

複数の実行や再試行（retry）をまたいでリクエストの開始間隔を制限するには、1つのリミッタを作成し、map のコールバック内で limiter.run を呼び出します。

import { from, map } from "mogami-stream";

export async function collectWithSharedRetries<T, U>(
  first: readonly T[],
  second: readonly T[],
  processJob: (job: T, signal: AbortSignal) => Promise<U>,
  isTransient: (error: unknown) => boolean,
) {
  // Both executions and every retry below use this same instance.
  const limiter = createSharedLimiter(100, 8);
  const execute = (jobs: readonly T[]) =>
    from(jobs)
      .pipe(
        map(
          (job, { signal }) =>
            limiter.run(signal, () => processJob(job, signal)),
          {
            concurrency: 2,
            capacity: 2,
            retry: {
              retries: 2,
              shouldRetry: (error) =>
                !(error instanceof LimiterQueueFullError) && isTransient(error),
            },
          },
        ),
      )
      .toArray();
  return Promise.all([execute(first), execute(second)]);
}

この2つの実行では開始間隔を最低 100 ms、待機中の呼び出しを最大8件に制限します。

processJob は呼び出されたら直ちに I/O を開始し、その I/O に signal を渡してください。リミッタは開始時刻の間隔を制御するものであり、先行する I/O の完了を待たずに次の呼び出しを許可します。各実行で同時に進行する項目数は、それぞれの map の concurrency で制限されます。

待機列とキャンセル

待機列は到着順（FIFO）です。再試行は末尾に追加され、他の実行より優先されることはありません。待機中にキャンセルされた呼び出しを取り除いても、残りの順序は保たれます。
maxWaiting は許可待ちの呼び出し数の上限であり、許可済みの1件や実行中の I/O は含みません。上限を超えると LimiterQueueFullError で失敗します（この例では再試行対象から除外しています）。0 を指定すると、即座に開始できる呼び出しだけを許可します。
バースト用の枠は蓄積されません。最初の呼び出しは直ちに開始できますが、空き時間やタイマーの遅延があった後でもまとめて開始することはありません。経過時間は performance.now() で計測し、タイマー発火時にも再確認します。
待機中にキャンセルされると signal.reason で reject し、abort リスナーを解除して待機枠を解放します（許可枠は消費しません）。許可が下りた後も処理関数を呼ぶ直前にキャンセルを再確認しますが、いったん消費した許可枠はキャンセルや処理失敗時にも返却されません。
待機列が空になるとタイマーは残らず、許可済みの呼び出しからは待機用の abort リスナーも解除されます。開始後の I/O の中断は渡された signal の処理に従います。

リミッタの実装

// Application recipe: not an export of mogami-stream.
export class LimiterQueueFullError extends Error {
  constructor() {
    super("Shared limiter waiting queue is full");
    this.name = "LimiterQueueFullError";
  }
}

export function createSharedLimiter(intervalMs: number, maxWaiting: number) {
  if (!Number.isFinite(intervalMs) || intervalMs <= 0)
    throw new RangeError("intervalMs must be finite and positive");
  if (!Number.isSafeInteger(maxWaiting) || maxWaiting < 0)
    throw new RangeError("maxWaiting must be a non-negative safe integer");

  type Waiter = { grant(): void };
  const waiting: Waiter[] = [];
  let nextStart = -Infinity;
  let starting = false; // A granted caller has not resumed yet.
  let timer: ReturnType<typeof setTimeout> | undefined;

  function drain() {
    if (timer !== undefined) clearTimeout(timer);
    timer = undefined;
    if (starting || waiting.length === 0) return;
    const remaining = nextStart - performance.now();
    if (remaining > 0) {
      // Long waits must not overflow the host's signed 32-bit timer delay.
      timer = setTimeout(drain, Math.min(remaining, 2_147_483_647));
      return;
    }
    starting = true;
    nextStart = performance.now() + intervalMs;
    waiting.shift()!.grant();
  }

  function acquire(signal: AbortSignal): Promise<void> {
    return new Promise((resolve, reject) => {
      signal.throwIfAborted();
      const canStart =
        !starting && waiting.length === 0 && performance.now() >= nextStart;
      if (!canStart && waiting.length >= maxWaiting)
        throw new LimiterQueueFullError();
      const abort = () => {
        waiting.splice(waiting.indexOf(waiter), 1);
        signal.removeEventListener("abort", abort);
        reject(signal.reason);
        drain();
      };
      const waiter: Waiter = {
        grant() {
          signal.removeEventListener("abort", abort);
          resolve();
        },
      };
      signal.addEventListener("abort", abort, { once: true });
      waiting.push(waiter);
      drain();
    });
  }

  return {
    async run<T>(signal: AbortSignal, operation: () => Promise<T>): Promise<T> {
      await acquire(signal);
      try {
        signal.throwIfAborted(); // Cancellation may occur after the grant.
        nextStart = performance.now() + intervalMs;
        return operation(); // Start I/O immediately; do not await other work first.
      } finally {
        // Release admission, not the operation: I/O may overlap.
        starting = false;
        drain();
      }
    },
  };
}

intervalMs には正の有限値、maxWaiting には 0 以上の安全な整数を指定します。状態はメモリ上で同じリミッタのインスタンスを参照する呼び出し間でのみ共有されるため、別プロセスや Worker スレッド間で制限を共有する場合は外部の仕組みを併用してください。

外部サービスが呼び出し全体で一定の開始間隔を求める場合はこのリミッタを使い、単一実行の上流読み取りペースだけを制限する場合は rateLimit を使ってください。再試行の設定はリトライを参照してください。

複数実行でリクエストの制限を共有する ​

待機列とキャンセル ​

リミッタの実装 ​

複数実行でリクエストの制限を共有する

待機列とキャンセル

リミッタの実装