Extract common worker commands dispatch logic

service/history/nexus_dispatch_response.go → common/nexus/dispatch_response.go

@@ -1,4 +1,4 @@
-package history
+package nexus
 
 import (
 	"github.com/nexus-rpc/sdk-go/nexus"
@@ -7,30 +7,30 @@ import (
 	"go.temporal.io/server/api/matchingservice/v1"
 )
 
-// dispatchResponseToError converts a DispatchNexusTaskResponse proto into a Go error.
+// DispatchResponseToError converts a DispatchNexusTaskResponse proto into a Go error.
 // Returns nil if the response indicates success.
 //
 // For failure cases (worker explicitly returned an error), the Temporal SDK's failure
 // converter is used to produce standard Go errors (ApplicationError, CanceledError).
 // For transport-level issues (timeout, internal), a nexus.HandlerError is returned
 // so the caller can check Retryable().
-func dispatchResponseToError(resp *matchingservice.DispatchNexusTaskResponse) error {
+func DispatchResponseToError(resp *matchingservice.DispatchNexusTaskResponse) error {
 	switch t := resp.GetOutcome().(type) {
 	case *matchingservice.DispatchNexusTaskResponse_Failure:
 		// Worker received the task and explicitly failed it (via RespondNexusTaskFailed).
 		return temporal.GetDefaultFailureConverter().FailureToError(t.Failure)
 	case *matchingservice.DispatchNexusTaskResponse_RequestTimeout:
 		return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeUpstreamTimeout, "upstream timeout")
 	case *matchingservice.DispatchNexusTaskResponse_Response:
-		return startOperationResponseToError(t.Response.GetStartOperation())
+		return StartOperationResponseToError(t.Response.GetStartOperation())
 	default:
 		return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeInternal, "empty or unknown dispatch outcome")
 	}
 }
 
-// startOperationResponseToError converts a StartOperationResponse proto into a Go error.
+// StartOperationResponseToError converts a StartOperationResponse proto into a Go error.
 // Returns nil for success variants (SyncSuccess, AsyncSuccess).
-func startOperationResponseToError(resp *nexuspb.StartOperationResponse) error {
+func StartOperationResponseToError(resp *nexuspb.StartOperationResponse) error {
 	switch t := resp.GetVariant().(type) {
 	case *nexuspb.StartOperationResponse_SyncSuccess:
 		return nil

service/history/nexus_dispatch_response_test.go → common/nexus/dispatch_response_test.go

@@ -1,4 +1,4 @@
-package history
+package nexus
 
 import (
 	"testing"
@@ -25,7 +25,7 @@ func TestDispatchResponseToError_SyncSuccess(t *testing.T) {
 			},
 		},
 	}
-	err := dispatchResponseToError(resp)
+	err := DispatchResponseToError(resp)
 	require.NoError(t, err)
 }
 
@@ -45,7 +45,7 @@ func TestDispatchResponseToError_AsyncSuccess(t *testing.T) {
 			},
 		},
 	}
-	err := dispatchResponseToError(resp)
+	err := DispatchResponseToError(resp)
 	require.NoError(t, err)
 }
 
@@ -55,7 +55,7 @@ func TestDispatchResponseToError_RequestTimeout(t *testing.T) {
 			RequestTimeout: &matchingservice.DispatchNexusTaskResponse_Timeout{},
 		},
 	}
-	err := dispatchResponseToError(resp)
+	err := DispatchResponseToError(resp)
 	require.Error(t, err)
 
 	var handlerErr *nexus.HandlerError
@@ -76,7 +76,7 @@ func TestDispatchResponseToError_WorkerFailure(t *testing.T) {
 			},
 		},
 	}
-	err := dispatchResponseToError(resp)
+	err := DispatchResponseToError(resp)
 	require.Error(t, err)
 
 	var appErr *temporal.ApplicationError
@@ -105,7 +105,7 @@ func TestDispatchResponseToError_OperationFailure_ApplicationError(t *testing.T)
 			},
 		},
 	}
-	err := dispatchResponseToError(resp)
+	err := DispatchResponseToError(resp)
 	require.Error(t, err)
 
 	var appErr *temporal.ApplicationError
@@ -132,7 +132,7 @@ func TestDispatchResponseToError_OperationFailure_CanceledError(t *testing.T) {
 			},
 		},
 	}
-	err := dispatchResponseToError(resp)
+	err := DispatchResponseToError(resp)
 	require.Error(t, err)
 
 	var cancelErr *temporal.CanceledError
