Testing your jobs

MemoryBackend implements Backend in-process, with no broker. It honours retry delays and deferrals through tokio::time, so a test can run on paused virtual time and a five-minute backoff costs no wall time at all.

#[tokio::test(start_paused = true)]
async fn flaky_email_is_retried_then_delivered() -> queuey::Result<()> {
    let backend = Arc::new(MemoryBackend::new());
    let producer = Producer::<AppQueues, _>::new(backend.clone()).await?;
    let worker = Worker::<AppQueues, _>::builder(backend.clone())
        .handler(EmailHandler)
        .build()
        .await?;
    let handle = worker.handle();
    let running = tokio::spawn(worker.run());

    producer.enqueue(&SendEmail { to: "a@b.c".into(), body: "hi".into() }).await?;
    tokio::time::sleep(Duration::from_secs(10)).await; // virtual

    handle.shutdown();
    running.await.unwrap()?;

    assert_eq!(backend.pending("myapp.emails"), 0);
    assert!(backend.dead_letters("myapp.emails").is_empty());
    Ok(())
}

What you can assert on

Helper	Returns
`pending(queue)`	Messages waiting to be consumed
`acked(queue)`	Every envelope acked, in order
`deferred(queue)`	Envelopes still in hold
`dead_letters(queue)`	Dead-lettered envelopes with their reason
`queue_names()`	Every declared queue name, sorted
`queue_config(queue)`	The `QueueConfig` a queue was declared with
`is_closed()`	Whether `close()` has been called

A retry acks the envelope it replaces, so acked counts attempts, not jobs. A job that succeeds on its third attempt appears three times, all with the same job_id and with attempt 1, 2 and 3.

let acked = backend.acked("e2e.emails");
assert_eq!(acked.iter().map(|e| e.attempt).collect::<Vec<_>>(), vec![1, 2, 3]);
assert!(acked.iter().all(|e| e.job_id == id), "same job id throughout");

Waiting on a paused clock

With start_paused = true the clock only advances when every task is idle, so a plain sleep is enough to let a backoff or a hold expire. When a test needs to wait for a condition rather than a duration, poll with a short virtual sleep:

/// Polls `cond` while letting the paused clock (and so retry and hold delays) advance.
async fn wait_for(what: &str, mut cond: impl FnMut() -> bool) {
    for _ in 0..1_000 {
        if cond() {
            return;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    panic!("timed out waiting for: {what}");
}

Asserting on the configuration itself

The derived QueueSet implementation is ordinary code, so the attributes are testable without running anything:

assert_eq!(AppQueues::all(), &[AppQueues::Emails, AppQueues::Images, AppQueues::SlowApi]);
assert_eq!(AppQueues::Emails.name(), "e2e.emails");
assert_eq!(AppQueues::from_name("e2e.img"), Some(AppQueues::Images));

let emails: QueueConfig = AppQueues::Emails.config();
assert_eq!(emails.prefetch, 4);
assert_eq!(emails.retry.max_attempts, 3);
assert!(matches!(emails.retry.backoff, Backoff::Exponential { .. }));

// Priority levels: the default unless the attribute says otherwise, and `0`
// means "not a priority queue".
assert_eq!(emails.max_priority, Some(DEFAULT_MAX_PRIORITY));
assert_eq!(AppQueues::SlowApi.config().max_priority, None);

// The queue itself does not retry; the job overrides it.
assert_eq!(AppQueues::Images.config().retry, RetryPolicy::none());
assert_eq!(
    ResizeImage::retry_policy(),
    Some(RetryPolicy::fixed(2, Duration::from_millis(500)))
);
assert_eq!(SendEmail::retry_policy(), None);
assert_eq!(SendEmail::QUEUE, AppQueues::Emails);

Making ordering observable

While a consumer is running, a backlog drains the instant it is enqueued, so the order in which work comes off a queue is not observable. To test that a deferred job overtakes the backlog, stop the worker, build the backlog, let the hold expire, then start a fresh worker on the same backend:

handle.shutdown();
task.await.unwrap().unwrap();

producer.enqueue(&CallApi::new("backlog-1")).await.unwrap();
producer.enqueue(&CallApi::new("backlog-2")).await.unwrap();
assert_eq!(backend.pending(EMAILS), 2);

wait_for("the hold to expire", || backend.deferred(EMAILS) == 0).await;
assert_eq!(backend.pending(EMAILS), 3);

let (handle, task) = spawn_worker(backend.clone(), log.clone()).await;
wait_for("all four runs", || log.runs() == 4).await;

assert_eq!(
    log.order(),
    vec!["limited", "limited", "backlog-1", "backlog-2"],
    "the deferred job must overtake the backlog that built up while it waited"
);

Setting concurrency(1) on the worker also helps: with one job in flight at a time, the order the backend hands work out is the order your handler sees it.

Two behaviours worth knowing

After close(), declare, publish, defer, retry and dead_letter all return Error::ShutDown, while a plain ack still records. That models a broker going away mid-flight.

Dropping a consumer stream requeues whatever it had pulled but not yet delivered, so a worker that stops does not swallow messages.

Running the library’s own tests

cargo test --workspace                       # unit + in-memory tests; broker tests print "skipping"
docker run --rm -d -p 5672:5672 -p 15672:15672 rabbitmq:4-management
AMQP_URL=amqp://guest:guest@localhost:5672/%2f cargo test --workspace   # + RabbitMQ integration tests

The broker-backed tests skip themselves when AMQP_URL is unset, so a plain cargo test needs nothing installed.