beginnerintermediateadvanced

Django · Section 63

Celery

Level: advanced
Read: 44 min
Concepts: 5

A Celery deployment is an app object created in `config/celery.py` and imported from `config/__init__.py`, tasks registered with `@shared_task`, and worker processes that consume messages — and the thing to internalise first is that calling a task does not run it. `.delay()` serialises the name and arguments, publishes a message, and returns; the body executes later in a different process with its own imports, settings and database connections, which is why you pass ids rather than instances and why `autodiscover_tasks()` importing `tasks.py` specifically decides whether your task exists in the worker at all. Broker and result backend are two services with two meanings: without the broker work never happens, while without the backend the work runs and you simply cannot see the outcome — so `task_ignore_result = True` with per-task opt-in matches how most tasks are actually used. Retries belong only on failures that could succeed unchanged, which makes `autoretry_for=(Exception,)` a way of turning every bug into a slow one; `retry_backoff` with jitter is what stops synchronised retries becoming a thundering herd, and Beat publishes periodic tasks and must run exactly once. Everything lands on a single `celery` queue by default, so splitting queues by *duration* — with a worker, a concurrency and a pair of time limits each — is what stops a transcode delaying a password reset, and only the soft limit gives a hung task the chance to mark itself failed. Finally, Celery acknowledges before executing by default, so a killed worker loses the task silently; `acks_late=True` converts that into redelivery, which the docs note requires an idempotent task, and since nobody is waiting on a task, alerting belongs on queue depth, failure rate and last-success age rather than on exceptions.

Django overview

Python Django Full Course for Beginners | Complete All-in-One Tutorial | 3 HoursDave Gray

What is true here

.delay() publishes a message and returns — the body runs later, in a process that shares nothing with the request.
Autodiscovery imports tasks.py only; a task defined elsewhere is NotRegistered in the worker.
The broker is required, the result backend is optional, and storing unread results is a write per task.
Retry only transient failures; split queues by duration and give each its own worker and time limits.
acks_late=True trades silent loss for redelivery and therefore requires idempotency.

What you will be able to do

Wire Celery into a Django project so tasks are discovered and settings are shared
Choose broker and result-backend configuration from what each failure would mean
Write a retry policy that distinguishes transient failures from bugs
Split queues so slow work cannot starve fast work, with limits that fail visibly
Make a task safe to run twice, and alert on the signals nobody is waiting for

A task from publish to acknowledgement, and everything that can go wrong on the way

Web process · task.delay(id)

serialises name + args, returns immediately

transaction.on_commit()

or the worker races the COMMIT and finds no row

task_routes → a queue

by duration: default · slow · maintenance

Broker

its own Redis database — cache.clear() would FLUSHDB it

No worker with -Q for that queue

no error; the messages simply accumulate

Worker reserves the message

acks_late: not acknowledged yet

Registry lookup by name

populated from tasks.py by autodiscover_tasks()

NotRegistered

Task body runs

soft_time_limit raises inside; time_limit kills the child

Transient failure → retry with backoff

only exceptions that could succeed unchanged

Worker killed mid-task

unacknowledged → redelivered → must be idempotent

Acknowledged

Depth · failure rate · last-success age

the only things that notice any of this

App, tasks, and workers

The three pieces, how discovery works, and why calling a task does not run it.

The Celery app, tasks, workers, and task discovery

coreintermediate

A Celery deployment has three pieces. The **app** is a `Celery(...)` instance created once in `config/celery.py` and imported from `__init__.py` so it exists before any task is defined. A **task** is a function registered with that app — in a Django project you use `@shared_task`, which registers with whichever app is configured rather than binding to one instance, so reusable apps work. A **worker** is a separate process (`celery -A config worker`) that consumes messages and runs those functions; it is not part of your web server and does not share its memory. **Task discovery** is `app.autodiscover_tasks()`, which imports `tasks.py` from every entry in `INSTALLED_APPS` — which is why a task in a file with any other name is invisible.

Think of it as

The single most useful thing to internalise is that calling a task does not run it. `send_email.delay(pk)` serialises the name and the arguments, publishes a message, and returns immediately — the function body executes later, in a different process, possibly on a different machine, possibly after your web process has been replaced. Everything else follows from that separation. The worker has its own imports, so a task must be importable from the worker's environment; it has its own settings load, so anything read at import time is read there too; and it shares nothing in memory with the request that enqueued it, so state has to travel through the message or through the database. `@shared_task` versus `@app.task` is a smaller decision than it looks: `@shared_task` avoids importing the app object into every module, which keeps reusable apps portable, and in a Django project it is the default choice for that reason alone. Autodiscovery is where most "my task is not registered" confusion comes from — it looks in `tasks.py` specifically, so a task defined in `services.py` exists in your web process (which imported it) and does not exist in the worker (which did not).

