Refactor: Move worker commands dispatcher to common/workercommands

service/history/worker_commands_task_dispatcher.go → common/workercommands/dispatcher.go

@@ -1,4 +1,4 @@
-package history
+package workercommands
 
 import (
 	"context"
@@ -26,18 +26,18 @@ import (
 )
 
 const (
-	workerCommandsTaskTimeout    = time.Second * 10 * debug.TimeoutMultiplier
-	workerCommandsMaxTaskAttempt = 3
+	DispatchTimeout = time.Second * 10 * debug.TimeoutMultiplier
+	MaxTaskAttempts = 3
 
 	// Nexus service and operation names for worker commands.
 	// TODO: Replace with workerservicepb.WorkerService.ServiceName and
 	// workerservicepb.WorkerService.ExecuteCommands.Name() once the Nexus service
 	// descriptor is published in go.temporal.io/api.
-	workerCommandsServiceName   = "temporal.api.nexusservices.workerservice.v1.WorkerService"
-	workerCommandsOperationName = "ExecuteCommands"
+	ServiceName   = "temporal.api.nexusservices.workerservice.v1.WorkerService"
+	OperationName = "ExecuteCommands"
 )
 
-// workerCommandsTaskDispatcher dispatches worker commands to workers via Nexus.
+// Dispatcher dispatches worker commands to workers via Nexus.
 //
 // Failure scenarios:
 //   - No worker polling: matching returns RequestTimeout -> *nexus.HandlerError{Type: UpstreamTimeout}.
@@ -51,37 +51,37 @@ const (
 //     *temporal.CanceledError. Permanent — the worker contract requires success for all
 //     defined commands, so this indicates a bug or version incompatibility.
 //
-// Retryable errors are capped at workerCommandsMaxTaskAttempt attempts (in-memory). These
+// Retryable errors are capped at MaxTaskAttempts attempts (in-memory). These
 // commands are best-effort — the activity will eventually time out anyway — so excessive
 // retries waste resources. The counter resets on shard movement, which is acceptable.
-type workerCommandsTaskDispatcher struct {
+type Dispatcher struct {
 	matchingClient resource.MatchingClient
 	config         *configs.Config
 	metricsHandler metrics.Handler
 	logger         log.Logger
 }
 
-func newWorkerCommandsTaskDispatcher(
+func NewDispatcher(
 	matchingClient resource.MatchingClient,
 	config *configs.Config,
 	metricsHandler metrics.Handler,
 	logger log.Logger,
-) *workerCommandsTaskDispatcher {
-	return &workerCommandsTaskDispatcher{
+) *Dispatcher {
+	return &Dispatcher{
 		matchingClient: matchingClient,
 		config:         config,
 		metricsHandler: metricsHandler,
 		logger:         logger,
 	}
 }
 
-func (d *workerCommandsTaskDispatcher) execute(
+func (d *Dispatcher) Execute(
 	ctx context.Context,
 	task *tasks.WorkerCommandsTask,
 	attempt int,
 	namespaceName string,
 ) error {
-	if attempt > workerCommandsMaxTaskAttempt {
+	if attempt > MaxTaskAttempts {
 		d.logger.Info("Worker commands task exceeded max attempts, dropping",
 			tag.WorkflowID(task.WorkflowID),
 			tag.WorkflowRunID(task.RunID),
@@ -107,13 +107,13 @@ func (d *workerCommandsTaskDispatcher) execute(
 		return nil
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, workerCommandsTaskTimeout)
+	ctx, cancel := context.WithTimeout(ctx, DispatchTimeout)
 	defer cancel()
 
 	return d.dispatchToWorker(ctx, task, namespaceName)
 }
 
-func (d *workerCommandsTaskDispatcher) dispatchToWorker(
+func (d *Dispatcher) dispatchToWorker(
 	ctx context.Context,
 	task *tasks.WorkerCommandsTask,
 	namespaceName string,
@@ -141,8 +141,8 @@ func (d *workerCommandsTaskDispatcher) dispatchToWorker(
 		Header: map[string]string{},
 		Variant: &nexuspb.Request_StartOperation{
 			StartOperation: &nexuspb.StartOperationRequest{
-				Service:   workerCommandsServiceName,
-				Operation: workerCommandsOperationName,
+				Service:   ServiceName,
+				Operation: OperationName,
 				Payload:   requestPayload,
 			},
 		},
@@ -170,7 +170,7 @@ func (d *workerCommandsTaskDispatcher) dispatchToWorker(
 	return d.handleError(nexusErr, task, namespaceName)
 }
 
-func (d *workerCommandsTaskDispatcher) handleError(nexusErr error, task *tasks.WorkerCommandsTask, namespaceName string) error {
+func (d *Dispatcher) handleError(nexusErr error, task *tasks.WorkerCommandsTask, namespaceName string) error {
 	var handlerErr *nexus.HandlerError
 	if errors.As(nexusErr, &handlerErr) {
 		// Handler-level error (transport, timeout, internal). These are constructed by
@@ -211,18 +211,18 @@ func (d *workerCommandsTaskDispatcher) handleError(nexusErr error, task *tasks.W
 	return nil
 }
 
-func (d *workerCommandsTaskDispatcher) recordCommandMetrics(commands []*workerpb.WorkerCommand, namespaceName string, outcome string) {
+func (d *Dispatcher) recordCommandMetrics(commands []*workerpb.WorkerCommand, namespaceName string, outcome string) {
 	for _, cmd := range commands {
 		metrics.WorkerCommandsSent.With(d.metricsHandler).Record(
 			1,
 			metrics.NamespaceTag(namespaceName),
 			metrics.OutcomeTag(outcome),
-			metrics.StringTag("command_type", workerCommandTypeName(cmd)),
+			metrics.StringTag("command_type", CommandTypeName(cmd)),
 		)
 	}
 }
 
-func workerCommandTypeName(cmd *workerpb.WorkerCommand) string {
+func CommandTypeName(cmd *workerpb.WorkerCommand) string {
 	switch cmd.GetType().(type) {
 	case *workerpb.WorkerCommand_CancelActivity:
 		return "cancel_activity"

service/history/worker_commands_task_dispatcher_test.go → common/workercommands/dispatcher_test.go

@@ -1,4 +1,4 @@
-package history
+package workercommands
 
 import (
 	"context"
@@ -44,20 +44,20 @@ func requireMetricValue(t *testing.T, snap map[string][]*metricstest.CapturedRec
 }
 
 func TestExecute_FeatureFlagOff_DropsTask(t *testing.T) {
-	d := &workerCommandsTaskDispatcher{
+	d := &Dispatcher{
 		config: &configs.Config{
 			EnableCancelActivityWorkerCommand: func(string) bool { return false },
 		},
 		logger: log.NewNoopLogger(),
 	}
 
 	task := testWorkerCommandsTask()
-	err := d.execute(context.Background(), task, 1 /* attempt */, "test-namespace")
+	err := d.Execute(context.Background(), task, 1 /* attempt */, "test-namespace")
 	require.NoError(t, err, "task should be silently dropped when feature flag is off")
 }
 
 func TestExecute_EmptyCommands_DropsTask(t *testing.T) {
-	d := &workerCommandsTaskDispatcher{
+	d := &Dispatcher{
 		config: &configs.Config{
 			EnableCancelActivityWorkerCommand: func(string) bool { return true },
 		},
@@ -66,7 +66,7 @@ func TestExecute_EmptyCommands_DropsTask(t *testing.T) {
 
 	task := testWorkerCommandsTask()
 	task.Commands = nil
-	err := d.execute(context.Background(), task, 1 /* attempt */, "test-namespace")
+	err := d.Execute(context.Background(), task, 1 /* attempt */, "test-namespace")
 	require.NoError(t, err, "task with no commands should be dropped")
 }
 
@@ -75,7 +75,7 @@ func TestExecute_ExceedsMaxAttempts_DropsTask(t *testing.T) {
 	capture := metricsHandler.StartCapture()
 	defer metricsHandler.StopCapture(capture)
 
-	d := &workerCommandsTaskDispatcher{
+	d := &Dispatcher{
 		config: &configs.Config{
 			EnableCancelActivityWorkerCommand: func(string) bool { return true },
 		},
@@ -84,7 +84,7 @@ func TestExecute_ExceedsMaxAttempts_DropsTask(t *testing.T) {
 	}
 
 	task := testWorkerCommandsTask()
-	err := d.execute(context.Background(), task, workerCommandsMaxTaskAttempt+1, "test-namespace")
+	err := d.Execute(context.Background(), task, MaxTaskAttempts+1, "test-namespace")
 	require.NoError(t, err, "task should be dropped when max attempts exceeded")
 
 	requireMetricValue(t, capture.Snapshot(), "max_attempts_exceeded")
@@ -97,7 +97,7 @@ func TestExecute_AtMaxAttempt_StillExecutes(t *testing.T) {
 	capture := metricsHandler.StartCapture()
 	defer metricsHandler.StopCapture(capture)
 
-	d := &workerCommandsTaskDispatcher{
+	d := &Dispatcher{
 		matchingClient: mockClient,
 		config: &configs.Config{
 			EnableCancelActivityWorkerCommand: func(string) bool { return true },
@@ -122,7 +122,7 @@ func TestExecute_AtMaxAttempt_StillExecutes(t *testing.T) {
 		}, nil)
 
 	task := testWorkerCommandsTask()
-	err := d.execute(context.Background(), task, workerCommandsMaxTaskAttempt, "test-namespace")
+	err := d.Execute(context.Background(), task, MaxTaskAttempts, "test-namespace")
 	require.NoError(t, err, "task at exactly max attempt should still execute")
 
 	requireMetricValue(t, capture.Snapshot(), "success")
@@ -135,7 +135,7 @@ func TestExecute_DispatchSuccess(t *testing.T) {
 	capture := metricsHandler.StartCapture()
 	defer metricsHandler.StopCapture(capture)
 
-	d := &workerCommandsTaskDispatcher{
+	d := &Dispatcher{
 		matchingClient: mockClient,
 		config: &configs.Config{
 			EnableCancelActivityWorkerCommand: func(string) bool { return true },
@@ -164,7 +164,7 @@ func TestExecute_DispatchSuccess(t *testing.T) {
 		})
 
 	task := testWorkerCommandsTask()
-	err := d.execute(context.Background(), task, 1 /* attempt */, "test-namespace")
+	err := d.Execute(context.Background(), task, 1 /* attempt */, "test-namespace")
 	require.NoError(t, err)
 
 	require.NotNil(t, capturedReq)
@@ -182,7 +182,7 @@ func TestExecute_DispatchRPCError(t *testing.T) {
 	capture := metricsHandler.StartCapture()
 	defer metricsHandler.StopCapture(capture)
 
-	d := &workerCommandsTaskDispatcher{
+	d := &Dispatcher{
 		matchingClient: mockClient,
 		config: &configs.Config{
 			EnableCancelActivityWorkerCommand: func(string) bool { return true },
@@ -195,7 +195,7 @@ func TestExecute_DispatchRPCError(t *testing.T) {
 		nil, errors.New("connection refused"))
 
 	task := testWorkerCommandsTask()
-	err := d.execute(context.Background(), task, 1 /* attempt */, "test-namespace")
+	err := d.Execute(context.Background(), task, 1 /* attempt */, "test-namespace")
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "connection refused")
 
@@ -209,7 +209,7 @@ func TestExecute_UpstreamTimeout(t *testing.T) {
 	capture := metricsHandler.StartCapture()
 	defer metricsHandler.StopCapture(capture)
 
-	d := &workerCommandsTaskDispatcher{
+	d := &Dispatcher{
 		matchingClient: mockClient,
 		config: &configs.Config{
 			EnableCancelActivityWorkerCommand: func(string) bool { return true },
@@ -226,7 +226,7 @@ func TestExecute_UpstreamTimeout(t *testing.T) {
 		}, nil)
 
 	task := testWorkerCommandsTask()
-	err := d.execute(context.Background(), task, 1 /* attempt */, "test-namespace")
+	err := d.Execute(context.Background(), task, 1 /* attempt */, "test-namespace")
 	require.Error(t, err)
 
 	var he *nexus.HandlerError
@@ -241,7 +241,7 @@ func TestHandleError_WorkerError_ReturnNil(t *testing.T) {
 	capture := metricsHandler.StartCapture()
 	defer metricsHandler.StopCapture(capture)
 
-	d := &workerCommandsTaskDispatcher{
+	d := &Dispatcher{
 		metricsHandler: metricsHandler,
 		logger:         log.NewNoopLogger(),
 	}
@@ -260,7 +260,7 @@ func TestHandleError_UpstreamTimeout_ReturnRetryable(t *testing.T) {
 	capture := metricsHandler.StartCapture()
 	defer metricsHandler.StopCapture(capture)
 
-	d := &workerCommandsTaskDispatcher{
+	d := &Dispatcher{
 		metricsHandler: metricsHandler,
 		logger:         log.NewNoopLogger(),
 	}
@@ -282,7 +282,7 @@ func TestHandleError_NonRetryableHandlerError_ReturnNil(t *testing.T) {
 	capture := metricsHandler.StartCapture()
 	defer metricsHandler.StopCapture(capture)
 
-	d := &workerCommandsTaskDispatcher{
+	d := &Dispatcher{
 		metricsHandler: metricsHandler,
 		logger:         log.NewNoopLogger(),
 	}
@@ -300,7 +300,7 @@ func TestHandleError_OtherHandlerError_ReturnRetryable(t *testing.T) {
 	capture := metricsHandler.StartCapture()
 	defer metricsHandler.StopCapture(capture)
 
-	d := &workerCommandsTaskDispatcher{
+	d := &Dispatcher{
 		metricsHandler: metricsHandler,
 		logger:         log.NewNoopLogger(),
 	}

service/history/outbound_queue_active_task_executor.go

@@ -11,6 +11,7 @@ import (
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/resource"
+	"go.temporal.io/server/common/workercommands"
 	"go.temporal.io/server/service/history/consts"
 	historyi "go.temporal.io/server/service/history/interfaces"
 	"go.temporal.io/server/service/history/queues"
@@ -25,8 +26,8 @@ const (
 
 type outboundQueueActiveTaskExecutor struct {
 	stateMachineEnvironment
-	chasmEngine                  chasm.Engine
-	workerCommandsTaskDispatcher *workerCommandsTaskDispatcher
+	chasmEngine              chasm.Engine
+	workerCommandsDispatcher *workercommands.Dispatcher
 }
 
 var _ queues.Executor = &outboundQueueActiveTaskExecutor{}
@@ -50,7 +51,7 @@ func newOutboundQueueActiveTaskExecutor(
 			metricsHandler: scopedMetricsHandler,
 		},
 		chasmEngine: chasmEngine,
-		workerCommandsTaskDispatcher: newWorkerCommandsTaskDispatcher(
+		workerCommandsDispatcher: workercommands.NewDispatcher(
 			matchingClient,
 			shardCtx.GetConfig(),
 			scopedMetricsHandler,
@@ -104,7 +105,7 @@ func (e *outboundQueueActiveTaskExecutor) Execute(
 	case *tasks.ChasmTask:
 		return respond(e.executeChasmSideEffectTask(ctx, task))
 	case *tasks.WorkerCommandsTask:
-		return respond(e.workerCommandsTaskDispatcher.execute(ctx, task, executable.Attempt(), namespaceTag.Value))
+		return respond(e.workerCommandsDispatcher.Execute(ctx, task, executable.Attempt(), namespaceTag.Value))
 	}
 
 	return respond(queueserrors.NewUnprocessableTaskError(fmt.Sprintf("unknown task type '%T'", task)))