LibsqlTransport — libSQL / Turso (experimental)

A clone of SqliteTransport over the libsql driver (Turso’s SQLite fork): the same messages + groups tables, the same BEGIN IMMEDIATE claim, the same lease, round-robin fairness and priority filtering. What changes is where the queue lives — a local file, a sqld server, or an embedded Turso replica.

Experimental: the local-file path is covered in-process by the test suite; the Turso/sqld embedded-replica path is wired but not yet validated against a real account.

Connection modes

The constructor opens libsql.connect(database, isolation_level=None, _check_same_thread=False, …). With sync_url= (+ auth_token=, optional sync_interval=) it is an embedded replica against a Turso/sqld primary; without it, a plain local file. isolation_level=None is autocommit, so claim drives BEGIN IMMEDIATE/COMMIT by hand (exactly as SqliteTransport).

Schema

Two tables (identical to SqliteTransport):

CREATE TABLE IF NOT EXISTS messages (
  seq          INTEGER PRIMARY KEY AUTOINCREMENT,  -- FIFO order + the Lease handle
  group_id     TEXT NOT NULL,                      -- the execution id (exclusivity group)
  event        TEXT NOT NULL,                      -- the Event JSON
  locked_by    TEXT,                               -- worker id / "__parked__" / NULL
  lock_expiry  REAL                                -- lease/park deadline; 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) — round-robin
  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

Identical to SqliteTransport: under BEGIN IMMEDIATE (libSQL serializes writers, so the per-group selection is race-free) select the oldest-claimed deliverable group and lease it:

BEGIN IMMEDIATE
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 error

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

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 -> locked_by="__parked__", lock_expiry=now+delay (park)
                              # delay==0 -> locked_by=NULL, lock_expiry=NULL (retry now)
close()                       # close the connection

Async twin

AsyncLibsqlTransport is not native async: the libsql package is synchronous, so it wraps the sync LibsqlTransport and off-loads each call with asyncio.to_thread, serialized by an asyncio.Lock (one connection, one op at a time — the BEGIN IMMEDIATE claim is single-writer anyway). Build with await AsyncLibsqlTransport.create(database, sync_url=…, auth_token=…).

When to pick it

Like SqliteTransport (claims serialize on the write-lock), but libSQL-hosted — a path to a managed distributed SQLite (Turso) later. Experimental today. See the transports hub and distribution.