Do not generate scheduled task with timestamp in the past

Author: Shaddoll

service/history/shard/context.go

@@ -1295,6 +1295,7 @@ func (s *contextImpl) allocateTimerIDsLocked(
 	workflowID string,
 	timerTasks []persistence.Task,
 ) error {
+	now := s.GetTimeSource().Now().Truncate(persistence.DBTimestampMinPrecision)
 	// assign IDs for the timer tasks. They need to be assigned under shard lock.
 	cluster := s.GetClusterMetadata().GetCurrentClusterName()
 	for _, task := range timerTasks {
@@ -1324,6 +1325,11 @@ func (s *contextImpl) allocateTimerIDsLocked(
 		}
 
 		readCursorTS := s.scheduledTaskMaxReadLevelMap[cluster]
+		// make sure scheduled task timestamp is higher than
+		// 1. max read level, so that queue processor can read the task back.
+		// 2. current time. Otherwise the task timestamp is in the past and causes aritical load latency in queue processor metrics.
+		// 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.
 		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.
@@ -1336,6 +1342,14 @@ func (s *contextImpl) allocateTimerIDsLocked(
 				tag.ValueShardAllocateTimerBeforeRead)
 			ts = readCursorTS.Add(persistence.DBTimestampMinPrecision)
 		}
+		if ts.Before(now) {
+			s.logger.Warn("New timer generated is in the past",
+				tag.WorkflowDomainID(domainEntry.GetInfo().ID),
+				tag.WorkflowID(workflowID),
+				tag.Timestamp(ts),
+				tag.ValueShardAllocateTimerBeforeRead)
+			ts = now
+		}
 		task.SetVisibilityTimestamp(ts)
 
 		seqNum, err := s.generateTaskIDLocked()

service/history/shard/context_test.go

@@ -1575,3 +1575,127 @@ func (s *contextTestSuite) TestAllocateTimerIDsLocked_WhenMultipleTasksProvidedA
 	s.True(task1.GetTaskID() > 0, "Task 1 ID should be positive")
 	s.True(task2.GetTaskID() > 0, "Task 2 ID should be positive")
 }
+
+func (s *contextTestSuite) TestAllocateTimerIDsLocked() {
+	s.mockResource.TimeSource = clock.NewMockedTimeSourceAt(time.Now())
+	testTimeNow := s.mockResource.TimeSource.Now()
+	testVisibilityTime := testTimeNow.Add(-time.Hour).Truncate(persistence.DBTimestampMinPrecision)
+	testReadCursorTime := testTimeNow.Add(time.Minute).Truncate(persistence.DBTimestampMinPrecision)
+	testFutureTime := testTimeNow.Add(30 * time.Minute).Truncate(persistence.DBTimestampMinPrecision)
+
+	domainInfo := &persistence.DomainInfo{
+		ID:   testDomainID,
+		Name: testDomain,
+	}
+
+	tests := []struct {
+		name             string
+		setupContext     func(*contextImpl)
+		setupMocks       func()
+		domainEntry      *cache.DomainCacheEntry
+		workflowID       string
+		timerTasks       []persistence.Task
+		expectError      bool
+		expectedErrorMsg string
+		validateResults  func(*testing.T, *contextImpl, []persistence.Task)
+	}{
+		{
+			name: "Queue V2 enabled - simple case",
+			setupContext: func(ctx *contextImpl) {
+				ctx.config.EnableTimerQueueV2 = func(shardID int) bool { return true }
+				ctx.scheduledTaskMaxReadLevelMap[cluster.TestCurrentClusterName] = testReadCursorTime
+			},
+			domainEntry: cache.NewLocalDomainCacheEntryForTest(domainInfo, &persistence.DomainConfig{}, cluster.TestCurrentClusterName),
+			workflowID:  testWorkflowID,
+			timerTasks: []persistence.Task{
+				&persistence.UserTimerTask{
+					TaskData: persistence.TaskData{
+						VisibilityTimestamp: testFutureTime,
+					},
+				},
+			},
+			expectError: false,
+			validateResults: func(t *testing.T, ctx *contextImpl, tasks []persistence.Task) {
+				require.Len(t, tasks, 1)
+				// Task should keep its future timestamp
+				assert.Equal(t, testFutureTime, tasks[0].GetVisibilityTimestamp())
+			},
+		},
+		{
+			name: "Task timestamp before read cursor - should adjust timestamp",
+			setupContext: func(ctx *contextImpl) {
+				ctx.config.EnableTimerQueueV2 = func(shardID int) bool { return true }
+				ctx.scheduledTaskMaxReadLevelMap[cluster.TestCurrentClusterName] = testReadCursorTime
+				ctx.logger = testlogger.New(s.T())
+			},
+			domainEntry: cache.NewLocalDomainCacheEntryForTest(domainInfo, &persistence.DomainConfig{}, cluster.TestCurrentClusterName),
+			workflowID:  testWorkflowID,
+			timerTasks: []persistence.Task{
+				&persistence.UserTimerTask{
+					TaskData: persistence.TaskData{
+						VisibilityTimestamp: testVisibilityTime,
+					},
+				},
+			},
+			expectError: false,
+			validateResults: func(t *testing.T, ctx *contextImpl, tasks []persistence.Task) {
+				require.Len(t, tasks, 1)
+				// Timestamp should have been adjusted to read cursor + precision
+				expectedTime := testReadCursorTime.Add(persistence.DBTimestampMinPrecision)
+				assert.Equal(t, expectedTime, tasks[0].GetVisibilityTimestamp())
+			},
+		},
+		{
+			name: "Task timestamp before current time - should use current time as base",
+			setupContext: func(ctx *contextImpl) {
+				ctx.config.EnableTimerQueueV2 = func(shardID int) bool { return true }
+				// Set read cursor to past time
+				ctx.scheduledTaskMaxReadLevelMap[cluster.TestCurrentClusterName] = testTimeNow.Add(-2 * time.Hour)
+			},
+			domainEntry: cache.NewLocalDomainCacheEntryForTest(domainInfo, &persistence.DomainConfig{}, cluster.TestCurrentClusterName),
+			workflowID:  testWorkflowID,
+			timerTasks: []persistence.Task{
+				&persistence.UserTimerTask{
+					TaskData: persistence.TaskData{
+						VisibilityTimestamp: testVisibilityTime,
+					},
+				},
+			},
+			expectError: false,
+			validateResults: func(t *testing.T, ctx *contextImpl, tasks []persistence.Task) {
+				require.Len(t, tasks, 1)
+				// Should use current time as base for adjustment
+				expectedTime := testTimeNow.Truncate(persistence.DBTimestampMinPrecision).Add(persistence.DBTimestampMinPrecision)
+				assert.Equal(t, expectedTime, tasks[0].GetVisibilityTimestamp())
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		s.Run(tt.name, func() {
+			// Reset context to clean state
+			context := s.newContext()
+
+			// Apply test-specific setup
+			if tt.setupContext != nil {
+				tt.setupContext(context)
+			}
+
+			// Execute the method under test
+			err := context.allocateTimerIDsLocked(tt.domainEntry, tt.workflowID, tt.timerTasks)
+
+			// Validate results
+			if tt.expectError {
+				s.Error(err)
+				if tt.expectedErrorMsg != "" {
+					s.Contains(err.Error(), tt.expectedErrorMsg)
+				}
+			} else {
+				s.NoError(err)
+				if tt.validateResults != nil {
+					tt.validateResults(s.T(), context, tt.timerTasks)
+				}
+			}
+		})
+	}
+}