move the check of domain of timer task from ack manager to processor,…

… processor will try to process a standby task indefinitely, until task is finished.

common/persistence/cassandraPersistence.go

@@ -2020,6 +2020,7 @@ func (d *cassandraPersistence) GetTimerIndexTasks(request *GetTimerIndexTasksReq
 		minTimestamp,
 		maxTimestamp,
 		request.BatchSize)
+	query.PageState(request.NextPageToken)
 
 	iter := query.Iter()
 	if iter == nil {

common/persistence/dataInterfaces.go

@@ -638,9 +638,10 @@ type (
 	// GetTimerIndexTasksRequest is the request for GetTimerIndexTasks
 	// TODO: replace this with an iterator that can configure min and max index.
 	GetTimerIndexTasksRequest struct {
-		MinTimestamp time.Time
-		MaxTimestamp time.Time
-		BatchSize    int
+		MinTimestamp  time.Time
+		MaxTimestamp  time.Time
+		BatchSize     int
+		NextPageToken []byte
 	}
 
 	// GetTimerIndexTasksResponse is the response for GetTimerIndexTasks

service/history/MockTimerQueueAckMgr.go

@@ -88,28 +88,24 @@ func (_m *MockTimerQueueAckMgr) readTimerTasks() ([]*persistence.TimerTaskInfo,
 	return r0, r1, r2, r3
 }
 
-func (_m *MockTimerQueueAckMgr) retryTimerTask(timerTask *persistence.TimerTaskInfo) {
+func (_m *MockTimerQueueAckMgr) completeTimerTask(timerTask *persistence.TimerTaskInfo) {
 	_m.Called(timerTask)
 }
 
-func (_m *MockTimerQueueAckMgr) readRetryTimerTasks() []*persistence.TimerTaskInfo {
+func (_m *MockTimerQueueAckMgr) getAckLevel() TimerSequenceID {
 	ret := _m.Called()
 
-	var r0 []*persistence.TimerTaskInfo
-	if rf, ok := ret.Get(0).(func() []*persistence.TimerTaskInfo); ok {
+	var r0 TimerSequenceID
+	if rf, ok := ret.Get(0).(func() TimerSequenceID); ok {
 		r0 = rf()
 	} else {
 		if ret.Get(0) != nil {
-			r0 = ret.Get(0).([]*persistence.TimerTaskInfo)
+			r0 = ret.Get(0).(TimerSequenceID)
 		}
 	}
 	return r0
 }
 
-func (_m *MockTimerQueueAckMgr) completeTimerTask(timerTask *persistence.TimerTaskInfo) {
-	_m.Called(timerTask)
-}
-
 func (_m *MockTimerQueueAckMgr) updateAckLevel() {
 	_m.Called()
 }

service/history/historyEngineInterfaces.go

@@ -44,6 +44,11 @@ type (
 		isWorkflowRunning bool
 		timestamp         time.Time
 	}
+
+	// error which will be thrown if the timer / transfer task should be
+	// retries due to various of reasons
+	taskRetryError struct{}
+
 	// Engine represents an interface for managing workflow execution history.
 	Engine interface {
 		common.Daemon
@@ -122,10 +127,9 @@ type (
 
 	timerQueueAckMgr interface {
 		getFinishedChan() <-chan struct{}
-		readRetryTimerTasks() []*persistence.TimerTaskInfo
 		readTimerTasks() ([]*persistence.TimerTaskInfo, *persistence.TimerTaskInfo, bool, error)
-		retryTimerTask(timerTask *persistence.TimerTaskInfo)
 		completeTimerTask(timerTask *persistence.TimerTaskInfo)
+		getAckLevel() TimerSequenceID
 		updateAckLevel()
 	}
 
@@ -136,3 +140,12 @@ type (
 		UnwatchHistoryEvent(identifier *workflowIdentifier, subscriberID string) error
 	}
 )
+
+// newTaskRetryError create a error which indicate the task should be retry
+func newTaskRetryError() *taskRetryError {
+	return &taskRetryError{}
+}
+
+func (e *taskRetryError) Error() string {
+	return "passive task should retry due to condition in mutable state is not met."
+}

service/history/historyTestBase.go

@@ -190,7 +190,24 @@ func (s *TestShardContext) UpdateReplicatorAckLevel(ackLevel int64) error {
 }
 
 // GetTimerAckLevel test implementation
-func (s *TestShardContext) GetTimerAckLevel(cluster string) time.Time {
+func (s *TestShardContext) GetTimerAckLevel() time.Time {
+	s.RLock()
+	defer s.RUnlock()
+
+	return s.shardInfo.TimerAckLevel
+}
+
+// UpdateTimerAckLevel test implementation
+func (s *TestShardContext) UpdateTimerAckLevel(ackLevel time.Time) error {
+	s.Lock()
+	defer s.Unlock()
+
+	s.shardInfo.TimerAckLevel = ackLevel
+	return nil
+}
+
+// GetTimerClusterAckLevel test implementation
+func (s *TestShardContext) GetTimerClusterAckLevel(cluster string) time.Time {
 	s.RLock()
 	defer s.RUnlock()
 
@@ -203,14 +220,11 @@ func (s *TestShardContext) GetTimerAckLevel(cluster string) time.Time {
 	return s.shardInfo.TimerAckLevel
 }
 
-// UpdateTimerAckLevel test implementation
-func (s *TestShardContext) UpdateTimerAckLevel(cluster string, ackLevel time.Time) error {
+// UpdateTimerClusterAckLevel test implementation
+func (s *TestShardContext) UpdateTimerClusterAckLevel(cluster string, ackLevel time.Time) error {
 	s.Lock()
 	defer s.Unlock()
 
-	if cluster == s.GetService().GetClusterMetadata().GetCurrentClusterName() {
-		s.shardInfo.TimerAckLevel = ackLevel
-	}
 	s.shardInfo.ClusterTimerAckLevel[cluster] = ackLevel
 	return nil
 }

service/history/service.go

@@ -49,12 +49,14 @@ type Config struct {
 	DefaultStartToCloseActivityTimeoutInSecs    int32
 
 	// TimerQueueProcessor settings
-	TimerTaskBatchSize                    int
-	ProcessTimerTaskWorkerCount           int
-	TimerProcessorUpdateFailureRetryCount int
-	TimerProcessorGetFailureRetryCount    int
-	TimerProcessorUpdateAckInterval       time.Duration
-	TimerProcessorForceUpdateInterval     time.Duration
+	TimerTaskBatchSize                           int
+	ProcessTimerTaskWorkerCount                  int
+	TimerProcessorUpdateFailureRetryCount        int
+	TimerProcessorGetFailureRetryCount           int
+	TimerProcessorCompleteTimerFailureRetryCount int
+	TimerProcessorUpdateAckInterval              time.Duration
+	TimerProcessorForceUpdateInterval            time.Duration
+	TimerProcessorCompleteTimerInterval          time.Duration
 
 	// TransferQueueProcessor settings
 	TransferTaskBatchSize                int
@@ -86,37 +88,39 @@ type Config struct {
 // NewConfig returns new service config with default values
 func NewConfig(dc *dynamicconfig.Collection, numberOfShards int) *Config {
 	return &Config{
-		NumberOfShards:                              numberOfShards,
-		HistoryCacheInitialSize:                     128,
-		HistoryCacheMaxSize:                         512,
-		HistoryCacheTTL:                             time.Hour,
-		RangeSizeBits:                               20, // 20 bits for sequencer, 2^20 sequence number for any range
-		AcquireShardInterval:                        time.Minute,
-		DefaultScheduleToStartActivityTimeoutInSecs: 10,
-		DefaultScheduleToCloseActivityTimeoutInSecs: 10,
-		DefaultStartToCloseActivityTimeoutInSecs:    10,
-		TimerTaskBatchSize:                          100,
-		ProcessTimerTaskWorkerCount:                 30,
-		TimerProcessorUpdateFailureRetryCount:       5,
-		TimerProcessorGetFailureRetryCount:          5,
-		TimerProcessorUpdateAckInterval:             10 * time.Second,
-		TimerProcessorForceUpdateInterval:           10 * time.Minute,
-		TransferTaskBatchSize:                       10,
-		TransferProcessorMaxPollRPS:                 100,
-		TransferProcessorMaxPollInterval:            60 * time.Second,
-		TransferProcessorUpdateAckInterval:          10 * time.Second,
-		TransferProcessorForceUpdateInterval:        10 * time.Minute,
-		TransferTaskWorkerCount:                     10,
-		TransferTaskMaxRetryCount:                   100,
-		ReplicatorTaskBatchSize:                     10,
-		ReplicatorProcessorMaxPollRPS:               100,
-		ReplicatorProcessorMaxPollInterval:          60 * time.Second,
-		ReplicatorProcessorUpdateAckInterval:        10 * time.Second,
-		ReplicatorProcessorForceUpdateInterval:      10 * time.Minute,
-		ReplicatorTaskWorkerCount:                   10,
-		ReplicatorTaskMaxRetryCount:                 100,
-		ExecutionMgrNumConns:                        100,
-		HistoryMgrNumConns:                          100,
+		NumberOfShards:                               numberOfShards,
+		HistoryCacheInitialSize:                      128,
+		HistoryCacheMaxSize:                          512,
+		HistoryCacheTTL:                              time.Hour,
+		RangeSizeBits:                                20, // 20 bits for sequencer, 2^20 sequence number for any range
+		AcquireShardInterval:                         time.Minute,
+		DefaultScheduleToStartActivityTimeoutInSecs:  10,
+		DefaultScheduleToCloseActivityTimeoutInSecs:  10,
+		DefaultStartToCloseActivityTimeoutInSecs:     10,
+		TimerTaskBatchSize:                           100,
+		ProcessTimerTaskWorkerCount:                  30,
+		TimerProcessorUpdateFailureRetryCount:        5,
+		TimerProcessorGetFailureRetryCount:           5,
+		TimerProcessorCompleteTimerFailureRetryCount: 10,
+		TimerProcessorUpdateAckInterval:              10 * time.Second,
+		TimerProcessorForceUpdateInterval:            10 * time.Minute,
+		TimerProcessorCompleteTimerInterval:          5 * time.Second,
+		TransferTaskBatchSize:                        10,
+		TransferProcessorMaxPollRPS:                  100,
+		TransferProcessorMaxPollInterval:             60 * time.Second,
+		TransferProcessorUpdateAckInterval:           10 * time.Second,
+		TransferProcessorForceUpdateInterval:         10 * time.Minute,
+		TransferTaskWorkerCount:                      10,
+		TransferTaskMaxRetryCount:                    100,
+		ReplicatorTaskBatchSize:                      10,
+		ReplicatorProcessorMaxPollRPS:                100,
+		ReplicatorProcessorMaxPollInterval:           60 * time.Second,
+		ReplicatorProcessorUpdateAckInterval:         10 * time.Second,
+		ReplicatorProcessorForceUpdateInterval:       10 * time.Minute,
+		ReplicatorTaskWorkerCount:                    10,
+		ReplicatorTaskMaxRetryCount:                  100,
+		ExecutionMgrNumConns:                         100,
+		HistoryMgrNumConns:                           100,
 		// history client: client/history/client.go set the client timeout 30s
 		LongPollExpirationInterval: dc.GetDurationProperty(
 			dynamicconfig.HistoryLongPollExpirationInterval, time.Second*20,

service/history/shardContext.go

@@ -51,6 +51,10 @@ type (
 		UpdateTransferAckLevel(ackLevel int64) error
 		GetReplicatorAckLevel() int64
 		UpdateReplicatorAckLevel(ackLevel int64) error
+		GetTimerAckLevel() time.Time
+		UpdateTimerAckLevel(ackLevel time.Time) error
+		GetTimerClusterAckLevel(cluster string) time.Time
+		UpdateTimerClusterAckLevel(cluster string, ackLevel time.Time) error
 		CreateWorkflowExecution(request *persistence.CreateWorkflowExecutionRequest) (
 			*persistence.CreateWorkflowExecutionResponse, error)
 		UpdateWorkflowExecution(request *persistence.UpdateWorkflowExecutionRequest) error
@@ -59,8 +63,6 @@ type (
 		GetConfig() *Config
 		GetLogger() bark.Logger
 		GetMetricsClient() metrics.Client
-		GetTimerAckLevel(cluster string) time.Time
-		UpdateTimerAckLevel(cluster string, ackLevel time.Time) error
 		GetTimeSource() common.TimeSource
 		SetCurrentTime(cluster string, currentTime time.Time)
 		GetCurrentTime(cluster string) time.Time
@@ -164,7 +166,23 @@ func (s *shardContextImpl) UpdateReplicatorAckLevel(ackLevel int64) error {
 	return s.updateShardInfoLocked()
 }
 
-func (s *shardContextImpl) GetTimerAckLevel(cluster string) time.Time {
+func (s *shardContextImpl) GetTimerAckLevel() time.Time {
+	s.RLock()
+	defer s.RUnlock()
+
+	return s.shardInfo.TimerAckLevel
+}
+
+func (s *shardContextImpl) UpdateTimerAckLevel(ackLevel time.Time) error {
+	s.RLock()
+	defer s.RUnlock()
+
+	s.shardInfo.TimerAckLevel = ackLevel
+	s.shardInfo.StolenSinceRenew = 0
+	return s.updateShardInfoLocked()
+}
+
+func (s *shardContextImpl) GetTimerClusterAckLevel(cluster string) time.Time {
 	s.RLock()
 	defer s.RUnlock()
 
@@ -177,13 +195,10 @@ func (s *shardContextImpl) GetTimerAckLevel(cluster string) time.Time {
 	return s.shardInfo.TimerAckLevel
 }
 
-func (s *shardContextImpl) UpdateTimerAckLevel(cluster string, ackLevel time.Time) error {
+func (s *shardContextImpl) UpdateTimerClusterAckLevel(cluster string, ackLevel time.Time) error {
 	s.Lock()
 	defer s.Unlock()
 
-	if cluster == s.GetService().GetClusterMetadata().GetCurrentClusterName() {
-		s.shardInfo.TimerAckLevel = ackLevel
-	}
 	s.shardInfo.ClusterTimerAckLevel[cluster] = ackLevel
 	s.shardInfo.StolenSinceRenew = 0
 	return s.updateShardInfoLocked()