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.

The workflow Cargo feature is enabled by default. Create a builder from a ProClient, add named tasks, and insert the graph atomically:

use riverqueue::{InsertOpts, JobArgs};
use riverqueue_pro::workflow::{TaskOpts, WorkflowOpts};
use serde::{Deserialize, Serialize};

#[derive(Deserialize, JobArgs, Serialize)]
#[river(kind = "my_job")]
struct MyJobArgs {}

let mut workflow = client.new_workflow(WorkflowOpts::default().with_name("My first workflow"));
workflow.add("a", MyJobArgs {})?;

// Fan out: both tasks depend on a.
workflow.add_with(
    "b1",
    MyJobArgs {},
    InsertOpts::default(),
    TaskOpts::default().with_dependencies(["a"]),
)?;
workflow.add_with(
    "b2",
    MyJobArgs {},
    InsertOpts::default(),
    TaskOpts::default().with_dependencies(["a"]),
)?;

// Fan in: c waits for both b1 and b2.
workflow.add_with(
    "c",
    MyJobArgs {},
    InsertOpts::default(),
    TaskOpts::default().with_dependencies(["b1", "b2"]),
)?;
let inserted = workflow.insert().await?;
let run = client.workflow(&inserted.id);

Register workers for the task kinds and start the client as in getting started. Started Pro clients run workflow evaluation automatically; the leader also processes timers and recovery. client.workflow(id) returns a handle for operations on a stored workflow.

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.

client.new_workflow(WorkflowOpts::default().with_name("fulfil order")) generates a sortable workflow ID. If you set with_id(...), supply a globally unique ID for each run.

add(name, args) adds a task without explicit options. add_with(name, args, insert_opts, task_opts) accepts core InsertOpts for settings like queue and priority, plus TaskOpts for dependencies and waits. Dependencies refer to task names and may refer forward to tasks added later.

workflow.insert().await? consumes the builder, validates dependencies and cycles, and inserts every task in one transaction. To include the workflow in an application transaction:

let mut tx = pool.begin().await?;
let inserted = workflow.insert().tx(&mut tx).await?;
tx.commit().await?;

Roll back the transaction on failure. Workflow tasks cannot use sequences, per-job ephemeral options, or ephemeral queues. Keep jobs distinct between workflow runs: avoid unique-job options that could deduplicate a task against another workflow. Build a new workflow for each run; insertion is not an idempotency mechanism.

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.

Inside a workflow worker, obtain its Pro client from the work context and build an extension from the job. Insert new tasks and complete the current task in one transaction:

use riverqueue::InsertOpts;
use riverqueue_pro::ProClient;
use riverqueue_pro::workflow::{TaskOpts, WorkflowOpts, task_from_job};

let client = ProClient::from_context(&context)?;
let current_task = task_from_job(&job.row).expect("job belongs to a workflow");
let mut extension = client.workflow_from_existing(&job.row, WorkflowOpts::default())?;
extension.add_with(
    "notify",
    NotifyArgs {},
    InsertOpts::default(),
    TaskOpts::default().with_dependencies([current_task]),
)?;

let mut tx = pool.begin().await?;
extension.insert().tx(&mut tx).await?;
context.job_complete_tx(&mut tx).await?;
tx.commit().await?;

Return Ok(WorkOutcome::Complete) from the worker after committing, and propagate errors so the attempt is retried. Outside a worker, use client.workflow(id).extend(WorkflowOpts::default()).await?. extend reads the existing workflow outside your transaction; the returned builder’s .insert().tx(&mut tx) inserts the new tasks inside it. Task names must be unique within the workflow.

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.

Use a workflow handle to load dependencies, individual tasks, or output:

use riverqueue_pro::workflow::{LoadAllOpts, LoadDepsOpts};

let run = client.workflow(&workflow_id);
let tasks = run
    .load_deps("c", LoadDepsOpts::default().with_recursive(true))
    .await?;
let task_a = tasks.get("a").expect("a is a dependency");
let task_a_output: serde_json::Value = tasks.output("a")?;

let output: serde_json::Value = run.load_output("a").await?;
let page = run
    .load_all(LoadAllOpts::default().with_limit(100).with_offset(200))
    .await?;
let task = run.load_task("a").await?;

Workers record output with context.record_output(value)?. Output is persisted with completion and can be decoded into a concrete type instead of serde_json::Value. LoadDepsOpts::default() loads only direct dependencies; with_recursive(true) includes transitive ones.

Within a worker, ProClient::from_context(&context)?, workflow_id_from_job(&job.row), and task_from_job(&job.row) identify its client, workflow, and task. Loaded WorkflowTask values expose job, pending_reason, and an optional wait snapshot. Read output with task.output::<T>()?. The handle’s read requests also accept .tx(&mut tx).

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.

