Bottleneck rate limiting with concurrency, spacing and retries: working versions (tested 2026-10-15)

{
  "name": "bottleneck-rate-limiting",
  "private": true,
  "type": "module",
  "scripts": {
    "test": "node --import tsx --test test/*.test.ts"
  },
  "dependencies": {
    "bottleneck": "2.19.5"
  },
  "devDependencies": {
    "@types/node": "26.6.5",
    "tsx": "4.23.15"
  }
}

import Bottleneck from "bottleneck";

export function createLimiter(opts: { maxConcurrent?: number; minTime?: number; retries?: number; retryDelay?: number } = {}) {
  const limiter = new Bottleneck({ maxConcurrent: opts.maxConcurrent ?? 1, minTime: opts.minTime ?? 0 });
  const retries = opts.retries ?? 0;

  // Returning a number from "failed" tells Bottleneck to retry the job after that many ms.
  limiter.on("failed", (error: Error & { status?: number }, info) => {
    if (error.status === 429 && info.retryCount < retries) return opts.retryDelay ?? 5;
    return undefined;
  });

  return limiter;
}

export type Call = { at: number; id: string };

export async function callAll(limiter: Bottleneck, ids: string[], work: (id: string) => Promise<void>) {
  return Promise.all(ids.map((id) => limiter.schedule(() => work(id))));
}

import { test } from "node:test";
import assert from "node:assert/strict";
import { createLimiter, callAll } from "../src/limiter.ts";

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

test("never exceeds maxConcurrent", async () => {
  const limiter = createLimiter({ maxConcurrent: 2 });
  let active = 0, peak = 0;
  await callAll(limiter, ["a", "b", "c", "d", "e"], async () => {
    active++; peak = Math.max(peak, active);
    await sleep(15);
    active--;
  });
  assert.equal(peak, 2);
  await limiter.stop({ dropWaitingJobs: false });
});

test("minTime spaces job starts", async () => {
  const limiter = createLimiter({ maxConcurrent: 5, minTime: 40 });
  const starts: number[] = [];
  await callAll(limiter, ["a", "b", "c"], async () => { starts.push(Date.now()); });
  for (let i = 1; i < starts.length; i++) assert.ok(starts[i] - starts[i - 1] >= 30, `gap ${starts[i] - starts[i - 1]}ms`);
  await limiter.stop({ dropWaitingJobs: false });
});

test("higher priority (lower number) runs first", async () => {
  const limiter = createLimiter({ maxConcurrent: 1 });
  const order: string[] = [];
  const blocker = limiter.schedule(() => sleep(20));
  const jobs = [
    limiter.schedule({ priority: 9 }, async () => { order.push("low"); }),
    limiter.schedule({ priority: 1 }, async () => { order.push("high"); }),
    limiter.schedule({ priority: 5 }, async () => { order.push("mid"); }),
  ];
  await Promise.all([blocker, ...jobs]);
  assert.deepEqual(order, ["high", "mid", "low"]);
  await limiter.stop({ dropWaitingJobs: false });
});

test("retries 429 responses then succeeds; other errors fail fast", async () => {
  const limiter = createLimiter({ retries: 3, retryDelay: 5 });
  let calls = 0;
  const value = await limiter.schedule(async () => {
    calls++;
    if (calls < 3) throw Object.assign(new Error("slow down"), { status: 429 });
    return "ok";
  });
  assert.equal(value, "ok");
  assert.equal(calls, 3);
  await assert.rejects(limiter.schedule(async () => { throw Object.assign(new Error("bad"), { status: 500 }); }), /bad/);
  assert.deepEqual(limiter.counts(), { RECEIVED: 0, QUEUED: 0, RUNNING: 0, EXECUTING: 0 });
  await limiter.stop({ dropWaitingJobs: false });
});