Help us shape the future of Go! Take the 2026 Go Developer Survey

workflow

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Nov 1, 2026 License: MIT Imports: 21 Imported by: 0

README

Workflow

arkoren.dev/core/service/workflow runs durable, typed workflows. A workflow orchestrates steps, waits for signals, sleeps, and starts other workflows. The facade owns what runs; a driver owns how it survives restarts and where it runs.

Install with go get arkoren.dev/core. The examples import "arkoren.dev/core/service/workflow" and "arkoren.dev/core/service/workflow/memoryworkflow", plus the standard context, errors, and time packages. Follow the workflow guide for signal delivery, driver selection, and lifecycle setup.

When onboarding must wait for email confirmation, put the mail I/O in a step and the wait in the workflow. Here User and sendWelcome are your application type and step handler:

var SendWelcome = workflow.NewStep("send-welcome", sendWelcome, func(step *workflow.StepBuilder) {
	step.Retry(5)
	step.Backoff(time.Second, 2, time.Minute)
	step.Timeout(30 * time.Second)
})

var EmailConfirmed = workflow.NewSignal[string]("email-confirmed")

Write the workflow's control flow separately:

func onboard(execution *workflow.Context, user User) (string, error) {
	if _, err := execution.Step(SendWelcome, user); err != nil {
		return "", err
	}

	token, err := execution.Wait(EmailConfirmed, workflow.Timeout(72*time.Hour))
	if errors.Is(err, workflow.ErrTimeout) {
		return "expired", nil
	}

	return token, err
}

Publish the function with its maximum duration:

var Onboard = workflow.New("onboard", onboard, func(definition *workflow.Builder) {
	definition.MaxDuration(96 * time.Hour)
})

Register the definitions with a driver. The memory driver is process-local, not durable:

registry, err := workflow.Handlers(Onboard, SendWelcome)
driver := memoryworkflow.New(registry) // or a durable driver built from the same registry

After starting your selected driver as its lifecycle requires, start and talk to runs with a client. Separate server and worker processes need a shared durable backend; they cannot share a memory driver:

client := workflow.NewClient(driver)

run, err := client.Start(ctx, Onboard, "user-42", user)
if errors.Is(err, workflow.ErrAlreadyStarted) {
	// run is the existing run; the error is safe to ignore
}

err = client.Signal(ctx, Onboard, "user-42", EmailConfirmed, "abc123")
result, err := run.Result(ctx) // ctx bounds the wait, not the run
status, err := run.Status(ctx)
err = run.Cancel(ctx)
  • Names are 1-64 lowercase letters, digits, or hyphens with no hyphen at either end. New and NewStep keep the first invalid name, nil function, or bad duration on the value (Err()); Handlers returns it as ErrInvalidName, ErrNilFunc, or ErrInvalidOptions. Workflows and steps share one namespace. Every workflow needs Builder.MaxDuration; omission fails only at registration, while an explicit nonpositive duration is invalid construction. NewSignal, which has no registration step, still panics with ErrInvalidName.
  • Definitions are immutable snapshots. Construction callbacks share one fresh draft, run directly once in order, and return no value. Settings return that draft. Nil entries and callback panics propagate; expanded nil or empty slices omit edits. Retaining a draft cannot change a published definition.
  • Change a step for one call with Step.Derive, which preserves its name, function, inherited policy, and first error. Registration snapshots the function and settings independently of the published pointer.
  • Workflow functions must be deterministic: use execution.Now(), execution.UUID(), and steps instead of the clock, random numbers, and I/O.
  • execution.Wait and execution.Select remember the first value of a signal per run.
  • execution.Child acquires cancellation ownership only when it creates the child: parent cancellation or timeout also stops that new run. Joining an existing ID ignores the new input and acquires no ownership, so parent cancellation or timeout ends only the wait. Drivers unable to decide ownership atomically and retain it across replay reject with ErrUnsupported before submission. Spawn intentionally creates detached work instead.
  • Return workflow.Permanent(err) from a step to stop retries; check with errors.Is(err, workflow.ErrPermanent).
  • ErrCanceled, ErrTimeout, and *workflow.Error keep their kind across the wire, so errors.Is works where the run was started.
  • Payloads are JSON objects {"v": ...} made by a Codec; the default is encoding/json/v2.

The package uses only the standard library. Drivers implement workflow.Driver and workflow.Runtime, and run the definitions from workflow.Handlers. To test a custom driver with workflowtest.Run, create its registry with workflowtest.Registry(t) inside the supplied driver factory.

Inside a workflow, use its execution context to disable retries for one call without changing the registered default:

attempt := SendWelcome.Derive(func(step *workflow.StepBuilder) {
	step.Retry(1)
})
_, err := execution.Step(attempt, user)

Documentation

Overview

Package workflow defines durable, typed workflows built from retried steps, signals, and sleeps, run by a swappable engine driver. A Workflow is a function that orchestrates Step values, waits for Signal values, sleeps, and starts other workflows; a Driver runs it and decides how it survives restarts. A Client starts runs and talks to them. Workflow signals are unrelated to the payload-less signals of package arkoren.dev/core/event. 

var SendWelcome = workflow.NewStep("send-welcome", sendWelcome, func(step *workflow.StepBuilder) {
	step.Retry(5)
	step.Backoff(time.Second, 2, time.Minute)
})
var EmailConfirmed = workflow.NewSignal[string]("email-confirmed")

var Onboard = workflow.New("onboard", func(c *workflow.Context, user User) (string, error) {
	if _, err := c.Step(SendWelcome, user); err != nil {
		return "", err
	}

	token, err := c.Wait(EmailConfirmed, workflow.Timeout(72*time.Hour))
	if errors.Is(err, workflow.ErrTimeout) {
		return "expired", nil
	}

	return token, err
}, func(definition *workflow.Builder) {
	definition.MaxDuration(96 * time.Hour)
})

A worker process registers the definitions with the driver: 

registry, err := workflow.Handlers(Onboard, SendWelcome)

Registry.Workflows and Registry.Steps return eager collection snapshots sorted by name. Their records are detached from registration maps, not live queries; functions and their captures remain shared, not cloned by wrappers.

A server process only needs a client: 

client := workflow.NewClient(driver)
run, err := client.Start(ctx, Onboard, "user-42", user)
err = client.Signal(ctx, Onboard, "user-42", EmailConfirmed, "abc")
result, err := run.Result(ctx)

Determinism

A driver may replay a workflow function from the start to rebuild its state. The function must therefore not read the clock, random numbers, the network, or global state directly. Use Context.Now, Context.UUID, and steps, which a driver records. Inputs and outputs must round-trip through the Codec, which is encoding/json/v2 by default.

Signals

Context.Wait returns the first value of a signal and remembers it: a later Wait or Context.Select on the same signal in the same run sees the same value. A signal sent before the workflow waits for it is not lost.

Child ownership

Context.Child waits for a newly created or existing run. Only the caller that creates the child owns its cancellation: parent cancellation or timeout stops newly created children. Joining an existing run ignores the new input and acquires no ownership; cancellation or timeout ends only the joining wait. Drivers must preserve that atomic decision across replay or reject Child with ErrUnsupported before submitting work. Use Context.Spawn for intentionally detached work, not as a substitute for unsupported Child semantics.

Errors

A failure of a run reaches Run.Result as an error. ErrCanceled, ErrTimeout, and any *Error keep their kind across the wire, so errors.Is works in the process that started the run. A step that returns Permanent is not retried.

Failure messages

A failing run stores the text of its error in the engine's history, and Run.Result returns it to every caller, in any process. Error text is persisted and exported, so keep secrets and personal data out of the messages of errors a workflow function or step returns; wrap them with NewError and a generic message when the underlying error may carry values.

Registration

Definitions are closures registered by name in every worker process. New and NewStep publish immutable definitions from Builder and StepBuilder callbacks. Settings return the same draft, and the first error (an invalid name, a nil function, a bad duration) is kept and reported by Workflow.Err and Step.Err and returned by Handlers as ErrInvalidName, ErrNilFunc, or ErrInvalidOptions. Callbacks run directly once, synchronously in order on one fresh draft; supplied nil entries and application panics propagate. Expanded nil or empty callback slices omit edits. Publication detaches the settings, so retained drafts cannot change definitions. Registration captures the name, function, and settings independently of the published pointer. Names are 1-64 lowercase letters, digits, or hyphens with no hyphen at either end, and share one namespace for workflows and steps. Run ids are free-form but not empty and free of ASCII control characters, else ErrInvalidID. A workflow must set Builder.MaxDuration; omission is rejected only at registration. Zero workflows and steps are invalid and report ErrInvalidName. Client.Start, Context.Step, Context.Child, and Context.Spawn reject stored declaration errors before encoding or dispatch, even when a worker registered a valid definition under that same name. Deriving a zero step preserves its declaration error. To change the settings of one step call, use Step.Derive: 

singleAttempt := SendWelcome.Derive(func(step *workflow.StepBuilder) {
	step.Retry(1)
})
_, err := workflowContext.Step(singleAttempt, user)

