workflow

package module
v0.7.0 Latest Latest
Warning

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

 Go to latest Published: Nov 3, 2026 License: MIT Imports: 15 Imported by: 0

README

go-workflow

go-workflow is a durable workflow library for Go, inspired by the Workflow SDK. Workflows are Go functions whose progress is persisted step by step. A run can suspend on a timer or a hook without holding a process and resumes when it can continue. Runs can be inspected with the Workflow SDK web UI.

go get github.com/eriicafes/go-workflow

Read the docs at go-workflow.vercel.app.

Agent skills are available for writing workflows and implementing World backends.

Workflows

A workflow is a struct type that embeds workflow.Struct[Out] to declare its output Out. Its package path and type name are the workflow name, and a value of it is the input of a run. Keep workflows in a workflows package and their runners in a flows package inside it, workflows/flows, so request handlers can start workflows without importing their runners.

package workflows

type GenerateReport struct {
	workflow.Struct[GenerateReportResult]
	Topic string
}

type GenerateReportResult struct {
	Summary string
}

Runners and steps

A runner executes a workflow. It is a plain function that takes the workflow as input and calls steps, which do the real work and side effects. Keep each runner in a file with the steps it calls, in workflows/flows.

package flows

func GenerateReport(ctx context.Context, input workflows.GenerateReport) (workflows.GenerateReportResult, error) {
	sources, err := workflow.Step(ctx, findSources, input.Topic)
	if err != nil {
		return workflows.GenerateReportResult{}, err
	}
	return workflows.GenerateReportResult{Summary: strings.Join(sources, ", ")}, nil
}

func findSources(ctx context.Context, topic string) ([]string, error) {
	return search(ctx, topic)
}

Register runners with workflow.Use and collect them with worlds.Runners.

runners := worlds.Runners(
	workflow.Use(flows.GenerateReport),
)

Each step's result is persisted once it completes, so a run continues from its last completed step and never repeats finished work. Waiting on a step, a timer, or a hook suspends the run without consuming any resources until it can continue. For this to work, a runner does side effects only through steps and reads time only through workflow.Now, and the analyzer checks these rules.

If a deployed change makes a runner replay differently from what its run recorded, the runtime detects the divergence and fails the run with a worlds.ErrCorruptedEventLog instead of matching results to the wrong calls. Runs can also be pinned to the deployment that started them using DeploymentID, so in-flight runs keep replaying the code that recorded them, while workflows that embed workflow.StartOnLatest start new runs on the latest deployment.

Steps run at least once and retry on error, three times by default. Set workflow.StepOptions{MaxRetries: n} to change that, return workflow.FatalError(err) to fail without retrying, or return workflow.RetryableError(err, after) to retry after a delay. These wrappers only apply to errors returned by steps, inline steps, and rollbacks: a runner that returns an error fails its run without retrying, wrapped or not. Step functions should be idempotent, and workflow.StepID(ctx) is a stable idempotency key.

workflow.InlineStep(ctx, fn, input) takes the same arguments as Step but runs a short step, such as generating an ID, in the flow itself, without a separate step delivery. Failed attempts are retried like a step's.

Running workflows

Every run is recorded in a World. The local World stores runs as JSON files and delivers work in-process, so it needs no database or worker.

world, err := localworld.Open(localworld.Config{Dir: ".workflows"}, runners)
if err != nil {
	log.Fatal(err)
}
defer world.Close()
if err := world.Prepare(ctx); err != nil {
	log.Fatal(err)
}

ctx = workflow.NewContext(ctx, workflow.NewClient(world))
run, err := workflow.Start(ctx, workflows.GenerateReport{Topic: "workflows"})
if err != nil {
	log.Fatal(err)
}
result, err := run.Value(ctx)

workflow.Start needs only the workflow type and a Client in ctx. A Run handle reads and controls the run with Value, Status, Cancel, and WakeUp, and workflow.Get returns the handle of an existing run by ID.

Worlds

Package Storage and delivery Entry points
worlds/local JSON files, in-process delivery World.Prepare
worlds/postgres PostgreSQL, LISTEN/NOTIFY workers postgresworld.Worker
worlds/vercel Redis, Vercel Queues, optional QStash vercelworld.Queue
worlds/aws DynamoDB, SQS, optional EventBridge Scheduler awsworld.Queue

Every World package also exports HTTP(world), which serves the RPC endpoint that clients and the web UI use and the webhook endpoint under /v1/workflow. HTTPPrefix(world, prefix) serves them under another prefix, and RPC(world) and Webhook(world) serve each endpoint on its own. Set Config.BaseURL to the public URL of that prefix, since webhook URLs are built from it.

Postgres

Any number of application and worker processes share one database. Worker runs the loop that executes deliveries.

world, err := postgresworld.Open(ctx, postgresworld.Config{BaseURL: os.Getenv("WORKFLOW_BASE_URL"), URL: os.Getenv("DATABASE_URL")})
if err != nil {
	log.Fatal(err)
}
defer world.Close()

go func() {
	if err := postgresworld.Worker(ctx, world, runners); err != nil {
		log.Printf("workflow worker stopped: %v", err)
	}
}()
log.Fatal(http.ListenAndServe(":8081", postgresworld.HTTP(world)))
Vercel

Deploy Queue as a function with a Vercel Queues trigger for Config.Topic, and HTTP as a separate function for clients. Redis is reached over TCP with RedisURL or over the Upstash REST API with RedisRESTURL and RedisRESTToken. OIDCToken defaults to reading VERCEL_OIDC_TOKEN on every request.

world, err := vercelworld.New(vercelworld.Config{
	BaseURL:  os.Getenv("WORKFLOW_BASE_URL"),
	RedisURL: os.Getenv("REDIS_URL"),
	QueueURL: os.Getenv("VERCEL_QUEUE_URL"),
	Topic:    os.Getenv("VERCEL_WORKFLOW_QUEUE_TOPIC"),
})
if err != nil {
	log.Fatal(err)
}

// The queue consumer function.
log.Fatal(http.ListenAndServe(":8081", vercelworld.Queue(world, runners)))

// The client-facing function.
log.Fatal(http.ListenAndServe(":8081", vercelworld.HTTP(world)))

Vercel Queues delays deliveries by at most seven days. Set Config.QStash to deliver work due later.

