Workflows

For River Pro for Go v0.24.0+, read the workflow rollout notes before using signals, timers, or workflow-aware retention in production.

Workflows are created with a workflow builder struct using Client.NewWorkflow, and tasks are added to the workflow until it is prepared for insertion. Jobs and args are defined like any other River job.

import (
    "riverqueue.com/riverpro"
)

// MyJobArgs is a sample River JobArgs struct
type MyJobArgs struct {
    // ...
}

func (MyJobArgs) Kind() string { return "my_job" }

func SampleWorkflow(client *riverpro.Client[pgx.Tx]) *riverpro.Workflow[pgx.Tx] {
    // Create a new workflow:
    workflow := client.NewWorkflow(&riverpro.WorkflowOpts{Name: "My first workflow"})

    // Add a first task to the workflow, named "a":
    taskA := workflow.Add("a", MyJobArgs{}, nil, nil)

    // Fan-out to tasks b1 and b2, which both depend on task a:
    taskB1 := workflow.Add("b1", MyJobArgs{}, nil, &riverpro.WorkflowTaskOpts{Deps: []string{taskA.Name}})
    taskB2 := workflow.Add("b2", MyJobArgs{}, nil, &riverpro.WorkflowTaskOpts{Deps: []string{taskA.Name}})

    // Fan-in to task c, which depends on both b1 and b2:
    taskC := workflow.Add("c", MyJobArgs{}, nil, &riverpro.WorkflowTaskOpts{Deps: []string{taskB1.Name, taskB2.Name}})

    var _ = taskC // avoids "declared and not used" error

    return workflow
}

func main() {
    ctx := context.Background()
    riverClient, err := riverpro.NewClient(riverpropgxv5.New(dbPool), &riverpro.Config{
        Config: river.Config{
            Queues: map[string]river.QueueConfig{
                river.QueueDefault: {MaxWorkers: 100},
            },
            Workers: workers,
        },
    })

    // Prepare the workflow for insertion and validate it:
    workflow := SampleWorkflow(riverClient)
    result, err := workflow.Prepare(ctx)
    if err != nil {
        panic(err)
    }

    // The result.Jobs field holds a slice of river.InsertManyParams. Enqueue
    // them with InsertMany or InsertManyTx:
    if _, err := riverClient.InsertMany(ctx, result.Jobs); err != nil {
        panic(err)
    }

    // continue execution, stop client, etc...
}

Create a builder with River::Pro.workflow, add named tasks with add, and insert workflow.jobs through a River::Pro::Client. The builder itself needs no client or database:

require "riverqueue-pro"

class MyJobArgs
  def kind = "my_job"
  def to_json = JSON.generate({})
end

workflow = River::Pro.workflow(name: "My first workflow") do |flow|
  task_a = flow.add(:a, MyJobArgs.new)

  # Fan out: both tasks depend on a.
  task_b1 = flow.add(:b1, MyJobArgs.new, after: task_a)
  task_b2 = flow.add(:b2, MyJobArgs.new, after: task_a)

  # Fan in: c waits for both b1 and b2.
  flow.add(:c, MyJobArgs.new, after: [task_b1, task_b2])
end

client = River::Pro::Client.new(driver,
  config: River::Pro::Config.new(
    core: River::Config.new(queues: {default: 100}, workers: workers)
  ))

# Validate dependencies and cycles, then insert all tasks atomically.
client.insert_many(workflow.jobs)

Register workers for each task kind as with any other Ruby job.

New workflows are created with Client.NewWorkflow, which takes an optional WorkflowOpts struct. A workflow Name is not required, but a readable name makes workflows easier to identify in River UI and operational logs.

The workflow’s ID is automatically generated and does not need to be specified. It can be customized through workflow options, but custom IDs must be globally unique because the workflow ID is used for task scheduling. Prefer ID schemes that sort lexicographically, like the default ULID scheme.

Tasks are added with WorkflowT.Add. Each task must have a unique name within the workflow, and task names are used to declare dependencies through the Deps field on WorkflowTaskOpts. When the workflow is ready, call WorkflowT.Prepare or WorkflowT.PrepareTx to validate the graph and produce the river.InsertManyParams enqueued with Client.InsertMany or Client.InsertManyTx.

River::Pro.workflow(name: ...) creates a workflow with an automatically generated ID. A readable name helps identify it in River UI. If you supply id:, it must be globally unique; prefer a lexicographically sortable ID such as a ULID.

