Повторы и ключ dead-letter

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

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

Пример - сервис отгрузки, который отгружает заказ, когда его оплата прошла.

Подписчик

Макросы

src/main.rs

/// How long an order waits before the service asks the payment service again.
const RECHECK: Duration = Duration::from_secs(2);

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
}

#[subscriber(FileStream::new("orders"))]
async fn ship(order: &Order, State(payments): State<Arc<Payments>>) -> HandlerOutcome {
    if !payments.has_cleared(order.id) {
        // The payment may clear later: ask again after a pause.
        return HandlerOutcome::retry_after(RECHECK);
    }
    tracing::info!(order.id, "order shipped");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// How long an order waits before the service asks the payment service again.
const RECHECK: Duration = Duration::from_secs(2);

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Order {
    id: u64,
}

struct Ship;

impl Handle<Order, (), (), (), Clients> for Ship {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, (), Clients>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if !ctx.state().payments.has_cleared(order.id) {
            // The payment may clear later: ask again after a pause.
            return ready(Err(HandlerOutcome::retry_after(RECHECK)));
        }
        tracing::info!(order.id, "order shipped");
        ready(Ok(()))
    }
}

Payments - клиент сервиса платежей, который хранится в состоянии приложения. Пока оплата заказа не прошла, ship отвечает retry_after, и сервис спрашивает снова через две секунды.

retry_after завершает доставку, выжидает задержку и дописывает копию заказа под ключом, который читает подписка. Подписка читает копию как новое сообщение, после всего, что уже было в файле. Поэтому заказы после ожидающего не задерживаются. У копии есть заголовок с числом попыток (RETRY_COUNT_HEADER), увеличенным на единицу. До конца задержки копия ждёт в памяти сервиса, поэтому процесс, который остановился раньше, её теряет. Воспроизведение читает копию, только если само ещё не дошло до конца файла.

Предел и ключ dead-letter

Макросы

src/main.rs

fn shipping(file: FileBroker) -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("shipping", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(file, |b| {
            // Three deliveries of one order, the first included; then the dead-letter key.
            b.include(ship)
                .max_attempts(nonzero!(3u32))
                .dead_letter("orders.dead");
            b.include(review);
        })
}

fn app() -> impl App<State = Clients> {
    shipping(FileBroker::new("/var/lib/shop/orders.ss"))
}

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

Вручную

src/main.rs

fn shipping(file: FileBroker) -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("shipping", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(file, |b| {
            // Three deliveries of one order, the first included; then the dead-letter key.
            b.include(subscriber(FileStream::new("orders"), Ship).build())
                .max_attempts(nonzero!(3u32))
                .dead_letter("orders.dead");
            b.include(subscriber(FileStream::new("orders.dead"), Review).build());
        })
}

fn app() -> impl App<State = Clients> {
    shipping(FileBroker::new("/var/lib/shop/orders.ss"))
}

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

max_attempts считает все доставки одного заказа, включая первую. Третья доставка не повторяется: dead_letter дописывает её под ключом orders.dead с полезной нагрузкой и заголовками в том виде, в каком они пришли. На файле потока ключ dead-letter - ещё один ключ потока того же файла. Предел без ключа dead-letter рядом с ним никуда не записывает заказ, у которого кончились попытки.

Чтение ключа dead-letter

Макросы

src/main.rs

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

Вручную

src/main.rs

struct Review;

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

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

Повторы в канале

Стандартный вывод ступени доходит до следующего процесса и никогда до её собственного ввода, поэтому копия не может вернуться на ступень, которая её отложила. На StdioBroker точка монтирования называет ключ, под который уходят копии, b.include(ship).out_retry(Publish).to("orders.retry"), и их читает одна из следующих ступеней (out_retry). Так смонтирована ступень конвейера из руководства Начало работы.

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

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

Тесты в файле примера запускают приложение на внутрипроцессном транспорте с остановленными часами, поэтому tb.advance(RECHECK) проматывает две секунды мгновенно. Один тест проводит оплату между двумя доставками, и заказ отгружается на второй. Тест dead-letter выполняется и внутри процесса, и на настоящем файле потока через TestApp::start_live, где задержки проходят по настоящим часам. Работу с обвязкой разбирает руководство Тестирование сервиса на файле или в конвейере.