Гарантии доставки

Это руководство выбирает качество обслуживания для каждой подписки, показывает, что каждый исход обработчика отправляет серверу, и сохраняет сообщения, которые сервис пропускает, пока остановлен.

Пример - сервис расчётов. Счётчики сообщают о расходе энергии, и каждый отчёт - это деньги, поэтому терять их нельзя. Ещё они сообщают о потребляемой мощности для графика в реальном времени, где потерянный замер заменяет следующий.

Качество обслуживания

В MQTT три уровня качества обслуживания:

Уровень	Гарантия	Пакеты протокола
Qos::AtMostOnce (0)	не более одного раза: сообщение может потеряться	PUBLISH
Qos::AtLeastOnce (1)	хотя бы один раз: сообщение может прийти дважды	PUBLISH, PUBACK
Qos::ExactlyOnce (2)	ровно один раз между двумя сторонами	PUBLISH, PUBREC, PUBREL, PUBCOMP

Подписка называет уровень, на котором она читает, а издатель - уровень, на котором отправляет. Доставка идёт на меньшем из двух. Поэтому счётчик, который публикует расход с QoS 1, доходит до подписки с QoS 2 на уровне QoS 1.

Завершение доставки отчёта

Макросы

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()
}

Вручную

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.
struct Bill;

impl Handle<Usage> for Bill {
    async fn handle(
        &self,
        usage: &Usage,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        if usage.kwh < 0.0 {
            tracing::warn!(usage.meter, "negative usage dropped");
            return Err(HandlerOutcome::drop());
        }
        if let Err(err) = post(usage).await {
            tracing::warn!(usage.meter, %err, "usage left for the next session");
            return Err(HandlerOutcome::retry());
        }
        tracing::info!(usage.meter, usage.kwh, "usage billed");
        Ok(())
    }
}

fn usage_of_every_meter() -> MqttFilter {
    MqttFilter::new("meters/+/usage").qos(Qos::ExactlyOnce)
}

Брокер подтверждает доставку, когда обработчик возвращает управление, а не когда она приходит. Каждый исход отправляет то, что для него есть в протоколе:

• ack отправляет PUBACK при QoS 1 и PUBREC при QoS 2. Обмен QoS 2 клиент доводит до конца сам.
• drop тоже подтверждает: отрицательного подтверждения в MQTT нет, и подтверждение - единственный способ закончить работу с сообщением.
• retry ничего не отправляет. Отчёт остаётся неподтверждённым, и сервер доставит его повторно, только когда сессия сервиса возобновится на новом соединении.

У доставки с QoS 0 подтверждения нет совсем. Её исход возвращает AckError::Unsupported, и рантайм пишет предупреждение в журнал.

bill возвращает retry, когда учётная система не отвечает. Тогда отчёт ждёт следующей сессии: после перезапуска сервер передаст его снова. Как попробовать снова в той же сессии, показывает руководство Переобработка и темы dead-letter.

Пакеты при 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()
}

Вручную

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.
struct Chart;

impl Handle<[Sample]> for Chart {
    fn handle(
        &self,
        samples: &[Sample],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(count = samples.len(), "samples charted");
        ready(Ok(()))
    }
}

fn power_of_every_meter() -> MqttFilter {
    MqttFilter::new("meters/+/power").qos(Qos::AtMostOnce)
}

Обработчик, который принимает срез, - это пакетный обработчик. Пакет протокола PUBLISH содержит одно сообщение, поэтому пакеты брокер собирает на стороне клиента. Пакет закрывается, когда в нём набирается размер, указанный в точке монтирования, или через 20 миллисекунд после первого замера. Подтверждение остаётся у каждой доставки своим.

Постоянная сессия

Макросы

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"))
}

Вручную

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(
            subscriber("meters/+/usage", Bill)
                .map_source(|_| usage_of_every_meter())
                .build(),
        )
        .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(
            subscriber("meters/+/power", Chart)
                .batch(nonzero!(64))
                .build()
                .map_source(|_| power_of_every_meter()),
        )
        .out_retry(Publish::default())
        .to("meters/retry/power");
    })
}

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

fn main() -> ExitCode {
    run_main(app)
}

Сессия - это то, что сервер хранит для одного идентификатора клиента: его подписки и неподтверждённые сообщения с QoS 1 и 2. clean_start(false) возобновляет сессию, которую сохранил сервер, вместо того чтобы начать новую. session_expiry - сколько сервер хранит её после отключения сервиса, здесь сутки. Пока сервис остановлен, сервер копит в очереди сообщения с QoS 1 и 2, которые подходят к его подпискам. Когда сервис снова подключается с тем же идентификатором клиента, сервер первыми отправляет их. Брокер придерживает их, пока не откроется подписка, которой они принадлежат, поэтому при запуске ни одно не теряется. Идентификатор клиента - это имя сессии, поэтому каждому процессу сервиса нужен свой. Руководство Подключение к промышленному серверу задаёт свой идентификатор каждому развёртыванию.

По умолчанию Mosquitto не копит сообщения с QoS 0 для отсутствующей сессии. Замеры, опубликованные, пока сервис остановлен, пропадают, и графику в реальном времени это и нужно.

Порядок по счётчику

Одна подписка передаёт обработчику доставки по одной, в порядке поступления. workers(n) запускает n обработчиков сразу, и этот порядок теряется. workers(n, by_key) сохраняет его для каждого ключа: доставки с одним ключом выполняются одна за другой, а с разными ключами - параллельно. Ключ - заголовок partition-key, PARTITION_KEY_HEADER. Счётчик ставит его в каждый отчёт, например свой идентификатор. Отчёт без заголовка попадает в следующую партицию воркеров по кругу.

Управление потоком

receive_maximum(IN_FLIGHT) сообщает серверу, сколько доставок с QoS 1 и 2 сервис принимает, прежде чем подтвердит одну из них. Достигнув предела, сервер перестаёт отправлять, пока обработчик не завершит какую-нибудь доставку. Значение по умолчанию - 1000. Неподтверждённая доставка занимает своё место в этом окне до конца соединения, поэтому каждый retry без объявленного предела попыток отнимает одно место до следующей сессии. У доставок с QoS 0 такого окна нет: если обработчик медленнее своих издателей, их сообщения копятся в памяти без ограничения.

Проверка гарантий

Внутрипроцессные тесты проверяют, как bill завершает каждый отчёт и что три замера, опубликованные разом, приходят одним пакетом. Сессия, её возобновление и окно Receive Maximum существуют только на сервере, поэтому один тест работает на Mosquitto:

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

Он запускает сервис, обрывает его соединение, публикует отчёт о расходе, пока сервис остановлен, и снова запускает сервис с тем же идентификатором клиента. Отчёт, опубликованный в отсутствие сервиса, bill получает первым. Без MQTT_TEST_URL тест сразу завершается. Этот приём показывает руководство Тестирование сервиса на MQTT.