task

package
v1.1.1 Latest Latest
Warning

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

 Go to latest Published: Sep 9, 2026 License: Apache-2.0 Imports: 22 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

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

ErrTaskBlocked is not an error, but rather a control flow signal indicating that a workflow 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.

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 NewTaskExecutor 

func NewTaskExecutor(registry *TaskRegistry) backend.Executor

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

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
	GetTaskID() int32
	GetTaskExecutionID() string
	Context() context.Context
	GetTraceContext() *protos.TraceContext
	GetPropagatedHistory() *api.PropagatedHistory
}

ActivityContext is the context parameter type for activity implementations.

type CallActivityOption 

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

CallActivityOption is the interface for options passed to CallActivity.

type CallActivityOptionFunc 

type CallActivityOptionFunc func(*callActivityOptions) error

func WithActivityAppID 

func WithActivityAppID(targetAppID string) CallActivityOptionFunc

func WithActivityAppNamespace 

func WithActivityAppNamespace(namespace string) CallActivityOptionFunc

WithActivityAppNamespace specifies the Dapr namespace that hosts the target activity. When set, the routing envelope carries a targetAppNamespace so the caller sidecar performs a durable cross-namespace dispatch (service invocation with per-hop reminders) rather than a direct actor call via placement. Must be combined with WithActivityAppID; setting a namespace without an app ID is rejected when the activity is scheduled. Cross-namespace calls are gated by the WorkflowAccessPolicy feature: a policy on the target side must explicitly permit the caller's (namespace, appID).

func WithActivityInput 

func WithActivityInput(input any) CallActivityOptionFunc

WithActivityInput configures an input for an activity invocation. The specified input must be JSON serializable.

func WithActivityRetryPolicy 

func WithActivityRetryPolicy(policy *RetryPolicy) CallActivityOptionFunc

func WithRawActivityInput 

func WithRawActivityInput(input *wrapperspb.StringValue) CallActivityOptionFunc

WithRawActivityInput configures a raw input for an activity invocation.

type ChildWorkflowOption 

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

ChildWorkflowOption is the interface for options passed to CallChildWorkflow.

func WithRawSubOrchestratorInput deprecated 

func WithRawSubOrchestratorInput(input *wrapperspb.StringValue) ChildWorkflowOption

Deprecated: Use WithRawChildWorkflowInput instead.

func WithSubOrchestrationInstanceID deprecated 

func WithSubOrchestrationInstanceID(instanceID string) ChildWorkflowOption

Deprecated: Use WithChildWorkflowInstanceID instead.

func WithSubOrchestrationRetryPolicy deprecated 

func WithSubOrchestrationRetryPolicy(policy *RetryPolicy) ChildWorkflowOption

Deprecated: Use WithChildWorkflowRetryPolicy instead.

func WithSubOrchestratorAppID deprecated 

func WithSubOrchestratorAppID(appID string) ChildWorkflowOption

Deprecated: Use WithChildWorkflowAppID instead.

func WithSubOrchestratorInput deprecated 

func WithSubOrchestratorInput(input any) ChildWorkflowOption

Deprecated: Use WithChildWorkflowInput instead.

type ChildWorkflowOptionFunc 

type ChildWorkflowOptionFunc func(*callChildWorkflowOptions) error

ChildWorkflowOptionFunc adapts a function to the ChildWorkflowOption interface.

func WithChildWorkflowAppID 

func WithChildWorkflowAppID(appID string) ChildWorkflowOptionFunc

WithChildWorkflowAppID is a functional option type for the CallChildWorkflow workflow method that specifies the app ID of the target activity.

func WithChildWorkflowAppNamespace 

func WithChildWorkflowAppNamespace(namespace string) ChildWorkflowOptionFunc

WithChildWorkflowAppNamespace specifies the Dapr namespace that hosts the target child workflow. When set, the routing envelope carries a targetAppNamespace so that the caller sidecar performs a durable cross-namespace dispatch (service invocation with per-hop reminders) instead of a direct actor call via placement. Must be combined with WithChildWorkflowAppID; setting a namespace without an app ID is rejected when the child workflow is scheduled. Cross-namespace calls are gated by the WorkflowAccessPolicy feature: a policy on the target side must explicitly permit the caller's (namespace, appID).

