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

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

В AMQP 1.0 нет отложенной повторной доставки, а связь не может объявить предел доставок или адрес недоставленных сообщений. И то и другое делает фреймворк: точка монтирования объявляет предел и адрес недоставленных сообщений, а отложенный повтор публикуется как копия сообщения с номером повтора в заголовке.

Пример - сервис возвратов, который проводит возвраты через процессинг карт.

Подписчик

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

Макросы

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

#[subscriber(AmqpAddress::queue("refunds"))]
async fn process_refund(refund: &Refund) -> HandlerOutcome {
    match send_refund(refund) {
        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, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Refund {
    id: u64,
    card: String,
    amount_cents: u64,
}

impl OutgoingDestination for Refund {
    type Form = CallerName;
}

impl MessageHeaders for Refund {
    type Contract = NoHeaders;
}

impl MessageInfo for Refund {
    const NAME: &'static str = "Refund";
}

struct ProcessRefund;

impl Handle<Refund> for ProcessRefund {
    fn handle(
        &self,
        refund: &Refund,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        ready(match send_refund(refund) {
            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())
            }
        })
    }
}

retry_after урегулирует доставку и по истечении задержки публикует копию в refunds. В копии заголовок с номером повтора (RETRY_COUNT_HEADER) увеличен на единицу. Отложенная копия доставляется at-most-once: процесс, который остановился до конца задержки, её теряет. В очереди копия достаётся одному потребителю, как любое другое сообщение.

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

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("refunds", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            // Five deliveries of one refund, the first included; then the dead-letter address.
            b.include(process_refund)
                .max_attempts(nonzero!(5u32))
                .dead_letter("refunds.dead")
                .out_retry(Publish)
                .transform(Retried);
            b.include(review);
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("refunds", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            // Five deliveries of one refund, the first included; then the dead-letter address.
            b.include(subscriber(AmqpAddress::queue("refunds"), ProcessRefund).build())
                .max_attempts(nonzero!(5u32))
                .dead_letter("refunds.dead")
                .out_retry(Publish)
                .transform(Retried);
            b.include(subscriber(AmqpAddress::queue("refunds.dead"), Review).build());
        },
    )
}

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

max_attempts считает все доставки одного возврата, включая первую. Пятая доставка не повторяется: dead_letter публикует её в refunds.dead с полезной нагрузкой и заголовками в том виде, в каком они пришли. Отброшенный возврат отклоняется и попадает под собственную политику недоставленных сообщений брокера, а не в refunds.dead.

При объявленном пределе фреймворк читает число попыток из доставки. Это delivery-count брокера, если у сообщения есть секция AMQP header. У сообщения, которое опубликовал этот крейт, её нет, поэтому retry в этом случае сразу публикует копию, и попытки считает заголовок с номером повтора.

Предел без dead_letter вместо этого отклоняет последнюю доставку, и дальше решает политика недоставленных сообщений брокера. У брокера есть и собственный предел: Artemis останавливается после max-delivery-attempts, по умолчанию 10. Выберите одного владельца предела: при пределе ниже брокерского действует счёт фреймворка.

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

Макросы

src/main.rs

#[subscriber(AmqpAddress::queue("refunds.dead"))]
async fn review(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()
}

Вручную

src/main.rs

struct Review;

impl Handle<Refund> for Review {
    fn handle(
        &self,
        refund: &Refund,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> 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(()))
    }
}

refunds.dead - обычная очередь, и review читает её, как любую другую. Заголовок с номером повтора у недоставленного сообщения хранит число доставок, через которые оно прошло, здесь 5. review читает его через ctx.headers() и пишет возврат в журнал, чтобы его посмотрел человек.

Пометка копий

Макросы

src/main.rs

/// Marks every copy with the address its delivery came from.
struct Retried;

impl<C, Options> PublishTransform<ForReply<C>, Options> for Retried {
    type Destination = Reads;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, C>,
    ) {
        out.headers_mut()
            .insert("x-retried-from", cx.name().to_owned());
    }
}

Вручную

src/main.rs

/// Marks every copy with the address its delivery came from.
struct Retried;

impl<C, Options> PublishTransform<ForReply<C>, Options> for Retried {
    type Destination = Reads;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, C>,
    ) {
        out.headers_mut()
            .insert("x-retried-from", cx.name().to_owned());
    }
}

Копию в обработчике не публикует ни один вызов, поэтому добраться до неё может только преобразование публикации. Приложение монтирует process_refund через .out_retry(Publish).transform(Retried). .out_retry(Publish) называет издателя копий, и шаги после него применяются к ним. Retried записывает в x-retried-from адрес, из которого пришла доставка. Каждая копия возвращается в адрес своей подписки, поэтому преобразование в этой позиции не может менять адрес назначения.

Повторы в теме

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

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

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