Keeping Other Systems in Sync: Outbox and CDC

Scenario

Example app: when an order is placed, the order service saves it to Postgres and publishes OrderPlaced to Kafka; the search index, email service and warehouse consume it.

What you see in production

• About 1 in 20,000 orders never reaches the warehouse — the publish failed after the commit.
• A few warehouse shipments exist for orders that don't exist — the publish succeeded, then the database transaction rolled back.
• The search index occasionally shows stale order status.

The naive code

with transaction.atomic():
    order = Order.objects.create(...)
    producer.send("orders", {"type": "OrderPlaced", "id": order.id})   # inside: may publish then roll back
# or after commit: may commit then fail to publish

@Transactional
public Order place(Cart cart) {
    Order order = orders.save(Order.from(cart));
    kafka.send("orders", new OrderPlaced(order.getId()));   // not part of the DB transaction
    return order;
}

Reproduce it

Inject failures at the two dangerous points and compare a dual write with a transactional outbox plus relay:

def dual_write(fail):
    db, broker = [], []
    for order in (1, 2, 3):
        if fail == ("publish-then-rollback", order):
            broker.append(order); continue          # published, then the transaction rolled back
        db.append(order)
        if fail == ("commit-then-crash", order):
            continue                                # committed, then crashed before publishing
        broker.append(order)
    return db, broker

def outbox(fail):
    db, outbox_rows, broker = [], [], []
    for order in (1, 2, 3):
        if fail == ("publish-then-rollback", order):
            continue                                # order + outbox row roll back together
        db.append(order); outbox_rows.append(order) # one transaction
    for row in outbox_rows:                         # relay: retries until published (at-least-once)
        broker.append(row)
    return db, broker

for fail in (("publish-then-rollback", 2), ("commit-then-crash", 3)):
    for name, fn in (("dual write", dual_write), ("outbox", outbox)):
        db, broker = fn(fail)
        print(f"{fail[0]:22} {name:10} db={db} broker={broker} -> {'consistent' if db == broker else 'MISMATCH'}")

publish-then-rollback  dual write db=[1, 3] broker=[1, 2, 3] -> MISMATCH
publish-then-rollback  outbox     db=[1, 3] broker=[1, 3] -> consistent
commit-then-crash      dual write db=[1, 2, 3] broker=[1, 2] -> MISMATCH
commit-then-crash      outbox     db=[1, 2, 3] broker=[1, 2, 3] -> consistent

Root cause

The database and the broker are two systems without a shared transaction. Whatever order you write them in, a crash between the two leaves them inconsistent: lost events or phantom events. Retrying blindly creates duplicates.

The fix

Transactional outbox: write the event to an outbox table in the same transaction as the business change; a relay publishes outbox rows and marks them sent. Or use change data capture (CDC): stream the database's log (Postgres logical decoding, MySQL binlog) to the broker with a tool such as Debezium.

CREATE TABLE outbox (
  id bigserial PRIMARY KEY,
  aggregate_id bigint NOT NULL,
  type text NOT NULL,
  payload jsonb NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now(),
  published_at timestamptz
);
CREATE INDEX outbox_unpublished ON outbox (id) WHERE published_at IS NULL;

with transaction.atomic():
    order = Order.objects.create(...)
    Outbox.objects.create(aggregate_id=order.id, type="OrderPlaced", payload={"id": order.id})

def relay():                                     # separate worker; FOR UPDATE SKIP LOCKED lets several run
    with transaction.atomic():
        rows = Outbox.objects.select_for_update(skip_locked=True).filter(published_at=None).order_by("id")[:100]
        for row in rows:
            producer.send("orders", key=str(row.aggregate_id), value=row.payload).get(timeout=5)
            row.published_at = timezone.now()
            row.save(update_fields=["published_at"])

@Transactional
public Order place(Cart cart) {
    Order order = orders.save(Order.from(cart));
    outbox.save(OutboxEvent.of("OrderPlaced", order.getId(), toJson(order)));   // same transaction
    return order;
}
// Debezium's outbox event router can publish these rows via CDC instead of a polling relay.

Consumers must be idempotent (dedupe by event id — the inbox pattern), because relays publish at least once. This lesson builds on Idempotency & Retry-Safe Systems and Event Sourcing & CQRS.

Trade-offs

• Polling relay: simple, adds a little latency and load (index the unpublished rows, clean up old ones).
• CDC: low latency and no polling, but more infrastructure (Kafka Connect/Debezium, replication slots that must be monitored — an abandoned slot makes Postgres retain WAL indefinitely).
• Ordering is per aggregate key; use it as the message key.

Detect it in production

• Outbox backlog (unpublished rows, age of the oldest) and relay errors.
• Reconciliation: counts of orders vs OrderPlaced events per hour.
• Replication slot lag for CDC.

Key points

• Writing to a database and a broker separately loses or invents events on failures.
• Write events to an outbox in the same transaction; publish via a relay or CDC.
• Publishing is at-least-once: make consumers idempotent.