without_durability¶
Durable workflows for without: a checkpoint any process can read, and the store interfaces that make one writer at a time enforceable.
Bases: Protocol
How a checkpointed value crosses into a store and back out.
Encoded is what this store can hold: text for the three shipped here, since a
Redis hash field, a SQLite TEXT column, and a Postgres jsonb all take it. It is a
type parameter rather than a fixed str because that is a fact about each store and
not about codecs, and a store that holds bytes should be able to say so.
Two requirements, and the second is the one that is easy to miss.
decode(encode(value))MUST equalvaluefor every value a workflow's steps return. A codec that does not round-trip makes a resumed pass see something the first pass did not, silently, one crash later. The stdlibJsonCodecbelow does not round-trip a tuple (it comes back a list) or a mapping with non-string keys, which is why a workflow using it must keep its step results JSON-native.encodeMUST be deterministic: values that are equal and of the same type at every level encode equal.Checkpointer.recorddecides who won a race by comparing encodings, so a codec that renders one value two ways reports a conflict that did not happen.
Both qualifiers are load-bearing, and the second is easy to drop. Python holds
1 == 1.0 while JSON tells 1 and 1.0 apart, and a codec cannot both encode them
identically and give each of them back, since one of the two would round-trip to the
other; so the plain reading is not available to any codec whose format distinguishes
what Python's equality does not, and asking for it would be asking for the round trip
to be broken. But the same pair inside a container is the same problem with a
container around it: [1] and [1.0] are equal, are both list, and still encode
apart. So the requirement is the property without_durability.graph.survives checks,
which is what a graph run already holds its node results to.
What it leaves is a store comparing text answering first=False for a tie between
1 and 1.0 while one comparing jsonb answers first=True, over values a
workflow should not be producing for one key anyway.
Both are properties of the pair, which is why a codec is one object rather than two functions: the stores do not merely encode, they compare encodings to decide who won a race and hand the decoded form back so a pass reads what the next pass will.
Only the encoded side is a parameter, and that asymmetry is real rather than an
oversight. Encoded genuinely varies: the stores here hold text, and one that held
bytes would say so. The decoded side cannot, because a checkpoint is heterogeneous by
construction: a workflow's "charged" holds a string, its "items" a mapping, its
"settling" a deadline, and one codec carries all of them. A Decoded parameter
would sit in encode's argument and decode's return, making it invariant, so a
CheckpointCodec[Step, str] would be refused by the very store it was written for.
Precision belongs inside a codec instead, where it costs nothing: a pydantic codec's
TypeAdapter can be as exact as it likes about what a workflow returns while still
presenting object here. That is the move without_dag.Node already makes, crossing
the executor interface as object with a typed frontend restoring precision above it.
dataclass
¶
The stdlib's JSON, as a CheckpointCodec[str], and the default every store here takes.
JSON because it is what makes a checkpoint readable by an operator with redis-cli or
psql and by a service written in something other than Python, which is most of what a
durable workflow's state is for. The stdlib because a default should add no
dependency; it is the slowest of the reasonable choices and the narrowest, and both are
the point of the codec being swappable.
What it costs is stated rather than hidden: a step result MUST be JSON-native, and
"JSON-serializable" is not the same thing. A tuple encodes and comes back a list, and a
mapping with integer keys comes back with string ones, so both break the round trip the
protocol requires. A codec that knows the application's types (a pydantic
TypeAdapter, msgspec with a schema) is how a workflow gets to return domain values,
and swapping one in changes the store's construction and nothing else.
Two arguments to json.dumps are what make it meet the protocol rather than merely
resemble it, and each one is a requirement rather than a preference.
sort_keys is determinism. A mapping's encoding would otherwise follow its insertion
order, so two passes that computed the same mapping by different routes encode it two
ways, and a store deciding who won by comparing encodings (record) reports a conflict
between values that are equal. Sorting also makes a key order the thing it should be, a
fact about the value rather than about how it was built. What it costs is that a mapping
whose keys are not mutually orderable ({1: ..., "a": ...}) now raises here instead of
encoding, which is the round trip failing where it can be seen rather than one crash
later.
allow_nan is the round trip. Left on, float("nan") encodes to the bare token NaN,
which is not JSON: it decodes back unequal to itself, the checkpoint stops being
readable by anything that parses JSON strictly, and a store whose column is jsonb
refuses the write at the far end of a workflow the double accepted. Off, the value that
cannot survive is refused where it is produced.
Bases: Protocol
Where a workflow's completed work is kept, and who is currently allowed to add to it.
The narrow interface a durable runner talks through, so the store is injected rather than
reached for: a Redis hash, a Postgres table, or a SQLite file in production, a plain
dict in a test. Its keys are plain names rather than without_dag's NodeKey,
because the store is the piece the two mechanisms share: a graph records under its
node names (run_durably) and an ordinary function under its step names
(stepwise), and the store cannot tell, nor should it.
The requirements are the whole reason this protocol is not just a mapping. A runner cannot construct exclusion out of an interface with no way to express it, so a store that cannot meet them cannot make a workflow safe to run.
loadMUST return the values recorded for that workflow so far, and an empty mapping for one that has never run.claimMUST grant at most one livePassper workflow, and MUST issue tokens that strictly increase per workflow, so that a later claim always outranks an earlier one. It returnsNonewhen someone else holds the workflow.recordMUST refuse a write whose token is below the highest claimed for that workflow, raisingFenced, and MUST NOT overwrite a key that is already recorded. It returns aRecorded: the value stored after the call, so two passes that both ran an effect at least agree on its result rather than diverging, and whether that value is this pass's own.recordandsupplyMUST make the value durable before returning.- Every value MUST cross the store's
CheckpointCodecin both directions, so that whatloadandrecordhand back is what a later pass will read rather than what this one happened to pass in. A store that skips the round trip on the way out is a store whose tests pass and whose resumed workflows see something else. transactMUST run the effect at most once across every pass of a workflow, returning the recorded value without re-running when the step is already recorded, and it MUST NOT leave the effect applied without its record or the reverse.
transact is the one that changes the guarantee rather than protecting it. record
is a second round trip after an effect already happened, so a crash in between leaves
the effect done and unrecorded: at-least-once, the bound every durable engine lands
on. Performing the work and writing the record in one commit closes that, for the
effects a store can perform itself. What bounds that is neither this interface nor a
store's feature list but the fact that you can only transact within one datastore
(see docs/without-durability/guarantees.md).
Effect is how a store expresses such a piece of work, and it is a type parameter
because there is no shared answer: a Lua script over keys in the same Redis, a
callback handed a cursor inside an open SQL transaction, a function over an in-memory
store's own dict. A store with nothing to offer here uses Never, which makes
transact uncallable rather than absent, since a caller cannot produce a value of
that type. That is also the default, so bare Checkpointer reads as "any store,
never mind what it can co-commit": an effect only ever goes in, so the parameter is
contravariant and Checkpointer[Never] is the supertype every concrete store
satisfies.
Only record reports who won, and the asymmetry is deliberate. It is the one write
whose caller has a decision to make, since run_durably has already handed a node's
result to that node's dependents by the time it writes. supply is called from
outside any pass by a client that wants the stored value and nothing else, and
transact runs at most once by construction, so neither has a race to report.
Bases: Interruption
Another pass holds this workflow, so this caller does not get to run one.
dataclass
¶
One wakeup, taken by a worker and not yet acknowledged.
The receipt is what makes the queue crash-safe: it names the entry the store is
still holding on this worker's behalf, so acknowledging is a separate act from
receiving and a worker that dies between them leaves the wakeup to be taken over
rather than losing it.
Bases: Protocol
Both stores a workflow needs, and the transitions that have to be atomic across them.
This exists because holding a Checkpointer and a Scheduler side by side is not
simpler to use correctly. Making a workflow runnable is two writes to two places,
and a caller that does them in the wrong order, or does the first and dies, leaves a
workflow that is recorded and unreachable. Scheduler.wake_due already answers that
shape of problem by naming the transition rather than exposing its halves, so that
a caller cannot hold a claimed-but-unqueued id at all; arrive is the same move one
level up.
The two stores stay separate underneath, because they genuinely can be separate: a
Postgres checkpoint beside an SQS queue is an ordinary deployment, and forbidding it
would be bundling a mechanism to fix an interface. What this interface changes is who
carries the coupling. Callers get one call with no ordering to get right, and what
varies between implementations is not whether arrive exists but what it
guarantees.
arriveMUST recordvalueunderkeywith first-writer-wins, exactly asCheckpointer.supplydoes, and MUST make the workflow ready. It returns the value stored after the call, the caller's if it won and the existing one if it did not.arriveSHOULD be a single commit where the two stores are one datastore, which is what a store built on one database or one file can offer.- Where they are not, it MUST record before it queues (
SplitDurable). The two failures are not symmetric: recorded-and-unqueued is a workflow waiting for a wakeup that a resubmission supplies, and queued-with-nothing-recorded is a pass that wakes, finds nothing to do, and drops the value on the floor.
Effect is Checkpointer's, threaded through so that a caller holding a Durable
can still reach a store's transact, and defaulting to Never for the same reason.
Bases: Interruption
A write from a pass that has been superseded, refused rather than applied.
Raised when a Pass outlives its claim and someone else has since taken the
workflow. It means this pass has lost, not that the workflow has: whoever holds the
newer claim carries on, and the right response is to stop, since every subsequent
write would be refused too. Which is exactly why it is an Interruption, since
compensating a saga or logging a failure are responses to the workflow going wrong
and both are wrong here.
Bases: BaseException
A control-flow signal from the durable machinery, not a failure of the work.
It descends from BaseException for the reason asyncio.CancelledError does: an
except Exception written to handle a workflow's own errors (a gateway declined, a
row was missing) must not silently absorb a signal about whether this pass may run at
all. Catching them is deliberate, by name, or not at all.
dataclass
¶
The right to run one pass at a workflow, and the proof of it.
token is a fencing token, not an identifier: it rises with every claim on this
workflow, so comparing two of them says which pass is the newer one. That is what
makes the exclusion survive a stalled process, which a lease alone cannot. A holder
that pauses past its lease keeps its Pass and believes it still owns the workflow;
the store is what knows better, because the next claim raised the number and every
write carries one. See Checkpointer.record.
dataclass
¶
What a store holds under a step's key after a write, and whether this pass put it there.
value is what the store holds, decoded: the caller's when it won and the winner's
when it did not. It is read back through the codec either way rather than handed back
as it came, so a value that does not survive the round trip shows that on the first
pass rather than surprising the second.
first is the part a caller cannot work out for itself, and the reason record
returns a value rather than a bare object. Equality between what a pass handed in
and what came back answers a different question, since a result crosses a
CheckpointCodec both ways: a step returning a tuple gets a list back from a JSON
codec having won outright, and a runner comparing the two would report a race that
never happened. Only the store sees both encodings, so only the store can say. It is
true when the encoding stored is this pass's own, which counts a tie as a win for
both: two passes that ran the same effect have nothing to disagree about.
Separating the two is what makes the equality worth testing rather than something to
avoid. Once first answers "did I win", comparing value against what went in
answers "did this value survive its own store", which is the check run_durably
makes.
Bases: Protocol
Where a workflow's right to run is kept, apart from what it has done.
The interface the API and the worker share: the API makes a workflow ready, a worker
takes the next ready one and says when it is done with it. Injected like the
checkpoint store, so the worker is drivable from a dict in a test.
The requirements are about not losing a wakeup, since a lost one is a workflow that
never runs again, and they are stated as properties rather than as mechanics because
the implementations reach them by different routes. One is a Redis stream beside a
sorted set; the rest are a single structure scored by when a workflow becomes
visible, where wake_due, reclaim, and prepare all have little or nothing to do.
An implementation MUST guarantee that:
- a workflow passed to
make_readyis eventually yielded by somenext_ready, even if the worker holding it dies mid-pass, and even if the wakeup arrives while a pass on that workflow is running; wake_duemoves each workflow it reports in one durable step, since one that removes a deadline and then queues the workflow loses it whenever it dies in between (an implementation with nothing to move satisfies this trivially);wake_atanswers for its delivery and sets the workflow's next pass in one step, and MUST NOT overwrite a wakeup that arrived since that delivery was taken.
That last one is why wake_at takes a Delivery rather than a workflow id, and it
is the same move wake_due and Durable.arrive make. Scheduling and acknowledging
were two calls with a rule about their order, which a protocol cannot enforce and a
caller can get wrong; worse, on a store that holds one entry per workflow they are a
read-modify-write over a value somebody else may have just written, so a confirmation
that landed while the pass was ending was overwritten by the deadline the pass chose
and waited days for a clock instead of running at once. Naming the transition instead
means the store compares the receipt it handed out, which is the one thing that can
tell the two apart.
What is deliberately not required is that a workflow reach only one worker at a
time. The stream will happily hand two deliveries for one workflow to two consumers,
and that is safe because exclusion belongs to Checkpointer.claim rather than here:
this interface answers "who owes a pass", the checkpoint store answers "who may write".
lease is how long a delivery stays its taker's, and it is on the store rather than
an argument to every call because the same number has to bound the checkpoint
claim. The implementations reach it by different routes (an idle threshold reclaim
measures against, or the invisibility a visibility-scored queue writes when it takes
one) and it is the answer to the same question either way, so work reads it here
and claims the workflow for exactly as long. Taking the two from different places
fails quietly: a delivery reclaimed while its holder can still write is a pass spent
finding out that somebody else owns the workflow.
A wake_at deadline is the caller's clock, where a lease is measured by the
store's, because a workflow chooses its own deadline (Run.sleep records one) and
nothing else can say what it meant. So a wakeup lands early or late by whatever the
two clocks disagree by, which bounds how promptly a sleep ends rather than
correctness: a pass that wakes too early finds its deadline unreached and suspends
again.
dataclass
¶
SplitDurable(
checkpointer: Checkpointer[Effect], scheduler: Scheduler
)
A Durable over two stores that are not one datastore, so arrive is two writes.
The general composition, and the one that admits it cannot co-commit. It is what a Redis deployment uses (the checkpoint hash and the queue live in one Redis, but on a cluster they are deliberately in different slots, which is the same thing as being in different stores), and what any pairing of unrelated products uses.
The order is the whole of what it can offer, and it is not arbitrary: the record goes first, so a crash in the window leaves a workflow that has the value and lacks the wakeup, which anything asking again supplies. The reverse would queue a pass that wakes to find nothing recorded and answers for the delivery, losing the value outright.
dataclass
¶
MemoryCheckpointer(
hashes: dict[str, dict[str, str]] = dict(),
tokens: dict[str, int] = dict(),
held_until: dict[str, float] = dict(),
codec: CheckpointCodec[str] = JSON,
data: dict[str, object] = dict(),
)
A Checkpointer keeping one dict per workflow, and one claim beside it.
It meets the protocol's requirements rather than approximating them, which is the
only way a test against it says anything about a real store: tokens rise per
workflow, a write below the fence raises Fenced, and a key already recorded is
never overwritten. Every method is synchronous between its awaits, which is this
store's version of a Lua script or a transaction.
A workflow whose checkpoint is a dict in this process is durable across exactly
nothing, so this is for tests and for driving a workflow in a script, not for a
deployment. hashes holds what the codec produced rather than what a step returned,
so reading a checkpoint back means load rather than reaching into it.
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
async
¶
transact(
holder: Pass, key: str, effect: MemoryEffect
) -> object
Run effect over this store's own data and record it, without an await between.
The in-memory answer to the question Redis answers with a script and SQL with a
transaction: this store's datastore is data, so an effect is a function over
data, and single-threaded code with no suspension point is its transaction.
Which is the point of Effect being a type parameter, since nothing about
LuaEffect would fit here.
Having no suspension point is only half a transaction, and the other half is the
rollback. An effect that raises partway, or one whose result the codec refuses,
would otherwise leave data moved and nothing recorded, which is the state the
protocol forbids outright and the state a replay then compounds by running the
effect again over data it already moved. So the mapping is snapshotted first and
put back on any exception, which is what transacted gets from ROLLBACK in the
SQLite store and a script gets from Redis running it to completion or not at all.
The snapshot is shallow, which bounds what this double can stand in for: an
effect that reaches inside a value in data (appending to a list it holds)
mutates something the restore hands back unchanged. That is the same bound the
rest of this store has, since a dict is not a datastore, and an effect written
the way a real one is (replace the entry, do not edit it in place) stays inside
it.
dataclass
¶
MemoryScheduler(
queue: deque[str] = deque(),
sleeping: dict[str, datetime] = dict(),
outstanding: dict[str, tuple[Delivery, float]] = dict(),
arrived: Event = Event(),
receipts: count[int] = count(),
lease: timedelta = LEASE,
)
A Scheduler keeping the queue in a deque, the sleepers in a dict, and, like the
stream it stands in for, the deliveries nobody has answered for yet.
outstanding is the part worth having a double for: a delivery stays there until
done, so a test can drop one on the floor the way a dying worker would and watch
reclaim pick it up. next_ready waits on an event rather than returning
immediately, mirroring the blocking read: a worker with nothing to do parks instead
of spinning, and a test that hands it a workflow gets a pass the moment it does.
dataclass
¶
The pass ran the workflow to the end, and value is what it returned.
InputNeeded(key: StepKey)
Bases: Suspended
The pass is waiting on a value only something outside it can supply (Run.awaiting).
Nobody schedules a wakeup for this, because no clock will satisfy it: the thing that writes the value is what makes the workflow ready again.
dataclass
¶
Run(
holder: Pass,
checkpointer: Checkpointer[Effect],
recorded: dict[StepKey, object],
now: Callable[[], datetime] = now_utc,
claimed: set[StepKey] = set(),
waking: dict[StepKey, datetime] = dict(),
)
One pass at a workflow: what it has already committed to, and how to commit more.
Built by resume and threaded through the workflow function as its first
argument, so what a step is stays visible at the call site rather than being
inferred from a decorator. recorded is the checkpoint loaded once at the top of
the pass and kept current as the pass adds to it, so a step reads memory rather
than the store. holder is this pass's claim, and carrying it is what lets a step
write at all: there is no way to record without one.
class-attribute
instance-attribute
¶
async
¶
Run effect once across every pass of this workflow, recording what it returns.
parse is what makes the return type true rather than asserted, and it is
required for that reason. What this hands back is never the object effect
produced: it is what the store holds, read back through the store's
CheckpointCodec, so a step returning a tuple is handed a list under the
default JsonCodec on the very pass that ran it. A cast here would be a lie
on every path rather than only after a crash.
The effect's own return type is deliberately not tied to parse's. What goes
in and what comes out are related by encode-then-decode, which is not the
identity, so requiring one type for both would assert something false. sleep
is the proof rather than the exception: it records an ISO string and reads back
a datetime. A richer codec narrows what a parser has to repair and no codec
removes it, since pydantic_core renders a tuple as a JSON array too: the
codec is a transport concern uniform over every key, and this is a meaning
concern particular to one.
The record is written before the step returns, so the workflow never proceeds
on a result the store has not accepted. When two passes both ran the effect,
the first to record wins and the second is handed the winner's value, so from
there they proceed identically rather than diverging on which capture id is
real. That makes the duplicate harmless downstream rather than preventing it,
which is what the claim is for. Which of the two happened is on the Recorded
and is deliberately ignored: a step has no dependents holding the loser's
value, where run_durably fed its node's result downstream before the write.
Cancellation is what separates those two sentences, and the write is held past
it deliberately. Once effect has returned, the thing it did has happened:
cancelling the write now does not undo the charge, it only removes the record
of it, so the next pass performs it again. And a step is cancelled in the
ordinary course of a fan-out, not only in a crash, since a workflow that spawns
a capture per line item ends its siblings when one of them is declined. Every
sibling that had already called the gateway would be charged twice.
So the write is a task the step shields rather than an ordinary await, and a cancelled step waits for it before unwinding. It is still cancellation: what the caller waits for is one store round trip, which is the same bound the worker's own release has, rather than whatever the cancelled effect was stuck on. The claim is still held while it lands (a release keeps the token, so a write in flight is not fenced by it), which is what makes the record valid.
It waits through repeated cancellation, which is not stubbornness but the
ordinary case. One cancelled pass delivers two: a fan-out gathered under
asyncio.gather cancels its children when the pass is cancelled, and the gather
returns as soon as the first child answers, so the caller's own teardown cancels
the rest a second time. Honouring the second one is dropping a write whose
gateway call has already happened, which is the charge this whole shape exists
to keep. The wait is still bounded by the store, not by the workflow.
async
¶
transact(
key: StepKey, effect: Effect, parse: Parse[T]
) -> T
Perform effect and record it in one commit, so the step is exactly once.
The difference from step is the failure it removes rather than the work it
does. step runs the effect and then writes the record, so a crash in between
leaves the effect done and unrecorded and the next pass repeats it:
at-least-once. Here the store performs the work and writes the record together,
so there is no in-between for a crash to land in.
The price is that effect has to be something the store can perform, which
means it has to live in the store: a Lua script over keys in the same Redis, a
callback over a cursor in the same SQL transaction. An effect that leaves the
datastore (a payment gateway, a carrier) cannot be in the commit, is not a
transaction anyone can offer, and belongs in step with an idempotency key.
That boundary is a fact about distributed transactions rather than a limitation
of this interface, which is why Effect is a type parameter and not a shared
interface.
parse is required for the reason it is on step, and more plainly: what the
effect returns is produced by the store (a Lua script's reply, a cursor's
row), so there is no Python type to infer even before the codec touches it.
async
¶
Wait out duration, across crashes, by suspending until the recorded deadline.
The deadline is what gets recorded, not the duration, which is the whole point: a crash on day two of a three-day wait must not restart the clock. The first pass computes and stores it, every later pass reads it back, and the wait ends when a pass arrives after it.
The deadline is noted on the Run, and noted on the cancellation path too,
because the raise is the part that can go missing. A sleep in one branch of a
task group is cancelled the moment another branch raises, and the write is held
past that cancellation (see step), so the ordinary way to end up with a durable
deadline and no ScheduledWakeup is not a crash but an ordinary fan-out. Left
unnoted, no pass reports it and no driver schedules it: the workflow holds a
deadline that was supposed to end a wait, and waits out a clock that will never
fire.
async
¶
awaiting(key: StepKey, parse: Parse[T]) -> T
The value another process recorded under key, suspending until there is one.
A signal, without a mailbox: whoever has the answer (an HTTP handler taking an approval, a webhook) writes one field into this workflow's checkpoint and asks for another pass. Because the wait is a recorded value rather than a message delivered to a running process, it outlives the process that was waiting and can be satisfied by any other.
This is the value a caller is least able to assume anything about, since it crossed a trust boundary: a step at least chose its own effect, where here the workflow reads what an HTTP handler put there.
claim(key: StepKey) -> None
Reserve key for this pass, refusing a name already used in it.
Two steps sharing a name is the failure this mechanism is most exposed to: the second silently inherits the first's result, and no amount of re-running reveals it. The graph rejects a duplicate node key when the graph is built; the closest thing available here is to reject it the moment the second one is reached, which happens on every pass rather than only after a crash.
Bases: Suspended
The pass is waiting out a deadline it chose itself (Run.sleep).
due is that deadline, and it is not optional, which is the reason this is its own
type rather than a field on Suspended. The two ways of waiting are structurally
different: this one carries a moment to schedule, and InputNeeded carries nothing
because there is nothing to schedule. One class with a nullable due would make
those two states the same shape and leave every consumer to re-derive them, which is
the same reason Sleeping and Waiting are separate on the way back out.
dataclass
¶
The pass stopped at a deadline the workflow chose, and nothing is owed but time.
due is that deadline, read back from the checkpoint rather than recomputed, so a
crash on day two of a three-day wait does not restart the clock. A driver schedules
a wakeup for it.
Bases: Interruption
The pass cannot go further until key is recorded. Nothing has failed.
A control-flow signal wearing an exception's clothes, so a workflow written as
straight-line code can stop in the middle of itself. It is how a suspension travels
through a workflow body, not how it is reported: resume catches these and hands
back an Outcome, so a driver matches over three values rather than catching this.
Error handling around a workflow must let this through: a suspended workflow is
one that is going fine, and unwinding it would undo work it is still counting on.
Being an Interruption is what makes that structural rather than a rule to
remember, and it is why these stay public despite resume absorbing them: a
workflow author needs to know what must not be caught.
Public to name, then, and not to raise. The two ways of waiting are its
subclasses, and each carries what a driver needs to answer it; this base carries
only the fact that a pass stopped, which no driver can act on. resume turns one
raised directly into an ordinary failure of that workflow rather than letting it
through (see there for why that is the kinder of the two).
dataclass
¶
Waiting(key: StepKey)
The pass stopped on a value only something outside it can supply.
There is no deadline here and deliberately none to invent: no clock satisfies this
wait, so a driver schedules nothing and whoever writes the value under key is what
makes the workflow ready again.
async
¶
run_durably(
run: CompiledGraph[*Ins, Out],
checkpointer: Checkpointer,
holder: Pass,
*values: *Ins,
) -> Out
Run run under a claimed workflow, recording each step and resuming from what
is already recorded.
Call it again with a fresh claim on the same workflow after a crash (or a timeout, or a redeploy) and the steps that finished are not re-entered: their results come back from the store, and only what was in flight or unstarted runs. Call it again after a completed run and nothing runs at all, which is what makes the whole call idempotent rather than merely restartable. The inputs are passed positionally every time, because an entry is not part of the checkpoint (it lives wherever the request itself does).
The record is written before the next result is pulled, and stream is
pull-driven, so no step downstream of a completed one starts until that
one's result is durable: the write is a barrier, not a background flush.
Siblings already in flight keep running, which is the point of the fan-out.
A record the store did not take from this pass means another pass recorded
that node first. Unlike stepwise, this cannot simply adopt the winner's
value: the graph handed its own to the node's dependents the moment the node
finished, so the run is already downstream of a value the store rejected, and
the only honest move left is to stop. Holding a claim makes that rare, and it
is Fenced rather than this when the claim has lapsed. Whether the store took
the value is Recorded.first rather than something inferred by comparison,
since a result crosses a CheckpointCodec and a run that won outright can be
handed back something unequal.
With that separated out, the comparison becomes the other check worth making,
and this is the one place able to make it, because it holds both values at once.
A graph feeds a node's result straight to its dependents, so without it they see
a tuple on the pass that computed the node and a list on the pass that
restored it, with no crash needed for the two to disagree. So a node whose
result does not survive its own store fails the run on the pass that wrote it,
naming the node. That is also why a graph needs no per-node parser where
stepwise does: verifying beats parsing when you still hold what you sent.
What that check is, exactly, is a diagnostic and not a repair, and the difference
is worth stating because the failure reads like one. The store took the value
before it could be compared (record is what produces the value to compare
against), so the reshaped result is durable by the time this raises and a later
pass will resume from it and run to completion, feeding dependents the restored
shape with nothing left to complain. The run that discovers it is therefore the
only one that can, which is what makes raising on it worth doing and why the
answer is a codec that carries the value rather than a retry.
A graph gets no transact, and the reason is the graph rather than the store.
Closing the at-least-once gap means making the effect and the record one call,
and a node is an ordinary async function this runner only sees the result of.
A step reaches it because it names its effect at the call site
(run.transact(...)); expressing that here would mean a node type that hands
the graph an effect instead of running one. So a crash between a node's effect
and its record repeats the effect, and the answer for anything leaving the
datastore is the ordinary one: make it idempotent under the workflow id.
A graph whose output is one of its own entries is refused rather than run.
evaluate supports that identity plan, because an entry it was handed is a
value it can return; here the output is read back out of the checkpoint, and an
entry is the one thing a checkpoint never holds (it is fed positionally on
every call, which is why Graph.of keys entries by position). So the run would
record every node correctly and then fail looking for a key that was never
going to be there.
Refuse a duration that is not positive, where the value enters.
Every timing here is an amount of time to let pass, and not one of them has a
meaningful zero: a lease already expired when granted excludes nobody, a poll or a
tick of zero spins, and a blocking read bounded by zero turns the worker's pull
into a busy loop. At the boundary rather than at the point of use, because a duration
assembled from an unset setting (timedelta(seconds=settings.lease_seconds)) is
exactly how a zero arrives.
Two durations are deliberately not run through this, because they are thresholds
rather than intervals and their zero means something: reclaim's idle (take over
anything outstanding, however recently it was delivered) and SQLite's busy_timeout
(do not wait for the write lock at all).
What no check reaches is the bound that decides correctness, that a lease exceed the longest a pass can honestly take. Only the deployment knows that, so this rules out the values that are nonsense rather than certifying the ones that are not.
async
¶
claimed(
checkpointer: Checkpointer,
workflow: str,
lease: timedelta = LEASE,
) -> Pass
Claim workflow, or raise because someone else has it.
The form for a caller that expects to win: a test, or a runner driving a workflow it
owns outright. A worker taking deliveries off a queue wants claim itself, because
losing the race is ordinary there and the answer is to come back later rather than to
fail.
The deadline sleep recorded, or a loud failure if the store holds something else.
async
¶
resume(
holder: Pass,
checkpointer: Checkpointer[Effect],
body: Callable[[Run[Effect]], Awaitable[T]],
*,
now: Callable[[], datetime] = now_utc,
) -> Outcome[T]
Make one pass at a claimed workflow, from whatever it has already recorded.
Call it after a crash, after a wakeup, or after a value it was waiting on arrives:
each call runs body from the top and reaches further than the last, and calling it
on a finished workflow performs no effects at all.
It returns what the pass came to rather than raising when the pass stops short, so
"the workflow finished", "it is waiting out a deadline", and "it is waiting to be
told" arrive as three values a caller matches over. What to do about each is still
the caller's (schedule the wakeup, do nothing until the approval lands, hold the
process open until due), which is the point: this reports, the driver decides.
Only the two suspensions are converted. Anything the workflow's own code raises
propagates untouched, including Fenced and Contended, because losing the
workflow is not an outcome of a pass but a statement that this pass was never
entitled to one.
A Suspended that is neither of the two is the one thing rewritten rather than
passed along, and the reason is what it would otherwise cost. Outcome has no arm
for it and a driver has no way to answer it, so it can only travel outward; and it
travels as an Interruption, which every sensible except Exception in a driver is
built to miss, so one workflow raising it would take down the loop running every
other workflow. Re-raised as an ordinary exception it is what it actually is: that
workflow's mistake, and nobody else's.
It takes a claim rather than making one, for the same reason it returns rather than
raises. Whether to wait for a contended workflow, come back later, or fail is the
driver's call: a worker holding a queue delivery wants one answer and a test driving
a workflow it owns wants another. claimed is the second of those.
An interruption raised inside a task group arrives wrapped, and is unwrapped here
rather than being left to the driver, because wrapped is exactly where the harm is. A
workflow that fans its steps out with asyncio.TaskGroup raises a
BaseExceptionGroup when one of them suspends or loses the claim, and a group whose
leaves are BaseExceptions is not itself an Exception: it is neither an Outcome,
nor something a driver's except (Fenced, Contended) matches, nor something its
except Exception can reach. So the one workflow that fanned out would take down the
loop running every other one, whichever of the two happened to it. Both are unwrapped,
for the same reason and by the same rule.
Losing the claim wins over everything else in the group. It says this pass may not write at all, so a sibling's failure beside it is a consequence rather than a second piece of news, and reporting the sibling instead would tell a driver to log a workflow failure for a workflow that is fine and being advanced by somebody else.
Several branches can suspend in one group, and a pass has one outcome, so a deadline wins over a wait for input. Both are answered eventually (the wakeup fires, and whoever writes the value queues the workflow either way), but only the deadline is answered by this driver: reporting the wait would leave nothing scheduled for a branch that asked for a clock. The earliest deadline wins among several, since a pass that wakes too early suspends again and one that wakes too late has kept a branch waiting for nothing.
A deadline Run.sleep recorded counts even when its own suspension never arrived,
which is what waking is for. A group cancels its remaining branches the instant one
of them raises, so a branch that had just written its deadline can be cancelled
between the write and the raise: the deadline is durable, and without this nothing
reports it, so the driver schedules nothing and a workflow waits out a clock that will
never fire. Reading it off the Run instead of off the exception is what makes the
deadline-wins rule deliverable rather than merely stated.
passes(
durable: Durable,
body: Callable[[Run], Awaitable[object]],
limit: int = POOL,
*,
lease: timedelta = LEASE,
contended: timedelta = CONTENDED,
now: Callable[[], datetime] = now_utc,
) -> Sink[Delivery]
The data plane: up to limit passes at once, and one delivery pulled per free slot.
A Sink because a pass produces nothing another stage consumes; what it produces is
recorded. A pass that raises is logged rather than propagated, since a workflow
failing is this service's data (a gateway declined) and not a bug in the loop that
ran it. What does propagate is a failure of the loop itself, such as the store
refusing an ack.
The pool is limit_concurrency over a lazy mapping of the delivery stream, which
is what keeps "pull one at a time" and "run twenty at a time" the same statement: the
generator that turns a delivery into a pass is only advanced when a slot frees, so
the queue is never read past what this worker can start.
The acknowledgement comes last, on every path this process saw through: a completed
pass, a suspended one, a failed one, and a contended one are all answers. Only
cancellation skips it, which is why it is not in a finally, since a half-run pass
should be reclaimed rather than forgotten. Releasing the claim is attempted on
every path including cancellation, because a shutting-down worker that keeps its
claim makes every other worker wait out the lease for nothing. On that path the
release is best effort: its await is a suspension point inside a task already being
cancelled, so a second cancellation can interrupt it, and nothing is lost when it
does because the claim expires with its lease anyway. That is why it is worth an
attempt and not worth shielding.
Losing the workflow is handled by name rather than falling into the failure arm, and
it has to be, since Fenced and Contended are Interruptions that except
Exception no longer reaches. They say another pass owns the workflow, which is what
a refused claim says too, so the two paths share look_again and neither gets a
warning. That they are still caught here while a suspension is not is the honest
split: a suspension is something the pass did, so it comes back as a value, and
losing the claim means there was no pass to have an outcome.
async
¶
ready(
scheduler: Scheduler,
within: timedelta = BLOCKING,
idle: timedelta = LEASE,
) -> AsyncGenerator[Delivery]
The stream of deliveries this worker should act on: taken over, then new.
A source stream like any other, so everything downstream is ordinary wiring, and
swapping one queue for another changes this function alone. It merges the two
sources because reclaim assigns a dead worker's delivery to this one, which
obliges it to run it.
Every pull answers the same question: is there work someone abandoned, and if not, is there anything new? One of each, never a batch, so a pull is always exactly the one delivery the caller has a slot for. Abandoned work goes first because it has been waiting the longest, and it is bounded work: the pending list is ordinarily empty, so the blocking read is what paces the loop.
idle is deliberately not checked for being positive as the other durations are: it
is a threshold rather than an interval, and a zero one is meaningful (take over
anything outstanding, however recently it was delivered).
The control plane: make every workflow whose deadline has passed ready again.
A Sink over a stream of moments rather than its own timer, so when it runs is the
caller's to decide and this only says what happens each time. Driven off ticks in
work; driven off a list in a test.
Safe to run in every worker, and safe to be killed at any point in it, because the
move is the store's single operation rather than this sink's two (see wake_due).
Whether it does anything is the queue's business. Over a stream beside a sorted set this is what carries a workflow from the sleepers to the queue; over a single structure scored by visibility it spins against a no-op, because being due and being ready are then the same score and nothing has to move. The worker runs it either way rather than asking which queue it has.
async
¶
work(
durable: Durable,
body: Callable[[Run], Awaitable[object]],
*,
tick: timedelta = TICK,
within: timedelta = BLOCKING,
contended: timedelta = CONTENDED,
limit: int = POOL,
now: Callable[[], datetime] = now_utc,
) -> None
Run the worker: the timer alongside the pass loop, until cancelled.
Both halves live for the process, so they are a task group rather than a
foreground and a background: cancelling either (a shutdown, a failed timer) takes
the other down with it instead of leaving a worker that runs passes nobody wakes.
prepare first, because reading a queue takes setup that writing to it does not.
Both halves are also the same shape, which is the point of the vocabulary: a sink over a stream. One consumes deliveries, the other consumes moments, and a deployment that wants a third (trimming a Redis stream, sweeping old checkpoints) adds a task to this group rather than a mechanism.
The lease is the scheduler's rather than an argument here, and it is the one number
that is not a knob on this call. It bounds two things that have to agree (how long
a delivery stays this worker's, and how long its claim on the workflow is good for),
and the queue is where the first one already lives: a visibility-scored store writes
it into the row it takes. Reading it back and claiming for exactly as long is what
keeps a store constructed with a ten-minute lease from being reclaimed after one.
Turning it means PostgresScheduler(pool, lease=...), which is also where the
matching poll and the store's own timings are set, so the passes a deployment can
honestly run are described in one place.