task

package
v1.0.0-beta2 Latest Latest
Warning

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

 Go to latest Published: Sep 29, 2026 License: Apache-2.0 Imports: 26 Imported by: 12

Documentation

Overview

Package task implements deterministic orchestrator, activity, and durable entity execution for Durable Task Scheduler workers.

Index

Constants

View Source 
const DefaultMaximumTimerInterval = 3 * 24 * time.Hour

DefaultMaximumTimerInterval is the default maximum duration of one physical durable timer action.

View Source 
const UnversionedTaskVersion = ""

UnversionedTaskVersion explicitly selects an unversioned task.

Variables

View Source 
var ErrHistoryLimitExceeded = errors.New("orchestration history limit exceeded")

ErrHistoryLimitExceeded is the sentinel error for deterministic history-budget failures.

View Source 
var ErrTaskBlocked = errors.New("the current task is blocked")

ErrTaskBlocked is not an error, but rather a control flow signal indicating that an orchestrator function has executed as far as it can and that it now needs to unload, dispatch any scheduled tasks, and commit its current execution progress to durable storage. Durable operations also panic with this signal when called by deferred code during forced coroutine unloading, before creating or modifying durable work.

View Source 
var ErrTaskCanceled = errors.New("the task was canceled") // CONSIDER: More specific info about the task

ErrTaskCanceled is used to indicate that a task was canceled. Tasks can be canceled, for example, when configured timeouts expire.

Functions

func LoggerFromContext 

func LoggerFromContext(ctx context.Context) *slog.Logger

LoggerFromContext returns the task logger associated with an activity or entity context, or slog.Default. Orchestrators must use OrchestrationContext.Logger; their replay-stable Context excludes loggers.

func WithEntityInput 

func WithEntityInput(input any) callEntityOption

WithEntityInput configures an input for an entity operation invocation.

func WithRawEntityInput 

func WithRawEntityInput(input string) callEntityOption

WithRawEntityInput configures a raw input for an entity operation invocation.

func WithRawSignalEntityInput 

func WithRawSignalEntityInput(input string) signalEntityOption

WithRawSignalEntityInput configures a raw input for a signal entity invocation.

func WithSignalEntityInput 

func WithSignalEntityInput(input any) signalEntityOption

WithSignalEntityInput configures an input for a signal entity invocation.

func WithSignalEntityScheduledTime 

func WithSignalEntityScheduledTime(scheduledTime time.Time) signalEntityOption

WithSignalEntityScheduledTime configures a scheduled time for an entity signal.

Types

type Activity 

type Activity func(ctx ActivityContext) (any, error)

Activity is the functional interface for activity implementations.

type ActivityContext 

type ActivityContext interface {
	GetInput(resultPtr any) error
	Context() context.Context
}

ActivityContext is the context parameter type for activity implementations.

type CallActivityOption 

type CallActivityOption func(*callActivityOptions, api.DataConverter) error

CallActivityOption configures an activity invocation.

func WithActivityInput added in v0.3.0 

func WithActivityInput(input any) CallActivityOption

WithActivityInput configures an input for an activity invocation. The configured data converter must be able to serialize the input.

func WithActivityRetryPolicy added in v0.6.0 

func WithActivityRetryPolicy(policy *RetryPolicy) CallActivityOption

WithActivityRetryPolicy snapshots policy when this option is created. Later caller mutations do not affect activity retries.

func WithActivityTags 

func WithActivityTags(tags map[string]string) CallActivityOption

WithActivityTags adds user tags to an activity invocation. Explicit activity tags override orchestration tags with the same key.

func WithActivityVersion 

func WithActivityVersion(version string) CallActivityOption

WithActivityVersion configures the activity version.

func WithRawActivityInput added in v0.3.0 

func WithRawActivityInput(input string) CallActivityOption

WithRawActivityInput configures a raw input for an activity invocation.

type ContinueAsNewOption added in v0.2.1 

type ContinueAsNewOption func(*OrchestrationContext)

ContinueAsNewOption is a functional option type for the ContinueAsNew orchestrator method.

func WithContinueAsNewVersion 

func WithContinueAsNewVersion(version string) ContinueAsNewOption

WithContinueAsNewVersion migrates the next execution to a new orchestration version.

func WithKeepUnprocessedEvents added in v0.2.1 

func WithKeepUnprocessedEvents() ContinueAsNewOption

WithKeepUnprocessedEvents instructs the runtime to carry forward external events that this execution did not deliver. That includes events that arrive after the orchestrator function returns, even if a wait for the same event name is still pending. An event that completed a wait before the function returned was delivered, even if the wait was never awaited.

type Entity 

type Entity func(ctx *EntityContext) (any, error)

Entity is the functional interface for entity implementations. An entity function receives an EntityContext and returns a result and error.

func NewEntityFor 

func NewEntityFor[S any]() Entity

NewEntityFor creates an entity function that dispatches operations to methods on a converter-serializable state struct of type S.

Supported method parameters are an optional *EntityContext, an optional api.EntityID, and at most one required input value or OptionalEntityInput, in any order. Supported return shapes are (), (result), (error), and (result, error).

