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

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

Отслеживание публикаций: плагин outbox

Это руководство доводит каждое сообщение, которое публикует сервис, до обработчика, который его обработает, на брокере, который ничего не хранит.

Некоторые брокеры доставляют сообщение только тем, кто слушает в этот момент, среди них Redis Pub/Sub и NATS core, а получатель может остановиться на середине любого сообщения. Outbox записывает каждое сообщение, которое сервис публикует под зарегистрированным именем, отмечает запись, когда получатель обработал сообщение, и при старте снова публикует необработанные записи. Это плагин поверх любого брокера RustStream, и ему не нужен процесс-ретранслятор. Его фича outbox не зависит от inbox.

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

Таблица​

src/main.rs
const SCHEMA: &str = "CREATE TABLE IF NOT EXISTS outbox (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    payload BLOB NOT NULL,
    processed_at TEXT
)";

Таблица принадлежит сервису: он создаёт её сам, в промышленной среде - собственными миграциями. processed_at остаётся пустым, пока получатель не обработал сообщение.

Запись​

src/main.rs
use ruststream_sqlx::{Outbox, outbox};

/// One record of the `outbox` table.
#[derive(Outbox, sqlx::FromRow)]
#[outbox(table = "outbox")]
struct PaymentOutbox {
    #[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 and returns its id; the outbox sends the id with the message.
impl outbox::Publish<Sqlite> for PaymentOutbox {
    async fn publish(conn: &mut SqliteConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
        sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES (?, ?) RETURNING id")
            .bind(msg.name())
            .bind(msg.payload())
            .fetch_one(conn)
            .await
    }
}

#[derive(Outbox)] описывает запись: id, name и payload обязательны, headers и processed_at необязательны. Сервис сохраняет запись собственным outbox::Publish и возвращает её id; остальные операторы строит крейт. В ручной форме то же описывает OutboxSpec, а отслеживаемое имя называет тип TrackedName.

Обработчики​

src/main.rs
use ruststream::memory::prelude::*;

/// The reply to `payments` is recorded before it is sent.
#[subscriber("checkout", reply)]
async fn take_payment(checkout: &Checkout) -> PaymentTaken {
    tracing::info!(order = checkout.order, "payment taken");
    PaymentTaken {
        order: checkout.order,
    }
}

/// The acknowledgement marks the payment's record processed.
#[subscriber("payments")]
async fn ship(payment: &PaymentTaken) -> HandlerOutcome {
    tracing::info!(order = payment.order, "shipping");
    HandlerOutcome::ack()
}

Обработчики ничего не знают об outbox: отслеживание выполняют middleware приложения.

Приложение​

src/main.rs
use std::convert::Infallible;
use std::io;

use sqlx::SqlitePool;

fn service(url: String) -> impl App {
    // The registry starts without a pool: sqlx builds one only inside the Tokio runtime.
    let tracking = outbox! { "payments" => PaymentOutbox };
    let registry = tracking.clone();

    RustStream::new(AppInfo::new("payments", "0.1.0"))
        .on_startup(async move |()| {
            let pool = SqlitePool::connect(&url).await.map_err(io::Error::other)?;
            sqlx::raw_sql(SCHEMA)
                .execute(&pool)
                .await
                .map_err(io::Error::other)?;
            registry.set_pool(pool.clone()).map_err(io::Error::other)?;
            Ok::<_, io::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(take_payment).out_reply(Publish);
            b.include(ship);
            // Once the subscriptions are open, the unprocessed records are sent again.
            b.after_startup(Publish, tracking.republish());
        })
}

#[ruststream::app]
fn app() -> impl App {
    service("sqlite:payments.db?mode=rwc".to_owned())
}

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

Отслеживаемое сообщение проходит четыре шага:

  1. Слой публикации сохраняет запись, а затем отправляет сообщение с id записи в заголовке x-ruststream-outbox-id. Если запись не удалось сохранить, публикация возвращает ошибку.
  2. Слой подписки берёт в работу доставку с id по этому id и запускает обработчик; копия, запись которой уже обработана, подтверждается без обработчика.
  3. Подтверждение или drop обработчика отмечают запись обработанной; переобработка оставляет её как была.
  4. При старте republish снова отправляет каждую необработанную запись, когда подписки уже открыты.

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

Что гарантирует outbox​

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

Запись сохраняется на собственном соединении и фиксируется сразу. Она не входит в транзакцию сервиса: транзакционный обработчик inbox, который уходит на переобработку, откатывает то, что записал сам, а запись outbox о его публикации остаётся. На SQLite такой обработчик публикует до своей первой записи через свою транзакцию, потому что вставке записи outbox нужна блокировка записи, которую держит эта транзакция. Публикация после этой записи завершается ошибкой OutboxError::WriteLockHeld (Взаимодействие). Событие, которое должно фиксироваться вместе с бизнес-строками, - это строка бизнес-транзакции, которую затем ретранслируют: Ретрансляция событий в другой брокер.

Публикация вне обработчика, например из конечной точки HTTP, берёт издателя работающего приложения и оборачивает его в реестр через wrap, поэтому она отслеживается так же. Отслеживание стоит одно соединение на каждую отслеживаемую публикацию и два на каждую отслеживаемую доставку: одно на получение записи, одно на завершение. Сообщение под именем, которого нет в реестре, стоит одно сравнение имён. examples/outbox в репозитории выполняет тот же сценарий поверх Redis Pub/Sub, с миграциями и собственным получением записи.

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

Тестовая сборка оставляет outbox выключенным, если его не включает RUSTSTREAM_SQLX_OUTBOX=on (Переключатель), поэтому собственные тесты сервиса запускают его обработчики без отслеживания. Тесты в файле примера, которые проверяют отслеживание, без переключателя пропускаются; ещё один тест проверяет, что при выключенном переключателе сервис работает без отслеживания:

cargo test
RUSTSTREAM_SQLX_OUTBOX=on cargo test

При включённом переключателе один тест проверяет, что оплата записана, отправлена с id своей записи и отмечена после отгрузки. Другой записывает запись, которую предыдущий экземпляр оставил необработанной, и находит её отгруженной после старта. Оба работают через TestApp::start на базе данных SQLite в памяти, которую открывает по URL собственный пул сервиса.