riverworkflow

README

river-workflow

Durable, Temporal-style workflows as code on top of the River job queue for Go and Postgres.

Write an ordinary Go function that dispatches activities, waits on and branches on their results at runtime, and runs them in parallel. River makes it durable: the workflow survives process crashes and arbitrarily long waits, and every activity runs exactly once.

river-workflow is built entirely on River's public API — it requires no changes to River or the River UI, no extra database tables, and no separate service to operate.

func OrderWorkflow(ctx riverworkflow.Context, in OrderInput) (OrderResult, error) {
    // Two independent checks in parallel: is the item in stock, and will the card
    // authorize? Dispatch both before awaiting either — they run concurrently.
    // Both are read-only, so nothing is committed yet.
    stock := riverworkflow.Execute[CheckStockArgs, StockResult](ctx, CheckStockArgs{Item: in.Item}, nil)
    auth := riverworkflow.Execute[AuthorizeArgs, AuthResult](ctx, AuthorizeArgs{Card: in.Card, Amount: in.Amount}, nil)

    stocked, err := stock.Get()
    if err != nil {
        return OrderResult{}, err // retries exhausted; nothing committed, so fail fast
    }
    authed, err := auth.Get()
    if err != nil {
        return OrderResult{}, err
    }

    // Branch on the outcomes at runtime. "Out of stock" and "declined" are
    // business results, not errors — complete with a status, don't fail.
    if !stocked.Available {
        return OrderResult{Status: "out_of_stock"}, nil
    }
    if !authed.Approved {
        return OrderResult{Status: "declined"}, nil
    }

    // Everything cleared. Capturing payment and shipping is the one committing
    // step, so it goes last — any failure before here strands nothing.
    fulfilled, err := riverworkflow.Execute[FulfilArgs, FulfilResult](ctx,
        FulfilArgs{Item: in.Item, Auth: authed.ID}, nil).Get()
    if err != nil {
        return OrderResult{}, err
    }
    return OrderResult{Status: "shipped", Tracking: fulfilled.Tracking}, nil
}

Features

• Workflows as code — imperative Go with real runtime control flow (if, loops, early return) driven by activity results.
• Durable execution — a workflow is a River job that deterministically replays against committed history; crashes and long waits are transparent, and completed activities never re-run.
• Parallelism — dispatch multiple activities before awaiting them and they run concurrently across your workers.
• Durable timers — Sleep(ctx, d) pauses a workflow for a wall-clock duration without holding a goroutine, surviving restarts.
• Side effects — SideEffect/Now/Random capture a nondeterministic value once and reproduce it on replay, no activity round-trip.
• Child workflows — ChildWorkflow starts and awaits another workflow; cancelling a parent stops the whole tree.
• ContinueAsNew — restart a long-running or looping workflow with fresh history to keep replay cost bounded.
• Signals — deliver a durable external event into a running workflow with Signal/AwaitSignal.
• Select — wait for whichever of several things happens first (a signal, a timeout, or a future) — e.g. "approve within an hour, else escalate".
• Activities are just River jobs — reuse your existing workers, queues, retry policies, and telemetry. Define one with RegisterActivity (a typed function whose return value is recorded for you) or a raw River worker that calls river.RecordOutput.
• Optional durable store — the opt-in pgstore package adds same-database tables for results that outlive River's job retention, push-based Await (no polling), an indexed lookup path, List, reap-proof replay history (a workflow suspended past River's retention still replays), and a targeted cross-process reactivation backstop — without modifying River. See the design RFC.
• Programmatic control — Cancel and GetStatus alongside Result.
• Reactive — a workflow advances the instant an activity finishes (via River subscriptions), with no polling on the hot path.
• Stoppable from the River UI — cancel the workflow's run job and the stop cascades to its in-flight activities.
• No new infrastructure — no extra tables, no migrations, no separate orchestration service.

Install

go get github.com/nlm/river-workflow

Requires Go 1.25+ and a River client (Postgres via riverpgxv5, or the database/sql driver).

Quick start

reg := riverworkflow.NewRegistry()
riverworkflow.RegisterWorkflow(reg, "order", OrderWorkflow)

workers := river.NewWorkers()
riverworkflow.RegisterActivity(workers, CheckStock) // recommended: return value is recorded for you
riverworkflow.RegisterActivity(workers, Authorize)
river.AddWorker(workers, &FulfilWorker{})           // or a raw River worker (see below)

config := &river.Config{
    Queues:  map[string]river.QueueConfig{river.QueueDefault: {MaxWorkers: 100}},
    Workers: workers,
}

engine := riverworkflow.NewEngine[pgx.Tx](reg, nil)
engine.Install(config, workers) // registers the run worker + the workflow queue

client, err := river.NewClient(riverpgxv5.New(pool), config)
if err != nil {
    log.Fatal(err)
}
if err := client.Start(ctx); err != nil {
    log.Fatal(err)
}

engine.Start(ctx, client) // reactive advancement + backstop
defer engine.Stop()

// Launch a workflow.
workflowID, err := riverworkflow.Start(ctx, client, "order", OrderInput{Item: "SKU-42", Card: "…", Amount: 4200}, nil)
if err != nil {
    log.Fatal(err)
}

// Later, read its typed result.
result, done, err := riverworkflow.Result[pgx.Tx, OrderResult](ctx, client, workflowID)

The simplest way to define an activity is a typed function — RegisterActivity records whatever it returns, so there is no RecordOutput to forget:

func Authorize(ctx context.Context, args AuthorizeArgs) (AuthResult, error) {
    return authorize(args.Card, args.Amount) // return value is recorded, decoded by Future.Get
}

riverworkflow.RegisterActivity(workers, Authorize)

Since an activity is just a River job, you can also register a raw worker and call river.RecordOutput yourself — do this when the activity needs custom worker behavior (Timeout, NextRetry, or Middleware):

// Fulfilment calls a slow external carrier, so it overrides Timeout — a concrete
// reason to use a raw worker instead of RegisterActivity.
type FulfilWorker struct{ river.WorkerDefaults[FulfilArgs] }

func (FulfilWorker) Timeout(*river.Job[FulfilArgs]) time.Duration { return 30 * time.Second }

func (FulfilWorker) Work(ctx context.Context, job *river.Job[FulfilArgs]) error {
    result := fulfil(job.Args.Item, job.Args.Auth)
    return river.RecordOutput(ctx, result) // decoded by Future.Get
}

Either way, an activity that completes without recording an output fails its workflow with a *MissingOutputError from Future.Get, rather than silently handing back a zero value.

Examples

The examples/ directory has seven small, self-contained programs you can run against a Postgres — an order-fulfilment saga with compensations, a parallel batch with a concurrency cap, child workflows, human-in-the-loop signals, a ContinueAsNew poller, a Select signal-or-timeout, and the durable store:

go run ./examples/order-pipeline

See examples/README.md for the full list.

Documentation

Guide	Contents
Getting started	Install, wire up an engine, run your first workflow end to end.
Concepts	Workflows, activities, futures, outputs, and the workflow ID.
Architecture	How durable replay works, the run job, reactive advancement, and the backstop.
Determinism	The rules for writing correct workflow functions, and how divergence is detected.
Failure & cancellation	Activity retries, failure surfacing, sagas, and stopping workflows from the UI.
Operations	Queues, tuning, scaling, cleanup, and deploying workflow code changes.
Caveats & limitations	The sharp edges: retention bounds, timing resolution, replay cost, and other things to know before production.
Testing	Unit-testing the replay engine and integration-testing against Postgres.
Roadmap	What is intentionally not in this version yet.

How it works, in one paragraph

A running workflow is a single River job (river_workflow_run) whose Work function re-executes your workflow function from the top every time it wakes. Execute dispatches an activity as an ordinary child River job; the child job rows are the workflow's event history. On each run the engine loads that committed history and hands recorded results back through Future.Get, so already-finished activities are skipped and the function makes forward progress. When it reaches an activity that hasn't finished, the run job snoozes; a River subscription wakes it the moment the activity commits. This is the same deterministic-replay technique Temporal uses, implemented on River primitives. See Architecture for the full picture.

Status

Version 0 — the engine (activities, typed futures, parallelism, runtime branching, durable replay, durable timers, side effects, child workflows, ContinueAsNew, signals, Select, cancellation, and stop-from-UI) is complete and tested. An optional supplemental store (pgstore) adds durable results, push-based Await, listing, reap-proof replay history, and a targeted cross-process reactivation backstop (the store RFC, fully implemented). Queries, workflow versioning, and other advanced features are on the roadmap.

License

MIT — see LICENSE.

Documentation

Overview

Package riverworkflow adds Temporal-style durable "workflows as code" on top of the River job queue (github.com/riverqueue/river), using only River's public API — it requires no changes to River or the River UI.

Model

A workflow is an ordinary Go function that dispatches activities, awaits and branches on their results at runtime, and runs them in parallel. It is made durable by River: the workflow runs as a River job that deterministically replays its function against committed history, so it survives process crashes and arbitrarily long waits.

func OrderWorkflow(ctx riverworkflow.Context, in OrderInput) (OrderResult, error) {
	charge, err := riverworkflow.Execute[ChargeArgs, ChargeResult](ctx, ChargeArgs{Card: in.Card}, nil).Get()
	if err != nil {
		return OrderResult{}, err
	}
	if charge.Declined {
		return OrderResult{Cancelled: true}, nil // runtime branch on a result
	}

	ship := riverworkflow.Execute[ShipArgs, ShipResult](ctx, ShipArgs{Order: in.ID}, nil)   // parallel...
	invoice := riverworkflow.Execute[InvoiceArgs, InvoiceResult](ctx, InvoiceArgs{Order: in.ID}, nil)
	if _, err := ship.Get(); err != nil {
		return OrderResult{}, err
	}
	if _, err := invoice.Get(); err != nil {
		return OrderResult{}, err
	}
	return OrderResult{OK: true}, nil
}

Activities

An activity is an ordinary River job that does the real work; its result is what Future.Get decodes. The recommended way to define one is RegisterActivity, which takes a plain typed function and records the value it returns automatically:

riverworkflow.RegisterActivity(workers, func(ctx context.Context, args ChargeArgs) (ChargeResult, error) {
	return charge(ctx, args) // returned value becomes the activity's output
})

Because an activity is just a River job, you can also register a raw github.com/riverqueue/river.Worker with github.com/riverqueue/river.AddWorker and call github.com/riverqueue/river.RecordOutput yourself — do this when the activity needs custom worker behavior (Timeout, NextRetry, or Middleware):

func (ChargeWorker) Work(ctx context.Context, job *river.Job[ChargeArgs]) error {
	return river.RecordOutput(ctx, ChargeResult{...})
}