type EntityBatch 

type EntityBatch struct {
	Entity Entity
	Close  func(context.Context) error
}

EntityBatch contains one batch-scoped entity implementation and optional cleanup.

type EntityBatchCloser 

type EntityBatchCloser interface {
	CloseEntityBatch(context.Context) error
}

EntityBatchCloser can release object resources after an entity batch completes.

type EntityContext 

type EntityContext struct {
	ID        api.EntityID
	Operation string
	RequestID string
	IsSignal  bool
	// contains filtered or unexported fields
}

EntityContext provides the execution context for an entity operation.

func (*EntityContext) Context 

func (ctx *EntityContext) Context() context.Context

Context returns the Go context for the current entity operation.

func (*EntityContext) CurrentTimeUTC 

func (ctx *EntityContext) CurrentTimeUTC() time.Time

CurrentTimeUTC returns the durable timestamp associated with this operation.

func (*EntityContext) DeleteState 

func (ctx *EntityContext) DeleteState()

DeleteState removes the entity state.

func (*EntityContext) GetInput 

func (ctx *EntityContext) GetInput(v any) error

GetInput unmarshals the serialized entity operation input and saves the result into [v].

func (*EntityContext) GetRawState 

func (ctx *EntityContext) GetRawState() (string, bool)

GetRawState returns the entity's pre-serialized state and whether state is set. Callers are responsible for deserializing the returned value.

func (*EntityContext) GetState 

func (ctx *EntityContext) GetState(v any) error

GetState unmarshals the entity state and saves the result into [v].

func (*EntityContext) HasInput 

func (ctx *EntityContext) HasInput() bool

HasInput reports whether the caller supplied an operation input.

func (*EntityContext) HasState 

func (ctx *EntityContext) HasState() bool

HasState returns true if the entity has state set.

func (*EntityContext) Logger 

func (ctx *EntityContext) Logger() *slog.Logger

Logger returns a logger enriched with entity operation identity.

func (*EntityContext) SerializeInput 

func (ctx *EntityContext) SerializeInput(value any) (string, error)

SerializeInput serializes a value using the entity's configured data converter. It is useful when an entity needs to persist input for a later operation.

func (*EntityContext) SetRawState 

func (ctx *EntityContext) SetRawState(state string)

SetRawState stores pre-serialized entity state. Callers are responsible for producing data accepted by all readers of the entity state.

func (*EntityContext) SetState 

func (ctx *EntityContext) SetState(state any) error

SetState serializes and stores entity state with the configured converter. Passing nil deletes the entity state.

func (*EntityContext) SignalEntity 

func (ctx *EntityContext) SignalEntity(entityID api.EntityID, operationName string, input any) error

SignalEntity sends a fire-and-forget signal to another entity.

func (*EntityContext) SignalEntityAt 

func (ctx *EntityContext) SignalEntityAt(entityID api.EntityID, scheduledTime time.Time, operationName string, input any) error

SignalEntityAt schedules a fire-and-forget signal to another entity.

func (*EntityContext) StartNewOrchestration 

func (ctx *EntityContext) StartNewOrchestration(name string, opts ...EntityStartOrchestrationOption) error

StartNewOrchestration schedules a new orchestration from within an entity operation.

type EntityExecutor 

type EntityExecutor interface {
	ExecuteEntity(context.Context, *protos.EntityBatchRequest) (*protos.EntityBatchResult, error)
}

EntityExecutor processes normalized operation batches from V2 entity work items. EntityBatchRequest is an internal execution model, not a supported legacy work-item envelope.

type EntityFactory 

type EntityFactory func(EntityFactoryContext) (EntityBatch, error)

EntityFactory creates one entity implementation for an operation batch. Factories may run concurrently for different entity batches.

func NewEntityObjectFactory 

func NewEntityObjectFactory[S any, E EntityObjectBinding[S]](
	factory func(EntityFactoryContext) (E, error),
	configure ...EntityObjectOption[S],
) EntityFactory

NewEntityObjectFactory creates one persistent entity object per batch while storing only S as durable state.

Operation binding is validated when the factory is created, so a malformed entity object fails at startup instead of failing every batch at runtime.

type EntityFactoryContext 

type EntityFactoryContext struct {
	Context context.Context
	ID      api.EntityID
}

EntityFactoryContext identifies the entity batch for which an implementation is created.

type EntityObjectBase 

type EntityObjectBase[S any] struct {
	// contains filtered or unexported fields
}

EntityObjectBase binds a persistent entity object to its current state and operation context. Embed it in an entity object used with NewEntityObjectFactory.

func (*EntityObjectBase[S]) Context 

func (base *EntityObjectBase[S]) Context() *EntityContext

Context returns the context for the current operation.

func (*EntityObjectBase[S]) State 

func (base *EntityObjectBase[S]) State() *S

State returns the mutable state for the current operation.

type EntityObjectBinding 

type EntityObjectBinding[S any] interface {
	// contains filtered or unexported methods
}

EntityObjectBinding is implemented by entity objects that embed EntityObjectBase.

