beginnerintermediateadvanced

Python · Section 30

Background Jobs and Task Queues

Level: intermediate
Read: 150 min
Concepts: 8

Moving slow or unreliable work off the request path: the queue/worker architecture, the named tools (Celery, Redis, RabbitMQ, Kafka, RQ, Dramatiq), and the delivery guarantees — retries, backoff, dead-letter queues, scheduling, concurrency, at-least-once delivery, idempotency, monitoring, and failure recovery — a background job system is actually built around.

Python overview

Learn Python - Full Course for Beginners [Tutorial]freeCodeCamp.org

What is true here

API → Queue → Worker → Database/external service: enqueueing returns immediately, decoupling request latency from job execution time.
A retry only helps a transient failure; exponential backoff and a max_attempts cap keep retries from making an outage worse.
At-least-once delivery means a worker will sometimes see the same job twice — idempotency is what makes a redelivery safe.
A job that exhausts its retries belongs in a dead-letter queue for inspection, not dropped silently or retried forever.
Celery/RQ/Dramatiq are task queue libraries; Redis/RabbitMQ/Kafka are the brokers underneath them.

What you will be able to do

Explain the API/queue/worker split and why enqueueing a job keeps a request handler fast
Choose between Celery, RQ, and Dramatiq, and identify which broker(s) each supports
Implement a retry with exponential backoff and a hard cap on attempts
Route a job that exhausts its retries to a dead-letter queue instead of dropping or endlessly retrying it
Distinguish a scheduled (repeating) job from a delayed (one-time, relative-wait) job
Set worker concurrency appropriately for I/O-bound versus CPU-bound job types
Design a job handler to be idempotent using an operation-scoped idempotency key, given at-least-once delivery
Track a job through its status lifecycle and describe what a usable failure-recovery path needs beyond a log line

Tools

The named technologies from the roadmap — task queue libraries and the message brokers underneath them.

Task queue tools landscape

standardintermediate

Celery, RQ, and Dramatiq are Python task queue libraries that hand work to background workers. Redis and RabbitMQ are the message brokers they run on top of; Kafka is a log-based streaming platform sometimes used the same way.

Think of it as

A task queue library and a broker are different layers. The broker (Redis, RabbitMQ, Kafka) is the mail room that durably holds messages until a worker collects them. Celery, RQ, and Dramatiq are the mail-sorting software — they define how a Python function becomes a message, gets picked up, and gets retried on failure.

python

from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379/0")

@app.task
def my_task(x, y):
    return x + y

Six technologies from the roadmap, and the layer each occupies

Six technologies from the roadmap, and the layer each occupies
Name	Layer	What it is for
Celery	task queue library	the most widely used Python distributed task queue; supports Redis, RabbitMQ, and other brokers
RQ	task queue library	a simpler, Redis-only task queue — smaller API surface than Celery
Dramatiq	task queue library	a Celery alternative with automatic retries and a simpler configuration model
Redis	message broker / store	an in-memory data store often used as a broker for Celery, RQ, or Dramatiq
RabbitMQ	message broker	a dedicated message broker implementing AMQP; a common Celery broker choice
Kafka	distributed log / streaming platform	a durable, replayable event log — used for streaming pipelines as well as task-like workloads

Together

python

# Celery app configured against a Redis broker (from Celery's own docs)
from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379/0")

@app.task
def send_welcome_email(user_id):
    ...  # deliver the email

send_welcome_email.delay(42)  # enqueues the task; a separate worker process runs it

Remember: Celery/RQ/Dramatiq are task queue libraries; Redis/RabbitMQ/Kafka are the brokers underneath them — pick the library first, then a broker it supports.

Queue and worker fundamentals

The core producer/queue/worker vocabulary every task queue library automates over a real broker.

Job queues and workers

coreintermediate

A job queue holds work items until something is free to run them. A worker is a separate process that pulls a job off the queue, runs it, and moves to the next one — decoupling "request the work" from "do the work" in time.

Think of it as

