Skip to main content

Retries and dead-letter streams

This guide retries a record now or later, caps the attempts and moves what keeps failing to a dead-letter stream.

Kinesis does not hold a record back, does not count deliveries and has no dead-letter queue. The framework does all three: the mount site declares the cap and the dead-letter stream, and a retry travels as a copy of the record with a retry count in a header.

The example is a refunds service that sends refunds through a card processor.

The subscriber​

The processor refuses a refund in three ways, and each needs a different outcome. Busy means its rate limit is used up: a retry right now would be refused again, so retry_after waits 30 seconds. Unavailable means it did not answer: retry tries again at once. Declined means the card account is closed, and another try gets the same answer, so the refund is dropped.

src/main.rs
/// How long a busy processor asks its callers to wait.
const BACKOFF: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "refunds")]
struct Refund {
    id: u64,
    card: String,
    amount_cents: u64,
}

#[subscriber(KinesisStream::new("refunds").create_if_missing(1))]
async fn process_refund(
    refund: &Refund,
    State(processor): State<Arc<Processor>>,
) -> HandlerOutcome {
    match processor.refund(refund).await {
        Ok(()) => HandlerOutcome::ack(),
        // Past the rate limit: a delivery right now would be refused again.
        Err(Refusal::Busy) => HandlerOutcome::retry_after(BACKOFF),
        // No answer: try again at once, up to the attempt cap.
        Err(Refusal::Unavailable) => HandlerOutcome::retry(),
        Err(Refusal::Declined) => {
            tracing::warn!(refund.id, "card declined, refund dropped");
            HandlerOutcome::drop()
        }
    }
}

retry_after settles the record and publishes a copy back to refunds once the delay has passed. The copy carries the retry-count header (RETRY_COUNT_HEADER), raised by one. The copy waits in the service's memory, so it is at-most-once: a process that stops before the delay ends loses it, and the original is already behind the checkpoint.

retry under a declared cap also becomes a copy, published at once. Without a cap it holds the shard's checkpoint in front of the record, as Checkpoints and delivery guarantees shows.

The cap and the dead-letter stream​

src/main.rs
fn service(broker: KinesisBroker) -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("refunds", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(broker, |b| {
            // Five deliveries of one refund, the first included; then the dead-letter stream.
            b.include(process_refund)
                .max_attempts(nonzero!(5u32))
                .dead_letter("refunds.review");
            b.include(review);
        })
}

#[ruststream::app]
fn app() -> impl App<State = Clients> {
    service(
        KinesisBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
    )
}

max_attempts counts every delivery of one refund, the first included. The fifth delivery is not retried: dead_letter publishes it to refunds.review, payload and headers as they arrived. A cap without a dead-letter stream checkpoints past the spent record instead, as drop does. A dropped refund settles and goes nowhere; the dead-letter stream holds only the refunds that ran out of attempts. The streams belong to the service's infrastructure; the example creates them with create_if_missing for a local stack.

Reading the dead-letter stream​

The dead-letter stream is an ordinary stream, and review reads it like any other.

src/main.rs
#[subscriber(KinesisStream::new("refunds.review").create_if_missing(1))]
async fn review(refund: &Refund, ctx: &mut Context<'_>) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
    tracing::error!(refund.id, deliveries, "refund needs a person to look at it");
    HandlerOutcome::ack()
}

The retry-count header of a dead-lettered record holds the number of deliveries it went through, 5 here. review reads it with ctx.headers() and logs the refund for a person to look at.

Copies and order​

A copy goes back to the stream the subscription reads, so a registration on Kinesis needs no retry destination of its own. It keeps the record's partition key, so it lands on the shard of that key. It lands at the tip of that shard, though, after the records published while it waited. A refund that waits 30 seconds is therefore handled after the refunds of the same card that came later.

Where the order of a key matters more than moving on, declare no cap and let retry hold the shard: the checkpoint stops at the record, and the shard is read again from it when its lease is next taken. out_retry replaces the publisher the copies leave through, so a transform can stamp each copy on its way out.

Testing the delay​

The tests in the example file run the app in process with a paused clock, so tb.advance(BACKOFF) lets the 30 seconds pass at once. TestApp::with_state starts the app with a stand-in processor that is busy or silent. Testing a Kinesis service covers the harness.