Idempotency Is a Feature: Designing Fan-Out That Survives Retries

A while back, three users on one large tenant received the same billing alert three times in a row. Not three different alerts. The same one, with the same amount, the same timestamp, the same “Action Required” subject line. One of them replied to the support channel with a screenshot of their inbox and the message: “Are you sure I owe you this much?”

The root cause wasn’t a bug in the notification code. The send logic was fine. What happened was simpler and more annoying: a network timeout during a batch of notification inserts caused a partial failure. The job runner retried the entire batch. Most rows were duplicates of rows that had already been written, and those duplicates made it all the way to the user’s inbox because nothing in the pipeline checked whether a notification had already been sent.

I’ve written before about worker pools and concurrency patterns at the code level. This post goes one level up the stack, into the design patterns that make distributed writes safe when retries are inevitable. Because retries are always inevitable.

The Myth of Exactly-Once

Let’s get this out of the way: exactly-once delivery does not exist in distributed systems. The phrase sounds reassuring in vendor docs and conference talks, but the underlying reality is harsher. Network calls can fail after the remote side has already processed the request. You can’t distinguish “the request failed” from “the request succeeded but the acknowledgment was lost.” This is a fundamental property of distributed systems, not an engineering limitation you can overcome with enough cleverness.

What you can actually achieve is at-least-once delivery with idempotent consumers. You accept that messages might arrive more than once, and you design the receiving side so that processing the same message twice produces the same result as processing it once. The retry is safe because the duplicate is a no-op.

This sounds simple in theory. The engineering is in the details.

Idempotency Keys

The core mechanism for making writes idempotent is the idempotency key: a deterministic identifier derived from the operation itself, stored in the database with a uniqueness constraint. If the key already exists, the write is a duplicate and gets skipped.

Designing the Key

A good idempotency key answers the question: “What makes this operation unique?” For a notification system, a notification is unique per entity, event type, and recipient. If user Alice should get one billing alert for invoice #1234, the idempotency key is:

billing_alert:inv_1234:user_alice

The key must be deterministic. Given the same inputs, it must produce the same key every time. No UUIDs, no timestamps, nothing random. If you include a timestamp, a retried operation produces a different key and the duplicate isn’t caught.

def notification_idempotency_key(
    event_type: str,
    entity_id: str,
    recipient_id: str,
) -> str:
    """Deterministic key: same inputs always produce the same key."""
    return f"{event_type}:{entity_id}:{recipient_id}"

The Enforcement Layer

The idempotency key lives in a table with a unique constraint. The insert uses ON CONFLICT DO NOTHING, which means Postgres silently skips the row if the key already exists. No exception, no retry logic, no race condition.

CREATE TABLE notification_log (
    idempotency_key TEXT PRIMARY KEY,
    entity_id UUID NOT NULL,
    event_type TEXT NOT NULL,
    recipient_id UUID NOT NULL,
    sent_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ DEFAULT now()
);

-- The insert is the enforcement layer
INSERT INTO notification_log (idempotency_key, entity_id, event_type, recipient_id)
VALUES ($1, $2, $3, $4)
ON CONFLICT (idempotency_key) DO NOTHING
RETURNING idempotency_key;

If the RETURNING clause returns a row, the insert succeeded and this is a new notification. If it returns nothing, the key already existed and the notification is a duplicate. Your application checks this before doing the actual send:

async def send_notification_idempotently(
    db: AsyncSession,
    event_type: str,
    entity_id: str,
    recipient_id: str,
    payload: dict,
):
    key = notification_idempotency_key(event_type, entity_id, recipient_id)

    result = await db.execute(
        text("""
            INSERT INTO notification_log (idempotency_key, entity_id, event_type, recipient_id)
            VALUES (:key, :entity_id, :event_type, :recipient_id)
            ON CONFLICT (idempotency_key) DO NOTHING
            RETURNING idempotency_key
        """),
        {"key": key, "entity_id": entity_id, "event_type": event_type, "recipient_id": recipient_id},
    )

    if result.rowcount == 0:
        # Duplicate detected, skip silently
        logger.info("Duplicate notification skipped", key=key)
        return

    # First time seeing this key, send the notification
    await send_email(recipient_id, payload)

    # Mark as sent
    await db.execute(
        text("UPDATE notification_log SET sent_at = now() WHERE idempotency_key = :key"),
        {"key": key},
    )
    await db.commit()

This pattern is simple, reliable, and fast. The unique constraint is enforced by Postgres at the storage engine level, which means it’s safe under concurrent inserts. Two workers can try to insert the same key at the same time, and exactly one will succeed.

Fan-Out Patterns

With idempotency keys handling the duplicate problem, let’s look at the fan-out itself. When an event occurs (say, a new invoice is created), you need to notify multiple recipients. There are two fundamental approaches.

Fan-Out-on-Write

Create one notification row per recipient at the time the event occurs:

async def fan_out_on_write(
    db: AsyncSession,
    event_type: str,
    entity_id: str,
    recipient_ids: list[str],
    payload: dict,
):
    """Create notification rows for all recipients immediately."""
    rows = [
        {
            "idempotency_key": notification_idempotency_key(event_type, entity_id, rid),
            "entity_id": entity_id,
            "event_type": event_type,
            "recipient_id": rid,
        }
        for rid in recipient_ids
    ]

    # Batch insert with conflict handling
    await db.execute(
        text("""
            INSERT INTO notification_log (idempotency_key, entity_id, event_type, recipient_id)
            SELECT
                unnest(:keys) AS idempotency_key,
                unnest(:entity_ids) AS entity_id,
                unnest(:event_types) AS event_type,
                unnest(:recipient_ids) AS recipient_id
            ON CONFLICT (idempotency_key) DO NOTHING
        """),
        {
            "keys": [r["idempotency_key"] for r in rows],
            "entity_ids": [str(r["entity_id"]) for r in rows],
            "event_types": [r["event_type"] for r in rows],
            "recipient_ids": [r["recipient_id"] for r in rows],
        },
    )
    await db.commit()

Fan-out-on-write is simple and predictable. The cost is paid upfront: if you have 10,000 recipients, you insert 10,000 rows immediately. For notification workloads where the recipient list is bounded (team members, account admins), this is usually fine.

Fan-Out-on-Read

For high-fanout scenarios (activity feeds, social-style notifications), inserting one row per recipient per event doesn’t scale. Instead, you store the event once and compute per-user views at read time:

-- Store the event once
INSERT INTO events (id, tenant_id, event_type, entity_id, payload, created_at)
VALUES ($1, $2, $3, $4, $5, now());

-- At read time, compute the user's feed
SELECT e.*
FROM events e
JOIN team_memberships tm ON tm.tenant_id = e.tenant_id
WHERE tm.user_id = $1
  AND e.created_at > $2  -- Only events after the user's last read timestamp
ORDER BY e.created_at DESC
LIMIT 50;

Fan-out-on-read trades write cost for read cost. The event is written once regardless of how many users might see it. The downside: reads become joins, which are slower and harder to paginate. The split I’d default to: fan-out-on-write for transactional notifications, where the recipient list is small and delivery has to be auditable, and fan-out-on-read for activity feeds, where it isn’t.

Advisory Locks for Race-Prone Edges

Some fan-out operations aren’t safe under concurrent execution. If two workers pick up the same event from a queue and both try to fan out, you get duplicate rows (mitigated by idempotency keys) and wasted work (not mitigated by anything).

Postgres advisory locks let you serialize access to a specific operation without locking any table rows:

async def fan_out_with_lock(
    db: AsyncSession,
    event_id: str,
    recipient_ids: list[str],
):
    """Use an advisory lock to prevent concurrent fan-out for the same event."""
    lock_key = hash(event_id) % (2**31)  # Advisory locks use int4 keys

    # Try to acquire the lock without waiting
    result = await db.execute(
        text("SELECT pg_try_advisory_xact_lock(:key)"),
        {"key": lock_key},
    )
    acquired = result.scalar()

    if not acquired:
        # Another worker is already processing this event
        logger.info("Skipping fan-out, lock held by another worker", event_id=event_id)
        return

    # We have the lock. Proceed with fan-out.
    for rid in recipient_ids:
        await send_notification_idempotently(db, "invoice", event_id, rid, {})

pg_try_advisory_xact_lock is the non-blocking variant. It returns False immediately if another transaction holds the lock, rather than waiting. The lock is released automatically when the transaction commits or rolls back. No manual cleanup, no risk of orphaned locks.

Race-Safe Bulk Operations

There’s a class of operations that looks simple but hides nasty race conditions: “mark all notifications as read,” “dismiss all alerts,” “acknowledge all items up to this point.” These are bulk writes that affect a set of rows, and they interact badly with concurrent inserts.

The Problem with Row Flags

The naive approach sets a boolean flag on each row:

-- Mark all notifications as read for this user
UPDATE notification_log
SET read = true
WHERE recipient_id = $1 AND read = false;

If a new notification is inserted between when the user clicks “mark all read” and when the update executes, that new notification gets marked as read even though the user never saw it. The user misses the notification entirely.

Monotonic Watermarks

The fix: instead of flagging individual rows, store a timestamp watermark. Everything before the watermark is considered read. Everything after it is unread.

CREATE TABLE read_watermarks (
    user_id UUID PRIMARY KEY REFERENCES users(id),
    watermark TIMESTAMPTZ NOT NULL DEFAULT '1970-01-01'
);

-- Mark all as read: set the watermark to the most recent notification's timestamp
UPDATE read_watermarks
SET watermark = (
    SELECT max(created_at) FROM notification_log
    WHERE recipient_id = $1
)
WHERE user_id = $1;

-- Query unread notifications: everything after the watermark
SELECT * FROM notification_log n
JOIN read_watermarks w ON w.user_id = n.recipient_id
WHERE n.recipient_id = $1
  AND n.created_at > w.watermark
ORDER BY n.created_at DESC;

This is race-safe because the watermark is set to the max(created_at) at the time of the update. Any notification inserted after that moment has a later timestamp and stays unread. No flags to get out of sync, no lost notifications.

The watermark pattern generalizes to any “dismiss everything up to now” operation: notification badges, activity feed read markers, alert acknowledgments.

The Outbox Pattern

All the patterns above assume you’re writing to Postgres and that’s the whole story. But real systems cross boundaries. You need to insert a row in your database and also send an email, publish a Kafka message, or call a webhook. The dangerous version looks like this:

# DON'T DO THIS
async def handle_invoice_created(db: AsyncSession, invoice: Invoice):
    # Step 1: Write to database
    notification = Notification(...)
    db.add(notification)
    await db.commit()

    # Step 2: Send email
    await send_email(invoice.customer_email, ...)  # What if this fails?

    # Step 3: Publish to Kafka
    await kafka_producer.send(...)  # What if THIS fails?

If step 2 or step 3 fails, you have a row in your database but the email was never sent or the Kafka message was never published. Retrying the entire function might duplicate the database write (mitigated by idempotency keys) but you’ve lost the guarantee that all three operations succeed or none do.

The outbox pattern solves this by writing everything to the database first, then processing the side effects asynchronously:

CREATE TABLE outbox (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type TEXT NOT NULL,    -- 'notification', 'webhook', etc.
    aggregate_id TEXT NOT NULL,
    event_type TEXT NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ DEFAULT now(),
    processed_at TIMESTAMPTZ          -- NULL = unprocessed
);

async def handle_invoice_created(db: AsyncSession, invoice: Invoice):
    """Write everything to the database in one transaction."""
    async with db.begin():
        # Write the notification
        notification = Notification(...)
        db.add(notification)

        # Write the outbox entry for the email
        outbox_entry = Outbox(
            aggregate_type="notification",
            aggregate_id=str(notification.id),
            event_type="send_email",
            payload={"to": invoice.customer_email, "subject": "Invoice Created"},
        )
        db.add(outbox_entry)

        # Write another outbox entry for Kafka
        kafka_entry = Outbox(
            aggregate_type="invoice",
            aggregate_id=str(invoice.id),
            event_type="publish_event",
            payload={"topic": "invoices", "data": invoice.to_dict()},
        )
        db.add(kafka_entry)
    # Single commit: all or nothing

A separate worker polls the outbox table for unprocessed entries and handles the side effects:

async def outbox_worker(db: AsyncSession):
    """Poll the outbox and process entries, with advisory lock to prevent concurrency."""
    while True:
        async with db.begin():
            # Grab the next unprocessed entry, locking it
            result = await db.execute(
                text("""
                    SELECT * FROM outbox
                    WHERE processed_at IS NULL
                    ORDER BY created_at ASC
                    LIMIT 1
                    FOR UPDATE SKIP LOCKED
                """)
            )
            entry = result.first()
            if entry is None:
                await asyncio.sleep(1)
                continue

            try:
                await process_outbox_entry(entry)
                await db.execute(
                    text("UPDATE outbox SET processed_at = now() WHERE id = :id"),
                    {"id": entry.id},
                )
            except Exception:
                logger.exception("Outbox processing failed", entry_id=entry.id)
                # The transaction rolls back, the entry stays unprocessed,
                # and the worker will retry on the next poll.

The FOR UPDATE SKIP LOCKED clause is important. It locks the row so no other worker grabs it, and SKIP LOCKED means workers don’t wait for each other. If you have multiple outbox workers for throughput, they each pick up different entries without contention.

The outbox pattern turns cross-system consistency into a database problem. Postgres gives you atomicity, and the outbox worker gives you at-least-once delivery of side effects. Combined with idempotency keys on the consumer side, you get reliable, exactly-once-effective processing across system boundaries.

Takeaways

1. Exactly-once delivery is a myth. Design for at-least-once delivery with idempotent consumers. This is the only contract you can reliably enforce in a distributed system.
2. Idempotency keys must be deterministic. Derive them from the operation’s natural key (entity + event + recipient), never from random values or timestamps.
3. Let Postgres enforce uniqueness. ON CONFLICT DO NOTHING is your friend. It’s atomic, race-safe, and handles concurrent inserts correctly.
4. Choose your fan-out strategy by scale. Fan-out-on-write for bounded recipient lists, fan-out-on-read for high-fanout scenarios like activity feeds.
5. Use advisory locks for serialization. pg_try_advisory_xact_lock prevents wasted work when multiple workers pick up the same event.
6. Prefer monotonic watermarks over row flags. For “mark all as read” operations, a timestamp watermark is simpler and race-safe. Row flags invite subtle bugs.
7. Cross system boundaries with the outbox pattern. Write everything to the database in one transaction, then process side effects asynchronously with FOR UPDATE SKIP LOCKED.
8. Idempotency is a feature you design, not a property you hope for. It requires deliberate key design, schema support, and application discipline. The payoff is a system that handles retries gracefully instead of spamming your users at 2 AM.

---

This post draws on patterns from building notification and event systems on a multi-tenant SaaS platform. The duplicate-notification incident was real; the fix shipped the same week.