Define CancelActivityNexusTask transfer task type

rkannan82 · rkannan82 · commit 54b6d1a658e1 · 2026-02-05T16:36:13.000-08:00
diff --git a/api/enums/v1/task.go-helpers.pb.go b/api/enums/v1/task.go-helpers.pb.go
diff --git a/api/enums/v1/task.pb.go b/api/enums/v1/task.pb.go
diff --git a/api/persistence/v1/executions.pb.go b/api/persistence/v1/executions.pb.go
diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go
@@ -189,6 +189,11 @@ config as the other services.`,
 		false,
 		`EnableActivityEagerExecution indicates if activity eager execution is enabled per namespace`,
 	)
+	EnableActivityCancellationNexusTask = NewGlobalBoolSetting(
+		"system.enableActivityCancellationNexusTask",
+		false,
+		`EnableActivityCancellationNexusTask enables pushing activity cancellation to workers via Nexus task`,
+	)
 	NamespaceMinRetentionGlobal = NewGlobalDurationSetting(
 		"system.namespaceMinRetentionGlobal",
 		24*time.Hour,
diff --git a/common/persistence/serialization/task_serializers.go b/common/persistence/serialization/task_serializers.go
@@ -38,6 +38,8 @@ func serializeTransferTask(
 		transferTask = transferDeleteExecutionTaskToProto(task)
 	case *tasks.ChasmTask:
 		transferTask = transferChasmTaskToProto(task)
+	case *tasks.CancelActivityNexusTask:
+		transferTask = transferCancelActivityNexusTaskToProto(task)
 	default:
 		return nil, serviceerror.NewInternalf("Unknown transfer task type: %v", task)
 	}
@@ -86,6 +88,8 @@ func deserializeTransferTask(
 		task = transferDeleteExecutionTaskFromProto(transferTask)
 	case enumsspb.TASK_TYPE_CHASM:
 		task = transferChasmTaskFromProto(transferTask)
+	case enumsspb.TASK_TYPE_TRANSFER_CANCEL_ACTIVITY_NEXUS:
+		task = transferCancelActivityNexusTaskFromProto(transferTask)
 	default:
 		return nil, serviceerror.NewInternalf("Unknown transfer task type: %v", transferTask.TaskType)
 	}
@@ -106,6 +110,40 @@ func transferChasmTaskFromProto(task *persistencespb.TransferTaskInfo) tasks.Tas
 	}
 }
 
+func transferCancelActivityNexusTaskToProto(task *tasks.CancelActivityNexusTask) *persistencespb.TransferTaskInfo {
+	return &persistencespb.TransferTaskInfo{
+		NamespaceId:    task.WorkflowKey.NamespaceID,
+		WorkflowId:     task.WorkflowKey.WorkflowID,
+		RunId:          task.WorkflowKey.RunID,
+		TaskId:         task.TaskID,
+		TaskType:       task.GetType(),
+		Version:        task.Version,
+		VisibilityTime: timestamppb.New(task.VisibilityTimestamp),
+		TaskDetails: &persistencespb.TransferTaskInfo_CancelActivityNexusTaskDetails_{
+			CancelActivityNexusTaskDetails: &persistencespb.TransferTaskInfo_CancelActivityNexusTaskDetails{
+				ScheduledEventIds: task.ScheduledEventIDs,
+				WorkerInstanceKey: task.WorkerInstanceKey,
+			},
+		},
+	}
+}
+
+func transferCancelActivityNexusTaskFromProto(task *persistencespb.TransferTaskInfo) tasks.Task {
+	details := task.GetCancelActivityNexusTaskDetails()
+	return &tasks.CancelActivityNexusTask{
+		WorkflowKey: definition.NewWorkflowKey(
+			task.NamespaceId,
+			task.WorkflowId,
+			task.RunId,
+		),
+		VisibilityTimestamp: task.VisibilityTime.AsTime(),
+		TaskID:              task.TaskId,
+		Version:             task.Version,
+		ScheduledEventIDs:   details.GetScheduledEventIds(),
+		WorkerInstanceKey:   details.GetWorkerInstanceKey(),
+	}
+}
+
 func serializeTimerTask(
 	encoder Encoder,
 	task tasks.Task,
diff --git a/common/persistence/serialization/task_serializers_test.go b/common/persistence/serialization/task_serializers_test.go
@@ -169,6 +169,19 @@ func (s *taskSerializerSuite) TestTransferResetTask() {
 	s.assertEqualTasks(resetTask)
 }
 
+func (s *taskSerializerSuite) TestTransferCancelActivityNexusTask() {
+	cancelActivityNexusTask := &tasks.CancelActivityNexusTask{
+		WorkflowKey:         s.workflowKey,
+		VisibilityTimestamp: time.Unix(0, rand.Int63()).UTC(),
+		TaskID:              rand.Int63(),
+		Version:             rand.Int63(),
+		ScheduledEventIDs:   []int64{rand.Int63(), rand.Int63(), rand.Int63()},
+		WorkerInstanceKey:   "test-worker-instance-key",
+	}
+
+	s.assertEqualTasks(cancelActivityNexusTask)
+}
+
 func (s *taskSerializerSuite) TestTimerWorkflowTask() {
 	workflowTaskTimer := &tasks.WorkflowTaskTimeoutTask{
 		WorkflowKey:         s.workflowKey,
diff --git a/proto/internal/temporal/server/api/enums/v1/task.proto b/proto/internal/temporal/server/api/enums/v1/task.proto
@@ -59,6 +59,9 @@ enum TaskType {
 
     // A task with side effects generated by a CHASM component.
     TASK_TYPE_CHASM = 33;
+
+    // A task to cancel a running activity via Nexus control queue.
+    TASK_TYPE_TRANSFER_CANCEL_ACTIVITY_NEXUS = 34;
 }
 
 // TaskPriority is only used for replication task as of May 2024
diff --git a/proto/internal/temporal/server/api/persistence/v1/executions.proto b/proto/internal/temporal/server/api/persistence/v1/executions.proto
@@ -349,11 +349,21 @@ message TransferTaskInfo {
         // by some other task, so this task doesn't need to worry about it.
         bool can_skip_visibility_archival = 1;
     }
+
+    // Details for a Nexus task that cancels activities belonging to a specific worker.
+    message CancelActivityNexusTaskDetails {
+        // Scheduled event IDs of activities to cancel.
+        repeated int64 scheduled_event_ids = 1;
+        string worker_instance_key = 2;
+    }
+
     oneof task_details {
         CloseExecutionTaskDetails close_execution_task_details = 16;
 
         // If the task addresses a CHASM component, this field will be set.
         ChasmTaskInfo chasm_task_info = 18;
+
+        CancelActivityNexusTaskDetails cancel_activity_nexus_task_details = 19;
     }
     // Stamp represents the "version" of the entity's internal state for which the transfer task was created.
     // It increases monotonically when the entity's options are modified.
diff --git a/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go b/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go
@@ -39,6 +39,7 @@ import (
 	"go.temporal.io/server/service/history/api"
 	"go.temporal.io/server/service/history/configs"
 	historyi "go.temporal.io/server/service/history/interfaces"
+	"go.temporal.io/server/service/history/tasks"
 	"go.temporal.io/server/service/history/workflow"
 	"go.temporal.io/server/service/history/workflow/update"
 	"google.golang.org/protobuf/proto"
@@ -658,6 +659,13 @@ func (handler *workflowTaskCompletedHandler) handleCommandRequestCancelActivity(
 				return nil, err
 			}
 			handler.activityNotStartedCancelled = true
+		} else if ai.WorkerInstanceKey != "" && handler.config.EnableActivityCancellationNexusTask() {
+			// Activity has started and worker supports Nexus tasks - create cancel task.
+			handler.mutableState.AddTasks(&tasks.CancelActivityNexusTask{
+				WorkflowKey:       handler.mutableState.GetWorkflowKey(),
+				ScheduledEventIDs: []int64{ai.ScheduledEventId},
+				WorkerInstanceKey: ai.WorkerInstanceKey,
+			})
 		}
 	}
 	return actCancelReqEvent, nil
diff --git a/service/history/configs/config.go b/service/history/configs/config.go
@@ -354,11 +354,12 @@ type Config struct {
 	ESProcessorFlushInterval          dynamicconfig.DurationPropertyFn
 	ESProcessorAckTimeout             dynamicconfig.DurationPropertyFn
 
-	EnableCrossNamespaceCommands      dynamicconfig.BoolPropertyFn
-	EnableActivityEagerExecution      dynamicconfig.BoolPropertyFnWithNamespaceFilter
-	EnableActivityRetryStampIncrement dynamicconfig.BoolPropertyFn
-	EnableEagerWorkflowStart          dynamicconfig.BoolPropertyFnWithNamespaceFilter
-	NamespaceCacheRefreshInterval     dynamicconfig.DurationPropertyFn
+	EnableCrossNamespaceCommands              dynamicconfig.BoolPropertyFn
+	EnableActivityEagerExecution              dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	EnableActivityRetryStampIncrement         dynamicconfig.BoolPropertyFn
+	EnableActivityCancellationNexusTask dynamicconfig.BoolPropertyFn
+	EnableEagerWorkflowStart                  dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	NamespaceCacheRefreshInterval             dynamicconfig.DurationPropertyFn
 
 	// ArchivalQueueProcessor settings
 	ArchivalProcessorSchedulerWorkerCount               dynamicconfig.TypedSubscribable[int]
@@ -721,11 +722,12 @@ func NewConfig(
 		ESProcessorFlushInterval: dynamicconfig.WorkerESProcessorFlushInterval.Get(dc),
 		ESProcessorAckTimeout:    dynamicconfig.WorkerESProcessorAckTimeout.Get(dc),
 
-		EnableCrossNamespaceCommands:      dynamicconfig.EnableCrossNamespaceCommands.Get(dc),
-		EnableActivityEagerExecution:      dynamicconfig.EnableActivityEagerExecution.Get(dc),
-		EnableActivityRetryStampIncrement: dynamicconfig.EnableActivityRetryStampIncrement.Get(dc),
-		EnableEagerWorkflowStart:          dynamicconfig.EnableEagerWorkflowStart.Get(dc),
-		NamespaceCacheRefreshInterval:     dynamicconfig.NamespaceCacheRefreshInterval.Get(dc),
+		EnableCrossNamespaceCommands:              dynamicconfig.EnableCrossNamespaceCommands.Get(dc),
+		EnableActivityEagerExecution:              dynamicconfig.EnableActivityEagerExecution.Get(dc),
+		EnableActivityRetryStampIncrement:         dynamicconfig.EnableActivityRetryStampIncrement.Get(dc),
+		EnableActivityCancellationNexusTask: dynamicconfig.EnableActivityCancellationNexusTask.Get(dc),
+		EnableEagerWorkflowStart:                  dynamicconfig.EnableEagerWorkflowStart.Get(dc),
+		NamespaceCacheRefreshInterval:             dynamicconfig.NamespaceCacheRefreshInterval.Get(dc),
 
 		// Archival related
 		ArchivalTaskBatchSize:                               dynamicconfig.ArchivalTaskBatchSize.Get(dc),
diff --git a/service/history/tasks/cancel_activity_nexus_task.go b/service/history/tasks/cancel_activity_nexus_task.go
@@ -0,0 +1,71 @@
+package tasks
+
+import (
+	"fmt"
+	"time"
+
+	enumsspb "go.temporal.io/server/api/enums/v1"
+	"go.temporal.io/server/common/definition"
+)
+
+var _ Task = (*CancelActivityNexusTask)(nil)
+
+type (
+	CancelActivityNexusTask struct {
+		definition.WorkflowKey
+		VisibilityTimestamp time.Time
+		TaskID              int64
+		Version             int64
+
+		// ScheduledEventIDs of activities to cancel (batched by worker).
+		ScheduledEventIDs []int64
+		WorkerInstanceKey string
+	}
+)
+
+func (t *CancelActivityNexusTask) GetKey() Key {
+	return NewImmediateKey(t.TaskID)
+}
+
+func (t *CancelActivityNexusTask) GetVersion() int64 {
+	return t.Version
+}
+
+func (t *CancelActivityNexusTask) SetVersion(version int64) {
+	t.Version = version
+}
+
+func (t *CancelActivityNexusTask) GetTaskID() int64 {
+	return t.TaskID
+}
+
+func (t *CancelActivityNexusTask) SetTaskID(id int64) {
+	t.TaskID = id
+}
+
+func (t *CancelActivityNexusTask) GetVisibilityTime() time.Time {
+	return t.VisibilityTimestamp
+}
+
+func (t *CancelActivityNexusTask) SetVisibilityTime(timestamp time.Time) {
+	t.VisibilityTimestamp = timestamp
+}
+
+func (t *CancelActivityNexusTask) GetCategory() Category {
+	return CategoryTransfer
+}
+
+func (t *CancelActivityNexusTask) GetType() enumsspb.TaskType {
+	return enumsspb.TASK_TYPE_TRANSFER_CANCEL_ACTIVITY_NEXUS
+}
+
+func (t *CancelActivityNexusTask) String() string {
+	return fmt.Sprintf("CancelActivityNexusTask{WorkflowKey: %s, VisibilityTimestamp: %v, TaskID: %v, ScheduledEventIDs: %v, WorkerInstanceKey: %v, Version: %v}",
+		t.WorkflowKey.String(),
+		t.VisibilityTimestamp,
+		t.TaskID,
+		t.ScheduledEventIDs,
+		t.WorkerInstanceKey,
+		t.Version,
+	)
+}
diff --git a/service/history/tasks/utils.go b/service/history/tasks/utils.go
@@ -78,6 +78,12 @@ func GetTransferTaskEventID(
 		eventID = common.FirstEventID
 	case *ChasmTask:
 		return getChasmTaskEventID()
+	case *CancelActivityNexusTask:
+		if len(task.ScheduledEventIDs) > 0 {
+			eventID = task.ScheduledEventIDs[0]
+		} else {
+			eventID = common.FirstEventID
+		}
 	case *FakeTask:
 		// no-op
 	default:
diff --git a/service/history/transfer_queue_active_task_executor.go b/service/history/transfer_queue_active_task_executor.go
@@ -142,6 +142,9 @@ func (t *transferQueueActiveTaskExecutor) Execute(
 		err = t.processDeleteExecutionTask(ctx, task)
 	case *tasks.ChasmTask:
 		err = t.executeChasmSideEffectTransferTask(ctx, task)
+	case *tasks.CancelActivityNexusTask:
+		// TODO: Implement dispatch to worker control queue
+		err = nil
 	default:
 		err = errUnknownTransferTask
 	}
diff --git a/service/history/transfer_queue_standby_task_executor.go b/service/history/transfer_queue_standby_task_executor.go
@@ -108,6 +108,9 @@ func (t *transferQueueStandbyTaskExecutor) Execute(
 		err = t.processDeleteExecutionTask(ctx, task, false)
 	case *tasks.ChasmTask:
 		err = t.executeChasmSideEffectTransferTask(ctx, task)
+	case *tasks.CancelActivityNexusTask:
+		// Cancel activity nexus task is best-effort and only processed in active cluster
+		err = nil
 	default:
 		err = errUnknownTransferTask
 	}

Original file line number	Diff line number	Diff line change
`@@ -59,6 +59,9 @@ enum TaskType {`
`59`	`59`
`60`	`60`	`// A task with side effects generated by a CHASM component.`
`61`	`61`	`TASK_TYPE_CHASM = 33;`
	`62`	`+`
	`63`	`+ // A task to cancel a running activity via Nexus control queue.`
	`64`	`+ TASK_TYPE_TRANSFER_CANCEL_ACTIVITY_NEXUS = 34;`
`62`	`65`	`}`
`63`	`66`
`64`	`67`	`// TaskPriority is only used for replication task as of May 2024`