Backend and Delivery

The core is transport-agnostic. MemoryBackend and RabbitMqBackend are two implementations of the same pair of traits, and nothing stops you writing a third.

Backend

#[async_trait]
pub trait Backend: Send + Sync + 'static {
    async fn declare(&self, queues: &[QueueConfig]) -> Result<()>;
    async fn publish(&self, envelope: &Envelope, delay: Option<Duration>) -> Result<()>;
    async fn defer(&self, envelope: &Envelope, delay: Duration) -> Result<()>;
    async fn consume(&self, queue: &QueueConfig) -> Result<DeliveryStream>;
    async fn close(&self) -> Result<()>;
}

Method	Contract
`declare`	Idempotently create all queues, plus any retry or dead-letter infrastructure
`publish`	Publish `envelope` to `envelope.queue`, optionally after `delay`
`defer`	Publish into a hold that releases onto `envelope.queue` after `delay`
`consume`	Start consuming `queue` with its prefetch. The stream ends when the backend is closed or the connection is lost
`close`	Stop all consumers, flush, close connections

publish with a delay and defer differ only in the priority the envelope carries, which the caller has already set. Both wait; only one comes back to the front.

Delivery

#[async_trait]
pub trait Delivery: Send + 'static {
    fn envelope(&self) -> &Envelope;
    async fn ack(self: Box<Self>) -> Result<()>;
    async fn dead_letter(self: Box<Self>, reason: &str) -> Result<()>;
    async fn retry(self: Box<Self>, next: Envelope, delay: Duration) -> Result<()>;
    async fn defer(self: Box<Self>, next: Envelope, delay: Duration) -> Result<()>;
}

pub type DeliveryStream =
    Pin<Box<dyn Stream<Item = Result<Box<dyn Delivery>>> + Send>>;

Every settling method takes self: Box<Self>, so a delivery can be settled exactly once and the type system enforces it.

What the worker expects

An implementation that honours these holds no surprises for the worker:

Prefetch is respected. No more than QueueConfig::prefetch unacknowledged deliveries per consumer. Unbounded delivery defeats the worker’s concurrency cap.

Delays never fire early. Round up, never down. A job released before its Retry-After has elapsed defeats the whole point of a deferral.

Priority is honoured when the queue has levels. A deferred envelope arriving behind the backlog is not a deferral.

Dropping a stream does not lose messages. Whatever was pulled but not delivered must become available again. MemoryBackend requeues on drop; RabbitMQ does it by closing the channel.

After close, publishing operations fail with Error::ShutDown and consumer streams end.

Reporting errors

Wrap transport failures with the helper, which boxes the source:

impl Error {
    pub fn backend<E: std::error::Error + Send + Sync + 'static>(e: E) -> Self;
}

Use Error::UnknownQueue(name) when a delayed publish targets a queue this backend has never declared, since it needs the queue to exist to release into.

Never panic on configuration. Clamp an implausible value and carry on, the way the RabbitMQ backend clamps a sub-millisecond granularity to 1 ms.

The two in-tree implementations

	`MemoryBackend`	`RabbitMqBackend`
Transport	in-process	AMQP, via `lapin`
Delays	`tokio::time`, so virtual time works	broker-side hold queues
Durability	none	persistent messages, durable queues
Inspection	`pending`, `acked`, `deferred`, `dead_letters`	the management UI
Use	tests, examples	production

MemoryBackend is worth reading before writing your own: it is the smallest complete implementation of the contract above.