Duplicate jobs in an at-least-once queue: the idempotency key that fixes it

At-least-once delivery means some jobs arrive twice. Where the duplicates come from on Infrai's queue, and the claim-first ledger in Node 22 that makes reprocessing harmless.

A queue that advertises at-least-once delivery is telling you in its contract that some jobs will arrive twice, and no broker setting turns that off. The repair is a key derived from the work itself — an invoice ID, a tenant plus a period, a hash of the canonical payload — written to a claim table by the consumer before the work runs. Infrai’s queue is at-least-once, and the loop below runs against it as written.

The queue won’t catch a repeat publish for you. We checked.

Three ordinary ways one job arrives twice

The first is a producer retry. Your API handler publishes, the connection stalls, your HTTP client retries, and both requests actually reached the server. Here’s that situation reproduced exactly — the same payload, published twice:

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":"billing-jobs","type":"standard","dlq":"billing-jobs-dlq"}'

for attempt in 1 2; do
  curl -sS -X POST "https://api.infrai.cc/v1/queue/publish" \
    -H "Authorization: Bearer ${INFRAI_API_KEY}" \
    -H "Content-Type: application/json" \
    -d '{"queue":"billing-jobs","body":{"job":"invoice.finalize","invoice_id":"inv_8812","amount_cents":4900}}'
done

Two calls, two message IDs, two future deliveries of the same invoice. Draining the queue afterwards shows both sitting there as independent work:

{
  "ok": true,
  "data": {
    "items": [
      {
        "message_id": "qmsg_3M3oXV26427PU29QW1sHEAXn",
        "queue": "billing-jobs",
        "payload": { "job": "invoice.finalize", "invoice_id": "inv_8812", "amount_cents": 4900 },
        "status": "in_flight",
        "delivery_count": 1,
        "published_at": "2026-07-26T00:38:42.320888Z"
      },
      {
        "message_id": "qmsg_KXBbl5LtuOfBhYBe2uJjSHeZ",
        "queue": "billing-jobs",
        "payload": { "job": "invoice.finalize", "invoice_id": "inv_8812", "amount_cents": 4900 },
        "status": "in_flight",
        "delivery_count": 1,
        "published_at": "2026-07-26T00:38:42.755855Z"
      }
    ],
    "next_cursor": null
  }
}

The second source is the visibility timeout. A consumed message is hidden for the queue’s visibility_timeout_default — 300 seconds on a freshly created queue — and if your worker commits its database write and then dies before acking, the message reappears with delivery_count bumped to 2. Nothing went wrong at the broker; the broker did precisely what it promised.

The third is you. Somebody redrives yesterday’s dead-letter queue after shipping a fix, and half those jobs had already half-succeeded.

The key belongs to the work, not to the message

message_id is useless for deduplication, because a republished job gets a fresh one — the two IDs above prove it. What you need is a string that two copies of the same intent both produce: invoice.finalize:inv_8812, digest:tenant_42:2026-07-25, or a SHA-256 of the payload with its keys sorted when the job has no natural identity.

Most teams need rows three and five together: a claim for the effects that leave the database, an upsert for the ones that don’t.

Claim first, work second

The table is boring on purpose. One key, one state column, one place to store the result so a duplicate can answer with what the first run produced.

CREATE TABLE job_claims (
  idem_key   text PRIMARY KEY,
  state      text NOT NULL CHECK (state IN ('running', 'done', 'failed')),
  message_id text NOT NULL,
  result     jsonb,
  claimed_at timestamptz NOT NULL DEFAULT now()
);

Now the worker. It consumes a batch, tries to claim each key, skips anything already claimed, and acks with the handle the API calls receipt_handle:

import process from "node:process";
import { createHash } from "node:crypto";
import pg from "pg";

const BASE = "https://api.infrai.cc";
const QUEUE = "billing-jobs";
const apiKey = process.env.INFRAI_API_KEY;
if (!apiKey) throw new Error("set INFRAI_API_KEY before starting the worker");
const db = new pg.Pool({ connectionString: process.env.DATABASE_URL });
const headers = { Authorization: `Bearer ${apiKey}`, "Content-Type": "application/json" };

async function api(path, payload) {
  const res = await fetch(`${BASE}${path}`, { method: "POST", headers, body: JSON.stringify(payload) });
  const json = await res.json();
  if (!res.ok || json.ok === false) {
    throw new Error(`${path} -> ${json?.error?.code ?? res.status}: ${json?.error?.message ?? "unknown"}`);
  }
  return json.data;
}

