Background Tasks: Your Job Queue Was a Decorator All Along

A Prefect task can run with no flow at all — submitted to the API from a web handler, executed by a task worker somewhere else. This is the feature that walks Prefect straight into Celery's territory.

Ten chapters in, this series has trained you on an assumption so foundational you probably stopped noticing it: a task lives inside a flow. The flow is the unit of work, the thing you deploy, the thing that gets scheduled; tasks are the steps within it. Everything from flows and tasks through transactions has taken that for granted. Airflow takes it for granted too — a task without a DAG isn’t a thing Airflow has a word for.

Prefect 3 breaks the assumption. A task can run with no flow at all:

resize_cover.delay(book_id=42)

No flow. No DAG. No schedule. That line, called from a FastAPI request handler or a Django view or a python -c one-liner, creates a task run in the Prefect API and returns immediately. Somewhere else — a different process, possibly a different machine — a task worker picks it up and executes it, with the retries, the caching, the result persistence and the state machine that task was decorated with.

You have just written a job queue. You did not stand up a broker, define a Celery app, configure a result backend, or learn a second framework’s idea of what a “task” is. You used the @task decorator you were already using, and called a different method on it.

This is the least Airflow-shaped thing in Prefect, and depending on your architecture it’s either a curiosity or the single feature that makes Prefect worth adopting on its own.

What a background task actually is

Strip the marketing off and the mechanism is simple. Three parties:

  1. The producer calls .delay(). That does not run anything. It creates a task run in the Prefect API in a Scheduled state, writes the call’s parameters to storage, and hands back a future. It returns in milliseconds.
  2. The API holds the run. It is the queue. There is no separate broker — no Redis, no RabbitMQ, no SQS — because the Prefect server you already have is doing that job.
  3. The task worker is a long-lived process that has called serve() on one or more tasks. It subscribes to the API for scheduled runs of exactly those tasks, pulls them, and executes them in its own process.

The producer and the worker never talk to each other. They don’t share memory, they don’t share a process, and in production they usually don’t share a machine. All they share is the task’s definition (both import it from the same module) and the API in between. That’s it. That’s the whole architecture.

The thing worth noticing is what you didn’t have to do. The task is not a special “background task” type. It’s the same @task you’d put in a flow — same decorator, same arguments, same everything — and it can be called all three ways: directly (resize_cover(42) runs it inline), submitted (resize_cover.submit(42) runs it on the flow’s task runner), or delayed (resize_cover.delay(42) ships it out of the process entirely). One definition, three execution modes. That’s a genuinely nice piece of design, and it’s the payoff of Prefect treating a task as a durable, addressable, keyed thing rather than a node in someone’s graph.

.delay(): the producer side

# bgtasks.py — the shared definition, imported by both sides
from prefect import task

@task(retries=2, log_prints=True)
def resize_cover(book_id: int) -> str:
    print(f"resizing cover for book {book_id}")
    src = fetch_original(book_id)
    out = resize(src, width=800)
    return upload(out, f"covers/{book_id}-800x1200.webp")
# anywhere at all — no flow in sight
from bgtasks import resize_cover

future = resize_cover.delay(book_id=42)
print(future.task_run_id)

I ran exactly this against a local Prefect server, and the producer printed a task run id and exited in well under a second while the worker did the work. What comes back is a PrefectDistributedFuture — “distributed” because, unlike the PrefectFuture you get from .submit(), the thing it’s a handle on isn’t running in your process and never will.

.delay() is a thin alias for apply_async(args=…, kwargs=…), which exists if you need the explicit form (and if that name rings a Celery bell, that is not an accident — the API is a deliberate nod). .delay() just lets you pass arguments naturally, so use it.

Note what carries over untouched: retries=2 still applies — the worker retries the run. log_prints=True still routes prints to the run’s logs, viewable in the UI. Cache policies still apply. State hooks still fire. Everything you know about a Prefect task remains true; only the execution venue changed.

The task worker, and the serve() that isn’t the other serve()

The consumer half is a process that stays alive:

# worker.py
from prefect.task_worker import serve
from bgtasks import resize_cover, rebuild_thumbnails

if __name__ == "__main__":
    serve(resize_cover, rebuild_thumbnails, limit=10)
