Idempotent Consumers in Go: Handling Duplicate Messages Safely

The problem

A message consumer that assumes it sees each message exactly once is wrong by construction, not by bad luck. Most brokers — Kafka included — give you at-least-once delivery: a message is redelivered whenever the broker can't confirm the consumer finished with it, and "can't confirm" covers a lot of completely normal situations, not just rare disasters.

This is a different problem from Idempotency in APIs. That article is about a client retrying an HTTP request it isn't sure succeeded. This one is about a broker redelivering a message the consumer already fully processed, because the acknowledgment never made it back.

Why it matters

Here's the sequence that causes it, and there's nothing exotic about any step:

The database write already committed. The business effect already happened. The consumer just didn't get to say so before it crashed, lost its connection, or hit a rebalance. From the broker's point of view, an unacknowledged message might as well never have been processed — so it sends it again. Without protection, "again" means double-crediting an account, sending a duplicate notification, or double-incrementing a counter. This is routine operation under at-least-once delivery, not an edge case to shrug off.

Deduplication strategies

Three pieces, in the order they actually get built:

1. A unique event ID the producer attaches to every message — a UUID generated once, at creation time, that stays the same across every redelivery of that same logical event.
2. A processed_events table in PostgreSQL keyed on that event ID, with a unique constraint — not an in-memory map, not a check-then-act query. The constraint is what actually closes the race (more on this below).
3. One transaction wrapping both the dedup insert and the business write, so either both happen or neither does.

Practical implementation

CREATE TABLE processed_events (
    event_id    UUID PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

func (c *PaymentConsumer) HandleEvent(ctx context.Context, event Event) error {
	tx, err := c.db.Begin(ctx)
	if err != nil {
		return fmt.Errorf("begin transaction: %w", err)
	}
	defer tx.Rollback(ctx)
 
	_, err = tx.Exec(ctx,
		`INSERT INTO processed_events (event_id) VALUES ($1)`,
		event.ID,
	)
	if isUniqueViolation(err) {
		// Someone already processed this event ID — this is a normal,
		// expected outcome of at-least-once delivery, not an error.
		return nil
	}
	if err != nil {
		return fmt.Errorf("record processed event: %w", err)
	}
 
	if err := applyPayment(ctx, tx, event); err != nil {
		return fmt.Errorf("apply payment: %w", err)
	}
 
	return tx.Commit(ctx)
}
 
func isUniqueViolation(err error) bool {
	var pgErr *pgconn.PgError
	return errors.As(err, &pgErr) && pgErr.Code == "23505"
}

If the insert into processed_events fails with a unique-constraint violation, the transaction rolls back — including whatever applyPayment had started — and the handler returns cleanly. The business write and the dedup record either both land or neither does, in the same way the outbox pattern uses one transaction to make two separate writes succeed or fail together.

The race condition in check-then-act

The bug is that the check and the act aren't atomic. If the same event is redelivered while the first attempt is still mid-processing — which happens more than people expect during a consumer-group rebalance, when partitions get reassigned mid-flight — both attempts can call alreadyProcessed, both see "not yet processed" (because neither has recorded anything yet), and both proceed to apply the effect. The gap between checking and recording is exactly where the duplicate slips through.

The fix is the code above: attempt the INSERT first, inside the transaction, and let the database's unique constraint be the actual decision-maker. Two concurrent attempts both try to insert the same event_id; the database guarantees only one succeeds. There's no window for both to think they're first, because there's no separate check step at all — the write attempt is the check.

Making the business operation itself idempotent

Deduplication at the message layer is the primary defense, but a second layer costs little when the operation naturally supports it. An UPSERT is idempotent by definition. A guarded update like:

UPDATE accounts
SET balance = balance + $1, last_event_id = $2
WHERE id = $3 AND last_event_id != $2;

only applies if this exact event hasn't already been applied to this row, regardless of what the dedup table says. This isn't a replacement for the processed_events table — it's a second guard that catches the case where dedup bookkeeping is somehow bypassed (a migration, a manual backfill, a bug in the dedup logic itself).

Production considerations

Retention matters, but shouldn't be aggressive. A duplicate delivered weeks after the original almost never happens in practice for most brokers, so a 7–30 day retention window on processed_events is reasonable — delete rows older than the window on a schedule, the same way the outbox pattern cleans up published events.

If the dedup table becomes a hot spot under very high message volume, reach for partitioning by time range or a shorter retention window before considering removing the check — the check is what prevents duplicate side effects; a performance problem there is a scaling problem to solve, not a correctness guarantee to drop.

Common mistakes

• Bare check-then-act (alreadyProcessed as a separate query before the write) instead of letting a unique constraint do the atomic check.
• Recording "processed" in a different transaction than the business write, which reopens the same atomicity gap the outbox pattern exists to close.
• Unbounded processed_events growth with no cleanup job, quietly becoming a multi-million-row table nobody budgeted for.

Testing

func TestHandleEvent_DuplicateDelivery_AppliesOnce(t *testing.T) {
	db := setupTestDB(t)
	consumer := NewPaymentConsumer(db)
	event := Event{ID: uuid.New(), AccountID: "acct_1", AmountCents: 500}
 
	var wg sync.WaitGroup
	for i := 0; i < 2; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = consumer.HandleEvent(context.Background(), event)
		}()
	}
	wg.Wait()
 
	balance := getBalance(t, db, "acct_1")
	if balance != 500 {
		t.Fatalf("expected balance 500 after duplicate delivery, got %d", balance)
	}
}

Firing the same event concurrently is the actual failure mode this code defends against — a sequential test would pass even with the unsafe check-then-act version.

Summary

At-least-once delivery means every consumer will eventually see a message more than once — that's the contract, not a bug in the broker. A unique event ID, a processed_events table with a real unique constraint, and one transaction wrapping both the dedup record and the business write closes the race that a bare alreadyProcessed check leaves open. Layer an idempotent business operation on top where you can, clean up old dedup records on a schedule, and duplicate delivery stops being something to fear.