Skip to main content

Retries and dead letters

This guide retries a message later, caps the attempts and parks what fails on a dead-letter queue.

ZeroMQ delivers a message at most once and stores nothing. A message that a handler fails on is gone unless the service publishes it again. The framework does that: the mount site declares the cap and where the copies go, and each copy carries a retry count in a header.

The example is a billing service with two workers. The job worker retries a failed job into its own queue. The invoice worker parks a failed invoice on a queue of its own, for a person to look at.

At most once​

ack and nack have nothing to tell the socket: ZeroMQ has no acknowledgement. Each of them reports AckError::Unsupported, and the runtime logs a warning. A plain retry with no cap therefore brings nothing back, and the handler is not called again. A retry that must happen goes through a copy, as below.

The job worker​

src/main.rs
use std::time::Duration;

use ruststream_zeromq::queue::prelude::*;
use serde::Deserialize;

/// How long a failed job waits before its next delivery.
const BACKOFF: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Deserialize)]
struct Job {
    id: u64,
}

#[subscriber("jobs")]
async fn work(job: &Job) -> HandlerOutcome {
    match run(job) {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(job.id, %err, "the job failed, retrying later");
            HandlerOutcome::retry_after(BACKOFF)
        }
    }
}

A job that fails asks for another try in 30 seconds with retry_after. The runtime waits out the delay and publishes a copy of the delivery, its payload and headers as they arrived. The copy waits in the memory of the service, so a process that stops before the delay is over loses it.

The invoice worker​

src/main.rs
#[derive(Debug, PartialEq, Deserialize)]
struct Invoice {
    id: u64,
    amount_cents: u64,
}

#[subscriber("invoices")]
async fn send_invoice(invoice: &Invoice) -> HandlerOutcome {
    if invoice.amount_cents == 0 {
        tracing::warn!(invoice.id, "an invoice without an amount, parking it");
        return HandlerOutcome::retry();
    }
    tracing::info!(invoice.id, invoice.amount_cents, "invoice sent");
    HandlerOutcome::ack()
}

/// Keeps the invoices nobody could send, for a person to look at.
#[subscriber("invoices.dead")]
async fn park(invoice: &Invoice) -> HandlerOutcome {
    tracing::error!(invoice.id, "invoice parked");
    HandlerOutcome::ack()
}

An invoice without an amount cannot be sent, and sending it again changes nothing. send_invoice returns retry, and the mount site turns that into a dead letter. park reads the dead letters and logs each one.

The mount sites​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    let dead = ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5561")).bindable();
    let parked = dead.bind(Publish);
    RustStream::new(AppInfo::new("billing", "0.1.0"))
        .with_broker_labeled(
            "jobs",
            ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
            |b| {
                // Five deliveries at most, the first one included; the copies return to this queue.
                b.include(work)
                    .max_attempts(nonzero!(5u32))
                    .out_retry(Publish);
            },
        )
        .with_broker_labeled(
            "invoices",
            ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5556")),
            |b| {
                // A failed invoice leaves for the second queue, under the name it is read by.
                b.include(send_invoice)
                    .dead_letter("invoices.dead")
                    .out_retry(parked);
            },
        )
        .with_broker_labeled("dead", dead, |b| {
            b.include(park);
        })
}

max_attempts counts every delivery of one job, the first included. ZMTP has no delivery counter, so the framework counts through its x-ruststream-retry-count header, and the cap holds wherever the copies travel. The fifth delivery of a failing job is not retried; with no dead-letter destination it is dropped with a warning that names the subscription.

out_retry picks the publisher the copies leave through. The jobs queue is bound by this service, so .out_retry(Publish) sends each copy back into it, and whichever job worker is free takes it.

The invoice mount declares dead_letter("invoices.dead") and no cap, so every failed invoice goes straight there. A dead letter is a publish like any other, and a bound queue takes only its own subscription's name. So the dead letters leave through parked, a policy on the dead broker, and park reads them there. The retry copies of a mount leave through the same publisher, so a mount trades its retries for its dead letters. A mount that wants both keeps its retries on its own queue and leaves dead-lettering out.

A worker that dials​

A worker that dials a ventilator cannot send a copy back: the ventilator only sends. Its mount names where the copies go, with .out_retry(policy).to("jobs") over a queue the worker binds for itself, and mounts the same handler on that queue. A mount that names no destination is refused at startup, and a Router chain that names none does not compile. out_retry shows this mount, and Pipelines with PUSH/PULL uses it.

Testing the delays​

The tests run the app in process. The job test starts on a paused clock, so tb.advance(BACKOFF) lets 30 seconds pass at once. It checks five deliveries and a last copy that carries the retry count 4. The invoice test checks that the invoice reached park after one delivery. Testing a ZeroMQ service covers the harness.