python worker.py

Mind that import, because Prefect has two functions called serve and they take completely different things.

  • from prefect import serve — the one chapter 05 used — takes deployments (RunnerDeployment objects, what you get from flow.to_deployment(...)). It polls the API for scheduled flow runs and runs them.
  • from prefect.task_worker import serve takes tasks. It subscribes for delayed task runs and runs them.

They are not interchangeable, they are not overloads of each other, and handing tasks to the top-level serve will not do what you want. The full signature of the one you want here:

serve(*tasks, limit=10, status_server_port=None, timeout=None)
  • limit caps how many task runs this process executes concurrently. It defaults to 10. Pass None to remove the cap entirely, which you should do approximately never.
  • status_server_port stands up a small HTTP status endpoint — point a Kubernetes liveness probe at it.
  • timeout makes the worker exit after N seconds. Useful for a worker you run as a periodic job rather than a daemon.

One constraint that catches people: you must serve the same task object you submit. The worker imports resize_cover from bgtasks and so does the producer, and that shared import is how the two sides agree on what’s being run. If you redefine the task in the worker file, or wrap it in a decorator that changes its identity, the worker will sit there politely receiving nothing.

Getting the answer back

.delay() gives you a future, and you have three ways to interrogate it:

future = slow_report.delay(month="2026-05")

future.wait(timeout=60)          # block until it's final; returns None
print(future.state.name)         # "Completed" / "Failed" / "Crashed" / …
result = future.result(timeout=60)   # block, then hand back the return value

Run for real, against a worker in another process:

delay() returned immediately; task_run_id = 0eb382bd-a4dc-4c64-b495-8b15791fdc72
type: PrefectDistributedFuture
state: Completed
result: covers/42-800x1200.webp

.result() takes raise_on_failure=True by default — a failed background run re-raises its exception in your process when you ask for its result, which is usually what you want and is occasionally a surprise, because the traceback you get is from code that ran on a different machine. raise_on_failure=False hands you the exception as a value instead. And .add_done_callback(fn) exists if you’d rather not block at all.

Now the part that matters in production and that nobody tells you until it breaks. The worker ran in a different process. The return value is not in your memory and never was — it has to travel. It travels through Prefect’s result storage, and the parameters you passed travel the same way, written by the producer and read by the worker.

Which means both sides must be able to read the same storage.

.delay() quietly forces persist_result=True on the submission path when the call has parameters — you don’t have to remember it, and that’s why a naive .delay() on your laptop Just Works. It works because the default storage is a LocalFileSystem at ~/.prefect/storage, and on your laptop the producer and the worker are the same laptop. Ship that to production with the web app in one pod and the task worker in another, and the worker cannot read the parameters the producer wrote, because they were written to a filesystem that no longer exists from where it’s standing.

So configure it. Point background-task parameter storage and result storage at something both sides can see:

# both the producer (your API pods) and the task worker need these
export PREFECT_TASKS_SCHEDULING_DEFAULT_STORAGE_BLOCK="s3-bucket/bookshop-tasks"
export PREFECT_RESULTS_PERSIST_BY_DEFAULT=true
export PREFECT_DEFAULT_RESULT_STORAGE_BLOCK="s3-bucket/bookshop-results"

PREFECT_TASKS_SCHEDULING_DEFAULT_STORAGE_BLOCK is the one that’s specific to this feature and the one nobody sets: it names the block where the parameters of a delayed run get stashed. Leave it unset and it silently falls back to the local filesystem. That’s the single highest-value line in this chapter, and it’s the one that will otherwise cost you a confused afternoon watching a worker fail to hydrate a task run it can plainly see.

.submit() versus .delay(), made rigorous

Both return a future. They are not the same tool, and the difference is not a matter of degree.

.submit() hands the task to the task runner of the current flow run. It requires a flow. The work happens inside your flow’s process (on a thread, by default, or on Dask/Ray if you’ve configured that runner), the future resolves within that flow run, and the flow won’t finish until you’ve collected it. It’s for parallelism inside one pipeline — fan out over forty files, rejoin, sum the counts.

