autocore

package
v19.2.0 Latest Latest
Warning

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

Go to latest
Published: Jul 15, 2026 License: MIT Imports: 41 Imported by: 0

Documentation

Index

Constants

View Source
const (
	SignalTypeUnknown        = persistence.SignalTypeUnknown
	SignalTypeCancelWorkflow = persistence.SignalTypeCancelWorkflow
	SignalTypeChannel        = persistence.SignalTypeChannel
	SignalTypeChildTerminal  = persistence.SignalTypeChildTerminal
)
View Source
const (

	// DefaultScheduleToCompleteTimeout is applied to every workflow whose
	// ExecuteWorkflowOptions.ScheduleToCompleteTimeout is zero. The deadline
	// cannot be disabled per workflow.
	DefaultScheduleToCompleteTimeout = 30 * 24 * time.Hour
	// MaxScheduleToCompleteTimeout is the hard upper bound on any per-workflow
	// schedule-to-complete deadline (the default or a per-call override).
	MaxScheduleToCompleteTimeout = 60 * 24 * time.Hour
)

Variables

View Source
var (
	ErrCanceled = errors.New("workflow canceled")
	// ErrTimedOut marks a workflow terminated because its schedule-to-complete
	// deadline elapsed. Surfaced to pending futures like ErrCanceled, but maps
	// to the TimedOut terminal rather than Canceled.
	ErrTimedOut = errors.New("workflow timed out")
	// ErrSystemFailure marks a failure the system caused (e.g. an activity whose
	// abort retry budget was exhausted or that hit a deterministic internal
	// error) rather than a deliberate error return. A workflow that propagates
	// it unhandled terminates as SystemFailed instead of Failed.
	ErrSystemFailure             = errors.New("system failure")
	ErrReservedIdentityKeyPrefix = fmt.Errorf("identity_key starting with %q is reserved for engine-internal use", persistence.ReservedIdentityKeyPrefix)
	ErrChildFanOutExceeded       = errors.New("child workflow fan-out cap exceeded")
	ErrWorkflowNotFound          = persistence.ErrWorkflowNotFound
	// ErrInvalidWorkflowKey is returned by ParseWorkflowKey when the input is not a
	// well-formed workflow key. Callers can map it to a permanent client error.
	ErrInvalidWorkflowKey = errors.New("invalid workflow key")
)
View Source
var (
	WorkflowExecutionResultCompleted    = WorkflowExecutionResult(persistence.WorkflowExecutionStateCompleted.String())
	WorkflowExecutionResultFailed       = WorkflowExecutionResult(persistence.WorkflowExecutionStateFailed.String())
	WorkflowExecutionResultCanceled     = WorkflowExecutionResult(persistence.WorkflowExecutionStateCanceled.String())
	WorkflowExecutionResultTimedOut     = WorkflowExecutionResult(persistence.WorkflowExecutionStateTimedOut.String())
	WorkflowExecutionResultSystemFailed = WorkflowExecutionResult(persistence.WorkflowExecutionStateSystemFailed.String())
)
View Source
var ErrPollExhausted = errors.New("poll exhausted without satisfying predicate")

ErrPollExhausted is the error a poll activity's future resolves to when its poll deadline passes without the predicate ever passing.

Functions

func ExecuteInlineActivity

func ExecuteInlineActivity[I, O proto.Message, N ActivityName[I, O]](ctx WorkflowContext, name N, input I, opts ...ExecuteInlineActivityOption) (O, error)

ExecuteInlineActivity invokes a registered activity synchronously inside the workflow goroutine. The body runs inline (no activity_task row, no separate worker), and its outcome is recorded as a single InlineActivity{Completed, Failed} event in workflow history in a dedicated fenced TX committed before this function returns. Replay reads the outcome from history; the body does NOT run on replay.

No retries. Use WithInlineTimeout for an execution-time bound. The caller is responsible for retry policy if needed -- e.g. wrap in a loop with NewTimer for backoff, or fall back to ExecuteActivity after a failure.

IDEMPOTENCY: the activity body MUST be idempotent at its destination. On a worker crash between body execution and the history-commit TX (a window of a single TX commit, typically milliseconds), the body is re-invoked on round retry. Autocore guarantees the result is recorded at most once in history; it does NOT guarantee the body runs only once.

