Workflows

Workflows keep their state in four tables. A fresh pgbus:install creates them; an existing install adds them with the generator (or rails generate pgbus:update, which detects the missing tables):

An install that ran pgbus:add_workflows before waiting steps existed adds them (the pgbus_workflow_signals table and five step columns) with rails generate pgbus:add_workflow_signals; pgbus:update detects that too. start! of a class with a fan_out or wait_for: step raises Pgbus::Workflow::WaitingStepsUnmigrated until it has run.

A workflow is a class with declared steps. Each step runs once per attempt, in order, as its own job; wait: delays it after the one before. Inside a step, subject is the record the run was started for and arguments the keywords start! received.

Step forms: step :name do … end, an anonymous step { … } (named step_1, step_2, …), step :name to run the instance method of that name, and step def name = …. Options, all checked when the class loads:

• wait: — seconds or a Duration after the previous step (for the first step, after start!). Underneath it is a PGMQ delay: the StepJob is sent with wait_until:.
• skip_if: — a lambda or method name; truthy skips the step. On a wait_for: step it runs on every claim, the one after signal! woke the step included: see "Skip a wait the record already passed" under Signals.
• timeout: — how long the step may stay in progress before the sweep treats it as stalled (default config.workflow_stall_threshold).
• job_options: — queue, priority, wait, wait_until for this step's StepJob; set_step_job_options sets them for the class. wait/wait_until only ever move a step later: they are folded into its scheduled_at when it is scheduled. Anything else raises.
• on_exception: — a step-level exception policy (below).
• wait_for:, wait_timeout:, on_timeout: — wait for a signal, with a fixed deadline or one per run (below).
• before_step: / after_step: — insert the step before or after another (useful in a subclass).

cancel_if runs before every step; truthy cancels the run. Both it and skip_if run under the workflow's row lock, before the body.

The subject is optional and loaded through GlobalID; a record that is gone makes subject return nil (react in cancel_if). Arguments go through ActiveJob::Arguments (GlobalID-aware, never Marshal) and are capped at config.workflow_max_payload_size (128 KB). A subject has at most one ongoing run per workflow class: a second start! raises Pgbus::Workflow::AlreadyRunning, whose #workflow is the run holding the slot. Subject-less runs are never limited.

idempotency_key:, actor: and reason: are reserved keywords of start!: they never reach arguments. actor: (default "app") and reason: go on the run's started event, and the actor on its first step_scheduled, as on every outside verb, so the audit trail says who started a run and why. A workflow that took an argument named actor or reason renames it (requested_by:).

start! writes the run and its first step row in one transaction; the StepJob is sent by the step row's after-commit callback, so calling start! inside your own transaction sends nothing until it commits (and nothing at all if it rolls back) when the workflow tables share your database. On a separate pgbus database (connects_to) the commit is the pgbus connection's: the run is written and its first step sent on their own, so a step can run before your transaction commits the subject (and even if it rolls back). Wait for the subject in the step body — reattempt!(wait: 5) if subject.nil? — rather than in cancel_if, which cancels the run.

Inside a step body, each verb ends the body on the spot:

• pause! — the step completes and the run pauses; resume! continues with the next step.
• cancel! — the run is cancelled.
• reattempt!(wait: 30) — this attempt fails and a new one is scheduled.
• skip! — the step is skipped and the run moves on.
• finished! — the step completes and the run finishes now.

From outside (the console, your code; the dashboard later), every verb takes actor: and reason: for the audit trail. Outside a step, skip! skips the step a waiting run waits on:

A verb that is not legal in the run's state raises Pgbus::Workflow::InvalidTransition (NotPaused for resume! and resume_at!); the table at the bottom of this page is the rule.

A fan_out step adds a group of jobs and waits for all of them, as one batch:

