Skip to main content

Acknowledging messages

This guide settles Pulsar messages by the handler's outcome, bounds an unsettled delivery with an acknowledgement timeout, and settles a batch at once.

The example is an inventory service. It reserves stock for orders and records the stock counts of warehouse scans. Stock stands in for the client of the stock service, and the app opens it at startup.

The outcome is the settlement​

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()
        }
    }
}

The value reserve returns decides what happens to the message on Pulsar:

OutcomeOn Pulsar
ackthe message is acknowledged and never comes back
retrythe message is negatively acknowledged and delivered again at once
retry_after(delay)the message is held for delay, then negatively acknowledged
dropthe message is acknowledged

A stock service that does not answer may answer the next call, so reserve retries. The message goes back to the subscription, and on a Shared subscription another consumer may take it. A SKU that no item carries stays unknown, so reserve drops the reservation. Pulsar has no verb to reject a message for good, so drop acknowledges it, and the message is gone. A message that keeps failing leaves the subscription through a dead-letter topic instead.

The client sends acknowledgements in the background. A settlement that returns is queued on the consumer, and the broker records it a moment later. A service that stops in between gets the message again on its next start, so Pulsar delivers each message at least once. Handle a message so that a second delivery does no harm, for example by keying the work on the order's id.

The acknowledgement timeout​

A delivery that nobody settles stays with its consumer. ack_timeout on the descriptor gives it a limit: the client asks the broker to deliver the message again once the limit has passed. Without it, such a message comes back only when its consumer disconnects.

Set the timeout above the longest time the handler takes. A handler still at work when the timeout passes gets a second delivery of the same message. Each redelivery counts toward the retry cap, like a retry.

Batches​

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()
}

A handler over a slice is a batch handler. The Pulsar client hands over one message at a time, so the service assembles the batch. A batch closes when it is full or when batch_wait has passed since its first message, here 250 milliseconds. One outcome settles every message of the batch. In the manual form, Ok(()) acknowledges the batch, and Err(outcomes) settles each message with its own outcome.

The app​

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)));
        })
}

batch(n) at the mount site is the largest batch, 64 counts here. on_startup opens the clients once, and the handlers read them as the app's state.

Testing the outcomes​

The tests in the example file run the app in process, with TestApp::with_state in place of on_startup. They make the stock service fail on purpose and check the outcome of each delivery. The in-process Pulsar hands a negatively acknowledged message back to the subscription, as a server does. The redelivery after the acknowledgement timeout is the server's, and the crate's live tests check it against a running Pulsar.