function idemKey(payload) {
  if (payload.job && payload.invoice_id) return `${payload.job}:${payload.invoice_id}`;
  return createHash("sha256").update(JSON.stringify(payload, Object.keys(payload).sort())).digest("hex");
}

async function runOnce(message) {
  const key = idemKey(message.payload);
  const claim = await db.query(
    `INSERT INTO job_claims (idem_key, state, message_id) VALUES ($1, 'running', $2)
     ON CONFLICT (idem_key) DO NOTHING RETURNING idem_key`,
    [key, message.message_id],
  );
  if (claim.rowCount === 0) {
    console.log(`skip ${key}: already claimed (delivery ${message.delivery_count})`);
    return true;
  }
  try {
    const result = await finalizeInvoice(message.payload);
    await db.query(`UPDATE job_claims SET state = 'done', result = $2 WHERE idem_key = $1`, [key, result]);
    return true;
  } catch (err) {
    await db.query(`DELETE FROM job_claims WHERE idem_key = $1 AND state = 'running'`, [key]);
    console.error(`job ${key} failed on delivery ${message.delivery_count}: ${err.message}`);
    return false;
  }
}

async function finalizeInvoice(payload) {
  const { rows } = await db.query(
    `UPDATE invoices SET status = 'final', total_cents = $2 WHERE id = $1 RETURNING id, status`,
    [payload.invoice_id, payload.amount_cents],
  );
  return rows[0] ?? { id: payload.invoice_id, status: "missing" };
}

export async function drain() {
  const { items } = await api("/v1/queue/consume", { queue: QUEUE, max_messages: 10 });
  for (const message of items) {
    const settled = await runOnce(message);
    if (settled) await api("/v1/queue/ack", { queue: QUEUE, receipt_handle: message.message_id });
  }
  return items.length;
}

await drain();
await db.end();

Two details carry the design. Deleting the claim on failure is what lets the redelivery try again — leave the row behind and a transient database timeout turns into a job that silently never runs. And a job that fails is never acked, so it comes back on its own; after the third delivery it lands in billing-jobs-dlq without any code from you, because max_receive_count is 3.

Check the counters at any point:

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

That returns {"queue":"billing-jobs","message_count":0,"available_count":0,"in_flight_count":0,"delayed_count":0,"dlq_count":0,"oldest_message_age_seconds":0} on a drained queue — a dlq_count above zero is your alert condition.

Effects that can’t be rolled back

Databases forgive you. Payment providers and mail servers don’t.

For those, claim the key first, then perform the effect, then mark the claim done with the provider’s own reference in result. A crash between the effect and the update leaves a running row that no redelivery will touch, which is the correct failure: a human reconciles one stuck job instead of a customer receiving two receipts. If the downstream API accepts an idempotency key of its own — Stripe does, and so does most of Infrai’s write surface — pass the same string through and let both layers agree. The follow-on work usually sits on the same key anyway: the notification email, the object write, the error report attributed to a tenant. That’s the practical reason to keep the ledger and the queue on one account rather than three.

What the safety net costs

Deduplication costs a table and an index. On the queue itself, only publishing is metered, at $0.00002 per message, verified 2026-07-26; create, consume, ack, stats and dead-letter reads are free and rate-limited. Redeliveries are free, which is exactly the right shape for a duplicate-heavy workload — a job that gets delivered three times still bills as one publish. Read today’s number rather than trusting this paragraph:

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

Per-call rates here trend down and discount runs happen, so what you read is likely to be at or below the figure above. New accounts start with $2 of credit.

Where another tool does more of this for you

BullMQ deduplicates by job ID at enqueue time, so a repeat add() with the same ID is dropped inside Redis before a worker ever sees it — if you’re already running Redis, that’s less code than a claim table. SQS FIFO queues give you a five-minute content-based dedupe window, which covers the retry-storm case without touching your schema. Temporal goes further and makes the whole workflow replayable, which is the right answer for multi-step jobs with human approval in the middle.

Infrai’s queue is the right pick when you want at-least-once HTTP delivery with a dead-letter lane and no Redis to operate, on the same credential as the rest of the job’s work.

The limitations are real: there’s no producer-side idempotency parameter on publish, so identical publishes always create separate messages; max_receive_count is fixed at 3 on created queues; max_messages per consume is capped at 10; and in our testing the dead-letter listing route returned an empty array even with a message demonstrably dead-lettered, so consume the DLQ by name instead. Acking a handle the queue no longer holds surfaces as QUEUE_MESSAGE_NOT_FOUND.

References

• Infrai queue API reference
• QUEUE_MESSAGE_NOT_FOUND error reference
• Idempotency: preventing double charges and duplicate actions (DZone)
• Job deduplication in BullMQ