workflow

package
v1.32.0 Latest Latest
Warning

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

 Go to latest Published: Sep 11, 2026 License: MIT Imports: 42 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrWorkflowIDNotSet = serviceerror.NewInvalidArgument("WorkflowId is not set on request.")

	ErrCronAndStartDelaySet             = serviceerror.NewInvalidArgument("CronSchedule and WorkflowStartDelay may not be used together.")
	ErrInvalidWorkflowStartDelaySeconds = serviceerror.NewInvalidArgument("An invalid WorkflowStartDelaySeconds is set on request.")
)
View Source 
var ErrCommandNotSupported = errors.New("command not supported")

ErrCommandNotSupported is returned by a CommandHandler when the command type is registered but not supported; for example, because of a disabled feature flag.

View Source 
var ErrCommandTargetNotFound = errors.New("command target not found in chasm tree")

ErrCommandTargetNotFound is returned by a CommandHandler when the command type is supported but the entity it targets is not owned by the CHASM tree (for example, a Nexus operation that lives in the HSM tree or no longer exists). The dispatcher falls back to the HSM command handler so the command follows the tree that owns the entity.

View Source 
var ErrDuplicateRegistration = errors.New("duplicate registration")

ErrDuplicateRegistration is returned by a Registry when it detects duplicate registration.

View Source 
var ErrEventNotCherryPickable = errors.New("event not cherry pickable")

ErrEventNotCherryPickable should be returned by CherryPick if an event should not be cherry picked for whatever reason.

View Source 
var ErrSignalWithStartOperationDisabled = serviceerror.NewUnimplemented("SignalWithStart operation is disabled")
View Source 
var HistoryHandlerModule = fx.Invoke(func(library *library, historyHandler historyservice.HistoryServiceServer) {
	library.workflowServiceNexusHandler.setHistoryHandler(historyHandler)
})

HistoryHandlerModule wires the workflow library's Nexus handler to the history service. Only include this in services that provide historyservice.HistoryServiceServer (the history service).

View Source 
var Module = fx.Module(
	"chasm.lib.workflow",
	fx.Provide(NewConfig),
	fx.Provide(NewRegistry),
	fx.Provide(newLibrary),
	fx.Invoke(func(
		chasmRegistry *chasm.Registry,
		library *library,
		config *nexusoperation.Config,
	) error {
		if err := library.registry.Register(
			newNexusLibrary(config, chasmRegistry.NexusEndpointProcessor),
		); err != nil {
			return err
		}
		return chasmRegistry.Register(library)
	}),
)

Functions

func NewLibrary added in v1.30.0 

func NewLibrary(registry *Registry) chasm.Library

NewLibrary creates a new CHASM library for the workflow package. Use newLibrary (via fx) for the full setup including Nexus services.

func NewWorkflowServiceNexusServiceProcessor 

func NewWorkflowServiceNexusServiceProcessor(
	config Config,
	saMapperProvider searchattribute.MapperProvider,
	saValidator *searchattribute.Validator,
) *chasm.NexusServiceProcessor

func SetEventRegistryOnContext 

func SetEventRegistryOnContext[C chasm.Context](ctx C, registry *Registry) C

SetEventRegistryOnContext injects the event registry into a CHASM context. This is primarily useful for tests that construct MockMutableContext directly.

Types

type CancelRequestCompletedEventDefinition 

type CancelRequestCompletedEventDefinition struct {
}

CancelRequestCompletedEventDefinition handles the NexusOperationCancelRequestCompleted history event.

func (CancelRequestCompletedEventDefinition) Apply 

func (d CancelRequestCompletedEventDefinition) Apply(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent) error

func (CancelRequestCompletedEventDefinition) CherryPick 

