LibsqlStore — libSQL / Turso (experimental)

LibsqlStore is a durable ExecutionStore backend over libSQL — Turso’s open-source fork of SQLite — reached through the libsql Python package. The libsql driver is DB-API compatible (a sqlite3-style driver), so for this backend the SQL, the version-CAS write and the one-transaction commit are byte-for-byte identical to SqliteStore. The same six tables, the same CAS, the same trace ring. What LibsqlStore adds over plain SQLite is purely where the database lives: it can be a local file (an embed, like SQLite) or an embedded replica of a remote Turso/sqld primary — selected entirely by constructor arguments. One backend is therefore both a single-file embed and a distributed store.

Experimental status

This backend is EXPERIMENTAL, and the boundary is precise:

  • Local-file path — tested. Driving LibsqlStore against a local file (or :memory:) is covered in-process by the test suite. This is the SQLite-equivalent path and behaves like SqliteStore.

  • Turso / sqld embedded-replica path — wired but unvalidated. Passing sync_url= (and auth_token=) builds an embedded replica against a Turso/sqld primary. The code path is in place but has not been validated against a real Turso account.

  • Replication is eventually consistent. A Turso primary-follower deployment replicates asynchronously: a follower (the embedded replica) may lag the primary. The version-CAS reads the local replica, so a CAS decided against a stale follower can be wrong. In practice that means: read from the primary for the CAS, or expect extra StoreConflict retries while the follower catches up.

The docstring states the same constraints verbatim:

EXPERIMENTAL: the local-file path is covered in-process by the test suite; the Turso/
`sqld` embedded-replica path (sync_url) is wired but not yet validated against a real
Turso account, and its primary-follower replication is eventually consistent (read from the
primary for CAS, or expect extra StoreConflict retries).

Connection modes

The constructor adapts to one of two modes by argument. There is no separate class for “local” vs “replica” — the kwargs decide:

def __init__(
    self,
    database: Union[str, Path] = ":memory:",
    *,
    auth_token: str = "",
    sync_url: Optional[str] = None,
    sync_interval: Optional[float] = None,
) -> None:
    import libsql

    kwargs: dict[str, Any] = {"_check_same_thread": False}
    if sync_url is not None:  # embedded replica against a Turso/sqld primary
        kwargs["sync_url"] = sync_url
        kwargs["auth_token"] = auth_token
        if sync_interval is not None:
            kwargs["sync_interval"] = sync_interval
    self._conn = libsql.connect(str(database), **kwargs)

_check_same_thread=False is always passed (so the one connection can be used from a worker thread — the async twin off-loads each call to a thread; see below). The two modes:

  1. Plain local file — LibsqlStore("state.db") (or LibsqlStore(":memory:"), the test variant). sync_url is None, so libsql.connect("state.db", _check_same_thread=False) opens a local SQLite-compatible database file. Reads and writes are all local. This is the SQLite-equivalent embed.

  2. Embedded replica — LibsqlStore("local.db", sync_url="libsql://…", auth_token="…"). With sync_url set, the kwargs grow sync_url + auth_token (and sync_interval if given), so the connection becomes libsql.connect("local.db", _check_same_thread=False, sync_url=…, auth_token=…). Operationally:

    • reads come from the local replica file (database) — fast, local, possibly stale;

    • writes are routed to the Turso/sqld primary and synced back into the local replica;

    • sync_interval (seconds) controls how often the replica pulls the primary’s latest state in the background.

    This is the path that makes the single backend a distributed store: the same SQL runs, but the authoritative copy lives on a remote primary and is replicated to every embedded replica.

After connecting, __init__ creates the six tables (CREATE TABLE IF NOT EXISTS …), sets self.trace_max = DEFAULT_TRACE_MAX (200), and self._conn.commit()s the DDL.

Schema — the six tables

The schema is created verbatim in __init__ (identical shape to SqliteStore):