flow.add(name, args, after: ...) returns a task reference. after: accepts a reference, a string or symbol name, or an array of either. Names permit forward references, but references must belong to the same builder. Pass insertion options such as queue:, priority:, scheduled_at:, and max_attempts: directly to add.

Calling workflow.jobs validates the graph and returns fresh River::InsertManyParams values. Insert them with client.insert_many(workflow.jobs); to include them in an application transaction, use an ActiveRecord or Sequel transaction on the driver’s connection. Workflow tasks cannot enable uniqueness, and repeated insertion of the same builder is not idempotent. Build a new workflow for each run.

Start from a job already in the workflow with Client.WorkflowFromExisting or, outside a worker, from a workflow ID with Client.WorkflowFromExistingID. Add tasks with WorkflowT.Add, then validate and insert them with WorkflowT.Prepare or WorkflowT.PrepareTx.

When extending a workflow from inside a task worker, prefer a transaction that prepares the new tasks, inserts them with Client.InsertManyTx, and completes the current job with river.JobCompleteTx. That keeps the workflow extension and the task completion atomic.

Use client.workflow_from_existing(job.row) inside a worker to reconstruct a builder from its current job. Add tasks to the builder, then insert its jobs.

When extending from a worker, keep insertion and completion atomic with the driver’s application transaction. For ActiveRecord:

ActiveRecord::Base.transaction do
  extension = client.workflow_from_existing(job.row)
  extension.add(:notify, NotifyArgs.new, after: job.metadata.fetch("task"))
  client.insert_many(extension.jobs)
  client.job_complete_tx(job)
end

With Sequel, use DB.transaction. Outside a worker, first load a task with client.workflow(workflow_id).load_task(:task_name) and pass its job to workflow_from_existing. Existing task names are captured when the builder is created, so coordinate concurrent extensions.

Within an existing workflow, or with a reference to an existing job in the workflow, a task’s dependencies can be loaded using WorkflowT.LoadDeps and WorkflowT.LoadDepsTx. These methods return a WorkflowTasks collection where loaded tasks and outputs can be fetched by task name:

// Load all dependencies recursively:
tasks, err := workflow.LoadDeps(ctx, taskName, &riverpro.WorkflowLoadDepsOpts{Recursive: true})
if err != nil {
    return err
}

// Fetch a dependent task by name:
taskA := tasks.Get("dependency_a")

// Fetch the output of a dependent task:
var taskAOutput MyJobArgs
if err := tasks.Output("dependency_a", &taskAOutput); err != nil {
    return err
}

To load the full set of tasks, use WorkflowT.LoadAll or WorkflowT.LoadAllTx. To load one task, use WorkflowT.LoadTask or WorkflowT.LoadTaskTx. Loaded WorkflowTaskWithJob values expose a Wait field, which is nil for tasks without wait conditions; for wait-specific status, see Wait diagnostics.

Use client.workflow(workflow_id) to get a handle for database operations. Load dependencies with load_deps; recursive: true includes all transitive dependencies:

run = client.workflow(workflow.id)
tasks = run.load_deps(:c, recursive: true)

task_a = tasks.fetch(:a)
task_a_output = tasks.output(:a)

Workers record output through job.output = value. The handle’s load_output(:a) loads one task’s output directly.

Use run.load_all to load every task, or run.load_all(limit: 100, offset: 200) to page by job ID. Use run.load_task(:a) for a single task. Loaded tasks expose job, output, pending_reason, and a wait snapshot, which is nil for tasks without waits. See wait diagnostics.

For whole-workflow retries, WorkflowT.Retry starts a new workflow attempt, clears live per-attempt wait metadata on retried jobs, keeps prior signals queryable by attempt, and lets the evaluator repair derived timer state.

After a workflow is finalized, client.workflow(workflow_id).retry starts a new attempt. Use mode: :failed_only to retry failed tasks, :failed_and_downstream to also retry their dependents, or the default :all. reset_history: true clears job attempts and errors while preserving maximum attempts.

Retries clear live wait evidence while retaining prior signals by attempt. Emit fresh signals with fresh idempotency keys for the new attempt.

By default, all tasks with failed dependencies are cancelled. This behavior can be customized at the level of an individual workflow using the IgnoreCancelledDeps, IgnoreDiscardedDeps, and IgnoreDeletedDeps options on either the WorkflowOpts or the WorkflowTaskOpts. These options allow you to control whether a task’s dependency should be considered successful despite it having entered into one of these failed states.