AWS

Install Queue as the SQS handler of one Lambda function and HTTP behind a Function URL of another. The Terraform module creates the table, queues, scheduler group, and IAM policies, and awsworld.CreateTable creates the table for tests and DynamoDB Local.

world, err := awsworld.New(awsworld.Config{BaseURL: os.Getenv("WORKFLOW_BASE_URL"), TableName: os.Getenv("WORKFLOW_TABLE"), QueueURL: os.Getenv("WORKFLOW_QUEUE_URL")}, dynamodb.NewFromConfig(cfg), sqs.NewFromConfig(cfg))
if err != nil {
	log.Fatal(err)
}

// The queue consumer function.
lambda.Start(awsworld.Queue(world, runners))

// The client-facing function.
lambdaurl.Start(awsworld.HTTP(world))

SQS delays deliveries by at most 15 minutes. Set Config.Scheduler to deliver work due later through EventBridge Scheduler.

Retention

Runs are kept forever by default. A World that supports retention deletes a run's data Config.Retention after it completes, fails, or is cancelled, so storage stays bounded. Config.Retentions sets it per workflow, where zero keeps that workflow's runs forever:

config.Retention = 24 * time.Hour
config.Retentions = worlds.Retentions(
	worlds.Retain[workflows.Chat](10 * time.Minute),
	worlds.Retain[workflows.Invoice](0),
)

Runs still pending or running are never deleted, and a hook's token retention keeps its run until it ends. A few minutes is the shortest practical retention, since callers read a run's result and streams right after it finishes. Each World's README describes how and when it deletes.

Application and worker

An application that only starts and reads runs needs the workflow types and a Client, not the runners. NewHTTPClient talks to a World served by HTTP, at the server's URL followed by the handler's prefix, and Middleware puts the client in every request context.

client := workflow.NewHTTPClient("http://workflow-worker:8081/v1/workflow")

mux := http.NewServeMux()
mux.HandleFunc("POST /reports", func(w http.ResponseWriter, r *http.Request) {
	run, err := workflow.Start(r.Context(), workflows.GenerateReport{Topic: r.FormValue("topic")})
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	w.WriteHeader(http.StatusAccepted)
	w.Write([]byte(run.ID()))
})
log.Fatal(http.ListenAndServe(":8080", workflow.Middleware(mux, client)))

examples/basic has the full application with a worker for each World.

Concurrency and time

workflow.Background starts a step without waiting and returns a *Result[T]. Map starts one step per input. All collects outputs in input order and cancels the remaining steps if any fails. Join, JoinOk, and the Select functions wait on several promises at once.

results, err := workflow.All(ctx, findSources, []string{"go", "rust"})

MapResults converts a Map result slice for Join, Select, or group Add. WaitGroup waits for every outcome; ErrGroup waits for every outcome and returns the first failure. Use workflow.Go(ctx, &group, step, input) to start and track steps. workflow.ErrGroupWithContext(ctx) returns an ErrGroup and a durable context cancelled on failure or when its Wait returns.

workflow.Sleep suspends the run for a duration, Timer returns a promise that settles after one, and WithTimeout and WithDeadline return a cancellation context and cancel function, using a timer to cancel work that uses that context. A sleeping run consumes no resources, so it can sleep for days or months.

Errors

A failed step's error is persisted with the step and comes back as a *workflow.Error, and so does a failed run's error from Run.Value. It keeps the message, type, and exported fields of every error in the original chain, so errors.Is and errors.As work on it in the runner and across child runs.

var ErrCardDeclined = errors.New("card declined")

type LimitError struct{ Limit int }

func (e *LimitError) Error() string { return fmt.Sprintf("over limit %d", e.Limit) }

Inspect the recorded error in the runner:

_, err := workflow.Step(ctx, charge, order)
var limit *LimitError
switch {
case errors.Is(err, ErrCardDeclined):
	// ...
case errors.As(err, &limit):
	// limit.Limit is decoded from the recorded error.
}

Rollbacks

Defer a compensation after the work it reverses. If the run later fails, the compensations run as durable steps in Go's defer order, with the last defer first.

invoice, err := workflow.Step(ctx, createInvoice, input)
if err != nil {
	return Report{}, err
}
defer workflow.Rollback(ctx, voidInvoice, invoice.ID)

Streams

Steps write run-owned byte streams that readers consume while the run is still going. Writers and readers use the run's default stream unless their option names another. A stream has one writer at a time, and another writer waits until the current one releases or closes it.

func writeProgress(ctx context.Context, count int) (struct{}, error) {
	writer, release, err := workflow.Writer(ctx)
	if err != nil {
		return struct{}{}, err
	}
	defer release()
	_, err = fmt.Fprintf(writer, "found %d sources\n", count)
	return struct{}{}, err
}

Outside a run, such as in a route handler, run.Reader reads a stream. Each write is one chunk, and ReaderOption.StartIndex picks the first chunk, where a negative index counts from the end. Reads wait for new data until the stream closes.

// The last three writes onward.
reader, err := run.Reader(ctx, workflow.ReaderOption{StartIndex: -3})

Hooks and webhooks

A hook receives repeated typed payloads. Declare reusable hook definitions beside workflow structs in a shared package. Pass a token when the caller must know it in advance.

hook := workflow.CreateHook[Approval](ctx, workflow.HookOption{Token: "approve:" + input.OrderID})
defer hook.Dispose(ctx)

approval, err := hook.Next(ctx) // registers the hook and waits for a payload

Creation is local and returns only a handle. Receiving from or selecting that hook registers it; unrelated waits do not. Use GetConflict(ctx) before starting work that requires ownership or exposes the token; it registers without waiting for a payload and returns the conflicting run, or nil. Dispose propagates failures by panicking, including when deferred.

err := workflow.ResumeHook(ctx, "approve:"+orderID, Approval{Approved: true})

A webhook is a hook resumed by an HTTP request to its URL, with the JSON body decoded as the payload. WebhookOption can require a method, set a static response, or hold the request until a step calls WebhookRequest.RespondWith. The token in the URL is its only authorization, so verify provider signatures before trusting the payload.

webhook, err := workflow.CreateWebhook[Payment](ctx)
if err != nil {
	return Result{}, err
}
defer webhook.Dispose(ctx)

