Express to worker: the three idempotency keys a background job needs

One key stops a double-tapped POST creating two jobs, one stops a retried publish, one stops the worker doing the work twice. Where each lives, with Postgres and Infrai's queue.

An Express route that accepts a request and hands the slow part to a worker has three separate places a duplicate can be born, and one idempotency key won’t cover all three. The caller can retry the HTTP request. Your publish to Infrai’s queue can time out after the broker already accepted it. And the queue itself is at-least-once, so the worker will occasionally see the same message twice even when everything upstream behaved.

Each needs its own key, and they’re keyed on different things.

The three layers

Skip the first and a user who taps “Export” twice gets two exports. Skip the third and a redelivered message charges a card twice. They are not substitutes.

The request ledger

CREATE TABLE api_requests (
  idempotency_key text PRIMARY KEY,
  route           text NOT NULL,
  request_hash    text NOT NULL,
  status_code     int,
  response_body   jsonb,
  message_id      text,
  created_at      timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE job_results (
  job_fact   text PRIMARY KEY,
  finished_at timestamptz NOT NULL DEFAULT now(),
  detail      jsonb
);

request_hash matters more than it looks. A caller that reuses a key with a different body is a bug on their side, and returning the first response silently would hide it — answer 409 instead.

The Express route

The route does three things: claim the key, publish, record. It never does the work.

import express from "express";
import { createHash, randomUUID } from "node:crypto";
import pg from "pg";

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

const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
const app = express();
app.use(express.json({ limit: "256kb" }));

const hash = (o) => createHash("sha256").update(JSON.stringify(o)).digest("hex");

async function publish(payload, idemKey) {
  const res = await fetch(`${BASE}/v1/queue/publish`, {
    method: "POST",
    headers: { authorization: `Bearer ${KEY}`, "content-type": "application/json" },
    body: JSON.stringify({ queue: "api-enqueue-jobs", payload: payload, idempotency_key: idemKey }),
  });
  const json = await res.json();
  if (!res.ok) throw new Error(`publish HTTP ${res.status}: ${JSON.stringify(json.error ?? json)}`);
  return json.data.message_id;
}

app.post("/api/exports", async (req, res) => {
  const idem = req.get("Idempotency-Key") ?? randomUUID();
  const bodyHash = hash(req.body);

  const prior = await pool.query("SELECT * FROM api_requests WHERE idempotency_key = $1", [idem]);
  if (prior.rowCount === 1) {
    const row = prior.rows[0];
    if (row.request_hash !== bodyHash) return res.status(409).json({ error: "key reused with a different body" });
    return res.status(row.status_code).json(row.response_body);
  }

  try {
    await pool.query(
      "INSERT INTO api_requests (idempotency_key, route, request_hash) VALUES ($1, $2, $3)",
      [idem, "POST /api/exports", bodyHash],
    );
  } catch {
    return res.status(409).json({ error: "request already in flight" });
  }

  try {
    const messageId = await publish({ ...req.body, job_fact: `export:${idem}` }, idem);
    const body = { status: "queued", message_id: messageId };
    await pool.query(
      "UPDATE api_requests SET status_code = 202, response_body = $2, message_id = $3 WHERE idempotency_key = $1",
      [idem, body, messageId],
    );
    res.status(202).json(body);
  } catch (err) {
    await pool.query("DELETE FROM api_requests WHERE idempotency_key = $1", [idem]);
    res.status(503).json({ error: String(err.message ?? err) });
  }
});

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

Deleting the ledger row on a failed publish is the part worth copying. Without it, a broker hiccup leaves a claimed key that will forever replay a response that was never produced — a poison entry that looks like success.

Two calls with the same header, one job:

curl -s -X POST http://localhost:3000/api/exports \
  -H "Idempotency-Key: exp_2026_07_26_acme" \
  -H "Content-Type: application/json" \
  -d '{"tenant":"acme","format":"csv"}'

curl -s -X POST http://localhost:3000/api/exports \
  -H "Idempotency-Key: exp_2026_07_26_acme" \
  -H "Content-Type: application/json" \
  -d '{"tenant":"acme","format":"csv"}'

What the publish key really does

idempotency_key on the publish call is a declared parameter, and it works — but the response won’t tell you it fired. Send the same key twice:

curl -s -X POST https://api.infrai.cc/v1/queue/publish \
  -H "Authorization: Bearer $INFRAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"queue":"api-enqueue-jobs","payload":{"job":"export","tenant":"acme"},"idempotency_key":"req_probe_abc123"}'

Run it twice and both responses carry the identical id:

{
  "ok": true,
  "data": {
    "message_id": "qmsg_qCMa6RhnsluHr3pOZjHL2Jqi",
    "queue": "api-enqueue-jobs",
    "payload": { "job": "export", "tenant": "acme" },
    "status": "available",
    "delivery_count": 0
  },
  "metadata": { "idempotent_replay": false, "cost_usd": 0.00002 }
}

One message lands on the queue — GET /v1/queue/stats/api-enqueue-jobs confirmed message_count: 1 after both calls in our testing. But idempotent_replay stayed false and the replay was still metered. The caveat, then: the repeated message_id is your only reliable signal that deduplication happened, so compare it against the one you stored rather than reading the metadata flag.

The worker’s own guard

The third key is derived from the work, not from the message. job_fact in the payload above is export:<idempotency-key>, and the worker inserts it before doing anything expensive.

import pg from "pg";

const BASE = "https://api.infrai.cc";
const KEY = process.env.INFRAI_API_KEY;
if (!KEY) throw new Error("set INFRAI_API_KEY (your_infrai_api_key)");
const QUEUE = "api-enqueue-jobs";
const H = { authorization: `Bearer ${KEY}`, "content-type": "application/json" };
const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });

