Skip to main content

Delivery guarantees

This guide chooses a quality of service per subscription, shows what each handler outcome sends to the server, and keeps the messages a service misses while it is down.

The example is a billing service. Meters report their energy usage, and every report is money, so none may be lost. They also report their power draw for a live chart, where a lost sample is replaced by the next one.

Quality of service​

MQTT has three levels of quality of service:

LevelGuaranteePackets
Qos::AtMostOnce (0)at most once: a message may be lostPUBLISH
Qos::AtLeastOnce (1)at least once: a message may come twicePUBLISH, PUBACK
Qos::ExactlyOnce (2)exactly once between two partiesPUBLISH, PUBREC, PUBREL, PUBCOMP

The subscription names the level it reads at, and the publisher names the level it sends at. A delivery arrives at the lower of the two. So a meter that publishes its usage at QoS 1 reaches a QoS 2 subscription at QoS 1.

Settling a report​

src/main.rs
/// Usage is money: the server hands it over in the four-packet handshake, and the service
/// acknowledges it only once it is posted.
#[subscriber(MqttFilter::new("meters/+/usage").qos(Qos::ExactlyOnce))]
async fn bill(usage: &Usage) -> HandlerOutcome {
    if usage.kwh < 0.0 {
        tracing::warn!(usage.meter, "negative usage dropped");
        return HandlerOutcome::drop();
    }
    if let Err(err) = post(usage).await {
        tracing::warn!(usage.meter, %err, "usage left for the next session");
        return HandlerOutcome::retry();
    }
    tracing::info!(usage.meter, usage.kwh, "usage billed");
    HandlerOutcome::ack()
}

The broker acknowledges a delivery when the handler returns, not when it arrives. Each outcome sends what the protocol has for it:

  • ack sends PUBACK at QoS 1 and PUBREC at QoS 2; the client finishes the QoS 2 handshake itself.
  • drop acknowledges too: MQTT has no negative acknowledgement, and acknowledging is its only way to be done with a message.
  • retry sends nothing. The report stays unacknowledged, and the server sends it again only when the service's session resumes on a new connection.

A QoS 0 delivery has no acknowledgement at all. Its outcome reports AckError::Unsupported, and the runtime logs a warning.

bill returns retry when the ledger does not answer. The report then waits for the next session: after a restart, the server hands it over again. Retries and dead letters shows how to try again within the same session.

Batches at QoS 0​

src/main.rs
/// Power samples feed a live chart: a lost one is replaced by the next, so they travel at
/// `QoS` 0 and are written in batches.
#[subscriber(MqttFilter::new("meters/+/power").qos(Qos::AtMostOnce))]
async fn chart(samples: &[Sample]) -> HandlerOutcome {
    tracing::info!(count = samples.len(), "samples charted");
    HandlerOutcome::ack()
}

A handler over a slice is a batch handler. A PUBLISH packet carries one message, so the broker assembles the batches on the client. A batch closes when it holds the size the mount site names, or 20 milliseconds after its first sample. Acknowledgement stays per delivery.

A persistent session​

src/main.rs
const IN_FLIGHT: NonZeroU16 = NonZeroU16::new(100).unwrap();

/// The server keeps the session of this client id for a day after the service disconnects.
fn broker(url: &str, client_id: &str) -> MqttBroker {
    MqttBroker::new(url, client_id)
        .clean_start(false)
        .session_expiry(Duration::from_hours(24))
        .receive_maximum(IN_FLIGHT)
}

fn service(broker: MqttBroker) -> RustStream {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(broker, |b| {
        b.include(bill)
            .out_retry(Publish::default())
            .to("meters/retry/usage");
        // At most 64 samples per call; a batch with fewer closes 20 ms after its first sample.
        b.include(chart.batch(nonzero!(64)))
            .out_retry(Publish::default())
            .to("meters/retry/power");
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(broker("mqtt://localhost:1883", "billing"))
}

A session is what the server keeps for one client id: its subscriptions and the QoS 1 and 2 messages it has not acknowledged. clean_start(false) resumes the session the server kept, instead of starting a new one. session_expiry is how long the server keeps it after the service disconnects, a day here. While the service is down, the server queues the QoS 1 and 2 messages that match its subscriptions. When the service connects again under the same client id, the server sends them first. The broker holds them until the subscription that owns them opens, so none is lost at startup. The client id is the session's name, so each process of the service needs its own; Connecting to production shows one per deployment.

Mosquitto does not queue QoS 0 messages for an absent session by default. The samples published while the service is down are gone, which is what a live chart wants.

Order per meter​

One subscription hands its deliveries to the handler one at a time, in the order they arrived. workers(n) runs n handlers at once and gives that order up. workers(n, by_key) keeps it per key: the deliveries of one key run one after another, and different keys run side by side. The key is the partition-key header, PARTITION_KEY_HEADER, which the meter sets on each report, such as its own id. A report without the header takes the next lane in turn.

Flow control​

receive_maximum(IN_FLIGHT) tells the server how many QoS 1 and 2 deliveries the service takes before it acknowledges one. At the limit the server stops sending until a handler settles a delivery. The default is 1000. An unacknowledged delivery holds its place in that window until the connection ends, so every retry without a declared cap takes one place away until the next session. QoS 0 deliveries have no such window: a handler slower than its publishers lets them queue in memory without a bound.

Testing the guarantees​

The in-process tests check what bill settles for each report and that three samples published at once arrive as one batch. A session, its resumption and the receive-maximum window live on a server, so one test runs against Mosquitto:

MQTT_TEST_URL=mqtt://localhost:1883 cargo test -- --test-threads=1

It starts the service, ends its connection abruptly, publishes a usage report while the service is down, and starts the service again under the same client id. The report published while the service was down is the first one bill receives. Without MQTT_TEST_URL the test returns at once. Testing an MQTT service shows the pattern.