Handlers

A handler is user code that processes exactly one job type. There are two ways to write one, and they are equivalent.

struct EmailHandler { client: SmtpClient }

#[async_trait]
impl JobHandler for EmailHandler {
    type Job = SendEmail;
    async fn handle(&self, job: SendEmail, ctx: JobContext) -> Result<(), JobError> {
        self.client.send(&job.to, &job.body).await.map_err(JobError::retryable)
    }
}

// The same thing without a struct.
let resize = FnHandler::<ResizeImage, _>::new(|job: ResizeImage, ctx: JobContext| async move {
    tracing::info!(path = %job.path, attempt = ctx.attempt, "resizing");
    Ok(())
});

A struct is the right choice when the handler holds dependencies, which is most of the time. FnHandler is for the handlers that hold nothing.

Capturing in a closure

FnHandler needs Fn, not FnOnce, so a closure that uses an Arc has to clone it per call:

FnHandler::<ResizeImage, _>::new({
    let settled = settled.clone();
    move |job: ResizeImage, ctx: JobContext| {
        let settled = settled.clone();
        async move {
            settled.bump();
            Ok(())
        }
    }
})

Handlers are registered on the worker builder, keyed by Job::NAME:

let worker = Worker::<AppQueues, _>::builder(backend)
    .handler(EmailHandler { client })
    .handler(resize)
    .build()
    .await?;

Registering two handlers for the same job type is an error, but it surfaces at build() rather than at handler(), as Error::DuplicateHandler.

What a handler returns

Result	Effect
Ok(())	Acked
Err(JobError::Retryable(_))	The retry policy decides: re-published after the backoff, or dead-lettered once max_attempts is reached
Err(JobError::Fatal(_))	Dead-lettered immediately, policy ignored
Err(JobError::Deferred { delay, .. })	Held for exactly delay, then re-delivered ahead of the backlog; no attempt spent

The distinction between retryable and fatal is the one judgement call a handler has to make, and it is worth making deliberately. A connection reset, a 503, a lock timeout: retryable, because the same input may well succeed later. A malformed address, a validation failure, a row that no longer exists: fatal, because nothing about waiting will change the outcome, and retrying it three times only delays the dead letter and burns the API quota.

if job.fatal {
    // Fatal errors skip the retry policy entirely.
    return Err(JobError::fatal_msg(format!("{} is not a mailbox", job.to)));
}
if ctx.attempt <= job.flaky_for {
    return Err(JobError::retryable_msg("smtp connection reset"));
}
Ok(())

Constructing a JobError

JobError::retryable(err)                  // wraps any std::error::Error
JobError::fatal(err)
JobError::retryable_msg("smtp reset")     // a plain string
JobError::fatal_msg("not a mailbox")
JobError::deferred(delay)                 // reason defaults to "deferred"
JobError::deferred_msg(delay, "rate limited")

Box<dyn Error> converts into a retryable error, so ? on most fallible calls inside a handler does the reasonable thing without a map_err. When the default is wrong, be explicit with map_err(JobError::fatal).

JobContext

Every delivery hands the handler a context describing the attempt:

pub struct JobContext {
    pub job_id: Uuid,
    pub job_type: &'static str,
    pub queue: &'static str,
    pub attempt: u32,          // 1-based
    pub max_attempts: u32,
    pub deferrals: u32,
    pub priority: u8,
    pub age: Duration,         // since first enqueue
}

job_id is stable across retries and deferrals, which makes it the right thing to log and the right thing to key an idempotency record on. attempt is 1-based: the first delivery is attempt 1. age is measured from the first enqueue, not from this delivery, so it is what to check before doing work that is only useful if it is timely.

ctx.is_last_attempt() returns attempt >= max_attempts. It lets a handler do something different when a failure now means the job is gone for good:

if let Err(e) = self.send(&job).await {
    if ctx.is_last_attempt() {
        self.alert_ops(&job, &e).await;
    }
    return Err(JobError::retryable(e));
}

Handlers should be idempotent

Delivery is at-least-once. A handler can run twice for the same job_id for reasons that have nothing to do with your code: the broker redelivers if a worker dies mid-job, and an ack that fails to reach the broker means the message comes back even though the work completed. WorkerHandle::settle_failures() counts exactly that case.

So: make the work safe to repeat. Upsert rather than insert, check before sending, or record the job_id you have already processed.

Panics and timeouts

A handler that panics does not take the worker down. The panic is caught and turned into Retryable("handler panicked"), so the retry policy decides what happens next.

A handler that exceeds the worker’s job_timeout is aborted and becomes Retryable("job timed out after {limit:?}"). There is no default timeout; set one if a hung handler would otherwise occupy a concurrency slot forever:

Worker::<AppQueues, _>::builder(backend)
    .handler(EmailHandler { client })
    .job_timeout(Duration::from_secs(30))
    .build()
    .await?

Because an abort drops the future at its next await point, a handler that must clean up should do so with a guard rather than with code after the await.

Tracing

Every delivery runs inside info_span!("job", job_id, job_type, queue, attempt), so any tracing event a handler emits is automatically attributed to the job. Successes are logged at DEBUG, deferrals at INFO, retries and dead letters at WARN, and failures to settle at ERROR.