func (d CancelRequestCompletedEventDefinition) CherryPick(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent, excludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error

func (CancelRequestCompletedEventDefinition) IsWorkflowTaskTrigger 

func (d CancelRequestCompletedEventDefinition) IsWorkflowTaskTrigger() bool

func (CancelRequestCompletedEventDefinition) Type 

func (d CancelRequestCompletedEventDefinition) Type() enumspb.EventType

type CancelRequestFailedEventDefinition 

type CancelRequestFailedEventDefinition struct {
}

CancelRequestFailedEventDefinition handles the NexusOperationCancelRequestFailed history event.

func (CancelRequestFailedEventDefinition) Apply 

func (d CancelRequestFailedEventDefinition) Apply(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent) error

func (CancelRequestFailedEventDefinition) CherryPick 

func (d CancelRequestFailedEventDefinition) CherryPick(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent, excludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error

func (CancelRequestFailedEventDefinition) IsWorkflowTaskTrigger 

func (d CancelRequestFailedEventDefinition) IsWorkflowTaskTrigger() bool

func (CancelRequestFailedEventDefinition) Type 

func (d CancelRequestFailedEventDefinition) Type() enumspb.EventType

type CancelRequestedEventDefinition 

type CancelRequestedEventDefinition struct {
}

CancelRequestedEventDefinition handles the NexusOperationCancelRequested history event.

func (CancelRequestedEventDefinition) Apply 

func (d CancelRequestedEventDefinition) Apply(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent) error

func (CancelRequestedEventDefinition) CherryPick 

func (d CancelRequestedEventDefinition) CherryPick(_ chasm.MutableContext, _ *Workflow, _ *historypb.HistoryEvent, _ map[enumspb.ResetReapplyExcludeType]struct{}) error

func (CancelRequestedEventDefinition) IsWorkflowTaskTrigger 

func (d CancelRequestedEventDefinition) IsWorkflowTaskTrigger() bool

func (CancelRequestedEventDefinition) Type 

func (d CancelRequestedEventDefinition) Type() enumspb.EventType

type CanceledEventDefinition 

type CanceledEventDefinition struct {
}

CanceledEventDefinition handles the NexusOperationCanceled history event.

func (CanceledEventDefinition) Apply 

func (d CanceledEventDefinition) Apply(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent) error

func (CanceledEventDefinition) CherryPick 

func (d CanceledEventDefinition) CherryPick(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent, excludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error

func (CanceledEventDefinition) IsWorkflowTaskTrigger 

func (d CanceledEventDefinition) IsWorkflowTaskTrigger() bool

func (CanceledEventDefinition) Type 

func (d CanceledEventDefinition) Type() enumspb.EventType

type CommandHandler 

type CommandHandler func(
	chasmCtx chasm.MutableContext,
	wf *Workflow,
	validator Validator,
	command *commandpb.Command,
	opts CommandHandlerOptions,
) error

CommandHandler is a function for handling a workflow command as part of processing a RespondWorkflowTaskCompleted worker request.

type CommandHandlerOptions 

type CommandHandlerOptions struct {
	WorkflowTaskCompletedEventID int64
}

type CompletedEventDefinition 

type CompletedEventDefinition struct {
}

CompletedEventDefinition handles the NexusOperationCompleted history event.

func (CompletedEventDefinition) Apply 

func (d CompletedEventDefinition) Apply(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent) error

func (CompletedEventDefinition) CherryPick 

func (d CompletedEventDefinition) CherryPick(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent, excludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error

func (CompletedEventDefinition) IsWorkflowTaskTrigger 

func (d CompletedEventDefinition) IsWorkflowTaskTrigger() bool

func (CompletedEventDefinition) Type 

func (d CompletedEventDefinition) Type() enumspb.EventType

type Config 

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

func NewConfig 

func NewConfig(dc *dynamicconfig.Collection) Config

type EventDefinition 

type EventDefinition interface {
	Type() enumspb.EventType
	// IsWorkflowTaskTrigger returns a boolean indicating whether this event type should trigger a workflow task.
	IsWorkflowTaskTrigger() bool
	// Apply a history event to the state machine. Triggered during replication and workflow reset.
	Apply(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent) error
	// CherryPick (a.k.a "reapply") an event from a different history branch.
	// Implementations should apply the event to the machine state and return nil in case the event is cherry-pickable.
	// Command events should never be cherry picked as we rely on the workflow to reschedule them.
	// Return [ErrEventNotCherryPickable] to skip cherry picking. Any other error is considered fatal and will abort the
	// cherry pick process.
	CherryPick(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent, resetReapplyExcludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error
}

EventDefinition is a definition for a history event for a given event type.

type FailWorkflowTaskError 

type FailWorkflowTaskError struct {
	Cause             enumspb.WorkflowTaskFailedCause
	Message           string
	TerminateWorkflow bool
}

FailWorkflowTaskError is an error that can be returned from a CommandHandler to fail the current workflow task and optionally terminate the entire workflow.

func (FailWorkflowTaskError) Error 

func (e FailWorkflowTaskError) Error() string

type FailedEventDefinition 

type FailedEventDefinition struct {
}

FailedEventDefinition handles the NexusOperationFailed history event.

func (FailedEventDefinition) Apply 

func (d FailedEventDefinition) Apply(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent) error

func (FailedEventDefinition) CherryPick 

func (d FailedEventDefinition) CherryPick(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent, excludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error

func (FailedEventDefinition) IsWorkflowTaskTrigger 

func (d FailedEventDefinition) IsWorkflowTaskTrigger() bool

func (FailedEventDefinition) Type 

func (d FailedEventDefinition) Type() enumspb.EventType

type Library added in v1.30.0 

type Library interface {
	CommandHandlers() map[enumspb.CommandType]CommandHandler
	EventDefinitions() []EventDefinition
}

Library is an interface for registering command handlers and event definitions with a Registry.

type Registry 

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

Registry maintains a the following mappings for a workflow: CommandType -> Handler EventType -> EventDefinition

func NewRegistry 

func NewRegistry() *Registry

NewRegistry creates a new Registry.

func (*Registry) CommandHandler 

func (r *Registry) CommandHandler(t enumspb.CommandType) (handler CommandHandler, ok bool)

CommandHandler returns a CommandHandler for a given command type and a boolean indicating whether it was found.

func (*Registry) EventDefinitionByEventType 

func (r *Registry) EventDefinitionByEventType(t enumspb.EventType) (EventDefinition, bool)

EventDefinitionByEventType returns an EventDefinition for a given event type and a boolean indicating whether it was found.

func (*Registry) Register 

func (r *Registry) Register(lib Library) error

Register registers all command handlers and event definitions from a Library. Returns an ErrDuplicateRegistration if a handler or definition is already registered. All registration is expected to happen in a single thread on process initialization.

type RequestValidator 

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

func NewValidator 

func NewValidator(
	config Config,
	saMapperProvider searchattribute.MapperProvider,
	saValidator *searchattribute.Validator,
) *RequestValidator

func (*RequestValidator) UnaliasedSearchAttributesFrom 

func (v *RequestValidator) UnaliasedSearchAttributesFrom(
	attributes *commonpb.SearchAttributes,
	namespaceName string,
) (*commonpb.SearchAttributes, error)

func (*RequestValidator) ValidateRetryPolicy 

func (v *RequestValidator) ValidateRetryPolicy(namespaceName string, retryPolicy *commonpb.RetryPolicy) error

func (*RequestValidator) ValidateSearchAttributes 

func (v *RequestValidator) ValidateSearchAttributes(searchAttributes *commonpb.SearchAttributes, namespaceName string) error

func (*RequestValidator) ValidateSignalWithStartRequest 

func (v *RequestValidator) ValidateSignalWithStartRequest(request *workflowservice.SignalWithStartWorkflowExecutionRequest) error

func (*RequestValidator) ValidateWorkflowID 

func (v *RequestValidator) ValidateWorkflowID(
	workflowID string,
) error

func (*RequestValidator) ValidateWorkflowIDReusePolicy 

func (v *RequestValidator) ValidateWorkflowIDReusePolicy(
	reusePolicy enumspb.WorkflowIdReusePolicy,
	conflictPolicy enumspb.WorkflowIdConflictPolicy,
) error

func (*RequestValidator) ValidateWorkflowStartDelay 

func (v *RequestValidator) ValidateWorkflowStartDelay(
	cronSchedule string,
	startDelay *durationpb.Duration,
) error

func (*RequestValidator) ValidateWorkflowTimeouts 

func (v *RequestValidator) ValidateWorkflowTimeouts(
	request StartWorkflowTimeoutLikeRequest,
) error

type ScheduledEventDefinition 

type ScheduledEventDefinition struct{}

ScheduledEventDefinition handles the NexusOperationScheduled history event.

func (ScheduledEventDefinition) Apply 

func (d ScheduledEventDefinition) Apply(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent) error

func (ScheduledEventDefinition) CherryPick 

func (d ScheduledEventDefinition) CherryPick(_ chasm.MutableContext, _ *Workflow, _ *historypb.HistoryEvent, _ map[enumspb.ResetReapplyExcludeType]struct{}) error

func (ScheduledEventDefinition) IsWorkflowTaskTrigger 

func (d ScheduledEventDefinition) IsWorkflowTaskTrigger() bool

func (ScheduledEventDefinition) Type 

func (d ScheduledEventDefinition) Type() enumspb.EventType

type SignalWithStartOperationProcessor 

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

func (SignalWithStartOperationProcessor) ProcessInput 

func (o SignalWithStartOperationProcessor) ProcessInput(ctx chasm.NexusOperationProcessorContext, request *workflowservice.SignalWithStartWorkflowExecutionRequest) (*chasm.NexusOperationProcessorResult, error)

type StartWorkflowTimeoutLikeRequest 

type StartWorkflowTimeoutLikeRequest interface {
	GetWorkflowExecutionTimeout() *durationpb.Duration
	GetWorkflowRunTimeout() *durationpb.Duration
	GetWorkflowTaskTimeout() *durationpb.Duration
}

type StartedEventDefinition 

type StartedEventDefinition struct {
}

StartedEventDefinition handles the NexusOperationStarted history event.

func (StartedEventDefinition) Apply 

func (d StartedEventDefinition) Apply(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent) error

func (StartedEventDefinition) CherryPick 

func (d StartedEventDefinition) CherryPick(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent, excludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error

func (StartedEventDefinition) IsWorkflowTaskTrigger 

func (d StartedEventDefinition) IsWorkflowTaskTrigger() bool

func (StartedEventDefinition) Type 

func (d StartedEventDefinition) Type() enumspb.EventType

type TimedOutEventDefinition 

type TimedOutEventDefinition struct {
}

TimedOutEventDefinition handles the NexusOperationTimedOut history event.

func (TimedOutEventDefinition) Apply 

func (d TimedOutEventDefinition) Apply(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent) error

func (TimedOutEventDefinition) CherryPick 

func (d TimedOutEventDefinition) CherryPick(ctx chasm.MutableContext, wf *Workflow, event *historypb.HistoryEvent, excludeTypes map[enumspb.ResetReapplyExcludeType]struct{}) error

func (TimedOutEventDefinition) IsWorkflowTaskTrigger 

func (d TimedOutEventDefinition) IsWorkflowTaskTrigger() bool

func (TimedOutEventDefinition) Type 

func (d TimedOutEventDefinition) Type() enumspb.EventType

type Validator 

type Validator interface {
	// IsValidPayloadSize validates that a payload size is within the configured limits.
	IsValidPayloadSize(size int) bool
}

Validator is a helper for validating workflow commands.

type Workflow added in v1.30.0 

type Workflow struct {
	chasm.UnimplementedComponent

	// For now, workflow state is managed by mutable_state_impl, not CHASM engine, leaving it empty as CHASM expects a
	// state object.
	*emptypb.Empty

	// MSPointer is a special in-memory field for accessing the underlying mutable state.
	chasm.MSPointer

	// Callbacks map is used to store the callbacks for the workflow.
	Callbacks chasm.Map[string, *callback.Callback]

	// Operations map is used to store the Nexus operations for the workflow, keyed by scheduled event ID.
	Operations chasm.Map[int64, *nexusoperation.Operation]

	// IncomingSignals map is used to track incoming signals, keyed by request ID,
	// to allow DescribeWorkflow to resolve RequestIDRef signal backlinks.
	IncomingSignals chasm.Map[string, *chasmworkflowpb.IncomingSignalData]

	// Updates indexed by update ID, used to store the update components.
	Updates chasm.Map[string, *WorkflowUpdate]
}

func NewWorkflow added in v1.30.0 

func NewWorkflow(
	_ chasm.MutableContext,
	msPointer chasm.MSPointer,
) *Workflow

func (*Workflow) AddCompletionCallbacks added in v1.30.0 

func (w *Workflow) AddCompletionCallbacks(
	ctx chasm.MutableContext,
	eventTime *timestamppb.Timestamp,
	requestID string,
	completionCallbacks []*commonpb.Callback,
	maxCallbacksPerWorkflow int,
) error

AddCompletionCallbacks creates completion callbacks using the CHASM implementation. maxCallbacksPerWorkflow is the configured maximum number of callbacks allowed per workflow.

func (*Workflow) AddIncomingSignalEvent 

func (w *Workflow) AddIncomingSignalEvent(
	ctx chasm.MutableContext,
	requestID string,
	eventID int64,
) error

AddIncomingSignalEvent adds an entry for the signal requestID -> eventID mapping to track all signals that have been received by the workflow. Note that since signals are buffered, the eventID may the common.BufferedEventID, which will be updated to a concrete eventID once this signal is flushed to the DB. If caller tries to add an already-existing eventID, this function will ignore and silently return instead of overwriting -- use UpdateIncomingSignalEvent to update existing entries.

func (*Workflow) AddUpdateCompletionCallbacks 

func (w *Workflow) AddUpdateCompletionCallbacks(
	ctx chasm.MutableContext,
	eventTime *timestamppb.Timestamp,
	updateID string,
	requestID string,
	completionCallbacks []*commonpb.Callback,
	maxCallbacksPerWorkflow int,
	maxCallbacksPerUpdateID int,
) error

AddUpdateCompletionCallbacks creates completion callbacks using the CHASM implementation. maxCallbacksPerWorkflow is the configured maximum number of callbacks allowed per workflow. maxCallbacksPerUpdateID is the configured maximum number of callbacks allowed per update ID.

func (*Workflow) BuildPendingNexusOperationInfos 

func (w *Workflow) BuildPendingNexusOperationInfos(
	ctx chasm.Context,
	circuitBreaker func(endpoint string) bool,
) ([]*workflowpb.PendingNexusOperationInfo, error)

BuildPendingNexusOperationInfos reads nexus operations from the workflow and converts them to API format.

func (*Workflow) ContextMetadata added in v1.31.0 

func (w *Workflow) ContextMetadata(_ chasm.Context) map[string]string

func (*Workflow) GetNexusCompletion added in v1.30.0 

func (w *Workflow) GetNexusCompletion(
	ctx chasm.Context,
	requestID string,
) (nexusrpc.CompleteOperationOptions, error)

func (*Workflow) HasAnyBufferedEvent 

func (w *Workflow) HasAnyBufferedEvent(filter historybuilder.BufferedEventFilter) bool

HasAnyBufferedEvent returns true if the workflow has any buffered event matching the given filter.

func (*Workflow) HasIncomingSignalEvent 

func (w *Workflow) HasIncomingSignalEvent(_ chasm.Context, requestID string) bool

HasIncomingSignalEvent returns true if a signal with this requestID is already persisted in this CHASM tree.

func (*Workflow) LifecycleState added in v1.30.0 

func (w *Workflow) LifecycleState(
	_ chasm.Context,
) chasm.LifecycleState

func (*Workflow) NexusOperationInvocationData 

func (w *Workflow) NexusOperationInvocationData(
	ctx chasm.Context,
	op *nexusoperation.Operation,
) (nexusoperation.InvocationData, error)

NexusOperationInvocationData loads invocation data from the scheduled history event.

func (*Workflow) OnNexusOperationCanceled 

func (w *Workflow) OnNexusOperationCanceled(
	ctx chasm.MutableContext,
	op *nexusoperation.Operation,
	cause *failurepb.Failure,
) error

OnNexusOperationCanceled adds a NexusOperationCanceled history event to the workflow and applies the corresponding event definition.

func (*Workflow) OnNexusOperationCancellationCompleted 

func (w *Workflow) OnNexusOperationCancellationCompleted(ctx chasm.MutableContext, op *nexusoperation.Operation) error

func (*Workflow) OnNexusOperationCancellationFailed 

func (w *Workflow) OnNexusOperationCancellationFailed(ctx chasm.MutableContext, op *nexusoperation.Operation, failure *failurepb.Failure) error

func (*Workflow) OnNexusOperationCompleted 

func (w *Workflow) OnNexusOperationCompleted(
	ctx chasm.MutableContext,
	op *nexusoperation.Operation,
	result *commonpb.Payload,
	links []*commonpb.Link,
) error

OnNexusOperationCompleted adds a NexusOperationCompleted history event to the workflow and applies the corresponding event definition.

func (*Workflow) OnNexusOperationFailed 

func (w *Workflow) OnNexusOperationFailed(
	ctx chasm.MutableContext,
	op *nexusoperation.Operation,
	cause *failurepb.Failure,
) error

OnNexusOperationFailed adds a NexusOperationFailed history event to the workflow and applies the corresponding event definition.

func (*Workflow) OnNexusOperationStarted 

func (w *Workflow) OnNexusOperationStarted(
	ctx chasm.MutableContext,
	op *nexusoperation.Operation,
	operationToken string,
	startTime *time.Time,
	links []*commonpb.Link,
) error

OnNexusOperationStarted adds a NexusOperationStarted history event to the workflow and applies the corresponding event definition.

func (*Workflow) OnNexusOperationTimedOut 

func (w *Workflow) OnNexusOperationTimedOut(
	ctx chasm.MutableContext,
	op *nexusoperation.Operation,
	cause *failurepb.Failure,
	_ bool,
) error

OnNexusOperationTimedOut adds a NexusOperationTimedOut history event to the workflow and applies the corresponding event definition.

func (*Workflow) ProcessAllUpdateCloseCallbacks 

func (w *Workflow) ProcessAllUpdateCloseCallbacks(ctx chasm.MutableContext) error

ProcessAllUpdateCloseCallbacks triggers callbacks for all updates without touching workflow-level callbacks. This is used when the workflow is continuing to a new run (ContinueAsNew, retry, cron): workflow-level callbacks are inherited by the new run, but update callbacks must fire now because the update was aborted on the old run.

func (*Workflow) ProcessCloseCallbacks added in v1.30.0 

func (w *Workflow) ProcessCloseCallbacks(ctx chasm.MutableContext) error

ProcessCloseCallbacks triggers "WorkflowClosed" callbacks using the CHASM implementation. It schedules all workflow-level and update-level callbacks that are in STANDBY state.

func (*Workflow) ProcessUpdateCallbacks 

func (w *Workflow) ProcessUpdateCallbacks(ctx chasm.MutableContext, updateID string) error

ProcessUpdateCallbacks triggers callbacks for a single updateID if exists.

func (*Workflow) RejectUpdate 

func (w *Workflow) RejectUpdate(ctx chasm.MutableContext, updateID string, rejectionFailure *failurepb.Failure) error

RejectUpdate stores the rejection failure on the WorkflowUpdate component and fires any pending callbacks. This is used when a reapplied update (after reset) is rejected by the worker's validator - the callbacks need to deliver the rejection failure to the caller.

func (*Workflow) Terminate added in v1.31.0 

func (w *Workflow) Terminate(
	_ chasm.MutableContext,
	_ chasm.TerminateComponentRequest,
) (chasm.TerminateComponentResponse, error)

func (*Workflow) UpdateIncomingSignalEvent 

func (w *Workflow) UpdateIncomingSignalEvent(
	ctx chasm.MutableContext,
	requestID string,
	eventID int64,
) error

UpdateIncomingSignalEvent updates the eventID for an existing signal requestID in the map. If the requestID is not in the map, this is a no-op (e.g. when called for non-signal request IDs during buffer flush).

func (*Workflow) WorkflowTypeName 

func (w *Workflow) WorkflowTypeName() string

type WorkflowUpdate 

type WorkflowUpdate struct {
	chasm.UnimplementedComponent

	*workflowpb.UpdateState

	// MSPointer is a special in-memory field for accessing the underlying mutable state.
	chasm.MSPointer

	// Callbacks map is used to store the callbacks for the update.
	Callbacks chasm.Map[string, *callback.Callback]
}

func NewWorkflowUpdate 

func NewWorkflowUpdate(
	_ chasm.MutableContext, updateID string, msPointer chasm.MSPointer,
) *WorkflowUpdate

func (*WorkflowUpdate) GetNexusCompletion 

func (u *WorkflowUpdate) GetNexusCompletion(
	ctx chasm.Context,
	requestID string,
) (nexusrpc.CompleteOperationOptions, error)

func (*WorkflowUpdate) LifecycleState 

func (u *WorkflowUpdate) LifecycleState(
	_ chasm.Context,
) chasm.LifecycleState

Source Files

View all Source files

Directories

Path Synopsis
gen
workflowpb/v1
Code generated by protoc-gen-go-helpers.
 Code generated by protoc-gen-go-helpers.

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.