without_durability_sqlite

A without-durability checkpoint store and queue backed by one SQLite file, with no server and no third-party driver.

without_durability_sqlite

SCHEMA module-attribute

SCHEMA = "\nCREATE TABLE IF NOT EXISTS workflow_checkpoint (\n    seq INTEGER PRIMARY KEY,\n    workflow TEXT NOT NULL,\n    step TEXT NOT NULL,\n    value TEXT NOT NULL,\n    written_at REAL NOT NULL DEFAULT (unixepoch('now', 'subsec')),\n    UNIQUE (workflow, step)\n);\n\nCREATE TABLE IF NOT EXISTS workflow_claim (\n    workflow TEXT PRIMARY KEY,\n    token INTEGER NOT NULL,\n    -- The budget: the latest this claim can lapse at, whatever its holder does.\n    held_until REAL NOT NULL,\n    -- When it lapses if nothing more is heard, which is what `CLAIM` tests. Every write of\n    -- it is a `MIN` against `held_until`, so a sign of life cannot carry a pass past its\n    -- budget or take a workflow back after a `RELEASE`.\n    alive_until REAL NOT NULL,\n    -- What one sign of life is worth, carried here because `RECORD` is one and is not told:\n    -- the statement making a write has only the workflow to go on.\n    alive_for REAL NOT NULL\n) WITHOUT ROWID;\n\nCREATE TABLE IF NOT EXISTS workflow_queue (\n    namespace TEXT NOT NULL,\n    workflow TEXT NOT NULL,\n    visible_at REAL NOT NULL,\n    PRIMARY KEY (namespace, workflow)\n) WITHOUT ROWID;\n\nCREATE INDEX IF NOT EXISTS workflow_queue_visible_at ON workflow_queue (namespace, visible_at);\n"

SqliteEffect

SqliteEffect = Callable[[Cursor], object]

Database dataclass

Database(connection: Connection, guard: Lock = Lock())

One SQLite connection and the lock that keeps one caller in it at a time.

The analogue of the Postgres store's connection pool, and the opposite shape for the opposite reason: a pool exists so several statements run at once, and this exists so they do not. Not because a connection would corrupt (SQLite is built serialized here, so it would not), but because a transaction belongs to the connection: without this, a caller arriving mid-BEGIN IMMEDIATE writes into somebody else's transaction and loses its write to that transaction's rollback. See the note at the top of this module for why the event loop's single thread does not already prevent that.

Build it with connect, which applies the pragmas that make this durable rather than merely persistent. Share one between the checkpoint store and the queue: that is what makes SqliteDurable.arrive a single commit, and it is checked rather than assumed. Close it with aclose, never connection.close(), for the reason given there.

connection instance-attribute

connection: Connection

guard class-attribute instance-attribute

guard: Lock = field(
    default_factory=asyncio.Lock, repr=False, compare=False
)

run async

run(work: Callable[[Connection], T]) -> T

Do work against the connection, on a thread, with nobody else inside it.

Cancellation is where "nobody else" has to be arranged rather than assumed, and it is the reason this is not simply async with self.guard. A thread is not cancellable: cancelling the caller unwinds this coroutine at once while the thread runs on, so releasing the guard on the way out would hand the connection to the next caller while the last one is still inside it. That is not a theoretical race. The statement in flight may be a BEGIN IMMEDIATE transaction, and a write that lands in somebody else's open transaction is committed or rolled back with it: record returns, a read sees the row, and the rollback takes it away again, which is precisely the guarantee this store exists to make.

So the guard is released by the thread finishing rather than by this coroutine returning. The work is a task, the caller awaits a shield of it (so cancelling the caller leaves the task alone), and a done-callback lets go of the connection when it is genuinely free. A cancelled caller still unwinds immediately; what it no longer does is take the connection with it.

What the shield adds beyond that is the reporting. A statement that fails after its caller has gone has nobody left to raise to, and shield hands it to the loop's exception handler rather than dropping it, so a write that failed on the way out of a process is in the log instead of nowhere.

aclose async

aclose() -> None

Close the connection once nobody is inside it.

The other half of run's handshake, and the reason a caller must never reach for connection.close() itself. sqlite3.close() frees the connection and finalizes its statements; a thread still executing one is then reading freed memory, which segfaults the process rather than raising. run makes that reachable by design, since a cancelled caller unwinds while its thread runs on, so a shutdown that follows a cancellation is exactly when the two meet: the worker's task is cancelled, the statement it left behind is still in flight, and the close lands on top of it.

Taking the guard is what waits that out, because run releases it from the thread rather than from its caller. That wait is unbounded: an effect hung inside a run thread holds the guard until it returns, and cancelling this coroutine while it is parked on the guard abandons the close, leaving the connection open. The guard is then let go again, so a run arriving after this fails on a closed connection, which is the loud version of a bug that would otherwise be silent.

