[ULTRA EXPERIMENTAL — DO NOT REVIEW] Durable workflows exploration - #6912
Draft
Alek99 wants to merge 112 commits into
Draft
[ULTRA EXPERIMENTAL — DO NOT REVIEW] Durable workflows exploration#6912Alek99 wants to merge 112 commits into
Alek99 wants to merge 112 commits into
Conversation
Implements the first slice of the Reflex Workflows design: durable automation on the existing Reflex programming model. Authoring contract: - rx.WorkflowConfig on a workflow-focused rx.State class (reserved __workflow__ attribute, excluded from state schemas) - @rx.event(durable=True, effect=...) with id/trigger/retry/timeout/ queue/on_failure/on_timeout options, validated at decoration time - rx.Retry with per-effect-class defaults materialized at compile (TransientWorkflowError is the explicit retryable signal) - declarative triggers rx.manual() / rx.webhook() / rx.schedule() - control returns rx.complete/fail/needs_attention and rx.after delays Runtime: - compile_workflow() validates the class contract (durable-only public handlers, no substates/backend vars/mixed scopes, resolvable hooks, stable unique ids) and produces an immutable digest-pinned definition - app.add_workflow() registers the class and detaches it from the session state tree: no per-session instances, no browser setters, no frontend event dispatch to durable handlers - WorkflowKernel executes runs against a RunStore: single-writer ordered mailbox with preallocated ordinals, fenced claims, atomic commit of state snapshot + successor slots + history, retry with exponential backoff and jitter as persisted timers, per-attempt execution timeouts, failure/timeout hooks after tombstoning, NEEDS_ATTENTION suspension for uncertain non-idempotent effects, drain-based cancellation, run deadlines, max_steps bounds, and orphan recovery with a separate infrastructure recovery budget - MemoryRunStore for tests; SqliteRunStore (stdlib, WAL) for crash-safe local persistence, including admission dedupe by request_key surviving restarts - rx.workflows.start/cancel/get_run namespace served by the app lifespan; WorkflowTestHarness runs definitions deterministically on virtual time Deliberately out of scope for this slice (per the design's MVP/Beta ledger): webhook/schedule ingress execution, the connector broker and effect evidence records, mixed-scope classes, UI run projections, operator commands beyond cancel, and multi-worker kernels.
A second worker starting while the first was mid-attempt reclaimed the live claim and executed the same step concurrently: recovery treated every CLAIMED row as an orphan left by a dead process. Proven with two OS processes against one SQLite file, where the peer ran the handler a second time and the original worker's commit was then discarded as stale. Claims now carry a lease. claim_next stamps lease_expires_at, the executing kernel renews it in the background while the attempt runs, and recovery reclaims only claims whose lease has lapsed, so a peer mid-attempt is never disturbed. Lease loss consumes the infrastructure recovery budget, never a business attempt, and is recorded as attempt_abandoned evidence. Also: - Recovery is now periodic rather than startup-only, so a peer that dies long after this worker booted is still reclaimed. - Renewal cadence runs on real time while expiry is measured on the injected clock, so virtual-time tests stay deterministic and a jumping clock cannot expire a live attempt (recover() renews own claims before sweeping). - Cancellation is now disambiguated three ways: a lost lease abandons the attempt, an operator cancel releases the step as CANCELLED, and any other cancellation (worker shutdown) re-raises and leaves the step claimed for lease recovery. Previously every cancellation terminally cancelled the step and wedged the run at RUNNING with no path forward. - SqliteRunStore migrates databases written before this change; a step left claimed by the previous binary has no lease and is a genuine orphan. - The worker loop no longer dies on a transient store error. Crash recovery is no longer instant: it is bounded below by the lease duration (default 30s, sweeping every 15s). That is the price of not double-executing. Multi-worker SQLite remains unsupported and is now documented as one worker process per database file: the store's calls are synchronous, so cross-process write contention blocks the caller's event loop including its own renewals.
rx.Retry(max_attempts=5) on a handler raising an ordinary exception executed the handler exactly once. Both the effect-class defaults and any explicit policy that did not name retry_on were resolved to retry only on TransientWorkflowError, so a flaky HTTP call or a dropped connection failed the run immediately while the declared policy promised five attempts. Failures now retry by default: none, read, and idempotent_write resolve to three attempts with exponential backoff on any Exception, and an explicit policy without retry_on retries on Exception too. Narrow it with do_not_retry_on to fail fast. non_idempotent_write still gets exactly one attempt and routes to NEEDS_ATTENTION, since the runtime cannot prove the external effect did not already land. This is a deliberate divergence from the design doc's 'unknown code defects do not retry by default'. That rule is incoherent with an idempotent_write declaring three attempts that can never fire, and it makes the common case -- surviving a flaky dependency, which is the whole point of a durable step -- require boilerplate. Every comparable engine retries by default. TransientWorkflowError remains as an explicit marker for intent and for staying retryable under a narrowed policy.
Runs were pinned to a hash of the whole compiled definition, so adding a state field, retuning a retry policy, or changing a timeout parked every live run of that workflow in NEEDS_ATTENTION -- with no API to get it back. That fires on the second deploy of any real app, not on an exotic one. Runs are now gated on what can actually strand a step: the handler it names is gone, or its persisted payload no longer fits that handler's parameters. Both suspend with a precise, actionable reason instead of a bare digest mismatch. Everything else -- new fields, retuned retries and timeouts, changed hooks, effects, and triggers -- deploys without disturbing work in flight. The definition digest is still recorded on each run as provenance. Adds rx.workflows.resume(run_id) so suspension is a door rather than a wall: it clears the error, grants the frontier step a fresh attempt budget, and makes it claimable. A handler that returns rx.needs_attention() now leaves its step NEEDS_ATTENTION rather than SUCCEEDED, so resuming re-runs the handler that suspended and lets it take a different branch once a human has acted.
Calling another durable handler directly (self.charge()) ran it inline inside the caller's attempt: no retry policy of its own, no effect tracking, no step in the mailbox, and a silent re-execution of its side effect whenever the caller retried. The run still reported success, which is what makes it dangerous -- and it is exactly the shape a code generator reaches for. The compiler now parses each handler body and rejects two traps with an actionable message: an inline call to a sibling durable handler (pointing at 'return MyFlow.charge' instead), and a return of a plain literal (listing the transitions a durable handler may return). Both fail at compile time, before a run exists. Handlers whose source is unavailable, as in a REPL, are skipped rather than guessed at.
Runs could only be fetched one at a time by id, so labels were recorded but never readable and nothing could build an operator view, a CLI listing, or a customer-facing 'your jobs' page. Adds RunQuery and RunStore.list_runs on both stores, surfaced as rx.workflows.list_runs(workflow_id=..., statuses=..., labels=..., limit=...), newest first with a created_before cursor for pagination. SQLite filters labels through json_extract so it stays a single indexed scan rather than loading every run.
rx.webhook(...) compiled but could never fire: only manual roots were
reachable, so the whole provider-driven half of the product was declarative
decoration.
Adds the ingress endpoint at POST /_workflow/webhook/{topic}, registered only
when a workflow actually declares a webhook root. It preserves the raw request
body, verifies the provider signature over those exact bytes, validates the
payload against the declared model, and durably admits the run before
acknowledging -- so a provider that never sees a 202 can safely redeliver.
Redelivery reaches the same run through dedupe_by rather than starting a
second one.
Authentication is not optional by default: a webhook trigger without a
verifier is a compile error naming the fix, and an endpoint that really is
public must say so with allow_unverified plus a reason. rx.hmac_signature()
covers the Stripe/GitHub/Shopify shape, reading the secret from the
environment at request time so it never enters workflow state, history, or a
browser bundle.
Trigger kind is now part of admission: a webhook root cannot be started by
application code, and a manual root is not reachable over HTTP.
rx.schedule(...) validated its five fields and then did nothing; scheduled workflows never ran. Adds a dependency-free UTC cron evaluator supporting the standard five fields with ranges, steps and lists, including the day-of-month OR day-of-week rule that every other cron implementation follows. Expressions are validated when the workflow compiles, so a bad expression fails at add_workflow() rather than silently never firing. The kernel admits one run per occurrence under a request key derived from the occurrence time, which reuses the existing dedupe path: a restart, a second process, or an overlapping sweep all converge on exactly one run per occurrence. Cursors are seeded when the kernel is constructed, so deploying a schedule never backfills history, and catch-up after an outage is capped at ten occurrences so a restart cannot stampede. The worker wakes for the next occurrence rather than polling for it.
A run could only move forward on its own timers, so human approvals, event-driven waits, and anything needing an answer from outside were simply inexpressible -- the gap that most separated this from Temporal and Inngest. A wait is now a BLOCKED slot at the run's frontier, which keeps the mailbox strictly serial: no second open slot, no change to the per-run fence, no concurrent commits. The slot carries the address a delivery must match, and its due_at doubles as the deadline, so a wait timeout fires through the same timer path that rx.after() already used and the virtual clock drives it with no new machinery. The race is settled on one row. A delivery compare-and-swaps the blocked slot to ready and hands the payload to the resume handler; a deadline instead makes the slot claimable, and claiming it *is* the timeout branch. Whichever lands first erases the other's trigger, so a late signal to a finished run is refused rather than silently dropped. A signal that arrives before the run reaches its wait is buffered and consumed by the arming commit itself, so a sender faster than the workflow cannot block it forever. Deliveries never write run state or the state version, so a delivery can never fence a live attempt. BLOCKED is deliberately not a claimable status: a wait with no deadline would otherwise report itself due at time zero and spin the worker against the database, starving lease renewal. Claimability is now one predicate both stores share, with a test asserting an unbounded wait leaves nothing claimable and nothing scheduled.
One dunning workflow written the way a user would: a manual root, a flaky charge with retries and a failure hook, a durable delay, a human decision with a deadline, and completion, failure, and suspension outcomes -- plus a second class reached by a verified webhook and a cron schedule. It caught a real authoring subtlety worth keeping in front of us: run state cannot count attempts, because a failed attempt's patch is discarded by design. The example now models gateway flakiness outside the run, which is where it actually lives.
Nothing about workflows was documented, which also matters because this page is the surface a text-to-workflow generator will learn from. Covers the whole shipped surface: durable steps and effect classes, retries and timeouts, the transition table, waits and typed signals with the approval example, manual/webhook/schedule triggers, inspecting and steering runs, the virtual-clock harness, and what a redeploy does to runs in flight. Every example was run against the real engine rather than written from memory, which caught one error in the draft: rx.Base no longer exists on main, so payload models use pydantic BaseModel.
Runs could only do one thing at a time: the mailbox is strictly serial, so there was no way to enrich and score a lead concurrently, or to do anything Temporal and Inngest express with child workflows. Concurrency now lives in the run graph rather than the mailbox. rx.parallel() commits a BLOCKED join slot in the parent and admits one child run per branch, each with its own state, retries, timers, and history. A child that finishes reports its outcome to the parent's join slot through a compare-and-swap on an arrival counter, so a redelivered result cannot be counted twice, and the slot becomes claimable exactly when the last expected branch lands. The join handler receives one entry per branch carrying run_id, status, result, and error. Keeping each run's mailbox serial is what makes this safe: the parent never has two open slots, so the per-run fence and the one-claim-per-run invariant are untouched, and a failing branch fails its own run rather than the parent's. Children are admitted after the parent's commit lands, so a crash in between leaves a join with no children -- which recovery re-runs -- rather than orphans with no parent.
Nothing bounded how often a root could start, so a chatty webhook produced one
run per delivery and two clicks produced two concurrent syncs of the same
customer. Flow control is the main thing Inngest sells and we had none of it.
A root now declares one start policy, applied at admission and grouped by a
payload field:
singleton one active run per key; a second start either returns the first
(skip) or replaces it (cancel)
debounce a burst collapses into one run, each start pushing it out until
things go quiet
rate_limit starts beyond the cap are refused with retry_after, which is what
you want when a provider can flood you
throttle the excess is delayed rather than dropped, for when every start
matters but the downstream is slow
The grouping key must name a real parameter of the root, checked at compile
time, and a policy without a trigger is rejected since it governs starting.
Only one policy per root, so its behavior stays predictable.
throttle= and debounce= are overloaded by type rather than given second names:
an int is still the browser event action on a session handler, while the
policy object is the durable start policy. Same word, same meaning, and a
session handler passing an int is untouched.
RunStore is a public extension point -- a deployment can back workflows with Postgres or a hosted kernel -- but the protocol's signatures say nothing about the invariants that make durable execution correct. Two implementations were already drifting apart with nothing but shared test files to hold them together, across 22 methods. reflex.workflow.CONFORMANCE_CHECKS is now the specification: 22 checks, each taking a fresh store and asserting one property. Frontier ordering, atomic commit, fenced claims discarding their work, failed attempts discarding state, lease renewal sparing a live claim, recovery reclaiming only lapsed ones, next_due never promising work that is not claimable, a deadline-less wait never becoming due, delivery never touching run state, joins counting each arrival once, finalize refusing while a step is claimed, and the queries start policies depend on. Both shipped stores pass all 22. Anyone adding a store runs the same suite; it is exported and documented for that purpose. Postgres is deliberately not in this commit: no server was available to test against, and an unverified store implementation is worse than none.
Runs could only be reached from inside the app, so diagnosing one meant writing a script against the store. That is the wrong tool at 2am. reflex workflows list filters by workflow, status, and label; show renders a run's state, steps with their attempt and recovery counts, and optionally its history; cancel and resume steer a run without opening the app. Both list and show take --json so the output is scriptable. The commands read the same SQLite database the app writes, so they work against a running deployment or a stopped one, and resume refuses a run that is not actually suspended rather than pretending to act.
Runs recorded a full history but nothing could watch them happen: diagnosing a production workflow meant querying the database after the fact, and there was no way to get workflow activity into metrics or tracing. WorkflowObserver receives every transition the kernel records -- admission, each attempt and its outcome, retries, waits, joins, and terminal dispositions -- with the correlation a durable system needs: run, workflow, step, and attempt. Install one with rx.App(workflow_observer=...). LoggingObserver is bundled for the common case. Instrumentation is deliberately not allowed to break execution: an observer that raises is reported and ignored, which a test asserts by running a workflow to completion under an observer that always throws.
An audit of the whole feature reproduced four real problems, each of which could stop a run permanently. A handler that raised CancelledError itself -- as any handler wrapping its own asyncio work might -- propagated past the cancellation branch and killed the worker task, so every later run in the process silently never executed. The kernel now distinguishes a task that was cancelled from a coroutine that raised: the first is a control signal, the second is an ordinary handler failure that retries. A crash between a fan-out's commit and the creation of its children left a join blocked on children that did not exist, with nothing to recover it; the original comment claiming recovery handled this was simply wrong. Children are now inserted in the same transaction as the join slot, so the window is gone rather than merely narrowed. A child that was cancelled or blew its run deadline never reported to its parent's join, because reporting only happened on commit and those paths finalize without one. Both now report, so a join can no longer wait forever on a child that already stopped. Singleton with mode='cancel' left the superseded run in CANCELLING, so a burst of starts produced several simultaneously active runs under one key. The replacement now waits for the old run to reach a terminal state first. Also fixes a store divergence the conformance suite missed: the memory store kept only the most recent buffered signal per wait key while SQLite queued them, so a second early signal was silently dropped. Two conformance checks now cover buffered-delivery ordering and children being created with their join.
A second, adversarial pass over the whole feature reproduced four problems, one of which was a fix from the previous commit that did not actually work. The guard added for 'a handler that raises CancelledError kills the worker' was a branch that could never be true: asyncio marks a task cancelled whether the kernel cancelled it or the coroutine let CancelledError escape, so the task's own flag cannot tell them apart. Verified directly -- both cases report cancelled() is True -- so the worker still died and every later run in the process silently never ran. The kernel now discriminates on its own control signals plus whether cancellation was requested on the executing task: a handler that raises is an ordinary failure, while a real shutdown still leaves the step claimed for lease recovery. The worker loop no longer dies on any exception, and a dead worker can be replaced. Admission dedupe ran after start policies, so a provider redelivering an event was judged as a new start: with singleton(mode='cancel') the redelivery cancelled the very run it deduplicated to, and answered the provider 202. With debounce, a provider retrying one event faster than the window starved it forever. request_key is now resolved before any policy. rx.fail(details=...) and rx.needs_attention(details=...) passed user values straight to the store, so a datetime raised inside the transaction recording the failure and left the run stuck RUNNING. Details are normalized first, and an unserializable value is recorded as its repr rather than losing the failure. timeout= on a synchronous handler was a lie: asyncio.wait_for cancels the wrapper while the thread runs on, so a timed-out step kept executing and its retries ran concurrently with it. It is now a compile error naming the fix.
Every kernel-level test used the in-memory store, so the harness was certifying semantics production does not necessarily have: any divergence between the two stores could ship green, and one already had. Harness-based tests are now parametrised over both stores, doubling that coverage to 478 workflow tests. Turning it on immediately caught a real divergence: the memory store handed callers live references to the values it was storing, so mutating a returned run's state silently changed committed data -- something a database-backed store cannot do. Reads now detach their mutable payloads, and a conformance check pins it. It also exposed an order-dependent test of its own, which assumed the first child listed was the branch that had already reported to the join. This is the gap that let earlier store divergences through, so it goes in before any further features.
Working through the confirmed findings, highest severity first. A class with __workflow__ that was never passed to app.add_workflow() stayed a session substate, so its durable handlers -- including non_idempotent_write -- remained dispatchable from a browser, and the only feedback was a later error from rx.workflows.start(). Tying detachment to registration meant the omitted line, exactly the one a generator drops, left money-moving handlers exposed. Detachment now happens when the class is created, so registration only adds the definition to the kernel. A wait whose deadline had already fallen due still accepted its signal, because delivery matched on BLOCKED alone and never compared the deadline. A seven-day approval that expired three weeks ago would be approved, and a sender that lost the race was told 'buffered' and then refused as 'duplicate'. Delivery now refuses with a distinct 'expired' disposition, which also stops a stale signal resolving a later wait on the same channel. A child failed by exhausting its recovery budget never reported to its parent's join: that path fails the run inside the store, and recover() only returned a count. The parent, and every ancestor, waited forever. recover_orphans now returns the runs it failed so the kernel can report them. Start policies of the wrong type vanished instead of raising: debounce='30s' -- a very plausible generation given every other duration in the API is a string -- silently disabled debouncing, and singleton='cid' failed only in production once two runs overlapped. All four are now type-checked at decoration, discriminating the browser event action on int. Run pagination used created_at alone, so runs sharing a timestamp -- the shape every fan-out produces -- were silently skipped; the cursor is now (created_at, run_id). The SQLite label filter interpolated user-supplied keys into a JSON path expression and now matches them as values.
Two more confirmed findings. Lease renewal failures were swallowed at debug level, so a store that kept failing -- which a contended SQLite file does, raising after a five second block -- left the attempt running while its lease quietly lapsed. Recovery then handed the step to someone else and the external effect ran twice, with a log line that is off by default as the only evidence. The kernel now tracks when the lease actually expires and abandons the attempt once too little of it remains to survive another failed round-trip: better to stop work you can no longer prove you own than to race the worker that is about to take it. rx.parallel resolved its branches through a path that skipped the trigger check start() enforces, so a webhook-only root -- one that exists precisely because only a verified provider may start it -- could be started from application code by naming it as a branch. Branches are now held to the same manual-root rule as a direct start. Fanning out to a handler of your own class is now a compile error. Each branch becomes a child run with fresh state, so a same-class branch cannot see anything the parent did; it is the most natural spelling and it silently did the wrong thing.
Every store call is synchronous on the caller's event loop, which also serves HTTP, websockets, and session state. SQLite's default busy timeout is multiple seconds, so a second process writing the same file could stall the whole app before raising -- and the error then landed in lease renewal, where a silent failure used to mean the kernel re-executed its own attempt. The busy timeout is now short: contention surfaces quickly as a transient error the kernel retries, rather than freezing everything first. Combined with the previous commit, a store that cannot be reached now degrades to abandoning the attempt instead of duplicating its effect. This bounds the symptom rather than removing the cause. Offloading the store's calls to a thread is the real fix, and one worker process per database file remains the supported deployment -- now stated plainly in the docs alongside why horizontal scale wants a different store.
The kernel claimed one step and awaited it before claiming again, so a single process executed one step at a time no matter how many runs were waiting. One ten-minute step stalled every other run in the deployment -- including their timers and deadlines -- which is not a throughput story any durable engine can be compared on. The scheduler now fills up to max_concurrency slots, eight by default. This needs no new locking: each run has exactly one claimable frontier step, so two concurrent claims are necessarily different runs and a run's own mailbox stays strictly serial. A test asserts both halves -- that attempts really do overlap, and that one run's steps still complete in order. Three details that had to be right: finished attempts are pruned synchronously rather than by a done-callback, or the scheduler spins on work it already finished; a round waits only on the attempts it started, so a second caller pumping the same kernel is not blocked by an attempt it does not own; and cancelling the scheduler cancels the attempts it started, since asyncio.wait does not cancel what it waits on. The test harness stays single-slot so virtual-clock tests remain deterministic; production defaults to eight.
rx.parallel always waited for every branch, so the one shape everybody reaches for -- ask two vendors, take whoever answers first -- could not be expressed. Worse, the losing branch kept running and kept calling out to a vendor after the order was already booked. mode="first" sets the join's expected count to one. The arrival that resolves the join now says so, and the kernel cancels the siblings still in flight, found through a new list_children(parent_run_id, parent_ordinal) store query rather than a scan of the run table -- indexed in sqlite, and covered by a conformance check so any future store has to answer it too. Two compile-time diagnostics came out of writing the tests, both failure modes a code generator will hit: self.book passed as then= is a bound method, not a routable transition, so it failed at runtime -- as a retrying durable step, which is the worst place to learn about a typo. The handler-body guard already rejected self.step() calls; it now rejects bare self.step references too and names the class form. Passing a list of workflow classes where varargs were expected died with "cannot use 'list' as a dict key" from inside the registry. register() now says what it wanted. The docs example was executed before committing.
Throttle deferred every excess start by exactly one window, so a burst of a hundred with limit=10 admitted ten now and scheduled ninety for the same instant one window later. That is not throttling, it is a delay line: the downstream the throttle exists to protect sees the same spike, just later, and the next window then holds ninety starts against a limit of ten. Each start is now placed at least a window after the limit-th most recent scheduled start under its key, which spaces the backlog at exactly the configured rate and holds the sliding-window bound rather than a per-window one: any window of length period contains at most limit starts, because a start is always a full window after its limit-th predecessor. This needs a new store query, nth_recent_start, since counting admissions in a window cannot see where the deferred ones are already scheduled. A run's scheduled start is when its root slot comes due, so a debounced or throttled run counts at the time it will run, not the time it was admitted. Covered by a conformance check, so a future store has to answer it the same way. The regression test asserts the schedule directly (0, 0, 10, 10, 20, 20 for six starts at limit=2 over ten seconds) and then advances the clock to confirm the runs execute on it.
SQLite takes one writer, so a deployment was capped at one worker process per database file. That is the ceiling that keeps this from being comparable to Temporal or Inngest at all, and it is not something the kernel can fix -- it is the store. PostgresRunStore claims a run's frontier step with FOR UPDATE ... SKIP LOCKED, so workers never queue behind each other and never take the same step. Adding a process adds throughput. A worker that dies mid-step holds a lease that another worker may only reclaim once it lapses, so a slow step is never duplicated -- there is a test for exactly that, and one asserting twenty non_idempotent_write runs across two kernels execute exactly once each, with both kernels provably taking a share of the work. The store is not trusted on assertion: the 29 conformance checks run against it, and the whole workflow suite -- 260 harness tests -- now runs a third time against a real server whenever REFLEX_TEST_POSTGRES names one. That third parameter immediately paid for itself twice. It caught a test of mine that asserted a race always starts every branch, which is not an invariant: a loser cancelled before its first step is the better outcome, and only Postgres's ordering exposed the assumption. And driving the real CLI against a Postgres URL surfaced that each command ran its own asyncio.run, which is fine for a file and fails for a pool, whose connections belong to the loop that opened them. Commands now run in one loop. Two things came out of the port. Pyright rejected the schema name spliced into DDL, since psycopg types raw SQL as LiteralString; the name is now composed as an identifier, which is a better guarantee than the check it replaces. And the observer turned out to see only admissions and commits -- not attempt starts, which are the spans a tracer actually wants -- so every recording site now reports, correlated to its workflow. A throwaway schema per test isolates them. Dropping one first evicts its own backends, because a test loop that dies mid-transaction leaves a pooled connection idle holding locks, and the DROP would otherwise wait on it forever. That was a real hang, reproducible only under random test ordering. Postgres is optional: pip install 'psycopg[binary,pool]'.
A run waiting on a decision is the most common human-in-the-loop shape, and the naive implementation -- a URL carrying a run id -- is an open door. An approval link here is an HMAC token over run, channel, payload, delivery key, and expiry, so an edited link is refused rather than believed; it is spent once; and it never decides on GET, because mail clients and scanners fetch URLs before a person reads the message, so a link that approved on GET would approve itself in transit. The secret comes from REFLEX_WORKFLOW_APPROVAL_SECRET with deliberately no default, and a server missing it says 'not configured' instead of masquerading as an expired link. Executing the docs example before committing caught a real defect: a channel declared with a pydantic model failed to serialize into the token -- and every realistic channel is typed -- so payloads now go through the same reduction the signal path uses. Links need to know which run built them, which is a capability handlers were missing generally. rx.current_run() now exposes the attempt's identity -- run, workflow, slot, attempt, epoch -- bound per attempt via a ContextVar that to_thread carries into sync handlers, plus idempotency_key(): stable across retries of a step, distinct across steps, which is exactly the contract a payment API's idempotency header wants. Chasing an intermittent test hang also root-caused a real defect: the worker loop's treated CancelledError as a retryable error, so anything that cancelled the worker task without calling aclose() -- a task group, a supervisor, an event loop tearing down -- waited forever on a task that had gone back to polling. The hang reproduced with the fix removed and disappears with it. The kernel also cancels its in-flight attempts on close instead of leaving them running against a store nobody reads, and the test harness closes stores it created, which previously leaked a Postgres pool into every later test in the process.
A handler is the unit of retry, so a handler that makes three calls and fails
after the second repeats the first two on retry -- three charges for one
order. rx.step(name, fn, ...) runs a callable once, records its result
durably at the moment it returns, and replays it to every later attempt of
the same handler, including one recovered from a crashed worker. The journal
is epoch-fenced: an attempt whose lease was reclaimed cannot record, so a
zombie stops instead of duplicating a side effect. Results round-trip through
serialization before the handler ever sees them, so the first execution and a
replay produce identical shapes -- a difference there would only surface
during retries, the worst place to find it. Async handlers await the call;
sync handlers call it bare and block, bounded so a stalled loop fails the
step rather than pinning the worker thread forever. A name reused in a loop
is numbered per occurrence. Recorded keys appear in run history.
queue= on a durable handler has been accepted since the first commit and
silently ignored, which is the worst state a parameter can be in. It now
routes: every step is stamped with its handler's queue ('default' when none),
a worker claims only from queues it serves (rx.App(workflow_queues=...)), and
per-run order holds across queues -- a run whose frontier sits on an unserved
queue waits for the right worker rather than running the step somewhere it
was configured not to. Wait and join slots take the queue of the handler that
resumes them; children take their root's.
Both are covered by conformance checks, so all three stores answer the same
way, and the whole workflow suite passes against memory, SQLite, and
Postgres. The teardown watchdog added to the workflow conftest (REFLEX_DEBUG_HANG=1)
names any task that outlives its cancellation instead of hanging the run
silently -- built while chasing an intermittent suite hang that so far only
reproduces when several suites contend for one database.
Contributor
Greptile SummaryThe PR introduces an experimental durable-workflow engine for Reflex, including persistent stores, leased execution, triggers, flow control, composition, operator APIs, testing utilities, and documentation.
Confidence Score: 3/5The PR is not safe to merge because two outstanding same-run lifecycle races can leave attempts untracked during shutdown or remove a live replacement attempt’s lease renewal.
Files Needing Attention: reflex/workflow/kernel.py and focused lease/shutdown regression tests in tests/units/workflow/test_lease.py
|
| Filename | Overview |
|---|---|
| reflex/workflow/kernel.py | Implements concurrent leased workflow execution, recovery, retries, transitions, shutdown, and worker scheduling. |
| reflex/workflow/store.py | Defines the workflow store contract and the in-memory and SQLite persistence implementations. |
| reflex/workflow/postgres.py | Adds the multi-worker Postgres implementation with transactional claiming and schema namespacing. |
| packages/reflex-base/src/reflex_base/workflow.py | Adds public authoring-time workflow configuration, trigger, policy, and transition value types. |
| reflex/workflow/definition.py | Compiles workflow state classes and durable handlers into validated runtime definitions. |
| reflex/workflow/ingress.py | Adds durable webhook admission with payload validation, signature verification, and deduplication. |
| tests/units/workflow/test_lease.py | Covers lease renewal, loss, recovery, and shutdown behavior, but does not exercise overlapping attempts for one run. |
Reviews (16): Last reviewed commit: "Pin two more contract clauses with confo..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
I told the reviewer I did not understand this one, so here is what it actually is. Returning a list from a handler preallocates an immediate sequential chain -- kernel._interpret_return -- so [middle(), finish()] creates both slots up front. A terminal failure in middle tombstones finish behind it, and retry() reopens only the failed step. The retried step succeeds, returns nothing to allocate, and the run completes having never run the finalizer the chain existed for. My earlier reasoning was wrong because I assumed chains are built a step at a time, which is true of rx.after and not of a returned list. Captured as a strict xfail rather than fixed: the fix belongs in retry_run and skip_step in all three stores, restoring the successors that failure tombstoned without reviving anything an unrelated cancellation cancelled, and that is not a change to start on a budget that cannot finish it. The first version of this test xfailed for the wrong reason -- the test harness has no retry(), so it was failing on AttributeError, not on the defect. It goes through the kernel now. That the harness exposes cancel but not retry/skip/resume is its own gap: operator recovery is exactly what someone would want to test.
A provider numbers its events per object, not per topic, so invoice_failed and invoice_paid for one invoice both arrive carrying that invoice's id. The ingress used that value as the whole admission key, and stores enforce uniqueness on (workflow_id, request_key), so the payment was taken for a redelivery of the failure and dropped. One workflow class serving an object's lifecycle -- the shape the API invites -- silently loses events. Keys are webhook:<handler_id>:<value> now. Two deliveries are the same event only if they would start the same handler. Changing a key format is itself a hazard: every run admitted under the old spelling would stop being found, and the provider's next redelivery would start it again. kernel.start() takes superseded_keys, matched for deduplication and never written, and the ingress passes the old bare value. Nothing replays across the upgrade. Found by a live-fire review over real sockets and processes, which also confirmed the parts that hold: 401 on forged signatures, 64 concurrent redeliveries collapsing to one run across two servers, and recovery of a WAITING run across a full process restart.
My own regression, an hour old. Scoping fresh webhook keys by handler fixed the cross-topic collision; the compatibility lookup I added in the same commit matched the old unqualified key against any run in the workflow, which is precisely the collision again -- invoice_paid deduplicating against a legacy invoice_failed run and being dropped. A superseded key is a less specific spelling than the one that replaced it, so a match on it means the same event only if it started the same root. It is checked against the existing run's first step now. Caught by a live-fire review over real sockets, which reproduced it as 202 deduplicated pointing at the wrong run.
The production-readiness review's P0: every advertised start policy violated its invariant in 50 of 50 synchronized trials across two OS processes and independent PostgreSQL pools. The policy decision was a read in one store call and a write in another, guarded by an asyncio lock -- which serializes one process, and a fleet is not one process. Postgres row locks cannot fix it alone: with no active run there is no row to lock, so both admitters count zero and both insert. The whole decision -- dedupe, every policy read, any policy mutation, and the insert -- now executes inside one store transaction under a durable lock on (workflow_id, flow_key): pg_advisory_xact_lock on Postgres, BEGIN IMMEDIATE's write lock on SQLite, the store lock in memory. The kernel builds a FlowGate and hands the store the records; admit(max_active=) is gone, one contract with one enforcer. Semantics pinned by conformance checks and by cross-instance race tests that open two stores -- two pools, no shared Python state -- over one database: skip admits exactly one; rate limit of one rejects the loser; throttle spaces a racing pair a full window apart; debounce coalesces the loser into the winner and, fixing the review's P1 alongside, takes the burst's LATEST payload; singleton cancel leaves at most one uncancelled run. All pass against real Postgres 16. Fan-out branches are written by the parent's commit, not through policy admission, so a policy-decorated branch root is now refused with a teaching error instead of having its policy silently bypassed -- and because that is a definition problem, WorkflowDefinitionError joins BUG_EXCEPTIONS and fails the run on attempt one instead of retrying a deterministic error. Singleton cancel changes shape slightly: the incumbent's cancellation intent is recorded in the admitting transaction and the replacement is admitted immediately, so at most one non-cancelling run exists per key at every instant, from any number of processes; incumbents drain to CANCELLED asynchronously as any cancelled run does. The contract states all of this in section 1.
The review's second P0. A handler returning a list preallocates an immediate sequential chain, so [middle(), finish()] creates both slots up front; middle's terminal failure tombstones finish, and retry or skip reopened only middle. The retried step succeeded, returned nothing to allocate, and the run completed COMPLETED/None having never run the finalizer -- an operator repair that silently dropped the cleanup step is worse than the failure it repaired. Both actions now restore every CANCELLED slot in the run, with fresh budgets, recorded as step_restored. The causality argument for restoring all of them rather than tracking which failure cancelled what: these actions accept only FAILED or NEEDS_ATTENTION runs, run-level cancellation ends in a CANCELLED run they refuse, force-finalization leaves no failed or suspended step for them to target, and suspension tombstones nothing -- so a CANCELLED slot in an acceptable run has exactly one possible source. No schema change needed. Fixed in memory, SQLite and Postgres; the strict xfail that pinned the defect is now the passing regression test, joined by the skip variant, and the operator suite passes against real Postgres 16.
The review's SQLite scaling cliff. claim_next selected every active run and loaded each run's steps until it found a claimable frontier; next_due did the same with one more query per run. Ten thousand durable timers -- the thing a workflow engine exists to hold -- meant ~114ms per idle poll, a worker burning a third to half a core doing nothing, and burst drains that looked quadratic (5,000 roots: 188s). Both queries are now one SQL statement: filter to steps whose status and due time can wake -- through a new (status, due_at, queue) index -- then pay the frontier check (NOT EXISTS a lower unresolved ordinal, a primary-key lookup) only for those candidates, and stop at the first winner. A store full of sleepers answers from the index range and touches nothing: measured on the reviewer's 10k-sleeper shape, next_due 61.8ms -> 4.8ms and claim_next 88.1ms -> 0.05ms, both under their proposed 5ms release gate. Claim order among concurrently-due runs changes from run creation time to due time, which is scheduling fairness the contract does not promise and arguably the better order -- oldest-due first. A plan-pinning test asserts EXPLAIN QUERY PLAN uses the wake index for both query shapes, because a timing assertion flakes on CI and the plan is the actual performance contract. The full workflow suite, conformance included, passes on all three stores with Postgres real.
Three review findings, one boundary. dedupe_by could only name a payload field, and GitHub's canonical delivery identity is the X-GitHub-Delivery header -- it appears nowhere in the body. Keyed on a payload field, two distinct deliveries sharing that field collapsed into one run, and with no dedupe_by at all a true redelivery executed twice. dedupe_by="header:Name" reads the header; distinct GUIDs are distinct runs and a repeated GUID deduplicates, pinned by test. A configured identity that could not be extracted silently disabled deduplication -- the request was admitted with no key, so every redelivery of that event ran again, defeating the exact thing the configuration asked for, invisibly. That is a 400 now, naming the missing source, so it surfaces as the config problem it is. rx.hmac_signature claimed to cover Stripe and cannot: Stripe signs timestamp.body, sends a structured header, and requires a replay window. The claim is corrected -- with a test that the docstring stays honest -- and rx.stripe_signature() implements the real scheme: t=/v1= parsing, multiple v1 digests for secret rotation, tolerance enforced both directions, constant-time comparison.
Fan-out children had no admission history: their record began at attempt_started, so runs_started reported one start for a four-run graph and every history reader met runs that were apparently never admitted. The creating transaction now writes run_admitted and step_scheduled per child -- one shared helper, three stores -- and the kernel reports the same events to the observer, so the dashboard's starts reconcile with its terminals. Opening the SQLite store ran CREATEs and an immediate-mode migration unconditionally, so an operator's read-only stats against a busy worker failed 4 times in 5 with 'database is locked'. The schema version is stamped in PRAGMA user_version now; a current database opens with no write lock at all, and a locked store renders as an actionable one-line error instead of a traceback. The legacy-migration test now also resets user_version in its simulation, because a genuinely old database has no stamp -- that is the very thing that triggers DDL. There was no retention surface: terminal data grew forever at ~1.7KB a run unless an operator did out-of-band SQL. purge_runs(before) deletes stale terminal runs and their steps, history, inbox, substeps and dedupe rows in one transaction on all three stores, conformance-pinned, with 'reflex workflows purge --older-than 30d' in front of it. The documented tradeoff: purging forgets request keys, so retention must exceed the provider's redelivery horizon. run_until_idle() could return with attempts from its own scheduling round still live -- a round starts several and waits only for the first -- handing tests a half-processed graph whose survivors the harness then cancelled. It now drains exactly the attempts it started, tracked per pump so a concurrent caller's work is never waited on; the first version of this fix waited on all in-flight attempts and deadlocked the lease tests, which own deliberately-hanging handlers.
The review's P3. GitHub can be configured to deliver application/x-www-form-urlencoded bodies carrying payload=<json>, and ingress unconditionally json.loads'd the raw body, so a correctly signed delivery came back 400 'payload is not JSON' -- an error that tells the operator nothing about which knob to turn. The signature was never the problem: it covers the raw form bytes and is checked against exactly those bytes before any parsing. Form bodies now unwrap the payload field before JSON parsing; a form body without one is a 400 that names it.
A 48-agent review over the six commits -- four lenses, two refuters per finding -- confirmed five defects in the new code itself. Postgres singleton-cancel could resurrect a finished run: the active SELECT is an unlocked snapshot, workers committing runs never hold the flow lock, and the cancel UPDATE had no non-terminal guard -- so under READ COMMITTED it waits out a concurrent completing commit and flips the COMPLETED run back to CANCELLING, which the next finalize sweep drives to CANCELLED over a terminal status. The guard request_cancel always had is now on this UPDATE too, and the cancelled list comes from RETURNING instead of the stale snapshot, so the kernel is never told a run was cancelled that actually completed. Postgres admit_flow's dedupe reservation was a plain INSERT after a plain SELECT; the advisory lock is keyed on the flow key, and two admissions can share a request key while computing different flow keys, so the loser raised UniqueViolation out of kernel.start where the contract promises a deduplicated admission. Same ON CONFLICT shape as admit() now. Memory purge_runs never deleted substep journals -- and popping by run_id would have silently missed anyway, since the journal is keyed by (run_id, ordinal). SQLite and Postgres purged theirs; parity restored. run_until_idle kept every finished task in its own-set for the whole call and its tail wait had no CancelledError handler, so cancelling a pump parked there left its attempts running unsupervised -- the exact contract _tick's wait already honors. FlowGate refuses combined policies in __post_init__. The decorator already enforces one policy per root, which is why the review's three divergence findings about combinations were refuted as unreachable -- but the stores genuinely disagree on what a half-applied combination leaves behind, and unreachable-because-callers-are-polite is not an invariant. Now it is unrepresentable. stripe_signature rejects non-finite timestamps: NaN compares false against both window bounds and sailed through -- harmless today only because the signature check follows, and the same NaN class bug was already found once in approval expiry.
The adversarial review's last confirmed finding was a contradiction between two sentences this branch added: section 1 promised a singleton holds 'at every instant', and section 9's retry re-opens a failed run without re-checking the gate -- so an operator retrying next to an already-admitted replacement puts two runs on one key. The behavior is the right one: a policy governs admissions, and an operator re-opening a run is a human override, not an admission. An engine that silently refused a repair because a policy would have is harder to operate, not safer. The contract now says exactly that, in both places, instead of promising an instant-by-instant invariant the operator surface deliberately does not enforce.
Third live review round, three regressions from tonight's own commits. The Postgres flow-gate reserved its dedupe key AFTER the policy mutations. Two deliveries of one event can compute different flow keys -- different advisory locks, no serialization between them -- so the loser ran the singleton-cancel branch against the OTHER key's incumbent, then hit the reservation conflict, returned 'deduplicated', and committed the cancellation anyway: an unrelated live run killed by an event that had already been handled, 20 times in 20 trials. The reservation is now the first write, exactly as in admit(), so a duplicate exits before it has mutated anything. Regression test races the exact scenario on all three stores and fails on the old ordering. The user_version gate used inequality, so an older binary opening a newer database re-ran its DDL and stamped its own OLDER version over the newer one -- a silent downgrade that would make the newer binary re-migrate. Strictly upward now; a future stamp is left alone. Both cancellation-cleanup sites in the kernel -- _tick's wait and run_until_idle's tail wait -- called _cancel_inflight(), which stops every attempt in the kernel, including ones a concurrent pump owns. One caller's timeout became another's abandoned work. Each site now cancels only the attempts that pump started; aclose() still sweeps everything, which is its job.
The reviewer's time-edge list starts with NaN/infinity durations, and
this is the third bug of exactly this class in the engine -- approval
expiry and Stripe timestamps had it first. NaN compares false against
every bound, so float('nan') sailed past parse_duration's negativity
check and poisoned every due-time comparison downstream; infinity
turned timers into never-fires.
parse_duration is the single choke point every timeout, retry delay,
debounce window, rx.after target, and tolerance flows through, so the
guard lives there: non-finite is a WorkflowDefinitionError, which since
tonight also fails a run on its first attempt instead of retrying.
The verified reviewer's top remaining blocker: worker wall clocks skew, and every scheduling comparison assumed they did not. A worker running fast saw a peer's live lease as lapsed and reclaimed the claim -- duplicating exactly the work leases exist to prevent -- and admitted schedule occurrences before their time. The store is the one thing every worker shares, so its clock is the authority. RunStore.epoch_time() answers with the store's own time -- clock_timestamp() on Postgres; None from SQLite and memory, whose single-host process clock already is the shared authority. A kernel constructed with the DEFAULT clock measures its offset against that answer, taken against the request midpoint so the measurement is off by at most half a round trip, at startup and again on every recovery pass. All time reads go through the offset, so lease expiry, due times, and occurrence keys use one time base across the fleet; measured offset against a real Postgres 16 was +9.7ms. Injectability is preserved exactly: an explicitly provided clock -- the test harness's virtual time, the dev CLI's fast-forward -- is authoritative as given and never synced, which is why all ~1,560 existing workflow tests pass unchanged. Regression tests pin the sync math, that store time decides due-ness rather than the worker's local clock, that injected clocks are never second-guessed, and that a store with no clock of its own is asked exactly once.
The verified reviewer's second blocker: work could commit after its deadline, and a signal could report success and then be discarded. Both were the same hole. The contract says a run past its deadline finalizes TIMED_OUT once drained -- but nothing guaranteed drained. An attempt that outran cooperative cancellation committed anyway, so a run the caller was told had timed out could quietly become COMPLETED. And a delivery to a past-deadline run was answered 'resolved' even though the continuation could never execute: claims exclude past-deadline runs, so the sweep finalized TIMED_OUT and the recorded decision evaporated. Commit now re-reads the run's deadline inside the same transaction that validates the claim fence -- one added column on the existing claim check, no extra round trip -- and refuses with DeadlinePassedError when it has passed. The kernel abandons the attempt (recorded substeps stand, crash-equivalent), releases the slot so the run drains immediately instead of waiting out a lease, and the sweep's TIMED_OUT is the only reachable outcome. Deliveries to past-deadline runs are refused as 'expired' in all three stores, so 'resolved' is never said of a decision about to be discarded. The contract's failure matrix gains both rows -- caught first by its own bidirectional reason guard, which is exactly the drift it exists to stop. Both fences are regression-tested and fail on the unfenced tree; the full workflow suite passes on all three stores against real Postgres 16.
Three from the reviewer's list, each verified before fixing. Retry backoff overflowed: multiplier ** 499 raises OverflowError, and delay_for_attempt runs inside the kernel's completion path, so a long-retrying step would have broken the worker loop rather than the run. The overflow saturates to max_delay, which is what a delay astronomically past the cap means anyway. Schedule catch-up loss was silent AND the contract lied about it: the failure matrix claimed the skipped remainder got a history record, and no such record was ever written -- there is no run to attach one to. A worker back from a week of downtime now warns with the count and the window when it skips past the cap, the cursor jump is no longer mistakable for coverage, and the contract describes what actually happens. My bidirectional reason guard could not catch this drift -- no reason string involved -- which is a useful reminder of its limits. The sync-handler drain question resolves to wording, not code: a thread cannot be interrupted, so no drain budget can bound how fast a sync handler stops -- the same fact that makes timeout= a compile error on sync handlers. The contract now says the budget bounds how long cancellable work is waited for, never how fast a thread can be made to stop, and points long sync work at async-around-rx.step.
The bounded half of the reviewer's typed-durability item, shippable
without the wire-format decision the rest of it needs.
Decimal('10.10') was stored as float 10.1: the serializer registry
converts it, so the type people reach for to avoid precision loss lost
precision silently -- in a refund handler that is a money bug with no
error anywhere on its path. bytes and bytearray became lists of
integers nothing ever turns back into bytes. Both are refused at record
time with errors that name the fix (str+Decimal or integer minor units;
explicit base64), and since TypeError is a bug-class error the attempt
fails once instead of burning retries. Tuples and sets still become
lists -- that is JSON's shape, round-trips losslessly enough, and
refusing it would break every workflow returning a tuple.
The HTTP start endpoint accepted arguments the handler's signature
refuses -- a 202 and a poison run instead of the 400 that names the
caller's bug. Supplied arguments now validate against the handler's
declared type hints; an exotic hint pydantic cannot adapt skips
validation rather than blaming the caller for it.
What remains of typed durability -- a type-preserving state encoding
and schema evolution -- changes the wire format and intersects the
API freeze, and stays a design decision rather than a patch.
Cancelling a rollout left its regional deploys deploying. Two unrelated
real-world scenarios -- a CI/CD rollback and a fleet rollout -- hit this
as a bug, and they are right: an operator presses cancel to stop the
blast radius, and a button that stops only the bookkeeping run has not
done the one thing it exists to do. Section 5 had chosen the other side
deliberately ("delegation is not ownership"); that default is wrong and
this changes it before anyone depends on it.
Any terminal transition of a run -- cancelled, failed, timed out,
force-finalized, completed -- now requests cancellation of every branch
it fanned out that is still running, in the SAME store transaction. Not
follow-up: a worker that dies mid-follow-up is exactly the case where
the deploys keep going. rx.parallel(..., parent_close="abandon") keeps
the old behaviour for work that should genuinely outlive its starter.
Each level closes only its own branches. That looks like it could
deadlock -- a tier waiting on shards that only get closed once the tier
finalizes -- but a run blocked on its own join holds no claim, so it is
control-pending the moment it is marked, finalizes, and closes the level
beneath it. A three-level test pins that.
Also backstops mode="first" race losers, whose cancellation was
best-effort follow-up from one worker; a loser is still not fenced at
commit while the parent lives on, so the side-effect warning stands.
The conformance suite -- the mechanism that is supposed to keep the
three stores identical -- was only wired to memory and sqlite. Postgres,
the one store whose SQL is hand-written, was not in it. It now joins
whenever REFLEX_TEST_POSTGRES is set; all 48 checks pass against a real
server, including the two new cascade checks. Overriding the harness
store fixture drops the run from 432 cases to 144 by not crossing two
independent store parameters.
The old all-mode test that asserted branches were left alone now runs
under parent_close="abandon", where the bug it actually guards -- a
tombstoned join misread as a decided race -- still shows.
retry, skip, force_complete and force_fail, matching what the CLI exposes. Rehearsing repair is most of what a workflow test needs to do, and reaching through harness.kernel for it read as private API -- which also meant a test reaching for a helper that was not there failed with AttributeError, and an xfail swallows that as a pass. That has bitten this suite before. The test that covers them documents three engine behaviours that make the obvious version of it wrong: a failed attempt discards its state, so an attempt counter has to live outside the run; skip only restores a successor that was preallocated, so the fixture needs a real chain; and force_* is refused on a terminal run, so it needs a run that is nonterminal and drained -- one sitting on a timer.
Section 4 documents that the engine does not pin a run to the code that started it, which invites the reading that pinning is a missing feature. It is not: it is a deployment concern, done by routing admissions, and the engine rules hold underneath whatever routing exists. Saying so keeps the contract and the (unbuilt) deploy layer from drifting into two different answers to the same question.
The suite simulated crashes in-process -- abandon a claim, expire a lease, commit behind a fence. That tests the store's logic and says nothing about what reached the disk, because the process that was supposed to have died is still there to tidy up. "Kill any process at any boundary" was an untested claim. These SIGKILL a worker subprocess at a named boundary and make a fresh process produce the documented outcome: before the effect -> work simply undone, runs once on recovery after an unguarded one -> repeats (section 2, and the reason rx.step exists) after a journal write -> replays, never charges twice after parent finalize -> branches already CANCELLING on disk The last is the one the cascade design turns on: the process that finalized dies with no chance to do anything else, so if branch cancellation were follow-up work the regions would still deploy. Deleting the _close_children_sql call makes exactly that test fail, which is the check that it is measuring what it claims to. Two ways this kind of test lies, both closed: an effect lost to the page cache would make a repeat look exactly-once, so the ledger is fsynced before each kill; and a worker that exits cleanly would pass every assertion that follows, so each scenario asserts it died by signal. Contract section 8 now says which claims rest on simulation and which on real kills, because the difference is the whole point.
Auditing section 1's exit criterion by re-reading the prose was not an audit. Enumerating what the engine can actually show an operator -- run statuses, step statuses, start and delivery dispositions, history events -- and checking each against CONTRACT.md found real holes, in both directions. wait_expired was declared and never emitted. A resolved wait records wait_resolved; an expired one recorded nothing, so "did the approval come through, or did nobody answer?" -- the exact question history exists to answer -- had to be inferred from which handler ran next. Now recorded, with the wait key and the timeout branch that ran. A signal deduplicated by sender key was silent too, which makes "the provider says it delivered" indistinguishable from a delivery that never arrived. Recorded now in the store's own transaction, so it is durable like every other history event, in all three stores. Getting there took two wrong patches: the memory and Postgres deliver() paths sit next to _apply_arrival(), whose duplicate branch is textually identical and means something else entirely -- a duplicate child arrival, not a duplicate signal. The conformance check added here is what caught the Postgres one, on a real server. signal_delivered is deleted rather than emitted: every accepted delivery already records wait_resolved or signal_buffered, and a third event for the same fact is noise. A vocabulary word that never appears makes its own absence uninformative, which is worse than not having it. Section 10 now defines the whole vocabulary, and test_contract_vocabulary.py fails if a member is added without being documented, or documented without anything emitting it. Both directions are mutation-checked. That turns "every failure scenario has one documented outcome" from a claim into something the suite enforces.
Two phase-3 items, found by walking the thing as a new developer rather than by reading it. RunHandle.result() returned Any. A result crosses the store as JSON, so callers got dicts back whatever the handler passed to rx.complete, and the type checker had nothing to say about it. It is now generic, and result(as_type=Receipt) returns a Receipt -- pyright infers it (checked with reveal_type) and pydantic validates it, so a result that does not fit raises here naming the run instead of becoming an AttributeError in the caller two frames later. Full inference from Orders.place(order) to RunHandle[Receipt] would mean threading generics through Reflex's @rx.event descriptor machinery, which every Reflex app shares. That is a bigger and riskier change than this one and is Alek's call, not mine. The CLI took full run ids only. "workflows dev" prints eight-character prefixes and "workflows list" prints full ids, so reading one and typing into the other -- the normal way these get used together -- gave "No run 'ca40d354' in this database" for an id the tool had just printed. Every command that takes a run id now resolves a prefix the way git does. An exact id is looked up directly and pays nothing; an ambiguous prefix refuses and names the candidates, because acting on the wrong run is worse than being asked to be specific. That also splits an error that used to run three causes together. A run that does not exist now says so; "unknown, already finished, or held by a worker" stays for a run that does exist and still cannot be finalized.
Alek99
force-pushed
the
alek/workflows-mvp
branch
from
August 20, 2026 18:14
b775638 to
399277b
Compare
The observer seam was built for this -- MetricsObserver's docstring promises "a metrics endpoint or an OpenTelemetry exporter is a few lines over snapshot()" -- and nothing had cashed it. A span here covers one ATTEMPT, not one run. That is the whole design question and it only has one answer: a span is in-process and time-bounded, and a durable run can wait a day, survive a restart, and execute its steps on different machines. Modelling a run as a span would mean holding one open across processes, which OpenTelemetry cannot do and no backend would draw. Every span carries workflow.run_id instead, so "everything that happened to this run" is an attribute search across however many attempts it took. Steps that never ran produce no spans, because nothing executed to time. Counters reuse MetricsObserver's event-to-name mapping rather than restating it. Two exporters that disagree about one deployment are worse than one exporter, and a test runs both observers over the same runs and asserts they report the same numbers. OpenTelemetry is not a Reflex dependency and this does not make it one: the import is lazy, so importing the module without it works, and constructing the observer raises with the pip line. Both paths are verified. It is in the dev group so CI actually exercises the code rather than skipping the file and shipping it untested. Tested through the real SDK's in-memory exporter and reader, not a mock -- whether a span actually ends, whether a failure survives into its status, and whether the counters agree are not questions a mock can answer.
Phase 5's bar is that each feature graduates only once its crash, race, authorization and versioning semantics are tested. Checking that directly, rather than assuming the general tests cover the specific features, turned up two gaps and one pleasant surprise. Authorization: I expected an unverified webhook to be an open run-starter. It is not -- compiling refuses one outright unless someone passes allow_unverified with a non-empty reason, which is better than a warning. But once opted in, doctor said nothing, and doctor exists for "the things whose absence is silent". The refusal protects whoever writes the webhook; whoever deploys it a year later is a different person reading a different surface. It is a note, not an error, because they did opt in -- it just should not be invisible. Versioning: the existing tests all deploy over plain sequential steps. A wait's continuation and a fan-out's join reach their handler by another route, carrying an injected payload or results list, so the compatibility gate applying to them as well was an unverified assumption. It does hold, and both are now pinned: removing the handler suspends the run as unknown_handler instead of raising inside a worker that is serving every other run in the deployment. The join test is mutation-checked -- restoring the removed handler makes it fail -- because a versioning test that passes for the wrong reason looks exactly like one that passes.
I had been reporting reflex deploy as blocked in "the hosting repo". That was wrong and I should have checked earlier: the deploy command is in reflex/reflex.py and the hosting CLI is vendored at packages/reflex-hosting-cli. Reading them settles what is actually external and what is not. What is not: the first thing a workflow-only project hits is assert_in_reflex_dir(), failing with "rxconfig.py not found. Move to the root folder of your project, or run reflex init to start a new project." For someone who just ran `reflex workflows init` -- which writes one module and no rxconfig.py on purpose, because there is no frontend to configure -- that instruction means scaffolding the web app they deliberately did not ask for. Deploy now recognises the case, names the modules holding workflows, says why it cannot proceed, and gives the command that does work. What is external, and stays external: hosting zips a backend and then unconditionally zips a frontend (cli.py, the two export_fn calls), and the service has no notion of an app that is workers plus a headless ingress. Making that work is a change to a running service I cannot reach or verify against from here, so this refuses honestly rather than half-implementing a path that would fail further in. The guard reads the .py files in the working directory, so it tolerates one that is not decodable text -- a check that crashed deploy before it started would be worse than the message it replaces.
All reproduced first, all fixed, all verified against a real server. 1. Deadline signals lied on Postgres. deliver() selected only status, never deadline, so a delivery to a past-deadline run answered "resolved" where Memory and SQLite answered "expired". The run then finalized TIMED_OUT and dropped the payload -- a person clicking approve was told their decision landed, moments before it was discarded. The conformance suite had no check for it, which is why three stores diverged quietly; it has one now and it failed on Postgres alone before the fix. 2. Cancelling a fan-out deadlocked against its branches. Closing a parent takes the parent row then its children; a branch reporting home takes itself then its parent. Same two rows, opposite orders. A probe measured 130 aborted transactions across 40 rounds. Fixed by ordering, not by retrying: both transactions now take branch rows first, so everything acquires children before self before parent -- deeper before shallower, which cannot cycle. 0 in 150 rounds after. Retrying would have left the latency spike and, where recovery had to converge, a cancelled rollout still running until a lease lapsed, which is the outcome cascade cancellation exists to prevent. 3. Store time moved when the wall clock moved. The offset was added to time.time() on every read, so an NTP step or a resumed snapshot carried the worker with it; it would renew its lease to a moment the store considered past, lose the claim mid-attempt, and let a peer reclaim the step. Time is now carried from the last sync by time.monotonic, which cannot jump. Only real drift remains, and the next recovery corrects it. The test for 3 initially passed while measuring nothing: the kernel decides whether to sync by identity against time.time, and monkeypatching it replaces the object the default argument was bound to, so the sync path silently turned off. It now passes the clock explicitly and asserts the path is live before testing it.
Five smaller defects, all reproduced before fixing. force_complete took the operator's result unchecked, so Memory stored a live Decimal that no other store could hold and SQLite raised a bare "not JSON serializable" from inside json.dumps. Same input, three behaviours, none of them actionable -- and the Memory case only surfaced on the day someone moved to Postgres. It now goes through the same strict serde as a handler's result and all three refuse identically. CLI prefix resolution scanned the newest 10,000 runs and treated a single hit as unique. On a bigger store another run just outside that window could share the prefix, and the next thing the resolver feeds is cancel or complete. There is no prefix query in the store protocol to do better, so it now refuses what it cannot prove and says to pass the full id. Retry.multiplier accepted nan and inf, because every comparison against nan is False and "< 1.0" was the only guard. The backoff then scheduled a step for a moment that never arrives. Cron could not see across a skipped leap century: 2100 is not a leap year, so 2096 to 2104 is eight years and the 1500-day horizon reported no occurrence for a perfectly good expression. Separately, 0 0 30 2 * parsed happily and then never fired, which is indistinguishable from a schedule that is merely waiting -- now a definition error. A weekday restriction still makes it legal, because cron matches day-of-month OR day-of-week and the date has a second path. Occurrences dropped past the catch-up cap now increment a counter on both exporters. They have no run to attach history to, so a log line was the only trace, and noticing that a nightly job silently stopped a week ago is exactly what a counter is for.
Every item reproduced before fixing, verified against a real server. Arrival deadline parity. record_arrival accepted arrivals to a past-deadline parent on all three stores, and then the atomic path disagreed: Postgres refused, Memory and SQLite resolved the join. The same fan-out behaved differently depending on what was behind it, which is worse than either answer alone. All four paths now refuse it as expired, and a conformance check covers them. Schedule seeding. _started_at was captured in __init__, before the first clock sync, so a worker running slow seeded a brand new schedule behind store time and backfilled occurrences from before the deploy. Taken at the end of the first recovery pass instead -- still "when this worker started", now on the store's clock. My first attempt moved it to the first sweep, which is not the same thing and quietly broke restart catch-up; the schedule tests caught it. Empty run id. `cancel "$RUN_ID"` with the variable unset arrives as an empty string, which prefixes every run, and with one run in the database it cancelled it and reported success. That was mine, introduced with prefix support. Refused now. The second Postgres deadlock. recover_orphans locked step rows and then updated their runs; commit takes the run and then the step. The store's stated invariant is run-first and recovery was the single path breaking it, so it now locks the run rows -- which serializes exactly as well, because every writer already obeys that order. 56 aborts in 30 rounds before, 0 in 60 after. Ordering again, not retries. Also: forced-failure error payloads go through strict serde like results; the prefix scan reads one past its cap so a database holding exactly the cap is not treated as truncated; missed occurrences are counted without a ceiling, because the count is what an alert fires on and the bound existed to limit catch-up, not accounting. And Postgres runs in CI. The conformance suite tests every store it can reach, so with no server the Postgres rows skipped and said nothing -- which is how store-specific divergence got as far as review. The workflow suite now runs a second time against a real server on Linux.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Warning
ULTRA EXPERIMENTAL — DO NOT REVIEW, DO NOT MERGE.
This is a messing-around branch for collaborating on ideas between agents. We are not at all sure any of this comes in. No review wanted; nothing here is final.
What this is
An exploration of durable workflows embedded in Reflex — a Temporal / Inngest / Trigger.dev / Zapier-class engine where a workflow is an
rx.Statesubclass, steps are@rx.event(durable=True)handlers, and control flow is the handler's return value.Core design bet: no replay — state is snapshotted per step, never rebuilt by re-executing user code, so handlers are ordinary Python with no determinism constraints.
What's in the branch (37 commits)
FOR UPDATE ... SKIP LOCKED, multi-worker, schema-namespaced) + a runnable conformance suite (32 checks) any store must passREFLEX_WORKFLOW_DATABASEresolves the store for the app, its workers, and the CLI alikerx.parallel, incl.mode="first"racing), signed single-use approval links for email (GET never decides)queue=+rx.App(workflow_queues=...))rx.steprecorded substeps (epoch-fenced journal replayed to retries and crash recoveries);rx.current_run()+ retry-stable idempotency keys; policy keys reach into model payloadsrx.workflows.*,reflex workflows list/show/cancel/resume/check(SQLite path or Postgres URL;check --jsonis the generation-loop validator)WorkflowTestHarnesswith a virtual clock; every workflow test runs against all three stores (~930 tests); Playwright integration tests prove the engine inside real dev and prod servers (browser → durable timer → completion; signed webhook over HTTP)Notable war story: an intermittent teardown hang was root-caused to a swallowed task cancellation during lease release (
task.cancelled()is never a valid "who cancelled" discriminator — third instance of that bug class here), fixed with a deterministic repro and verified over 30 clean suite runs.🤖 Generated with Claude Code