Подтверждения, переобработка и недоставленные сообщения

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

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

Что отправляет исход

Исход обработчика становится кадром AMQP в канале доставки:

Исход	Кадр	Что делает RabbitMQ
HandlerOutcome::ack()	basic.ack	удаляет сообщение из очереди
HandlerOutcome::retry()	basic.reject с возвратом в очередь	возвращает сообщение в очередь
HandlerOutcome::drop()	basic.reject без возврата в очередь	удаляет сообщение или, если у очереди есть обменник недоставленных сообщений, переносит его туда
HandlerOutcome::retry_after(delay)	подтверждённая доставка и отложенная копия	доставляет копию снова, когда задержка истекла

Каждое отклонение RabbitMQ засчитывает сообщению как израсходованную доставку. Поэтому обработчик, который раз за разом просит вернуть сообщение, исчерпывает собственный предел кворумной очереди. Доставка, которую сервис взял и так и не подтвердил, возвращается в очередь, когда закрывается канал или соединение.

Подписчик

Макросы

src/main.rs

/// How long a busy processor asks its callers to wait.
const BACKOFF: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Refund {
    id: u64,
    card: String,
    amount_cents: u64,
}

/// Refunds wait out a delay in the broker's waiting queue, `refunds.retry`.
#[subscriber(RabbitQueue::new("refunds").delay(Delay::dlx_ttl()))]
async fn process_refund(
    refund: &Refund,
    State(processor): State<Arc<Processor>>,
) -> HandlerOutcome {
    match processor.refund(&refund.card).await {
        Ok(()) => HandlerOutcome::ack(),
        // Past the rate limit: a delivery right now would be refused again.
        Err(Refusal::Busy) => HandlerOutcome::retry_after(BACKOFF),
        // No answer: try again at once, up to the attempt cap.
        Err(Refusal::Unavailable) => HandlerOutcome::retry(),
        Err(Refusal::Declined) => {
            tracing::warn!(refund.id, "card declined, refund dropped");
            HandlerOutcome::drop()
        }
    }
}

Вручную

src/main.rs

/// How long a busy processor asks its callers to wait.
const BACKOFF: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Refund {
    id: u64,
    card: String,
    amount_cents: u64,
}

struct ProcessRefund;

impl Handle<Refund, (), (), (), Clients> for ProcessRefund {
    async fn handle(
        &self,
        refund: &Refund,
        _outs: &(),
        ctx: &mut Context<'_, (), Clients>,
    ) -> Result<(), HandlerOutcome> {
        match ctx.state().processor.refund(&refund.card).await {
            Ok(()) => Ok(()),
            // Past the rate limit: a delivery right now would be refused again.
            Err(Refusal::Busy) => Err(HandlerOutcome::retry_after(BACKOFF)),
            // No answer: try again at once, up to the attempt cap.
            Err(Refusal::Unavailable) => Err(HandlerOutcome::retry()),
            Err(Refusal::Declined) => {
                tracing::warn!(refund.id, "card declined, refund dropped");
                Err(HandlerOutcome::drop())
            }
        }
    }
}

/// Refunds wait out a delay in the broker's waiting queue, `refunds.retry`.
fn refunds_queue() -> RabbitQueue {
    RabbitQueue::new("refunds").delay(Delay::dlx_ttl())
}

Процессинговый центр отказывает в возврате тремя способами, и каждому нужен свой исход. Busy означает, что лимит запросов исчерпан: немедленная переобработка снова получит отказ, поэтому retry_after ждёт 30 секунд. Unavailable означает, что процессинговый центр не ответил: retry пробует снова сразу. Declined означает, что карточный счёт закрыт и следующая попытка получит тот же ответ, поэтому возврат отбрасывается.

Предел попыток на классической очереди

Макросы

src/main.rs

fn service(broker: LapinBroker) -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("refunds", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(broker, |b| {
            // The runtime counts these five deliveries, on the copies it publishes back.
            b.include(process_refund)
                .max_attempts(nonzero!(5u32))
                .dead_letter("refunds.dead");
            b.include(review_refund);
            // The server counts these: they become the queue's delivery limit and route.
            b.include(send_payout)
                .max_attempts(nonzero!(5u32))
                .dead_letter("payouts.dead");
            b.include(review_payout);
        })
}

