09 - Idempotency in Consumers: Deduplication and Idempotency Key

Why the confirmation email goes out twice

The system works in development and in production, until a customer receives the confirmation email twice, or a card is charged twice. There is no bug to find: the brokers you use, starting with SQS, guarantee at-least-once delivery, and copies are expected. A consumer that gets copies wrong is a consumer written for a world that does not exist. The defence is idempotency: processing the same message once, twice or ten times must produce the same result as processing it only once.

There are three tools, and this article runs all of them: an idempotency key written together with the effect, an operation that is idempotent by nature, and the inbox pattern, which separates receiving from processing. For each tool I show where it holds and where it breaks (the window of a key with an expiry, an external effect that the transaction does not cover). The programs run in memory or on SQLite, without a broker: where a real broker changes things, I say so. The context is in the article on dead letter queues, which explains where many of the redelivered messages come from.

Where duplicates come from

The typical case: the consumer charges the payment, then dies before confirming the message to the broker. To the broker the message has not been processed, and it delivers it again. The second case is the timeout: in SQS a received message becomes invisible for the visibility timeout, 30 seconds by default according to the guide; if processing takes 40, the message becomes visible again and another consumer picks it up while the first is still working (the proof is in the article on SQS, SNS and EventBridge).

A third case is the producer that retries a send for which it received no confirmation. In none of the three cases is there an error to fix; the remedy lies in the consumer, as with Kafka consumers, covered in the dedicated series. How a Kafka consumer saves its position and what happens to a record that will not go through is in the article on the dead letter queue in Kafka; where the guarantee of transactions ends is explained in the article on exactly-once; records reread after a failover between clusters are the case in the production guide.

The key and the effect in the same transaction

The idea is simple: every message has an identifier, and before processing it the consumer checks whether it has already seen it. The delicate point is where you write the key. If you write it before the effect, a crash halfway leaves the key without the effect, and the retry is discarded as a duplicate. If you write it after, a crash between the two leaves the effect without the key and the retry repeats it.

The solution is to write key and effect in the same transaction: both or neither. The program does it with a processed_messages table and INSERT OR IGNORE, which on SQLite does the same thing as PostgreSQL's INSERT ... ON CONFLICT DO NOTHING (the PostgreSQL guide defines it as the alternative that simply avoids inserting the row).

import { DatabaseSync } from 'node:sqlite';

const db = new DatabaseSync(':memory:');
db.exec(`
  CREATE TABLE processed_messages (message_id TEXT PRIMARY KEY, consumer TEXT NOT NULL);
  CREATE TABLE inventory (product_id TEXT PRIMARY KEY, quantity INTEGER NOT NULL);
  CREATE TABLE orders (id TEXT PRIMARY KEY, payment_status TEXT NOT NULL);
  INSERT INTO inventory VALUES ('pen', 100);
  INSERT INTO orders VALUES ('o-1', 'pending');
`);
const quantity = () => (db.prepare("SELECT quantity FROM inventory WHERE product_id = 'pen'").get() as { quantity: number }).quantity;

// The consumer writes the key and the effect in the SAME transaction: both or neither.
function handle(messageId: string, handler: () => void): 'processed' | 'duplicate' {
  db.exec('BEGIN');
  try {
    const r = db.prepare('INSERT OR IGNORE INTO processed_messages VALUES (?, ?)').run(messageId, 'inventory');
    if (r.changes === 0) { db.exec('ROLLBACK'); return 'duplicate'; }
    handler();
    db.exec('COMMIT');
    return 'processed';
  } catch (e) {
    db.exec('ROLLBACK');                 // the key disappears together with the effect: the retry is possible
    throw e;
  }
}
const subtract = (n: number) => () => db.prepare("UPDATE inventory SET quantity = quantity - ? WHERE product_id = 'pen'").run(n);

// 1) Same message twice.
console.log('1) first delivery :', handle('msg-1', subtract(5)), '| quantity', quantity());
console.log('1) second delivery:', handle('msg-1', subtract(5)), '| quantity', quantity());