type EntityObjectOption 

type EntityObjectOption[S any] func(*entityObjectOptions[S])

EntityObjectOption configures reflected entity-object dispatch.

func WithEntityStateDispatch 

func WithEntityStateDispatch[S any]() EntityObjectOption[S]

WithEntityStateDispatch enables reflected operation dispatch to the state after no matching entity-object method is found.

func WithEntityStateInitializer 

func WithEntityStateInitializer[S any](
	initializer func(*EntityContext) (S, error),
) EntityObjectOption[S]

WithEntityStateInitializer configures state creation when an operation starts without state.

type EntityOperationFailedError 

type EntityOperationFailedError struct {
	EntityID       api.EntityID
	OperationName  string
	FailureDetails *api.FailureDetails
}

EntityOperationFailedError is returned when a called entity operation fails.

func (*EntityOperationFailedError) DurableTaskErrorType 

func (*EntityOperationFailedError) DurableTaskErrorType() api.ErrorType

func (*EntityOperationFailedError) DurableTaskFailureDetails 

func (e *EntityOperationFailedError) DurableTaskFailureDetails() *api.FailureDetails

func (*EntityOperationFailedError) Error 

func (e *EntityOperationFailedError) Error() string

func (*EntityOperationFailedError) Is 

func (e *EntityOperationFailedError) Is(target error) bool

func (*EntityOperationFailedError) NonRetriable 

func (e *EntityOperationFailedError) NonRetriable() bool

type EntityStartOrchestrationOption 

type EntityStartOrchestrationOption func(*entityStartOrchestrationOptions, api.DataConverter) error

EntityStartOrchestrationOption is a functional option for StartNewOrchestration.

func WithEntityStartOrchestrationInput 

func WithEntityStartOrchestrationInput(input any) EntityStartOrchestrationOption

WithEntityStartOrchestrationInput sets the input for the new orchestration.

func WithEntityStartOrchestrationInstanceID 

func WithEntityStartOrchestrationInstanceID(instanceID string) EntityStartOrchestrationOption

WithEntityStartOrchestrationInstanceID sets the instance ID for the new orchestration.

func WithEntityStartOrchestrationScheduledTime 

func WithEntityStartOrchestrationScheduledTime(scheduledTime time.Time) EntityStartOrchestrationOption

WithEntityStartOrchestrationScheduledTime schedules the new orchestration.

func WithEntityStartOrchestrationVersion 

func WithEntityStartOrchestrationVersion(version string) EntityStartOrchestrationOption

WithEntityStartOrchestrationVersion sets the version for the new orchestration.

func WithRawEntityStartOrchestrationInput 

func WithRawEntityStartOrchestrationInput(input string) EntityStartOrchestrationOption

WithRawEntityStartOrchestrationInput sets pre-serialized input for the new orchestration. Callers are responsible for ensuring the payload is accepted by the target orchestration's data converter.

type EventChannel 

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

EventChannel receives a named external event repeatedly while preserving durable event order. Use ReceiveErr or TryReceiveErr when payload errors should be handled by the orchestrator.

func NewEventChannel 

func NewEventChannel[T any](ctx *OrchestrationContext, name string) *EventChannel[T]

NewEventChannel returns the durable channel for a named external event. Repeated calls using the same event name and value type return the same channel.

func (*EventChannel[T]) Receive 

func (c *EventChannel[T]) Receive(ctx *OrchestrationContext) T

Receive waits for and consumes the next event value.

func (*EventChannel[T]) ReceiveErr 

func (c *EventChannel[T]) ReceiveErr(ctx *OrchestrationContext) (T, error)

ReceiveErr waits for and consumes the next event value, returning payload decoding and cancellation errors to the orchestrator.

func (*EventChannel[T]) TryReceive 

func (c *EventChannel[T]) TryReceive() (T, bool)

TryReceive consumes a buffered event without blocking.

func (*EventChannel[T]) TryReceiveErr 

func (c *EventChannel[T]) TryReceiveErr() (T, bool, error)

TryReceiveErr consumes a buffered event without blocking and returns payload decoding errors.

type ExecutionResults 

type ExecutionResults struct {
	Response *protos.OrchestratorResponse
}

ExecutionResults carries the orchestrator actions produced by a single orchestrator execution turn. It is transport-neutral: the DTS worker sends ExecutionResults.Response back to the scheduler over gRPC.

type Executor 

type Executor interface {
	ExecuteOrchestrator(
		ctx context.Context,
		iid api.InstanceID,
		oldEvents []*protos.HistoryEvent,
		newEvents []*protos.HistoryEvent,
		entityParameters *protos.OrchestratorEntityParameters,
	) (*ExecutionResults, error)
	ExecuteActivity(context.Context, api.InstanceID, *protos.HistoryEvent) (*protos.HistoryEvent, error)
	Shutdown(ctx context.Context) error
}

Executor is the internal collaboration contract between the task runtime and the gRPC worker. It is exported only so those packages can share it; its internal protobuf parameters intentionally prevent application implementations. NewTaskExecutor is the implementation used by DTS workers.

func NewTaskExecutor 

