activity

package
v1.32.1-x.2 Latest Latest
Warning

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

 Go to latest Published: Oct 15, 2026 License: MIT Imports: 56 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// WorkflowTypeTag is a required workflow tag for standalone activities to ensure consistent
	// metric labeling between workflows and activities.
	WorkflowTypeTag = "__temporal_standalone_activity__"

	// ByIDTokenAttempt is used in synthesized tokens for by-ID API calls where the caller does not specify the attempt.
	// The validator skips the attempt check when it sees this value.
	// 0 is safe because polled tokens always carry Count >= 1 (TransitionScheduled increments from 0).
	ByIDTokenAttempt int32 = 0
)

Variables

View Source 
var (
	TypeSearchAttribute   = chasm.NewSearchAttributeKeyword("ActivityType", chasm.SearchAttributeFieldKeyword01)
	StatusSearchAttribute = chasm.NewSearchAttributeKeyword("ExecutionStatus", chasm.SearchAttributeFieldLowCardinalityKeyword01)
)
View Source 
var (
	Enabled = dynamicconfig.NewNamespaceBoolSetting(
		"activity.enableStandalone",
		true,
		`Toggles standalone activity functionality on the server.`,
	)

	LongPollTimeout = dynamicconfig.NewNamespaceDurationSetting(
		"activity.longPollTimeout",
		common.DefaultLongPollTimeout,
		`Timeout for activity long-poll requests.`,
	)

	LongPollBuffer = dynamicconfig.NewNamespaceDurationSetting(
		"activity.longPollBuffer",
		common.DefaultLongPollBuffer,
		`A buffer used to adjust the activity long-poll timeouts.
 Specifically, activity long-poll requests are timed out at a time which leaves at least the buffer's duration
 remaining before the caller's deadline, if permitted by the caller's deadline.`,
	)

	StartDelayEnabled = dynamicconfig.NewNamespaceBoolSetting(
		"activity.startDelayEnabled",
		true,
		`Allows non-zero start_delay on StartActivityExecution requests.`,
	)

	EnableCallbacks = dynamicconfig.NewNamespaceBoolSetting(
		"activity.enableCallbacks",
		false,
		`Allows attaching completion callbacks to standalone activity executions.`,
	)

	EnableStandaloneActivityOperatorCommands = dynamicconfig.NewNamespaceBoolSetting(
		"history.enableStandaloneActivityOperatorCommands",
		false,
		`Enables reset, pause, unpause, and update options commands for standalone activities.`,
	)
)
View Source 
var (
	Archetype   = chasm.FullyQualifiedName(libraryName, componentName)
	ArchetypeID = chasm.GenerateTypeID(Archetype)
)
View Source 
var ErrStandaloneActivityDisabled = serviceerror.NewUnimplemented("Standalone activity is disabled")
View Source 
var ErrStandaloneActivityOperatorCommandsDisabled = serviceerror.NewUnimplemented("Standalone activity operator commands are disabled")
View Source 
var FrontendModule = fx.Module(
	"activity-frontend",
	fx.Provide(ConfigProvider),
	fx.Provide(linkValidatorProvider),
	fx.Provide(activitypb.NewActivityServiceLayeredClient),
	fx.Provide(NewFrontendHandler),
	fx.Provide(resource.SearchAttributeValidatorProvider),
	fx.Provide(newComponentOnlyLibrary),
	fx.Invoke(func(l *componentOnlyLibrary, registry *chasm.Registry) error {

		return registry.Register(l)
	}),
)
View Source 
var HistoryModule = fx.Module(
	"activity-history",
	fx.Provide(
		ConfigProvider,
		linkValidatorProvider,
		newActivityDispatchTaskHandler,
		newScheduleToStartTimeoutTaskHandler,
		newScheduleToCloseTimeoutTaskHandler,
		newStartToCloseTimeoutTaskHandler,
		newHeartbeatTimeoutTaskHandler,
		newHandler,
		newLibrary,
	),
	fx.Invoke(func(l *library, registry *chasm.Registry) error {
		return registry.Register(l)
	}),
)
View Source 
var TransitionAttemptFailedWhilePauseRequested = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSE_REQUESTED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED,
	func(a *Activity, ctx chasm.MutableContext, event rescheduleEvent) error {
		return a.applyFailedAttempt(ctx, event)
	},
)

TransitionAttemptFailedWhilePauseRequested transitions PAUSE_REQUESTED → PAUSED. It is performed instead of TransitionReschedule, when the activity is in PAUSE_REQUESTED and the worker yields (failure or timeout) with retries remaining. The failed attempt is recorded and Count is incremented.

