Use the Transactional Outbox Pattern With Django and Celery

Problem statement

A view creates an order and enqueues send_confirmation.delay(order.pk). Sometimes the task runs before the transaction commits and fails with DoesNotExist. You switch to transaction.on_commit(), which fixes that, but now if the broker is unreachable at commit time, or the process dies right after commit, the email is never sent, with no record that it was supposed to be. When side effects must not be lost (payments, inventory sync, outbound webhooks), you need the intent to send stored in the same transaction as the data: the transactional outbox.

1. For most tasks, transaction.on_commit(lambda: task.delay(pk)) is enough and avoids the "task runs too early" bug.
2. When losing a side effect is unacceptable, insert an OutboxMessage row in the same transaction as the business change.
3. A relay process polls the outbox with SELECT ... FOR UPDATE SKIP LOCKED, publishes each message to Celery (or a broker), and marks it sent.
4. Consumers must be idempotent, because the relay can publish a message more than once.
5. Delete or archive sent outbox rows on a schedule.

Step-by-step solution

1. Know where on_commit stops

with transaction.atomic():
    order = Order.objects.create(...)
    transaction.on_commit(lambda: send_confirmation.delay(order.pk))

This guarantees the task never sees uncommitted data. It does not guarantee the task is enqueued: delay() runs after commit, in the same process, and can fail (broker down, network blip, process killed). The order exists; the email intent is gone.

2. Write an outbox row atomically

class OutboxMessage(models.Model):
    topic = models.CharField(max_length=100)          # e.g. "orders.send_confirmation"
    payload = models.JSONField()
    created_at = models.DateTimeField(auto_now_add=True)
    sent_at = models.DateTimeField(null=True, db_index=True)
    attempts = models.PositiveIntegerField(default=0)

    class Meta:
        indexes = [models.Index(fields=["created_at"], name="outbox_unsent", condition=models.Q(sent_at__isnull=True))]

with transaction.atomic():
    order = Order.objects.create(...)
    OutboxMessage.objects.create(topic="orders.send_confirmation", payload={"order_id": order.pk})

Either both rows commit or neither does. The intent is now durable.

3. Relay messages to Celery

A small worker (a management command under systemd, or a Celery beat task running frequently) publishes unsent messages:

from django.db import transaction
from django.utils import timezone

HANDLERS = {"orders.send_confirmation": send_confirmation}

def relay_batch(limit=100):
    with transaction.atomic():
        batch = list(OutboxMessage.objects
                     .select_for_update(skip_locked=True)
                     .filter(sent_at__isnull=True)
                     .order_by("created_at")[:limit])
        for msg in batch:
            HANDLERS[msg.topic].delay(**msg.payload)
            msg.sent_at = timezone.now()
            msg.attempts += 1
        OutboxMessage.objects.bulk_update(batch, ["sent_at", "attempts"])
    return len(batch)

skip_locked lets several relay processes run without taking the same rows. If publishing fails, the transaction rolls back and the messages stay unsent for the next run. If the process dies after publishing but before commit, messages are published again later, which is why consumers must be idempotent.

4. Make consumers idempotent

@shared_task
def send_confirmation(order_id):
    order = Order.objects.get(pk=order_id)
    if order.confirmation_sent_at:
        return
    send_email(order)
    Order.objects.filter(pk=order_id, confirmation_sent_at__isnull=True).update(confirmation_sent_at=timezone.now())

For external APIs, pass an idempotency key derived from the outbox message ID.

5. Clean up and monitor

Delete sent rows older than a few days in batches. Alert when the oldest unsent message is older than a threshold (for example 5 minutes), which means the relay is down or failing.

Explanation

The outbox pattern solves the dual-write problem: you cannot atomically commit to a database and publish to a separate broker. By writing the message to the database in the same transaction as the data, durability and consistency come from the database. A relay turns stored messages into broker messages with at-least-once semantics, and idempotent consumers turn at-least-once into effectively-once. transaction.on_commit covers the more common problem of tasks running too early but cannot make the publish itself durable.

Edge cases / notes

• Latency. Relays polling every second add up to a second of delay; LISTEN/NOTIFY can wake the relay immediately.
• Ordering. Process messages in created_at order per aggregate if consumers depend on order; otherwise allow parallelism.
• Volume. For very high message rates, change data capture (Debezium) reading the outbox table can replace polling.

Next steps

• Make consumers safe with Make Celery Tasks Idempotent With Retries.
• Get outbox-style guarantees from a queue with Run Procrastinate as a Postgres Task Queue.
• Handle the inbound direction with Process Incoming Webhooks Reliably in Django.

FAQ

Is transaction.on_commit enough?

For most tasks, yes. It prevents tasks from running before commit, but the enqueue can still fail after commit.

What problem does the outbox pattern solve?

Atomically saving data and the intent to trigger a side effect, so neither happens without the other.

Can the relay send a message twice?

Yes, if it crashes after publishing but before marking the row sent. Consumers must be idempotent.

Do I need a separate broker with an outbox?

No. You can relay to Celery, or use a Postgres-based queue that combines both roles.