Week 10: Background Work & Messaging

1. Celery Tasks & RabbitMQ

Some work genuinely doesn't need to finish before you respond to the client — sending a welcome email, generating a report, calling a slow third-party API. Doing that work inline, inside the request handler, ties your API's latency and availability to however long (and however reliably) that work runs. A task queue decouples the two: the request handler just publishes a message describing the work and returns immediately, while a separate process actually does it.

Celery splits this into three roles. The broker — RabbitMQ here — is a durable message queue that holds pending task messages until a worker is ready for them. The worker is a separate, long-running process (or several) that consumes messages from the broker and executes the matching task code. The result backend (often Redis) is an optional store for a task's return value and status, needed only if something later wants to poll for a result — a fire-and-forget task like sending an email usually skips it entirely.

from celery import Celery

celery = Celery("app", broker="amqp://guest:guest@rabbitmq:5672//")

@celery.task
def send_welcome_email(user_id: int) -> None:
    mailer.send_welcome(user_id)

from fastapi import APIRouter, Depends
from app.tasks import send_welcome_email

router = APIRouter()

@router.post("/users", status_code=201)
def create_user(payload: UserCreate, session: Session = Depends(get_session)) -> UserRead:
    user = create_user_record(session, payload)
    send_welcome_email.delay(user.id)   # publish and return -- don't wait for it
    return user

.delay(user.id) serializes the task's name and arguments into a message, publishes it to RabbitMQ, and returns an AsyncResult handle immediately — it does not wait for a worker to pick it up, let alone finish. The HTTP response goes out before the email has even been attempted, which is exactly the decoupling you want. Contrast that with calling send_welcome_email(user.id) directly (no .delay), which would just run the function synchronously in the request, defeating the entire point. The worker itself runs as its own process, separate from the FastAPI app: celery -A app.tasks worker --loglevel=info.

2. Delivery Guarantees & Idempotent Consumers

Message delivery comes in three theoretical flavors. At-most-once means a message is delivered zero or one times — simple, but messages can be lost outright (e.g., the broker marks a message delivered right when the worker picks it up, then the worker crashes before finishing; the message is gone). At-least-once means a message may be delivered — and processed — more than once, because the broker only removes it once it's confirmed processed, and will redeliver if that confirmation never arrives (say, the worker crashes after doing the work but before acknowledging it). Exactly-once is the one everyone wants and, in a real distributed system, effectively doesn't exist: guaranteeing it would require atomically coordinating "the message was delivered" with "the side effect happened," across a network, with crashes possible at any point in between — coordination overhead that's rarely worth paying for.

The pragmatic default is at-least-once delivery — configure Celery with task_acks_late = True so RabbitMQ only removes a message once the task has actually finished, not the moment a worker starts it — paired with an idempotent consumer: a task written so that processing the same message twice produces the same end state as processing it once.

class TemporaryError(Exception):
    """Raised for failures worth retrying, e.g. an SMTP timeout."""

@celery.task(bind=True, autoretry_for=(TemporaryError,), retry_backoff=True, max_retries=5)
def send_welcome_email(self, user_id: int):
    if email_log.already_sent(user_id):
        return
    try:
        mailer.send_welcome(user_id)
    except SMTPTimeout as exc:
        raise TemporaryError(str(exc)) from exc
    email_log.mark_sent(user_id)

bind=True gives the task access to self, which Celery needs internally to manage retries. autoretry_for=(TemporaryError,) combined with retry_backoff=True means Celery automatically catches a raised TemporaryError and re-queues the task with exponentially increasing delay, up to max_retries=5 — you don't write the retry loop yourself. The actual idempotency check is email_log.already_sent(user_id): a durable record, in the database, that survives worker restarts and persists across retries, checked before mailer.send_welcome() runs. If this message gets redelivered — a worker crash after sending but before mark_sent(), for example — the second delivery sees already_sent return True and exits as a no-op instead of sending a second email.

3. Retries, Dead-Letter Handling & the Outbox Pattern

When a task exhausts max_retries, Celery raises MaxRetriesExceededError and the task ends up in a FAILURE state. In production, failed or rejected messages should be routed to a dead-letter queue (DLQ) — a separate queue, configured on RabbitMQ via a dead-letter exchange, that captures messages which were rejected, expired, or ran out of retries — so they're visible for a human or an alert to inspect, rather than silently vanishing after the last retry.

There's a subtler reliability problem underneath all of this: you often need to write a row to your database and publish an event about it (e.g., "task created") — but a database commit and a broker publish are two separate network calls to two separate systems, and you can't make them atomic together. If your process crashes between them, you can end up having written the row but never published the event, or published the event for a write that then rolled back. The transactional outbox pattern solves this by writing the event into an outbox table in the same database transaction as the business write — so both commit together or neither does — and having a separate relay process read unpublished outbox rows and publish them afterward.

def create_task(session: Session, payload: TaskCreate) -> Task:
    task = Task(title=payload.title, owner_id=payload.owner_id)
    session.add(task)
    session.flush()   # task.id is now populated

    session.add(OutboxEvent(
        event_type="task.created",
        payload={"task_id": task.id, "title": task.title},
    ))
    session.commit()   # task row + outbox row commit atomically, or neither does
    return task

def relay_outbox_events(session: Session, publisher: EventPublisher) -> None:
    pending = session.query(OutboxEvent).filter_by(published_at=None).all()
    for event in pending:
        publisher.publish(event.event_type, event.payload)
        event.published_at = datetime.utcnow()
        session.commit()

The relay's worst case is publishing an event twice — a crash after publisher.publish() but before the commit that marks it published would cause exactly that on the next relay run. That's an acceptable failure mode precisely because the consumer on the other end (Section 2) is already idempotent, which is why the outbox pattern and idempotent consumers are usually adopted together rather than one without the other.

4. Hands-on Exercise

5. Knowledge Check

Four quick questions. Expand each to check your answer.