func WithChildWorkflowInput 

func WithChildWorkflowInput(input any) ChildWorkflowOptionFunc

WithChildWorkflowInput is a functional option type for the CallChildWorkflow workflow method that takes an input value and marshals it to JSON.

func WithChildWorkflowInstanceID 

func WithChildWorkflowInstanceID(instanceID string) ChildWorkflowOptionFunc

WithChildWorkflowInstanceID is a functional option type for the CallChildWorkflow workflow method that specifies the instance ID of the child workflow.

func WithChildWorkflowRetryPolicy 

func WithChildWorkflowRetryPolicy(policy *RetryPolicy) ChildWorkflowOptionFunc

func WithRawChildWorkflowInput 

func WithRawChildWorkflowInput(input *wrapperspb.StringValue) ChildWorkflowOptionFunc

WithRawChildWorkflowInput is a functional option type for the CallChildWorkflow workflow method that takes a raw input value.

type ContinueAsNewOption 

type ContinueAsNewOption func(*WorkflowContext)

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

func WithKeepUnprocessedEvents 

func WithKeepUnprocessedEvents() ContinueAsNewOption

WithKeepUnprocessedEvents returns a ContinueAsNewOptions struct that instructs the runtime to carry forward any unprocessed external events to the new instance.

type CreateTimerOption 

type CreateTimerOption func(*createTimerOptions) error

func WithTimerName 

func WithTimerName(name string) CreateTimerOption

type DetachedWorkflowOptions 

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

DetachedWorkflowOptions is the interface for options passed to ScheduleNewWorkflow. Unlike CallChildWorkflow, the spawned instance is fire-and-forget: the caller receives the instance ID synchronously but does not wait on (and is not notified of) the spawned workflow's completion.

type DetachedWorkflowOptionsFunc 

type DetachedWorkflowOptionsFunc func(*scheduleNewWorkflowOptions) error

DetachedWorkflowOptionsFunc adapts a function to the DetachedWorkflowOptions interface.

func WithDetachedWorkflowAppID 

func WithDetachedWorkflowAppID(appID string) DetachedWorkflowOptionsFunc

WithDetachedWorkflowAppID specifies the app ID hosting the detached workflow. When set, the action carries a routing envelope so the dispatcher delivers the new instance to the target app.

func WithDetachedWorkflowAppNamespace 

func WithDetachedWorkflowAppNamespace(namespace string) DetachedWorkflowOptionsFunc

WithDetachedWorkflowAppNamespace specifies the Dapr namespace hosting the detached workflow. Must be combined with WithDetachedWorkflowAppID; see WithChildWorkflowAppNamespace for full semantics.

func WithDetachedWorkflowInput 

func WithDetachedWorkflowInput(input any) DetachedWorkflowOptionsFunc

WithDetachedWorkflowInput sets the input for the detached workflow, marshalling it to JSON.

func WithDetachedWorkflowInstanceID 

func WithDetachedWorkflowInstanceID(instanceID string) DetachedWorkflowOptionsFunc

WithDetachedWorkflowInstanceID sets the instance ID of the detached workflow. When omitted, ScheduleNewWorkflow generates a deterministic ID of the form "<callerInstanceID>-<n>" where n increments per default-ID spawn within the execution. The '-' separator keeps the generated ID safe for consumers (e.g. dapr) that propagate the instance id into RFC 1123 subdomain identifiers. Passing an empty string is rejected as an error: callers either set a non-empty ID or omit the option entirely to opt into the default.

func WithDetachedWorkflowStartTime 

func WithDetachedWorkflowStartTime(startTime time.Time) DetachedWorkflowOptionsFunc

WithDetachedWorkflowStartTime defers the start of the detached workflow until the given time. Mirrors api.WithStartTime on the client API.

func WithRawDetachedWorkflowInput 

func WithRawDetachedWorkflowInput(input *wrapperspb.StringValue) DetachedWorkflowOptionsFunc

