Idempotent Consumers
At-least-once delivery means your handler will run twice, so make running it twice harmless.
INSERT INTO processed_messages (message_id) VALUES ($1) ON CONFLICT DO NOTHINGQueues and webhooks promise at-least-once delivery: every message arrives, and some arrive more than once. The reason is simple. The consumer does the work and then tells the queue it is done; that confirmation is called an ack. If the process crashes before the ack, or the ack is lost on the network, or the claim on the job expires first, the message is handed out again. The work already happened, and the queue has no way to know.
Idempotent means doing something twice has the same effect as doing it once. That is the property your handler needs.
Three ways to get it
Set a value instead of changing one. SET state = 'shipped' is idempotent. SET balance = balance + 10 is not. Prefer the first shape where the data allows it.
Deduplicate. The message carries a stable id. Keep a table of ids you have processed, with the id as the primary key, and insert it in the same transaction as the effect. If the insert conflicts, you have seen this message before: skip the work and ack.
Make the write conditional. UPDATE ... WHERE state = 'pending' affects zero rows the second time, and that count tells you which run you are in.
What trips people up
Do not SELECT to check and then INSERT. Two workers can both see nothing and both do the work. Let the unique constraint decide and read RowsAffected. And prune the dedupe table only beyond your queue's longest retry window, never sooner.
Example
package main
import "fmt"
// account holds the effect we must not apply twice, plus the ids already
// applied. In a real service seen is a table with message_id as primary key.
type account struct {
balance int
seen map[string]bool
}
// credit applies a message once. A second call with the same id does nothing.
func (a *account) credit(msgID string, amount int) {
// Reading a missing key from a map gives the zero value, false here.
// The longhand is: applied, ok := a.seen[msgID]; ok is false when absent.
if a.seen[msgID] {
fmt.Println("skipped duplicate", msgID)
return
}
a.balance += amount
a.seen[msgID] = true
fmt.Println("applied", msgID, "balance", a.balance)
}
func main() {
a := &account{seen: make(map[string]bool)}
a.credit("msg-1", 50)
a.credit("msg-1", 50) // redelivered because the ack was lost
a.credit("msg-2", 25)
fmt.Println("final balance:", a.balance)
}
// Output:
// applied msg-1 balance 50
// skipped duplicate msg-1
// applied msg-2 balance 75
// final balance: 75
When to use it
- Receive the same queue message twice without charging the card twice.
- Make a payment webhook safe when the sender retries after a timeout.
- Resume a worker that crashed after doing the work but before acking.
- Replay a topic from the start without corrupting the data it feeds.
More examples
Dedupe row and effect in one transaction
The general solution, and the one to use when the effect itself cannot be made idempotent, such as adding to a balance. Both statements commit together, so there is no moment where the message is marked processed but the money has not moved.
// apply credits an account at most once per message.
func apply(ctx context.Context, db *sql.DB, msgID string, accountID, amount int64) error {
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
// message_id is the primary key, so the database does the deduplicating.
res, err := tx.ExecContext(ctx,
`INSERT INTO processed_messages (message_id) VALUES ($1)
ON CONFLICT DO NOTHING`, msgID)
if err != nil {
return err
}
n, err := res.RowsAffected()
if err != nil {
return err
}
if n == 0 {
return nil // seen before: ack the message and do nothing else
}
if _, err := tx.ExecContext(ctx,
`UPDATE accounts SET balance = balance + $1 WHERE id = $2`,
amount, accountID); err != nil {
return err
}
return tx.Commit()
}A conditional write needs no extra table
Cheaper and simpler than a dedupe table when the row already carries a state you can test. Use it for state machines, where every transition has a precondition that stops being true once the transition has happened.
// The second delivery matches no rows, because state is no longer 'paid'.
res, err := db.ExecContext(ctx,
`UPDATE orders SET state = 'shipped', shipped_at = $1
WHERE id = $2 AND state = 'paid'`, shippedAt, orderID)
if err != nil {
return err
}
n, err := res.RowsAffected()
if err != nil {
return err
}
if n == 0 {
// Already shipped, or never paid. Both are fine to ack; neither is a retry.
log.Printf("order %d not shipped by message %s", orderID, msgID)
}
// Compare: this one is not safe to run twice.
// UPDATE accounts SET balance = balance - $1 WHERE id = $2An in-process filter in front of the real check
Useful when a sender retries aggressively and you want to avoid a database round trip per duplicate. The mutex matters: several workers read this map at once, and a map that is written while being read is a data race the race detector will report.
// seen suppresses duplicates inside one process. It is lost on restart, so it
// is a cheap filter in front of the database check, never a replacement for it.
type seen struct {
mu sync.Mutex
ids map[string]bool
}
func (s *seen) first(id string) bool {
s.mu.Lock()
defer s.mu.Unlock()
if s.ids[id] {
return false
}
s.ids[id] = true
return true
}
Discussion