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 written into the checkpoint verbatim; under the
default JsonCodec that means JSON text, and cjson.encode is the usual way to
produce it. That is the one place an effect has to know which codec 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 a hash field 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
¶
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
¶
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,
)
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 is written into the checkpoint verbatim, 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.
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
¶
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
¶
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,
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
¶
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 here, 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.
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
¶
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.