Работа в промышленной среде

Это руководство выводит очередь в промышленную среду на Postgres: схема и её индекс, проверки, которые ловят неверную схему, пул, версии баз данных и пулеры, завершение работы.

Сервис - рассылка писем с восемью воркерами, пределом в пять попыток и группой dead letter.

Схема

src/main.rs

/// The service's first migration; in a service crate, `migrations/0001_email_jobs.sql`, run by
/// `sqlx::migrate!`.
const MIGRATION: &str = "
CREATE TABLE IF NOT EXISTS email_jobs (
    job_id  BIGSERIAL PRIMARY KEY,
    name    TEXT NOT NULL,
    attempt SMALLINT NOT NULL DEFAULT 1,
    payload BYTEA NOT NULL
);
CREATE INDEX IF NOT EXISTS email_jobs_claim ON email_jobs (name, job_id);
";

Крейт не создаёт таблиц: схема принадлежит сервису, и создают её его миграции, как sqlx::migrate! запускает их в examples/outbox. Индекс повторяет порядок захвата: (group, priority, retry_after, id) из тех столбцов, что есть в таблице, здесь (name, job_id). Захват блокирует каждую прочитанную строку своей группы до того, как их отсортировать, поэтому без такого индекса захват, удерживаемый на время обработчика, закрывает остальную часть группы от других захватов. В таблице с группами FIFO тот же индекс ограничивает чтение группы при захвате собственными строками группы.

Проверяемые операторы

Макросы

src/main.rs

use ruststream_sqlx::prelude::*;
use sqlx::Postgres;

/// One row of `email_jobs`: every statement it determines goes through sqlx's compile-time check.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs", checked, db = postgres)]
struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(payload)]
    payload: Vec<u8>,
}

Вручную

src/main.rs

use ruststream_sqlx::prelude::*;
use sqlx::Postgres;

