SqliteTransport — durable, one host

A durable queue over stdlib sqlite3 (WAL). The whole point: claim runs inside a BEGIN IMMEDIATE transaction, so SQLite’s global write-lock serializes claims across processes — the per-group exclusivity selection is then race-free with plain SQL, no row or advisory locks. The lease (lock_expiry) recovers a message a crashed worker was holding. The connection is opened with isolation_level=None (autocommit) so the code drives BEGIN IMMEDIATE/COMMIT by hand in claim; PRAGMA busy_timeout=5000 makes a contended writer wait for the lock instead of erroring.

Schema

Two tables:

CREATE TABLE IF NOT EXISTS messages (
  seq          INTEGER PRIMARY KEY AUTOINCREMENT,  -- monotonic id: FIFO order + the Lease handle
  group_id     TEXT NOT NULL,                      -- the execution id (the exclusivity group)
  event        TEXT NOT NULL,                      -- the Event JSON (event.model_dump_json())
  locked_by    TEXT,                               -- worker id while leased / "__parked__" / NULL when free
  lock_expiry  REAL                                -- epoch when the lease (or park) ends; NULL when free
)

CREATE TABLE IF NOT EXISTS groups (
  group_id       TEXT PRIMARY KEY,    -- one row per group that has messages
  last_claimed_at REAL NOT NULL DEFAULT 0.0,  -- epoch of last claim (0 = never claimed)
  priority        INT  NOT NULL DEFAULT 0     -- set on first publish; 0–4
)

(locked_by, lock_expiry) is the lease. The groups table drives round-robin fairness and priority filtering: claim sorts by last_claimed_at ASC (oldest-claimed first) and filters by priority >= min_priority.

The claim — atomic select-then-lease with round-robin and priority

BEGIN IMMEDIATE                                   -- take SQLite's write-lock: claims serialize
SELECT m.seq, m.group_id, m.event
  FROM messages m JOIN groups g ON g.group_id = m.group_id
  WHERE (m.locked_by IS NULL OR m.lock_expiry < ?)          -- message free / lease lapsed (recovery)
    AND m.group_id NOT IN (                                  -- group has nothing in flight
      SELECT group_id FROM messages WHERE locked_by IS NOT NULL AND lock_expiry >= ?)
    AND g.priority >= ?                                      -- priority floor (min_priority)
  ORDER BY g.last_claimed_at ASC, m.seq ASC LIMIT 1         -- oldest-claimed group first (round-robin)
-- if a row matched:
UPDATE groups SET last_claimed_at = ? WHERE group_id = ?    -- record claim time (round-robin)
UPDATE messages SET locked_by = ?, lock_expiry = now+visibility WHERE seq = ?
COMMIT                                              -- (ROLLBACK on any error)

Sorting by g.last_claimed_at ASC makes groups that have never been claimed (0) always go first; last_claimed_at is set to now on claim so a just-serviced group yields to others. AND g.priority >= ? skips groups below the priority floor (pass min_priority=0 for normal operation). Returns Lease(seq, group_id, event) or None when nothing is deliverable.

Operations

publish(group_id, event, priority=0)
    # INSERT INTO messages (group_id, event) VALUES (?, ?)
    # INSERT OR IGNORE INTO groups (group_id, priority) VALUES (?, ?)  -- first publish sets priority
claim(worker_id, visibility, min_priority=0)  # the BEGIN IMMEDIATE select-then-lease above
ack(lease)
    # DELETE FROM messages WHERE seq = ?
    # DELETE FROM groups WHERE group_id = ? AND NOT EXISTS (SELECT 1 FROM messages WHERE group_id = ?)
nack(lease, delay=0)          # delay>0  -> UPDATE locked_by="__parked__", lock_expiry=now+delay  (park)
                              # delay==0 -> UPDATE locked_by=NULL, lock_expiry=NULL              (retry now)
close()                       # close the connection

ack removes the message and drops the group row when no messages remain. nack(delay>0) parks the message (_PARKED sentinel keeps its group blocked) until the delay elapses — the portable queue-park the control plane uses for a suspended group. nack(0) frees it for immediate retry (e.g. after a StoreConflict).

FIFO

Delivery within a group is ORDER BY seq (oldest first), and a group is advanced by one consumer at a time, so per-group order is preserved.

Async twin

AsyncSqliteTransport (aio_transport/sqlite.py) mirrors this over aiosqlite — every cursor op awaited, the same hand-driven BEGIN IMMEDIATE/COMMIT. aiosqlite serializes a connection’s operations on its own worker thread, so the multi-statement claim stays atomic.

When to pick it

The global write-lock means claims serialize — perfect for one host or a shared volume, but it won’t let workers claim different groups in parallel. For that, use Postgres (SKIP LOCKED) or Redis. See the transports hub and distribution.