Keep a queue claim selection and update in one transaction

By Batu · English technical notes

Also published in our Blogger archive.

Claim selection and mutation in one transaction

BEGIN IMMEDIATE starts the local write transaction before the oldest queued id is selected. The UPDATE repeats state = queued while setting that id to claimed, and rowcount must be one before commit. The repeated predicate documents the transition and resists overwriting a row that is no longer queued.

This fixture always has a candidate. A worker must handle fetchone returning None for an empty queue, and separately define lease expiry and crash recovery.

Example

import sqlite3
from pathlib import Path
from tempfile import TemporaryDirectory

with TemporaryDirectory() as directory:
    connection = sqlite3.connect(Path(directory) / "queue.db")
    connection.execute("CREATE TABLE job(id INTEGER PRIMARY KEY, state TEXT, created INTEGER)")
    connection.execute("INSERT INTO job VALUES (1, 'queued', 10)")
    connection.commit()
    connection.execute("BEGIN IMMEDIATE")
    job_id = connection.execute("SELECT id FROM job WHERE state = 'queued' ORDER BY created LIMIT 1").fetchone()[0]
    changed = connection.execute("UPDATE job SET state = 'claimed' WHERE id = ? AND state = 'queued'", (job_id,)).rowcount
    connection.commit()
    assert changed == 1
    assert connection.execute("SELECT state FROM job WHERE id = 1").fetchone()[0] == "claimed"
    print("claim=id-1 state=claimed")

Expected stdout:

claim=id-1 state=claimed

Sources

- SQLite Transaction Control

- SQLite UPDATE

Prepared with AI assistance. The example uses synthetic data; its stated limits apply.