By default, tasks with failed dependencies are cancelled. Set ignore_cancelled_deps:, ignore_discarded_deps:, or ignore_deleted_deps: to true on River::Pro.workflow to set workflow-wide defaults, or on flow.add to override one task. Each option treats the corresponding failed dependency state as satisfied.

Workflow tasks may define a wait condition with the Wait field on WorkflowTaskOpts, using a riverworkflow.WaitSpec. A task with a wait condition remains pending until both its dependencies are satisfied and the wait condition resolves. This includes root tasks with no dependencies, which can wait on a signal or timer before they ever become available.

Pass a River::Pro::WorkflowWait as wait: to flow.add. A task stays pending until both its dependencies and its wait condition are satisfied. Root tasks can also wait for signals or timers before becoming available.

The preferred model is to declare named terms, then combine those terms with a CEL expression. A WaitSpec has three parts: Terms for named predicates and timers, an optional Expr for combining terms, and raw Inputs for CEL access to declared signals and timers. Declaring terms tells River which signal keys and timers can affect the task, while CEL provides the boolean logic.

CEL expressions are single expressions, not Go statements; use ||, &&, and ! for boolean logic, parentheses for grouping, dot access for object fields, and bracket access for maps or dynamic keys:

workflow.Add("decide_shipment", ShipmentDecisionArgs{OrderID: "ord_123"}, nil, &riverpro.WorkflowTaskOpts{
    Wait: &riverworkflow.WaitSpec{
        Expr: "manual_review_received || review_sla",
        Terms: []riverworkflow.WaitTermSpec{
            riverworkflow.WaitTermSignal(
                "manual_review_received",
                "manual_review",
                `payload.approved == true && payload.reviewer != ""`,
            ).Label("Manual review approved"),
            riverworkflow.WaitTermTimer(
                riverworkflow.TimerAfterWaitStarted("review_sla", 30*time.Minute),
            ).Label("Review SLA elapsed"),
        },
    },
})

WorkflowT.Prepare and WorkflowT.PrepareTx validate the wait definition along with the rest of the workflow graph. They check term names, declared inputs, timer anchors, CEL syntax, and dependency references before any jobs are inserted. See WaitSpec.Validate for the validation behavior and WorkflowTaskOpts for the task option that attaches a wait to a task.

Build named signal or timer terms, then combine them with River::Pro::WorkflowWait.cel. CEL expressions use ||, &&, and !, with dot or bracket access for object fields:

wait = River::Pro::WorkflowWait.cel(
  "manual_review_received || review_sla",
  terms: [
    River::Pro::WorkflowWait.for_signal(
      :manual_review,
      name: "manual_review_received",
      predicate: 'payload.approved == true && payload.reviewer != ""'
    ),
    River::Pro::WorkflowWait.after_wait_started(30 * 60, name: "review_sla")
  ]
)

workflow.add(:decide_shipment, ShipmentDecisionArgs.new(order_id: "ord_123"), wait: wait)

Here, approval or a 30-minute timeout makes the task eligible. Durations are seconds. workflow.jobs validates the graph and waits before insertion, including names, declared inputs, CEL syntax, timer anchors, and dependency references.

Signals are emitted through WorkflowT.Signals and WorkflowSignals.Emit:

_, err := workflow.Signals().Emit(ctx, "manual_review", ManualReviewSignal{
    Approved: true,
    Reviewer: "alice",
}, &riverpro.WorkflowSignalEmitOpts{
    IdempotencyKey: requestID,
    Source: map[string]any{
        "request_id": requestID,
        "actor":      "alice",
    },
})
if err != nil {
    return err
}

Use IdempotencyKey on WorkflowSignalEmitOpts when retries or webhook replays should create only one signal row for the same logical event. Reusing a key with a different payload returns a riverworkflow.SignalPayloadMismatchError.

Inside a task, prefer the task-scoped read APIs when loading signal data that caused the task to run. WorkflowSignals.LatestForTask and WorkflowSignals.ListForTask only expose keys declared by that task’s wait. Before wait evidence exists, task-scoped reads use current-attempt visibility; after persisted evidence exists, they stop at the evidence boundary captured when the wait resolved unless IncludeAfterResolution is set:

workflow, err := riverClient.WorkflowFromExisting(job.JobRow, nil)
if err != nil {
    return err
}

taskName := riverworkflow.TaskFromJobRow(job.JobRow)
signal, err := workflow.Signals().LatestForTask(ctx, taskName, "manual_review", nil)
if err != nil {
    return err
}

Set IncludeAfterResolution through WorkflowSignalLatestForTaskOpts or WorkflowSignalListForTaskParams only when the task intentionally wants to include signals emitted after the wait resolved.