func RegisterActivity

func RegisterActivity[I, O proto.Message, N ActivityName[I, O]](r *Registry, name N, fn func(ctx ActivityContext, input I) (O, error), opts ...RegisterActivityOption[O])

RegisterActivity registers a typed activity function. Options attach optional per-activity capabilities such as a poll predicate (see WithPollPredicate).

func RegisterWorkflow

func RegisterWorkflow[I, O proto.Message, N WorkflowName[I, O]](r *Registry, name N, fn func(ctx WorkflowContext, input I) (O, error), opts ...RegisterWorkflowOption)

RegisterWorkflow registers a typed workflow function.

Types

type AbortCause

type AbortCause string

AbortCause categorizes why an in-flight task run produced no persisted result. Aborted runs do not consume retry budget; the row stays in 'running' state for recovery to sweep back to 'pending'.

const (
	// AbortCauseFenced: this instance lost the shard to another instance
	// mid-run (errFencedOut), or a fenced transaction returned ErrFenced.
	AbortCauseFenced AbortCause = "fenced"
	// AbortCauseShardReleased: this instance voluntarily released the shard
	// mid-run (errShardReleased).
	AbortCauseShardReleased AbortCause = "released"
	// AbortCauseShutdown: parent context canceled (typically pool shutdown)
	// without a more specific cause.
	AbortCauseShutdown AbortCause = "shutdown"
	// AbortCausePersistFailed: a fenced transaction returned a non-fence
	// error before a result was committed.
	AbortCausePersistFailed AbortCause = "persist_failed"
	// AbortCauseInternalError: a pre-execute setup failure (proto decode,
	// unknown workflow/activity, build-context failure). Indicates a bug
	// or configuration drift, not infrastructure flakiness.
	AbortCauseInternalError AbortCause = "internal_error"
	// AbortCauseUnknown: cause could not be classified -- e.g. parent
	// context canceled with a cause we don't recognize, or shard ownership
	// lost without a cancellation cause having propagated yet (race with
	// RemoveShard). Indicates a gap in classification rather than a known
	// failure mode; investigate if it stops being rare.
	AbortCauseUnknown AbortCause = "unknown"
)

type AbortError

type AbortError struct {
	Cause AbortCause
	Err   error
}

AbortError pairs an AbortCause with the underlying error (if any). Passed by value on the hot path so the struct stays on the stack; metric records still go through the pre-computed workflowAbortAddOpts / activityAbortAddOpts maps keyed on Cause, so observability is allocation-free aside from the struct copy.

func (AbortError) Error

func (e AbortError) Error() string

func (AbortError) LogAttrs

func (e AbortError) LogAttrs() slog.Attr

LogAttrs returns the abort cause and underlying error (if any) as a single slog.Attr. The empty group name flattens both fields to the top level of the log record so they appear as plain abort_cause and error keys, not nested under a group.

func (AbortError) Unwrap

func (e AbortError) Unwrap() error

type ActivityContext

type ActivityContext interface {
	context.Context
	Logger() *slog.Logger
	WorkflowKey() WorkflowKey
	// IdentityKey returns the opaque identity key of this activity execution.
	IdentityKey() string
}

type ActivityExecutionDisposition

type ActivityExecutionDisposition string

ActivityExecutionDisposition is the second dimension of the activity outcome (for non-aborted runs only): whether the persisted attempt was the final outcome or a retry was scheduled. Only meaningful when ActivityExecutionResult is not Aborted.

const (
	// ActivityExecutionDispositionSucceeded: result was Completed; no further
	// attempts.
	ActivityExecutionDispositionSucceeded ActivityExecutionDisposition = "succeeded"
	// ActivityExecutionDispositionRetried: result was Failed or TimedOut and the
	// activity has remaining attempts; a retry was scheduled.
	ActivityExecutionDispositionRetried ActivityExecutionDisposition = "retried"
	// ActivityExecutionDispositionExhausted: result was Failed or TimedOut and no
	// attempts remain; the activity is terminally failed.
	ActivityExecutionDispositionExhausted ActivityExecutionDisposition = "exhausted"
)

