Expectations

The expect steps make the stream refuse a publish unless it, or the message's subject, is still at the position the writer last saw.

They are publish steps of JetStreamPublishSteps, written into the call's JetStreamOptions:

Step	Header	The stream stores the message only if
`expect_last_sequence(n)`	`Nats-Expected-Last-Sequence`	its last sequence is `n`
`expect_last_subject_sequence(n)`	`Nats-Expected-Last-Subject-Sequence`	the last sequence on the message's subject is `n`
`expect_last_message_id(id)`	`Nats-Expected-Last-Msg-Id`	the `Nats-Msg-Id` of its last message is `id`

A refused publish fails with NatsError::JetStream, and nothing is stored. This is optimistic concurrency: a writer that read the stream at some position writes only if nobody wrote after it. A handler reading a stream knows where its delivery sits through the StreamSequence key, so a write to the subject it read can expect that sequence. The publish error wraps the client's error, so a handler that treats a refusal apart from a failed send finds NatsError::JetStream in its source chain. As with message_id, the body imports this crate's prelude and bounds its slot on JetStreamOptions.

Example

A reservation is confirmed once its cancellation window has passed, and only if nothing was written to the item's subject since; a cancellation in the window makes the stream refuse the confirmation.

Macros
Manual

examples/nats_publishing_expectations.rs
/// Confirms a reservation once its window has passed. The confirmation expects the item's
/// subject to be still at the reservation's sequence, so a cancellation written in the window
/// makes the stream refuse it.
#[subscriber(JetStreamSubject::new("stock.*", "STOCK").durable("confirmer"))]
async fn confirm(
    event: &StockEvent,
    Ctx(sequence): Ctx<StreamSequence>,
    Ctx(delivered): Ctx<Delivered>,
    Out(confirmations): Out<impl Publisher<Options = JetStreamOptions>, Confirmations>,
) -> HandlerOutcome {
    let (Change::Reserved, Some(sequence)) = (event.change, sequence) else {
        return HandlerOutcome::ack();
    };
    if delivered == Some(1) {
        return HandlerOutcome::retry_after(CANCEL_WINDOW);
    }
    let confirmed = StockEvent {
        item: event.item.clone(),
        order_id: event.order_id,
        change: Change::Confirmed,
    };
    let sent = confirmations
        .message(&confirmed)
        .to(format!("stock.{}", event.item))
        .expect_last_subject_sequence(sequence)
        .publish()
        .await;
    match sent {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) if refused_by_stream(&err) => {
            tracing::info!(event.order_id, %err, "the item changed during the window");
            HandlerOutcome::drop()
        }
        Err(err) => {
            tracing::warn!(event.order_id, %err, "confirmation not sent, redelivering");
            HandlerOutcome::retry()
        }
    }
}

examples/manual/nats_publishing_expectations.rs
/// Confirms a reservation once its window has passed. The confirmation expects the item's
/// subject to be still at the reservation's sequence, so a cancellation written in the window
/// makes the stream refuse it.
struct Confirm;

impl<Entry> Handle<StockEvent, (), Outs<(Entry,)>, JetStreamContext> for Confirm
where
    Entry: OutEntry<Confirmations, Wire: Publisher<Options = JetStreamOptions>>,
{
    async fn handle(
        &self,
        event: &StockEvent,
        outs: &Outs<(Entry,)>,
        ctx: &mut Context<'_, JetStreamContext>,
    ) -> Result<(), HandlerOutcome> {
        let (Change::Reserved, Some(sequence)) = (event.change, ctx.context(keys::STREAM_SEQUENCE))
        else {
            return Ok(());
        };
        if ctx.context(keys::DELIVERED) == Some(1) {
            return Err(HandlerOutcome::retry_after(CANCEL_WINDOW));
        }
        let confirmed = StockEvent {
            item: event.item.clone(),
            order_id: event.order_id,
            change: Change::Confirmed,
        };
        let sent = outs
            .get(Confirmations)
            .message(&confirmed)
            .to(format!("stock.{}", event.item))
            .expect_last_subject_sequence(sequence)
            .publish()
            .await;
        match sent {
            Ok(()) => Ok(()),
            Err(err) if refused_by_stream(&err) => {
                tracing::info!(event.order_id, %err, "the item changed during the window");
                Err(HandlerOutcome::drop())
            }
            Err(err) => {
                tracing::warn!(event.order_id, %err, "confirmation not sent, redelivering");
                Err(HandlerOutcome::retry())
            }
        }
    }
}

Example​

Example