The body adds job instances with b.add, as often as it likes (each keeps its own .set(queue:, wait:, priority:); a block given to b.add runs inside the batch, so a perform_later in it joins too). When the body returns, one batch opens with those jobs, described OrderWorkflow#notify_vendors <workflow_id> and carrying { workflow_id, step_id } as its properties, so it shows on the Batches page. The step is waiting, and so is the run; no message is sent for it.

When every member has finished (completed or dead-lettered), the batch's on_complete callback, Pgbus::Workflow::BatchFinishedJob on the step's StepJob queue, resumes the run: with no dead-lettered member the step completes and the next one is scheduled; with any, on_batch_failure applies (:pause! by default, :cancel! or :skip!) and Pgbus::Workflow::FanOutFailed ("3 of 12 jobs dead-lettered") is recorded on the step. A batch that stalls is repaired by the batch sweep, whose finish fires the same callback; a callback that is lost (dead-lettered, or never sent) is made up by the workflow sweep once its batch has been finished for config.workflow_stall_threshold. If the batch row itself is gone (batch retention ran before a long-delayed callback), its outcome is unknown: the step fails with FanOutFailed and the run pauses, whatever on_batch_failure says, for a person to resume or skip.

• The workflow owns the batch's callbacks: b.on_complete, b.on_success and b.on_failure raise ArgumentError in the body.
• A flow-control verb (skip!, cancel!, …) or an exception in the body ends the step before any batch exists. Members join only through b.add: a bare perform_later in the body is not a member.
• An empty batch finishes at once; the step completes without waiting.
• A reattempt, or resume! after a failure, runs the whole body again and opens a fresh batch. Every member runs again, so member jobs must be idempotent: derive each member's key from step_idempotency_key (stable across attempts) plus something that tells the members apart, such as the vendor id.
• fan_out takes every step option except wait_for:, wait_timeout: and on_timeout:, and needs the batch callback-job columns (rails generate pgbus:add_batch_callback_jobs): without them start! raises Pgbus::Workflow::BatchCallbacksUnmigrated.

A wait_for: step waits for an event from outside, delivered with signal!:

The step reaches its gate after its wait: and its guards. It takes the oldest pending signal its matcher takes and runs with it as received_signal; with none it parks as waiting and nothing is sent. signal! stores the signal under the run's row lock and, when the run's step is waiting for one it matches, claims it and schedules the step; its StepJob is sent once the signal commits (so a signal! inside your transaction wakes nothing before it commits), into a queue the workers already LISTEN on, so the step runs at once.

• Buffering. A signal sent before its step waits stays pending and is taken when the step reaches its gate (unless a guard skips the step first; see below).
• Idempotency. With an idempotency_key, a redelivery stores nothing, wakes nothing and returns the stored signal, whose duplicate_delivery? is true. A blank key ("") is no key.
• Audit. actor: and reason: (both optional, on the handle and by subject) go on the run's signal_received event: wf.signal!(:shipped, actor: "webhook", reason: "carrier callback").
• Matchers. wait_for: takes a signal name, a Pgbus::Workflow::SignalMatcher.new(:shipped) { |payload| payload[:region] == "eu" }, or any object answering matches?(signal). A matcher runs under the run's row lock, so it must be fast and free of side effects. One that raises at the gate is handed to the exception policies; in signal! the error reaches the caller and nothing is stored.
• Payloads go through ActiveJob::Arguments, are capped at config.workflow_max_payload_size, and read back with indifferent access (received_signal.payload["amount"]).
• A finished or cancelled run raises Pgbus::Workflow::NotOngoing. By subject, missing: :raise (default) raises it when the subject has no ongoing run of the class; missing: :ignore returns nil. A subject-less run is signalled through its handle (OrderWorkflow.find(workflow_id).signal!).
• Lifecycle. A claimed signal is consumed when its step ends, and kept when the attempt reattempts or fails into a pause: the next attempt (reattempt!, resume!, or resume_at! of that step) runs with the same signal, without matching again.
• A paused run still takes a signal for its waiting step, but stays paused; resume! runs the step. resume! of a run paused while it waited puts it back to waiting.