type ActivityExecutionResult

type ActivityExecutionResult string

ActivityExecutionResult is the typed outcome of a single activity execution. Aligned across metrics, traces, and logs via pre-computed attribute KVs.

const (
	ActivityExecutionResultCompleted ActivityExecutionResult = "completed"
	ActivityExecutionResultFailed    ActivityExecutionResult = "failed"
	ActivityExecutionResultTimedOut  ActivityExecutionResult = "timed_out"
	// ActivityExecutionResultAborted: execution produced no persisted
	// result (shard fenced, shard released, shutdown, persist failure,
	// internal error). Recorded via the activity aborts counter, not the
	// execution duration histogram. Aborted runs do not consume retry
	// budget; the activity_task row stays in 'running' for recovery.
	ActivityExecutionResultAborted ActivityExecutionResult = "aborted"
	// ActivityExecutionResultNotReady: a poll activity completed but its
	// predicate was not satisfied. It re-polls on the poll interval without
	// consuming the failure RetryPolicy attempt budget, until the predicate
	// passes or the poll deadline is reached (then it exhausts).
	ActivityExecutionResultNotReady ActivityExecutionResult = "not_ready"
)

type ActivityName

type ActivityName[I, O proto.Message] string

type ActivityPollPredicate added in v19.2.0

type ActivityPollPredicate[O proto.Message] func(ctx ActivityContext, output O, config []byte) (bool, error)

ActivityPollPredicate decides, from a typed activity output and the per-call opaque config, whether a poll is satisfied. It runs on the activity worker at completion, before any history is written.

(true, nil)  -> satisfied; record terminal success
(false, nil) -> not satisfied; re-poll via the activity retry path
(_, err)     -> terminal failure

Register one per pollable activity via RegisterActivity + WithPollPredicate.

type CancelWorkflowOptions

type CancelWorkflowOptions struct {
	WorkflowKey WorkflowKey
}

type Case

type Case interface {
	IsReadier
}

type Channel added in v19.2.0

type Channel[O proto.Message] interface {
	IsReadier
	Receive() (O, error)
	Name() string
}

Channel yields successive values over time, similar to a Go channel. Receive may be called repeatedly to obtain successive values.

func NewChannel added in v19.2.0

func NewChannel[O proto.Message](ctx WorkflowContext) Channel[O]

NewChannel constructs a typed channel sourcing values from the underlying channel on the workflow context.

type ChildFuture

type ChildFuture[O proto.Message] interface {
	Future[O]
	WorkflowID() WorkflowID
}

ChildFuture is a Future plus the deterministic child workflow ID, available synchronously (before persistence).

func ExecuteChildWorkflow

func ExecuteChildWorkflow[I, O proto.Message, N WorkflowName[I, O]](ctx WorkflowContext, name N, input I) ChildFuture[O]

ExecuteChildWorkflow schedules an awaitable child workflow on the same shard as the parent. The returned ChildFuture exposes the deterministic child WorkflowID synchronously and resolves once the child reaches a terminal state.

Parent termination cancels pending awaitable children via the signal inbox: each child observes the cancellation on its next yield and unwinds.

type Client

type Client interface {
	// ExecuteWorkflow executes the named workflow.
	// Do not use this method directly, use the ExecuteWorkflow function.
	ExecuteWorkflow(ctx context.Context, name string, input proto.Message, opts *ExecuteWorkflowOptions) (*ExecuteWorkflowResult, error)
	CancelWorkflow(ctx context.Context, opts *CancelWorkflowOptions) error
	// TODO: I think we want to align the name here with the workflow context.
	// Maybe SendMessageToWorkflow and ReceiveMessage (instead of ReceiveChannel)?
	// Not sure yet.
	SendToWorkflow(ctx context.Context, opts *SendToWorkflowOptions) error
	// GetWorkflowInfo returns a read-only snapshot of a workflow's execution
	// state and terminal result. Returns ErrWorkflowNotFound if no workflow
	// exists for the key.
	GetWorkflowInfo(ctx context.Context, key WorkflowKey) (*WorkflowInfo, error)
}

