A token bucket in the queue worker: Node example with 429 penalties

Bursty traffic plus a 5-per-second vendor cap needs a bucket, not a sleep. A refill-on-read TokenBucket, an Express endpoint that enqueues, and the retry path.

Put the bucket in the worker, not in the request path. An Express route that calls a 5-requests-per-second vendor API inline will either block your event loop waiting for a token or return an error to a user who did nothing wrong; the same route that publishes to a queue returns in a millisecond and lets a background consumer spend the budget properly. Infrai’s queue gives you the buffer and the retry semantics — the bucket itself is about forty lines you own, because there’s no server-side rate limiter to configure.

Forty lines is the whole feature. Here they are.

Why a bucket and not a window

Real traffic arrives in clumps. A fixed-window counter that permits 300 calls per minute will happily let all 300 through in the first two seconds and then stall for fifty-eight, which is exactly the burst most vendors are limiting against.

Token bucket wins because it encodes both rules a vendor actually publishes: a sustained rate and a burst allowance.

The bucket

No timers, no interval, no background refill. Compute how many tokens should have accumulated when someone asks — that’s the whole trick, and it makes the class safe to construct per worker process.

export class TokenBucket {
  constructor({ ratePerSecond, burst }) {
    this.rate = ratePerSecond;
    this.capacity = burst ?? ratePerSecond;
    this.tokens = this.capacity;
    this.updatedAt = Date.now();
  }

  #refill() {
    const now = Date.now();
    const gained = ((now - this.updatedAt) / 1000) * this.rate;
    this.tokens = Math.min(this.capacity, this.tokens + gained);
    this.updatedAt = now;
  }

  /** Milliseconds to wait before one token is available; 0 if it already is. */
  delayMs() {
    this.#refill();
    if (this.tokens >= 1) return 0;
    return Math.ceil(((1 - this.tokens) / this.rate) * 1000);
  }

  take() {
    this.#refill();
    if (this.tokens < 1) return false;
    this.tokens -= 1;
    return true;
  }

  /** Vendor said 429: throw the budget away so we stop pushing immediately. */
  penalise(seconds) {
    this.#refill();
    this.tokens = -this.rate * seconds;
    this.updatedAt = Date.now();
  }
}

penalise is the part most examples leave out. When a vendor returns 429 despite your pacing, you were wrong about the limit — draining the bucket into negative territory buys back the time without a sleep that blocks every other message in the batch.

The endpoint enqueues and gets out of the way

import express from "express";
import process from "node:process";

const BASE = "https://api.infrai.cc";
const key = process.env.INFRAI_API_KEY;
if (!key) throw new Error("INFRAI_API_KEY is not set");

const app = express();

app.post("/sync/:accountId", express.json(), async (req, res) => {
  const r = await fetch(`${BASE}/v1/queue/publish`, {
    method: "POST",
    headers: { Authorization: `Bearer ${key}`, "Content-Type": "application/json" },
    body: JSON.stringify({
      queue: "outbound-api-calls",
      body: { account_id: req.params.accountId, action: "sync", requested_at: Date.now() },
    }),
  });
  const out = await r.json();
  if (!out.ok) {
    console.error(`enqueue failed: ${out.error.code}`);
    return res.status(503).json({ error: "could not queue the sync" });
  }
  res.status(202).json({ accepted: true, message_id: out.data.message_id });
});

app.listen(3000, () => console.log("api listening on :3000"));

202 with a message id is the honest status. The caller knows the work is durable and knows nothing has happened yet.

Create the queue before any of that runs:

export INFRAI_API_KEY="your_infrai_api_key"

curl -sS -X POST "https://api.infrai.cc/v1/queue/create" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{"name":"outbound-api-calls","type":"standard","dlq":"outbound-api-calls.dlq"}'

You can confirm a publish by hand with a concrete call:

curl -sS -X POST "https://api.infrai.cc/v1/queue/publish" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{"queue":"outbound-api-calls","body":{"account_id":"acct_314","action":"sync"}}'

{
  "ok": true,
  "data": {
    "message_id": "qmsg_odPGgSXvopGcK7nviCWqzrWV",
    "queue": "outbound-api-calls",
    "payload": { "account_id": "acct_314", "action": "sync" },
    "status": "available",
    "delivery_count": 0,
    "published_at": "2026-07-26T00:31:05.392215Z"
  }
}

