Graceful shutdown

WorkerHandle::shutdown() asks a running worker to stop. It is cloneable, Send, idempotent, and safe to call before run() has even started.

let worker = Worker::<AppQueues, _>::builder(backend)
    .handler(EmailHandler)
    .build()
    .await?;
let handle = worker.handle();
let running = tokio::spawn(worker.run());

tokio::signal::ctrl_c().await?;
handle.shutdown();
running.await??;

The contract

shutdown() makes run() finish in this order, and only then return:

1. The consumer tasks stop pulling from their streams. A delivery is pulled and forwarded to the dispatch loop in one step, so none is ever held and then discarded.
2. Everything already pulled is dispatched to its handler and settled normally. “Stop consuming” never means “throw away what is in hand”.
3. The jobs that were already running are awaited. The concurrency cap still applies, so this drains at the same rate normal work does.

There is no timeout on step 3. A handler that hangs forever holds shutdown open forever, which is what job_timeout is for.

The backend stays open

The backend is an Arc, normally shared with a Producer and possibly other workers, so run() does not close it. The owner closes it once run() has returned:

handle.shutdown();
running.await??;
backend.close().await?;

When the process exists only to run the worker, opt in instead:

Worker::<AppQueues, _>::builder(backend)
    .handler(EmailHandler)
    // This process exists to run the worker, so let it close the connection too.
    // (The default is to leave the shared backend alone.)
    .close_backend_on_shutdown(true)
    .build()
    .await?

Catching a signal

A worker process wants both: stop on SIGINT, and stop when the work is done.

let timed_out = tokio::select! {
    () = all_settled => false,
    _ = tokio::time::sleep(DEADLINE) => true,
    _ = tokio::signal::ctrl_c() => {
        tracing::warn!("interrupted; shutting down");
        false
    }
};

// Graceful: stop consuming, finish the jobs in flight, close the connection.
handle.shutdown();
running.await??;

In a container, make sure the signal actually reaches the process: a shell-form CMD puts /bin/sh at PID 1 and it will not forward SIGTERM. Use the exec form, and set the orchestrator’s termination grace period longer than your slowest handler.

What run() returns

Outcome	Return
shutdown(), everything drained	Ok(())
A consumer stream ended on its own	Err(Error::ConsumerStopped(queue))
A backend stream produced an error	That error
Only backend.close() failed	That error

ConsumerStopped is the one to understand: it means the broker or the connection went away without anyone asking for a shutdown.

Reconnection is out of scope

When the connection to the broker drops, consumer streams end and run() returns an error. queuey does not reconnect. Supervise the process — systemd, Kubernetes, a container restart policy — and let it start again, which re-declares the topology and resumes consuming.

settle_failures

handle.settle_failures()   // -> u64

This counts failures to ack, retry, defer or dead-letter. Each one is also logged at ERROR.

It means: the job ran, and the broker was not told. The message will therefore be redelivered and the work will happen twice. A non-zero, growing count is a signal that duplicate processing is ahead — not that work was lost. It is the single most useful number to put on a dashboard, because it is normally exactly zero.

This is also the concrete reason handlers should be idempotent. See Handlers.

Checking the state

assert!(!h.handle.is_shutdown());
h.stop().await.expect("graceful shutdown is not an error");

// The backend is shared with the producer, so the worker leaves it open unless
// `close_backend_on_shutdown(true)` says otherwise; closing it is the caller's call.
assert!(!backend.is_closed());
backend.close().await.unwrap();
assert!(backend.is_closed());

Deploying a new version

Because retries and deferrals wait in broker-side hold queues rather than in the worker, a rolling restart loses none of them. The sequence that avoids dead letters:

1. Deploy workers that understand the new job types first.
2. Then deploy the producers that enqueue them.

The reverse order puts jobs on a queue whose workers have no handler for them, and those get dead-lettered rather than waiting politely. See Dead letters.