.delay() hands the task to the API. It does not require a flow, does not run in your process, and does not resolve within any particular run. Whoever calls it gets a receipt and gets on with their life. It’s for decoupling work out of the caller — a background job, not a pipeline step.

The distinction that actually decides it for you: does the caller need to wait?

  • If the caller’s job is not done until this work is done — the flow needs the transformed rows to load them — you want .submit(). The work is part of the caller.
  • If the caller’s job is done the moment the work is accepted — the HTTP response should go out now, the resize can happen whenever — you want .delay(). The work is triggered by the caller.

A second axis worth naming: .submit() scales with your flow’s task runner (threads, or a Dask cluster). .delay() scales with how many task workers you’re running, which is an infrastructure decision made outside the code entirely. .submit() scales a flow. .delay() escapes the flow.

And yes, you can call .delay() from inside a flow. It does exactly what it says: fires a run off to the workers and doesn’t wait, so the flow can complete while the work continues elsewhere. That’s a strange and occasionally perfect thing to do — a nightly flow whose last act is to enqueue ten thousand thumbnail regenerations that it has no intention of supervising.

Chaining background work with wait_for

Background runs aren’t required to be independent. apply_async — the explicit form under .delay() — takes a wait_for of futures, and the second run won’t start until the first is done:

from bgtasks import extract_cover, resize_cover

extracted = extract_cover.delay(book_id=42)
resized = resize_cover.apply_async(
    kwargs={"book_id": 42},
    wait_for=[extracted],        # doesn't start until `extracted` is final
)

Both calls return instantly; the ordering is enforced by the API, not by your process blocking between them. I ran this with two tasks and one worker and got exactly that: both submitted in the same breath, the second executing only after the first reached a terminal state, and both results retrievable afterwards.

This is a genuinely useful shape — a small dependency chain of background work that no flow supervises — but resist the urge to build a DAG out of it. If you find yourself hand-wiring wait_for through five delayed tasks, you have written a pipeline with none of the pipeline features, and you should have written a flow. Use wait_for for a link or two of ordering. Use a flow for a graph.

Failure, retries, and the worker that dies

Everything the states chapter says about failure applies here, with one twist worth internalizing: the failure happens somewhere you aren’t.

Retries work exactly as declared, and they are executed by the worker. A @task(retries=2, retry_delay_seconds=30) that fails in the background waits thirty seconds and comes back, with the producer none the wiser — the producer left ages ago. Only when the retries are exhausted does the run land in Failed, at which point future.result() (if anyone is still holding the future) re-raises, and any on_failure hook on the task fires in the worker’s process.

If the worker itself dies mid-run — OOM-killed, pod evicted, SIGKILL — the run goes Crashed, not Failed, which is exactly the distinction Prefect draws everywhere else and exactly the one Celery has never drawn. And this is where the automation you already wrote pays off: a server-side rule watching for prefect.task-run.Crashed fires whether or not any Python of yours survived, which is the only kind of alerting that’s worth anything when the failure mode is “the process is gone.”

The observability story is the quiet win here. A background job that failed is a run in the Prefect UI with a state, a message, a traceback, its logs, its retry history, and its parameters. You click on it. Compare the Celery experience — grep the worker’s stdout in your log aggregator, hope the traceback wasn’t interleaved with three other workers’, and reconstruct what arguments it was called with from a task_id you don’t have.

One setting to know while you’re here: PREFECT_TASKS_SCHEDULING_DELETE_FAILED_SUBMISSIONS (default true) controls whether runs that fail to submit — as opposed to fail to run — are cleaned out. Leave it alone unless you’re debugging why a .delay() didn’t produce a run.

Scaling: the API is the queue

The task worker is a plain Python process. It scales the plain way: run more of them.

Several serve() processes subscribed to the same tasks share the incoming runs. Two workers give you roughly twice the throughput; ten give you ten. There is no coordination to configure, no consistent-hash ring, no broker partitions to rebalance, because the Prefect API is the queue and the workers are just subscribers to it. kubectl scale deployment task-worker --replicas=8 is the entire scaling story.

Within one worker, limit= is your throttle:

serve(resize_cover, limit=20)     # I/O-bound: crank it
serve(train_model, limit=2)       # memory-hungry: don't

