Background Workers Explained: Designing Reliable Asynchronous Processing

Background workers execute tasks outside the main request-response flow of an application. Instead of making an API client wait while a report is generated, an email is sent, or a video is processed, the application places a job into a queue and returns a response quickly. A separate worker receives the job and performs the work asynchronously.

This architecture improves responsiveness and allows expensive workloads to scale independently, but it also introduces new failure modes. Jobs can be delivered more than once, workers can crash halfway through processing, deployments can interrupt active work, and queue backlogs can grow faster than workers can consume them. Reliable background processing therefore requires careful handling of acknowledgements, retries, idempotency, timeouts, concurrency, shutdown, and observability.

Table of Contents

• What Are Background Workers?
• Synchronous vs Asynchronous Processing
• How Background Workers Work
  • Producer
  • Job Queue
  • Worker
  • Result Storage
• Background Worker Lifecycle
• Common Background Worker Architectures
  • Single Shared Queue
  • Separate Queues by Workload
  • Workflow Orchestration
• Job Payload Design
• Acknowledgements and Delivery Guarantees
• Idempotent Job Processing
• Retries and Backoff
• Timeouts, Leases, and Heartbeats
• Handling Long-Running Jobs
• Concurrency and Parallelism
• Job Ordering
• Priority, Delayed, and Scheduled Jobs
• Graceful Shutdown and Deployments
• Worker Scaling
• Backpressure and Overload Protection
• Observability and Monitoring
• Real-World Report Generation Example
• Common Mistakes
• Production Checklist
• Conclusion
• Key Takeaway
• More Articles to Read

What Are Background Workers?

A background worker is a process that continuously waits for asynchronous jobs, executes them, and reports whether they succeeded or failed. It usually runs separately from web servers, API containers, or frontend applications.

Typical background jobs include:

• Sending emails and notifications.
• Generating PDF or CSV reports.
• Processing uploaded images or videos.
• Synchronising data with external services.
• Updating search indexes.
• Importing large files.
• Calculating analytics.
• Executing scheduled maintenance.
• Creating shipments or invoices.

The application that creates a job is usually called the producer. The process that executes it is the worker or consumer. A queue or broker connects them.

def request_report(report_id: str) -> dict:
    job = {
        "job_id": generate_job_id(),
        "job_type": "generate_report",
        "report_id": report_id
    }

    job_queue.publish(job)

    return {
        "report_id": report_id,
        "status": "queued"
    }

The API does not generate the report directly. It records or publishes the work and returns control to the client. A worker later processes the job.

Synchronous vs Asynchronous Processing

Synchronous processing keeps the client connected until the operation finishes. Asynchronous processing accepts the request, creates a background job, and allows the result to become available later.

A database lookup that completes in a few milliseconds usually belongs in the request path. A data export that scans millions of rows should normally run in the background.

Background processing is not automatically better. Moving a simple operation into a queue adds latency, infrastructure, deployment concerns, retries, and monitoring requirements. It is most useful when the work is slow, bursty, independently scalable, or able to continue after the original request ends.

How Background Workers Work

A background-processing system normally contains four main parts: a producer, a job queue, one or more workers, and optional result storage. Each part has a separate responsibility.

Producer

The producer decides that asynchronous work is required and creates a job. Producers may be API servers, event consumers, scheduled processes, administrative tools, or other workers.

def queue_thumbnail_job(image_id: str) -> None:
    job_queue.publish({
        "job_id": generate_job_id(),
        "job_type": "create_thumbnail",
        "image_id": image_id,
        "attempt": 0
    })

The producer should include enough information to identify the operation without embedding large or sensitive objects unnecessarily.

Job Queue

The queue stores jobs until a worker is ready. Depending on the technology, it may also provide acknowledgements, visibility timeouts, retries, delayed delivery, ordering, priorities, and dead-letter queues.

The queue separates production speed from processing speed. An API can create 10,000 jobs during a traffic spike while workers process them gradually, assuming the queue has sufficient capacity and the backlog remains within acceptable limits.