Either way, an activity that completes without recording an output fails its workflow with a *MissingOutputError from Future.Get (rather than silently yielding a zero value) — RegisterActivity makes that impossible to forget.

Activities retry per their own River retry policy — a Future only resolves once its activity is terminal (completed yields the value; a terminal failure yields an *ActivityError that workflow code may handle).

Determinism

Workflow functions replay from the top on every resume, so they must be deterministic: perform all I/O inside activities, never inline; do not spawn goroutines; and do not recover panics (Future.Get suspends the workflow by panicking with an internal sentinel). Divergence from recorded history is detected and fails the workflow. For cheap nondeterministic values use SideEffect, Now, or Random, which capture the value once and reproduce it on replay; to wait for a duration use Sleep, a durable timer.

Controlling a workflow

Result returns a finished workflow's output; GetStatus returns its high-level state; Cancel stops it (equivalent to cancelling its run job in the River UI), cascading to in-flight activities.

Wiring

reg := riverworkflow.NewRegistry()
riverworkflow.RegisterWorkflow(reg, "order", OrderWorkflow)

workers := river.NewWorkers()
riverworkflow.RegisterActivity(workers, ChargeActivity) // recommended
river.AddWorker(workers, &ShipWorker{})                 // or a raw worker
// ...

config := &river.Config{Queues: map[string]river.QueueConfig{river.QueueDefault: {MaxWorkers: 100}}, Workers: workers}
engine := riverworkflow.NewEngine[pgx.Tx](reg, nil)
engine.Install(config, workers) // adds the run worker + workflow queue

client, _ := river.NewClient(riverpgxv5.New(pool), config)
client.Start(ctx)
engine.Start(ctx, client) // reactive advancement + backstop
defer engine.Stop()

workflowID, _ := riverworkflow.Start(ctx, client, "order", OrderInput{ID: 1, Card: "..."}, nil)

Stopping from the River UI

Each workflow is represented by its run job (kind "river_workflow_run"), which is its row in the River UI. Cancelling that job stops the workflow: the engine cascades the cancel to the workflow's still-live activities.

Roadmap

Not yet implemented: queries, activity heartbeats/timeouts, workflow versioning, and list/search APIs.

Example

Example demonstrates defining a durable workflow as code, wiring it into a River client, and reading its result.

package main

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	riverworkflow "github.com/nlm/river-workflow"
	"github.com/nlm/river-workflow/internal/wfdbtest"

	"github.com/riverqueue/river"
	"github.com/riverqueue/river/riverdriver/riverpgxv5"
	"github.com/riverqueue/river/rivershared/riversharedtest"
	"github.com/riverqueue/river/rivershared/util/slogutil"
	"github.com/riverqueue/river/rivershared/util/testutil"
)

// --- workflow input/output and activities ---

type OrderInput struct {
	Amount int `json:"amount"`
}

type OrderResult struct {
	Shipped bool `json:"shipped"`
}

type ChargeArgs struct {
	Amount int `json:"amount"`
}

func (ChargeArgs) Kind() string { return "example_charge" }

type ChargeResult struct {
	Approved bool `json:"approved"`
}

type ChargeWorker struct {
	river.WorkerDefaults[ChargeArgs]
}

func (ChargeWorker) Work(ctx context.Context, job *river.Job[ChargeArgs]) error {
	return river.RecordOutput(ctx, ChargeResult{Approved: job.Args.Amount > 0})
}

type ShipArgs struct{}

func (ShipArgs) Kind() string { return "example_ship" }

type ShipWorker struct{ river.WorkerDefaults[ShipArgs] }

func (ShipWorker) Work(ctx context.Context, job *river.Job[ShipArgs]) error {
	return river.RecordOutput(ctx, struct{}{})
}

// OrderWorkflow charges the card and, only if the charge is approved, ships —
// branching on the activity result at runtime.
func OrderWorkflow(ctx riverworkflow.Context, in OrderInput) (OrderResult, error) {
	charge, err := riverworkflow.Execute[ChargeArgs, ChargeResult](ctx, ChargeArgs(in), nil).Get()
	if err != nil {
		return OrderResult{}, err
	}
	if !charge.Approved {
		return OrderResult{Shipped: false}, nil
	}
	if _, err := riverworkflow.Execute[ShipArgs, struct{}](ctx, ShipArgs{}, nil).Get(); err != nil {
		return OrderResult{}, err
	}
	return OrderResult{Shipped: true}, nil
}

// Example demonstrates defining a durable workflow as code, wiring it into a
// River client, and reading its result.
func main() {
	ctx := context.Background()

	dbPool, err := pgxpool.New(ctx, riversharedtest.TestDatabaseURL())
	if err != nil {
		panic(err)
	}
	defer dbPool.Close()

	driver := riverpgxv5.New(dbPool)

	reg := riverworkflow.NewRegistry()
	riverworkflow.RegisterWorkflow(reg, "order", OrderWorkflow)

	workers := river.NewWorkers()
	river.AddWorker(workers, &ChargeWorker{}) // activities are ordinary workers
	river.AddWorker(workers, &ShipWorker{})

	config := &river.Config{
		Logger:   slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelWarn, ReplaceAttr: slogutil.NoLevelTime})),
		Queues:   map[string]river.QueueConfig{river.QueueDefault: {MaxWorkers: 10}},
		Schema:   wfdbtest.Schema(ctx, testutil.PanicTB(), driver), // only for the example test
		TestOnly: true,                                             // remove outside of tests
		Workers:  workers,
	}

	engine := riverworkflow.NewEngine[pgx.Tx](reg, nil)
	engine.Install(config, workers)

	client, err := river.NewClient(driver, config)
	if err != nil {
		panic(err)
	}
	if err := client.Start(ctx); err != nil {
		panic(err)
	}
	defer func() { _ = client.Stop(ctx) }()

	engine.Start(ctx, client)
	defer engine.Stop()

	workflowID, err := riverworkflow.Start(ctx, client, "order", OrderInput{Amount: 100}, nil)
	if err != nil {
		panic(err)
	}

	// Wait for the workflow to finish, then read its typed result.
	var result OrderResult
	for {
		out, done, err := riverworkflow.Result[pgx.Tx, OrderResult](ctx, client, workflowID)
		if err != nil {
			panic(err)
		}
		if done {
			result = out
			break
		}
		time.Sleep(20 * time.Millisecond)
	}

	fmt.Printf("shipped: %v\n", result.Shipped)
}

