common

Documentation

Index

• Constants
• Variables
• func AwaitWaitGroup(wg *sync.WaitGroup, timeout time.Duration) bool
• func CheckEventBlobSizeLimit(actualSize int, warnLimit int, errorLimit int, namespace string, ...) error
• func CloneProto[T proto.Message](v T) T
• func CreateCompleteTaskRetryPolicy() backoff.RetryPolicy
• func CreateDependencyTaskNotCompletedReschedulePolicy() backoff.RetryPolicy
• func CreateFrontendClientRetryPolicy() backoff.RetryPolicy
• func CreateFrontendHandlerRetryPolicy() backoff.RetryPolicy
• func CreateHistoryClientRetryPolicy() backoff.RetryPolicy
• func CreateHistoryHandlerRetryPolicy() backoff.RetryPolicy
• func CreateHistoryStartWorkflowRequest(namespaceID string, ...) *historyservice.StartWorkflowExecutionRequest
• func CreateMatchingClientRetryPolicy() backoff.RetryPolicy
• func CreateMatchingHandlerRetryPolicy() backoff.RetryPolicy
• func CreateMatchingPollWorkflowTaskQueueResponse(historyResponse *historyservice.RecordWorkflowTaskStartedResponse, ...) *matchingservice.PollWorkflowTaskQueueResponse
• func CreatePersistenceClientRetryPolicy() backoff.RetryPolicy
• func CreateReadTaskRetryPolicy() backoff.RetryPolicy
• func CreateSdkClientFactoryRetryPolicy() backoff.RetryPolicy
• func CreateTaskNotReadyReschedulePolicy() backoff.RetryPolicy
• func CreateTaskReschedulePolicy() backoff.RetryPolicy
• func CreateTaskResourceExhaustedReschedulePolicy() backoff.RetryPolicy
• func EnsureRetryPolicyDefaults(originalPolicy *commonpb.RetryPolicy, defaultSettings DefaultRetrySettings)
• func GenerateRandomString(n int) string
• func GetDefaultRetryPolicyConfigOptions() map[string]interface{}
• func GetPayloadsMapSize(data map[string]*commonpb.Payloads) int
• func InterruptibleSleep(ctx context.Context, timeout time.Duration)
• func IsContextCanceledErr(err error) bool
• func IsContextDeadlineExceededErr(err error) bool
• func IsInternalError(err error) bool
• func IsNotFoundError(err error) bool
• func IsPersistenceTransientError(err error) bool
• func IsResourceExhausted(err error) bool
• func IsServiceClientTransientError(err error) bool
• func IsServiceHandlerRetryableError(err error) bool
• func IsServiceTransientError(err error) bool
• func IsStickyWorkerUnavailable(err error) bool
• func IsValidContext(ctx context.Context) error
• func MakeVersionDirectiveForActivityTask(stamp *commonpb.WorkerVersionStamp, useCompatibleVersion bool) *taskqueuespb.TaskVersionDirective
• func MakeVersionDirectiveForWorkflowTask(stamp *commonpb.WorkerVersionStamp, lastWorkflowTaskStartedEventID int64) *taskqueuespb.TaskVersionDirective
• func MapShardID(sourceShardCount int32, targetShardCount int32, sourceShardID int32) []int32
• func OverrideWorkflowRunTimeout(workflowRunTimeout time.Duration, workflowExecutionTimeout time.Duration) time.Duration
• func OverrideWorkflowTaskTimeout(namespace string, taskStartToCloseTimeout time.Duration, ...) time.Duration
• func PrettyPrint[T proto.Message](msgs []T, header ...string)
• func StampIfUsingVersioning(stamp *commonpb.WorkerVersionStamp) *commonpb.WorkerVersionStamp
• func ValidateLongPollContextTimeout(ctx context.Context, handlerName string, logger log.Logger) error
• func ValidateLongPollContextTimeoutIsSet(ctx context.Context, handlerName string, logger log.Logger) (time.Time, error)
• func ValidateRetryPolicy(policy *commonpb.RetryPolicy) error
• func VerifyShardIDMapping(thisShardCount int32, thatShardCount int32, thisShardID int32, ...) error
• func WorkflowIDToHistoryShard(namespaceID string, workflowID string, numberOfShards int32) int32
• type ClientCache
  • func NewClientCache(keyResolver keyResolver, clientProvider clientProvider) ClientCache
