Skip to main content

Acknowledgement and delivery guarantees

This guide shows what each handler outcome does to an SQS message, how a handler stays correct when a message arrives twice, and how a batch settles.

SQS delivers at least once. A received message stays on the queue, invisible, until the consumer deletes it. A standard queue may also deliver a message twice, and in another order than it was sent (AWS: at-least-once delivery).

The example is a payments service that books payments into a ledger and pays refunds.

What each outcome does​

Every outcome is one SQS call on the message the handler received:

OutcomeSQS callWhat happens to the message
ackDeleteMessageit leaves the queue
dropDeleteMessageit leaves the queue: SQS has no call that rejects a message and keeps it
retryChangeMessageVisibility to zeroit is handed out again at once
retry_afterChangeMessageVisibility to the delayit is handed out again when the delay ends

A message the handler never settles, because the process stopped in the middle, gets no call. It comes back when its visibility timeout ends. So the work of a handler happens before its outcome, and a message is deleted only after the work is done.

A handler that books once​

src/main.rs
/// How long a payment on an account under review waits before the next try.
const REVIEW: Duration = Duration::from_secs(60);

/// The receive after which a payment under review is given up.
const LAST_RECEIVE: u32 = 3;

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "payments")]
struct Payment {
    id: u64,
    account: String,
    amount_cents: u64,
    under_review: bool,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "ledger-entries")]
struct Entry {
    payment: u64,
    account: String,
    amount_cents: u64,
}

#[derive(OutSlot)]
#[publishes(Entry)]
struct Entries;

/// The ids of the payments already booked.
#[derive(Debug, Default, Clone)]
struct Ledger {
    booked: Arc<Mutex<HashSet<u64>>>,
}

#[subscriber(SqsQueue::new("payments"))]
async fn book(
    payment: &Payment,
    ctx: &mut Context<'_, (), Ledger>,
    Out(entries): Out<impl Publisher, Entries>,
) -> HandlerOutcome {
    // SQS counts the receives of a message, this one included.
    let receives: u32 = ctx
        .headers()
        .get_str(RECEIVE_COUNT_HEADER)
        .and_then(|count| count.parse().ok())
        .unwrap_or(1);
    if payment.under_review {
        if receives >= LAST_RECEIVE {
            tracing::error!(payment.id, receives, "payment dropped, still under review");
            return HandlerOutcome::drop();
        }
        return HandlerOutcome::retry_after(REVIEW);
    }
    let booked = Arc::clone(&ctx.state().booked);
    if booked
        .lock()
        .unwrap_or_else(PoisonError::into_inner)
        .contains(&payment.id)
    {
        tracing::info!(payment.id, receives, "payment already booked");
        return HandlerOutcome::ack();
    }
    let entry = Entry {
        payment: payment.id,
        account: payment.account.clone(),
        amount_cents: payment.amount_cents,
    };
    if entries.message(&entry).publish().await.is_err() {
        // Nothing is booked yet: the payment comes back and the work runs again.
        return HandlerOutcome::retry();
    }
    booked
        .lock()
        .unwrap_or_else(PoisonError::into_inner)
        .insert(payment.id);
    tracing::info!(payment.id, "payment booked");
    HandlerOutcome::ack()
}

book writes a ledger entry for a payment and then acknowledges it. A payment SQS delivers a second time is found among the booked ids and acknowledged without a second entry. The set in the app state stands in for the store a service keeps its booked ids in; the id is the payment's own, not the SQS message id, because a producer that sends twice sends two messages. When the entry cannot be published, nothing is booked, and retry hands the payment back for another try.

The handler reads RECEIVE_COUNT_HEADER, the queue's own ApproximateReceiveCount. It counts this receive too, so the first delivery reads 1. A payment on an account under review waits a minute with retry_after, and on its third receive the handler gives up and drops it. Retries and dead-letter queues shows how the queue itself caps the receives and keeps what failed.

A batch settles message by message​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "refunds")]
struct Refund {
    payment: u64,
    amount_cents: u64,
}

/// The largest refund paid without a person's approval.
const APPROVAL_LIMIT_CENTS: u64 = 50_000;

/// How long a refund waits for approval before it is looked at again.
const APPROVAL: Duration = Duration::from_secs(300);

/// Settles each refund of a batch on its own: a small refund is paid and deleted, a large one
/// waits on the queue for its approval.
#[subscriber(SqsQueue::new("refunds"))]
async fn settle_refunds(refunds: &[Refund]) -> Vec<HandlerOutcome> {
    refunds
        .iter()
        .map(|refund| {
            if refund.amount_cents > APPROVAL_LIMIT_CENTS {
                tracing::info!(refund.payment, "refund waits for approval");
                HandlerOutcome::retry_after(APPROVAL)
            } else {
                tracing::info!(refund.payment, refund.amount_cents, "refund paid");
                HandlerOutcome::ack()
            }
        })
        .collect()
}

A batch handler returns one outcome per message, in the order of the slice. Each outcome is a call of its own on its message. Here the small refunds are deleted, and a large one stays on the queue and comes back alone after five minutes. A batch handler that returns a single outcome settles every message of the batch with it.

The app​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = Ledger> {
    RustStream::new(AppInfo::new("payments", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Ledger::default()))
        .with_broker(
            SqsBroker::new()
                .endpoint("http://localhost:4566")
                .region("us-east-1")
                .test_credentials(),
            |b| {
                b.include(book).out(Entries, Publish::default()).build();
                b.include(settle_refunds.batch(nonzero!(10)));
            },
        )
}

book publishes its entries through the Entries slot, bound to the queue publisher at the mount site. A FIFO queue drops a second send with the same deduplication id for five minutes, which keeps a producer from sending one message twice; FIFO queues shows it. A handler on a standard queue still keeps its own record of what it has done.

The tests in the example file deliver a payment twice, let a payment under review run out its receives on a paused clock, and send three refunds in one batch. Testing an SQS service covers the harness.