python

@shared_task
def send_welcome_email(user_id):
    ...

send_welcome_email.delay(user.pk)     # publishes a message; does not run anything

What we're doing: The three files every Django + Celery project needs, wired so autodiscovery actually finds the tasks.

config/celery.py + config/__init__.py + accounts/tasks.pypython

# config/celery.py
import os
from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")

app = Celery("config")
# Every CELERY_* setting in Django settings becomes a Celery setting.
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()          # imports <app>/tasks.py for each INSTALLED_APPS entry


# config/__init__.py
from .celery import app as celery_app

__all__ = ("celery_app",)


# accounts/tasks.py           <- the filename matters; services.py is NOT discovered
from celery import shared_task


@shared_task
def send_welcome_email(user_id):
    user = User.objects.get(pk=user_id)
    send_mail("Welcome", render_welcome(user), FROM, [user.email])

5: Set before the app is created: the worker starts without `manage.py`, so nothing else points it at your settings module.
8–9: The `CELERY` namespace means `CELERY_BROKER_URL` in Django settings becomes `broker_url` in Celery — one settings file rather than two.
10: Autodiscovery looks for `tasks.py` specifically. This one line is why a task defined in `services.py` runs fine in a test and raises `NotRegistered` in production.
13–16: Importing the app from `__init__.py` guarantees it exists before Django imports any application module, so `@shared_task` has something to register against.
23–26: The task takes an id and reads the row itself — the worker has its own database connection and should see current state, not a snapshot from the message.

Why this works: These three files are almost entirely boilerplate, and every one of the common Celery-in-Django failures — settings not found, tasks not registered, app imported too late — is caused by one of them being wrong.

Defining a task outside `tasks.py`

Wrong

python

# accounts/services.py
@shared_task
def send_welcome_email(user_id):
    ...
# Works in tests (the module gets imported) and raises NotRegistered in the worker.

Better

python

# accounts/tasks.py
from .services import build_welcome_email

@shared_task
def send_welcome_email(user_id):
    build_welcome_email(user_id)

What you see: `celery.exceptions.NotRegistered: accounts.services.send_welcome_email` in the worker log, while the web process enqueues happily and every test passes — because the test suite imported the module and the worker never did.

Why: `autodiscover_tasks()` imports a fixed filename from each installed app, so registration depends on where the function lives rather than on the decorator. The web process registers it as a side effect of importing `services.py` for something else; the worker imports only `tasks.py`. Keeping tasks in `tasks.py` as thin wrappers over ordinary functions elsewhere gives you discovery and keeps the logic testable without Celery.

What happens between .delay() and the function body running

View calls send_email.delay(57)

returns in microseconds — nothing has run

Serialize name + args

JSON by default, so arguments must be JSON-safe

Broker (Redis / RabbitMQ)

the message waits here — possibly across a deploy

Worker process

separate imports, separate settings, separate connections

Task registry lookup by name

populated by autodiscover_tasks() from tasks.py only

The function body finally runs

NotRegistered

the task exists in the web process and not in the worker

Result backend (optional)

only if configured — otherwise the return value is discarded

The three pieces, and where each one lives

The three pieces, and where each one lives
Piece	Where	Created by	Fails as
App	`config/celery.py`	`Celery("config")` + `config.from_object`	tasks register against no app
Task	`<app>/tasks.py`	`@shared_task`	`NotRegistered` in the worker
Worker	its own process	`celery -A config worker`	messages queue up unconsumed
Discovery	`config/celery.py`	`app.autodiscover_tasks()`	silently skips non-`tasks.py` files

Together

bash

celery -A config worker -l info
# [tasks]
#   . accounts.tasks.send_welcome_email
#   . reports.tasks.build_report

Remember: The app is created in `config/celery.py` and imported from `config/__init__.py` so it exists before any task is defined. `@shared_task` registers with the current app, which is what makes reusable apps work. `.delay()` publishes a message and returns — the body runs later, in a separate process with its own imports, settings and connections, so pass ids and read current state. And autodiscovery imports `tasks.py` specifically: a task defined anywhere else works in your web process and raises `NotRegistered` in the worker.