Output:
shipped: true

Index

• Variables
• func AwaitSignal[T any](ctx Context, name string) T
• func Cancel[TTx any](ctx context.Context, client *river.Client[TTx], workflowID string) error
• func ContinueAsNew[I any](ctx Context, in I) error
• func Now(ctx Context) time.Time
• func Random(ctx Context) int64
• func RegisterActivity[TArgs river.JobArgs, TOut any](workers *river.Workers, fn func(context.Context, TArgs) (TOut, error))
• func RegisterWorkflow[I, O any](reg *Registry, kind string, fn func(ctx Context, in I) (O, error))
• func Result[TTx any, O any](ctx context.Context, client *river.Client[TTx], workflowID string) (O, bool, error)
• func Select(ctx Context, cases ...SelectCase)
• func SideEffect[T any](ctx Context, fn func() T) T
• func Signal[TTx any, T any](ctx context.Context, client *river.Client[TTx], workflowID, name string, ...) error
• func Sleep(ctx Context, d time.Duration)
• func Start[TTx any, I any](ctx context.Context, client *river.Client[TTx], kind string, in I, ...) (string, error)
• type ActivityError
  • func (e *ActivityError) Error() string
• type ActivityOpts
• type CommandRecord
• type Context
  • func (c Context) Context() context.Context
• type Engine
  • func NewEngine[TTx any](reg *Registry, opts *EngineOpts) *Engine[TTx]
  • func (e *Engine[TTx]) Install(config *river.Config, workers *river.Workers)
  • func (e *Engine[TTx]) Start(ctx context.Context, client *river.Client[TTx])
  • func (e *Engine[TTx]) Stop()
• type EngineOpts
• type Future
  • func ChildWorkflow[I any, O any](ctx Context, kind string, in I) *Future[O]
  • func Execute[TArgs river.JobArgs, TOut any](ctx Context, args TArgs, opts *ActivityOpts) *Future[TOut]
  • func (f *Future[TOut]) Get() (TOut, error)
• type HistoryStore
• type MissingOutputError
  • func (e *MissingOutputError) Error() string
• type Registry
  • func NewRegistry() *Registry
• type RunWorker
  • func (w *RunWorker[TTx]) Work(ctx context.Context, job *river.Job[runArgs]) error
• type SelectCase
  • func OnFuture[T any](f *Future[T], handler func(T, error)) SelectCase
  • func OnSignal[T any](name string, handler func(T)) SelectCase
  • func OnTimeout(d time.Duration, handler func()) SelectCase
• type SignalRecord
• type StartOpts
• type Status
  • func GetStatus[TTx any](ctx context.Context, client *river.Client[TTx], workflowID string) (Status, error)
• type Store
• type Wake
• type WakeStore
• type WorkflowState
  • func (s WorkflowState) Terminal() bool

Examples

• Package

Variables

var ErrWorkflowFailed = errors.New("riverworkflow: workflow did not complete successfully")

ErrWorkflowFailed is returned by Result when a workflow finished in a terminal state other than completed — that is, it was stopped from the UI or failed. The returned error wraps this sentinel, so errors.Is can be used to detect it.

Functions

func AwaitSignal

func AwaitSignal[T any](ctx Context, name string) T

AwaitSignal waits for the next unconsumed signal of the given name and returns its payload, decoded into T. If no such signal has arrived yet, the workflow suspends and resumes once one does. Repeated calls consume successive signals of that name in send order.

AwaitSignal blocks indefinitely. To bound the wait (e.g. "approve within an hour, else escalate") you currently need a separate mechanism; a select-style await of a signal or a timer is on the roadmap.

func Cancel

func Cancel[TTx any](ctx context.Context, client *river.Client[TTx], workflowID string) error

Cancel stops a running workflow: it cancels the workflow's run job and any of its activities still in flight. It is a no-op if the workflow has already finished, and returns rivertype.ErrNotFound if no workflow with the given ID exists.

This is the programmatic equivalent of cancelling the workflow's run job from the River UI.

func ContinueAsNew

func ContinueAsNew[I any](ctx Context, in I) error

ContinueAsNew ends the current workflow run and immediately starts a fresh generation of the same workflow (same ID and kind) with the given input, discarding the accumulated history. Return its result from the workflow function:

func Loop(ctx riverworkflow.Context, in State) (State, error) {
	next, done, err := step(ctx, in)
	if err != nil || done {
		return next, err
	}
	return next, riverworkflow.ContinueAsNew(ctx, next)
}

