Concept: The Atomic Claim and Transactions

Kind: Concept. Builds on the store you just wrote — no new crate, one new idea (two, really, and they are cousins).

The last build gave a run a durable home. It did not give it a safe one. Every method so far — insert_run, get_run, insert_jobs — assumed exactly one actor touching the database at a time, and for a single-process coordinator that assumption holds right up until the moment it doesn't: the moment there are two schedulers, or one scheduler that restarted while the old one was still draining, or two worker results landing at the same instant. This chapter is about the two SQL tools that make those moments safe. Both are one idea wearing two hats — this happens as one indivisible step, or it does not happen at all.

We build them on a toy tasks queue so nothing here is the answer key, then map it one-for-one onto the real store in the next chapter.

Why this matters the instant there is more than one worker

Picture the store as it stands, and two schedulers both looking for work. Each wants "the oldest queued run." With the tools you have, the honest way to grab one is two steps:

scheduler A                         scheduler B
-----------                         -----------
SELECT id FROM runs
  WHERE status='queued'
  ORDER BY created_at LIMIT 1   ->  run r7
                                    SELECT id FROM runs
                                      WHERE status='queued'
                                      ORDER BY created_at LIMIT 1  -> run r7
UPDATE runs SET status='running'
  WHERE id = r7
                                    UPDATE runs SET status='running'
                                      WHERE id = r7

Both read r7 as queued — the read happened before either write — so both claim r7 and run it. Every vignette in that run gets evaluated twice, tokens spent twice, and the run's counters are now a lie. Nothing here is a bug in either scheduler; each did the obvious correct thing. The bug is the gap between the read and the write, a gap another actor slipped into. That gap is called a race, and it is the defining hazard of the whole distributed half of this course.

The atomic claim: UPDATE … RETURNING

The race exists because "find the oldest queued run" and "mark it running" were two statements. SQL lets you fuse them. A single UPDATE can do the selecting and the writing, and RETURNING hands you back the row it just changed:

UPDATE tasks SET status = 'running'
WHERE id = (SELECT id FROM tasks WHERE status = 'queued' ORDER BY created_at LIMIT 1)
RETURNING *;

Read it as one motion: among the queued tasks, take the oldest, flip it to running, and give it back to me. The subquery picks the target; the UPDATE writes it; RETURNING * yields the post-update row so the caller learns which task it got — all inside one statement the database runs to completion before any other connection sees the table. There is no gap for a second scheduler to read the same row as still queued, because by the time anyone else looks, it is already running.

Two properties fall out, and both are the point:

• No double-claim. Because the flip is part of the same statement as the selection, a concurrent claim either runs before this one (and picks a different oldest row) or after (and sees this row as no longer queued). It can never land in the middle. Two schedulers hammering claim_next divide the queue between them; they never split a row.
• The empty queue answers honestly. When nothing is queued, the subquery returns no id, the UPDATE matches no row, and RETURNING yields nothing — which sqlx surfaces as fetch_optional returning None. "Nothing to claim" and "here is your claim" come back through the same call, distinguished by Some/None.

Transactions: all-or-nothing across several statements

The claim fuses a read and a write. The other hazard is the opposite shape: several writes that must all land together. When a job finishes, the store does two things — mark the job done, and bump its run's done_count. Do them as two independent statements and a crash (or a rolled-back connection, or a failed second statement) between them leaves the database in a state that can never be true: a job marked done that the run's counter never counted, or a counter that ran ahead of a job still marked pending. The run's progress is now wrong forever.

A transaction is the fix. You open one with pool.begin(), run as many statements as you like against the returned handle, and finish with commit(). Until you commit, none of it is visible to anyone else; and if you never commit — you call rollback(), or the connection drops, or the process dies — every statement in the transaction is undone as a unit. All of them, or none of them. There is no half.

let mut tx = pool.begin().await?;
sqlx::query("UPDATE tasks SET status = 'done' WHERE id = ?")
    .bind("t1")
    .execute(&mut *tx)            // note: &mut *tx, not &pool
    .await?;