func NewTaskExecutor(registry *TaskRegistry, opts ...TaskExecutorOption) Executor

NewTaskExecutor returns a Executor implementation that executes orchestrator and activity functions in-memory.

type HistoryLimitError 

type HistoryLimitError struct {
	InstanceID       api.InstanceID
	HistoryLength    int
	MaxHistoryEvents int
	PolicyError      error
}

HistoryLimitError describes the history budget that was exceeded.

func (*HistoryLimitError) DurableTaskErrorType 

func (*HistoryLimitError) DurableTaskErrorType() api.ErrorType

func (*HistoryLimitError) Error 

func (e *HistoryLimitError) Error() string

func (*HistoryLimitError) NonRetriable 

func (*HistoryLimitError) NonRetriable() bool

func (*HistoryLimitError) Unwrap 

func (e *HistoryLimitError) Unwrap() error

type HistoryLimitHandler 

type HistoryLimitHandler func(HistoryLimitInfo) (any, error)

HistoryLimitHandler supplies safe input for a ContinueAsNew transition. It must be deterministic. Unconsumed external events are preserved automatically.

type HistoryLimitInfo 

type HistoryLimitInfo struct {
	InstanceID            api.InstanceID
	OrchestrationName     string
	OrchestrationVersion  string
	HistoryLength         int
	MaxHistoryEvents      int
	UnprocessedEventCount int
	SerializedInput       string
	Converter             api.DataConverter
}

HistoryLimitInfo is the immutable input passed to a history limit handler.

func (HistoryLimitInfo) GetInput 

func (info HistoryLimitInfo) GetInput(v any) error

GetInput unmarshals the current orchestration input.

type HistoryMetric 

type HistoryMetric struct {
	InstanceID           api.InstanceID
	OrchestrationName    string
	OrchestrationVersion string
	HistoryLength        int
	ProcessedEvents      int
	HistoryLimitExceeded bool
}

HistoryMetric reports orchestration history usage for one execution turn.

type MetricsHooks 

type MetricsHooks struct {
	Retry   func(RetryMetric)
	History func(HistoryMetric)
}

MetricsHooks contains optional transport-neutral metric callbacks. Callbacks must return quickly and must not block worker progress.

type OptionalEntityInput 

type OptionalEntityInput[T any] struct {
	Value   T
	Present bool
}

OptionalEntityInput binds an entity method input that may be omitted by the caller.

func (OptionalEntityInput[T]) Or 

func (input OptionalEntityInput[T]) Or(defaultValue T) T

Or returns the input value when present, or defaultValue when absent.

type OrchestrationContext 

type OrchestrationContext struct {
	ID             api.InstanceID
	Name           string
	Version        string
	IsReplaying    bool
	CurrentTimeUtc time.Time
	// contains filtered or unexported fields
}

OrchestrationContext is the parameter type for orchestrator functions.

Deferred functions can issue and await durable work when their coroutine returns normally. When the runtime forcibly unloads a waiting coroutine, durable operations instead panic with ErrTaskBlocked before changing state. Applications must not recover that control-flow signal.

func (*OrchestrationContext) CallActivity 

func (ctx *OrchestrationContext) CallActivity(activity any, opts ...CallActivityOption) Task

CallActivity schedules an asynchronous invocation of an activity function. The [activity] parameter can be either the name of an activity as a string or can be a pointer to the function that implements the activity, in which case the name is obtained via reflection.

func (*OrchestrationContext) CallEntity 

func (ctx *OrchestrationContext) CallEntity(entityID api.EntityID, operationName string, opts ...callEntityOption) Task

CallEntity sends an operation request to an entity and waits for its response.

func (*OrchestrationContext) CallSubOrchestrator added in v0.3.0 

func (ctx *OrchestrationContext) CallSubOrchestrator(orchestrator any, opts ...SubOrchestratorOption) Task

func (*OrchestrationContext) Context 

func (ctx *OrchestrationContext) Context() context.Context

Context returns a replay-stable Go context containing only persisted orchestration identity and caller fields. Host values, deadlines, and cancellation are intentionally excluded from deterministic orchestrator code.

func (*OrchestrationContext) ContinueAsNew 

func (ctx *OrchestrationContext) ContinueAsNew(newInput any, options ...ContinueAsNewOption)

ContinueAsNew requests a new execution when the root coroutine finishes.

func (*OrchestrationContext) CreateTimer 

func (ctx *OrchestrationContext) CreateTimer(delay time.Duration) Task

CreateTimer schedules a durable timer that expires after the specified delay.

func (*OrchestrationContext) GetInput 

func (octx *OrchestrationContext) GetInput(v any) error

GetInput unmarshals the serialized orchestration input and stores it in [v].

func (*OrchestrationContext) Go 

func (ctx *OrchestrationContext) Go(fn func(ctx *OrchestrationContext))

Go starts a coroutine that is cooperatively scheduled with the orchestration. Only one orchestration coroutine runs at a time, in monotonically increasing ID order. A callback whose scope is canceled before it starts is not invoked. Join children explicitly when their cleanup must finish before the root returns.

