tasks/service/history/queues/executable_test.go

package queues_test

import (
	"context"
	"errors"
	"fmt"
	"math"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"
	enumspb "go.temporal.io/api/enums/v1"
	"go.temporal.io/api/serviceerror"
	"go.temporal.io/server/chasm"
	"go.temporal.io/server/common/clock"
	"go.temporal.io/server/common/cluster"
	"go.temporal.io/server/common/definition"
	"go.temporal.io/server/common/dynamicconfig"
	"go.temporal.io/server/common/headers"
	"go.temporal.io/server/common/log"
	"go.temporal.io/server/common/metrics"
	"go.temporal.io/server/common/metrics/metricstest"
	"go.temporal.io/server/common/namespace"
	"go.temporal.io/server/common/persistence"
	"go.temporal.io/server/common/persistence/serialization"
	ctasks "go.temporal.io/server/common/tasks"
	"go.temporal.io/server/common/telemetry"
	"go.temporal.io/server/service/history/consts"
	"go.temporal.io/server/service/history/queues"
	"go.temporal.io/server/service/history/queues/queuestest"
	"go.temporal.io/server/service/history/tasks"
	"go.temporal.io/server/service/history/tests"
	"go.uber.org/mock/gomock"
)

type (
	executableSuite struct {
		suite.Suite
		*require.Assertions

		controller            *gomock.Controller
		mockExecutor          *queues.MockExecutor
		mockScheduler         *queues.MockScheduler
		mockRescheduler       *queues.MockRescheduler
		mockNamespaceRegistry *namespace.MockRegistry
		mockClusterMetadata   *cluster.MockMetadata
		chasmRegistry         *chasm.Registry
		metricsHandler        *metricstest.CaptureHandler

		timeSource *clock.EventTimeSource
	}
	params struct {
		dlqWriter                  *queues.DLQWriter
		dlqEnabled                 dynamicconfig.BoolPropertyFn
		priorityAssigner           queues.PriorityAssigner
		maxUnexpectedErrorAttempts dynamicconfig.IntPropertyFn
		dlqInternalErrors          dynamicconfig.BoolPropertyFn
		dlqErrorPattern            dynamicconfig.StringPropertyFn
	}
	option func(*params)
)

func TestExecutableSuite(t *testing.T) {
	s := new(executableSuite)
	suite.Run(t, s)
}

func (s *executableSuite) SetupTest() {
	s.Assertions = require.New(s.T())

	s.controller = gomock.NewController(s.T())
	s.mockExecutor = queues.NewMockExecutor(s.controller)
	s.mockScheduler = queues.NewMockScheduler(s.controller)
	s.mockRescheduler = queues.NewMockRescheduler(s.controller)
	s.mockNamespaceRegistry = namespace.NewMockRegistry(s.controller)
	s.mockClusterMetadata = cluster.NewMockMetadata(s.controller)
	s.metricsHandler = metricstest.NewCaptureHandler()

	s.mockNamespaceRegistry.EXPECT().GetNamespaceName(gomock.Any()).Return(tests.Namespace, nil).AnyTimes()
	s.mockNamespaceRegistry.EXPECT().GetNamespaceByID(gomock.Any()).Return(tests.GlobalNamespaceEntry, nil).AnyTimes()
	s.mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes()
	s.mockClusterMetadata.EXPECT().GetAllClusterInfo().Return(map[string]cluster.ClusterInformation{
		cluster.TestCurrentClusterName: {
			ShardCount: 1,
		},
	}).AnyTimes()

	s.timeSource = clock.NewEventTimeSource()
	s.chasmRegistry = chasm.NewRegistry(log.NewTestLogger())
}

func (s *executableSuite) TearDownSuite() {
	s.controller.Finish()
}

func (s *executableSuite) TestExecute_TaskExecuted() {
	executable := s.newTestExecutable()

	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    true,
		ExecutionErr:        errors.New("some random error"),
	})
	s.Error(executable.Execute())

	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    true,
		ExecutionErr:        nil,
	})
	s.NoError(executable.Execute())
}