if _, err := workflow.Step(ctx, registerCallback, webhook.URL()); err != nil {
	return Result{}, err
}
request, err := webhook.Request(ctx)
if err != nil {
	return Result{}, err
}
payment, err := request.JSON()

WithCancel and WithCancelCause return a durable cancellation context, a cancel function, and a creation error. Their cancel functions also return errors. Pass cancellation contexts to steps; cancel runs explicitly with Run.Cancel; use CancelAny(ctx, sources...) to combine cancellation while preserving base values. Cancelled steps fail without retrying; cancelled runs let running steps finish naturally. Then follows the result returned by func(T, error) *Result[U]; use NewResult(value, err) for immediate values and failures.

Composing workflows

A runner can compose another workflow in two ways, as in the Workflow SDK. Calling the other workflow's runner directly runs its steps in the current run.

report, err := GenerateReport(ctx, workflows.GenerateReport{Topic: input.Topic})

workflow.Start in workflow context starts a child run, which carries the $parentRunId and $rootRunId attributes. In workflow context Run.Value returns workflow.ErrRunNotFinished instead of waiting, so to wait for a child, pass it a hook token and have it resume the hook when it finishes.

func Parent(ctx context.Context, input workflows.Parent) (workflows.ParentResult, error) {
	hook := workflow.CreateHook[string](ctx)
	defer hook.Dispose(ctx)

	if owner, err := hook.GetConflict(ctx); err != nil {
		return workflows.ParentResult{}, err
	} else if owner != nil {
		return workflows.ParentResult{}, workflow.ErrHookConflict
	}
	if _, err := workflow.Start(ctx, workflows.Child{Token: hook.Token()}); err != nil {
		return workflows.ParentResult{}, err
	}
	value, err := hook.Next(ctx)
	return workflows.ParentResult{Value: value}, err
}

func Child(ctx context.Context, input workflows.Child) (workflows.ChildResult, error) {
	value, err := workflow.Step(ctx, work, input)
	if err != nil {
		return workflows.ChildResult{}, err
	}
	_, err = workflow.Step(ctx, resumeParent, completion{Token: input.Token, Value: value})
	return workflows.ChildResult{Value: value}, err
}

func resumeParent(ctx context.Context, c completion) (struct{}, error) {
	return struct{}{}, workflow.ResumeHook(ctx, c.Token, c.Value)
}

AI agents

The workflowai package runs AI SDK style agents as durable runs: every model call and tool step is recorded, the response streams through the run's stream, and tool approvals wait for the user across turns. It matches @ai-sdk/workflow's WorkflowAgent and is built on the Go AI SDK. examples/durable-agent serves it to a React useChat client, and the AI Agents guide covers it in depth.

Development

The analyzer checks that every runner passed to Use is a named function whose input is a workflow, and reports goroutines, select, map iteration, time.Now, I/O outside steps, and steps that capture variables instead of taking them as input. Run it as a vet tool.

go install github.com/eriicafes/go-workflow/cmd/workflowcheck@latest
go vet -vettool=$(which workflowcheck) ./...

The workflowdev package integrates with go-dev. workflowdev.Check runs the analyzer before every start or reload, and workflowdev.Web runs the Workflow SDK web UI against a World served by HTTP. The web UI needs the Workflow SDK CLI and this repository's World adapter for it, installed in the directory go-dev runs from.

npm install --save-dev workflow go-workflow

session := dev.New()
task := session.NewTask(dev.Cmd{
	Run:        dev.Package("./cmd/app"),
	Watch:      dev.Values("cmd", "workflows"),
	ServerAddr: ":8080",
	Commands: dev.Commands(
		dev.Cmd{Run: dev.Package("./cmd/worker/pg"), Prepare: dev.Hooks(workflowdev.Check("./...")), Watch: dev.Values("cmd", "workflows"), Phase: dev.Before},
		workflowdev.Web("http://localhost:8081/v1/workflow"),
	),
})
session.Catch(task.Run())
session.Catch(session.Wait())

Documentation

Overview

Package workflow runs durable workflows whose runners are ordinary Go functions.

Workflows and runners

A workflow is a struct type that embeds Struct[Out], or has its own Workflow(Out) method, to declare the workflow's output. Its package path and type name are the workflow name, and a value of it is the input of a run, so code can Start a workflow from the type alone.

A runner executes the workflow. It is a function func(context.Context, In) (Out, error) registered with Use, and it calls steps through Step, Background, Map, and All. Steps do the real work and side effects.

Runs

Start begins a run of a workflow and returns a Run handle. Every run is recorded in a World, the backend that stores its history, queues its deliveries, and holds its streams.

A run executes in one of two contexts. In workflow context, the runner's own code executes as a flow, which replays the runner from the beginning, records the operations it starts (steps, timers, hooks, webhooks, and child runs), queues the steps it needs, and records the run's result once the runner returns. In step context, a step function executes as a step, which replays the runner up to that step and calls the step function.

Replay

Operations return promises. Once an operation is recorded, replay returns its recorded result instead of running it again, and awaiting a promise that has not settled suspends the run until it does.

A runner must therefore be deterministic. It does side effects only through steps and reads time only through Now. Steps may retry and are executed at least once, so step functions should be idempotent. Package check reports runners that break these rules.

Clients

Code outside a run uses a Client, carried in a context by NewContext or Middleware, to start, read, cancel, and resume runs.

Index

Constants

View Source 
const (
	// StatusPending means the run or step is recorded but has not started.
	StatusPending = worlds.StatusPending
	// StatusRunning means the run or step has started and has not finished.
	StatusRunning = worlds.StatusRunning
	// StatusCompleted means the run or step finished with an output.
	StatusCompleted = worlds.StatusCompleted
	// StatusFailed means the run or step finished with an error.
	StatusFailed = worlds.StatusFailed
	// StatusCancelled means the run or step was cancelled before it finished.
	StatusCancelled = worlds.StatusCancelled
)

Variables

