Clean up

Signed-off-by: Ignat Tubylov <[email protected]>

Author: tubignat

common/dynamicconfig/dynamicproperties/constants.go

@@ -2113,7 +2113,6 @@ const (
 	EnableTimerQueueV2
 	EnableTransferQueueV2PendingTaskCountAlert
 	EnableTimerQueueV2PendingTaskCountAlert
-	EnableTimerProcessorInMemoryQueue
 
 	// LastBoolKey must be the last one in this const group
 	LastBoolKey
@@ -2673,6 +2672,12 @@ const (
 	// Default value: 1s (1*time.Second)
 	// Allowed filters: N/A
 	TimerProcessorMaxTimeShift
+	// TimerProcessorInMemoryQueueMaxTimeShift is the max shift timer processor in memory queue can have. If set to 0, in memory queue is disabled.
+	// KeyName: history.timerProcessorInMemoryQueueMaxTimeShift
+	// Value type: Duration
+	// Default value: 0
+	// Allowed filters: ShardID
+	TimerProcessorInMemoryQueueMaxTimeShift
 	// TransferProcessorFailoverMaxStartJitterInterval is the max jitter interval for starting transfer
 	// failover queue processing. The actual jitter interval used will be a random duration between
 	// 0 and the max interval so that timer failover queue across different shards won't start at
@@ -4681,12 +4686,6 @@ var BoolKeys = map[BoolKey]DynamicBool{
 		Filters:      []Filter{ShardID},
 		DefaultValue: false,
 	},
-	EnableTimerProcessorInMemoryQueue: {
-		KeyName:      "history.enableTimerProcessorInMemoryQueue",
-		Description:  "EnableTimerProcessorInMemoryQueue is the flag to enable in-memory queue for timer processor",
-		Filters:      []Filter{ShardID},
-		DefaultValue: false,
-	},
 }
 
 var FloatKeys = map[FloatKey]DynamicFloat{
@@ -5205,6 +5204,12 @@ var DurationKeys = map[DurationKey]DynamicDuration{
 		Description:  "TimerProcessorMaxTimeShift is the max shift timer processor can have",
 		DefaultValue: time.Second,
 	},
+	TimerProcessorInMemoryQueueMaxTimeShift: {
+		KeyName:      "history.timerProcessorInMemoryQueueMaxTimeShift",
+		Filters:      []Filter{ShardID},
+		Description:  "TimerProcessorInMemoryQueueMaxTimeShift is the max shift timer processor in memory queue can have. If set to 0, in memory queue is disabled.",
+		DefaultValue: 0,
+	},
 	TransferProcessorFailoverMaxStartJitterInterval: {
 		KeyName:      "history.transferProcessorFailoverMaxStartJitterInterval",
 		Description:  "TransferProcessorFailoverMaxStartJitterInterval is the max jitter interval for starting transfer failover queue processing. The actual jitter interval used will be a random duration between 0 and the max interval so that timer failover queue across different shards won't start at the same time",

service/history/config/config.go

@@ -154,7 +154,7 @@ type Config struct {
 	TimerProcessorSplitQueueIntervalJitterCoefficient dynamicproperties.FloatPropertyFn
 	TimerProcessorMaxRedispatchQueueSize              dynamicproperties.IntPropertyFn
 	TimerProcessorMaxTimeShift                        dynamicproperties.DurationPropertyFn
-	EnableTimerProcessorInMemoryQueue                 dynamicproperties.BoolPropertyFnWithShardIDFilter
+	TimerProcessorInMemoryQueueMaxTimeShift           dynamicproperties.DurationPropertyFnWithShardIDFilter
 	TimerProcessorHistoryArchivalSizeLimit            dynamicproperties.IntPropertyFn
 	TimerProcessorArchivalTimeLimit                   dynamicproperties.DurationPropertyFn
 	DisableTimerFailoverQueue                         dynamicproperties.BoolPropertyFn
@@ -454,7 +454,7 @@ func New(dc *dynamicconfig.Collection, numberOfShards int, maxMessageSize int, i
 		TimerProcessorSplitQueueIntervalJitterCoefficient:    dc.GetFloat64Property(dynamicproperties.TimerProcessorSplitQueueIntervalJitterCoefficient),
 		TimerProcessorMaxRedispatchQueueSize:                 dc.GetIntProperty(dynamicproperties.TimerProcessorMaxRedispatchQueueSize),
 		TimerProcessorMaxTimeShift:                           dc.GetDurationProperty(dynamicproperties.TimerProcessorMaxTimeShift),
-		EnableTimerProcessorInMemoryQueue:                    dc.GetBoolPropertyFilteredByShardID(dynamicproperties.EnableTimerProcessorInMemoryQueue),
+		TimerProcessorInMemoryQueueMaxTimeShift:              dc.GetDurationPropertyFilteredByShardID(dynamicproperties.TimerProcessorInMemoryQueueMaxTimeShift),
 		TimerProcessorHistoryArchivalSizeLimit:               dc.GetIntProperty(dynamicproperties.TimerProcessorHistoryArchivalSizeLimit),
 		TimerProcessorArchivalTimeLimit:                      dc.GetDurationProperty(dynamicproperties.TimerProcessorArchivalTimeLimit),
 		DisableTimerFailoverQueue:                            dc.GetBoolProperty(dynamicproperties.DisableTimerFailoverQueue),

service/history/config/config_test.go

@@ -139,6 +139,7 @@ func TestNewConfig(t *testing.T) {
 		"TimerProcessorSplitQueueIntervalJitterCoefficient":    {dynamicproperties.TimerProcessorSplitQueueIntervalJitterCoefficient, 4.0},
 		"TimerProcessorMaxRedispatchQueueSize":                 {dynamicproperties.TimerProcessorMaxRedispatchQueueSize, 45},
 		"TimerProcessorMaxTimeShift":                           {dynamicproperties.TimerProcessorMaxTimeShift, time.Second},
+		"TimerProcessorInMemoryQueueMaxTimeShift":              {dynamicproperties.TimerProcessorInMemoryQueueMaxTimeShift, 0},
 		"TimerProcessorHistoryArchivalSizeLimit":               {dynamicproperties.TimerProcessorHistoryArchivalSizeLimit, 46},
 		"TimerProcessorArchivalTimeLimit":                      {dynamicproperties.TimerProcessorArchivalTimeLimit, time.Second},
 		"TransferTaskBatchSize":                                {dynamicproperties.TransferTaskBatchSize, 47},
@@ -278,7 +279,6 @@ func TestNewConfig(t *testing.T) {
 		"QueueMaxVirtualQueueCount":                            {dynamicproperties.QueueMaxVirtualQueueCount, 101},
 		"VirtualSliceForceAppendInterval":                      {dynamicproperties.VirtualSliceForceAppendInterval, time.Second},
 		"ReplicationTaskProcessorLatencyLogThreshold":          {dynamicproperties.ReplicationTaskProcessorLatencyLogThreshold, time.Duration(0)},
-		"EnableTimerProcessorInMemoryQueue":                    {dynamicproperties.EnableTimerProcessorInMemoryQueue, false},
 	}
 	client := dynamicconfig.NewInMemoryClient()
 	for fieldName, expected := range fields {

service/history/queuev2/virtual_queue.go

@@ -464,7 +464,11 @@ func (q *virtualQueueImpl) RemoveScheduledTasksAfter(t time.Time) {
 			continue
 		}
 
-		// TODO: remove scheduled tasks from virtual slices
+		s.CancelTasks(func(task task.Task) bool {
+			taskTime := task.GetTaskKey().GetScheduledTime()
+			return taskTime.After(t) || taskTime.Equal(t)
+		})
+
 		q.monitor.SetSlicePendingTaskCount(s, s.GetPendingTaskCount())
 	}
 }

service/history/queuev2/virtual_slice.go

@@ -43,6 +43,7 @@ type (
 		GetPendingTaskCount() int
 		Clear()
 		PendingTaskStats() PendingTaskStats
+		CancelTasks(predicate func(task.Task) bool)
 
 		TrySplitByTaskKey(persistence.HistoryTaskKey) (VirtualSlice, VirtualSlice, bool)
 		TrySplitByPredicate(Predicate) (VirtualSlice, VirtualSlice, bool)
@@ -196,6 +197,15 @@ func (s *virtualSliceImpl) UpdateAndGetState() VirtualSliceState {
 	return s.state
 }
 
+func (s *virtualSliceImpl) CancelTasks(predicate func(task.Task) bool) {
+	taskMap := s.pendingTaskTracker.GetTasks()
+	for _, task := range taskMap {
+		if predicate(task) {
+			task.Cancel()
+		}
+	}
+}
+
 func (s *virtualSliceImpl) TrySplitByTaskKey(taskKey persistence.HistoryTaskKey) (VirtualSlice, VirtualSlice, bool) {
 	leftState, rightState, ok := s.state.TrySplitByTaskKey(taskKey)
 	if !ok {

service/history/queuev2/virtual_slice_mock.go

@@ -246,7 +246,12 @@ func (s *contextImpl) updateScheduledTaskMaxReadLevel(cluster string) persistenc
 		currentTime = s.remoteClusterCurrentTime[cluster]
 	}
 
-	newMaxReadLevel := currentTime.Add(s.config.TimerProcessorMaxTimeShift()).Truncate(persistence.DBTimestampMinPrecision)
+	maxTimeShift := s.config.TimerProcessorMaxTimeShift()
+	if s.config.TimerProcessorInMemoryQueueMaxTimeShift(s.shardID) > 0 {
+		maxTimeShift = s.config.TimerProcessorInMemoryQueueMaxTimeShift(s.shardID)
+	}
+
+	newMaxReadLevel := currentTime.Add(maxTimeShift).Truncate(persistence.DBTimestampMinPrecision)
 	if newMaxReadLevel.After(s.scheduledTaskMaxReadLevelMap[cluster]) {
 		s.scheduledTaskMaxReadLevelMap[cluster] = newMaxReadLevel
 	}
@@ -1334,8 +1339,8 @@ func (s *contextImpl) allocateTimerIDsLocked(
 		// Above cases can happen if shard move and new host have a time SKU,
 		// or there is db write delay, or we are simply (re-)generating tasks for an old workflow.
 
-		// we need to skip this check for in-memory queue because it is going to be populated before the db is read
-		if !s.config.EnableTimerProcessorInMemoryQueue(s.shardID) {
+		if s.config.TimerProcessorInMemoryQueueMaxTimeShift(s.shardID) == 0 {
+			// this check is only required when an in memory queue is disabled. If it's enabled, we expect timers below read level to be enqueued in memory
 			if ts.Before(readCursorTS) {
 				// This can happen if shard move and new host have a time SKU, or there is db write delay.
 				// We generate a new timer ID using timerMaxReadLevel.