Переобработка и темы dead-letter

Это руководство обрабатывает сообщение повторно сразу или позже, ограничивает число попыток и переносит то, что раз за разом не обрабатывается, в тему dead-letter.

MQTT не придерживает сообщения, не считает доставки, и очереди dead-letter в нём нет. Всё это делает фреймворк: точка монтирования объявляет предел попыток и тему dead-letter, а для переобработки публикуется копия сообщения с числом попыток в заголовке.

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

Подписчик

Устройство отказывается от обновления тремя способами, и каждому нужен свой исход. Busy значит, что устройство выполняет задание. Обновление, предложенное прямо сейчас, снова получит отказ, поэтому retry_after предлагает его через 30 секунд. Interrupted значит, что передача оборвалась: retry пробует снова сразу. Incompatible значит, что образ не подходит устройству. Новая попытка получит тот же ответ, поэтому обновление отбрасывается.

Макросы

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

Вручную

src/main.rs

struct Install;

impl Handle<Update> for Install {
    async fn handle(
        &self,
        update: &Update,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        match flash(update).await {
            Ok(()) => Ok(()),
            Err(Refusal::Busy) => {
                tracing::info!(update.device, "device busy, offering the update later");
                Err(HandlerOutcome::retry_after(BACKOFF))
            }
            Err(Refusal::Interrupted) => {
                tracing::warn!(update.device, "transfer interrupted, retrying");
                Err(HandlerOutcome::retry())
            }
            Err(Refusal::Incompatible) => {
                tracing::warn!(update.device, update.version, "incompatible image dropped");
                Err(HandlerOutcome::drop())
            }
        }
    }
}

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

retry_after подтверждает доставку, ждёт и затем публикует копию обратно в updates/install. В копии есть заголовок с числом попыток (RETRY_COUNT_HEADER), увеличенным на единицу. Это работает при любом качестве обслуживания: у доставки с QoS 0 подтверждать нечего, а копия всё равно публикуется. Копия ждёт в памяти сервиса: процесс, который остановился до конца задержки, её теряет.

retry тоже становится копией, которая публикуется сразу, потому что точка монтирования ниже объявляет предел попыток. MQTT не считает повторные доставки, поэтому предел считает копии по их заголовку. Без предела retry оставляет доставку неподтверждённой до следующей сессии, как показывает руководство Гарантии доставки.

Предел попыток и тема dead-letter

Макросы

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);
        },
    )
}

Вручную

src/main.rs

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(
                subscriber("updates/install", Install)
                    .map_source(|_| updates())
                    .build(),
            )
            .max_attempts(nonzero!(4u32))
            .dead_letter("updates/dead");
            b.include(
                subscriber("updates/dead", Review)
                    .map_source(|_| dead_updates())
                    .build(),
            );
            b.include(
                subscriber("devices/+/reports", Record)
                    .map_source(|_| reports())
                    .build(),
            )
            .max_attempts(nonzero!(4u32))
            .dead_letter("reports/dead")
            .out_retry(Publish::default())
            .transform(ToDeliveryTopic);
        },
    )
}

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

max_attempts считает все доставки одного обновления, включая первую. Четвёртая доставка не обрабатывается повторно: dead_letter публикует её в updates/dead с полезной нагрузкой и заголовками в том виде, в каком они пришли. Отброшенное обновление завершается и никуда не уходит. В теме dead-letter лежат только обновления, у которых кончились попытки.

Копия - обычная публикация, поэтому её получает каждая подписка, фильтр которой подходит к её теме, в этом сервисе и в любом другом. Поэтому ни тема переобработки, ни тема dead-letter не подходит ни к одному другому фильтру, который читает сервис. Подписка на updates/# рядом с install читала бы каждую копию каждого обновления и заодно каждое сообщение dead-letter.

Чтение темы dead-letter

Макросы

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

Вручную

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

impl Handle<Update> for Review {
    fn handle(
        &self,
        update: &Update,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
        tracing::error!(
            update.device,
            update.version,
            deliveries,
            "update needs a person"
        );
        ready(Ok(()))
    }
}

fn dead_updates() -> MqttTopic {
    MqttTopic::new("updates/dead").qos(Qos::AtLeastOnce)
}

Тема dead-letter - обычная тема, и review читает её, как любую другую. Заголовок с числом попыток у обновления из темы dead-letter хранит число доставок, через которые оно прошло, здесь 4.

Переобработка на фильтре

Макросы

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

Вручную

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

/// A body typed over `MqttContext` makes the crate's context the registration's, which is what
/// the transform reads.
struct Record;

impl Handle<Report, (), (), MqttContext> for Record {
    async fn handle(
        &self,
        report: &Report,
        _outs: &(),
        ctx: &mut Context<'_, MqttContext>,
    ) -> Result<(), HandlerOutcome> {
        if let Err(err) = store(report).await {
            let topic = ctx.context(DeliveryTopic);
            tracing::warn!(%topic, %err, "report not stored, retrying");
            return Err(HandlerOutcome::retry_after(BACKOFF));
        }
        Ok(())
    }
}

fn reports() -> MqttFilter {
    MqttFilter::new("devices/+/reports").qos(Qos::AtLeastOnce)
}

Копия доставки на MqttTopic возвращается в эту тему. MqttFilter подходит ко многим темам и ни одной не называет, поэтому его регистрация называет, куда уходит копия, а без этого сервис не запускается. out_retry(Publish::default()).to(..) называет одну тему для всех копий. ToDeliveryTopic вместо этого называет тему для каждой доставки: он берёт тему, на которую пришёл отчёт, из контекста MQTT и отправляет копию обратно туда. Поэтому отчёт из devices/dev7/reports возвращается в devices/dev7/reports, и только к record. Обработчик принимает Ctx<DeliveryTopic>, и через это регистрация получает контекст MQTT. В ручной форме тип его контекста - MqttContext.

Проверка задержек

Тесты в файле примера запускают приложение на внутрипроцессном сервере с остановленными часами, поэтому tb.advance(BACKOFF) проматывает 30 секунд мгновенно. Они считают доставки каждого обновления, читают заголовок с числом попыток у каждой копии и проверяют, что дошло до updates/dead. Работу с обвязкой описывает руководство Тестирование сервиса на MQTT.