Worker

pub struct Worker<Q: QueueSet, B: Backend>;

impl<Q: QueueSet, B: Backend> Worker<Q, B> {
    pub fn builder(backend: Arc<B>) -> WorkerBuilder<Q, B>;
    pub fn handle(&self) -> WorkerHandle;
    pub fn queues(&self) -> &[QueueConfig];
    pub fn concurrency(&self) -> usize;
    pub async fn run(self) -> Result<()>;
}

run takes self, so take the handle before you start it.

let worker = Worker::<AppQueues, _>::builder(backend)
    .handler(EmailHandler)
    .handler(FnHandler::<Ping, _>::new(|job, ctx| async move { Ok(()) }))
    .queues(&[AppQueues::Emails])
    .concurrency(32)
    .job_timeout(Duration::from_secs(30))
    .close_backend_on_shutdown(true)
    .build().await?;
let handle = worker.handle();
worker.run().await?;

WorkerBuilder

pub fn handler<H>(mut self, handler: H) -> Self
where H: JobHandler, H::Job: Job<Queue = Q>;
pub fn queues(mut self, queues: &[Q]) -> Self;
pub fn concurrency(mut self, concurrency: usize) -> Self;
pub fn job_timeout(mut self, timeout: Duration) -> Self;
pub fn close_backend_on_shutdown(mut self, close: bool) -> Self;
pub async fn build(self) -> Result<Worker<Q, B>>;

Method	Default	Meaning
handler(h)	—	Register a handler. Its job must belong to Q
queues(&[..])	every queue in Q	Consume only these queues
concurrency(n)	sum of the consumed queues’ prefetch, at least 1	Cap on jobs running at once
job_timeout(d)	none	Abort a handler running longer than d and treat it as a retryable failure
close_backend_on_shutdown(b)	false	Close the shared backend once run() returns

build() declares the selected queues, de-duplicates them by resolved name, and fails with Error::DuplicateHandler if two handlers share a Job::NAME. A duplicate is reported at build(), not at handler().

concurrency(0) is treated as 1.

Prefetch and concurrency are different knobs

prefetch is per queue and set in #[queue(prefetch = n)]; it is how many messages the broker will let one consumer hold unacknowledged. concurrency is per worker and caps how many handlers run at once. The default ties them together, which is usually right; set concurrency explicitly when a handler is CPU-bound or holds a scarce resource such as a database connection.

WorkerHandle

#[derive(Clone)]
pub struct WorkerHandle;

impl WorkerHandle {
    pub fn shutdown(&self);
    pub fn is_shutdown(&self) -> bool;
    pub fn settle_failures(&self) -> u64;
}

shutdown() is idempotent and safe to call before run() starts. settle_failures() counts failures to ack, retry, defer or dead-letter — the job ran and the broker was not told, so it will be redelivered. See Graceful shutdown.

What happens to a delivery

In order, for each message:

Condition	Action
No handler for job_type	dead_letter("no handler for job type `X`")
Payload does not decode	dead_letter("decode error: ...")
Handler returns Ok(())	ack
JobError::Fatal	dead_letter("fatal error: ...")
JobError::Deferred { delay, reason }	logged at INFO, then defer with deferrals + 1 and the queue’s max priority; the attempt is unchanged and the retry policy is never consulted
JobError::Retryable	policy.decide(attempt): either retry with attempt + 1 after the backoff, or dead_letter("max attempts (N) exhausted after attempt M: ...")
Handler panicked	becomes Retryable("handler panicked")
job_timeout elapsed	the task is aborted; becomes Retryable("job timed out after {limit:?}")

The effective retry policy is Job::retry_policy() when it is Some, and otherwise Job::QUEUE.config().retry.

What run() returns

Outcome	Return
Shutdown, everything drained	Ok(())
A consumer stream ended by itself	Err(Error::ConsumerStopped(queue))
A backend stream errored	that error
Only backend.close() failed	that error

Tracing

Each delivery runs inside:

info_span!("job", job_id, job_type, queue, attempt)

so any event a handler emits is attributed to the job. Deferrals log at INFO, retries and dead letters at WARN, settle failures at ERROR.

RUST_LOG=info,queuey_core=debug cargo run

Splitting queues across processes

// Process A: only emails, high concurrency, short timeout.
Worker::<AppQueues, _>::builder(backend.clone())
    .queues(&[AppQueues::Emails])
    .handler(EmailHandler { client })
    .concurrency(64)
    .job_timeout(Duration::from_secs(10))
    .build().await?

// Process B: only image resizing, CPU-bound, so a small cap.
Worker::<AppQueues, _>::builder(backend)
    .queues(&[AppQueues::Images])
    .handler(ResizeHandler)
    .concurrency(4)
    .build().await?

A worker only needs handlers for the job types on the queues it consumes.

See also

Handlers, Graceful shutdown, and Worker on docs.rs.