Brokers and result backends

Two services, two failure meanings — and why storing every result is usually wrong.

Brokers and result backends

coreintermediate

These are two separate services doing two unrelated jobs, and they are constantly confused because they are usually configured to the same Redis. The **broker** carries messages from your web process to a worker — it is required, and it is where a task lives between `.delay()` and execution. The **result backend** stores what a task *returned*, so somebody can ask for it later; it is optional, and if you never call `.get()` on an `AsyncResult` you do not need one. Storing results you never read is pure cost: every task writes a row or a key that expires on a timer, which on a busy queue is a substantial amount of traffic for data nobody looks at.

Think of it as

Ask what happens if each one is lost, and the difference becomes obvious. Losing the broker means work never happens — the enqueue fails, or messages sit unconsumed. Losing the result backend means you cannot find out what happened, while the work itself completed normally. That asymmetry is why the broker deserves durability and the result backend usually does not. It also explains the standard advice to set `task_ignore_result = True` globally and opt in per task: most tasks are fired for their side effect, not their return value, so the default of storing everything is backwards for the common case. On broker choice, Redis and RabbitMQ differ mainly in what they guarantee when things go wrong. RabbitMQ is a real message broker with acknowledgements, durable queues and dead-lettering built in; Redis is a data structure server being used as a queue, which is entirely workable and is what most Django projects already have running — the trade is that Redis persistence is weaker and its dead-letter story is something you build. The genuinely important operational rule is smaller and easier: give the broker its own Redis database, because a `cache.clear()` pointed at the same one runs `FLUSHDB` and deletes every queued task.

python

CELERY_BROKER_URL = env("REDIS_URL") + "/0"
CELERY_TASK_IGNORE_RESULT = True

@shared_task(ignore_result=False)      # opt in only where a result is read
def build_report(job_id): ...

What we're doing: Separate the broker from the cache, discard results by default, and keep them only where something reads them.

config/settings.py + reports/tasks.pypython

# Broker: its own database, so a cache.clear() cannot flush the queue.
CELERY_BROKER_URL = env("REDIS_URL") + "/0"
CACHES = {"default": {"BACKEND": "...redis.RedisCache",
                      "LOCATION": env("REDIS_URL") + "/1"}}

CELERY_RESULT_BACKEND = env("REDIS_URL") + "/3"
CELERY_TASK_IGNORE_RESULT = True        # the default for every task
CELERY_RESULT_EXPIRES = 60 * 60 * 6     # six hours — bounds the store

CELERY_BROKER_TRANSPORT_OPTIONS = {"visibility_timeout": 3600}


@shared_task                            # side effect only — no result stored
def send_welcome_email(user_id):
    ...


@shared_task(ignore_result=False)       # a result the API polls for
def build_report(job_id):
    rows = compute(job_id)
    return {"rows": len(rows), "url": upload(rows)}

2–4: Database 0 for the broker, 1 for the cache. This is the single most damaging Redis mistake in a Django stack, and it is one character to avoid.
7: Ignoring results by default inverts Celery's assumption to match reality: most tasks are fired for a side effect, and storing a result nobody reads is a write per task.
8: Without `result_expires` the backend grows without bound. Six hours is far longer than any poller waits and far shorter than the default day.
10: `visibility_timeout` is Redis-specific: a message not acknowledged within it is redelivered. Set it longer than your slowest task, or long tasks are run twice.
18–21: The one task whose return value is read opts back in explicitly, so the exception is visible in the code rather than implied by a global.

Why this works: Two settings — database separation and ignoring results by default — remove the two most common Celery-on-Redis problems, and the per-task opt-in keeps the exception where a reader will notice it.

Leaving `visibility_timeout` shorter than the slowest task

Wrong

python

CELERY_BROKER_TRANSPORT_OPTIONS = {"visibility_timeout": 300}   # 5 minutes
# build_report routinely takes 12 minutes.

Better

python

CELERY_BROKER_TRANSPORT_OPTIONS = {"visibility_timeout": 3600}  # > the slowest task
# and route genuinely long work to its own queue with its own settings.

What you see: Long reports are generated two or three times over, each run producing a duplicate file. There is no error, no retry in the logs, and short tasks are entirely unaffected.

