retry_after

retry_after(delay) acknowledges the delivery, waits, then publishes a copy of the message with its retry count in a header.

MQTT holds no message back and counts no deliveries, so the runtime does both. It acknowledges the original, waits out the delay and publishes the payload and the headers as they arrived. The copy carries RETRY_COUNT_HEADER, raised by one on every copy, and the handler reads it like any other header. An MqttTopic subscription receives its copies on its own topic; an MqttFilter names where they go with out_retry. The copy waits in the service's memory, so a process that exits before the delay is over loses it. A QoS 0 delivery has no acknowledgement, and its copy is published all the same. A copy is an ordinary publish, so every subscription whose filter matches its topic receives it, in this service and in any other. A retried topic therefore lies outside every other filter the service reads, or another handler reads the copy too. max_attempts caps the deliveries of one message.

Example

A reading the database timed out on is written again five seconds later.

Macros

examples/mqtt_retries.rs

/// One topic is a name a publisher can use, so the copy goes back to this topic and this
/// subscription reads it again, with the retry-count header incremented.
#[subscriber(MqttTopic::new("meters/readings").qos(Qos::AtLeastOnce))]
async fn store(reading: &Reading) -> HandlerOutcome {
    if let Err(err) = write(reading).await {
        tracing::warn!(reading.meter, %err, "write failed, retrying");
        return HandlerOutcome::retry_after(BACKOFF);
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/mqtt_retries.rs

/// One topic is a name a publisher can use, so the copy goes back to this topic and this
/// subscription reads it again, with the retry-count header incremented.
struct Store;

impl Handle<Reading> for Store {
    async fn handle(
        &self,
        reading: &Reading,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        if let Err(err) = write(reading).await {
            tracing::warn!(reading.meter, %err, "write failed, retrying");
            return Err(HandlerOutcome::retry_after(BACKOFF));
        }
        Ok(())
    }
}

fn readings() -> MqttTopic {
    MqttTopic::new("meters/readings").qos(Qos::AtLeastOnce)
}