func (s *executableSuite) TestExecute_InMemoryNoUserLatency_SingleAttempt() {
	scheduleLatency := 100 * time.Millisecond
	userLatency := 500 * time.Millisecond
	attemptLatency := time.Second
	attemptNoUserLatency := scheduleLatency + attemptLatency - userLatency

	testCases := []struct {
		name                         string
		taskErr                      error
		expectError                  bool
		expectedAttemptNoUserLatency time.Duration
		expectBackoff                bool
	}{
		{
			name:                         "NoError",
			taskErr:                      nil,
			expectError:                  false,
			expectedAttemptNoUserLatency: attemptNoUserLatency,
			expectBackoff:                false,
		},
		{
			name:                         "UnavailableError",
			taskErr:                      serviceerror.NewUnavailable("some random error"),
			expectError:                  true,
			expectedAttemptNoUserLatency: attemptNoUserLatency,
			expectBackoff:                true,
		},
		{
			name:                         "NotFoundError",
			taskErr:                      serviceerror.NewNotFound("not found error"),
			expectError:                  false,
			expectedAttemptNoUserLatency: 0,
			expectBackoff:                false,
		},
		{
			name:                         "NotFoundErrorWrapped",
			taskErr:                      fmt.Errorf("%w: some reason", consts.ErrWorkflowCompleted),
			expectError:                  false,
			expectedAttemptNoUserLatency: 0,
			expectBackoff:                false,
		},
		{
			name:                         "BusyWorkflowError",
			taskErr:                      consts.ErrResourceExhaustedBusyWorkflow,
			expectError:                  true,
			expectedAttemptNoUserLatency: 0,
			expectBackoff:                false,
		},
		{
			name:                         "APSLimitError",
			taskErr:                      consts.ErrResourceExhaustedBusyWorkflow,
			expectError:                  true,
			expectedAttemptNoUserLatency: 0,
			expectBackoff:                false,
		},
		{
			name: "OPSLimitError",
			taskErr: &serviceerror.ResourceExhausted{
				Cause:   enumspb.RESOURCE_EXHAUSTED_CAUSE_OPS_LIMIT,
				Scope:   enumspb.RESOURCE_EXHAUSTED_SCOPE_NAMESPACE,
				Message: "Namespace Max OPS Limit Reached.",
			},
			expectError:                  true,
			expectedAttemptNoUserLatency: 0,
			expectBackoff:                false,
		},
		{
			name:                         "PersistenceNamespaceLimitExceeded",
			taskErr:                      persistence.ErrPersistenceNamespaceLimitExceeded,
			expectError:                  true,
			expectedAttemptNoUserLatency: 0,
			expectBackoff:                false,
		},
	}

	for _, tc := range testCases {
		s.Run(tc.name, func() {

			executable := s.newTestExecutable()

			now := time.Now()
			s.timeSource.Update(now)
			executable.SetScheduledTime(now)

			now = now.Add(scheduleLatency)
			s.timeSource.Update(now)

			s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Do(func(ctx context.Context, taskInfo any) {
				metrics.ContextCounterAdd(
					ctx,
					metrics.HistoryWorkflowExecutionCacheLatency.Name(),
					int64(userLatency),
				)

				now = now.Add(attemptLatency)
				s.timeSource.Update(now)
			}).Return(queues.ExecuteResponse{
				ExecutionMetricTags: nil,
				ExecutedAsActive:    true,
				ExecutionErr:        tc.taskErr,
			})

			err := executable.Execute()
			if err != nil {
				err = executable.HandleErr(err)
			}

			if tc.expectError {
				s.Error(err)
				s.mockScheduler.EXPECT().TrySubmit(executable).Return(false)
				s.mockRescheduler.EXPECT().Add(executable, gomock.Any())
				executable.Nack(err)
				return
			}

			s.NoError(err)
			capture := s.metricsHandler.StartCapture()
			executable.Ack()
			snapshot := capture.Snapshot()
			recordings := snapshot[metrics.TaskLatency.Name()]
			if tc.expectedAttemptNoUserLatency == 0 {
				// invalid task, no noUserLatency will be recorded.
				s.Empty(recordings)
				return
			}

			s.Len(recordings, 1)
			actualAttemptNoUserLatency, ok := recordings[0].Value.(time.Duration)
			s.True(ok)
			if tc.expectBackoff {
				// the backoff duration is random, so we can't compare the exact value
				s.Less(tc.expectedAttemptNoUserLatency, actualAttemptNoUserLatency)
			} else {
				s.Equal(tc.expectedAttemptNoUserLatency, actualAttemptNoUserLatency)
			}
		})
	}
}

func (s *executableSuite) TestExecute_InMemoryNoUserLatency_MultipleAttempts() {
	numAttempts := 3
	scheduleLatencies := []time.Duration{100 * time.Millisecond, 150 * time.Millisecond, 200 * time.Millisecond}
	userLatencies := []time.Duration{10 * time.Millisecond, 20 * time.Millisecond, 30 * time.Millisecond}
	attemptLatencies := []time.Duration{time.Second, 2 * time.Second, 3 * time.Second}
	taskErrors := []error{
		serviceerror.NewUnavailable("test unavailable error"),
		consts.ErrResourceExhaustedBusyWorkflow,
		nil,
	}
	expectedInMemoryNoUserLatency := scheduleLatencies[0] + attemptLatencies[0] - userLatencies[0] +
		scheduleLatencies[2] + attemptLatencies[2] - userLatencies[2]
	_ = expectedInMemoryNoUserLatency

	executable := s.newTestExecutable()

	now := time.Now()
	s.timeSource.Update(now)
	executable.SetScheduledTime(now)

	for i := 0; i != numAttempts; i++ {
		now = now.Add(scheduleLatencies[i])
		s.timeSource.Update(now)

		s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Do(func(ctx context.Context, taskInfo any) {
			metrics.ContextCounterAdd(
				ctx,
				metrics.HistoryWorkflowExecutionCacheLatency.Name(),
				int64(userLatencies[i]),
			)

			now = now.Add(attemptLatencies[i])
			s.timeSource.Update(now)
		}).Return(queues.ExecuteResponse{
			ExecutionMetricTags: nil,
			ExecutedAsActive:    true,
			ExecutionErr:        taskErrors[i],
		})

		err := executable.Execute()
		if err != nil {
			err = executable.HandleErr(err)
		}

		if taskErrors[i] != nil {
			s.Error(err)
			s.mockScheduler.EXPECT().TrySubmit(executable).Return(true)
			executable.Nack(err)
		} else {
			s.NoError(err)
			capture := s.metricsHandler.StartCapture()
			executable.Ack()
			snapshot := capture.Snapshot()
			recordings := snapshot[metrics.TaskLatency.Name()]
			s.Len(recordings, 1)
			actualInMemoryNoUserLatency, ok := recordings[0].Value.(time.Duration)
			s.True(ok)
			s.Equal(expectedInMemoryNoUserLatency, actualInMemoryNoUserLatency)
		}
	}
}