Worker

The worker reserves or receives a job, validates it, executes the operation, records the result, and acknowledges completion.

def run_worker() -> None:
    while True:
        job = job_queue.receive(timeout_seconds=10)

        if job is None:
            continue

        try:
            process_job(job)
            job_queue.acknowledge(job)
        except TemporaryError:
            job_queue.retry(job)
        except PermanentError:
            job_queue.move_to_dead_letter_queue(job)

Production workers also need shutdown handling, structured logs, metrics, tracing, timeout protection, and retry limits.

Result Storage

Some jobs produce a result that must be retrieved later. The result may be stored in a database, object storage, cache, or dedicated job-status table.

CREATE TABLE background_jobs (
    job_id VARCHAR(64) PRIMARY KEY,
    job_type VARCHAR(100) NOT NULL,
    status VARCHAR(30) NOT NULL,
    progress_percent INTEGER NOT NULL DEFAULT 0,
    result_location TEXT,
    error_code VARCHAR(100),
    created_at TIMESTAMP NOT NULL,
    started_at TIMESTAMP,
    completed_at TIMESTAMP
);

A status endpoint can return queued, running, completed, or failed without requiring the client to remain connected during processing.

Background Worker Lifecycle

A reliable worker follows a defined lifecycle rather than treating every job as a single unstructured function call.

1. Receive: obtain a job from the queue.
2. Reserve: temporarily prevent another worker from processing it.
3. Validate: confirm that the payload and job type are supported.
4. Check idempotency: determine whether the operation already completed.
5. Mark running: record the processing state if status tracking is required.
6. Execute: perform the business operation.
7. Persist result: save durable output or state changes.
8. Acknowledge: tell the queue that processing completed.
9. Retry or dead-letter: handle failed execution according to error type.

def process_worker_job(job: dict) -> None:
    validate_job(job)

    if processed_jobs.exists(job["job_id"]):
        job_queue.acknowledge(job)
        return

    jobs.mark_running(job["job_id"])

    try:
        result = execute_job(job)

        with database.transaction():
            jobs.mark_completed(
                job_id=job["job_id"],
                result=result
            )
            processed_jobs.record(job["job_id"])

        job_queue.acknowledge(job)

    except TemporaryError as error:
        jobs.mark_retrying(job["job_id"], str(error))
        schedule_retry(job)

    except PermanentError as error:
        jobs.mark_failed(job["job_id"], str(error))
        job_queue.move_to_dead_letter_queue(job)

The queue acknowledgement should generally occur only after the durable business result has been stored. Otherwise, a worker crash can remove the job before its work is complete.

Common Background Worker Architectures

Worker systems can be organised in several ways. The correct structure depends on workload size, isolation requirements, ordering, latency targets, and operational complexity.

Single Shared Queue

All job types are published to one queue and processed by a shared worker pool.

Advantages

• Simple infrastructure.
• Easy initial deployment.
• Idle workers can process any job type.

Disadvantages

• Slow jobs can delay urgent jobs.
• One poison workload can reduce capacity for every job type.
• Scaling cannot target one workload independently.

When to Use

Use a shared queue for small systems with similar job durations, priorities, and resource requirements.

Real-World Use Cases

• Small administrative applications.
• Low-volume email and notification processing.
• Early-stage products with limited worker traffic.

Example

handlers = {
    "send_email": send_email,
    "generate_invoice": generate_invoice,
    "update_search_index": update_search_index
}

def process_job(job: dict) -> None:
    handler = handlers[job["job_type"]]
    handler(job)

Separate Queues by Workload

Different job categories use separate queues and worker pools.

Advantages

• Resource-heavy jobs cannot block lightweight jobs.
• Each queue can scale independently.
• Retry, timeout, and priority settings can differ.
• Deployments can target one worker type.

Disadvantages

• More queues and services must be operated.
• Idle capacity may exist in one pool while another is overloaded.
• Routing configuration becomes more complex.