View Source 
var (
	// ErrNoClient reports that the context has no client.
	ErrNoClient = errors.New("workflow: context has no client")
	// ErrNoWorkflow reports that an operation requires workflow context.
	ErrNoWorkflow = errors.New("workflow: context has no active workflow")
	// ErrNilResult reports a nil result where an operation's result was expected.
	ErrNilResult = errors.New("workflow: nil result")
	// ErrNotInStep reports that an operation cannot be called in workflow context.
	ErrNotInStep = errors.New("workflow: must be called inside a step")
	// ErrRunCancelled is returned when awaiting a run that was cancelled.
	ErrRunCancelled = worlds.ErrRunCancelled
	// ErrResumeNotQueued reports that a hook payload was recorded but its run was not queued.
	ErrResumeNotQueued = worlds.ErrResumeNotQueued
	// ErrRunNotFinished reports that a run is still waiting or running.
	ErrRunNotFinished = errors.New("workflow: run has not finished")
	// ErrHookNotFound reports that no active hook owns the requested token.
	ErrHookNotFound = worlds.ErrHookNotFound
	// ErrHookNotActive reports that a hook can no longer receive payloads.
	ErrHookNotActive = worlds.ErrHookNotActive
	// ErrHookNotRegistered reports that a hook does not belong to the current
	// workflow execution.
	ErrHookNotRegistered = errors.New("workflow: hook is not registered")
	// ErrHookConflict is returned when claiming a hook whose token another
	// active hook already has. Hook.GetConflict reports the run that owns it.
	ErrHookConflict = worlds.ErrHookConflict
	// ErrHookForceClaimed reports that another hook took over this hook's token.
	ErrHookForceClaimed = worlds.ErrHookForceClaimed
)

Functions

func All 

func All[In, Out any](ctx context.Context, step func(context.Context, In) (Out, error), inputs []In) ([]Out, error)

All starts every step and collects successful values in input order. If any step fails, it cancels the others and returns that failure.

func CancelAny 

func CancelAny(ctx context.Context, sources ...context.Context) context.Context

CancelAny retains ctx's values and cancels when ctx or a source cancels. Outside workflow context, it returns ctx unchanged.

func CancelFunc 

func CancelFunc(ctx context.Context) (func() error, bool)

CancelFunc returns the controller carried by a managed step context.

func FatalError 

func FatalError(cause error) error

FatalError wraps cause so that a step returning it fails without retrying. It returns nil when cause is nil.

func InlineStep 

func InlineStep[In, Out any](ctx context.Context, fn func(context.Context, In) (Out, error), input In, options ...StepOptions) (Out, error)

InlineStep calls the step fn with input as an inline step and waits for it. It is InlineBackground followed by Value. It behaves like Step, except that the flow always runs it, and its retries, itself, never as a step delivery. It returns the step's final error as an *Error, and ErrNoWorkflow when ctx does not belong to a run.

Use InlineStep for short work that rarely fails, such as generating an ID or reading another run, and Step for anything else. fn is executed at least once, so it should be idempotent or free of side effects.

func Join 

func Join(ctx context.Context, results ...promise) error

Join waits for every input to settle, ignoring operation failures.

func JoinOk 

func JoinOk(ctx context.Context, results ...promise) error

JoinOk waits for every input to succeed and returns the first recorded failure. It does not cancel the remaining inputs when one fails.

func MapResults 

func MapResults[T any](results []*Result[T]) []promise

MapResults converts results to promise inputs, preserving their order.

func Middleware 

func Middleware(handler http.Handler, client Client) http.Handler

Middleware returns an http.Handler that serves each request with handler, adding client to the request context as NewContext does.

func NewContext 

func NewContext(ctx context.Context, client Client) context.Context

NewContext returns a copy of ctx that carries client. Start, Get, GetHook, ResumeHook, and Run methods use it, and in step context it takes precedence over the World that records the run.

func NewHookToken 

func NewHookToken() string

NewHookToken returns a random 21-character hook token. DefineHook generates replay-stable tokens when HookOption.Token is empty.

func Now 

func Now(ctx context.Context) (time.Time, error)

Now returns the current time in UTC, recorded the first time it runs so that replay returns the same time. In workflow context, a runner reads time only through Now, never time.Now. Now returns ErrNoWorkflow outside workflow context, including step context, where time.Now is safe to use.

func ResumeHook 

func ResumeHook[T any](ctx context.Context, token string, value T) error

ResumeHook resumes the active hook with token, sending value as its payload encoded as JSON, and queues a delivery so its run resumes. It uses the Client in ctx and returns ErrHookNotFound when no active hook has the token. It is for steps and external callers; workflow context returns ErrNotInStep.

func ResumeHookWithRetries 

func ResumeHookWithRetries[T any](ctx context.Context, token string, value T, attempts int, interval time.Duration) error

ResumeHookWithRetries attempts ResumeHook up to attempts times, waiting interval between attempts while it returns ErrHookNotFound. The first attempt is immediate. Use it when a run may not have registered its hook yet, such as right after Start. It stops when ctx is done and rejects workflow context with ErrNotInStep.

func RetryableError 

func RetryableError(cause error, after time.Duration) error

RetryableError wraps cause so that a step returning it is retried no sooner than after, while StepOptions.MaxRetries allows another retry. It returns nil when cause is nil.

func Rollback 

func Rollback[In any](ctx context.Context, fn func(context.Context, In) error, input In, options ...StepOptions)

Rollback registers fn with input as a durable compensation for workflow failure. Defer it after the work it reverses. Compensations run in registration order and retry according to options. It panics with ErrNoWorkflow outside workflow context.

func Select 

func Select(ctx context.Context, results ...promise) (promise, error)

Select waits for the first recorded settlement and returns the original input.

func SelectOk 

func SelectOk(ctx context.Context, results ...promise) (promise, error)

SelectOk skips failed inputs until one succeeds or every input fails.

func SetMaxInlineSteps added in v0.3.0 

func SetMaxInlineSteps(ctx context.Context, n int) context.Context

SetMaxInlineSteps returns a copy of ctx that sets how many new steps a flow runs inline at once, instead of queueing a step delivery for each. The default is 3. Zero runs none inline, and other values are clamped to 1 through 16. Steps beyond the limit are queued as usual.

Set it on the ctx a World's handler executes deliveries with to set the limit for that worker, or in a runner for the steps it declares with the returned ctx.

func SetMaxInlineStepsDuration added in v0.5.1 

func SetMaxInlineStepsDuration(ctx context.Context, d time.Duration) context.Context

SetMaxInlineStepsDuration returns a copy of ctx that sets how long after a flow delivery starts it may start new steps inline. The default is 2 minutes, and zero starts none inline. After it, new steps are queued, so a long-running flow hands them to other deliveries.