Use it to keep a long-running or looping workflow's replay cost and history bounded: each generation replays only its own commands. Any activities still in flight when ContinueAsNew is reached are abandoned, so call it at a point where nothing you care about is pending. The output value returned alongside it is ignored.

ContinueAsNew is only supported in a top-level workflow. A continued generation is a new run job that is not linked back to a parent, so a child workflow that calls ContinueAsNew fails (surfacing to its parent as an *ActivityError) rather than silently resolving the parent with an empty result. A child that needs to loop unboundedly should be a top-level workflow started with Start instead.

func Now

func Now(ctx Context) time.Time

Now returns a wall-clock time captured once for the workflow. Unlike time.Now, the value is fixed on first execution and reused on every replay, so it is safe to use inside a workflow function.

func Random

func Random(ctx Context) int64

Random returns a non-negative random int64 captured once for the workflow. Like Now, the value is fixed on first execution and stable across replays.

func RegisterActivity

func RegisterActivity[TArgs river.JobArgs, TOut any](
	workers *river.Workers,
	fn func(context.Context, TArgs) (TOut, error),
)

RegisterActivity registers a typed activity worker. It is the recommended way to define an activity: the value fn returns is recorded automatically as the activity's output (via river.RecordOutput), so a workflow's Execute[TArgs, TOut] can decode it — there is no RecordOutput call to remember or forget.

A non-nil error from fn fails the activity job, which then retries per River's retry policy and, once terminal, surfaces to the workflow as an *ActivityError from Future.Get. A void activity returns (struct{}{}, nil).

RegisterActivity is a thin wrapper over river.AddWorker; like AddWorker it panics if an activity of the same kind is already registered. Activities that need custom worker behavior (Timeout, NextRetry, or Middleware) should instead be registered as ordinary River workers with river.AddWorker and call river.RecordOutput themselves; either way, an activity that completes without recording output fails its workflow with a *MissingOutputError.

func RegisterWorkflow

func RegisterWorkflow[I, O any](reg *Registry, kind string, fn func(ctx Context, in I) (O, error))

RegisterWorkflow registers a workflow function under a stable kind string. The function takes a typed input and returns a typed output, both JSON-encoded for storage. Registering the same kind twice panics.

func Result

func Result[TTx any, O any](ctx context.Context, client *river.Client[TTx], workflowID string) (O, bool, error)

Result returns a finished workflow's output, decoded into O.

The boolean is true only when the workflow has completed successfully. While the workflow is still running it returns (zero, false, nil). If the workflow finished unsuccessfully (stopped or failed) it returns an error wrapping ErrWorkflowFailed. It returns rivertype.ErrNotFound if no workflow with the given ID exists.

func Select

func Select(ctx Context, cases ...SelectCase)

Select waits until one of the given cases is ready, runs exactly that case's handler, and returns. If several are ready at once, the earliest-declared wins. The decision is recorded, so replays take the same branch even though the underlying signal or timer state has moved on.

A timeout case's timer keeps running even if another case wins; it fires harmlessly later. Select must be called within a workflow.

func SideEffect

func SideEffect[T any](ctx Context, fn func() T) T

SideEffect runs fn exactly once for a workflow and returns its result, even across replays and restarts. Use it to bring a nondeterministic value (a timestamp, a random number, a generated ID) into an otherwise deterministic workflow without a full activity round-trip.

fn runs only the first time this call is reached; the value is recorded durably and every later replay returns the recorded value instead of running fn again. fn must therefore not have externally-visible side effects (use an activity for those) — it should only compute a value. The value must be JSON-serializable.

SideEffect does not suspend the workflow: it returns immediately.

func Signal

func Signal[TTx any, T any](ctx context.Context, client *river.Client[TTx], workflowID, name string, payload T) error

Signal delivers a named signal with a payload to a running workflow. It is sent from outside the workflow (ordinary application code), is durable, and is delivered to whichever generation of the workflow consumes it with AwaitSignal. Signals of the same name are delivered in send order. Sending a signal to a workflow that doesn't exist (or has finished) simply leaves an unconsumed signal that is eventually reaped.

func Sleep

func Sleep(ctx Context, d time.Duration)

Sleep durably pauses the workflow for the given duration. Unlike time.Sleep, it does not block a goroutine: the workflow suspends and resumes after the duration elapses, surviving restarts in between.

The wake time is fixed when Sleep is first reached (as now plus d) and stored on a timer job, so replays reuse it — Sleep does not read the clock again. A duration of zero or less resumes as soon as possible.

func Start

func Start[TTx any, I any](ctx context.Context, client *river.Client[TTx], kind string, in I, opts *StartOpts) (string, error)

Start launches a workflow of the given registered kind with the given typed input, inserting its run job. It returns the workflow ID, which also identifies the run job in the River UI. opts may be nil.

Types

type ActivityError

type ActivityError struct {
	Kind  string
	State rivertype.JobState
}

ActivityError is returned from Future.Get when the underlying activity job reached a terminal failure state (discarded or cancelled) rather than completing successfully. Workflow code may inspect it to run compensating logic, or return it to fail the workflow.

func (*ActivityError) Error

func (e *ActivityError) Error() string

type ActivityOpts

type ActivityOpts struct {
	// InsertOpts are the River insert options for the activity job (queue,
	// priority, max attempts, etc.). Workflow bookkeeping metadata is merged in
	// automatically.
	InsertOpts *river.InsertOpts
}

ActivityOpts are options for a dispatched activity.

