Delivery

Two knobs, not one enum

Timing and execution site are separate questions, and conflating them is how "put it on a queue" ends up meaning "and also change when it runs".

Knob 1 - timing and guarantee

Declared per receiver as mode=.

Mode	Runs	Can veto the change	Recoverable
INLINE	inside the firing transaction	yes, by raising	not needed: its failure is a rollback
ON_COMMIT	after commit, in the firing process	no	no
DURABLE (default)	after commit, at-least-once, retried	no	yes

INLINE needs no durability precisely because its failure mode is a rollback: if it raises, the change never committed, so nothing can be owed. That is also the only mode that may legitimately abort the caller's work.

ON_COMMIT is best effort and honest about it. A process that dies between commit and callback loses the delivery, with no row to say so.

DURABLE writes a row per receiver. One failing receiver must not replay or block the other four, which is why the debt is per-receiver rather than one outbox row per event. A receiver declared with targets= takes that one step further and writes a row per target, for the same reason.

Effectively once, for database-only receivers

For a receiver that touches only this database, the work and the acknowledgement commit together. The duplicate an at-least-once system owes you cannot be observed. Receivers with side effects outside the database - an email, a webhook - are at-least-once, as promised, and need their own idempotency.

Knob 2 - execution site

Meaningful only for DURABLE, because it is the only mode with a row to hand somewhere else.

• site="relay" (default) - the relay process runs the receiver.
• site="task" - the relay enqueues the delivery id, with the claim it holds, to the configured task backend and moves on. The task takes the row under that claim before running it, so a queue that delivers a message twice or late runs the receiver once.

A queue is only ever an answer to where. It is not a timing mode, and adopting one does not change what the event promised.

Eager delivery

@receiver(OrderPlaced, eager=True)
def notify(evt: OrderPlaced) -> None: ...

eager=True attempts the receiver immediately after commit, in the firing process, with the relay as the fallback. The row still exists; a failed eager attempt is simply owed to the relay like any other. It buys latency without giving up the crash story.

The relay

python manage.py deliver_events            # forever
python manage.py deliver_events --once     # one pass, for cron or CI
python manage.py deliver_events --limit 100 --worker-id box-1

The relay claims with SELECT ... FOR UPDATE SKIP LOCKED under a lease, so you can run as many as you like:

• Two workers never take the same row.
• A worker that dies without acknowledging has its rows reclaimed when the lease lapses.
• The acknowledgement is a compare-and-set on (claimed_by, claimed_at), so a worker whose lease already lapsed and was stolen cannot overwrite the new owner's result.
• A relay told to stop (SIGTERM or SIGINT) finishes the delivery in hand and hands back the rest of its batch. Another relay can claim those rows at once instead of waiting for the lease to lapse. See Stopping it.
• A relay whose database goes away stays up: it closes the dead connection, backs off and claims again. See When the database goes away.

A DURABLE receiver runs inside the transaction that carries its acknowledgement, so a call it makes outside the database holds that transaction open. See A receiver holds a transaction open for what that means behind pgbouncer and under idle_in_transaction_session_timeout, and Side effects inside the acknowledging transaction for why a call made there can happen twice.

SKIP LOCKED is Postgres and MySQL 8

SQLite has neither the statement nor the concurrency model that would make it meaningful. Declaration, INLINE, ON_COMMIT, fire(), the tables and drain_outbox() all work on every backend, so a SQLite test suite is fully supported - but running more than one relay is not.

Failure

A DURABLE receiver that raises is retried with exponential backoff and full jitter, up to max_attempts, then dead-lettered - unless it says otherwise, which is when the receiver knows better:

ceiling      = min(BACKOFF_BASE_SECONDS * 2 ** (attempt - 1), BACKOFF_CAP_SECONDS)
available_at = now + ceiling * random()

Full jitter draws from the whole window up to the ceiling rather than adding a little on top of it. Retrying a shared downstream at ceiling-plus-a-bit keeps every failed delivery in the same cohort, which is the thundering herd the backoff was meant to break up.

An async def receiver fails the same way, and a coroutine that is cancelled before it finishes counts as a failure: the attempt is spent and the row retried, with a RuntimeError naming the receiver as last_error.

The backoff is a floor, not a schedule. available_at is the earliest a retry can run, and the relay claims in order of available_at and then id, so a retry due in a minute also waits for every row that came due before it. With an empty queue that is the same thing; behind a backlog of twenty thousand rows, a retry due in a minute runs after the twenty thousand. There is deliberately no option to claim retries first: a destination failing on every attempt would then be retried ahead of every new event, and the new events would starve behind it. Work that must not queue behind another kind of work belongs in its own lane.