A flow also stops starting steps inline halfway to the deadline of the ctx it executes with, so the steps it started can finish before it.

Set it like SetMaxInlineSteps: on the ctx a World's handler executes deliveries with, or in a runner for the steps it declares with the returned ctx.

func Sleep 

func Sleep(ctx context.Context, duration time.Duration) error

Sleep suspends the run for duration. Use it instead of time.Sleep in a workflow. It is Timer followed by Value.

func Step 

func Step[In, Out any](ctx context.Context, fn func(context.Context, In) (Out, error), input In, options ...StepOptions) (Out, error)

Step calls the step fn with input from a workflow and returns its result, suspending the run until the step has settled. Step is Background followed by Value. It returns ErrNoWorkflow when ctx does not belong to a run.

When the step fails, Step returns the step's error as an *Error, which errors.Is and errors.As match against the error the step returned.

func StepID 

func StepID(ctx context.Context) string

StepID returns the ID of the step ctx belongs to, or "" outside step context. It is shorthand for GetStepMetadata(ctx).ID.

func Use 

func Use[In worlds.Workflow[Out], Out any](fn func(context.Context, In) (Out, error)) worlds.Runner

Use registers fn as the runner of workflow In. Pass the returned Runner to worlds.Runners so the runtime can execute the workflow. The workflow name is the package path and type name of In. fn replays on every delivery, so it must be deterministic. It does side effects only through steps and reads time only through Now.

func WithCancel 

func WithCancel(ctx context.Context) (context.Context, func() error, error)

WithCancel creates a durable cancellation context in workflow context. It returns ErrNoWorkflow outside workflow context.

func WithCancelCause 

func WithCancelCause(ctx context.Context) (context.Context, func(error) error, error)

WithCancelCause creates a durable cancellation context whose first cancellation cause wins. It returns ErrNoWorkflow outside workflow context.

func WithDeadline 

func WithDeadline(ctx context.Context, deadline time.Time) (context.Context, func() error, error)

WithDeadline creates a durable cancellation context that is cancelled at deadline. Expiry sets context.Cause to context.DeadlineExceeded; manual cancellation uses context.Canceled. Cancellation affects steps and timers using the context, and does not cancel runs. It returns ErrNoWorkflow outside workflow context.

func WithDeadlineCause 

func WithDeadlineCause(ctx context.Context, deadline time.Time, cause error) (context.Context, func() error, error)

WithDeadlineCause is WithDeadline with a custom expiry cause. A nil cause uses context.DeadlineExceeded. Manual cancellation uses context.Canceled and disarms the timer.

func WithTimeout 

func WithTimeout(ctx context.Context, duration time.Duration) (context.Context, func() error, error)

WithTimeout creates a durable cancellation context that is cancelled after duration. The cancel function cancels the context early and disarms its timer. It returns ErrNoWorkflow outside workflow context.

func WithTimeoutCause 

func WithTimeoutCause(ctx context.Context, duration time.Duration, cause error) (context.Context, func() error, error)

WithTimeoutCause is WithTimeout with a custom expiry cause. A nil cause uses context.DeadlineExceeded. Manual cancellation uses context.Canceled and disarms the timer.

func Writer 

func Writer(ctx context.Context, options ...WriterOption) (io.WriteCloser, func() error, error)

Writer returns a writer for a stream of the run ctx belongs to. It must be called in step context. It returns ErrNotInStep in workflow context and ErrNoWorkflow outside a run.

The writer holds the stream name, so Writer waits while another writer holds it. Close closes the stream for all writers, while the returned release function only gives up this writer's hold so a retry or another step can continue the stream.

Types

type Client 

type Client = worlds.Client

Client is the API for starting, reading, cancelling, and resuming runs, and for reading and writing their streams, from outside a run.

func NewClient 

func NewClient(world World) Client

NewClient returns a Client that calls world directly, in the same process.

func NewHTTPClient 

func NewHTTPClient(endpoint string, clients ...*http.Client) Client

NewHTTPClient returns a Client that calls a World served by worlds.HTTPHandler. endpoint is the absolute URL of the server followed by the handler's prefix, such as http://worker:8081/v1/workflow. It panics when endpoint is not an absolute HTTP URL. Requests use the last of clients, or http.DefaultClient when none is given.

type ErrGroup 

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

ErrGroup waits for all its members and returns their first recorded failure. Its zero value is ready to use without cancellation. ErrGroupWithContext enables cancellation. After completion, the outcome stays fixed and new work is ignored. Create a new ErrGroup to wait for more work.

func ErrGroupWithContext 

func ErrGroupWithContext(ctx context.Context) (*ErrGroup, context.Context, error)

ErrGroupWithContext returns an ErrGroup and a durable cancellation context. The group begins observing members when first awaited. Its context is cancelled on the first observed failure or when Wait returns. It returns ErrNoWorkflow outside workflow context.

func (*ErrGroup) Add 

func (g *ErrGroup) Add(members ...promise)

Add includes existing results, hooks, webhooks, or groups in the wait. It does not change their cancellation contexts and ignores new members after settlement.

func (*ErrGroup) Wait 

func (g *ErrGroup) Wait(ctx context.Context) error

Wait waits for every member, returns the first failure, and cancels the context returned by ErrGroupWithContext, including on success.

type Error 

type Error = worlds.Error

Error represents a recorded failure. It keeps the type, message, and exported fields of each error in the original chain, so errors.Is matches sentinel errors and errors.As decodes error types after replay.

type Hook 

type Hook[T any] struct {

	// ClaimedFrom identifies the previous owner of a force-claimed token.
	ClaimedFrom *worlds.HookForceClaim
	// contains filtered or unexported fields
}

Hook receives repeated typed payloads addressed by its token.

func CreateHook 

func CreateHook[T any](ctx context.Context, options ...HookOption) *Hook[T]

CreateHook creates an unregistered hook handle for payloads of type T. It is DefineHook followed by Create.

func GetHook 

func GetHook[T any](ctx context.Context, token string) (*Hook[T], bool, error)

GetHook looks up the active hook with token using the Client in ctx. It returns false when no active hook has the token. The returned Hook only reports identity and claim metadata, and it cannot be awaited. It is for steps and external callers; workflow context returns ErrNotInStep.