// 2) The handler fails halfway: no key, no effect, and the retry works.
try { handle('msg-2', () => { subtract(5)(); throw new Error('crash after the UPDATE'); }); } catch (e) { console.log('2) error:', (e as Error).message, '| quantity', quantity()); }
console.log('2) retry          :', handle('msg-2', subtract(5)), '| quantity', quantity());

// 3) Effect outside the database: the transaction does not cover it.
let charges = 0;
const gateway = { seen: new Set<string>(), charge(key?: string) { if (key && this.seen.has(key)) return; if (key) this.seen.add(key); charges++; } };
function payWithoutKey(id: string) {
  try { handle(id, () => { gateway.charge(); throw new Error('commit failed after the charge'); }); } catch { /* the broker redelivers */ }
  handle(id, () => gateway.charge());
}
payWithoutKey('pay-1');
console.log('3) without a key at the gateway: charges =', charges);
charges = 0;
function payWithKey(id: string) {
  try { handle(id, () => { gateway.charge(id); throw new Error('commit failed after the charge'); }); } catch { /* the broker redelivers */ }
  handle(id, () => gateway.charge(id));
}
payWithKey('pay-2');
console.log('3) with the key at the gateway: charges =', charges);

// 4) Natural idempotency: absolute state instead of a delta, a transition that does nothing the second time.
const markPaid = () => db.prepare("UPDATE orders SET payment_status = 'paid' WHERE id = 'o-1' AND payment_status = 'pending'").run().changes;
console.log('4) mark as paid: changes', markPaid(), 'then', markPaid());

// Output:
// 1) first delivery : processed | quantity 95
// 1) second delivery: duplicate | quantity 95
// 2) error: crash after the UPDATE | quantity 95
// 2) retry          : processed | quantity 90
// 3) without a key at the gateway: charges = 2
// 3) with the key at the gateway: charges = 1
// 4) mark as paid: changes 1 then 0

In the first scenario the second delivery of msg-1 is discarded and the quantity stays 95: the changes === 0 of the INSERT OR IGNORE is the duplicate signal, and the ROLLBACK closes the transaction without running the handler. The second scenario is the reason the key lives in the transaction: the handler fails after the UPDATE, the ROLLBACK undoes both the UPDATE and the key (quantity still 95), and the retry succeeds (90).

The third scenario is the most expensive trap: the charge on the gateway is not part of the database transaction. If the commit fails after the charge, the retry charges again (charges = 2). The only remedy is to pass the key to the gateway as well: Stripe, for example, accepts the Idempotency-Key header on requests and recognises a repeated request; in the program the fake gateway behaves that way and the charges stay at 1. A database transaction guarantees only what lives in the database.

On PostgreSQL the behaviour under concurrency (two consumers inserting the same key at the same instant) is a uniqueness constraint that the database enforces; I did not run it here, and it must be covered by a test on the real database. A note on the key: use the identifier that identifies the business event (for example the id of the event or of the payment), not the identifier the broker assigns to the individual send. If the producer retries and resends the same event, the broker may assign it a new message identifier, and the broker's key does not recognise the copy.

Idempotency by nature: the state, not the delta

The most robust solution is not needing one. A command like "subtract 5 from the stock" changes the result on every repetition; a command like "mark the order as paid if it is pending" does not. The last line of the output shows it: the first run changes one row, the second zero, because the WHERE payment_status = 'pending' clause no longer finds anything to change.

The same holds for an UPSERT that writes the absolute value ("the reservation of order o-1 for the pen is 5") instead of an increment. It is not always possible: a charge is by nature an action, and there you need the key. But every operation you manage to turn into an assertion about state is one mechanism fewer to manage, to expire and to make survive a restart.

Keys with an expiry: the window and the restart

If messages are many, a table of keys kept forever is not practical and you use a store with an expiry. Redis has the SET command with the options NX (the guide: set the key only if it does not already exist) and EX (the expiry in seconds): in a single atomic command it tells you whether the key was new, with no need for Lua scripts.

The model below reproduces it with a virtual clock and shows two limits: the key expires, and a store without persistence loses the keys on restart (the Redis guide warns that an RDB snapshot can lose the last few minutes of data). The second part concerns the producer side: SQS FIFO queues discard a send with the same identifier within five minutes.

// Virtual time in seconds.
const clock = { t: 0 };