View Source 
var TransitionCancelRequested = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_STARTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED,
		activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED,
		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSE_REQUESTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_RESET_REQUESTED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED,
	func(a *Activity, ctx chasm.MutableContext, req *workflowservice.RequestCancelActivityExecutionRequest) error {
		a.CancelState = &activitypb.ActivityCancelState{
			Identity:    req.GetIdentity(),
			RequestId:   req.GetRequestId(),
			Reason:      req.GetReason(),
			RequestTime: timestamppb.New(ctx.Now(a)),
		}

		a.ResetRestoreOptions = false
		a.ResetShouldClearHeartbeat = false

		return nil
	},
)

TransitionCancelRequested transitions to CancelRequested status.

View Source 
var TransitionCanceled = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_CANCELED,
	func(a *Activity, ctx chasm.MutableContext, event cancelEvent) error {
		return a.StoreOrSelf(ctx).RecordCompleted(ctx, func(ctx chasm.MutableContext) error {
			outcome := a.Outcome.Get(ctx)
			failure := &failurepb.Failure{
				Message: "Activity canceled",
				FailureInfo: &failurepb.Failure_CanceledFailureInfo{
					CanceledFailureInfo: &failurepb.CanceledFailureInfo{
						Details:  event.details,
						Identity: a.GetCancelState().GetIdentity(),
					},
				},
			}
			outcome.Variant = &activitypb.ActivityOutcome_Failed_{
				Failed: &activitypb.ActivityOutcome_Failed{
					Failure: failure,
				},
			}

			a.emitOnCanceledMetrics(ctx, event.metricsHandler, event.fromStatus)

			return nil
		})
	},
)

TransitionCanceled transitions to Canceled status.

View Source 
var TransitionCompleted = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED,
		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED,
		activitypb.ACTIVITY_EXECUTION_STATUS_STARTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSE_REQUESTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_RESET_REQUESTED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_COMPLETED,
	func(a *Activity, ctx chasm.MutableContext, event completeEvent) error {
		return a.StoreOrSelf(ctx).RecordCompleted(ctx, func(ctx chasm.MutableContext) error {
			req := event.req.GetCompleteRequest()

			attemptWasStarted := a.hasAttemptInProgress()
			attempt := a.LastAttempt.Get(ctx)
			if !attemptWasStarted {

				attempt.StartedTime = timestamppb.New(ctx.Now(a))
				if a.FirstAttemptStartedTime == nil {
					a.FirstAttemptStartedTime = attempt.StartedTime
				}
			}
			attempt.CompleteTime = timestamppb.New(ctx.Now(a))
			attempt.LastWorkerIdentity = req.GetIdentity()
			outcome := a.Outcome.Get(ctx)
			outcome.Variant = &activitypb.ActivityOutcome_Successful_{
				Successful: &activitypb.ActivityOutcome_Successful{
					Output: req.GetResult(),
				},
			}

			a.emitOnCompletedMetrics(ctx, event.baseHandler, event.enrichedHandler, req.GetResult(), attemptWasStarted)

			return nil
		})
	},
)

TransitionCompleted transitions to Completed status. SCHEDULED and PAUSED are included because RespondActivityTaskCompletedById can force-complete an activity that has no attempt in progress, mirroring workflow-activity behavior.

View Source 
var TransitionFailed = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_STARTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSE_REQUESTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_RESET_REQUESTED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_FAILED,
	func(a *Activity, ctx chasm.MutableContext, event failedEvent) error {
		return a.StoreOrSelf(ctx).RecordCompleted(ctx, func(ctx chasm.MutableContext) error {
			req := event.req.GetFailedRequest()
			a.Outcome.Get(ctx).RetryState = event.retryState

			attempt := a.LastAttempt.Get(ctx)
			attempt.LastWorkerIdentity = req.GetIdentity()

			failure := req.GetFailure()
			if failure == nil {
				failure = &failurepb.Failure{Message: "activity task failed without failure details"}
			}

			if err := a.recordFailedAttempt(ctx, 0, activitypb.ACTIVITY_RETRY_INTERVAL_SOURCE_UNSPECIFIED, failure, ctx.Now(a), true); err != nil {
				return err
			}

			a.emitOnFailedMetrics(ctx, event.baseHandler, event.enrichedHandler, req.GetFailure())

			return nil
		})
	},
)

TransitionFailed transitions to Failed status.

View Source 
var TransitionPauseRequested = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_STARTED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_PAUSE_REQUESTED,
	func(a *Activity, ctx chasm.MutableContext, event pauseEvent) error {
		a.recordPauseState(ctx, event)
		return nil
	},
)

TransitionPauseRequested transitions a STARTED activity to PAUSE_REQUESTED. The worker is still in charge of the activity. It will be notified via ActivityPaused=true on its next heartbeat response, its task token is not invalidated by this transition, and there is no stamp bump since StartToCloseTimeoutTask and HeartbeatTimeoutTask must stay valid.

