Dispatch activity cancellation to worker using Nexus

common/metrics/metric_defs.go

@@ -863,6 +863,7 @@ var (
 			" latency of the final attempt to send the task to the DLQ, not the cumulative latency of all attempts."),
 	)
 	TaskDiscarded                   = NewCounterDef("task_errors_discarded")
+	WorkerCommandsSent              = NewCounterDef("worker_commands_sent")
 	TaskSkipped                     = NewCounterDef("task_skipped")
 	TaskVersionMisMatch             = NewCounterDef("task_errors_version_mismatch")
 	TasksDependencyTaskNotCompleted = NewCounterDef("task_dependency_task_not_completed")

proto/internal/temporal/server/api/persistence/v1/executions.proto

@@ -668,6 +668,21 @@ message ActivityInfo {
   // A dedicated per-worker Nexus task queue on which the server sends control
   // tasks (e.g. activity cancellation) to this specific worker instance.
   string worker_control_task_queue = 51;
+
+  // The shard clock at the time this activity was started (RecordActivityTaskStarted).
+  // Matching uses this clock to build the task token sent to the worker. Stored here so
+  // that history can later reconstruct the same task token (e.g. for cancel worker commands).
+  //
+  // IMPORTANT: The clock approach requires history to reconstruct the token using
+  // the same fields and logic as matching — if NewActivityTaskToken changes, both
+  // call sites must stay in sync or the tokens will silently diverge. An alternative
+  // is to store the full serialized task token (~150-300 bytes), which avoids
+  // reconstruction entirely and is immune to token format changes. We chose the
+  // clock approach to keep the per-activity memory footprint minimal (~24 bytes).
+  //
+  // Replication: This field is part of ActivityInfo and is automatically replicated
+  // via state-based replication. No special handling is needed.
+  temporal.server.api.clock.v1.VectorClock started_clock = 52;
 }
 
 // timer_map column

service/history/api/recordactivitytaskstarted/api.go

@@ -152,6 +152,7 @@ func recordActivityTaskStarted(
 		if ai.RequestId == requestID {
 			response.StartedTime = ai.StartedTime
 			response.Attempt = ai.Attempt
+			response.Clock = ai.StartedClock
 			return response, rejectCodeAccepted, nil
 		}
 
@@ -238,6 +239,17 @@ func recordActivityTaskStarted(
 		}
 	}
 
+	// Create the shard clock before recording the start event. Matching uses the returned clock
+	// to build the task token sent to the worker. We store it in ActivityInfo so that history
+	// can later reconstruct the same task token that matching created (e.g. for cancel worker
+	// commands). On retries of this RPC, we return the stored clock from the early return
+	// path above.
+	clock, err := shardContext.NewVectorClock()
+	if err != nil {
+		return nil, rejectCodeUndefined, err
+	}
+	ai.StartedClock = clock
+
 	versioningStamp := worker_versioning.StampFromCapabilities(request.PollRequest.WorkerVersionCapabilities, request.PollRequest.DeploymentOptions) //nolint:staticcheck // SA1019: WorkerVersionCapabilities is deprecated but still used for old versioning [cleanup-old-wv]
 	if _, err := mutableState.AddActivityTaskStartedEvent(
 		ai, scheduledEventID, requestID, request.PollRequest.GetIdentity(),
@@ -261,6 +273,8 @@ func recordActivityTaskStarted(
 		),
 	).Record(scheduleToStartLatency)
 
+	response.Clock = clock
+
 	response.StartedTime = ai.StartedTime
 	response.Attempt = ai.Attempt
 	response.HeartbeatDetails = ai.LastHeartbeatDetails

service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go