type Config

type Config struct {
	Log                   *slog.Logger
	MeterProvider         otelmetric.MeterProvider
	TracerProvider        oteltrace.TracerProvider
	ErrCapturer           errz.ErrCapturer
	Validator             protovalidate.Validator
	RedisClient           rueidis.Client
	RedisKeyPrefix        string
	InstanceID            int64
	MigrationCentralDBCfg pgtool.DBConnConfig
	CentralDBCfg          pgtool.DBConnConfig
	Registry              *Registry

	// Workflow worker pool size.
	WorkflowPoolSize      int
	MaxSuspendedWorkflows int

	// Activity worker pool size.
	ActivityPoolSize int
}

type Engine

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

func New

func New(ctx context.Context, cfg *Config) (*Engine, error)

func (*Engine) Client

func (e *Engine) Client() Client

func (*Engine) Run

func (e *Engine) Run(ctx context.Context) error

func (*Engine) RunMigrations

func (e *Engine) RunMigrations(ctx context.Context) error

type ExecuteActivityOption

type ExecuteActivityOption func(*executeActivityOptions)

func WithClaimToCompleteTimeout

func WithClaimToCompleteTimeout(d time.Duration) ExecuteActivityOption

func WithPollConfig added in v19.2.0

func WithPollConfig(config []byte, interval time.Duration, deadline time.Time) ExecuteActivityOption

WithPollConfig marks the activity invocation as a poll. config is the opaque config handed to the activity's registered poll predicate after each successful execution. A not-ready result re-polls after interval until deadline, at which point the poll is exhausted. Polling is bounded by deadline, independently of the failure RetryPolicy (which still governs genuine activity failures).

func WithRetryPolicy

func WithRetryPolicy(p *RetryPolicy) ExecuteActivityOption

WithRetryPolicy overrides the default retry policy entirely. The caller must set all fields - zero-valued fields are not merged with defaults.

type ExecuteInlineActivityOption

type ExecuteInlineActivityOption func(*executeInlineActivityOptions)

func WithInlineTimeout

func WithInlineTimeout(d time.Duration) ExecuteInlineActivityOption

WithInlineTimeout overrides the default execution timeout for an inline activity. On expiry the activity's context.Context is canceled; a well- behaved body returns (typically context.DeadlineExceeded). The error is recorded in history and replay returns the same recorded error.

type ExecuteWorkflowOptions

type ExecuteWorkflowOptions struct {
	ExternalNamespaceID ExternalNamespaceID
	IdentityKey         string
	// ScheduleToCompleteTimeout overrides the engine's default per-workflow
	// schedule-to-complete deadline. Zero uses the engine default; a positive
	// value overrides it. The deadline cannot be disabled per workflow.
	ScheduleToCompleteTimeout time.Duration
	// Metadata is extra information about the workflow execution, available to interceptors.
	Metadata proto.Message
}

type ExecuteWorkflowResult

type ExecuteWorkflowResult struct {
	WorkflowKey   WorkflowKey
	AlreadyExists bool
}

func ExecuteWorkflow

func ExecuteWorkflow[I, O proto.Message, N WorkflowName[I, O]](ctx context.Context, client Client, name N, input I, opts *ExecuteWorkflowOptions) (*ExecuteWorkflowResult, error)

type ExternalNamespaceID

type ExternalNamespaceID = persistence.ExternalNamespaceID

ExternalNamespaceID is the namespace id supplied by the caller when starting a workflow; autocore maps it to an internal namespace surrogate.

type Future

type Future[O proto.Message] interface {
	IsReadier
	Get() (O, error)
}

Future represents the result of an asynchronous operation (activity or timer). It is safe to call Get() multiple times.

func ExecuteActivity

func ExecuteActivity[I, O proto.Message, N ActivityName[I, O]](ctx WorkflowContext, name N, input I, opts ...ExecuteActivityOption) Future[O]

func ReceiveSignal

func ReceiveSignal[O proto.Message](ctx WorkflowContext, channelName string) Future[O]

type IsReadier

type IsReadier interface {
	IsReady() bool
}

IsReadier can announce that it is ready TODO: rename this to something sane :)

type ReadinessAnnouncer

