queues

May 31, 2024

Changes in this version

+ func FromPersistenceDestinationPredicate(attributes *persistencespb.DestinationPredicateAttributes) tasks.Predicate

+ func FromPersistenceStateMachineTypePredicate(attributes *persistencespb.StateMachineTaskTypePredicateAttributes) tasks.Predicate

+ func ToPersistenceDestinationPredicate(taskDestinationPredicate *tasks.DestinationPredicate) *persistencespb.Predicate

+ func ToPersistenceStateMachineTaskTypePredicate(pred *tasks.StateMachineTaskTypePredicate) *persistencespb.Predicate

+ type CircuitBreakerExecutable struct

+ func NewCircuitBreakerExecutable(e Executable, cb *gobreaker.TwoStepCircuitBreaker) *CircuitBreakerExecutable

+ func (e *CircuitBreakerExecutable) Execute() error

+ type CommonSchedulerWrapper struct

+ TaskKeyFn func(e Executable) TaskChannelKey

+ func (s *CommonSchedulerWrapper) TaskChannelKeyFn() TaskChannelKeyFn

+ type DestinationDownError struct

+ Message string

+ func NewDestinationDownError(msg string, err error) *DestinationDownError

+ func (e *DestinationDownError) Error() string

+ func (e *DestinationDownError) Unwrap() error

type ExecutableParams

+ DLQErrorPattern dynamicconfig.StringPropertyFn

+ DLQInternalErrors dynamicconfig.BoolPropertyFn

+ type Grouper interface

+ Key func(task tasks.Task) (key any)

+ Predicate func(keys []any) tasks.Predicate

+ type GrouperNamespaceID struct

+ func (GrouperNamespaceID) Key(task tasks.Task) (key any)

+ func (GrouperNamespaceID) Predicate(keys []any) tasks.Predicate

+ type GrouperStateMachineNamespaceIDAndDestination struct

+ func (GrouperStateMachineNamespaceIDAndDestination) KeyTyped(task tasks.Task) (key StateMachineTaskTypeNamespaceIDAndDestination)

+ func (GrouperStateMachineNamespaceIDAndDestination) Predicate(keys []any) tasks.Predicate

+ func (g GrouperStateMachineNamespaceIDAndDestination) Key(task tasks.Task) (key any)

+ type MaybeTerminalTaskError interface

+ IsTerminalTaskError func() bool

+ type MockMaybeTerminalTaskError struct

+ func NewMockMaybeTerminalTaskError(ctrl *gomock.Controller) *MockMaybeTerminalTaskError

+ func (m *MockMaybeTerminalTaskError) EXPECT() *MockMaybeTerminalTaskErrorMockRecorder

+ func (m *MockMaybeTerminalTaskError) IsTerminalTaskError() bool

+ type MockMaybeTerminalTaskErrorMockRecorder struct

+ func (mr *MockMaybeTerminalTaskErrorMockRecorder) IsTerminalTaskError() *gomock.Call

+ type NamespaceIDAndDestination struct

+ Destination string

+ NamespaceID string

+ type StateMachineTaskTypeNamespaceIDAndDestination struct

+ Destination string

+ NamespaceID string

+ StateMachineTaskType int32

type TaskStats

+ PendingPerKey map[any]int

+ type UnprocessableTaskError struct

+ Message string

+ func NewUnprocessableTaskError(message string) UnprocessableTaskError

+ func (UnprocessableTaskError) IsTerminalTaskError() bool

+ func (e UnprocessableTaskError) Error() string

Mar 22, 2024

Changes in this version

+ var ErrCreateDLQ = errors.New("failed to create DLQ")

+ var ErrGetClusterMetadata = errors.New("failed to get cluster metadata")

+ var ErrSendTaskToDLQ = errors.New("failed to send task to DLQ")

+ var ErrTerminalTaskFailure = errors.New("original task failed and this task is now to send the original to the DLQ")

+ func EstimateTaskMetricTag(e Executable, namespaceRegistry namespace.Registry, currentClusterName string) []metrics.Tag

+ func NewExecutableFactory(executor Executor, scheduler Scheduler, rescheduler Rescheduler, ...) *executableFactoryImpl

+ type DLQWriter struct

+ func NewDLQWriter(w QueueWriter, m cluster.Metadata, h metrics.Handler, l log.SnTaggedLogger, ...) *DLQWriter

+ func (q *DLQWriter) WriteTaskToDLQ(ctx context.Context, sourceCluster, targetCluster string, task tasks.Task) error

+ type ExecutableFactory interface

+ NewExecutable func(task tasks.Task, readerID int64) Executable

+ type ExecutableFactoryFn func(readerID int64, t tasks.Task) Executable

+ func (f ExecutableFactoryFn) NewExecutable(task tasks.Task, readerID int64) Executable

+ type ExecutableOption func(*ExecutableParams)

+ type ExecutableParams struct

+ DLQEnabled dynamicconfig.BoolPropertyFn

+ DLQWriter *DLQWriter

+ MaxUnexpectedErrorAttempts dynamicconfig.IntPropertyFn

+ type ExecuteResponse struct

+ ExecutedAsActive bool

+ ExecutionErr error

+ ExecutionMetricTags []metrics.Tag

+ type MockTerminalTaskError struct

+ func NewMockTerminalTaskError(ctrl *gomock.Controller) *MockTerminalTaskError

+ func (m *MockTerminalTaskError) EXPECT() *MockTerminalTaskErrorMockRecorder

+ func (m *MockTerminalTaskError) IsTerminalTaskError()

+ type MockTerminalTaskErrorMockRecorder struct

+ func (mr *MockTerminalTaskErrorMockRecorder) IsTerminalTaskError() *gomock.Call