Why: On Redis, Celery emulates acknowledgement with a visibility timeout: a message taken by a worker becomes invisible for that long, and reappears if it has not been acknowledged. A task that runs longer than the timeout has its message redelivered while it is still running, so a second worker starts the same job. The timeout has to exceed your slowest task — which is also a good argument for putting long work on its own queue rather than raising the value for everything.

Two services, usually one Redis — and why they need separating

Broker · required

Holds name + arguments

the message waits here, possibly across a deploy

Its own Redis database

cache.clear() runs FLUSHDB and would delete the queue

RabbitMQ adds durability and DLQs

Redis leaves dead-lettering to you

Down means work never happens

delay() raises — decide whether that is a 503

Result backend · optional

Only needed for .get()

most tasks are fired for a side effect, not a value

result_expires bounds growth

one day by default; without it the store grows forever

task_ignore_result = True

the right default; opt in per task

Down means you cannot see outcomes

the work itself completed normally

Two services, two failure meanings

Two services, two failure meanings
Property	Broker	Result backend
Required?	yes	no
Holds	the message: task name + arguments	the return value and task state
If it is down	work never happens	work happens; you cannot see the outcome
Typical choice	Redis or RabbitMQ	Redis, or the database, or none
Cost of getting it wrong	lost or undelivered jobs	wasted writes for data nobody reads
Setting	`CELERY_BROKER_URL`	`CELERY_RESULT_BACKEND`

Together

python

CELERY_BROKER_URL = env("REDIS_URL") + "/0"
CELERY_RESULT_BACKEND = None
CELERY_TASK_IGNORE_RESULT = True        # opt in per task where a result is read

Remember: The broker is required and carries the message; the result backend is optional and stores the return value. If the broker is down work never happens, while if the backend is down the work still ran and you simply cannot see the outcome. Set `task_ignore_result = True` and opt in per task, because most tasks are fired for a side effect and storing unread results is a write each. Give the broker its own Redis database — `cache.clear()` runs `FLUSHDB` — and on Redis make `visibility_timeout` longer than your slowest task, or it is redelivered mid-run.

Retries and scheduling

What deserves a retry, how backoff and jitter behave, and what Beat actually does.

Retries, backoff, autoretry_for, ETA/countdown, and Beat

coreadvanced

A task that talks to anything outside your database will eventually fail for a reason that goes away on its own, which is what retries are for. `self.retry(exc=..., countdown=...)` retries manually and needs `bind=True`; `autoretry_for=(SomeError,)` does it declaratively. `retry_backoff=True` spaces the attempts exponentially instead of hammering a struggling service, `retry_backoff_max` caps the delay (600 seconds by default), and `retry_jitter` — on by default — randomises it so a thousand tasks that failed together do not retry together. `max_retries` defaults to 3. For scheduling *ahead*, `countdown=` and `eta=` delay a single task; **Celery Beat** is the separate process that publishes tasks on a recurring schedule.

Think of it as

Retry only what can succeed unchanged. That one rule sorts the failures cleanly: a connection reset, a 503, a timeout — retry, because the input was fine and the world was temporarily not. A `ValidationError`, a `KeyError`, a 400 from an API — do not retry, because attempting the identical call four more times produces four identical failures plus a delay before anyone finds out. `autoretry_for` is a list of the first kind, and being specific in it is the whole point; `autoretry_for=(Exception,)` converts every bug into a slow, quiet, four-times-repeated bug. Backoff exists for a second reason beyond politeness: a dependency that fell over usually fell over because of load, so retrying immediately from every failed task is the thundering herd that keeps it down — and jitter is what stops your retries from being synchronised, which matters more than the exponent. Beat is worth understanding as a *publisher* rather than a scheduler: it does not run anything, it just enqueues tasks at the right moments, so it is a single process that must not be run twice (two Beats mean every periodic task fires twice) and whose own failure is invisible, because a task that was never published raises nothing.

python

@shared_task(bind=True, autoretry_for=(RequestException,),
             retry_backoff=True, retry_backoff_max=600,
             retry_jitter=True, max_retries=5)
def notify_partner(self, order_id): ...

What we're doing: Retry only the transient failures, honour a 429's own advice, and schedule the recurring work from Beat.

integrations/tasks.py + config/celery.pypython

