Skip to main content

Long jobs, crashes and the claim form

This guide chooses how a queue table holds a job while a handler works on it, and shows what happens to a job that runs for minutes or whose process dies.

A claimed job must stay out of other claims until it settles, and come back if its process stops. The table's description chooses one of three forms for that, and the handler is the same in all of them.

The three forms​

Row lockLeaseAdvisory lock
Selected bynothing: the default#[field(locked_until)]#[inbox(advisory_lock = "..")]
Holds a job bythe claim's transaction, FOR UPDATE SKIP LOCKEDan expiry written into the rowa lock on the row's key, held by a session
Connectionsone per job in work, for the whole handlerone per statementone per job in work
After a crashthe job returns at oncethe job returns when its lease runs outthe job returns at once
A crash spends an attemptnoyesyes
DatabasesPostgres, MySQL, MariaDBallall

The row lock keeps the claim's transaction open while the handler runs, and the settlement commits it. A crashed process's transaction rolls back, so its jobs return at once and their attempts are not counted. The price is a connection per job in work. SQLite has no row locks, so a SQLite table names one of the other forms.

The lease commits the claim at once, so a job in work holds no connection and no transaction. It suits long jobs, and it is the form of a SQLite queue. The advisory lock holds each job by a key: a key on the id holds each job alone, and a key on another column keeps the jobs that share it in work one at a time.

A report that renders for minutes​

src/main.rs
const SCHEMA: &str = "CREATE TABLE IF NOT EXISTS report_jobs (
    id INTEGER PRIMARY KEY,
    attempt INTEGER NOT NULL DEFAULT 1,
    locked_until TEXT,
    payload BLOB NOT NULL
)";
src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `report_jobs`: `locked_until` claims it under a lease.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "report_jobs")]
struct RenderReport {
    #[field(id, generated)]
    id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// How long the row of a crashed process stays out of the queue.
const LEASE: Duration = Duration::from_mins(1);

#[subscriber(InboxQueue::<RenderReport>::new("reports").lease(LEASE))]
async fn render(report: &Report, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    if attempt > Some(1) {
        tracing::warn!(report.id, ?attempt, "rendering again after a lost attempt");
    }
    draw(report).await;
    tracing::info!(report.id, "report rendered");
    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("reports", "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::<RenderReport>("reports"), |b| {
            b.include(render);
        })
}

A claim writes the lease's expiry into locked_until and counts the attempt. While render runs, the subscription extends the lease every half lease, so a report may render for five minutes on a lease of one. The expiry is the delivery's ownership token: the acknowledgement deletes the row only while it still holds that lease. InboxQueue::lease sets the lease of one subscription, and SqlxBroker::lease the broker's; it is thirty seconds unless set, in whole seconds.

What a crash does​

A process that dies mid-report stops extending its lease. The row stays out of the queue until the lease runs out, a minute here, then the next claim takes it. That claim counts one more attempt, so render sees a second attempt and logs it. A lease is therefore how long a crash delays a job: a short lease brings it back sooner, at the price of more extensions. A process that stops the extensions while alive, cut off from the database for longer than its lease, loses the job to another claim, and its late settlement fails with SqlxBrokerError::LeaseLost.

A row lock or an advisory lock ends with its session, so a crash returns the jobs at once. At shutdown the broker stops extending leases: a job in work keeps its lease and settles if it can, or returns when the lease runs out. Advisory locks are released at shutdown (Shutdown).

The pool each form needs​

A row lock or an advisory lock holds a connection per job in work, so a subscription with workers(8) needs eight connections for its jobs alone. A lease holds one only for each statement. Every subscription leaves the pool one connection for the handlers, their publishes and the settlements (Connections). On Postgres, the advisory lock needs a direct connection or PgBouncer in session pooling, because the lock lives in the session. On SQLite, the process keeps the advisory keys in work, so one process serves a database file. Running in production sizes the pool.

Testing​

The tests in the example file run the service on SQLite in memory. The first pauses the clock, so five minutes of rendering pass at once and the lease's extensions run as the clock reaches them; the report is delivered once. The second writes two rows another process had claimed: one whose lease ran out, one whose lease is still in force. Only the first comes back. Those rows wake no subscription, so this test runs on the real clock with TestApp::start_live past the poll interval.