Skip a wait the record already passed. A run that starts late, or a later signal that overtakes an earlier wait, finds the record past the state the step waits for. The guard for that reads signal_ready?:

Guards run before the gate, on every claim of the step. Without the reader, skip_if: -> { subject.shipped? } would also skip the step signal! woke once the order shipped (its body would never run), and a signal sent before the step's first claim would stay pending for good. Two step readers, usable in skip_if:, cancel_if and the body, tell the cases apart:

• signal_ready? — the gate runs this attempt with a signal: one the step holds, or a pending one its wait_for: matcher takes. The guard above skips only when no signal is held or pending for the step; a woken step and one whose signal came first both complete: the gate claims the signal, and it is consumed when the step ends.
• woken? — narrower: this attempt holds a signal before its gate (signal! woke it, or an earlier attempt took one and reattempted or failed into a pause). False on a first claim, even when a buffered signal waits for the gate, so !woken? in that guard skips the step and leaves that signal pending.
• Both only read (the gate claims, after the guards), keep one answer for the whole attempt, are false after an on_timeout: :run wake (the gate takes no signal then) and in a step without wait_for:, and raise Pgbus::Workflow::NotInStep outside a step.
• A guard that skips (or a cancel_if that cancels) takes no signal: a pending one stays pending for a later wait_for: step that matches it, as with an outside skip!. A signal the step already holds is consumed, as when the body calls skip!.

Timeouts. A signal that never comes is the usual way a workflow gets stuck, so a waiting step can carry a deadline: wait_timeout: (seconds or a Duration) and on_timeout:, one of :pause! (default: the step fails with Pgbus::Workflow::WaitTimedOut), :cancel!, :skip!, :run (the body runs with received_signal nil and timed_out? true) or :notify (below). timeout: keeps its meaning on every step: it bounds the in-progress attempt from its claim, while wait_timeout: bounds the wait before it. The sweep fires timeouts, so one fires within config.workflow_sweep_interval of its deadline. A paused run's wait does not time out while it is paused, and the time spent paused does not count against it: resume! moves the deadline by the length of the pause. A signal that arrives after the timeout fired stays pending, except under :notify.

A deadline per run. wait_timeout: also takes a method name or a lambda, evaluated on the workflow instance (subject, arguments) when the step parks, returning a duration or a Time, or nil:

• A Time (or anything with #to_time that is not a String) is the deadline itself, however long the run took to reach the wait: the steps before it, queue latency, a reattempt. A deadline already past is timeout_at = now, so the next sweep applies on_timeout.
• A Duration or a number of seconds is counted from park, as a static wait_timeout: is, and must be positive.
• nil parks with no deadline for this attempt.
• Anything else raises ArgumentError at the gate, which, like an error the callable raises, goes to the step's exception policies (the default pauses the run): the step never parks silently.

It is evaluated once per attempt, when the step parks, under the run's row lock, so it must be fast and free of side effects, as a matcher is. It runs in a savepoint: a query of its own that fails in the database goes to the exception policies like any other error. Under on_timeout: :run, a step that reattempts parks again and evaluates it again: a callable that keeps returning the same past deadline times out each attempt at once. A paused run's deadline moves by the length of the pause, a Time one too. A static wait_timeout: is unchanged, and checked when the class loads.

A soft deadline. on_timeout: :notify reports a late signal and keeps waiting for it: a late notice writes a wait_timed_out event (action: "notify"), sends pgbus.workflow.wait_timed_out and counts once in pgbus_workflow_wait_overdue_total, while the same attempt stays waiting and its deadline moves one wait_timeout: on. No attempt fails, so the step's outcomes and the step-failing alert below are untouched, and a signal that arrives later wakes that attempt as usual (timed_out? is never true under :notify). A sweep that missed several deadlines (the dispatcher was down) reports them once and moves to the next one ahead. A per-run wait_timeout: is evaluated again at each notice: a Time after now is the next deadline, a duration moves the deadline one period on as a static one does, and nil or a Time not after now (the same expires_at) leaves the wait without a deadline: it was reported once and keeps waiting for its signal (the Workflows list says since when it is late, and that no further late notice is due). One that raises there, or returns anything else, pauses the run with that error: the sweep holds the run's row lock, under which no on_exception block may run.

A wait without a timeout has three escape hatches: skip! (the run moves on), cancel!, or pause! then resume_at!. One waiting longer than config.workflow_waiting_visibility_threshold (7 days) counts as stuck (waiting_overdue).

An exception in a step body never reaches ActiveJob. The step row records the error (class, message, cleaned backtrace, cause chain, the worker's hostname:pid and the app revision), the error goes to Pgbus::ErrorReporter, and by default the run pauses. A policy says otherwise:

Resolution order: the step's own policies, then class-level policies that name the error, then blanket ones, then pause!. matching: takes classes, class names or a list of either. max_reattempts: counts consecutive reattempted attempts of the step; once spent, terminal_action: (default pause!) applies. report: is :always (default), :never, or :terminal_only (not on a reattempt that still has attempts left). A plain rescue in the body bypasses the policies: your code decides.

To the executor the StepJob always succeeds, whatever the step's outcome; job stats count it as completed. The step row and the pgbus.workflow.* events are the truth about the step.

A step can run more than once: a worker that dies mid-step leaves part of the body done, and the stall action may run it again. Four things make that safe:

• Outputs. output(**values) records values on the attempt, in the same UPDATE that completes the step, so they exist exactly when the step does. output_of(:name) returns the latest completed attempt's output (with indifferent access), or nil; outputs returns all of them. A failed attempt keeps its output on its own row for debugging, but output_of never returns it. Outputs are capped like arguments; an unserialisable value raises inside the step and goes through the policies. The body's return value is ignored.
• Step keys. step_idempotency_key is "pgbus-wf:<workflow_id>:<step>", the same on every attempt and on a resume_at! re-run: hand it to the external API so the side effect happens once. attempt_idempotency_key adds :<attempt> for the rare call that must differ per attempt.
• Idempotent start. start!(…, idempotency_key:) returns the run that already has that key, in whatever status, with duplicate_start? true, no new step and no new event: the first caller's actor: stays on the trail. The key is a reserved keyword. A different key for a subject with an ongoing run still raises AlreadyRunning.
• Completed steps stay completed. resume! and the policies never re-run a completed step; only resume_at! does, on purpose.

The engine's own messages are idempotent too: the claim (scheduled → in_progress, only once due) makes a duplicate StepJob a no-op — except a redelivery of the very message that claimed a still-running step, which applies the stall action (see Housekeeping) — and the unique (workflow_id, step_name, attempt) index makes a duplicate attempt impossible. A step still has to be idempotent: the key is how.

A run that is scheduled on a step the class no longer declares pauses with Pgbus::Workflow::StepNotDefined. To retire a step, replace it with a gravestone where it used to be:

Every run stores the SHA256 of its class's ordered step list when it starts (definition_digest); Pgbus::WorkflowEntry#definition_outdated? tells a run that began on an older definition.

The dispatcher looks after workflows on its own:

• A dead worker is noticed in seconds, while config.visibility_heartbeat is on (the default). StepJob keeps the heartbeat, so a live worker keeps its message invisible; when the very message that claimed a step comes back while the row is still in progress, its worker died (or its heartbeat lapsed, for example through a long pause), and config.workflow_stall_action applies at once. With the heartbeat off, only the sweep below catches a dead worker.
• A hung step (in progress past its timeout:, or past config.workflow_stall_threshold) gets the same stall action from the sweep, every config.workflow_sweep_interval. :pause (default) records Pgbus::Workflow::StepStalled, :reattempt schedules a new attempt, :cancel cancels the run.
• A lost message — a scheduled step of a ready run, due more than the threshold ago and not sent within it — is sent again.
• A wait past its wait_timeout: gets its on_timeout: (see Signals). A waiting step is never treated as stalled.
• A lost fan-out callback — a fan_out step still waiting on a batch that finished more than the threshold ago — is resolved as its BatchFinishedJob would have. One whose batch row is gone pauses with FanOutFailed (the outcome is unknown).
• Retention: finished and cancelled runs older than config.workflow_retention (30 days) are deleted hourly, with their steps and events.

A step that was declared stalled while it was in fact still running finds its row no longer in progress when it ends: its outcome and output are discarded and a late_completion_ignored event says so.

The dashboard's Workflows page (/pgbus/workflows) lists every run, newest first: the ongoing ones by default, or one state, or all. Each row says in words what the run is doing and what happens next: "Performing capture (attempt 2)", "Waiting for signal shipped since 3h ago", "Waiting for batch OrderWorkflow#notify_vendors (7 of 12 jobs done)", "Paused on capture: Timeout::Error — resume or cancel", "Scheduled: reminder runs in 2d". Filter by class, current step, error class, subject, workflow id and a created-at range; the filters combine and survive paging.

A run's page shows its arguments, a timeline (one range bar per attempt, split into waited, queued and ran, coloured by how it ended, with waits and pauses as their own segments; zoom for runs that span days), its newest 100 attempts with the same durations as text, its worker, app revision, error, backtrace and cause chain, the signals it received, and the audit trail ("Resumed by dashboard:alice: SMTP fixed"). A run that started on an older step list says so.

Actions show only when available_actions allows them: Resume (its label says whether it retries the failed step or continues with the next), Resume at a chosen step, Run now (a scheduled step's wait cut short), Skip step (a waiting step), Send signal (a JSON object; the form's idempotency key makes a double submit send one signal), Rerun (finished or cancelled runs; the new run links back) and Cancel. Each takes an optional reason and reports the real outcome: when the run moved on first, the flash says where it is now.

Needs attention (/pgbus/workflows/attention) groups every stuck run by class, step, stuck reason and error class, with a count, when the group was first and last seen and the newest error message (the 200 biggest groups; past that the page is not exhaustive). Once the cause is fixed, Resume or Cancel the whole group (at most 500 runs per press, each through its own guarded transition): "Resumed 212; 2 had already moved on". The home page counts ongoing, paused and stuck runs.

Two settings connect the page to your app:

The MCP server's read tools have the same views: pgbus_workflows (the list, with the page's filters), pgbus_workflow (one run, with available_actions) and pgbus_workflows_attention. Arguments, step outputs, signal payloads and error text are redacted unless the server allows payloads and the call sets include_payloads.

Acting over MCP is a separate opt-in. With the write tools on, an agent that found the paused runs can resume them after the fix:

That adds pgbus_workflow_resume, pgbus_workflow_cancel, pgbus_workflow_skip_step, pgbus_workflow_resume_at (with step), pgbus_workflow_run_now and pgbus_workflow_signal (with name, a JSON object payload of at most workflow_max_payload_size and an optional idempotency_key, which a retried call should pass; a blank one is no key). Each takes workflow_id, a required reason (1–500 characters after trimming whitespace) and an optional expected_status, the status the agent read: a run that has moved on by the time the call starts answers ok: false and nothing is done (the check runs just before the action, whose transition is guarded again under the run's row lock). A verb the run's available_actions does not allow answers ok: false with the allowed list. Otherwise the call goes through the same action as the dashboard's button and lands on the audit trail with actor mcp (mcp:<client name> when the transport knows the client) and the reason it was given; a refused call changes nothing and records nothing. Every call answers { ok:, reason:, workflow: } with the run as it is after the call (payloads always redacted; a transition refused under the row lock adds code and the status the run is in now; an unexpected failure answers a fixed reason, the error is in the pgbus log). They are marked destructiveHint: true (idempotentHint: true on cancel only), and there are no bulk tools: pgbus_workflows_attention groups the stuck runs, pgbus_workflows lists a group's runs (status, workflow_class, step, error_class), and each run gets one audited call.

Enable them only on an authenticated endpoint: Pgbus::MCP.rack_app with token: or auth:, or mounted inside your app's authenticated routes, never public. An unauthenticated rack_app with the write tools on logs a warning saying anyone who reaches it can change workflow runs.

Workflow insights (/pgbus/workflows/definitions, linked from the Workflows page as "Insights by class") answers how a workflow class is doing, not where one run is. The list has one row per class (every registered Pgbus::Workflow subclass and every class name still in pgbus_workflows): runs going on now, paused, needing attention, the runs that finished or were cancelled in the window, and the p50/p95 end-to-end duration (created → finished). A class that no longer loads says "definition not loaded"; one with ongoing runs on a removed_step says how many.

A class's page (/pgbus/workflows/definitions/OrderWorkflow, or Billing::OrderWorkflow) shows its steps in order as a pipeline: each step with the runs sitting at it now (scheduled, running, waiting, paused), the p50/p95/p99 run time and the pickup latency (scheduled → started) of its attempts that finished in the window, and their failure (an error was recorded), reattempt and skip rates. A step that waits for a signal also shows its wait timeout: its wait_timeout:, or "dynamic (per run)" for a method name or a lambda. Percentiles over fewer than 20 attempts carry their sample size and a "few samples" note. Below: runs started, finished and cancelled per bucket (5 min for 1 h, 1 h for 24 h, 6 h for 7 d, 1 d for 30 d), the end-to-end duration distribution, a funnel (of the runs started in the window, how many reached each step and where each one is now) and the same numbers for the last five app revisions that finished an attempt, newest first, so a regression after a deploy shows as a row (a failure rate more than five points above the revision before is marked). The window is 1 h, 24 h (default), 7 d or 30 d, capped at workflow_retention. Every count of runs a list filter can express (runs at a step now, by status, started or ended in the window, where the funnel's runs are now) links to the Workflows list filtered to exactly those runs; attempt-level numbers (percentiles, rates, the funnel's "reached", the per-revision rows) are text. The list gained ended_from/ended_to (when a run finished or was cancelled) for it, and the need-attention counts open Needs attention narrowed to the class (/pgbus/workflows/attention?workflow_class=OrderWorkflow).

The numbers come from the step and run rows, on the indexes the tables already have: on about 1 M step rows over 20 classes a class's page reads in about 25 ms and the list in about 100 ms over 24 hours (the budgets rake bench:workflow_insights checks: 250 and 500 ms); over 30 days both take about 300 ms, which the bench reports without a budget. The MCP server's read tools have the same: pgbus_workflow_definitions(window:) and pgbus_workflow_stats(workflow_class:, window:), and pgbus_workflows_attention takes workflow_class:.

Alerts belong in Prometheus or AppSignal, on the metrics below. Three starting points: a step whose attempts end in a reattempt or a pause (an exception, or the body's own reattempt!/pause!) more than 5 % of the time for 15 minutes, runs needing a person for an hour, and a :notify wait that missed its deadline:

On AppSignal, the bundled main dashboard (pgbus dashboard main) has a workflows row: steps per outcome, step run time, workflows by status, stuck workflows and late waits.

Audit trail. Every transition and every outside verb writes a pgbus_workflow_events row in the same transaction, with its actor ("app" by default, "system" for the StepJob and the sweep, or the actor: you pass) and reason. wf.events returns them in order: started, step_scheduled, step_started, step_completed, step_failed, step_skipped, step_reattempted, paused, resumed, cancelled, finished, run_now, rescheduled, rerun, stall_detected, late_completion_ignored, and for waiting steps step_waiting (details: wait_for or batch_id), signal_received (name, idempotency_key, duplicate), signal_claimed, wait_timed_out (wait_for, action, the missed timeout_at), fan_out_opened (batch_id, size) and fan_out_finished (failed_jobs, total_jobs). start! and signal! take actor: too.

Metrics reach Prometheus, StatsD and AppSignal with the same names (one exception for StatsD, below the table):

StatsD reads timer values as milliseconds, so it receives the step histogram as pgbus_workflow_step_duration_ms (seconds × 1000). pgbus_workflow_wait_overdue_total counts the late notices of on_timeout: :notify waits (one per deadline, or one for a sweep that missed several); the other on_timeout: values end the wait and show in the step and run counters instead.

Stuck reasons (the pgbus_workflows_stuck gauge): paused_error (paused on a step that failed), stalled (paused on StepStalled), paused_manual (the step called pause!, or an outside pause!), overdue (a scheduled step nobody picked up), waiting_overdue (a wait without a deadline older than workflow_waiting_visibility_threshold: no wait_timeout:, or a per-run one that set none or cleared it after a :notify notice; a :notify wait with a deadline has pgbus_workflow_wait_overdue_total instead).

Errors reach Pgbus::ErrorReporter with { workflow_class:, workflow_id:, step_name:, attempt:, outcome: } as context. Logs: while a step runs, Pgbus's own log lines carry workflow=… step=… attempt=… through Pgbus::LogFormatter, and your code reads Pgbus::Workflow::StepContext.workflow_id (and step_name, attempt) to tag its own. The instrumentation events are pgbus.workflow.step, pgbus.workflow_finished, pgbus.workflow_paused, pgbus.workflow_cancelled, pgbus.workflow.stall_detected, pgbus.workflow.signal, pgbus.workflow.wait_timed_out and pgbus.workflow.gauge.

Pgbus::Workflow::TestHelpers runs steps inline against your test database, ignoring their wait:, through the real StepJob:

speedrun_workflow stops at paused, finished and cancelled, and raises Pgbus::Workflow::StepWaiting on waiting, naming the signal or the batch the step waits for. complete_fan_out completes every member of the waiting step's batch (without running them) and runs its BatchFinishedJob inline; deliver_signal calls signal! and runs the step it scheduled; expire_wait applies the waiting step's on_timeout whatever its deadline (under :notify the step stays waiting: assert the wait_timed_out event and the moved timeout_at).

• pgbus_workflows — one row per run (Pgbus::WorkflowEntry): the class, the subject, the arguments, the status and the current step. A partial unique index keeps one ongoing run per class and subject; another dedupes idempotency_keys.
• pgbus_workflow_steps — one row per step attempt (Pgbus::WorkflowStep), unique on (workflow_id, step_name, attempt), with the timings, the worker, the app revision, the output and the error.
• pgbus_workflow_events — the append-only audit trail (Pgbus::WorkflowEvent).
• pgbus_workflow_signals — one row per signal (Pgbus::WorkflowSignal), unique on (workflow_id, idempotency_key).

Every StepJob is sent by WorkflowStep's after-commit callback, and only there: a new scheduled row, a row going back to scheduled, or a bumped enqueued_at (how the sweep and resume! re-send). A StepJob claims its row (scheduled → in_progress, only once due) and the run (ready → performing) under the run's row lock, runs the guards and the body, then applies the outcome under the lock again: one step at a time per run, every change a guarded UPDATE from the table below. A waiting step goes back to scheduled when it is woken (a signal, or a timeout with :run), so its resume is a StepJob like any other; a fan-out's batch, a timeout or an operator skip! resolves the waiting row directly.

Pgbus::Workflow::States is the one rule for which verb is legal in which state, for the runner, available_actions and this page. The outside verbs:

The StepJob's and the sweep's own moves, never offered as actions:

Workflow states: ready, performing, waiting, paused (the four ongoing ones), finished, cancelled. Step states: scheduled, waiting, in_progress, completed, failed, cancelled, skipped. A run is waiting while its step waits for a signal or a fan-out batch.