View Source 
var TransitionPaused = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED,
	func(a *Activity, ctx chasm.MutableContext, event pauseEvent) error {
		a.recordPauseState(ctx, event)
		attempt := a.LastAttempt.Get(ctx)
		attempt.Stamp++
		return nil
	},
)

TransitionPaused transitions a SCHEDULED activity to PAUSED status. The stamp is bumped to invalidate any pending dispatch task so the activity is not dispatched while paused.

View Source 
var TransitionRescheduled = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_STARTED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED,
	func(a *Activity, ctx chasm.MutableContext, event rescheduleEvent) error {
		if err := a.applyFailedAttempt(ctx, event); err != nil {
			return err
		}

		attempt := a.LastAttempt.Get(ctx)
		retryScheduledTime := dispatchTimeForRetry(attempt).AsTime()
		attempt.DispatchTime = timestamppb.New(retryScheduledTime)

		if timeout := a.GetScheduleToStartTimeout().AsDuration(); timeout > 0 {
			ctx.AddTask(
				a,
				chasm.TaskAttributes{
					ScheduledTime: retryScheduledTime.Add(timeout),
				},
				&activitypb.ScheduleToStartTimeoutTask{
					Stamp: attempt.GetStamp(),
				})
		}

		ctx.AddTask(
			a,
			chasm.TaskAttributes{
				ScheduledTime: retryScheduledTime,
			},
			a.newActivityDispatchTask(ctx))

		return nil
	},
)

TransitionRescheduled transitions to Scheduled from Started, which happens on retries. The event to pass in is the failure to be recorded from the previously failed attempt.

View Source 
var TransitionReset = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED,
		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED,
	func(a *Activity, ctx chasm.MutableContext, event resetEvent) error {
		a.reset(ctx, event)
		return nil
	},
)

TransitionReset resets a SCHEDULED or PAUSED activity back to attempt 1. The stamp is bumped to invalidate any pending dispatch task, then a new dispatch task is added at the given schedule time. For STARTED activities the reset is deferred — the activity transitions to RESET_REQUESTED via TransitionResetRequested and lands back in SCHEDULED via TransitionResetAttemptFailedToScheduled when the worker yields.

View Source 
var TransitionResetAttemptFailedToPaused = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_RESET_REQUESTED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED,
	func(a *Activity, ctx chasm.MutableContext, event rescheduleEvent) error {
		attempt := a.LastAttempt.Get(ctx)
		a.ResetShouldPause = false
		a.applyDeferredOptionRestore(ctx)
		a.applyDeferredHeartbeatClear(ctx)
		attempt.Count = 1
		attempt.Stamp++
		if err := a.recordFailedAttempt(ctx, event.retryInterval, event.retryIntervalSource, event.failure, ctx.Now(a), false); err != nil {
			return err
		}

		attempt.CurrentRetryInterval = nil
		attempt.CurrentRetryIntervalSource = activitypb.ACTIVITY_RETRY_INTERVAL_SOURCE_UNSPECIFIED
		attempt.DispatchTime = nil
		return nil
	},
)

TransitionResetAttemptFailedToPaused transitions RESET_REQUESTED → PAUSED. It is performed when the worker yields in RESET_REQUESTED with ResetShouldPause set (i.e. reset was issued with keepPaused=true while the activity was in PAUSE_REQUESTED). The failed attempt is recorded, the attempt count is reset to 1, and no dispatch task is emitted — the activity stays paused until an explicit unpause.

View Source 
var TransitionResetAttemptFailedToScheduled = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_RESET_REQUESTED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED,
	func(a *Activity, ctx chasm.MutableContext, event rescheduleEvent) error {
		attempt := a.LastAttempt.Get(ctx)
		currentTime := ctx.Now(a)

		a.ResetShouldPause = false
		a.applyDeferredOptionRestore(ctx)
		a.applyDeferredHeartbeatClear(ctx)

		attempt.Count = 1
		attempt.Stamp++

		if err := a.recordFailedAttempt(ctx, event.retryInterval, event.retryIntervalSource, event.failure, currentTime, false); err != nil {
			return err
		}

		attempt.CurrentRetryInterval = nil
		attempt.CurrentRetryIntervalSource = activitypb.ACTIVITY_RETRY_INTERVAL_SOURCE_UNSPECIFIED

		dispatchTime := a.dispatchTimeRespectingStartDelay(currentTime)
		attempt.DispatchTime = timestamppb.New(dispatchTime)
		if timeout := a.GetScheduleToStartTimeout().AsDuration(); timeout > 0 {
			ctx.AddTask(
				a,
				chasm.TaskAttributes{
					ScheduledTime: dispatchTime.Add(timeout),
				},
				&activitypb.ScheduleToStartTimeoutTask{
					Stamp: attempt.GetStamp(),
				})
		}
		ctx.AddTask(
			a,
			chasm.TaskAttributes{
				ScheduledTime: dispatchTime,
			},
			a.newActivityDispatchTask(ctx))

		return nil
	},
)