A curve per receiver

The two settings are the curve for every receiver. One that talks to a destination with its own idea of patience declares its own:

@receiver(OrderPlaced, backoff_base_seconds=120, backoff_cap_seconds=1800, max_attempts=10)
def email_receipt(evt: OrderPlaced) -> None: ...

Either may be declared alone, and the other comes from its setting. Declaring both with the cap below the base is refused at the decorator, because every retry would wait for the cap and the base would never be read.

The curve is read when an attempt fails, not copied onto the row, so a deploy that changes it applies to deliveries already in flight. max_attempts is the opposite on purpose: it is copied at fire time, so lowering it cannot dead-letter rows already owed.

Full jitter applies to a declared curve too, and it is easy to misread. A base of 60 does not mean "retry in a minute"; it means "retry somewhere in the next minute", and the first retry can land after a few seconds. The ceilings above are the longest each wait can be, and on average a wait is half its ceiling. So the span a retry budget covers is a distribution, not a number:

• the sum of the ceilings is the longest it can last;
• half of that is how long it lasts on average.

The curve above, with ten attempts, has nine waits with ceilings of 120, 240, 480, 960 and then 1800 seconds five times: at most three hours, an hour and a half on average. It is the curve the example shop gives its receipt mailer, and the demo there checks both numbers. Eight attempts on the default two-second base wait at most 254 seconds in all, so a mail provider down for an hour dead-letters every delivery sent into it.

A list of delays or a callable schedule is not offered, because the catalogue publishes the curve and could not publish either.

Dead is where a delivery stops on its own, not where it stops for good - see requeue.

Status	Means
pending	owed, waiting for available_at
claimed	leased by a worker
succeeded	ran, acknowledged
failed	raised with attempts remaining, or deferred
dead	out of attempts, or deferred past give_up_after
orphaned	addressed to a receiver key the registry no longer has
cancelled	withdrawn by cancel_deliveries before it ran

failed is distinct from pending so that "has this ever failed" is answerable without reading the attempt count.

Recording a failed attempt

A receiver cannot keep a record of its own failure by writing one: it runs inside the transaction that carries its acknowledgement, so everything it wrote is rolled back the moment it raises. on_failure is called afterwards, outside that transaction and after the delivery row is updated, so what it writes survives.

from django_domain_events import DeliveryFailure, receiver


def log_failure(failure: DeliveryFailure) -> None:
    DeliveryLog.objects.create(
        delivery_id=failure.delivery_id,
        attempt=failure.attempt,
        status=failure.status,
        error=failure.error,
    )


@receiver(OrderPlaced, on_failure=log_failure)
def notify_partner(evt: OrderPlaced) -> None: ...

It is called for failed and for dead, because "it failed again" and "it will not be tried again" are different things to record. DeliveryFailure carries the delivery's identity - delivery_id, event_id, event_name, receiver_key, and the target for a fan-out receiver - with the attempt, the status and the stored error, rather than the row itself, since another worker may own the row by the time the hook runs.

A worker whose lease lapsed does not call it: whoever holds the row now will report its own outcome. A hook that raises is logged and swallowed, because a failure path that fails leaves the operator a traceback about logging instead of about the delivery.

The failure log

Whether or not a receiver declares on_failure, every failed attempt the relay records is logged once, at WARNING, on the django_domain_events.delivery.deliver logger:

receiver shop.email_receipt failed delivery 4182 on attempt 3 with SMTPServerDisconnected; the row is failed

The line names the receiver, the delivery, the attempt, the exception's type and what the row is now - failed while it will be retried, dead once it will not. A deferral is logged the same way, and so is the give-up that ends one. There is no traceback: the message is in the row's last_error, and a receiver failing on every retry would otherwise write a stack each time. Nor is it rate limited - the attempt budget and the backoff already bound how often one delivery can fail.

A payload that will not decode is logged too, naming the decoder's exception, as the row goes straight to dead; the receiver never ran, so its on_failure is not told. A row whose receiver is no longer registered goes to orphaned and is deliberately not logged as a failed attempt: no attempt is counted, and no receiver failed.

It follows the same rule as the hook: a worker whose lease lapsed recorded nothing and logs nothing, so the log never shows an attempt the row does not. An attempt whose connection was lost before anything could be written is not logged either, because nothing recorded it; the lease lapsing is what puts it back in the queue. See Side effects inside the acknowledging transaction.

