Базы данных SQL¶

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

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

ruststream-sqlx подключает базы данных SQL к сервису RustStream через sqlx. Крейт состоит из двух компонентов:

inbox - очереди задач в собственных таблицах сервиса в Postgres, MySQL/MariaDB и SQLite. Их обслуживает брокер RustStream;
транзакционный outbox. Сервис публикует сообщения через любой брокер RustStream, а каждое сообщение хранится в отдельной таблице сервиса, пока подписчик его не обработает.

ruststream = { version = "0.7", features = ["macros"] }
ruststream-sqlx = { version = "0.7", features = ["inbox", "outbox", "postgres"] }
sqlx = { version = "0.9", features = ["runtime-tokio", "postgres", "derive", "chrono"] }
chrono = "0.4"
serde = { version = "1", features = ["derive"] }

Функции inbox и outbox включают компоненты, и каждую можно включить отдельно от другой. Функция драйвера выбирает базу данных: postgres, mysql (MySQL и MariaDB), sqlite или any для AnyPool.

Очереди задач в таблицах сервиса¶

crates/ruststream-sqlx/examples/inbox.rs

use std::error::Error;

use ruststream::OutgoingMessage;
use ruststream_sqlx::prelude::*;
use serde::{Deserialize, Serialize};
use sqlx::postgres::{PgConnectOptions, PgConnection, PgPool, Postgres};

/// One row of `email_jobs`: the struct names the table and the role of each column.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

/// A publish to `emails` writes a row through the service's own statement.
impl Publish<Postgres> for SendEmail {
    async fn publish(
        conn: &mut PgConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO email_jobs (name, payload) VALUES ($1, $2)")
            .bind(message.name())
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

#[derive(Deserialize, Serialize, Outgoing)]
struct Email {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    println!("sending to {}", email.to);
    HandlerOutcome::ack()
}

fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(
        SqlxBroker::new(pool).route::<SendEmail>("emails"),
        |b| {
            b.include(send);
        },
    )
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    // sqlx builds a pool only inside a Tokio runtime, so the service builds it here, before the
    // app.
    let pool = PgPool::connect_with(PgConnectOptions::new()).await?;
    app(pool.clone()).run().await?;
    pool.close().await;
    Ok(())
}

#[derive(Inbox)] описывает таблицу очереди: её имя и роль каждого столбца. SqlxBroker обслуживает очереди из пула соединений сервиса. Подписка берёт строки в работу, а от результата обработчика зависит, что станет со строкой: подтверждение удаляет строку, переобработка возвращает её в очередь. Публикация записывает строку собственным запросом сервиса.

Подписка берёт строки в работу в одном из трёх режимов. В режиме блокировки строк транзакция остаётся открытой, пока работает обработчик. В режиме аренды подписка сразу фиксирует транзакцию. В режиме рекомендательных блокировок подписка блокирует ключ строки. Обработчик, при регистрации которого указан .transactional(), записывает данные через транзакцию своей доставки, и подтверждение фиксирует его изменения вместе со строкой. Каждая роль столбца включает одно поведение: группы, порядок, в котором строки берутся в работу, отложенную переобработку, ограничение числа попыток с переносом строки в другую группу или таблицу и отметку, при которой обработанная строка остаётся в таблице.

Ручной API - второй способ описать ту же таблицу, наравне с #[derive]: структура реализует трейт, а все настройки таблицы задаёт типизированный билдер. Компилятор проверяет такое описание так же строго, как #[derive]. Оба способа дают одни и те же запросы и одинаковые затраты.

Команда, которая проверяет свой SQL во время компиляции, добавляет в #[derive] параметр checked. Тогда cargo sqlx prepare проверяет запросы #[derive] вместе с собственными запросами сервиса. Тест запускает собственное приложение сервиса в TestApp на настоящей базе данных, потому что SQL сервиса тоже входит в то, что проверяет тест. Для SQLite сервер не нужен.

Транзакционный outbox¶

crates/ruststream-sqlx/examples/outbox.rs

use std::convert::Infallible;

use chrono::{DateTime, Utc};
use ruststream::OutgoingMessage;
use ruststream::memory::prelude::*;
use ruststream_sqlx::{Outbox, outbox};
use serde::{Deserialize, Serialize};
use sqlx::postgres::{PgConnectOptions, PgConnection, PgPool, Postgres};

/// One record of the `outbox` table.
#[derive(Outbox, sqlx::FromRow)]
#[outbox(table = "outbox")]
struct OrderOutbox {
    #[field(id)]
    id: i64,
    #[field(name)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
}

/// The service writes the record; the outbox sends its id with the message.
impl outbox::Publish<Postgres> for OrderOutbox {
    async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
        sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
            .bind(msg.name())
            .bind(msg.payload())
            .fetch_one(conn)
            .await
    }
}