CREATE TABLE IF NOT EXISTS harel_executions
  (id TEXT PRIMARY KEY, definition_id TEXT NOT NULL, data TEXT NOT NULL, version INTEGER NOT NULL)

CREATE TABLE IF NOT EXISTS harel_outbox
  (seq INTEGER PRIMARY KEY AUTOINCREMENT, target_id TEXT, event TEXT NOT NULL)

CREATE TABLE IF NOT EXISTS harel_processed_events
  (execution_id TEXT NOT NULL, event_id TEXT NOT NULL, PRIMARY KEY (execution_id, event_id))

CREATE TABLE IF NOT EXISTS harel_timers
  (execution_id TEXT NOT NULL, path TEXT NOT NULL, fire_at REAL NOT NULL,
   PRIMARY KEY (execution_id, path))

CREATE TABLE IF NOT EXISTS harel_spawns
  (seq INTEGER PRIMARY KEY AUTOINCREMENT, parent_id TEXT NOT NULL, child_id TEXT NOT NULL,
   root_path TEXT NOT NULL, context TEXT NOT NULL)

CREATE TABLE IF NOT EXISTS harel_trace
  (execution_id TEXT NOT NULL, idx INTEGER NOT NULL, entry TEXT NOT NULL,
   PRIMARY KEY (execution_id, idx))

What each holds:

  • harel_executions — one row per Execution. data is the full Execution serialized as JSON (exe.model_dump_json()); version is the optimistic-concurrency counter; definition_id lets list_executions filter without parsing the blob. PK is id.

  • harel_outbox — the transactional outbox of emitted events awaiting delivery. seq is a monotonic auto-increment used to order and ack entries; target_id is the Execution the event is delivered to (nullable = no target); event is the event JSON. The relay drains it after the commit, so a crash never loses a Finished.

  • harel_processed_events — the dedupe ledger: (execution_id, event_id) for every event already handled, PK on both. Makes at-least-once delivery effect-once.

  • harel_timers — durable timers, one per (execution_id, path) (PK), firing at fire_at (a REAL epoch second). Re-arming the same path replaces the row (upsert).

  • harel_spawns — pending orthogonal child-Execution creation intents, persisted in the same transaction as the parent’s advance + join expectations. seq orders/acks; parent_id, child_id, root_path and the JSON context describe the child the relay must create idempotently.

  • harel_trace — the opt-in execution timeline ring. (execution_id, idx) PK; entry is the JSON step (event/transition/actions/context_out). Off by default.

The CAS write + commit

_write — the version-CAS

_write persists the Execution without committing the transaction, so it can be batched atomically with the outbox/dedupe/timer/spawn/trace writes inside commit:

def _write(self, exe: Execution) -> None:
    old = exe.version
    exe.version = old + 1
    data = exe.model_dump_json()
    cur = self._conn.execute(
        "UPDATE harel_executions SET data = ?, version = ? WHERE id = ? AND version = ?",
        (data, exe.version, exe.id, old),
    )
    if cur.rowcount == 0:
        found = self._conn.execute("SELECT version FROM harel_executions WHERE id = ?", (exe.id,)).fetchone()
        if found is None and old == 0:
            self._conn.execute(
                "INSERT INTO harel_executions (id, definition_id, data, version) VALUES (?, ?, ?, ?)",
                (exe.id, exe.definition_id, data, exe.version),
            )
        else:
            exe.version = old
            raise StoreConflict(exe.id, expected=old, found=found[0] if found else None)

The mechanism: bump exe.version to old + 1, then UPDATE … WHERE id=? AND version=old. If a row matched, exactly one writer won and the version advanced. If rowcount == 0, no row was at old — disambiguate by existence:

  • brand-new Execution (found is None and old == 0): there is no row yet, so INSERT it at version 1.

  • stale write (a row exists but its version moved past old, or old != 0 with no row): another writer won. Roll back the in-memory exe.version = old (the commit did not happen) and raise StoreConflict.