func (s *executableSuite) TestExecute_CapturePanic() {
	executable := s.newTestExecutable()

	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).DoAndReturn(
		func(_ context.Context, _ queues.Executable) queues.ExecuteResponse {
			panic("test panic during execution")
		},
	)
	s.Error(executable.Execute())
}

func (s *executableSuite) TestExecute_CallerInfo() {
	executable := s.newTestExecutable()

	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).DoAndReturn(
		func(ctx context.Context, _ queues.Executable) queues.ExecuteResponse {
			s.Equal(headers.CallerTypeBackgroundHigh, headers.GetCallerInfo(ctx).CallerType)
			return queues.ExecuteResponse{
				ExecutionMetricTags: nil,
				ExecutedAsActive:    true,
				ExecutionErr:        nil,
			}
		},
	)
	s.NoError(executable.Execute())

	executable = s.newTestExecutable(func(p *params) {
		p.priorityAssigner = queues.NewStaticPriorityAssigner(ctasks.PriorityLow)
	})
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).DoAndReturn(
		func(ctx context.Context, _ queues.Executable) queues.ExecuteResponse {
			s.Equal(headers.CallerTypeBackgroundLow, headers.GetCallerInfo(ctx).CallerType)
			return queues.ExecuteResponse{
				ExecutionMetricTags: nil,
				ExecutedAsActive:    true,
				ExecutionErr:        nil,
			}
		},
	)
	s.NoError(executable.Execute())

	executable = s.newTestExecutable(func(p *params) {
		p.priorityAssigner = queues.NewStaticPriorityAssigner(ctasks.PriorityPreemptable)
	})
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).DoAndReturn(
		func(ctx context.Context, _ queues.Executable) queues.ExecuteResponse {
			s.Equal(headers.CallerTypePreemptable, headers.GetCallerInfo(ctx).CallerType)
			return queues.ExecuteResponse{
				ExecutionMetricTags: nil,
				ExecutedAsActive:    true,
				ExecutionErr:        nil,
			}
		},
	)
	s.NoError(executable.Execute())
}

func (s *executableSuite) TestExecute_TaskProcessingNoPersistenceLatency_RecordedWhenPersistenceInContext() {
	scheduleLatency := 100 * time.Millisecond
	persistenceDuration := 50 * time.Millisecond
	attemptLatency := 200 * time.Millisecond
	expectedNoPersistence := attemptLatency - persistenceDuration

	executable := s.newTestExecutable()

	now := time.Now()
	s.timeSource.Update(now)
	executable.SetScheduledTime(now)

	now = now.Add(scheduleLatency)
	s.timeSource.Update(now)

	capture := s.metricsHandler.StartCapture()

	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Do(func(ctx context.Context, _ any) {
		metrics.ContextCounterAdd(ctx, metrics.TaskPersistenceLatency.Name(), int64(persistenceDuration))
		now = now.Add(attemptLatency)
		s.timeSource.Update(now)
	}).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    true,
		ExecutionErr:        nil,
	})

	s.NoError(executable.Execute())

	snapshot := capture.Snapshot()
	recordings := snapshot[metrics.TaskProcessingNoPersistenceLatency.Name()]
	s.Require().Len(recordings, 1)
	actualNoPersistence, ok := recordings[0].Value.(time.Duration)
	s.Require().True(ok)
	s.Equal(expectedNoPersistence, actualNoPersistence)
}

func (s *executableSuite) TestExecute_TaskProcessingNoPersistenceLatency_NotRecordedWhenNoPersistence() {
	scheduleLatency := 100 * time.Millisecond
	attemptLatency := 200 * time.Millisecond

	executable := s.newTestExecutable()

	now := time.Now()
	s.timeSource.Update(now)
	executable.SetScheduledTime(now)

	now = now.Add(scheduleLatency)
	s.timeSource.Update(now)

	capture := s.metricsHandler.StartCapture()

	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Do(func(ctx context.Context, _ any) {
		// Do not add TaskPersistenceLatency to context (simulates transfer/timer task that never calls visibility).
		now = now.Add(attemptLatency)
		s.timeSource.Update(now)
	}).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    true,
		ExecutionErr:        nil,
	})

	s.NoError(executable.Execute())

	snapshot := capture.Snapshot()
	recordings := snapshot[metrics.TaskProcessingNoPersistenceLatency.Name()]
	s.Empty(recordings)
}

func (s *executableSuite) TestExecuteHandleErr_ResetAttempt() {
	testCases := []struct {
		name            string
		executionResp   queues.ExecuteResponse
		expectedAttempt int64
	}{
		{
			name: "no failover",
			executionResp: queues.ExecuteResponse{
				ExecutionMetricTags: nil,
				ExecutedAsActive:    true,
				ExecutionErr:        errors.New("some random error"),
			},
			expectedAttempt: 2,
		},
		{
			name: "with failover",
			executionResp: queues.ExecuteResponse{
				ExecutionMetricTags: nil,
				ExecutedAsActive:    false,
				ExecutionErr:        nil,
			},
			expectedAttempt: 1,
		},
	}

	for _, tc := range testCases {
		s.Run(tc.name, func() {
			executable := s.newTestExecutable()
			s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(tc.executionResp)
			err := executable.Execute()
			if tc.executionResp.ExecutionErr != nil {
				s.Error(err)
				s.Error(executable.HandleErr(err))
			} else {
				s.NoError(err)
			}

			capture := s.metricsHandler.StartCapture()
			executable.Ack()
			snapshot := capture.Snapshot()
			s.Equal(tc.expectedAttempt, snapshot[metrics.TaskAttempt.Name()][0].Value)
			s.metricsHandler.StopCapture(capture)
		})
	}
}