limit bounds how many runs that process executes concurrently. Raise it for I/O-bound work where the process is mostly waiting on the network. Lower it for anything that holds a lot of memory, because ten concurrent image resizes in one container is a very effective way to discover your memory limit.

And note that the two knobs stack in a useful way. limit is a per-process cap. The tag-based concurrency limits from the concurrency chapter are global — they apply to task runs by tag, across every worker, everywhere. So:

@task(tags=["thumbnails"], retries=2)
def resize_cover(book_id: int) -> str:
    ...
prefect concurrency-limit create thumbnails 30

Now you can run twelve workers with limit=20 each — 240 slots of capacity — while the global limit guarantees at most 30 resizes are actually executing at once, because that’s what your image service can take. Capacity and permission are separate concerns, and Prefect lets you set them separately. Celery makes you approximate this with queue routing and prefetch multipliers, and it’s never quite the thing you meant.

Serve the right tasks together

serve() takes a list of tasks, and which tasks you group into one worker is a real design decision rather than an afterthought. A worker’s limit is shared across everything it serves — so if you serve a memory-hungry train_model alongside a cheap send_email in a process with limit=10, ten model trainings can start at once and take the pod with them.

The rule: one worker per resource profile. Group tasks that want the same limit, the same memory, the same image. Cheap I/O tasks in a high-limit worker; heavy tasks in a low-limit worker with a bigger memory request; anything needing a GPU in a worker scheduled onto GPU nodes. They’re separate serve() scripts and separate deployments, and they all talk to the same API. That’s it — that’s the whole routing story, and it’s a lot less machinery than Celery queues plus routing keys plus worker -Q flags.

Running task workers in production

A task worker is a long-lived Python process with no HTTP surface, which makes it the most boring possible thing to deploy — a Deployment with replicas, or a systemd unit, or an ECS service.

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "worker.py"]

Three operational details separate a worker that survives a bad Tuesday from one that doesn’t:

Health checks. serve(..., status_server_port=8080) stands up a small HTTP endpoint reporting the worker’s status. Point a Kubernetes livenessProbe at it and a wedged worker gets restarted instead of quietly holding slots forever. Without it, your orchestrator’s only signal is “the process hasn’t exited,” which a deadlocked worker satisfies beautifully.

Shared storage, again. Every worker replica and every producer needs the same PREFECT_TASKS_SCHEDULING_DEFAULT_STORAGE_BLOCK and result storage. This is not optional in a multi-pod deployment and it is the number one way this feature fails in production — so put it in the shared ConfigMap where both the API pods and the worker pods pick it up, not in one Dockerfile.

Graceful shutdown. A worker that gets SIGTERM’d mid-run leaves a task run that never reaches a terminal state until Prefect’s lost-heartbeat detection marks it Crashed. Give the pod a terminationGracePeriodSeconds long enough for in-flight runs to finish, and set retries on the tasks that matter so a genuinely-killed run comes back rather than being lost. The timeout= argument to serve() is also worth knowing here: it makes the worker exit cleanly after N seconds, which turns a daemon into a periodic job — useful when you’d rather recycle processes on a timer than trust them to stay healthy indefinitely.

The payoff: a web handler that returns immediately

This is the case the whole feature exists for.

# api.py
from fastapi import FastAPI
from bgtasks import resize_cover

app = FastAPI()

@app.post("/books/{book_id}/cover", status_code=202)
def upload_cover(book_id: int):
    future = resize_cover.delay(book_id=book_id)
    return {"status": "queued", "task_run_id": str(future.task_run_id)}
# status.py — because you returned an id, the client can poll
from prefect.client.orchestration import get_client

@app.get("/tasks/{task_run_id}")
async def task_status(task_run_id: str):
    async with get_client() as client:
        run = await client.read_task_run(task_run_id)
        return {"state": run.state.name}

The handler enqueues and returns 202 Accepted in single-digit milliseconds. The image resize — slow, failure-prone, and absolutely not something that should hold an HTTP connection open — happens in a worker pod, with two retries, with its state visible in the Prefect UI, with its logs collected, and with a result you can fetch by id.

