Node.js worker at concurrency 1: batch enqueue, paced retries, idempotent jobs

A single-flight queue worker in Node 22 that respects a downstream rate limit, enqueues in batches, and survives redelivery without doing the work twice.

Running one job at a time is the simplest way to hold a rate limit, and it only works if redelivery can’t double-charge a customer. Infrai’s queue gives at-least-once delivery, which is the honest guarantee every managed queue offers, so the worker below pairs a single-flight loop with a dedupe table keyed on message_id. Enqueueing happens in batches; the actual work happens one call at a time.

Three separate mechanisms, and it’s worth being clear about which does what. Pacing is the loop. Deduplication at enqueue is idempotency_key. Deduplication of the effect is your own database.

Concurrency 1 is a budget, not a bottleneck

A worker that handles one message at a time and sleeps 200ms between them is pinned at 5 jobs per second and cannot exceed it, whatever the backlog does. No shared token bucket, no coordination between instances, no surprise burst when an autoscaler decides to double your pods. If the queue holds 40,000 jobs, that’s about two hours of steady draining — predictable in a way that a concurrency-10 worker with a limiter never quite is.

The cost is throughput, obviously. One slow job blocks the rest.

Batch the enqueue, not the work

Publishing accepts up to a batch per call, and each element can set its own delay, so a nightly fan-out that would have been 500 round trips becomes 50:

import process from "node:process";

const KEY = process.env.INFRAI_API_KEY;
if (!KEY) throw new Error("INFRAI_API_KEY is missing from the environment");

const headers = { Authorization: `Bearer ${KEY}`, "Content-Type": "application/json" };

export async function enqueueAll(accountIds, runId) {
  const chunks = [];
  for (let i = 0; i < accountIds.length; i += 10) chunks.push(accountIds.slice(i, i + 10));

  let queued = 0;
  for (const [chunkIndex, chunk] of chunks.entries()) {
    const batch = {
      queue: "billing-sync",
      messages: chunk.map((accountId, offset) => ({
        payload: { account_id: accountId, run_id: runId },
        idempotency_key: `${runId}:${accountId}`,
        delay_seconds: chunkIndex * 2 + offset,
      })),
    };

    const res = await fetch("https://api.infrai.cc/v1/queue/publish_batch", {
      method: "POST",
      headers,
      body: JSON.stringify(batch),
    });
    const out = await res.json();
    if (!out.ok) throw new Error(`publish_batch: ${out.error.code} ${out.error.message}`);
    queued += out.data.items.length;
  }
  return queued;
}

Spreading delay_seconds across the batch spaces the work out before a worker ever sees it — the queue becomes the scheduler and your loop just drains whatever is due.

One honest caveat we confirmed by testing on 2026-07-26: idempotency_key deduplicates on the single-message route (POST /v1/queue/publish returns the same message_id for a repeat, and the queue depth stays at one), but the same key inside a batch produced a second, distinct message. So a retried batch enqueue can double-publish. Either enqueue singly where duplicates are unacceptable, or accept it and let the consumer-side guard catch it.

{
  "ok": true,
  "data": {
    "items": [
      {
        "message_id": "qmsg_HuXwbvgQ1Im8PEJJ0avwaVOc",
        "queue": "billing-sync",
        "payload": { "account_id": "acct_11", "run_id": "2026-07-26T02:00Z" },
        "status": "available",
        "delivery_count": 0,
        "available_at": "2026-07-26T00:41:57Z",
        "priority": 0
      }
    ],
    "next_cursor": null
  }
}

The worker

The dedupe table first, because the loop depends on it:

CREATE TABLE IF NOT EXISTS processed_messages (
  message_id TEXT PRIMARY KEY,
  seen_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS processed_messages_seen_at_idx
  ON processed_messages (seen_at);

Prune it on the same cadence as message retention — 14 days is the queue’s default, so anything older than that can’t come back to haunt you.

import process from "node:process";
import pg from "pg";

const KEY = process.env.INFRAI_API_KEY;
const QUEUE = "billing-sync";
const MIN_GAP_MS = 200; // 5 jobs/second, hard ceiling
const db = new pg.Pool({ connectionString: process.env.DATABASE_URL });
const wait = (ms) => new Promise((done) => setTimeout(done, ms));

async function queueCall(path, payload) {
  const res = await fetch(`https://api.infrai.cc${path}`, {
    method: "POST",
    headers: { Authorization: `Bearer ${KEY}`, "Content-Type": "application/json" },
    body: JSON.stringify(payload),
  });
  const parsed = await res.json();
  if (!parsed.ok) throw new Error(`${path} failed: ${parsed.error.code}`);
  return parsed.data;
}

async function claim(messageId) {
  const { rowCount } = await db.query(
    "INSERT INTO processed_messages (message_id, seen_at) VALUES ($1, now()) ON CONFLICT DO NOTHING",
    [messageId],
  );
  return rowCount === 1;
}

async function runOnce() {
  const { items } = await queueCall("/v1/queue/consume", { queue: QUEUE, max_messages: 10 });
  for (const message of items) {
    const first = await claim(message.message_id);
    if (!first) {
      await queueCall("/v1/queue/ack", { queue: QUEUE, message_id: message.message_id });
      continue;
    }

    try {
      await syncAccount(message.payload.account_id);
      await queueCall("/v1/queue/ack", { queue: QUEUE, message_id: message.message_id });
    } catch (err) {
      console.error(`job ${message.message_id} failed on delivery ${message.delivery_count}:`, err.message);
      await db.query("DELETE FROM processed_messages WHERE message_id = $1", [message.message_id]);
      await queueCall("/v1/queue/nack", { queue: QUEUE, message_id: message.message_id, requeue: true });
    }
    await wait(MIN_GAP_MS);
  }
  return items.length;
}

async function syncAccount(accountId) {
  const target = "https://billing.example.com/v1/accounts/" + accountId + "/sync";
  const res = await fetch(target, { method: "POST" });
  if (!res.ok) throw new Error(`upstream ${res.status}`);
}

while (true) {
  const handled = await runOnce();
  if (handled === 0) await wait(2000);
}

The claim insert is the load-bearing line. Because delivery is at-least-once, the same message can arrive twice — a worker that dies mid-job, a lease that expires, a redrive from the dead-letter queue — and the unique constraint turns the second arrival into a free ack. Rolling the claim back inside the catch block is what keeps a genuine failure retryable.

Leases, and the error that tells you one expired

A consumed message stays invisible for the queue’s visibility timeout, 300 seconds by default. Exceed it and the message is handed to someone else; your late ack comes back as QUEUE_MESSAGE_NOT_IN_FLIGHT. If a single job routinely takes four minutes, raise the ceiling before you tune anything else:

export INFRAI_API_KEY="your_infrai_api_key"

curl -sS -X PATCH "https://api.infrai.cc/v1/queue/update/billing-sync" \
  -H "Authorization: Bearer ${INFRAI_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{"visibility_timeout_default":900,"message_retention_days":7}'

Two fields we watched change and one we didn’t: retention and the visibility timeout both took, while max_receive_count stayed pinned at 3 whatever we sent. Three failed deliveries dead-letter a message, and that number is not currently yours to move.

Where to put the guarantee

Skipping the third row is the common mistake, and it’s the only one that shows up as a customer complaint rather than a graph.

Cost, and what the pacing does to it

Publishing is the only billed call — $0.00002 per message, verified 2026-07-26 — while consume, ack, nack and stats are free and rate-limited instead of metered. That means a slow, paced drain costs exactly the same as a fast one; you’re paying for jobs created, not seconds spent. Forty thousand nightly syncs run about $0.80. New accounts start with $2 in credit. Rates trend downward and promotions happen, so pull the current figure before you build a spreadsheet on it:

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

When something else fits better

BullMQ’s limiter gives you per-group rate limiting and job classes in-process, and if you’re already running Redis it’s less machinery than an HTTP queue. Celery is the same argument in Python. SQS FIFO buys ordering guarantees this queue doesn’t offer — Infrai queues are standard-type and don’t support strict per-group ordering, so if two jobs for the same account must never overlap, you need that ordering somewhere else. Sidekiq remains the best answer inside a Rails monolith.

The reason to keep it here is the rest of the job. The same credential that drains this queue writes the sync report to storage, emails the finance team when it finishes, and files the stack trace when it doesn’t — one account, one bill, one usage query to attribute the run to a tenant.

References

• Infrai Message Queue API reference
• QUEUE_MESSAGE_NOT_IN_FLIGHT error reference
• BullMQ: rate limiting
• Celery: task idempotency and retries