A job queue is a restaurant order ticket rail. The API is the waiter — it takes an order (the job) and clips it to the rail, then immediately goes back to serving other tables. A worker is a cook who pulls the next ticket, cooks it, and reaches for the next one — the waiter never waits for the food to finish.

python

import queue, threading

work_queue = queue.Queue()

def worker():
    while (job := work_queue.get()) is not None:
        ...  # run job
        work_queue.task_done()

What we're doing: Enqueue three jobs from a producer, and have one worker thread dequeue and run each in order, standing in for the API/queue/worker split a real broker provides.

job_queue_demo.pypython

import queue
import threading

work_queue = queue.Queue()
results = []

def worker(name):
    while True:
        job = work_queue.get()
        if job is None:
            work_queue.task_done()
            break
        results.append((name, job["id"], job["payload"] * 2))
        work_queue.task_done()


for i in range(3):
    work_queue.put({"id": i, "payload": i})
work_queue.put(None)  # signals the worker to stop

t = threading.Thread(target=worker, args=("worker-1",))
t.start()
work_queue.join()
t.join()

print(results)

4: work_queue.Queue() is the durable hand-off point between the producer and the worker — thread-safe by design.
12: job = work_queue.get() is the dequeue step — it blocks until a job is available.
14: The worker executes the job here — a real worker would call the job's actual function with its stored arguments.
16: work_queue.task_done() is the acknowledge step — it tells the queue this job is finished.

Output

[('worker-1', 0, 0), ('worker-1', 1, 2), ('worker-1', 2, 4)]

Why this works: work_queue.put(...) returns immediately — the loop enqueueing three jobs never blocks on their execution, the same way an API handler enqueues a job and responds to the client without waiting for a worker. The worker thread runs independently, pulling one job at a time with get() and only reporting done with task_done(), which is what work_queue.join() waits on before the main thread continues.

Running the job inline instead of enqueueing it

Wrong

python

def handle_signup_request(user_id):
    send_welcome_email(user_id)   # runs inline — request blocks on email delivery
    return {"status": "created"}

Better

python

def handle_signup_request(user_id):
    work_queue.put({"fn": "send_welcome_email", "args": [user_id]})  # enqueue
    return {"status": "created"}   # returns immediately; a worker sends the email

What you see: The request handler is only as fast as the slowest thing it does inline — a slow email provider makes every signup request slow, and an email provider outage makes signup fail entirely.

Why: Calling send_welcome_email(user_id) directly ties the request's response time, and its success or failure, to a completely unrelated system (the email provider). Enqueueing decouples them: the request succeeds once the job is durably queued, and a worker retries the email delivery independently if it fails.

API enqueues; a separate worker dequeues and runs the job

API

enqueues, returns immediately

Queue

durably holds pending jobs

Worker

dequeues, runs one job at a time

Database / external service

the job's actual side effect

The four stages a job passes through

The four stages a job passes through
Stage	What happens
Enqueue	the producer serializes the job (function + args) and puts it on the queue
Dequeue	an idle worker pulls the next job off the queue
Execute	the worker runs the job's function with its arguments
Acknowledge	the worker tells the queue the job finished, so it is not redelivered

Together

python

import queue, threading

work_queue = queue.Queue()

def worker():
    while True:
        job = work_queue.get()      # dequeue (blocks until one exists)
        if job is None:
            break
        print("running job", job)   # execute
        work_queue.task_done()      # acknowledge

t = threading.Thread(target=worker)
t.start()
work_queue.put({"id": 1})           # enqueue
work_queue.put(None)                # tell the worker to stop
t.join()

Remember: A queue durably holds jobs between "request the work" and "do the work"; a worker is the separate process that actually does it — that gap is what makes the API response fast.

Worker concurrency

standardintermediate

Worker concurrency is how many jobs a worker runs at the same time. A concurrency of 1 processes jobs strictly one after another; a concurrency of 8 runs up to 8 jobs in parallel using threads, processes, or async tasks.

Think of it as

Concurrency is how many checkout lanes a store opens. One lane (concurrency 1) serves customers strictly in order — simple, but slow under load. Eight lanes (concurrency 8) serve eight customers at once, trading more staff (memory, CPU) for higher throughput.

python