// Model of SET key value NX EX ttl: creates the key only if it is missing and makes it expire after ttl seconds.
class ExpiringStore {
  private keys = new Map<string, number>();            // key -> expiry time
  setNxEx(key: string, ttl: number): boolean {
    const expiry = this.keys.get(key);
    if (expiry !== undefined && expiry > clock.t) return false;   // it exists and has not expired: duplicate
    this.keys.set(key, clock.t + ttl);
    return true;
  }
  restartWithoutPersistence(): void { this.keys.clear(); }
}

const TTL = 24 * 3600;
const store = new ExpiringStore();
let effects = 0;
const consume = (key: string) => { if (store.setNxEx(key, TTL)) effects++; };

consume('orders-0-17');
clock.t = 60;           consume('orders-0-17');
console.log('1) duplicate after 1 minute  : effects =', effects);
clock.t = 25 * 3600;    consume('orders-0-17');
console.log('2) duplicate after 25 hours  : effects =', effects, '(the key has expired)');
store.restartWithoutPersistence(); consume('orders-0-17');
console.log('3) after a store restart     : effects =', effects, '(the key was lost)');

// Producer-side deduplication: the key must describe the event, not the instant of the send.
const WINDOW = 300;
class QueueWithDedup {
  private seen = new Map<string, number>();
  accepted: string[] = [];
  send(dedupId: string): void {
    const before = this.seen.get(dedupId);
    if (before !== undefined && clock.t - before < WINDOW) return;
    this.seen.set(dedupId, clock.t);
    this.accepted.push(dedupId);
  }
}
clock.t = 0;
const wrong = new QueueWithDedup(), right = new QueueWithDedup();
for (const attempt of [0, 1]) {                         // the producer retries the send after 1 second
  clock.t = attempt;
  wrong.send(`OrderCreated-o-9-${1_700_000_000 + clock.t}`);   // with the instant in the key
  right.send('OrderCreated-o-9');                               // with the identity of the event
}
console.log('4) key with the instant :', wrong.accepted.length, 'messages accepted');
console.log('4) key of the event     :', right.accepted.length, 'message accepted');

// Output:
// 1) duplicate after 1 minute  : effects = 1
// 2) duplicate after 25 hours  : effects = 2 (the key has expired)
// 3) after a store restart     : effects = 3 (the key was lost)
// 4) key with the instant : 2 messages accepted
// 4) key of the event     : 1 message accepted

The duplicate after one minute is discarded (effects = 1); the one after 25 hours goes through, because the TTL was 24 hours, and the third goes through after the restart. The operational conclusion is that the TTL must be longer than the longest plausible delay of a redelivery (for SQS the default retention is 4 days, so a 24-hour TTL may not be enough) and that for irreversible operations, such as payments, the key lives in the transactional database and not only in Redis.

The last two lines are the classic producer mistake: if you put the instant of the send in the deduplication key, every retry has a new key and goes through; a key that describes the event (type and order id) makes the second send be discarded.

Inbox: receiving and processing are two steps

The inbox pattern moves the key into a table that also holds the message. The first step saves the message and immediately gives the broker its acknowledgement: if the same message arrives twice, the primary key constraint discards the copy. The second step, a worker, reads the messages not yet processed and runs them in a transaction that includes the effect and the "processed" mark. Compared with the key alone you get two advantages: the broker never sees a processing failure (the message is already safe) and you have a history with errors and number of attempts.

import { DatabaseSync } from 'node:sqlite';

const db = new DatabaseSync(':memory:');
db.exec(`
  CREATE TABLE inbox (id TEXT PRIMARY KEY, event_type TEXT NOT NULL, payload TEXT NOT NULL,
                      processed_at TEXT, error TEXT, retry_count INTEGER NOT NULL DEFAULT 0);
  CREATE TABLE inventory (product_id TEXT PRIMARY KEY, reserved INTEGER NOT NULL);
  INSERT INTO inventory VALUES ('pen', 0);
`);

// Phase 1: receiving = saving. The broker can now get the ack: the message is safe, even if it arrives twice.
function receive(id: string, eventType: string, payload: object): boolean {
  return db.prepare('INSERT OR IGNORE INTO inbox (id, event_type, payload) VALUES (?, ?, ?)').run(id, eventType, JSON.stringify(payload)).changes === 1;
}