After a workflow has no active tasks, retry it through its handle:

use riverqueue_pro::workflow::{RetryMode, RetryOpts};

let retried = client
    .workflow(&workflow_id)
    .retry(RetryOpts::default().with_mode(RetryMode::FailedOnly))
    .await?;

RetryMode::All is the default and retries every task. FailedOnly retries cancelled and discarded tasks and downstream tasks that failed to complete because of them. FailedAndDownstream also retries completed downstream tasks. Add with_reset_history(true) to clear attempts and errors while preserving maximum attempts.

A retry starts a new workflow attempt and clears live wait evidence. Prior signals remain available for audit by attempt; emit fresh signals with fresh idempotency keys for the new attempt. .retry(...).tx(&mut tx) participates in an application transaction.

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.

Set with_ignore_cancelled_dependencies(true), with_ignore_discarded_dependencies(true), or with_ignore_deleted_dependencies(true) on WorkflowOpts to establish workflow-wide defaults. Set the same methods on TaskOpts to override an individual task, including setting an inherited option back to false.

Each enabled option treats that failed dependency state as satisfied. Without an override, a failed dependency cancels its dependent tasks.

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.

Pass TaskOpts::default().with_wait(wait) to workflow.add_with, where wait is a WaitSpec. The task stays pending until both its dependencies and wait are satisfied. A root task can wait without dependencies.

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.

Build named WaitTerm values and combine them with a WaitSpec expression:

use riverqueue::InsertOpts;
use riverqueue_pro::workflow::{TaskOpts, Timer, WaitInputs, WaitSpec, WaitTerm};
use std::time::Duration;

let wait = WaitSpec::new(
    "manual_review_received || review_sla",
    WaitInputs::default(),
)
.with_term(WaitTerm::signal(
    "manual_review_received",
    "manual_review",
    "payload.approved == true && payload.reviewer != \"\"",
)?)
.with_term(WaitTerm::timer(Timer::after_wait_started(
    "review_sla",
    Duration::from_secs(30 * 60),
)?));
workflow.add_with(
    "decide_shipment",
    ShipmentDecisionArgs {
        order_id: "ord_123".into(),
    },
    InsertOpts::default(),
    TaskOpts::default().with_wait(wait),
)?;

Approval or a 30-minute timeout makes the task eligible. Structured signal and timer terms declare their own inputs. Adding the task validates its wait; inserting the graph also validates dependencies and cycles.

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.

Emit serializable payloads through a workflow handle:

use riverqueue_pro::workflow::SignalOpts;
use serde_json::json;

let run = client.workflow(&workflow_id);
let result = run
    .signals()
    .emit(
        "manual_review",
        json!({"approved": true, "reviewer": "alice"}),
        SignalOpts::default()
            .with_idempotency_key(&request_id)
            .with_source(serde_json::Map::from_iter([
                ("request_id".into(), json!(request_id)),
                ("actor".into(), json!("alice")),
            ])),
    )
    .await?;

Replaying an idempotency key for the same signal key with the same payload returns the original signal; a different payload returns workflow::Error::SignalPayloadMismatch. Replays retain the original source. with_attempt(attempt) rejects a late signal intended for an earlier attempt. Emission also accepts .tx(&mut tx).

Inside a worker, read the signal evidence visible to that task:

use riverqueue_pro::ProClient;
use riverqueue_pro::workflow::{SignalLatestOpts, task_from_job, workflow_id_from_job};

let client = ProClient::from_context(&context)?;
let workflow_id = workflow_id_from_job(&job.row).expect("workflow task");
let task = task_from_job(&job.row).expect("workflow task");
let signal = client
    .workflow(workflow_id)
    .signals()
    .latest_for_task(&task, "manual_review", SignalLatestOpts::default())
    .await?;

latest_for_task and list_for_task expose only keys declared by the task’s wait. Before resolution they use current-attempt visibility; afterward they stop at the recorded evidence boundary. Opt into later signals with with_include_after_resolution(true) only when intended.

For audit views, run.signals().list(SignalListOpts::default()).await? returns a SignalPage. Pass next_cursor_id to SignalListOpts::with_cursor while has_more is true. Use with_attempt(...) to inspect earlier attempts. Task-scoped lists take SignalListForTaskOpts.

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.

Timers take std::time::Duration or an absolute jiff::Timestamp. Timer::after_wait_started starts counting when dependencies are satisfied and the wait becomes active:

use riverqueue_pro::workflow::{Timer, WaitInputs, WaitSpec, WaitTerm};
use std::time::Duration;

