without_durability_postgres¶
A without-durability checkpoint store and queue backed by Postgres, where every guarantee is an ordinary transaction.
module-attribute
¶
SCHEMA = "\nCREATE TABLE IF NOT EXISTS workflow_checkpoint (\n workflow text NOT NULL,\n step text NOT NULL,\n value jsonb NOT NULL,\n PRIMARY KEY (workflow, step)\n);\n\nCREATE TABLE IF NOT EXISTS workflow_claim (\n workflow text PRIMARY KEY,\n token bigint NOT NULL,\n held_until timestamptz NOT NULL\n);\n\nCREATE TABLE IF NOT EXISTS workflow_queue (\n namespace text NOT NULL,\n workflow text NOT NULL,\n visible_at timestamptz NOT NULL,\n PRIMARY KEY (namespace, workflow)\n);\n\nCREATE INDEX IF NOT EXISTS workflow_queue_visible_at ON workflow_queue (namespace, visible_at);\n"
dataclass
¶
PostgresCheckpointer(
pool: AsyncConnectionPool,
codec: CheckpointCodec[str] = JSON,
)
A workflow's completed steps as rows in one table, and its claim as a row in another.
The Checkpointer implementation for the deployment that already has a Postgres, and
the one that can co-commit with the application's own tables, which is the capability
the whole Effect parameter exists for. SqlEffect is a callback over transact's
open transaction, so a step whose effect is a write to this database happens exactly
once rather than at least once.
It holds a pool rather than a connection, because a pass is one short transaction and
several passes run at once: a worker with a pool of ten runs ten passes without them
queueing behind each other, and next_ready's poll is not blocking a connection while
it waits. Call migrate once against the same pool before anything else, at the
entrypoint that built it.
The durability question RedisCheckpointer has to hedge on does not arise here.
record returning means the transaction committed, and a default Postgres has
synchronous_commit on, so the write is on disk and survives a crash of the server
rather than only of the client. That is exactly what run_durably's reasoning about
the window between an effect and its record assumes.
A workflow id carries no constraints here at all, since it is bound as a query parameter rather than 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 document in a jsonb column and comes
back, defaulting to the stdlib's JSON. The column type narrows what a codec here may
be in a way it does not for the other two stores: it MUST render JSON text, because
that is what jsonb will accept. What that still leaves free is the library and the
value mapping, which is the part worth changing. What it MUST keep, as everywhere, is
the round trip.
The column narrows the values too, and this is the one place where "store it as
jsonb" is not free. jsonb holds a parsed document rather than the text it was
given, so what comes back is jsonb's rendering of the value rather than the codec's,
and three things change with it:
- a number goes through
numeric, so a step returning1e16is read back as the integer10000000000000000. Above 2^53 it is not even the same number, sincejson.dumpswrites the shortest decimal that round-trips as a float andnumerickeeps that decimal exactly:2.024478232766865e+16returns as20244782327668650, which is a different value and not merely a different type. - keys are reordered by
jsonb's own rule (length, then bytes), so a mapping comes back in an order the codec did not choose. Equality survives it; iteration order does not, so a workflow that iterates a recorded mapping should sort it. - a string
jsonbcannot hold is refused outright rather than narrowed: aNULescape or a lone surrogate is valid JSON and valid to every other store here, andrecordraises on the cast.
Nothing about the codec can repair any of it, since it happens after encode and
before decode. So the round trip a step result MUST survive here is jsonb's and
not only JSON's. run_durably catches the first of the three rather than a comment,
by comparing what a node returned against what the store reads back, type included,
on the pass that wrote it; Run.step's parser is where a stepwise workflow says what
it expects.
async
¶
Run effect and record it in one transaction, so the step happens once.
The order is the Lua script's, for the same reasons: 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. What differs is that none of it needed a
mechanism. BEGIN and COMMIT are the atomicity, the connection pool's context
manager is what issues them, and an exception anywhere inside (the fence, the
effect's own SQL, a constraint the effect violated) rolls the whole thing back
including the record.
The effect's result is written and read back through the column rather than
returned as it came, so it round-trips through the codec exactly as a later pass
will see it. A step that returns something jsonb renders differently (a tuple,
which comes back a list) then does so on the first pass rather than surprising the
second.
The fence is held for as long as the effect runs, which is what makes it a fence
and is worth stating because of what it costs elsewhere: claim takes the same
row, so a worker trying to take this workflow over waits for the effect rather
than being told the workflow is held, and each one that waits holds a pool
connection while it does. A long effect and a small pool is a worker that stops
pulling work for unrelated namespaces. Size the pool for the passes a deployment
runs concurrently plus the takeovers it expects, or keep the effects short.
The fence excludes every other pass, and one writer is left over: supply is
ungated on purpose, so an approval can land under this key between the read and the
write. That is what the retry below is for. The insert declines to overwrite, the
transaction rolls back so the effect goes with it, and the value that did land is
read and returned, which is what "the recorded value without re-running" means when
the recording was somebody else's. Rare enough to pay a second transaction for, and
it costs nothing on the path that wins.
dataclass
¶
PostgresDurable(
checkpointer: PostgresCheckpointer,
scheduler: PostgresScheduler,
)
A Durable whose two stores are one database, so arrive is a single commit.
This is the row SplitDurable cannot fill in. Recording the value a workflow is
waiting on and making the workflow runnable are two writes with a crash window
between them everywhere else; here they are two statements in one transaction, so the
window does not exist. That is the same capability transact offers a step, arriving
at the interface above rather than inside a pass, and it is available for the same reason:
both things live in one datastore.
Which is why the two stores MUST share a pool, checked at construction rather than
documented. It is the exact question LuaEffect asks with its hash tag, and it does
not stop being asked because SQL hides it: a checkpoint and a queue in two Postgres
databases are two datastores, and a transaction across them is a distributed
transaction whatever the connection string suggests. Sharded Postgres asks it again
at the next level down, where the answer is that both tables must be distributed by
the workflow id and co-located, or the "one commit" here becomes a two-phase commit
across nodes.
async
¶
Record the value and make the workflow ready, together or not at all.
The order within the transaction does not matter, which is the point: a commit has no halfway. What does matter is that both statements go through the same cursor, since a second connection would be a second transaction wearing the same method's name.
dataclass
¶
PostgresScheduler(
pool: AsyncConnectionPool,
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 either Redis queue: the same protocol, the same worker, the same API.
It is modelled on the sorted-set one rather than on the stream, so queued now is a
visible_at in the past, sleeping is one in the future, and being worked on is one a
lease ahead, which leaves wake_due, reclaim, and prepare's queue half with
nothing to do.
What Postgres adds over the sorted set is SKIP LOCKED, which is what lets several
workers poll one queue without serializing on its head, and what a ZRANGEBYSCORE in
a Lua script gets instead by being the only thing running.
What it does not add is the blocking read. This polls on poll, so an idle worker
costs a round trip per interval and a submitted order waits up to one interval to be
picked up. Postgres can close that (LISTEN/NOTIFY on a dedicated connection, woken
by a trigger or by the writer) and this does not, which is the honest state of it
rather than a claim that a table cannot wait.
namespace separates queues rather than deployments, and it is a column rather than
part of a table name, so a queue name is data here as a workflow id is.
class-attribute
instance-attribute
¶
async
¶
Create the tables, which every worker does at boot and all but the first find done.
It creates the checkpoint tables too, because there is one database and one DDL
for it. That is a little more than this interface is asked for, and it is the right
place anyway: the worker already calls prepare before reading a queue, so a
deployment gets its schema from the same call whichever queue it runs, and an
entrypoint that would rather be explicit calls migrate itself.
async
¶
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.
async
¶
Nothing to do: a workflow whose visible_at has passed is already visible.
async
¶
The next visible workflow, waiting up to within for one to appear.
Polling, because nothing here is listening. within bounds how long a cancelled
worker sits in this call before it can notice, but unlike a blocking read it is
spent in round trips rather than in one parked call, which is the cost of the
design and the reason poll is a knob.
async
¶
Nothing to take over by hand: an abandoned workflow becomes visible on its own.
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. That is why
a worker may call wake_at and then done in that order without the second
undoing the first.
async
¶
Create the three tables, from every process, as often as it likes.
Idempotent by IF NOT EXISTS and safe against itself by the advisory lock, which is
the part that is easy to skip: concurrent CREATE TABLE IF NOT EXISTS is a
duplicate-key error on the system catalog rather than a no-op, and a fleet of workers
booting together is exactly a race. pg_advisory_xact_lock is held to the end of the
surrounding transaction and released by the commit, so there is nothing to unlock.
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, which is the ordinary thing a deployment would want and the ordinary tool (Alembic, sqitch, plain numbered SQL files) is where it belongs.