type PriorityAssigner

+ func NewStaticPriorityAssigner(priority tasks.Priority) PriorityAssigner

+ type QueueWriter interface

+ CreateQueue func(ctx context.Context, request *persistence.CreateQueueRequest) (*persistence.CreateQueueResponse, error)

+ EnqueueTask func(ctx context.Context, request *persistence.EnqueueTaskRequest) (*persistence.EnqueueTaskResponse, error)

+ type RateLimitedSchedulerOptions struct

+ EnableShadowMode dynamicconfig.BoolPropertyFn

+ StartupDelay dynamicconfig.DurationPropertyFn

type Scheduler

+ func NewRateLimitedScheduler(baseScheduler Scheduler, options RateLimitedSchedulerOptions, ...) Scheduler

type SchedulerOptions

+ ActiveNamespaceWeights dynamicconfig.MapPropertyFnWithNamespaceFilter

+ StandbyNamespaceWeights dynamicconfig.MapPropertyFnWithNamespaceFilter

+ WorkerCount dynamicconfig.IntPropertyFn

type SchedulerRateLimiter

+ func NewPrioritySchedulerRateLimiter(namespaceRateFn quotas.NamespaceRateFn, hostRateFn quotas.RateFn, ...) (SchedulerRateLimiter, error)

+ type TerminalTaskError interface

+ IsTerminalTaskError func()

Feb 29, 2024 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Feb 22, 2024 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Feb 22, 2024 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Feb 2, 2024 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Jan 12, 2024 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Dec 7, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Nov 30, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Nov 28, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Nov 22, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Nov 13, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Nov 3, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Nov 1, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Oct 31, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Oct 13, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Oct 6, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Nov 1, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Sep 5, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Changes in this version

type Executor

+ func NewActiveStandbyExecutor(currentClusterName string, registry namespace.Registry, ...) Executor

+ type ExecutorWrapper interface

+ Wrap func(delegate Executor) Executor

+ type MockExecutorWrapper struct

+ func NewMockExecutorWrapper(ctrl *gomock.Controller) *MockExecutorWrapper

+ func (m *MockExecutorWrapper) EXPECT() *MockExecutorWrapperMockRecorder

+ func (m *MockExecutorWrapper) Wrap(delegate Executor) Executor

+ type MockExecutorWrapperMockRecorder struct

+ func (mr *MockExecutorWrapperMockRecorder) Wrap(delegate interface{}) *gomock.Call

type Queue

+ Start func()

+ Stop func()

type Reader

+ Start func()

+ Stop func()

type Rescheduler

+ Start func()

+ Stop func()

type Scheduler

+ Start func()

+ Stop func()

Aug 28, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Aug 21, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Aug 14, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Aug 14, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Aug 14, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Aug 8, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Jul 28, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Jul 24, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Jul 14, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Jul 1, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Jun 23, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Changes in this version

+ var NoopReaderCompletionFn = func(_ int64)

type Action

+ Name func() string

type Iterator

+ HasNext func(readerID int64) bool

+ Next func(readerID int64) (tasks.Task, error)

type MockExecutable

+ func (m *MockExecutable) Abort()

type MockExecutableMockRecorder

+ func (mr *MockExecutableMockRecorder) Abort() *gomock.Call

type Monitor

+ SilenceAlert func(AlertType)

type Reader

+ Notify func()

+ type ReaderCompletionFn func(readerID int64)

type ReaderGroup

+ func (g *ReaderGroup) ForEach(f func(int64, Reader))

+ func (g *ReaderGroup) GetOrCreateReader(readerID int64) (Reader, error)

type ReaderImpl

+ func (r *ReaderImpl) Notify()

+ type SpeculativeWorkflowTaskTimeoutQueue struct

+ func NewSpeculativeWorkflowTaskTimeoutQueue(scheduler ctasks.Scheduler[ctasks.Task], priorityAssigner PriorityAssigner, ...) *SpeculativeWorkflowTaskTimeoutQueue

+ func (q SpeculativeWorkflowTaskTimeoutQueue) Category() tasks.Category

+ func (q SpeculativeWorkflowTaskTimeoutQueue) FailoverNamespace(_ string)

+ func (q SpeculativeWorkflowTaskTimeoutQueue) NotifyNewTasks(ts []tasks.Task)

+ func (q SpeculativeWorkflowTaskTimeoutQueue) Start()

+ func (q SpeculativeWorkflowTaskTimeoutQueue) Stop()

Jul 13, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

May 15, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Apr 17, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Mar 29, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Feb 17, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Changes in this version

type NamespacePrioritySchedulerOptions

+ DispatchThrottleDuration dynamicconfig.DurationPropertyFn

+ EnableRateLimiterShadowMode dynamicconfig.BoolPropertyFn

type PrioritySchedulerOptions

+ DispatchThrottleDuration dynamicconfig.DurationPropertyFn

+ EnableRateLimiterShadowMode dynamicconfig.BoolPropertyFn

Jan 13, 2023 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Dec 1, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Changes in this version

+ func GetArchivalTaskTypeTagValue(task tasks.Task) string

type NamespacePrioritySchedulerOptions

+ EnableRateLimiter dynamicconfig.BoolPropertyFn

+ MaxDispatchThrottleDuration time.Duration

+ type PrioritySchedulerOptions struct

+ EnableRateLimiter dynamicconfig.BoolPropertyFn

+ MaxDispatchThrottleDuration time.Duration

+ Weight dynamicconfig.MapPropertyFn

+ WorkerCount dynamicconfig.IntPropertyFn

type Scheduler