Count what you didn’t build. No broker. No result backend. No task registry. No separate serialization contract between the web app and the worker. No second UI to go and look at when something failed at 2 a.m., because the run is in Prefect, next to your pipelines, with the same states, the same retries, and the same automations watching it — which means the Crashed automation you wrote in the last chapter already covers your web app’s background jobs. You didn’t do anything to make that true. It’s true because it’s the same task system.

Where this competes with Celery and RQ, honestly

Now let me be the bad salesman again, because “Prefect replaces Celery” is a claim that deserves scrutiny rather than applause.

What Prefect genuinely wins. One system instead of two. If you’re already running Prefect for pipelines, adding a job queue costs you a .delay() and a worker deployment. Your background jobs get the whole platform for free: states, retries with exponential backoff, result persistence, caching, transactions, the UI, the event stream, automations, and the ability to call the same task from a flow when you want it in a pipeline instead. Celery gives you none of that. Celery gives you a queue, and everything else is yours to build — which is why every Celery shop I’ve ever seen has a homegrown Flower-adjacent admin page and a Slack bot that nobody maintains.

Where Celery still wins, and it isn’t close. Throughput. Celery over RabbitMQ or Redis is a purpose-built broker doing one job at machine speed; Prefect’s queue is an HTTP API backed by Postgres, and every task run is a database row with a state history. If you are enqueuing tens of thousands of jobs a second, this is not your tool, and no amount of tuning will make it your tool — the per-run overhead is the price of the observability, and you cannot have one without the other. Prefect’s background tasks are for jobs that are worth a database row: the image resize, the report build, the email send, the webhook retry. Not the fifty-thousand-a-second telemetry writes.

RQ sits in between and its pitch is simplicity — it’s a few hundred lines you can read in an afternoon. If your entire requirement is “run this function later, on this box,” RQ is smaller than Prefect and always will be.

The honest decision rule. Already running Prefect and need background jobs at human scale? Use this; the marginal cost is nearly zero and the marginal benefit is a real control plane. Running Celery at high throughput with no orchestrator? Don’t rip it out; you’d be trading a fast queue for a slower one to get features you may not need. Greenfield, and you need both pipelines and background jobs? That’s the sweet spot, and running one system instead of two is worth a great deal more than the benchmark difference will ever cost you.

And notice the ground this covers that Airflow simply doesn’t have a story for. Airflow is a batch scheduler for graphs of work on a timetable. There is no supported way to say “run this one function, right now, from my web handler, with no DAG” — the closest you get is triggering a DAG run via the API and then explaining to your teammates why a single image resize has a scheduler, an executor, and a five-second startup tax. Prefect covers the scheduled-pipeline ground and this ground with the same task abstraction, and that’s not a nicer skin on the same tool. It’s a wider tool.

Final thoughts

The three chapters that ended here — transactions, automations, background tasks — don’t share a line of code, but they share a thesis, and it’s the thesis this whole series has been building toward.

Because a Prefect task is a durable, addressable, first-class thing — a value that persists, a state that emits an event, a run you can enqueue by id — you get capabilities Airflow’s static graph cannot express, not because Airflow’s engineers were less clever, but because a graph parsed ahead of time can only describe work it knew about at parse time. Transactions make “these steps are atomic, and here’s the undo” a property of the code. Automations make “when this happens, do that” a rule on a live event stream instead of a callback trapped inside the process that’s already dead. Background tasks make “run this now, off to the side, no pipeline required” a method call.

Airflow’s bet is the opposite one, and it’s just as coherent: a workflow is a static graph on a schedule, and committing to that shape up front buys you a picture you can see before it runs, a decade of operators for every system you’ll ever touch, and an operational story that large batch shops have already hardened at a scale Prefect hasn’t been tested at. For a fixed nightly graph feeding a warehouse, that certainty and that ecosystem are a feature, not a limitation you’re tolerating.

So don’t read these three chapters as Prefect winning. Read them as the honest answer to “what do I get that I couldn’t have?” If your pipelines write to multiple systems, react to things that happen, or double as your application’s background jobs, this is exactly the ground Airflow left you to build by hand — and building it by hand is how you ended up with a cleanup task guarded by a trigger rule, a Slack webhook pasted into six DAGs, and a Celery cluster nobody wants to own.

Next: States and State Hooks

Comments