Batch enqueue in Node: 100 jobs per publish call, then a paced worker

Enqueue thousands of background jobs from Node with queue.publish_batch — chunking, the 100-message ceiling, DLQ wiring and a rate-limited consumer.

Publishing 4,300 import rows one HTTP call at a time is simply slow. Infrai’s queue takes up to 100 messages in a single POST /v1/queue/publish_batch, so a 4,300-row CSV import becomes 43 round trips instead of 4,300 — and the worker that drains them is paced separately from however fast you managed to enqueue. What follows is the whole loop in Node 22, against the live API, including the edges that surprised us.

Producing and consuming are opposite problems. Enqueueing wants throughput. Sending wants restraint, because the email provider on the other end has a rate limit and doesn’t care that your importer had a good afternoon.

One call carries 100 messages, and the 101st is an error

The request is {queue, messages[]}, where each element needs a payload object and may carry delay_seconds, priority, headers, and — on FIFO queues — message_group_id and deduplication_id. A top-level idempotency_key covers the whole batch, so a retried POST after a network timeout won’t double-enqueue.

curl -sS -X POST https://api.infrai.cc/v1/queue/publish_batch \
  -H "Authorization: Bearer ${INFRAI_API_KEY:-your_infrai_api_key}" \
  -H "Content-Type: application/json" \
  -d '{
    "queue": "csv-import-batch",
    "idempotency_key": "import-2026-07-26-chunk-0",
    "messages": [
      {"payload": {"row": 1, "email": "a@example.com"}},
      {"payload": {"row": 2, "email": "b@example.com"}, "delay_seconds": 30},
      {"payload": {"row": 3, "email": "c@example.com"}}
    ]
  }'

Every message comes back with its own id, so you can log the mapping from source row to message_id before the worker ever runs:

{
  "ok": true,
  "data": {
    "items": [
      {"message_id": "qmsg_FHXCzWU3pZ9JFDMC2ClFCHcB", "queue": "csv-import-batch", "status": "available", "delivery_count": 0, "available_at": "2026-07-26T01:28:30Z"},
      {"message_id": "qmsg_QwK472Qxw8vQ6bC8EUGD9eRR", "queue": "csv-import-batch", "status": "available", "delivery_count": 0, "available_at": "2026-07-26T01:29:00Z"}
    ],
    "next_cursor": null
  }
}

Send 101 and you get an HTTP 400 with publish_batch accepts at most 100 messages. Note the second item above: its status says available even though available_at is 30 seconds out. That echo is misleading, and the fix is in the drain section below.

The batch route won’t create the queue for you

Single POST /v1/queue/publish auto-creates a missing queue. publish_batch doesn’t — it answers 400 with queue '...' not found, which is a genuinely useful asymmetry once you know about it, because a typo in a batch producer would otherwise scatter 4,300 messages into a queue no worker is watching.

So declare the queue, and its dead-letter lane, before the import runs:

curl -sS -X POST https://api.infrai.cc/v1/queue/create \
  -H "Authorization: Bearer ${INFRAI_API_KEY:-your_infrai_api_key}" \
  -H "Content-Type: application/json" \
  -d '{"name": "csv-import-batch.dlq"}'

curl -sS -X POST https://api.infrai.cc/v1/queue/create \
  -H "Authorization: Bearer ${INFRAI_API_KEY:-your_infrai_api_key}" \
  -H "Content-Type: application/json" \
  -d '{"name": "csv-import-batch", "dead_letter_queue": "csv-import-batch.dlq", "max_retries": 3}'

Defaults you inherit if you skip that: 14-day retention, 256 KB per message, a 300-second visibility timeout, and three delivery attempts before dead-lettering. The retry count is fixed at 3 on standard queues — worth flagging if your runbook assumes you can dial it to 10.

Chunking a 4,300-row import

The producer’s only jobs are to slice, to stagger, and to retry its own HTTP failures. Staggering matters more than it looks: if all 4,300 messages become available in the same second, your worker’s first consume returns a full lease and everything downstream queues up behind the provider’s limit anyway. Spreading them with delay_seconds turns a spike into a ramp.

import { readFileSync } from "node:fs";
import { setTimeout as sleep } from "node:timers/promises";

const KEY = process.env.INFRAI_API_KEY;
if (!KEY) throw new Error("set INFRAI_API_KEY");
const QUEUE = "csv-import-batch";
const CHUNK = 100;
const PER_SECOND = 5; // what the mail provider allows us

const rows = readFileSync(new URL("./contacts.csv", import.meta.url), "utf8")
  .split("\n").slice(1).filter(Boolean)
  .map((line, i) => { const [email, name] = line.split(","); return { row: i + 1, email, name }; });