type ReadinessAnnouncer interface {
	// Announce registers this instance as ready in the shared store.
	Announce(ctx context.Context) error
	// ReadyCount returns the number of ready instances with hysteresis applied.
	ReadyCount(ctx context.Context) int32
	// Deregister removes this instance from the ready set.
	Deregister(ctx context.Context) error
	// GC removes expired entries from the backing store.
	GC(ctx context.Context)
}

ReadinessAnnouncer handles instance readiness announcements and peer counting.

type ReadinessTracker

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

ReadinessTracker implements ReadinessAnnouncer using a Redis hash via the ExpiringHashAPI abstraction. The tracker applies hysteresis to the observed ready count to dampen transient fleet-size fluctuations.

func NewReadinessTracker

func NewReadinessTracker(
	log *slog.Logger,
	client rueidis.Client,
	redisKeyPrefix string,
	instanceID int64,
	meter otelmetric.Meter,
	opts ...ReadinessTrackerOption,
) (*ReadinessTracker, error)

func (*ReadinessTracker) Announce

func (r *ReadinessTracker) Announce(ctx context.Context) error

func (*ReadinessTracker) Deregister

func (r *ReadinessTracker) Deregister(ctx context.Context) error

func (*ReadinessTracker) GC

func (r *ReadinessTracker) GC(ctx context.Context)

func (*ReadinessTracker) ReadyCount

func (r *ReadinessTracker) ReadyCount(ctx context.Context) int32

type ReadinessTrackerOption

type ReadinessTrackerOption func(*ReadinessTracker)

func WithReadinessTTL

func WithReadinessTTL(d time.Duration) ReadinessTrackerOption

func WithStabilityThreshold

func WithStabilityThreshold(n int) ReadinessTrackerOption

type RegisterActivityOption added in v19.2.0

type RegisterActivityOption[O proto.Message] func(r *Registry, name string)

RegisterActivityOption configures an activity registration. O is the activity's output type, so options can be typed against it.

func WithPollPredicate added in v19.2.0

func WithPollPredicate[O proto.Message](predicate ActivityPollPredicate[O]) RegisterActivityOption[O]

WithPollPredicate registers a typed poll predicate for the activity. When the activity is invoked as a poll (ExecuteActivity with WithPollConfig), the worker hands its output to predicate, which decides whether the poll is satisfied. The output is unmarshaled into O here, so the predicate is fully typed.

type RegisterWorkflowOption added in v19.2.0

type RegisterWorkflowOption func(r *Registry, name string)

func WithSignalPromotionInterceptor added in v19.2.0

func WithSignalPromotionInterceptor[C proto.Message](f func(ctx context.Context, log *slog.Logger, sCtx *SignalPromotionContext, cookie C) (bool, error)) RegisterWorkflowOption

WithSignalPromotionInterceptor sets an interceptor that is called for each signal that is being promoted. This interceptor can be used to drop signals by not promoting them. The returned error is logged and captured.

type Registry

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

Registry is not thread-safe. All registrations must happen before the registry is used by workers.

func NewRegistry

func NewRegistry() *Registry

type RetryPolicy

type RetryPolicy = persistence.RetryPolicy

type Selector

type Selector interface {
	AddCase(c Case, fn func()) Selector
	AddDefault(fn func()) Selector
	Select()
	HasPending() bool
}

type SendToWorkflowOptions

type SendToWorkflowOptions struct {
	WorkflowKey WorkflowKey
	IdentityKey string // caller-provided dedup key; auto-generated if empty
	ChannelName string
	Payload     proto.Message
}

type ShardID added in v19.2.0

type ShardID = persistence.ShardID

type SignalID added in v19.2.0

type SignalID = persistence.SignalID

type SignalPromotionContext added in v19.2.0

type SignalPromotionContext struct {
	ShardID     ShardID
	WorkflowID  WorkflowID
	SignalID    SignalID
	IdentityKey string
	SignalType  SignalType
	SignalName  string
	Payload     []byte
	CreatedAt   time.Time
}

type SignalType added in v19.2.0

type SignalType = persistence.SignalType

type TimerFuture

type TimerFuture Future[*emptypb.Empty]

