Подтверждение и переобработка

Это руководство завершает каждую доставку Pub/Sub по исходу обработчика, запрашивает переобработку сразу или после задержки и ограничивает число доставок темой dead-letter.

Pub/Sub завершает каждое сообщение отдельно. Подтверждённое сообщение обработано. Возвращённое сообщение доставляется снова, возможно другому потребителю. Подписка считает доставки и переносит сообщение, у которого они кончились, в свою тему dead-letter. Крейт переводит исход обработчика в эти действия, а предел из точки монтирования - в политику dead-letter подписки.

Пример - платёжный сервис, который списывает деньги с карт через банк.

Обработчик

Банк отвечает на списание четырьмя способами, и каждому нужен свой исход. Charged значит, что платёж проведён: ack. Held значит, что банк проверяет платёж, и это занимает время: retry_after спрашивает снова через 30 секунд. Unreachable значит, что банк не ответил: retry спрашивает снова сразу. Declined значит, что карта закрыта. Новая попытка получит тот же ответ, поэтому платёж отбрасывается.

Макросы

src/main.rs

/// How long the bank's review of a held payment takes.
const HOLD: Duration = Duration::from_secs(30);

#[subscriber(GooglePubSub::new("payments-workers").create_with_topic("payments"))]
async fn charge(payment: &Payment, ctx: &mut Context<'_>) -> HandlerOutcome {
    let attempt = ctx
        .headers()
        .get_str(DELIVERY_ATTEMPT_HEADER)
        .unwrap_or("1");
    match ask_the_bank(payment) {
        Answer::Charged => HandlerOutcome::ack(),
        Answer::Held => {
            tracing::info!(
                payment.id,
                attempt,
                "payment held by the bank, asked again later"
            );
            HandlerOutcome::retry_after(HOLD)
        }
        Answer::Unreachable => {
            tracing::warn!(payment.id, attempt, "bank unreachable, payment retried");
            HandlerOutcome::retry()
        }
        Answer::Declined => {
            tracing::warn!(payment.id, "card declined, payment dropped");
            HandlerOutcome::drop()
        }
    }
}

Вручную

src/main.rs

/// How long the bank's review of a held payment takes.
const HOLD: Duration = Duration::from_secs(30);

struct Charge;

impl Handle<Payment> for Charge {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let attempt = ctx
            .headers()
            .get_str(DELIVERY_ATTEMPT_HEADER)
            .unwrap_or("1");
        let settled = match ask_the_bank(payment) {
            Answer::Charged => Ok(()),
            Answer::Held => {
                tracing::info!(
                    payment.id,
                    attempt,
                    "payment held by the bank, asked again later"
                );
                Err(HandlerOutcome::retry_after(HOLD))
            }
            Answer::Unreachable => {
                tracing::warn!(payment.id, attempt, "bank unreachable, payment retried");
                Err(HandlerOutcome::retry())
            }
            Answer::Declined => {
                tracing::warn!(payment.id, "card declined, payment dropped");
                Err(HandlerOutcome::drop())
            }
        };
        ready(settled)
    }
}

В Pub/Sub исходы становятся такими вызовами:

Исход	Что делает Pub/Sub
ack	подтверждает, и сообщение больше не доставляется
retry	отправляет nack, и подписка доставляет сообщение снова
retry_after(delay)	сервис держит сообщение в аренде на время задержки, затем отправляет nack
drop	подтверждает. На последней доставке, которую допускает предел, отправляет nack, и сообщение уходит в тему dead-letter

Подписка без политики повторных попыток доставляет сообщение после nack снова сразу. Подписка с политикой экспоненциальной задержки, которую задала эксплуатация, сначала ждёт, и тогда retry тоже ждёт.

У Pub/Sub нет отложенного nack, поэтому retry_after держит сообщение в сервисе. Клиент продолжает продлевать его аренду на время задержки и отправляет nack, когда задержка истекла. Задержка должна укладываться в max_lease подписки, по умолчанию час. Сообщение с более долгой задержкой возвращается сразу. Сервис, который останавливается во время задержки, перестаёт продлевать аренду, и сообщение возвращается, когда аренда истекает. Аренду описывает руководство Управление потоком и аренда.

DELIVERY_ATTEMPT_HEADER - номер попытки доставки, у первой доставки он равен 1. Pub/Sub считает попытки только на подписке с политикой dead-letter, и его счётчик приблизителен: он может пропустить доставку.

Предел и тема dead-letter

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            // Five deliveries of one payment, the first included; then `payments-dead`.
            b.include(charge)
                .max_attempts(nonzero!(5u32))
                .dead_letter("payments-dead");
            b.include(review);
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            // Five deliveries of one payment, the first included; then `payments-dead`.
            b.include(
                subscriber(
                    GooglePubSub::new("payments-workers").create_with_topic("payments"),
                    Charge,
                )
                .build(),
            )
            .max_attempts(nonzero!(5u32))
            .dead_letter("payments-dead");
            b.include(
                subscriber(
                    GooglePubSub::new("payments-dead-review").create_with_topic("payments-dead"),
                    Review,
                )
                .build(),
            );
        },
    )
}

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

max_attempts и dead_letter становятся собственной политикой dead-letter подписки, которая записывается при запуске. После пятой доставки Pub/Sub перестаёт доставлять платёж и публикует его в payments-dead с полезной нагрузкой и атрибутами в том виде, в каком они пришли. Сам сервис ничего не публикует, поэтому сбой не может потерять перенос. Pub/Sub принимает предел от 5 до 100. Любое другое значение или только одно из двух объявлений прерывает запуск. Отброшенный платёж подтверждается и никуда не уходит. В теме dead-letter лежат только платежи, у которых кончились доставки.

Подписка payments-workers создаётся с этой политикой, а вместе с ней создаётся тема payments-dead. В подписку, которую создала эксплуатация, политика вместо этого записывается при запуске. В промышленной среде служебному аккаунту Pub/Sub проекта нужны права на публикацию в тему dead-letter и на подписку на исходную подписку. Роли перечисляет руководство Работа в промышленной среде.

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

Макросы

src/main.rs

/// Reads the dead-letter topic through a subscription of its own.
#[subscriber(GooglePubSub::new("payments-dead-review").create_with_topic("payments-dead"))]
async fn review(payment: &Payment) -> HandlerOutcome {
    tracing::error!(
        payment.id,
        "payment ran out of attempts, a person looks at it"
    );
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Reads the dead-letter topic through a subscription of its own.
struct Review;

impl Handle<Payment> for Review {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::error!(
            payment.id,
            "payment ran out of attempts, a person looks at it"
        );
        ready(Ok(()))
    }
}

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

Подписки exactly-once

Подписка с доставкой exactly-once гарантирует, что подтверждённое сообщение не доставляется снова. Это настройка подписки, которую задаёт эксплуатация при её создании. Сервису своя настройка не нужна. Крейт читает её из каждой доставки, и на такой подписке ack и retry ждут подтверждения от Pub/Sub. Подтверждение, которое Pub/Sub отклонил, например после истечения аренды, завершает доставку ошибкой с указанием причины, и сообщение может вернуться. Гарантия действует для подписчиков в регионе подписки, поэтому сервис подключается через конечную точку этого региона.

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

Тесты в файле примера публикуют по одному платежу на каждый ответ банка. Задержанный платёж возвращается, только когда на остановленных часах прошло 30 секунд. Недоступный банк стоит пяти доставок, после чего платёж получает review. Отклонённый платёж обрабатывается один раз и не доходит до review. Работу с обвязкой описывает руководство Тестирование сервиса на Pub/Sub.