/// One row of `email_jobs`, claimed by row lock: the description names no other form. Each
/// subscription prepares its statements at startup; the compile-time check is the derive's.
#[derive(sqlx::FromRow)]
struct SendEmail {
    job_id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

impl InboxTable for SendEmail {
    type Id = i64;
    type Table = InboxSpec<(Attempt, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("email_jobs", Column::new("job_id").generated())
        .group(Column::new("name"))
        .attempt(Column::new("attempt").generated())
        .payload(Column::new("payload"));

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

impl AttemptRow for SendEmail {
    type Attempt = i16;

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

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

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

checked, db = postgres передаёт каждый оператор, который определяет структура, проверке sqlx во время компиляции: захват, завершения, перемещение в dead letter группы и вставку. Столбец, который называет структура и которого нет в таблице, останавливает сборку. Сборка проверяет операторы по DATABASE_URL или по данным .sqlx, которые пишет cargo sqlx prepare и которые хранятся в репозитории рядом с крейтом; cargo sqlx prepare --check в CI останавливает сборку с устаревшими данными. Подписка выполняет те же операторы, что и без checked, поэтому сообщение стоит столько же. Эту проверку выполняет derive: для таблицы, описанной вручную, остаётся только проверка при старте.

Каждая подписка к тому же подготавливает свои операторы при старте. Таблица или столбец, которых нет в базе данных, останавливают сервис с ошибкой SqlxBrokerError::Schema, которая называет подписку, таблицу и оператор. Точка монтирования, которая объявляет предел без dead letter или наоборот, тоже останавливает его. Поэтому on_startup выполняет миграции до открытия подписок, и проверки видят ту схему, которую читает выпуск.

Пул

Макросы

src/main.rs

use std::num::NonZeroUsize;
use std::time::Duration;

use sqlx::Connection;
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};

/// The emails in work at once; each holds its row's lock and a connection until it settles.
const WORKERS: NonZeroUsize = nonzero!(8);

/// A connection for each worker, and one the subscription leaves to settlements and publishes.
fn pool() -> PgPoolOptions {
    let size = u32::try_from(WORKERS.get() + 1).unwrap_or(u32::MAX);
    PgPoolOptions::new()
        .max_connections(size)
        .acquire_timeout(Duration::from_secs(5))
}

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

/// The service on the database `options` reach.
fn service(options: PgConnectOptions) -> RustStream {
    let migrate = options.clone();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_connect_options(options, pool()).route::<SendEmail>("emails");
    RustStream::new(AppInfo::new("mailer", "0.1.0"))
        // The schema the release reads exists before any subscription prepares its statements.
        .on_startup(async move |()| {
            let mut conn = PgConnection::connect_with(&migrate).await?;
            sqlx::raw_sql(MIGRATION).execute(&mut conn).await?;
            conn.close().await
        })
        // The label names the server in the generated AsyncAPI document.
        .with_broker_labeled("postgres", broker, |b| {
            b.include(send.workers(WORKERS))
                .max_attempts(nonzero!(5u32))
                .dead_letter("emails.dead");
        })
}

Вручную

src/main.rs

use std::num::NonZeroUsize;
use std::time::Duration;

use sqlx::Connection;
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};

/// The emails in work at once; each holds its row's lock and a connection until it settles.
const WORKERS: NonZeroUsize = nonzero!(8);

/// A connection for each worker, and one the subscription leaves to settlements and publishes.
fn pool() -> PgPoolOptions {
    let size = u32::try_from(WORKERS.get() + 1).unwrap_or(u32::MAX);
    PgPoolOptions::new()
        .max_connections(size)
        .acquire_timeout(Duration::from_secs(5))
}

struct Mailer;

impl Handle<Email, (), (), InboxContext> for Mailer {
    async fn handle(
        &self,
        email: &Email,
        _outs: &(),
        ctx: &mut Context<'_, InboxContext>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(attempt) = Ctx::<keys::Attempt>::from_context(ctx).await?;
        tracing::info!(to = %email.to, ?attempt, "sending");
        Ok(())
    }
}

/// The service on the database `options` reach.
fn service(options: PgConnectOptions) -> RustStream {
    let migrate = options.clone();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_connect_options(options, pool()).route::<SendEmail>("emails");
    RustStream::new(AppInfo::new("mailer", "0.1.0"))
        // The schema the release reads exists before any subscription prepares its statements.
        .on_startup(async move |()| {
            let mut conn = PgConnection::connect_with(&migrate).await?;
            sqlx::raw_sql(MIGRATION).execute(&mut conn).await?;
            conn.close().await
        })
        // The label names the server in the generated AsyncAPI document.
        .with_broker_labeled("postgres", broker, |b| {
            let send = subscriber("emails", Mailer)
                .map_source(|_| InboxQueue::<SendEmail>::new("emails"))
                .workers(WORKERS);
            b.include(send.build())
                .max_attempts(nonzero!(5u32))
                .dead_letter("emails.dead");
        })
}

Таблица не задаёт форму, поэтому захватывает под блокировкой строк: каждое письмо в работе держит блокировку своей строки и соединение, пока не завершится. Подписка никогда не держит больше соединений, чем размер пула минус одно, и оставляет это одно для собственных запросов обработчиков, их публикаций и завершений. Поэтому пул рассчитывают на все подписки сервиса: n на каждый workers(n), ещё одно и соединения, которые обработчики берут для собственной работы (Соединения). Таблица аренды держит соединение только на время каждого оператора и подходит для пула, который не может выделить по соединению на задачу. На Postgres уведомления держат ещё одно соединение на всё время жизни брокера. pool возвращает параметры пула, а брокер строит по ним пул при старте сервиса.

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    // The connection's settings come from the `PG*` environment variables.
    service(PgConnectOptions::new())
}

Вручную

src/main.rs

fn main() -> ExitCode {
    // The connection's settings come from the `PG*` environment variables.
    run_main(|| service(PgConnectOptions::new()))
}

Базы данных и пулеры

База данных	Требования
Postgres	9.5 для SKIP LOCKED; 11 для формы рекомендательной блокировки и групп FIFO, которые блокируют 64-битные хеши
MySQL, MariaDB	MySQL 8.0.1 или MariaDB 10.6, версия читается при старте; более старый сервер останавливает старт с ошибкой ServerTooOld
SQLite	форма аренды или рекомендательной блокировки; в форме рекомендательной блокировки один процесс на файл базы данных

На MySQL и MariaDB захват выполняется на уровне READ COMMITTED, если таблица блокировки строк не объявляет другой уровень. Сервер, который записывает в журнал операторы, binlog_format = STATEMENT, отклоняет записи на этом уровне; форматы row и mixed их принимают. На Postgres практичный уровень изоляции таблицы блокировки строк - read_committed: более строгий уровень даёт в захватах ошибки сериализации, когда захваты и завершения выполняются одновременно. Форма рекомендательной блокировки хранит блокировку в сеансе базы данных, поэтому ей нужно прямое соединение или PgBouncer в режиме сеанса; в режиме транзакций пулер между операторами отдаёт сеанс другому клиенту. Брокер с listen_notify тоже слушает на сеансе.

Завершение работы

При завершении работы рантайм останавливает подписки, даёт обработчикам в работе закончить в пределах своего тайм-аута завершения, а затем завершает работу брокера. Доставка в форме блокировки строк держит свою транзакцию, пока не завершится; та, что ещё не завершена к выходу процесса, откатывается вместе со своим соединением, и её строка возвращается. Доставка в форме аренды сохраняет свою аренду и завершается, если успевает, а иначе её строка возвращается, когда аренда истекает. Брокер снимает рекомендательные блокировки доставок в работе. Свой пул брокер закрывает последним; соединение, которое ещё держит доставка в работе, закрывается, когда возвращается.

Документ AsyncAPI

С фичей asyncapi генерируемый документ AsyncAPI описывает сервер, а также таблицу и группу каждой подписки. Метка with_broker_labeled называет сервер, а запись сервера содержит хост и порт базы данных, но никогда не его пользователя или пароль: документ публикуется и передаётся другим.

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

Один тест в файле примера строит документ сервиса по URL с паролем, без ввода-вывода, и находит в нём хост и порт, а пароль - нигде. Ещё два работают на собственной базе данных Postgres: таблица прежнего выпуска, без столбца, который называет структура, останавливает старт с ошибкой, которая называет таблицу, а проверяемые операторы обслуживают очередь. Они читают адрес сервера из POSTGRES_TEST_URL и без него пропускаются (Тестирование сервиса на SQL).