+ func NewPriorityScheduler(options PrioritySchedulerOptions, rateLimiter SchedulerRateLimiter, ...) Scheduler

+ type SchedulerRateLimiter quotas.RequestRateLimiter

+ func NewSchedulerRateLimiter(namespaceMaxQPS dynamicconfig.IntPropertyFnWithNamespaceFilter, ...) SchedulerRateLimiter

Nov 15, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Oct 31, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Oct 20, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Oct 18, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Oct 10, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Sep 16, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Changes in this version

+ const DefaultReaderId

+ const QueueActionCounter

+ const QueueReaderCountHistogram

+ const QueueScheduleLatency

+ const QueueSliceCountHistogram

+ const TaskLoadLatency

+ const TaskScheduleLatency

+ func IsTaskAcked(task tasks.Task, persistenceQueueState *persistencespb.QueueState) bool

+ func IsTimeExpired(referenceTime time.Time, testingTime time.Time) bool

+ func NewReaderPriorityRateLimiter(rateFn quotas.RateFn, maxReaders int) quotas.RequestRateLimiter

+ type Action interface

+ Run func(*ReaderGroup)

+ type Alert struct

+ AlertAttributesQueuePendingTaskCount *AlertAttributesQueuePendingTaskCount

+ AlertAttributesReaderStuck *AlertAttributesReaderStuck

+ AlertAttributesSliceCount *AlertAttributesSlicesCount

+ AlertType AlertType

+ type AlertAttributesQueuePendingTaskCount struct

+ CiriticalPendingTaskCount int

+ CurrentPendingTaskCount int

+ type AlertAttributesReaderStuck struct

+ CurrentWatermark tasks.Key

+ ReaderID int32

+ type AlertAttributesSlicesCount struct

+ CriticalSliceCount int

+ CurrentSliceCount int

+ type AlertType int

+ const AlertTypeQueuePendingTaskCount

+ const AlertTypeReaderStuck

+ const AlertTypeSliceCount

+ const AlertTypeUnspecified

+ func (a AlertType) String() string

+ type ChannelWeightFn = tasks.ChannelWeightFn[TaskChannelKey]

type Executable

+ GetPriority func() ctasks.Priority

+ GetScheduledTime func() time.Time

+ SetScheduledTime func(time.Time)

+ type FIFOSchedulerOptions struct

+ QueueSize int

+ WorkerCount dynamicconfig.IntPropertyFn

+ type Mitigator interface

+ Mitigate func(Alert) Action

type MockExecutable

+ func (m *MockExecutable) GetScheduledTime() time.Time

+ func (m *MockExecutable) SetScheduledTime(arg0 time.Time)

type MockExecutableMockRecorder

+ func (mr *MockExecutableMockRecorder) GetScheduledTime() *gomock.Call

+ func (mr *MockExecutableMockRecorder) SetScheduledTime(arg0 interface{}) *gomock.Call

type MockScheduler

+ func (m *MockScheduler) ChannelWeightFn() ChannelWeightFn

+ func (m *MockScheduler) TaskChannelKeyFn() TaskChannelKeyFn

type MockSchedulerMockRecorder

+ func (mr *MockSchedulerMockRecorder) ChannelWeightFn() *gomock.Call

+ func (mr *MockSchedulerMockRecorder) TaskChannelKeyFn() *gomock.Call

+ type Monitor interface

+ AlertCh func() <-chan *Alert

+ Close func()

+ GetReaderWatermark func(readerID int32) (tasks.Key, bool)

+ GetSliceCount func(readerID int32) int

+ GetSlicePendingTaskCount func(slice Slice) int

+ GetTotalPendingTaskCount func() int

+ GetTotalSliceCount func() int

+ RemoveReader func(readerID int32)

+ RemoveSlice func(slice Slice)

+ ResolveAlert func(AlertType)

+ SetReaderWatermark func(readerID int32, watermark tasks.Key)

+ SetSliceCount func(readerID int32, count int)

+ SetSlicePendingTaskCount func(slice Slice, count int)

+ type MonitorOptions struct

+ PendingTasksCriticalCount dynamicconfig.IntPropertyFn

+ ReaderStuckCriticalAttempts dynamicconfig.IntPropertyFn

+ SliceCountCriticalThreshold dynamicconfig.IntPropertyFn

+ type NamespacePrioritySchedulerOptions struct

+ ActiveNamespaceWeights dynamicconfig.MapPropertyFnWithNamespaceFilter

+ StandbyNamespaceWeights dynamicconfig.MapPropertyFnWithNamespaceFilter

+ WorkerCount dynamicconfig.IntPropertyFn

type Options

+ MaxPollRPS dynamicconfig.IntPropertyFn

+ MaxReaderCount dynamicconfig.IntPropertyFn

type Reader

+ AppendSlices func(...Slice)

+ CompactSlices func(SlicePredicate)

type ReaderGroup

+ func (g *ReaderGroup) RemoveReader(readerID int32)

type ReaderImpl

+ func (r *ReaderImpl) AppendSlices(incomingSlices ...Slice)

+ func (r *ReaderImpl) CompactSlices(predicate SlicePredicate)

type Scheduler

+ func NewFIFOScheduler(options FIFOSchedulerOptions, logger log.Logger) Scheduler

+ func NewNamespacePriorityScheduler(currentClusterName string, options NamespacePrioritySchedulerOptions, ...) Scheduler

+ ChannelWeightFn func() ChannelWeightFn

+ TaskChannelKeyFn func() TaskChannelKeyFn

type Slice

+ CompactWithSlice func(Slice) Slice

+ ShrinkScope func()

+ TaskStats func() TaskStats

type SliceImpl

+ func (s *SliceImpl) CompactWithSlice(slice Slice) Slice