This is the single-writer-per-execution backstop. On the Turso replica path the UPDATE is evaluated against the local follower, which is exactly why an eventually-consistent follower can produce spurious conflicts — see Experimental status.

save is the standalone version (one Execution, no side-writes):

def save(self, exe: Execution) -> None:
    old = exe.version
    try:
        self._write(exe)
        self._conn.commit()
    except BaseException:
        self._conn.rollback()  # any failure: release the write lock, keep nothing
        exe.version = old
        raise

commit — one atomic write per event

commit is the one atomic write per event boundary: it persists the Execution advance, the emitted events, the dedupe marker, the spawn intents, the timer mutations and the (optional) trace step all in a single SQLite transaction, then commits once. Either every piece lands or none does — which is the property that makes a crash safe: a fork’s children and the parent’s join expectations commit together; a Finished is durable before delivery.

def commit(
    self,
    exe: Execution,
    emits: list[tuple[Optional[str], Event]],
    processed_event_id: Optional[str] = None,
    timers: tuple[TimerOp, ...] = (),
    spawns: tuple[tuple[str, str, dict], ...] = (),
    trace: Optional[dict] = None,
) -> list[int]:
    old = exe.version
    try:
        self._write(exe)
        seqs = []
        for target_id, event in emits:
            cur = self._conn.execute(
                "INSERT INTO harel_outbox (target_id, event) VALUES (?, ?)",
                (target_id, event.model_dump_json()),
            )
            seqs.append(cur.lastrowid)
        if processed_event_id is not None:
            self._conn.execute(
                "INSERT OR IGNORE INTO harel_processed_events (execution_id, event_id) VALUES (?, ?)",
                (exe.id, processed_event_id),
            )
        for child_id, root_path, context in spawns:
            self._conn.execute(
                "INSERT INTO harel_spawns (parent_id, child_id, root_path, context) VALUES (?, ?, ?, ?)",
                (exe.id, child_id, root_path, json.dumps(context)),
            )
        for op in timers:
            if op.action == "schedule":
                self._conn.execute(
                    "INSERT INTO harel_timers (execution_id, path, fire_at) VALUES (?, ?, ?) "
                    "ON CONFLICT(execution_id, path) DO UPDATE SET fire_at = excluded.fire_at",
                    (exe.id, op.path, op.fire_at),
                )
            else:
                self._conn.execute(
                    "DELETE FROM harel_timers WHERE execution_id = ? AND path = ?", (exe.id, op.path)
                )
        if trace is not None:
            self._write_trace(exe.id, trace)
        self._conn.commit()
        return seqs
    except BaseException:
        self._conn.rollback()  # any failure discards the whole batch and its write lock
        exe.version = old
        raise

Statement by statement:

  1. self._write(exe) — the version-CAS UPDATE/INSERT above (uncommitted). If it raises StoreConflict, the whole transaction is rolled back and the error re-raised — nothing else in the batch is applied.

  2. Outbox — for each (target_id, event), INSERT INTO harel_outbox (target_id, event). The seq auto-increments, and commit returns the new seqs in emits order (so a caller that delivers an entry itself can ack it). These are the deferred events the relay delivers post-commit.

  3. Dedupe — if processed_event_id is given, INSERT OR IGNORE INTO harel_processed_events. The OR IGNORE makes recording the handled event idempotent against the PK (a re-delivery is a no-op).

  4. Spawns — for each (child_id, root_path, context), INSERT INTO harel_spawns, with context JSON-encoded. Persisted alongside the parent’s advance so the orthogonal fork is atomic.

  5. Timers — for each TimerOp: schedule does an upsert (ON CONFLICT(execution_id, path) DO UPDATE SET fire_at = excluded.fire_at, so re-arming the same path replaces its fire_at); cancel does DELETE FROM harel_timers WHERE execution_id=? AND path=?. Arming/cancelling happens in the same transaction as the transition that caused it — no dual-write, a scheduled timer cannot be lost.

  6. Trace — if a trace step is given, _write_trace appends it (still inside this txn; see below).

  7. self._conn.commit() — the single atomic commit. Up to here nothing is durable.

  8. except BaseException: rollback; restore version; raise — any failure rolls the whole transaction back, so the connection never keeps the file’s write lock (which would block every other writer) and a later commit can’t persist a half-written batch. The in-memory version goes back to old, so a retry isn’t a false conflict. On a StoreConflict the caller reloads and retries (or drops the stale work).

