Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
- API
  
  Reference documentation for the pkg.go.dev API
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

engine

package

Go to main page

Versions in this module

v0

v0.7.0

Jun 25, 2026

Changes in this version

type Engine

+ func (e *Engine) Recover(ctx context.Context, flowKey string, stateOverrides any) error

v0.6.3

Jun 22, 2026

v0.6.2

Jun 22, 2026

v0.6.1

Jun 22, 2026

v0.6.0

Jun 22, 2026

v0.5.0

Jun 18, 2026

v0.4.1

Jun 16, 2026

v0.4.0

Jun 16, 2026

Changes in this version

type Engine

+ func (e *Engine) DeliverSignal(ctx context.Context, op string, payload []byte) error

+ func (e *Engine) SetDSN(dsn string) error

+ func (e *Engine) SetDebugLogger() error

+ func (e *Engine) SetDefaultPriority(p int) error

+ func (e *Engine) SetHost(h Host) error

+ func (e *Engine) SetLogger(l *slog.Logger) error

+ func (e *Engine) SetMaxOpenConns(n int) error

+ func (e *Engine) SetMeterProvider(mp metric.MeterProvider) error

+ func (e *Engine) SetNumShards(num int) error

+ func (e *Engine) SetTimeBudget(d time.Duration) error

+ func (e *Engine) SetTracerProvider(tp trace.TracerProvider) error

+ func (e *Engine) SetWorkers(n int) error

+ func (e *Engine) StartupInTest(ctx context.Context, testID string) error

+ func (e *Engine) ValveCount() int

+ type Host interface

+ ExecuteTask func(ctx context.Context, taskURL string, flow *workflow.Flow) error

+ FlowStopped func(ctx context.Context, outcome *workflow.FlowOutcome)

+ LoadGraph func(ctx context.Context, workflowURL string) (*workflow.Graph, error)

+ SignalPeers func(ctx context.Context, op string, payload []byte)

type TestProxy

+ func (p *TestProxy) AddPeer(peer *Engine)

+ func (p *TestProxy) FlowStopped(ctx context.Context, outcome *workflow.FlowOutcome)

+ func (p *TestProxy) OnFlowStopped(cb func(ctx context.Context, outcome *workflow.FlowOutcome))

+ func (p *TestProxy) SignalPeers(ctx context.Context, op string, payload []byte)

v0.3.2

Jun 14, 2026

v0.3.1

Jun 14, 2026

v0.3.0

Jun 14, 2026

v0.2.0

Jun 13, 2026

v0.1.0

Jun 13, 2026

Changes in this version

+ type Engine struct

+ func NewEngine() *Engine

+ func (e *Engine) Await(ctx context.Context, flowKey string) (*workflow.FlowOutcome, error)

+ func (e *Engine) BreakBefore(ctx context.Context, flowKey string, taskName string, enabled bool) error

+ func (e *Engine) BreakerTripped(taskName string) bool

+ func (e *Engine) Cancel(ctx context.Context, flowKey string, reason string) error

+ func (e *Engine) Continue(ctx context.Context, threadKey string, additionalState any, ...) (string, error)

+ func (e *Engine) Create(ctx context.Context, workflowName string, initialState any, ...) (flowKey string, err error)

+ func (e *Engine) CreateTask(ctx context.Context, taskName string, initialState any, ...) (flowKey string, err error)

+ func (e *Engine) Delete(ctx context.Context, flowKey string) error

+ func (e *Engine) Fingerprint(ctx context.Context, flowKey string) (fingerprint string, status string, err error)

+ func (e *Engine) HandleEnqueue(ctx context.Context, shard, stepID int) error

+ func (e *Engine) HandleNotifyStatusChange(ctx context.Context, flowKey string, status string) error

+ func (e *Engine) HandleSyncValve(ctx context.Context, taskName string, wCong int, tCong time.Time) error

+ func (e *Engine) HandleTripBreaker(ctx context.Context, taskName string) error