A hook around each delivery

Tracing, a log context, a tenant schema or a database role: anything that has to be in place for the length of one attempt, for every receiver, goes in the AROUND_DELIVERY setting rather than in each receiver. It names a callable that takes the attempt's DeliveryContext and returns a context manager:

# myproject/delivery_log.py
import logging
from collections.abc import Iterator
from contextlib import contextmanager
from contextvars import ContextVar

from django_domain_events import DeliveryContext

current_delivery: ContextVar[str] = ContextVar("current_delivery", default="-")


@contextmanager
def delivery_log_context(ctx: DeliveryContext) -> Iterator[None]:
    token = current_delivery.set(f"delivery={ctx.delivery_id} receiver={ctx.receiver_key}")
    try:
        yield
    finally:
        current_delivery.reset(token)


class DeliveryLogFilter(logging.Filter):
    """Put the ids on every line logged while a delivery runs."""

    def filter(self, record: logging.LogRecord) -> bool:
        record.delivery = current_delivery.get()
        return True

DJANGO_DOMAIN_EVENTS = {"AROUND_DELIVERY": "myproject.delivery_log.delivery_log_context"}

DeliveryContext.receiver_key is there for this: the hook is handed the context, not the receiver, and the key is what a span, a metric or a log line is labelled by. The path is read on every delivery and not cached, as CODEC is, so override_settings in a test takes effect at once. A path that does not import, or that names something not callable, fails the system check E011.

Where it sits. Around the attempt, outside the transaction that carries the acknowledgement. The receiver runs, the outcome is recorded, and only then is __exit__ called. So:

• __exit__ sees the receiver's exception, the same object, with its traceback, so a span can take its error status.
• Its return value is ignored. Returning True cannot suppress a failure, because the failure has already been recorded.
• An Exception from __exit__ is logged at ERROR and swallowed. It cannot undo an acknowledgement for work already done, or turn a recorded success into a failure that would run the work again. A BaseException from __exit__ propagates, and the outcome recorded before it stands.
• on_failure, and the failure log line above, run inside the hook, so a log context it sets is on those lines too.
• What the hook writes is not part of the receiver's transaction, so it is not rolled back when the receiver raises. If it has to commit with the receiver's work, it belongs in the receiver.
• The hook must not open a transaction. The outcome is written after the receiver's transaction closes and before __exit__ runs, so a transaction.atomic() the hook entered would hold that write, and exiting it with the receiver's exception would roll the recorded failure back after the log line and on_failure reported it. Set a tenant schema or role at session level (SET search_path, SET ROLE) and reset it on exit, rather than with SET LOCAL.
• Recorded is not always committed. Run from the relay or a task, the delivery's transaction is the outermost one and has committed by the time __exit__ runs. Run inside a transaction of your own - deliver_pending() under atomic() in a test, say - the outcome is written into that transaction, which is still open.
• A BaseException - KeyboardInterrupt, a task runner's timeout - reaches __exit__ and then propagates. The row stays claimed until its lease lapses, as it would without the hook.
• An async def receiver changes none of this. The hook is entered on the calling thread, before the coroutine starts, a ContextVar it sets is visible inside the coroutine, and __exit__ gets the coroutine's own exception.

Entering it fails the attempt if it raises. The receiver does not run, the attempt counts, the row records the hook's error and on_failure is told: read exactly as if the receiver had raised. That is deliberate. A hook that sets the tenant a receiver writes to must not let the receiver run without it, and a broken tracer dead-lettering loudly is recoverable with requeue_dead, where receivers writing to the wrong tenant are not. So an instrumentation hook should guard its own exporter calls. A path that no longer imports, or a factory returning something that is not a context manager, fails the attempt the same way. Because the verdict is read as a receiver's, a hook raising NotYet or RetryAfter on entry gets their semantics.

What it wraps. Durable deliveries, each attempt once, wherever they run: the relay, the eager attempt after commit, deliver_pending, drain_outbox, deliver_one, and the task worker for a site="task" receiver. The relay's hand-off of a site="task" row runs no receiver and is not wrapped, so that row is wrapped once, in the worker. What it never wraps: INLINE and ON_COMMIT receivers, which run in the caller's own request under whatever context it already has, and a row that never reaches its receiver - orphaned, undecodable, for an event no longer declared, not owed, lost to another worker, or a task message whose claim is stale. Each of those is logged already.

One callable. Several hooks compose in your own callable, with contextlib.ExitStack:

from contextlib import ExitStack


