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

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

Постановка задач в очередь

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

У очереди в базе данных сервиса есть одно преимущество перед отдельным брокером: задача фиксируется вместе с данными, которые она обслуживает. Заказ, который откатился, не оставляет после себя письма с подтверждением, а письмо никогда не уходит по заказу, который не сохранился.

Пример - интернет-магазин. Подписка оформления заказа размещает каждый заказ, а подписка писем сразу отправляет подтверждение, а через три дня - просьбу оставить отзыв.

Таблица​

src/main.rs
const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY, email TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS checkout_jobs (
    id INTEGER PRIMARY KEY,
    locked_until TEXT,
    payload BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS email_jobs (
    id INTEGER PRIMARY KEY,
    priority INTEGER NOT NULL,
    retry_after TEXT NOT NULL,
    attempt INTEGER NOT NULL DEFAULT 1,
    processed_at TEXT,
    locked_until TEXT,
    payload BLOB NOT NULL
);
";
src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Sqlite, SqlitePool};

/// One row of `email_jobs`.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
struct SendEmail {
    #[field(id, generated)]
    id: i64,
    #[field(priority)]
    priority: i64,
    #[field(retry_after)]
    retry_after: DateTime<Utc>,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// A smaller priority is claimed first.
const URGENT: i64 = 0;
const ROUTINE: i64 = 10;

impl SendEmail {
    /// A job for `email`, due at `due`.
    fn new(email: &Email, priority: i64, due: DateTime<Utc>) -> serde_json::Result<Self> {
        Ok(Self {
            id: 0,
            priority,
            retry_after: due,
            attempt: 1,
            processed_at: None,
            locked_until: None,
            payload: serde_json::to_vec(email)?,
        })
    }
}

Каждый столбец включает одно поведение очереди (Роли). retry_after хранит время, когда строка становится готовой к захвату: захват берёт только строки, время которых наступило. priority задаёт порядок захвата: меньшее значение идёт первым, затем более ранний retry_after, затем меньший id. processed_at сохраняет выполненную строку: подтверждение записывает в него время, а не удаляет строку. generated отмечает столбцы, которые заполняет база данных, поэтому вставка их пропускает. В ручной форме сервис сам пишет эту вставку как реализацию Insert.

Задача в транзакции заказа​

src/main.rs
/// The order, its confirmation and a reminder three days later commit together, or not at all.
async fn place_order(pool: &SqlitePool, order: &Order) -> Result<(), Box<dyn Error>> {
    let mut tx = pool.begin().await?;
    sqlx::query("INSERT INTO orders (id, email) VALUES (?, ?)")
        .bind(order.id)
        .bind(&order.email)
        .execute(&mut *tx)
        .await?;
    let now = Utc::now();
    let confirmation = Email::confirmation(order);
    SendEmail::new(&confirmation, URGENT, now)?
        .insert(&mut *tx)
        .await?;
    let reminder = Email::review_request(order);
    SendEmail::new(&reminder, ROUTINE, now + TimeDelta::days(3))?
        .insert(&mut *tx)
        .await?;
    tx.commit().await?;
    Ok(())
}

place_order открывает одну транзакцию sqlx и записывает в ней три строки: заказ, подтверждение и напоминание. insert из derive принимает любое соединение с базой данных таблицы, поэтому &mut *tx помещает задачи в транзакцию заказа. Они фиксируются вместе или вместе откатываются: заказ, вставка которого вернула ошибку, не оставляет в таблице ни одного письма. Конечная точка HTTP, которая размещает заказы, вызывает ту же функцию с собственным пулом сервиса; сервис, у которого один пул на конечные точки и очереди, передаёт его в SqlxBroker::new.

Подтверждение срочное и готово к захвату сразу. Напоминание станет готовым через три дня, и до тех пор его никто не захватит. Повторяющаяся задача устроена так же: её обработчик вставляет следующую строку с более поздним retry_after. Задача, которая должна существовать в единственном экземпляре, - это уникальный индекс сервиса и INSERT .. ON CONFLICT DO NOTHING в собственном операторе.

Подписка​

src/main.rs
#[subscriber(InboxQueue::<Checkout>::new("checkouts"))]
async fn checkout(order: &Order, Ctx(pool): Ctx<keys::Pool<Sqlite>>) -> HandlerOutcome {
    match place_order(&pool, order).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(error) => {
            tracing::warn!(subscription = "checkouts", order.id, %error, "order not placed");
            HandlerOutcome::drop()
        }
    }
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(order = email.order, template = %email.template, "sending");
    HandlerOutcome::ack()
}

/// 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("shop", "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::<Checkout>("checkouts"), |b| {
            b.include(checkout);
            b.include(send);
        })
}

checkout берёт пул из своего контекста и размещает заказ; заказ, размещение которого вернуло ошибку, отбрасывается с предупреждением. Его собственная транзакция отделена от захвата строки оформления: обработчик, записи которого должны фиксироваться вместе с завершением его строки, работает в транзакционном режиме. send читает каждую строку email_jobs в порядке захвата. Его подтверждение отмечает строку через processed_at, поэтому отправленные письма остаются в таблице для аудита, а захват их пропускает. Таблица без этого столбца теряет строку при подтверждении.

Когда задачу видно​

Публикация через брокер сразу будит подписки своей таблицы в том же процессе: маршрут, Repository, ответ или слот Out. Так оформление, опубликованное витриной магазина, будит checkout. Строка, которую сервис пишет собственным SQL, как place_order, никого не будит. Подписка находит её по истечении своего интервала опроса, по умолчанию одной секунды. Публикация брокера не входит в транзакцию сервиса, поэтому задача, которая должна фиксироваться вместе с заказом, вставляется через insert и ждёт этот интервал. На Postgres ожидание раньше заканчивает уведомление от сервиса или триггера.

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

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