Sharing a request budget across executions

To share a request budget across multiple executions and retry attempts, create a single limiter instance and call limiter.run inside map:

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

The two executions share a minimum 100 ms start interval and at most eight waiting callers.

processJob must start its I/O immediately and pass signal to that I/O. The limiter paces start times; it does not wait for in-flight I/O to finish before admitting the next call. Each execution's map concurrency still bounds its own in-flight items.

Queue and cancellation

• Waiting callers are admitted in FIFO order. A retry joins the tail of the queue without priority over other executions, and cancelled callers are removed without disturbing the remaining order.
• maxWaiting bounds callers awaiting admission (excluding the granted caller and running I/O). Exceeding it rejects with LimiterQueueFullError, which the example does not retry. Setting maxWaiting: 0 allows only immediate admission.
• No burst credit accumulates: the first call starts immediately, and idle periods or delayed timers never trigger catch-up bursts. Intervals are measured with performance.now() and verified when the timer fires.
• Cancellation while queued rejects with signal.reason, removes the abort listener, and frees the queue slot without consuming a permit. Cancellation is checked again just before invoking a granted operation; once granted, the permit remains consumed even if the caller aborts or the operation fails.
• No timer remains active once the queue empties, and granted callers have their queue abort listener removed before I/O begins.

Limiter implementation

// 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 must be finite and positive; maxWaiting must be a non-negative safe integer. State is shared only among callers holding the same limiter instance in memory; coordinating across separate processes or worker threads requires an external store.

Use this pattern when an external service enforces a minimum start interval across callers. For per-execution upstream read pacing without retries, use rateLimit. See Retries for retry configuration.