The worker spends the budget

import process from "node:process";
import { setTimeout as sleep } from "node:timers/promises";
import { TokenBucket } from "./token-bucket.mjs";

const BASE = "https://api.infrai.cc";
const headers = {
  Authorization: `Bearer ${process.env.INFRAI_API_KEY}`,
  "Content-Type": "application/json",
};
const bucket = new TokenBucket({ ratePerSecond: 5, burst: 20 });

async function call(path, payload) {
  const res = await fetch(`${BASE}${path}`, { method: "POST", headers, body: JSON.stringify(payload) });
  const out = await res.json();
  if (!out.ok) throw new Error(`${path}: ${out.error.code} — ${out.error.message}`);
  return out.data;
}

export async function run(callVendor) {
  for (;;) {
    const { items } = await call("/v1/queue/consume", { queue: "outbound-api-calls", max_messages: 10 });
    if (!items.length) { await sleep(4000); continue; }

    for (const msg of items) {
      const wait = bucket.delayMs();
      if (wait > 0) await sleep(wait);
      if (!bucket.take()) continue;

      const vendorRes = await callVendor(msg.payload);
      if (vendorRes.status === 429) {
        const after = Number(vendorRes.headers.get("retry-after") ?? 5);
        bucket.penalise(after);
        console.warn(`vendor 429 on delivery ${msg.delivery_count}; bucket drained for ${after}s`);
        break;
      }
      if (!vendorRes.ok) { console.error(`vendor ${vendorRes.status}; leaving message queued`); continue; }
      await call("/v1/queue/ack", { queue: "outbound-api-calls", receipt_handle: msg.message_id });
    }
  }
}

Messages that aren’t acked come back. That’s the delayed retry — no setTimeout, no retry table, no scheduler. The interval is the queue’s visibility timeout (300 seconds by default), and after three deliveries the message goes to outbound-api-calls.dlq instead of cycling forever. If 300 seconds is too patient for you, POST /v1/queue/nack returns a message immediately, at the cost of burning a delivery.

The trade-off worth naming: this bucket lives in one process. Run four workers and you’re pacing at 4 × 5 requests per second unless you either divide the rate by the worker count or move the counter into shared storage. For a single consumer — which is what most of these jobs need — the in-process version is right, and it’s testable without a network.

Watching it work

curl -sS "https://api.infrai.cc/v1/queue/stats/outbound-api-calls" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}"

{
  "ok": true,
  "data": {
    "queue": "outbound-api-calls",
    "message_count": 340,
    "available_count": 330,
    "in_flight_count": 10,
    "dlq_count": 0,
    "oldest_message_age_seconds": 67
  }
}

If oldest_message_age_seconds climbs without bound, the arrival rate has overtaken the vendor’s ceiling and no algorithm fixes that — you need a higher limit or fewer requests.

Running cost

Publishing is the metered route at $0.00002 per message, verified 2026-07-26; consume, ack, nack, stats and dead-letter reads are free and rate-limited rather than billed. That asymmetry is the reason redelivery is a reasonable retry strategy — a message that bounces off three 429s costs the same $0.00002 as one that succeeds first time. 1,000,000 syncs a month is $20. New accounts get $2 of credit to start.

curl -sS "https://api.infrai.cc/v1/discovery" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  | jq '.capabilities[] | select(.id == "queue.publish") | .billing'

Prices in this category trend downwards and campaigns run, so today’s figure may be under what’s printed here. The rate isn’t the argument, though: the vendor response you’re about to store, the alert you’ll send when the dead-letter count moves and the error you’ll record when one of these throws are all reachable with the same key.

Where a different tool wins

If your stack already has Redis and Node workers, bullmq’s built-in limiter does this without the class above and coordinates across workers, which is genuinely better than a per-process bucket. Ruby shops get the same from sidekiq’s enterprise rate limiting. Both are the right answer if a queue is all you’re buying.

The limitations here, stated plainly: no server-side rate limiter, 10 messages per consume call, one retry interval rather than a backoff ladder, and on 2026-07-26 POST /v1/queue/dlq/redrive/{queue} returned an error for us — draining the .dlq queue with POST /v1/queue/consume and republishing worked, so that’s the recovery path we’d script.

References

• Infrai queue API reference
• QUEUE_NOT_FOUND error reference
• BullMQ rate limiting guide
• HackerNoon: the token bucket algorithm in Node.js