func (*Hook[T]) Dispose 

func (h *Hook[T]) Dispose(ctx context.Context)

Dispose durably deactivates the hook and releases its token. It may be deferred and panics on failure.

func (*Hook[T]) GetConflict 

func (h *Hook[T]) GetConflict(ctx context.Context) (*Run[any], error)

GetConflict registers the hook and attempts to claim its token, without waiting for a payload. It returns the conflicting owner, or nil on success. Repeated calls return the recorded result without attempting another claim.

func (*Hook[T]) ID 

func (h *Hook[T]) ID() string

ID returns the hook ID after registration.

func (*Hook[T]) Next 

func (h *Hook[T]) Next(ctx context.Context) (T, error)

Next waits for and consumes one payload. A disposed or displaced hook returns an error; payloads received before takeover are delivered first. It registers this hook on first use; a conflicting token returns ErrHookConflict.

func (*Hook[T]) Token 

func (h *Hook[T]) Token() string

Token returns the hook token.

func (*Hook[T]) Values 

func (h *Hook[T]) Values(ctx context.Context) iter.Seq2[T, error]

Values yields successive payloads, stopping after an error or a range break. Ending iteration does not dispose the hook.

type HookDefinition 

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

HookDefinition is a hook declared with DefineHook but not yet created in a run. Receiving or selecting registers it; GetConflict waits only for ownership.

func DefineHook 

func DefineHook[T any](options ...HookOption) HookDefinition[T]

DefineHook declares a hook that receives a payload of type T. The last of options applies. An empty token is generated when the handle is created.

func (HookDefinition[T]) Create 

func (d HookDefinition[T]) Create(ctx context.Context) *Hook[T]

Create creates a hook handle in workflow context without claiming its token. Receiving or selecting this hook registers it. Use GetConflict to claim ownership before starting work that depends on the token. Outside workflow context, the handle reports ErrNoWorkflow when used.

type HookOption 

type HookOption struct {
	// Token is the hook token callers pass to ResumeHook. When empty, a
	// token is generated at creation and kept across replays.
	Token string
	// MinRetention reserves the token until creation time plus this duration,
	// including after run completion. Explicit Dispose releases it immediately.
	MinRetention time.Duration
	// Force takes the token over from another run's active hook instead of
	// failing with ErrHookConflict. Awaiting the displaced hook returns
	// ErrHookForceClaimed.
	Force bool
}

HookOption holds the options to DefineHook and CreateHook.

type ReaderOption 

type ReaderOption struct {
	// Name is the stream name. The zero value names the run's default stream.
	Name string
	// StartIndex is the index of the chunk to start reading at. Each write is
	// one chunk. A negative StartIndex counts back from the end of the
	// stream, so -3 reads the last three chunks.
	StartIndex int
}

ReaderOption holds the options to Reader and Run.Reader.

type Result 

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

Result holds the outcome of a background step, timer, or continuation. Reading a pending result suspends the workflow until it settles.

func Background 

func Background[In, Out any](ctx context.Context, fn func(context.Context, In) (Out, error), input In, options ...StepOptions) *Result[Out]

Background starts the step fn with input from a workflow and returns a promise for its result without waiting for it. Outside workflow context, the promise fails with ErrNoWorkflow.

The step is identified by fn's function name and its position in the runner. It is retried on error according to options, where the last option applies and the default allows 3 retries. fn is executed at least once and should be idempotent.

func Go 

func Go[In, Out any](ctx context.Context, g group, step func(context.Context, In) (Out, error), input In, options ...StepOptions) *Result[Out]

Go starts a background step and adds its result to group. It returns nil without starting a step if the group has completed.

func InlineBackground added in v0.3.0 

func InlineBackground[In, Out any](ctx context.Context, fn func(context.Context, In) (Out, error), input In, options ...StepOptions) *Result[Out]

InlineBackground starts the step fn with input as an inline step and returns a promise for its result without waiting for it, like Background. When the runner waits, the flow runs it, at the same time as the other steps it runs inline, and the runner continues as each one finishes. It always runs inline, whatever SetMaxInlineSteps allows, and a retry runs in a later flow once it is due. Outside workflow context, the promise fails with ErrNoWorkflow.

func Map 

func Map[In, Out any](ctx context.Context, step func(context.Context, In) (Out, error), inputs []In) []*Result[Out]

Map starts one background step per input, returning results in input order.

func NewResult 

func NewResult[T any](value T, err error) *Result[T]

NewResult returns an already-settled result containing value and err.

func Then 

func Then[T, U any](ctx context.Context, source *Result[T], fn func(T, error) *Result[U]) *Result[U]

Then follows the result returned by fn after source settles. The callback runs in workflow context during replay. Return pending work without waiting inside the callback so other Select inputs can still win. A nil result from fn settles the chain with ErrNilResult.

func Timer 

func Timer(ctx context.Context, duration time.Duration) *Result[struct{}]

Timer returns a timer, which is a promise that settles after duration. Outside workflow context, the promise fails with ErrNoWorkflow.

func (*Result[T]) Unwrap 

func (p *Result[T]) Unwrap(ctx context.Context) T

Unwrap waits for the result and panics if it failed.

func (*Result[T]) Value 

func (p *Result[T]) Value(ctx context.Context) (T, error)

Value waits for the result and returns its value or operation error.

type Run 

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

Run is a handle to a run of a workflow whose output type is Out. Start and Get return one. A Run has an independent lifetime; use Cancel to cancel it.

A Run started in workflow context is a child run. In workflow context, every Run method reads or changes the run once in an inline step and replay returns the recorded result, so a method does not wait for the run to finish. Value returns ErrRunNotFinished for a run that had not finished when it was called. These inline steps do not inherit managed context cancellation. To wait for a child run, create a hook and have the child resume it.

func Get 

func Get[Out any](ctx context.Context, runID string) (*Run[Out], error)

Get returns a handle to the run with runID, using the Client in ctx. It does not check that the run exists (see Run.Exists). It returns ErrNoClient when ctx carries no Client and does not belong to a run.

func Start 

func Start[In worlds.Workflow[Out], Out any](ctx context.Context, input In) (*Run[Out], error)

Start starts a run of workflow In with input and returns a handle to the run without waiting for it. It needs only the workflow type, not its runner.