func (s *executableSuite) TestExecuteHandleErr_Corrupted() {
	queueWriter := &queuestest.FakeQueueWriter{}
	executable := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return false
		}
	})

	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).DoAndReturn(
		func(_ context.Context, _ queues.Executable) queues.ExecuteResponse {
			panic(serialization.NewUnknownEncodingTypeError("unknownEncoding", enumspb.ENCODING_TYPE_PROTO3))
		},
	).Times(2)
	err := executable.Execute()
	s.Error(err)
	s.NoError(executable.HandleErr(err))
	s.Error(executable.Execute())
}

func (s *executableSuite) TestExecute_SendToDLQAfterMaxAttempts() {
	queueWriter := &queuestest.FakeQueueWriter{}
	executable := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return true
		}
		p.maxUnexpectedErrorAttempts = func() int {
			return 2
		}
	})
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr:        errors.New("some random error"),
	}).Times(2)

	// Attempt 1
	err := executable.Execute()
	s.Error(executable.HandleErr(err))

	// Attempt 2
	err = executable.Execute()
	err2 := executable.HandleErr(err)
	s.ErrorIs(err2, queues.ErrTerminalTaskFailure)
	s.NoError(executable.Execute())
	s.Len(queueWriter.EnqueueTaskRequests, 1)
}

func (s *executableSuite) TestExecute_DontSendToDLQAfterMaxAttemptsDLQDisabled() {
	queueWriter := &queuestest.FakeQueueWriter{}
	executable := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return false
		}
		p.maxUnexpectedErrorAttempts = func() int {
			return 1
		}
	})

	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr:        errors.New("some random error"),
	}).Times(2)

	// Attempt 1
	err := executable.Execute()
	s.Error(executable.HandleErr(err))

	// Attempt 2
	s.Error(executable.Execute())
	s.Error(executable.HandleErr(err))
	s.Empty(queueWriter.EnqueueTaskRequests)
}

func (s *executableSuite) TestExecute_DontSendToDLQAfterMaxAttemptsExpectedError() {
	queueWriter := &queuestest.FakeQueueWriter{}
	executable := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return true
		}
		p.maxUnexpectedErrorAttempts = func() int {
			return 1
		}
	})

	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr: &serviceerror.ResourceExhausted{
			Cause:   enumspb.RESOURCE_EXHAUSTED_CAUSE_BUSY_WORKFLOW,
			Scope:   enumspb.RESOURCE_EXHAUSTED_SCOPE_NAMESPACE,
			Message: "test",
		},
	}).Times(2)

	// Attempt 1
	err := executable.Execute()
	s.Error(executable.HandleErr(err))

	// Attempt 2
	s.Error(executable.Execute())
	s.Error(executable.HandleErr(err))
	s.Empty(queueWriter.EnqueueTaskRequests)
}

func (s *executableSuite) TestExecute_SendToDLQAfterMaxAttemptsThenDisableDropCorruption() {
	queueWriter := &queuestest.FakeQueueWriter{}
	dlqEnabled := true
	executable := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return dlqEnabled
		}
		p.maxUnexpectedErrorAttempts = func() int {
			return 1
		}
	})
	execError := serialization.NewDeserializationError(enumspb.ENCODING_TYPE_PROTO3, errors.New("random error"))
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr:        execError,
	}).Times(1)

	err := executable.Execute()
	err2 := executable.HandleErr(err)

	s.ErrorIs(err2, queues.ErrTerminalTaskFailure)
	s.Contains(err2.Error(), execError.Error())

	dlqEnabled = false
	s.NoError(executable.Execute())
	s.Empty(queueWriter.EnqueueTaskRequests)
}

func (s *executableSuite) TestExecute_SendToDLQAfterMaxAttemptsThenDisable() {
	queueWriter := &queuestest.FakeQueueWriter{}
	dlqEnabled := true
	executable := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return dlqEnabled
		}
		p.maxUnexpectedErrorAttempts = func() int {
			return 1
		}
	})
	execError := errors.New("some random error")
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr:        execError,
	}).Times(1)

	err := executable.Execute()
	err2 := executable.HandleErr(err)

	s.ErrorIs(err2, queues.ErrTerminalTaskFailure)

	dlqEnabled = false

	// Make sure task is retried this time
	execError2 := errors.New("some other random error")
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr:        execError2,
	}).Times(1)
	s.ErrorIs(executable.Execute(), execError2)
	s.Empty(queueWriter.EnqueueTaskRequests)
}

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, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return true
		}
		p.dlqInternalErrors = func() bool {
			return true
		}
	})

	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr:        serviceerror.NewInternal("injected error"),
	}).Times(1)

	err := executable.HandleErr(executable.Execute())
	s.ErrorIs(err, queues.ErrTerminalTaskFailure)
	s.NoError(executable.Execute())
	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, s.chasmRegistry)
		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
	executable := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return dlqEnabled
		}
		p.dlqInternalErrors = func() bool {
			return true
		}
	})

	injectedErr := serviceerror.NewInternal("injected error")
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr:        injectedErr,
	}).Times(2)

	s.ErrorIs(executable.HandleErr(executable.Execute()), queues.ErrTerminalTaskFailure)

	// The task should be dropped but not sent to DLQ
	dlqEnabled = false
	s.ErrorIs(executable.Execute(), injectedErr)
	s.Empty(queueWriter.EnqueueTaskRequests)
}