The trace ring

_write_trace appends one timeline step without committing (so it batches into commit’s transaction), in two statements: the index is computed inline (no pre-read), then the ring is trimmed.

def _write_trace(self, execution_id: str, entry: dict) -> None:
    self._conn.execute(
        "INSERT INTO harel_trace (execution_id, idx, entry) "
        "SELECT ?, COALESCE((SELECT MAX(idx) FROM harel_trace WHERE execution_id = ?), -1) + 1, ?",
        (execution_id, execution_id, json.dumps(entry)),
    )
    if self.trace_max:
        self._conn.execute(
            "DELETE FROM harel_trace WHERE execution_id = ? AND idx <= "
            "(SELECT MAX(idx) FROM harel_trace WHERE execution_id = ?) - ?",
            (execution_id, execution_id, self.trace_max),
        )

  • Inline idx — the INSERT … SELECT MAX(idx)+1 computes the next index in SQL (COALESCE(MAX(idx), -1) + 1, so the first step is 0). No round-trip to read the current max first; the idx is monotonic and survives the ring delete.

  • Ring trim — when trace_max is set, DELETE … WHERE idx <= MAX(idx) - trace_max keeps only the last trace_max steps per execution. The index is not reset, so read_trace always returns ascending, contiguous-from-the-tail indices.

append_trace is the standalone seam (demo/test): write one step and commit it on its own. read_trace returns the steps in order, stamping each with its stored idx as index (so the JSON entry need not carry it):

def append_trace(self, execution_id: str, entry: dict) -> None:
    self._write_trace(execution_id, entry)
    self._conn.commit()

def read_trace(self, execution_id: str) -> list[dict]:
    rows = self._conn.execute(
        "SELECT idx, entry FROM harel_trace WHERE execution_id = ? ORDER BY idx", (execution_id,)
    ).fetchall()
    return [{**json.loads(entry), "index": idx} for idx, entry in rows]

Reads & sweeps

load and load_for_event

load rehydrates an Execution from its JSON blob. load_for_event folds the dedupe check into the same query (the worker’s per-event pair): it returns (execution, already_processed) in one round-trip via a correlated subquery + EXISTS.

def load(self, execution_id: str) -> Optional[Execution]:
    row = self._conn.execute("SELECT data FROM harel_executions WHERE id = ?", (execution_id,)).fetchone()
    return Execution.model_validate_json(row[0]) if row is not None else None

def load_for_event(self, execution_id: str, event_id: str) -> tuple[Optional[Execution], bool]:
    row = self._conn.execute(
        "SELECT (SELECT data FROM harel_executions WHERE id = ?), "
        "EXISTS(SELECT 1 FROM harel_processed_events WHERE execution_id = ? AND event_id = ?)",
        (execution_id, execution_id, event_id),
    ).fetchone()
    if row is None or row[0] is None:
        return None, False
    return Execution.model_validate_json(row[0]), bool(row[1])

list_executions

A page of lightweight ExecutionSummary for the monitor, projecting only the scalar fields out of the JSON blob via json_extract (never pulling the full data). Filters: definition_id (exact), status (any-of, json_extract(data,'$.status') IN (…)), roots_only (json_extract(data,'$.parent_id') IS NULL). Pagination is offset-based with an opaque cursor, fetching limit + 1 rows to know whether a next page exists:

