Store worker attributes needed by server to propagate nexus tasks to worker (#9231)

## What changed?
As part of RecordActivityTaskStarted flow, store
worker_control_task_queue for an activity in the mutable state
(ActivityInfo).

Main changes:
- executions.proto: Added the new worker_control_task_queue field.
- mutable_state_impl.go: Update mutable state.
- matching/forwarder.go: Propagate worker_control_task_queue when polls
get forwarded. Otherwise, RecordActivityTaskStarted request will not
have it set when invoked from a forwarded poll.

## Why?
To support activity cancellation without activity heartbeat.

Overall flow:
- [This PR] Store worker attributes in ActivityInfo as part of
RecordActivityTaskStarted call.
- [#9232] When user cancels a workflow, create 1 or more tasks. Group
all activities belonging to a worker into the task (for efficiency).
- [#9233] Lookup the Nexus task queue for each worker, and send a Nexus
operation for each transfer task.
- [SDK] Worker will receive this cancel task and cancel the running
activities.

## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [x] added new unit test(s)
- [ ] added new functional test(s)

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: rkannan82

api/persistence/v1/executions.pb.go

@@ -655,6 +655,10 @@ message ActivityInfo {
   bool reset_heartbeats = 48;
 
   int64 start_version = 50;
+
+  // 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;
 }
 
 // timer_map column

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

@@ -242,6 +242,7 @@ func recordActivityTaskStarted(
 	if _, err := mutableState.AddActivityTaskStartedEvent(
 		ai, scheduledEventID, requestID, request.PollRequest.GetIdentity(),
 		versioningStamp, pollerDeployment, request.GetBuildIdRedirectInfo(),
+		request.PollRequest.GetWorkerControlTaskQueue(),
 	); err != nil {
 		return nil, rejectCodeUndefined, err
 	}

service/history/api/recordactivitytaskstarted/api.go

@@ -99,6 +99,7 @@ func Invoke(
 					// TODO (shahab): do we need to do anything with wf redirect in this case or any
 					// other case where an activity starts?
 					nil,
+					"", // workerControlTaskQueue not available for force complete
 				)
 				if err != nil {
 					return nil, err

service/history/api/respondactivitytaskcompleted/api.go

@@ -393,6 +393,7 @@ func (handler *WorkflowTaskCompletedHandler) Invoke(
 
 		workflowTaskHandler := newWorkflowTaskCompletedHandler(
 			request.GetIdentity(),
+			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

@@ -54,6 +54,7 @@ type (
 
 	workflowTaskCompletedHandler struct {
 		identity                string
+		workerControlTaskQueue  string
 		workflowTaskCompletedID int64
 
 		// internal state
@@ -106,6 +107,7 @@ type (
 
 func newWorkflowTaskCompletedHandler(
 	identity string,
+	workerControlTaskQueue string,
 	workflowTaskCompletedID int64,
 	mutableState historyi.MutableState,
 	updateRegistry update.Registry,
@@ -126,6 +128,7 @@ func newWorkflowTaskCompletedHandler(
 ) *workflowTaskCompletedHandler {
 	return &workflowTaskCompletedHandler{
 		identity:                identity,
+		workerControlTaskQueue:  workerControlTaskQueue,
 		workflowTaskCompletedID: workflowTaskCompletedID,
 
 		// internal state
@@ -585,6 +588,7 @@ func (handler *workflowTaskCompletedHandler) handlePostCommandEagerExecuteActivi
 		stamp,
 		nil,
 		nil,
+		handler.workerControlTaskQueue, // Eager: activity runs on the same worker that completed the WFT.
 	); err != nil {
 		return nil, err
 	}

service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go

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

service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler_test.go

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

service/history/history_engine_test.go

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