Revert "Treat serviceerror.Internal as non-retryable" (#6729)

Reverts #6622

common/dynamicconfig/constants.go

@@ -2195,6 +2195,12 @@ Do not turn this on if you aren't using Cassandra as the history task DLQ is not
 		70, // 70 attempts takes about an hour
 		`HistoryTaskDLQUnexpectedErrorAttempts is the number of task execution attempts before sending the task to DLQ.`,
 	)
+	HistoryTaskDLQInternalErrors = NewGlobalBoolSetting(
+		"history.TaskDLQInternalErrors",
+		false,
+		`HistoryTaskDLQInternalErrors causes history task processing to send tasks failing with serviceerror.Internal to
+the dlq (or will drop them if not enabled)`,
+	)
 	HistoryTaskDLQErrorPattern = NewGlobalStringSetting(
 		"history.TaskDLQErrorPattern",
 		"",

common/util.go

@@ -343,7 +343,8 @@ func IsServiceHandlerRetryableError(err error) bool {
 	}
 
 	switch err.(type) {
-	case *serviceerror.Unavailable:
+	case *serviceerror.Internal,
+		*serviceerror.Unavailable:
 		return true
 	}
 

service/history/archival_queue_factory.go

@@ -184,6 +184,7 @@ func (f *archivalQueueFactory) newScheduledQueue(shard shard.Context, executor q
 		f.DLQWriter,
 		f.Config.TaskDLQEnabled,
 		f.Config.TaskDLQUnexpectedErrorAttempts,
+		f.Config.TaskDLQInternalErrors,
 		f.Config.TaskDLQErrorPattern,
 	)
 	return queues.NewScheduledQueue(

service/history/configs/config.go

@@ -115,6 +115,7 @@ type Config struct {
 
 	TaskDLQEnabled                 dynamicconfig.BoolPropertyFn
 	TaskDLQUnexpectedErrorAttempts dynamicconfig.IntPropertyFn
+	TaskDLQInternalErrors          dynamicconfig.BoolPropertyFn
 	TaskDLQErrorPattern            dynamicconfig.StringPropertyFn
 
 	TaskSchedulerEnableRateLimiter           dynamicconfig.BoolPropertyFn
@@ -456,6 +457,7 @@ func NewConfig(
 
 		TaskDLQEnabled:                 dynamicconfig.HistoryTaskDLQEnabled.Get(dc),
 		TaskDLQUnexpectedErrorAttempts: dynamicconfig.HistoryTaskDLQUnexpectedErrorAttempts.Get(dc),
+		TaskDLQInternalErrors:          dynamicconfig.HistoryTaskDLQInternalErrors.Get(dc),
 		TaskDLQErrorPattern:            dynamicconfig.HistoryTaskDLQErrorPattern.Get(dc),
 
 		TaskSchedulerEnableRateLimiter:           dynamicconfig.TaskSchedulerEnableRateLimiter.Get(dc),

service/history/outbound_queue_factory.go

@@ -294,6 +294,7 @@ func (f *outboundQueueFactory) CreateQueue(
 		f.DLQWriter,
 		f.Config.TaskDLQEnabled,
 		f.Config.TaskDLQUnexpectedErrorAttempts,
+		f.Config.TaskDLQInternalErrors,
 		f.Config.TaskDLQErrorPattern,
 	)
 	return queues.NewImmediateQueue(

service/history/queues/executable.go

@@ -171,12 +171,14 @@ type (
 		terminalFailureCause       error
 		unexpectedErrorAttempts    int
 		maxUnexpectedErrorAttempts dynamicconfig.IntPropertyFn
+		dlqInternalErrors          dynamicconfig.BoolPropertyFn
 		dlqErrorPattern            dynamicconfig.StringPropertyFn
 	}
 	ExecutableParams struct {
 		DLQEnabled                 dynamicconfig.BoolPropertyFn
 		DLQWriter                  *DLQWriter
 		MaxUnexpectedErrorAttempts dynamicconfig.IntPropertyFn
+		DLQInternalErrors          dynamicconfig.BoolPropertyFn
 		DLQErrorPattern            dynamicconfig.StringPropertyFn
 	}
 	ExecutableOption func(*ExecutableParams)
@@ -204,6 +206,9 @@ func NewExecutable(
 		MaxUnexpectedErrorAttempts: func() int {
 			return math.MaxInt
 		},
+		DLQInternalErrors: func() bool {
+			return false
+		},
 		DLQErrorPattern: func() string {
 			return ""
 		},
@@ -235,6 +240,7 @@ func NewExecutable(
 		dlqWriter:                  params.DLQWriter,
 		dlqEnabled:                 params.DLQEnabled,
 		maxUnexpectedErrorAttempts: params.MaxUnexpectedErrorAttempts,
+		dlqInternalErrors:          params.DLQInternalErrors,
 		dlqErrorPattern:            params.DLQErrorPattern,
 	}
 	executable.updatePriority()
@@ -446,14 +452,14 @@ func (e *executableImpl) isUnexpectedNonRetryableError(err error) bool {
 	}
 
 	if _, isDataLoss := err.(*serviceerror.DataLoss); isDataLoss {
-		metrics.TaskCorruptionCounter.With(e.taggedMetricsHandler).Record(1)
 		return true
 	}
 
 	isInternalError := common.IsInternalError(err)
 	if isInternalError {
 		metrics.TaskInternalErrorCounter.With(e.taggedMetricsHandler).Record(1)
-		return true
+		// Only DQL/drop when configured to
+		return e.dlqInternalErrors()
 	}
 
 	return false
@@ -520,8 +526,10 @@ func (e *executableImpl) HandleErr(err error) (retErr error) {
 	e.logger.Warn("Fail to process task", tag.Error(err), tag.ErrorType(err), tag.UnexpectedErrorAttempts(int32(e.unexpectedErrorAttempts)), tag.LifeCycleProcessingFailed)
 
 	if e.isUnexpectedNonRetryableError(err) {
+		// Terminal errors are likely due to data corruption.
 		// Drop the task by returning nil so that task will be marked as completed,
 		// or send it to the DLQ if that is enabled.
+		metrics.TaskCorruptionCounter.With(e.taggedMetricsHandler).Record(1)
 		if e.dlqEnabled() {
 			// Keep this message in sync with the log line mentioned in Investigation section of docs/admin/dlq.md
 			e.logger.Error("Marking task as terminally failed, will send to DLQ", tag.Error(err), tag.ErrorType(err))

service/history/queues/executable_factory.go

@@ -77,6 +77,7 @@ func NewExecutableFactory(
 	dlqWriter *DLQWriter,
 	dlqEnabled dynamicconfig.BoolPropertyFn,
 	attemptsBeforeSendingToDlq dynamicconfig.IntPropertyFn,
+	dlqInternalErrors dynamicconfig.BoolPropertyFn,
 	dlqErrorPattern dynamicconfig.StringPropertyFn,
 ) *executableFactoryImpl {
 	return &executableFactoryImpl{
@@ -92,6 +93,7 @@ func NewExecutableFactory(
 		dlqWriter:                  dlqWriter,
 		dlqEnabled:                 dlqEnabled,
 		attemptsBeforeSendingToDlq: attemptsBeforeSendingToDlq,
+		dlqInternalErrors:          dlqInternalErrors,
 		dlqErrorPattern:            dlqErrorPattern,
 	}
 }
@@ -113,6 +115,7 @@ func (f *executableFactoryImpl) NewExecutable(task tasks.Task, readerID int64) E
 			params.DLQEnabled = f.dlqEnabled
 			params.DLQWriter = f.dlqWriter
 			params.MaxUnexpectedErrorAttempts = f.attemptsBeforeSendingToDlq
+			params.DLQInternalErrors = f.dlqInternalErrors
 			params.DLQErrorPattern = f.dlqErrorPattern
 		},
 	)

service/history/queues/executable_test.go

@@ -75,6 +75,7 @@ type (
 		dlqEnabled                 dynamicconfig.BoolPropertyFn
 		priorityAssigner           queues.PriorityAssigner
 		maxUnexpectedErrorAttempts dynamicconfig.IntPropertyFn
+		dlqInternalErrors          dynamicconfig.BoolPropertyFn
 		dlqErrorPattern            dynamicconfig.StringPropertyFn
 	}
 	option func(*params)
@@ -539,13 +540,16 @@ func (s *executableSuite) TestExecute_SendToDLQAfterMaxAttemptsThenDisable() {
 	s.Empty(queueWriter.EnqueueTaskRequests)
 }
 
-func (s *executableSuite) TestExecute_SendsInternalErrorsToDLQ() {
+func (s *executableSuite) TestExecute_SendsInternalErrorsToDLQ_WhenEnabled() {
 	queueWriter := &queuestest.FakeQueueWriter{}
 	executable := s.newTestExecutable(func(p *params) {
 		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry)
 		p.dlqEnabled = func() bool {
 			return true
 		}
+		p.dlqInternalErrors = func() bool {
+			return true
+		}
 	})
 
 	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
@@ -560,6 +564,35 @@ func (s *executableSuite) TestExecute_SendsInternalErrorsToDLQ() {
 	s.Len(queueWriter.EnqueueTaskRequests, 1)
 }
 
+func (s *executableSuite) TestExecute_DoesntSendInternalErrorsToDLQ_WhenDisabled() {
+	queueWriter := &queuestest.FakeQueueWriter{}
+	executable := s.newTestExecutable(func(p *params) {
+		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry)
+		p.dlqEnabled = func() bool {
+			return true
+		}
+		p.dlqInternalErrors = func() bool {
+			return false
+		}
+	})
+
+	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
+		ExecutionMetricTags: nil,
+		ExecutedAsActive:    false,
+		ExecutionErr:        serviceerror.NewInternal("injected error"),
+	}).Times(2)
+
+	// Attempt 1
+	err := executable.Execute()
+	s.Error(executable.HandleErr(err))
+
+	// Attempt 2
+	err = executable.Execute()
+	s.Error(err)
+	s.Error(executable.HandleErr(err))
+	s.Empty(queueWriter.EnqueueTaskRequests)
+}
+
 func (s *executableSuite) TestExecute_SendInternalErrorsToDLQ_ThenDisable() {
 	queueWriter := &queuestest.FakeQueueWriter{}
 	dlqEnabled := true
@@ -568,6 +601,9 @@ func (s *executableSuite) TestExecute_SendInternalErrorsToDLQ_ThenDisable() {
 		p.dlqEnabled = func() bool {
 			return dlqEnabled
 		}
+		p.dlqInternalErrors = func() bool {
+			return true
+		}
 	})
 
 	injectedErr := serviceerror.NewInternal("injected error")
@@ -1005,6 +1041,9 @@ func (s *executableSuite) newTestExecutable(opts ...option) queues.Executable {
 		dlqEnabled: func() bool {
 			return false
 		},
+		dlqInternalErrors: func() bool {
+			return false
+		},
 		priorityAssigner: queues.NewNoopPriorityAssigner(),
 		maxUnexpectedErrorAttempts: func() int {
 			return math.MaxInt
@@ -1040,6 +1079,7 @@ func (s *executableSuite) newTestExecutable(opts ...option) queues.Executable {
 			params.DLQEnabled = p.dlqEnabled
 			params.DLQWriter = p.dlqWriter
 			params.MaxUnexpectedErrorAttempts = p.maxUnexpectedErrorAttempts
+			params.DLQInternalErrors = p.dlqInternalErrors
 			params.DLQErrorPattern = p.dlqErrorPattern
 		},
 	)

service/history/queues/queue_base_test.go

@@ -574,6 +574,9 @@ func (s *queueBaseSuite) newQueueBase(
 		func() int {
 			return math.MaxInt
 		},
+		func() bool {
+			return false
+		},
 		func() string {
 			return ""
 		},

service/history/queues/queue_scheduled_test.go

@@ -146,6 +146,9 @@ func (s *scheduledQueueSuite) SetupTest() {
 		func() int {
 			return math.MaxInt
 		},
+		func() bool {
+			return false
+		},
 		func() string {
 			return ""
 		},

service/history/timer_queue_factory.go

@@ -242,6 +242,7 @@ func (f *timerQueueFactory) CreateQueue(
 		f.DLQWriter,
 		f.Config.TaskDLQEnabled,
 		f.Config.TaskDLQUnexpectedErrorAttempts,
+		f.Config.TaskDLQInternalErrors,
 		f.Config.TaskDLQErrorPattern,
 	)
 	return queues.NewScheduledQueue(

service/history/transfer_queue_factory.go

@@ -236,6 +236,7 @@ func (f *transferQueueFactory) CreateQueue(
 		f.DLQWriter,
 		f.Config.TaskDLQEnabled,
 		f.Config.TaskDLQUnexpectedErrorAttempts,
+		f.Config.TaskDLQInternalErrors,
 		f.Config.TaskDLQErrorPattern,
 	)
 	return queues.NewImmediateQueue(

service/history/visibility_queue_factory.go

@@ -142,6 +142,7 @@ func (f *visibilityQueueFactory) CreateQueue(
 		f.DLQWriter,
 		f.Config.TaskDLQEnabled,
 		f.Config.TaskDLQUnexpectedErrorAttempts,
+		f.Config.TaskDLQInternalErrors,
 		f.Config.TaskDLQErrorPattern,
 	)
 	return queues.NewImmediateQueue(