When to Use

Use separate queues when workloads have different latency targets, execution times, resource profiles, or failure behaviour.

Real-World Use Cases

• Fast email jobs separated from long video-processing jobs.
• Customer-facing jobs separated from maintenance tasks.
• CPU-intensive document generation separated from network-heavy integrations.

Example

def enqueue_job(job: dict) -> None:
    queue_by_type = {
        "send_email": "notifications",
        "resize_video": "media-processing",
        "generate_report": "reports"
    }

    queue_name = queue_by_type[job["job_type"]]
    job_queue.publish(queue_name, job)

Workflow Orchestration

A workflow engine coordinates several dependent background steps. Each step records state and triggers the next one.

Advantages

• Makes long-running business workflows visible.
• Supports retries and compensation per step.
• Prevents large jobs from becoming one fragile function.

Disadvantages

• Requires workflow state management.
• Introduces more components and transitions.
• Debugging requires correlation across several jobs.

When to Use

Use orchestration when a process contains several dependent stages, may run for a long time, or requires compensation after partial failure.

Real-World Use Cases

• Order fulfilment.
• Customer onboarding.
• Large data imports.
• Insurance claim processing.

Example

def start_order_workflow(order_id: str) -> None:
    workflow.create({
        "workflow_id": generate_workflow_id(),
        "order_id": order_id,
        "current_step": "reserve_inventory",
        "status": "running"
    })

    job_queue.publish({
        "job_type": "reserve_inventory",
        "order_id": order_id
    })

Job Payload Design

A job payload should be small, versioned, traceable, and safe to retry. Large payloads increase queue storage, network use, and deserialisation cost. Sensitive payloads may also create security and compliance risks.

A practical payload contains identifiers and metadata rather than complete database objects.

job = {
    "job_id": "job-4f83a",
    "job_type": "generate_customer_report",
    "job_version": 2,
    "customer_id": "customer-910",
    "report_id": "report-221",
    "requested_at": "2026-07-25T14:30:00Z",
    "correlation_id": "request-82ad"
}

The worker loads current state from the authoritative database. This avoids processing stale customer details copied into the queue several hours earlier.

However, storing only an entity ID can also change semantics. If the job must process the exact state that existed at creation time, the payload may need a snapshot version, object-storage reference, or immutable event data.

Useful payload fields include:

• job_id for deduplication and tracing.
• job_type for handler routing.
• job_version for schema evolution.
• Business entity identifiers.
• correlation_id for distributed tracing.
• created_at for queue-age monitoring.
• Tenant or account identifier where required.

Credentials, large binary data, and mutable application objects should not normally be embedded directly in queue messages.

Acknowledgements and Delivery Guarantees

An acknowledgement tells the queue that a job finished and no longer needs delivery. The timing of this acknowledgement determines whether the system prefers avoiding message loss or avoiding duplicate execution.

def process_at_least_once(job: dict) -> None:
    result = execute_job(job)

    # Store the result before acknowledging the job.
    save_result(job["job_id"], result)

    job_queue.acknowledge(job)

If the worker crashes after saving the result but before acknowledgement, the queue may redeliver the job. That behaviour provides at-least-once delivery and requires idempotent processing.

Acknowledging before execution creates at-most-once behaviour:

def process_at_most_once(job: dict) -> None:
    job_queue.acknowledge(job)

    # A crash here permanently loses the job.
    execute_job(job)

Most important background workloads use at-least-once delivery because losing work is more harmful than receiving a duplicate. Exactly-once business effects are usually created through database constraints, transactions, operation IDs, or external idempotency keys rather than through delivery alone.

Idempotent Job Processing

An idempotent job can run several times without applying the same business effect more than once. Idempotency is essential because a worker may complete the operation and crash before acknowledging the job.

Setting a report status to completed is naturally idempotent. Incrementing a balance, charging a payment, or sending a reward is not.