func (*OrchestrationContext) HistoryLength 

func (ctx *OrchestrationContext) HistoryLength() int

HistoryLength returns the total old and new history supplied to this execution.

func (*OrchestrationContext) HistoryLimitExceeded 

func (ctx *OrchestrationContext) HistoryLimitExceeded() bool

HistoryLimitExceeded reports whether MaxHistoryEvents was exceeded.

func (*OrchestrationContext) IsInCriticalSection 

func (ctx *OrchestrationContext) IsInCriticalSection() bool

IsInCriticalSection reports whether the orchestration currently holds entity locks.

func (*OrchestrationContext) LockEntities 

func (ctx *OrchestrationContext) LockEntities(entityIDs ...api.EntityID) (func(), error)

LockEntities acquires an ordered critical section over a set of entities. If cancellation follows a committed request, critical-section restrictions remain in effect until the eventual grant is received and automatically released.

func (*OrchestrationContext) Logger 

func (ctx *OrchestrationContext) Logger() *slog.Logger

Logger returns a slog logger that suppresses output while replaying history and while waiting coroutines are forcibly unloaded.

func (*OrchestrationContext) NewGuid 

func (ctx *OrchestrationContext) NewGuid() string

NewGuid returns a deterministic UUID that is stable across orchestration replay.

func (*OrchestrationContext) NewWaitGroup 

func (ctx *OrchestrationContext) NewWaitGroup() WaitGroup

NewWaitGroup creates a deterministic coroutine wait group.

func (*OrchestrationContext) Select 

func (ctx *OrchestrationContext) Select(cases ...SelectCase)

Select waits until one case is ready and invokes its handler. All cases must belong to the same orchestration, including its child scopes.

func (*OrchestrationContext) SendEvent 

func (ctx *OrchestrationContext) SendEvent(instanceID api.InstanceID, eventName string, payload any) error

SendEvent sends an event to another orchestration instance as part of the current durable orchestration transaction.

func (*OrchestrationContext) SetCustomStatus added in v0.6.0 

func (octx *OrchestrationContext) SetCustomStatus(cs string)

SetCustomStatus stores a raw, pre-serialized custom status string. Use SetCustomStatusValue to apply the configured data converter.

func (*OrchestrationContext) SetCustomStatusValue 

func (octx *OrchestrationContext) SetCustomStatusValue(value any) error

SetCustomStatusValue serializes and stores a typed custom status value.

func (*OrchestrationContext) SetRawCustomStatus 

func (octx *OrchestrationContext) SetRawCustomStatus(payload string)

SetRawCustomStatus stores a pre-serialized custom status value.

func (*OrchestrationContext) SignalEntity 

func (ctx *OrchestrationContext) SignalEntity(entityID api.EntityID, operationName string, opts ...signalEntityOption) error

SignalEntity sends a fire-and-forget entity operation.

func (*OrchestrationContext) WaitForSingleEvent 

func (ctx *OrchestrationContext) WaitForSingleEvent(eventName string, timeout time.Duration) Task

WaitForSingleEvent creates a task that is completed only after an event named [eventName] is received by this orchestration or when the specified timeout expires.

The [timeout] parameter can be used to define a timeout for receiving the event. If the timeout expires before the named event is received, the task will be completed and will return a timeout error value ErrTaskCanceled when awaited. Otherwise, the awaited task will return the deserialized payload of the received event. A Duration value of zero returns a canceled task if the event isn't already available in the history. Use a negative Duration to wait indefinitely for the event to be received.

Orchestrators can wait for the same event name multiple times, so waiting for multiple events with the same name is allowed. Each event received by an orchestrator will complete just one task returned by this method. Live waiters use LIFO ordering, so the newest waiter receives the next event. Events buffered before any waiter exists retain FIFO arrival order. This ordering is part of the deterministic replay contract.

Note that event names are case-insensitive.

func (*OrchestrationContext) WhenAll 

func (ctx *OrchestrationContext) WhenAll(tasks ...Task) error

WhenAll blocks until every task completes and returns the earliest failure, ordered by orchestration history.

func (*OrchestrationContext) WhenAny 

func (ctx *OrchestrationContext) WhenAny(tasks ...Task) Task

WhenAny blocks until the first task completes and returns that task. If multiple tasks are ready, the task completed first in orchestration history wins.

func (*OrchestrationContext) WithCancel 

func (ctx *OrchestrationContext) WithCancel() (*OrchestrationContext, func())

WithCancel creates a child orchestration context whose tasks, nested scopes, and coroutines are canceled together at the next scheduler step. A child of an already-canceled scope is canceled immediately. Await cleanup from an uncanceled wrapper, cancel only its body operations, and join the wrapper before returning.

type OrchestrationOptions 