@shared_task(
    bind=True,
    autoretry_for=(requests.ConnectionError, requests.Timeout),
    retry_backoff=True, retry_backoff_max=600, retry_jitter=True, max_retries=5,
)
def notify_partner(self, order_id):
    order = Order.objects.get(pk=order_id)
    response = requests.post(PARTNER_URL, json=order.as_payload(), timeout=10)

    if response.status_code == 429:
        # Their advice beats our exponent.
        raise self.retry(countdown=int(response.headers.get("Retry-After", 60)))

    if 400 <= response.status_code < 500:
        # Our payload is wrong. Retrying sends the identical body again.
        raise PartnerRejected(response.text[:200])

    response.raise_for_status()          # 5xx -> RequestException -> autoretry


# Delay a single task without a schedule:
#   send_reminder.apply_async(args=[booking.pk], countdown=3600)
#   send_reminder.apply_async(args=[booking.pk], eta=booking.starts_at - timedelta(hours=1))


app.conf.beat_schedule = {
    "reconcile-nightly": {
        "task": "billing.tasks.reconcile",
        "schedule": crontab(hour=2, minute=30),
        "options": {"queue": "maintenance", "expires": 3600},
    },
}

3–4: Two specific exception classes, not `RequestException` broadly and certainly not `Exception` — an `HTTPError` from a 400 must not be retried, and listing the base class would retry it.
12: A 429 carries its own advice. Honouring `Retry-After` is both more polite and more effective than an exponent that knows nothing about their limit.
14–16: A 4xx is a permanent failure by definition: the same body will be rejected the same way. Raising a distinct exception fails it immediately and visibly.
18: Only the 5xx path reaches `autoretry_for`, which is exactly the set of failures that can succeed unchanged.
30: `expires` on a Beat entry means a task that has not started within the hour is discarded rather than piling up — important when workers were down overnight.

Why this works: The retry policy distinguishes three outcomes that a single `autoretry_for=(Exception,)` would collapse into one: retry with backoff, retry when they say so, and fail now.

`autoretry_for=(Exception,)`

Wrong

python

@shared_task(autoretry_for=(Exception,), retry_backoff=True, max_retries=5)
def import_row(row):
    Product.objects.create(sku=row["sku"], price=Decimal(row["price"]))
# A malformed price now retries five times over ~30 seconds, then fails.

Better

python

@shared_task(autoretry_for=(OperationalError, requests.ConnectionError),
             retry_backoff=True, max_retries=5)
def import_row(row):
    Product.objects.create(sku=row["sku"], price=Decimal(row["price"]))
# A malformed price fails immediately, once, with a usable traceback.

What you see: A bad import file produces five times the log volume, takes minutes instead of seconds to report, and the alert arrives long after the upload — while every traceback is identical.

Why: Retrying is only meaningful when the same input might succeed later. A `decimal.InvalidOperation` on a malformed field is deterministic: five more attempts produce five more identical failures plus the backoff delay before anyone is told. Naming specific transient exceptions keeps retries where they help and lets real bugs fail fast, which is also what keeps a queue moving during an incident.

One failing task under exponential backoff with jitter

t = 0s
Attempt 1 — partner returns 503
autoretry_for catches RequestException; the task raises Retry rather than failing.
t ≈ 1s
Attempt 2
retry_backoff=True starts at roughly one second. Jitter spreads simultaneous failures apart from here on.
t ≈ 2s
Attempt 3
The delay doubles each time, so a dependency recovering after a few seconds costs almost nothing.
t ≈ 4s
Attempt 4
Still cheap. Most transient failures are already resolved by this point.
t ≈ 8s
Attempt 5
max_retries=5 means this is the last one. Without a cap, "retry forever" hides an outage instead of reporting it.
capped
retry_backoff_max = 600s
For longer-running retry policies the delay stops doubling at ten minutes rather than growing to hours.
exhausted
MaxRetriesExceededError
Now it is a real failure: alert on it, record it, and let a dead-letter path or a human decide what happens next.

Retry it, or fail it?

Retry it, or fail it?
Failure	Retry?	Why
Connection reset, timeout, 502/503	yes	the input was fine; the world was temporarily not
429 rate limited	yes, with backoff	honour `Retry-After` if the response carries one
400 / 422 from an API	no	the same payload fails identically every time
`ValidationError`, `KeyError`, `TypeError`	no	a bug — retrying hides it behind a delay
`IntegrityError` on a unique key	usually no	often means the work already happened
Database deadlock	yes	the classic transient — retry the whole transaction

Together

python

@shared_task(autoretry_for=(requests.RequestException,),
             retry_backoff=True, retry_backoff_max=600, max_retries=5)
def notify_partner(order_id): ...

