Retry standby tasks immediately on failover (#7199)

Shaddoll · web-flow · commit 8430e3432cab · 2025-08-21T22:20:48.000Z
diff --git a/common/task/fifo_task_scheduler.go b/common/task/fifo_task_scheduler.go
@@ -151,7 +151,7 @@ func (f *fifoTaskSchedulerImpl) dispatcher() {
 		case task := <-f.taskCh:
 			if err := f.processor.Submit(task); err != nil {
 				f.logger.Error("failed to submit task to processor", tag.Error(err))
-				task.Nack()
+				task.Nack(err)
 			}
 		case <-f.shutdownCh:
 			return
@@ -167,7 +167,7 @@ func (f *fifoTaskSchedulerImpl) drainAndNackTasks() {
 	for {
 		select {
 		case task := <-f.taskCh:
-			task.Nack()
+			task.Nack(nil)
 		default:
 			return
 		}
diff --git a/common/task/interface.go b/common/task/interface.go
@@ -57,7 +57,7 @@ type (
 		// Ack marks the task as successful completed
 		Ack()
 		// Nack marks the task as unsuccessful completed
-		Nack()
+		Nack(err error)
 		// Cancel marks the task as canceled
 		Cancel()
 		// State returns the current task state
diff --git a/common/task/interface_mock.go b/common/task/interface_mock.go
diff --git a/common/task/parallel_task_processor.go b/common/task/parallel_task_processor.go
@@ -158,7 +158,7 @@ func (p *parallelTaskProcessorImpl) executeTask(task Task, shutdownCh chan struc
 		if r := recover(); r != nil {
 			p.logger.Error("recovered panic in task execution", tag.Dynamic("recovered-panic", r))
 			task.HandleErr(fmt.Errorf("recovered panic: %v", r))
-			task.Nack()
+			task.Nack(nil)
 		}
 	}()
 
@@ -185,7 +185,7 @@ func (p *parallelTaskProcessorImpl) executeTask(task Task, shutdownCh chan struc
 
 	if err := throttleRetry.Do(context.Background(), op); err != nil {
 		// non-retryable error or exhausted all retries or worker shutdown
-		task.Nack()
+		task.Nack(err)
 		return
 	}
 
@@ -245,7 +245,7 @@ func (p *parallelTaskProcessorImpl) drainAndNackTasks() {
 	for {
 		select {
 		case task := <-p.tasksCh:
-			task.Nack()
+			task.Nack(nil)
 		default:
 			return
 		}
diff --git a/common/task/parallel_task_processor_test.go b/common/task/parallel_task_processor_test.go
@@ -146,7 +146,7 @@ func (s *parallelTaskProcessorSuite) TestExecuteTask_NonRetryableError() {
 		mockTask.EXPECT().Execute().Return(errNonRetryable),
 		mockTask.EXPECT().HandleErr(errNonRetryable).Return(errNonRetryable),
 		mockTask.EXPECT().RetryErr(errNonRetryable).Return(false).AnyTimes(),
-		mockTask.EXPECT().Nack(),
+		mockTask.EXPECT().Nack(gomock.Any()),
 	)
 
 	s.processor.executeTask(mockTask, make(chan struct{}))
@@ -157,7 +157,7 @@ func (s *parallelTaskProcessorSuite) TestExecuteTask_WorkerStopped() {
 	mockTask.EXPECT().Execute().Return(errRetryable).AnyTimes()
 	mockTask.EXPECT().HandleErr(errRetryable).Return(errRetryable).AnyTimes()
 	mockTask.EXPECT().RetryErr(errRetryable).Return(true).AnyTimes()
-	mockTask.EXPECT().Nack().Times(1)
+	mockTask.EXPECT().Nack(gomock.Any()).Times(1)
 
 	done := make(chan struct{})
 	workerShutdownCh := make(chan struct{})
@@ -265,7 +265,7 @@ func (s *parallelTaskProcessorSuite) TestProcessorContract() {
 			taskStatus[mockTask] = TaskStateAcked
 			taskWG.Done()
 		}).MaxTimes(1)
-		mockTask.EXPECT().Nack().Do(func() {
+		mockTask.EXPECT().Nack(gomock.Any()).Do(func(err error) {
 			taskStatusLock.Lock()
 			defer taskStatusLock.Unlock()
 
@@ -315,7 +315,7 @@ func (s *parallelTaskProcessorSuite) TestExecuteTask_PanicHandling() {
 		panic("A panic occurred")
 	})
 	mockTask.EXPECT().HandleErr(gomock.Any()).Return(errRetryable).AnyTimes()
-	mockTask.EXPECT().Nack().Times(1)
+	mockTask.EXPECT().Nack(gomock.Any()).Times(1)
 	done := make(chan struct{})
 	workerShutdownCh := make(chan struct{})
 	go func() {
diff --git a/common/task/sequential_task_processor.go b/common/task/sequential_task_processor.go
@@ -185,7 +185,7 @@ func (t *sequentialTaskProcessorImpl) processTaskOnce(taskqueue SequentialTaskQu
 			taskqueue.Add(task)
 		} else {
 			t.logger.Error("Unable to process task", tag.Error(err))
-			task.Nack()
+			task.Nack(err)
 		}
 	} else {
 		task.Ack()
diff --git a/common/task/sequential_task_processor_test.go b/common/task/sequential_task_processor_test.go
@@ -276,7 +276,7 @@ func (t *testSequentialTaskImpl) NumAcked() int {
 	return t.acked
 }
 
-func (t *testSequentialTaskImpl) Nack() {
+func (t *testSequentialTaskImpl) Nack(err error) {
 	t.lock.Lock()
 	defer t.lock.Unlock()
 
diff --git a/common/task/weighted_round_robin_task_scheduler.go b/common/task/weighted_round_robin_task_scheduler.go
@@ -191,7 +191,7 @@ func (w *weightedRoundRobinTaskSchedulerImpl[K]) dispatchTasks() {
 				hasTask = true
 				if err := w.processor.Submit(task); err != nil {
 					w.logger.Error("fail to submit task to processor", tag.Error(err))
-					task.Nack()
+					task.Nack(err)
 				}
 			case <-w.shutdownCh:
 				return
@@ -218,7 +218,7 @@ func drainAndNackPriorityTask(taskCh <-chan PriorityTask) {
 	for {
 		select {
 		case task := <-taskCh:
-			task.Nack()
+			task.Nack(nil)
 		default:
 			return
 		}
diff --git a/common/task/weighted_round_robin_task_scheduler_test.go b/common/task/weighted_round_robin_task_scheduler_test.go
@@ -227,7 +227,7 @@ func (s *weightedRoundRobinTaskSchedulerSuite) TestDispatcher_SubmitWithNoError(
 func (s *weightedRoundRobinTaskSchedulerSuite) TestDispatcher_FailToSubmit() {
 	mockTask := NewMockPriorityTask(s.controller)
 	mockTask.EXPECT().Priority().Return(0)
-	mockTask.EXPECT().Nack()
+	mockTask.EXPECT().Nack(gomock.Any())
 
 	var taskWG sync.WaitGroup
 	s.scheduler.Submit(mockTask)
@@ -328,7 +328,7 @@ func testSchedulerContract(
 			taskStatus[mockTask] = TaskStateAcked
 			taskWG.Done()
 		}).MaxTimes(1)
-		mockTask.EXPECT().Nack().Do(func() {
+		mockTask.EXPECT().Nack(gomock.Any()).Do(func(err error) {
 			taskStatusLock.Lock()
 			defer taskStatusLock.Unlock()
 
diff --git a/service/history/task/interface_mock.go b/service/history/task/interface_mock.go
diff --git a/service/history/task/standby_task_util.go b/service/history/task/standby_task_util.go
@@ -43,6 +43,10 @@ type (
 	standbyCurrentTimeFn func(persistence.Task) (time.Time, error)
 )
 
+var (
+	errDomainBecomesActive = errors.New("domain becomes active when processing task as standby")
+)
+
 func standbyTaskPostActionNoOp(
 	ctx context.Context,
 	taskInfo persistence.Task,
@@ -222,15 +226,15 @@ func getRemoteClusterName(
 		}
 		if resp.ClusterName == currentCluster {
 			// domain has turned active, retry the task
-			return "", errors.New("domain becomes active when processing task as standby")
+			return "", errDomainBecomesActive
 		}
 		return resp.ClusterName, nil
 	}
 
 	remoteClusterName := domainEntry.GetReplicationConfig().ActiveClusterName
 	if remoteClusterName == currentCluster {
 		// domain has turned active, retry the task
-		return "", errors.New("domain becomes active when processing task as standby")
+		return "", errDomainBecomesActive
 	}
 	return remoteClusterName, nil
 }
diff --git a/service/history/task/task.go b/service/history/task/task.go
@@ -343,14 +343,14 @@ func (t *taskImpl) Ack() {
 	}
 }
 
-func (t *taskImpl) Nack() {
+func (t *taskImpl) Nack(err error) {
 	if t.State() != ctask.TaskStatePending {
 		return
 	}
 
 	logEvent(t.eventLogger, "Nacked task")
 
-	if t.shouldResubmitOnNack() {
+	if t.shouldResubmitOnNack(err) {
 		if submitted, _ := t.taskProcessor.TrySubmit(t); submitted {
 			return
 		}
@@ -415,10 +415,13 @@ func (t *taskImpl) SetInitialSubmitTime(submitTime time.Time) {
 	t.initialSubmitTime = submitTime
 }
 
-func (t *taskImpl) shouldResubmitOnNack() bool {
+func (t *taskImpl) shouldResubmitOnNack(err error) bool {
 	// TODO: for now only resubmit active task on Nack()
 	// we can also consider resubmit standby tasks that fails due to certain error types
 	// this may require change the Nack() interface to Nack(error)
+	if errors.Is(err, errDomainBecomesActive) {
+		return true
+	}
 	return t.GetAttempt() < activeTaskResubmitMaxAttempts &&
 		(t.queueType == QueueTypeActiveTransfer || t.queueType == QueueTypeActiveTimer || ((t.queueType == QueueTypeTransfer || t.queueType == QueueTypeTimer) && t.isPreviousExecutorActive))
 }
diff --git a/service/history/task/task_rate_limiter_test.go b/service/history/task/task_rate_limiter_test.go
@@ -78,7 +78,7 @@ func (s *noopTask) Ack() {
 	s.state = ctask.TaskStateAcked
 }
 
-func (s *noopTask) Nack() {
+func (s *noopTask) Nack(err error) {
 	s.Lock()
 	defer s.Unlock()
 	if s.state != ctask.TaskStatePending {
diff --git a/service/history/task/task_test.go b/service/history/task/task_test.go
@@ -273,7 +273,7 @@ func (s *taskSuite) TestTaskCancel() {
 	taskBase.Ack()
 	s.Equal(ctask.TaskStateCanceled, taskBase.State())
 
-	taskBase.Nack()
+	taskBase.Nack(nil)
 	s.Equal(ctask.TaskStateCanceled, taskBase.State())
 
 	s.False(taskBase.RetryErr(errors.New("some random error")))
@@ -289,7 +289,7 @@ func (s *taskSuite) TestTaskState() {
 	taskBase.Ack()
 	s.Equal(ctask.TaskStateAcked, taskBase.State())
 
-	taskBase.Nack()
+	taskBase.Nack(nil)
 	s.Equal(ctask.TaskStateAcked, taskBase.State())
 }
 
@@ -312,7 +312,21 @@ func (s *taskSuite) TestTaskNack_ResubmitSucceeded() {
 
 	s.mockTaskProcessor.EXPECT().TrySubmit(task).Return(true, nil).Times(1)
 
-	task.Nack()
+	task.Nack(nil)
+	s.Equal(ctask.TaskStatePending, task.State())
+}
+
+func (s *taskSuite) TestTaskNack_DomainBecomesActive() {
+	task := s.newTestTask(
+		func(task persistence.Task) (bool, error) {
+			return true, nil
+		},
+	)
+	task.queueType = QueueTypeTransfer
+
+	s.mockTaskProcessor.EXPECT().TrySubmit(task).Return(true, nil).Times(1)
+
+	task.Nack(errDomainBecomesActive)
 	s.Equal(ctask.TaskStatePending, task.State())
 }
 
@@ -326,7 +340,7 @@ func (s *taskSuite) TestTaskNack_ResubmitFailed() {
 	s.mockTaskProcessor.EXPECT().TrySubmit(task).Return(false, errTaskProcessorNotRunning).Times(1)
 	s.mockTaskRedispatcher.EXPECT().RedispatchTask(task, gomock.Any()).Times(1)
 
-	task.Nack()
+	task.Nack(nil)
 	s.Equal(ctask.TaskStatePending, task.State())
 }
 

Original file line number	Diff line number	Diff line change
`@@ -185,7 +185,7 @@ func (t *sequentialTaskProcessorImpl) processTaskOnce(taskqueue SequentialTaskQu`
`185`	`185`	`taskqueue.Add(task)`
`186`	`186`	`} else {`
`187`	`187`	`t.logger.Error("Unable to process task", tag.Error(err))`
`188`		`- task.Nack()`
	`188`	`+ task.Nack(err)`
`189`	`189`	`}`
`190`	`190`	`} else {`
`191`	`191`	`task.Ack()`
Original file line number	Diff line number	Diff line change
`@@ -276,7 +276,7 @@ func (t *testSequentialTaskImpl) NumAcked() int {`
`276`	`276`	`return t.acked`
`277`	`277`	`}`
`278`	`278`
`279`		`-func (t *testSequentialTaskImpl) Nack() {`
	`279`	`+func (t *testSequentialTaskImpl) Nack(err error) {`
`280`	`280`	`t.lock.Lock()`
`281`	`281`	`defer t.lock.Unlock()`
`282`	`282`