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

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

common/dynamicconfig/constants.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/persistence/serialization/task_serializers.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_test.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.