# Celery: run a worker with 4 concurrent child processes (from Celery's own docs)
# celery -A tasks worker --concurrency=4

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=4) as pool:
    results = list(pool.map(run_job, jobs))

What we're doing: Run six jobs through a thread pool capped at 4 concurrent workers, showing the same "N jobs, M concurrency" shape a real worker pool schedules.

worker_concurrency_demo.pypython

from concurrent.futures import ThreadPoolExecutor


def do_work(n):
    return n * n


with ThreadPoolExecutor(max_workers=4) as pool:
    out = list(pool.map(do_work, range(6)))

print(out)

8: max_workers=4 is the concurrency setting — at most 4 of the 6 jobs run at the same instant.
9: pool.map schedules all 6 calls, running the next one as soon as a worker slot frees up.

Output

[0, 1, 4, 9, 16, 25]

Why this works: ThreadPoolExecutor(max_workers=4) creates a fixed pool of 4 worker threads. pool.map(do_work, range(6)) submits all 6 calls, but only 4 run at once — the pool queues the remaining 2 and starts each as a slot frees up. pool.map preserves input order in its results regardless of which worker finished first, which is why the output matches range(6) squared in order.

Setting concurrency far above what the job type can use

Wrong

python

# CPU-bound job (e.g. image resizing) run with high THREAD concurrency
with ThreadPoolExecutor(max_workers=32) as pool:
    pool.map(resize_image, images)   # threads share one GIL for pure Python CPU work

Better

python

# CPU-bound job: use processes instead, which do not share a GIL
from concurrent.futures import ProcessPoolExecutor

with ProcessPoolExecutor(max_workers=8) as pool:
    pool.map(resize_image, images)

What you see: Adding more threads does not speed up a CPU-bound job past a low ceiling — throughput plateaus while memory use keeps climbing.

Why: Python's Global Interpreter Lock (GIL) lets only one thread run Python bytecode at a time in the standard interpreter, so threads help with I/O waits but not CPU-bound work. A CPU-bound job needs process-based concurrency (or C-extension code that releases the GIL) to actually use more than one core.

Remember: Concurrency is how many jobs a worker runs at once — match the mechanism to the job: threads/async for I/O-bound work, processes for CPU-bound work, since threads share one GIL.

Reliability patterns

What happens when a job fails: retrying with backoff, and where a permanently failing job ends up.

Retries and exponential backoff

coreintermediate

A retry re-runs a failed job instead of giving up immediately. Exponential backoff doubles the wait between each retry, so a struggling dependency gets progressively more breathing room instead of being hit again instantly.

Think of it as

Retrying immediately after a failure is knocking on a door again the instant no one answers — if the problem was "everyone is busy," knocking harder and faster only adds to the load. Exponential backoff is waiting longer before each next knock, giving whatever is overloaded time to recover before the next attempt arrives.

python

import functools, time

def retry_with_backoff(max_attempts=4, base_delay=1):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except RuntimeError:
                    if attempt == max_attempts:
                        raise
                    time.sleep(base_delay * (2 ** (attempt - 1)))
        return wrapper
    return decorator

What we're doing: Wrap a job function that fails twice before succeeding in a retry-with-backoff decorator, and show the exhausted-retries case raising after the cap.

retry_backoff.pypython

import functools
import time


def retry_with_backoff(max_attempts=4, base_delay=0.01, sleep_fn=time.sleep):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            attempt = 0
            while True:
                attempt += 1
                try:
                    return fn(*args, **kwargs)
                except RuntimeError as exc:
                    if attempt >= max_attempts:
                        raise
                    delay = base_delay * (2 ** (attempt - 1))
                    print(f"attempt {attempt} failed ({exc}); sleeping {delay:.3f}s")
                    sleep_fn(delay)
        return wrapper
    return decorator


calls = {"n": 0}


@retry_with_backoff(max_attempts=4, base_delay=0.01)
def flaky_call():
    calls["n"] += 1
    if calls["n"] < 3:
        raise RuntimeError("upstream unavailable")
    return "ok"


print(flaky_call())
print("total calls:", calls["n"])