func (s *executableSuite) TestExecute_DLQ() {
	queueWriter := &queuestest.FakeQueueWriter{}
	executable := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return true
		}
	})

	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).DoAndReturn(
		func(_ context.Context, _ queues.Executable) queues.ExecuteResponse {
			panic(serialization.NewUnknownEncodingTypeError("unknownEncoding", enumspb.ENCODING_TYPE_PROTO3))
		},
	)
	err := executable.Execute()
	s.Error(err)
	s.Error(executable.HandleErr(err))
	s.NoError(executable.Execute())
	s.Len(queueWriter.EnqueueTaskRequests, 1)
}

func (s *executableSuite) TestExecute_DLQThenDisable() {
	queueWriter := &queuestest.FakeQueueWriter{}
	dlqEnabled := true
	executable := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return dlqEnabled
		}
	})

	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).DoAndReturn(
		func(_ context.Context, _ queues.Executable) queues.ExecuteResponse {
			panic(serialization.NewUnknownEncodingTypeError("unknownEncoding", enumspb.ENCODING_TYPE_PROTO3))
		},
	)
	err := executable.Execute()
	s.Error(err)
	s.Error(executable.HandleErr(err))
	dlqEnabled = false
	s.NoError(executable.Execute())
	s.Empty(queueWriter.EnqueueTaskRequests)
}

func (s *executableSuite) TestExecute_DLQFailThenRetry() {
	queueWriter := &queuestest.FakeQueueWriter{}
	executable := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return true
		}
	})

	capture := s.metricsHandler.StartCapture()
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).DoAndReturn(
		func(_ context.Context, _ queues.Executable) queues.ExecuteResponse {
			panic(serialization.NewUnknownEncodingTypeError("unknownEncoding", enumspb.ENCODING_TYPE_PROTO3))
		},
	)
	err := executable.Execute()
	s.Error(err)
	s.Error(executable.HandleErr(err))
	queueWriter.EnqueueTaskErr = errors.New("some random error")
	err = executable.Execute()
	s.Error(err)
	s.Error(executable.HandleErr(err))
	queueWriter.EnqueueTaskErr = nil
	err = executable.Execute()
	s.NoError(err)
	snapshot := capture.Snapshot()
	s.Len(snapshot[metrics.TaskTerminalFailures.Name()], 1)
	s.Len(snapshot[metrics.TaskDLQFailures.Name()], 1)
	s.Len(snapshot[metrics.TaskDLQSendLatency.Name()], 2)
}

func (s *executableSuite) TestHandleErr_EntityNotExists() {
	executable := s.newTestExecutable()

	s.NoError(executable.HandleErr(serviceerror.NewNotFound("")))
}

func (s *executableSuite) TestHandleErr_ErrTaskRetry() {
	executable := s.newTestExecutable()

	s.Equal(consts.ErrTaskRetry, executable.HandleErr(consts.ErrTaskRetry))
}

func (s *executableSuite) TestHandleErr_ErrDeleteOpenExecution() {
	executable := s.newTestExecutable()

	s.Equal(consts.ErrDependencyTaskNotCompleted, executable.HandleErr(consts.ErrDependencyTaskNotCompleted))
}

func (s *executableSuite) TestHandleErr_ErrTaskDiscarded() {
	executable := s.newTestExecutable()

	s.NoError(executable.HandleErr(consts.ErrTaskDiscarded))
}

func (s *executableSuite) TestHandleErr_ErrTaskVersionMismatch() {
	executable := s.newTestExecutable()

	s.NoError(executable.HandleErr(consts.ErrTaskVersionMismatch))
}

func (s *executableSuite) TestHandleErr_NamespaceNotActiveError() {
	err := serviceerror.NewNamespaceNotActive("", "", "")

	s.Equal(err, s.newTestExecutable().HandleErr(err))
}

func (s *executableSuite) TestHandleErr_RandomErr() {
	executable := s.newTestExecutable()

	s.Error(executable.HandleErr(errors.New("random error")))
}

func (s *executableSuite) TestTaskAck_ValidTask_NoRetry() {
	executable := s.newTestExecutable()

	s.Equal(ctasks.TaskStatePending, executable.State())

	capture := s.metricsHandler.StartCapture()

	executable.Ack()
	s.Equal(ctasks.TaskStateAcked, executable.State())

	snapshot := capture.Snapshot()
	s.Len(snapshot[metrics.TaskAttempt.Name()], 1)
	s.Len(snapshot[metrics.TaskLatency.Name()], 1)
	s.Len(snapshot[metrics.TaskQueueLatency.Name()], 1)
}

