Skip to main content

Enqueueing jobs

This guide writes jobs into a queue table: in the same transaction as the data they serve, for a later time, and by priority.

A queue in the service's database has one advantage over a broker of its own: a job commits with the data it serves. An order that rolls back leaves no confirmation email behind, and an email never goes out for an order that was not saved.

The example is a shop. A checkout subscription places each order, and an email subscription sends its confirmation at once and a request for a review three days later.

The table​

src/main.rs
const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY, email TEXT 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 email_jobs (
    id INTEGER PRIMARY KEY,
    priority INTEGER NOT NULL,
    retry_after TEXT NOT NULL,
    attempt INTEGER NOT NULL DEFAULT 1,
    processed_at TEXT,
    locked_until TEXT,
    payload BLOB NOT NULL
);
";
src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Sqlite, SqlitePool};

/// One row of `email_jobs`.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
struct SendEmail {
    #[field(id, generated)]
    id: i64,
    #[field(priority)]
    priority: i64,
    #[field(retry_after)]
    retry_after: DateTime<Utc>,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// A smaller priority is claimed first.
const URGENT: i64 = 0;
const ROUTINE: i64 = 10;

impl SendEmail {
    /// A job for `email`, due at `due`.
    fn new(email: &Email, priority: i64, due: DateTime<Utc>) -> serde_json::Result<Self> {
        Ok(Self {
            id: 0,
            priority,
            retry_after: due,
            attempt: 1,
            processed_at: None,
            locked_until: None,
            payload: serde_json::to_vec(email)?,
        })
    }
}

Each column turns on one behaviour of the queue (Roles). retry_after holds the time a row becomes due: a claim takes only rows whose time has come. priority orders the claims: a smaller value goes first, then the earlier retry_after, then the smaller id. processed_at keeps a finished row: an acknowledgement writes the time into it instead of deleting the row. generated marks the columns the database fills, so the insert leaves them out. In the manual form, the service writes that insert itself, as an Insert impl.

A job in the order's transaction​

src/main.rs
/// The order, its confirmation and a reminder three days later commit together, or not at all.
async fn place_order(pool: &SqlitePool, order: &Order) -> Result<(), Box<dyn Error>> {
    let mut tx = pool.begin().await?;
    sqlx::query("INSERT INTO orders (id, email) VALUES (?, ?)")
        .bind(order.id)
        .bind(&order.email)
        .execute(&mut *tx)
        .await?;
    let now = Utc::now();
    let confirmation = Email::confirmation(order);
    SendEmail::new(&confirmation, URGENT, now)?
        .insert(&mut *tx)
        .await?;
    let reminder = Email::review_request(order);
    SendEmail::new(&reminder, ROUTINE, now + TimeDelta::days(3))?
        .insert(&mut *tx)
        .await?;
    tx.commit().await?;
    Ok(())
}

place_order opens one sqlx transaction and writes three rows in it: the order, the confirmation and the reminder. The derive's insert takes any connection of the table's database, so &mut *tx puts the jobs in the order's transaction. They commit together, or roll back together: an order whose insert fails leaves no email in the table. An HTTP endpoint that places orders calls the same function with a pool of the service's own; a service that shares one pool between its endpoints and its queues hands it to SqlxBroker::new.

The confirmation is due now and urgent. The reminder is due in three days, and nothing claims it until then. A job that recurs is the same move: its handler inserts the next row with a later retry_after. A job that must exist once is the service's unique index and an INSERT .. ON CONFLICT DO NOTHING in its own statement.

The subscription​

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

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(order = email.order, template = %email.template, "sending");
    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("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);
            b.include(send);
        })
}

checkout takes the pool from its context and places the order; an order that fails is dropped with a warning. Its own transaction is separate from the claim of its checkout row: a handler whose writes must commit with its row's settlement runs in transactional mode. send reads every row of email_jobs in claim order. Its acknowledgement marks the row with processed_at, so the sent emails stay in the table for an audit, and a claim passes over them. A table without that column loses the row on acknowledgement.

When a job is seen​

A publish through the broker wakes the subscriptions of its table in the same process at once: a route, a Repository, a reply or an Out slot. The checkout published by the front end wakes checkout that way. A row the service writes with its own SQL, as place_order does, wakes nobody. The subscription finds it after its poll interval, one second unless set. The broker's publish does not join the service's transaction, so a job that must commit with the order goes in through insert and waits that interval. On Postgres, a notification from the service or a trigger ends the wait early.

Testing​

The tests in the example file publish checkouts through the harness and check the table and the handlers. An email row reaches its subscription on the next poll, so the tests that wait for an email start the app with TestApp::start_live and let two seconds pass with tb.advance. One test checks that the confirmation went out and the reminder still waits. One places the same order twice and finds only the first order's emails. One writes a routine and an urgent email before the service starts and checks that the urgent one was sent first. Testing a service on SQL covers the harness.