Address review feedback: backward compat, standby executor, lock safety

- Add StartedClock nil check in cancel handler (backward compat for
  activities started before deploy)
- Add WorkerCommandsTask case to standby executor (drop task)
- Use shardCtx.GetConfig() instead of passing config param
- Add lock around Executable.Attempt() for thread safety
- Add replication comment on started_clock proto field
- Add tests for cancel command with/without StartedClock

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: rkannan82

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

@@ -655,6 +655,9 @@ message ActivityInfo {
     // is to store the full serialized task token (~150-300 bytes), which avoids
     // reconstruction entirely and is immune to token format changes. We chose the
     // clock approach to keep the per-activity memory footprint minimal (~24 bytes).
+    //
+    // Replication: This field is part of ActivityInfo and is automatically replicated
+    // via state-based replication. No special handling is needed.
     temporal.server.api.clock.v1.VectorClock started_clock = 52;
 }
 

service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go

@@ -670,36 +670,47 @@ func (handler *workflowTaskCompletedHandler) handleCommandRequestCancelActivity(
 			}
 			handler.activityNotStartedCancelled = true
 		} else if ai.StartedEventId != common.EmptyEventID && ai.WorkerControlTaskQueue != "" {
-			// Activity has started and worker supports Nexus control tasks - collect for batched dispatch.
-			taskToken, err := handler.tokenSerializer.Serialize(tasktoken.NewActivityTaskToken(
-				handler.mutableState.GetNamespaceEntry().ID().String(),
-				handler.mutableState.GetWorkflowKey().WorkflowID,
-				handler.mutableState.GetWorkflowKey().RunID,
-				ai.ScheduledEventId,
-				ai.ActivityId,
-				ai.ActivityType.GetName(),
-				ai.Attempt,
-				ai.StartedClock,
-				ai.Version,
-				ai.StartVersion,
-				nil,
-			))
-			if err != nil {
-				return nil, err
-			}
-			if handler.pendingWorkerCommandsByControlQueue == nil {
-				handler.pendingWorkerCommandsByControlQueue = make(map[string][]*workerpb.WorkerCommand)
-			}
-			handler.pendingWorkerCommandsByControlQueue[ai.WorkerControlTaskQueue] = append(
-				handler.pendingWorkerCommandsByControlQueue[ai.WorkerControlTaskQueue],
-				&workerpb.WorkerCommand{
-					Type: &workerpb.WorkerCommand_CancelActivity{
-						CancelActivity: &workerpb.CancelActivityCommand{
-							TaskToken: taskToken,
+			if ai.StartedClock == nil {
+				// StartedClock may be nil for activities started before this feature was deployed.
+				// Skip cancel command; the activity will time out normally.
+				handler.logger.Info("Skipping worker cancel command: activity missing StartedClock (pre-deploy)",
+					tag.WorkflowNamespaceID(handler.mutableState.GetWorkflowKey().NamespaceID),
+					tag.WorkflowID(handler.mutableState.GetWorkflowKey().WorkflowID),
+					tag.WorkflowRunID(handler.mutableState.GetWorkflowKey().RunID),
+					tag.WorkflowScheduledEventID(ai.ScheduledEventId),
+				)
+			} else {
+				// Activity has started and worker supports Nexus control tasks - collect for batched dispatch.
+				taskToken, err := handler.tokenSerializer.Serialize(tasktoken.NewActivityTaskToken(
+					handler.mutableState.GetNamespaceEntry().ID().String(),
+					handler.mutableState.GetWorkflowKey().WorkflowID,
+					handler.mutableState.GetWorkflowKey().RunID,
+					ai.ScheduledEventId,
+					ai.ActivityId,
+					ai.ActivityType.GetName(),
+					ai.Attempt,
+					ai.StartedClock,
+					ai.Version,
+					ai.StartVersion,
+					nil,
+				))
+				if err != nil {
+					return nil, err
+				}
+				if handler.pendingWorkerCommandsByControlQueue == nil {
+					handler.pendingWorkerCommandsByControlQueue = make(map[string][]*workerpb.WorkerCommand)
+				}
+				handler.pendingWorkerCommandsByControlQueue[ai.WorkerControlTaskQueue] = append(
+					handler.pendingWorkerCommandsByControlQueue[ai.WorkerControlTaskQueue],
+					&workerpb.WorkerCommand{
+						Type: &workerpb.WorkerCommand_CancelActivity{
+							CancelActivity: &workerpb.CancelActivityCommand{
+								TaskToken: taskToken,
+							},
 						},
 					},
-				},
-			)
+				)
+			}
 		}
 	}
 	return actCancelReqEvent, nil

service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	commandpb "go.temporal.io/api/command/v1"
 	commonpb "go.temporal.io/api/common/v1"
@@ -17,12 +18,15 @@ import (
 	"go.temporal.io/api/serviceerror"
 	updatepb "go.temporal.io/api/update/v1"
 	workerpb "go.temporal.io/api/worker/v1"
+	clockspb "go.temporal.io/server/api/clock/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
+	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/backoff"
 	"go.temporal.io/server/common/collection"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/effect"
 	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/tasktoken"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/namespace/nsregistry"
@@ -470,3 +474,120 @@ func TestFlushWorkerCommandsTasks(t *testing.T) {
 		require.NoError(t, err)
 	})
 }