rows = self._conn.execute(
    "SELECT id, definition_id, version, json_extract(data,'$.status'), "
    "json_extract(data,'$.outcome'), json_extract(data,'$.active_path'), "
    "json_extract(data,'$.parent_id'), json_extract(data,'$.finished_at') FROM harel_executions "
    f"WHERE {' AND '.join(where)} ORDER BY id LIMIT ? OFFSET ?",
    (*params, limit + 1, off),
).fetchall()

is_processed

The dedupe lookup, a simple existence check on harel_processed_events:

def is_processed(self, execution_id: str, event_id: str) -> bool:
    row = self._conn.execute(
        "SELECT 1 FROM harel_processed_events WHERE execution_id = ? AND event_id = ?",
        (execution_id, event_id),
    ).fetchone()
    return row is not None

Outbox relay

pending_outbox returns undelivered entries oldest-first (by seq); ack_outbox removes one by seq after delivery:

def pending_outbox(self) -> list[OutboxEntry]:
    rows = self._conn.execute("SELECT seq, target_id, event FROM harel_outbox ORDER BY seq").fetchall()
    return [
        OutboxEntry(seq, target_id, Event.model_validate_json(event)) for seq, target_id, event in rows
    ]

def ack_outbox(self, seq: int) -> None:
    self._conn.execute("DELETE FROM harel_outbox WHERE seq = ?", (seq,))
    self._conn.commit()

Spawn relay

pending_spawns / ack_spawn mirror the outbox for orthogonal child-creation intents (the relay creates each child idempotently, then acks):

def pending_spawns(self) -> list[SpawnEntry]:
    rows = self._conn.execute(
        "SELECT seq, parent_id, child_id, root_path, context FROM harel_spawns ORDER BY seq"
    ).fetchall()
    return [SpawnEntry(seq, pid, cid, rp, json.loads(ctx)) for seq, pid, cid, rp, ctx in rows]

def ack_spawn(self, seq: int) -> None:
    self._conn.execute("DELETE FROM harel_spawns WHERE seq = ?", (seq,))
    self._conn.commit()

Timer sweep

due_timers returns every timer whose fire_at <= now, soonest-first; delete_timer removes one only if it still holds the same fire_at, so a concurrent re-schedule to a new time survives a stale sweep:

def due_timers(self, now: float) -> list[tuple[str, str, float]]:
    rows = self._conn.execute(
        "SELECT execution_id, path, fire_at FROM harel_timers WHERE fire_at <= ? ORDER BY fire_at", (now,)
    ).fetchall()
    return [(eid, path, fa) for eid, path, fa in rows]

def delete_timer(self, execution_id: str, path: str, fire_at: float) -> None:
    self._conn.execute(
        "DELETE FROM harel_timers WHERE execution_id = ? AND path = ? AND fire_at = ?",
        (execution_id, path, fire_at),
    )
    self._conn.commit()

close() closes the underlying connection.

purge

purge(execution_id, expected_version) is the same single transaction as SqliteStore.purge (shared SQL, _PURGE_COMPANIONS_SQL):

DELETE FROM harel_executions WHERE id = ? AND version = ?        -- the CAS
-- 0 rows and the id still exists -> it moved on: roll back, return False
DELETE FROM harel_processed_events WHERE execution_id = ?
DELETE FROM harel_trace            WHERE execution_id = ?
DELETE FROM harel_timers           WHERE execution_id = ?
DELETE FROM harel_outbox           WHERE target_id    = ?
DELETE FROM harel_spawns           WHERE parent_id    = ?
COMMIT

Any error rolls it back. It returns True if the Execution was deleted. If no row exists under the id at all, the companion deletes still run (a no-op, or the leftovers of an earlier interrupted purge) and it returns False. The control plane’s purge calls it once per member of a finished tree. The async twin runs it on a thread under its lock, like every other call.

ids_with_prefix(prefix) is the same escaped LIKE query as SqliteStore’s.

