after review

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

Author: tubignat

config/development.yaml

@@ -172,4 +172,4 @@ shardDistribution:
       prefix: "store"
   process:
     period: 1s
-    heartbeatTTL: 2s
+    heartbeatTTL: 2s

service/history/common/type.go

@@ -38,7 +38,9 @@ type (
 		PersistenceError bool
 
 		// if true, the task will be scheduled in memory for the current execution, otherwise
-		// it will only be scheduled after the next DB scan
+		// it will only be scheduled after the next DB scan. This notification is sometimes passed with a fake
+		// timer with the sole purpose of resetting the next scheduled DB read, that's why sometimes we want to
+		// avoid scheduling the task in memory.
 		ScheduleInMemory bool
 	}
 )

service/history/queuev2/queue_base.go

@@ -278,8 +278,8 @@ func (q *queueBase) insertSingleTask(task task.Task) bool {
 	return q.virtualQueueManager.InsertSingleTaskToRootQueue(task)
 }
 
-func (q *queueBase) removeScheduledTasksAfter(key persistence.HistoryTaskKey) {
-	q.virtualQueueManager.RemoveScheduledTasksAfter(key)
+func (q *queueBase) resetProgress(key persistence.HistoryTaskKey) {
+	q.virtualQueueManager.ResetProgress(key)
 }
 
 func (q *queueBase) updateQueueState(ctx context.Context) {

service/history/queuev2/queue_scheduled.go

@@ -175,7 +175,7 @@ func (q *scheduledQueue) NotifyNewTask(clusterName string, info *hcommon.NotifyT
 	}
 
 	if !nextReadTime.IsZero() {
-		q.base.removeScheduledTasksAfter(persistence.NewHistoryTaskKey(nextReadTime, 0))
+		q.base.resetProgress(persistence.NewHistoryTaskKey(nextReadTime, 0))
 		q.notify(nextReadTime)
 	}
 

service/history/queuev2/virtual_queue.go

@@ -69,8 +69,8 @@ type (
 		Pause(time.Duration)
 		// InsertSingleTask inserts a single task to the virtual queue. Return false if the task's timestamp is out of range of the current queue slice..
 		InsertSingleTask(task task.Task) bool
-		// RemoveScheduledTasksAfter removes the scheduled tasks after the given time
-		RemoveScheduledTasksAfter(persistence.HistoryTaskKey)
+		// ResetProgress removes the scheduled tasks after the given time
+		ResetProgress(persistence.HistoryTaskKey)
 	}
 
 	VirtualQueueOptions struct {
@@ -445,21 +445,24 @@ func (q *virtualQueueImpl) InsertSingleTask(task task.Task) bool {
 
 	now := q.timeSource.Now()
 	if err := q.submitTask(now, task); err != nil {
-		q.logger.Error("Virtual queue failed to submit task", tag.Error(err))
-		return false
+		q.logger.Error("Error submitting task to virtual queue", tag.Error(err))
 	}
 
 	return true
 }
 
-func (q *virtualQueueImpl) RemoveScheduledTasksAfter(key persistence.HistoryTaskKey) {
+func (q *virtualQueueImpl) ResetProgress(key persistence.HistoryTaskKey) {
 	q.Lock()
 	defer q.Unlock()
 
 	for e := q.virtualSlices.Front(); e != nil; e = e.Next() {
 		slice := e.Value.(VirtualSlice)
-		slice.CancelTasksAfter(key)
+		slice.ResetProgress(key)
 		q.monitor.SetSlicePendingTaskCount(slice, slice.GetPendingTaskCount())
+
+		if e == q.sliceToRead {
+			break
+		}
 	}
 }
 
@@ -482,16 +485,13 @@ func (q *virtualQueueImpl) submitTask(now time.Time, task task.Task) error {
 	q.metricsScope.RecordHistogramDuration(metrics.TaskEnqueueToFetchLatency, now.Sub(task.GetVisibilityTimestamp()))
 	task.SetInitialSubmitTime(now)
 	submitted, err := q.processor.TrySubmit(task)
-	if err != nil {
-		return err
-	}
 
 	if !submitted {
 		q.metricsScope.IncCounter(metrics.ProcessingQueueThrottledCounter)
 		q.rescheduler.RescheduleTask(task, q.timeSource.Now().Add(taskSchedulerThrottleBackoffInterval))
 	}
 
-	return nil
+	return err
 }
 
 func (q *virtualQueueImpl) resetNextReadSliceLocked() {

service/history/queuev2/virtual_queue_manager.go

@@ -59,7 +59,7 @@ type (
 		AddNewVirtualSliceToRootQueue(VirtualSlice)
 		// Insert a single task to the current slice. Return false if the task's timestamp is out of range of the current slice.
 		InsertSingleTaskToRootQueue(task.Task) bool
-		RemoveScheduledTasksAfter(persistence.HistoryTaskKey)
+		ResetProgress(persistence.HistoryTaskKey)
 	}
 
 	virtualQueueManagerImpl struct {
@@ -233,11 +233,11 @@ func (m *virtualQueueManagerImpl) InsertSingleTaskToRootQueue(t task.Task) bool
 	return false
 }
 
-func (m *virtualQueueManagerImpl) RemoveScheduledTasksAfter(key persistence.HistoryTaskKey) {
+func (m *virtualQueueManagerImpl) ResetProgress(key persistence.HistoryTaskKey) {
 	m.Lock()
 	defer m.Unlock()
 	for _, vq := range m.virtualQueues {
-		vq.RemoveScheduledTasksAfter(key)
+		vq.ResetProgress(key)
 	}
 }
 

service/history/queuev2/virtual_queue_manager_mock.go

@@ -43,7 +43,7 @@ type (
 		GetPendingTaskCount() int
 		Clear()
 		PendingTaskStats() PendingTaskStats
-		CancelTasksAfter(key persistence.HistoryTaskKey)
+		ResetProgress(key persistence.HistoryTaskKey)
 
 		TrySplitByTaskKey(persistence.HistoryTaskKey) (VirtualSlice, VirtualSlice, bool)
 		TrySplitByPredicate(Predicate) (VirtualSlice, VirtualSlice, bool)
@@ -197,7 +197,10 @@ func (s *virtualSliceImpl) UpdateAndGetState() VirtualSliceState {
 	return s.state
 }
 
-func (s *virtualSliceImpl) CancelTasksAfter(key persistence.HistoryTaskKey) {
+// this function is used when we are not sure if our in-memory state after the given key is correct,
+// we want to cancel all the tasks after the given key and reset the progress to the given key,
+// so that in the next poll, we will read the tasks from the DB starting from the given key.
+func (s *virtualSliceImpl) ResetProgress(key persistence.HistoryTaskKey) {
 	taskMap := s.pendingTaskTracker.GetTasks()
 	for _, task := range taskMap {
 		if task.GetTaskKey().Compare(key) >= 0 {