type TypedWorkflowInfo

type TypedWorkflowInfo[O proto.Message] struct {
	WorkflowKey WorkflowKey
	Name        string
	IdentityKey string
	State       WorkflowState
	CreatedAt   time.Time
	UpdatedAt   time.Time
	Output      O
	Error       string
}

TypedWorkflowInfo is the generic counterpart of WorkflowInfo returned by the GetWorkflowInfo helper. Output is the workflow's output decoded into O, set only when State is Completed.

func GetWorkflowInfo

func GetWorkflowInfo[I, O proto.Message, N WorkflowName[I, O]](ctx context.Context, client Client, name N, key WorkflowKey) (*TypedWorkflowInfo[O], error)

GetWorkflowInfo returns a snapshot of the workflow addressed by key, with its output decoded into O. The name binds the output type and is verified against the stored workflow so the bytes are never decoded into the wrong type. Returns ErrWorkflowNotFound if no such workflow exists.

type WorkerManager

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

func NewWorkerManager

func NewWorkerManager(
	log *slog.Logger,
	validator protovalidate.Validator,
	meter otelmetric.Meter,
	tracer trace.Tracer,
	errCapturer errz.ErrCapturer,
	store *persistence.WorkloadStore,
	registry *Registry,
	workflowPool *workflowWorkerPool,
	activityPool *activityWorkerPool,
	opts ...WorkerManagerOption,
) (*WorkerManager, error)

func (*WorkerManager) AddShard

func (wm *WorkerManager) AddShard(ctx context.Context, shardID persistence.ShardID, fenceID persistence.FenceID)

func (*WorkerManager) DrainShard

func (wm *WorkerManager) DrainShard(shardID persistence.ShardID)

DrainShard marks an owned shard as draining: it is dropped from the claim snapshot and skipped by the scanners so no new work is dispatched onto it, while it stays in the fence map with its fence_id so in-flight fenced work keeps committing. No in-flight goroutine is canceled. A shard that is not owned cannot be drained.

func (*WorkerManager) InFlightByShard

func (wm *WorkerManager) InFlightByShard() map[persistence.ShardID]int

InFlightByShard returns the number of active in-flight tasks per shard across both worker pools, computed in a single pass per pool. Only active goroutines are counted: suspended workflows hold no running row and are evicted when the shard is released, so the pools' active-only counts already exclude them. The drain machinery uses this to pick the least-loaded shard to drain and to tell when a draining shard has no in-flight work left (count zero / absent).

func (*WorkerManager) NotifySignalReady

func (wm *WorkerManager) NotifySignalReady() bool

func (*WorkerManager) NotifyWorkflowTaskReady

func (wm *WorkerManager) NotifyWorkflowTaskReady() bool

func (*WorkerManager) RemoveShard

func (wm *WorkerManager) RemoveShard(ctx context.Context, shardID persistence.ShardID, cause error)

func (*WorkerManager) Run

func (wm *WorkerManager) Run(ctx context.Context)

Run starts all goroutines and blocks until ctx is canceled.

Goroutine layout:

  • 2 claim-and-dispatch loops (1 per task type), each issues a single cross-shard LATERAL query per poll tick and spawns worker goroutines
  • 0 to N workflow workers + 0 to M activity workers (spawned on demand)
  • 2 scanners (timers, signals), each fans out across owned shards per tick up to cfg.shardScanConcurrency

Fixed goroutines: 4. Dynamic goroutines: up to N + M.

type WorkerManagerOption

type WorkerManagerOption func(*WorkerManager)

func WithWMActivityPollInterval

func WithWMActivityPollInterval(d time.Duration) WorkerManagerOption

func WithWMMaxChildrenPerParent

func WithWMMaxChildrenPerParent(n int) WorkerManagerOption

WithWMMaxChildrenPerParent caps the lifetime number of children (every ChildWorkflowScheduled event in the parent's history plus the current yield round's child schedules) a single parent may schedule. Calls to ExecuteChildWorkflow / StartChildWorkflow above the cap surface ErrChildFanOutExceeded via the returned future / error.

func WithWMMinimumTick

func WithWMMinimumTick(d time.Duration) WorkerManagerOption