func (s *executableSuite) TestTaskAck_ValidTask_WithRetry() {
	executable := s.newTestExecutable()

	s.Equal(ctasks.TaskStatePending, executable.State())

	// For retried tasks, they are not considered invalid even
	// if their last attempt completed with a invalid task error.
	_ = executable.HandleErr(context.DeadlineExceeded)
	_ = executable.HandleErr(consts.ErrActivityNotFound)

	capture := s.metricsHandler.StartCapture()

	executable.Ack()
	s.Equal(ctasks.TaskStateAcked, executable.State())

	snapshot := capture.Snapshot()
	s.Len(snapshot[metrics.TaskAttempt.Name()], 1)
	s.Len(snapshot[metrics.TaskLatency.Name()], 1)
	s.Len(snapshot[metrics.TaskQueueLatency.Name()], 1)
}

func (s *executableSuite) TestTaskAck_InvalidTask() {
	executable := s.newTestExecutable()

	s.Equal(ctasks.TaskStatePending, executable.State())

	// This will mark the task as invalid
	_ = executable.HandleErr(consts.ErrActivityNotFound)

	capture := s.metricsHandler.StartCapture()

	executable.Ack()
	s.Equal(ctasks.TaskStateAcked, executable.State())

	snapshot := capture.Snapshot()
	s.Empty(snapshot[metrics.TaskAttempt.Name()])
	s.Empty(snapshot[metrics.TaskLatency.Name()])
	s.Empty(snapshot[metrics.TaskQueueLatency.Name()])
}

func (s *executableSuite) TestTaskNack_Resubmit_Success() {
	executable := s.newTestExecutable()

	s.mockScheduler.EXPECT().TrySubmit(executable).DoAndReturn(func(_ queues.Executable) bool {
		s.Equal(ctasks.TaskStatePending, executable.State())

		go func() {
			// access internal state in a separate goroutine to check if there's any race condition
			// between reschdule and nack.
			s.accessInternalState(executable)
		}()
		return true
	})

	executable.Nack(errors.New("some random error"))
}

func (s *executableSuite) TestTaskNack_Resubmit_Fail() {
	executable := s.newTestExecutable()

	s.mockScheduler.EXPECT().TrySubmit(executable).Return(false)
	s.mockRescheduler.EXPECT().Add(executable, gomock.AssignableToTypeOf(time.Now())).Do(func(_ queues.Executable, _ time.Time) {
		s.Equal(ctasks.TaskStatePending, executable.State())

		go func() {
			// access internal state in a separate goroutine to check if there's any race condition
			// between reschdule and nack.
			s.accessInternalState(executable)
		}()
	}).Times(1)

	executable.Nack(errors.New("some random error"))
}

func (s *executableSuite) TestTaskNack_Reschedule() {

	testCases := []struct {
		name    string
		taskErr error
	}{
		{
			name:    "ErrTaskRetry",
			taskErr: consts.ErrTaskRetry, // this error won't trigger re-submit
		},
		{
			name:    "ErrDeleteOpenExecErr",
			taskErr: consts.ErrDependencyTaskNotCompleted, // this error won't trigger re-submit
		},
		{
			name:    "ErrNamespaceHandover",
			taskErr: consts.ErrNamespaceHandover, // this error won't trigger re-submit
		},
	}

	for _, tc := range testCases {
		s.Run(tc.name, func() {
			executable := s.newTestExecutable()

			s.mockRescheduler.EXPECT().Add(executable, gomock.AssignableToTypeOf(time.Now())).Do(func(_ queues.Executable, _ time.Time) {
				s.Equal(ctasks.TaskStatePending, executable.State())

				go func() {
					// access internal state in a separate goroutine to check if there's any race condition
					// between reschdule and nack.
					s.accessInternalState(executable)
				}()
			}).Times(1)

			executable.Nack(tc.taskErr)
		})
	}
}

func (s *executableSuite) TestTaskAbort() {
	executable := s.newTestExecutable()

	executable.Abort()

	s.NoError(executable.Execute()) // should be no-op and won't invoke executor

	executable.Ack() // should be no-op
	s.Equal(ctasks.TaskStateAborted, executable.State())

	executable.Nack(errors.New("some random error")) // should be no-op and won't invoke scheduler or rescheduler

	executable.Reschedule() // should be no-op and won't invoke rescheduler

	// all error should be treated as non-retryable to break retry loop
	s.False(executable.IsRetryableError(errors.New("some random error")))
}

func (s *executableSuite) TestTaskCancellation() {
	executable := s.newTestExecutable()

	executable.Cancel()

	s.NoError(executable.Execute()) // should be no-op and won't invoke executor

	executable.Ack() // should be no-op
	s.Equal(ctasks.TaskStateCancelled, executable.State())

	executable.Nack(errors.New("some random error")) // should be no-op and won't invoke scheduler or rescheduler

	executable.Reschedule() // should be no-op and won't invoke rescheduler

	// all error should be treated as non-retryable to break retry loop
	s.False(executable.IsRetryableError(errors.New("some random error")))
}

func (s *executableSuite) TestExecute_SendToDLQErrPatternDoesNotMatch() {
	queueWriter := &queuestest.FakeQueueWriter{}
	executable := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return true
		}
		p.dlqErrorPattern = func() string {
			return "testpattern"
		}
	})
	executionError := errors.New("some random error")
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr:        executionError,
	}).Times(2)

	// Attempt 1
	err := executable.Execute()
	err2 := executable.HandleErr(err)
	s.Error(err2)
	s.NotErrorIs(err2, queues.ErrTerminalTaskFailure)
	s.Contains(err2.Error(), executionError.Error())

	// Attempt 2
	s.Error(executable.Execute())
	s.Error(executable.HandleErr(err))
	s.Empty(queueWriter.EnqueueTaskRequests)
}