Derive preserves the name, function, inherited policy, and first error. Backoff initial zero requests immediate step retries without a deliberate positive delay. An engine unable to express the completed multi-attempt policy returns ErrUnsupported before scheduling or executing the affected step. It must not substitute a positive default or retry the whole workflow instead. Single attempts and positive backoff retain their normal support. Provider duration limits remain provider constraints; this defines no new core maximum. Names, JSON payloads, and an explicit serialized trace carrier cross the driver; context.Context itself never crosses processes.

Events

NewClient uses ClientBuilder callbacks for Codec, Telemetry, and Events, separate from definition and per-start options. Callbacks run once in order on one fresh draft; nil entries panic and nil/empty slice expansion omits edits. Publication detaches retained roots, not shared codecs or event buses. A final nil codec selects JSON and must match the worker registry's encoding.

With ClientBuilder.Events, a Client publishes RunStarted ("workflow.run.started") after Client.Start starts a run, RunSignaled ("workflow.run.signaled") after Client.Signal delivers a signal, and RunCanceled ("workflow.run.canceled") after Run.Cancel on a run from that client. Payloads carry the workflow name, the run ID, and for signals the signal name; never inputs, outputs, metadata, or signal values. Failed calls publish nothing. These are local successful-action notifications, not a distributed engine completion feed. Polling Result or Status does not publish terminal events.

Telemetry

ClientBuilder.Telemetry observes start, signal, cancel, status, and result calls. Result duration is wait latency, not execution time. Configure worker observations separately through NewRegistry and RegistryBuilder.Telemetry: 

registry, err := workflow.NewRegistry([]workflow.Definition{Onboard, SendWelcome},
	func(worker *workflow.RegistryBuilder) {
		worker.Telemetry(observations)
	})
if err != nil {
	return err
}
_ = registry.Steps()

Nil telemetry follows the current process default per call/attempt. Actual StepHandler.Handle invocations produce workflow step_attempt spans, a count, and a seconds histogram. Each retry is a new attempt; replayed orchestration, suspension, sleeps, and recorded step results produce no attempt observations. The step context supports dependency child spans. No payload, signal value, output, metadata, run ID, or raw error text enters automatic telemetry; metric labels contain only operation, outcome, and classified error type.

StartOptions.Trace carries propagation independently of application metadata. Drivers forward it as deterministic data to children/spawns where supported, then Registry.NewTraceContext extracts it onto each native attempt context. Signals do not replace this start parent. Configure an exporter in every pod and use a common backend for distributed traces; local event buses and memory capture do not aggregate remote pods. Only trusted submitters should be able to supply carriers. No terminal engine subscription exists in this contract.

Index

Constants

This section is empty.

Variables

View Source 
var ErrAlreadyStarted = NewError("workflow.already_started", "workflow: already started")

ErrAlreadyStarted reports a run that already exists under the same name and identifier. Client.Start also returns the existing run handle for idempotent callers. 

run, err := client.Start(ctx, Onboard, "user-42", user)
if err != nil && !errors.Is(err, workflow.ErrAlreadyStarted) {
	return err
}
_ = run.ID()
View Source 
var ErrCanceled = NewError("workflow.canceled", "workflow: canceled")

ErrCanceled reports a run, step, or wait that was canceled. Its kind survives the driver wire representation. 

if errors.Is(err, workflow.ErrCanceled) {
	return nil // Cancellation was the requested outcome.
}
View Source 
var ErrDuplicateName = errors.New("duplicate workflow name")

ErrDuplicateName reports two definitions registered under one name. Workflows and steps share the registry's name namespace. 

_, err := workflow.Handlers(Onboard, Onboard)
if errors.Is(err, workflow.ErrDuplicateName) {
	return err
}
View Source 
var ErrInvalidID = errors.New("invalid run id")

ErrInvalidID reports an empty run identifier, one containing ASCII control characters, or a child that uses the identifier of its own run. It rejects unsafe identities before a client submits work. 

if _, err := client.Start(ctx, Onboard, "", user); errors.Is(err, workflow.ErrInvalidID) {
	return err
}
View Source 
var ErrInvalidName = errors.New("invalid name")

ErrInvalidName reports a name that is empty or has characters outside [a-z0-9-], or is otherwise rejected by the shared identifier rule. The definition constructors keep it on the value; Handlers returns it. 

definition := workflow.New("Invalid", onboard)
if errors.Is(definition.Err(), workflow.ErrInvalidName) {
	return definition.Err()
}
View Source 
var ErrInvalidOptions = errors.New("invalid options")

ErrInvalidOptions reports an out-of-range duration in workflow or step settings. Later declaration edits cannot clear the first invalid setting. 

definition := workflow.New("onboard", onboard, func(definition *workflow.Builder) {
	definition.MaxDuration(-time.Second)
})
if errors.Is(definition.Err(), workflow.ErrInvalidOptions) {
	return definition.Err()
}
View Source 
var ErrMissingMaxDuration = errors.New("workflow requires MaxDuration")

ErrMissingMaxDuration reports a workflow registered without Builder.MaxDuration. Err deliberately leaves this required worker registration check to Handlers. 

_, err := workflow.Handlers(workflow.New("onboard", onboard))
if errors.Is(err, workflow.ErrMissingMaxDuration) {
	return err
}
View Source 
var ErrNilFunc = errors.New("nil function")

ErrNilFunc reports a definition built with a nil function, or a nil definition passed to Handlers. This is a declaration error, not construction-callback misuse. 

definition := workflow.NewStep[string, string]("charge", nil)
if errors.Is(definition.Err(), workflow.ErrNilFunc) {
	return definition.Err()
}
View Source 
var ErrNoCases = errors.New("select without cases")

ErrNoCases reports a Context.Select without cases. Supply at least one signal or timer alternative before waiting. 

if err := workflowContext.Select(); errors.Is(err, workflow.ErrNoCases) {
	return "", err
}
View Source 
var ErrNotFound = NewError("workflow.not_found", "workflow: not found")

ErrNotFound reports an unknown workflow or run, or a signal sent to a run that is not running. 

if errors.Is(err, workflow.ErrNotFound) {
	return nil // The run has already disappeared or ended.
}
View Source 
var ErrPermanent = errors.New("permanent failure")

ErrPermanent matches, through errors.Is, an error marked with Permanent. A step that fails with such an error stops at once even when attempts remain. 

if errors.Is(err, workflow.ErrPermanent) {
	return err // Do not apply an application-level retry around the step.
}
View Source 
var ErrTimeout = NewError("workflow.timeout", "workflow: timeout")

ErrTimeout reports an expired wait, step timeout, or workflow Builder.MaxDuration. Match the error rather than its persisted human-readable text. 

if errors.Is(err, workflow.ErrTimeout) {
	return workflow.NewError("onboarding.expired", "confirmation expired")
}
View Source 
var ErrUnsupported = NewError("workflow.unsupported", "workflow: unsupported by driver")

ErrUnsupported reports an operation the driver cannot perform. Select a compatible engine rather than retrying the same unsupported call. 

if errors.Is(err, workflow.ErrUnsupported) {
	return fmt.Errorf("this workflow needs a different engine: %w", err)
}
View Source 
var RunCanceled = event.NewTopic[RunEvent]("workflow.run.canceled")

RunCanceled is published when Run.Cancel has asked the driver to cancel a run. A failed cancel publishes nothing. Canceling a finished run is not an error and publishes the event as well. The observation is not a distributed subscription to engine terminal states. 

unsubscribe, err := bus.Listen(workflow.RunCanceled, func(_ context.Context, notice workflow.RunEvent) error {
	fmt.Println(notice.Workflow)
	return nil
})
if err != nil {
	return err
}
defer unsubscribe()
View Source 
var RunSignaled = event.NewTopic[RunSignaledEvent]("workflow.run.signaled")

RunSignaled is published when Client.Signal has delivered a signal to a run. A failed signal publishes nothing. It never includes the signal payload or replaces a run's trace parent. 

unsubscribe, err := bus.Listen(workflow.RunSignaled, func(_ context.Context, notice workflow.RunSignaledEvent) error {
	fmt.Println(notice.Signal)
	return nil
})
if err != nil {
	return err
}
defer unsubscribe()
View Source 
var RunStarted = event.NewTopic[RunEvent]("workflow.run.started")

RunStarted is published when Client.Start has started a run. A start that fails, including one that finds the run already started, publishes nothing. Configure ClientBuilder.Events with the same bus before observing commands. 

unsubscribe, err := bus.Listen(workflow.RunStarted, func(_ context.Context, notice workflow.RunEvent) error {
	fmt.Println(notice.Workflow)
	return nil
})
if err != nil {
	return err
}
defer unsubscribe()

Functions

func DecodeError 

func DecodeError(data []byte) error

DecodeError restores an error from EncodeError. The result is an *Error that matches the sentinels such as ErrCanceled and ErrTimeout through errors.Is, and is marked with Permanent when the original was. Data that is not an envelope becomes an *Error with the data as message. DecodeError of empty data is nil. Decoded messages remain persisted application data, not safe telemetry text. 