The close itself goes to a thread like every other driver call: under WAL it runs the final checkpoint (and, with synchronous=FULL, an fsync), which is real disk I/O that does not belong on the event loop.

SqliteCheckpointer dataclass

SqliteCheckpointer(
    database: Database, codec: CheckpointCodec[str] = JSON
)

A workflow's completed steps as rows in one file, and its claim as a row beside them.

The Checkpointer implementation for a deployment that is one machine, and the one that needs nothing installed. It meets the same requirements as the others by the simplest route any of them take: SQLite admits one writer, so a single statement or a single BEGIN IMMEDIATE transaction is already all the exclusion this needs.

SqliteEffect is a callback over the open transaction's cursor, so a step whose effect is a write to this file happens exactly once. Since the file is the whole datastore, that covers every table an application on this machine keeps here, which is a broader reach than it sounds: it is the same guarantee DBOS gets from Postgres, for an application that never needed Postgres.

A workflow id carries no constraints at all: it is bound as a query parameter, never parsed as key structure. Nothing here derives one id from another either, so an application is free to name a workflow's sibling (a saga's rollback, say) however it likes out of its own namespace.

codec is how a step's result becomes the TEXT in a row and comes back, defaulting to the stdlib's JSON. Swap it to widen what a step may return or to speed the encoding up; what it MUST keep is the round trip, since a resumed pass reads what it produced.

database instance-attribute

database: Database

codec class-attribute instance-attribute

codec: CheckpointCodec[str] = JSON

load async

load(workflow: str) -> dict[str, object]

history async

history(workflow: str) -> dict[str, Written]

The same records load returns, each with the moment it was written.

claim async

claim(
    workflow: str, budget: timedelta, alive: timedelta
) -> Pass | None

extend async

extend(
    holder: Pass, budget: timedelta, alive: timedelta
) -> bool

renew async

renew(holder: Pass, alive: timedelta) -> bool

record async

record(holder: Pass, key: str, value: object) -> Recorded

transact async

transact(
    holder: Pass, key: str, effect: SqliteEffect
) -> object

Run effect and record it in one transaction, so the step happens once.

The order is the other stores': fence first, because a superseded pass must not act; then the existence check, because a step already recorded must not run again, which is what makes a replay perform nothing at all; then the effect; then the record. BEGIN IMMEDIATE holds the write lock across all four, so no other writer can land between them and any exception rolls back the effect along with its record.

The effect's result is written and read back through the codec rather than returned as it came, so it round-trips exactly as a later pass will see it.

What the write lock costs is stated rather than hidden: it is the connection's, and there is one connection, so nothing else in this process reaches the store until the effect returns, the worker's own renewal included. The claim is safe regardless, since the sign of life at the end of the transaction lands before any queued claim runs; what a long effect does lose is its delivery, which is not renewed meanwhile and is redelivered after a lease to a pass that finds the workflow held. Keep effects short, or accept that redelivery.

supply async

supply(workflow: str, key: str, value: object) -> object

append async

append(workflow: str, value: object) -> Entry

File value in this workflow's inbox, under the next key in the table.

discard async

discard(workflow: str) -> int

Forget every record this workflow has, and raise its fence, in one transaction.

One commit rather than two statements, because the two are only right together: a crash between them either leaves the records deleted with the fence unraised, so the pass that was mid-flight writes them back one at a time, or the reverse, which fences a live pass for a deletion that never happened.

What is left behind is one claim row carrying a number. Nothing here sweeps it, in keeping with the rest of this store, where nothing expires and tidying the file is the deployment's homework.

release async

release(holder: Pass) -> None

SqliteDurable dataclass

SqliteDurable(
    checkpointer: SqliteCheckpointer,
    scheduler: SqliteScheduler,
)

A Durable whose two stores are one file, so arrive is a single commit.

The strongest form of the guarantee, reached by the least machinery: there is nothing to co-locate, no pool to share by accident, and no sharding to grow into. The two stores MUST hold the same Database, checked at construction, which here is less a warning about distributed transactions than a way of saying that two SQLite files are two datastores however adjacent they sit on disk.

checkpointer instance-attribute

checkpointer: SqliteCheckpointer

scheduler instance-attribute

scheduler: SqliteScheduler

arrive async

arrive(workflow: str, key: str, value: object) -> object

Record the value and make the workflow ready, together or not at all.

deliver async

deliver(workflow: str, value: object) -> Entry

Append the message and make the workflow ready, together or not at all.

delete async

delete(workflow: str) -> int

Cancel the workflow's wakeups and forget its records, together or not at all.

Three statements in one commit, so the ordering SplitDurable has to reason about does not arise: there is no window in which the records are gone and a wakeup is not, and none in which the fence has been raised for a deletion that did not happen. Which is the same thing arrive gets from this store and for the same reason, one file.

SqliteScheduler dataclass

SqliteScheduler(
    database: Database,
    namespace: str = "workflow",
    lease: timedelta = LEASE,
    poll: timedelta = POLL,
    now: Callable[[], datetime] = now_utc,
)