• type Daemon
• type DefaultRetrySettings
  • func FromConfigToDefaultRetrySettings(options map[string]interface{}) DefaultRetrySettings
• type PingCheck
• type Pingable
• type RPCFactory
• type TaskTokenSerializer
  • func NewProtoTaskTokenSerializer() TaskTokenSerializer

Constants

const (
	// FirstEventID is the id of the first event in the history
	FirstEventID int64 = 1
	// LastEventID is the id of the last possible event in the history
	LastEventID int64 = math.MaxInt64
	// EmptyEventID is the id of the empty event
	EmptyEventID int64 = 0
	// EmptyVersion is used as the default value for failover version when no value is provided
	EmptyVersion int64 = 0
	// EndEventID is the id of the end event, here we use the int64 max
	EndEventID int64 = 1<<63 - 1
	// BufferedEventID is the id of the buffered event
	BufferedEventID int64 = -123
	// EmptyEventTaskID is uninitialized id of the task id within event
	EmptyEventTaskID int64 = 0
	// TransientEventID is the id of the transient event
	TransientEventID int64 = -124
	// FirstBlobPageToken is the page token identifying the first blob for each history archival
	FirstBlobPageToken = 1
	// LastBlobNextPageToken is the next page token on the last blob for each history archival
	LastBlobNextPageToken = -1
	// EndMessageID is the id of the end message, here we use the int64 max
	EndMessageID int64 = 1<<63 - 1
)

const (
	// GetHistoryMaxPageSize is the max page size for get history
	GetHistoryMaxPageSize = 256
	// ReadDLQMessagesPageSize is the max page size for read DLQ messages
	ReadDLQMessagesPageSize = 1000
)

const (
	// MinLongPollTimeout is the minimum context timeout for long poll API, below which
	// the request won't be processed
	MinLongPollTimeout = time.Second * 2
	// CriticalLongPollTimeout is a threshold for the context timeout passed into long poll API,
	// below which a warning will be logged
	CriticalLongPollTimeout = time.Second * 20
)

const (
	// DefaultWorkflowTaskTimeout sets the Default Workflow Task timeout for a Workflow
	DefaultWorkflowTaskTimeout = 10 * time.Second * debug.TimeoutMultiplier

	// MaxWorkflowTaskStartToCloseTimeout sets the Max Workflow Task start to close timeout for a Workflow
	MaxWorkflowTaskStartToCloseTimeout = 120 * time.Second
)

const (

	// DaemonStatusInitialized coroutine pool initialized
	DaemonStatusInitialized int32 = 0
	// DaemonStatusStarted coroutine pool started
	DaemonStatusStarted int32 = 1
	// DaemonStatusStopped coroutine pool stopped
	DaemonStatusStopped int32 = 2
)

const (

	// FailureReasonCompleteResultExceedsLimit is failureReason for complete result exceeds limit
	FailureReasonCompleteResultExceedsLimit = "Complete result exceeds size limit."
	// FailureReasonFailureDetailsExceedsLimit is failureReason for failure details exceeds limit
	FailureReasonFailureExceedsLimit = "Failure exceeds size limit."
	// FailureReasonCancelDetailsExceedsLimit is failureReason for cancel details exceeds limit
	FailureReasonCancelDetailsExceedsLimit = "Cancel details exceed size limit."
	// FailureReasonHeartbeatExceedsLimit is failureReason for heartbeat exceeds limit
	FailureReasonHeartbeatExceedsLimit = "Heartbeat details exceed size limit."
	// FailureReasonHistorySizeExceedsLimit is reason to fail workflow when history size or count exceed limit
	FailureReasonHistorySizeExceedsLimit = "Workflow history size / count exceeds limit."
	// FailureReasonMutableStateSizeExceedsLimit is reason to fail workflow when mutable state size exceeds limit
	FailureReasonMutableStateSizeExceedsLimit = "Workflow mutable state size exceeds limit."
	// FailureReasonTransactionSizeExceedsLimit is the failureReason for when transaction cannot be committed because it exceeds size limit
	FailureReasonTransactionSizeExceedsLimit = "Transaction size exceeds limit."
)

