Documentation
¶
Index ¶
- Constants
- Variables
- func ExecuteInlineActivity[I, O proto.Message, N ActivityName[I, O]](ctx WorkflowContext, name N, input I, opts ...ExecuteInlineActivityOption) (O, error)
- func RegisterActivity[I, O proto.Message, N ActivityName[I, O]](r *Registry, name N, fn func(ctx ActivityContext, input I) (O, error), ...)
- func RegisterWorkflow[I, O proto.Message, N WorkflowName[I, O]](r *Registry, name N, fn func(ctx WorkflowContext, input I) (O, error), ...)
- type AbortCause
- type AbortError
- type ActivityContext
- type ActivityExecutionDisposition
- type ActivityExecutionResult
- type ActivityName
- type ActivityPollPredicate
- type CancelWorkflowOptions
- type Case
- type Channel
- type ChildFuture
- type Client
- type Config
- type Engine
- type ExecuteActivityOption
- type ExecuteInlineActivityOption
- type ExecuteWorkflowOptions
- type ExecuteWorkflowResult
- type ExternalNamespaceID
- type Future
- type IsReadier
- type ReadinessAnnouncer
- type ReadinessTracker
- type ReadinessTrackerOption
- type RegisterActivityOption
- type RegisterWorkflowOption
- type Registry
- type RetryPolicy
- type Selector
- type SendToWorkflowOptions
- type ShardID
- type SignalID
- type SignalPromotionContext
- type SignalType
- type TimerFuture
- type TypedWorkflowInfo
- type WorkerManager
- func (wm *WorkerManager) AddShard(ctx context.Context, shardID persistence.ShardID, fenceID persistence.FenceID)
- func (wm *WorkerManager) DrainShard(shardID persistence.ShardID)
- func (wm *WorkerManager) InFlightByShard() map[persistence.ShardID]int
- func (wm *WorkerManager) NotifySignalReady() bool
- func (wm *WorkerManager) NotifyWorkflowTaskReady() bool
- func (wm *WorkerManager) RemoveShard(ctx context.Context, shardID persistence.ShardID, cause error)
- func (wm *WorkerManager) Run(ctx context.Context)
- type WorkerManagerOption
- func WithWMActivityPollInterval(d time.Duration) WorkerManagerOption
- func WithWMMaxChildrenPerParent(n int) WorkerManagerOption
- func WithWMMinimumTick(d time.Duration) WorkerManagerOption
- func WithWMRecoveryBatchSize(n int64) WorkerManagerOption
- func WithWMShardScanConcurrency(n int) WorkerManagerOption
- func WithWMSignalScanInterval(d time.Duration) WorkerManagerOption
- func WithWMSuspensionTTL(d time.Duration) WorkerManagerOption
- func WithWMTimerScanInterval(d time.Duration) WorkerManagerOption
- func WithWMWorkflowPollInterval(d time.Duration) WorkerManagerOption
- type WorkflowContext
- type WorkflowExecutionResult
- type WorkflowID
- type WorkflowInfo
- type WorkflowKey
- type WorkflowName
- type WorkflowState
Constants ¶
const ( SignalTypeUnknown = persistence.SignalTypeUnknown SignalTypeCancelWorkflow = persistence.SignalTypeCancelWorkflow SignalTypeChannel = persistence.SignalTypeChannel SignalTypeChildTerminal = persistence.SignalTypeChildTerminal )
const ( WorkflowStateUnknown = persistence.WorkflowExecutionStateUnknown WorkflowStateRunning = persistence.WorkflowExecutionStateRunning WorkflowStateCompleted = persistence.WorkflowExecutionStateCompleted WorkflowStateFailed = persistence.WorkflowExecutionStateFailed WorkflowStateCanceled = persistence.WorkflowExecutionStateCanceled WorkflowStateTimedOut = persistence.WorkflowExecutionStateTimedOut WorkflowStateSystemFailed = persistence.WorkflowExecutionStateSystemFailed )
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 ¶
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") )
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()) )
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 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 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 Channel ¶ added in v19.2.0
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 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 ¶
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) 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
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
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 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 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 WorkflowState ¶
type WorkflowState = persistence.WorkflowExecutionState
WorkflowState is the lifecycle state of a workflow execution.
Source Files
¶
- abort_cause.go
- activity_context.go
- activity_execution_result.go
- activity_worker_pool.go
- api.go
- api_activity.go
- api_workflow.go
- channel.go
- client.go
- db_manager.go
- db_manager_with_workers.go
- db_manager_without_workers.go
- engine.go
- future.go
- local_notifier.go
- otel.go
- partition_manager.go
- poll.go
- readiness_tracker.go
- redis_notifier.go
- redis_shard_subscriber.go
- registry.go
- replay_aware_log_handler.go
- selector.go
- timer_hint_tracker.go
- worker_manager.go
- worker_manager_children.go
- worker_manager_noop.go
- worker_manager_tasks.go
- workflow_context.go
- workflow_db_manager.go
- workflow_db_worker.go
- workflow_execution_result.go
- workflow_worker_pool.go