sqlx::query("UPDATE counters SET value = value + 1 WHERE name = 'done'")
    .execute(&mut *tx)
    .await?;
tx.commit().await?;              // both land now, atomically — or neither did

The one syntactic thing to notice: inside a transaction you execute against &mut *tx, not &pool. That is how sqlx routes both statements down the same connection and holds them open as one unit; hand either statement &pool and it would run on its own connection, outside the transaction, defeating the whole point. (You saw this already in insert_jobs last chapter — the batch insert was a transaction so a run's jobs land all-or-nothing. Same tool, now load-bearing.)

The whole toy, working

Here is tasks end to end: two queued rows, the atomic claim called three times, a committed two-statement transaction, and the same transaction rolled back to show all-or-nothing from the other side. It is the exact shape the real store will have — claim_next is claim_next_run, the commit is the happy path of record_job_outcome, the rollback is what a crash mid-record does for you.

use std::str::FromStr;
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions, SqliteRow};
use sqlx::{Row, SqlitePool};

#[derive(Debug)]
struct Task { id: String, status: String, created_at: String }

fn row_to_task(row: SqliteRow) -> anyhow::Result<Task> {
    Ok(Task {
        id: row.try_get("id")?,
        status: row.try_get("status")?,
        created_at: row.try_get("created_at")?,
    })
}

async fn connect() -> anyhow::Result<SqlitePool> {
    let opts = SqliteConnectOptions::from_str("sqlite::memory:")?.create_if_missing(true);
    let pool = SqlitePoolOptions::new().max_connections(1).connect_with(opts).await?;
    sqlx::migrate!().run(&pool).await?;
    Ok(pool)
}

// The atomic claim: select the oldest queued row and flip it to running in one
// statement. RETURNING hands back the row it just changed — or nothing.
async fn claim_next(pool: &SqlitePool) -> anyhow::Result<Option<Task>> {
    let row = sqlx::query(
        "UPDATE tasks SET status = 'running' \
         WHERE id = (SELECT id FROM tasks WHERE status = 'queued' ORDER BY created_at LIMIT 1) \
         RETURNING *",
    )
    .fetch_optional(pool)
    .await?;
    Ok(row.map(row_to_task).transpose()?)
}

async fn status(pool: &SqlitePool, id: &str) -> anyhow::Result<String> {
    let row = sqlx::query("SELECT status FROM tasks WHERE id = ?").bind(id)
        .fetch_one(pool).await?;
    Ok(row.try_get("status")?)
}
async fn counter(pool: &SqlitePool) -> anyhow::Result<i64> {
    let row = sqlx::query("SELECT value FROM counters WHERE name = 'done'")
        .fetch_one(pool).await?;
    Ok(row.try_get("value")?)
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let pool = connect().await?;
    // Seed: two queued tasks (t1 older than t2) and a counter at zero.
    for (id, at) in [("t1", "2026-07-21T10:00:00Z"), ("t2", "2026-07-21T10:01:00Z")] {
        sqlx::query("INSERT INTO tasks (id, status, created_at) VALUES (?, 'queued', ?)")
            .bind(id).bind(at).execute(&pool).await?;
    }
    sqlx::query("INSERT INTO counters (name, value) VALUES ('done', 0)")
        .execute(&pool).await?;

    // --- The atomic claim, three times over a two-item queue ---
    println!("claim 1 -> {:?}", claim_next(&pool).await?.map(|t| t.id));
    println!("claim 2 -> {:?}", claim_next(&pool).await?.map(|t| t.id));
    println!("claim 3 -> {:?}", claim_next(&pool).await?.map(|t| t.id));

    // --- Two statements, committed together ---
    let mut tx = pool.begin().await?;
    sqlx::query("UPDATE tasks SET status = 'done' WHERE id = ?").bind("t1")
        .execute(&mut *tx).await?;
    sqlx::query("UPDATE counters SET value = value + 1 WHERE name = 'done'")
        .execute(&mut *tx).await?;
    tx.commit().await?;
    println!("after commit:   t1={}, counter={}", status(&pool, "t1").await?, counter(&pool).await?);

    // --- The same two statements, rolled back (what a crash mid-record does) ---
    let mut tx = pool.begin().await?;
    sqlx::query("UPDATE tasks SET status = 'done' WHERE id = ?").bind("t2")
        .execute(&mut *tx).await?;
    sqlx::query("UPDATE counters SET value = value + 1 WHERE name = 'done'")
        .execute(&mut *tx).await?;
    tx.rollback().await?;
    println!("after rollback: t2={}, counter={}", status(&pool, "t2").await?, counter(&pool).await?);
    Ok(())
}

