keiro
Safe HaskellNone
LanguageGHC2024

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 #1
  sleepNamed (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 suspendsrunWorkflow 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

  • sleepNamed is 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.
  • sleep is an ordinal convenience built on sleepNamed via freshOrdinal (the Nth sleep in a run gets sleep: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. Prefer sleepNamed for 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 its payload (see sleepTimerPayload / 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 without gen remain 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 uses scheduleTimerOnceTx, so the first arm's fire_at wins 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_timers schema 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 (via runWorkflowTimerWorker, or by passing workflowSleepFireAction directly to runTimerWorker). A sleep whose timer is never drained — or one whose timer an operator cancelTimers — stays suspended forever until an operator intervenes. Workflow sleeps otherwise inherit the timer subsystem's recovery surface (findStuckTimers requeueStuckTimer deadLetterTimer) for free.
Synopsis

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. sleepStepName (StepName "cool") == "sleep:cool". The prefix keeps the journal self-describing — an operator scanning it sees 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.