Idempotent Consumers
Brokers deliver at least once, so a handler will run twice; make the second run have no extra effect.
const { rowCount } = await pool.query('INSERT INTO processed_jobs (job_id) VALUES ($1) ON CONFLICT DO NOTHING', [job.id])Brokers promise at-least-once delivery: every job arrives at least once, and sometimes more than once. That is not a defect they plan to fix. A worker takes a job, does the work, then acknowledges it, and that ack is what tells the broker to delete the job. A crash in the gap between the work and the ack leaves the job undeleted, so it comes back and the work runs again. SQS does the same on a clock: if no ack arrives within the visibility timeout, the period the job is hidden from other workers, it becomes visible again and another worker takes it.
Design for the second run
An idempotent handler has the same effect run twice as run once. Some work already does: setting a status to paid, or writing a file to a fixed name. Some never does: adding credits, sending an email, charging a card.
For those, give each job a stable id and record that you handled it. The usual line is an insert that quietly loses the race: INSERT INTO processed_jobs (job_id) VALUES ($1) ON CONFLICT DO NOTHING. A rowCount of 0 means another delivery got there first, so return without doing anything. When the marker and the effect live in the same database, put both in one transaction, so either both happen or neither does.
These two read the same: const { rowCount } = await pool.query(sql, [job.id]) is shorthand for const result = await pool.query(sql, [job.id]); const rowCount = result.rowCount.
For effects in someone else's system, pass the id onward. Most payment and mail APIs accept an idempotency key and return the first result rather than acting twice.
Example
When to use it
- Stop a redelivered payment job from charging a customer twice.
- Make a webhook handler safe when the sender retries on a timeout.
- Keep a double-clicked 'send invoice' button from queueing two emails.
- Let a worker crash mid-job knowing the retry will not duplicate the effect.
More examples
The marker and the effect in one transaction
The only arrangement with no window: the claim and the balance change commit as one unit, so a crash either loses both or keeps both. Note the single client rather than pool.query, since a transaction must run on one connection.
import { Pool } from 'pg';
const pool = new Pool({ connectionString: process.env.DATABASE_URL });
// processed_jobs has: job_id text PRIMARY KEY, handled_at timestamptz
export async function applyTopUp(job) {
const client = await pool.connect(); // one connection for the whole transaction
try {
await client.query('BEGIN');
const claim = await client.query(
'INSERT INTO processed_jobs (job_id) VALUES ($1) ON CONFLICT DO NOTHING',
[job.id]
);
if (claim.rowCount === 0) { // a previous delivery already did this
await client.query('ROLLBACK');
return 'duplicate';
}
await client.query('UPDATE accounts SET credits = credits + $1 WHERE id = $2', [
job.data.credits,
job.data.accountId,
]);
await client.query('COMMIT'); // marker and effect become visible together
return 'applied';
} catch (err) {
await client.query('ROLLBACK'); // nothing recorded, so a retry is safe
throw err;
} finally {
client.release();
}
}Deduplicating at the point of adding the job
Cheap protection against duplicates created on the producer side: two requests, one job. It is not a substitute for an idempotent handler, because the guarantee ends when the job leaves the queue. Use both.
// BullMQ ignores an add() whose jobId already exists in the queue, which
// turns a double-clicked button into one job.
await emails.add(
'send-receipt',
{ orderId: order.id },
{ jobId: `send-receipt:${order.id}` } // derived from the data, not random
);
// This only holds while that job is still in the queue. Once it has
// completed and been removed, the same id can be added again, so the
// worker still needs its own check for anything that must not repeat.Passing an idempotency key to an outside API
You cannot roll back someone else's system, so you make the second call a no-op instead. The timeout matters as much as the key: without it a stalled request holds the job past its visibility timeout and the broker hands a copy to another worker.
// The job id is stable across redeliveries, so send it as the key and the
// provider returns the FIRST result instead of charging again.
async function chargeOnce(job) {
const res = await fetch('https://api.example-payments.test/v1/charges', {
method: 'POST',
headers: {
authorization: `Bearer ${process.env.PAYMENTS_API_KEY}`,
'content-type': 'application/json',
'Idempotency-Key': job.id, // check your provider's exact header name
},
body: JSON.stringify({ amount: job.data.amount, currency: 'gbp' }),
signal: AbortSignal.timeout(10_000), // never hang a worker on a dead API
});
if (!res.ok) throw new Error(`charge failed: ${res.status}`); // let it retry
return res.json();
}
Discussion