failure := workflow.DecodeError(workflow.EncodeError(workflow.ErrTimeout))
if errors.Is(failure, workflow.ErrTimeout) {
	return failure
}

func DecodeValue 

func DecodeValue(codec Codec, data []byte, v any) error

DecodeValue decodes an envelope made by EncodeValue into v, which is a pointer. Empty data leaves v unchanged. Data that is not an envelope, or has no "v" member, is an error. A nil codec uses JSON. JSON null is a present value, including in whitespace-formatted envelopes. 

var greeting string
if err := workflow.DecodeValue(nil, []byte(`{"v":"welcome"}`), &greeting); err != nil {
	return err
}
fmt.Println(greeting)

func EncodeError 

func EncodeError(err error) []byte

EncodeError encodes err as {"kind","message","permanent"} so a driver can carry it across the wire. The kind and message come from the first *Error in the chain, so wrapping context is dropped. Without one, context.Canceled maps to the kind of ErrCanceled and context.DeadlineExceeded to the kind of ErrTimeout. EncodeError(nil) is nil. Only the serialized kind and message cross the boundary, not arbitrary Go error identities. Keep secrets out of messages before calling this function. 

failure := workflow.NewError("billing.card-declined", "payment declined")
encoded := workflow.EncodeError(workflow.Permanent(failure))
_ = encoded // Store or transport this envelope, not the original cause.

func EncodeValue 

func EncodeValue(codec Codec, v any) ([]byte, error)

EncodeValue encodes v with codec inside the envelope {"v": ...}. A nil codec uses JSON. The output of the codec must be valid JSON; it is not re-validated. Empty codec output is carried as JSON null; the returned envelope owns its bytes independently of the codec's buffer. 

data, err := workflow.EncodeValue(nil, "welcome")
if err != nil {
	return err
}
_ = data // {"v":"welcome"}

func Permanent 

func Permanent(err error) error

Permanent marks err as non-retryable: a step that returns it fails at once even when attempts remain. Check for the mark with errors.Is(err, ErrPermanent). Permanent(nil) is nil.

Example: 

return Receipt{}, workflow.Permanent(errors.New("card declined"))

Types

type Builder 

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

Builder configures one workflow declaration before New publishes its immutable snapshot. Retaining this draft does not permit later changes to the published workflow. Settings preserve the first declaration error.

Example: 

workflow.New("onboard", onboard, func(definition *workflow.Builder) {
	definition.MaxDuration(24 * time.Hour)
})

func (*Builder) Description 

func (definition *Builder) Description(text string) *Builder

Description sets free text that documents the workflow, replacing any earlier description. Empty text clears it. It returns the same draft without clearing an earlier validation error.

Example: 

workflow.New("onboard", onboard, func(definition *workflow.Builder) {
	definition.MaxDuration(time.Hour)
	definition.Description("Welcomes a new user.")
})

func (*Builder) MaxDuration 

func (definition *Builder) MaxDuration(duration time.Duration) *Builder

MaxDuration bounds the whole run. It is required: Handlers fails for a workflow without it. When it expires the run fails with ErrTimeout. A value of zero or less is an error. It returns the same draft and preserves any earlier declaration error.

Example: 

workflow.New("onboard", onboard, func(definition *workflow.Builder) {
	definition.MaxDuration(24 * time.Hour)
})

func (*Builder) Retention 

func (definition *Builder) Retention(duration time.Duration) *Builder

Retention sets how long the engine keeps a finished run. Zero uses the engine default and a negative value is an error. It returns the same draft and preserves the first declaration error.

Example: 

workflow.New("onboard", onboard, func(definition *workflow.Builder) {
	definition.MaxDuration(time.Hour)
	definition.Retention(7 * 24 * time.Hour)
})

type BuilderFunc 

type BuilderFunc = func(*Builder)

BuilderFunc edits the fresh workflow declaration draft used by New. Callbacks run synchronously once in order, even for invalid declarations; supplied nil callbacks panic. Set MaxDuration before registering the result.

Example: 

policy := workflow.BuilderFunc(func(definition *workflow.Builder) {
	definition.MaxDuration(time.Minute)
})
onboard := workflow.New("onboard", onboardUser, policy)
if err := onboard.Err(); err != nil {
	return err
}

type Case 

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

Case is one alternative of Context.Select. Create it with On or After. Case callbacks execute synchronously on the workflow's orchestration path and must follow the same replay/determinism restrictions as its body. 

approved := workflow.NewSignal[bool]("approved")
var accepted bool
choice := workflow.On(approved, func(value bool) {
	accepted = value
})
if err := workflowContext.Select(choice); err != nil {
	return "", err
}
_ = accepted

func After 

func After(d time.Duration, fn func()) Case

After returns a case that calls fn once d has passed. fn must not be nil; After panics with ErrNilFunc otherwise. A nonpositive delay becomes an immediately eligible timer alternative; the engine still arbitrates it against available signals. 

var expired bool
if err := workflowContext.Select(workflow.After(time.Minute, func() {
	expired = true
})); err != nil {
	return "", err
}
_ = expired

func On 

func On[T any](s Signal[T], fn func(T)) Case

On returns a case that calls fn with the value of signal s. fn must not be nil; On panics with ErrNilFunc otherwise. Decoding errors stop Select before fn is invoked. The first remembered value for the signal is reused by later waits and selections in the same run. 

approved := workflow.NewSignal[bool]("approved")
var accepted bool
choice := workflow.On(approved, func(value bool) {
	accepted = value
})
if err := workflowContext.Select(choice); err != nil {
	return "", err
}
_ = accepted

type Client 

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

Client starts workflows and talks to their runs through a Driver. Create it with NewClient. A Client is safe for concurrent use. The driver and codec remain shared; their lifecycle belongs to their owner. 

client := workflow.NewClient(driver)
run, err := client.Start(ctx, Onboard, "user-42", user)
if err != nil {
	return err
}
_ = run.ID()

func NewClient 

func NewClient(driver Driver, configure ...ClientBuilderFunc) *Client

NewClient returns a client using driver without invoking handlers or doing I/O. Callbacks run directly once in order on one fresh draft. Nil entries panic; expanded nil/empty slices omit edits. Retaining a draft cannot change clients. A final nil codec uses JSON; registry compatibility remains the caller's job. 

client := workflow.NewClient(driver, func(client *workflow.ClientBuilder) {
	client.Codec(workflow.JSON{})
})

func (*Client) Driver 

func (client *Client) Driver() Driver

Driver returns the underlying Driver. Raw operations bypass the client's typed encoding and action observations. 

backend := client.Driver()
status, err := backend.Status(ctx, "onboard", "user-42")
if err != nil {
	return err
}
_ = status

func (*Client) DriverAs 

func (client *Client) DriverAs[T any]() (T, bool)

DriverAs returns the first driver of type T in the chain of the underlying driver. A driver that wraps another exposes it with an Unwrap() Driver method. The returned value has the original driver's ownership; lookup does no I/O. 

backend, ok := client.DriverAs[*memoryworkflow.Driver]()
if ok {
	_ = backend // The composition owner remains responsible for Stop.
}

func (*Client) Run 

func (client *Client) Run[In, Out any](w *Workflow[In, Out], id string) *Run[Out]

Run returns the handle of run id of workflow w without any I/O. Its methods fail with ErrInvalidID when id is empty or contains control characters. Obtaining a handle does not establish that the run exists. 

run := client.Run(Onboard, "user-42")
status, err := run.Status(ctx)
if err != nil {
	return err
}
_ = status

func (*Client) Signal 

func (client *Client) Signal[In, Out, T any](ctx context.Context, w *Workflow[In, Out], id string, s Signal[T], v T) (err error)

Signal sends value v of signal s to run id of workflow w. It returns ErrNotFound when the run does not exist or has ended. Telemetry observes delivery acceptance only. A signal does not replace the durable run's start parent, and its value is never recorded. 

if err := client.Signal(ctx, Onboard, "user-42", Approved, approval); err != nil {
	return err
}

func (*Client) Start 

func (client *Client) Start[In, Out any](ctx context.Context, w *Workflow[In, Out], id string, in In, opts ...StartOption) (_ *Run[Out], err error)

Start begins run id of workflow w with input in and returns its handle. When the run already exists it returns the handle of the existing run together with an error that satisfies errors.Is with ErrAlreadyStarted, so callers may ignore that error to start idempotently. One observation includes validation and encoding; ErrAlreadyStarted is an expected telemetry outcome. The start span is injected into a separate carrier, never into application metadata or payload values. Invalid declarations fail with Workflow.Err before encoding or driver I/O. 

run, err := client.Start(ctx, Onboard, "user-42", user)
if err != nil && !errors.Is(err, workflow.ErrAlreadyStarted) {
	return err
}
_ = run.ID()

type ClientBuilder 

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

ClientBuilder edits client-only settings, not workflow definitions or starts. NewClient copies the final draft while keeping codecs and event buses shared. 

client := workflow.NewClient(driver, func(client *workflow.ClientBuilder) {
	client.Codec(workflow.JSON{})
})

