Retries

max_attempts caps how many deliveries a message gets, counted in the framework's retry-count header that every copy carries.

max_attempts(n) follows include at the mount site; n counts the first delivery too. Neither transport counts redeliveries, so the count is RETRY_COUNT_HEADER, which the runtime writes on every copy. A handler reads the header off its context to know which attempt it runs. Under a cap, an immediate retry travels as a copy too, because the count lives in the header. The delivery that reaches the cap is rejected and no copy follows it. A cap with no dead-letter key beside it writes the spent delivery nowhere. A registration mounted by a bare stream key takes the cap the same way as one mounted by a FileStream. The guide Retries and dead letters caps a service's retries on a stream file and on a pipe.

Example

The printer is jammed, so every attempt asks for another; the third delivery is the last.

Macros

examples/file_max_attempts.rs

/// Prints a parcel label; the printer is jammed, so every attempt asks for another.
#[subscriber(FileStream::new("labels"))]
async fn print(label: &Label, ctx: &mut Context<'_>) -> HandlerOutcome {
    let attempt = ctx
        .headers()
        .get_str(RETRY_COUNT_HEADER)
        .and_then(|count| count.parse::<u32>().ok())
        .unwrap_or(0);
    tracing::info!(label.parcel, attempt, "printer jammed");
    HandlerOutcome::retry_after(RETRY_DELAY)
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("labels", "0.1.0")).with_broker(
        FileBroker::new("/var/lib/shop/labels.ss"),
        |b| {
            b.include(print).max_attempts(nonzero!(3u32));
        },
    )
}

Manual

examples/manual/file_max_attempts.rs

/// Prints a parcel label; the printer is jammed, so every attempt asks for another.
struct Print;

impl Handle<Label> for Print {
    fn handle(
        &self,
        label: &Label,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let attempt = ctx
            .headers()
            .get_str(RETRY_COUNT_HEADER)
            .and_then(|count| count.parse::<u32>().ok())
            .unwrap_or(0);
        tracing::info!(label.parcel, attempt, "printer jammed");
        ready(Err(HandlerOutcome::retry_after(RETRY_DELAY)))
    }
}

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("labels", "0.1.0")).with_broker(
        FileBroker::new("/var/lib/shop/labels.ss"),
        |b| {
            b.include(subscriber(FileStream::new("labels"), Print).build())
                .max_attempts(nonzero!(3u32));
        },
    )
}