5: sleep_fn is injected so a caller (or test) can swap in a no-op instead of a real time.sleep — used only to keep this example fast.
16: attempt >= max_attempts is the cap — without it, a permanently failing job would retry forever.
17: The delay doubles each attempt: base_delay * 2^(attempt-1) — 0.01s, then 0.02s here.
28: flaky_call fails its first two calls, succeeding on the third — a realistic transient-failure shape.

Output

attempt 1 failed (upstream unavailable); sleeping 0.010s
attempt 2 failed (upstream unavailable); sleeping 0.020s
ok
total calls: 3

Why this works: wrapper() catches RuntimeError and re-invokes fn(*args, **kwargs) rather than propagating the exception immediately, up to max_attempts times. Each caught failure computes a delay that doubles from the last (base_delay * 2 raised to attempt-1), so attempt 2 waits longer than attempt 1 — exactly the shrinking-frequency retry behavior backoff is for. flaky_call succeeds on its third call, so the loop returns that result instead of ever reaching the re-raise.

Retrying with a fixed delay instead of backing off

Wrong

python

def retry_fixed(max_attempts=5, delay=0.1):
    def decorator(fn):
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return fn(*args, **kwargs)
                except RuntimeError:
                    time.sleep(delay)   # same wait every time
            raise RuntimeError("exhausted retries")
        return wrapper
    return decorator

Better

python

def retry_backoff(max_attempts=5, base_delay=0.1):
    def decorator(fn):
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return fn(*args, **kwargs)
                except RuntimeError:
                    time.sleep(base_delay * (2 ** attempt))  # grows each time
            raise RuntimeError("exhausted retries")
        return wrapper
    return decorator

What you see: A struggling dependency gets hit at the exact same rate by every retrying caller — a fixed delay does nothing to relieve load during an outage, and can prolong it.

Why: A fixed delay retries at a constant rate regardless of how long the dependency has been failing, so many callers retrying together keep re-applying the same load that caused the failure. Backoff spaces retries further apart the longer a failure persists, giving the dependency progressively more room to recover.

Delay doubles each attempt, capped by max_attempts

attempt 1
runs immediately
0s wait
attempt 2
waits 1s
base_delay * 2^0
attempt 3
waits 2s
base_delay * 2^1
attempt 4
waits 4s
base_delay * 2^2 — max_attempts reached, raise if still failing

How the delay grows across attempts, base_delay=1s

How the delay grows across attempts, base_delay=1s
Attempt	Delay before this attempt	Formula
1	0s (first try)	—
2	1s	base_delay * 2^0
3	2s	base_delay * 2^1
4	4s	base_delay * 2^2
5	8s	base_delay * 2^3

Together

python

base_delay = 1
for attempt in range(1, 6):
    delay = base_delay * (2 ** (attempt - 2)) if attempt > 1 else 0
    print(f"attempt {attempt}: wait {max(delay, 0)}s")

Remember: A retry only helps a transient failure; exponential backoff (delay doubling each attempt, capped by max_attempts) keeps retries from hammering an already-struggling dependency.

Dead-letter queues

standardintermediate

A dead-letter queue (DLQ) is a separate queue that holds jobs which failed every retry attempt. It exists so a permanently failing job is held for inspection, instead of being lost or retried forever.

Think of it as

A DLQ is the undeliverable-mail bin at a post office. A letter (job) that could not be delivered after several attempts does not vanish and does not get redelivered forever — it is set aside in a specific bin so a person can look at it and decide what to do.

python

def process_with_dlq(jobs, handler, max_attempts=3):
    dead_letter_queue = []
    for job in jobs:
        for attempt in range(1, max_attempts + 1):
            try:
                handler(job)
                break
            except ValueError as exc:
                if attempt == max_attempts:
                    dead_letter_queue.append({"job": job, "error": str(exc)})
    return dead_letter_queue

What we're doing: Process two jobs where one always fails, and confirm it lands in the dead-letter queue with its error and attempt count after exhausting retries.

dlq_demo.pypython

