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

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

Порядок: группы FIFO и ключи партиционирования

Это руководство сохраняет порядок задач: всех задач группы или задач каждого ключа, пока задачи других ключей выполняются параллельно.

SKIP LOCKED ускоряет захваты, потому что они пропускают строки в работе. Из-за этого же следующая строка может уйти в работу, пока более ранняя ещё в работе или ждёт переобработки. Очередь, задачи которой должны применяться по порядку, объявляет это в своей таблице.

Пример - сервис счетов. Он отправляет изменения контактов в CRM, которая должна получать их в том порядке, в каком они произошли, и применяет проводки по счетам, которые должны сохранять порядок в пределах одного счёта.

Таблицы​

src/main.rs
const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS crm_jobs (
    id INTEGER PRIMARY KEY,
    target TEXT NOT NULL,
    attempt INTEGER NOT NULL DEFAULT 1,
    locked_until TEXT,
    payload BLOB NOT NULL
);
CREATE INDEX IF NOT EXISTS crm_jobs_claim ON crm_jobs (target, id);
CREATE TABLE IF NOT EXISTS ledger_jobs (
    id INTEGER PRIMARY KEY,
    account TEXT NOT NULL,
    locked_until TEXT,
    payload BLOB NOT NULL
);
";

Индекс по crm_jobs повторяет порядок захвата, начиная с группы. Захват в группе FIFO читает незавершённые строки группы, и индекс ограничивает это чтение ими (Работа в промышленной среде).

Группа FIFO​

src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `crm_jobs`; each group holds the changes bound for one system, in order.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "crm_jobs")]
struct CrmChange {
    #[field(id, generated)]
    id: i64,
    #[field(group, fifo = true)]
    target: String,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// The CRM receives its changes one at a time, in the order they were queued.
#[subscriber(InboxQueue::<CrmChange>::new("crm"))]
async fn sync(change: &Contact) -> HandlerOutcome {
    if push(change).await {
        return HandlerOutcome::ack();
    }
    // At once: the change keeps its place at the head of the group.
    HandlerOutcome::retry()
}

#[field(group, fifo = true)] делает каждую группу target группой FIFO. В группе FIFO в работе не больше одной строки, сколько бы воркеров, подписок и процессов её ни читали. Первая незавершённая строка группы в порядке захвата - её голова, и захват берёт только голову. Строки за ней ждут, пока голова в работе, в том числе под арендой другого экземпляра. retry() оставляет изменение в голове, поэтому оно снова уходит раньше тех, что за ним. retry_after(d) переместил бы его за строки, готовые раньше него.

Голова, обработка которой раз за разом не удаётся, держит всю свою группу. Точка монтирования ограничивает число её попыток:

src/main.rs
/// 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());
    let broker = broker
        .route::<CrmChange>("crm")
        .route::<LedgerPosting>("postings");
    RustStream::new(AppInfo::new("accounts", "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, |b| {
            // A change the CRM refuses three times moves out of the group, which then moves on.
            b.include(sync)
                .max_attempts(nonzero!(3u32))
                .dead_letter("crm.parked");
            // Four workers, and the postings of one account on one of them, one after another.
            b.include(post.workers_by_key(nonzero!(4)));
        })
}

После трёх отказов изменение перемещается в группу crm.parked, и головой становится следующее изменение crm. Отложенное изменение ждёт там человека. Изменения за ним уже ушли без него, и это цена освобождения группы.

Группы работают бок о бок, каждая через собственную подписку, поэтому группа подходит для нескольких заранее известных потоков: по одной на внешнюю систему, по одной на клиента. workers(n) не даёт подписке FIFO параллелизма: её обслуживает один воркер.

Ключ на счёт​

Счетов в учёте много, слишком много, чтобы заводить подписку на каждый. Важен порядок внутри одного счёта, а разные счета могут обрабатываться параллельно.

src/main.rs
/// One row of `ledger_jobs`: the postings of one account apply in the order they were written.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "ledger_jobs")]
struct LedgerPosting {
    #[field(id, generated)]
    id: i64,
    #[field(partition_key)]
    account: String,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<LedgerPosting>::new("postings"))]
async fn post(posting: &Posting) -> HandlerOutcome {
    tracing::info!(account = %posting.account, posting.n, "posting");
    HandlerOutcome::ack()
}

#[field(partition_key)] делает счёт ключом каждой строки. Приложение монтирует post с workers_by_key(4): четыре воркера, а проводки одного счёта идут на одном из них в порядке захвата. На такой таблице подписка держит не больше одного захвата одновременно, поэтому захваты не обгоняют друг друга. Этот порядок действует внутри одной подписки. Проводки одного счёта, которые должны сохранять порядок на нескольких экземплярах, помещаются в группу FIFO или в форму рекомендательной блокировки с ключом по счёту: она пускает строки одного ключа в работу по одной во всех процессах на сервере.

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

Тесты в файле примера запускают сервис на SQLite в памяти. Первый публикует изменение, которое CRM отклоняет, и изменение, которое она принимает: первое уходит три раза и откладывается, затем уходит второе. Второй записывает голову crm как арендованную другим экземпляром и находит, что изменение за ней ждёт. Третий записывает вперемешку проводки двух счетов и находит, что проводки каждого счёта применены по порядку. Строки последних двух тестов не будят ни одну подписку, поэтому эти тесты работают на реальных часах через TestApp::start_live и ждут дольше интервала опроса.