+ func (s *SliceImpl) ShrinkScope()

+ func (s *SliceImpl) TaskStats() TaskStats

+ type TaskChannelKey struct

+ NamespaceID string

+ Priority tasks.Priority

+ type TaskChannelKeyFn = tasks.TaskChannelKeyFn[Executable, TaskChannelKey]

+ type TaskStats struct

+ PendingPerNamespace map[namespace.ID]int

Sep 14, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Aug 30, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Aug 16, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Aug 17, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Changes in this version

+ const AckLevelUpdateCounter

+ const AckLevelUpdateFailedCounter

+ const FactoryFxGroup

+ const NewTimerNotifyCounter

+ const OperationTimerQueueProcessor

+ const OperationTransferQueueProcessor

+ const PendingTasksCounter

+ const TaskBatchCompleteCounter

+ func FromPersistenceAndPredicate(attributes *persistencespb.AndPredicateAttributes) tasks.Predicate

+ func FromPersistenceEmptyPredicate(_ *persistencespb.EmptyPredicateAttributes) tasks.Predicate

+ func FromPersistenceNamespaceIDPredicate(attributes *persistencespb.NamespaceIdPredicateAttributes) tasks.Predicate

+ func FromPersistenceNotPredicate(attributes *persistencespb.NotPredicateAttributes) tasks.Predicate

+ func FromPersistenceOrPredicate(attributes *persistencespb.OrPredicateAttributes) tasks.Predicate

+ func FromPersistencePredicate(predicate *persistencespb.Predicate) tasks.Predicate

+ func FromPersistenceQueueState(state *persistencespb.QueueState) *queueState

+ func FromPersistenceTaskKey(key *persistencespb.TaskKey) tasks.Key

+ func FromPersistenceTaskTypePredicate(attributes *persistencespb.TaskTypePredicateAttributes) tasks.Predicate

+ func FromPersistenceUniversalPredicate(_ *persistencespb.UniversalPredicateAttributes) tasks.Predicate

+ func NewImmediateQueue(shard shard.Context, category tasks.Category, scheduler Scheduler, ...) *immediateQueue

+ func NewScheduledQueue(shard shard.Context, category tasks.Category, scheduler Scheduler, ...) *scheduledQueue

+ func ToPersistenceAndPredicate(andPredicate *predicates.AndImpl[tasks.Task]) *persistencespb.Predicate

+ func ToPersistenceEmptyPredicate(_ *predicates.EmptyImpl[tasks.Task]) *persistencespb.Predicate

+ func ToPersistenceNamespaceIDPredicate(namespaceIDPredicate *tasks.NamespacePredicate) *persistencespb.Predicate

+ func ToPersistenceNotPredicate(notPredicate *predicates.NotImpl[tasks.Task]) *persistencespb.Predicate

+ func ToPersistenceOrPredicate(orPredicate *predicates.OrImpl[tasks.Task]) *persistencespb.Predicate

+ func ToPersistencePredicate(predicate tasks.Predicate) *persistencespb.Predicate

+ func ToPersistenceQueueState(queueState *queueState) *persistencespb.QueueState

+ func ToPersistenceRange(r Range) *persistencespb.QueueSliceRange

+ func ToPersistenceScope(scope Scope) *persistencespb.QueueSliceScope

+ func ToPersistenceTaskKey(key tasks.Key) *persistencespb.TaskKey

+ func ToPersistenceTaskTypePredicate(taskTypePredicate *tasks.TypePredicate) *persistencespb.Predicate

+ func ToPersistenceUniversalPredicate(_ *predicates.UniversalImpl[tasks.Task]) *persistencespb.Predicate

+ type ExecutableInitializer func(tasks.Task) Executable

+ type Factory interface

+ CreateQueue func(shard shard.Context, engine shard.Engine, cache workflow.Cache) Queue

type MockExecutable

+ func (m *MockExecutable) Cancel()

type MockExecutableMockRecorder

+ func (mr *MockExecutableMockRecorder) Cancel() *gomock.Call

+ type MockFactory struct

+ func NewMockFactory(ctrl *gomock.Controller) *MockFactory

+ func (m *MockFactory) CreateQueue(shard shard.Context, engine shard.Engine, cache workflow.Cache) Queue

+ func (m *MockFactory) EXPECT() *MockFactoryMockRecorder

+ func (m *MockFactory) Start()

+ func (m *MockFactory) Stop()

+ type MockFactoryMockRecorder struct

+ func (mr *MockFactoryMockRecorder) CreateQueue(shard, engine, cache interface{}) *gomock.Call

+ func (mr *MockFactoryMockRecorder) Start() *gomock.Call

+ func (mr *MockFactoryMockRecorder) Stop() *gomock.Call

+ type MockQueue struct

+ func NewMockQueue(ctrl *gomock.Controller) *MockQueue

+ func (m *MockQueue) Category() tasks.Category

+ func (m *MockQueue) EXPECT() *MockQueueMockRecorder

+ func (m *MockQueue) FailoverNamespace(namespaceIDs map[string]struct{})

+ func (m *MockQueue) LockTaskProcessing()

+ func (m *MockQueue) NotifyNewTasks(clusterName string, tasks []tasks.Task)

+ func (m *MockQueue) Start()

+ func (m *MockQueue) Stop()

+ func (m *MockQueue) UnlockTaskProcessing()

+ type MockQueueMockRecorder struct

+ func (mr *MockQueueMockRecorder) Category() *gomock.Call

+ func (mr *MockQueueMockRecorder) FailoverNamespace(namespaceIDs interface{}) *gomock.Call

+ func (mr *MockQueueMockRecorder) LockTaskProcessing() *gomock.Call

