Lesson 5 · Keep it correct — reliability

At-least-once & idempotency

Message brokers deliver the same message more than once — on purpose. This is the pattern that makes a handler survive that, and the one line in the repo's version that everyone gets wrong.

Your win: explain why duplicates are unavoidable, describe the idempotent-consumer fix, and pinpoint when this repo records the "already processed" marker — and why the timing matters.

Why the same message arrives twice

Kafka and NATS JetStream both guarantee at-least-once delivery, not exactly-once. If a consumer processes a message but crashes (or is slow) before acknowledging it, the broker can't tell "done" from "lost" — so it redelivers. That's the safe choice (better a duplicate than a dropped message), but it means every handler must tolerate seeing the same message twice.1

The idempotent consumer pattern Make processing the same message twice have the same effect as once. Chris Richardson's canonical fix: record the IDs of messages you've processed; on each message, check the record and discard duplicates. If the work isn't naturally idempotent (sending an email, incrementing a counter), you bolt on this dedup layer.1

The repo's IdempotentService

The mechanism is a small DB-backed dedup: a table of processed IDs, wrapped by a service with just two methods:

internal/notification/services/idempotent.go:30,39
func (s *IdempotentService) GetOne(ctx, idempotentID string, _type consts.IdempotentType) (*Idempotent, error)
func (s *IdempotentService) Insert(ctx, idempotentID string, _type consts.IdempotentType) error

A row is keyed by (idempotent_id, type) — the ID is an upstream-supplied ULID on the message, the type an enum (today only upsert_async_notification_kafka). Two things worth knowing: the table lives in the bob DB (not notificationmgmt — query the wrong one and it's always empty), and only the async-notification Kafka consumer uses it. spike has no idempotency table at all.2

The crux: when the marker is written

Here's the line everyone gets wrong. It's tempting to insert the "processed" marker as soon as you start — but look at exactly when this consumer does it:

internal/notification/transports/kafka/consumer_upsert_async_notification.go:129-147
idempotent, err := c.idempotentService.GetOne(ctx, notification.IdempotentID, …)
if err != nil        { return true,  … }   // lookup failed → retryable
if idempotent != nil { return false, … }   // already processed → DROP (don't retry)

defer func() {
    if err == nil || !retryable {          // ← insert marker ONLY on success or non-retryable
        c.idempotentService.Insert(ctx, notification.IdempotentID, …)
    }
}()
Why "only on success or non-retryable" is the whole point If processing fails with a retryable error, the marker is not written — so when the broker redelivers, GetOne finds nothing and the handler tries again. If processing succeeds, or fails in a way that's pointless to retry (a malformed message), the marker is written, so any later duplicate is dropped. Writing the marker too early would mark a message "done" that a retry still needs to complete — silently losing work. The timing is the correctness.2
Two dedup layers, don't confuse them The DB idempotent table is the Kafka-path dedup. On the NATS path, dedup is the Nats-Msg-Id header (a ULID) that JetStream itself uses within a dedup window. Same goal, different layer — application-level table vs broker-level header.
Read this next

Idempotent Consumer — Chris Richardson

The canonical pattern: record processed message IDs, detect and discard duplicates. Two screens, and the reference every interviewer has in mind.

→ microservices.io — Idempotent Consumer
→ in-repo internal/notification/services/idempotent.go, .../transports/kafka/consumer_upsert_async_notification.go:127-148

Check yourself (from memory)

Q1. Why must a consumer tolerate seeing a message twice?

At-least-once: if a consumer might have processed-but-not-acked, the broker redelivers rather than risk a lost message. Duplicates are the price.

Q2. When does this consumer write the idempotent marker?

if err == nil || !retryable. A retryable failure leaves no marker so the redelivery re-processes; success/non-retryable marks it done.

Q3. What happens on a duplicate once the marker exists?

GetOne returns a row → the handler returns false (don't retry) and drops it. Same effect as processing once.
Recall: at-least-once + the idempotent consumer + the marker timing.
the problem + the pattern + the crux, then reveal
Problem: Kafka/NATS = at-least-once; a consumer that processes-then-crashes-before-ack gets the message redelivered → every handler must tolerate duplicates. Pattern (Richardson): idempotent consumer — record processed message IDs, discard duplicates. Repo: IdempotentService (GetOne/Insert, keyed by (idempotent_id, type)); table in the bob DB; one Kafka consumer uses it. Crux: insert the marker in a defer only if err == nil || !retryable — a retryable failure leaves no marker so the redelivery re-processes; marking too early would lose work. NATS path uses Nats-Msg-Id instead.
🎯 Interview one-liner "How do you make a consumer idempotent?" → "Brokers are at-least-once, so I record processed message IDs in a table and drop duplicates on lookup. The subtlety is when you write the marker — only after success or a non-retryable failure, so a retryable failure leaves no marker and the redelivery actually re-runs. Mark it too early and you silently lose work."
Next: the other half of the story — retries & redelivery: what "retryable" actually triggers, and why it retries in-process before it ever goes back to the broker. Ask me if the marker-timing logic needs another pass — it's the crux.

1. Chris Richardson — Idempotent Consumer.

2. In-repo (verified): internal/notification/services/idempotent.go:17-46; .../transports/kafka/consumer_upsert_async_notification.go:127-148; table in bob DB (migrations/bob/1720_migrate.up.sql).