Versions in this module Expand all Collapse all v0 v0.7.0 Oct 20, 2026 Changes in this version + const StatusCancelled + const StatusCompleted + const StatusFailed + const StatusRunning + var ErrActivityNameConflict = errors.New("activity name already registered with another definition") + var ErrAppNotStarted = errors.New("app not started") + var ErrDeterminismViolation = replay.ErrDeterminismViolation + var ErrFenced = storage.ErrFenced + var ErrHistoryKeyConflict = storage.ErrHistoryKeyConflict + var ErrInstanceClosed = storage.ErrInstanceClosed + var ErrInstanceNotFound = storage.ErrInstanceNotFound + var ErrInvalidSignalChannel = storage.ErrInvalidSignalChannel + var ErrInvalidSignalPayload = errors.New("invalid signal payload") + var ErrInvalidWorkflowState = errors.New("invalid workflow state") + var ErrLeaseLost = replay.ErrLeaseLost + var ErrNotActivityContext = errors.New("not an activity context") + var ErrQueryCommand = replay.ErrQueryCommand + var ErrUnsupportedHistoryFormat = errors.New("unsupported history format") + var ErrWorkflowAlreadyCancelled = storage.ErrWorkflowAlreadyCancelled + var ErrWorkflowAlreadyCompleted = storage.ErrWorkflowAlreadyCompleted + var ErrWorkflowAlreadyFailed = storage.ErrWorkflowAlreadyFailed + var ErrWorkflowAlreadyStarted = storage.ErrWorkflowAlreadyStarted + var ErrWorkflowStartRejected = errors.New("workflow start rejected") + func Await(ctx *WorkflowContext, cond func() bool) error + func CancelWorkflow(ctx context.Context, app *App, instanceID, reason string) error + func EmbeddedMigrationsFS() fs.FS + func IsTerminalError(err error) bool + func Now(ctx *WorkflowContext) (value time.Time, err error) + func Pending(ctx *WorkflowContext, channel string) int + func Query[R any](ctx context.Context, app *App, instanceID, name string) (R, error) + func ReceiveAsync[T any](ctx *WorkflowContext, channel string) (value T, received bool, err error) + func ReceiveWithTimeout[T any](ctx *WorkflowContext, channel string, timeout time.Duration) (T, error) + func Receive[T any](ctx *WorkflowContext, channel string) (T, error) + func RecordHeartbeat(ctx context.Context) error + func RegisterActivity[I, O any](app *App, a *Activity[I, O]) error + func RegisterWorkflow[I, O any](app *App, workflow Workflow[I, O]) + func ReplayHistory(ctx context.Context, app *App, history *RecordedHistory) error + func ReplayInstance(ctx context.Context, app *App, id string) error + func SetQueryHandler[R any](ctx *WorkflowContext, name string, handler func() (R, error)) error + func Sleep(ctx *WorkflowContext, duration time.Duration, opts ...SleepOption) error + func SleepUntil(ctx *WorkflowContext, t time.Time, opts ...SleepOption) error + func StartWorkflowByName(ctx context.Context, app *App, workflow string, input any, opts ...StartOption) (string, error) + func StartWorkflow[I, O any](ctx context.Context, app *App, workflow Workflow[I, O], input I, ...) (string, error) + type Activity struct + func DefineActivity[I, O any](name string, fn func(ctx context.Context, input I) (O, error), ...) *Activity[I, O] + func (a *Activity[I, O]) Execute(ctx *WorkflowContext, input I, opts ...ExecuteOption) (O, error) + func (a *Activity[I, O]) ExecuteAsync(ctx *WorkflowContext, input I, opts ...ExecuteOption) (*Future[O], error) + func (a *Activity[I, O]) Name() string + type ActivityInfo struct + ActivityID string + ActivityName string + Attempt int + InstanceID string + Options ActivityOptions + WorkflowName string + func GetActivityInfo(ctx context.Context) *ActivityInfo + type ActivityOption func(*Activity[I, O]) + func WithActivityDefaults[I, O any](options ActivityOptions) ActivityOption[I, O] + func WithRetryPolicy[I, O any](policy *retry.Policy) ActivityOption[I, O] + func WithTimeout[I, O any](d time.Duration) ActivityOption[I, O] + type ActivityOptions struct + HeartbeatTimeout time.Duration + Queue string + RetryPolicy *retry.Policy + ScheduleToStartTimeout time.Duration + StartToCloseTimeout time.Duration + type ActivityTaskInfo struct + Attempts int + Failures int + Key string + LastError error + Name string + Options ActivityOptions + Queue string + Seq int64 + Status string + func ListActivityTasks(ctx context.Context, app *App, id string) ([]ActivityTaskInfo, error) + type ActivityTimeoutError struct + Activity string + Attempt int + Kind TimeoutKind + func (e *ActivityTimeoutError) Error() string + func (e *ActivityTimeoutError) Unwrap() error + type App struct + func NewApp(opts ...Option) *App + func (a *App) FindInstances(ctx context.Context, inputFilters map[string]any) ([]*storage.WorkflowInstance, error) + func (a *App) FindInstancesWithOptions(ctx context.Context, opts storage.ListInstancesOptions) ([]*storage.WorkflowInstance, error) + func (a *App) GetInstance(ctx context.Context, instanceID string) (*storage.WorkflowInstance, error) + func (a *App) Shutdown(ctx context.Context) error + func (a *App) Signal(ctx context.Context, instanceID, channel string, payload any) error + func (a *App) Start(ctx context.Context) error + func (a *App) Storage() storage.Storage + func (a *App) WorkerID() string + type ApplicationError struct + Cause error + Details any + Message string + NonRetryable bool + Type string + func NewApplicationError(message, errType string, details any, cause error) *ApplicationError + func NewNonRetryableApplicationError(message, errType string, details any, cause error) *ApplicationError + func (e *ApplicationError) Error() string + func (e *ApplicationError) Unwrap() error + type Child struct + func StartChild[O any](ctx *WorkflowContext, workflow string, input any, opts ...ChildOption) (*Child[O], error) + func (c *Child[O]) Cancel(ctx *WorkflowContext) error + func (c *Child[O]) Get(ctx *WorkflowContext) (O, error) + func (c *Child[O]) ID() string + func (c *Child[O]) IsReady() bool + func (c *Child[O]) Key() string + type ChildOption func(*childOptions) + func WithChildID(id string) ChildOption + type ExecuteOption func(*executeOptions) + func WithActivityID(id string) ExecuteOption + func WithActivityOptions(options ActivityOptions) ExecuteOption + type FindInstancesOptions = storage.ListInstancesOptions + type Future struct + func (f *Future[O]) Get(ctx *WorkflowContext) (O, error) + func (f *Future[O]) IsReady() bool + func (f *Future[O]) Key() string + type InstanceClosedError = storage.InstanceClosedError + type LockAcquisitionError struct + InstanceID string + Reason string + WorkerID string + func (e *LockAcquisitionError) Error() string + type NondeterminismError = replay.NondeterminismError + type Option func(*appConfig) + func WithActivityPollInterval(d time.Duration) Option + func WithActivityQueues(queues ...string) Option + func WithAutoMigrate(enabled bool) Option + func WithClock(clock func() time.Time) Option + func WithDatabase(url string) Option + func WithHooks(h hooks.WorkflowHooks) Option + func WithLeaderHeartbeatInterval(d time.Duration) Option + func WithLeaderLeaseDuration(d time.Duration) Option + func WithLease(ttl, renewInterval, margin, maxSkew time.Duration) Option + func WithListenNotify(enabled *bool) Option + func WithMaxConcurrentActivities(n int) Option + func WithMaxConcurrentResumptions(n int) Option + func WithMaxConcurrentTimers(n int) Option + func WithMaxPollInterval(d time.Duration) Option + func WithMaxTimersPerBatch(n int) Option + func WithMaxWorkflowsPerBatch(n int) Option + func WithMigrationsFS(migrationsFS fs.FS) Option + func WithNotifyReconnectDelay(d time.Duration) Option + func WithServiceName(name string) Option + func WithShutdownTimeout(d time.Duration) Option + func WithTimerCheckInterval(d time.Duration) Option + func WithWaitPollInterval(d time.Duration) Option + func WithWorkerID(id string) Option + func WithWorkflowResumptionInterval(d time.Duration) Option + type QueryFailedError struct + Err error + Name string + func (e *QueryFailedError) Error() string + func (e *QueryFailedError) Unwrap() error + type QueryNotFoundError struct + Name string + func (e *QueryNotFoundError) Error() string + type RecordedEvent struct + Binary []byte + Command string + Data json.RawMessage + DataType string + DeliveredSeq int64 + EventType string + Key string + Reservation json.RawMessage + Seq int64 + type RecordedHistory struct + Events []RecordedEvent + Format string + Instance RecordedInstance + Signals []RecordedSignal + func ExportHistory(ctx context.Context, app *App, id string) (*RecordedHistory, error) + func ParseHistory(data []byte) (*RecordedHistory, error) + type RecordedInstance struct + CancelReason string + CancelRequested bool + Input json.RawMessage + InstanceID string + Output json.RawMessage + ParentInstanceID string + ParentKey string + Status WorkflowStatus + WorkflowName string + type RecordedSignal struct + Channel string + ConsumedSeq int64 + DeliveredSeq int64 + ID int64 + type ReplayDecodeError struct + Err error + GoType string + InstanceID string + Key string + func (e *ReplayDecodeError) Error() string + func (e *ReplayDecodeError) Unwrap() error + type ReplayError struct + Divergence error + InstanceID string + WorkflowErr error + func (e *ReplayError) Error() string + func (e *ReplayError) Unwrap() []error + type ReplayOutcomeError struct + InstanceID string + Recorded json.RawMessage + RecordedStatus WorkflowStatus + Replayed json.RawMessage + ReplayedStatus WorkflowStatus + func (e *ReplayOutcomeError) Error() string + type RetryExhaustedError struct + ActivityName string + Attempts int + LastErr error + func (e *RetryExhaustedError) Error() string + func (e *RetryExhaustedError) Unwrap() error + type SleepOption func(*sleepOptions) + func WithSleepID(id string) SleepOption + type StartOption func(*startOptions) + func WithInstanceID(id string) StartOption + type TerminalError struct + Err error + func NewTerminalError(err error) *TerminalError + func NewTerminalErrorf(format string, args ...any) *TerminalError + func (e *TerminalError) Error() string + func (e *TerminalError) Unwrap() error + type TimeoutKind string + const TimeoutHeartbeat + const TimeoutScheduleToStart + const TimeoutStartToClose + type Timer struct + func NewTimer(ctx *WorkflowContext, d time.Duration, opts ...SleepOption) (*Timer, error) + func (t *Timer) Get(ctx *WorkflowContext) (time.Time, error) + func (t *Timer) IsReady() bool + func (t *Timer) Key() string + type Workflow interface + Execute func(ctx *WorkflowContext, input I) (O, error) + Name func() string + func GetWorkflow[I, O any](name string) Workflow[I, O] + type WorkflowCancelledError struct + InstanceID string + Reason string + func (e *WorkflowCancelledError) Error() string + func (e *WorkflowCancelledError) Unwrap() error + type WorkflowContext struct + func Detached(ctx *WorkflowContext) *WorkflowContext + func NewWorkflowContext(ctx context.Context, instanceID, workerID, workflowName string) *WorkflowContext + func NewWorkflowContextFromExecution(execCtx *replay.ExecutionContext) *WorkflowContext + func WithCancel(ctx *WorkflowContext) (scoped *WorkflowContext, cancel func()) + func (c *WorkflowContext) App() *App + func (c *WorkflowContext) Context() context.Context + func (c *WorkflowContext) Done() <-chan struct{} + func (c *WorkflowContext) Err() error + func (c *WorkflowContext) Hooks() hooks.WorkflowHooks + func (c *WorkflowContext) InstanceID() string + func (c *WorkflowContext) IsReplaying() bool + func (c *WorkflowContext) SetApp(app *App) + func (c *WorkflowContext) Storage() WorkflowStorage + func (c *WorkflowContext) WithContext(ctx context.Context) *WorkflowContext + func (c *WorkflowContext) WorkerID() string + func (c *WorkflowContext) WorkflowName() string + type WorkflowDescription struct + CancelReason string + CancelRequested bool + InstanceID string + ParentInstanceID string + ParentKey string + StartedAt time.Time + Status WorkflowStatus + UpdatedAt time.Time + WorkflowName string + func Describe(ctx context.Context, app *App, instanceID string) (*WorkflowDescription, error) + type WorkflowFunc struct + func DefineWorkflow[I, O any](name string, fn func(ctx *WorkflowContext, input I) (O, error)) *WorkflowFunc[I, O] + func (w *WorkflowFunc[I, O]) Execute(ctx *WorkflowContext, input I) (O, error) + func (w *WorkflowFunc[I, O]) Name() string + type WorkflowNotFoundError struct + InstanceID string + func (e *WorkflowNotFoundError) Error() string + func (e *WorkflowNotFoundError) Unwrap() error + type WorkflowNotRegisteredError struct + WorkflowName string + func (e *WorkflowNotRegisteredError) Error() string + type WorkflowPanicError = replay.WorkflowPanicError + type WorkflowResult struct + Error error + InstanceID string + Output O + Status string + func GetWorkflowResult[O any](ctx context.Context, app *App, instanceID string) (*WorkflowResult[O], error) + func WaitWorkflow[O any](ctx context.Context, app *App, instanceID string) (*WorkflowResult[O], error) + type WorkflowStatus = storage.WorkflowStatus + type WorkflowStorage interface + GetHistoryEvent func(context.Context, string, string) (*storage.HistoryEvent, error)
Go to main page