let fault = true; // the first attempt to process "OrderPlaced" fails
function dispatch(type: string, p: { productId: string; quantity: number }): void {
  if (type !== 'OrderPlaced') throw new Error('Unknown type: ' + type);
  db.prepare('UPDATE inventory SET reserved = reserved + ? WHERE product_id = ?').run(p.quantity, p.productId);
  if (fault) { fault = false; throw new Error('fault after the UPDATE'); }
}

// Phase 2: processing. Effect and "processed" mark in the same transaction.
function processNext(): string {
  const row = db.prepare('SELECT id, event_type, payload FROM inbox WHERE processed_at IS NULL AND retry_count < 3 ORDER BY rowid LIMIT 1').get() as
    { id: string; event_type: string; payload: string } | undefined;
  if (!row) return 'nothing to do';
  db.exec('BEGIN');
  try {
    dispatch(row.event_type, JSON.parse(row.payload));
    db.prepare("UPDATE inbox SET processed_at = 'now', error = NULL WHERE id = ?").run(row.id);
    db.exec('COMMIT');
    return `${row.id} processed`;
  } catch (e) {
    db.exec('ROLLBACK');
    db.prepare('UPDATE inbox SET retry_count = retry_count + 1, error = ? WHERE id = ?').run((e as Error).message, row.id);
    return `${row.id} failed: ${(e as Error).message}`;
  }
}

console.log('received 1st time :', receive('evt-1', 'OrderPlaced', { productId: 'pen', quantity: 3 }));
console.log('received 2nd time :', receive('evt-1', 'OrderPlaced', { productId: 'pen', quantity: 3 }));
console.log('round 1           :', processNext());
console.log('round 2           :', processNext());
console.log('round 3           :', processNext());
console.log('reserved          :', (db.prepare("SELECT reserved FROM inventory WHERE product_id = 'pen'").get() as { reserved: number }).reserved);
console.log('inbox state       :', JSON.stringify(db.prepare('SELECT id, processed_at, retry_count, error FROM inbox').all()));

// Output:
// received 1st time : true
// received 2nd time : false
// round 1           : evt-1 failed: fault after the UPDATE
// round 2           : evt-1 processed
// round 3           : nothing to do
// reserved          : 3
// inbox state       : [{"id":"evt-1","processed_at":"now","retry_count":1,"error":null}]

The second receive returns false: the message is not saved twice. The first round fails after the inventory UPDATE, the ROLLBACK undoes the UPDATE and the worker increments retry_count; the second round succeeds and the reserved units are 3, not 6. The retry_count < 3 limit keeps a poisoned message from cycling forever, but then someone has to look at the failed rows, as with a DLQ.

On PostgreSQL with several workers you typically use SELECT ... FOR UPDATE SKIP LOCKED (the guide points to it for queue-like tables), but the selection and the processing must be in the same transaction: if the read happens in autocommit, the lock disappears immediately and two workers can take the same row. This is the point where people go wrong most often and which SQLite, with a single writer, does not let you try.

How to test idempotency

The normal test is not enough: you have to force redeliveries. The minimum cases are these:

• the same message delivered twice in a row: the effect appears only once;
• a crash after the effect and before the confirmation: the retry neither duplicates it nor loses it;
• two simultaneous copies: only one processes, the other is discarded;
• a copy after the key has expired: the behaviour is decided and documented, not accidental.

The first two are written with an in-memory database, as in the examples above. The third requires the real database with two connections, because it depends on how that database serialises conflicts: it is the test that tells you whether your key holds. The fourth is the one that is usually missing, and that turns the "window" from an occasional bug into a declared limit.

What you can do now

You know why duplicates are expected, you can write key and effect in the same transaction and you can recognise the effects that the transaction does not cover. You can choose between an expiring key, a durable key and an inbox based on the cost of a duplicate, and you have the four tests that show the consumer holds up. The next step concerns the opposite side of the problem: publishing the event without duplicating or losing it, with the outbox pattern. For the side that publishes events see also EventBridge; for what the user sees while the data catches up, eventual consistency.