Перейти к основному содержимому
Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Переобработка и dead letter

Это руководство переобрабатывает задачу сразу или позже, ограничивает число её попыток и откладывает то, что раз за разом не обрабатывается, в группу dead letter той же таблицы.

Переобработка здесь - это оператор над строкой задачи. Строка сохраняет свой 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()
        }
    }
}

Отсрочку выбирает обработчик. Он читает попытку доставки и с каждой попыткой просит на секунду больше: retry_after(d) записывает «сейчас» плюс d в столбец retry_after строки, и до этого времени строку не берёт ни один захват. Таблица без этого столбца возвращает строку сразу, а рантайм пишет предупреждение в журнал. retry() сразу возвращает строку в очередь; её следующая доставка читает попытку на единицу больше. drop() завершает строку так же, как подтверждение, поэтому отклонённое письмо уходит из таблицы. Строка, полезная нагрузка которой не декодируется, тратит свои попытки, как при любой другой ошибке (Декодирование).

Предел и группа dead letter​

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 читает attempt строки. Пятая неудачная доставка не переобрабатывается: dead_letter перемещает строку в группу emails.dead с полезной нагрузкой и попыткой в том виде, в каком они были. Отложенная переобработка засчитывается в предел так же, как немедленная. Два шага задаются вместе: точка монтирования, которая объявляет один из них без другого, останавливает сервис при старте. Таблица без столбца группы принимает в dead_letter имя таблицы, и строка перемещается в эту таблицу с теми же столбцами.

Чтение dead letter​

Группа 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()
}

review пишет письмо в журнал, чтобы его посмотрел человек, и подтверждает его. Сервис, который хранит dead letter для оператора, не монтирует подписку на эту группу, и строки ждут в таблице. Чтобы отправить их снова, достаточно одного собственного оператора сервиса, который возвращает строки и сбрасывает их счётчик:

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

В таблице с группами FIFO retry() оставляет строку в голове её группы, а retry_after(d) пропускает вперёд строки, готовые раньше (Порядок).

Тестирование​

Тесты в файле примера публикуют письмо через обвязку так, как его записывает собственный Publish сервиса. Письмо на отклонённый адрес отбрасывается сразу, и TestApp::start показывает это, когда публикация возвращает управление. Письму без ответа и занятому провайдеру нужно, чтобы прошло время: отложенная строка и dead letter возвращаются при следующем опросе подписок. Эти тесты работают на реальных часах через TestApp::start_live с отсрочкой в одну секунду, и tb.advance выжидает её. Работу с обвязкой описывает руководство Тестирование сервиса на SQL.