Skip to main content

Retries and dead letters

This guide retries a job now or later, caps its attempts and parks what keeps failing in a dead-letter group of the same table.

A retry here is a statement on the job's row. The row keeps its id and its place in the table, counts one more attempt, and a delay is a time in one of its columns. Nothing lives only in the process: a job waiting for its retry survives a restart.

The example is a mailer that sends emails through a provider.

The table​

src/main.rs
const SCHEMA: &str = "CREATE TABLE IF NOT EXISTS mail_jobs (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    attempt INTEGER NOT NULL DEFAULT 1,
    retry_after TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%f+00:00', 'now')),
    locked_until TEXT,
    payload BLOB NOT NULL
)";

name is the group a row waits in: the subscription to emails reads the rows whose name is emails. attempt counts the deliveries of a row, the first one included. retry_after is the time a row becomes due; its default makes a new row due at once.

The subscriber​

The provider refuses an email in three ways, and each needs a different outcome. Busy means its rate limit is spent: a retry right now would be refused again, so the email waits. Unavailable means it did not answer: retry tries again at once. Rejected means the mailbox does not exist, and another try gets the same answer, so the email is dropped.

src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `mail_jobs`; `name` is the queue the row waits in.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "mail_jobs")]
struct SendMail {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    name: String,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(retry_after)]
    retry_after: DateTime<Utc>,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// How much longer each attempt waits after the provider said it was busy.
const BACKOFF: Duration = Duration::from_secs(1);

#[subscriber(InboxQueue::<SendMail>::new("emails"))]
async fn send(mail: &Mail, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    let attempt = attempt.unwrap_or(1);
    match deliver(mail, attempt).await {
        Ok(()) => HandlerOutcome::ack(),
        // The provider's rate limit is spent: a retry right now would be refused again.
        Err(Refusal::Busy) => {
            HandlerOutcome::retry_after(BACKOFF * u32::try_from(attempt).unwrap_or(u32::MAX))
        }
        // The provider did not answer: the next attempt may reach it.
        Err(Refusal::Unavailable) => HandlerOutcome::retry(),
        // The mailbox does not exist: every attempt gets the same answer.
        Err(Refusal::Rejected) => {
            tracing::warn!(to = %mail.to, "address rejected, dropped");
            HandlerOutcome::drop()
        }
    }
}

The handler owns the backoff. It reads the delivery's attempt and asks for a second more with each one: retry_after(d) writes "now" plus d into the row's retry_after, and no claim takes the row before then. A table without that column returns the row at once, and the runtime logs a warning. retry() returns the row to the queue at once; its next delivery reads one more attempt. drop() finishes the row as an acknowledgement does, so a rejected email leaves the table. A row whose payload does not decode spends its attempts like any other failure (Decoding).

The cap and the dead-letter group​

src/main.rs
/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("mailer", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<SendMail>("emails"), |b| {
            // Five deliveries, the first included; the fifth failure moves the row into the
            // `emails.dead` group.
            b.include(send)
                .max_attempts(nonzero!(5u32))
                .dead_letter("emails.dead");
            b.include(review);
        })
}

max_attempts reads the row's attempt. The fifth delivery that fails is not retried: dead_letter moves the row into the emails.dead group, payload and attempt as they were. A delayed retry counts against the cap as an immediate one does. The two steps come together: a mount that declares one without the other stops the service at startup. A table without a group column takes a table name in dead_letter instead, and the row moves into that table, which has the same columns.

Reading the dead letters​

The dead-letter group is an ordinary queue of the table, and review reads it like any other.

src/main.rs
#[subscriber(InboxQueue::<SendMail>::new("emails.dead"))]
async fn review(mail: &Mail, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    tracing::error!(to = %mail.to, ?attempt, "email not sent, needs a person");
    HandlerOutcome::ack()
}

review logs the email for a person to look at and acknowledges it. A service that keeps its dead letters for an operator mounts no subscription on the group; the rows wait in the table. Sending them again is one statement of the service's own, which moves them back and resets their count:

UPDATE mail_jobs SET name = 'emails', attempt = 1, locked_until = NULL WHERE name = 'emails.dead';

In a table with FIFO groups, retry() keeps a row at the head of its group, and retry_after(d) lets the rows due earlier pass it (Ordering).

Testing​

The tests in the example file publish an email through the harness, as the service's own Publish writes it. A rejected address is dropped at once, which TestApp::start shows when the publish returns. An unanswered email and a busy provider need time to pass: the delayed row and the dead letter come back on the subscriptions' next poll. Those tests run on the real clock with TestApp::start_live, with a backoff of one second, and tb.advance waits it out. Testing a service on SQL covers the harness.