Remember: Retry only what can succeed unchanged: connection errors, timeouts, 5xx and deadlocks — never a 400, a `ValidationError`, or anything deterministic. Name specific exceptions in `autoretry_for`, because `Exception` turns every bug into a slow one. `retry_backoff` spaces attempts, `retry_backoff_max` caps at 600 seconds, and jitter (on by default) is what stops synchronised retries becoming a thundering herd. `countdown`/`eta` delay one task; Beat publishes on a schedule and must run exactly once.

Queues, routing, and limits

Splitting work by duration, sizing concurrency, and the two kinds of time limit.

Queues, routing, worker concurrency, and time limits

coreadvanced

By default every task goes to one queue named `celery` and every worker consumes it, which means a twenty-minute video transcode sits in front of a password-reset email. **Routing** fixes that: `task_routes` sends tasks to named queues by pattern, and each worker is started with `-Q` naming the queues it consumes, so slow work and fast work stop competing. **Concurrency** is `--concurrency=N`, the number of child processes (or threads, with a different pool) each worker runs — sized by CPU for compute and higher for I/O-bound work. **Time limits** come in two kinds: `task_soft_time_limit` raises `SoftTimeLimitExceeded` inside the task so it can clean up, while `task_time_limit` kills the process outright.

Think of it as

Queues exist to stop one class of work starving another, and the useful way to draw the boundaries is by *duration*, not by feature. A queue whose tasks all take under a second behaves completely differently from one whose tasks take minutes: the first needs low latency and the second needs throughput, and mixing them means the fast queue inherits the slow one's worst case. So the usual split is something like `default`, `slow` and `maintenance`, with dedicated workers per queue and different concurrency and time limits on each. That also makes concurrency a per-queue decision rather than a global one — CPU-bound transcoding wants roughly one child per core, while a queue that mostly waits on HTTP can run far more. The time limits are the part most projects skip and then need during an incident: without one, a task that hangs on a socket holds a worker child forever, and enough of them silently reduce your worker pool to nothing. Set the soft limit first, because a task that can catch `SoftTimeLimitExceeded` gets to mark its job row as failed and release its lock, which is the difference between a visible failure and a job stuck in "running" until someone investigates.

python

CELERY_TASK_ROUTES = {
    "media.tasks.*": {"queue": "slow"},
    "billing.tasks.reconcile": {"queue": "maintenance"},
}

What we're doing: Separate slow work from fast work, and give the slow queue limits that let a hung task fail visibly.

config/settings.py + media/tasks.pypython

CELERY_TASK_DEFAULT_QUEUE = "default"
CELERY_TASK_ROUTES = {
    "media.tasks.*": {"queue": "slow"},
    "reports.tasks.*": {"queue": "slow"},
    "billing.tasks.reconcile": {"queue": "maintenance"},
}

# Global fallbacks; per-task values override them.
CELERY_TASK_SOFT_TIME_LIMIT = 30
CELERY_TASK_TIME_LIMIT = 60


@shared_task(
    bind=True,
    soft_time_limit=900,          # 15 min: raise inside the task
    time_limit=1000,              # ~17 min: kill the child process
)
def transcode(self, asset_id):
    asset = Asset.objects.get(pk=asset_id)
    try:
        run_ffmpeg(asset)
    except SoftTimeLimitExceeded:
        # The only chance to leave things tidy — a hard limit skips this entirely.
        logger.error("transcode_timeout", extra={"task_id": self.request.id})
        Asset.objects.filter(pk=asset_id).update(status=Asset.Status.TIMED_OUT)
        cleanup_partial_output(asset)
        raise

3–4: Routing by name pattern rather than per call, so a new task in `media.tasks` lands on the right queue without anyone remembering to say so.
9–10: Global limits sized for the *default* queue. Without them a hung task holds a worker child indefinitely, and enough of those quietly shrink the pool to nothing.
15–16: The slow queue overrides both. The gap between them is deliberate: the soft limit must fire first, with enough room for cleanup before the hard kill.
22–27: Catching `SoftTimeLimitExceeded` is what turns a stuck job into a visible failed one. A hard limit kills the process, so no `except` and no `finally` runs — the row would stay "running" forever.

Why this works: The split stops a transcode from delaying a password reset, and the two-tier limit means a hung task marks itself failed and cleans up instead of being killed silently.

Creating a queue nothing consumes

Wrong

python