type OrchestrationOptions struct {
	// MaximumTimerInterval limits one physical durable timer action. Longer
	// timers are split into deterministic sequential actions that retain the
	// original deadline. Zero uses [DefaultMaximumTimerInterval].
	//
	// Upgrading from a release that did not split long timers is replay-compatible
	// once the historical logical deadline has fired. Changing this value between
	// two splitting configurations remains replay-breaking for in-flight
	// orchestrations that have timers longer than either the old or new value.
	MaximumTimerInterval time.Duration

	// MaxEventsPerTurn limits new DTS work-item events processed in one
	// execution turn. Orchestration control markers are outside DTS's processed
	// event count. Remaining events are left for DTS to redeliver. Zero
	// disables the limit. Old replay history is always processed.
	MaxEventsPerTurn int

	// MaxHistoryEvents limits the total old and new history supplied to an execution.
	// Zero disables the limit.
	MaxHistoryEvents int

	// OnHistoryLimitExceeded handles MaxHistoryEvents when the orchestration
	// remains incomplete after the turn. It opts into ContinueAsNew by supplying
	// serializable state; when nil, the orchestration fails with a non-retriable
	// HistoryLimitExceeded failure.
	OnHistoryLimitExceeded HistoryLimitHandler
}

OrchestrationOptions configures deterministic orchestration engine policies.

type Orchestrator 

type Orchestrator func(ctx *OrchestrationContext) (any, error)

Orchestrator is the functional interface for orchestrator functions.

type OrchestratorNotFoundStrategy 

type OrchestratorNotFoundStrategy int

OrchestratorNotFoundStrategy determines whether missing orchestrators fail or reject work items. 

const (
	// OrchestratorNotFoundFail completes the orchestration with an OrchestratorTaskNotFound failure.
	OrchestratorNotFoundFail OrchestratorNotFoundStrategy = iota
	// OrchestratorNotFoundReject abandons the work item so another worker or deployment can process it.
	OrchestratorNotFoundReject
)

type RetryContext 

type RetryContext struct {
	LastAttemptNumber int
	// LastFailure is never nil while Handle is running and must be treated as read-only.
	LastFailure    *api.FailureDetails
	TotalRetryTime time.Duration
}

RetryContext contains durable inputs for a retry decision. Retry handlers execute during replay and must not perform I/O or depend on wall-clock time.

type RetryMetric 

type RetryMetric struct {
	InstanceID           api.InstanceID
	OrchestrationName    string
	OrchestrationVersion string
	TaskKind             WorkItemKind
	TaskName             string
	TaskVersion          string
	FailedAttempt        int
	NextAttempt          int
	MaxAttempts          int
	Delay                time.Duration
	ErrorType            string
	ErrorMessage         string
}

RetryMetric reports a durable task retry that was scheduled.

type RetryPolicy added in v0.6.0 

type RetryPolicy struct {
	// Max number of attempts to try the activity call, first execution inclusive
	MaxAttempts int
	// Timespan to wait for the first retry
	InitialRetryInterval time.Duration
	// Used to determine rate of increase of back-off
	BackoffCoefficient float64
	// Max timespan to wait for a retry
	MaxRetryInterval time.Duration
	// Total timeout across all the retries performed
	RetryTimeout time.Duration
	// Optional deterministic function that controls whether retries should proceed.
	Handle func(RetryContext) bool
}

func (*RetryPolicy) Normalized 

func (policy *RetryPolicy) Normalized() (RetryPolicy, error)

Normalized validates the retry policy and returns an independent copy with default values applied. The receiver is not modified.

func (*RetryPolicy) Validate added in v0.6.0 

func (policy *RetryPolicy) Validate() error

Validate reports whether the retry policy is valid without modifying it.

type SelectCase 

type SelectCase interface {
	// contains filtered or unexported methods
}

SelectCase is a durable selection case created by OnTask or OnEvent.

func OnEvent 

func OnEvent[T any](channel *EventChannel[T], handler func(T)) SelectCase

OnEvent creates a Select case that receives from a durable event channel.

func OnTask 

func OnTask(task Task, handler func(Task)) SelectCase

OnTask creates a Select case that becomes ready when task completes.

type SubOrchestratorOption 

type SubOrchestratorOption func(*callSubOrchestratorOptions, api.DataConverter) error

SubOrchestratorOption configures OrchestrationContext.CallSubOrchestrator.

func WithRawSubOrchestratorInput added in v0.3.0 

func WithRawSubOrchestratorInput(input string) SubOrchestratorOption

WithRawSubOrchestratorInput is a functional option type for the CallSubOrchestrator orchestrator method that takes a raw input value.

func WithSubOrchestrationContextFields 

func WithSubOrchestrationContextFields(fields api.ContextFields) SubOrchestratorOption

WithSubOrchestrationContextFields adds immutable context fields to a sub-orchestration.

func WithSubOrchestrationInstanceID added in v0.3.0 

func WithSubOrchestrationInstanceID(instanceID string) SubOrchestratorOption

WithSubOrchestrationInstanceID is a functional option type for the CallSubOrchestrator orchestrator method that specifies the instance ID of the sub-orchestration.

func WithSubOrchestrationRetryPolicy added in v0.6.0 

func WithSubOrchestrationRetryPolicy(policy *RetryPolicy) SubOrchestratorOption

WithSubOrchestrationRetryPolicy snapshots policy when this option is created. Later caller mutations do not affect sub-orchestration retries.