WithRawDetachedWorkflowInput sets a pre-marshalled raw input on the detached workflow.

type HistoryPropagation 

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

HistoryPropagation implements both CallActivityOption and ChildWorkflowOption, allowing WithHistoryPropagation to be used with both CallActivity and CallChildWorkflow.

func WithHistoryPropagation 

func WithHistoryPropagation(opt api.PropagationOption) HistoryPropagation

WithHistoryPropagation creates an option that configures history propagation for a child workflow or activity call. It can be passed to both CallActivity and CallChildWorkflow. Pass one of PropagateOwnHistory() or PropagateLineage().

Usage: 

ctx.CallChildWorkflow("child",
    task.WithChildWorkflowInput(input),
    task.WithHistoryPropagation(task.PropagateLineage()),
)

ctx.CallActivity("activity",
    task.WithActivityInput(input),
    task.WithHistoryPropagation(task.PropagateOwnHistory()),
)

type OrchestrationContext deprecated 

type OrchestrationContext = WorkflowContext

Deprecated: Use WorkflowContext instead.

type Orchestrator deprecated 

type Orchestrator = Workflow

Deprecated: Use Workflow instead.

type RetryPolicy 

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 function to control if retries should proceed
	Handle func(error) bool
}

func (*RetryPolicy) Validate 

func (policy *RetryPolicy) Validate() error

type SubOrchestratorOption deprecated 

type SubOrchestratorOption = ChildWorkflowOption

Deprecated: Use ChildWorkflowOption instead.

type Task 

type Task interface {
	Await(v any) error
	TaskExecutionId() string
}

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

type TaskRegistry 

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

TaskRegistry contains maps of names to corresponding workflow and activity 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. Returns an error if an activity with the same name is already registered.

func (*TaskRegistry) AddOrchestrator deprecated 

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

Deprecated: Use AddWorkflow instead.

func (*TaskRegistry) AddOrchestratorN deprecated 

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

Deprecated: Use AddWorkflowN instead.

func (*TaskRegistry) AddVersionedOrchestrator deprecated 

func (r *TaskRegistry) AddVersionedOrchestrator(canonicalName string, isLatest bool, o Workflow) error

Deprecated: Use AddVersionedWorkflow instead.

func (*TaskRegistry) AddVersionedOrchestratorN deprecated 

func (r *TaskRegistry) AddVersionedOrchestratorN(canonicalName string, name string, isLatest bool, o Workflow) error

Deprecated: Use AddVersionedWorkflowN instead.

func (*TaskRegistry) AddVersionedWorkflow 

func (r *TaskRegistry) AddVersionedWorkflow(canonicalName string, isLatest bool, o Workflow) error

AddVersionedWorkflow adds a versioned workflow function to the registry with a specified name.

func (*TaskRegistry) AddVersionedWorkflowN 

func (r *TaskRegistry) AddVersionedWorkflowN(canonicalName string, name string, isLatest bool, o Workflow) error

AddVersionedWorkflowN adds a versioned workflow function to the registry. Returns an error if a workflow with the same version name is already registered.

func (*TaskRegistry) AddWorkflow 

func (r *TaskRegistry) AddWorkflow(o Workflow) error

AddWorkflow adds a workflow function to the registry. The name of the workflow function is determined using reflection.

func (*TaskRegistry) AddWorkflowN 

func (r *TaskRegistry) AddWorkflowN(name string, o Workflow) error

AddWorkflowN adds a workflow function to the registry with a specified name.

func (*TaskRegistry) RemoveActivity 

func (r *TaskRegistry) RemoveActivity(name string)

RemoveActivity removes an activity from the registry.

func (*TaskRegistry) RemoveVersionedWorkflow 

func (r *TaskRegistry) RemoveVersionedWorkflow(canonicalName string)

RemoveVersionedWorkflow removes all versions of a workflow from the registry.

func (*TaskRegistry) ResolveWorkflow 

func (r *TaskRegistry) ResolveWorkflow(name string, pinnedVersion *string) (Workflow, *string, error)

ResolveWorkflow looks up a workflow by name under a single read lock. It checks: exact match → versioned match → wildcard, returning the first hit. Returns the resolved workflow, an optional version string, or an error.

