retry_after

retry_after(delay) asks for the delivery back no sooner than the delay, for a condition that will pass.

It is one of the outcomes, next to an immediate retry(). A broker with native delayed redelivery holds the delivery and returns it after the delay. The in-memory broker is one: the delivery comes back to the same subscriber, and nothing is published. On a broker without it, the runtime drops the delivery and publishes a copy back to the subscription after the delay, through the Retry slot. That copy carries the RETRY_COUNT_HEADER (x-ruststream-retry-count), incremented on every copy. The copy is at most once over the delay: a process that exits before the timer fires loses it. A graceful shutdown waits for pending copies within the app's shutdown_timeout. A zero delay redelivers at once, as retry() does; write retry() when no delay is meant. A batch body sets a delay per element, and only those elements come back. A test runs on a paused clock (#[tokio::test(start_paused = true)]), and tb.advance(delay) brings the delivery back, as the advance page shows. How many times a delivery may come back is the registration's declaration, max_attempts.

Example

Macros

examples/retry.rs

const RECHECK: Duration = Duration::from_secs(5);

#[subscriber("payments")]
async fn reconcile(
    payment: &Payment,
    ctx: &mut Context<'_, (), Arc<Ledger>>,
) -> HandlerOutcome {
    if !ctx.state().is_cleared(payment.id) {
        // The bank has not cleared it yet: an immediate redelivery would only spin.
        return HandlerOutcome::retry_after(RECHECK);
    }
    tracing::info!(payment.id, "reconciled");
    HandlerOutcome::ack()
}

Manual

examples/manual/retry.rs

const RECHECK: Duration = Duration::from_secs(5);

struct Reconcile;

impl Handle<Payment, (), (), (), Arc<Ledger>> for Reconcile {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        ctx: &mut Context<'_, (), Arc<Ledger>>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if !ctx.state().is_cleared(payment.id) {
            // The bank has not cleared it yet: an immediate redelivery would only spin.
            return ready(Err(HandlerOutcome::retry_after(RECHECK)));
        }
        tracing::info!(payment.id, "reconciled");
        ready(Ok(()))
    }
}

Per element of a batch

Macros

examples/retry.rs

const RECHECK: Duration = Duration::from_secs(30);

/// One outcome per element: a cleared payment is acked, a pending one alone comes back.
#[subscriber("payments")]
async fn reconcile(
    payments: &[Payment],
    ctx: &mut Context<'_, (), Arc<Ledger>>,
) -> Vec<HandlerOutcome> {
    payments
        .iter()
        .map(|payment| {
            if ctx.state().is_cleared(payment.id) {
                HandlerOutcome::ack()
            } else {
                HandlerOutcome::retry_after(RECHECK)
            }
        })
        .collect()
}

Manual

examples/manual/retry.rs

const RECHECK: Duration = Duration::from_secs(30);

/// One outcome per element: a cleared payment is acked, a pending one alone comes back.
struct Reconcile;

impl Handle<[Payment], (), (), (), Arc<Ledger>> for Reconcile {
    fn handle(
        &self,
        payments: &[Payment],
        _outs: &(),
        ctx: &mut Context<'_, (), Arc<Ledger>>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        ready(Err(payments
            .iter()
            .map(|payment| {
                if ctx.state().is_cleared(payment.id) {
                    HandlerOutcome::ack()
                } else {
                    HandlerOutcome::retry_after(RECHECK)
                }
            })
            .collect()))
    }
}

fn app(broker: MemoryBroker<Retaining>) -> impl App<State = Arc<Ledger>> {
    RustStream::new(AppInfo::new("payments", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Arc::new(Ledger::default())))
        .with_broker(broker, |b| {
            b.include(
                subscriber("payments", Reconcile)
                    .batch(nonzero!(2))
                    .start_at(MemoryPosition::start())
                    .build(),
            );
        })
}