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

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

Ретрансляция событий в другой брокер

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

У сервиса, который сохраняет заказ, а затем публикует в Kafka «заказ размещён», две записи, которые могут разойтись. Заказ может зафиксироваться, а публикация вернуть ошибку, или публикация может уйти для заказа, который откатился. Транзакционный outbox объединяет их: событие - это строка, записанная в транзакции заказа, а ретранслятор публикует зафиксированные строки. С этим крейтом таблица событий - это таблица очереди, а ретранслятор - обычная подписка.

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

Заказ и его событие​

src/main.rs
const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY, total_cents INTEGER 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 order_events (
    id INTEGER PRIMARY KEY,
    aggregate TEXT NOT NULL,
    locked_until TEXT,
    payload BLOB NOT NULL
);
";
src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Sqlite, SqliteConnection};

/// One row of `order_events`: an event waiting for the relay, keyed by its order.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "order_events")]
struct OrderEvent {
    #[field(id, generated)]
    id: i64,
    #[field(partition_key)]
    aggregate: String,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// The order and its event, in the checkout's transaction: both commit, or neither does.
async fn place(
    conn: &mut SqliteConnection,
    order: &Order,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    sqlx::query("INSERT INTO orders (id, total_cents) VALUES (?, ?)")
        .bind(order.id)
        .bind(order.total_cents)
        .execute(&mut *conn)
        .await?;
    let placed = OrderPlaced {
        order: order.id,
        total_cents: order.total_cents,
    };
    let event = OrderEvent {
        id: 0,
        aggregate: order.id.to_string(),
        locked_until: None,
        payload: serde_json::to_vec(&placed)?,
    };
    event.insert(conn).await?;
    Ok(())
}

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

checkout работает в транзакционном режиме: он записывает заказ и событие через транзакцию своей доставки. Его подтверждение фиксирует обе строки вместе с завершением оформления. Заказ, вставка которого вернула ошибку, откатывает своё событие, и для него ничего не публикуется. Код вне обработчика, например конечная точка HTTP, записывает те же две строки в собственной транзакции sqlx, как в руководстве Постановка задач в очередь. Строка события и в этом случае не будит ни одну подписку: ретранслятор находит её в пределах своего интервала опроса.

Ретранслятор​

src/main.rs
use ruststream::memory::{MemoryBroker, MemoryPublish};

/// Each event goes out on the bus as it was written; the acknowledgement deletes its row.
#[subscriber(InboxQueue::<OrderEvent>::new("order-events"), reply)]
async fn relay(event: &OrderPlaced) -> OrderPlaced {
    event.clone()
}

/// 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());
    // The target broker is wrapped first, so its publish token exists before the scopes.
    let bus = MemoryBroker::new().bindable();
    let to_bus = bus.bind(MemoryPublish);
    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.transactional());
            b.include(relay.workers_by_key(nonzero!(4)))
                .out_reply(to_bus);
        })
        .with_broker(bus, |_b| {})
}

relay читает order_events и отвечает на каждое событие им самим; тип ответа называет своё место назначения, orders. Ответ уходит через шину: bindable() оборачивает целевой брокер, bind(MemoryPublish) выпускает для него токен публикации, а .out_reply(to_bus) отправляет ответы relay через этот токен. Сервис на Kafka или NATS указывает в тех же двух строках свой брокер и свою политику публикации. Ответ публикуется до того, как завершится строка события, а подтверждение удаляет строку.

Ретранслятор работает по схеме at-least-once. Ответ, публикация которого вернула ошибку, возвращает событие в очередь, а сбой после публикации и до подтверждения публикует его снова. Получатель, который не должен действовать дважды, хранит идентификатор события, здесь заказ, и пропускает уже виденное.

Порядок по агрегату​

aggregate - ключ партиционирования события, id заказа. workers_by_key(4) ретранслирует четыре заказа одновременно и сохраняет события одного заказа в том порядке, в каком они записаны. Этот порядок действует внутри одной подписки ретранслятора. Ретранслятор на нескольких экземплярах, события которого должны сохранять порядок по агрегату на всех них, держит их в группе FIFO или в форме рекомендательной блокировки.

Ретранслятору не нужны ни флаг «отправлено», ни собственный опрос: захват выполняет очередь, задержка после неудачной публикации - это переобработка, а предел с группой dead letter откладывает событие, которое цель раз за разом отклоняет. Отслеживание публикаций - другой вариант: плагин outbox записывает то, что публикует обработчик, и отмечает запись, когда получатель обработал сообщение, но его запись фиксируется сама по себе, вне бизнес-транзакции.

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

Тесты в файле примера публикуют оформления через обвязку и читают, что дошло до шины, через tb.broker::<MemoryBroker>().published(..). Один заказ доходит до шины один раз, а строка его события удалена. Тот же заказ, оформленный дважды, публикует одно событие: второе оформление возвращает ошибку, и его событие откатывается. Строки событий не будят ни одну подписку, поэтому тесты работают на реальных часах через TestApp::start_live и ждут дольше интервала опроса.