visibility

visibility(duration) sets how long a received message stays hidden; the crate re-arms it halfway through each timeout while the handler runs.

The visibility timeout is what each receive asks for: how long a received message stays hidden from every other receive. SQS counts the timeout in whole seconds, from one second to twelve hours. A value outside that range stops the service at startup with SqsError::InvalidQueue. A fraction above one second rounds up to the next whole second, so a message never comes back sooner than the timeout asked for. A subscription that names none takes the queue's own timeout, which it reads once with GetQueueAttributes when it opens, so it needs the sqs:GetQueueAttributes permission. A handler may take longer than the timeout. While a handler holds a delivery, the crate re-arms the timeout halfway through each period with ChangeMessageVisibility, so no other consumer receives the message in the meantime. The other half of the period is left for the call to reach the queue, so a one-second timeout is re-armed every half second. SQS keeps a message hidden for at most twelve hours from its receive, and an extension does not restart them. A handler that holds a delivery that long gets one last extension to the end of the twelve hours and a warning in the log; after that the queue hands the message to another consumer. The timeout also decides how soon an abandoned delivery comes back: a message dropped unsettled, or one a stopped process held, is received again once the timeout lapses. In process, a subscription that names no timeout gets the 30 seconds of a new queue; the timeout an operator set on a queue shows only in a live test. The step is on SqsQueue, and on the mount chain through SqsSubscription::visibility, as in the example.

Example

A report takes five seconds to render, and a receive hides it for two; the report is still handled once.

Macros

examples/sqs_subscriptions_visibility.rs

/// Rendering a report takes longer than the two seconds a receive hides it for. The crate
/// re-arms the visibility every second while the handler runs, so no other consumer gets the
/// report in the meantime.
#[subscriber(SqsQueue)]
async fn publish_report(report: &Report) -> HandlerOutcome {
    render(report).await;
    tracing::info!(report.id, "report published");
    HandlerOutcome::ack()
}

fn service(broker: SqsBroker, queue: String) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(broker, |b| {
        b.include(
            publish_report
                .name(queue)
                .visibility(Duration::from_secs(2)),
        );
    })
}

Manual

examples/manual/sqs_subscriptions_visibility.rs

/// Rendering a report takes longer than the two seconds a receive hides it for. The crate
/// re-arms the visibility every second while the handler runs, so no other consumer gets the
/// report in the meantime.
struct PublishReport;

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

fn service(broker: SqsBroker, queue: String) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(Unnamed::<SqsQueue>::new(), PublishReport)
                .build()
                .name(queue)
                .visibility(Duration::from_secs(2)),
        );
    })
}