@@ -141,7 +141,7 @@ func TestDispatchResponseToError_OperationFailure_CanceledError(t *testing.T) {
 
 func TestDispatchResponseToError_EmptyOutcome(t *testing.T) {
 	resp := &matchingservice.DispatchNexusTaskResponse{}
-	err := dispatchResponseToError(resp)
+	err := DispatchResponseToError(resp)
 	require.Error(t, err)
 
 	var handlerErr *nexus.HandlerError
@@ -151,7 +151,7 @@ func TestDispatchResponseToError_EmptyOutcome(t *testing.T) {
 
 func TestStartOperationResponseToError_EmptyVariant(t *testing.T) {
 	resp := &nexuspb.StartOperationResponse{}
-	err := startOperationResponseToError(resp)
+	err := StartOperationResponseToError(resp)
 	require.Error(t, err)
 
 	var handlerErr *nexus.HandlerError

common/workercommands/dispatch.go

@@ -0,0 +1,139 @@
+package workercommands
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/nexus-rpc/sdk-go/nexus"
+	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	nexuspb "go.temporal.io/api/nexus/v1"
+	workerservicepb "go.temporal.io/api/nexusservices/workerservice/v1"
+	taskqueuepb "go.temporal.io/api/taskqueue/v1"
+	workerpb "go.temporal.io/api/worker/v1"
+	"go.temporal.io/server/api/matchingservice/v1"
+	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/log/tag"
+	"go.temporal.io/server/common/metrics"
+	commonnexus "go.temporal.io/server/common/nexus"
+	"go.temporal.io/server/common/resource"
+	"google.golang.org/protobuf/proto"
+)
+
+const (
+	// 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.
+	ServiceName   = "temporal.api.nexusservices.workerservice.v1.WorkerService"
+	OperationName = "ExecuteCommands"
+)
+
+// DispatchToWorker dispatches worker commands to a worker's control queue via Nexus.
+// It encodes the commands as binary/protobuf (which SDK Core can decode natively via prost),
+// sends them via DispatchNexusTask to matching, and handles the response.
+// Returns nil on success or permanent (non-retryable) errors. Returns an error for
+// retryable failures so the caller can retry.
+func DispatchToWorker(
+	ctx context.Context,
+	matchingClient resource.MatchingClient,
+	metricsHandler metrics.Handler,
+	logger log.Logger,
+	namespaceID string,
+	controlQueue string,
+	commands []*workerpb.WorkerCommand,
+) error {
+	request := &workerservicepb.ExecuteCommandsRequest{
+		Commands: commands,
+	}
+	// Encode as binary/protobuf using the standard Temporal payload format.
+	// Worker commands are handled directly by SDK Core (not by lang-SDK Nexus handlers),
+	// so we use binary/protobuf which Core can decode natively via prost. The standard
+	// payload.Encode() uses json/protobuf encoding, which Core does not support because
+	// it normally delegates Nexus payload deserialization to the lang SDK.
+	requestData, err := proto.Marshal(request)
+	if err != nil {
+		return fmt.Errorf("failed to encode worker commands request: %w", err)
+	}
+	requestPayload := &commonpb.Payload{
+		Metadata: map[string][]byte{
+			"encoding": []byte("binary/protobuf"),
+		},
+		Data: requestData,
+	}
+
+	nexusRequest := &nexuspb.Request{
+		Header: map[string]string{},
+		Variant: &nexuspb.Request_StartOperation{
+			StartOperation: &nexuspb.StartOperationRequest{
+				Service:   ServiceName,
+				Operation: OperationName,
+				Payload:   requestPayload,
+			},
+		},
+	}
+
+	resp, err := matchingClient.DispatchNexusTask(ctx, &matchingservice.DispatchNexusTaskRequest{
+		NamespaceId: namespaceID,
+		TaskQueue: &taskqueuepb.TaskQueue{
+			Name: controlQueue,
+			Kind: enumspb.TASK_QUEUE_KIND_WORKER_COMMANDS,
+		},
+		Request: nexusRequest,
+	})
+	if err != nil {
+		logger.Warn("Failed to dispatch worker commands",
+			tag.NewStringTag("control_queue", controlQueue),
+			tag.Error(err))
+		metrics.WorkerCommandsSent.With(metricsHandler).Record(1, metrics.OutcomeTag("rpc_error"))
+		return err
+	}
+
+	nexusErr := commonnexus.DispatchResponseToError(resp)
+	if nexusErr == nil {
+		metrics.WorkerCommandsSent.With(metricsHandler).Record(1, metrics.OutcomeTag("success"))
+		return nil
+	}
+
+	return HandleDispatchError(nexusErr, controlQueue, metricsHandler, logger)
+}
+
+// HandleDispatchError classifies a Nexus dispatch error and records the appropriate metric.
+// Returns nil for permanent errors (caller should not retry) and the original error for
+// retryable failures.
+func HandleDispatchError(nexusErr error, controlQueue string, metricsHandler metrics.Handler, logger log.Logger) error {
+	var handlerErr *nexus.HandlerError
+	if errors.As(nexusErr, &handlerErr) {
+		if handlerErr.Type == nexus.HandlerErrorTypeUpstreamTimeout {
+			logger.Warn("No worker polling control queue",
+				tag.NewStringTag("control_queue", controlQueue))
+			metrics.WorkerCommandsSent.With(metricsHandler).Record(1, metrics.OutcomeTag("no_poller"))
+			return nexusErr
+		}
+
+		if !handlerErr.Retryable() {
+			logger.Error("Worker commands non-retryable handler error",
+				tag.NewStringTag("control_queue", controlQueue),
+				tag.Error(nexusErr))
+			metrics.WorkerCommandsSent.With(metricsHandler).Record(1, metrics.OutcomeTag("non_retryable_error"))
+			return nil
+		}
+
+		logger.Warn("Worker commands transport failure",
+			tag.NewStringTag("control_queue", controlQueue),
+			tag.Error(nexusErr))
+		metrics.WorkerCommandsSent.With(metricsHandler).Record(1, metrics.OutcomeTag("transport_error"))
+		return nexusErr
+	}
+
+	// Worker-returned failure (ApplicationError, CanceledError, etc.). The worker received
+	// and processed the request but returned an error. Permanent — the worker contract
+	// requires success for all defined commands, so this indicates a bug or version
+	// incompatibility. Retrying won't help.
+	logger.Error("Worker returned failure for worker commands",
+		tag.NewStringTag("control_queue", controlQueue),
+		tag.Error(nexusErr))
+	metrics.WorkerCommandsSent.With(metricsHandler).Record(1, metrics.OutcomeTag("worker_error"))
+	return nil
+}