@contextmanager
def around_delivery(ctx: DeliveryContext) -> Iterator[None]:
    with ExitStack() as stack:
        stack.enter_context(delivery_log_context(ctx))
        stack.enter_context(tracing_span(ctx))
        yield

When the receiver knows better

The backoff curve is a guess, and the attempt budget is the only thing that ends a delivery on its own. Two exceptions let a durable receiver replace either with what it actually knows.

from django_domain_events import PermanentFailure, RetryAfter, receiver


@receiver(OrderPlaced)
def notify_partner(evt: OrderPlaced) -> None:
    response = post_to_partner(evt)
    if response.status_code == 410:
        raise PermanentFailure("410 Gone: the partner retired this endpoint")
    if response.status_code == 429:
        seconds = float(response.headers["Retry-After"])
        raise RetryAfter(seconds=seconds, reason=f"rate limited for {seconds:g}s")
    response.raise_for_status()

PermanentFailure dead-letters the row on the attempt that raised it. A destination that has said it will never accept another request is not asked four more times across the next hour. The row is dead, and on_failure is called with DEAD, exactly as it is when the budget runs out.

RetryAfter schedules the next attempt seconds from now instead of drawing from the curve. A rate limiter that says "in two minutes" is answering the question the curve is guessing at, and arriving early only earns another refusal.

By default, a requested retry consumes an attempt

Otherwise a destination answering 429 to every request would keep a row alive forever and nothing would ever declare it dead. Every delivery stays bounded by the budget it was fired with. A receiver expecting to be rate limited should declare a wider max_attempts - or defer without counting, bounded by time instead.

The request is capped at MAX_RECEIVER_RETRY_DELAY_SECONDS, a day by default, and a longer one is clamped with a warning naming both numbers. The cap is its own setting rather than BACKOFF_CAP_SECONDS, which bounds a curve: clamping an hour-long Retry-After to a backoff ceiling would hammer a destination that asked to be left alone.

Both are read by the relay and by nothing else. An INLINE receiver raising either fails the caller's transaction like any other exception, and an ON_COMMIT one has it logged: neither has a row to schedule or dead-letter. The type is the signal, never the message - a RuntimeError saying "410 Gone" is an ordinary failure.

A deferral that spends no attempt

A throttle or a quota is a limit on the destination, not a failure of the row. Counting it against max_attempts dead-letters good deliveries for being sent at a busy moment, and widening the budget to survive a burst also widens it for the failures it exists to end. RetryAfter(seconds, counts=False) says so:

from datetime import timedelta

from django_domain_events import RetryAfter, receiver


@receiver(ReceiptQueued, lane="mail", give_up_after=timedelta(days=2))
def send_receipt(evt: ReceiptQueued) -> None:
    try:
        provider.send(render_receipt(evt))
    except provider.Throttled as exc:
        raise RetryAfter(exc.retry_after, reason="sending rate exceeded", counts=False)
    except provider.DailyQuotaExceeded:
        raise RetryAfter(seconds_until_quota_resets(), reason="daily quota", counts=False)

Three things happen, and each answers a way a plain retry goes wrong in a burst of twenty thousand rows:

• The attempt is not counted. attempts is unchanged and max_attempts is not spent, so the row is still owed its whole budget for real failures. It is recorded as failed, with the message as last_error, and on_failure is called with FAILED and the attempt number the receiver's context carried - which the next run is given again, because nothing was spent.
• The next attempt is jittered upwards: between seconds and twice that from now, clamped to MAX_RECEIVER_RETRY_DELAY_SECONDS. Never earlier than the destination asked, and spread, so rows deferred together do not all come back in the same second and get throttled together again.
• The relay pauses the receiver's lane for seconds, and hands back the rest of the batch it had claimed rather than attempting it. Without that, learning that a daily quota is gone would cost one call per row: twenty thousand calls to hear the same answer twenty thousand times. With it, the relay that heard it spends no more calls on the lane until the delay has passed; the rows it gave back keep their place at the head of the queue and are attempted when the pause ends, one at a time, so if the quota is still gone it costs one call per pause, not one per row.

give_up_after keeps every delivery ending

A deferral that does not count cannot be ended by max_attempts, and a destination that throttles forever would keep its rows owed forever - holding their events past retention, since a prune waits for every delivery of an event. So it is bounded by time: the receiver declares give_up_after, and a deferral arriving once the row has been owed that long dead-letters it, exactly as a spent budget does - dead, completed_at, on_failure with DEAD, and a last_error naming the bound.