TransitionResetAttemptFailedToScheduled transitions RESET_REQUESTED → SCHEDULED. It is performed instead of TransitionRescheduled when the activity is in RESET_REQUESTED and the worker yields (failure or timeout) with retries remaining. The failed attempt is recorded and the count is reset to 1 (so the next attempt is "attempt 1").

View Source 
var TransitionResetRequested = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_STARTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSE_REQUESTED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_RESET_REQUESTED,
	func(a *Activity, ctx chasm.MutableContext, _ any) error {
		return nil
	},
)

TransitionResetRequested transitions a STARTED or PAUSE_REQUESTED activity to RESET_REQUESTED. PAUSE_REQUESTED is allowed when the operator issues reset with keepPaused=true: ResetShouldPause is set so the activity lands back in PAUSED (not SCHEDULED) when the worker yields. The worker is still in charge of the activity; it will be notified via ActivityReset=true on its next heartbeat response, its task token is not invalidated by this transition, and there is no stamp bump since StartToCloseTimeoutTask and HeartbeatTimeoutTask must stay valid.

View Source 
var TransitionScheduled = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_UNSPECIFIED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED,
	func(a *Activity, ctx chasm.MutableContext, _ any) error {
		attempt := a.LastAttempt.Get(ctx)

		attempt.Count++
		attempt.Stamp++

		dispatchTime := a.firstDispatchTime()
		attempt.DispatchTime = timestamppb.New(dispatchTime)

		if timeout := a.GetScheduleToStartTimeout().AsDuration(); timeout > 0 {
			ctx.AddTask(
				a,
				chasm.TaskAttributes{
					ScheduledTime: dispatchTime.Add(timeout),
				},
				&activitypb.ScheduleToStartTimeoutTask{
					Stamp: attempt.GetStamp(),
				})
		}

		if deadline := a.scheduleToCloseDeadline(); !deadline.IsZero() {
			a.ScheduleToCloseStamp++
			ctx.AddTask(
				a,
				chasm.TaskAttributes{
					ScheduledTime: deadline,
				},
				&activitypb.ScheduleToCloseTimeoutTask{Stamp: a.GetScheduleToCloseStamp()})
		}

		dispatchAttrs := chasm.TaskAttributes{}
		if dispatchTime.After(a.ScheduleTime.AsTime()) {
			dispatchAttrs.ScheduledTime = dispatchTime
		}
		ctx.AddTask(
			a,
			dispatchAttrs,
			a.newActivityDispatchTask(ctx))

		return nil
	},
)

TransitionScheduled transitions to Scheduled status. This is only called on the initial scheduling of the activity.

View Source 
var TransitionStarted = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_STARTED,
	func(a *Activity, ctx chasm.MutableContext, request *historyservice.RecordActivityTaskStartedRequest) error {
		attempt := a.LastAttempt.Get(ctx)
		attempt.StartedTime = timestamppb.New(ctx.Now(a))
		attempt.StartedStamp = request.GetStamp()

		if a.FirstAttemptStartedTime == nil {
			a.FirstAttemptStartedTime = attempt.GetStartedTime()
		}
		attempt.StartRequestId = request.GetRequestId()
		attempt.LastWorkerIdentity = request.GetPollRequest().GetIdentity()
		attempt.SdkName = ctx.RequestHeader(headers.ClientNameHeaderName)
		attempt.SdkVersion = ctx.RequestHeader(headers.ClientVersionHeaderName)
		if versionDirective := request.GetVersionDirective().GetDeploymentVersion(); versionDirective != nil {
			attempt.LastDeploymentVersion = &deploymentpb.WorkerDeploymentVersion{
				BuildId:        versionDirective.GetBuildId(),
				DeploymentName: versionDirective.GetDeploymentName(),
			}
		}
		startTime := attempt.GetStartedTime().AsTime()
		ctx.AddTask(
			a,
			chasm.TaskAttributes{
				ScheduledTime: startTime.Add(a.GetStartToCloseTimeout().AsDuration()),
			},
			&activitypb.StartToCloseTimeoutTask{
				Stamp: a.LastAttempt.Get(ctx).GetStamp(),
			})

		if heartbeatTimeout := a.GetHeartbeatTimeout().AsDuration(); heartbeatTimeout > 0 {
			ctx.AddTask(
				a,
				chasm.TaskAttributes{
					ScheduledTime: startTime.Add(heartbeatTimeout),
				},
				&activitypb.HeartbeatTimeoutTask{
					Stamp: attempt.GetStamp(),
				})
		}

		return nil
	},
)

