Skip to main content

Ordering: FIFO groups and partition keys

This guide keeps jobs in order: all the jobs of a group, or the jobs of each key while other keys run in parallel.

SKIP LOCKED makes claims fast because they pass over rows in work. It also lets the next row go out while an earlier one is still in work or waiting for its retry. A queue whose jobs must apply in order says so in its table.

The example is an accounts service. It sends contact changes to a CRM, which must receive them in the order they happened, and applies ledger postings, which must keep their order within an account.

The tables​

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
);
";

The index on crm_jobs follows the claim order, the group first. A claim of a FIFO group reads the group's unfinished rows, and the index keeps that read to them (Running in production).

A FIFO group​

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)] makes each group of target a FIFO group. A FIFO group has at most one row in work, however many workers, subscriptions and processes read it. Its first unfinished row in claim order is its head, and a claim takes the head alone. The rows behind it wait while the head is in work, under a lease of another instance included. retry() keeps the change at the head, so it goes again before the ones behind it. retry_after(d) would move it behind the rows due before it.

A head that keeps failing holds its whole group. The mount caps it:

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)));
        })
}

After three refusals the change moves into the crm.parked group, and the next change of crm becomes the head. A parked change waits there for a person; the changes behind it have gone out without it, and that is the price of freeing the group.

Groups run side by side, each through a subscription of its own, so a group suits a few streams known in advance: one per downstream system, one per tenant. workers(n) gives a FIFO subscription no parallelism, and one worker serves it.

A key per account​

A ledger has many accounts, too many for a subscription each. The order that matters is within one account, and accounts may run in parallel.

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)] gives each row the account as its key. The app mounts post with workers_by_key(4): four workers, and the postings of one account on one of them, in the order they were claimed. The subscription keeps one claim in flight on such a table, so the claims do not overtake each other. That order holds inside one subscription. Postings of one account that must stay in order across several instances go into a FIFO group, or into the advisory lock form with a key on the account, which keeps the rows of one key in work one at a time across every process on a server.

Testing​

The tests in the example file run the service on SQLite in memory. The first publishes a change the CRM refuses and one it takes: the first goes three times and is parked, then the second goes out. The second writes the head of crm as leased by another instance and finds that the change behind it waits. The third writes postings of two accounts, interleaved, and finds each account's postings applied in order. The rows of the last two tests wake no subscription, so they run on the real clock with TestApp::start_live past the poll interval.