func (*TaskRegistry) UpsertActivityN 

func (r *TaskRegistry) UpsertActivityN(name string, a Activity)

UpsertActivityN adds or replaces an activity function in the registry. Used by callers that intentionally re-register (e.g. hot-reload).

func (*TaskRegistry) UpsertVersionedWorkflowN 

func (r *TaskRegistry) UpsertVersionedWorkflowN(canonicalName string, name string, isLatest bool, o Workflow)

UpsertVersionedWorkflowN adds or replaces a versioned workflow function in the registry. Used by callers that intentionally re-register (e.g. hot-reload).

type Workflow 

type Workflow func(ctx *WorkflowContext) (any, error)

Workflow is the functional interface for workflow functions.

type WorkflowContext 

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

WorkflowContext is the parameter type for workflow functions.

func NewWorkflowContext 

func NewWorkflowContext(registry *TaskRegistry, id api.InstanceID, oldEvents []*protos.HistoryEvent, newEvents []*protos.HistoryEvent) *WorkflowContext

NewWorkflowContext returns a new WorkflowContext struct with the specified parameters.

func (*WorkflowContext) CallActivity 

func (ctx *WorkflowContext) CallActivity(activity interface{}, 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 (*WorkflowContext) CallChildWorkflow 

func (ctx *WorkflowContext) CallChildWorkflow(workflow interface{}, opts ...ChildWorkflowOption) Task

func (*WorkflowContext) CallSubOrchestrator deprecated 

func (ctx *WorkflowContext) CallSubOrchestrator(workflow interface{}, opts ...ChildWorkflowOption) Task

Deprecated: Use CallChildWorkflow instead.

func (*WorkflowContext) ContinueAsNew 

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

func (*WorkflowContext) CreateTimer 

func (ctx *WorkflowContext) CreateTimer(delay time.Duration, opts ...CreateTimerOption) Task

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

func (*WorkflowContext) GetInput 

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

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

func (*WorkflowContext) GetPropagatedHistory 

func (octx *WorkflowContext) GetPropagatedHistory() *api.PropagatedHistory

GetPropagatedHistory returns the propagated history from a parent workflow, or nil if no history was propagated. The propagated history contains events from the parent (and optionally ancestor) workflows that opted to propagate their execution context.

func (*WorkflowContext) IsPatched 

func (ctx *WorkflowContext) IsPatched(patchName string) bool

func (*WorkflowContext) ScheduleNewWorkflow 

func (ctx *WorkflowContext) ScheduleNewWorkflow(workflow any, opts ...DetachedWorkflowOptions) (api.InstanceID, error)

ScheduleNewWorkflow schedules a new, fully decoupled workflow instance from the calling workflow. Unlike CallChildWorkflow, the spawned workflow has no parent linkage: its history's ExecutionStartedEvent carries no ParentInstanceInfo, completion and failure do not flow back to the caller, and the caller receives the new instance ID synchronously rather than an awaitable Task.

If WithDetachedWorkflowInstanceID is not provided, a deterministic ID of the form "<callerInstanceID>-<n>" is generated, where n increments only for default-ID spawns within this execution (so the suffix reflects the order of those calls and is stable across replays). The '-' separator keeps the generated ID safe for consumers that propagate the instance id into RFC 1123 subdomain identifiers.

If the workflow author needs the spawned workflow's result, they should model the dependency through external events (RaiseEvent / WaitForEvent) or shared state — there is no built-in completion channel.

func (*WorkflowContext) SetCustomStatus 

func (octx *WorkflowContext) SetCustomStatus(cs string)

func (*WorkflowContext) SetPropagatedHistory 

func (octx *WorkflowContext) SetPropagatedHistory(ph *api.PropagatedHistory)

SetPropagatedHistory sets the propagated history on the context.

func (*WorkflowContext) WaitForSingleEvent 

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

WaitForSingleEvent creates a task that is completed only after an event named [eventName] is received by this workflow 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.

Workflows 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 workflow will complete just one task returned by this method.

Note that event names are case-insensitive.

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.