Use WorkflowSignals.List for workflow-wide audit or export views that should not apply task visibility rules. Signal rows are stamped with a workflow attempt, so workflow retries can start a fresh attempt while prior signals remain queryable by explicit attempt filters.

The generated docs include complete examples for emitting a manual review signal, listing task-visible signal evidence, reading the latest task-visible signal, and workflow-wide signal pagination.

Emit signals through client.workflow(workflow_id).signals:

run = client.workflow(workflow.id)
result = run.signals.emit(
  "manual_review",
  {"approved" => true, "reviewer" => "alice"},
  idempotency_key: request_id,
  source: {"request_id" => request_id, "actor" => "alice"}
)

idempotency_key: prevents webhook replays or retries from creating duplicate events. Reusing a key with a different payload raises River::Pro::SignalPayloadMismatchError. source: records a JSON object for audit and correlation; replays retain the original source.

Inside a worker, prefer task-scoped reads of the signal evidence that made the task eligible:

run = job.client.workflow(job.metadata.fetch("workflow_id"))
signal = run.signals.latest_for_task(
  job.metadata.fetch("task"),
  key: "manual_review"
)

latest_for_task and list_for_task only expose keys declared by the task’s wait. Before resolution, they use current-attempt visibility; afterward, they stop at the evidence boundary recorded when the wait resolved. Set include_after_resolution: true only to intentionally include later signals.

For audit views, run.signals.list and list_page read workflow-wide signals. Use attempt: to inspect an earlier attempt, and cursor_id: with the page’s next_cursor_id to paginate while has_more? is true.

Timers let a wait resolve after an absolute time or after a relative duration anchored to a workflow event. The most common helper is TimerAfterWaitStarted, which starts counting only after the task’s dependencies are satisfied and the wait becomes active. Timer wake-up latency is controlled by WorkflowTimerPollerInterval on Config, which defaults to 1 second.

Other helpers include TimerAt, TimerAfterWorkflowCreated, and TimerAfterTaskFinalized. TimerAfterTaskFinalized may only reference a direct dependency of the task.

Wait: &riverworkflow.WaitSpec{
    Expr: "risk_hold_received || hold_timeout",
    Terms: []riverworkflow.WaitTermSpec{
        riverworkflow.WaitTermSignal("risk_hold_received", "risk_hold", `payload.active == true`),
        riverworkflow.WaitTermTimer(riverworkflow.TimerAfterWaitStarted("hold_timeout", 15*time.Minute)),
    },
},

For a full executable example, see WorkflowWaitTimerFallback.

Timers wait for an absolute time or a duration anchored to a workflow event. River::Pro::WorkflowWait.after_wait_started counts from when dependencies are satisfied and the wait becomes active:

wait = River::Pro::WorkflowWait.cel(
  "risk_hold_received || hold_timeout",
  terms: [
    River::Pro::WorkflowWait.for_signal(
      :risk_hold,
      name: "risk_hold_received",
      predicate: "payload.active == true"
    ),
    River::Pro::WorkflowWait.after_wait_started(15 * 60, name: "hold_timeout")
  ]
)

Other helpers include until_time(time), after_workflow_created(seconds), and after_task_finalized(task, seconds). A task used as a finalization anchor must also be a direct after: dependency. Timer state persists across restarts and is processed by the Pro runtime.

CEL expressions are evaluated by River Pro when declared inputs change. Structured signal terms created with WaitTermSignal evaluate their expression once per matching signal row, where fields like attempt, created_at, id, key, payload, and source are available. A signal-term expression cannot access deps, signals, timers, workflow, or other term names.

Timer terms created with WaitTermTimer become true when the timer has fired. Generic WaitTerm expressions use the raw wait environment and can inspect direct dependency outputs through deps["task_name"].output, declared signals["key"] and timers["name"] inputs, and the current workflow through workflow. Signal and timer keys are declared through WaitInputs or structured terms.

For example, a task can wait for a dependency’s recorded output before it runs:

workflow.Add("ship_order", ShipOrderArgs{OrderID: "ord_123"}, nil, &riverpro.WorkflowTaskOpts{
    Deps: []string{scoreFraudTask.Name},
    Wait: &riverworkflow.WaitSpec{
        Terms: []riverworkflow.WaitTermSpec{
            riverworkflow.WaitTerm("fraud_score_acceptable", `deps["score_fraud"].output.score < 90`),
        },
    },
})