const (
	// DefaultQueueReaderID is the default readerID when loading history tasks
	DefaultQueueReaderID int64 = 0
)

const (
	// DefaultTransactionSizeLimit is the largest allowed transaction size to persistence
	DefaultTransactionSizeLimit = 4 * 1024 * 1024
)

const (
	// Limit for schedule notes field
	ScheduleNotesSizeLimit = 1000
)

const (
	// TimeoutFailureTypePrefix is the prefix for timeout failure types
	// used in retry policy
	// the actual failure type will be prefix + enums.TimeoutType.String()
	// e.g. "TemporalTimeout:StartToClose" or "TemporalTimeout:Heartbeat"
	TimeoutFailureTypePrefix = "TemporalTimeout:"
)

Variables

var (
	// ErrBlobSizeExceedsLimit is error for event blob size exceeds limit
	ErrBlobSizeExceedsLimit = serviceerror.NewInvalidArgument("Blob data size exceeds limit.")
	// ErrMemoSizeExceedsLimit is error for memo size exceeds limit
	ErrMemoSizeExceedsLimit = serviceerror.NewInvalidArgument("Memo size exceeds limit.")
	// ErrContextTimeoutTooShort is error for setting a very short context timeout when calling a long poll API
	ErrContextTimeoutTooShort = serviceerror.NewFailedPrecondition("Context timeout is too short.")
	// ErrContextTimeoutNotSet is error for not setting a context timeout when calling a long poll API
	ErrContextTimeoutNotSet = serviceerror.NewInvalidArgument("Context timeout is not set.")
)

var (
	// ErrNamespaceHandover is error indicating namespace is in handover state and cannot process request.
	ErrNamespaceHandover = serviceerror.NewUnavailable(fmt.Sprintf("Namespace replication in %s state.", enumspb.REPLICATION_STATE_HANDOVER.String()))
)

Functions

func AwaitWaitGroup

func AwaitWaitGroup(wg *sync.WaitGroup, timeout time.Duration) bool

AwaitWaitGroup calls Wait on the given wait Returns true if the Wait() call succeeded before the timeout Returns false if the Wait() did not return before the timeout

func CheckEventBlobSizeLimit

func CheckEventBlobSizeLimit(
	actualSize int,
	warnLimit int,
	errorLimit int,
	namespace string,
	workflowID string,
	runID string,
	metricsHandler metrics.Handler,
	logger log.Logger,
	blobSizeViolationOperationTag tag.ZapTag,
) error

CheckEventBlobSizeLimit checks if a blob data exceeds limits. It logs a warning if it exceeds warnLimit, and return ErrBlobSizeExceedsLimit if it exceeds errorLimit.

func CloneProto

func CloneProto[T proto.Message](v T) T

CloneProto is a generic typed version of proto.Clone from gogoproto.

func CreateCompleteTaskRetryPolicy

func CreateCompleteTaskRetryPolicy() backoff.RetryPolicy

CreateCompleteTaskRetryPolicy creates a retry policy for completing background tasks

func CreateDependencyTaskNotCompletedReschedulePolicy

func CreateDependencyTaskNotCompletedReschedulePolicy() backoff.RetryPolicy

CreateDependencyTaskNotCompletedReschedulePolicy creates a retry policy for rescheduling task with ErrDependencyTaskNotCompleted

func CreateFrontendClientRetryPolicy

func CreateFrontendClientRetryPolicy() backoff.RetryPolicy

CreateFrontendClientRetryPolicy creates a retry policy for calls to frontend service

