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.