TransitionStarted transitions to Started status.

View Source 
var TransitionTerminated = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED,
		activitypb.ACTIVITY_EXECUTION_STATUS_STARTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED,
		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSE_REQUESTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_RESET_REQUESTED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_TERMINATED,
	func(a *Activity, ctx chasm.MutableContext, event terminateEvent) error {
		return a.StoreOrSelf(ctx).RecordCompleted(ctx, func(ctx chasm.MutableContext) error {
			a.TerminateState = &activitypb.ActivityTerminateState{
				RequestId: event.request.RequestID,
			}
			outcome := a.Outcome.Get(ctx)
			failure := &failurepb.Failure{
				Message: event.request.Reason,
				FailureInfo: &failurepb.Failure_TerminatedFailureInfo{
					TerminatedFailureInfo: &failurepb.TerminatedFailureInfo{
						Identity: event.request.Identity,
					},
				},
			}
			outcome.Variant = &activitypb.ActivityOutcome_Failed_{
				Failed: &activitypb.ActivityOutcome_Failed{
					Failure: failure,
				},
			}

			a.emitOnTerminatedMetrics(event.metricsHandler)

			return nil
		})
	},
)

TransitionTerminated transitions to Terminated status.

View Source 
var TransitionTimedOut = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED,
		activitypb.ACTIVITY_EXECUTION_STATUS_STARTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED,
		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSE_REQUESTED,
		activitypb.ACTIVITY_EXECUTION_STATUS_RESET_REQUESTED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_TIMED_OUT,
	func(a *Activity, ctx chasm.MutableContext, event timeoutEvent) error {
		timeoutType := event.timeoutType

		return a.StoreOrSelf(ctx).RecordCompleted(ctx, func(ctx chasm.MutableContext) error {
			a.Outcome.Get(ctx).RetryState = event.retryState
			priorAttemptFailure := a.LastAttempt.Get(ctx).GetLastFailureDetails().GetFailure()
			var err error
			switch timeoutType {
			case enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START,
				enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE:
				err = a.recordScheduleToStartOrCloseTimeoutFailure(
					ctx,
					timeoutType,
					fmt.Sprintf(common.FailureReasonActivityTimeout, timeoutType.String()),
					priorAttemptFailure,
				)
			case enumspb.TIMEOUT_TYPE_START_TO_CLOSE:
				failure := createStartToCloseTimeoutFailure()
				failure.GetTimeoutFailureInfo().LastHeartbeatDetails = a.lastHeartbeatDetails(ctx)
				failure.Cause = priorAttemptFailure
				err = a.recordFailedAttempt(ctx, 0, activitypb.ACTIVITY_RETRY_INTERVAL_SOURCE_UNSPECIFIED, failure, ctx.Now(a), true)
			case enumspb.TIMEOUT_TYPE_HEARTBEAT:
				failure := createHeartbeatTimeoutFailure()
				failure.GetTimeoutFailureInfo().LastHeartbeatDetails = a.lastHeartbeatDetails(ctx)
				failure.Cause = priorAttemptFailure
				err = a.recordFailedAttempt(ctx, 0, activitypb.ACTIVITY_RETRY_INTERVAL_SOURCE_UNSPECIFIED, failure, ctx.Now(a), true)
			default:
				err = fmt.Errorf("unhandled activity timeout: %v", timeoutType)
			}
			if err != nil {
				return err
			}

			retryPreventedByScheduleToClose := event.retryState == enumspb.RETRY_STATE_TIMEOUT &&
				(timeoutType == enumspb.TIMEOUT_TYPE_START_TO_CLOSE ||
					timeoutType == enumspb.TIMEOUT_TYPE_HEARTBEAT)
			if retryPreventedByScheduleToClose {
				if err := a.recordScheduleToStartOrCloseTimeoutFailure(
					ctx,
					enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE,
					common.FailureReasonActivityRetryScheduleToCloseTimeout,
					priorAttemptFailure,
				); err != nil {
					return err
				}
			}

			a.emitOnTimedOutMetrics(event.metricsHandler, timeoutType)

			return nil
		})
	},
)

TransitionTimedOut transitions to TimedOut status.

View Source 
var TransitionUnpaused = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED,
	func(a *Activity, ctx chasm.MutableContext, event unpauseEvent) error {
		a.unpause(ctx, event)
		return nil
	},
)

TransitionUnpaused transitions PAUSED → SCHEDULED, triggering a dispatch task that will lead to another activity attempt.

View Source 
var TransitionUnpausedWhilePauseRequested = chasm.NewTransition(
	[]activitypb.ActivityExecutionStatus{
		activitypb.ACTIVITY_EXECUTION_STATUS_PAUSE_REQUESTED,
	},
	activitypb.ACTIVITY_EXECUTION_STATUS_STARTED,
	func(a *Activity, ctx chasm.MutableContext, event unpauseEvent) error {
		return nil
	},
)

TransitionUnpausedWhilePauseRequested transitions PAUSE_REQUESTED → STARTED. The worker is still in charge of the activity. Its task token is not invalidated by this transition, and there is no stamp bump since StartToCloseTimeoutTask and HeartbeatTimeoutTask must stay valid.

Functions

func InternalStatusToAPIStatus 

func InternalStatusToAPIStatus(status activitypb.ActivityExecutionStatus) enumspb.ActivityExecutionStatus

InternalStatusToAPIStatus converts internal activity execution status to API status.

func NewEmbeddedActivity 

func NewEmbeddedActivity(
	ctx chasm.MutableContext,
	state *activitypb.ActivityState,
	parent ActivityStore,
)

func NewNilLibrary 

func NewNilLibrary() chasm.Library

NewNilLibrary creates a Library with all nil handlers. Useful for registration-only contexts like tdbg where no task execution is needed.

func ValidateAndNormalizeEmbeddedActivity 

func ValidateAndNormalizeEmbeddedActivity(
	activityID string,
	activityType string,
	getDefaultActivityRetrySettings dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings],
	maxIDLengthLimit int,
	namespaceName namespace.Name,
	options *activitypb.ActivityOptions,
	priority *commonpb.Priority,
	runTimeout *durationpb.Duration,
	workflowTaskQueueName string,
) error

ValidateAndNormalizeEmbeddedActivity validates and normalizes the attributes for an embedded activity.

func ValidateAndNormalizeStandaloneActivity 

func ValidateAndNormalizeStandaloneActivity(
	activityID string,
	activityType string,
	getDefaultActivityRetrySettings dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings],
	maxIDLengthLimit int,
	namespaceName namespace.Name,
	options *activitypb.ActivityOptions,
	priority *commonpb.Priority,
) error

ValidateAndNormalizeStandaloneActivity validates and normalizes the attributes for a standalone activity.

Types

type Activity 

type Activity struct {
	chasm.UnimplementedComponent

	*activitypb.ActivityState

	Visibility    chasm.Field[*chasm.Visibility]
	LastAttempt   chasm.Field[*activitypb.ActivityAttemptState]
	LastHeartbeat chasm.Field[*activitypb.ActivityHeartbeatState]
	// Standalone only
	RequestData chasm.Field[*activitypb.ActivityRequestData]
	Outcome     chasm.Field[*activitypb.ActivityOutcome]
	// Pointer to an implementation of the "store". For a workflow activity this would be a parent
	// pointer back to the workflow. For a standalone activity this is nil (Activity itself
	// implements the ActivityStore interface).
	// TODO(saa-preview): figure out better naming.
	Store chasm.ParentPtr[ActivityStore]

	// Callbacks holds completion callbacks to be invoked when this standalone activity reaches a terminal state. Nil
	// for workflow-embedded activities as the workflow handles its own callbacks.
	Callbacks chasm.Map[string, *callback.Callback]
}

Activity component represents an activity execution persistence object and can be either standalone activity or one embedded within a workflow.

func NewStandaloneActivity 

func NewStandaloneActivity(
	ctx chasm.MutableContext,
	request *workflowservice.StartActivityExecutionRequest,
) (*Activity, error)

NewStandaloneActivity creates a new activity component and adds associated tasks to start execution.

func (*Activity) ContextMetadata 

func (a *Activity) ContextMetadata(_ chasm.Context) map[string]string

func (*Activity) GenerateRecordActivityTaskStartedResponse 

func (a *Activity) GenerateRecordActivityTaskStartedResponse(
	ctx chasm.Context,
	namespace string,
) (*historyservice.RecordActivityTaskStartedResponse, error)

GenerateRecordActivityTaskStartedResponse generates the response for HandleStarted.

func (*Activity) GetNexusCompletion 

func (a *Activity) GetNexusCompletion(ctx chasm.Context, _ string) (nexusrpc.CompleteOperationOptions, error)

GetNexusCompletion returns the activity's completion data in the format required by the Nexus callback invocation. Implements callback.CompletionSource.

func (*Activity) HandleCanceled 

func (a *Activity) HandleCanceled(
	ctx chasm.MutableContext,
	event RespondCancelledEvent,
) (*historyservice.RespondActivityTaskCanceledResponse, error)

HandleCanceled updates the activity on activity canceled.

func (*Activity) HandleCompleted 

func (a *Activity) HandleCompleted(
	ctx chasm.MutableContext,
	event RespondCompletedEvent,
) (*historyservice.RespondActivityTaskCompletedResponse, error)

HandleCompleted updates the activity on activity completion.

func (*Activity) HandleFailed 

func (a *Activity) HandleFailed(
	ctx chasm.MutableContext,
	event RespondFailedEvent,
) (*historyservice.RespondActivityTaskFailedResponse, error)

HandleFailed updates the activity on activity failure. if the activity is retryable, it will be rescheduled for retry instead.

func (*Activity) HandleStarted 

func (a *Activity) HandleStarted(ctx chasm.MutableContext, request *historyservice.RecordActivityTaskStartedRequest) (
	*historyservice.RecordActivityTaskStartedResponse, error,
)

HandleStarted updates the activity on recording activity task started and populates the response.

func (*Activity) IsDispatchTaskValid 

func (a *Activity) IsDispatchTaskValid(
	ctx chasm.Context,
	stamp int32,
) (bool, error)

IsDispatchTaskValid reports whether the attempt identified by stamp is still the activity's current scheduled attempt.

func (*Activity) LifecycleState 

func (a *Activity) LifecycleState(_ chasm.Context) chasm.LifecycleState

LifecycleState implements the chasm.Component interface.

func (*Activity) RecordCompleted 

func (a *Activity) RecordCompleted(ctx chasm.MutableContext, applyFn func(ctx chasm.MutableContext) error) error

RecordCompleted applies the provided function to record activity completion. For standalone activities, it also triggers any registered completion callbacks.

func (*Activity) RecordHeartbeat 

func (a *Activity) RecordHeartbeat(
	ctx chasm.MutableContext,
	input WithToken[*historyservice.RecordActivityTaskHeartbeatRequest],
) (*historyservice.RecordActivityTaskHeartbeatResponse, error)

RecordHeartbeat records a heartbeat for the activity.

func (*Activity) SearchAttributes 

func (a *Activity) SearchAttributes(_ chasm.Context) []chasm.SearchAttributeKeyValue

SearchAttributes implements chasm.VisibilitySearchAttributesProvider interface. Returns the current search attribute values for this activity execution.

func (*Activity) SetStateMachineState 

func (a *Activity) SetStateMachineState(state activitypb.ActivityExecutionStatus)

SetStateMachineState sets the status of the activity.

func (*Activity) StateMachineState 

func (a *Activity) StateMachineState() activitypb.ActivityExecutionStatus

StateMachineState returns the current status of the activity.

func (*Activity) StoreOrSelf 

func (a *Activity) StoreOrSelf(ctx chasm.Context) ActivityStore

StoreOrSelf returns the store for the activity. If the store is not set as a field (e.g. standalone activities), it returns the activity itself.

func (*Activity) Terminate 

func (a *Activity) Terminate(
	ctx chasm.MutableContext,
	req chasm.TerminateComponentRequest,
) (chasm.TerminateComponentResponse, error)

Terminate implements the chasm.RootComponent interface.

func (*Activity) UpdateActivityExecutionOptions 

func (a *Activity) UpdateActivityExecutionOptions(
	ctx chasm.MutableContext,
	req *activitypb.UpdateActivityExecutionOptionsRequest,
) (*activitypb.UpdateActivityExecutionOptionsResponse, error)

type ActivityStore 

type ActivityStore interface {
	// RecordCompleted applies the provided function to record activity completion
	RecordCompleted(ctx chasm.MutableContext, applyFn func(ctx chasm.MutableContext) error) error
}

type Config 

type Config struct {
	BlobSizeLimitError                        dynamicconfig.IntPropertyFnWithNamespaceFilter
	BlobSizeLimitWarn                         dynamicconfig.IntPropertyFnWithNamespaceFilter
	BreakdownMetricsByTaskQueue               dynamicconfig.TypedPropertyFnWithTaskQueueFilter[bool]
	EnableCallbacks                           dynamicconfig.BoolPropertyFnWithNamespaceFilter
	Enabled                                   dynamicconfig.BoolPropertyFnWithNamespaceFilter
	EnableStandaloneActivityOperatorCommands  dynamicconfig.BoolPropertyFnWithNamespaceFilter
	LongPollBuffer                            dynamicconfig.DurationPropertyFnWithNamespaceFilter
	LongPollTimeout                           dynamicconfig.DurationPropertyFnWithNamespaceFilter
	MaxIDLengthLimit                          dynamicconfig.IntPropertyFn
	MaxCallbacksPerExecution                  dynamicconfig.IntPropertyFnWithNamespaceFilter
	MutableStateActivityFailureSizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter
	DefaultActivityRetryPolicy                dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings]
	MaxUserMetadataDetailsSize                dynamicconfig.IntPropertyFnWithNamespaceFilter
	MaxUserMetadataSummarySize                dynamicconfig.IntPropertyFnWithNamespaceFilter
	StartDelayEnabled                         dynamicconfig.BoolPropertyFnWithNamespaceFilter
	VisibilityMaxPageSize                     dynamicconfig.IntPropertyFnWithNamespaceFilter
}