func (*ClientBuilder) Codec 

func (client *ClientBuilder) Codec(value Codec) *ClientBuilder

Codec replaces the payload codec. A final nil selects JSON at construction; use the same codec for the worker Registry. It returns the same draft. 

client := workflow.NewClient(driver, func(client *workflow.ClientBuilder) {
	client.Codec(workflow.JSON{})
})

func (*ClientBuilder) Events 

func (client *ClientBuilder) Events(value *event.Bus) *ClientBuilder

Events selects the shared bus for run outcomes. Nil publishes nothing; it does not affect per-start metadata. It returns the same draft. 

client := workflow.NewClient(driver, func(client *workflow.ClientBuilder) {
	client.Events(app.Events())
})

func (*ClientBuilder) Telemetry 

func (client *ClientBuilder) Telemetry(value *telemetry.Telemetry) *ClientBuilder

Telemetry selects client-call observation and start-carrier injection. It does not configure workers in this or another process. Nil follows the process default on each call. The service remains shared after publication. Panics still propagate unchanged; an interrupted call closes its started observation as workflow.panic without recording the panic value or type. 

client := workflow.NewClient(driver, func(client *workflow.ClientBuilder) {
	client.Telemetry(observations)
})

type ClientBuilderFunc 

type ClientBuilderFunc = func(*ClientBuilder)

ClientBuilderFunc edits the fresh construction draft used by NewClient. Callbacks run synchronously once in order; retaining a draft cannot replace the published client's settings. A supplied nil callback panics.

Example: 

policy := workflow.ClientBuilderFunc(func(client *workflow.ClientBuilder) {
	client.Codec(workflow.JSON{})
})
client := workflow.NewClient(driver, policy)

type Codec 

type Codec interface {
	// Marshal encodes v as JSON.
	Marshal(v any) ([]byte, error)

	// Unmarshal decodes JSON data into v, which is a pointer.
	Unmarshal(data []byte, v any) error
}

Codec converts values to and from the bytes that cross a driver. The output of Marshal must be valid JSON, because payloads are wrapped in a JSON object envelope. Implementations remain shared by clients and registered handlers and must support their concurrent use. Use the same encoding in both processes. 

var codec workflow.Codec = workflow.JSON{}
client := workflow.NewClient(driver, func(client *workflow.ClientBuilder) {
	client.Codec(codec)
})
_ = client.Driver()

type Context 

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

Context is what a workflow function uses to run steps, wait for signals, sleep, and start other workflows. It wraps the Runtime of the driver. Do not use the clock, random numbers, or the network directly in a workflow function; use Now, UUID, and steps so a driver can replay it. Drivers supply the Context to a registered workflow handler; callers do not construct it directly. Side effects belong in registered steps. 

definition := workflow.New("welcome", func(ctx *workflow.Context, user string) (string, error) {
	if err := ctx.Sleep(time.Minute); err != nil {
		return "", err
	}
	return user, nil
}, func(definition *workflow.Builder) {
	definition.MaxDuration(time.Hour)
})
_ = definition.Err()

func (*Context) Child 

func (workflowContext *Context) Child[In, Out any](w *Workflow[In, Out], id string, in In) (Out, error)

Child runs workflow w as run id and waits for its output. If this call creates the run, parent cancellation or MaxDuration also stops that child. When a run with the same id already exists, Child joins it and ignores in: joining acquires no cancellation ownership. Parent cancellation or MaxDuration ends only that wait and must not stop the existing run, even if another parent created it. Creation versus joining must be decided atomically and retained across replay. A driver unable to preserve these guarantees returns ErrUnsupported before submitting child work. An empty id, or the id of the calling run itself, fails with ErrInvalidID. Workflow.Err is checked before encoding or asking the runtime to start/join.

Example: 

receipt, err := workflowContext.Child(fulfill, "order-42", order)
if err != nil {
	return "", err
}

func (*Context) Done 

func (workflowContext *Context) Done() <-chan struct{}

Done is closed when the run is canceled or timed out. Poll it with a non-blocking select; blocking on it is not replay-safe on every engine. Durable calls such as Sleep and Wait return early on cancellation.

Example: 

select {
case <-workflowContext.Done():
	return "", workflowContext.Err()
default:
}

func (*Context) Err 

func (workflowContext *Context) Err() error

Err returns why Done is closed, ErrCanceled or ErrTimeout, or nil.

Example: 

if err := workflowContext.Err(); err != nil {
	return "", err
}

func (*Context) ID 

func (workflowContext *Context) ID() string

ID returns the run identifier.

Example: 

runID := workflowContext.ID()

func (*Context) Logger 

func (workflowContext *Context) Logger() *slog.Logger

Logger returns the logger of the run.

Example: 

workflowContext.Logger().Info("processing order")

func (*Context) Name 

func (workflowContext *Context) Name() string

Name returns the workflow name.

Example: 

name := workflowContext.Name()

func (*Context) Now 

func (workflowContext *Context) Now() time.Time

Now returns the time of the run. Use it instead of time.Now.

Example: 

startedAt := workflowContext.Now()

func (*Context) Raw 

func (workflowContext *Context) Raw() any

Raw returns the native context of the engine behind the driver, for example a Temporal workflow.Context. Its type depends on the driver.

Example: 

nativeContext := workflowContext.Raw()

func (*Context) Select 

func (workflowContext *Context) Select(cases ...Case) error

Select blocks until one case fires and runs its function. A signal case whose signal was already received fires at once, in case order; otherwise the first signal or timer to fire wins. A signal resolves once per run: a signal case records the value, and later Context.Wait or Select calls for that signal return the same value without waiting for a new one. A timer case with a zero or negative delay fires after the shortest delay.

Example: 

confirmed := workflow.NewSignal[string]("confirmed")
var result string
err := workflowContext.Select(workflow.On(confirmed, func(token string) {
	result = token
}), workflow.After(time.Minute, func() {
	result = "expired"
}))
if err != nil {
	return "", err
}
return result, nil

func (*Context) Sleep 

func (workflowContext *Context) Sleep(d time.Duration) error

Sleep pauses the run for d. It returns early with the error of Context.Err when the run ends.

Example: 

if err := workflowContext.Sleep(time.Minute); err != nil {
	return "", err
}

func (*Context) Spawn 

func (workflowContext *Context) Spawn[In, Out any](w *Workflow[In, Out], id string, in In) (string, error)

Spawn starts workflow w as run id without waiting and returns id. The run outlives the parent. A run that already exists is not an error, so a replay of the parent spawns once; use Client.Run with the id to query it. A driver that cannot start runs from a workflow returns ErrUnsupported. An empty id, or the id of the calling run itself, fails with ErrInvalidID. Workflow.Err is checked before encoding or submitting detached work.

Example: 

childID, err := workflowContext.Spawn(sendReceipt, "receipt-42", order)
if err != nil {
	return "", err
}

func (*Context) Step 

func (workflowContext *Context) Step[In, Out any](s *Step[In, Out], in In) (Out, error)

Step runs step s with input in and returns its output. To change the settings for one call, use Step.Derive; it keeps the name. A step with a non-nil Step.Err fails without running. A failure that the step marked with Permanent is returned without further retries.

Example: 

singleAttempt := charge.Derive(func(step *workflow.StepBuilder) {
	step.Retry(1)
})
receipt, err := workflowContext.Step(singleAttempt, order)
if err != nil {
	return "", err
}

func (*Context) UUID 

func (workflowContext *Context) UUID() string

UUID returns a new random identifier. Use it instead of random numbers.

Example: 

childID := workflowContext.UUID()

func (*Context) Wait 

func (workflowContext *Context) Wait[T any](s Signal[T], opts ...WaitOption) (T, error)

Wait blocks until signal s arrives and returns its value. A signal resolves once per run: the first value received per signal name is kept, and every later Wait or Select for the same signal returns that value at once, even if the sender sent more. Timeout bounds the wait and yields ErrTimeout; zero or no Timeout waits without limit.

Example: 

confirmed := workflow.NewSignal[string]("confirmed")
token, err := workflowContext.Wait(confirmed, workflow.Timeout(time.Minute))
if err != nil {
	return "", err
}

type Definition 

type Definition interface {
	// contains filtered or unexported methods
}

Definition is a workflow or step that a Driver can run. *Workflow and *Step implement it; pass them to Handlers. The interface is sealed so registry validation and typed decoding remain owned by this package rather than arbitrary caller implementations. 

definitions := []workflow.Definition{Onboard, SendWelcome}
registry, err := workflow.NewRegistry(definitions)
if err != nil {
	return err
}
_ = registry.Workflows()

type Driver 

