diff --git a/service/history/worker_commands_task_dispatcher.go b/common/workercommands/dispatcher.go similarity index 84% rename from service/history/worker_commands_task_dispatcher.go rename to common/workercommands/dispatcher.go index baaf9f25640..79c3f2c714b 100644 --- a/service/history/worker_commands_task_dispatcher.go +++ b/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,23 +51,23 @@ 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, @@ -75,13 +75,13 @@ func newWorkerCommandsTaskDispatcher( } } -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" diff --git a/service/history/worker_commands_task_dispatcher_test.go b/common/workercommands/dispatcher_test.go similarity index 91% rename from service/history/worker_commands_task_dispatcher_test.go rename to common/workercommands/dispatcher_test.go index 16e7c846fc5..6da6346269c 100644 --- a/service/history/worker_commands_task_dispatcher_test.go +++ b/common/workercommands/dispatcher_test.go @@ -1,4 +1,4 @@ -package history +package workercommands import ( "context" @@ -44,7 +44,7 @@ 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 }, }, @@ -52,12 +52,12 @@ func TestExecute_FeatureFlagOff_DropsTask(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, "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(), } diff --git a/service/history/outbound_queue_active_task_executor.go b/service/history/outbound_queue_active_task_executor.go index a5cfe6844df..12f523d51f4 100644 --- a/service/history/outbound_queue_active_task_executor.go +++ b/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)))