#[ruststream::app]
fn app() -> impl App<State = Clients> {
    service(LapinBroker::new("amqp://localhost:5672").declare_topology(true))
}

Вручную

src/main.rs

fn service(broker: LapinBroker) -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("refunds", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(broker, |b| {
            // The runtime counts these five deliveries, on the copies it publishes back.
            b.include(subscriber(refunds_queue(), ProcessRefund).build())
                .max_attempts(nonzero!(5u32))
                .dead_letter("refunds.dead");
            b.include(subscriber("refunds.dead", ReviewRefund).build());
            // The server counts these: they become the queue's delivery limit and route.
            b.include(subscriber(payouts_queue(), SendPayout).build())
                .max_attempts(nonzero!(5u32))
                .dead_letter("payouts.dead");
            b.include(subscriber("payouts.dead", ReviewPayout).build());
        })
}

fn app() -> impl App<State = Clients> {
    service(LapinBroker::new("amqp://localhost:5672").declare_topology(true))
}

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

max_attempts считает каждую доставку одного возврата, включая первую. Пятая доставка не переобрабатывается: dead_letter публикует её в refunds.dead. Адрес назначения - ключ маршрутизации в обменнике по умолчанию, поэтому возврат попадает в очередь с этим именем.

Классическая очередь ничего не считает, поэтому предел попыток ведёт сервис. Пока предел не достигнут, переобработка подтверждает доставку и публикует копию обратно в очередь под её собственным именем. В заголовке копии x-ruststream-retry-count хранится число попыток, увеличенное на единицу (RETRY_COUNT_HEADER). Копии уходят через политику публикации брокера по умолчанию. .out_retry(policy) задаёт другую, например TransactionalPublish, чтобы ждать подтверждения RabbitMQ для каждой копии.

Отброшенный возврат не попадает в refunds.dead: очередь недоставленных сообщений хранит только возвраты, исчерпавшие попытки. Чтобы сохранять и отброшенные сообщения, дайте очереди собственный обменник недоставленных сообщений (Топология).

Предел попыток на кворумной очереди

Макросы

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Payout {
    id: u64,
    card: String,
    amount_cents: u64,
}

/// Payouts live in a quorum queue, which counts their deliveries itself.
#[subscriber(RabbitQuorumQueue::new("payouts"))]
async fn send_payout(payout: &Payout, State(processor): State<Arc<Processor>>) -> HandlerOutcome {
    match processor.refund(&payout.card).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(Refusal::Busy | Refusal::Unavailable) => HandlerOutcome::retry(),
        Err(Refusal::Declined) => HandlerOutcome::drop(),
    }
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Payout {
    id: u64,
    card: String,
    amount_cents: u64,
}

struct SendPayout;

impl Handle<Payout, (), (), (), Clients> for SendPayout {
    async fn handle(
        &self,
        payout: &Payout,
        _outs: &(),
        ctx: &mut Context<'_, (), Clients>,
    ) -> Result<(), HandlerOutcome> {
        match ctx.state().processor.refund(&payout.card).await {
            Ok(()) => Ok(()),
            Err(Refusal::Busy | Refusal::Unavailable) => Err(HandlerOutcome::retry()),
            Err(Refusal::Declined) => Err(HandlerOutcome::drop()),
        }
    }
}

/// Payouts live in a quorum queue, which counts their deliveries itself.
fn payouts_queue() -> RabbitQuorumQueue {
    RabbitQuorumQueue::new("payouts")
}

Кворумная очередь сама считает доставки каждого сообщения. На очереди payouts те же два шага в точке монтирования становятся аргументом очереди x-delivery-limit и её маршрутом для недоставленных сообщений. Тогда сервер сам переносит выплату в payouts.dead после пятой доставки, работает сервис или нет, а сервис ничего не публикует. Переобработка остаётся обычным basic.reject с возвратом в очередь.

Аргументы попадают в очередь, только если её объявляет сервис, поэтому здесь нужен declare_topology(true). На брокере, который ничего не объявляет, сервис не запускается: у очереди, объявленной в другом месте, аргументы её владельца, и предел не действовал бы нигде. В этом случае задайте предел и маршрут на самой очереди, в её объявлении или политикой сервера, и монтируйте обработчик без этих двух шагов. Половину объявления сервис тоже не принимает, потому что предел без маршрута для израсходованного сообщения отбрасывает его. .out_retry(..) на кворумной очереди не компилируется: публиковать копию не нужно.