type Driver interface {
	// Start begins run id of the workflow called name with input in. It
	// returns [ErrAlreadyStarted] when the run exists, finished or not, until
	// the engine forgets it after its retention, and [ErrNotFound] when the
	// workflow is not registered. Concurrent starts of one id have exactly one
	// winner.
	// Trace is a separate, owned propagation carrier: retain a copy, carry it
	// through durable child/spawn scheduling, and extract it only at actual
	// step attempts. Never merge it with Metadata or retain the caller context.
	//
	//	err := driver.Start(ctx, "onboard", "user-42", input, workflow.StartOptions{})
	//	if err != nil && !errors.Is(err, workflow.ErrAlreadyStarted) {
	//		return err
	//	}
	Start(ctx context.Context, name, id string, in []byte, o StartOptions) error

	// Result blocks until the run ends, or until ctx is done, and returns the
	// encoded output or the failure of the run. [ErrNotFound] for an unknown
	// run.
	Result(ctx context.Context, name, id string) ([]byte, error)

	// Status returns the state of the run: [Running] while it has not ended,
	// [Failed] after an error or [ErrTimeout], [Canceled] after a cancel.
	// [ErrNotFound] for an unknown run.
	Status(ctx context.Context, name, id string) (Status, error)

	// Cancel stops the run. Its result becomes [ErrCanceled]. Canceling a
	// finished run is not an error; an unknown run is [ErrNotFound].
	Cancel(ctx context.Context, name, id string) error

	// Signal delivers a named value to a running run. It returns
	// [ErrNotFound] for an unknown or finished run.
	Signal(ctx context.Context, name, id, signal string, v []byte) error
}

Driver runs workflows. It works on bytes: names, run identifiers, and payloads made by EncodeValue. Failures cross it as errors that EncodeError and DecodeError can carry; a driver returns the sentinels such as ErrNotFound and ErrAlreadyStarted for the matching conditions. Applications normally use Client; raw callers own encoding and engine lifecycle. Driver methods must preserve error identity through wrapping. 

client := workflow.NewClient(driver)
run, err := client.Start(ctx, Onboard, "user-42", user)
if err != nil {
	return err
}
_ = run.ID()

type Error 

type Error struct {
	// Kind classifies the failure, for example "billing.card_declined".
	Kind string

	// Message describes the failure for people.
	Message string
}

Error is a failure with a stable Kind that survives the wire. Drivers carry it with EncodeError and DecodeError, and errors.Is matches two errors of the same non-empty Kind. Message text crosses process boundaries; use a stable, nonsensitive message rather than embedding credentials or input values in a failure. 

declined := workflow.NewError("billing.card-declined", "payment declined")
if errors.Is(err, declined) {
	return declined
}

func NewError 

func NewError(kind, msg string) *Error

NewError returns an Error with the given kind and message. Applications use it for failures that callers must recognize with errors.Is after the error crossed a process boundary. Kinds that start with "workflow." are reserved for the sentinels of this package; do not use them. Empty kinds do not match other errors by kind. NewError does not retain an underlying cause; it deliberately publishes only the supplied safe message. 

declined := workflow.NewError("billing.card-declined", "payment declined")
return Receipt{}, declined

func (*Error) Error 

func (err *Error) Error() string

Error returns the message, or the kind when the message is empty. It adds no wrapping context, which keeps the cross-process failure readable. 

failure := workflow.NewError("billing.card-declined", "payment declined")
fmt.Println(failure.Error()) // payment declined

func (*Error) Is 

func (err *Error) Is(target error) bool

Is reports whether target is an *Error with the same non-empty Kind. Use errors.Is to find matching failures through wrapping and wire decoding. 

target := workflow.NewError("billing.card-declined", "payment declined")
received := workflow.DecodeError(workflow.EncodeError(target))
fmt.Println(errors.Is(received, target)) // true

type JSON 

type JSON struct{}

JSON is the default Codec. It uses encoding/json/v2. Its zero value is ready to use and holds no mutable state. 

codec := workflow.JSON{}
data, err := codec.Marshal("welcome")
if err != nil {
	return err
}
_ = data

func (JSON) Marshal 

func (JSON) Marshal(v any) ([]byte, error)

Marshal encodes v with encoding/json/v2. Unsupported values return the JSON encoder's error rather than a payload. 

data, err := (workflow.JSON{}).Marshal(map[string]string{"greeting": "welcome"})
if err != nil {
	return err
}
_ = data

func (JSON) Unmarshal 

func (JSON) Unmarshal(data []byte, v any) error

Unmarshal decodes data with encoding/json/v2. v must be an appropriate nonnil destination pointer; invalid JSON and incompatible destinations return the decoder's error. 

var greeting string
if err := (workflow.JSON{}).Unmarshal([]byte(`"welcome"`), &greeting); err != nil {
	return err
}
fmt.Println(greeting)

type RawCase 

type RawCase struct {
	// Signal is the signal to wait for. Empty for a timer case.
	Signal string

	// After is the delay of a timer case.
	After time.Duration
}

RawCase is one alternative of Runtime.Select: a signal name or, when Signal is empty, a timer of After. This raw record is for driver adapters. Application workflows create typed alternatives with On and After rather than decoding signal bytes themselves. 

cases := []workflow.RawCase{{Signal: "approved"}, {After: time.Minute}}
index, value, err := runtime.Select(cases)
if err != nil {
	return err
}
_, _ = index, value

type Registry 

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

Registry holds the handlers of validated definitions for a driver to register. Use NewRegistry to select worker telemetry or a custom codec; Handlers uses JSON and follows the dynamic process telemetry default. Published handlers snapshot settings; service implementations stay shared. 

registry, err := workflow.NewRegistry(definitions, func(worker *workflow.RegistryBuilder) {
	worker.Telemetry(observations)
})
if err != nil {
	return err
}
_ = registry.Steps()

func Handlers 

func Handlers(defs ...Definition) (*Registry, error)

Handlers validates defs and returns their handlers, encoded with JSON. It returns the Workflow.Err or Step.Err of a definition, so an invalid name satisfies ErrInvalidName and a nil function ErrNilFunc, then ErrDuplicateName when two definitions share a name and ErrMissingMaxDuration for a workflow without Builder.MaxDuration. Each handler captures the name, function, and settings independently of the published pointer; subsequent assignment through that pointer does not change the registry. Captured function closures retain their own ownership rules.

Example: 

registry, err := workflow.Handlers(Onboard, SendWelcome)
if err != nil {
	return err
}
_ = registry.Workflows()

func HandlersWith 

func HandlersWith(codec Codec, defs ...Definition) (*Registry, error)

HandlersWith is Handlers with a custom Codec. Use the same codec for the Client. Nil selects JSON. Codec implementations stay shared, while registrations snapshot the validated name, function, and policy of each definition. 

registry, err := workflow.HandlersWith(workflow.JSON{}, Onboard, SendWelcome)
if err != nil {
	return err
}
_ = registry.Workflows()

func NewRegistry 

func NewRegistry(definitions []Definition, configure ...RegistryBuilderFunc) (*Registry, error)

NewRegistry validates definitions and publishes an immutable worker registry. It has the same validation and handler-snapshot semantics as HandlersWith. Configure worker telemetry here even when clients run in a different pod. Without callbacks JSON and the dynamic process telemetry default apply. 

registry, err := workflow.NewRegistry([]workflow.Definition{Onboard, SendWelcome},
	func(worker *workflow.RegistryBuilder) {
		worker.Telemetry(observations)
	})
if err != nil {
	return err
}
_ = registry.Workflows()

func (*Registry) NewTraceContext 

func (registry *Registry) NewTraceContext(ctx context.Context, carrier map[string]string) context.Context

NewTraceContext extracts a serialized start carrier onto a native attempt context, preserving its cancellation/deadline. Drivers call this only at the worker-to-step boundary; extraction must never run in replayed orchestration. The registry's worker telemetry service performs extraction. Nil/empty carriers are valid; no caller context is transported or retained. 

ctx = registry.NewTraceContext(ctx, options.Trace)
output, err := handler.Handle(ctx, input)
if err != nil {
	return err
}
_ = output

func (*Registry) Step 

func (registry *Registry) Step(name string) (StepHandler, bool)

Step returns the handler of the step called name. The boolean is false for an absent name. Drivers use this lookup for actual attempts, with per-call derived policy supplied separately by Runtime.Step. 

handler, ok := registry.Step("charge")
if !ok {
	return workflow.ErrNotFound
}
_ = handler.Options.Timeout

func (*Registry) Steps 

func (registry *Registry) Steps() collection.Slice[StepHandler]

Steps returns the step handlers sorted by name. The slice and records are copies; the registered typed functions are shared. This eager snapshot is not a live query; wrapping it adds collection operations without acquiring ownership of functions or their captures. 

retried := registry.Steps().Filter(func(_ int, handler workflow.StepHandler) bool {
	return handler.Options.Attempts > 1
})
for _, handler := range retried {
	fmt.Println(handler.Name, handler.Options.Attempts)
}

func (*Registry) Workflow 

func (registry *Registry) Workflow(name string) (WorkflowHandler, bool)

Workflow returns the handler of the workflow called name. The boolean is false for an absent name. Editing the returned record cannot replace the registry's snapshot or its scalar policy. 

handler, ok := registry.Workflow("onboard")
if !ok {
	return workflow.ErrNotFound
}
_ = handler.Options.MaxDuration

