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.loadMUST return them in the order they were first recorded. A workflow's records have two independent writers, the pass throughrecordand anything outside it throughsupply, and neither can order itself against the other: a counter either side keeps is read from a stale snapshot or observed and then raced, so both reach for the same next number and the tie has to be invented. The store is the only thing that sees every write, which makes it the only thing that can say. First-writer-wins already decides what a key holds; this says the same writer decides where it sits, so a losing write moves neither the value nor the position. The order is the guarantee and the number behind it is not: it is adict, which preserves insertion order, so a caller reads the order by iterating and no implementation owes a sequence anyone can see.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. A claim lapses at the earlier of onealivepast its holder's last sign of life and itsbudgetrunning out, so a holder that stops renewing frees the workflow withinalive, and one that renews forever frees it at the budget regardless.extendMUST grantbudgetfrom now and count as a sign of life, and MUST NOT bring a budget already granted forward: what it sets is the later of the two, so a step asking for less than what is left costs nothing and takes nothing away. A claim that has lapsed without being taken is still its holder's to stretch, since nobody else has raised the fence.renewMUST count as a sign of life without touching the budget, and MUST reportFalseonce the budget has run out whatever the token says: a pass past its budget holds nothing however alive it looks, and that report is what ends a hung pass rather than leaving it to renew a claim anybody may take.- Both MUST refuse a holder below the fence, exactly as
recorddoes, and both MUST report whether it still holds the workflow rather than raisingFenced: the caller is deciding what to do about having lost it, whereFencedis the answer to a write it will not get to make. - A sign of life MUST NOT carry a claim past its budget or past a
release. Holding the liveness deadline at or below the budget is the direct way to get both, sincereleasebrings the budget down to now and a write still in flight then renews nothing rather than taking the workflow back. recordandtransactMUST renew the liveness deadline of the pass that wins, and of no other pass: a write refused at the fence MUST leave the holder's deadline where it was, or a superseded pass's stray writes would keep a dead holder's claim alive past the silence that should have freed it. A write is the plainest sign of life there is, so this makes a workflow of ordinary short steps renew itself for free, and leaves the worker's own renewal with the case it is actually needed for, a single step long enough that no write falls inside oneLEASE.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.historyMUST return the recordsloadreturns, in the same order, each carried with the moment the store wrote it. That moment is the store's own clock read at the winning write, so a losing write moves the value, the position, and the time equally not at all.discardMUST remove every record a workflow has, and MUST raise its fence rather than lowering it: aPassoutstanding when it runs MUST find its writesFenced. It returns how many records it removed.appendMUST filevalueunder a key it assigns out of theINBOXspace, MUST assign a distinct key to every concurrent append to one workflow without losing either value, and MUST assign keys that sort into append order within that workflow. They need not be contiguous and need not order across workflows, which is why a shared counter with gaps is a sound implementation and usually the easiest one to make atomic. The entry it writes is an ordinary record: it MUST appear inloadlike any other, since a consumer renders a workflow fromloadand forks one by copying whatloadreturns, and an entry invisible there neither renders nor forks.- 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.
append is supply's sibling, and the only difference between them is who picks the
key. supply is a value from outside a pass under a key the caller names, which is
the shape of an approval or a webhook: one named answer, written once, read by the
Run.awaiting that named it in advance. append is a value under a key the store
names, which is the shape of a stream: a caller with a message and no place to put it
is told where it went. Neither is gated on a claim, for the same reason, since input
must not fail because a worker happens to be mid-pass.
What that buys is a workflow that takes input it did not name in advance, without
inventing a key space and allocating out of it by trying. The store already sees every
write and already decides where each one sits (load's ordering), so it is the only
party that can hand out a name nobody else is about to take.
history is load's other reading, and it is a second method rather than a richer
return because the two have different readers. Every pass calls load at its top and
wants what a step recorded; nothing inside a pass has any use for when a step
recorded it, and a mapping of wrappers on that path would cost a construction per key
per pass to carry a field the runners would immediately drop. What wants the times is
a status view, an operator asking how long a settlement actually took, a sweep
deciding which workflows are old enough to forget: all outside a pass, all reading one
workflow at a time. So the write path stamps every record and the read path splits.
discard and the fence are the reason deletion is a store method rather than a
caller's loop over load's keys. Removing a workflow's records is easy; removing them
safely is not, because a pass in flight holds a Pass and is about to write. Doing
it as a caller can (delete the records, delete the claim) hands the next claim token 1
on the stores whose tokens are a counter, so the pass still holding token 7 outranks
it and writes its remaining steps back into a workflow that has been deleted, one row
at a time and with nothing to show that it happened.
So a discard raises the token and keeps the claim row rather than removing it, which
fences that pass on its next write and leaves the workflow claimable by anything new.
What is left behind is one number per deleted workflow: a Redis pass hash carries a
ttl and expires on its own, and a SQL claim row stays until something sweeps it,
which is the same homework those stores already have (see
docs/without-durability/index.md).
A claim carries two deadlines because one number was being asked two questions it cannot answer at once. A single lease has to exceed the longest a pass can honestly take, or a slow-but-healthy pass is fenced mid-flight and the step it was in the middle of runs twice; and it has to be short, or a dead worker's workflow waits that long before anyone else may touch it. Those pull in opposite directions, and a deployment that sizes for the first gets a takeover latency it never chose.
Splitting them lets each be answerable. alive is how long a sign of life is good for,
so it measures how fast a death is noticed and has nothing to do with how long the work
takes; the pass renews inside it for as long as it is running. budget is how long
this pass may hold the workflow at all, so it measures the work and is what a step
declares when it knows better than the default. Renewal cannot lift budget, which is
what keeps a pass that is hung rather than slow from holding a workflow forever: it
goes on renewing happily until the budget runs out, at which point renew says so
and the worker ends the pass, and the workflow is free either way.
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. deliverMUST appendvalueto the workflow's inbox exactly asCheckpointer.appenddoes, and MUST make the workflow ready, under all three of the clauses above. It returns theEntrythe append wrote.deleteMUST cancel the workflow's wakeups and discard its records, exactly asScheduler.cancelandCheckpointer.discarddo, returning the countdiscardreports. It SHOULD be a single commit where the two stores are one datastore, and where they are not it MUST cancel before it discards.
deliver is to append what arrive is to supply, and it exists for the same
reason rather than for symmetry's sake: a caller that appends and then separately
schedules can die between the two and leave a message nobody will ever wake for. One
call with no ordering to get right is the whole of what this interface adds.
delete is the third of those, and its ordering is the reverse of arrive's for the
reason arrive's is what it is: the two halves fail asymmetrically, and the order
puts the survivable failure in the window. Cancelled-but-not-discarded is a workflow
holding records nothing will wake, which asking again finishes. Discarded-but-not-
cancelled is a wakeup for a workflow with nothing recorded, which a worker answers by
running the body from the top: the deleted workflow starts over, performing every
effect again. So the wakeup goes first, and a crash in the window costs a second call
rather than a second run.
What no ordering reaches is a pass that is already in flight, and that is the
Checkpointer's half rather than this one's: discard raises the fence, so the pass
is refused at its next write, and Scheduler.wake_at declines to queue a workflow
whose delivery was cancelled underneath it. Both are stated as requirements there.
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.
dataclass
¶
One value delivered to a workflow's inbox, and the key the store filed it under.
Both halves, because the key is the part the caller could not have known: it is the store that assigns it, and it is what a workflow records to say how far it has read. A pass carries the key rather than a copy of the value for the same reason the entry is never consumed: entries are immutable and first-writer-wins, so "I took entries 7 and 8" replays to the same two values, where a copy would be a second thing to keep in step with the first.
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.
Fenced and Contended may be caught by name, by a driver deciding what to do about
losing a workflow. Suspended may not be caught at all, by anyone: a pass that
handles one and carries on is claiming a wait was answered when it was not, and
resume refuses it (see Swallowed).
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;cancelremoves every wakeup the store holds for a workflow, so nothing it has cancelled reaches a laternext_readyorreclaimuntil something makes the workflow ready again;wake_atMUST NOT reinstate a workflow whose wakeups have been cancelled since its delivery was taken;extendkeeps a delivery its taker's for anotherwithinfrom now, and returns the delivery to use from then on. It MUST be silent about one the store no longer holds, returning it unchanged, since a delivery taken over or cancelled underneath a worker is not that worker's to renew and not an error to have tried.
The last two are one requirement seen from both ends, and the second half is what
makes the first half worth anything. A worker holding a delivery answers for it
after the pass, so a cancel that swept the queue clean is undone a moment later by
the wake_at that pass reaches: the workflow whose records have just been discarded
is queued again, wakes with nothing recorded, and runs from the top. Which is the
resurrection deleting a workflow is supposed to rule out, arriving through the one
door a queue sweep cannot close. So it is closed at the other end, where the store
still holds the receipt and can tell a live delivery from a cancelled one.
The overwrite clause 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 gives the workflow's claim exactly as long to live. 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.
It is the liveness window that has to match, not the pass's budget, and the pairing
is what extend exists for. Both answer the one question a second worker asks, whether
the holder is still there, so the worker renewing its claim renews its delivery on the
same tick and the two lapse together. A budget is the other question, how long this
pass may run, which only the checkpoint store is asked: a two-hour step that keeps
renewing holds its delivery two hours by renewing it, not by having said two hours up
front, so a worker that dies inside one still loses both within a lease.
That extend hands back a delivery rather than nothing is a fact about these queues
rather than ceremony. Three of the four here make the visibility a delivery is taken
under be its receipt, which is what lets wake_at tell its own delivery from a wakeup
that arrived since; moving the visibility therefore renames the delivery, and a worker
still holding the old name would find its done silently declined and the workflow
redelivered for nothing. So the store says what the delivery is called now, the same
way record says who won: it is the party that knows.
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.
That repair is weaker for deliver than for arrive, and the difference is the key.
A resubmitted arrive names the key it named before, so first-writer-wins makes it
the same write and the retry costs nothing; a resent message has no key to name, so
it appends a second entry beside the first and the workflow reads both. A caller
that cannot tolerate that wants a store whose deliver is one commit.
delete runs the other way round, and the reversal is the same reasoning rather than
an exception to it: the order puts the recoverable failure in the crash window. Here
that is a workflow left with records nothing will wake, which asking again finishes,
against a wakeup for a workflow with nothing recorded, which runs it from the top.
dataclass
¶
One record as history reads it back: what the store holds, and when it landed there.
at is the store's own clock read at the write, which is the same clock every lease
here is measured by and for the same reason: the writer is a different machine, and a
moment stamped by whichever process happened to record it is only as good as the
agreement between the two. So the times across one workflow's records are comparable
with each other, which is what a duration between two steps is read off, and are
not comparable with a datetime.now() taken here.
It is the moment of the winning write, since first-writer-wins decides what a key holds and where it sits, and a time that moved under a losing write would say a step ran when it was merely replayed.
dataclass
¶
Who holds a workflow, until when, and what a word from them is worth.
One record rather than a mapping per field, for the reason Stored is one: four
mappings kept in step are three states that can disagree, and this cannot. It is the
claim row the SQL stores keep, with the column names spelled out.
alive_until is when the claim lapses if nothing more is heard and is what claim
tests; held_until is the budget it can never pass. Holding the first at or below the
second is what makes a renewal unable to do the two things a renewal must not: carry a
pass beyond its budget, or take a workflow back after its holder released it.
alive travels with the claim because a write is a sign of life and record is not
told how long one should count for. Taking it from the claim rather than from a
store-wide setting is what keeps a workflow claimed with a short liveness window from
quietly renewing itself on a long one.
Both moments are monotonic, as every lease here is, so a clock that steps cannot
hand a workflow to two writers.
A copy of the claim, with a sign of life noted at at and good for alive past it.
dataclass
¶
MemoryCheckpointer(
hashes: dict[str, dict[str, Stored]] = dict(),
claims: dict[str, Claim] = dict(),
codec: CheckpointCodec[str] = JSON,
now: Callable[[], datetime] = now_utc,
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
¶
wrote(holder: Pass) -> None
Note that this pass wrote, which is the plainest sign of life a claim can get.
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.
async
¶
File value under the next key in this workflow's inbox.
The position is how many records the workflow already has, which is the same
number the Redis store takes from HLEN and for the same reasons: nothing here
ever removes a record and first-writer-wins means nothing is ever replaced, so the
count only rises and a key it yields cannot already be taken. Being synchronous
between its awaits is what makes it atomic, which is this store's version of a
Lua script.
It counts every record rather than only the inbox's, so a workflow's entries are numbered with gaps wherever a step was recorded between two appends. That is exactly what the interface allows: the keys sort into append order, which is the whole of what a consumer reads them for.
async
¶
Forget every record this workflow has, and raise its fence so a live pass cannot write more.
The token is taken up rather than removed, which is the whole of what makes this
safe against a pass in flight: that pass keeps its Pass and believes it still
owns the workflow, and the number it carries is now below the fence, so its next
record raises Fenced. Deleting the token instead would hand the next claim a 1
that a pass holding 7 outranks, and the deleted workflow would fill back up.
The lease goes with it, so the workflow is claimable again immediately: what is being kept is the ordering, not the claim.
Only a workflow that has a token gets one, so discarding an id nobody has claimed
writes nothing: a Pass exists only because claim wrote a token, so where there
is no token there is no pass to fence, and minting one would leave a tombstone for
a workflow that never ran.
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.
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
async
¶
Keep this delivery its taker's for another within, by restarting its idle clock.
Silent about a delivery the store no longer holds, which is the requirement and not merely tolerance: a worker renewing on a tick has no way to know it was reclaimed or cancelled a moment ago, and putting it back would be this store handing out a delivery nobody is waiting on.
What this store keeps is the moment a delivery was taken, and reclaim measures
idle back from now against it, so keeping one for another within means putting
that moment where one lease past it lands within from now. For the within the
worker actually passes, its own lease, that is exactly "freshly taken"; spelling
out the arithmetic is what makes a test that passes something else get what it
asked for rather than the window this store happens to run on.
The receipt does not change here, which is the one place this double is weaker than what it stands in for rather than equal to it: a receipt is a counter, where three of the four real schedulers make the visibility the receipt and so rename a delivery whenever they move it. A worker that wrongly held on to the old name would pass against this store and fail against those, which is what the cross-store suite is for.
async
¶
cancel(workflow: str) -> None
Drop every wakeup this store holds for the workflow, whichever structure it is in.
All three, because a workflow can be in any of them and the caller has no way to
know which: waiting in the queue, waiting on a clock, or out with a worker that
has not answered for it yet. That last one is why wake_at checks outstanding,
since dropping the delivery here is what tells the pass still running that its
workflow is gone.
dataclass
¶
One record as this store keeps it: the encoding, and when the winning write landed.
A value per record rather than a second mapping beside hashes, because two mappings
kept in step are a state that can be wrong and this cannot: a key either has a record
or has none, and the record carries its own time. It is the row the SQL stores keep,
with the column names spelled out.
dataclass
¶
The pass stopped on the outside world, and this is everything that would move it.
There is no deadline here and deliberately none to invent: no clock satisfies any of these, so a driver schedules nothing and the next write is what makes the workflow ready again. Which write is the whole content of this type, and the reason it holds two sets rather than one:
waitingare addresses, fromRun.awaiting. A client holding one answers witharrive(workflow, key, value).listeningname the read steps that stopped, fromRun.receive. Nobody writes to those keys, and the answer isdeliver(workflow, value), addressed to the workflow rather than to any key.
Two fields rather than two types, because a driver's response to both is identical (acknowledge the delivery, schedule nothing) and a pass can be stopped on both at once. A type per kind forced a pass blocked on an approval and an inbox to report one and discard the other, and a single set would have left a key that is sometimes somewhere to write and sometimes a diagnostic. Naming the two collections keeps that distinction exactly where it is load-bearing while letting a pass say both.
Reporting all of them rather than one is the point. A fan-out that suspends in several branches is blocked on every one of them, so a client asking what would advance this workflow needs the set, and picking a representative made the answer both incomplete and unstable, since which branch reached its raise first decided it.
A pass is never Blocked on nothing, which is checked rather than documented: an
empty one would say a workflow stopped for no reason, and a driver reading it would
park a workflow that nothing will ever wake.
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.
MessageNeeded(key: StepKey)
Bases: Suspended
The pass is waiting on the workflow's inbox having something new in it (Run.receive).
Nobody schedules a wakeup for this either, and for InputNeeded's reason: no clock
satisfies it, and whoever delivers the next message is what makes the workflow ready.
What differs is how it is answered. InputNeeded's key is an address, so a client
holding it calls arrive(workflow, key, value); this one's key names the read step
that stopped, which nobody writes to, and the answer is deliver(workflow, value).
Sharing a type would mean a key that is sometimes somewhere to write and sometimes a
diagnostic, which is what keeps them apart here and what Blocked keeps apart on the
way back out, in two fields rather than two types.
dataclass
¶
Run(
holder: Pass,
checkpointer: Checkpointer[Effect],
recorded: dict[StepKey, object],
extend: Extend,
now: Callable[[], datetime] = now_utc,
claimed: set[StepKey] = set(),
reached: list[Suspended] = list(),
)
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.
extend is how a step buys more time than that claim was granted, and it has no
default because a silent one would be worse than none: a body annotating its slow steps
with within and getting no extension would read as working and behave exactly as it
did before, one fenced pass per slow step. resume builds the ordinary one.
async
¶
step(
key: StepKey,
effect: Callable[[], Awaitable[object]],
parse: Parse[T],
*,
within: timedelta | None = None,
) -> T
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.
within is how long this step says it may honestly take, and giving it turns the
one number a deployment had to guess for every workflow at once into a statement by
the code that knows. Without it the step spends whatever the pass was claimed for,
which is the deployment's default and is the right answer for the steps that are
over in milliseconds. With it the claim is stretched to cover the step before the
effect runs, so a gateway that takes four minutes is not a pass fenced at one and a
charge made twice.
It is a bound and not a timeout: nothing here stops an effect that outruns it. What it buys is the right to still be holding the workflow when the effect returns, and what it costs is that a step which hangs holds the workflow for this long before anything else may take it.
async
¶
perform(
key: StepKey,
effect: Callable[[], Awaitable[object]],
parse: Parse[T],
*,
within: timedelta | None = None,
) -> T
step once the name has been claimed: the lookup, the effect, and the write.
Split out for receive, which has to claim its key before it knows whether it
will run an effect at all, since a pass that suspends on an empty inbox should
still have reported a duplicate step name. Claiming inside here instead would
make that either a double claim or no claim.
The recorded check comes first, so a pass replaying fifty finished steps to reach an unfinished one pays for none of their budgets: a step that is not going to run needs no time to run in.
async
¶
covered(within: timedelta | None) -> None
Stretch the claim to cover a step that named a window, or lose the pass trying.
Nothing at all for a step that named none, so the steps that are over in milliseconds pay nothing for the mechanism. A zero is the one value that would quietly do the opposite of what it says: a budget already spent lapses the claim on the spot, so the step that asked for more time gets none at all.
async
¶
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.
within is step's, and means the same thing: the store is running this work, so
the claim has to outlive it exactly as it would an effect of the caller's own.
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. This is the sharpest case of the rule every wait here follows: what a pass
reports is what it reached, not what happened to propagate out of it.
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.
async
¶
receive(
key: StepKey,
*,
after: StepKey | None = None,
limit: int | None = None,
) -> tuple[Entry, ...]
The entries delivered to this workflow after after, suspending until there are any.
awaiting's role over a stream. Where that waits on one named value written once,
this waits on a log the outside world appends to (Durable.deliver), and the
workflow reads it with a cursor of its own rather than naming each message in
advance.
after is that cursor, and it is the caller's to carry: pass the key of the last
entry you took, and the next call picks up behind it. Threading it explicitly is
what makes two independent readers inside one workflow work without a rule about
it, and what keeps the cursor a value rather than hidden state on the Run.
It never returns empty. With nothing new it raises MessageNeeded, so the pass
ends Blocked with this key among its listening and whoever delivers next
re-queues the workflow, which means received[-1].key is total and threading the
cursor needs no branch. pending is
the variant for a caller that wants whatever is there and would rather carry on.
limit bounds the take, and is how a workflow consumes part of what has
arrived: a consumer that treats the first new message as opening a unit of work
and everything behind it as belonging to that unit cannot advance its cursor past
the lot, so it takes one and leaves the rest for the next call.
Reading the inbox is a step, and for the reason everything else here is one: a live read of a log somebody is still writing to gives two passes different answers. What gets recorded is the key of the last entry taken, so a replay hands back the same entries rather than whatever has arrived since. It is a reference rather than a copy, which is sound precisely because entries are immutable: the keys it names still hold what they held.
No store round trip is needed to read them, either. Entries are ordinary records,
so they are already in recorded, which is the snapshot this pass loaded at the
top. An entry appended mid-pass is therefore invisible to this pass, which is
correct rather than a limitation: the append made the workflow ready, so the next
pass sees it.
async
¶
pending(
key: StepKey,
*,
after: StepKey | None = None,
limit: int | None = None,
) -> tuple[Entry, ...]
The entries delivered after after, or nothing at all, without suspending.
receive for a workflow that wants to fold in anything waiting and carry on
regardless: a long-running unit of work checking for a cancellation, a reducer
draining what has piled up since its last turn.
It records how far it read exactly as receive does, including when it read
nothing, and that write is the point rather than bookkeeping. Left unrecorded, a
replay would re-evaluate against a fuller inbox and hand this pass entries the
first one never saw, which is the divergence the whole step mechanism exists to
prevent.
The inbox past after, as this pass sees it, in the order the entries were appended.
Two guarantees carry the order, and both are the store's. The records arrive from
load in the order they were first recorded, which is what puts these in append
order among everything else the workflow has done; and append mints keys that
sort, which is what makes the > against a cursor mean "later than". A store
meeting one and not the other is a store this reads wrongly, which is why the
conformance suite pins both.
Write a suspension down on the pass, and hand it back to be raised.
raise self.note(InputNeeded(key)) rather than a bare raise, so that noting
cannot drift out of step with raising: the two are one expression, at the one
place that knows the pass stopped. What the pass reports is then built from what
it reached, which is what makes the report independent of whatever combinator
the workflow wrapped its waits in. See reached.
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.
The INBOX prefix is refused here for the same reason and at the same moment. A
step named inbox:3 would be read back by receive as a message somebody
delivered, silently, and no amount of re-running would reveal it. The store owns
that key space, so a workflow reaching into it is a collision worth failing on
rather than a naming style.
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 ways of waiting are structurally
different: this one carries a moment to schedule, and the other two carry nothing
because there is nothing to schedule. One class with a nullable due would make
those states the same shape and leave every consumer to re-derive them, which is the
same reason Sleeping and Blocked 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.
Must not be caught at all, which is narrower than it once was here and is now
enforced rather than asked for. A workflow that catches one and returns anyway is
saying a wait was answered when it was not, and resume refuses that outcome
(Swallowed) instead of reporting a finished workflow that is still waiting on the
world. BaseException stops an except Exception from absorbing one by accident; it
cannot stop asyncio.wait or gather(return_exceptions=True), which capture
exceptions as values by design, so the check is what covers those.
Public to name, then, and not to raise. The 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).
Bases: Exception
The workflow caught a suspension and carried on, so the pass cannot be believed.
A pass that reached a wait and then returned a value has had one of its suspensions
handled by the workflow's own code: an asyncio.wait or a gather(return_exceptions=
True) that captured it as a value, or an except that named it. Reporting
Completed there would mark a workflow finished while it is still waiting on the
world, which is unrecoverable in the quietest possible way, since nothing wakes a
finished workflow and no record says a wait went unanswered.
An ordinary Exception rather than an Interruption, because that is what it is: the
workflow's own bug, caught by a driver's except Exception and logged against that
workflow, where an Interruption would say something about this pass's right to run
and slip past every driver written to handle failures.
What it costs is that a suspension can no longer be handled at all, only named. There
is no awaiting that returns a default, so a workflow wanting "carry on if it is not
there yet" has to be written another way; Run.pending is that shape for the inbox.
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 budget exceed the
longest a step can honestly take. Only the deployment knows that, and only for the
steps that named no within of their own, so this rules out the values that are
nonsense rather than certifying the ones that are not.
async
¶
claimed(
checkpointer: Checkpointer,
workflow: str,
budget: timedelta = BUDGET,
alive: 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 key an entry at position in a workflow's inbox is stored under.
extending(
checkpointer: Checkpointer,
granted: timedelta | None = None,
alive: timedelta | None = None,
now: Callable[[], datetime] = now_utc,
) -> Extend
An Extend over one claim, skipping the round trip when that claim already covers it.
Built per pass, because what it remembers is that pass's claim: how long the store last
granted and when. A step asking for less than what is left goes through untouched, so a
body can annotate every step with a within and pay for none of the fast ones.
granted is what the claim was taken for, and it is what the first skip is measured
against. Left unset, nothing is assumed: the first window a pass names is always bought
from the store, and only what the store has granted since is ever skipped over. That is
the honest default for a caller that did not make the claim itself, since a Pass
carries no budget and a guess that overstates it is a step whose extension is skipped
as already covered and fenced after its effect ran.
alive is what one sign of life is worth, and it is the caller's to state because it
is the caller's tick: a worker renewing a few times per lease passes its lease.
Left unset, the caller has no tick, and each window it buys is its own sign of life,
good for the whole of the step it covers, because nothing will speak for the pass
again before the write that ends that step. Every store holds the liveness deadline at
or below the budget, so a window bought under a tick shorter than the step and never
renewed would lapse partway through it however long the budget said.
The elapsed time is a difference between two reads of the same clock, which is what makes it safe to compare against a budget the store granted by its own. An offset between the two clocks cancels; only a difference in their rate survives, and that is small enough to ignore where an offset is not. Neither is trusted very far: the margin is a whole liveness window, so nothing is skipped unless it fits with a window to spare, and a clock wrong enough to defeat that costs a fenced pass rather than a lost one.
Skipping leaves a longer grant standing, so the cap a pass is running under is the
largest budget it has asked for recently rather than the current step's exactly. That
is the right way round: it can only ever be too generous, and being too tight is what
fences a pass that was doing nothing wrong. The store keeps the same promise from its
side, since extend never brings a budget forward.
How far a read got, or a loud failure if the store holds something else.
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]],
*,
extend: Extend | None = None,
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 on the
outside world" 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 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 on the outside world. All are answered eventually (the wakeup fires,
and whoever writes queues the workflow either way), but only the deadline is answered
by this driver: reporting the waits 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. That is the one place information is still dropped, and it is bounded: the
keys a Sleeping pass was also blocked on are not reported, and are reached again by
the pass the wakeup produces.
Waits on the outside world do not choose between themselves. Blocked carries every
one of them, because a fan-out is blocked on all of them at once and because choosing
was unstable as well as lossy: the winner was whichever branch reached its raise
first, so two passes at one suspended workflow could name different keys on scheduling
alone.
A suspension counts even when it never arrived, which is what Run.reached is for and
why every arm below reads it. Three things lose one on the way out: a group cancels its
remaining branches the instant one raises, so a branch that had just written its
deadline can be cancelled between the write and the raise; asyncio.gather propagates
only the first exception, so a fan-out's other suspensions never get here; and a
combinator that captures exceptions as values propagates none. Building the report from
what the pass reached rather than from what came out is what makes it true regardless
of which of those the workflow used.
The third one is also the reason a body that returned is not automatically
Completed. Returning normally having reached a suspension means something caught one,
and a pass reporting Completed there would mark a workflow finished that is still
waiting on the world, so it is refused as the workflow's own error (Swallowed). That
is a deliberate narrowing of what a workflow may do with a Suspended: they may be
named, and they may not be handled.
extend defaults to one over this checkpointer alone, which is the whole of what a
caller driving its own workflow needs: there is no delivery to keep alive beside the
claim, and no tick renewing it, so every window a step names is bought outright and
counts as a sign of life for as long as the step. A worker passes its own, because it
holds both and a claim renewed without its delivery is a pass another worker is about
to be handed a wakeup for. What the default cannot know is the budget this pass was
claimed for, so it assumes nothing and buys the first window it is asked for; a caller
that wants that round trip skipped builds its own with extending.
passes(
durable: Durable,
body: Callable[[Run], Awaitable[object]],
limit: int = POOL,
*,
lease: timedelta = LEASE,
budget: timedelta = BUDGET,
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,
budget: timedelta = BUDGET,
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 outlives the
last word from it), 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 renewing
both on that window 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.
budget is the other number and is deliberately not the scheduler's, because the
queue has no opinion about it. It caps how long one pass may hold a workflow however
alive it looks, so what it has to exceed is the longest a step can honestly take, where
the lease has to exceed nothing at all and is sized for how fast a dead worker should
be noticed. A step that knows better than this default says so with
Run.step(..., within=...), so this only has to cover the ones nobody annotated.