CREATE TABLE processed_jobs (
    worker_name VARCHAR(100) NOT NULL,
    job_id VARCHAR(64) NOT NULL,
    processed_at TIMESTAMP NOT NULL,
    PRIMARY KEY (worker_name, job_id)
);

The worker can insert the job ID and apply the business update in one database transaction.

def process_reward_job(job: dict) -> None:
    with database.transaction():
        inserted = database.execute(
            """
            INSERT INTO processed_jobs (
                worker_name,
                job_id,
                processed_at
            )
            VALUES (%s, %s, NOW())
            ON CONFLICT DO NOTHING
            """,
            ["reward-worker", job["job_id"]]
        )

        if inserted.row_count == 0:
            return

        database.execute(
            """
            UPDATE customer_accounts
            SET reward_points = reward_points + %s
            WHERE customer_id = %s
            """,
            [job["points"], job["customer_id"]]
        )

A stable business-operation ID is often stronger than a queue-generated job ID. The same payment request may accidentally be published as two different jobs. A unique payment operation ID protects the real business action.

CREATE UNIQUE INDEX unique_payment_capture
ON payment_operations (capture_operation_id);

For external APIs, the same idempotency key should be reused for every retry.

payment_gateway.capture(
    payment_id=job["payment_id"],
    amount=job["amount"],
    idempotency_key=job["capture_operation_id"]
)

Retries and Backoff

Retries help workers recover from temporary failures such as network timeouts, rate limits, database failovers, or short service outages. They should not be used blindly for every error.

Failures should be classified into at least three categories:

• Transient: likely to succeed later, such as a timeout.
• Permanent: retrying will not help, such as invalid input.
• Unknown: requires inspection, reconciliation, or cautious retrying.

def handle_failure(job: dict, error: Exception) -> None:
    if isinstance(error, RateLimitError):
        schedule_retry(job, delay_seconds=calculate_backoff(job))
        return

    if isinstance(error, ValidationError):
        move_to_dead_letter_queue(job, reason=str(error))
        return

    schedule_retry(job, delay_seconds=calculate_backoff(job))

Exponential backoff increases the delay after every failure.

import random

