Build: claim_next_run + Transactional Recording

Maps to: Phase 2 (control-store). Kind: Build.

Objective

Add the two methods this crate exists for: claim_next_run, the atomic claim that lets any number of schedulers pull work without ever grabbing the same run twice, and record_job_outcome, the transactional-and-idempotent record that advances a run's progress exactly once per job even when a result is delivered twice. Three tests pin the behavior — the claim drains the queue and then answers None, a full run's outcomes advance done_count and flip the run to done on the last job, and the same job recorded twice counts once. When these are green, the store is safe under concurrency and under redelivery, which is everything the scheduler and cluster arcs will lean on.

The atomic claim and the two-statement transaction are the concept toy, translated: tasks becomes runs, claim_next becomes claim_next_run, the committed transfer becomes the outcome record. The one genuinely new piece is the idempotency guard on the record — a single AND status != 'done' clause, plus a check on how many rows it touched — and it is worth understanding before you type it.

Scaffold

Add to crates/control-store/src/lib.rs (no new files, no new deps): two methods on Store — claim_next_run and record_job_outcome — and three tests. You will also lean on insert_jobs from the last chapter, which now gets its first real exercise.

Expected result: cargo test -p control-store → 4 tests pass (the previous insert_then_get_run plus the three below):

• claim_moves_run_to_running_then_none
• record_outcome_advances_done_count_and_finishes_run
• recording_the_same_job_twice_counts_once

The spec (givens)

claim_next_run — the atomic claim

Signature: async fn claim_next_run(&self) -> Result<Option<Run>, ControlError>. One statement, run with fetch_optional, mapped through the same row_to_run you already wrote:

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

The subquery picks the oldest queued run (the idx_runs_status index makes this a seek); the UPDATE flips it to running; RETURNING * yields the post-update row so the caller gets the Run it claimed. No ? placeholders here — there is nothing to bind. fetch_optional gives None when the queue is empty (the UPDATE matched no row), which is how "nothing to claim" comes back through the same call as a successful claim. This is the concept toy's claim_next with tasks → runs.

[→ Answer key](../appendix-answer-key.md#store-claim)

record_job_outcome — transactional and idempotent

Signature: async fn record_job_outcome(&self, run_id: RunId, outcome: &JobOutcome) -> Result<(), ControlError>. One transaction, two statements — but the second runs only conditionally. The full logic:

-- statement 1: mark the job done, but only if it is not already done
UPDATE jobs SET status = 'done', outcome = ?
WHERE id = ? AND status != 'done';

-- statement 2: run this ONLY when statement 1 changed exactly one row
UPDATE runs SET done_count = done_count + 1,
    status = CASE WHEN done_count + 1 >= job_count THEN 'done' ELSE status END
WHERE id = ?;

Wrap both in pool.begin() … tx.commit(), executing each against &mut *tx. Between the two, branch on res.rows_affected() from statement 1: run statement 2 only if rows_affected() == 1. Bind the JSON-encoded outcome and outcome.job_id into statement 1, and run_id into statement 2.

Three clauses carry the whole method, and each is deliberate:

• AND status != 'done' — the guard. A job already marked done is not matched by statement 1, so the UPDATE changes zero rows. A first-time job is matched, so it changes one. This one clause is the entire difference between "recorded once" and "recorded twice."
• if res.rows_affected() == 1 — the gate. rows_affected() reports how many rows statement 1 actually changed: 1 the first time a job lands, 0 on any redelivery of an already-done job. The run's done_count advances only when a job newly transitions to done. A redelivered outcome rewrites the (identical) outcome JSON on... nothing — it matched no row — and leaves done_count exactly where it was.
• CASE WHEN done_count + 1 >= job_count THEN 'done' ELSE status END — the finish. In the same statement that bumps the counter, check whether this landing is the last one; if the incremented count reaches job_count, flip the run to done. (done_count + 1 because the arithmetic in SET reads the old value; the whole SET sees the pre-update row.) No separate "is the run finished?" query, no second round trip — the completion check rides along with the increment, inside the transaction.

[→ Answer key](../appendix-answer-key.md#store-record)

Why idempotency, in one paragraph (full story: Part VIII)

The guard exists because the coordinator will deliver jobs at-least-once: if a worker finishes a job but dies before its Result is acknowledged, the scheduler cannot tell "finished but unacked" from "never finished," so it re-dispatches — and the job genuinely runs twice, or its outcome arrives twice. At-least-once is the only delivery guarantee that loses nothing under worker failure, but it makes duplicate recording a normal event rather than a bug. Idempotent recording is the escape: written this way, recording the same job's outcome twice is indistinguishable from recording it once, so at-least-once delivery becomes safe. The full argument — heartbeats, reaping a dead worker's lease, why exactly-once is a fiction — is the capstone of Part VIII; here you build the store-side half that makes it all sound, and prove it with recording_the_same_job_twice_counts_once.

Concepts exercised

• The atomic claim (UPDATE … RETURNING) as concurrency-safe work pulling — the concept toy, on runs.
• A multi-statement transaction whose second statement is conditional on the first's rows_affected().
• Idempotency by a WHERE … AND status != 'done' guard plus a rows_affected() == 1 gate.
• A CASE expression folding a completion check into the same UPDATE that advances the counter.
• Reading rows_affected() to distinguish "newly changed" from "already in that state."

The build loop (you drive)

Write each test first, predict, run red, implement to green.

Test 1 — claim_moves_run_to_running_then_none (#[tokio::test])

1. Write it. Store::in_memory(), insert one a_run(0) (queued). Call claim_next_run().await.unwrap().unwrap(); assert the claimed run's id matches and its status == RunStatus::Running. Then call claim_next_run() again and assert it is_none().
2. Predict: the second claim runs the identical statement against the identical table — nothing was inserted or deleted between the calls. Why does it return None rather than re-claiming the same run? Point at the exact word in the SQL that changed the answer.
3. Run red (no claim_next_run yet), implement the one statement, run green, commit.

Test 2 — record_outcome_advances_done_count_and_finishes_run (#[tokio::test])

1. Write it. Insert a_run(2) (two jobs), insert_jobs two jobs j1/j2 for it, and claim_next_run() to move the run to running (a run reports outcomes while it is running). Record j1's outcome; get_run and assert done_count == 1, status == Running. Record j2's outcome; get_run and assert done_count == 2, status == Done.
2. Predict: which record call flips the run to Done, and what makes the CASE fire on that call and not the first? (Compute done_count + 1 versus job_count for each.)
3. Run red, implement the transaction + conditional second statement, run green, commit.

Test 3 — recording_the_same_job_twice_counts_once (#[tokio::test])

1. Write it. Insert a_run(2), insert_jobs one job j1, claim_next_run(). Record j1's outcome, then record the same outcome again — same job_id. get_run and assert done_count == 1 (not 2) and status == Running.
2. Predict: on the second call, rows_affected() from statement 1 is what number — and therefore does statement 2 run at all? This is the at-least-once safety property in one assertion; name what would break without the guard.
3. Run red — and note that with a naive record (no guard) this test is the one that fails while tests 1 and 2 still pass, which is exactly why it exists. Implement the guard + gate, run green, commit.

Done when

cargo test -p control-store shows all four tests green; claim_next_run drains the queue and then answers None; record_job_outcome advances done_count once per job and finishes the run on the last one; and recording_the_same_job_twice_counts_once proves a redelivered outcome does not double-count. You can state, in one sentence, why the rows_affected() == 1 gate and the AND status != 'done' guard are a pair — and you know that the reason the coordinator needs this at all is the at-least-once delivery that Part VIII builds on top of it. Commit; the persistence arc is done.