+ func (e *Engine) History(ctx context.Context, flowKey string) ([]workflow.FlowStep, error)

+ func (e *Engine) HistoryMermaid(ctx context.Context, flowKey string, w io.StringWriter) error

+ func (e *Engine) List(ctx context.Context, query workflow.Query) ([]workflow.FlowSummary, string, error)

+ func (e *Engine) Purge(ctx context.Context, query workflow.Query) (int, error)

+ func (e *Engine) Restart(ctx context.Context, flowKey string, stateOverrides any) error

+ func (e *Engine) RestartFrom(ctx context.Context, stepKey string, stateOverrides any) error

+ func (e *Engine) Resume(ctx context.Context, flowKey string, resumeData any) error

+ func (e *Engine) ResumeBreak(ctx context.Context, flowKey string, stateOverrides any) error

+ func (e *Engine) Run(ctx context.Context, workflowName string, initialState any, ...) (*workflow.FlowOutcome, error)

+ func (e *Engine) RunInTest(t *testing.T)

+ func (e *Engine) ShardInfo(ctx context.Context) ([]ShardSummary, error)

+ func (e *Engine) Shutdown(ctx context.Context) error

+ func (e *Engine) Snapshot(ctx context.Context, flowKey string) (*workflow.FlowOutcome, error)

+ func (e *Engine) Start(ctx context.Context, flowKey string) error

+ func (e *Engine) StartNotify(ctx context.Context, flowKey string, notifyHostname string) error

+ func (e *Engine) Startup(ctx context.Context) error

+ func (e *Engine) Step(ctx context.Context, stepKey string) (*workflow.FlowStep, error)

+ func (e *Engine) WithDSN(dsn string) *Engine

+ func (e *Engine) WithDefaultPriority(p int) *Engine

+ func (e *Engine) WithFlowStoppedCallback(cb FlowStoppedCallback) *Engine

+ func (e *Engine) WithGraphLoader(gl GraphLoader) *Engine

+ func (e *Engine) WithLogger(l *slog.Logger) *Engine

+ func (e *Engine) WithMaxOpenConns(n int) *Engine

+ func (e *Engine) WithMeterProvider(mp metric.MeterProvider) *Engine

+ func (e *Engine) WithNumShards(n int) *Engine

+ func (e *Engine) WithPeerNotifier(pn PeerNotifier) *Engine

+ func (e *Engine) WithTaskExecutor(te TaskExecutor) *Engine

+ func (e *Engine) WithTimeBudget(d time.Duration) *Engine

+ func (e *Engine) WithTracerProvider(tp trace.TracerProvider) *Engine

+ func (e *Engine) WithWorkers(n int) *Engine

+ type FlowStoppedCallback func(ctx context.Context, hostname string, outcome *workflow.FlowOutcome)

+ type GraphLoader func(ctx context.Context, workflowName string) (*workflow.Graph, error)

+ type PeerNotifier interface

+ Enqueue func(ctx context.Context, shard, stepID int)

+ NotifyStatusChange func(ctx context.Context, flowKey string, status string)

+ SyncValve func(ctx context.Context, taskName string, wCong int, tCong time.Time)

+ TripBreaker func(ctx context.Context, taskName string)

+ type ShardSummary struct

+ Error string

+ Flows int

+ LatencyMs int

+ Shard int

+ Steps int

+ type TaskExecutor func(ctx context.Context, taskName string, flow *workflow.Flow) error

+ type TaskHandler func(ctx context.Context, flow *workflow.Flow) error

+ type TestProxy struct

+ func NewTestProxy() *TestProxy

+ func (p *TestProxy) ExecuteTask(ctx context.Context, taskName string, flow *workflow.Flow) error

+ func (p *TestProxy) HandleGraph(name string, graph *workflow.Graph)

+ func (p *TestProxy) HandleTask(name string, handler TaskHandler)

+ func (p *TestProxy) LoadGraph(ctx context.Context, workflowName string) (*workflow.Graph, error)