+ func (mr *MockQueueMockRecorder) NotifyNewTasks(clusterName, tasks interface{}) *gomock.Call

+ func (mr *MockQueueMockRecorder) Start() *gomock.Call

+ func (mr *MockQueueMockRecorder) Stop() *gomock.Call

+ func (mr *MockQueueMockRecorder) UnlockTaskProcessing() *gomock.Call

type MockRescheduler

+ func (m *MockRescheduler) Start()

+ func (m *MockRescheduler) Stop()

type MockReschedulerMockRecorder

+ func (mr *MockReschedulerMockRecorder) Start() *gomock.Call

+ func (mr *MockReschedulerMockRecorder) Stop() *gomock.Call

+ type Options struct

+ CheckpointInterval dynamicconfig.DurationPropertyFn

+ CheckpointIntervalJitterCoefficient dynamicconfig.FloatPropertyFn

+ MaxPollInterval dynamicconfig.DurationPropertyFn

+ MaxPollIntervalJitterCoefficient dynamicconfig.FloatPropertyFn

+ TaskMaxRetryCount dynamicconfig.IntPropertyFn

+ type PaginationFnProvider func(Range) collection.PaginationFn[tasks.Task]

+ type Queue interface

+ Category func() tasks.Category

+ FailoverNamespace func(namespaceIDs map[string]struct{})

+ LockTaskProcessing func()

+ NotifyNewTasks func(clusterName string, tasks []tasks.Task)

+ UnlockTaskProcessing func()

type Range

+ func FromPersistenceRange(r *persistencespb.QueueSliceRange) Range

+ func NewRandomOrderedRangesInRange(r Range, numRanges int) []Range

+ func (r *Range) Equals(input Range) bool

+ type Reader interface

+ ClearSlices func(SlicePredicate)

+ MergeSlices func(...Slice)

+ Pause func(time.Duration)

+ Scopes func() []Scope

+ ShrinkSlices func()

+ SplitSlices func(SliceSplitter)

+ WalkSlices func(SliceIterator)

+ type ReaderGroup struct

+ func NewReaderGroup(initializer ReaderInitializer) *ReaderGroup

+ func (g *ReaderGroup) NewReader(readerID int32, slices ...Slice) Reader

+ func (g *ReaderGroup) ReaderByID(readerID int32) (Reader, bool)

+ func (g *ReaderGroup) Readers() map[int32]Reader

+ func (g *ReaderGroup) Start()

+ func (g *ReaderGroup) Stop()

+ type ReaderImpl struct

+ func NewReader(slices []Slice, options *ReaderOptions, scheduler Scheduler, ...) *ReaderImpl

+ func (r *ReaderImpl) ClearSlices(selector SlicePredicate)

+ func (r *ReaderImpl) MergeSlices(incomingSlices ...Slice)

+ func (r *ReaderImpl) Pause(duration time.Duration)

+ func (r *ReaderImpl) Scopes() []Scope

+ func (r *ReaderImpl) ShrinkSlices()

+ func (r *ReaderImpl) SplitSlices(splitter SliceSplitter)

+ func (r *ReaderImpl) Start()

+ func (r *ReaderImpl) Stop()

+ func (r *ReaderImpl) WalkSlices(iterator SliceIterator)

+ type ReaderInitializer func(readerID int32, slices []Slice) Reader

+ type ReaderOptions struct

+ BatchSize dynamicconfig.IntPropertyFn

+ MaxPendingTasksCount dynamicconfig.IntPropertyFn

+ PollBackoffInterval dynamicconfig.DurationPropertyFn

type Scope

+ func FromPersistenceScope(scope *persistencespb.QueueSliceScope) Scope

+ func NewRandomScopes(numScopes int) []Scope

+ func (s *Scope) Equals(scope Scope) bool

+ func (s *Scope) IsEmpty() bool

+ type Slice interface

+ CanMergeWithSlice func(Slice) bool

+ CanSplitByRange func(tasks.Key) bool

+ Clear func()

+ MergeWithSlice func(Slice) []Slice

+ MoreTasks func() bool

+ Scope func() Scope

+ SelectTasks func(int) ([]Executable, error)

+ ShrinkRange func()

+ SplitByPredicate func(tasks.Predicate) (pass Slice, fail Slice)

+ SplitByRange func(tasks.Key) (left Slice, right Slice)

+ type SliceImpl struct

+ func NewSlice(paginationFnProvider PaginationFnProvider, ...) *SliceImpl

+ func (s *SliceImpl) CanMergeWithSlice(slice Slice) bool

+ func (s *SliceImpl) CanSplitByRange(key tasks.Key) bool

+ func (s *SliceImpl) Clear()

+ func (s *SliceImpl) MergeWithSlice(slice Slice) []Slice

+ func (s *SliceImpl) MoreTasks() bool

+ func (s *SliceImpl) Scope() Scope

+ func (s *SliceImpl) SelectTasks(batchSize int) ([]Executable, error)

+ func (s *SliceImpl) ShrinkRange()

+ func (s *SliceImpl) SplitByPredicate(predicate tasks.Predicate) (pass Slice, fail Slice)

+ func (s *SliceImpl) SplitByRange(key tasks.Key) (left Slice, right Slice)

+ type SliceIterator func(s Slice)

+ type SlicePredicate func(s Slice) bool

+ type SliceSplitter func(s Slice) (remaining []Slice, split bool)

Aug 2, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Jul 11, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Jun 17, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Changes in this version

+ const OperationTaskPriorityAssigner

+ const OperationTimerActiveQueueProcessor

+ const OperationTimerStandbyQueueProcessor

+ const OperationTransferActiveQueueProcessor

