Подтверждение сообщений

Это руководство завершает сообщения Pulsar по исходу обработчика, ограничивает незавершённую доставку тайм-аутом подтверждения и завершает пакет целиком.

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

Исход и есть завершение

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize)]
struct Reservation {
    order: u64,
    sku: String,
    quantity: u32,
}

/// Reserves stock for an order. A reservation left unsettled for a minute comes back.
#[subscriber(
    PulsarSubscription::new("reservations", "inventory").ack_timeout(Duration::from_secs(60))
)]
async fn reserve(reservation: &Reservation, State(stock): State<Arc<Stock>>) -> HandlerOutcome {
    match stock.reserve(reservation).await {
        Ok(()) => {
            tracing::info!(reservation.order, reservation.quantity, "stock reserved");
            HandlerOutcome::ack()
        }
        // No answer: the subscription hands the reservation out again at once.
        Err(Refusal::Unavailable) => HandlerOutcome::retry(),
        Err(Refusal::UnknownSku) => {
            tracing::warn!(reservation.order, "unknown SKU, reservation dropped");
            HandlerOutcome::drop()
        }
    }
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
struct Reservation {
    order: u64,
    sku: String,
    quantity: u32,
}

/// A reservation left unsettled for a minute comes back.
fn reservations() -> PulsarSubscription {
    PulsarSubscription::new("reservations", "inventory").ack_timeout(Duration::from_secs(60))
}

/// Reserves stock for an order.
struct Reserve;

impl Handle<Reservation, (), (), (), Clients> for Reserve {
    async fn handle(
        &self,
        reservation: &Reservation,
        _outs: &(),
        ctx: &mut Context<'_, (), Clients>,
    ) -> Result<(), HandlerOutcome> {
        match ctx.state().stock.reserve(reservation).await {
            Ok(()) => {
                tracing::info!(reservation.order, reservation.quantity, "stock reserved");
                Ok(())
            }
            // No answer: the subscription hands the reservation out again at once.
            Err(Refusal::Unavailable) => Err(HandlerOutcome::retry()),
            Err(Refusal::UnknownSku) => {
                tracing::warn!(reservation.order, "unknown SKU, reservation dropped");
                Err(HandlerOutcome::drop())
            }
        }
    }
}

Значение, которое возвращает reserve, решает, что происходит с сообщением в Pulsar:

Исход	В Pulsar
ack	сообщение подтверждается и больше не возвращается
retry	сообщение подтверждается отрицательно и сразу доставляется снова
retry_after(delay)	сообщение удерживается на время delay, затем подтверждается отрицательно
drop	сообщение подтверждается

Сервис остатков, который не ответил, может ответить на следующий вызов, поэтому reserve повторяет. Сообщение возвращается в подписку, и на подписке Shared его может взять другой консьюмер. Артикул, которого нет ни у одного товара, так и останется неизвестным, поэтому reserve отбрасывает резервирование. В Pulsar нет команды, которая отвергает сообщение окончательно, поэтому drop подтверждает его, и сообщение исчезает. Сообщение, которое раз за разом не обрабатывается, покидает подписку через топик dead-letter.

Клиент отправляет подтверждения в фоне. Завершение возвращает управление, когда подтверждение встало в очередь консьюмера, а брокер записывает его мгновением позже. Сервис, остановленный в этом промежутке, получит сообщение снова при следующем запуске, поэтому Pulsar доставляет каждое сообщение at-least-once. Обрабатывайте сообщение так, чтобы вторая доставка не навредила, например привязывайте работу к номеру заказа.

Тайм-аут подтверждения

Доставка, которую никто не завершил, остаётся у своего консьюмера. ack_timeout в дескрипторе ограничивает её срок: когда срок истёк, клиент просит брокер доставить сообщение снова. Без тайм-аута такое сообщение возвращается, только когда его консьюмер отключается.

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

Пакеты

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize)]
struct Count {
    sku: String,
    on_hand: u32,
}

/// Records the stock counts of a warehouse scan, many counts per call.
#[subscriber(
    PulsarSubscription::new("stock-counts", "inventory").batch_wait(Duration::from_millis(250))
)]
async fn record_counts(counts: &[Count], State(stock): State<Arc<Stock>>) -> HandlerOutcome {
    stock.record(counts).await;
    tracing::info!(counts = counts.len(), "stock counts recorded");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
struct Count {
    sku: String,
    on_hand: u32,
}

/// A batch that is not full closes 250 ms after its first count.
fn stock_counts() -> PulsarSubscription {
    PulsarSubscription::new("stock-counts", "inventory").batch_wait(Duration::from_millis(250))
}

/// Records the stock counts of a warehouse scan; `Ok` settles the whole batch.
struct RecordCounts;

impl Handle<[Count], (), (), (), Clients> for RecordCounts {
    async fn handle(
        &self,
        counts: &[Count],
        _outs: &(),
        ctx: &mut Context<'_, (), Clients>,
    ) -> Result<(), Vec<HandlerOutcome>> {
        ctx.state().stock.record(counts).await;
        tracing::info!(counts = counts.len(), "stock counts recorded");
        Ok(())
    }
}

Обработчик, который принимает срез, - это пакетный обработчик. Клиент Pulsar выдаёт сообщения по одному, поэтому пакет собирает сервис. Пакет закрывается, когда он заполнен или когда с его первого сообщения прошло batch_wait, здесь 250 миллисекунд. Один исход завершает все сообщения пакета. В ручной форме Ok(()) подтверждает пакет, а Err(outcomes) завершает каждое сообщение его собственным исходом.

Приложение

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("inventory", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(PulsarBroker::new("pulsar://localhost:6650"), |b| {
            b.include(reserve);
            b.include(record_counts.batch(nonzero!(64)));
        })
}

Вручную

src/main.rs

fn app() -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("inventory", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(PulsarBroker::new("pulsar://localhost:6650"), |b| {
            b.include(subscriber(reservations(), Reserve).build());
            b.include(
                subscriber(stock_counts(), RecordCounts)
                    .batch(nonzero!(64))
                    .build(),
            );
        })
}

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

batch(n) в точке монтирования задаёт наибольший размер пакета, здесь 64 записи остатков. on_startup открывает клиенты один раз, а обработчики читают их как состояние приложения.

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

Тесты в файле примера запускают приложение внутри процесса, с TestApp::with_state вместо on_startup. Они намеренно заставляют сервис остатков отказывать и проверяют исход каждой доставки. Pulsar внутри процесса возвращает отрицательно подтверждённое сообщение в подписку, как это делает сервер. Повторная доставка по тайм-ауту подтверждения принадлежит серверу, и её проверяют живые тесты крейта на работающем Pulsar.