type CommandRecord

type CommandRecord struct {
	CommandID int
	Role      string
	Kind      string
	Args      json.RawMessage
	Output    json.RawMessage
	State     string
}

CommandRecord is a durable copy of one completed command's outcome (an activity, timer, side effect, or child workflow). It carries everything replay needs to reconstruct the command's result — including the args, which is where a side effect's captured value lives — so a workflow can replay correctly even after River has reaped the underlying job.

type Context

type Context struct {
	// contains filtered or unexported fields
}

Context is passed to a workflow function in place of a bare context.Context. It carries the standard context along with the replay engine's state.

Workflow functions must be deterministic: every run of the function replays from the top, so all nondeterminism and side effects must happen inside activities dispatched with Execute, never inline. Do not read the wall clock, generate randomness, perform I/O, spawn goroutines, or recover panics inside a workflow function.

func (Context) Context

func (c Context) Context() context.Context

Context returns the underlying context.Context. It's occasionally useful for deadline propagation, but workflow code must not use it to perform I/O.

type Engine

type Engine[TTx any] struct {
	// contains filtered or unexported fields
}

Engine wires workflow support into a River client and runs the reactive advancement loop. It's generic over the driver's transaction type, which must match the client the workflows run on.

Advancement is driven by River subscriptions rather than a worker hook because subscription events are emitted only after the completer commits an activity's state change. A hook runs before that commit, so it would observe uncommitted history and could leave a downstream step blocked on a fan-in. The blocked run job's snooze (see RunWorker) is the durability backstop if an event is ever missed (dropped under load, or a process crash between commit and delivery).

func NewEngine

func NewEngine[TTx any](reg *Registry, opts *EngineOpts) *Engine[TTx]

NewEngine returns an Engine for the given workflow registry. opts may be nil.

func (*Engine[TTx]) Install

func (e *Engine[TTx]) Install(config *river.Config, workers *river.Workers)

Install registers the workflow run worker and ensures the dedicated workflow queue exists on the config. Call it before river.NewClient, passing the same Workers bundle used for config.Workers. Activity workers are registered by the user as ordinary River workers.

func (*Engine[TTx]) Start

func (e *Engine[TTx]) Start(ctx context.Context, client *river.Client[TTx])

Start begins the reactive advancement loop and the backstop reconciler. Call it after the client has started. The loops run until Stop is called or ctx is cancelled.

func (*Engine[TTx]) Stop

func (e *Engine[TTx]) Stop()

Stop halts the advancement loop and reconciler and waits for them to exit. It is safe to call more than once.

type EngineOpts

type EngineOpts struct {
	// ReconcileInterval is how often the backstop checks for UI-initiated
	// workflow stops. Defaults to defaultReconcileInterval.
	ReconcileInterval time.Duration

	// SnoozeInterval is how long a blocked workflow run parks before re-checking
	// itself. Reactive wakes normally resume it sooner; this backstops a missed
	// wake. Defaults to defaultSnooze.
	SnoozeInterval time.Duration

	// Store, if set, is a supplemental persistence backend the engine records
	// workflow lifecycle into (see Store and the pgstore package). It enables
	// durable results, an indexed wake path, listing, and push-based Await. Nil
	// keeps the default all-state-in-river_job behavior.
	Store Store
}

EngineOpts configure an Engine.

type Future

type Future[TOut any] struct {
	// contains filtered or unexported fields
}

Future is a handle to an activity dispatched with Execute. Call Get to await its result.

func ChildWorkflow

func ChildWorkflow[I any, O any](ctx Context, kind string, in I) *Future[O]

ChildWorkflow starts another registered workflow as a child of the current one and returns a Future for its result. The child runs as its own workflow (with its own history, timers, activities, and even further child workflows); the parent awaits it exactly like an activity:

res, err := riverworkflow.ChildWorkflow[SubInput, SubResult](ctx, "sub", SubInput{…}).Get()

kind must be registered on the same registry as the parent. The child is dispatched immediately and, like Execute, several children (or activities) can be started before any is awaited to run them in parallel. The child's ID is derived deterministically from the parent, so replay reuses the same child rather than starting a new one. A child that fails or is stopped surfaces at Get as an *ActivityError.

func Execute

func Execute[TArgs river.JobArgs, TOut any](ctx Context, args TArgs, opts *ActivityOpts) *Future[TOut]

Execute dispatches an activity — an ordinary River job of type TArgs — as part of the workflow and returns a Future for its result of type TOut. The activity must record a result: register it with RegisterActivity (which records the value it returns) or, for a raw worker, call river.RecordOutput. That recorded value is what Future.Get decodes into TOut; an activity that completes without one fails the workflow with a *MissingOutputError.

Execute returns immediately. Multiple activities dispatched before their results are awaited run in parallel.

func (*Future[TOut]) Get

func (f *Future[TOut]) Get() (TOut, error)

Get returns the activity's output, decoded into TOut. If the activity hasn't finished yet, Get suspends the workflow (it does not return); the workflow resumes and replays once the activity completes. If the activity failed terminally, Get returns an *ActivityError.

type HistoryStore

