queuey-rabbitmq

RabbitMQ backend for queuey, built on lapin 4.x.

RabbitMqBackend implements queuey_core::Backend:

• 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 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.

Reconnection is out of scope for v1: when the connection drops, consumer streams end and further calls fail.

Topology

For each logical queue q:

Delayed publishes and retries go to q.retry with a per-message expiration; when the message expires RabbitMQ routes it back to q. Because a classic queue only expires messages at its head, a long delay at the head can hold up shorter delays behind it. This is a known trade-off for v1.

Dead-lettered envelopes are published to q.dead with headers x-death-reason, x-original-queue and x-attempts. Bodies that do not decode as an Envelope are copied verbatim to q.dead with x-death-reason = "malformed envelope" and then acked, so one poison message cannot stall a consumer.

q, q.retry and q.dead are created by declare. Hold queues are not: their names depend on the delays jobs actually ask for, so defer creates them on demand and the broker deletes them again once idle.

The suffixes and whether q.dead is declared are configurable:

use queuey_rabbitmq::{RabbitMqBackend, RabbitMqOptions};

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

Deferral

Backend::defer and Delivery::defer hold a job for a delay and then put it back on q ahead of the backlog. That is what a 429 Too Many Requests with Retry-After: 30 calls for: the job did not fail and must not burn an attempt.

Two mechanisms do that:

• Hold queues. The deferral is published to q.deferred.{ttl_ms}, whose only job is to dead-letter its contents back onto q after ttl_ms. The delay is the queue's x-message-ttl, never a per-message expiration, so every message in one hold queue expires in publish order: unlike q.retry, a short deferral can never be stuck behind a long one. The price is one queue per distinct delay, so delays are rounded up to deferred_granularity (default 1s), so 29.2s and 30s share q.deferred.30000. They are never rounded down, so a job is never released early.
• Priorities. q carries x-max-priority from QueueConfig::max_priority (default Some(10)) and every publish carries the envelope's priority. Normal work is 0, a deferred envelope carries the queue's top level, so it is served before everything that piled up while it waited. "Ahead of the backlog" means ahead of what is still on the queue: a consumer with prefetch N already holds up to N backlog messages, and the returning deferral is first among what is still on the queue.

The hold queue is declared immediately before every deferred publish and never cached: an idle hold queue deletes itself one TTL after the last deferred publish to it (x-expires = 2 * TTL), and every declare resets that timer. So a delay still in occasional use keeps its queue, and one that falls out of use is cleaned up by the broker.

x-expires is deliberately not tunable. A hold queue's arguments are a pure function of its name, so two processes running different builds compute identical arguments for q.deferred.30000. Were the expiry a setting, a process with a different value would be answered PRECONDITION_FAILED on every deferral, for ever, with no way out but deleting the queue.

use std::time::Duration;
use queuey_rabbitmq::RabbitMqOptions;

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

A zero deferred_granularity is clamped to one millisecond rather than rejected, because library code does not panic on configuration. It does mean up to one hold queue per distinct millisecond, which is almost never what you want.

What deferral requires

• The queue must have been declared through this backend, in this process. Otherwise the hold queue's durability and the queue it dead-letters back to would be guesses, and a TTL expiry into a queue that does not exist is discarded silently by the broker. Unlike a mandatory publish, nothing is returned and nothing is logged. Deferring onto an unknown queue is Error::UnknownQueue instead. Producer::new and WorkerBuilder::build declare the queue set; Producer::new_undeclared deliberately does not, so a producer built that way can enqueue but not defer.
• The delay must fit. It is capped at topology::MAX_DEFERRAL_MS, about 24.8 days. That is half of what a 32-bit millisecond TTL can express, because a hold queue's x-expires is twice its TTL. A longer delay is refused, not clamped: releasing a job early is the one thing a deferral promises not to do. Rounding up to the granularity happens first, so a delay just under the cap can be refused too.
• The queue name must leave room for its hold queues. A 250-byte queue name is legal, but {q}.deferred.2147483647 is not, so declare refuses such a name up front rather than letting deferrals fail one job at a time later.

All of these fail before anything is acked, so from Delivery::defer the original message stays unacknowledged and the broker redelivers it.

Upgrading

x-max-priority is a breaking topology change. It is a declaration argument, and RabbitMQ refuses to change the arguments of an existing queue: the declaration comes back as PRECONDITION_FAILED, which closes the channel and surfaces as an error from declare.

A q created before this feature has no x-max-priority, so declaring it again with the default config will fail. Two options:

• drain and delete q, then let the backend redeclare it. Deferred jobs then come back ahead of the backlog; or
• set max_priority = 0 (QueueConfig::max_priority(0), or #[queue(max_priority = 0)]), which declares q exactly as before. Deferral still works; the returning job queues up FIFO with everything else.

q.retry, q.dead and hold queues never carry x-max-priority, so only q is affected.

RabbitMqOptions::deferred_queue_grace is gone. A hold queue's x-expires is now always 2 * TTL, computed from the TTL in its name and nothing else. The setting was unsafe by construction: two processes with different graces agreed on the name q.deferred.30000 but disagreed on its arguments, and the broker then refused one of them with PRECONDITION_FAILED on every single deferral until somebody deleted the queue. Drop the builder call; there is no replacement and no broker-side migration. Existing hold queues expire on their own.

Tests

Unit tests (topology naming, queue arguments including x-max-priority and the hold queue's TTL / x-expires arithmetic, property and header mapping, option defaults) need no broker and run with a plain:

cargo test -p queuey-rabbitmq

The integration tests in tests/broker.rs need a real RabbitMQ. They are not #[ignore]d. Each one early-returns with a skip notice when AMQP_URL is unset. To run them:

docker run --rm -p 5672:5672 rabbitmq:4-management
AMQP_URL=amqp://guest:guest@localhost:5672/%2f cargo test -p queuey-rabbitmq

Each test uses queue names carrying a fresh UUID and deletes them at the end (hold queues included), so runs can share a broker.