max_ack_pending

max_ack_pending caps how many unacknowledged deliveries the server lets a consumer hold at once; the default is 1024.

A delivery is pending from the moment the server hands it out until it is settled. Once the consumer holds that many, the server stops delivering until a settlement frees a place. The cap belongs to the consumer, so every subscription that shares a durable counts against one cap. It bounds the work in flight from the server's side, whatever the service does. workers(n) runs up to n handlers at once in one subscription; with a cap below n, the cap is the limit. A delivery the consumer holds counts against its ack_wait from the moment it is handed out, so a large cap over slow handlers lets deliveries wait out their ack_wait and come back. The cap is enforced by the server, so a test of it runs against a live server.

Example

Six workers render reports, and the server hands out two at a time.

Macros
Manual

examples/nats_subscriptions_pending.rs
/// Six workers, but the server hands out at most two reports before they are acknowledged, so
/// two render at a time.
#[subscriber(
    JetStreamSubject::new("reports.requested", "REPORTS")
        .durable("renderers")
        .max_ack_pending(2),
    workers(6)
)]
async fn produce(report: &Report) -> HandlerOutcome {
    render(report).await;
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(produce);
        },
    )
}

examples/manual/nats_subscriptions_pending.rs
/// Six workers, but the server hands out at most two reports before they are acknowledged, so
/// two render at a time.
struct Produce;

impl Handle<Report> for Produce {
    async fn handle(
        &self,
        report: &Report,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        render(report).await;
        Ok(())
    }
}

fn reports() -> JetStreamSubject {
    JetStreamSubject::new("reports.requested", "REPORTS")
        .durable("renderers")
        .max_ack_pending(2)
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(subscriber(reports(), Produce).workers(nonzero!(6)).build());
        },
    )
}

Example​

Example