def process_with_dlq(jobs, handler, max_attempts=3):
    dead_letter_queue = []
    for job in jobs:
        attempts = 0
        while attempts < max_attempts:
            attempts += 1
            try:
                handler(job)
                break
            except ValueError as exc:
                if attempts >= max_attempts:
                    dead_letter_queue.append(
                        {"job": job, "error": str(exc), "attempts": attempts}
                    )
    return dead_letter_queue


def handler(job):
    if job["id"] == "bad-job":
        raise ValueError("invalid payload: missing 'amount'")


jobs = [{"id": "good-job"}, {"id": "bad-job"}]
dlq = process_with_dlq(jobs, handler)
print(dlq)

5: Each job gets up to max_attempts tries before being given up on.
10: attempts >= max_attempts is the point where retries are exhausted — this is where the DLQ routing happens.
11: The failed job, its error, and its attempt count are recorded together — enough context to debug later without rerunning it blind.

Output

[{'job': {'id': 'bad-job'}, 'error': "invalid payload: missing 'amount'", 'attempts': 3}]

Why this works: good-job succeeds on its first attempt and breaks out of the retry loop without ever reaching the dead_letter_queue.append call. bad-job's handler always raises ValueError, so the loop runs it max_attempts times; only once attempts reaches that cap does the except block record it — with its error message and attempt count preserved — instead of retrying a fourth time.

Retrying a DLQ-bound job forever instead of routing it out of the main queue

Wrong

python

while True:
    try:
        handler(job)
        break
    except ValueError:
        pass   # never gives up, never records anything

Better

python

for attempt in range(1, max_attempts + 1):
    try:
        handler(job)
        break
    except ValueError as exc:
        if attempt == max_attempts:
            dead_letter_queue.append({"job": job, "error": str(exc)})

What you see: A permanently broken job spins forever, consuming a worker slot that could process healthy jobs — and no one is ever notified it is failing.

Why: Without a cap and a DLQ, a job with bad data or a real bug retries identically forever, since the failure is not transient. Routing it to a DLQ after max_attempts frees the worker and creates a visible record someone can act on.

Remember: A dead-letter queue holds a job that exhausted its retries — not deleted, not retried forever, just set aside for a human or automated process to inspect.

Scheduling

Running a job later — once after a delay, or repeatedly on a schedule.

Job scheduling and delayed jobs

standardintermediate

Job scheduling runs a job at a specific time or on a repeating interval, like "every day at midnight." A delayed job runs once, after a relative wait, like "in 10 minutes" — both hold the job back from a worker until its time arrives.

Think of it as

A delayed job is a kitchen timer — set it for 10 minutes, and after it rings, the one thing happens once. A scheduled job is a recurring calendar reminder — it fires at midnight every day, indefinitely, not just once.

python

# Celery: delay a single call (from Celery's own docs)
send_welcome_email.apply_async(args=[user_id], countdown=600)   # in 10 minutes
send_welcome_email.apply_async(args=[user_id], eta=specific_datetime)

# Celery: a periodic (scheduled) task via celery beat
app.conf.beat_schedule = {
    "cleanup-every-midnight": {
        "task": "tasks.cleanup_temp_files",
        "schedule": crontab(hour=0, minute=0),
    },
}

What we're doing: Simulate scheduling three jobs with different relative delays, and confirm a min-heap orders them by when they are actually due — the same ordering a real scheduler must produce.

schedule_demo.pypython

import heapq

scheduled = []
now = 1000.0


def schedule_job(job_id, run_at):
    heapq.heappush(scheduled, (run_at, job_id))


schedule_job("send-email", now + 5)
schedule_job("cleanup-temp-files", now + 1)
schedule_job("send-reminder", now + 3)

due_order = [job_id for run_at, job_id in sorted(scheduled)]
print(due_order)

8: heapq.heappush keeps the earliest run_at at the front — a scheduler needs to always find the next-due job cheaply.
12: send-email is scheduled 5 seconds out, but is enqueued first — enqueue order is unrelated to run order.
16: Sorting by run_at, not enqueue order, gives the real execution order — cleanup-temp-files (delay 1) runs before send-reminder (delay 3) before send-email (delay 5).

