Deterministic Shard-Aware Outbox For Idempotent Event Publishing

I ran into a maddening failure mode in a system design I built: event publishing “worked” in tests, but under production retries it occasionally delivered duplicate side effects out of order—despite having an outbox table and “at least once” delivery assumptions.

The root cause wasn’t the outbox itself; it was how I mapped events to partitions (shards) during retries. The fix was to make publishing deterministic per event and stable across retries, so the same event always goes to the same shard and gets the same processing order key.

This post documents the niche design I ended up implementing: a deterministic shard-aware outbox that uses a stable key derived from the outbox row, making retries safe and preserving ordering per business entity.

The specific failure: duplicates + reordering during retry

My system had:

A database table outbox_events where the app wrote events in the same transaction as state changes.
A background publisher that read unapublished rows and sent them to a message broker.
A consumer that performed side effects (e.g., calling downstream services).

Even with an outbox pattern, things can go wrong when:

The publisher retries after timeouts.
The publisher assigns the message to a shard/partition based on current time or a changing value (or based on “next available partition”).
The consumer’s ordering guarantee is only per partition.

Result:

The “same” logical event can be delivered twice (expected with at-least-once).
The duplicates land in different partitions across retries, so the consumer observes them out of order relative to other events for the same entity.

The design: deterministic shard + stable ordering key

Core idea

For each outbox row, I compute:

dedupe_key: a stable idempotency key for the consumer (unique per logical event).
partition_key: a stable value that decides which broker partition/shard receives it.
ordering_key: a stable value that helps the consumer enforce in-order processing for a given entity.

To make this robust:

The publisher never changes partition assignment for the same event.
The consumer can safely ignore duplicates using dedupe_key.

Why “deterministic shard-aware” matters

Most message brokers guarantee ordering only within a partition. So if my app accidentally routes the same event to a different partition on retry, ordering can be violated even if dedupe exists.

Data model (Postgres) with stable identifiers

I used Postgres. The outbox table stores events plus the derived keys.

-- schema.sql
create table if not exists outbox_events (
  id bigserial primary key,

  -- business entity id; ordering is per entity
  entity_id text not null,

  -- logical event type, e.g. "OrderPaid"
  event_type text not null,

  -- stable, consumer-side idempotency key
  dedupe_key text not null unique,

  -- stable partition/routing key (string; broker will hash internally)
  partition_key text not null,

  -- stable ordering key (string; consumer can sort/compare)
  ordering_key text not null,

  payload jsonb not null,

  -- delivery state
  status text not null default 'pending', -- pending | published | failed
  attempts integer not null default 0,
  last_error text,

  created_at timestamptz not null default now(),
  published_at timestamptz
);

create index if not exists outbox_pending_idx
  on outbox_events(status, created_at);

How I compute keys

For correctness, the keys must be stable for each outbox row forever.

I generated:

dedupe_key = "{event_type}:{entity_id}:{id}"
(where id is the outbox row id; once the row exists, id is stable).
partition_key = entity_id
(all events for the same entity go to the same partition).
ordering_key = a string that increases with creation time + id:
"{created_at_epoch_ms}:{id}"
(simple to compare as strings only if zero-padded, so I used a numeric representation in practice; in code below I’ll keep it as an integer and serialize carefully).

This approach is particularly effective when your consumer processes per entity and broker ordering is partition-scoped.

End-to-end example (app writes + outbox publisher)

Below is a runnable example using:

psycopg for Postgres
a toy “broker” implemented as a table to show partition routing
a publisher that atomically claims rows, computes routing deterministically, and marks them published

1) Toy broker table (to visualize shard routing)

-- broker.sql
create table if not exists broker_messages (
  broker_partition integer not null,
  key text not null,
  dedupe_key text not null,
  ordering_key text not null,
  payload jsonb not null,
  created_at timestamptz not null default now()
);

create index if not exists broker_partition_created_idx
  on broker_messages(broker_partition, created_at);

In a real system, the broker would be Kafka or similar, and key would map to a partition via hashing. Here I’ll simulate hashing in the publisher.

2) Deterministic sharding + claiming outbox rows (Python)

# publisher.py
import hashlib
import json
import os
from dataclasses import dataclass
from typing import Optional

import psycopg
from psycopg.rows import dict_row


@dataclass
class OutboxEvent:
    id: int
    entity_id: str
    event_type: str
    payload: dict
    created_at: str  # ISO string from DB


def shard_for_key(partition_count: int, key: str) -> int:
    """
    Deterministic shard mapping.
    In Kafka, partitioning is typically done by hashing the key.
    I mimic that with SHA-256 for a stable result.
    """
    h = hashlib.sha256(key.encode("utf-8")).digest()
    num = int.from_bytes(h[:8], byteorder="big", signed=False)
    return num % partition_count


def compute_keys(outbox_id: int, entity_id: str, event_type: str, created_at_epoch_ms: int):
    dedupe_key = f"{event_type}:{entity_id}:{outbox_id}"
    partition_key = entity_id
    ordering_key = f"{created_at_epoch_ms:013d}:{outbox_id:020d}"  # lexicographically sortable
    return dedupe_key, partition_key, ordering_key


def publish_batch(conn, partition_count: int, limit: int = 20):
    """
    Claim rows atomically and publish them.
    The key requirement:
    - partition_key and ordering_key are derived deterministically from stable outbox fields
      (including the stable outbox row id).
    """
    with conn.cursor(row_factory=dict_row) as cur:
        # Claim: select pending events and lock them to avoid concurrent publishers.
        # status='pending' ensures idempotent behavior for retries.
        cur.execute(
            """
            with cte as (
              select id
              from outbox_events
              where status = 'pending'
              order by created_at asc
              limit %s
              for update skip locked
            )
            update outbox_events
            set status = 'publishing'
            where id in (select id from cte)
            returning id, entity_id, event_type, payload, created_at
            """,
            (limit,),
        )
        rows = cur.fetchall()

        if not rows:
            return 0

        published_count = 0

        for r in rows:
            event = OutboxEvent(
                id=r["id"],
                entity_id=r["entity_id"],
                event_type=r["event_type"],
                payload=r["payload"],
                created_at=r["created_at"].isoformat() if hasattr(r["created_at"], "isoformat") else str(r["created_at"]),
            )

            # created_at to epoch ms
            # psycopg returns datetime; handle both representations for robustness.
            # If it's datetime, use timestamps.
            if hasattr(r["created_at"], "timestamp"):
                epoch_ms = int(r["created_at"].timestamp() * 1000)
            else:
                # Fallback: parse ISO string
                from datetime import datetime, timezone
                dt = datetime.fromisoformat(str(r["created_at"]))
                if dt.tzinfo is None:
                    dt = dt.replace(tzinfo=timezone.utc)
                epoch_ms = int(dt.timestamp() * 1000)

            dedupe_key, partition_key, ordering_key = compute_keys(
                outbox_id=event.id,
                entity_id=event.entity_id,
                event_type=event.event_type,
                created_at_epoch_ms=epoch_ms,
            )

            broker_partition = shard_for_key(partition_count, partition_key)

            # Publish to toy broker
            cur.execute(
                """
                insert into broker_messages(broker_partition, key, dedupe_key, ordering_key, payload)
                values (%s, %s, %s, %s, %s)
                """,
                (
                    broker_partition,
                    partition_key,
                    dedupe_key,
                    ordering_key,
                    json.dumps(event.payload),
                ),
            )

            # Mark as published. If the publisher retries after crash,
            # status transitions keep behavior explicit.
            cur.execute(
                """
                update outbox_events
                set status = 'published',
                    attempts = attempts + 1,
                    published_at = now()
                where id = %s
                """,
                (event.id,),
            )

            published_count += 1

        return published_count


def main():
    dsn = os.environ.get("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/postgres")
    partition_count = int(os.environ.get("PARTITION_COUNT", "12"))

    with psycopg.connect(dsn) as conn:
        # Ensure table has the publishing status used in this example
        with conn.cursor() as cur:
            cur.execute("update outbox_events set status='pending' where status='publishing'")

        # Publish a batch
        published = publish_batch(conn, partition_count=partition_count, limit=50)
        print(f"Published {published} events.")


if __name__ == "__main__":
    main()

What the important blocks are doing (and why)

shard_for_key(...): deterministically maps a string key to a shard.
compute_keys(...): builds dedupe_key, partition_key, ordering_key from stable values, including the outbox row id.
In the DB claim:
- I FOR UPDATE SKIP LOCKED so multiple publishers can run without stepping on each other.
- I update status to publishing so a crash mid-batch won’t result in ambiguous ownership.
When publishing:
- The shard is computed from partition_key = entity_id, so retries always land on the same partition.
- The consumer dedupes by dedupe_key.

3) Consumer-side idempotency (dedupe by key, order within partition)

Here’s a simple consumer simulation that processes messages sorted by broker partition then by created_at. It ignores duplicates by dedupe_key.

# consumer.py
import hashlib
import os
from typing import Set

