RECEIVE_COUNT_HEADER

RECEIVE_COUNT_HEADER carries the queue's ApproximateReceiveCount, one on the first delivery, so a handler can tell the last attempt.

The header's name is sqs-receive-count, and its value is the count as decimal text. SQS counts every receive of a message, the one in hand included. A retry or a deferred retry returns the message to the same queue, so the count goes on rising.

A redrive policy caps the same count. After max_attempts receives, SQS moves the message to the dead-letter queue on the receive that follows. A handler that compares the header with the cap knows when it holds the last attempt. SQS calls the count approximate: on a standard queue a message is delivered at least once, and a duplicate receive counts too.

Example

A declined payment is tried three times. On the third receive the handler sends it to review and acknowledges it, so it never reaches the dead-letter queue.

Macros
Manual

examples/sqs_delivery_receive_count.rs
/// How many times the queue hands a payment over before it moves it to the dead-letter queue.
const ATTEMPTS: u32 = 3;

#[subscriber(SqsQueue::new("payments"))]
async fn charge(
    payment: &Payment,
    cx: &mut Context<'_>,
    Out(review): Out<impl Publisher, Review>,
) -> HandlerOutcome {
    if payment.card_accepted {
        return HandlerOutcome::ack();
    }
    let receives: u32 = cx
        .headers()
        .get(RECEIVE_COUNT_HEADER)
        .and_then(|count| str::from_utf8(count).ok()?.parse().ok())
        .unwrap_or(1);
    if receives < ATTEMPTS {
        return HandlerOutcome::retry();
    }
    // The last attempt: a person looks at the payment before the queue gives up on it.
    tracing::warn!(payment.id, receives, "payment declined, sent to review");
    match review
        .message(payment)
        .to("payments-review")
        .publish()
        .await
    {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

#[app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(SqsBroker::new(), |b| {
        b.include(charge)
            .max_attempts(nonzero!(ATTEMPTS))
            .dead_letter("payments-dead")
            .out(Review, Publish::default())
            .build();
    })
}

examples/manual/sqs_delivery_receive_count.rs
/// How many times the queue hands a payment over before it moves it to the dead-letter queue.
const ATTEMPTS: u32 = 3;

struct Charge;

impl<Entry> Handle<Payment, (), Outs<(Entry,)>> for Charge
where
    Entry: OutEntry<Review, Wire: Publisher>,
{
    async fn handle(
        &self,
        payment: &Payment,
        outs: &Outs<(Entry,)>,
        ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        if payment.card_accepted {
            return Ok(());
        }
        let receives: u32 = ctx
            .headers()
            .get(RECEIVE_COUNT_HEADER)
            .and_then(|count| str::from_utf8(count).ok()?.parse().ok())
            .unwrap_or(1);
        if receives < ATTEMPTS {
            return Err(HandlerOutcome::retry());
        }
        // The last attempt: a person looks at the payment before the queue gives up on it.
        tracing::warn!(payment.id, receives, "payment declined, sent to review");
        let sent = outs
            .get(Review)
            .message(payment)
            .to("payments-review")
            .publish()
            .await;
        sent.map_err(|_| HandlerOutcome::retry())
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(SqsBroker::new(), |b| {
        b.include(
            subscriber(Unnamed::<SqsQueue>::new(), Charge)
                .build()
                .name("payments"),
        )
        .max_attempts(nonzero!(ATTEMPTS))
        .dead_letter("payments-dead")
        .out(Review, Publish::default())
        .build();
    })
}

Example​

Example