def calculate_backoff(job: dict) -> int:
    attempt = job.get("attempt", 0)
    base_delay = 5
    maximum_delay = 900

    delay = min(base_delay * (2 ** attempt), maximum_delay)

    # Jitter prevents many workers from retrying simultaneously.
    jitter = random.randint(0, max(1, delay // 4))

    return delay + jitter

Retries should have a maximum attempt count or maximum job age. A permanently failing job must eventually move to a dead-letter queue or failed state.

def schedule_retry(job: dict, delay_seconds: int) -> None:
    next_attempt = job.get("attempt", 0) + 1

    if next_attempt >= 8:
        job_queue.move_to_dead_letter_queue(
            job,
            reason="maximum_attempts_exceeded"
        )
        return

    job["attempt"] = next_attempt
    job_queue.publish_delayed(job, delay_seconds)

Timeouts, Leases, and Heartbeats

When a worker receives a job, many queues temporarily hide it from other workers. This period may be called a visibility timeout, reservation timeout, or lease.

If the worker does not acknowledge the job before the lease expires, the queue assumes the worker failed and makes the job available again.

A timeout that is too short causes duplicate concurrent execution. A timeout that is too long delays recovery after a real crash.

Long-running jobs can extend their leases through heartbeats.

def process_large_import(job: dict) -> None:
    lease = job_queue.reserve(
        job,
        visibility_timeout_seconds=120
    )

    for batch in load_import_batches(job["import_id"]):
        process_import_batch(batch)

        # Confirm that the worker is still active.
        lease.extend(seconds=120)

    save_import_completion(job["import_id"])
    job_queue.acknowledge(job)

Lease extension must stop if the worker no longer makes progress. A frozen process that continues sending heartbeats can block recovery indefinitely.

Handling Long-Running Jobs

Long jobs are harder to retry because a failure near completion may repeat substantial work. Large jobs should often be split into smaller, independently retryable units.

A single job that processes one million records creates several problems:

• Long lease requirements.
• Expensive retries.
• Poor progress visibility.
• Difficult horizontal scaling.
• Large transactions.

A better design creates a parent job and several batch jobs.

def create_import_jobs(import_id: str, total_rows: int) -> None:
    batch_size = 1000

    for start_row in range(0, total_rows, batch_size):
        job_queue.publish({
            "job_id": generate_job_id(),
            "job_type": "process_import_batch",
            "import_id": import_id,
            "start_row": start_row,
            "end_row": min(start_row + batch_size, total_rows)
        })

Each batch records completion separately. A coordinator marks the full import complete when every batch succeeds.

Checkpointing is another option. The job stores the latest completed position and resumes from that position after failure.

def process_export(job: dict) -> None:
    checkpoint = export_jobs.get_checkpoint(job["export_id"])

    for page in read_pages_after(checkpoint):
        write_export_page(job["export_id"], page)

        export_jobs.save_checkpoint(
            export_id=job["export_id"],
            page_number=page.number
        )

Checkpoint updates and output writes should be coordinated carefully. Otherwise, a crash between them may skip data or write the same page twice.

Concurrency and Parallelism

Worker throughput can increase by running more worker processes, threads, tasks, or containers. The correct model depends on the workload.

Increasing concurrency without measuring downstream capacity can make the system slower. Hundreds of workers may overload the database, exhaust connection pools, exceed API rate limits, or compete for the same rows.

WORKER_CONCURRENCY = 20
DATABASE_POOL_SIZE = 25
EXTERNAL_API_LIMIT_PER_SECOND = 50

Concurrency should be limited by the narrowest dependency. A worker pool capable of 500 requests per second should not send that rate to an API limited to 50 requests per second.

Jobs operating on the same entity may also require locking or serialisation.

SELECT *
FROM customer_accounts
WHERE customer_id = 'customer-910'
FOR UPDATE;

Database locks protect critical sections but can reduce throughput or create deadlocks. Partitioning jobs by entity key is often more scalable when strict per-entity order is required.

Job Ordering

Queues may preserve publication order, but worker concurrency, retries, delayed jobs, and multiple partitions can change completion order.

Consider two jobs:

1. Set shipment status to in_transit.
2. Set shipment status to delivered.

If the first job fails and is retried after the second succeeds, the shipment can incorrectly return to in_transit.

Version checks prevent stale updates.

def update_shipment_status(job: dict) -> None:
    shipment = shipments.get(job["shipment_id"])

    if job["version"] <= shipment.last_processed_version:
        return

    shipments.update_status(
        shipment_id=job["shipment_id"],
        status=job["status"],
        last_processed_version=job["version"]
    )

Strict ordering can be implemented by routing jobs with the same entity ID to the same partition or worker lane.

partition_key = job["customer_id"]

job_queue.publish(
    topic="customer-jobs",
    partition_key=partition_key,
    payload=job
)

Global ordering is expensive and usually unnecessary. Per-customer, per-order, or per-account ordering is more practical.

Priority, Delayed, and Scheduled Jobs

Not every job needs immediate FIFO execution. Worker systems often support priority, delayed delivery, and scheduled execution.

Priority Jobs

Priority queues allow urgent work to move ahead of normal jobs.

Advantages

• Protects customer-facing latency.
• Supports operational overrides.
• Keeps critical jobs separate from maintenance work.

Disadvantages

• Low-priority work may starve.
• Too many priority levels become difficult to manage.

When to Use

Use priority queues when workloads have genuinely different business urgency.

Real-World Use Cases

• Password-reset emails before marketing emails.
• Customer exports before internal analytics jobs.
• Incident-recovery tasks before routine maintenance.

Example

job_queue.publish(
    queue="notifications-high",
    payload=password_reset_job
)

Delayed Jobs

Delayed jobs become available after a specified period.

Advantages

• Supports retry backoff.
• Avoids busy waiting inside workers.
• Useful for follow-up actions.

Disadvantages

• Delivery time may not be exact.
• Large numbers of delayed jobs may require specialised broker support.

When to Use

Use delayed delivery for retries, grace periods, reminders, and deferred checks.

Real-World Use Cases

• Retrying a rate-limited API request.
• Checking payment status after several minutes.
• Sending a reminder before an appointment.

Example

job_queue.publish_delayed(
    payload=payment_status_job,
    delay_seconds=300
)

Scheduled Jobs

Scheduled jobs execute at a recurring or predefined time.

Advantages

• Supports recurring maintenance and reports.
• Separates scheduling from execution.
• Uses the same worker infrastructure as ordinary jobs.

Disadvantages

• Duplicate scheduler execution may create duplicate jobs.
• Timezone and daylight-saving rules require care.

When to Use

Use scheduled jobs for periodic workloads that do not require real-time execution.

Real-World Use Cases

• Nightly report generation.
• Expired-session cleanup.
• Daily invoice reminders.

Example

def enqueue_daily_report() -> None:
    report_date = current_date()

    job_queue.publish({
        "job_id": f"daily-report:{report_date}",
        "job_type": "generate_daily_report",
        "report_date": str(report_date)
    })

The deterministic job ID prevents two scheduler instances from creating duplicate daily reports.

Graceful Shutdown and Deployments

Workers must shut down without accepting new jobs and abandoning active work. This is especially important during container replacement, autoscaling, rolling deployments, and infrastructure maintenance.

A graceful shutdown normally follows this sequence:

1. Receive a termination signal.
2. Stop reserving new jobs.
3. Allow active jobs to finish within a shutdown deadline.
4. Extend leases if necessary.
5. Release unfinished jobs before process termination.

import signal

shutdown_requested = False

def request_shutdown(signum, frame) -> None:
    global shutdown_requested
    shutdown_requested = True

signal.signal(signal.SIGTERM, request_shutdown)

def run_worker() -> None:
    while not shutdown_requested:
        job = job_queue.receive(timeout_seconds=5)

        if job is None:
            continue

        process_job(job)

    wait_for_active_jobs(timeout_seconds=30)

The infrastructure termination grace period must be longer than the worker shutdown period. Otherwise, the platform kills the process before graceful shutdown finishes.

Very long jobs should support checkpoints or cooperative cancellation rather than depending on unlimited deployment delays.

Worker code must also remain compatible with jobs published by older application versions. During rolling deployments, old and new producers and consumers may run at the same time.

def process_generate_report(job: dict) -> None:
    version = job.get("job_version", 1)

    if version == 1:
        process_report_v1(job)
        return

    if version == 2:
        process_report_v2(job)
        return

    raise PermanentError(
        f"Unsupported job version: {version}"
    )

Worker Scaling

Background workers usually scale horizontally by increasing the number of worker instances. Scaling decisions should consider more than queue depth alone.

Useful scaling signals include:

• Number of queued jobs.
• Age of the oldest job.
• Average job execution time.
• Incoming job rate.
• Current processing rate.
• CPU and memory utilisation.
• Database and external API capacity.

Queue age is often more meaningful than queue depth. A queue containing 10,000 jobs may be healthy if workers process 20,000 jobs per minute. A queue containing 100 jobs may be unhealthy if the oldest job has waited for two hours.

A simple capacity estimate is:

def required_workers(
    jobs_per_second: float,
    average_duration_seconds: float,
    target_utilisation: float = 0.70
) -> int:
    raw_workers = jobs_per_second * average_duration_seconds
    return math.ceil(raw_workers / target_utilisation)

If 20 jobs arrive each second and each job takes 0.5 seconds, approximately 10 workers are needed at full utilisation. Targeting 70% utilisation requires about 15 workers, leaving capacity for traffic variation and slower jobs.

Autoscaling must also avoid rapid oscillation. Workers need time to start, connect, and become productive. Scale-down should be gradual so active jobs can finish safely.

Backpressure and Overload Protection

A queue can absorb temporary bursts, but it cannot fix a permanently overloaded system. If producers create work faster than workers can process it for an extended period, backlog and latency continue to grow.

Backpressure limits incoming work before the system becomes unstable.

Common techniques include:

• Rejecting or delaying non-critical job creation.
• Applying per-tenant rate limits.
• Limiting producer batch sizes.
• Pausing low-priority queues.
• Reducing worker concurrency when a dependency is overloaded.
• Using bounded queues.
• Applying circuit breakers around failing services.

def submit_export(account_id: str) -> dict:
    queued_exports = job_metrics.count_queued(
        account_id=account_id,
        job_type="generate_export"
    )

    if queued_exports >= 5:
        raise TooManyRequestsError(
            "Too many exports are already queued"
        )

    return create_export_job(account_id)

Backpressure should be applied near the producer, where the application can return a clear response or defer low-value work. Allowing an unlimited backlog may create hours of delayed processing and expensive recovery.

Observability and Monitoring

Background jobs happen outside the original request, so failures are not immediately visible to users or API clients. Reliable worker systems require explicit monitoring.

Important metrics include:

Every log entry should include identifiers that connect the job to its originating request and business entity.

logger.info(
    "background_job_started",
    extra={
        "job_id": job["job_id"],
        "job_type": job["job_type"],
        "correlation_id": job["correlation_id"],
        "attempt": job.get("attempt", 0)
    }
)

Distributed traces should connect the API request, queue publication, worker execution, database calls, and external API requests. Without correlation, asynchronous failures become difficult to investigate.

Alerts should focus on service impact rather than isolated errors. One failed job may be normal if it succeeds on retry. A rapidly growing oldest-job age or dead-letter backlog usually requires immediate attention.

Real-World Report Generation Example

Consider a system that generates large customer reports. Report creation may take several minutes and produce a file in object storage. Running the operation inside an HTTP request would create timeouts and consume API capacity.

The API first creates a report record and publishes a job.

def create_report(customer_id: str) -> dict:
    report_id = generate_report_id()
    job_id = f"generate-report:{report_id}"

    with database.transaction():
        database.reports.insert({
            "report_id": report_id,
            "customer_id": customer_id,
            "status": "queued"
        })

        database.outbox.insert({
            "event_id": job_id,
            "event_type": "report.generation_requested",
            "payload": {
                "job_id": job_id,
                "job_type": "generate_report",
                "report_id": report_id,
                "customer_id": customer_id,
                "job_version": 1
            }
        })

    return {
        "report_id": report_id,
        "status": "queued"
    }

The outbox ensures that the report record and job request are stored atomically. A publisher transfers the outbox event to the queue.

def publish_outbox_events() -> None:
    for event in database.outbox.get_unpublished(limit=100):
        job_queue.publish(event["payload"])

        database.outbox.mark_published(
            event_id=event["event_id"]
        )

The publisher can send the same job twice if it crashes after publishing but before marking the event as published. The worker therefore checks the report state and uses deterministic output paths.

def generate_report(job: dict) -> None:
    report = database.reports.get(job["report_id"])

    if report.status == "completed":
        job_queue.acknowledge(job)
        return

    database.reports.mark_running(
        report_id=job["report_id"]
    )

    temporary_path = create_temporary_file()

    try:
        for page in load_report_pages(
            customer_id=job["customer_id"],
            page_size=1000
        ):
            append_page_to_report(
                temporary_path,
                page
            )

        storage_key = (
            f"reports/{job['report_id']}/report.csv"
        )

        object_storage.upload(
            source_path=temporary_path,
            destination_key=storage_key
        )

        with database.transaction():
            database.reports.mark_completed(
                report_id=job["report_id"],
                storage_key=storage_key
            )

            database.outbox.insert({
                "event_id": f"report-ready:{job['report_id']}",
                "event_type": "report.ready",
                "payload": {
                    "report_id": job["report_id"],
                    "customer_id": job["customer_id"]
                }
            })

        job_queue.acknowledge(job)

    finally:
        delete_temporary_file(temporary_path)

The deterministic storage key means repeated uploads replace the same object instead of creating multiple files. The completed report status prevents unnecessary regeneration after redelivery.

Temporary failures use delayed retries with exponential backoff.

def handle_report_job(job: dict) -> None:
    try:
        generate_report(job)

    except ObjectStorageUnavailable:
        delay = calculate_backoff(job)
        schedule_retry(job, delay)

    except InvalidReportConfiguration as error:
        database.reports.mark_failed(
            report_id=job["report_id"],
            error_code="invalid_configuration"
        )

        job_queue.move_to_dead_letter_queue(
            job,
            reason=str(error)
        )

Monitoring tracks queue age, generation duration, retry count, failed reports, file size, and worker memory. Large reports can later be divided into batch jobs if single-job execution becomes too expensive.

Common Mistakes

Production Checklist

• Define job ownership: identify which service publishes, processes, retries, and resolves each job type.
• Use stable job IDs: support tracing and duplicate detection.
• Protect business operations: use operation IDs and database uniqueness constraints.
• Acknowledge after durable processing: avoid permanent job loss.
• Expect redelivery: design handlers for at-least-once execution.
• Classify failures: distinguish transient, permanent, and uncertain errors.
• Use bounded retries: stop retrying jobs that cannot recover automatically.
• Add exponential backoff and jitter: reduce pressure on failing dependencies.
• Configure dead-letter queues: isolate jobs requiring investigation.
• Set realistic execution timeouts: prevent workers from hanging indefinitely.
• Align visibility timeouts with job duration: prevent accidental concurrent execution.
• Extend leases for active long jobs: keep ownership while useful work continues.
• Split oversized jobs: use batches, checkpoints, or workflows.
• Separate incompatible workloads: isolate jobs with different priorities or resource needs.
• Limit concurrency: protect databases and external APIs.
• Preserve per-entity ordering where required: use partitions or version checks.
• Version job payloads: maintain compatibility during rolling deployments.
• Implement graceful shutdown: stop receiving new jobs before termination.
• Use queue-age-based scaling: scale according to processing latency, not only queue depth.
• Apply backpressure: prevent unlimited backlog growth.
• Track queue depth and oldest-job age: detect capacity problems early.
• Measure success, retry, timeout, and dead-letter rates: identify reliability regressions.
• Add structured logs and traces: connect producers, queues, workers, and business operations.
• Test crash scenarios: terminate workers before, during, and after durable state changes.
• Test duplicate delivery: verify that repeated jobs do not repeat business effects.
• Document replay procedures: define how failed or dead-letter jobs return to production safely.

Conclusion

Background workers move slow, expensive, or retryable operations outside the main request-response path. They improve API responsiveness and allow asynchronous workloads to scale independently, but they also create a distributed processing system with queues, retries, acknowledgements, timeouts, deployments, and partial failures.

A reliable worker does more than call a function after receiving a message. It validates the job, protects the business operation with idempotency, stores results durably, acknowledges only after completion, classifies failures, applies bounded retries, and exposes enough metrics and logs for operational support.

Worker architecture should match the workload. Small applications may use one shared queue, while larger systems usually separate queues by latency, priority, or resource requirements. Long-running work should be divided into batches, checkpoints, or explicit workflows. Scaling should account for queue age and downstream capacity rather than only the number of waiting jobs.

Key Takeaway

A background worker is reliable only when repeated delivery, partial failure, long execution, deployment interruption, and downstream overload are treated as normal operating conditions rather than exceptional cases.

More Articles to Read

• Message Queues Explained: Producers, Consumers, and Brokers
• Event-Driven Architecture in Distributed Systems
• Message Delivery Guarantees: At-Most-Once vs At-Least-Once vs Exactly-Once
• Dead-Letter Queues, Retries, and Poison Messages
• Transactional Outbox Pattern for Reliable Messaging
• Message Queue Best Practices for Production Systems