The migration is two small tables:

CREATE TABLE tasks (
    id         TEXT PRIMARY KEY,
    status     TEXT NOT NULL,   -- 'queued' | 'running' | 'done'
    created_at TEXT NOT NULL
);
CREATE TABLE counters (
    name  TEXT PRIMARY KEY,
    value INTEGER NOT NULL
);

Run it, and the two ideas print their signatures:

claim 1 -> Some("t1")
claim 2 -> Some("t2")
claim 3 -> None
after commit:   t1=done, counter=1
after rollback: t2=running, counter=1

Read every line against the concept:

• claim 1 -> Some("t1"), claim 2 -> Some("t2"), claim 3 -> None. Each claim flips its target to running inside the one statement, so the next claim never sees it as queued. The queue drains, oldest first; the empty queue answers None. Two schedulers alternating these calls would split the two tasks between them and never collide — that is the atomic claim earning its name.
• after commit: t1=done, counter=1. Both statements landed together. The job is done and the count moved — the pair that must always agree, agreeing.
• after rollback: t2=running, counter=1. This is all-or-nothing from the failure side. The transaction set t2 to done and bumped the counter, then rolled back — and the database undid both. t2 is back to running (its pre-transaction state, since it had been claimed) and the counter is untouched at 1. A crash between the two writes would have done exactly this for you: no orphaned half-write, no counter running ahead of reality.

This toy uses sqlx, so there is no playground button. To run it: cargo new claim-scratch, add sqlx = { version = "0.8", features = ["runtime-tokio", "sqlite", "macros", "migrate"] }, tokio = { version = "1", features = ["full"] }, and anyhow; put the two CREATE TABLEs in migrations/0001_init.sql; paste the code into main.rs. Same crate set and same runtime-API style as the store crate.

The seam to the build: one-for-one

The next chapter builds the two real store methods, and the toy maps straight across:

• toy tasks queue ↔ the runs table with its status/created_at columns (and the idx_runs_status index you already built for exactly this claim)
• claim_next (the UPDATE … RETURNING) ↔ claim_next_run — same statement, tasks becomes runs, Task becomes Run
• the committed two-statement transaction (mark task done + bump counter) ↔ record_job_outcome — mark the job done, then bump the run's done_count, in one begin/commit
• the rollback demo ↔ what a crash mid-record does: the job stays pending and the counter stays put, so redelivery can safely try again

The domain changes from anonymous tasks to eval runs and jobs; the two moves — fuse the read and write into one statement, fuse the several writes into one transaction — do not. The real record_job_outcome adds one twist the toy omits: a guard that makes recording the same job twice count once, so at-least-once redelivery is safe. That guard is the next build's centerpiece, and why it has to exist is the payoff waiting in Part VIII.

Questions to lock

Pause on each. A fuzzy answer is the signal to re-read the matching section.

1. Two schedulers each run SELECT oldest queued then UPDATE it to running as separate statements. Walk the interleaving that makes them both claim the same run. Then explain precisely how folding both into one UPDATE … RETURNING closes the gap.
2. Inside a transaction you .execute(&mut *tx) instead of .execute(&pool). What goes wrong if you accidentally pass &pool for one of the two statements in record_job_outcome?
3. record_job_outcome marks a job done and bumps its run's done_count in one transaction. If the process crashes between those two writes, what state is the database left in — and why is that the state that makes redelivery safe rather than corrupt?

Next chapter is the build: claim_next_run and record_job_outcome, the transactional-and-idempotent record that is the reason this whole crate exists.