+ const OperationTransferStandbyQueueProcessor

+ const OperationVisibilityQueueProcessor

+ const TaskAttempt

+ const TaskDiscarded

+ const TaskFailures

+ const TaskLatency

+ const TaskNoUserLatency

+ const TaskNoUserProcessingLatency

+ const TaskNoUserQueueLatency

+ const TaskNotActiveCounter

+ const TaskProcessingLatency

+ const TaskQueueLatency

+ const TaskRequests

+ const TaskReschedulerPendingTasks

+ const TaskSkipped

+ const TaskStandbyRetryCounter

+ const TaskThrottledCounter

+ const TaskTypeTimerActiveTaskActivityRetryTimer

+ const TaskTypeTimerActiveTaskActivityTimeout

+ const TaskTypeTimerActiveTaskDeleteHistoryEvent

+ const TaskTypeTimerActiveTaskUserTimer

+ const TaskTypeTimerActiveTaskWorkflowBackoffTimer

+ const TaskTypeTimerActiveTaskWorkflowTaskTimeout

+ const TaskTypeTimerActiveTaskWorkflowTimeout

+ const TaskTypeTimerStandbyTaskActivityRetryTimer

+ const TaskTypeTimerStandbyTaskActivityTimeout

+ const TaskTypeTimerStandbyTaskDeleteHistoryEvent

+ const TaskTypeTimerStandbyTaskUserTimer

+ const TaskTypeTimerStandbyTaskWorkflowBackoffTimer

+ const TaskTypeTimerStandbyTaskWorkflowTaskTimeout

+ const TaskTypeTimerStandbyTaskWorkflowTimeout

+ const TaskTypeTransferActiveTaskActivity

+ const TaskTypeTransferActiveTaskCancelExecution

+ const TaskTypeTransferActiveTaskCloseExecution

+ const TaskTypeTransferActiveTaskResetWorkflow

+ const TaskTypeTransferActiveTaskSignalExecution

+ const TaskTypeTransferActiveTaskStartChildExecution

+ const TaskTypeTransferActiveTaskWorkflowTask

+ const TaskTypeTransferStandbyTaskActivity

+ const TaskTypeTransferStandbyTaskCancelExecution

+ const TaskTypeTransferStandbyTaskCloseExecution

+ const TaskTypeTransferStandbyTaskResetWorkflow

+ const TaskTypeTransferStandbyTaskSignalExecution

+ const TaskTypeTransferStandbyTaskStartChildExecution

+ const TaskTypeTransferStandbyTaskWorkflowTask

+ const TaskTypeVisibilityTaskCloseExecution

+ const TaskTypeVisibilityTaskDeleteExecution

+ const TaskTypeVisibilityTaskStartExecution

+ const TaskTypeVisibilityTaskUpsertExecution

+ const TaskUserLatency

+ const TaskWorkflowBusyCounter

+ func GetActiveTimerTaskTypeTagValue(task tasks.Task) string

+ func GetActiveTransferTaskTypeTagValue(task tasks.Task) string

+ func GetStandbyTimerTaskTypeTagValue(task tasks.Task) string

+ func GetStandbyTransferTaskTypeTagValue(task tasks.Task) string

+ func GetVisibilityTaskTypeTagValue(task tasks.Task) string

+ func NewRandomKey() tasks.Key

+ func NewRandomKeyInRange(r Range) tasks.Key

+ func NewRescheduler(scheduler Scheduler, timeSource clock.TimeSource, ...) *reschedulerImpl

+ func NewScheduler(priorityAssigner PriorityAssigner, options SchedulerOptions, ...) *schedulerImpl

+ type Executable interface

+ Attempt func() int

+ GetTask func() tasks.Task

+ Logger func() log.Logger

+ QueueType func() QueueType

+ func NewExecutable(task tasks.Task, filter TaskFilter, executor Executor, scheduler Scheduler, ...) Executable

+ type Executor interface

+ Execute func(context.Context, Executable) (metrics.MetricsHandler, error)

+ func NewExecutorWrapper(currentClusterName string, registry namespace.Registry, ...) Executor

+ type Iterator interface

+ CanMerge func(Iterator) bool

+ CanSplit func(tasks.Key) bool

+ Merge func(Iterator) Iterator

+ Range func() Range

+ Remaining func() Iterator

+ Split func(key tasks.Key) (left Iterator, right Iterator)

+ type IteratorImpl struct

+ func NewIterator(paginationFnProvider paginationFnProvider, r Range) *IteratorImpl

+ func (i *IteratorImpl) CanMerge(iter Iterator) bool

+ func (i *IteratorImpl) CanSplit(key tasks.Key) bool

+ func (i *IteratorImpl) HasNext() bool

+ func (i *IteratorImpl) Merge(iter Iterator) Iterator

+ func (i *IteratorImpl) Next() (tasks.Task, error)

+ func (i *IteratorImpl) Range() Range

+ func (i *IteratorImpl) Remaining() Iterator

+ func (i *IteratorImpl) Split(key tasks.Key) (left Iterator, right Iterator)

+ type MockExecutable struct

+ func NewMockExecutable(ctrl *gomock.Controller) *MockExecutable

+ func (m *MockExecutable) Ack()

+ func (m *MockExecutable) Attempt() int

+ func (m *MockExecutable) EXPECT() *MockExecutableMockRecorder

+ func (m *MockExecutable) Execute() error

+ func (m *MockExecutable) GetCategory() tasks0.Category

+ func (m *MockExecutable) GetKey() tasks0.Key

+ func (m *MockExecutable) GetNamespaceID() string

+ func (m *MockExecutable) GetPriority() tasks.Priority

