Idempotency in Consumers: Deduplication and Idempotency Key
You've set up your event-driven system with SQS, Kafka, or EventBridge. The messages they flow, consumers process events, everything works perfectly in development. Then you go to production and find that some order confirmation emails are sent twice times, or worse, that certain payments are charged twice. The problem is not a bug in your code: it's a fundamental property of modern distributed systems.
All the major message brokers — SQS, Kafka, RabbitMQ, EventBridge — guarantee the delivery at-least-once: A message is delivered at least once time, but may be delivered multiple times. This happens for automatic retries, rebalancing of consumer groups, visibility timeout, consumer crash during the processing. The solution is not in the broker: it is in the consumer. The consumer must be idempotent.
What You Will Learn
- Because at-least-once systems inevitably produce duplicate messages
- Idempotency Key: the fundamental pattern for deduplication
- Database-level idempotency with INSERT ON CONFLICT
- Redis-based deduplication with TTL for high performance
- Inbox Pattern: the structured solution for exactly-once semantics
- Natural idempotency: how to design naturally idempotent operations
- Testing strategies to verify consumer idempotence
Why Duplicate Messages Are Inevitable
To understand why consumer idempotence is necessary, we need to understand when a message is delivered more than once. The main cases in production:
Case 1: Consumer Crash After Processing, Before ACK
Consumer processes message successfully (writes to DB, calls external API) but it crashes before sending the ACK to the broker. The broker considers the message not delivered and sends it back after the visibility timeout. A new consumer (or the same one after the restart) receives and reprocesses the message.
Case 2: Processing Timeout
SQS has a visibility timeout (default 30 seconds). If the consumer employs more than 30 seconds to process the message without extending the timeout, SQS makes the message visible to other consumers. The message is processed twice in parallel by two different consumers.
Case 3: Kafka Consumer Group Rebalancing
During a rebalancing of the Kafka consumer group (for adding/removing consumers, deploy rolling), some partitions are reassigned. If the consumer who comes removed has not yet committed the offset, the messages in that batch come reprocessed by the new consumer assigned to the partition.
// Simulazione: perche i duplicati sono inevitabili
// Questo codice mostra IL PROBLEMA, non la soluzione
async function processPayment(message: SQSMessage): Promise<void> {
const { paymentId, amount, customerId } = JSON.parse(message.Body);
// Step 1: chiama l'API di pagamento esterna
await paymentGateway.charge(customerId, amount);
// ^^^ SUCCESSO: il pagamento e stato addebitato
// -- CRASH QUI --
// Il processo muore per OOM, segfault, deploy, ecc.
// Il pagamento e gia stato addebitato MA non abbiamo
// ancora eliminato il messaggio dalla coda SQS.
// SQS considera il messaggio non processato e lo
// rimanda dopo il visibility timeout.
await sqs.deleteMessage({
QueueUrl: QUEUE_URL,
ReceiptHandle: message.ReceiptHandle,
});
// ^^^ Mai eseguito se crashiamo sopra
}
// Risultato: il cliente viene addebitato due volte.
// Nessun bug nel codice. E' la natura del sistema at-least-once.
The Pattern Idempotency Key
The most common solution is to use a idempotency key: an identifier unique for each operation that allows the consumer to detect whether it has already processed this message. The consumer checks the database before processing: if the key already exists, skip silent; if it does not exist, process and save the key.
// Pattern base: Idempotency Key con PostgreSQL
// Schema tabella per il tracking
CREATE TABLE processed_messages (
message_id VARCHAR(255) PRIMARY KEY,
processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
consumer_name VARCHAR(100) NOT NULL,
-- TTL gestito da un job di cleanup o da una policy di partizione
expires_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + INTERVAL '7 days'
);
CREATE INDEX idx_processed_messages_expires
ON processed_messages(expires_at);
-- Job di cleanup: esegui ogni ora
DELETE FROM processed_messages WHERE expires_at < NOW();
// TypeScript: consumer idempotente con PostgreSQL
interface ProcessedMessageRecord {
messageId: string;
processedAt: Date;
consumerName: string;
}
class IdempotentConsumer {
constructor(
private readonly db: Pool,
private readonly consumerName: string
) {}
async processMessage<T>(
messageId: string,
payload: T,
handler: (payload: T) => Promise<void>
): Promise<{ processed: boolean; skipped: boolean }> {
const client = await this.db.connect();
try {
await client.query('BEGIN');
// Tenta di inserire il messageId (fail se gia esiste)
const result = await client.query<ProcessedMessageRecord>(`
INSERT INTO processed_messages (message_id, consumer_name, expires_at)
VALUES ($1, $2, NOW() + INTERVAL '7 days')
ON CONFLICT (message_id) DO NOTHING
RETURNING message_id
`, [messageId, this.consumerName]);
if (result.rowCount === 0) {
// Gia processato: skip idempotente
await client.query('ROLLBACK');
console.log(`[${this.consumerName}] Skipping duplicate: ${messageId}`);
return { processed: false, skipped: true };
}
// Prima volta: esegui l'handler nella stessa transazione
await handler(payload);
await client.query('COMMIT');
return { processed: true, skipped: false };
} catch (error) {
await client.query('ROLLBACK');
throw error;
} finally {
client.release();
}
}
}
// Utilizzo nel consumer SQS
const consumer = new IdempotentConsumer(db, 'payment-processor');
async function handlePaymentEvent(message: SQSMessage): Promise<void> {
const payload = JSON.parse(message.Body);
const messageId = message.MessageId; // ID univoco SQS
const { processed, skipped } = await consumer.processMessage(
messageId,
payload,
async (data) => {
await paymentGateway.charge(data.customerId, data.amount);
await db.query(
'UPDATE orders SET payment_status = $1 WHERE id = $2',
['paid', data.orderId]
);
}
);
if (skipped) {
// Log ma non errore: comportamento atteso
metrics.increment('consumer.duplicate_skipped');
}
}
Redis-Based Deduplication: High Performance
The PostgreSQL check introduces a database query for each message. For high throughput systems (thousands of messages per second), this can become a bottleneck. Redis with TTL and the solution: O(1) operation, sub-millisecond latency, Native TTL for automatic expiration.
// Redis-based deduplication per alto throughput
import { Redis } from 'ioredis';
class RedisIdempotencyStore {
constructor(
private readonly redis: Redis,
private readonly ttlSeconds: number = 86400 // 24 ore default
) {}
// Ritorna true se e la PRIMA VOLTA che vediamo questa key
// Ritorna false se e un duplicato
async setIfAbsent(key: string): Promise<boolean> {
// SET key value NX EX ttl
// NX = solo se non esiste
// EX = TTL in secondi
const result = await this.redis.set(
`dedup:${key}`,
'1',
'EX',
this.ttlSeconds,
'NX'
);
return result === 'OK'; // 'OK' = primo inserimento, null = gia esisteva
}
async isProcessed(key: string): Promise<boolean> {
const exists = await this.redis.exists(`dedup:${key}`);
return exists === 1;
}
// Per operazioni atomiche: check + set in Lua script
async checkAndSet(key: string): Promise<boolean> {
const luaScript = `
local exists = redis.call('EXISTS', KEYS[1])
if exists == 0 then
redis.call('SET', KEYS[1], '1', 'EX', ARGV[1])
return 1
end
return 0
`;
const result = await this.redis.eval(
luaScript,
1,
`dedup:${key}`,
this.ttlSeconds.toString()
);
return result === 1;
}
}
// Consumer con Redis deduplication
class HighThroughputConsumer {
constructor(
private readonly dedup: RedisIdempotencyStore,
private readonly db: Pool
) {}
async handleKafkaMessage(
topic: string,
partition: number,
offset: string,
payload: OrderPayload
): Promise<void> {
// Componi una key univoca: topic + partition + offset
const messageKey = `${topic}-${partition}-${offset}`;
const isFirst = await this.dedup.setIfAbsent(messageKey);
if (!isFirst) {
// Duplicato: skip
return;
}
// Prima elaborazione: procedi
await this.processOrder(payload);
}
private async processOrder(payload: OrderPayload): Promise<void> {
await this.db.query(
'UPDATE inventory SET quantity = quantity - $1 WHERE product_id = $2',
[payload.quantity, payload.productId]
);
}
}
// ATTENZIONE: Redis ha durabilita limitata.
// Se Redis perde dati (AOF/RDB non aggiornati), i duplicati
// potrebbero passare dopo un crash Redis.
// Per operazioni critiche (pagamenti), usa sempre PostgreSQL.
Inbox Pattern: Exactly-Once Semantics
The pattern Inbox and the most robust version of idempotence: the message is first written to the inbox table (within a DB transaction), then tried. Guarantees exactly-once even in case of crash during processing.
-- Schema Inbox Pattern
CREATE TABLE inbox_messages (
id UUID PRIMARY KEY,
source VARCHAR(100) NOT NULL, -- nome del producer/queue
event_type VARCHAR(100) NOT NULL,
payload JSONB NOT NULL,
received_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
processed_at TIMESTAMPTZ, -- NULL = non ancora processato
error TEXT, -- NULL = successo o non processato
retry_count INTEGER NOT NULL DEFAULT 0
);
-- Index per il worker che processa i messaggi pendenti
CREATE INDEX idx_inbox_pending
ON inbox_messages(received_at)
WHERE processed_at IS NULL AND retry_count < 3;
// TypeScript: Inbox Pattern completo
class InboxProcessor {
constructor(private readonly db: Pool) {}
// Fase 1: scrivi nella inbox (idempotente grazie al PK)
async receiveMessage(message: IncomingMessage): Promise<void> {
await this.db.query(`
INSERT INTO inbox_messages (id, source, event_type, payload)
VALUES ($1, $2, $3, $4)
ON CONFLICT (id) DO NOTHING
`, [
message.id,
message.source,
message.eventType,
JSON.stringify(message.payload)
]);
// Se il messaggio arriva due volte, ON CONFLICT DO NOTHING
// lo scarta silenziosamente
}
// Fase 2: worker che processa i messaggi dalla inbox
async processPendingMessages(): Promise<void> {
// Prendi un messaggio con SELECT FOR UPDATE SKIP LOCKED
// Evita che piu worker prendano lo stesso messaggio
const { rows } = await this.db.query(`
SELECT id, event_type, payload
FROM inbox_messages
WHERE processed_at IS NULL
AND retry_count < 3
ORDER BY received_at
LIMIT 1
FOR UPDATE SKIP LOCKED
`);
if (rows.length === 0) return;
const message = rows[0];
const client = await this.db.connect();
try {
await client.query('BEGIN');
// Esegui l'handler specifico per il tipo di evento
await this.dispatch(message.event_type, message.payload);
// Marca come processato nella stessa transazione
await client.query(`
UPDATE inbox_messages
SET processed_at = NOW(), error = NULL
WHERE id = $1
`, [message.id]);
await client.query('COMMIT');
} catch (error) {
await client.query('ROLLBACK');
// Incrementa retry count e salva l'errore
await this.db.query(`
UPDATE inbox_messages
SET retry_count = retry_count + 1,
error = $1
WHERE id = $2
`, [(error as Error).message, message.id]);
} finally {
client.release();
}
}
private async dispatch(
eventType: string,
payload: unknown
): Promise<void> {
switch (eventType) {
case 'OrderPlaced':
await this.handleOrderPlaced(payload as OrderPayload);
break;
case 'PaymentReceived':
await this.handlePaymentReceived(payload as PaymentPayload);
break;
default:
throw new Error(`Unknown event type: ${eventType}`);
}
}
private async handleOrderPlaced(payload: OrderPayload): Promise<void> {
await this.db.query(
'UPDATE inventory SET reserved = reserved + $1 WHERE product_id = $2',
[payload.quantity, payload.productId]
);
}
private async handlePaymentReceived(payload: PaymentPayload): Promise<void> {
await this.db.query(
'UPDATE orders SET status = $1 WHERE id = $2',
['confirmed', payload.orderId]
);
}
}
Natural Idempotency: Designing Idempotent Operations for Nature
The most elegant solution to idempotence is to design operations so that are naturally idempotent: executing them multiple times produces the same result of the single execution. This eliminates the need for explicit tracking.
// Operazioni naturalmente idempotenti vs non idempotenti
// NON IDEMPOTENTE: aggiornamento relativo
// Se eseguita due volte, l'inventory diventa -2 invece di -1
async function decrementInventory(productId: string, qty: number): Promise<void> {
await db.query(
'UPDATE inventory SET quantity = quantity - $1 WHERE product_id = $2',
[qty, productId]
);
}
// IDEMPOTENTE: aggiornamento assoluto con versioning
// Usa il numero dell'ordine come "target state"
async function setInventoryForOrder(
productId: string,
orderId: string,
newQuantity: number
): Promise<void> {
await db.query(`
INSERT INTO inventory_reservations (order_id, product_id, quantity)
VALUES ($1, $2, $3)
ON CONFLICT (order_id, product_id)
DO UPDATE SET quantity = EXCLUDED.quantity
`, [orderId, productId, newQuantity]);
}
// NON IDEMPOTENTE: INSERT senza conflict handling
async function createPaymentRecord(payment: Payment): Promise<void> {
await db.query(
'INSERT INTO payments (id, order_id, amount) VALUES ($1, $2, $3)',
[payment.id, payment.orderId, payment.amount]
);
// Fallisce con unique constraint la seconda volta
}
// IDEMPOTENTE: UPSERT con ON CONFLICT DO NOTHING
async function upsertPaymentRecord(payment: Payment): Promise<void> {
await db.query(`
INSERT INTO payments (id, order_id, amount, status)
VALUES ($1, $2, $3, 'pending')
ON CONFLICT (id) DO NOTHING
`, [payment.id, payment.orderId, payment.amount]);
}
// IDEMPOTENTE: update a stato finale (state machine idempotente)
// Transitare da 'confirmed' a 'confirmed' non cambia nulla
async function markOrderAsShipped(orderId: string): Promise<void> {
await db.query(`
UPDATE orders
SET status = 'shipped', shipped_at = COALESCE(shipped_at, NOW())
WHERE id = $1
AND status IN ('confirmed', 'processing')
`, [orderId]);
// Se lo stato e gia 'shipped', la WHERE non matcha: no-op
}
SQS Level Deduplication
SQS FIFO Queue offers native deduplication via Message Deduplication ID. Messages with the same deduplication ID sent in the deduplication interval (5 minutes) are delivered only once. It does not eliminate the need for idempotence consumer side, but significantly reduces duplicates.
// AWS SDK v3: invio su SQS FIFO con MessageDeduplicationId
import { SQSClient, SendMessageCommand } from '@aws-sdk/client-sqs';
const sqs = new SQSClient({ region: 'eu-west-1' });
async function publishOrderEvent(
orderId: string,
eventType: string,
payload: unknown
): Promise<void> {
// MessageDeduplicationId: hash del contenuto o ID evento univoco
// Stesso ID = stesso messaggio entro 5 minuti = consegnato una sola volta
const deduplicationId = `${eventType}-${orderId}-${Date.now()}`;
await sqs.send(new SendMessageCommand({
QueueUrl: 'https://sqs.eu-west-1.amazonaws.com/123456/orders.fifo',
MessageBody: JSON.stringify(payload),
MessageGroupId: orderId, // Ordine FIFO per stesso ordine
MessageDeduplicationId: deduplicationId,
MessageAttributes: {
EventType: {
DataType: 'String',
StringValue: eventType,
},
},
}));
}
// SQS Standard Queue: nessuna deduplication nativa
// Puoi usare l'Attribute MessageId come idempotency key nel consumer
async function handleSqsStandardMessage(msg: SQSMessage): Promise<void> {
// msg.MessageId e univoco per invio, ma se SQS rimanda il messaggio
// il MessageId rimane lo stesso. Usalo come idempotency key.
const idempotencyKey = msg.MessageId;
await consumer.processMessage(idempotencyKey, JSON.parse(msg.Body), handler);
}
Idempotence Testing
An idempotent consumer must be explicitly tested: it is not enough to test the normal case. You need to test what happens when the same message arrives twice, ten times, with several parallel consumers.
// Test suite per consumer idempotente
describe('IdempotentPaymentConsumer', () => {
let consumer: IdempotentConsumer;
let db: Pool;
beforeEach(async () => {
db = await createTestDatabase();
consumer = new IdempotentConsumer(db, 'payment-test');
await db.query('DELETE FROM processed_messages');
});
it('should process message exactly once on first delivery', async () => {
const messageId = 'msg-001';
const payload = { orderId: 'ord-001', amount: 100 };
let callCount = 0;
const handler = async () => { callCount++; };
const result = await consumer.processMessage(messageId, payload, handler);
expect(result.processed).toBe(true);
expect(result.skipped).toBe(false);
expect(callCount).toBe(1);
});
it('should skip duplicate message silently', async () => {
const messageId = 'msg-001';
const payload = { orderId: 'ord-001', amount: 100 };
let callCount = 0;
const handler = async () => { callCount++; };
// Prima consegna
await consumer.processMessage(messageId, payload, handler);
// Seconda consegna (duplicato)
const result = await consumer.processMessage(messageId, payload, handler);
expect(result.processed).toBe(false);
expect(result.skipped).toBe(true);
expect(callCount).toBe(1); // Handler chiamato solo una volta
});
it('should handle concurrent duplicate messages correctly', async () => {
const messageId = 'msg-concurrent';
const payload = { orderId: 'ord-002', amount: 200 };
let callCount = 0;
const handler = async () => {
callCount++;
// Simula elaborazione lenta per forzare concorrenza
await new Promise((resolve) => setTimeout(resolve, 100));
};
// Simula 5 consumer che ricevono lo stesso messaggio in parallelo
const results = await Promise.allSettled([
consumer.processMessage(messageId, payload, handler),
consumer.processMessage(messageId, payload, handler),
consumer.processMessage(messageId, payload, handler),
consumer.processMessage(messageId, payload, handler),
consumer.processMessage(messageId, payload, handler),
]);
const processed = results.filter(
(r) => r.status === 'fulfilled' && r.value.processed
).length;
// Solo uno deve essere processato, gli altri skippati
expect(processed).toBe(1);
expect(callCount).toBe(1);
});
});
Anti-Pattern: Idempotence Only in Memory
Do not use an in-memory Set or Map to track processed messages. If the process is restarted, the memory and all messages are lost previously processed will be processed again. The idempotency store must be persistent (PostgreSQL, Redis with AOF, DynamoDB).
Conclusions and Next Steps
Consumer idempotency is not an optional optimization in a system event-driven production-grade: it is a fundamental requirement. The choice between PostgreSQL-based deduplication, Redis, and Inbox Pattern depends on the level required durability and system throughput. For critical operations (payments, irreversible status updates), the Inbox Pattern offers the maximum guarantee of exactly-once semantics.