let wait = WaitSpec::new("risk_hold_received || hold_timeout", WaitInputs::default())
    .with_term(WaitTerm::signal(
        "risk_hold_received",
        "risk_hold",
        "payload.active == true",
    )?)
    .with_term(WaitTerm::timer(Timer::after_wait_started(
        "hold_timeout",
        Duration::from_secs(15 * 60),
    )?));

Other constructors are Timer::at(name, timestamp), Timer::after_workflow_created(name, duration), and Timer::after_task_finalized(name, task, duration). A finalization anchor must also be a direct dependency of the waiting task. Timer state survives 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.

WaitTerm::signal predicates evaluate each signal row independently and can read attempt, created_at, id, key, payload, and source. They cannot read dependency outputs or other terms.

Generic terms can read direct dependency outputs, declared signals and timers, and workflow metadata:

use riverqueue::InsertOpts;
use riverqueue_pro::workflow::{TaskOpts, WaitSpec, WaitTerm};

workflow.add_with(
    "ship_order",
    ShipOrderArgs {
        order_id: "ord_123".into(),
    },
    InsertOpts::default(),
    TaskOpts::default()
        .with_dependencies(["score_fraud"])
        .with_wait(WaitSpec::from_term(WaitTerm::generic(
            "fraud_score_acceptable",
            "deps[\"score_fraud\"].output.score < 90.0",
        )?)),
)?;

WaitSpec::from_term uses one term as the entire wait. Combine multiple terms with WaitSpec::new(expression, inputs).with_term(...). For raw CEL inputs, use WaitInputs::with_signals and with_timers; structured terms declare their own inputs. Dependencies must be direct task dependencies. JSON numbers are CEL doubles, while IDs and attempts are integers, and wait expressions must return booleans.

For a quorum, set a count on the signal term to require two matching rows:

use riverqueue_pro::workflow::{WaitSpec, WaitTerm};
use std::num::NonZeroUsize;

let wait = WaitSpec::from_term(
    WaitTerm::signal(
        "reviews_received",
        "manual_review",
        "payload.approved == true",
    )?
    .with_count(NonZeroUsize::new(2).expect("nonzero quorum"))?,
);

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

A loaded task’s optional Wait contains its phase, inputs, terms, and persisted resolution evidence:

use riverqueue_pro::workflow::WaitPhase;

let task = client
    .workflow(&workflow_id)
    .load_task("decide_shipment")
    .await?;
if let Some(wait) = task.wait {
    if wait.phase == WaitPhase::Resolved {
        println!("{}", wait.summary);
    }
}

Use wait.term(name), wait.signal_input(key), wait.dep_input(task), and wait.timer_input(name) to inspect the evidence that made the task eligible.

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.

Evaluate live inputs without changing the task or recording new evidence:

use riverqueue_pro::workflow::WaitDiagnosticsOpts;

let diagnostics = client
    .workflow(&workflow_id)
    .wait_diagnostics("decide_shipment", WaitDiagnosticsOpts::default())
    .await?;
println!(
    "{:?}: {:?}",
    diagnostics.phase, diagnostics.expression_result
);

The result includes eval_error, inputs, terms, and truncated. WaitDiagnosticsOpts::with_signal_scan_limit defaults to 10,000 rows and accepts up to 100,000. A truncated result reflects only the scanned prefix. Use the loaded task’s wait for persisted evidence, and diagnostics for a current evaluation.

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.

Cancel unfinished tasks through the workflow handle:

let cancelled = client.workflow(&workflow_id).cancel().await?;

The CancelResult exposes cancelled_jobs. Already-finalized jobs are preserved; running workers receive a cancellation request and may finish before observing it. Use .cancel().tx(&mut tx) to coordinate cancellation with application writes.

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.

By default, workflow jobs follow ordinary retention. Enable cleanup of whole finalized workflows with .workflow_aware_retention(...):

use riverqueue::{QueueConfig, Retention, database::PostgresDatabase};
use riverqueue_pro::{ProClient, WorkflowRetention};
use std::time::Duration;

let retention = WorkflowRetention::default()
    .with_cancelled_retention(Retention::DeleteAfter(Duration::from_secs(24 * 60 * 60)))
    .with_closed_retention(Retention::DeleteAfter(Duration::from_secs(24 * 60 * 60)));
let client = ProClient::builder(PostgresDatabase::new(pool))
    .workflow_aware_retention(retention)
    .queue("default", QueueConfig::new(100))
    .workers(workers)
    .build()?;

Both periods default to 24 hours; Retention::Keep disables cleanup for that workflow state. With workflow-aware retention enabled, workflow tasks are excluded from ordinary job cleanup and dead letter maintenance. Cleanup removes finalized workflows together with their jobs, signals, timers, and attempt history.