tasks/service/history/worker_commands_task_dispatcher.go

package history

import (
	"context"
	"errors"
	"fmt"
	"time"

	"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"
	"go.temporal.io/server/api/matchingservice/v1"
	"go.temporal.io/server/common/debug"
	"go.temporal.io/server/common/log"
	"go.temporal.io/server/common/log/tag"
	"go.temporal.io/server/common/metrics"
	"go.temporal.io/server/common/resource"
	"go.temporal.io/server/service/history/configs"
	"go.temporal.io/server/service/history/tasks"
	"google.golang.org/protobuf/proto"
)

const (
	workerCommandsTaskTimeout    = time.Second * 10 * debug.TimeoutMultiplier
	workerCommandsMaxTaskAttempt = 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"
)

// workerCommandsTaskDispatcher dispatches worker commands to workers via Nexus.
//
// Failure scenarios:
//   - No worker polling: matching returns RequestTimeout -> *nexus.HandlerError{Type: UpstreamTimeout}.
//     Retryable -- worker may come up later.
//   - Worker crashes after receiving the task: matching blocks waiting for a response until
//     context deadline, then returns RequestTimeout. Indistinguishable from "no worker polling".
//     Safe to retry because commands are idempotent (e.g., cancelling a missing activity is a
//     no-op success per the worker contract).
//   - Transport/RPC failure: *nexus.HandlerError. Retryable.
//   - Worker failure (worker explicitly returns error): *temporal.ApplicationError or
//     *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
// 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 {
	matchingClient resource.MatchingClient
	config         *configs.Config
	metricsHandler metrics.Handler
	logger         log.Logger
}

func newWorkerCommandsTaskDispatcher(
	matchingClient resource.MatchingClient,
	config *configs.Config,
	metricsHandler metrics.Handler,
	logger log.Logger,
) *workerCommandsTaskDispatcher {
	return &workerCommandsTaskDispatcher{
		matchingClient: matchingClient,
		config:         config,
		metricsHandler: metricsHandler,
		logger:         logger,
	}
}

func (d *workerCommandsTaskDispatcher) execute(
	ctx context.Context,
	task *tasks.WorkerCommandsTask,
	attempt int,
) error {
	if attempt > workerCommandsMaxTaskAttempt {
		d.logger.Info("Worker commands task exceeded max attempts, dropping",
			tag.WorkflowID(task.WorkflowID),
			tag.WorkflowRunID(task.RunID),
			tag.NewStringTag("control_queue", task.Destination),
			tag.Attempt(int32(attempt)),
		)
		metrics.WorkerCommandsSent.With(d.metricsHandler).Record(1, metrics.OutcomeTag("max_attempts_exceeded"))
		return nil
	}

	if !d.config.EnableCancelActivityWorkerCommand() {
		d.logger.Info("Worker commands feature disabled, dropping task",
			tag.WorkflowID(task.WorkflowID),
			tag.WorkflowRunID(task.RunID),
			tag.NewStringTag("control_queue", task.Destination),
			tag.NewInt("command_count", len(task.Commands)),
		)
		return nil
	}

	if len(task.Commands) == 0 {
		return nil
	}

	ctx, cancel := context.WithTimeout(ctx, workerCommandsTaskTimeout)
	defer cancel()

	return d.dispatchToWorker(ctx, task)
}

func (d *workerCommandsTaskDispatcher) dispatchToWorker(
	ctx context.Context,
	task *tasks.WorkerCommandsTask,
) error {
	request := &workerservicepb.ExecuteCommandsRequest{
		Commands: task.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:   workerCommandsServiceName,
				Operation: workerCommandsOperationName,
				Payload:   requestPayload,
			},
		},
	}

	resp, err := d.matchingClient.DispatchNexusTask(ctx, &matchingservice.DispatchNexusTaskRequest{
		NamespaceId: task.NamespaceID,
		TaskQueue: &taskqueuepb.TaskQueue{
			Name: task.Destination,
			Kind: enumspb.TASK_QUEUE_KIND_WORKER_COMMANDS,
		},
		Request: nexusRequest,
	})
	if err != nil {
		metrics.WorkerCommandsSent.With(d.metricsHandler).Record(1, metrics.OutcomeTag("rpc_error"))
		return fmt.Errorf("failed to dispatch worker commands to control queue %s: %w", task.Destination, err)
	}

	nexusErr := dispatchResponseToError(resp)
	if nexusErr == nil {
		metrics.WorkerCommandsSent.With(d.metricsHandler).Record(1, metrics.OutcomeTag("success"))
		return nil
	}

	return d.handleError(nexusErr, task)
}

func (d *workerCommandsTaskDispatcher) handleError(nexusErr error, task *tasks.WorkerCommandsTask) error {
	var handlerErr *nexus.HandlerError
	if errors.As(nexusErr, &handlerErr) {
		// Handler-level error (transport, timeout, internal). These are constructed by
		// dispatchResponseToError for non-worker-returned failures.
		if handlerErr.Type == nexus.HandlerErrorTypeUpstreamTimeout {
			d.logger.Warn("No worker polling control queue",
				tag.NewStringTag("control_queue", task.Destination))
			metrics.WorkerCommandsSent.With(d.metricsHandler).Record(1, metrics.OutcomeTag("no_poller"))
			return nexusErr
		}

		if !handlerErr.Retryable() {
			d.logger.Error("Worker commands non-retryable handler error",
				tag.NewStringTag("control_queue", task.Destination),
				tag.Error(nexusErr))
			metrics.WorkerCommandsSent.With(d.metricsHandler).Record(1, metrics.OutcomeTag("non_retryable_error"))
			return nil
		}

		d.logger.Warn("Worker commands transport failure",
			tag.NewStringTag("control_queue", task.Destination),
			tag.Error(nexusErr))
		metrics.WorkerCommandsSent.With(d.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.
	d.logger.Error("Worker returned failure for worker commands",
		tag.WorkflowID(task.WorkflowID),
		tag.WorkflowRunID(task.RunID),
		tag.NewStringTag("control_queue", task.Destination),
		tag.NewInt("command_count", len(task.Commands)),
		tag.Error(nexusErr))
	metrics.WorkerCommandsSent.With(d.metricsHandler).Record(1, metrics.OutcomeTag("worker_error"))
	return nil
}