У кворумной очереди, объявленной без предела, действует значение RabbitMQ 4 по умолчанию - 20 доставок. Другое значение задаёт delivery_limit у дескриптора. Доставка из кворумной очереди сообщает, сколько доставок она израсходовала, через redelivery_count(). Доставка из классической очереди сообщает только Redelivered: выдавал ли её RabbitMQ раньше.

Отложенная переобработка

В AMQP нет задержки для отдельного сообщения. Если у очереди не задан delay, retry_after держит копию в сервисе и публикует её, когда задержка истекла. Такая копия доставляется по схеме at-most-once: если сервис остановится во время задержки, копия потеряется.

delay(Delay::dlx_ttl()) переносит ожидание в RabbitMQ. Копия ждёт в очереди refunds.retry с TTL на уровне сообщения, и когда TTL истекает, очередь ожидания переносит её обратно в refunds как недоставленное сообщение. Оригинал подтверждается после публикации копии в том же канале, поэтому сбой между двумя шагами дублирует возврат, а не теряет его. Копия - новое сообщение, поэтому число попыток в её заголовке увеличено на единицу, и предел останавливает отложенный цикл так же, как немедленный.

Очередь ожидания выпускает сообщения только из своей головы. Сообщение с долгой задержкой задерживает все сообщения за ним, поэтому для разных задержек нужна своя очередь ожидания на каждую. Её имя задаёт Delay::dlx_ttl_named. Фича plugin-dme добавляет Delay::plugin_dme: ожидание в плагине RabbitMQ delayed-message-exchange, где короткая задержка никогда не ждёт за долгой.

Очередь ожидания и отложенный обменник - это инфраструктура. При declare_topology(true) их объявляет сервис. Если объявление выключено, подписка с delay при запуске проверяет, что её очередь ожидания или отложенный обменник существует, и без них сервис не запускается.

Чтение очереди недоставленных сообщений

Макросы

src/main.rs

#[subscriber("refunds.dead")]
async fn review_refund(refund: &Refund, ctx: &mut Context<'_>) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
    tracing::error!(refund.id, deliveries, "refund needs a person to look at it");
    HandlerOutcome::ack()
}

#[subscriber("payouts.dead")]
async fn review_payout(payout: &Payout) -> HandlerOutcome {
    tracing::error!(payout.id, "payout needs a person to look at it");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

struct ReviewRefund;

impl Handle<Refund, (), (), (), Clients> for ReviewRefund {
    fn handle(
        &self,
        refund: &Refund,
        _outs: &(),
        ctx: &mut Context<'_, (), Clients>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
        tracing::error!(refund.id, deliveries, "refund needs a person to look at it");
        ready(Ok(()))
    }
}

struct ReviewPayout;

impl Handle<Payout, (), (), (), Clients> for ReviewPayout {
    fn handle(
        &self,
        payout: &Payout,
        _outs: &(),
        _ctx: &mut Context<'_, (), Clients>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::error!(payout.id, "payout needs a person to look at it");
        ready(Ok(()))
    }
}

Очередь недоставленных сообщений - обычная очередь, и обработчик читает её, как любую другую. У возврата, который сервис перенёс в очередь недоставленных сообщений, в заголовке с числом попыток записано, сколько доставок он прошёл, здесь 5. У выплаты, которую перенесла кворумная очередь, вместо него есть собственный заголовок RabbitMQ x-death. Оба обработчика пишут сообщение в журнал, чтобы его посмотрел человек.

Тестирование

Тесты в файле примера запускают приложение внутри процесса с остановленными часами, поэтому tb.advance(BACKOFF) проматывает 30 секунд сразу. RabbitMQ внутри процесса выпускает копию из очереди ожидания, считает доставки кворумной очереди и переносит исчерпавшее попытки сообщение по его маршруту, как это делает сервер. Плагин delayed-message-exchange работает только на сервере, поэтому его тесты выполняются на настоящем RabbitMQ. Обвязку разбирает руководство Тестирование сервиса на RabbitMQ.