func CreateFrontendHandlerRetryPolicy

func CreateFrontendHandlerRetryPolicy() backoff.RetryPolicy

CreateFrontendHandlerRetryPolicy creates a retry policy for calls to frontend service

func CreateHistoryClientRetryPolicy

func CreateHistoryClientRetryPolicy() backoff.RetryPolicy

CreateHistoryClientRetryPolicy creates a retry policy for calls to history service

func CreateHistoryHandlerRetryPolicy

func CreateHistoryHandlerRetryPolicy() backoff.RetryPolicy

CreateHistoryHandlerRetryPolicy creates a retry policy for calls to history service

func CreateHistoryStartWorkflowRequest

func CreateHistoryStartWorkflowRequest(
	namespaceID string,
	startRequest *workflowservice.StartWorkflowExecutionRequest,
	parentExecutionInfo *workflowspb.ParentExecutionInfo,
	now time.Time,
) *historyservice.StartWorkflowExecutionRequest

CreateHistoryStartWorkflowRequest create a start workflow request for history. Note: this mutates startRequest by unsetting the fields ContinuedFailure and LastCompletionResult (these should only be set on workflows created by the scheduler worker). Assumes startRequest is valid. See frontend workflow_handler for detailed validation logic.

func CreateMatchingClientRetryPolicy

func CreateMatchingClientRetryPolicy() backoff.RetryPolicy

CreateMatchingClientRetryPolicy creates a retry policy for calls to matching service

func CreateMatchingHandlerRetryPolicy

func CreateMatchingHandlerRetryPolicy() backoff.RetryPolicy

CreateMatchingHandlerRetryPolicy creates a retry policy for calls to matching service

func CreateMatchingPollWorkflowTaskQueueResponse

func CreateMatchingPollWorkflowTaskQueueResponse(historyResponse *historyservice.RecordWorkflowTaskStartedResponse, workflowExecution *commonpb.WorkflowExecution, token []byte) *matchingservice.PollWorkflowTaskQueueResponse

CreateMatchingPollWorkflowTaskQueueResponse create response for matching's PollWorkflowTaskQueue

func CreatePersistenceClientRetryPolicy

func CreatePersistenceClientRetryPolicy() backoff.RetryPolicy

CreatePersistenceClientRetryPolicy creates a retry policy for calls to persistence

func CreateReadTaskRetryPolicy

func CreateReadTaskRetryPolicy() backoff.RetryPolicy

CreateReadTaskRetryPolicy creates a retry policy for loading background tasks

func CreateSdkClientFactoryRetryPolicy

func CreateSdkClientFactoryRetryPolicy() backoff.RetryPolicy

CreateSdkClientFactoryRetryPolicy creates a retry policy to handle SdkClientFactory NewClient when frontend service is not ready

func CreateTaskNotReadyReschedulePolicy

func CreateTaskNotReadyReschedulePolicy() backoff.RetryPolicy

CreateTaskNotReadyReschedulePolicy creates a retry policy for rescheduling task with ErrTaskRetry

func CreateTaskReschedulePolicy

func CreateTaskReschedulePolicy() backoff.RetryPolicy

CreateTaskReschedulePolicy creates a retry policy for rescheduling task with errors not equal to ErrTaskRetry

func CreateTaskResourceExhaustedReschedulePolicy

func CreateTaskResourceExhaustedReschedulePolicy() backoff.RetryPolicy

CreateTaskResourceExhaustedReschedulePolicy creates a retry policy for rescheduling task with resource exhausted error

func EnsureRetryPolicyDefaults

func EnsureRetryPolicyDefaults(originalPolicy *commonpb.RetryPolicy, defaultSettings DefaultRetrySettings)

EnsureRetryPolicyDefaults ensures the policy subfields, if not explicitly set, are set to the specified defaults

func GenerateRandomString

func GenerateRandomString(n int) string

GenerateRandomString is used for generate test string

func GetDefaultRetryPolicyConfigOptions

func GetDefaultRetryPolicyConfigOptions() map[string]interface{}

func GetPayloadsMapSize

func GetPayloadsMapSize(data map[string]*commonpb.Payloads) int

func InterruptibleSleep

func InterruptibleSleep(ctx context.Context, timeout time.Duration)

InterruptibleSleep is like time.Sleep but can be interrupted by a context.

func IsContextCanceledErr

func IsContextCanceledErr(err error) bool

IsContextCanceledErr checks if the error is context.Canceled or serviceerror.Canceled error

func IsContextDeadlineExceededErr

func IsContextDeadlineExceededErr(err error) bool

IsContextDeadlineExceededErr checks if the error is context.DeadlineExceeded or serviceerror.DeadlineExceeded error

func IsInternalError

func IsInternalError(err error) bool

IsInternalError checks if the error is an internal error.

func IsNotFoundError

func IsNotFoundError(err error) bool

IsNotFoundError checks if the error is a not found error.

func IsPersistenceTransientError

func IsPersistenceTransientError(err error) bool

IsPersistenceTransientError checks if the error is a transient persistence error

func IsResourceExhausted

func IsResourceExhausted(err error) bool

IsResourceExhausted checks if the error is a service busy error.

func IsServiceClientTransientError

func IsServiceClientTransientError(err error) bool

IsServiceClientTransientError checks if the error is a transient error.

func IsServiceHandlerRetryableError

func IsServiceHandlerRetryableError(err error) bool

func IsServiceTransientError

func IsServiceTransientError(err error) bool

IsServiceTransientError checks if the error is a retryable error.

func IsStickyWorkerUnavailable

func IsStickyWorkerUnavailable(err error) bool

func IsValidContext

func IsValidContext(ctx context.Context) error

IsValidContext checks that the thrift context is not expired on cancelled. Returns nil if the context is still valid. Otherwise, returns the result of ctx.Err()

func MakeVersionDirectiveForActivityTask

func MakeVersionDirectiveForActivityTask(
	stamp *commonpb.WorkerVersionStamp,
	useCompatibleVersion bool,
) *taskqueuespb.TaskVersionDirective

func MakeVersionDirectiveForWorkflowTask

func MakeVersionDirectiveForWorkflowTask(
	stamp *commonpb.WorkerVersionStamp,
	lastWorkflowTaskStartedEventID int64,
) *taskqueuespb.TaskVersionDirective

func MapShardID

func MapShardID(
	sourceShardCount int32,
	targetShardCount int32,
	sourceShardID int32,
) []int32

func OverrideWorkflowRunTimeout

func OverrideWorkflowRunTimeout(
	workflowRunTimeout time.Duration,
	workflowExecutionTimeout time.Duration,
) time.Duration

OverrideWorkflowRunTimeout override the run timeout according to execution timeout

func OverrideWorkflowTaskTimeout

func OverrideWorkflowTaskTimeout(
	namespace string,
	taskStartToCloseTimeout time.Duration,
	workflowRunTimeout time.Duration,
	getDefaultTimeoutFunc dynamicconfig.DurationPropertyFnWithNamespaceFilter,
) time.Duration

OverrideWorkflowTaskTimeout override the workflow task timeout according to default timeout or max timeout

func PrettyPrint

func PrettyPrint[T proto.Message](msgs []T, header ...string)

func StampIfUsingVersioning

func StampIfUsingVersioning(stamp *commonpb.WorkerVersionStamp) *commonpb.WorkerVersionStamp

StampIfUsingVersioning returns the given WorkerVersionStamp if it is using versioning, otherwise returns nil.

func ValidateLongPollContextTimeout

func ValidateLongPollContextTimeout(
	ctx context.Context,
	handlerName string,
	logger log.Logger,
) error

ValidateLongPollContextTimeout check if the context timeout for a long poll handler is too short or below a normal value. If the timeout is not set or too short, it logs an error, and return ErrContextTimeoutNotSet or ErrContextTimeoutTooShort accordingly. If the timeout is only below a normal value, it just logs an info and return nil.

func ValidateLongPollContextTimeoutIsSet

func ValidateLongPollContextTimeoutIsSet(
	ctx context.Context,
	handlerName string,
	logger log.Logger,
) (time.Time, error)

ValidateLongPollContextTimeoutIsSet checks if the context timeout is set for long poll requests.

func ValidateRetryPolicy

func ValidateRetryPolicy(policy *commonpb.RetryPolicy) error

ValidateRetryPolicy validates a retry policy

func VerifyShardIDMapping

func VerifyShardIDMapping(
	thisShardCount int32,
	thatShardCount int32,
	thisShardID int32,
	thatShardID int32,
) error

func WorkflowIDToHistoryShard

func WorkflowIDToHistoryShard(
	namespaceID string,
	workflowID string,
	numberOfShards int32,
) int32

WorkflowIDToHistoryShard is used to map namespaceID-workflowID pair to a shardID.

Types

type ClientCache

type ClientCache interface {
	GetClientForKey(key string) (interface{}, error)
	GetClientForClientKey(clientKey string) (interface{}, error)
	GetAllClients() ([]interface{}, error)
}

ClientCache store initialized clients

func NewClientCache

func NewClientCache(
	keyResolver keyResolver,
	clientProvider clientProvider,
) ClientCache

NewClientCache creates a new client cache based on membership

type Daemon

type Daemon interface {
	Start()
	Stop()
}

Daemon is the base interfaces implemented by background tasks within Temporal

type DefaultRetrySettings

type DefaultRetrySettings struct {
	InitialInterval            time.Duration
	MaximumIntervalCoefficient float64
	BackoffCoefficient         float64
	MaximumAttempts            int32
}

DefaultRetrySettings indicates what the "default" retry settings are if it is not specified on an Activity or for any unset fields if a policy is explicitly set on a workflow

func FromConfigToDefaultRetrySettings

func FromConfigToDefaultRetrySettings(options map[string]interface{}) DefaultRetrySettings

type PingCheck

type PingCheck struct {
	// Name of this component.
	Name string
	// The longest time that Ping can take. If it doesn't return in that much time, that's
	// considered a deadlock and the deadlock detector may take actions to recover, like
	// killing the process.
	Timeout time.Duration
	// Perform the check. The typical implementation will just be Lock() and then Unlock()
	// on a mutex, returning nil. Ping can also return more Pingables for sub-components
	// that will be checked independently. These should form a tree and not lead to cycles.
	Ping func() []Pingable

	// Metrics recording:
	// Timer id within DeadlockDetectorScope (or zero for no metrics)
	MetricsName string
}

type Pingable

type Pingable interface {
	GetPingChecks() []PingCheck
}

Pingable is interface to check for liveness of a component, to detect deadlocks. This call should not block.

type RPCFactory

type RPCFactory interface {
	GetFrontendGRPCServerOptions() ([]grpc.ServerOption, error)
	GetInternodeGRPCServerOptions() ([]grpc.ServerOption, error)
	GetGRPCListener() net.Listener
	CreateRemoteFrontendGRPCConnection(rpcAddress string) *grpc.ClientConn
	CreateLocalFrontendGRPCConnection() *grpc.ClientConn
	CreateInternodeGRPCConnection(rpcAddress string) *grpc.ClientConn
}

RPCFactory creates gRPC listener and connection.

type TaskTokenSerializer

type TaskTokenSerializer interface {
	Serialize(token *tokenspb.Task) ([]byte, error)
	Deserialize(data []byte) (*tokenspb.Task, error)
	SerializeQueryTaskToken(token *tokenspb.QueryTask) ([]byte, error)
	DeserializeQueryTaskToken(data []byte) (*tokenspb.QueryTask, error)
}

TaskTokenSerializer serializes task tokens

func NewProtoTaskTokenSerializer

func NewProtoTaskTokenSerializer() TaskTokenSerializer

NewProtoTaskTokenSerializer creates a new instance of TaskTokenSerializer