Retries

retry_after(delay) publishes a copy after the delay, and max_attempts caps the copies through the framework's x-ruststream-retry-count header.

The runtime waits out the delay and publishes a copy of the delivery, its payload and headers as they arrived. The copy carries RETRY_COUNT_HEADER incremented, and a handler reads it like any other header. The copy waits in the memory of the service, so a process that stops before the delay is over loses it.

max_attempts(n) caps the deliveries of one message at the mount site, the first one included. ZMTP has no delivery counter, so the framework counts through that header, and the cap holds wherever the copies travel. Under a cap a plain retry travels as a copy too. The delivery that reaches the cap goes to the dead-letter destination, and with none it is dropped with a warning naming the subscription.

The copies leave through the Retry slot, which out_retry sets. On a queue the service binds, a copy goes back into the same queue. The copy carries the bytes of the delivery, so a codec named at the retry position encodes nothing. A .transform(..) there reads the delivery it retries, which is where a service marks a redelivery.

Retries and dead letters retries a failed job into its own queue with a cap.

Example

A job that fails gets five deliveries, 30 seconds apart, and each copy returns to the queue the worker binds.

Macros
Manual

examples/zmq_retries.rs
#[subscriber("jobs")]
async fn handle(job: &Job) -> HandlerOutcome {
    match run(job) {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(job.id, %err, "the job failed, retrying");
            HandlerOutcome::retry_after(Duration::from_secs(30))
        }
    }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
        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(handle)
                .max_attempts(nonzero!(5u32))
                .out_retry(Publish);
        },
    )
}

examples/manual/zmq_retries.rs
/// Works a job, and asks for another try in 30 seconds when it fails.
struct Work;

impl Handle<Job> for Work {
    fn handle(
        &self,
        job: &Job,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        ready(run(job).map_err(|err| {
            tracing::warn!(job.id, %err, "the job failed, retrying");
            HandlerOutcome::retry_after(Duration::from_secs(30))
        }))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
        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(subscriber("jobs", Work).build())
                .max_attempts(nonzero!(5u32))
                .out_retry(Publish);
        },
    )
}

Example​

Example