func (s *executableSuite) TestExecute_SendToDLQErrPatternEmptyString() {
	queueWriter := &queuestest.FakeQueueWriter{}
	executable := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return true
		}
		p.dlqErrorPattern = func() string {
			return ""
		}
	})
	executionError := errors.New("some random error")
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr:        executionError,
	}).Times(2)

	// Attempt 1
	err := executable.Execute()
	err2 := executable.HandleErr(err)
	s.Error(err2)
	s.NotErrorIs(err2, queues.ErrTerminalTaskFailure)
	s.Contains(err2.Error(), executionError.Error())

	// Attempt 2
	s.Error(executable.Execute())
	s.Error(executable.HandleErr(err))
	s.Empty(queueWriter.EnqueueTaskRequests)
}

func (s *executableSuite) TestExecute_SendToDLQErrPatternMatchesMultiple() {
	queueWriter := &queuestest.FakeQueueWriter{}
	executable1 := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return true
		}
		p.dlqErrorPattern = func() string {
			return "test substring 1|test substring 2"
		}
	})
	executionError1 := errors.New("some random error with test substring 1")
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable1).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr:        executionError1,
	}).Times(1)

	executable2 := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return true
		}
		p.dlqErrorPattern = func() string {
			return "test substring 1|test substring 2"
		}
	})
	executionError2 := errors.New("some random error with test substring 2")
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable2).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr:        executionError2,
	}).Times(1)

	// Attempt 1
	err := executable1.Execute()
	err2 := executable1.HandleErr(err)
	s.Error(err2)
	s.ErrorIs(err2, queues.ErrTerminalTaskFailure)
	s.Contains(err2.Error(), executionError1.Error())

	err = executable2.Execute()
	err2 = executable2.HandleErr(err)
	s.Error(err2)
	s.ErrorIs(err2, queues.ErrTerminalTaskFailure)
	s.Contains(err2.Error(), executionError2.Error())

	// Attempt 2
	s.NoError(executable1.Execute())
	s.NoError(executable2.Execute())
	s.Len(queueWriter.EnqueueTaskRequests, 2)
}

func (s *executableSuite) TestExecute_ErrPatternIfDLQDisabled() {
	queueWriter := &queuestest.FakeQueueWriter{}
	executable := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return false
		}
		p.dlqErrorPattern = func() string {
			return "test substring"
		}
	})
	executionError := errors.New("some random error with test substring")
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr:        executionError,
	}).Times(2)

	// Attempt 1
	err := executable.Execute()
	err2 := executable.HandleErr(err)
	s.Error(err2)
	s.ErrorIs(err2, queues.ErrTerminalTaskFailure)
	s.Contains(err2.Error(), executionError.Error())

	// Attempt 2
	s.Error(executable.Execute())
	s.Error(executable.HandleErr(err))
	s.Empty(queueWriter.EnqueueTaskRequests)
}

func (s *executableSuite) TestExecute_ErrorErrPatternThenDisableDLQ() {
	queueWriter := &queuestest.FakeQueueWriter{}
	dlqEnabled := true
	executable := s.newTestExecutable(func(p *params) {
		p.dlqWriter = queues.NewDLQWriter(queueWriter, metrics.NoopMetricsHandler, log.NewTestLogger(), s.mockNamespaceRegistry, s.chasmRegistry)
		p.dlqEnabled = func() bool {
			return dlqEnabled
		}
		p.dlqErrorPattern = func() string {
			return "test substring"
		}
	})
	execError := errors.New("some random error with test substring")
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr:        execError,
	}).Times(1)

	err := executable.Execute()
	err2 := executable.HandleErr(err)

	s.ErrorIs(err2, queues.ErrTerminalTaskFailure)

	dlqEnabled = false

	// Make sure task is retried this time
	execError2 := errors.New("some other random error")
	s.mockExecutor.EXPECT().Execute(gomock.Any(), executable).Return(queues.ExecuteResponse{
		ExecutionMetricTags: nil,
		ExecutedAsActive:    false,
		ExecutionErr:        execError2,
	}).Times(1)
	s.ErrorIs(executable.Execute(), execError2)
	s.Empty(queueWriter.EnqueueTaskRequests)
}

func (s *executableSuite) newTestExecutable(opts ...option) queues.Executable {
	p := params{
		dlqWriter: nil,
		dlqEnabled: func() bool {
			return false
		},
		dlqInternalErrors: func() bool {
			return false
		},
		priorityAssigner: queues.NewNoopPriorityAssigner(),
		maxUnexpectedErrorAttempts: func() int {
			return math.MaxInt
		},
		dlqErrorPattern: func() string {
			return ""
		},
	}
	for _, opt := range opts {
		opt(&p)
	}
	return queues.NewExecutable(
		queues.DefaultReaderId,
		tasks.NewFakeTask(
			definition.NewWorkflowKey(
				tests.NamespaceID.String(),
				tests.WorkflowID,
				tests.RunID,
			),
			tasks.CategoryTransfer,
			s.timeSource.Now(),
		),
		s.mockExecutor,
		s.mockScheduler,
		s.mockRescheduler,
		p.priorityAssigner,
		s.timeSource,
		s.mockNamespaceRegistry,
		s.mockClusterMetadata,
		s.chasmRegistry,
		queues.GetTaskTypeTagValue,
		log.NewTestLogger(),
		s.metricsHandler,
		telemetry.NoopTracer,
		func(params *queues.ExecutableParams) {
			params.DLQEnabled = p.dlqEnabled
			params.DLQWriter = p.dlqWriter
			params.MaxUnexpectedErrorAttempts = p.maxUnexpectedErrorAttempts
			params.DLQInternalErrors = p.dlqInternalErrors
			params.DLQErrorPattern = p.dlqErrorPattern
		},
	)
}