CELERY_TASK_ROUTES = {"reports.tasks.*": {"queue": "reports"}}
# Deployment still runs: celery -A config worker -Q default,slow

Better

bash

celery -A config worker -Q default,slow,reports -c 8
# or a dedicated deployment per queue

What you see: Reports are enqueued successfully, `.delay()` returns an id, and none of them ever run. No error is logged anywhere, because publishing to a queue with no consumer is a completely normal operation.

Why: Routing decides which queue a message goes to; `-Q` decides which queues a worker reads. The two are configured in different places — settings and the process command line — so adding a route without updating the deployment silently creates a queue that only fills up. Monitoring queue *depth* rather than only task failures is what catches this, since the failure mode is an absence of execution rather than an error.

One queue, or three — and where the head-of-line blocking goes

Tasks published

emails, reports, transcodes, nightly jobs

One queue: "celery"

the default — everything shares a line

A 20-minute transcode at the head

every password-reset email behind it waits

task_routes by name pattern

the routing decision, made once in settings

default

short tasks · -c 8 · soft limit 30 s

slow

reports and media · -c 4 · prefetch 1 · soft limit 900 s

maintenance

nightly and backfills · -c 2 · off-peak

Worker -Q default

Worker -Q slow

Worker -Q maintenance

A queue nobody consumes

no error anywhere — the tasks just accumulate

A three-queue split, and why each differs

A three-queue split, and why each differs
Queue	Typical task	Concurrency	Soft / hard limit	Prefetch
`default`	emails, notifications, webhooks	cores × 2	30 s / 60 s	default
`slow`	reports, exports, transcoding	cores	900 s / 1000 s	1
`maintenance`	nightly reconciliation, backfills	2	3600 s / 3700 s	1

Together

bash

celery -A config worker -Q default -c 8 --prefetch-multiplier 4
celery -A config worker -Q slow -c 4 --prefetch-multiplier 1

Remember: Everything lands on one `celery` queue by default, so a long task blocks short ones behind it — split queues by *duration* and give each its own worker, concurrency and limits. A worker only consumes what `-Q` names, and publishing to a queue nobody reads produces no error at all, so monitor queue depth. Size concurrency by CPU for compute and higher for I/O. And always set both limits: the soft one raises inside the task so it can mark itself failed and clean up, while the hard one kills the process and skips every `finally`.

Idempotency, monitoring, and recovery

The acks_late trade, making a task safe to run twice, and the three numbers worth alerting on.

Task idempotency, monitoring, recovery, and Flower

coreadvanced

Celery acknowledges a message *before* running the task by default, which means a worker killed mid-task loses that work silently. Setting `acks_late=True` moves the acknowledgement to after completion, so a crash causes redelivery instead of loss — and the Celery docs are explicit that this requires the task to be idempotent, because it may now run more than once. Monitoring is the other half: a queue can be deep, a worker can be dead, and tasks can be failing, and none of those raises anything in your application. **Flower** is the standard web UI for Celery's event stream — live workers, task history, queue lengths — and it is a debugging tool rather than an alerting system, so real alerts belong on queue depth, failure rate, and time since last success.

Think of it as

The `acks_late` decision is a straight choice between two ways to be wrong: acknowledge early and a crash loses the task, or acknowledge late and a crash duplicates it. There is no third option, so the question becomes which failure your task can absorb — and duplication is almost always the cheaper one to make safe, because you can engineer idempotency while you cannot engineer the recovery of work that was never recorded. That is why `acks_late=True` plus an idempotent body is the standard shape for anything that matters. Idempotency here means the same thing it does for webhooks: derive a stable identity for the unit of work, record it with a unique constraint inside the same transaction as the effect, and treat "already recorded" as success. On monitoring, the thing to internalise is that Celery failures are quiet by construction — the caller has already been given a response and is not waiting, so nothing surfaces a growing backlog or a worker that died an hour ago. Alert on the three numbers that describe the system rather than on exceptions: queue depth (work arriving faster than it leaves), failure rate (tasks dying), and last-success age per periodic task (schedules that stopped firing).

python

@shared_task(acks_late=True, autoretry_for=(RequestException,), max_retries=5)
def charge_order(order_id, key):
    with transaction.atomic():
        TaskRun.objects.create(key=key)     # unique constraint = the duplicate check
        charge(order_id)

What we're doing: A task that survives a worker being killed mid-run, without charging anyone twice.

billing/tasks.py + billing/monitoring.pypython