+
+func TestHandleCommandRequestCancelActivity_WorkerCommands(t *testing.T) {
+	t.Parallel()
+
+	cancelReqEvent := &historypb.HistoryEvent{EventId: 10}
+	scheduledEventID := int64(5)
+	controlQueue := "/_sys/worker-commands/test-ns/key1"
+
+	t.Run("started activity with clock collects worker command", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		ms := historyi.NewMockMutableState(ctrl)
+
+		ai := &persistencespb.ActivityInfo{
+			ScheduledEventId:      scheduledEventID,
+			StartedEventId:        7,
+			ActivityId:            "act-1",
+			WorkerControlTaskQueue: controlQueue,
+			StartedClock:          &clockspb.VectorClock{ClusterId: 1, ShardId: 1, Clock: 100},
+			Version:               1,
+			StartVersion:          1,
+			Attempt:               1,
+			ActivityType:          &commonpb.ActivityType{Name: "test-activity"},
+		}
+
+		ms.EXPECT().AddActivityTaskCancelRequestedEvent(int64(123), scheduledEventID, "test-identity").
+			Return(cancelReqEvent, ai, nil)
+		ms.EXPECT().GetNamespaceEntry().Return(tests.LocalNamespaceEntry).AnyTimes()
+		ms.EXPECT().GetWorkflowKey().Return(tests.WorkflowKey).AnyTimes()
+
+		handler := &workflowTaskCompletedHandler{
+			identity:                    "test-identity",
+			workflowTaskCompletedID:     123,
+			mutableState:                ms,
+			tokenSerializer:             tasktoken.NewSerializer(),
+			logger:                      log.NewNoopLogger(),
+		}
+
+		event, err := handler.handleCommandRequestCancelActivity(
+			context.Background(),
+			&commandpb.RequestCancelActivityTaskCommandAttributes{
+				ScheduledEventId: scheduledEventID,
+			},
+		)
+		require.NoError(t, err)
+		assert.Equal(t, cancelReqEvent, event)
+		assert.Len(t, handler.pendingWorkerCommandsByControlQueue[controlQueue], 1)
+	})
+
+	t.Run("started activity without clock skips worker command", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		ms := historyi.NewMockMutableState(ctrl)
+
+		ai := &persistencespb.ActivityInfo{
+			ScheduledEventId:      scheduledEventID,
+			StartedEventId:        7,
+			ActivityId:            "act-1",
+			WorkerControlTaskQueue: controlQueue,
+			StartedClock:          nil, // pre-deploy: no clock stored
+			Version:               1,
+			StartVersion:          1,
+			Attempt:               1,
+		}
+
+		ms.EXPECT().AddActivityTaskCancelRequestedEvent(int64(123), scheduledEventID, "test-identity").
+			Return(cancelReqEvent, ai, nil)
+
+		handler := &workflowTaskCompletedHandler{
+			identity:                    "test-identity",
+			workflowTaskCompletedID:     123,
+			mutableState:                ms,
+			logger:                      log.NewNoopLogger(),
+		}
+
+		event, err := handler.handleCommandRequestCancelActivity(
+			context.Background(),
+			&commandpb.RequestCancelActivityTaskCommandAttributes{
+				ScheduledEventId: scheduledEventID,
+			},
+		)
+		require.NoError(t, err)
+		assert.Equal(t, cancelReqEvent, event)
+		assert.Empty(t, handler.pendingWorkerCommandsByControlQueue, "no worker command should be created when StartedClock is nil")
+	})
+
+	t.Run("started activity without control queue does not collect worker command", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		ms := historyi.NewMockMutableState(ctrl)
+
+		ai := &persistencespb.ActivityInfo{
+			ScheduledEventId:      scheduledEventID,
+			StartedEventId:        7,
+			ActivityId:            "act-1",
+			WorkerControlTaskQueue: "", // worker doesn't support control tasks
+			StartedClock:          &clockspb.VectorClock{ClusterId: 1, ShardId: 1, Clock: 100},
+		}
+
+		ms.EXPECT().AddActivityTaskCancelRequestedEvent(int64(123), scheduledEventID, "test-identity").
+			Return(cancelReqEvent, ai, nil)
+
+		handler := &workflowTaskCompletedHandler{
+			identity:                    "test-identity",
+			workflowTaskCompletedID:     123,
+			mutableState:                ms,
+			logger:                      log.NewNoopLogger(),
+		}
+
+		event, err := handler.handleCommandRequestCancelActivity(
+			context.Background(),
+			&commandpb.RequestCancelActivityTaskCommandAttributes{
+				ScheduledEventId: scheduledEventID,
+			},
+		)
+		require.NoError(t, err)
+		assert.Equal(t, cancelReqEvent, event)
+		assert.Empty(t, handler.pendingWorkerCommandsByControlQueue)
+	})
+}