"Owed" is measured from when the row became owed: when it was written, or when replay or requeue last reopened it. Both reset that moment, so a month-old event replayed into a throttled destination is not dead-lettered on its first deferral for being a month old. Backoff never moves it, so deferring cannot keep resetting the clock. At exactly give_up_after the bound is reached.

The bound applies to deferrals alone. A receiver declaring it still has ordinary failures and counting RetryAfters ended by max_attempts, however long the row has been owed.

counts=False without give_up_after counts

A receiver raising counts=False that declares no give_up_after has nothing that would ever end its deliveries. The deferral is then treated as an ordinary RetryAfter - counted, no jitter, no pause - and a warning naming the receiver is logged once per process. Declare the bound to get the deferral.

The pause is per process

Nothing is shared between relays: each process pauses a lane when a deferral it ran says to, measured on its own clock, and every relay serving the lane learns of the throttle from its own first deferral - one call each. That is the price of having no shared state to keep consistent, and it is small next to one call per row. A relay serving the lane claims nothing while the pause lasts, and resumes at its first pass after it ends, so within POLL_SECONDS.

A relay serving every lane - run_relay(lane=None), or deliver_pending() and drain_outbox() from code - pauses only the deferring lane: its unstarted rows in the batch go back, the other lanes' rows in the same batch are still delivered, and later claims leave the paused lane out. deliver_pending() has no clock to wait on, so it sets the lane aside for the rest of the call. Neither the pause nor the hand-back reaches a row already given to a task backend, and a receiver running in a task defers there, in a process that holds no batch to pause; its deferral is recorded all the same. A direct deliver_one() likewise records and pauses nothing, and so does an eager=True receiver's attempt at commit: the firing process serves no lane, so in a burst every fire calls the throttled destination once before the relay's pause can apply.

For an endpoint's 429

django-outbound-webhooks raises RetryAfter for an endpoint answering 429. That is the case this exists for, and it can adopt counts=False with a give_up_after on its receiver, so a rate-limited endpoint stops costing deliveries their budget. A receiver declared with site="task" keeps the unspent attempt and loses the pause, for the reason above.

When the work is not due yet

RetryAfter is a destination saying when to come back. NotYet(until) is the receiver itself saying its work is not due until until, an aware datetime in the future - typically a scheduled receiver whose due time has moved since its row was written:

from django.utils import timezone

from django_domain_events import DeliveryContext, NotYet, receiver


@receiver(OfferMade, due=timedelta(hours=48), takes_context=True)
def expire_offer(evt: OfferMade, ctx: DeliveryContext) -> None:
    offer = Offer.objects.get(pk=evt.offer_id)
    if offer.expires_at > max(ctx.due_at, timezone.now()):
        raise NotYet(offer.expires_at, reason="the offer was extended")
    offer.expire()

The relay moves the row's available_at and due_at to until and sets it back to pending. No attempt is spent, nothing is capped or jittered, give_up_after does not apply, last_error and on_failure are left alone, and no lane is paused: it is the application's own statement of when its work is due, not a destination's advice to be distrusted. On the next run, ctx.due_at is until. A past until is refused when the exception is built, and so is a naive one unless the project runs with USE_TZ = False, so raising one records an ordinary, counted failure instead. Scheduled reactions covers it in full, with how it compares to RetryAfter.

Like the other two, it is read by the relay and by nothing else: an INLINE receiver raising it fails the caller's transaction, and an ON_COMMIT one has it logged.

In tests

from django_domain_events import assert_fired, drain_outbox


def test_placing_an_order_reserves_stock():
    place_order()
    assert_fired(OrderPlaced, times=1)
    drain_outbox()

drain_outbox() runs the real delivery path to completion - the same claim, encode, decode and acknowledgement the relay performs - so a test exercising it exercises production. assert_fired reads the log rather than a mock, which means it also proves the row was written inside the transaction that committed.

It skips the backoff, so a failing receiver is retried to its dead letter in one call, and never skips a receiver's due=: a scheduled row is left alone until it is due. Pass now= to move the claims forward in time instead:

def test_an_unanswered_offer_expires():
    make_offer()
    drain_outbox()  # the expiry is not due yet, so it is left alone
    drain_outbox(now=timezone.now() + timedelta(hours=48))  # now it is

Only the claims read now; the receiver runs on the real clock, so a scheduled receiver should compare against ctx.due_at, with the clock only as the floor that lets a late row proceed (max(ctx.due_at, timezone.now())). See scheduled reactions.