type HistoryStore interface {
	// LoadHistory returns the durably-checkpointed commands for one generation
	// and the durable signal inbox for the workflow.
	LoadHistory(ctx context.Context, id string, gen int) ([]CommandRecord, []SignalRecord, error)

	// Checkpoint durably records newly-terminal commands and newly-delivered
	// signals. It must be idempotent (keyed by command id and signal id); a
	// terminal command's data never changes, so re-checkpointing is a no-op.
	Checkpoint(ctx context.Context, id string, gen int, commands []CommandRecord, signals []SignalRecord) error

	// PruneGenerationsBefore deletes checkpointed commands for generations older
	// than gen. Called after ContinueAsNew, since those generations never replay
	// again; signals are retained (they are addressed to the workflow, not a
	// generation, and carry across generations).
	PruneGenerationsBefore(ctx context.Context, id string, gen int) error

	// PruneWorkflow deletes all checkpointed history (commands and signals) for a
	// finished workflow. Its summary row and result are kept.
	PruneWorkflow(ctx context.Context, id string) error
}

HistoryStore is an optional capability a Store may implement to make a workflow's replay history durable — surviving longer than River's job retention (see docs/design/supplemental-store.md, Phase 2). The engine detects it with a type assertion; a Store that does not implement it keeps the Phase-1 behavior, where replay history is read from river_job only.

The engine checkpoints newly-terminal commands and delivered signals as a workflow advances, merges them over what river_job still holds on each replay, and prunes them when a generation is superseded or the workflow finishes.

type MissingOutputError

type MissingOutputError struct {
	Kind string
}

MissingOutputError is returned from Future.Get (and delivered to an OnFuture handler) when a completed future carries no recorded output. Futures back both activities and child workflows, but a child workflow always records a result on normal completion, so in practice this means an activity's worker returned without calling river.RecordOutput. Without a recorded value there is nothing for Get to decode, so rather than silently yielding a zero value the workflow fails. The fix is to register the activity with RegisterActivity (which records the returned value automatically) or to call river.RecordOutput in the worker.

func (*MissingOutputError) Error

func (e *MissingOutputError) Error() string

type Registry

type Registry struct {
	// contains filtered or unexported fields
}

Registry holds the workflow functions known to an engine, keyed by workflow kind.

func NewRegistry

func NewRegistry() *Registry

NewRegistry returns an empty Registry.

type RunWorker

type RunWorker[TTx any] struct {
	river.WorkerDefaults[runArgs]
	// contains filtered or unexported fields
}

RunWorker works workflow run jobs by replaying their workflow function against committed activity history. It's generic over the driver's transaction type, matching the client the workflows run on.

func (*RunWorker[TTx]) Work

func (w *RunWorker[TTx]) Work(ctx context.Context, job *river.Job[runArgs]) error

Work executes one replay of the workflow and settles the run job into one of four outcomes:

• stopped: the run job was cancelled (e.g. from the River UI) — cancel any in-flight activities and finalize as cancelled.
• failed: the workflow function returned an error (including a terminal activity failure surfaced through Future.Get) — cancel in-flight activities and finalize as cancelled.
• blocked: the function is waiting on an unfinished activity — snooze until woken by the engine (or the snooze backstop).
• completed: the function returned — record its output and finish.

Committed activity history is loaded first so the replay observes only durable results; newly encountered activities are dispatched once the replay unwinds. See the package documentation for the full lifecycle.

type SelectCase

type SelectCase struct {
	// contains filtered or unexported fields
}

SelectCase is one branch of a Select. Build cases with OnSignal, OnTimeout, or OnFuture.

func OnFuture

func OnFuture[T any](f *Future[T], handler func(T, error)) SelectCase

OnFuture is a Select case that fires when an already-dispatched activity or child-workflow future is ready; the handler receives its result (or an *ActivityError). Create the future with Execute or ChildWorkflow before Select.

func OnSignal

func OnSignal[T any](name string, handler func(T)) SelectCase

OnSignal is a Select case that fires when a signal of the given name arrives; the handler receives its payload.

func OnTimeout

func OnTimeout(d time.Duration, handler func()) SelectCase

OnTimeout is a Select case that fires after the given duration if nothing else wins first. It's the way to bound a wait (e.g. wait for a signal OR time out).

type SignalRecord

type SignalRecord struct {
	SignalID int64
	Name     string
	Payload  json.RawMessage
}

SignalRecord is a durable copy of one delivered signal, so a workflow that suspends longer than River's retention still sees its full signal inbox on replay.

type StartOpts

type StartOpts struct {
	// ID optionally sets the workflow's identifier. A random (effectively unique)
	// one is generated when empty.
	//
	// Supplying a stable ID makes Start idempotent: while a workflow with that ID
	// still exists — including after it has finished, until River's job cleaner
	// reaps it — a second Start with the same ID does not launch another
	// workflow. It returns the same ID and leaves the existing one in place. The
	// ID becomes reusable only once the prior workflow's jobs have been reaped.
	ID string
}

StartOpts are options for launching a workflow.

type Status

type Status string

Status is the high-level state of a workflow, derived from its run job.

const (
	// StatusRunning means the workflow is active (executing or suspended).
	StatusRunning Status = "running"

	// StatusCompleted means the workflow function returned successfully.
	StatusCompleted Status = "completed"

	// StatusCancelled means the workflow was stopped (from the UI or via Cancel)
	// or failed (a workflow error or a terminal activity failure).
	//
	// Note the default (river_job-only) status model cannot distinguish a stop
	// from a failure — both finalize the run job as cancelled. A configured
	// [Store] records the two separately, surfacing failures as StatusFailed.
	StatusCancelled Status = "cancelled"

	// StatusFailed means the workflow's function returned an error or one of its
	// activities failed terminally. It is only ever reported by a [Store]; the
	// river_job-only model reports these as StatusCancelled.
	StatusFailed Status = "failed"

	// StatusDiscarded means the workflow's run job exhausted its attempts on
	// infrastructure errors.
	StatusDiscarded Status = "discarded"
)