func WithWMRecoveryBatchSize

func WithWMRecoveryBatchSize(n int64) WorkerManagerOption

func WithWMShardScanConcurrency

func WithWMShardScanConcurrency(n int) WorkerManagerOption

WithWMShardScanConcurrency caps how many per-shard scanner transactions run in parallel within one tick. Set to 1 for fully sequential scanning.

func WithWMSignalScanInterval

func WithWMSignalScanInterval(d time.Duration) WorkerManagerOption

func WithWMSuspensionTTL

func WithWMSuspensionTTL(d time.Duration) WorkerManagerOption

func WithWMTimerScanInterval

func WithWMTimerScanInterval(d time.Duration) WorkerManagerOption

func WithWMWorkflowPollInterval

func WithWMWorkflowPollInterval(d time.Duration) WorkerManagerOption

type WorkflowContext

type WorkflowContext interface {
	Logger() *slog.Logger
	Replaying() bool
	Now() time.Time
	NewSelector() Selector
	// Sleep blocks the workflow until the duration elapses. Returns ErrCanceled
	// if the workflow is canceled while sleeping. Cancellation is cooperative:
	// the caller chooses whether to propagate the error or ignore it. If the
	// underlying context is canceled (e.g. shard fence loss),
	// Sleep panics with terminateError so the workflow round terminates
	// without persisting completion.
	Sleep(d time.Duration) error
	NewTimer(d time.Duration) TimerFuture
	WorkflowKey() WorkflowKey
	// contains filtered or unexported methods
}

type WorkflowExecutionResult

type WorkflowExecutionResult string

WorkflowExecutionResult is the typed outcome of a single workflow round. Three values (Yield, Terminated, Aborted) are autocore-only. The other three (Completed, Failed, Canceled) are derived from persistence.WorkflowExecutionState.String() so the observability and persisted-state vocabularies stay in lockstep.

const (
	WorkflowExecutionResultYield      WorkflowExecutionResult = "yield"
	WorkflowExecutionResultTerminated WorkflowExecutionResult = "terminated"
	// WorkflowExecutionResultAborted: round produced no persisted outcome.
	// Recorded via the workflow aborts counter, not the round duration
	// histogram.
	WorkflowExecutionResultAborted WorkflowExecutionResult = "aborted"
)

type WorkflowID

type WorkflowID = persistence.WorkflowID

func NewWorkflowID added in v19.2.0

func NewWorkflowID() WorkflowID

NewWorkflowID generates a new unique WorkflowID.

func StartChildWorkflow

func StartChildWorkflow[I, O proto.Message, N WorkflowName[I, O]](ctx WorkflowContext, name N, input I) (WorkflowID, error)

StartChildWorkflow schedules a fire-and-forget child workflow on the parent's shard. The child runs independently; parent terminal state has no effect, and no completion event is ever delivered to the parent. Returns the deterministic child WorkflowID immediately.

type WorkflowInfo

type WorkflowInfo struct {
	WorkflowKey WorkflowKey
	Name        string
	IdentityKey string
	State       WorkflowState
	CreatedAt   time.Time
	UpdatedAt   time.Time
	Output      []byte
	Error       string
}

WorkflowInfo is a read-only snapshot of a workflow's execution state and terminal result, returned by Client.GetWorkflowInfo.

type WorkflowKey

type WorkflowKey struct {
	ShardID    ShardID
	WorkflowID WorkflowID
}

WorkflowKey uniquely addresses a workflow across all databases. The ShardID provides the routing key (shard_id → db_id via central DB).

func ParseWorkflowKey

func ParseWorkflowKey(s string) (WorkflowKey, error)

ParseWorkflowKey is the inverse of WorkflowKey.String: it accepts strings of the form "wk:<shard_id>/<workflow_id>" and returns ErrInvalidWorkflowKey for anything else.

func (WorkflowKey) String added in v19.2.0

func (wk WorkflowKey) String() string

type WorkflowName

type WorkflowName[I, O proto.Message] string

type WorkflowState

type WorkflowState = persistence.WorkflowExecutionState

WorkflowState is the lifecycle state of a workflow execution.

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL