Define CancelActivityNexusTask transfer task type

Author: rkannan82

api/enums/v1/task.go-helpers.pb.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,

api/enums/v1/task.pb.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,

api/persistence/v1/executions.pb.go

@@ -111,7 +111,7 @@ require (
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
-	github.com/golang/mock v1.6.0 // indirect
+	github.com/golang/mock v1.6.0
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/s2a-go v0.1.9 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.5 // indirect
@@ -171,4 +171,3 @@ require (
 	modernc.org/mathutil v1.7.1 // indirect
 	modernc.org/memory v1.11.0 // indirect
 )
-

common/dynamicconfig/constants.go

@@ -373,6 +373,8 @@ go.opentelemetry.io/otel/trace v1.34.0 h1:+ouXS2V8Rd4hp4580a8q23bg0azF2nI8cqLYnC
 go.opentelemetry.io/otel/trace v1.34.0/go.mod h1:Svm7lSjQD7kG7KJ/MUHPVXSDGz2OX4h0M2jHBhmSfRE=
 go.opentelemetry.io/proto/otlp v1.5.0 h1:xJvq7gMzB31/d406fB8U5CBdyQGw4P399D1aQWU/3i4=
 go.opentelemetry.io/proto/otlp v1.5.0/go.mod h1:keN8WnHxOy8PG0rQZjJJ5A2ebUoafqWp0eVQ4yIXvJ4=
+go.temporal.io/api v1.61.1-0.20260128230845-c246540cf2ed h1:g3CgsK5BXL2rQy0ZIJVRpNUDdtPM1y4bGv5ZoKsqR74=
+go.temporal.io/api v1.61.1-0.20260128230845-c246540cf2ed/go.mod h1:iaxoP/9OXMJcQkETTECfwYq4cw/bj4nwov8b3ZLVnXM=
 go.temporal.io/sdk v1.38.0 h1:4Bok5LEdED7YKpsSjIa3dDqram5VOq+ydBf4pyx0Wo4=
 go.temporal.io/sdk v1.38.0/go.mod h1:a+R2Ej28ObvHoILbHaxMyind7M6D+W0L7edt5UJF4SE=
 go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=

common/persistence/serialization/task_serializers.go

@@ -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

go.mod

@@ -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.

go.sum

@@ -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