Store worker_control_task_queue in ActivityInfo

Add workerControlTaskQueue parameter to AddActivityTaskStartedEvent
and persist it in ActivityInfo when an activity starts. This enables
routing activity cancellation requests to the correct worker's control
queue via Nexus.

Changes:
- Add worker_control_task_queue field to ActivityInfo proto
- Update MutableState interface and implementation
- Pass workerControlTaskQueue from poll request for regular activities
- Pass from RespondWorkflowTaskCompleted request for eager activities
- Update all test call sites

Author: rkannan82

api/persistence/v1/executions.pb.go

@@ -633,6 +633,9 @@ message ActivityInfo {
 
     // Unique identifier of the worker that is this activity.
     string worker_instance_key = 51;
+
+    // The task queue on which the server will send control tasks to the worker running this activity.
+    string worker_control_task_queue = 52;
 }
 
 // timer_map column

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

@@ -243,6 +243,7 @@ func recordActivityTaskStarted(
 		ai, scheduledEventID, requestID, request.PollRequest.GetIdentity(),
 		versioningStamp, pollerDeployment, request.GetBuildIdRedirectInfo(),
 		request.PollRequest.GetWorkerInstanceKey(),
+		request.PollRequest.GetWorkerControlTaskQueue(),
 	); err != nil {
 		return nil, rejectCodeUndefined, err
 	}

service/history/api/recordactivitytaskstarted/api.go

@@ -99,6 +99,7 @@ func Invoke(
 					// other case where an activity starts?
 					nil,
 					"", // workerInstanceKey not available for force complete
+					"", // workerControlTaskQueue not available for force complete
 				)
 				if err != nil {
 					return nil, err

service/history/api/respondactivitytaskcompleted/api.go

@@ -389,6 +389,8 @@ func (handler *WorkflowTaskCompletedHandler) Invoke(
 
 		workflowTaskHandler := newWorkflowTaskCompletedHandler(
 			request.GetIdentity(),
+			request.GetWorkerInstanceKey(),
+			request.GetWorkerControlTaskQueue(),
 			completedEvent.GetEventId(), // If completedEvent is nil, then GetEventId() returns 0 and this value shouldn't be used in workflowTaskHandler.
 			ms,
 			updateRegistry,

service/history/api/respondworkflowtaskcompleted/api.go

@@ -53,6 +53,8 @@ type (
 
 	workflowTaskCompletedHandler struct {
 		identity                string
+		workerInstanceKey       string
+		workerControlTaskQueue  string
 		workflowTaskCompletedID int64
 
 		// internal state
@@ -104,6 +106,8 @@ type (
 
 func newWorkflowTaskCompletedHandler(
 	identity string,
+	workerInstanceKey string,
+	workerControlTaskQueue string,
 	workflowTaskCompletedID int64,
 	mutableState historyi.MutableState,
 	updateRegistry update.Registry,
@@ -122,7 +126,9 @@ func newWorkflowTaskCompletedHandler(
 	versionMembershipCache worker_versioning.VersionMembershipCache,
 ) *workflowTaskCompletedHandler {
 	return &workflowTaskCompletedHandler{
-		identity:                identity,
+		identity:               identity,
+		workerInstanceKey:      workerInstanceKey,
+		workerControlTaskQueue: workerControlTaskQueue,
 		workflowTaskCompletedID: workflowTaskCompletedID,
 
 		// internal state
@@ -559,7 +565,8 @@ func (handler *workflowTaskCompletedHandler) handlePostCommandEagerExecuteActivi
 		stamp,
 		nil,
 		nil,
-		"", // workerInstanceKey not available for eager dispatch
+		handler.workerInstanceKey,
+		handler.workerControlTaskQueue,
 	); err != nil {
 		return nil, err
 	}

service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go

@@ -84,6 +84,8 @@ func TestCommandProtocolMessage(t *testing.T) {
 		)
 		out.handler = newWorkflowTaskCompletedHandler( // 😲
 			t.Name(), // identity
+			"",       // workerInstanceKey
+			"",       // workerControlTaskQueue
 			123,      // workflowTaskCompletedID
 			out.ms,
 			out.updates,

service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler_test.go

@@ -6679,6 +6679,7 @@ func addActivityTaskStartedEvent(ms historyi.MutableState, scheduledEventID int6
 		nil,
 		nil,
 		"",
+		"",
 	)
 	return event
 }

service/history/history_engine_test.go

@@ -58,6 +58,7 @@ type (
 			*deploymentpb.Deployment,
 			*taskqueuespb.BuildIdRedirectInfo,
 			string, // workerInstanceKey
+			string, // workerControlTaskQueue
 		) (*historypb.HistoryEvent, error)
 		AddActivityTaskTimedOutEvent(int64, int64, *failurepb.Failure, enumspb.RetryState) (*historypb.HistoryEvent, error)
 		AddChildWorkflowExecutionCanceledEvent(int64, *commonpb.WorkflowExecution, *historypb.WorkflowExecutionCanceledEventAttributes) (*historypb.HistoryEvent, error)