Called in workflow context, Start starts a child run in a step, so replay returns the same run. The child run carries worlds.ParentRunIDAttribute and worlds.RootRunIDAttribute. Cancellation of ctx does not cancel the child; use Run.Cancel explicitly. Called elsewhere, it uses the Client in ctx and returns ErrNoClient when ctx carries none.

func (*Run[Out]) Cancel 

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

Cancel cancels the run. Cancelling a run that has already finished is a no-op.

func (*Run[Out]) Exists 

func (r *Run[Out]) Exists(ctx context.Context) (bool, error)

Exists reports whether the run is recorded in its World.

func (*Run[Out]) ID 

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

ID returns the run ID.

func (*Run[Out]) Reader 

func (r *Run[Out]) Reader(ctx context.Context, options ...ReaderOption) (StreamReader, error)

Reader returns a reader for the run's stream from ReaderOption.StartIndex, for step context and for use outside a run, such as a route handler. Reads block until data is written and return io.EOF once the stream is closed. It returns ErrNotInStep in workflow context.

func (*Run[Out]) Status 

func (r *Run[Out]) Status(ctx context.Context) (Status, error)

Status returns the current status of the run. It returns worlds.ErrRunNotFound when the run does not exist.

func (*Run[Out]) Value 

func (r *Run[Out]) Value(ctx context.Context) (Out, error)

Value returns the output of the run once it has finished. It returns the run's error as an *Error when the run failed and ErrRunCancelled when it was cancelled.

In workflow context, Value reads the run once and returns ErrRunNotFinished when the run had not finished, so replay stays deterministic. In step context and outside a run, it blocks until the run finishes or ctx is done.

func (*Run[Out]) WakeUp 

func (r *Run[Out]) WakeUp(ctx context.Context, options ...WakeOptions) (int, error)

WakeUp settles the run's pending timers early. It queues a delivery so the run resumes, and returns the number of timers it settled. Waking a deadline timer triggers its cancellation continuation.

func (*Run[Out]) WorkflowName 

func (r *Run[Out]) WorkflowName(ctx context.Context) (string, error)

WorkflowName returns the name of the workflow the run executes. It returns worlds.ErrRunNotFound when the run does not exist.

func (*Run[Out]) Writer 

func (r *Run[Out]) Writer(ctx context.Context, options ...WriterOption) (io.WriteCloser, func() error, error)

Writer returns a writer for the run's stream, for step context and for use outside a run. It waits until no other writer holds the stream name. Close closes the stream for all writers, while the returned release function only gives up this writer's hold so another writer can continue the stream. Writer returns ErrNotInStep in workflow context.

type StartOnLatest added in v0.7.0 

type StartOnLatest struct{}

StartOnLatest marks a workflow, when embedded in the workflow struct, whose new runs start on the World's latest deployment instead of the deployment that starts them: 

type CheckIns struct {
	workflow.Struct[CheckInsResult]
	workflow.StartOnLatest
	UserID string
}

A run never moves once started, so its event log is only ever replayed by the code that wrote it. A workflow that starts its next run before it finishes, such as a daily chain, moves to new code at that boundary instead of staying on the deployment its first run started on until the deployment is deleted. Mark only workflows nothing waits on the output of, since a newer deployment may return a different output. On Worlds without deployment affinity, or that do not implement worlds.LatestDeployments, it changes nothing.

It has no fields, so it adds nothing to the workflow's JSON input.

type Status 

type Status = worlds.Status

Status is the state of a run or step.

type StepMetadata 

type StepMetadata struct {
	// ID is the step ID, stable across attempts. It is suitable as an
	// idempotency key.
	ID string
	// Name is the step function's name.
	Name string
	// StartedAt is when the current attempt started.
	StartedAt time.Time
	// Attempt is the current attempt number, starting at 1.
	Attempt int
}

StepMetadata describes the step attempt that step context belongs to.

func GetStepMetadata 

func GetStepMetadata(ctx context.Context) StepMetadata

GetStepMetadata returns metadata about the step ctx belongs to, or the zero StepMetadata when ctx does not belong to a step.

type StepOptions 

type StepOptions struct {
	// MaxRetries is the number of times a failed step is retried before it
	// fails. Omitting StepOptions allows 3 retries, while the zero value
	// allows none. A FatalError is never retried.
	MaxRetries int
}

StepOptions holds the options for a step.

type StreamChunk 

type StreamChunk = worlds.StreamChunk

StreamChunk is one chunk of a stream: the data of one write and its zero-based index in the stream.

type StreamReader 

type StreamReader interface {
	// Read reads the stream's bytes, blocking until data is written. It
	// returns io.EOF once the stream is closed and read.
	io.Reader
	// Next returns the next chunk, blocking until one is written. After a
	// partial Read, it returns the rest of the current chunk. It returns
	// io.EOF once the stream is closed and read.
	Next() (StreamChunk, error)
	// TailIndex returns the index of the stream's last chunk, or -1 when it
	// has none. A reconnecting client can resume from the index after the
	// last chunk it received.
	TailIndex() (int, error)
}

StreamReader reads a run's stream. Read joins the stream's chunks into bytes, while Next returns one chunk at a time with its index, so a reader can relay whole writes and resume after the last index it saw.

func Reader 

func Reader(ctx context.Context, options ...ReaderOption) (StreamReader, error)

Reader returns a reader for a stream of the run ctx belongs to, from ReaderOption.StartIndex. Reads block until data is written and return io.EOF once the stream is closed. It must be called in step context. It returns ErrNotInStep in workflow context and ErrNoWorkflow outside a run.

type Struct added in v0.3.0 

type Struct[Out any] struct{}

Struct declares a workflow's output type when embedded in the workflow struct, in place of writing its Workflow(Out) method: 

type GenerateReport struct {
	workflow.Struct[GenerateReportResult]
	Topic string
}

It has no fields, so it adds nothing to the workflow's JSON input.

func (Struct[Out]) Workflow added in v0.3.0 

func (Struct[Out]) Workflow(Out)

Workflow declares Out as the output of the workflow that embeds Struct.

type WaitGroup 

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

WaitGroup waits for all its members, ignoring operation failures. Its zero value is ready to use. After completion, new work is ignored. Create a new WaitGroup to wait for more work.