+ func (m *MockExecutable) GetRunID() string

+ func (m *MockExecutable) GetTask() tasks0.Task

+ func (m *MockExecutable) GetTaskID() int64

+ func (m *MockExecutable) GetType() v1.TaskType

+ func (m *MockExecutable) GetVersion() int64

+ func (m *MockExecutable) GetVisibilityTime() time.Time

+ func (m *MockExecutable) GetWorkflowID() string

+ func (m *MockExecutable) HandleErr(err error) error

+ func (m *MockExecutable) IsRetryableError(err error) bool

+ func (m *MockExecutable) Logger() log.Logger

+ func (m *MockExecutable) Nack(err error)

+ func (m *MockExecutable) QueueType() QueueType

+ func (m *MockExecutable) Reschedule()

+ func (m *MockExecutable) RetryPolicy() backoff.RetryPolicy

+ func (m *MockExecutable) SetPriority(arg0 tasks.Priority)

+ func (m *MockExecutable) SetTaskID(id int64)

+ func (m *MockExecutable) SetVersion(version int64)

+ func (m *MockExecutable) SetVisibilityTime(timestamp time.Time)

+ func (m *MockExecutable) State() tasks.State

+ type MockExecutableMockRecorder struct

+ func (mr *MockExecutableMockRecorder) Ack() *gomock.Call

+ func (mr *MockExecutableMockRecorder) Attempt() *gomock.Call

+ func (mr *MockExecutableMockRecorder) Execute() *gomock.Call

+ func (mr *MockExecutableMockRecorder) GetCategory() *gomock.Call

+ func (mr *MockExecutableMockRecorder) GetKey() *gomock.Call

+ func (mr *MockExecutableMockRecorder) GetNamespaceID() *gomock.Call

+ func (mr *MockExecutableMockRecorder) GetPriority() *gomock.Call

+ func (mr *MockExecutableMockRecorder) GetRunID() *gomock.Call

+ func (mr *MockExecutableMockRecorder) GetTask() *gomock.Call

+ func (mr *MockExecutableMockRecorder) GetTaskID() *gomock.Call

+ func (mr *MockExecutableMockRecorder) GetType() *gomock.Call

+ func (mr *MockExecutableMockRecorder) GetVersion() *gomock.Call

+ func (mr *MockExecutableMockRecorder) GetVisibilityTime() *gomock.Call

+ func (mr *MockExecutableMockRecorder) GetWorkflowID() *gomock.Call

+ func (mr *MockExecutableMockRecorder) HandleErr(err interface{}) *gomock.Call

+ func (mr *MockExecutableMockRecorder) IsRetryableError(err interface{}) *gomock.Call

+ func (mr *MockExecutableMockRecorder) Logger() *gomock.Call

+ func (mr *MockExecutableMockRecorder) Nack(err interface{}) *gomock.Call

+ func (mr *MockExecutableMockRecorder) QueueType() *gomock.Call

+ func (mr *MockExecutableMockRecorder) Reschedule() *gomock.Call

+ func (mr *MockExecutableMockRecorder) RetryPolicy() *gomock.Call

+ func (mr *MockExecutableMockRecorder) SetPriority(arg0 interface{}) *gomock.Call

+ func (mr *MockExecutableMockRecorder) SetTaskID(id interface{}) *gomock.Call

+ func (mr *MockExecutableMockRecorder) SetVersion(version interface{}) *gomock.Call

+ func (mr *MockExecutableMockRecorder) SetVisibilityTime(timestamp interface{}) *gomock.Call

+ func (mr *MockExecutableMockRecorder) State() *gomock.Call

+ type MockExecutor struct

+ func NewMockExecutor(ctrl *gomock.Controller) *MockExecutor

+ func (m *MockExecutor) EXPECT() *MockExecutorMockRecorder

+ func (m *MockExecutor) Execute(arg0 context.Context, arg1 Executable) (metrics.MetricsHandler, error)

+ type MockExecutorMockRecorder struct

+ func (mr *MockExecutorMockRecorder) Execute(arg0, arg1 interface{}) *gomock.Call

type MockProcessorFactory

+ func (m *MockProcessorFactory) Start()

+ func (m *MockProcessorFactory) Stop()

type MockProcessorFactoryMockRecorder

+ func (mr *MockProcessorFactoryMockRecorder) Start() *gomock.Call

+ func (mr *MockProcessorFactoryMockRecorder) Stop() *gomock.Call

+ type MockRescheduler struct

+ func NewMockRescheduler(ctrl *gomock.Controller) *MockRescheduler

+ func (m *MockRescheduler) Add(task Executable, backoff time.Duration)

+ func (m *MockRescheduler) EXPECT() *MockReschedulerMockRecorder

+ func (m *MockRescheduler) Len() int

+ func (m *MockRescheduler) Reschedule(targetRescheduleSize int)

+ type MockReschedulerMockRecorder struct

+ func (mr *MockReschedulerMockRecorder) Add(task, backoff interface{}) *gomock.Call

+ func (mr *MockReschedulerMockRecorder) Len() *gomock.Call

+ func (mr *MockReschedulerMockRecorder) Reschedule(targetRescheduleSize interface{}) *gomock.Call

+ type MockScheduler struct

+ func NewMockScheduler(ctrl *gomock.Controller) *MockScheduler

+ func (m *MockScheduler) EXPECT() *MockSchedulerMockRecorder

+ func (m *MockScheduler) Start()

+ func (m *MockScheduler) Stop()

+ func (m *MockScheduler) Submit(arg0 Executable) error

+ func (m *MockScheduler) TrySubmit(arg0 Executable) (bool, error)

+ type MockSchedulerMockRecorder struct