async function post(path, body) {
  const r = await fetch(`${BASE}${path}`, { method: "POST", headers: H, body: JSON.stringify(body) });
  if (!r.ok) throw new Error(`${path} → HTTP ${r.status}`);
  return (await r.json()).data;
}

export async function tick() {
  const { items } = await post("/v1/queue/consume", { queue: QUEUE, max_messages: 10 });
  for (const msg of items) {
    const fact = msg.payload.job_fact;
    const claim = await pool.query(
      "INSERT INTO job_results (job_fact) VALUES ($1) ON CONFLICT DO NOTHING RETURNING job_fact",
      [fact],
    );
    if (claim.rowCount === 0) {
      await post("/v1/queue/ack", { queue: QUEUE, message_id: msg.message_id });
      continue;
    }
    try {
      await buildExport(msg.payload);
      const { acked } = await post("/v1/queue/ack", { queue: QUEUE, message_id: msg.message_id });
      if (!acked) console.warn("ack false, will redeliver:", msg.message_id);
    } catch (err) {
      await pool.query("DELETE FROM job_results WHERE job_fact = $1", [fact]);
      await post("/v1/queue/nack", { queue: QUEUE, message_id: msg.message_id, requeue: true });
      console.error(fact, String(err.message ?? err));
    }
  }
}

async function buildExport(payload) {
  console.log("building export for", payload.tenant);
}

max_messages above 10 is rejected with a 400, and max_receive_count is fixed at 3 — the fourth failure dead-letters the message into api-enqueue-jobs-dlq rather than looping. Also mind the payload: messages are capped at 256 KB, so put a row id in the message and leave the CSV in storage.

Watch it drain:

curl -s https://api.infrai.cc/v1/queue/stats/api-enqueue-jobs \
  -H "Authorization: Bearer $INFRAI_API_KEY"

Cost, and when to use something else

Consume, ack, nack and stats are free and rate-limited; publish is the billable call at $0.00002 per message, verified 2026-07-26, against $2 of free credit on a new account. A million enqueues is $20, and — as measured above — an idempotent replay is charged like a first publish, so budget on requests rather than on distinct jobs. Prices move down over time, so read them live:

curl -s "https://api.infrai.cc/v1/discovery" -H "Authorization: Bearer $INFRAI_API_KEY"
curl -s "https://api.infrai.cc/v1/account/balance" -H "Authorization: Bearer $INFRAI_API_KEY"

If you’re already running Redis next to your Express app, BullMQ does all of this in-process with job ids as the dedupe key, and its per-job cost is zero — that’s the honest comparison, and for a single-service app it’s a good one. SQS is the right answer when the rest of the stack is AWS and the IAM story matters more than the ergonomics. The reason to enqueue over HTTP instead is that the same credential also fetches the rows, stores the finished export and emails the link, so the follow-on work doesn’t need a fourth vendor. If background jobs are all you’ll ever need, stick with the library.

References

• Infrai queue API reference
• QUEUE_MESSAGE_TOO_LARGE — Infrai error reference
• Queues — BullMQ