#[derive(Deserialize)]
struct PlaceOrder {
    id: u64,
}

#[derive(Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders")]
struct OrderPlaced {
    id: u64,
}

/// The reply to `orders` is recorded before it is sent.
#[subscriber("checkout", reply)]
async fn place(cmd: &PlaceOrder) -> OrderPlaced {
    OrderPlaced { id: cmd.id }
}

/// After the ack, the record is marked processed.
#[subscriber("orders")]
async fn fulfil(order: &OrderPlaced) -> HandlerOutcome {
    println!("fulfilling order {}", order.id);
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let tracking = outbox! {
        "orders" => OrderOutbox,
    };
    let registry = tracking.clone();

    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // sqlx builds a pool only inside the Tokio runtime, which starts after this builder.
        .on_startup(async move |()| {
            let pool = PgPool::connect_with(PgConnectOptions::new()).await?;
            registry
                .set_pool(pool.clone())
                .expect("the outbox is deferred and gets one pool");
            Ok::<_, sqlx::Error>(pool)
        })
        .layer(tracking.layer())
        .publish_layer(tracking.publish_layer())
        .after_shutdown(async move |pool| {
            pool.close().await;
            Ok::<_, Infallible>(())
        })
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
            b.include(fulfil);
            // Once the subscriptions are open, the unprocessed records are sent again.
            b.after_startup(Publish, tracking.republish());
        })
}

#[derive(Outbox)] описывает таблицу outbox сервиса, а outbox! регистрирует её под теми именами, которые она отслеживает. Middleware публикации записывает каждое отслеживаемое сообщение перед отправкой и передаёт идентификатор записи в заголовке. Middleware подписки по этому идентификатору берёт запись в работу и помечает её обработанной, когда обработчик подтверждает сообщение. При запуске сервис заново публикует все необработанные записи.

Сообщение доставляется хотя бы один раз. Два экземпляра сервиса, которые запускаются одновременно, могут оба заново отправить одну и ту же запись. Если подписчик не должен обрабатывать сообщение дважды, он проверяет это сам.

Внутри #[ruststream::app] приложение строится до запуска среды выполнения Tokio, а sqlx создаёт пул соединений только внутри неё. Поэтому реестр начинает работу без пула: on_startup создаёт пул и передаёт его через set_pool. В тестовой сборке outbox выключен, пока в окружении не задана переменная RUSTSTREAM_SQLX_OUTBOX=on.

Middleware outbox оборачивают обработчики inbox так же, как любые другие. Запись выполняется через отдельное соединение, поэтому она фиксируется, даже когда транзакция доставки откатывается. Сообщение, которое отслеживается и попадает в таблицу inbox, сохраняет идентификатор записи, если в таблице есть столбец headers и его заполняет запрос публикации сервиса. Подробности - в разделе о транзакционном outbox.

Что выбрать¶

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

Где всё остальное¶

Справочник на docs.rs начинается с учебника по крейту, по разделу на тему:

Брокер inbox: таблица очереди, её подписки и то, что результат обработчика делает со строкой.
Макрос или ручное описание: таблица очереди, описанная вручную через InboxTable и типизированный билдер.
Роли и время: что включает каждый столбец и откуда берётся текущее время.
Блокировки строк, аренда и рекомендательные блокировки и транзакционный режим: как подписка берёт строки в работу и как обработчик записывает данные через транзакцию своей доставки.
Режим строки, структура заголовков и пакеты: что обработчик получает из строки.
Базы данных и собственный диалект сервиса: какие запросы выполняет каждая база данных и как написать диалект для другой.
Проверка запросов во время компиляции: режим checked.
Пробуждение подписки: интервал опроса и LISTEN/NOTIFY в Postgres.
Тестирование сервиса на inbox: собственное приложение сервиса в TestApp.
Транзакционный outbox: гарантия доставки, запись и её реестр, middleware, повторная публикация, пул соединений, тестирование и затраты.

Во что крейт обходится на каждом сообщении и сколько сообщений он обрабатывает в секунду рядом с циклом на чистом sqlx, который делает ту же работу, показано на странице бенчмарков.

Обработчики, роутеры, кодеки и middleware даёт сам фреймворк, а его входные страницы начинаются с сайта RustStream.