Merge remote-tracking branch 'origin/kannan/worker-cmd-no-retry-timeout' into kannan/worker-cmd-dynamic-config

2026-08-30 18:41:49 -07:00 · 2026-08-29 11:31:25 -07:00
parent 55115d8179 315fee73a4
commit 9a03746351
4 changed files with 335 additions and 22 deletions
--- a/docs/architecture/worker-commands.md
+++ b/docs/architecture/worker-commands.md
@@ -1,6 +1,8 @@
 # Worker Commands

-Worker commands are server-initiated instructions sent to workers via Nexus. Each worker process has a dedicated control task queue that it polls via Nexus RPC. A single control queue serves one or more workers running within the same process. This enables the server to push actions to workers without relying on heartbeats or long-poll cycles.
+Worker commands are server-initiated commands sent to workers via Nexus. Each worker process has a dedicated task queue that it polls via Nexus RPC. This dedicated queue serves one or more workers running within the same process. This enables the server to push commands to workers without relying on activity heartbeats. For example, if user cancels an activity, server will send a cancel command to the worker that was running the activity.
+
+The dedicated queue (aka control queue in the code) uses a task queue kind: `TASK_QUEUE_KIND_WORKER_COMMANDS`. This distinguishes it from normal and sticky task queues in matching. The SDK sets this kind when polling for worker commands, and the dispatcher sets it when dispatching commands to matching. The server treats it as an internal task queue kind, meaning it is not directly accessible to users via standard task queue APIs.

 To route a command, the server needs to know the target worker's control queue. For commands that target activities, this information is stored directly in the mutable state (`ActivityInfo`) when the activity was dispatched to the worker.

