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

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

Ровно один раз внутри базы данных: транзакционные обработчики

Это руководство фиксирует записи обработчика вместе с завершением его задачи, поэтому задача, результат которой - строки в той же базе данных, вступает в силу ровно один раз.

Очередь доставляет каждую задачу хотя бы один раз. Обработчик, который записал свой результат и аварийно завершился до завершения задачи, запускается снова и записывает результат дважды. Когда результат - строки в той базе данных, где лежит очередь, этот дубликат убирает одна транзакция: записи и завершение фиксируются вместе или не фиксируется ничего.

Пример - сервис возвратов. Возврат записывает свою строку в refunds и задачу на чек для сервиса рассылки, а затем вызывает процессинг карт.

Таблицы​

src/main.rs
const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS refunds (order_id INTEGER PRIMARY KEY, amount_cents INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS refund_jobs (
    id INTEGER PRIMARY KEY,
    attempt INTEGER NOT NULL DEFAULT 1,
    locked_until TEXT,
    payload BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS receipt_jobs (
    id INTEGER PRIMARY KEY,
    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 `refund_jobs`, the refunds the shop's support queues.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "refund_jobs")]
struct IssueRefund {
    #[field(id, generated)]
    id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// One row of `receipt_jobs`, the mailer's queue.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "receipt_jobs")]
struct SendReceipt {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

refund_jobs - очередь, которую читает этот сервис. receipt_jobs - очередь сервиса рассылки: этот сервис только пишет в неё, через insert из derive, а в ручной форме - собственным оператором.

Обработчик​

src/main.rs
/// The refund's record and its receipt, written on the connection of the delivery's transaction.
async fn record(
    conn: &mut SqliteConnection,
    refund: &Refund,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    sqlx::query("INSERT INTO refunds (order_id, amount_cents) VALUES (?, ?)")
        .bind(refund.order)
        .bind(refund.amount_cents)
        .execute(&mut *conn)
        .await?;
    let receipt = SendReceipt {
        id: 0,
        locked_until: None,
        payload: serde_json::to_vec(&Receipt {
            order: refund.order,
        })?,
    };
    receipt.insert(conn).await?;
    Ok(())
}

#[subscriber(InboxQueue::<IssueRefund>::new("refunds"))]
async fn issue(
    refund: &Refund,
    Ctx(mut tx): Ctx<keys::Tx<Sqlite>>,
    Ctx(attempt): Ctx<keys::Attempt>,
) -> HandlerOutcome {
    if let Err(error) = record(&mut tx, refund).await {
        tracing::warn!(subscription = "refunds", refund.order, %error, "refund not recorded");
        return HandlerOutcome::drop();
    }
    // The processor takes the order as its idempotency key: a call repeated after a retry
    // refunds the card once.
    if !card_refund(refund, attempt).await {
        // The retry rolls the record and the receipt back first.
        return HandlerOutcome::retry();
    }
    // The commit keeps the record and the receipt, and finishes the refund's row.
    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("refunds", "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::<IssueRefund>("refunds"), |b| {
            b.include(issue.transactional());
        })
}

.transactional() в точке монтирования предоставляет обработчику транзакцию его доставки. Обработчик принимает её первым параметром Ctx, Ctx<keys::Tx<Sqlite>>, и пишет через &mut *tx. Обработчик, который принимает keys::Tx в подписке без этого шага, не компилируется.

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

Задача на чек образует цепочку: сервис рассылки видит её, только когда возврат зафиксирован, и никогда - для возврата, который откатился. Запись через Ctx<keys::Pool> фиксируется сама по себе, каким бы ни был исход, как и публикация через брокер; ни то ни другое в транзакцию не входит.

Что остаётся снаружи​

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

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

Изоляция​

Транзакция каждой доставки открывается на уровне изоляции таблицы, #[inbox(isolation = read_committed)] на сервере, или в её режиме SQLite, #[inbox(mode = immediate)]. Таблица, которая не объявляет ни того ни другого, получает уровень базы данных по умолчанию, а на MySQL и MariaDB - READ COMMITTED. Каждая доставка в работе держит соединение для своей транзакции, поэтому обработчику, который берёт второе через keys::Pool или публикацию, нужен пул больше его workers(n). Транзакционный режим обслуживает одиночные доставки; обработчик пакетов, смонтированный с ним, не компилируется. На MySQL и MariaDB обработчик, который прерывает оператор через tx, как это делает select! вокруг запроса, может нарушить следующее за ним завершение. Это ошибка sqlx-mysql 0.9.0: дайте каждому оператору завершиться.

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

Тесты в файле примера запускают сервис через TestApp::start на SQLite в памяти и читают таблицы после каждой публикации. Возврат фиксирует одну строку в refunds и один чек. Возврат, первый вызов процессинга которого истёк по тайм-ауту, переобрабатывается, и вторая попытка оставляет одну строку в refunds и один чек, а не по два. Тот же возврат, опубликованный дважды, во второй раз отбрасывается.