func ConfigProvider 

func ConfigProvider(dc *dynamicconfig.Collection) *Config

type FrontendHandler 

type FrontendHandler interface {
	StartActivityExecution(ctx context.Context, req *workflowservice.StartActivityExecutionRequest) (*workflowservice.StartActivityExecutionResponse, error)
	DescribeActivityExecution(ctx context.Context, req *workflowservice.DescribeActivityExecutionRequest) (*workflowservice.DescribeActivityExecutionResponse, error)
	PollActivityExecution(ctx context.Context, req *workflowservice.PollActivityExecutionRequest) (*workflowservice.PollActivityExecutionResponse, error)
	CountActivityExecutions(context.Context, *workflowservice.CountActivityExecutionsRequest) (*workflowservice.CountActivityExecutionsResponse, error)
	DeleteActivityExecution(context.Context, *workflowservice.DeleteActivityExecutionRequest) (*workflowservice.DeleteActivityExecutionResponse, error)
	ListActivityExecutions(context.Context, *workflowservice.ListActivityExecutionsRequest) (*workflowservice.ListActivityExecutionsResponse, error)
	RequestCancelActivityExecution(context.Context, *workflowservice.RequestCancelActivityExecutionRequest) (*workflowservice.RequestCancelActivityExecutionResponse, error)
	TerminateActivityExecution(context.Context, *workflowservice.TerminateActivityExecutionRequest) (*workflowservice.TerminateActivityExecutionResponse, error)
	PauseActivityExecution(context.Context, *workflowservice.PauseActivityExecutionRequest) (*workflowservice.PauseActivityExecutionResponse, error)
	UnpauseActivityExecution(context.Context, *workflowservice.UnpauseActivityExecutionRequest) (*workflowservice.UnpauseActivityExecutionResponse, error)
	ResetActivityExecution(context.Context, *workflowservice.ResetActivityExecutionRequest) (*workflowservice.ResetActivityExecutionResponse, error)
	UpdateActivityExecutionOptions(context.Context, *workflowservice.UpdateActivityExecutionOptionsRequest) (*workflowservice.UpdateActivityExecutionOptionsResponse, error)
	IsStandaloneActivityEnabled(namespaceName string) bool
}

func NewFrontendHandler 

func NewFrontendHandler(
	callbackValidator callback.Validator,
	linkValidator *linkValidator,
	client activitypb.ActivityServiceClient,
	config *Config,
	logger log.Logger,
	metricsHandler metrics.Handler,
	namespaceRegistry namespace.Registry,
	saMapperProvider searchattribute.MapperProvider,
	saValidator *searchattribute.Validator,
) FrontendHandler

NewFrontendHandler creates a new FrontendHandler instance for processing activity frontend requests.

type RespondCancelledEvent 

type RespondCancelledEvent struct {
	Request *historyservice.RespondActivityTaskCanceledRequest
	Token   *tokenspb.Task
}

RespondCancelledEvent wraps the RespondActivityTaskCanceledRequest with context-specific data.

type RespondCompletedEvent 

type RespondCompletedEvent struct {
	Request *historyservice.RespondActivityTaskCompletedRequest
	Token   *tokenspb.Task
}

RespondCompletedEvent wraps the RespondActivityTaskCompletedRequest with context-specific data.

type RespondFailedEvent 

type RespondFailedEvent struct {
	Request *historyservice.RespondActivityTaskFailedRequest
	Token   *tokenspb.Task
}

RespondFailedEvent wraps the RespondActivityTaskFailedRequest with context-specific data.

type WithToken 

type WithToken[R any] struct {
	Token   *tokenspb.Task
	Request R
}

WithToken wraps a request with its deserialized task token.

Source Files

View all Source files

Directories

Path Synopsis
gen
activitypb/v1
Code generated by protoc-gen-go-helpers.
 Code generated by protoc-gen-go-helpers.
Package model is an implementation-independent vocabulary for specifying a sequence of events (a 'trace') in the lifetime of an activity.
 Package model is an implementation-independent vocabulary for specifying a sequence of events (a 'trace') in the lifetime of an activity.

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.