--- a/service/history/ndc/transaction_manager.go
+++ b/service/history/ndc/transaction_manager.go
@@ -48,6 +48,7 @@ import (
 // create path (there will be only current branch)
 // 1. create as current															-> nDCTransactionPolicyCreateAsCurrent
 // 2. create as zombie															-> nDCTransactionPolicyCreateAsZombie
+// 3. create without changing state or current									-> nDCTransactionPolicyCreateBypassCurrent
 //
 // create path (there will be only current branch) + suppress current
 // 1. create as current & suppress current										-> nDCTransactionPolicySuppressCurrentAndCreateAsCurrent
@@ -74,6 +75,7 @@ type nDCTransactionPolicy int
 const (
 	nDCTransactionPolicyCreateAsCurrent nDCTransactionPolicy = iota
 	nDCTransactionPolicyCreateAsZombie
+	nDCTransactionPolicyCreateBypassCurrent
 	nDCTransactionPolicySuppressCurrentAndCreateAsCurrent

 	nDCTransactionPolicyUpdateAsCurrent
--- a/service/history/ndc/transaction_manager_new_workflow.go
+++ b/service/history/ndc/transaction_manager_new_workflow.go
@@ -6,6 +6,7 @@ import (
 	"context"

 	"go.temporal.io/api/serviceerror"
+	enumsspb "go.temporal.io/server/api/enums/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/persistence"
@@ -57,8 +58,9 @@ func (r *nDCTransactionMgrForNewWorkflowImpl) dispatchForNewWorkflow(
 	// NOTE: this function does NOT mutate current workflow or target workflow,
 	//  workflow mutation is done in methods within executeTransaction function

-	targetExecutionInfo := targetWorkflow.GetMutableState().GetExecutionInfo()
-	targetExecutionState := targetWorkflow.GetMutableState().GetExecutionState()
+	targetMutableState := targetWorkflow.GetMutableState()
+	targetExecutionInfo := targetMutableState.GetExecutionInfo()
+	targetExecutionState := targetMutableState.GetExecutionState()
 	namespaceID := namespace.ID(targetExecutionInfo.NamespaceId)
 	workflowID := targetExecutionInfo.WorkflowId
 	targetRunID := targetExecutionState.RunId
@@ -79,7 +81,24 @@ func (r *nDCTransactionMgrForNewWorkflowImpl) dispatchForNewWorkflow(
 	}

 	if currentRunID == "" {
-		// current record does not exists
+		// Preserve non-current snapshots without promoting them to current. A zombie must never own the
+		// current execution record. A completed or corrupted run with a known successor must also remain
+		// non-current; otherwise a deleted successor could cause the older run to be resurrected.
+		targetState := targetExecutionState.GetState()
+		targetHasSuccessor := targetExecutionInfo.GetNewExecutionRunId() != "" ||
+			targetExecutionInfo.GetSuccessorRunId() != ""
+		switch {
+		case targetState == enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE,
+			targetState == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED && targetHasSuccessor,
+			targetState == enumsspb.WORKFLOW_EXECUTION_STATE_CORRUPTED && targetHasSuccessor:
+			return r.executeTransaction(
+				ctx,
+				nDCTransactionPolicyCreateBypassCurrent,
+				nil, // no current workflow: the current record was deleted
+				targetWorkflow,
+			)
+		}
+		// current record does not exist, create as brand new
 		return r.executeTransaction(
 			ctx,
 			nDCTransactionPolicyCreateAsCurrent,
@@ -195,9 +214,7 @@ func (r *nDCTransactionMgrForNewWorkflowImpl) createAsZombie(
 	targetWorkflow Workflow,
 ) error {

-	targetWorkflowPolicy, err := targetWorkflow.SuppressBy(
-		currentWorkflow,
-	)
+	targetWorkflowPolicy, err := targetWorkflow.SuppressBy(currentWorkflow)
 	if err != nil {
 		return err
 	}
@@ -208,7 +225,25 @@ func (r *nDCTransactionMgrForNewWorkflowImpl) createAsZombie(
 	// release lock on current workflow, since current cluster maybe the active cluster
 	// and events maybe reapplied to current workflow
 	currentWorkflow.GetReleaseFn()(nil)
-	currentWorkflow = nil
+
+	return r.persistBypassCurrent(ctx, targetWorkflow, targetWorkflowPolicy)
+}
+
+// createBypassCurrent persists the incoming execution state as-is without creating or updating the
+// current execution record. Unlike createAsZombie, it deliberately performs no suppression and no
+// workflow state transition.
+func (r *nDCTransactionMgrForNewWorkflowImpl) createBypassCurrent(
+	ctx context.Context,
+	targetWorkflow Workflow,
+) error {
+	return r.persistBypassCurrent(ctx, targetWorkflow, historyi.TransactionPolicyPassive)
+}
+
+func (r *nDCTransactionMgrForNewWorkflowImpl) persistBypassCurrent(
+	ctx context.Context,
+	targetWorkflow Workflow,
+	targetWorkflowPolicy historyi.TransactionPolicy,
+) error {

 	ms := targetWorkflow.GetMutableState()

@@ -247,7 +282,7 @@ func (r *nDCTransactionMgrForNewWorkflowImpl) createAsZombie(
 		}
 	}

-	// target workflow is in zombie state, no need to update current record.
+	// The target workflow is non-current, so do not create or update the current record.
 	createMode := persistence.CreateWorkflowModeBypassCurrent
 	prevRunID := ""
 	prevLastWriteVersion := int64(0)
@@ -331,6 +366,12 @@ func (r *nDCTransactionMgrForNewWorkflowImpl) executeTransaction(
 			targetWorkflow,
 		)

+	case nDCTransactionPolicyCreateBypassCurrent:
+		return r.createBypassCurrent(
+			ctx,
+			targetWorkflow,
+		)
+
 	case nDCTransactionPolicySuppressCurrentAndCreateAsCurrent:
 		return r.suppressCurrentAndCreateAsCurrent(
 			ctx,
--- a/service/history/ndc/transaction_manager_new_workflow_test.go
+++ b/service/history/ndc/transaction_manager_new_workflow_test.go
@@ -8,6 +8,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 	historypb "go.temporal.io/api/history/v1"
+	enumsspb "go.temporal.io/server/api/enums/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/common"
@@ -78,7 +79,253 @@ func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_Dup() {
 	s.ErrorIs(err, consts.ErrDuplicate)
 }

-func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_BrandNew() {
+func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_NoCurrentRecord_StateAndSuccessorCombinations() {
+	successorCases := []struct {
+		name              string
+		newExecutionRunID string
+		successorRunID    string
+	}{
+		{name: "without successor"},
+		{name: "with new execution run ID", newExecutionRunID: "successor"},
+		{name: "with successor run ID", successorRunID: "successor"},
+	}
+
+	stateCases := []struct {
+		name          string
+		state         enumsspb.WorkflowExecutionState
+		expectedModes [3]persistence.CreateWorkflowMode
+	}{
+		{
+			name:  "created",
+			state: enumsspb.WORKFLOW_EXECUTION_STATE_CREATED,
+			expectedModes: [3]persistence.CreateWorkflowMode{
+				persistence.CreateWorkflowModeBrandNew,
+				persistence.CreateWorkflowModeBrandNew,
+				persistence.CreateWorkflowModeBrandNew,
+			},
+		},
+		{
+			name:  "running",
+			state: enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING,
+			expectedModes: [3]persistence.CreateWorkflowMode{
+				persistence.CreateWorkflowModeBrandNew,
+				persistence.CreateWorkflowModeBrandNew,
+				persistence.CreateWorkflowModeBrandNew,
+			},
+		},
+		{
+			name:  "completed",
+			state: enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,
+			expectedModes: [3]persistence.CreateWorkflowMode{
+				persistence.CreateWorkflowModeBrandNew,
+				persistence.CreateWorkflowModeBypassCurrent,
+				persistence.CreateWorkflowModeBypassCurrent,
+			},
+		},
+		{
+			name:  "zombie",
+			state: enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE,
+			expectedModes: [3]persistence.CreateWorkflowMode{
+				persistence.CreateWorkflowModeBypassCurrent,
+				persistence.CreateWorkflowModeBypassCurrent,
+				persistence.CreateWorkflowModeBypassCurrent,
+			},
+		},
+		{
+			name:  "corrupted",
+			state: enumsspb.WORKFLOW_EXECUTION_STATE_CORRUPTED,
+			expectedModes: [3]persistence.CreateWorkflowMode{
+				persistence.CreateWorkflowModeBrandNew,
+				persistence.CreateWorkflowModeBypassCurrent,
+				persistence.CreateWorkflowModeBypassCurrent,
+			},
+		},
+	}
+
+	for _, stateCase := range stateCases {
+		for successorIndex, successorCase := range successorCases {
+			expectedMode := stateCase.expectedModes[successorIndex]
+			s.Run(stateCase.name+" "+successorCase.name, func() {
+				ctx := context.Background()
+				namespaceID := namespace.ID("some random namespace ID")
+				workflowID := "some random workflow ID"
+				runID := "some random run ID"
+				releaseCalled := false
+
+				targetWorkflow := NewMockWorkflow(s.controller)
+				weContext := historyi.NewMockWorkflowContext(s.controller)
+				mutableState := historyi.NewMockMutableState(s.controller)
+				var releaseFn historyi.ReleaseWorkflowContextFunc = func(error) { releaseCalled = true }
+				targetWorkflow.EXPECT().GetContext().Return(weContext).AnyTimes()
+				targetWorkflow.EXPECT().GetMutableState().Return(mutableState).AnyTimes()
+				targetWorkflow.EXPECT().GetReleaseFn().Return(releaseFn).AnyTimes()
+
+				executionInfo := &persistencespb.WorkflowExecutionInfo{
+					NamespaceId:       namespaceID.String(),
+					WorkflowId:        workflowID,
+					NewExecutionRunId: successorCase.newExecutionRunID,
+					SuccessorRunId:    successorCase.successorRunID,
+				}
+				executionState := &persistencespb.WorkflowExecutionState{
+					RunId: runID,
+					State: stateCase.state,
+				}
+				workflowSnapshot := &persistence.WorkflowSnapshot{
+					ExecutionState: executionState,
+				}
+				workflowEventsSeq := []*persistence.WorkflowEvents{}
+
+				mutableState.EXPECT().GetExecutionInfo().Return(executionInfo).AnyTimes()
+				mutableState.EXPECT().GetExecutionState().Return(executionState).AnyTimes()
+				if expectedMode == persistence.CreateWorkflowModeBypassCurrent {
+					mutableState.EXPECT().GetReapplyCandidateEvents().Return(nil)
+				}
+				mutableState.EXPECT().CloseTransactionAsSnapshot(ctx, historyi.TransactionPolicyPassive).Return(
+					workflowSnapshot, workflowEventsSeq, nil,
+				)
+
+				s.mockTransactionMgr.EXPECT().GetCurrentWorkflowRunID(
+					ctx, namespaceID, workflowID, chasm.WorkflowArchetypeID,
+				).Return("", nil)
+
+				weContext.EXPECT().CreateWorkflowExecution(
+					gomock.Any(),
+					s.mockShard,
+					gomock.Any(),
+					"",
+					int64(0),
+					mutableState,
+					workflowSnapshot,
+					workflowEventsSeq,
+					historyi.TransactionPolicyPassive,
+				).DoAndReturn(func(
+					_ context.Context,
+					_ historyi.ShardContext,
+					createMode persistence.CreateWorkflowMode,
+					_ string,
+					_ int64,
+					_ historyi.MutableState,
+					workflowSnapshot *persistence.WorkflowSnapshot,
+					_ []*persistence.WorkflowEvents,
+					_ historyi.TransactionPolicy,
+				) error {
+					s.Equal(expectedMode, createMode)
+					s.Equal(stateCase.state, workflowSnapshot.ExecutionState.State)
+					return persistence.ValidateCreateWorkflowModeState(createMode, *workflowSnapshot)
+				})
+
+				err := s.createMgr.dispatchForNewWorkflow(ctx, chasm.WorkflowArchetypeID, targetWorkflow)
+				s.Require().NoError(err)
+				s.Equal(stateCase.state, executionState.State)
+				s.True(releaseCalled)
+			})
+		}
+	}
+}
+
+func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_NoCurrentRecord_CompletedWithNewExecutionRunID_CreatesBypassCurrentPreservingState() {
+	s.testDispatchForNewWorkflowNoCurrentRecordPreservesState(&persistencespb.WorkflowExecutionInfo{
+		NewExecutionRunId: "successor run ID",
+	}, enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED)
+}
+
+func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_NoCurrentRecord_CompletedWithSuccessorRunID_CreatesBypassCurrentPreservingState() {
+	s.testDispatchForNewWorkflowNoCurrentRecordPreservesState(&persistencespb.WorkflowExecutionInfo{
+		SuccessorRunId: "successor run ID",
+	}, enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED)
+}
+
+func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_NoCurrentRecord_CorruptedWithSuccessorRunID_CreatesBypassCurrentPreservingState() {
+	s.testDispatchForNewWorkflowNoCurrentRecordPreservesState(&persistencespb.WorkflowExecutionInfo{
+		SuccessorRunId: "successor run ID",
+	}, enumsspb.WORKFLOW_EXECUTION_STATE_CORRUPTED)
+}
+
+func (s *transactionMgrForNewWorkflowSuite) testDispatchForNewWorkflowNoCurrentRecordPreservesState(
+	executionInfo *persistencespb.WorkflowExecutionInfo,
+	executionStateValue enumsspb.WorkflowExecutionState,
+) {
+	ctx := context.Background()
+
+	namespaceID := namespace.ID("some random namespace ID")
+	workflowID := "some random workflow ID"
+	runID := "some random run ID"
+	executionInfo.NamespaceId = namespaceID.String()
+	executionInfo.WorkflowId = workflowID
+
+	releaseCalled := false
+
+	targetWorkflow := NewMockWorkflow(s.controller)
+	weContext := historyi.NewMockWorkflowContext(s.controller)
+	mutableState := historyi.NewMockMutableState(s.controller)
+	var releaseFn historyi.ReleaseWorkflowContextFunc = func(error) { releaseCalled = true }
+	targetWorkflow.EXPECT().GetContext().Return(weContext).AnyTimes()
+	targetWorkflow.EXPECT().GetMutableState().Return(mutableState).AnyTimes()
+	targetWorkflow.EXPECT().GetReleaseFn().Return(releaseFn).AnyTimes()
+
+	executionState := &persistencespb.WorkflowExecutionState{
+		RunId: runID,
+		State: executionStateValue,
+	}
+	workflowSnapshot := &persistence.WorkflowSnapshot{
+		ExecutionState: executionState,
+	}
+	// Non-empty event sequence so the reapply branch of createBypassCurrent is exercised (rather than
+	// short-circuited by empty lists). On the passive apply path reapply is forwarded to the active
+	// cluster; here the mock stands in for a successful reapply.
+	workflowEventsSeq := []*persistence.WorkflowEvents{{
+		Events: []*historypb.HistoryEvent{{
+			EventId: common.FirstEventID + rand.Int63(),
+		}},
+	}}
+	mutableState.EXPECT().GetExecutionInfo().Return(executionInfo).AnyTimes()
+	mutableState.EXPECT().GetExecutionState().Return(executionState).AnyTimes()
+	mutableState.EXPECT().GetReapplyCandidateEvents().Return(nil)
+	mutableState.EXPECT().CloseTransactionAsSnapshot(context.Background(), historyi.TransactionPolicyPassive).Return(
+		workflowSnapshot, workflowEventsSeq, nil,
+	)
+
+	s.mockTransactionMgr.EXPECT().GetCurrentWorkflowRunID(
+		ctx, namespaceID, workflowID, chasm.WorkflowArchetypeID,
+	).Return("", nil)
+
+	// A non-current run with a successor and no current record must not resurrect as current: even
+	// with events to reapply, it is persisted via bypass-current without touching the (absent)
+	// current record. SuppressBy is never called since there is no current workflow to suppress against.
+	weContext.EXPECT().ReapplyEvents(gomock.Any(), s.mockShard, workflowEventsSeq).Return(nil)
+	weContext.EXPECT().CreateWorkflowExecution(
+		gomock.Any(),
+		s.mockShard,
+		persistence.CreateWorkflowModeBypassCurrent,
+		"",
+		int64(0),
+		mutableState,
+		workflowSnapshot,
+		workflowEventsSeq,
+		historyi.TransactionPolicyPassive,
+	).DoAndReturn(func(
+		_ context.Context,
+		_ historyi.ShardContext,
+		createMode persistence.CreateWorkflowMode,
+		_ string,
+		_ int64,
+		_ historyi.MutableState,
+		workflowSnapshot *persistence.WorkflowSnapshot,
+		_ []*persistence.WorkflowEvents,
+		_ historyi.TransactionPolicy,
+	) error {
+		s.Equal(persistence.CreateWorkflowModeBypassCurrent, createMode)
+		s.Equal(executionStateValue, workflowSnapshot.ExecutionState.State)
+		return persistence.ValidateCreateWorkflowModeState(createMode, *workflowSnapshot)
+	})
+
+	err := s.createMgr.dispatchForNewWorkflow(ctx, chasm.WorkflowArchetypeID, targetWorkflow)
+	s.NoError(err)
+	s.Equal(executionStateValue, executionState.State)
+	s.True(releaseCalled)
+}
+
+func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_NoCurrentRecord_ZombieWithoutSuccessor_CreatesBypassCurrentPreservingState() {
 	ctx := context.Background()

 	namespaceID := namespace.ID("some random namespace ID")
@@ -87,15 +334,21 @@ func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_BrandNew(

 	releaseCalled := false

-	newWorkflow := NewMockWorkflow(s.controller)
+	targetWorkflow := NewMockWorkflow(s.controller)
 	weContext := historyi.NewMockWorkflowContext(s.controller)
 	mutableState := historyi.NewMockMutableState(s.controller)
 	var releaseFn historyi.ReleaseWorkflowContextFunc = func(error) { releaseCalled = true }
-	newWorkflow.EXPECT().GetContext().Return(weContext).AnyTimes()
-	newWorkflow.EXPECT().GetMutableState().Return(mutableState).AnyTimes()
-	newWorkflow.EXPECT().GetReleaseFn().Return(releaseFn).AnyTimes()
+	targetWorkflow.EXPECT().GetContext().Return(weContext).AnyTimes()
+	targetWorkflow.EXPECT().GetMutableState().Return(mutableState).AnyTimes()
+	targetWorkflow.EXPECT().GetReleaseFn().Return(releaseFn).AnyTimes()

-	workflowSnapshot := &persistence.WorkflowSnapshot{}
+	executionState := &persistencespb.WorkflowExecutionState{
+		RunId: runID,
+		State: enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE,
+	}
+	workflowSnapshot := &persistence.WorkflowSnapshot{
+		ExecutionState: executionState,
+	}
 	workflowEventsSeq := []*persistence.WorkflowEvents{{
 		Events: []*historypb.HistoryEvent{{
 			EventId: common.FirstEventID + rand.Int63(),
@@ -105,9 +358,8 @@ func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_BrandNew(
 		NamespaceId: namespaceID.String(),
 		WorkflowId:  workflowID,
 	}).AnyTimes()
-	mutableState.EXPECT().GetExecutionState().Return(&persistencespb.WorkflowExecutionState{
-		RunId: runID,
-	}).AnyTimes()
+	mutableState.EXPECT().GetExecutionState().Return(executionState).AnyTimes()
+	mutableState.EXPECT().GetReapplyCandidateEvents().Return(nil)
 	mutableState.EXPECT().CloseTransactionAsSnapshot(context.Background(), historyi.TransactionPolicyPassive).Return(
 		workflowSnapshot, workflowEventsSeq, nil,
 	)
@@ -116,20 +368,36 @@ func (s *transactionMgrForNewWorkflowSuite) TestDispatchForNewWorkflow_BrandNew(
 		ctx, namespaceID, workflowID, chasm.WorkflowArchetypeID,
 	).Return("", nil)

+	weContext.EXPECT().ReapplyEvents(gomock.Any(), s.mockShard, workflowEventsSeq).Return(nil)
 	weContext.EXPECT().CreateWorkflowExecution(
 		gomock.Any(),
 		s.mockShard,
-		persistence.CreateWorkflowModeBrandNew,
+		gomock.Any(),
 		"",
 		int64(0),
 		mutableState,
 		workflowSnapshot,
 		workflowEventsSeq,
 		gomock.Any(),
-	).Return(nil)
+	).DoAndReturn(func(
+		_ context.Context,
+		_ historyi.ShardContext,
+		createMode persistence.CreateWorkflowMode,
+		_ string,
+		_ int64,
+		_ historyi.MutableState,
+		workflowSnapshot *persistence.WorkflowSnapshot,
+		_ []*persistence.WorkflowEvents,
+		_ historyi.TransactionPolicy,
+	) error {
+		s.Equal(persistence.CreateWorkflowModeBypassCurrent, createMode)
+		s.Equal(enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE, workflowSnapshot.ExecutionState.State)
+		return persistence.ValidateCreateWorkflowModeState(createMode, *workflowSnapshot)
+	})

-	err := s.createMgr.dispatchForNewWorkflow(ctx, chasm.WorkflowArchetypeID, newWorkflow)
-	s.NoError(err)
+	err := s.createMgr.dispatchForNewWorkflow(ctx, chasm.WorkflowArchetypeID, targetWorkflow)
+	s.Require().NoError(err)
+	s.Equal(enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE, executionState.State)
 	s.True(releaseCalled)
 }