Skip to main content

Retries and dead letters

This guide retries a message now or later, caps the attempts and moves what keeps failing to a dead-letter topic.

MQTT holds no message back, counts no deliveries and has no dead-letter queue. The framework does all three: the mount site declares the cap and the dead-letter topic, and a retry travels as a copy of the message with a retry count in a header.

The example is a firmware service. It installs updates on devices, and a device may refuse an update for a while. It also stores the reports every device sends.

The subscriber​

A device refuses an update in three ways, and each needs a different outcome. Busy means it is running a job: an update offered now is refused again, so retry_after offers it in 30 seconds. Interrupted means the transfer broke off: retry tries again at once. Incompatible means the image does not fit the device, and another try gets the same answer, so the update is dropped.

src/main.rs
#[subscriber(MqttTopic::new("updates/install").qos(Qos::AtLeastOnce))]
async fn install(update: &Update) -> HandlerOutcome {
    match flash(update).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(Refusal::Busy) => {
            tracing::info!(update.device, "device busy, offering the update later");
            HandlerOutcome::retry_after(BACKOFF)
        }
        Err(Refusal::Interrupted) => {
            tracing::warn!(update.device, "transfer interrupted, retrying");
            HandlerOutcome::retry()
        }
        Err(Refusal::Incompatible) => {
            tracing::warn!(update.device, update.version, "incompatible image dropped");
            HandlerOutcome::drop()
        }
    }
}

retry_after acknowledges the delivery, waits, then publishes a copy back to updates/install. The copy carries the retry-count header (RETRY_COUNT_HEADER), raised by one. It works at every quality of service; a QoS 0 delivery has nothing to acknowledge, and its copy is published all the same. The copy waits in the service's memory: a process that stops before the delay ends loses it.

retry becomes a copy too, published at once, because the mount site below declares a cap. MQTT counts no redeliveries, so the cap counts the copies through their header. Without a cap, retry leaves the delivery unacknowledged for the next session, as Delivery guarantees shows.

The cap and the dead-letter topic​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("firmware", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "firmware-svc"),
        |b| {
            // Four deliveries of one update, the first included; the fourth refusal moves it to
            // "updates/dead", payload and headers as they arrived.
            b.include(install)
                .max_attempts(nonzero!(4u32))
                .dead_letter("updates/dead");
            b.include(review);
            b.include(record)
                .max_attempts(nonzero!(4u32))
                .dead_letter("reports/dead")
                .out_retry(Publish::default())
                .transform(ToDeliveryTopic);
        },
    )
}

max_attempts counts every delivery of one update, the first included. The fourth delivery is not retried: dead_letter publishes it to updates/dead, payload and headers as they arrived. A dropped update settles and goes nowhere; the dead-letter topic holds only the updates that ran out of attempts.

A copy is an ordinary publish, so every subscription whose filter matches its topic receives it, in this service and in any other. So each retried topic and each dead-letter topic lies outside every other filter the service reads. An updates/# subscription next to install would read every copy of every update, and every dead letter too.

Reading the dead-letter topic​

src/main.rs
/// The dead-letter topic is read like any other; the retry-count header says how many times the
/// update was delivered before it was moved there.
#[subscriber(MqttTopic::new("updates/dead").qos(Qos::AtLeastOnce))]
async fn review(update: &Update, ctx: &mut Context) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
    tracing::error!(
        update.device,
        update.version,
        deliveries,
        "update needs a person"
    );
    HandlerOutcome::ack()
}

The dead-letter topic is an ordinary topic, and review reads it like any other. The retry-count header of a dead-lettered update holds the number of deliveries it went through, 4 here.

Retries over a filter​

src/main.rs
/// Sends every copy back to the topic its delivery arrived on, the one topic of the filter's many
/// that the report belongs to.
struct ToDeliveryTopic;

impl<Options> PublishTransform<ForReply<MqttContext>, Options> for ToDeliveryTopic {
    type Destination = Names;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, MqttContext>,
    ) {
        out.set_name(cx.context(DeliveryTopic).to_owned());
    }
}

/// The `Ctx<DeliveryTopic>` parameter makes the crate's context the registration's, which is
/// what the transform reads.
#[subscriber(MqttFilter::new("devices/+/reports").qos(Qos::AtLeastOnce))]
async fn record(report: &Report, Ctx(topic): Ctx<DeliveryTopic>) -> HandlerOutcome {
    if let Err(err) = store(report).await {
        tracing::warn!(%topic, %err, "report not stored, retrying");
        return HandlerOutcome::retry_after(BACKOFF);
    }
    HandlerOutcome::ack()
}

A copy of a delivery on an MqttTopic goes back to that topic. An MqttFilter matches many topics and names none, so its registration names where a copy goes, and the service refuses to start without it. out_retry(Publish::default()).to(..) names one topic for every copy. ToDeliveryTopic names one per delivery instead: it reads the topic the report arrived on from the MQTT context and sends the copy back there. So a report from devices/dev7/reports comes back on devices/dev7/reports, to record alone. The handler takes Ctx<DeliveryTopic>, which gives the registration the MQTT context; in the manual form its context type is MqttContext.

Testing the delays​

The tests in the example file run the app in process with a paused clock, so tb.advance(BACKOFF) lets the 30 seconds pass at once. They count the deliveries of each update, read the retry-count header of each copy, and check what reaches updates/dead. Testing an MQTT service covers the harness.