class TaskRun(models.Model):
    key = models.CharField(max_length=128, unique=True)
    task = models.CharField(max_length=128)
    completed_at = models.DateTimeField(null=True)


@shared_task(acks_late=True, autoretry_for=(PSPUnavailable,),
             retry_backoff=True, max_retries=5)
def charge_order(order_id, key):
    try:
        with transaction.atomic():
            run = TaskRun.objects.create(key=key, task="charge_order")
            charge_customer(order_id)          # same transaction as the claim
            run.completed_at = timezone.now()
            run.save(update_fields=["completed_at"])
    except IntegrityError:
        # Redelivered after a crash, or a duplicate publish. Already handled.
        logger.info("charge_order_duplicate", extra={"key": key})
        return


@shared_task
def check_queue_health():
    for queue, sla_seconds in {"default": 60, "slow": 900}.items():
        depth = redis_client.llen(queue)
        oldest = oldest_message_age(queue)
        if depth > 1000 or oldest > sla_seconds:
            alert(f"queue {queue}: depth={depth} oldest={oldest}s")

7: `acks_late=True` is the line that changes the failure mode: a worker killed at any point below leaves the message unacknowledged, so it is redelivered rather than lost.
11–15: The claim and the charge share one transaction. A crash between them rolls both back, so the redelivered message finds no key and legitimately runs again.
16–19: The `IntegrityError` *is* the duplicate detection, and returning normally acknowledges the redelivery instead of retrying work already done.
22–28: Depth and oldest-message age together. Depth alone misses a queue that is small but stalled; age alone misses one that is keeping up but overwhelmed.

Why this works: `acks_late` converts "lost on crash" into "delivered twice", and the unique key converts "delivered twice" into "executed once" — which is the only combination that survives a worker being killed at an arbitrary moment.

Turning on `acks_late` without making the task idempotent

Wrong

python

@shared_task(acks_late=True)
def charge_order(order_id):
    charge_customer(order_id)      # no key, no claim, no guard

Better

python

@shared_task(acks_late=True)
def charge_order(order_id, key):
    with transaction.atomic():
        TaskRun.objects.create(key=key)     # unique -> IntegrityError on redelivery
        charge_customer(order_id)

What you see: Duplicate charges appear after every node eviction or OOM kill — and only then, so they cluster around deploys and traffic spikes and never reproduce in testing.

Why: `acks_late` deliberately changes the failure mode from "lost" to "run again", which is why the Celery documentation states it requires idempotent tasks. Without a guard, a worker killed after `charge_customer` returned but before the acknowledgement leaves a message that will be redelivered and charged a second time. The setting and the guard are one decision, not two.

One message under acks_late, including the crash

Queued in the broker

start

Reserved by a worker — NOT yet acknowledged

Task body running

Idempotency key inserted (unique constraint)

Key already present — this is a redelivery

Acknowledged · removed from the broker

end

Worker killed mid-task

Raised after max_retries — dead-letter or alert

end

Which failure do you want?

Which failure do you want?
Setting	Worker killed mid-task	Requires	Use for
`acks_late=False` (default)	the task is lost, silently	nothing	work that is cheap to lose and expensive to repeat
`acks_late=True`	the task is redelivered	an idempotent body	anything that matters
`acks_late=True` + `reject_on_worker_lost`	requeued even on process loss	a poison-message guard	critical work, with a retry cap

Together

python

@shared_task(acks_late=True, max_retries=5)
def charge_order(order_id, idempotency_key): ...

What to watch, and what each signal means

What to watch, and what each signal means
Signal	Means	Alert when
Queue depth	arrival rate exceeds completion rate	depth grows for N minutes
Oldest message age	latency, independent of depth	older than the SLA for that queue
Failure rate	tasks raising after retries	above baseline
Active worker count	a deployment that did not come back	below the expected replica count
Last success age (per schedule)	a periodic task that stopped firing	older than its period + margin

Together

bash

celery -A config inspect active_queues
celery -A config inspect stats | grep -E "pool|total"

Remember: Celery acknowledges before running by default, so a killed worker loses the task silently. `acks_late=True` swaps that for redelivery — and the docs are explicit that it requires an idempotent task, so the setting and a unique-constraint claim inside the same transaction as the effect are one decision. Nothing surfaces Celery failures on its own, because nobody is waiting: alert on queue depth, oldest-message age, failure rate and last-success age. Flower is a debugger for when the alert fires, not the alert.