Function job_worker 

pub async fn job_worker<AppState: AS>(
    app_state: AppState,
    registry: impl JobRegistry<AppState>,
    sleep_duration: Duration,
    max_retries: i32,
    shutdown_token: CancellationToken,
    config: JobWorkerConfig,
) -> Result<()>

Start a job worker that processes jobs from the queue.

The worker will continuously poll for jobs and execute them using the provided registry. Jobs are executed with automatic retry logic on failure.

Arguments

• app_state - The application state containing database connection and configuration
• registry - The job registry that maps job names to their implementations
• sleep_duration - How long to sleep when no jobs are available. With an idle poll gate, only permit holders sleep on this interval; permitless workers park at admission.
• max_retries - Maximum number of times to retry a failed job before moving to the dead letter queue (default: 20)
• shutdown_token - Cancellation token for graceful shutdown. When cancelled, the worker will stop accepting new jobs and release database locks before exiting.
• config - Heartbeat interval, reclaim window, local shutdown drain budget, and optional shared idle poll gate

Retry Behavior

When a job fails:

• The error count is incremented
• The error message and timestamp are recorded
• The job is requeued with exponential backoff: delay = 2^(error_count + 1) seconds (first retry: 2s, second: 4s, third: 8s, fourth: 16s, etc.)
• If error_count >= max_retries, the job is moved to the dead letter queue

Graceful Shutdown

When the shutdown_token is cancelled:

• The worker stops polling for new jobs
• Any currently executing job gets a bounded drain to complete
• Remaining owned locks are released after the drain without charging an attempt

Transient Error Resilience

Errors from the worker loop itself (e.g. the database being unreachable while trying to claim the next job, such as a pool acquire timeout during a brief database stall) are NOT fatal. They are logged at ERROR level and the worker retries with exponential backoff (starting at 1 second, doubling up to a 30 second cap, resetting after the next successful tick). This keeps a transient database hiccup from killing the worker task — and with it, apps that join on all worker tasks. Job execution failures are unaffected by this: they continue to use the per-job retry machinery described above.

Schema Check

Before polling, the worker confirms the jobs and dead_letter_jobs tables have the columns its queries use, and returns an error if the database is missing cja migrations. Without this an out-of-date schema boots fine and then fails every claim silently. A database that is merely unreachable at startup is not an error; see transient error resilience above.

Lease Timing

If a worker crashes or becomes unresponsive while processing a job, the job will remain locked in the database. JobWorkerConfig defaults to a 30-second heartbeat, a 120-second reclaim window, and the shutdown budget’s job drain (2 seconds when unset). An expired lock is charged as a failed attempt.

Example

use std::time::Duration;
use tokio_util::sync::CancellationToken;

let shutdown_token = CancellationToken::new();
let worker_token = shutdown_token.clone();

// Start worker with the default heartbeat, reclaim, and drain settings
tokio::spawn(async move {
    cja::jobs::worker::job_worker(
        app_state,
        registry,
        Duration::from_secs(60),      // poll every 60s when idle
        20,                            // max 20 retries
        worker_token,                  // for graceful shutdown
        Default::default(),
    ).await.unwrap();
});

// Later, trigger shutdown
shutdown_token.cancel();