ack_timeout

ack_timeout redelivers a message left unacknowledged longer than the timeout; it also bounds the delay a retry_after may ask for.

It is a setting of the PulsarSubscription descriptor and has no default: without it, an unsettled message comes back only when its consumer disconnects. With it, the client asks the broker for every message its handler has held longer than the timeout, whatever the reason: a handler that hangs, or a service that stopped mid-delivery. Each such redelivery counts toward the dead-letter cap, as a retry does. A retry_after delay that is not shorter than the timeout is refused when the handler returns it, with an error naming both values, because the timeout would bring the message back first. A handler keeps its backoff below the timeout for that reason. The in-process transport of a TestApp test applies the bound on retry_after; the redelivery itself happens on a broker.

Acknowledgement bounds an unsettled delivery of an inventory service with an acknowledgement timeout.

Example

An export the archive refused waits thirty seconds, inside the one-minute ack timeout of its subscription.

Macros
Manual

examples/pulsar_settlement_ack_timeout.rs
/// How long the archive gets before the next upload: shorter than the ack timeout, which would
/// bring the export back first.
const BACKOFF: Duration = Duration::from_secs(30);

/// Uploads each export. An export nobody settles within a minute, because the handler hangs or
/// the service stops, comes back from the broker.
#[subscriber(
    PulsarSubscription::new("exports", "archiver").ack_timeout(Duration::from_secs(60))
)]
async fn upload(export: &Export, State(archive): State<Archive>) -> HandlerOutcome {
    match archive.upload(export).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(export.id, %err, "upload failed, trying again later");
            HandlerOutcome::retry_after(BACKOFF)
        }
    }
}

examples/manual/pulsar_settlement_ack_timeout.rs
/// How long the archive gets before the next upload: shorter than the ack timeout, which would
/// bring the export back first.
const BACKOFF: Duration = Duration::from_secs(30);

/// Uploads each export.
struct Upload;

impl Handle<Export, (), (), (), AppState> for Upload {
    async fn handle(
        &self,
        export: &Export,
        _outs: &(),
        ctx: &mut Context<'_, (), AppState>,
    ) -> Result<(), HandlerOutcome> {
        ctx.state().archive.upload(export).await.map_err(|err| {
            tracing::warn!(export.id, %err, "upload failed, trying again later");
            HandlerOutcome::retry_after(BACKOFF)
        })
    }
}

fn app() -> impl App<State = AppState> {
    RustStream::new(AppInfo::new("archiver", "0.1.0"))
        .on_startup(async move |()| {
            Ok::<_, Infallible>(AppState {
                archive: Archive::default(),
            })
        })
        .with_broker(PulsarBroker::new("pulsar://localhost:6650"), |b| {
            // An export nobody settles within a minute, because the handler hangs or the service
            // stops, comes back from the broker.
            b.include(
                subscriber(
                    PulsarSubscription::new("exports", "archiver")
                        .ack_timeout(Duration::from_secs(60)),
                    Upload,
                )
                .build(),
            );
        })
}

Example​

Example