| Safe Haskell | None |
|---|---|
| Language | GHC2024 |
Keiro.Workflow.Sleep
Description
Durable sleep for workflows, backed by the existing keiro_timers table.
What this gives you
A workflow author can insert a durable pause between steps:
demo :: (Workflow:>es,Store:>es, IOE:>es) => Eff es (Int, Int) demo = do a <-step(StepName "a") (liftIO incr) -- side effect #1sleepNamed(StepName "cool") 300 -- durable wait, survives a restart b <-step(StepName "b") (liftIO incr) -- side effect #2 pure (a, b)
On the first run, demo executes a, journals it, arms a Postgres timer
for the sleep, and suspends — runWorkflow returns
Suspended and the journal wf:demo-<id> holds only the StepRecorded "a"
event. The b side effect has not run. The process can now crash, be
redeployed, or sit idle for the full delay: the only durable state of the pause
is a single row in keiro_timers, so the wait survives a restart with __no
external scheduler and no in-memory timer thread__.
When the timer becomes due, the existing timer worker
(runTimerWorker) fires it through workflowSleepFireAction,
which recognises the row as a workflow sleep (by the JSON payload
discriminator), reconstructs the journal stream, and appends a
StepRecorded "sleep:cool" completion. A later run replays: step "a"
short-circuits to its recorded result, sleepNamed "cool" sees its completion
already journaled and returns immediately, and only step "b" runs for real.
sleepNamed vs. sleep
sleepNamedis the stable primitive. Replay matches the sleep on its (prefixed)StepName, so the name must be deterministic across replays. A user-supplied name is unconditionally stable: the same source always produces the same name regardless of how surrounding code is reordered between deploys.
sleepis an ordinal convenience built onsleepNamedviafreshOrdinal(theNth sleep in a run getssleep:N). Its determinism is conditional: reordering or inserting sleeps between deploys shifts the ordinals and can make a resumed in-flight workflow re-arm a different timer. PrefersleepNamedfor anything that must survive a code change mid-flight.
Operational contract
- Payload discriminator. A workflow-sleep timer row carries
{"kind":"keiro.workflow.sleep","step":"sleep:<suffix>","gen":0}in itspayload(seesleepTimerPayload/parseSleepPayload). This is how a single timer worker distinguishes workflow sleeps from ordinary process-manager timers, routes each correctly, and pins a fire to the generation that armed it. Legacy payloads withoutgenremain supported.
- Deterministic timer id. The timer id is a v5 UUID over
("keiro":"workflow-sleep":name:id:generation:sleepStepName)for generation 1 and later, while generation 0 keeps the legacy("keiro":"workflow-sleep":name:id:sleepStepName)shape. The workflow sleep arm usesscheduleTimerOnceTx, so the first arm'sfire_atwins and every resume that re-enters the not-yet-resolved sleep leaves the row untouched. The sleep duration is measured from the first arm, not from the latest resume pass. - No
keiro_timersschema change. Routing is entirely a function of the caller-supplied fire action and the JSON payload; this module owns no migration. - A worker must drain the timers. A sleep only ever fires if some timer
worker runs
workflowSleepFireAction(viarunWorkflowTimerWorker, or by passingworkflowSleepFireActiondirectly torunTimerWorker). A sleep whose timer is never drained — or one whose timer an operatorcancelTimers — stays suspended forever until an operator intervenes. Workflow sleeps otherwise inherit the timer subsystem's recovery surface (findStuckTimersrequeueStuckTimerdeadLetterTimer) for free.
Synopsis
- sleepNamed :: forall (es :: [Effect]). (Workflow :> es, Store :> es, IOE :> es) => StepName -> NominalDiffTime -> Eff es ()
- sleep :: forall (es :: [Effect]). (Workflow :> es, Store :> es, IOE :> es) => NominalDiffTime -> Eff es ()
- workflowSleepFireAction :: forall (es :: [Effect]). (Store :> es, IOE :> es) => TimerRow -> Eff es (Maybe EventId)
- runWorkflowTimerWorker :: forall (es :: [Effect]). (IOE :> es, Store :> es) => Maybe KeiroMetrics -> UTCTime -> (TimerRow -> Eff es (Maybe EventId)) -> Eff es (Maybe TimerRow)
- drainWorkflowSleepTimers :: forall (es :: [Effect]). (IOE :> es, Store :> es) => Maybe KeiroMetrics -> UTCTime -> Int -> (TimerRow -> Eff es (Maybe EventId)) -> Eff es Int
- sleepTimerId :: WorkflowName -> WorkflowId -> Int -> Text -> TimerId
- sleepStepName :: StepName -> Text
- sleepTimerPayload :: Int -> Text -> Value
- parseSleepPayload :: Value -> Maybe (Text, Maybe Int)
- matchSleepTimerGeneration :: WorkflowName -> WorkflowId -> Int -> Text -> TimerId -> Maybe Int
- workflowSleepKind :: Text
Authoring surface
sleepNamed :: forall (es :: [Effect]). (Workflow :> es, Store :> es, IOE :> es) => StepName -> NominalDiffTime -> Eff es () Source #
Durably pause the workflow for delta under the stable name userStep.
On the first encounter this arms a deterministic keiro_timers row and
suspends the run; the suspension resolves when a timer worker fires the row
(see workflowSleepFireAction) and journals the sleep's completion, after which
a later run replays past the sleep without re-arming.
The arming action is idempotent (a resumed workflow re-runs it until the sleep
resolves): it inserts with a deterministic sleepTimerId only when the row is
absent, so the first fire_at persists across every resume pass.
sleep :: forall (es :: [Effect]). (Workflow :> es, Store :> es, IOE :> es) => NominalDiffTime -> Eff es () Source #
Durably pause the workflow for delta under an ordinal name (the Nth
sleep in a run becomes sleep:N). Convenient but its determinism is
conditional — see the module header. Prefer sleepNamed for anything that must
survive a code change mid-flight.
Firing and worker wiring
workflowSleepFireAction :: forall (es :: [Effect]). (Store :> es, IOE :> es) => TimerRow -> Eff es (Maybe EventId) Source #
The fire action for workflow-sleep timers. For a TimerRow whose payload
is a workflow-sleep discriminator, reconstruct the workflow identity from the
row's processManagerName (= workflow name) and correlationId (= workflow
id), resolve the generation that armed the timer, append a StepRecorded
completion (result = null) to that generation's journal, and return the
deterministic EventId so the worker marks the timer Fired. Returns
Nothing for a row whose payload is not a workflow sleep, or for a sleep
owned by a terminal workflow (whose timer is cancelled), so a mixed worker can
delegate that row to its process-manager fire action.
Idempotent: prepareJournalAppend pre-checks the generation-scoped step and
the event id is deterministic, so at-least-once timer firing yields
exactly-once journaling even after the workflow has rotated.
runWorkflowTimerWorker Source #
Arguments
| :: forall (es :: [Effect]). (IOE :> es, Store :> es) | |
| => Maybe KeiroMetrics | |
| -> UTCTime | |
| -> (TimerRow -> Eff es (Maybe EventId)) | Fallback fire action for non-sleep (process-manager) timers. |
| -> Eff es (Maybe TimerRow) |
A timer worker pass that handles both workflow-sleep timers and ordinary process-manager timers. For each claimed timer: if its payload is a workflow sleep, wake the workflow; otherwise delegate to the supplied process-manager fire action.
A deployment that runs only workflows can pass \_ -> pure Nothing as the
fallback (or use workflowSleepFireAction directly with
runTimerWorker). A deployment that mixes process-manager timers
and workflow sleeps passes its existing PM fire action so one worker drains
both kinds.
drainWorkflowSleepTimers Source #
Arguments
| :: forall (es :: [Effect]). (IOE :> es, Store :> es) | |
| => Maybe KeiroMetrics | |
| -> UTCTime | |
| -> Int | Maximum timers to process in this pass. |
| -> (TimerRow -> Eff es (Maybe EventId)) | Fallback fire action for non-sleep (process-manager) timers. |
| -> Eff es Int |
runWorkflowTimerWorker over a whole batch: drain up to limit due timers
in one pass, routing each to the workflow-sleep fire action or the supplied
process-manager fallback exactly as the single-claim worker does.
This is what a deployment wants when many sleeps come due together. The
single-claim worker drains a backlog of K due sleeps at one per poll tick,
so the last sleeping workflow waits K ticks to wake; one drain pass with a
large enough limit wakes them all. Per-timer semantics are unchanged
(drainDueTimersWith), so a sleep still fires at-least-once and
still pins to the generation that armed it.
Timer id, payload, and step-name helpers
sleepTimerId :: WorkflowName -> WorkflowId -> Int -> Text -> TimerId Source #
The deterministic timer id for a sleep. Generation 0 keeps the legacy v5
UUID over ("keiro":"workflow-sleep":name:id:fullStep) so in-flight
pre-change timers remain signalable. Generations 1 and later include the
generation component so a sleep after continueAsNew never
collides with a prior generation's terminal timer row.
The seed is hashed as UTF-8 bytes (identitySeedBytes), which is
byte-identical to the original codepoint encoding for ASCII seeds and
collision-free for the rest; see
docs/adr/0024-deterministic-ids-hash-utf-8-seed-bytes-and-are-frozen-replay-identity.md.
sleepStepName :: StepName -> Text Source #
The durable journal step name for a sleep: the user's suffix prefixed with
sleepStepPrefix. . The
prefix keeps the journal self-describing — an operator scanning it sees
sleepStepName (StepName "cool") == "sleep:cool"sleep: and knows the entry is a durable wait, not an ordinary step.
sleepTimerPayload :: Int -> Text -> Value Source #
Build the JSON payload carried on a workflow-sleep timer row. The first
argument is the generation that armed the timer; the second is the full
"sleep:<suffix>" journal step name the firing will record.
parseSleepPayload :: Value -> Maybe (Text, Maybe Int) Source #
Recognise and extract a workflow-sleep payload. The result contains the
full step name and the generation when the payload records one. Nothing in
the generation slot denotes a legacy workflow-sleep payload written before
generation pinning; an overall Nothing denotes any other timer (for example,
a process manager's).
matchSleepTimerGeneration :: WorkflowName -> WorkflowId -> Int -> Text -> TimerId -> Maybe Int Source #
Recover the generation represented by a deterministic workflow-sleep
timer id. Candidate generations from currentGen down to zero are tested
against sleepTimerId; this lets a new worker pin legacy payloads that do not
carry an explicit generation. Returns Nothing only for an operator-crafted
or otherwise non-matching timer id.
workflowSleepKind :: Text Source #
The payload "kind" tag that marks a keiro_timers row as a workflow
sleep, distinguishing it from an ordinary process-manager timer so a single
timer worker can route each correctly.