When a wait has exactly one term, the top-level Expr can be omitted. For more complex conditions, use Expr to combine terms with normal CEL boolean operators. See WaitSpec, WaitTermSpec, and the generated examples for signal quorum waits, mixed structured terms and raw CEL, and raw CEL waits.

Signal predicates from WorkflowWait.for_signal evaluate once per signal row and can read attempt, created_at, id, key, payload, and source. They cannot read dependency outputs or other terms.

Generic WorkflowWait.term expressions can read direct dependency outputs, declared signal collections and timers, and workflow metadata. For example:

workflow.add(
  :ship_order,
  ShipOrderArgs.new(order_id: "ord_123"),
  after: :score_fraud,
  wait: River::Pro::WorkflowWait.term(
    "fraud_score_acceptable",
    'deps["score_fraud"].output.score < 90.0'
  )
)

A single term can serve as the wait directly. Combine several with WorkflowWait.cel(expression, terms: [...]). Declare raw signal and timer inputs through signals: and timers:; structured terms declare their own inputs. JSON numbers in payloads, outputs, and metadata are CEL doubles, while IDs and attempts are integers. Expressions must return booleans.

Loaded workflow tasks include a riverworkflow.Wait snapshot when the task has a wait. It exposes the wait phase, declared inputs, terms, timer fire times, and final bounded evidence after resolution:

task, err := workflow.LoadTask(ctx, "decide_shipment")
if err != nil {
    return err
}

if task.Wait != nil && task.Wait.Phase == riverworkflow.WaitPhaseResolved {
    fmt.Println(task.Wait.Summary)
}

Loaded tasks expose a River::Pro::WorkflowWaitSnapshot with the phase, declared inputs, terms, and persisted resolution evidence:

task = client.workflow(workflow.id).load_task(:decide_shipment)
puts task.wait.summary if task.wait&.phase == :resolved

Use WorkflowT.WaitDiagnostics for live, read-only diagnostics of a waiting task without writing result evidence:

diagnostics, err := workflow.WaitDiagnostics(ctx, "decide_shipment", nil)
if err != nil {
    return err
}
fmt.Println(diagnostics.Phase, diagnostics.ExprResult)

For workflows with large signal histories, SignalScanLimit on WorkflowWaitDiagnosticsOpts controls how many signal rows diagnostics scan. The default is 10,000, the maximum is 100,000, and Truncated on WaitDiagnostics is true when diagnostics hit the cap.

Use wait_diagnostics to evaluate current inputs without changing the task or writing evidence:

diagnostics = client.workflow(workflow.id).wait_diagnostics(:decide_shipment)
puts diagnostics.phase
puts diagnostics.expr_result

Diagnostics expose eval_error, inputs, terms, and truncated. The signal_scan_limit: keyword defaults to 10,000 rows and accepts up to 100,000. A truncated result only reflects the signals scanned.

Workflows can be controlled using the riverpro.Client type. The Client.WorkflowCancel and Client.WorkflowCancelTx methods allow you to cancel all non-finalized tasks in a workflow. These methods are useful for cleaning up workflows that are no longer needed or have failed.

Cancellation targets non-finalized tasks. Already-running jobs may still complete if they finish before observing cancellation.

result, err := riverClient.WorkflowCancel(ctx, workflowID)
if err != nil {
    // handle error
}
fmt.Printf("cancelled %d jobs", len(result.CancelledJobs))

Cancel unfinished tasks with client.workflow_cancel(workflow_id):

jobs = client.workflow_cancel(workflow_id)

The method returns the workflow’s jobs. Already-finalized jobs are preserved, and running jobs may complete before observing cancellation. To coordinate cancellation with application writes, call it inside the driver’s transaction block.

By default, workflow task jobs are cleaned by the normal job cleaner according to their individual job state and retention settings. River Pro can instead keep finalized workflow data together and clean it as a unit with WorkflowAwareRetention on Config.

Workflow-aware retention matters most when you want task jobs, signals, timers, attempts, and wait evidence kept together for audit or debugging. See workflow-aware retention for the cleaner behavior and configuration.

By default, task jobs follow ordinary job retention settings. Set workflow_aware_retention: true on River::Pro::Config to retain and clean finalized workflows as a unit, including their jobs, signals, timers, and attempt history.

workflow_cancelled_retention_period and workflow_closed_retention_period default to 86,400 seconds (24 hours). Set either to nil or -1 to disable cleanup for that workflow state. With workflow-aware retention enabled, workflow jobs are excluded from ordinary job cleanup and dead letter maintenance.