After review #2

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

Author: tubignat

service/history/queuev2/queue_base.go

@@ -275,7 +275,7 @@ func (q *queueBase) processNewTasks() bool {
 }
 
 func (q *queueBase) insertSingleTask(task task.Task) bool {
-	return q.virtualQueueManager.InsertSingleTaskToRootQueue(task)
+	return q.virtualQueueManager.InsertSingleTask(task)
 }
 
 func (q *queueBase) resetProgress(key persistence.HistoryTaskKey) {

service/history/queuev2/virtual_queue.go

@@ -67,7 +67,7 @@ type (
 		SplitSlices(func(VirtualSlice) (remaining []VirtualSlice, split bool))
 		// Pause pauses the virtual queue for a while
 		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 inserts a single task to the virtual queue. Return false if the task's timestamp is out of range of the current queue slice or the task does not satisfy the predicate
 		InsertSingleTask(task task.Task) bool
 		// ResetProgress removes the scheduled tasks after the given time
 		ResetProgress(persistence.HistoryTaskKey)
@@ -398,7 +398,7 @@ func (q *virtualQueueImpl) loadAndSubmitTasks() {
 
 	now := q.timeSource.Now()
 	for _, task := range tasks {
-		if err := q.submitTask(now, task); err != nil {
+		if err := q.submitOrRescheduleTask(now, task); err != nil {
 			select {
 			case <-q.ctx.Done():
 				return
@@ -423,32 +423,21 @@ func (q *virtualQueueImpl) InsertSingleTask(task task.Task) bool {
 	q.Lock()
 	defer q.Unlock()
 
-	taskKey := task.GetTaskKey()
-	var slice VirtualSlice
-
 	for e := q.virtualSlices.Front(); e != nil; e = e.Next() {
-		s := e.Value.(VirtualSlice)
-		r := s.GetState().Range
-		if taskKey.Compare(r.InclusiveMinTaskKey) >= 0 && taskKey.Compare(r.ExclusiveMaxTaskKey) < 0 {
-			slice = s
-			break
-		}
-	}
-
-	if slice == nil {
-		// the new task is outside of the current range, it will be read from the DB on the next poll
-		return false
-	}
-
-	slice.InsertTask(task)
-	q.monitor.SetSlicePendingTaskCount(slice, slice.GetPendingTaskCount())
+		slice := e.Value.(VirtualSlice)
+		inserted := slice.InsertTask(task)
+		if inserted {
+			q.monitor.SetSlicePendingTaskCount(slice, slice.GetPendingTaskCount())
+			now := q.timeSource.Now()
+			if err := q.submitOrRescheduleTask(now, task); err != nil {
+				q.logger.Error("Error submitting task to virtual queue", tag.Error(err))
+			}
 
-	now := q.timeSource.Now()
-	if err := q.submitTask(now, task); err != nil {
-		q.logger.Error("Error submitting task to virtual queue", tag.Error(err))
+			return true
+		}
 	}
 
-	return true
+	return false
 }
 
 func (q *virtualQueueImpl) ResetProgress(key persistence.HistoryTaskKey) {
@@ -464,9 +453,11 @@ func (q *virtualQueueImpl) ResetProgress(key persistence.HistoryTaskKey) {
 			break
 		}
 	}
+
+	q.resetNextReadSliceLocked()
 }
 
-func (q *virtualQueueImpl) submitTask(now time.Time, task task.Task) error {
+func (q *virtualQueueImpl) submitOrRescheduleTask(now time.Time, task task.Task) error {
 	if persistence.IsTaskCorrupted(task) {
 		q.logger.Error("Virtual queue encountered a corrupted task", tag.Dynamic("task", task))
 		q.metricsScope.IncCounter(metrics.CorruptedHistoryTaskCounter)

service/history/queuev2/virtual_queue_manager.go

@@ -58,7 +58,8 @@ type (
 		// By default, all new tasks belong to the root queue, so we need to add a new virtual slice to the root queue.
 		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
+		InsertSingleTask(task.Task) bool
+		// ResetProgress resets the progress of the virtual queue to the given key. Pending tasks after the given key are cancelled.
 		ResetProgress(persistence.HistoryTaskKey)
 	}
 
@@ -222,15 +223,18 @@ func (m *virtualQueueManagerImpl) AddNewVirtualSliceToRootQueue(s VirtualSlice)
 	m.virtualQueues[rootQueueID].Start()
 }
 
-func (m *virtualQueueManagerImpl) InsertSingleTaskToRootQueue(t task.Task) bool {
+func (m *virtualQueueManagerImpl) InsertSingleTask(t task.Task) bool {
 	m.Lock()
 	defer m.Unlock()
-	if vq, ok := m.virtualQueues[rootQueueID]; ok {
-		return vq.InsertSingleTask(t)
+
+	inserted := false
+	for _, vq := range m.virtualQueues {
+		if vq.InsertSingleTask(t) {
+			inserted = true
+		}
 	}
 
-	// if a root queue is not created yet, no need to schedule an incoming task, it will be read from the slice
-	return false
+	return inserted
 }
 
 func (m *virtualQueueManagerImpl) ResetProgress(key persistence.HistoryTaskKey) {

service/history/queuev2/virtual_queue_manager_mock.go

@@ -37,7 +37,7 @@ type (
 		GetState() VirtualSliceState
 		IsEmpty() bool
 		GetTasks(context.Context, int) ([]task.Task, error)
-		InsertTask(task.Task)
+		InsertTask(task.Task) bool
 		HasMoreTasks() bool
 		UpdateAndGetState() VirtualSliceState
 		GetPendingTaskCount() int
@@ -117,8 +117,20 @@ func (s *virtualSliceImpl) Clear() {
 	}
 }
 
-func (s *virtualSliceImpl) InsertTask(task task.Task) {
+func (s *virtualSliceImpl) InsertTask(task task.Task) bool {
+	r := s.state.Range
+	key := task.GetTaskKey()
+
+	if key.Compare(r.InclusiveMinTaskKey) < 0 || key.Compare(r.ExclusiveMaxTaskKey) >= 0 {
+		return false
+	}
+
+	if !s.state.Predicate.Check(task) {
+		return false
+	}
+
 	s.pendingTaskTracker.AddTask(task)
+	return true
 }
 
 func (s *virtualSliceImpl) GetTasks(ctx context.Context, pageSize int) ([]task.Task, error) {
@@ -223,15 +235,21 @@ func (s *virtualSliceImpl) ResetProgress(key persistence.HistoryTaskKey) {
 	}
 
 	for i, progress := range s.progress {
+		// progress contains sorted non-overlapping ranges. If we found a range that contains the given key, we can reset
+		// this range's progress to the given key and merge the remaining ranges into it.
 		if progress.NextTaskKey.Compare(key) > 0 {
+			maxTaskKey := s.progress[len(s.progress)-1].Range.ExclusiveMaxTaskKey
 			s.progress[i] = &GetTaskProgress{
 				Range: Range{
 					InclusiveMinTaskKey: key,
-					ExclusiveMaxTaskKey: progress.Range.ExclusiveMaxTaskKey,
+					ExclusiveMaxTaskKey: maxTaskKey,
 				},
 				NextPageToken: nil,
 				NextTaskKey:   key,
 			}
+
+			s.progress = s.progress[:i+1]
+			break
 		}
 	}
 }

service/history/queuev2/virtual_slice.go

@@ -1363,6 +1363,15 @@ func (s *contextImpl) allocateTimerIDsLocked(
 				tag.ValueShardAllocateTimerBeforeRead)
 			ts = now.Add(persistence.DBTimestampMinPrecision)
 		}
+
+		if ts.Before(s.shardInfo.TimerAckLevel) {
+			s.logger.Warn("New timer generated is less than ack level",
+				tag.WorkflowDomainID(domainEntry.GetInfo().ID),
+				tag.WorkflowID(workflowID),
+				tag.Timestamp(ts))
+			ts = s.shardInfo.TimerAckLevel.Add(persistence.DBTimestampMinPrecision)
+		}
+
 		task.SetVisibilityTimestamp(ts)
 
 		seqNum, err := s.generateTaskIDLocked()