func (*Registry) Workflows 

func (registry *Registry) Workflows() collection.Slice[WorkflowHandler]

Workflows returns the workflow handlers sorted by name. The returned slice is independent of the registry; changing its records cannot replace registered handlers. Handler functions remain shared. This eager snapshot is not a live query; collection operations do not clone functions, their captures, or services used by those functions. 

short := registry.Workflows().Filter(func(_ int, handler workflow.WorkflowHandler) bool {
	return handler.Options.MaxDuration <= time.Hour
})
for _, handler := range short {
	fmt.Println(handler.Name, handler.Options.MaxDuration)
}

type RegistryBuilder 

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

RegistryBuilder configures worker registration independently of client pods. Its final state is copied before handlers are published; services stay shared. 

registry, err := workflow.NewRegistry(definitions, func(worker *workflow.RegistryBuilder) {
	worker.Telemetry(observations)
})
if err != nil {
	return err
}
_ = registry.Steps()

func (*RegistryBuilder) Codec 

func (worker *RegistryBuilder) Codec(value Codec) *RegistryBuilder

Codec replaces the worker payload codec; nil selects JSON. The codec is shared, not cloned, and must match the client. It returns the same draft. 

registry, err := workflow.NewRegistry(definitions, func(worker *workflow.RegistryBuilder) {
	worker.Codec(workflow.JSON{})
})
if err != nil {
	return err
}
_ = registry.Steps()

func (*RegistryBuilder) Telemetry 

func (worker *RegistryBuilder) Telemetry(value *telemetry.Telemetry) *RegistryBuilder

Telemetry selects worker attempt observations and carrier extraction. Nil follows the current process default at each attempt, independently of the client's choice. It does not observe orchestration or terminal run outcomes. 

registry, err := workflow.NewRegistry(definitions, func(worker *workflow.RegistryBuilder) {
	worker.Telemetry(observations)
})
if err != nil {
	return err
}
_ = registry.Steps()

type RegistryBuilderFunc 

type RegistryBuilderFunc = func(*RegistryBuilder)

RegistryBuilderFunc configures one fresh worker-registration draft. Callbacks run once in order, including for invalid definitions; nil callbacks panic. 

var configure workflow.RegistryBuilderFunc = func(worker *workflow.RegistryBuilder) {
	worker.Codec(workflow.JSON{})
}
registry, err := workflow.NewRegistry(definitions, configure)
if err != nil {
	return err
}
_ = registry

type Run 

type Run[Out any] struct {
	// contains filtered or unexported fields
}

Run is a handle to one run of a workflow that produces Out. Get it from Client.Start or Client.Run. A handle stores identity and shared client dependencies, not execution state; each observation queries the driver and can fail if the run no longer exists. 

run := client.Run(Onboard, "user-42")
output, err := run.Result(ctx)
if err != nil {
	return err
}
_ = output

func (*Run[Out]) Cancel 

func (run *Run[Out]) Cancel(ctx context.Context) (err error)

Cancel stops the run. Canceling a finished run is not an error. Telemetry observes the cancel call, not confirmation of engine termination. The existing RunCanceled topic reports the successful call only. 

if err := run.Cancel(ctx); err != nil {
	return err
}

func (*Run[Out]) ID 

func (run *Run[Out]) ID() string

ID returns the run identifier. This is the exact caller-supplied identity, not a new engine execution ID. 

run := client.Run(Onboard, "user-42")
fmt.Println(run.ID()) // user-42

func (*Run[Out]) Result 

func (run *Run[Out]) Result(ctx context.Context) (_ Out, err error)

Result waits for the run to end and returns its output, or the failure of the run: ErrCanceled after a cancel, ErrTimeout after Builder.MaxDuration, or the error of the workflow. ctx bounds only the wait; when it ends Result returns its error and the run continues. Telemetry measures this wait, including decoding, not workflow execution. Repeated waits are separate observations and never emit terminal events. 

output, err := run.Result(ctx)
if err != nil {
	return err
}
_ = output

func (*Run[Out]) Status 

func (run *Run[Out]) Status(ctx context.Context) (_ Status, err error)

Status returns the state of the run. Telemetry measures the query only; it never emits a terminal event. 

status, err := run.Status(ctx)
if err != nil {
	return err
}
_ = status == workflow.Completed

type RunEvent 

type RunEvent struct {
	// Workflow is the workflow name.
	Workflow string

	// RunID is the run identifier.
	RunID string
}

RunEvent is the payload of RunStarted and RunCanceled. It carries no input, output, or metadata of the run. RunID is application-supplied operational data: restrict subscriptions when those identities are sensitive. These events describe local accepted actions. 

unsubscribe, err := bus.Listen(workflow.RunStarted, func(_ context.Context, notice workflow.RunEvent) error {
	fmt.Println(notice.Workflow)
	return nil
})
if err != nil {
	return err
}
defer unsubscribe()

type RunSignaledEvent 

type RunSignaledEvent struct {
	// Workflow is the workflow name.
	Workflow string

	// RunID is the run identifier.
	RunID string

	// Signal is the signal name.
	Signal string
}

RunSignaledEvent is the payload of RunSignaled. It omits the signal value. The event reports accepted delivery, not workflow consumption or completion. 

unsubscribe, err := bus.Listen(workflow.RunSignaled, func(_ context.Context, notice workflow.RunSignaledEvent) error {
	fmt.Println(notice.Signal)
	return nil
})
if err != nil {
	return err
}
defer unsubscribe()

type Runtime 

type Runtime interface {
	// ID returns the run identifier.
	ID() string

	// Name returns the workflow name.
	Name() string

	// Now returns the current time of the run.
	Now() time.Time

	// UUID returns a new random identifier.
	UUID() string

	// Logger returns the logger of the run.
	Logger() *slog.Logger

	// Sleep pauses the run for d.
	Sleep(d time.Duration) error

	// Done is closed when the run is canceled or timed out.
	Done() <-chan struct{}

	// Err returns why Done is closed, or nil.
	Err() error

	// Step runs the registered step called name with completed per-call options o,
	// not the registered handler's default policy. Initial zero means no deliberate
	// positive backoff. An engine unable to express Attempts > 1 with Initial == 0
	// returns ErrUnsupported before scheduling or executing the affected step;
	// it must not substitute a positive delay or retry the whole workflow instead.
	// This policy rejection does not disable single attempts or positive backoff.
	//
	//	policy := workflow.StepOptions{Attempts: 3, Initial: 0, Factor: 1}
	//	output, err := runtime.Step("charge", input, policy)
	//	if err != nil {
	//		return err
	//	}
	//	_ = output
	Step(name string, in []byte, o StepOptions) ([]byte, error)

	// Child atomically creates or joins run id of workflow name and waits for
	// its result. Only creation acquires cancellation ownership: cancellation
	// or timeout of the caller also stops a newly created child. Joining ignores
	// in and never acquires ownership, including when another parent created
	// the existing run. Cancellation or timeout ends the joining wait without
	// stopping that run. Replay must retain the original ownership decision;
	// a lookup followed by a separate create is not sufficient. A driver unable
	// to guarantee this returns [ErrUnsupported] before submission, not a
	// detached child or a join that can cancel another owner's run.
	//
	//	output, err := runtime.Child("fulfill", "order-42", input)
	//	if err != nil {
	//		return err
	//	}
	//	_ = output // An existing run keeps its original input and owner.
	Child(name, id string, in []byte) ([]byte, error)

	// Spawn starts run id of workflow name without waiting; the run outlives
	// the caller. An existing run is not an error: [Context.Spawn] treats
	// [ErrAlreadyStarted] as success so a replay is idempotent. A driver that
	// cannot do it returns [ErrUnsupported].
	Spawn(name, id string, in []byte) error

	// Wait blocks for the next value of signal. A zero timeout waits without
	// limit. It returns [ErrTimeout] when the timeout expires.
	Wait(signal string, timeout time.Duration) ([]byte, error)

	// Select blocks until one case fires and returns its index and, for a
	// signal case, the value.
	Select(cases []RawCase) (index int, value []byte, err error)

	// Raw returns the native context of the engine, such as a
	// workflow.Context from Temporal.
	Raw() any
}

Runtime is the execution environment a driver gives to a workflow function. A Runtime may also implement Repanic(v any) bool; the workflow handler raises a recovered panic again when it returns true, so engine control-flow panics are not turned into failures. Every call that blocks returns early with an error when the run is canceled or timed out. A driver that replays workflows must make these calls deterministic and durable, including the ownership decision of Child. Engine implementations receive one from their own execution adapter; an application uses Context rather than constructing a native runtime. 

handler, ok := registry.Workflow("onboard")
if !ok {
	return workflow.ErrNotFound
}
out, err := handler.Handle(runtime, input) // runtime is supplied by the engine.
if err != nil {
	return err
}
_ = out

type Signal 

type Signal[T any] struct {
	// contains filtered or unexported fields
}

Signal names an external message of type T that a running workflow waits for. Create it with NewSignal; send it with Client.Signal. Definitions bind a shared name to the same payload type on both sides of the wire; using the same name with incompatible types is an application error. 