Published definitions

The published machine versions (see published versions) live in their own table, written outside any execution’s commit:

CREATE TABLE IF NOT EXISTS harel_definitions
  (definition_id TEXT NOT NULL, version INTEGER NOT NULL, digest TEXT NOT NULL,
   source TEXT NOT NULL, published_at REAL, PRIMARY KEY (definition_id, version))

publish_definition is INSERT OR IGNORE — the primary key keeps the version stored first — and then reads back what is stored; load_definition is a primary-key lookup. source is the record’s JSON (the machine’s file and the files it imports).

Async twin — AsyncLibsqlStore

harel/engine/aio_store/libsql.py is the async-facing store the async worker talks to — but it is not natively async. The libsql package is a synchronous sqlite3-style driver with no awaitable API, so AsyncLibsqlStore is a thread off-load wrapper around the sync LibsqlStore:

  • it holds one sync LibsqlStore (self._s) and one asyncio.Lock;

  • every method off-loads its sync call to a worker thread with asyncio.to_thread(...) (so the event loop is never blocked on libSQL IO);

  • the lock serializes those off-loaded calls — one libSQL connection used one operation at a time, which suits this single-writer-class backend.

def __init__(self, sync_store: Any) -> None:
    self._s = sync_store
    self._lock = asyncio.Lock()

A representative method shows the pattern (every method follows it): take the lock, then to_thread the corresponding sync call:

async def commit(
    self, exe, emits, processed_event_id=None, timers=(), spawns=(), trace=None
) -> list[int]:
    async with self._lock:
        return await asyncio.to_thread(
            self._s.commit, exe, emits,
            processed_event_id=processed_event_id, timers=timers, spawns=spawns, trace=trace,
        )

Construction is via the async factory create(), which builds the sync store on a thread too (opening the connection and creating the tables is itself blocking IO):

@classmethod
async def create(
    cls, database: str = ":memory:", *, auth_token="", sync_url=None, sync_interval=None
) -> "AsyncLibsqlStore":
    from harel.engine.store import LibsqlStore

    sync = await asyncio.to_thread(
        LibsqlStore, database, auth_token=auth_token, sync_url=sync_url, sync_interval=sync_interval
    )
    return cls(sync)

trace_max is a property delegating to the sync store (so configuring the ring on the async wrapper configures the underlying sync store):

@property
def trace_max(self) -> int:
    return self._s.trace_max

@trace_max.setter
def trace_max(self, value: int) -> None:
    self._s.trace_max = value

This is the one store whose async twin is not native — every other backend (aiosqlite, redis.asyncio, psycopg async pool, motor, aioboto3, httpx for rqlite) issues real awaitable IO; AsyncLibsqlStore is sync-on-a-thread because the libsql driver leaves no other option today.

When to pick it / tradeoffs

  • Today: a single-file embed. On the local-file path LibsqlStore is functionally SqliteStore reached through the libSQL driver — durable single-machine state, zero infrastructure, identical SQL and guarantees. If you only need that, SqliteStore is the battle-tested choice; LibsqlStore is interesting mainly as the same code path that can later point at Turso.

  • Tomorrow: managed distributed SQLite. The sync_url / embedded-replica path turns the same backend into a client of a managed, replicated SQLite primary (Turso/sqld) — local reads, remote-primary writes, background sync — without changing any of the engine’s persistence logic.

  • The caveats are the experimental ones. The replica path is wired but unvalidated against a real account, and primary-follower replication is eventually consistent: a CAS decided against a lagging follower can spuriously conflict, so read from the primary for the CAS or budget for extra StoreConflict retries. The async wrapper is sync-on-a-thread, serialized to one in-flight operation — fine for a single-writer-class store, but not a true concurrent-IO backend.

See the stores hub for the full backend comparison and the ExecutionStore contract, and durability for why the one-transaction commit, the outbox, the dedupe ledger and the version-CAS together make a crash safe.