重试与死信

本指南立即或稍后重试一个任务，限定它的尝试次数，并把一直失败的任务存放到同一张表的死信组中。

在这里，一次重试就是对任务所在行执行的一条语句。该行保留它的 id 和它在表中的位置，尝试次数加一，延迟则是它某一列中的一个时间。没有任何东西只存在于进程中：等待重试的任务在重启后依然存在。

示例是一个通过邮件服务商发送邮件的发信服务。

表

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 是一行所在的组：对 emails 的订阅读取 name 为 emails 的行。attempt 统计一行的投递次数，包括第一次。retry_after 是一行到期的时间；它的默认值让新行立即到期。

订阅者

服务商会以三种方式拒绝一封邮件，每种方式需要不同的结果。Busy 表示它的速率限额已经用完：现在重试还会被拒绝，所以这封邮件要等待。Unavailable 表示它没有应答：retry 立即再试一次。Rejected 表示邮箱不存在，再试一次也会得到同样的答复，所以这封邮件被丢弃。

宏

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()
        }
    }
}

手写

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(sqlx::FromRow)]
struct SendMail {
    id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

impl InboxTable for SendMail {
    type Id = i64;
    type Table = InboxSpec<(
        Attempt,
        RetryAfter<DateTime<Utc>>,
        Lease<DateTime<Utc>>,
        Payload,
    )>;
    const TABLE: Self::Table = InboxSpec::new("mail_jobs", Column::new("id").generated())
        .group(Column::new("name"))
        .attempt(Column::new("attempt").generated())
        .retry_after(Column::new("retry_after"))
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

    fn id(&self) -> &i64 {
        &self.id
    }
}

impl AttemptRow for SendMail {
    type Attempt = i16;

    fn attempt(&self) -> &i16 {
        &self.attempt
    }
}

impl PayloadRow for SendMail {
    type Column = Vec<u8>;

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

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

struct Sender;

impl Handle<Mail, (), (), InboxContext> for Sender {
    async fn handle(
        &self,
        mail: &Mail,
        _outs: &(),
        ctx: &mut Context<'_, InboxContext>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(attempt) = Ctx::<keys::Attempt>::from_context(ctx).await?;
        let attempt = attempt.unwrap_or(1);
        match deliver(mail, attempt).await {
            Ok(()) => Ok(()),
            // The provider's rate limit is spent: a retry right now would be refused again.
            Err(Refusal::Busy) => Err(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) => Err(HandlerOutcome::retry()),
            // The mailbox does not exist: every attempt gets the same answer.
            Err(Refusal::Rejected) => {
                tracing::warn!(to = %mail.to, "address rejected, dropped");
                Err(HandlerOutcome::drop())
            }
        }
    }
}

退避策略由处理器掌握。它读取投递的尝试次数，每多一次就多等一秒：retry_after(d) 把“现在”加上 d 写入该行的 retry_after，在此之前没有领取会取走这一行。没有这一列的表会立即放回该行，运行时记录一条警告。retry() 立即把该行放回队列；它的下一次投递读到的尝试次数多一。drop() 像确认一样结束该行，因此被拒绝的邮件离开这张表。载荷无法解码的行，与其他失败一样消耗尝试次数（解码）。

上限与死信组

宏

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);
        })
}

手写

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.
            let send =
                subscriber("emails", Sender).map_source(|_| InboxQueue::<SendMail>::new("emails"));
            b.include(send.build())
                .max_attempts(nonzero!(5u32))
                .dead_letter("emails.dead");
            let review = subscriber("emails.dead", Review)
                .map_source(|_| InboxQueue::<SendMail>::new("emails.dead"));
            b.include(review.build());
        })
}

max_attempts 读取该行的 attempt。第五次失败的投递不再重试：dead_letter 把该行转入 emails.dead 组，载荷和尝试次数保持原样。延迟重试与立即重试一样计入上限。这两个步骤必须一起出现：只声明其中一个的挂载会让服务在启动时停止。没有组列的表在 dead_letter 中改为给出一个表名，该行转入那张表，它具有相同的列。

读取死信

死信组是这张表中的一个普通队列，review 像读取其他队列一样读取它。

宏

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()
}

手写

src/main.rs

struct Review;

impl Handle<Mail, (), (), InboxContext> for Review {
    async fn handle(
        &self,
        mail: &Mail,
        _outs: &(),
        ctx: &mut Context<'_, InboxContext>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(attempt) = Ctx::<keys::Attempt>::from_context(ctx).await?;
        tracing::error!(to = %mail.to, ?attempt, "email not sent, needs a person");
        Ok(())
    }
}

review 把邮件记入日志，留给人来查看，然后确认它。为运维人员保留死信的服务，不在这个组上挂载任何订阅；这些行留在表中等待。重新发送它们只需服务自己的一条语句，它把这些行移回原组，并重置它们的计数：

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

在带 FIFO 组的表中，retry() 让一行保持在它的组的首位，而 retry_after(d) 让更早到期的行越过它（顺序）。

测试

示例文件中的测试通过测试框架发布一封邮件，写入的方式与服务自己的 Publish 相同。被拒绝的地址立即被丢弃，TestApp::start 在发布返回时就能看到这一点。没有应答的邮件和繁忙的服务商需要时间流逝：延迟的行和死信在订阅的下一次轮询时才回来。这些测试用 TestApp::start_live 在真实时钟上运行，退避时间为一秒，由 tb.advance 等它过去。在 SQL 上测试服务介绍了测试框架。