func WithSubOrchestrationTags 

func WithSubOrchestrationTags(tags map[string]string) SubOrchestratorOption

WithSubOrchestrationTags adds user tags to a sub-orchestration.

func WithSubOrchestrationVersion 

func WithSubOrchestrationVersion(version string) SubOrchestratorOption

WithSubOrchestrationVersion configures the sub-orchestration version.

func WithSubOrchestratorInput added in v0.3.0 

func WithSubOrchestratorInput(input any) SubOrchestratorOption

WithSubOrchestratorInput is a functional option type for the CallSubOrchestrator orchestrator method that serializes an input value with the configured converter.

type Task 

type Task interface {
	Await(v any) error
}

Task is an interface for asynchronous durable tasks. A task is conceptually similar to a future.

type TaskExecutorOption 

type TaskExecutorOption func(*taskExecutor)

TaskExecutorOption configures the in-memory task executor.

func WithContextFields 

func WithContextFields(fields api.ContextFields) TaskExecutorOption

WithContextFields configures worker-local fields for activity and entity contexts. Root-orchestration fields must use api.WithContextFields, and child fields must use WithSubOrchestrationContextFields, so they are persisted and remain stable across replay and worker deployments.

func WithDataConverter 

func WithDataConverter(converter api.DataConverter) TaskExecutorOption

WithDataConverter configures application payload serialization.

func WithErrorPropertiesProvider 

func WithErrorPropertiesProvider(provider api.ErrorPropertiesProvider) TaskExecutorOption

WithErrorPropertiesProvider configures custom durable failure properties.

func WithLogger 

func WithLogger(logger *slog.Logger) TaskExecutorOption

WithLogger configures the slog logger exposed to orchestration and activity code.

func WithMaximumTimerInterval 

func WithMaximumTimerInterval(interval time.Duration) TaskExecutorOption

WithMaximumTimerInterval configures the maximum duration of one physical durable timer action. Zero restores DefaultMaximumTimerInterval. A negative interval panics when the option is applied.

func WithMetricsHooks 

func WithMetricsHooks(hooks MetricsHooks) TaskExecutorOption

WithMetricsHooks configures optional transport-neutral metric callbacks.

func WithOrchestrationOptions 

func WithOrchestrationOptions(options OrchestrationOptions) TaskExecutorOption

WithOrchestrationOptions configures deterministic orchestration engine policies. It panics when a duration or event limit is negative.

func WithOrchestratorNotFoundStrategy 

func WithOrchestratorNotFoundStrategy(strategy OrchestratorNotFoundStrategy) TaskExecutorOption

WithOrchestratorNotFoundStrategy configures missing-orchestrator handling.

func WithUnversionedActivityNames 

func WithUnversionedActivityNames(names ...string) TaskExecutorOption

WithUnversionedActivityNames allows named system activities to bypass worker version matching when their work item explicitly selects no version. System orchestrations run unversioned, and an activity inherits its caller's version, so their activities are dispatched unversioned as well.

func WithUnversionedOrchestratorNames 

func WithUnversionedOrchestratorNames(names ...string) TaskExecutorOption

WithUnversionedOrchestratorNames allows named system orchestrators to bypass worker version matching when their work item explicitly selects no version.

func WithVersioning 

func WithVersioning(options VersioningOptions) TaskExecutorOption

WithVersioning configures version-aware orchestration and activity dispatch.

type TaskFailedError 

type TaskFailedError struct {
	TaskName       string
	TaskVersion    string
	TaskID         int32
	FailureDetails *api.FailureDetails
}

TaskFailedError is returned when an activity or sub-orchestration fails.

func (*TaskFailedError) DurableTaskErrorType 

func (*TaskFailedError) DurableTaskErrorType() api.ErrorType

func (*TaskFailedError) DurableTaskFailureDetails 

func (e *TaskFailedError) DurableTaskFailureDetails() *api.FailureDetails

func (*TaskFailedError) Error 

func (e *TaskFailedError) Error() string

func (*TaskFailedError) Is 

func (e *TaskFailedError) Is(target error) bool

func (*TaskFailedError) NonRetriable 

func (e *TaskFailedError) NonRetriable() bool

type TaskRegistration 

type TaskRegistration struct {
	Name    string
	Version string
}

TaskRegistration identifies one registered task implementation.

type TaskRegistry 

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

TaskRegistry contains registered orchestrator, activity, and entity functions.

func NewTaskRegistry 

func NewTaskRegistry() *TaskRegistry

NewTaskRegistry returns a new TaskRegistry struct.

func (*TaskRegistry) AddActivity 

func (r *TaskRegistry) AddActivity(a Activity) error

AddActivity adds an activity function to the registry. The name of the activity function is determined using reflection.

func (*TaskRegistry) AddActivityN 

func (r *TaskRegistry) AddActivityN(name string, a Activity) error

AddActivityN adds an activity function to the registry with a specified name.

func (*TaskRegistry) AddActivityNVersion 

func (r *TaskRegistry) AddActivityNVersion(name, version string, a Activity) error