import psycopg
from psycopg.rows import dict_row

PROCESSED_TABLE = """
create table if not exists processed_events (
  dedupe_key text primary key,
  processed_at timestamptz not null default now()
);
"""


def consume_all(conn, limit: int = 1000):
    with conn.cursor() as cur:
        cur.execute(PROCESSED_TABLE)

    processed: Set[str] = set()

    with conn.cursor(row_factory=dict_row) as cur:
        cur.execute(
            """
            select broker_partition, key, dedupe_key, ordering_key, payload
            from broker_messages
            order by broker_partition asc, created_at asc
            limit %s
            """,
            (limit,),
        )
        messages = cur.fetchall()

        for m in messages:
            dedupe_key = m["dedupe_key"]

            # idempotency check
            if dedupe_key in processed:
                continue

            # store dedupe key; if it already exists, ignore (duplicate delivery)
            with conn.cursor() as cur2:
                cur2.execute(
                    """
                    insert into processed_events(dedupe_key)
                    values (%s)
                    on conflict do nothing
                    """,
                    (dedupe_key,),
                )

                # If inserted, it means this is the first time we see it
                cur2.execute("select 1 where exists (select 1 from processed_events where dedupe_key=%s)", (dedupe_key,))
            processed.add(dedupe_key)

    return len(processed)


def main():
    dsn = os.environ.get("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/postgres")
    with psycopg.connect(dsn) as conn:
        n = consume_all(conn, limit=5000)
        print(f"Processed {n} unique events.")


if __name__ == "__main__":
    main()

This consumer logic demonstrates the intended invariant:

duplicates may appear (at-least-once delivery),
but duplicates are ignored via dedupe_key,
and because retries keep the same partition_key, ordering for a given entity remains stable.

A tiny reproducible walkthrough

Step 1: Create schema

Run the SQL files (or paste them into your SQL client). The crucial part is that outbox_events exists.

Step 2: Insert sample outbox rows

# seed.py
import os
import json
from datetime import datetime, timezone

import psycopg

def main():
    dsn = os.environ.get("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/postgres")
    now = datetime.now(timezone.utc)

    payload1 = {"order_id": "A-1", "amount": 10}
    payload2 = {"order_id": "A-1", "amount": 20}

    with psycopg.connect(dsn) as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                insert into outbox_events(entity_id, event_type, dedupe_key, partition_key, ordering_key, payload)
                values (%s,%s,%s,%s,%s,%s)
                returning id, created_at
                """,
                ("order:A-1", "OrderCreated", "OrderCreated:order:A-1:tmp1", "order:A-1", "0000000000000:00000000000000000001", json.dumps(payload1)),
            )
            row1 = cur.fetchone()

            cur.execute(
                """
                insert into outbox_events(entity_id, event_type, dedupe_key, partition_key, ordering_key, payload)
                values (%s,%s,%s,%s,%s,%s)
                returning id, created_at
                """,
                ("order:A-1", "OrderUpdated", "OrderUpdated:order:A-1:tmp2", "order:A-1", "0000000000000:00000000000000000002", json.dumps(payload2)),
            )
            row2 = cur.fetchone()

            print("Seeded outbox rows:", row1, row2)

if __name__ == "__main__":
    main()

Note: In a real app, you’d compute dedupe_key, partition_key, and ordering_key at insertion time using the outbox row id. In this mini demo, the publisher computes the routing keys deterministically at publish time; the existing columns still allow the publisher to run the way it’s written.

Step 3: Publish twice (simulate retry)

Run:

python publisher.py
python publisher.py

With the deterministic routing (partition_key = entity_id) and consumer dedupe (dedupe_key), duplicates won’t cause repeated side effects.

Step 4: Consume

python consumer.py

You should see that the consumer reports processing only unique events (based on processed_events.dedupe_key).

Why this works under retries (the invariant)

The invariant I enforced was:

For any given outbox row id outbox_events.id = X, the tuple
(dedupe_key, partition_key, ordering_key) is identical across all retry attempts.
Therefore:
- duplicates hit the same dedupe_key (consumer ignores duplicates),
- duplicates hit the same broker partition (consumer sees consistent ordering within the partition).

This removes the “partition hopping” problem that caused my original out-of-order side effects.

Conclusion

I learned that outbox patterns are necessary but not always sufficient: the hidden system-design lever is how routing keys are chosen under retries. By making shard/partition routing deterministic using stable keys derived from outbox row data—and pairing that with consumer-side deduplication via a stable dedupe_key—I eliminated duplicate side effects and preserved ordering within a partition even when the publisher retried after failures.