approved := workflow.NewSignal[bool]("approved")
if err := client.Signal(ctx, Onboard, "user-42", approved, true); err != nil {
	return err
}

func NewSignal 

func NewSignal[T any](identity string) Signal[T]

NewSignal returns the signal called identity that carries values of type T. The name must satisfy the shared identifier rule of 1-64 lowercase letters, digits, or hyphens with no hyphen at either end; NewSignal panics with ErrInvalidName otherwise. The constructor does no I/O; use a shared definition for senders and waiters. 

approved := workflow.NewSignal[bool]("approved")
value, err := workflowContext.Wait(approved, workflow.Timeout(time.Minute))
if err != nil {
	return "", err
}
_ = value

func (Signal[T]) Name 

func (signal Signal[T]) Name() string

Name returns the name of the signal. The name is the stable identity used by both senders and receivers.

Example: 

confirmed := workflow.NewSignal[string]("confirmed")
name := confirmed.Name()

type StartOption 

type StartOption func(*StartOptions)

StartOption changes one StartOptions setting of Client.Start. Options run in call order before trace injection and do not reconfigure the client. Custom options must not retain the per-call settings pointer. 

var option workflow.StartOption = workflow.Delay(time.Minute)
run, err := client.Start(ctx, Onboard, "user-42", user, option)
if err != nil {
	return err
}
_ = run.ID()

func Delay 

func Delay(d time.Duration) StartOption

Delay postpones the start of the run by d. The driver owns scheduling; this does not sleep in Client.Start. 

run, err := client.Start(ctx, Onboard, "user-42", user, workflow.Delay(time.Minute))
if err != nil {
	return err
}
_ = run.ID()

func Metadata 

func Metadata(data map[string]string) StartOption

Metadata attaches free-form data to the run. Applying the option copies the outer map. Keep sensitive data out of metadata that the selected engine exposes in its operational interfaces. 

run, err := client.Start(ctx, Onboard, "user-42", user,
	workflow.Metadata(map[string]string{"source": "signup"}))
if err != nil {
	return err
}
_ = run.ID()

type StartOptions 

type StartOptions struct {
	// Trace carries injected trace fields separately from application metadata.
	// Drivers copy retained maps and forward fields without invoking telemetry
	// in replayed orchestration. Nil means no propagated parent. This is not an
	// authorization boundary; only trusted submitters should reach workers.
	//
	//	options := workflow.StartOptions{Trace: map[string]string{"traceparent": parent}}
	Trace map[string]string
	// Delay postpones the start of the run.
	Delay time.Duration

	// Metadata is free-form data kept with the run.
	Metadata map[string]string
}

StartOptions are the settings of one Driver.Start. Trace is transport propagation state, not application metadata; Client.Start supplies it after applying start options. Drivers retain copies rather than caller contexts. 

options := workflow.StartOptions{Metadata: map[string]string{"source": "signup"}}
err := driver.Start(ctx, "onboard", "user-42", input, options)
if err != nil && !errors.Is(err, workflow.ErrAlreadyStarted) {
	return err
}

type Status 

type Status string

Status is the persisted, lowercase state of a run. Its empty zero value is invalid and does not identify a running workflow.

Example: _ = workflow.Status("running") == workflow.Running 

const (
	// Running means the run has not ended.
	//
	// Example: _ = workflow.Status("running") == workflow.Running
	Running Status = "running"

	// Completed means the run returned an output.
	//
	// Example: _ = workflow.Completed.String() // "completed"
	Completed Status = "completed"

	// Failed means the run returned an error or timed out.
	//
	// Example: _ = workflow.Failed.String() // "failed"
	Failed Status = "failed"

	// Canceled means the run was canceled.
	//
	// Example: _ = workflow.Canceled.String() // "canceled"
	Canceled Status = "canceled"
)

The states of a run.

func (Status) String 

func (status Status) String() string

String returns the status's persisted value, including empty and unknown values. It does not validate a value received from storage.

Example: _ = workflow.Completed.String() // "completed"

type Step 

type Step[In, Out any] struct {
	// contains filtered or unexported fields
}

Step is a named, typed unit of side-effecting work that a workflow runs through Context.Step. A driver may retry and record it. Create it with NewStep and configure it through StepBuilder.

A Step is an immutable definition: Step.Derive returns a modified copy that keeps the name, and the first error found is kept and reported by Step.Err. Handlers and Context.Step reject a step whose Err is not nil. To change the settings of one call, derive a step and pass that instead. Always create steps with NewStep; the zero value has no name and no function.

Example: 

var Charge = workflow.NewStep("charge-card", charge, func(step *workflow.StepBuilder) {
	step.Retry(5)
	step.Backoff(time.Second, 2, time.Minute)
	step.Timeout(30 * time.Second)
})

// Inside a workflow, for this call only:
singleAttempt := Charge.Derive(func(step *workflow.StepBuilder) {
	step.Retry(1)
})
receipt, err := workflowContext.Step(singleAttempt, order)

func NewStep 

func NewStep[In, Out any](identity string, fn func(context.Context, In) (Out, error), configure ...StepBuilderFunc) *Step[In, Out]

NewStep returns the step called identity that runs fn. The name must satisfy the shared identifier rule of 1-64 lowercase letters, digits, or hyphens with no hyphen at either end, and fn must not be nil. Violations are kept on the value and reported by Step.Err, Handlers, and Context.Step as ErrInvalidName or ErrNilFunc. Steps run once by default with a one minute timeout. Return Permanent from fn to stop retries. The default retry backoff starts at one second, multiplies by two, and caps at 30 seconds; it is used only when the configured attempts allow retries. Callbacks run directly once, synchronously in order on one fresh, nonnil draft, including invalid declarations. Nil entries and callback panics propagate; expanded nil or empty slices supply no edits. Publication copies the settings so a retained draft cannot change the step.

Example: 

charge := workflow.NewStep("charge", chargeCard, func(step *workflow.StepBuilder) {
	step.Retry(3)
})
registry, err := workflow.Handlers(charge)
if err != nil {
	return err
}
_ = registry

func (*Step[In, Out]) Derive 

func (step *Step[In, Out]) Derive(configure ...StepBuilderFunc) *Step[In, Out]

Derive returns a detached step with the same name, function, policy, and first error, followed by the supplied edits. Callbacks run directly once, synchronously in order on one fresh draft. Nil entries panic; expanded nil or empty slices supply no edits. The base is never mutated, and defaults are not reapplied. Use the result for one Context.Step call. The base is captured before callbacks run, including its typed function.

Example: 

singleAttempt := charge.Derive(func(step *workflow.StepBuilder) {
	step.Retry(1)
})
receipt, err := workflowContext.Step(singleAttempt, order)

func (*Step[In, Out]) Err 

func (step *Step[In, Out]) Err() error

Err returns the first declaration error, or nil for a valid step. Derive retains this error; later settings cannot repair an invalid declaration. The zero Step reports ErrInvalidName because it has no declaration identity.

Example: 

step := workflow.NewStep("Bad", charge)
if errors.Is(step.Err(), workflow.ErrInvalidName) {
	return step.Err()
}

func (*Step[In, Out]) Name 

func (step *Step[In, Out]) Name() string

Name returns the stable identifier supplied to NewStep. Deriving a policy preserves this name so the same registered handler serves the derived step.

Example: 

name := workflow.NewStep("charge", charge).Name()
fmt.Println(name) // charge

type StepBuilder 

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

StepBuilder configures the retry policy of a step before NewStep or Step.Derive publishes it. Each setting returns the same draft; publication copies its scalar settings so a retained draft cannot change the step.

Example: 

workflow.NewStep("charge", charge, func(step *workflow.StepBuilder) {
	step.Retry(3)
})

func (*StepBuilder) Backoff 

func (step *StepBuilder) Backoff(initial time.Duration, factor float64, maximum time.Duration) *StepBuilder

Backoff sets the delay before the first retry, the factor that multiplies it after each retry, and the cap on the delay. A zero maximum means no cap. A factor below one counts as one (constant delay). A negative initial or maximum is an error. Initial zero requests no deliberate positive delay, not an engine default. Engines unable to express it with multiple attempts return ErrUnsupported before scheduling or executing that call; the declaration itself remains valid. Defaults are an initial delay of one second, factor two, and a 30-second cap. Backoff returns the same draft and preserves the first error. NaN and infinite factors are normalized to one, like a factor below one, so the published policy remains usable by native engines and the memory driver.

Example: 

workflow.NewStep("charge", charge, func(step *workflow.StepBuilder) {
	step.Retry(3)
	step.Backoff(time.Second, 2, time.Minute)
})

func (*StepBuilder) Retry 

func (step *StepBuilder) Retry(attempts int) *StepBuilder

Retry sets the maximum number of executions of a step. One means no retry, which is the default. A smaller value counts as one. Retry returns the same draft and does not clear earlier errors.

Example: 

workflow.NewStep("charge", charge, func(step *workflow.StepBuilder) {
	step.Retry(3)
})

