The RabbitMQ backend

RabbitMqBackend is the production backend, built on lapin. It is behind the rabbitmq feature, which is on by default.

let backend = Arc::new(
    RabbitMqBackend::connect("amqp://guest:guest@localhost:5672/%2f").await?
);

The URI is a standard AMQP URI. Note the %2f: that is the default virtual host /, percent-encoded, and leaving it off connects to a vhost named nothing at all.

For local development:

docker run --rm -d -p 5672:5672 -p 15672:15672 rabbitmq:4-management

The examples read AMQP_URL with a localhost fallback, which is a pattern worth copying:

let url = std::env::var("AMQP_URL").unwrap_or_else(|_| DEFAULT_URL.to_owned());
tracing::info!(%url, "connecting");
let backend = Arc::new(RabbitMqBackend::connect(&url).await?);

Share one backend

One RabbitMqBackend is one connection. Wrap it in an Arc and give the same one to the producer and every worker in the process. Do not open a connection per producer:

let backend = Arc::new(RabbitMqBackend::connect(&url).await?);

let producer = Producer::<AppQueues, _>::new(backend.clone()).await?;
let worker = Worker::<AppQueues, _>::builder(backend.clone())
    .handler(EmailHandler)
    .build()
    .await?;

The channel model

Channels are cheap; the failure modes of sharing one are not. The backend splits them deliberately:

• one lapin::Connection;
• one publishing channel in confirm mode, shared behind a tokio::sync::Mutex. Every publish — enqueue, retry, defer, dead-letter — waits for the broker’s confirmation. Nothing is ever declared on it;
• one channel for the hold queue declarations that every retry, delayed enqueue and defer makes on demand. A declaration is the one thing the broker routinely refuses (PRECONDITION_FAILED closes the channel it ran on), so it is kept away from the publishes it would otherwise take down with it;
• one fresh channel per consume call, with basic_qos(prefetch, global = false);
• one throwaway channel per declare, so a rejected declaration cannot poison the other channels.

The publishing channel is reopened lazily if a channel exception closed it.

Publisher confirms

Every publish goes to the default exchange ("") with the queue name as the routing key, mandatory: true, and is awaited for a confirm. Only a bare ack counts as success. An unroutable message returned by the broker, a nack, or confirms not being enabled all become errors, so enqueue returning Ok means the broker has the message.

The errors you might see, all wrapped in Error::Backend:

broker nacked the message published to `myapp.emails`
broker returned the message published to `myapp.emails` as unroutable: 312 NO_ROUTE
publisher confirms are not enabled; cannot confirm publish to `myapp.emails`

312 NO_ROUTE means the queue does not exist. That is almost always a producer built with new_undeclared against a topology nobody created.

Reconnection is out of scope

When the connection drops, consumer streams end, Worker::run returns an error, and later calls fail. queuey does not reconnect. Run the worker under a supervisor and let the process restart; build() re-declares the topology on the way back up.

Options

use queuey_rabbitmq::{RabbitMqBackend, RabbitMqOptions};

let backend = RabbitMqBackend::with_options(
    "amqp://guest:guest@localhost:5672/%2f",
    RabbitMqOptions::default()
        .dead_suffix("-dlq")
        .declare_dead_letter_queues(true),
)
.await?;

Option	Default	Effect
connection_properties	ConnectionProperties::default()	Client properties, locale, executor
dead_suffix	".dead"	Suffix for the dead-letter queue name
declare_dead_letter_queues	true	Whether queuey declares and publishes to q.dead at all
deferred_suffix	".deferred"	Suffix in hold queue names
retry_granularity	1s	Rounding for retry backoffs and enqueue_after
deferred_granularity	1s	Rounding for defer

Every builder takes and returns self. Full field-level detail is in RabbitMqOptions.

Options must match across processes

dead_suffix and deferred_suffix are part of the queue names the backend computes and declares. Two processes with different suffixes declare different topologies against the same work queues. Set them in one place.

Granularity is a real trade-off

A backoff is a heuristic and tolerates coarse rounding. A Retry-After is a contract. That is why the two have separate knobs:

Option	Applies to	Default
retry_granularity	Delivery::retry (backoff) and enqueue_after	1s
deferred_granularity	Delivery::defer and Producer::defer	1s

Each distinct rounded delay gets its own hold queue. Exponential backoff with jitter produces a different delay on every retry, so retry_granularity bounds how many hold queues a busy, failing queue can have at once: with a policy capped at five minutes, 1s allows up to 300, 10s up to 30.

let options = RabbitMqOptions::default()
    .deferred_suffix(".deferred")                            // default
    .retry_granularity(Duration::from_secs(10))              // default 1s
    .deferred_granularity(Duration::from_secs(1));           // default

Rounding is always up, so raising retry_granularity makes retries later, never earlier. A zero or sub-millisecond granularity is clamped to 1 ms rather than rejected: library code does not panic on configuration.

What declare() creates

Producer::new and WorkerBuilder::build both call declare, which idempotently creates each work queue and its .dead companion. Hold queues are not created here; they appear on demand and expire on their own.

declare also pre-checks every queue name against the longest hold-queue name it could ever need, and fails before creating anything if that would exceed AMQP’s 255-byte limit:

queue `...` leaves no room for its hold queues: `...` is 261 bytes, over the 255-byte AMQP limit

That is a startup-time failure rather than a surprise the first time something retries.

Identifying consumers

Consumer tags are queuey.{queue}.{nanos:x}.{sequence}, with the queue name truncated to 160 bytes. They show up in the management UI’s Consumers tab, which makes it easy to see which process is consuming what.

Watching it work

With the management UI open on http://localhost:15672, run the end-to-end example and watch:

cargo run -p queuey --example rabbitmq_end_to_end

aq-example.emails and aq-example.emails.dead appear at startup. As the flaky job fails, aq-example.emails.deferred.500 and .deferred.1000 appear, hold one message each for their TTL, and then delete themselves.