async function publishChunk(chunk, offset) {
  const messages = chunk.map((r, i) => ({
    payload: r,
    delay_seconds: Math.floor((offset + i) / PER_SECOND),
  }));
  for (let attempt = 0; attempt < 4; attempt++) {
    const res = await fetch("https://api.infrai.cc/v1/queue/publish_batch", {
      method: "POST",
      headers: { authorization: `Bearer ${KEY}`, "content-type": "application/json" },
      body: JSON.stringify({ queue: QUEUE, messages, idempotency_key: `import-2026-07-26-chunk-${offset}` }),
    });
    if (res.ok) return (await res.json()).data.items.length;
    if (res.status !== 429 && res.status < 500) throw new Error(`chunk ${offset}: ${res.status} ${await res.text()}`);
    await sleep(500 * 2 ** attempt);
  }
  throw new Error(`chunk ${offset}: still failing after 4 attempts`);
}

let enqueued = 0;
for (let i = 0; i < rows.length; i += CHUNK) {
  enqueued += await publishChunk(rows.slice(i, i + CHUNK), i);
}
console.log(`enqueued ${enqueued} of ${rows.length} rows`);

delay_seconds accepts 0 to 604800 — seven days — so a ramp of any realistic import length fits. Push past the ceiling and the call is rejected rather than silently clamped.

The consumer goes at the provider’s speed, not yours

POST /v1/queue/consume leases messages for the visibility timeout and hands back message_id plus payload. The schema advertises max_messages up to 100; in practice the live API caps a lease at 10, so build the loop around repeated small leases rather than one giant one.

import { setTimeout as sleep } from "node:timers/promises";

const KEY = process.env.INFRAI_API_KEY;
if (!KEY) throw new Error("set INFRAI_API_KEY");
const QUEUE = "csv-import-batch";
const GAP_MS = 200; // 5 sends per second

const call = async (path, body) => {
  const res = await fetch(`https://api.infrai.cc${path}`, {
    method: "POST",
    headers: { authorization: `Bearer ${KEY}`, "content-type": "application/json" },
    body: JSON.stringify(body),
  });
  if (!res.ok) throw new Error(`${path} -> ${res.status} ${await res.text()}`);
  return (await res.json()).data;
};

async function sendWelcome(contact) {
  // your provider call goes here; throw to trigger a nack
  if (!contact.email?.includes("@")) throw new Error(`bad address: ${contact.email}`);
}

for (;;) {
  const { items } = await call("/v1/queue/consume", { queue: QUEUE, max_messages: 10 });
  if (!items.length) { await sleep(2000); continue; }
  for (const msg of items) {
    try {
      await sendWelcome(msg.payload);
      await call("/v1/queue/ack", { queue: QUEUE, message_id: msg.message_id });
    } catch (err) {
      console.error(`row ${msg.payload.row} failed (attempt ${msg.delivery_count}):`, err.message);
      await call("/v1/queue/nack", { queue: QUEUE, message_id: msg.message_id, requeue: true });
    }
    await sleep(GAP_MS);
  }
}

One trap: ack on an id that isn’t leased returns HTTP 200 with {"acked": false}. Checking only the status code will tell you every message was confirmed when none were.

Watching a batch drain

curl -sS https://api.infrai.cc/v1/queue/stats/csv-import-batch \
  -H "Authorization: Bearer ${INFRAI_API_KEY:-your_infrai_api_key}"

That returns message_count, available_count, in_flight_count, delayed_count and dlq_count. delayed_count is the field that tells the truth about a staggered publish — the publish response’s status doesn’t. If dlq_count climbs, consume csv-import-batch.dlq by name; the dedicated DLQ listing route came back empty for us even with messages sitting in it.

What the fan-out costs

Publishing is the only billable step here. A single publish is $0.00002 per message (verified 2026-07-26); consume, ack, nack, create and stats are all free, rate-limited calls. publish_batch is advertised at $0.001 per call, but metering charges per message — a three-message batch reported cost_usd: 0.00006 — so batching buys you round trips and latency, not a discount. New accounts start with $2 free, which covers a lot of imports.

Read today’s numbers rather than trusting this paragraph:

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

Rates on this platform move downward and discount campaigns run, so what you read there may well be lower than what’s printed here. The structural argument is steadier: enqueue is metered, everything else in the loop isn’t, and the same key already reaches the mail send, the object store for the source CSV, and error capture for the failures — so the second question doesn’t need a second vendor.

Where BullMQ or SQS still wins

If your bottleneck is the worker’s own throttle, BullMQ’s limiter: { max, duration } is a real feature and reimplementing it in a for loop is a step backwards — stick with it, and pay for the Redis. If your producers and consumers already live inside one AWS account, SQS’s IAM integration is worth more than anything here. The trade-off you’re accepting with Infrai’s queue is that there’s no server-side worker limiter and no addBulk-style unbounded chunk; what you get back is no broker to run, a seven-day delay ceiling instead of fifteen minutes, and one credential for the queue plus whatever the job actually does.

References

• Infrai queue API reference
• Adding jobs in bulk — BullMQ
• QUEUE_NOT_FOUND error reference