without_durability_redis¶
A without-durability checkpoint store and queue backed by Redis, with each guarantee enforced by a Lua script.
dataclass
¶
LuaEffect(
source: str,
keys: tuple[str, ...] = (),
args: tuple[str | int | float | bytes, ...] = (),
)
A piece of work this Redis can do, written as the Lua it would be on its own.
The Effect type for RedisCheckpointer. source is an ordinary script body: it
reads KEYS and ARGV from index 1 as if it were the only thing running, because
transact splices it into a wrapper that supplies the fence check and the record and
rebinds those two tables. It MUST return whatever the store's CheckpointCodec
decodes, since what it returns is what the checkpoint comes to hold; under the
default JsonCodec that means JSON text, and cjson.encode is the usual way to
produce it. The wrapper puts the field's position in front of that on the way in and
takes it off on the way out, so an effect neither writes a prefix nor ever sees one.
The codec is the one place an effect has to know which one its store was
built with, and it is unavoidable, because the encoding happens in the server where
the Python codec cannot reach.
Its keys MUST hash to the workflow's own slot, which on a single node is free and
on a cluster means carrying the same {id} tag. That is not a quirk of the wrapper:
it is what "the same datastore" reduces to once the datastore is partitioned, and a
script spanning two slots is a distributed transaction wearing a local disguise.
dataclass
¶
RedisCheckpointer(
redis: Redis,
namespace: str = "workflow",
ttl: timedelta = timedelta(days=1),
codec: CheckpointCodec[str] = JSON,
)
A workflow's completed steps as one Redis hash, and its claim as another.
The client MUST be built with decode_responses=True. That is this app's choice to
make (it owns both ends of this hash), and making it once here is what keeps every
read from carrying a bytes-or-text branch it would never take. It is also what fixes
codec to a CheckpointCodec[str]: a hash field can hold bytes, but a client
decoding every reply has already decided this store speaks text.
codec is how a step's result becomes the encoding a hash field carries and comes
back, and it defaults
to the stdlib's JSON. Change it to widen what a step may return or to speed the
encoding up; what it MUST keep is the round trip, since a resumed pass reads what it
produced. A LuaEffect under transact has to agree with it, which is the one thing
the type cannot check, because that encoding happens in the server.
namespace keeps the workflow keys clear of whatever else shares the database, and
ttl is the answer to the question a checkpoint store cannot dodge: these records
outlive the process that wrote them, so something has to decide when a workflow is
beyond resuming. Setting it on the hash rather than sweeping is what lets a finished
or abandoned workflow expire on its own.
It is re-armed only on a write, which makes it a bound on how long a workflow may
wait as much as on how long a finished one is kept: a workflow suspended for longer
than ttl writes nothing meanwhile, so its checkpoint expires while its entry in the
sleeping set (which carries no expiry) survives, and the wakeup it eventually gets
finds nothing recorded. So ttl MUST exceed the longest sleep or approval any
workflow using this store can sit in.
How durable a write actually is stops at what the server is configured for. It
returns when Redis has accepted the write, which with the default snapshotting and
asynchronous replication is not the same as surviving a failover, and nothing here
asks for more with WAIT. run_durably's reasoning about the window between an
effect and its record assumes that gap is closed; closing it is this store's job, not
the runner's.
What a workflow id has to be¶
A workflow id becomes key structure here rather than data, which is what gives it any constraints at all. They are not checked at run time, deliberately: the ordinary id is a UUID or a ULID and satisfies all of this without anyone thinking about it, so paying for a validation on every call to catch a caller who went out of their way would be the wrong trade. Enforce it where ids are minted if you need to.
- It MUST NOT contain
{or}. Those delimit the cluster hash tag, so an id carrying its own braces makes Redis take some prefix of it as the tag instead of the whole id. Both of a workflow's keys still agree on that prefix, so nothing breaks, but the slot is then chosen by an arbitrary fragment and keys stop spreading evenly across a cluster. - It SHOULD be bounded in length. Redis keys are held in memory and an id appears in
two of them per workflow, plus any key an effect derives from
hash_key.
Both of those are about this store, and neither applies to a scheduler here, which
holds an id as a stream field or a sorted-set member rather than in a key name. A SQL
store binds it as a query parameter and so asks nothing of it at all, which is the
tell: this is a property of building keys by interpolation, not of workflow ids. And
nothing in without-durability derives one id from another, so these two are the
whole list rather than the part of it one store happens to care about.
class-attribute
instance-attribute
¶
take: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
stretch: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
still_here: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
write: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
offer: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
file_away: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
forget: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
hand_back: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
transactions: dict[str, AsyncScript] = field(
default_factory=dict,
init=False,
repr=False,
compare=False,
)
async
¶
Every step this workflow has recorded, in the order they were first recorded.
async
¶
The same records, each with the moment the server stamped it as it was written.
transaction(source: str) -> AsyncScript
The wrapper script for one effect body, spliced and digested once.
It cannot be built at construction, because it is the effect that decides the body. What it can do is build each one only the first time it sees it: the splice and the SHA are pure functions of the source, and an application's effects are written in its source rather than derived from a request, so the set is small.
async
¶
Run effect and record it as key, in one script, so the step happens once.
Whatever the effect returns becomes the step's encoding, so it has to already be in the shape this store's codec reads back (JSON text by default). The encoding happens in the server, which is exactly why it cannot be the codec's job. The script puts the field's position in front of it on the way in and takes it off on the way out, so an effect neither writes that prefix nor sees one.
async
¶
File value in this workflow's inbox, under the field the script mints for it.
The encoding goes out and comes back rather than being round-tripped through the
store, which is the one place this differs from supply and is sound for the
reason supply's read-back is not: the field is known absent inside the script,
so there is no earlier writer whose value could be what is stored. Decoding what
was just encoded still runs, since that is what every other write here promises
and what makes a value that cannot survive its own codec fail on the way in.
async
¶
Delete this workflow's steps hash, and raise the token in its pass hash above anyone still holding one.
What is left behind is the pass hash, carrying a token and nothing else, and it
goes away on its own: every write here re-arms ttl, so a workflow nothing writes
to again is forgotten entirely once the expiry elapses. That is this store's answer
to the tombstone the SQL stores leave for a sweep, and it is the same expiry that
already collects a finished workflow.
dataclass
¶
RedisSetScheduler(
redis: Redis,
namespace: str = "workflow",
lease: timedelta = LEASE,
poll: timedelta = POLL,
now: Callable[[], datetime] = now_utc,
)
Scheduler as a single sorted set scored by when each workflow becomes visible.
A drop-in for RedisStreamScheduler: the same protocol, the same worker, one structure
instead of two. Like the other Redis stores here, the client MUST be built with
decode_responses=True.
Two methods do nothing, and that is the finding rather than an omission. prepare
has nothing to create, because a sorted set needs no consumer group. wake_due has
nothing to move, because being due and being ready are the same score. reclaim
likewise returns nothing: an abandoned workflow is picked up by next_ready along
with everything else, since its lease elapsing is indistinguishable from a deadline
arriving, and treating them the same is the point.
class-attribute
instance-attribute
¶
take: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
finish: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
suspend: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
keep: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
async
¶
make_ready(workflow: str) -> None
Make the workflow visible now, whatever it was waiting for before.
A plain write rather than a conditional one, including over a pass in flight. Landing on top of a running pass's score is what keeps the wakeup alive (that pass will now decline to remove the entry), and landing on top of a deadline is correct too: the workflow wakes, finds its wait unfinished, and reschedules itself.
async
¶
Suspend the workflow until when, unless something asked for a pass meanwhile.
The receipt is the score this pass took, so anything that rescheduled the workflow
since (a confirmation, another worker taking over an overrun) wrote a different
one and this leaves it be. Which is the right answer rather than a concession: the
deadline lives in the workflow's checkpoint, so the pass that runs sooner reaches
the same sleep and writes it again.
async
¶
Nothing to do: a workflow whose score has passed is already visible.
async
¶
The next visible workflow, waiting up to within for one to appear.
Polling, because a sorted set has no blocking read. within bounds how long a
cancelled worker sits here before it can notice, exactly as the blocking read's
argument did, but here it is also spent in round trips rather than in one parked
call, which is the cost of this whole design.
async
¶
Nothing to take over by hand: an abandoned workflow becomes visible on its own.
async
¶
Push this delivery's invisibility out, and say what it is called now.
The new score is the new receipt, since this set's receipt is its score, so the caller is handed a delivery to use from here rather than left to discover that the one it holds has been renamed.
Nothing back means this delivery is no longer this worker's: cancelled, or
rescheduled by a wakeup that arrived mid-pass, either of which wrote a score that is
not the one it took. The answer to both is to hand back what came in, exactly as
wake_at and done already do.
async
¶
cancel(workflow: str) -> None
Drop the workflow's entry, whichever of the three things its score currently means.
One ZREM covers queued, sleeping, and out with a worker, because this set holds
one entry per workflow and the score is the only thing that differs between them.
That is the same collapse that leaves wake_due and reclaim with nothing to do.
The half of cancel that a queue sweep cannot reach comes free here as well: a
pass still in flight answers with wake_at, which is conditional on the score
still being the one it took, and a removed entry has no score at all. So the
deadline it was about to write is declined and a deleted workflow is not put back
among the sleepers.
async
¶
done(delivery: Delivery) -> None
Drop the workflow, unless something asked for another pass while this one ran.
The receipt is the score this pass took, so anything that rescheduled the
workflow meanwhile (a confirmation, this pass's own wake_at, another worker
taking over an overrun) wrote a different one and this leaves it alone. That is
why a worker may call wake_at and then done in that order without the second
undoing the first.
dataclass
¶
RedisStreamScheduler(
redis: Redis,
namespace: str = "workflow",
group: str = "workers",
batch: int = 100,
consumer: str = (lambda: hex)(),
lease: timedelta = LEASE,
scan: int = 500,
scanned: list[str] = (lambda: ["0-0"])(),
)
Scheduler as one Redis stream (with a consumer group) and one sorted set.
Like RedisCheckpointer, the client MUST be built with decode_responses=True:
this app owns both ends of the queue, so it decides once here rather than every
read deciding again.
next_ready blocks in Redis rather than polling, so a worker with nothing to do
costs nothing and a submitted order is picked up the instant it is appended. Its
within bound is not a poll interval but a shutdown one: it caps how long a
cancelled worker sits in a blocking read before it can notice.
Every worker reads the same group under its own consumer name, which is how the
work distributes: the group hands each entry to exactly one of them, so scaling out
is starting another process rather than partitioning anything. A long-lived
deployment would name consumers after the host and process (and retire dead ones with
XGROUP DELCONSUMER) rather than minting one per instance as this does.
XACK clears an entry from the pending list but leaves it in the stream, so the thing
that bounds this queue is trim, run as its own control-plane task beside the worker
(see trimming). Without it the stream is correct and grows forever.
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
move: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
suspend: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
hold: AsyncScript = field(
init=False, repr=False, compare=False
)
class-attribute
instance-attribute
¶
async
¶
Create the consumer group, which every reader needs and no writer does.
From 0 rather than $, so an order submitted before any worker existed is
delivered rather than stranded: the group starts at the beginning of the stream
instead of at whatever happened to be its end when the first worker booted.
async
¶
Put the workflow among the sleepers, and answer for the delivery that got it there.
Nothing to compare about freshness, which is the sorted set's problem rather than the stream's: a wakeup that arrived mid-pass is a new entry in the stream, so writing a deadline into the sleepers cannot overwrite it and acknowledging this delivery cannot remove it.
What is compared is whether the delivery still exists, which is what a cancel
needs and what makes this a script rather than the two commands it reads as. See
SUSPEND.
async
¶
Take over one delivery a worker has been holding without acknowledging.
One, because a worker should never hold more than it is about to work on: taking a batch would mean owing several passes while running one, which is the thing pulling one at a time exists to avoid. A backlog of abandoned work is drained the same way any other work is, one free slot at a time.
idle is a lease: too short and a slow pass is overtaken while it is still
running, too long and a crashed worker's workflow waits that long to be picked
up. Overtaking is survivable and not free, so the bound should exceed how long a
pass can honestly take.
The cursor is kept rather than discarded or exhausted, because XAUTOCLAIM
bounds its own work: it scans about ten times count pending entries per call and
then stops, handing back where it got to. Both of the obvious ways to spend that
are wrong, in opposite directions.
Starting from 0-0 every time gives up after ten entries and reports "nothing
abandoned" while a dead worker's deliveries sit behind them, which is not a rare
arrangement but one this worker makes for itself: a pool of twenty holds twenty
entries whose idle clocks it keeps resetting, so the abandoned ones are exactly
the entries furthest down the list. Walking the cursor to the end within one call
finds them, and costs a full sweep of the pending list on the path that always
runs: with a fleet holding two thousand entries in flight, that is a fifth of a
second of round trips before every single pull, and it worsens as the fleet grows.
One step per call, resumed from where the last one stopped, is what both of those miss. Each pull costs a single round trip, and successive pulls sweep the whole pending list and wrap around, so an abandoned delivery is found within one sweep rather than immediately or never. Holding a cursor makes this scheduler stateful in a way nothing else here is, and the state is a hint rather than a fact: losing it (a restart, a second scheduler over the same group) costs a sweep, not a delivery.
async
¶
Reset this delivery's idle clock, so reclaim stops counting it as abandoned.
The only scheduler here whose delivery keeps its name across a renewal, because it
is the only one whose receipt is an identity rather than a deadline: a stream entry
id says when the entry was appended, which no amount of renewing changes. So the
delivery comes back exactly as it went in, and the Delivery this returns is the
argument.
XCLAIM with JUSTID is what says it, and the reset is the point rather than a
side effect: idle time is what reclaim measures, so claiming an entry this
consumer already holds sets it back to zero without moving the entry anywhere.
within therefore has nothing to say here: the clock is reset to now either way,
and how long that buys is whatever idle the next reclaim measures against.
Silent about an entry that is no longer this consumer's, whether acknowledged,
deleted, or taken over by another worker's reclaim, and the last of those is why
this is a script rather than the one command it reads as (see HOLD): claiming an
entry another consumer now holds would take it straight back, which is not renewing
a delivery but stealing one.
async
¶
cancel(workflow: str) -> None
Drop the workflow from the sleepers, and delete every entry it has in the stream.
The sleepers are one ZREM. The stream is the expensive half and there is no
cheaper way to do it: a stream is addressed by entry id, and an id says when an
entry was appended rather than what is in it, so finding a workflow's entries means
reading them. The sorted-set queue answers the same call with a single ZREM
because it holds one entry per workflow, which is the trade that whole design makes
(see RedisSetScheduler), and this is where the stream pays it back.
What bounds the scan is the trimmer rather than the design. trim removes
everything every group has acknowledged, so the stream a healthy deployment carries
is the backlog nobody has run yet, not its history. On a deployment that never
trims, this reads every entry ever appended.
Read in batches from here rather than looped inside a script, because a script runs
to completion with the server to itself: a scan of a large backlog would hold every
other client out for its whole duration, where batching costs a round trip per
scan entries and lets everything else through in between. That leaves the call
non-atomic, which costs nothing: entries appended after it starts are a
make_ready racing a delete, and every entry present when it starts is passed over
exactly once, since ids only go up and nothing else removes them.
async
¶
trim() -> int
Drop the entries every consumer group has finished with, and report how many.
XACK clears an entry from a group's pending list and leaves it in the stream,
so without this the queue is append-only: correct, and unbounded. ACKED is the
bound, and it is the server's own answer rather than one computed here: it
removes only entries that every group has read and acknowledged. Working that
floor out client-side is possible and strictly worse, because it races every ack
that lands between the read and the trim. Capping by length instead would be
the wrong bound entirely, since that drops the oldest entries, which are the ones
nobody has run yet.
MAXLEN 0 reads as "keep nothing", and with ACKED that is exactly right:
trimming still stops at the first entry somebody has not answered for, so the
threshold only says "as much as you are allowed to".
Note what ACKED does not do. With no consumer groups at all it has no effect
and the trim degrades to a plain MAXLEN 0, which would delete a queue nobody
has read yet - and orders can be queued before the first worker ever boots, which
is the case prepare creates its group from 0 to handle. So this refuses to
trim a stream that has no groups, which is the one hazard in an otherwise safe
command.
Safe to run from every process at once, and safe to never run at all: the trim is idempotent, what counts as acknowledged only grows, and a stream nobody trims is merely large.
Requires Redis 8.2 or newer, which is where ACKED arrives.
trimming(scheduler: RedisStreamScheduler) -> Sink[object]
Keep the stream tidy, once per event, over whatever stream you drive it with.
A Sink rather than a loop with a sleep in it, which is the same shape waking has
and for the same reason: what makes a trim happen is a value somebody supplies, so
this runs off a timer, off an operator poking a queue, off a Kubernetes cron hitting
an endpoint, or off three items in a test. A loop can only ever be a timer, and it
buries the schedule inside the thing being scheduled. It takes Sink[object] because
it reads nothing from the event: whatever the stream carries, a trim is a trim.
async with asyncio.TaskGroup() as group:
group.create_task(work(durable, body))
group.create_task(trimming(scheduler)(ticks(TRIM_EVERY)))
Control plane rather than data plane, and deliberately not folded into work:
whether an entry is still needed is a question about what every group has
acknowledged, not about the delivery a worker happens to be holding, so triggering it
by traffic would make housekeeping cost scale with load for no reason. Its
cardinality needs no arranging either, since the trim is idempotent: every process
may run one, and N of them just means the same trim happens N times.