Scheduler as one table, each row scored by when its workflow becomes visible.

A drop-in for every other queue here, and modelled on the same visibility scheme: queued now is a visible_at in the past, sleeping is one in the future, and being worked on is one a lease ahead, so wake_due, reclaim, and prepare's queue half all have nothing to do.

It polls, like the other visibility-scored queues, so the poll interval is a floor under how fast anything starts. SQLite offers no blocking read and no notification a process outside this one can wait on, so unlike the Postgres store there is not even a LISTEN/NOTIFY left on the table: within one process an asyncio.Event would do it, across processes on one machine it would take a filesystem watch, and neither is here.

database instance-attribute

database: Database

namespace class-attribute instance-attribute

namespace: str = 'workflow'

lease class-attribute instance-attribute

lease: timedelta = LEASE

poll class-attribute instance-attribute

poll: timedelta = POLL

now class-attribute instance-attribute

now: Callable[[], datetime] = now_utc

lease_seconds class-attribute instance-attribute

lease_seconds: float = field(
    init=False, repr=False, compare=False
)

poll_seconds class-attribute instance-attribute

poll_seconds: float = field(
    init=False, repr=False, compare=False
)

prepare async

prepare() -> None

Create the tables, which every worker does at boot and all but the first find done.

make_ready async

make_ready(workflow: str) -> None

wake_at async

wake_at(delivery: Delivery, when: datetime) -> None

Suspend the workflow until when, unless something asked for a pass meanwhile.

The receipt is the visibility this pass took, so anything that rescheduled the workflow since (a confirmation, another worker taking over an overrun) wrote a different one and this leaves it be. Which is the right answer rather than a concession: the deadline lives in the workflow's checkpoint, so the pass that runs sooner reaches the same sleep and writes it again.

schedule async

schedule(workflow: str, visible_at: datetime) -> None

wake_due async

wake_due(now: datetime) -> tuple[str, ...]

Nothing to do: a workflow whose visible_at has passed is already visible.

next_ready async

next_ready(within: timedelta) -> Delivery | None

The next visible workflow, waiting up to within for one to appear.

reclaim async

reclaim(idle: timedelta) -> Delivery | None

Nothing to take over by hand: an abandoned workflow becomes visible on its own.

extend async

extend(delivery: Delivery, within: timedelta) -> Delivery

Push this delivery's visibility out, and say what it is called now.

The new visibility is the new receipt, since this table's receipt is its visibility, so the caller is handed a delivery to use from here rather than left to discover that the one it holds has been renamed.

No row means this delivery is no longer this worker's: cancelled, or rescheduled by a wakeup that arrived mid-pass, either of which wrote a visible_at that is not the one it took. The answer to both is to hand back what came in, exactly as wake_at and done already do.

cancel async

cancel(workflow: str) -> None

Drop the workflow's row, whichever of the three things its visible_at means.

One DELETE covers queued, sleeping, and out with a worker, because this table holds one row per workflow and the visibility is the only thing that differs between them. That is the same collapse that leaves wake_due and reclaim with nothing to do.

The half of cancel a queue sweep cannot reach comes free with it: a pass still in flight answers with wake_at, which is an UPDATE conditional on the visibility still being the one it took, and a deleted row has none. So the deadline it was about to write updates nothing and a deleted workflow is not put back to sleep.

done async

done(delivery: Delivery) -> None

Drop the workflow, unless something asked for another pass while this one ran.

The receipt is the visibility this pass took, so anything that rescheduled the workflow meanwhile (a confirmation, this pass's own wake_at, another worker taking over an overrun) wrote a different one and this leaves it alone.

connect

connect(
    path: Path | str,
    *,
    timeout: Milliseconds = BUSY_TIMEOUT,
) -> Database

Open the database this store runs on, configured for durability rather than speed.

• journal_mode=WAL so a reader does not block the writer, which is what lets a status query run while a pass is mid-transaction.
• synchronous=FULL because this store's entire claim is that record returning means the value survives. NORMAL is the usual advice under WAL and it trades exactly that away: a commit can be lost on power loss or an OS crash. Everything run_durably reasons about assumes the commit held, so this pays the fsync.
• busy_timeout so a second process finding the write lock taken waits for it rather than failing immediately, which is the ordinary case when two processes share the file.

autocommit=True leaves transaction control here rather than in the driver: every statement below is either atomic on its own or wrapped in an explicit BEGIN IMMEDIATE, and nothing is left to a hidden implicit transaction.

migrate async

migrate(database: Database) -> None

Create the three tables, from every process, as often as it likes.

No advisory lock and no race to guard against, unlike the Postgres migration: SQLite runs the whole script in one exclusive transaction, so a second process either waits for it or finds the tables already there.

Schema migration as a whole is not what this is. There is no versioning and no path from one shape of these tables to another; user_version is where SQLite keeps that, and a deployment that needs it should use it.