@@ -682,7 +682,6 @@ func (handler *workflowTaskCompletedHandler) handleCommandRequestCancelActivity(
 	if ai != nil {
 		// If ai is nil, the activity has already been canceled/completed/timedout. The cancel request
 		// will be recorded in the history, but no further action will be taken.
-
 		if ai.StartedEventId == common.EmptyEventID {
 			// We haven't started the activity yet, we can cancel the activity right away and
 			// schedule a workflow task to ensure the workflow makes progress.
@@ -697,37 +696,48 @@ func (handler *workflowTaskCompletedHandler) handleCommandRequestCancelActivity(
 				return nil, err
 			}
 			handler.activityNotStartedCancelled = true
-		} else if ai.StartedEventId != common.EmptyEventID && ai.WorkerControlTaskQueue != "" {
-			// Activity has started and worker supports Nexus control tasks - collect for batched dispatch.
-			taskToken, err := handler.tokenSerializer.Serialize(tasktoken.NewActivityTaskToken(
-				handler.mutableState.GetNamespaceEntry().ID().String(),
-				handler.mutableState.GetWorkflowKey().WorkflowID,
-				handler.mutableState.GetWorkflowKey().RunID,
-				ai.ScheduledEventId,
-				ai.ActivityId,
-				ai.ActivityType.GetName(),
-				ai.Attempt,
-				nil, // Clock not needed for cancel
-				ai.Version,
-				ai.StartVersion,
-				nil,
-			))
-			if err != nil {
-				return nil, err
-			}
-			if handler.pendingWorkerCommandsByControlQueue == nil {
-				handler.pendingWorkerCommandsByControlQueue = make(map[string][]*workerpb.WorkerCommand)
-			}
-			handler.pendingWorkerCommandsByControlQueue[ai.WorkerControlTaskQueue] = append(
-				handler.pendingWorkerCommandsByControlQueue[ai.WorkerControlTaskQueue],
-				&workerpb.WorkerCommand{
-					Type: &workerpb.WorkerCommand_CancelActivity{
-						CancelActivity: &workerpb.CancelActivityCommand{
-							TaskToken: taskToken,
+		} else if ai.WorkerControlTaskQueue != "" {
+			if ai.StartedClock == nil {
+				// StartedClock may be nil for activities started before this feature was deployed.
+				// Skip cancel command; the activity will time out normally.
+				handler.logger.Info("Skipping worker cancel command: activity missing StartedClock (pre-deploy)",
+					tag.WorkflowNamespaceID(handler.mutableState.GetWorkflowKey().NamespaceID),
+					tag.WorkflowID(handler.mutableState.GetWorkflowKey().WorkflowID),
+					tag.WorkflowRunID(handler.mutableState.GetWorkflowKey().RunID),
+					tag.WorkflowScheduledEventID(ai.ScheduledEventId),
+				)
+			} else {
+				// Activity has started and worker supports Nexus control tasks - collect for batched dispatch.
+				taskToken, err := handler.tokenSerializer.Serialize(tasktoken.NewActivityTaskToken(
+					handler.mutableState.GetNamespaceEntry().ID().String(),
+					handler.mutableState.GetWorkflowKey().WorkflowID,
+					handler.mutableState.GetWorkflowKey().RunID,
+					ai.ScheduledEventId,
+					ai.ActivityId,
+					ai.ActivityType.GetName(),
+					ai.Attempt,
+					ai.StartedClock,
+					ai.Version,
+					ai.StartVersion,
+					nil,
+				))
+				if err != nil {
+					return nil, err
+				}
+				if handler.pendingWorkerCommandsByControlQueue == nil {
+					handler.pendingWorkerCommandsByControlQueue = make(map[string][]*workerpb.WorkerCommand)
+				}
+				handler.pendingWorkerCommandsByControlQueue[ai.WorkerControlTaskQueue] = append(
+					handler.pendingWorkerCommandsByControlQueue[ai.WorkerControlTaskQueue],
+					&workerpb.WorkerCommand{
+						Type: &workerpb.WorkerCommand_CancelActivity{
+							CancelActivity: &workerpb.CancelActivityCommand{
+								TaskToken: taskToken,
+							},
 						},
 					},
-				},
-			)
+				)
+			}
 		}
 	}
 	return actCancelReqEvent, nil

service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler_test.go

@@ -18,6 +18,7 @@ import (
 	"go.temporal.io/api/serviceerror"
 	updatepb "go.temporal.io/api/update/v1"
 	workerpb "go.temporal.io/api/worker/v1"
+	clockspb "go.temporal.io/server/api/clock/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/chasm"
 	chasmworkflow "go.temporal.io/server/chasm/lib/workflow"
@@ -30,6 +31,7 @@ import (
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/namespace/nsregistry"
 	"go.temporal.io/server/common/persistence"
+	"go.temporal.io/server/common/tasktoken"
 	"go.temporal.io/server/service/history/api"
 	"go.temporal.io/server/service/history/configs"
 	historyi "go.temporal.io/server/service/history/interfaces"
@@ -537,3 +539,115 @@ func TestFlushWorkerCommandsTasks(t *testing.T) {
 		require.NoError(t, err)
 	})
 }
+
+func TestHandleCommandRequestCancelActivity_WorkerCommands(t *testing.T) {
+	t.Parallel()
+
+	cancelReqEvent := &historypb.HistoryEvent{EventId: 10}
+	scheduledEventID := int64(5)
+	controlQueue := "/_sys/worker-commands/test-ns/key1"
+
+	t.Run("collects commands for activities with clock, skips those without", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		ms := historyi.NewMockMutableState(ctrl)
+
+		// Activity 1: started with clock (post-deploy) — should produce a cancel command.
+		ai1 := &persistencespb.ActivityInfo{
+			ScheduledEventId:       int64(5),
+			StartedEventId:         7,
+			ActivityId:             "act-1",
+			WorkerControlTaskQueue: controlQueue,
+			StartedClock:           &clockspb.VectorClock{ClusterId: 1, ShardId: 1, Clock: 100},
+			Version:                1,
+			StartVersion:           1,
+			Attempt:                1,
+			ActivityType:           &commonpb.ActivityType{Name: "test-activity"},
+		}
+		// Activity 2: started without clock (pre-deploy) — should be skipped.
+		ai2 := &persistencespb.ActivityInfo{
+			ScheduledEventId:       int64(6),
+			StartedEventId:         8,
+			ActivityId:             "act-2",
+			WorkerControlTaskQueue: controlQueue,
+			StartedClock:           nil,
+			Version:                1,
+			StartVersion:           1,
+			Attempt:                1,
+			ActivityType:           &commonpb.ActivityType{Name: "test-activity"},
+		}
+		// Activity 3: started with clock — should also produce a cancel command.
+		ai3 := &persistencespb.ActivityInfo{
+			ScheduledEventId:       int64(7),
+			StartedEventId:         9,
+			ActivityId:             "act-3",
+			WorkerControlTaskQueue: controlQueue,
+			StartedClock:           &clockspb.VectorClock{ClusterId: 1, ShardId: 1, Clock: 200},
+			Version:                1,
+			StartVersion:           1,
+			Attempt:                1,
+			ActivityType:           &commonpb.ActivityType{Name: "test-activity"},
+		}
+
+		ms.EXPECT().AddActivityTaskCancelRequestedEvent(int64(123), int64(5), "test-identity").
+			Return(&historypb.HistoryEvent{EventId: 10}, ai1, nil)
+		ms.EXPECT().AddActivityTaskCancelRequestedEvent(int64(123), int64(6), "test-identity").
+			Return(&historypb.HistoryEvent{EventId: 11}, ai2, nil)
+		ms.EXPECT().AddActivityTaskCancelRequestedEvent(int64(123), int64(7), "test-identity").
+			Return(&historypb.HistoryEvent{EventId: 12}, ai3, nil)
+		ms.EXPECT().GetNamespaceEntry().Return(tests.LocalNamespaceEntry).AnyTimes()
+		ms.EXPECT().GetWorkflowKey().Return(tests.WorkflowKey).AnyTimes()
+
+		handler := &workflowTaskCompletedHandler{
+			identity:                "test-identity",
+			workflowTaskCompletedID: 123,
+			mutableState:            ms,
+			tokenSerializer:         tasktoken.NewSerializer(),
+			logger:                  log.NewNoopLogger(),
+		}
+
+		for _, id := range []int64{5, 6, 7} {
+			_, err := handler.handleCommandRequestCancelActivity(
+				context.Background(),
+				&commandpb.RequestCancelActivityTaskCommandAttributes{ScheduledEventId: id},
+			)
+			require.NoError(t, err)
+		}
+
+		// Activities 1 and 3 (with clock) should produce commands; activity 2 (nil clock) should be skipped.
+		require.Len(t, handler.pendingWorkerCommandsByControlQueue[controlQueue], 2,
+			"only activities with StartedClock should produce cancel commands")
+	})
+
+	t.Run("started activity without control queue does not collect worker command", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		ms := historyi.NewMockMutableState(ctrl)
+
+		ai := &persistencespb.ActivityInfo{
+			ScheduledEventId:       scheduledEventID,
+			StartedEventId:         7,
+			ActivityId:             "act-1",
+			WorkerControlTaskQueue: "", // worker doesn't support control tasks
+			StartedClock:           &clockspb.VectorClock{ClusterId: 1, ShardId: 1, Clock: 100},
+		}
+
+		ms.EXPECT().AddActivityTaskCancelRequestedEvent(int64(123), scheduledEventID, "test-identity").
+			Return(cancelReqEvent, ai, nil)
+
+		handler := &workflowTaskCompletedHandler{
+			identity:                "test-identity",
+			workflowTaskCompletedID: 123,
+			mutableState:            ms,
+			logger:                  log.NewNoopLogger(),
+		}
+
+		event, err := handler.handleCommandRequestCancelActivity(
+			context.Background(),
+			&commandpb.RequestCancelActivityTaskCommandAttributes{
+				ScheduledEventId: scheduledEventID,
+			},
+		)
+		require.NoError(t, err)
+		require.Equal(t, cancelReqEvent, event)
+		require.Empty(t, handler.pendingWorkerCommandsByControlQueue)
+	})
+}

service/history/handler.go

@@ -459,10 +459,6 @@ func (h *Handler) RecordActivityTaskStarted(ctx context.Context, request *histor
 	if err != nil {
 		return nil, h.convertError(err)
 	}
-	response.Clock, err = shardContext.NewVectorClock()
-	if err != nil {
-		return nil, h.convertError(err)
-	}
 	return response, nil
 }
 

service/history/nexus_dispatch_response.go

@@ -0,0 +1,84 @@
+package history
+
+import (
+	"fmt"
+
+	"github.com/nexus-rpc/sdk-go/nexus"
+	failurepb "go.temporal.io/api/failure/v1"
+	nexuspb "go.temporal.io/api/nexus/v1"
+	"go.temporal.io/server/api/matchingservice/v1"
+	commonnexus "go.temporal.io/server/common/nexus"
+	"go.temporal.io/server/common/nexus/nexusrpc"
+)
+
+// convertTemporalFailure converts a Temporal API Failure proto into a Go error
+// via the Nexus SDK failure converter.
+func convertTemporalFailure(failure *failurepb.Failure) (nexusErr error, err error) {
+	nexusFailure, err := commonnexus.TemporalFailureToNexusFailure(failure)
+	if err != nil {
+		return nil, fmt.Errorf("failed to convert Temporal failure: %w", err)
+	}
+	nexusErr, err = nexusrpc.DefaultFailureConverter().FailureToError(nexusFailure)
+	if err != nil {
+		return nil, fmt.Errorf("failed to convert Nexus failure to error: %w", err)
+	}
+	return nexusErr, nil
+}
+
+// dispatchResponseToError converts a DispatchNexusTaskResponse proto into a Nexus SDK error.
+// Returns nil if the response indicates success.
+func dispatchResponseToError(resp *matchingservice.DispatchNexusTaskResponse) error {
+	switch t := resp.GetOutcome().(type) {
+	// Worker received the task and explicitly failed it (via RespondNexusTaskFailed).
+	case *matchingservice.DispatchNexusTaskResponse_Failure:
+		converted, err := convertTemporalFailure(t.Failure)
+		if err != nil {
+			return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeInternal, "%v", err)
+		}
+		return converted
+	case *matchingservice.DispatchNexusTaskResponse_RequestTimeout:
+		return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeUpstreamTimeout, "upstream timeout")
+	// Worker responded successfully; check the inner StartOperation response.
+	case *matchingservice.DispatchNexusTaskResponse_Response:
+		return startOperationResponseToError(t.Response.GetStartOperation())
+	default:
+		return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeInternal, "empty or unknown dispatch outcome")
+	}
+}
+
+// startOperationResponseToError converts a StartOperationResponse proto into a Nexus SDK error.
+// Returns nil for success variants (SyncSuccess, AsyncSuccess).
+func startOperationResponseToError(resp *nexuspb.StartOperationResponse) error {
+	switch t := resp.GetVariant().(type) {
+	case *nexuspb.StartOperationResponse_SyncSuccess:
+		return nil
+	case *nexuspb.StartOperationResponse_AsyncSuccess:
+		return nil
+	// Operation processed but failed — the worker returned an explicit failure.
+	case *nexuspb.StartOperationResponse_Failure:
+		return operationErrorFromFailure(t.Failure)
+	default:
+		return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeInternal, "empty or unknown start operation response variant")
+	}
+}
+
+// operationErrorFromFailure converts a Temporal API Failure into a Nexus SDK operation error.
+func operationErrorFromFailure(failure *failurepb.Failure) error {
+	state := nexus.OperationStateFailed
+	if failure.GetCanceledFailureInfo() != nil {
+		state = nexus.OperationStateCanceled
+	}
+	cause, err := convertTemporalFailure(failure)
+	if err != nil {
+		return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeInternal, "%v", err)
+	}
+	opError := &nexus.OperationError{
+		State:   state,
+		Message: fmt.Sprintf("operation error: %s", failure.GetMessage()),
+		Cause:   cause,
+	}
+	if err := nexusrpc.MarkAsWrapperError(nexusrpc.DefaultFailureConverter(), opError); err != nil {
+		return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeInternal, "failed to mark operation error as wrapper: %v", err)
+	}
+	return opError
+}