func GetStatus

func GetStatus[TTx any](ctx context.Context, client *river.Client[TTx], workflowID string) (Status, error)

GetStatus returns the current status of a workflow. It returns rivertype.ErrNotFound if no workflow with the given ID exists.

type Store

type Store interface {
	// RecordStarted upserts a workflow as running, recording its kind, current
	// generation, and the id of the run job currently orchestrating it. Called on
	// every replay; must not resurrect an already-finished workflow.
	RecordStarted(ctx context.Context, id, kind string, gen int, runJobID int64) error

	// RecordCompleted marks a workflow completed and stores its final result.
	RecordCompleted(ctx context.Context, id string, result json.RawMessage) error

	// RecordFailed marks a workflow failed (its function returned an error or an
	// activity failed terminally), storing a human-readable failure message.
	RecordFailed(ctx context.Context, id, failure string) error

	// RecordCancelled marks a workflow cancelled (stopped from the UI or via
	// Cancel). A no-op once the workflow has already finished.
	RecordCancelled(ctx context.Context, id string) error

	// RunJobID returns the run job id currently orchestrating an active workflow,
	// for the engine's wake path. ok is false if the workflow is unknown or has
	// already finished.
	RunJobID(ctx context.Context, id string) (runJobID int64, ok bool, err error)

	// LoadState returns the store's authoritative record of a workflow: its latest
	// recorded generation and its terminal outcome, if any. The run worker
	// consults it before replaying so that a workflow whose outcome is already
	// durable is never re-executed — closing the crash window between recording a
	// terminal outcome (and pruning history) and River committing the run job.
	LoadState(ctx context.Context, id string) (WorkflowState, error)
}

Store is an optional supplemental persistence backend for workflow state.

When an Engine is configured with a Store (via EngineOpts.Store), the run worker records each workflow's lifecycle into it as the workflow advances. That durable, indexed record is what powers results that outlive River's job retention, an indexed run-job lookup on the engine's wake path, listing and search, and push-based completion waiting — none of which the default (all-state-in-river_job) model provides. See docs/design/supplemental-store.md.

It is implemented by the pgstore package. Leaving EngineOpts.Store nil keeps the engine's original behavior unchanged, so the store is strictly opt-in.

All methods must be safe to call repeatedly with the same arguments: the engine may re-record a workflow across replays, retries, and its backstop reconciler, and relies on the implementation being idempotent (in particular, a terminal state must never be overwritten by a later "started" record).

type Wake

type Wake struct {
	ID       string
	RunJobID int64
}

Wake identifies a workflow that has a pending wake and the run job that should be resumed to service it.

type WakeStore

type WakeStore interface {
	// MarkWake sets the workflow's wake-pending flag and emits a cross-process
	// wake notification. A no-op once the workflow has finished.
	MarkWake(ctx context.Context, id string) error

	// ClearWake clears the wake-pending flag. The run worker calls it at the
	// start of a replay, before loading history.
	ClearWake(ctx context.Context, id string) error

	// PendingWakes returns up to limit still-running workflows whose wake-pending
	// flag is set, for the engine's targeted backstop sweep.
	PendingWakes(ctx context.Context, limit int) ([]Wake, error)

	// WakeEvents returns a channel that emits the id of each workflow whose wake
	// was notified cross-process (via MarkWake, on any process). The engine
	// ranges over it to wake runs promptly. Requires the store's listener to be
	// running; delivery is best-effort (the sweep is the backstop).
	WakeEvents() <-chan string
}

WakeStore is an optional capability that makes workflow reactivation targeted and cross-process, so the engine need not lean on the blind per-run snooze backstop (see docs/design/supplemental-store.md, §8 and Phase 3).

River's own completion subscription is lossy and process-local; without this, a dropped event is only recovered when the suspended run job snoozes and blindly re-checks itself (~30 s), reloading its whole history to usually find nothing changed. With a WakeStore the engine instead:

• marks a persistent wake-pending flag and emits a cross-process NOTIFY when a command completes (MarkWake), so every engine process can wake the run promptly; and
• runs a cheap targeted sweep (PendingWakes) that resumes only the workflows that actually have pending work, recovering a dropped NOTIFY far sooner and more cheaply than the blind snooze.

The flag is cleared at the start of each replay (ClearWake), before history is loaded, so a completion arriving during or after a replay re-arms it and is never lost. Reactivation stays correct without this capability — the snooze is the floor — so it is purely a scalability/latency optimization.

type WorkflowState

type WorkflowState struct {
	// Found is false when the workflow is not in the store yet (its first replay).
	Found bool

	// Gen is the latest generation recorded for the workflow. A run job whose own
	// generation is older has been superseded by a ContinueAsNew and must not
	// replay.
	Gen int

	// Status is the workflow's recorded status: StatusRunning while active, or a
	// terminal status once finished.
	Status Status

	// Result and Failure carry a finished workflow's outcome (Result for a
	// completed workflow, Failure for a failed one).
	Result  json.RawMessage
	Failure string
}

WorkflowState is the store's authoritative record of a workflow, used by the run worker to decide whether a replay may be skipped.

func (WorkflowState) Terminal

func (s WorkflowState) Terminal() bool

Terminal reports whether the recorded status is final.