// mockSchedulerWithBusyHandler is a mock scheduler that also implements BusyWorkflowHandler
type mockSchedulerWithBusyHandler struct {
	*queues.MockScheduler
	handleBusyWorkflowCalled bool
	handleBusyWorkflowReturn bool
}

func (m *mockSchedulerWithBusyHandler) HandleBusyWorkflow(executable queues.Executable) bool {
	m.handleBusyWorkflowCalled = true
	return m.handleBusyWorkflowReturn
}

func (s *executableSuite) TestTaskNack_BusyWorkflow_HandledByBusyWorkflowHandler() {
	// Create a mock scheduler that implements BusyWorkflowHandler
	mockScheduler := &mockSchedulerWithBusyHandler{
		MockScheduler:            queues.NewMockScheduler(s.controller),
		handleBusyWorkflowReturn: true, // Handler successfully handles the task
	}

	executable := queues.NewExecutable(
		queues.DefaultReaderId,
		tasks.NewFakeTask(
			definition.NewWorkflowKey(
				tests.NamespaceID.String(),
				tests.WorkflowID,
				tests.RunID,
			),
			tasks.CategoryTransfer,
			s.timeSource.Now(),
		),
		s.mockExecutor,
		mockScheduler,
		s.mockRescheduler,
		queues.NewNoopPriorityAssigner(),
		s.timeSource,
		s.mockNamespaceRegistry,
		s.mockClusterMetadata,
		s.chasmRegistry,
		queues.GetTaskTypeTagValue,
		log.NewTestLogger(),
		s.metricsHandler,
		telemetry.NoopTracer,
		func(params *queues.ExecutableParams) {
			params.DLQEnabled = func() bool { return false }
		},
	)

	// Nack with ErrResourceExhaustedBusyWorkflow should trigger HandleBusyWorkflow
	executable.Nack(consts.ErrResourceExhaustedBusyWorkflow)

	// Verify HandleBusyWorkflow was called
	s.True(mockScheduler.handleBusyWorkflowCalled, "HandleBusyWorkflow should be called for busy workflow errors")
}

func (s *executableSuite) TestTaskNack_BusyWorkflow_FallbackToReschedule() {
	// Create a mock scheduler that implements BusyWorkflowHandler but returns false
	mockScheduler := &mockSchedulerWithBusyHandler{
		MockScheduler:            queues.NewMockScheduler(s.controller),
		handleBusyWorkflowReturn: false, // Handler does not handle the task
	}

	executable := queues.NewExecutable(
		queues.DefaultReaderId,
		tasks.NewFakeTask(
			definition.NewWorkflowKey(
				tests.NamespaceID.String(),
				tests.WorkflowID,
				tests.RunID,
			),
			tasks.CategoryTransfer,
			s.timeSource.Now(),
		),
		s.mockExecutor,
		mockScheduler,
		s.mockRescheduler,
		queues.NewNoopPriorityAssigner(),
		s.timeSource,
		s.mockNamespaceRegistry,
		s.mockClusterMetadata,
		s.chasmRegistry,
		queues.GetTaskTypeTagValue,
		log.NewTestLogger(),
		s.metricsHandler,
		telemetry.NoopTracer,
		func(params *queues.ExecutableParams) {
			params.DLQEnabled = func() bool { return false }
		},
	)

	// When HandleBusyWorkflow returns false, normal Nack logic kicks in.
	// For busy workflow errors, shouldResubmitOnNack returns true, so TrySubmit is called first.
	// If TrySubmit fails (returns false), then rescheduler.Add is called.
	mockScheduler.EXPECT().TrySubmit(executable).Return(false).Times(1)
	s.mockRescheduler.EXPECT().Add(executable, gomock.AssignableToTypeOf(time.Now())).Times(1)

	executable.Nack(consts.ErrResourceExhaustedBusyWorkflow)

	// Verify HandleBusyWorkflow was called first
	s.True(mockScheduler.handleBusyWorkflowCalled, "HandleBusyWorkflow should be called for busy workflow errors")
}

func (s *executableSuite) TestTaskNack_BusyWorkflow_NoHandlerFallsBackToReschedule() {
	// Test that when scheduler does not implement BusyWorkflowHandler,
	// busy workflow errors fall back to normal Nack path (TrySubmit then reschedule)
	executable := s.newTestExecutable()

	// When scheduler doesn't implement BusyWorkflowHandler, normal Nack flow is used.
	// For busy workflow errors, shouldResubmitOnNack returns true, so TrySubmit is called first.
	// If TrySubmit fails (returns false), then rescheduler.Add is called.
	s.mockScheduler.EXPECT().TrySubmit(executable).Return(false).Times(1)
	s.mockRescheduler.EXPECT().Add(executable, gomock.AssignableToTypeOf(time.Now())).Times(1)

	executable.Nack(consts.ErrResourceExhaustedBusyWorkflow)
}

func (s *executableSuite) accessInternalState(executable queues.Executable) {
	_ = fmt.Sprintf("%v", executable)
}