Dead letters

max_attempts and dead_letter together become the queue's redrive policy, and SQS moves a spent message to the dead-letter queue itself.

The cap is the policy's maxReceiveCount, at most 1000, and the destination is the queue it points at. The subscription writes the policy onto its queue with SetQueueAttributes when it opens. SQS then counts the receives and carries the message away on the receive past the cap, so nothing leaves the service. A bare queue name takes the declaration the same way as an SqsQueue. Both halves are needed: a cap without a destination, or a destination without a cap, stops the service at startup with SqsError::IncompleteRedrive, which names the missing half. The dead-letter queue has to exist at startup, unless the subscription creates its queues. A FIFO queue takes a FIFO dead-letter queue, and a standard queue a standard one. A drop normally deletes the message, but on the delivery that has spent the policy it returns the message instead, because the next receive is what moves it to the dead-letter queue. The dead-letter queue is an ordinary queue, read by an ordinary subscription. The policy needs the permissions sqs:SetQueueAttributes on the queue and sqs:GetQueueAttributes on the dead-letter queue.

Example

An invoice that is never ready is received three times and then moves to invoices-dead, where a person takes it over.

Macros

examples/sqs_dead_letter.rs

/// A handler that is never ready. Every delivery asks to come back later, so what ends the
/// message is the cap: the queue hands it over three times and then moves it.
#[subscriber(SqsQueue::new("invoices").create_if_missing())]
async fn settle(invoice: &Invoice) -> HandlerOutcome {
    tracing::info!(invoice.id, "invoice is not ready yet");
    HandlerOutcome::retry_after(RECHECK)
}

/// The dead-letter queue is an ordinary queue, read by an ordinary subscription.
#[subscriber("invoices-dead")]
async fn escalate(invoice: &Invoice) -> HandlerOutcome {
    tracing::error!(invoice.id, "invoice escalated to a person");
    HandlerOutcome::ack()
}

Manual

examples/manual/sqs_dead_letter.rs

/// A handler that is never ready. Every delivery asks to come back later, so what ends the
/// message is the cap: the queue hands it over three times and then moves it.
struct Settle;

impl Handle<Invoice> for Settle {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(invoice.id, "invoice is not ready yet");
        ready(Err(HandlerOutcome::retry_after(RECHECK)))
    }
}

/// The dead-letter queue is an ordinary queue, read by an ordinary subscription.
struct Escalate;

impl Handle<Invoice> for Escalate {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::error!(invoice.id, "invoice escalated to a person");
        ready(Ok(()))
    }
}

The declaration

The cap and the dead-letter queue, written at the mount site.

Macros

examples/sqs_dead_letter.rs

fn service(broker: SqsBroker) -> impl App {
    RustStream::new(AppInfo::new("invoices", "0.1.0")).with_broker(broker, |b| {
        // The two steps are the queue's redrive policy: `maxReceiveCount` and the
        // dead-letter queue it points at. The subscription writes them when it opens.
        b.include(settle)
            .max_attempts(nonzero!(3u32))
            .dead_letter("invoices-dead");
        b.include(escalate);
    })
}

Manual

examples/manual/sqs_dead_letter.rs

fn service(broker: SqsBroker) -> impl App {
    RustStream::new(AppInfo::new("invoices", "0.1.0")).with_broker(broker, |b| {
        // The two steps are the queue's redrive policy: `maxReceiveCount` and the
        // dead-letter queue it points at. The subscription writes them when it opens.
        b.include(
            subscriber("invoices", Settle)
                .build()
                .map_source(|_| SqsQueue::new("invoices").create_if_missing()),
        )
        .max_attempts(nonzero!(3u32))
        .dead_letter("invoices-dead");
        b.include(subscriber("invoices-dead", Escalate).build());
    })
}