AddActivityNVersion adds an activity with an explicit name and version.

func (*TaskRegistry) AddActivityVersion 

func (r *TaskRegistry) AddActivityVersion(version string, a Activity) error

AddActivityVersion adds a versioned activity whose name is determined using reflection.

func (*TaskRegistry) AddEntity 

func (r *TaskRegistry) AddEntity(e Entity) error

AddEntity adds an entity function to the registry. The name of the entity function is determined using reflection.

func (*TaskRegistry) AddEntityFactory 

func (r *TaskRegistry) AddEntityFactory(factory EntityFactory) error

AddEntityFactory adds an entity factory whose name is determined using reflection.

func (*TaskRegistry) AddEntityFactoryN 

func (r *TaskRegistry) AddEntityFactoryN(name string, factory EntityFactory) error

AddEntityFactoryN adds a named factory that creates one entity implementation per batch.

func (*TaskRegistry) AddEntityN 

func (r *TaskRegistry) AddEntityN(name string, e Entity) error

AddEntityN adds a shared entity function with a specified name. The function can be invoked concurrently by different entity batches and must be thread-safe.

func (*TaskRegistry) AddOrchestrator 

func (r *TaskRegistry) AddOrchestrator(o Orchestrator) error

AddOrchestrator adds an orchestrator function to the registry. The name of the orchestrator function is determined using reflection.

func (*TaskRegistry) AddOrchestratorN 

func (r *TaskRegistry) AddOrchestratorN(name string, o Orchestrator) error

AddOrchestratorN adds an orchestrator function to the registry with a specified name.

func (*TaskRegistry) AddOrchestratorNVersion 

func (r *TaskRegistry) AddOrchestratorNVersion(name, version string, o Orchestrator) error

AddOrchestratorNVersion adds an orchestrator with an explicit name and version.

func (*TaskRegistry) AddOrchestratorVersion 

func (r *TaskRegistry) AddOrchestratorVersion(version string, o Orchestrator) error

AddOrchestratorVersion adds a versioned orchestrator whose name is determined using reflection.

func (*TaskRegistry) Snapshot 

func (r *TaskRegistry) Snapshot() TaskRegistrySnapshot

Snapshot returns a deterministic, immutable view of the registry.

type TaskRegistrySnapshot 

type TaskRegistrySnapshot struct {
	Orchestrators []TaskRegistration
	Activities    []TaskRegistration
	Entities      []string
}

TaskRegistrySnapshot is an immutable view of the tasks registered in a registry.

type VersionFailureStrategy 

type VersionFailureStrategy int

VersionFailureStrategy determines how a worker handles a version mismatch. 

const (
	VersionFailureReject VersionFailureStrategy = iota
	VersionFailureFail
)

type VersionMatchStrategy 

type VersionMatchStrategy int

VersionMatchStrategy determines which task versions a worker accepts. 

const (
	VersionMatchNone VersionMatchStrategy = iota
	VersionMatchStrict
	VersionMatchCurrentOrOlder
)

type VersionMismatchError 

type VersionMismatchError struct {
	TaskVersion   string
	WorkerVersion string
	Strategy      VersionMatchStrategy
}

VersionMismatchError indicates that a work item is incompatible with this worker.

func (*VersionMismatchError) DurableTaskErrorType 

func (*VersionMismatchError) DurableTaskErrorType() api.ErrorType

func (*VersionMismatchError) Error 

func (e *VersionMismatchError) Error() string

func (*VersionMismatchError) Is 

func (*VersionMismatchError) Is(target error) bool

func (*VersionMismatchError) NonRetriable 

func (*VersionMismatchError) NonRetriable() bool

func (*VersionMismatchError) WorkItemAbandonDelay 

func (*VersionMismatchError) WorkItemAbandonDelay() time.Duration

WorkItemAbandonDelay asks the worker to delay redelivery of a version-incompatible activity work item instead of retrying it immediately.

type VersioningOptions 

type VersioningOptions struct {
	Version         string
	DefaultVersion  string
	MatchStrategy   VersionMatchStrategy
	FailureStrategy VersionFailureStrategy
}

VersioningOptions configures version-aware orchestration and activity dispatch.

func (VersioningOptions) Validate 

func (o VersioningOptions) Validate() error

Validate checks that the versioning configuration is supported.

type WaitGroup 

type WaitGroup interface {
	Add(delta int)
	Done()
	Wait(ctx *OrchestrationContext)
}

WaitGroup coordinates orchestration coroutines. It behaves like sync.WaitGroup, but Wait yields to the deterministic orchestration scheduler. Do not pair Add with a Done deferred only inside a coroutine that can be skipped before starting because its scope was canceled. Use an uncanceled wrapper and a separate cancellation scope for its body operations instead.

type WorkItemKind 

type WorkItemKind string

WorkItemKind identifies an orchestration, activity, or entity work item. 

const (
	WorkItemKindOrchestration WorkItemKind = "orchestration"
	WorkItemKindActivity      WorkItemKind = "activity"
	WorkItemKindEntity        WorkItemKind = "entity"
)

Source Files

View all Source files

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.