CodePark/ questions / Delivery guarantees & idempotent consumers
START PRACTICINGPRACTICE
Backend question · Message queues & streaming › Delivery guarantees & idempotent consumers

How do you implement an idempotent consumer in practice? Walk through the code.

await db.transaction(async (tx) => {
  const inserted = await tx.query(
    `INSERT INTO processed_messages (message_id)
     VALUES ($1) ON CONFLICT DO NOTHING`, [msg.id]);
  if (inserted.rowCount === 0) return; // duplicate
  await tx.query(
    "UPDATE accounts SET balance = balance + $1 WHERE id = $2",
    [msg.amount, msg.accountId]);
});
await ack(msg);

Give every message a unique id, and record processed ids in a table with a unique constraint. An idempotent consumer is one that gives the same result if it gets the same message twice.

The producer sets the id, not the broker. So if the producer sends the message again, the copy has the same id. The consumer's table has a unique constraint: the database refuses a second row with the same id.

The steps, all inside one database transaction:

  • Insert the message id first.
  • If the insert hits a conflict, the message was already processed. The handler returns.
  • Otherwise, apply the business change.
  • Commit both writes together.
  • Only after the commit, ack the message. To ack means to tell the broker the message is done.

What happens on a crash:

  • A crash before the commit rolls back both writes. The broker redelivers the message, and the consumer processes it cleanly.
  • A crash after the commit but before the ack also causes a redelivery. This time the unique constraint catches the duplicate.

Delete old ids after a time window. Make the window longer than any realistic redelivery delay. Some operations are naturally idempotent: repeating them gives the same result, such as an upsert or setting a status. For those, you may not need the table at all.

  • The producer gives each message a unique id.
  • Insert the id, with a unique constraint, in the same transaction as the business write.
  • A conflict on insert means the message is a duplicate, so skip it.
  • Ack the message only after the commit.
  • Delete old ids after a time window. A naturally idempotent operation can replace the table.