Output

['cleanup-temp-files', 'send-reminder', 'send-email']

Why this works: Each schedule_job call stores a (run_at, job_id) tuple, and Python tuple comparison sorts by run_at first. Even though send-email was scheduled first, its run_at (now + 5) is the largest, so it sorts last — exactly matching the order a real scheduler must dispatch jobs in, by due time rather than by submission time.

Remember: A delayed job runs once after a relative wait (countdown); a scheduled job runs repeatedly on a fixed pattern (cron-style) — both are held back from a worker until their time arrives.

Delivery guarantees and operations

What a broker actually promises about delivery, why that makes idempotency necessary, and how to see and recover from failures.

At-least-once delivery, duplicate processing, and idempotency

coreintermediate

At-least-once delivery means a broker guarantees a job arrives one or more times, never zero — but "or more" means a worker can see the same job twice (duplicate processing). Idempotency is designing a job so running it twice has the same effect as running it once.

Think of it as

At-least-once delivery is a broker that would rather deliver a package twice than risk losing it — it keeps resending until it gets a clear acknowledgment. That trade means the receiving end (your job handler) has to be the one that treats a repeat delivery safely, the same way signing for a package twice should not charge the customer twice.

python

processed = {}

def handle_job(idempotency_key, *args):
    if idempotency_key in processed:
        return processed[idempotency_key]   # duplicate delivery — skip
    result = do_the_actual_work(*args)
    processed[idempotency_key] = result
    return result

What we're doing: Simulate a redelivered payment job with an idempotency key, and contrast it against an unsafe handler that double-charges when the same job is redelivered.

idempotency_demo.pypython

processed_payments = {}


def charge_card(idempotency_key, amount):
    if idempotency_key in processed_payments:
        return processed_payments[idempotency_key]
    result = {"status": "charged", "amount": amount}
    processed_payments[idempotency_key] = result
    return result


r1 = charge_card("order-42-charge", 19.99)
r2 = charge_card("order-42-charge", 19.99)  # redelivered message
print(r1)
print(r2)
print("charged only once:", len(processed_payments) == 1)

4: idempotency_key identifies the logical operation ("charge order #42"), not the individual delivery attempt.
5: If this key was already processed, the stored result is returned instead of re-running the charge.
13: The second call simulates a broker redelivering the same job — it must not result in a second charge.

Output

{'status': 'charged', 'amount': 19.99}
{'status': 'charged', 'amount': 19.99}
charged only once: True

Why this works: charge_card checks processed_payments[idempotency_key] before doing any work. The first call is a genuine miss, so it charges and stores the result. The second call — standing in for a broker redelivering the job after, say, a lost acknowledgment — finds the key already present and returns the stored result instead of charging again, which is why processed_payments has exactly one entry despite two calls.

Writing a job handler with no idempotency key, assuming delivery happens exactly once

Wrong

python

ledger = []

def charge_card_unsafe(amount):
    ledger.append(amount)                 # no dedup — every call charges
    return {"status": "charged", "amount": amount}

charge_card_unsafe(19.99)
charge_card_unsafe(19.99)   # broker redelivery — double-charges
print(len(ledger), sum(ledger))

Better

python

processed_payments = {}

def charge_card(idempotency_key, amount):
    if idempotency_key in processed_payments:
        return processed_payments[idempotency_key]
    result = {"status": "charged", "amount": amount}
    processed_payments[idempotency_key] = result
    return result

What you see: ledger entries: 2 total charged: 39.98 — the customer's card is charged twice for one order, from a single redelivered job.