service/history/outbound_queue_active_task_executor.go

@@ -11,7 +11,6 @@ import (
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/resource"
-	"go.temporal.io/server/service/history/configs"
 	"go.temporal.io/server/service/history/consts"
 	historyi "go.temporal.io/server/service/history/interfaces"
 	"go.temporal.io/server/service/history/queues"
@@ -39,7 +38,6 @@ func newOutboundQueueActiveTaskExecutor(
 	metricsHandler metrics.Handler,
 	chasmEngine chasm.Engine,
 	matchingRawClient resource.MatchingRawClient,
-	config *configs.Config,
 ) *outboundQueueActiveTaskExecutor {
 	scopedMetricsHandler := metricsHandler.WithTags(
 		metrics.OperationTag(metrics.OperationOutboundQueueProcessorScope),
@@ -54,7 +52,7 @@ func newOutboundQueueActiveTaskExecutor(
 		chasmEngine: chasmEngine,
 		workerCommandsTaskDispatcher: newWorkerCommandsTaskDispatcher(
 			matchingRawClient,
-			config,
+			shardCtx.GetConfig(),
 			scopedMetricsHandler,
 			logger,
 		),

service/history/outbound_queue_active_task_executor_test.go

@@ -116,7 +116,6 @@ func (s *outboundQueueActiveTaskExecutorSuite) SetupTest() {
 		s.metricsHandler,
 		s.mockChasmEngine,
 		nil, // matchingRawClient - not used in these tests
-		nil, // config - not used in these tests
 	)
 }
 

service/history/outbound_queue_factory.go

@@ -230,7 +230,6 @@ func (f *outboundQueueFactory) CreateQueue(
 		metricsHandler,
 		f.ChasmEngine,
 		f.MatchingRawClient,
-		shardContext.GetConfig(),
 	)
 
 	standbyExecutor := newOutboundQueueStandbyTaskExecutor(

service/history/outbound_queue_standby_task_executor.go

@@ -99,6 +99,10 @@ func (e *outboundQueueStandbyTaskExecutor) Execute(
 		return respond(e.executeStateMachineTask(ctx, task, nsName))
 	case *tasks.ChasmTask:
 		return respond(e.executeChasmSideEffectTask(ctx, task))
+	case *tasks.WorkerCommandsTask:
+		// Worker commands are best-effort and only executed on the active cluster.
+		// On standby, simply drop the task.
+		return respond(nil)
 	}
 
 	return respond(queueserrors.NewUnprocessableTaskError(fmt.Sprintf("unknown task type '%T'", task)))

service/history/queues/executable.go

@@ -716,6 +716,9 @@ func (e *executableImpl) State() ctasks.State {
 }
 
 func (e *executableImpl) Attempt() int {
+	e.Lock()
+	defer e.Unlock()
+
 	return e.attempt
 }