retry_after

retry_after(delay) holds the delivery under its lease for the delay, then nacks it, so the subscription redelivers it once the delay is over.

Pub/Sub has no delayed nack, so the crate keeps the delivery in the process for the delay. The client goes on extending the delivery's lease while it waits, so the service does not redeliver it early. The wait runs on a task of its own, and the subscription keeps delivering meanwhile. The held delivery counts toward max_outstanding until it comes back, and its redelivery counts as an attempt toward a dead-letter policy. Nothing is published: the message itself comes back, with its payload and its attributes, and with the next attempt number where the subscription counts attempts.

The delay must fit inside the subscription's max_lease, an hour by default. A longer one is refused with PubSubError::DelayBeyondLease, and the delivery comes back at once. A process that stops while it holds a delivery stops extending its lease, and the service redelivers the message when the lease runs out: the delay is lost, the message is not.

Example

A payment the bank has not settled yet comes back in thirty seconds. The test advances the paused clock and sees the second delivery only once the thirty seconds are over.

Macros

examples/pubsub_settlement_retry_after.rs

/// A payment the bank has not settled yet comes back in thirty seconds, not at once.
#[subscriber(GooglePubSub::new("payments-workers").create_with_topic("payments"))]
async fn reconcile(payment: &Payment) -> HandlerOutcome {
    if !payment.settled {
        return HandlerOutcome::retry_after(RECHECK);
    }
    tracing::info!(payment.id, "payment reconciled");
    HandlerOutcome::ack()
}

Manual

examples/manual/pubsub_settlement_retry_after.rs

/// A payment the bank has not settled yet comes back in thirty seconds, not at once.
struct Reconcile;

impl Handle<Payment> for Reconcile {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if !payment.settled {
            return ready(Err(HandlerOutcome::retry_after(RECHECK)));
        }
        tracing::info!(payment.id, "payment reconciled");
        ready(Ok(()))
    }
}