Why: Most brokers (including Celery's default acknowledgment modes, and cloud queues like SQS) are at-least-once by design — a lost acknowledgment, a worker crash, or a network partition can cause the same job to be delivered more than once. A handler with no way to recognize "I already did this" treats every delivery as new work.

Same redelivered job, with and without an idempotency key

Without idempotency

+Worker crashes after charging the card, before acknowledging
+Broker redelivers the job — it looks unhandled
+Handler charges the card again
+Customer is billed twice for one order

With an idempotency key

−Worker crashes after charging the card, before acknowledging
−Broker redelivers the job — it looks unhandled
−Handler checks the idempotency key first — already processed
−Handler returns the stored result; no second charge

Idempotent vs. non-idempotent job actions

Idempotent vs. non-idempotent job actions
Action	Idempotent?	Why
UPDATE orders SET status = 'shipped' WHERE id = 42	yes	setting an absolute value — running it twice leaves the same end state
UPDATE accounts SET balance = balance + 10 WHERE id = 1	no	each run changes the result — running it twice adds 20, not 10
INSERT INTO payments (idempotency_key, ...) ... ON CONFLICT DO NOTHING	yes	the unique key makes a duplicate insert a no-op instead of a second row
send_email(user_id)	no, by default	a duplicate call sends a second email unless deduplicated by a key

Together

python

processed_payments = {}

def charge_card(idempotency_key, amount):
    if idempotency_key in processed_payments:
        return processed_payments[idempotency_key]   # duplicate — return the original result
    result = {"status": "charged", "amount": amount}
    processed_payments[idempotency_key] = result
    return result

charge_card("order-42-charge", 19.99)
charge_card("order-42-charge", 19.99)   # redelivered message — not double-charged
print(len(processed_payments))

Remember: At-least-once delivery means a worker will sometimes see the same job twice — an idempotency key (tied to the operation, not the delivery attempt) is what makes that safe to run again.

Job monitoring and failure recovery

standardintermediate

Job monitoring tracks whether jobs are succeeding, failing, or piling up, usually through a dashboard or metrics. Failure recovery is what happens after a job fails for good — alerting someone, and providing a way to requeue it once the root cause is fixed.

Think of it as

Monitoring is the dashboard in a delivery truck showing how many packages are in the back and how many failed to deliver. Recovery is the depot process for a failed delivery — it does not just vanish; someone is notified, and there is a documented way to try again once the address is corrected.

python

job_states = {}

def run_job(job_id, fn):
    job_states[job_id] = {"status": "running"}
    try:
        fn()
        job_states[job_id]["status"] = "succeeded"
    except Exception as exc:
        job_states[job_id]["status"] = "failed"
        job_states[job_id]["error"] = str(exc)

What we're doing: Track a job through queued/running/failed states, capture its error for debugging, and identify it as needing recovery by filtering for the failed state.

job_monitoring_demo.pypython

class JobStatus:
    QUEUED = "queued"
    RUNNING = "running"
    SUCCEEDED = "succeeded"
    FAILED = "failed"


job_states = {}


def run_job(job_id, fn):
    job_states[job_id] = {
        "status": JobStatus.RUNNING,
        "attempts": job_states.get(job_id, {}).get("attempts", 0) + 1,
    }
    try:
        fn()
        job_states[job_id]["status"] = JobStatus.SUCCEEDED
    except Exception as exc:
        job_states[job_id]["status"] = JobStatus.FAILED
        job_states[job_id]["error"] = str(exc)


def failing_task():
    raise ConnectionError("could not reach payment gateway")


run_job("job-1", failing_task)
print(job_states["job-1"])

failed_ids = [jid for jid, s in job_states.items() if s["status"] == JobStatus.FAILED]
print("jobs needing recovery:", failed_ids)

12: job_states records the status transition before the job even runs, so a crash mid-job is still visible as "running", not silently missing.
18: The except block records both the FAILED status and the actual error message — enough to debug without re-running the job.
30: Filtering job_states for FAILED is the recovery step: it produces the exact list of jobs that need attention.

Output

{'status': 'failed', 'attempts': 1, 'error': 'could not reach payment gateway'}
jobs needing recovery: ['job-1']

Why this works: run_job wraps fn() in a try/except so a raised exception updates job_states to FAILED with the error message, rather than crashing the whole monitoring loop. Because the status and error are stored in a lookup keyed by job_id rather than just printed, a later pass can filter for every FAILED job — the same query a real monitoring dashboard runs to show what needs recovery.

Remember: Monitoring tracks what state every job is in (and whether the queue is draining); recovery turns a captured failure into an actionable next step — alert, context, and a safe way to requeue.