+ func (mr *MockSchedulerMockRecorder) Start() *gomock.Call

+ func (mr *MockSchedulerMockRecorder) Stop() *gomock.Call

+ func (mr *MockSchedulerMockRecorder) Submit(arg0 interface{}) *gomock.Call

+ func (mr *MockSchedulerMockRecorder) TrySubmit(arg0 interface{}) *gomock.Call

+ type PriorityAssigner interface

+ Assign func(Executable) error

+ func NewNoopPriorityAssigner() PriorityAssigner

+ func NewPriorityAssigner(currentClusterName string, namespaceRegistry namespace.Registry, ...) PriorityAssigner

+ type PriorityAssignerOptions struct

+ CriticalRetryAttempts dynamicconfig.IntPropertyFn

+ HighPriorityRPS dynamicconfig.IntPropertyFnWithNamespaceFilter

+ type QueueType int

+ const QueueTypeActiveTimer

+ const QueueTypeActiveTransfer

+ const QueueTypeStandbyTimer

+ const QueueTypeStandbyTransfer

+ const QueueTypeTimer

+ const QueueTypeTransfer

+ const QueueTypeUnknown

+ const QueueTypeVisibility

+ type Range struct

+ ExclusiveMax tasks.Key

+ InclusiveMin tasks.Key

+ func NewRandomRange() Range

+ func NewRange(inclusiveMin tasks.Key, exclusiveMax tasks.Key) Range

+ func (r *Range) CanMerge(input Range) bool

+ func (r *Range) CanSplit(key tasks.Key) bool

+ func (r *Range) ContainsKey(key tasks.Key) bool

+ func (r *Range) ContainsRange(input Range) bool

+ func (r *Range) Equal(input Range) bool

+ func (r *Range) IsEmpty() bool

+ func (r *Range) Merge(input Range) Range

+ func (r *Range) Split(key tasks.Key) (left Range, right Range)

+ type Rescheduler interface

+ Add func(task Executable, backoff time.Duration)

+ Len func() int

+ Reschedule func(targetRescheduleSize int)

+ type Scheduler interface

+ Submit func(Executable) error

+ TrySubmit func(Executable) (bool, error)

+ type SchedulerOptions struct

+ type Scope struct

+ Predicate tasks.Predicate

+ Range Range

+ func NewScope(r Range, predicate tasks.Predicate) Scope

+ func (s *Scope) CanMergeByPredicate(incomingScope Scope) bool

+ func (s *Scope) CanMergeByRange(incomingScope Scope) bool

+ func (s *Scope) CanSplitByRange(key tasks.Key) bool

+ func (s *Scope) Contains(task tasks.Task) bool

+ func (s *Scope) MergeByPredicate(incomingScope Scope) Scope

+ func (s *Scope) MergeByRange(incomingScope Scope) Scope

+ func (s *Scope) SplitByPredicate(predicate tasks.Predicate) (pass Scope, fail Scope)

+ func (s *Scope) SplitByRange(key tasks.Key) (left Scope, right Scope)

+ type TaskFilter func(task tasks.Task) bool

Jul 11, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

May 5, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Apr 19, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Apr 11, 2022 GO-2024-2689

  GO-2024-2689: Temporal Server Denial of Service in go.temporal.io/server

Changes in this version

+ const ProcessorFactoryFxGroup

+ type MockProcessor struct

+ func NewMockProcessor(ctrl *gomock.Controller) *MockProcessor

+ func (m *MockProcessor) Category() tasks.Category

+ func (m *MockProcessor) EXPECT() *MockProcessorMockRecorder

+ func (m *MockProcessor) FailoverNamespace(namespaceIDs map[string]struct{})

+ func (m *MockProcessor) LockTaskProcessing()

+ func (m *MockProcessor) NotifyNewTasks(clusterName string, tasks []tasks.Task)

+ func (m *MockProcessor) Start()

+ func (m *MockProcessor) Stop()

+ func (m *MockProcessor) UnlockTaskProcessing()

+ type MockProcessorFactory struct

+ func NewMockProcessorFactory(ctrl *gomock.Controller) *MockProcessorFactory

+ func (m *MockProcessorFactory) CreateProcessor(shard shard.Context, engine shard.Engine, cache workflow.Cache) Processor

+ func (m *MockProcessorFactory) EXPECT() *MockProcessorFactoryMockRecorder

+ type MockProcessorFactoryMockRecorder struct

+ func (mr *MockProcessorFactoryMockRecorder) CreateProcessor(shard, engine, cache interface{}) *gomock.Call

+ type MockProcessorMockRecorder struct

+ func (mr *MockProcessorMockRecorder) Category() *gomock.Call

+ func (mr *MockProcessorMockRecorder) FailoverNamespace(namespaceIDs interface{}) *gomock.Call

+ func (mr *MockProcessorMockRecorder) LockTaskProcessing() *gomock.Call

+ func (mr *MockProcessorMockRecorder) NotifyNewTasks(clusterName, tasks interface{}) *gomock.Call

+ func (mr *MockProcessorMockRecorder) Start() *gomock.Call

+ func (mr *MockProcessorMockRecorder) Stop() *gomock.Call

+ func (mr *MockProcessorMockRecorder) UnlockTaskProcessing() *gomock.Call

+ type Processor interface

+ Category func() tasks.Category

+ FailoverNamespace func(namespaceIDs map[string]struct{})

+ LockTaskProcessing func()

+ NotifyNewTasks func(clusterName string, tasks []tasks.Task)

+ UnlockTaskProcessing func()

+ type ProcessorFactory interface

+ CreateProcessor func(shard shard.Context, engine shard.Engine, cache workflow.Cache) Processor