riverworkflow

package module
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 10, 2026 License: MIT Imports: 12 Imported by: 0

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) {
    // Charge the card (an activity — an ordinary River job).
    charge, err := riverworkflow.Execute[ChargeArgs, ChargeResult](ctx, ChargeArgs{Card: in.Card}, nil).Get()
    if err != nil {
        return OrderResult{}, err
    }

    // Branch on the result at runtime.
    if charge.Declined {
        return OrderResult{Cancelled: true}, nil
    }

    // Fan out two activities in parallel, then await both.
    ship := riverworkflow.Execute[ShipArgs, ShipResult](ctx, ShipArgs{Order: in.ID}, nil)
    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
}

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.
  • Activities are just River jobs — reuse your existing workers, queues, retry policies, and telemetry. An activity returns a value with river.RecordOutput.
  • 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()
river.AddWorker(workers, &ChargeWorker{}) // activities are ordinary River workers
river.AddWorker(workers, &ShipWorker{})
river.AddWorker(workers, &InvoiceWorker{})

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{ID: 1, Card: "…"}, nil)
if err != nil {
    log.Fatal(err)
}

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

An activity worker is a normal River worker that records its return value:

type ChargeWorker struct{ river.WorkerDefaults[ChargeArgs] }

func (ChargeWorker) Work(ctx context.Context, job *river.Job[ChargeArgs]) error {
    result := charge(job.Args.Card)
    return river.RecordOutput(ctx, result) // decoded by Future.Get
}

See the runnable example for a complete program.

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.
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, cancellation, and stop-from-UI) is complete and tested. Signals, 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. Its worker does the real work and returns a value with github.com/riverqueue/river.RecordOutput; that value is what Future.Get decodes. 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()
river.AddWorker(workers, &ChargeWorker{}) // activities: normal workers
// ...

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: signals and 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

Examples

Constants

This section is empty.

Variables

View Source 
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 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 added in v0.2.0 

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.

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 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 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 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 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
}

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 added in v0.2.0 

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 worker is expected to record its result with river.RecordOutput; that value is what Future.Get decodes into TOut.

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 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 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).
	StatusCancelled Status = "cancelled"

	// 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.

Source Files

View all Source files

Directories

Path Synopsis
internal
wfdbtest
Package wfdbtest wraps riverdbtest for river-workflow's own tests.
 Package wfdbtest wraps riverdbtest for river-workflow's own tests.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.