Skip to main content

Retries and dead letters

This guide retries a message later, caps its deliveries and moves what keeps failing to a dead-letter key.

Neither transport holds a message back or counts its deliveries. The framework does both: a later delivery is a copy the runtime publishes once a delay has passed, and the count travels in a header of that copy.

The example is a shipping service that ships an order once its payment has cleared.

The subscriber​

src/main.rs
/// How long an order waits before the service asks the payment service again.
const RECHECK: Duration = Duration::from_secs(2);

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
}

#[subscriber(FileStream::new("orders"))]
async fn ship(order: &Order, State(payments): State<Arc<Payments>>) -> HandlerOutcome {
    if !payments.has_cleared(order.id) {
        // The payment may clear later: ask again after a pause.
        return HandlerOutcome::retry_after(RECHECK);
    }
    tracing::info!(order.id, "order shipped");
    HandlerOutcome::ack()
}

Payments is the service's client of the payment service, kept in the app's state. While an order's payment has not cleared, ship answers retry_after and the service asks again two seconds later.

retry_after settles the delivery, waits out the delay and appends a copy of the order under the key the subscription reads. The subscription reads the copy as a new message, after everything that was already in the file. So the orders behind a waiting one are not held up. The copy carries the retry-count header (RETRY_COUNT_HEADER), raised by one. The copy waits in the service's memory until the delay is over, so a process that stops before then loses it. A replay reads a copy only while it is still short of the end of the file.

The cap and the dead-letter key​

src/main.rs
fn shipping(file: FileBroker) -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("shipping", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(file, |b| {
            // Three deliveries of one order, the first included; then the dead-letter key.
            b.include(ship)
                .max_attempts(nonzero!(3u32))
                .dead_letter("orders.dead");
            b.include(review);
        })
}

fn app() -> impl App<State = Clients> {
    shipping(FileBroker::new("/var/lib/shop/orders.ss"))
}

fn main() -> ExitCode {
    run_main(app)
}

max_attempts counts every delivery of one order, the first included. The third delivery is not retried: dead_letter appends it under orders.dead, payload and headers as they arrived. On a stream file the dead-letter key is one more stream key of the same file. A cap with no dead-letter key beside it writes the spent order nowhere.

Reading the dead letters​

src/main.rs
#[subscriber(FileStream::new("orders.dead"))]
async fn review(order: &Order, ctx: &mut Context<'_>) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
    tracing::error!(
        order.id,
        deliveries,
        "unpaid order needs a person to look at it"
    );
    HandlerOutcome::ack()
}

review reads orders.dead like any other key. The retry-count header of a dead letter holds the number of deliveries the order got, 3 here. An operator who inspects the dead letters later reads them with a replay instead.

Retries on a pipe​

A stage's standard output reaches the next process and never its own input, so a copy cannot come back to the stage that deferred it. On StdioBroker the mount site names the key the copies go to, b.include(ship).out_retry(Publish).to("orders.retry"), and a later stage reads them (out_retry). The pipeline stage of Getting started is mounted this way.

A stage that names no key does not start, because any handler may answer retry_after. The cap and the dead-letter key work on a pipe as they do on a file; a dead letter is a line under its key for a downstream stage.

Testing the delay​

The tests in the example file run the app in process on a paused clock, so tb.advance(RECHECK) lets the two seconds pass at once. One test clears the payment between two deliveries, and the order ships on the second. The dead-letter test runs both in process and on a real stream file with TestApp::start_live, where the delays pass on the real clock. Testing a file or pipeline service covers the harness.