Add an Idempotent Consumer Handler

Record message ids in the same transaction as the business change so at-least-once delivery produces a single effect after retries and restarts.

~30 minmessagingidempotencyreliability

Task

Make a queue consumer safe to run more than once for the same message without duplicating writes or side effects.

When to use this

A worker processes a payment event, crashes before acknowledging, and the broker redelivers. The first transaction may already have applied the charge.

Prerequisites

A durable database available to the consumer
A stable message id from the broker or producer
A business operation that can share a transaction with the deduplication record

Steps

1. Create a processed-messages table keyed by consumer and message id.

CREATE TABLE processed_messages (consumer text, message_id text, processed_at timestamptz DEFAULT now(), PRIMARY KEY (consumer, message_id));

Expected output: The table enforces one row per consumer and message.

2. Insert the message id and apply the business change in one transaction; treat a unique violation as already processed.

await db.query('BEGIN');
try {
  await db.query(
    'INSERT INTO processed_messages (consumer, message_id) VALUES ($1, $2)',
    ['payments', message.id]
  );
  await applyPayment(message, db);
  await db.query('COMMIT');
} catch (error) {
  await db.query('ROLLBACK');
  if (error.code === '23505') return ack(message); // duplicate
  throw error;
}

Expected output: The first delivery applies the change; a duplicate is acknowledged without applying it again.

3. Test a redelivery after a simulated crash and a duplicate with the same id.

node --test consumer.test.mjs

Expected output: The business table contains exactly one effect for the message id.

Troubleshooting

Problem

Duplicates still apply twice

Solution

The dedup insert is outside the business transaction or the database cannot enforce the unique key. Use one transaction and a durable constraint.

Problem

The dedup table grows without bound

Solution

Retain ids longer than the broker's redelivery window and partition or archive old rows.

Limitations

Idempotency by message id does not deduplicate two different messages describing the same business action; use a domain key or version as well.
Some side effects, such as sending email, cannot join the database transaction; use an outbox or an idempotent provider API.

Deduplicate with the effect

At-least-once delivery means the consumer must assume duplicates. Store the message id with the business change in one transaction. If the insert conflicts, the message was processed; acknowledge it. If the transaction fails, both changes roll back and the broker can redeliver.

BEGIN;
INSERT INTO processed_messages (consumer, message_id) VALUES ('payments', $1);
UPDATE accounts SET balance = balance + $2 WHERE id = $3;
COMMIT;

The unique constraint is the arbiter, not an in-memory set. Use the broker’s message id or a producer-supplied event id, not a hash of the payload.

Keep the window bounded

Processed ids cannot live forever. Retain them longer than the broker’s maximum redelivery window, partition by time, and archive or delete older rows. If a message can be delivered after the window, the broker configuration must bound redelivery.

Handle non-transactional effects

Email, webhooks, and third-party APIs cannot join the database transaction. Use an outbox row written in the same transaction, then a relay sends the effect with retries. Make the external call idempotent with a provider-supported idempotency key where available.

Verify the duplicate path

Expected outcome: delivering the same message id twice leaves exactly one business effect, and a failed transaction leaves none so a later redelivery can succeed. Confirm with an integration test that runs the handler twice and queries the effect table, plus a crash simulation between the business update and the acknowledgement. Check that the deduplication row and the effect commit or roll back together.