func (*WaitGroup) Add 

func (g *WaitGroup) Add(members ...promise)

Add includes existing results, hooks, webhooks, or groups in the wait. It does not change their cancellation contexts and ignores new members after settlement.

func (*WaitGroup) Wait 

func (g *WaitGroup) Wait(ctx context.Context) error

Wait waits for every member to settle.

type WakeOptions 

type WakeOptions struct {
	// CorrelationIDs limits WakeUp to the waits with these IDs. When empty,
	// WakeUp settles every pending wait in the run.
	CorrelationIDs []string
}

WakeOptions holds the options to WakeUp.

type Webhook 

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

Webhook is a hook resumed by an HTTP request to its URL, whose JSON request body is decoded as T.

func CreateWebhook 

func CreateWebhook[T any](ctx context.Context, options ...WebhookOption) (*Webhook[T], error)

CreateWebhook creates a webhook in the run ctx belongs to. A webhook is a hook resumed once by an HTTP request to its URL. On replay it returns the same webhook. The last of options applies. It returns ErrNoWorkflow when ctx does not belong to a run.

func (*Webhook[T]) Dispose 

func (w *Webhook[T]) Dispose(ctx context.Context)

Dispose durably deactivates the webhook endpoint and panics on failure.

func (*Webhook[T]) ID 

func (w *Webhook[T]) ID() string

ID returns the hook ID.

func (*Webhook[T]) Request 

func (w *Webhook[T]) Request(ctx context.Context) (*WebhookRequest[T], error)

Request waits for the HTTP request and preserves its origin for a step response.

func (*Webhook[T]) Token 

func (w *Webhook[T]) Token() string

Token returns the webhook's hook token, which is part of its URL.

func (*Webhook[T]) URL 

func (w *Webhook[T]) URL() string

URL returns the webhook's absolute URL, under the World's base URL.

type WebhookOption 

type WebhookOption struct {
	// Method is the HTTP method the webhook accepts. When empty, it accepts
	// any method.
	Method string
	// Response is the response sent to the request that resumes the webhook.
	// Its body is limited to 128 KiB. When nil and Manual is false, the
	// request receives 202 Accepted.
	Response *http.Response
	// Manual holds the request open until the runner responds with
	// RespondWith.
	Manual bool
}

WebhookOption holds the options to CreateWebhook. A webhook always gets a random hook token, since its URL is public.

type WebhookRequest 

type WebhookRequest[T any] struct {
	*http.Request
	// contains filtered or unexported fields
}

WebhookRequest is a serializable HTTP request. Pass it to a step to inspect its body or send a manual response.

func (*WebhookRequest[T]) JSON 

func (r *WebhookRequest[T]) JSON() (T, error)

JSON decodes the stored body without consuming Request.Body.

func (*WebhookRequest[T]) MarshalJSON 

func (r *WebhookRequest[T]) MarshalJSON() ([]byte, error)

func (*WebhookRequest[T]) RespondWith 

func (r *WebhookRequest[T]) RespondWith(ctx context.Context, response *http.Response) error

RespondWith records the response for this request. It is step-only, and retrying the same response is idempotent.

func (*WebhookRequest[T]) UnmarshalJSON 

func (r *WebhookRequest[T]) UnmarshalJSON(data []byte) error

type WorkflowMetadata 

type WorkflowMetadata struct {
	// Name is the workflow name.
	Name string
	// RunID is the run's ID.
	RunID string
	// StartedAt is when the run started, or when it was created if it has
	// not started.
	StartedAt time.Time
}

WorkflowMetadata describes the run a workflow or step is executing in.

func GetWorkflowMetadata 

func GetWorkflowMetadata(ctx context.Context) WorkflowMetadata

GetWorkflowMetadata returns metadata about the run ctx belongs to, or the zero WorkflowMetadata when ctx does not belong to a run.

type World 

type World = worlds.World

World is the backend a run is recorded in. It provides storage, a delivery queue, and streams.

type WriterOption 

type WriterOption struct {
	// Name is the stream name. The zero value names the run's default stream.
	Name string
}

WriterOption holds the options to Writer and Run.Writer.

Source Files

View all Source files

Directories

Path Synopsis
Package check provides Analyzer, a static analyzer that reports runners that cannot replay deterministically and calls that misuse the workflow API.
 Package check provides Analyzer, a static analyzer that reports runners that cannot replay deterministically and calls that misuse the workflow API.
cmd
workflowcheck command
Command workflowcheck runs the check analyzer as a go vet tool.
 Command workflowcheck runs the check analyzer as a go vet tool.
Package workflowai runs AI agents as durable workflow runs.
 Package workflowai runs AI agents as durable workflow runs.
Package workflowdev integrates go-workflow's static analyzer with go-dev.
 Package workflowdev integrates go-workflow's static analyzer with go-dev.
Package workflowtest provides conformance tests for worlds.World implementations.
 Package workflowtest provides conformance tests for worlds.World implementations.
Package worlds defines the World contract that go-workflow records runs in, the runtime that executes deliveries against it, and the HTTP server that exposes a World to clients.
 Package worlds defines the World contract that go-workflow records runs in, the runtime that executes deliveries against it, and the HTTP server that exposes a World to clients.
aws
Package awsworld implements a World on DynamoDB, SQS, and EventBridge Scheduler, for deployments on AWS Lambda or other AWS compute where any number of functions and processes share one table.
 Package awsworld implements a World on DynamoDB, SQS, and EventBridge Scheduler, for deployments on AWS Lambda or other AWS compute where any number of functions and processes share one table.
local
Package localworld implements a World on JSON files in a local directory, for development and single-process deployments.
 Package localworld implements a World on JSON files in a local directory, for development and single-process deployments.
postgres
Package postgresworld implements a World on PostgreSQL, for production deployments where any number of application and worker processes share one database.
 Package postgresworld implements a World on PostgreSQL, for production deployments where any number of application and worker processes share one database.
testworld
Package testworld implements a World in memory, for tests.
 Package testworld implements a World in memory, for tests.
vercel
Package vercelworld implements a World on Redis (over TCP or the Upstash REST API) and Vercel Queues, for deployments on Vercel Functions.
 Package vercelworld implements a World on Redis (over TCP or the Upstash REST API) and Vercel Queues, for deployments on Vercel Functions.

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.