Soumabrata

1. The Problem: The Async Fire-and-Forget

The original gateway handler followed a straightforward pattern: accept the webhook, validate it, persist the event to PostgreSQL, then fire off an async task to publish to Kafka.

# Original code (simplified)
db.add(event)
await db.commit()

async def _publish():
    producer = get_kafka()
    if producer is None:
        return
    try:
        await producer.send_and_wait(...)
        # update status to 'queued'
    except Exception:
        # update status to 'failed'

asyncio.create_task(_publish())  # 🚨 fire-and-forget

At a glance this looks fine. The event is safely in PostgreSQL. The Kafka publish happens asynchronously. The 202 Accepted response is not blocked. But there's a critical vulnerability:

The event is persisted as pending. The background task never runs. Kafka never receives the message. The destination never gets the webhook. And nobody knows — the event sits in the database with a status of pending forever.

2. What Could Go Wrong

Four distinct failure modes, ordered by severity:

The fundamental problem is the dual-write pattern: one write to PostgreSQL, one write to Kafka, coordinated only by application code in a single process. If that process dies between the two writes, one of them is lost.

3. The Transactional Outbox Pattern

The outbox pattern eliminates the dual-write by making the Kafka publish someone else's problem.

Before: Gateway Owns Both Writes

After: Gateway Only Writes to DB

4. Implementation

The OutboxRecord Model

A single new table tracks all events waiting to be published:

class OutboxRecord(Base):
    __tablename__ = "outbox_records"

    id: Mapped[uuid.UUID]       # PK
    event_id: Mapped[uuid.UUID]  # FK → events.id
    status: Mapped[str]          # pending | completed | failed
    publish_key: Mapped[str]      # endpoint_id (kafka key)
    publish_topic: Mapped[str]    # kafka topic
    attempts: Mapped[int]         # retry counter
    last_error: Mapped[str | None] # last failure message
    created_at: Mapped[datetime]
    processed_at: Mapped[datetime | None]

The Gateway Handler (After)

The refactored handler is simpler and safer — no Kafka imports, no background tasks:

# Refactored handler — single transaction, zero Kafka dependency
event_id = uuid.uuid4()
event = Event(
    id=event_id,                          # explicit! see §5
    endpoint_id=endpoint.id,
    request_body=payload,
    request_headers=headers,
    status="pending",
)
db.add(event)

outbox = OutboxRecord(
    event_id=event_id,
    publish_key=str(endpoint.id),
    publish_topic=settings.kafka_topic_raw_events,
)
db.add(outbox)
await db.commit()

The Event and OutboxRecord are created in the same database transaction. Either both are persisted, or neither is.

5. The event.id Gotcha

This one cost us a NOT NULL constraint violation on the first run. The foreign key from outbox_records.event_id to events.id was being set to NULL.

# ❌ BROKEN — SQLAlchemy default=uuid.uuid4 fires at flush, not construction
event = Event(
    endpoint_id=...,
    request_body=...,
)
db.add(event)                       # event.id is None here
outbox = OutboxRecord(
    event_id=event.id,                # 🔥 FK = NULL → constraint violation
)

SQLAlchemy's default=uuid.uuid4 on the column definition is only invoked at flush time, not at object construction. When OutboxRecord(event_id=event.id) runs, event.id is None.

# ✅ FIXED — generate the ID explicitly before construction
event_id = uuid.uuid4()             # now event_id is available …
event = Event(
    id=event_id,                      # … before the Event object exists
    endpoint_id=...,
    request_body=...,
)
db.add(event)
outbox = OutboxRecord(
    event_id=event_id,                # ✅ valid FK
)

Rule of thumb: When you need an object's PK before it's flushed, generate the UUID explicitly. Don't rely on SQLAlchemy's column default.

6. The Outbox Relay Worker

The relay is a standalone process that runs in a tight loop:

# Poll loop (simplified)
while not _stop.is_set():
    async with async_session_factory() as db:
        result = await db.execute(
            select(OutboxRecord)
            .where(OutboxRecord.status == "pending")
            .order_by(OutboxRecord.created_at)
            .limit(BATCH_SIZE)                           # ← 10
            .with_for_update(skip_locked=True)           # ← key detail
        )
        records = result.scalars().all()

        for record in records:
            try:
                await _publish(record)
                record.status = "completed"
            except Exception as exc:
                record.attempts += 1
                record.last_error = f"{type(exc).__name__}: {exc}"
                if record.attempts >= 10:
                    record.status = "failed"            # give up

        await db.commit()

Key Design Decisions

Kafka Unavailability

If Kafka is down at relay startup, init_kafka() returns None. The relay exits its poll loop early:

producer = await init_kafka()
if producer is None:
    logger.warning("Kafka unavailable — outbox records will queue")
    return

Records stay at pending in PostgreSQL. When Kafka comes back, a container restart picks them up. No data loss.

7. Verifying Resilience

We tested three crash scenarios against the live system:

Scenario A: Normal Flow

curl -X POST /hooks/<endpoint_id> \
  -H "X-Hub-Signature-256: sha256=<sig>" \
  -d '{"action":"push"}'

# → {"status": "accepted", "event_id": "0df9e4db-..."}

# Outbox record:
 outbox_records: "completed"
 delivery_worker: delivered to collector → 200 (39ms)

Scenario B: Kill Outbox Relay Mid-Stream

docker stop outbox-relay

# Send event while relay is down
curl -X POST /hooks/<endpoint_id> ...
# → {"status": "accepted", "event_id": "4e28b9ca-..."}

# Outbox record stays pending:
 outbox_records: "pending"   ← queued in DB

docker start outbox-relay

# After 1-2 seconds:
 outbox_records: "completed"   ← drained
 delivery_worker: delivered → 200 (11ms)

Scenario C: Kafka Crash at Relay Start

# Kafka goes down → relay starts → init_kafka() returns None
# → relay exits → records stay pending

# Kafka comes back → restart relay
# → picks up pending records → publishes → marks completed

8. The Cost Question

"Isn't a DB table costly if the service goes boom?"

The outbox table is the cheapest insurance policy you can buy. Here's the breakdown:

Compare this to the cost of losing events:

• Stripe webhook missed → customer not notified of payment failure → chargeback fees
• GitHub push event lost → CI pipeline doesn't run → production bug deployed
• Slack alert dropped → on-call engineer not paged → incident response time blows up

The outbox pattern trades ~200 bytes of disk and one extra process for provable at-least-once delivery guarantees. It's not costly — it's insurance.

9. Lessons Learned

1. Fire-and-forget is a fire hazard. Any asyncio.create_task that touches external infrastructure is a gamble. The process can die between the DB commit and the network call. The outbox pattern makes this gap disappear.
2. Dual-write is the root of most distributed systems bugs. If you need to write to two systems atomically, you need either a distributed transaction (expensive, fragile) or an outbox (simple, proven). Choose the outbox.
3. FOR UPDATE SKIP LOCKED is a superpower. PostgreSQL 9.5+ lets you implement a reliable work queue with zero infrastructure. No Redis streams, no RabbitMQ — just a SQL query that's safe for concurrent consumers.
4. SQLAlchemy defaults are deceptive. default=uuid.uuid4 on a mapped column does NOT set the attribute at construction time. If you need the PK before the object is flushed, generate it yourself.
5. Test the crash, not just the happy path. Kill containers mid-operation. Verify events queue in the database. Restart and watch them drain. A system that works perfectly when everything is up is not a production system.
6. Decoupling is not free — but it's cheaper than recovery. The outbox table and relay worker added ~300 lines of code and one container. The alternative — recovering lost events from database logs or customer complaints — is infinitely more expensive.

The full source code is available on GitHub. The outbox relay lives in workers/outbox_relay.py (112 lines).