func (*StepBuilder) Timeout 

func (step *StepBuilder) Timeout(duration time.Duration) *StepBuilder

Timeout bounds one execution of a step. The default is one minute. Zero means no limit and a negative value is an error. Timeout returns the same draft and preserves the first declaration error.

Example: 

workflow.NewStep("charge", charge, func(step *workflow.StepBuilder) {
	step.Timeout(30 * time.Second)
})

type StepBuilderFunc 

type StepBuilderFunc = func(*StepBuilder)

StepBuilderFunc edits a step draft during NewStep or Step.Derive. Callbacks run synchronously once in order on one independent draft; supplied nil callbacks panic. Edits preserve an inherited declaration error.

Example: 

policy := workflow.StepBuilderFunc(func(step *workflow.StepBuilder) {
	step.Retry(3)
})
charge := workflow.NewStep("charge", chargeCard, policy)
if err := charge.Err(); err != nil {
	return err
}

type StepHandler 

type StepHandler struct {
	// Name is the step name.
	Name string

	// Options are the default settings of the step.
	Options StepOptions

	// Handle decodes in, runs the function, and returns the encoded output.
	// An error that satisfies [ErrPermanent] must not be retried.
	// Registered handlers observe each actual invocation with the registry's
	// telemetry service. Drivers must not invoke this callback for replayed
	// results; extract the carrier with Registry.NewTraceContext beforehand.
	//
	//	ctx = registry.NewTraceContext(ctx, options.Trace)
	//	out, err := handler.Handle(ctx, input)
	//	if err != nil {
	//		return err
	//	}
	//	_ = out
	Handle func(ctx context.Context, in []byte) ([]byte, error)
}

StepHandler runs one step definition on bytes. Call Handle only for actual attempts, not for replayed recorded results. Its context carries cancellation and worker trace-parent extraction. 

handler, ok := registry.Step("charge")
if !ok {
	return workflow.ErrNotFound
}
ctx = registry.NewTraceContext(ctx, carrier)
output, err := handler.Handle(ctx, input)
if err != nil {
	return err
}
_ = output

type StepOptions 

type StepOptions struct {
	// Attempts is the maximum number of executions. One means no retry.
	Attempts int

	// Initial is the delay before the first retry. Zero means no deliberate
	// positive backoff, not an engine default. A driver unable to express this
	// with multiple attempts rejects the completed policy with ErrUnsupported
	// before affected step scheduling or execution; see Runtime.Step.
	//
	//	policy := workflow.StepOptions{Attempts: 3, Initial: 0, Factor: 1}
	//	fmt.Println(policy.Backoff(1)) // 0s
	Initial time.Duration

	// Factor multiplies the delay after every retry.
	Factor float64

	// Max caps the delay between retries. Zero means no cap.
	Max time.Duration

	// Timeout bounds one execution.
	Timeout time.Duration
}

StepOptions are the settings of one step execution. NewStep publishes completed defaults. Raw adapters retain finite factors; Backoff normalizes nonfinite factors to one before calculating a delay. 

handler, ok := registry.Step("charge")
if !ok {
	return workflow.ErrNotFound
}
delay := handler.Options.Backoff(1)
_ = delay

func (StepOptions) Backoff 

func (options StepOptions) Backoff(n int) time.Duration

Backoff returns the delay before retry n, counting from 1. It never overflows: the result is capped by Max, or by the largest duration.

Example: 

policy := workflow.StepOptions{
	Initial: time.Second,
	Factor:  2,
	Max:     time.Minute,
}
delay := policy.Backoff(2)

type WaitOption 

type WaitOption func(*waitConfig)

WaitOption changes one setting of Context.Wait. It affects only this wait and cannot change the workflow's MaxDuration. 

var option workflow.WaitOption = workflow.Timeout(time.Minute)
value, err := workflowContext.Wait(Approved, option)
if err != nil {
	return "", err
}
_ = value

func Timeout 

func Timeout(d time.Duration) WaitOption

Timeout limits Context.Wait. When it expires Wait returns ErrTimeout. Without it Wait blocks until the signal arrives or the run ends. Zero or a negative duration disables the per-wait timeout; neither disables cancellation or the workflow's MaxDuration. An already remembered signal value returns immediately without scheduling another timer. 

value, err := workflowContext.Wait(Approved, workflow.Timeout(time.Minute))
if errors.Is(err, workflow.ErrTimeout) {
	return "expired", nil
}
if err != nil {
	return "", err
}
_ = value

type Workflow 

type Workflow[In, Out any] struct {
	// contains filtered or unexported fields
}

Workflow is a named, typed, durable function together with the run settings that are application code rather than transport data. Create it with New, configure it through Builder, pass it to Handlers on the worker, and start it with Client.Start.

A Workflow is an immutable published definition: callbacks edit a draft, and the first error found is kept and reported by Workflow.Err. Handlers rejects a workflow whose Err is not nil. Always create workflows with New; the zero value has no name and no function.

Example: 

var Onboard = workflow.New("onboard", onboard, func(definition *workflow.Builder) {
	definition.MaxDuration(24 * time.Hour)
	definition.Retention(7 * 24 * time.Hour)
	definition.Description("Welcomes a new user and waits for confirmation.")
})

func New 

func New[In, Out any](identity string, fn func(*Context, In) (Out, error), configure ...BuilderFunc) *Workflow[In, Out]

New returns the workflow called identity that runs fn. The name must satisfy the shared identifier rule of 1-64 lowercase letters, digits, or hyphens with no hyphen at either end, and fn must not be nil. Violations are kept on the value and reported by Workflow.Err and Handlers as ErrInvalidName or ErrNilFunc. fn must use the Context for time, randomness, and side effects, so a driver can replay it. Set Builder.MaxDuration before registering: it is required. Callbacks run directly once, synchronously in order on one fresh, nonnil draft, even when name or fn is invalid. Nil entries and callback panics propagate; expanded nil or empty slices supply no edits. Publication copies the completed settings; retained drafts cannot change the result.

Example: 

onboard := workflow.New("onboard", onboardUser, func(definition *workflow.Builder) {
	definition.MaxDuration(24 * time.Hour)
})
registry, err := workflow.Handlers(onboard)
if err != nil {
	return err
}
_ = registry

func (*Workflow[In, Out]) Err 

func (workflow *Workflow[In, Out]) Err() error

Err returns the first declaration error, or nil for valid construction. Missing MaxDuration is checked only by Handlers, not by Err. A zero Workflow reports ErrInvalidName before registration-only checks.

Example: 

definition := workflow.New("Bad", onboard)
if errors.Is(definition.Err(), workflow.ErrInvalidName) {
	return definition.Err()
}

func (*Workflow[In, Out]) Name 

func (workflow *Workflow[In, Out]) Name() string

Name returns the stable identifier supplied to New. Settings cannot change it; registration and client operations use this name across processes.

Example: 

name := workflow.New("onboard", onboard).Name()
fmt.Println(name) // onboard

type WorkflowHandler 

type WorkflowHandler struct {
	// Name is the workflow name.
	Name string

	// Options are the settings of the definition.
	Options WorkflowOptions

	// Handle decodes in, runs the function with a [Context] over rt, and
	// returns the encoded output. A failure is returned as the error of the
	// function; use [EncodeError] to carry it.
	Handle func(rt Runtime, in []byte) ([]byte, error)
}

WorkflowHandler runs one workflow definition on bytes. Registration snapshots the declaration; adapters must supply a replay-safe Runtime and preserve failures using EncodeError at process boundaries. 

handler, ok := registry.Workflow("onboard")
if !ok {
	return workflow.ErrNotFound
}
output, err := handler.Handle(runtime, input)
if err != nil {
	return err
}
_ = output

type WorkflowOptions 

type WorkflowOptions struct {
	// MaxDuration bounds the whole run. Required.
	MaxDuration time.Duration

	// Retention is how long the engine keeps a finished run. Zero uses the
	// engine default.
	Retention time.Duration

	// Description documents the workflow.
	Description string
}

WorkflowOptions are the settings of a workflow definition. Configure definitions through Builder; engine adapters read this completed reporting record from registered WorkflowHandler values. 

handler, ok := registry.Workflow("onboard")
if !ok {
	return workflow.ErrNotFound
}
limit := handler.Options.MaxDuration
_ = limit

Source Files

View all Source files

Directories

Path Synopsis
internal
envelope
Package envelope owns workflow payload framing and failure wire records.
 Package envelope owns workflow payload framing and failure wire records.
execution
Package execution protects registered workflow and step invocations.
 Package execution protects registered workflow and step invocations.
Package memoryworkflow provides an in-process workflow.Driver for local development and single-replica deployments.
 Package memoryworkflow provides an in-process workflow.Driver for local development and single-replica deployments.
internal/engine
Package engine owns in-process workflow runs, worker lifetime, and retention.
 Package engine owns in-process workflow runs, worker lifetime, and retention.
Package workflowtest helps test code that starts workflows and checks drivers of the workflow package.
 Package workflowtest helps test code that starts workflows and checks drivers of the workflow package.

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.