Handle delete history branch (#2591)

yux0 · web-flow · commit 4375f942860c · 2022-03-08T15:52:43.000-08:00
* Add branch token to workflow cleanup task
diff --git a/api/persistence/v1/executions.pb.go b/api/persistence/v1/executions.pb.go
diff --git a/common/persistence/persistence-tests/executionManagerTest.go b/common/persistence/persistence-tests/executionManagerTest.go
@@ -2392,10 +2392,10 @@ func (s *ExecutionManagerSuite) TestTimerTasksComplete() {
 	updatedState := copyWorkflowExecutionState(state0.ExecutionState)
 	updatedInfo.LastWorkflowTaskStartId = int64(2)
 	taskSlice := []tasks.Task{
-		&tasks.WorkflowTimeoutTask{workflowKey, now.Add(2 * time.Second), 2, 12},
-		&tasks.DeleteHistoryEventTask{workflowKey, now.Add(2 * time.Second), 3, 13},
-		&tasks.ActivityTimeoutTask{workflowKey, now.Add(3 * time.Second), 4, enumspb.TIMEOUT_TYPE_START_TO_CLOSE, 7, 0, 14},
-		&tasks.UserTimerTask{workflowKey, now.Add(3 * time.Second), 5, 7, 15},
+		&tasks.WorkflowTimeoutTask{WorkflowKey: workflowKey, VisibilityTimestamp: now.Add(2 * time.Second), TaskID: 2, Version: 12},
+		&tasks.DeleteHistoryEventTask{WorkflowKey: workflowKey, VisibilityTimestamp: now.Add(2 * time.Second), TaskID: 3, Version: 13},
+		&tasks.ActivityTimeoutTask{WorkflowKey: workflowKey, VisibilityTimestamp: now.Add(3 * time.Second), TaskID: 4, TimeoutType: enumspb.TIMEOUT_TYPE_START_TO_CLOSE, EventID: 7, Version: 14},
+		&tasks.UserTimerTask{WorkflowKey: workflowKey, VisibilityTimestamp: now.Add(3 * time.Second), TaskID: 5, EventID: 7, Version: 15},
 	}
 	err2 := s.UpdateWorkflowExecution(updatedInfo, updatedState, int64(5), []int64{int64(4)}, nil, int64(3), taskSlice, nil, nil, nil, nil)
 	s.NoError(err2)
@@ -2459,11 +2459,11 @@ func (s *ExecutionManagerSuite) TestTimerTasksRangeComplete() {
 	updatedState := copyWorkflowExecutionState(state0.ExecutionState)
 	updatedInfo.LastWorkflowTaskStartId = int64(2)
 	taskSlice := []tasks.Task{
-		&tasks.WorkflowTaskTimeoutTask{workflowKey, time.Now().UTC(), 1, 2, 3, enumspb.TIMEOUT_TYPE_START_TO_CLOSE, 11},
-		&tasks.WorkflowTimeoutTask{workflowKey, time.Now().UTC(), 2, 12},
-		&tasks.DeleteHistoryEventTask{workflowKey, time.Now().UTC(), 3, 13},
-		&tasks.ActivityTimeoutTask{workflowKey, time.Now().UTC(), 4, enumspb.TIMEOUT_TYPE_START_TO_CLOSE, 7, 0, 14},
-		&tasks.UserTimerTask{workflowKey, time.Now().UTC(), 5, 7, 15},
+		&tasks.WorkflowTaskTimeoutTask{WorkflowKey: workflowKey, VisibilityTimestamp: time.Now().UTC(), TaskID: 1, EventID: 2, ScheduleAttempt: 3, TimeoutType: enumspb.TIMEOUT_TYPE_START_TO_CLOSE, Version: 11},
+		&tasks.WorkflowTimeoutTask{WorkflowKey: workflowKey, VisibilityTimestamp: time.Now().UTC(), TaskID: 2, Version: 12},
+		&tasks.DeleteHistoryEventTask{WorkflowKey: workflowKey, VisibilityTimestamp: time.Now().UTC(), TaskID: 3, Version: 13},
+		&tasks.ActivityTimeoutTask{WorkflowKey: workflowKey, VisibilityTimestamp: time.Now().UTC(), TaskID: 4, TimeoutType: enumspb.TIMEOUT_TYPE_START_TO_CLOSE, EventID: 7, Version: 14},
+		&tasks.UserTimerTask{WorkflowKey: workflowKey, VisibilityTimestamp: time.Now().UTC(), TaskID: 5, EventID: 7, Version: 15},
 	}
 	err2 := s.UpdateWorkflowExecution(updatedInfo, updatedState, int64(5), []int64{int64(4)}, nil, int64(3), taskSlice, nil, nil, nil, nil)
 	s.NoError(err2)
diff --git a/common/persistence/serialization/task_serializer.go b/common/persistence/serialization/task_serializer.go
@@ -811,6 +811,7 @@ func (s *TaskSerializer) timerWorkflowCleanupTaskToProto(
 		EventId:             0,
 		TaskId:              workflowCleanupTimer.TaskID,
 		VisibilityTime:      &workflowCleanupTimer.VisibilityTimestamp,
+		BranchToken:         workflowCleanupTimer.BranchToken,
 	}
 }
 
@@ -826,6 +827,7 @@ func (s *TaskSerializer) timerWorkflowCleanupTaskFromProto(
 		VisibilityTimestamp: *workflowCleanupTimer.VisibilityTime,
 		TaskID:              workflowCleanupTimer.TaskId,
 		Version:             workflowCleanupTimer.Version,
+		BranchToken:         workflowCleanupTimer.BranchToken,
 	}
 }
 
diff --git a/common/persistence/serialization/task_serializer_test.go b/common/persistence/serialization/task_serializer_test.go
@@ -253,6 +253,7 @@ func (s *taskSerializerSuite) TestTimerWorkflowCleanupTask() {
 		VisibilityTimestamp: time.Unix(0, rand.Int63()).UTC(),
 		TaskID:              rand.Int63(),
 		Version:             rand.Int63(),
+		BranchToken:         []byte{123},
 	}
 
 	s.assertEqualTasks(workflowCleanupTimer)
diff --git a/proto/internal/temporal/server/api/persistence/v1/executions.proto b/proto/internal/temporal/server/api/persistence/v1/executions.proto
@@ -206,6 +206,7 @@ message TimerTaskInfo {
     int64 event_id = 9;
     int64 task_id = 10;
     google.protobuf.Timestamp visibility_time = 11 [(gogoproto.stdtime) = true];
+    bytes branch_token = 12;
 }
 
 // activity_map column
diff --git a/service/history/nDCTaskUtil.go b/service/history/nDCTaskUtil.go
@@ -42,35 +42,30 @@ const (
 	refreshTaskTimeout = 30 * time.Second
 )
 
-// verifyTaskVersion, will return true if failover version check is successful
-func verifyTaskVersion(
+// VerifyTaskVersion, will return true if failover version check is successful
+func VerifyTaskVersion(
 	shard shard.Context,
 	logger log.Logger,
-	namespaceID namespace.ID,
+	namespace *namespace.Namespace,
 	version int64,
 	taskVersion int64,
 	task interface{},
-) (bool, error) {
+) bool {
 
 	if !shard.GetClusterMetadata().IsGlobalNamespaceEnabled() {
-		return true, nil
+		return true
 	}
 
 	// the first return value is whether this task is valid for further processing
-	namespaceEntry, err := shard.GetNamespaceRegistry().GetNamespaceByID(namespaceID)
-	if err != nil {
-		logger.Debug("Cannot find namespaceID", tag.WorkflowNamespaceID(namespaceID.String()), tag.Error(err))
-		return false, err
-	}
-	if !namespaceEntry.IsGlobalNamespace() {
-		logger.Debug("NamespaceID is not active, task version check pass", tag.WorkflowNamespaceID(namespaceID.String()), tag.Task(task))
-		return true, nil
+	if !namespace.IsGlobalNamespace() {
+		logger.Debug("NamespaceID is not global, task version check pass", tag.WorkflowNamespaceID(namespace.ID().String()), tag.Task(task))
+		return true
 	} else if version != taskVersion {
-		logger.Debug("NamespaceID is active, task version != target version", tag.WorkflowNamespaceID(namespaceID.String()), tag.Task(task), tag.TaskVersion(version))
-		return false, nil
+		logger.Debug("NamespaceID is global, task version != target version", tag.WorkflowNamespaceID(namespace.ID().String()), tag.Task(task), tag.TaskVersion(version))
+		return false
 	}
-	logger.Debug("NamespaceID is active, task version == target version", tag.WorkflowNamespaceID(namespaceID.String()), tag.Task(task), tag.TaskVersion(version))
-	return true, nil
+	logger.Debug("NamespaceID is global, task version == target version", tag.WorkflowNamespaceID(namespace.ID().String()), tag.Task(task), tag.TaskVersion(version))
+	return true
 }
 
 // load mutable state, if mutable state's next event ID <= task ID, will attempt to refresh
@@ -81,7 +76,7 @@ func loadMutableStateForTransferTask(
 	metricsClient metrics.Client,
 	logger log.Logger,
 ) (workflow.MutableState, error) {
-	return loadMutableStateForTask(
+	return LoadMutableStateForTask(
 		context,
 		transferTask,
 		getTransferTaskEventIDAndRetryable,
@@ -98,7 +93,7 @@ func loadMutableStateForTimerTask(
 	metricsClient metrics.Client,
 	logger log.Logger,
 ) (workflow.MutableState, error) {
-	return loadMutableStateForTask(
+	return LoadMutableStateForTask(
 		context,
 		timerTask,
 		getTimerTaskEventIDAndRetryable,
@@ -107,7 +102,7 @@ func loadMutableStateForTimerTask(
 	)
 }
 
-func loadMutableStateForTask(
+func LoadMutableStateForTask(
 	context workflow.Context,
 	task tasks.Task,
 	taskEventIDAndRetryable func(task tasks.Task, executionInfo *persistencespb.WorkflowExecutionInfo) (int64, bool),
diff --git a/service/history/tasks/workflow_cleanup_timer.go b/service/history/tasks/workflow_cleanup_timer.go
@@ -39,6 +39,7 @@ type (
 		VisibilityTimestamp time.Time
 		TaskID              int64
 		Version             int64
+		BranchToken         []byte
 	}
 )
 
diff --git a/service/history/timerQueueActiveTaskExecutor.go b/service/history/timerQueueActiveTaskExecutor.go
@@ -314,9 +314,9 @@ func (t *timerQueueActiveTaskExecutor) executeWorkflowTaskTimeoutTask(
 	if !ok {
 		return nil
 	}
-	ok, err = verifyTaskVersion(t.shard, t.logger, namespace.ID(task.NamespaceID), workflowTask.Version, task.Version, task)
-	if err != nil || !ok {
-		return err
+	ok = VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), workflowTask.Version, task.Version, task)
+	if !ok {
+		return nil
 	}
 
 	if workflowTask.Attempt != task.ScheduleAttempt {
@@ -447,9 +447,9 @@ func (t *timerQueueActiveTaskExecutor) executeActivityRetryTimerTask(
 		}
 		return nil
 	}
-	ok, err = verifyTaskVersion(t.shard, t.logger, namespace.ID(task.NamespaceID), activityInfo.Version, task.Version, task)
-	if err != nil || !ok {
-		return err
+	ok = VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), activityInfo.Version, task.Version, task)
+	if !ok {
+		return nil
 	}
 
 	targetNamespaceID := activityInfo.NamespaceId
@@ -509,9 +509,9 @@ func (t *timerQueueActiveTaskExecutor) executeWorkflowTimeoutTask(
 	if err != nil {
 		return err
 	}
-	ok, err := verifyTaskVersion(t.shard, t.logger, namespace.ID(task.NamespaceID), startVersion, task.Version, task)
-	if err != nil || !ok {
-		return err
+	ok := VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), startVersion, task.Version, task)
+	if !ok {
+		return nil
 	}
 
 	eventBatchFirstEventID := mutableState.GetNextEventID()
diff --git a/service/history/timerQueueStandbyTaskExecutor.go b/service/history/timerQueueStandbyTaskExecutor.go
@@ -291,9 +291,9 @@ func (t *timerQueueStandbyTaskExecutor) executeWorkflowTaskTimeoutTask(
 			return nil, nil
 		}
 
-		ok, err := verifyTaskVersion(t.shard, t.logger, namespace.ID(timerTask.NamespaceID), workflowTask.Version, timerTask.Version, timerTask)
-		if err != nil || !ok {
-			return nil, err
+		ok := VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), workflowTask.Version, timerTask.Version, timerTask)
+		if !ok {
+			return nil, nil
 		}
 
 		return getHistoryResendInfo(mutableState)
@@ -369,9 +369,9 @@ func (t *timerQueueStandbyTaskExecutor) executeWorkflowTimeoutTask(
 		if err != nil {
 			return nil, err
 		}
-		ok, err := verifyTaskVersion(t.shard, t.logger, namespace.ID(timerTask.NamespaceID), startVersion, timerTask.Version, timerTask)
-		if err != nil || !ok {
-			return nil, err
+		ok := VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), startVersion, timerTask.Version, timerTask)
+		if !ok {
+			return nil, nil
 		}
 
 		return getHistoryResendInfo(mutableState)
diff --git a/service/history/timerQueueTaskExecutorBase.go b/service/history/timerQueueTaskExecutorBase.go
@@ -25,13 +25,16 @@
 package history
 
 import (
+	"bytes"
 	"context"
 
 	commonpb "go.temporal.io/api/common/v1"
-
+	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
+	"go.temporal.io/server/common/persistence"
 	"go.temporal.io/server/service/history/configs"
 	"go.temporal.io/server/service/history/shard"
 	"go.temporal.io/server/service/history/tasks"
@@ -91,17 +94,35 @@ func (t *timerQueueTaskExecutorBase) executeDeleteHistoryEventTask(
 	defer func() { release(retError) }()
 
 	mutableState, err := loadMutableStateForTimerTask(weContext, task, t.metricsClient, t.logger)
-	if err != nil {
+	switch err.(type) {
+	case nil:
+		if mutableState == nil {
+			return nil
+		}
+	case *serviceerror.NotFound:
+		// the mutable state is deleted and delete history branch operation failed.
+		// use task branch token to delete the leftover history branch
+		return t.deleteHistoryBranch(task.BranchToken)
+	default:
 		return err
 	}
 
 	lastWriteVersion, err := mutableState.GetLastWriteVersion()
 	if err != nil {
 		return err
 	}
-	ok, err := verifyTaskVersion(t.shard, t.logger, namespace.ID(task.NamespaceID), lastWriteVersion, task.Version, task)
-	if err != nil || !ok {
-		return err
+	if ok := VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), lastWriteVersion, task.Version, task); !ok {
+		currentBranchToken, err := mutableState.GetCurrentBranchToken()
+		if err != nil {
+			return err
+		}
+		// the mutable state has a newer version and the branch token is updated
+		// use task branch token to delete the original branch
+		if !bytes.Equal(task.BranchToken, currentBranchToken) {
+			return t.deleteHistoryBranch(task.BranchToken)
+		}
+		t.logger.Error("Different mutable state versions have the same branch token", tag.TaskVersion(task.Version), tag.LastEventVersion(lastWriteVersion))
+		return serviceerror.NewInternal("Mutable state has different version but same branch token")
 	}
 
 	return t.deleteManager.DeleteWorkflowExecutionByRetention(
@@ -122,3 +143,13 @@ func (t *timerQueueTaskExecutorBase) getNamespaceIDAndWorkflowExecution(
 		RunId:      task.GetRunID(),
 	}
 }
+
+func (t *timerQueueTaskExecutorBase) deleteHistoryBranch(branchToken []byte) error {
+	if len(branchToken) > 0 {
+		return t.shard.GetExecutionManager().DeleteHistoryBranch(&persistence.DeleteHistoryBranchRequest{
+			ShardID:     t.shard.GetShardID(),
+			BranchToken: branchToken,
+		})
+	}
+	return nil
+}
diff --git a/service/history/timerQueueTaskExecutorBase_test.go b/service/history/timerQueueTaskExecutorBase_test.go
@@ -128,6 +128,7 @@ func (s *timerQueueTaskExecutorBaseSuite) Test_executeDeleteHistoryEventTask_NoE
 	mockMutableState.EXPECT().GetLastWriteVersion().Return(int64(1), nil)
 	mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{})
 	mockMutableState.EXPECT().GetNextEventID().Return(int64(2))
+	mockMutableState.EXPECT().GetNamespaceEntry().Return(tests.LocalNamespaceEntry)
 	s.testShardContext.Resource.ClusterMetadata.EXPECT().IsGlobalNamespaceEnabled().Return(false)
 
 	s.mockDeleteManager.EXPECT().DeleteWorkflowExecutionByRetention(
@@ -169,6 +170,7 @@ func (s *timerQueueTaskExecutorBaseSuite) TestArchiveHistory_DeleteFailed() {
 	mockMutableState.EXPECT().GetLastWriteVersion().Return(int64(1), nil)
 	mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{})
 	mockMutableState.EXPECT().GetNextEventID().Return(int64(2))
+	mockMutableState.EXPECT().GetNamespaceEntry().Return(tests.LocalNamespaceEntry)
 	s.testShardContext.Resource.ClusterMetadata.EXPECT().IsGlobalNamespaceEnabled().Return(false)
 
 	s.mockDeleteManager.EXPECT().DeleteWorkflowExecutionByRetention(
diff --git a/service/history/transferQueueActiveTaskExecutor.go b/service/history/transferQueueActiveTaskExecutor.go
@@ -170,9 +170,9 @@ func (t *transferQueueActiveTaskExecutor) processActivityTask(
 	if !ok {
 		return nil
 	}
-	ok, err = verifyTaskVersion(t.shard, t.logger, namespace.ID(task.NamespaceID), ai.Version, task.Version, task)
-	if err != nil || !ok {
-		return err
+	ok = VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), ai.Version, task.Version, task)
+	if !ok {
+		return nil
 	}
 
 	timeout := timestamp.DurationValue(ai.ScheduleToStartTimeout)
@@ -215,9 +215,9 @@ func (t *transferQueueActiveTaskExecutor) processWorkflowTask(
 	if !found {
 		return nil
 	}
-	ok, err := verifyTaskVersion(t.shard, t.logger, namespace.ID(task.NamespaceID), workflowTask.Version, task.Version, task)
-	if err != nil || !ok {
-		return err
+	ok := VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), workflowTask.Version, task.Version, task)
+	if !ok {
+		return nil
 	}
 
 	executionInfo := mutableState.GetExecutionInfo()
@@ -284,9 +284,9 @@ func (t *transferQueueActiveTaskExecutor) processCloseExecution(
 	if err != nil {
 		return err
 	}
-	ok, err := verifyTaskVersion(t.shard, t.logger, namespace.ID(task.NamespaceID), lastWriteVersion, task.Version, task)
-	if err != nil || !ok {
-		return err
+	ok := VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), lastWriteVersion, task.Version, task)
+	if !ok {
+		return nil
 	}
 
 	executionInfo := mutableState.GetExecutionInfo()
@@ -402,9 +402,9 @@ func (t *transferQueueActiveTaskExecutor) processCancelExecution(
 	if !ok {
 		return nil
 	}
-	ok, err = verifyTaskVersion(t.shard, t.logger, namespace.ID(task.NamespaceID), requestCancelInfo.Version, task.Version, task)
-	if err != nil || !ok {
-		return err
+	ok = VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), requestCancelInfo.Version, task.Version, task)
+	if !ok {
+		return nil
 	}
 
 	targetNamespaceEntry, err := t.shard.GetNamespaceRegistry().GetNamespaceByID(namespace.ID(task.TargetNamespaceID))
@@ -490,9 +490,9 @@ func (t *transferQueueActiveTaskExecutor) processSignalExecution(
 		// To do that, probably need to add the SignalRequestID in transfer task.
 		return nil
 	}
-	ok, err = verifyTaskVersion(t.shard, t.logger, namespace.ID(task.NamespaceID), signalInfo.Version, task.Version, task)
-	if err != nil || !ok {
-		return err
+	ok = VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), signalInfo.Version, task.Version, task)
+	if !ok {
+		return nil
 	}
 
 	initiatedEvent, err := mutableState.GetSignalExternalInitiatedEvent(task.InitiatedID)
@@ -630,9 +630,9 @@ func (t *transferQueueActiveTaskExecutor) processStartChildExecution(
 	if !ok {
 		return nil
 	}
-	ok, err = verifyTaskVersion(t.shard, t.logger, namespace.ID(task.NamespaceID), childInfo.Version, task.Version, task)
-	if err != nil || !ok {
-		return err
+	ok = VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), childInfo.Version, task.Version, task)
+	if !ok {
+		return nil
 	}
 
 	initiatedEvent, err := mutableState.GetChildExecutionInitiatedEvent(task.InitiatedID)
@@ -752,9 +752,9 @@ func (t *transferQueueActiveTaskExecutor) processResetWorkflow(
 	if err != nil {
 		return err
 	}
-	ok, err := verifyTaskVersion(t.shard, t.logger, namespace.ID(task.NamespaceID), currentStartVersion, task.Version, task)
-	if err != nil || !ok {
-		return err
+	ok := VerifyTaskVersion(t.shard, t.logger, currentMutableState.GetNamespaceEntry(), currentStartVersion, task.Version, task)
+	if !ok {
+		return nil
 	}
 
 	executionInfo := currentMutableState.GetExecutionInfo()
diff --git a/service/history/transferQueueStandbyTaskExecutor.go b/service/history/transferQueueStandbyTaskExecutor.go
diff --git a/service/history/transferQueueTaskExecutorBase.go b/service/history/transferQueueTaskExecutorBase.go
diff --git a/service/history/visibilityQueueTaskExecutor.go b/service/history/visibilityQueueTaskExecutor.go
diff --git a/service/history/workflow/task_generator.go b/service/history/workflow/task_generator.go

Original file line number	Diff line number	Diff line change
`@@ -811,6 +811,7 @@ func (s *TaskSerializer) timerWorkflowCleanupTaskToProto(`
`811`	`811`	`EventId: 0,`
`812`	`812`	`TaskId: workflowCleanupTimer.TaskID,`
`813`	`813`	`VisibilityTime: &workflowCleanupTimer.VisibilityTimestamp,`
	`814`	`+ BranchToken: workflowCleanupTimer.BranchToken,`
`814`	`815`	`}`
`815`	`816`	`}`
`816`	`817`
`@@ -826,6 +827,7 @@ func (s *TaskSerializer) timerWorkflowCleanupTaskFromProto(`
`826`	`827`	`VisibilityTimestamp: *workflowCleanupTimer.VisibilityTime,`
`827`	`828`	`TaskID: workflowCleanupTimer.TaskId,`
`828`	`829`	`Version: workflowCleanupTimer.Version,`
	`830`	`+ BranchToken: workflowCleanupTimer.BranchToken,`
`829`	`831`	`}`
`830`	`832`	`}`
`831`	`833`
Original file line number	Diff line number	Diff line change
`@@ -253,6 +253,7 @@ func (s *taskSerializerSuite) TestTimerWorkflowCleanupTask() {`
`253`	`253`	`VisibilityTimestamp: time.Unix(0, rand.Int63()).UTC(),`
`254`	`254`	`TaskID: rand.Int63(),`
`255`	`255`	`Version: rand.Int63(),`
	`256`	`+ BranchToken: []byte{123},`
`256`	`257`	`}`
`257`	`258`
`258`	`259`	`s.assertEqualTasks(workflowCleanupTimer)`
Original file line number	Diff line number	Diff line change
`@@ -206,6 +206,7 @@ message TimerTaskInfo {`
`206`	`206`	`int64 event_id = 9;`
`207`	`207`	`int64 task_id = 10;`
`208`	`208`	`google.protobuf.Timestamp visibility_time = 11 [(gogoproto.stdtime) = true];`
	`209`	`+ bytes branch_token = 12;`
`209`	`210`	`}`
`210`	`211`
`211`	`212`	`// activity_map column`
Original file line number	Diff line number	Diff line change
`@@ -39,6 +39,7 @@ type (`
`39`	`39`	`VisibilityTimestamp time.Time`
`40`	`40`	`TaskID int64`
`41`	`41`	`Version int64`
	`42`	`+ BranchToken []byte`
`42`	`43`	`}`
`43`	`44`	`)`
`44`	`45`
Original file line number	Diff line number	Diff line change
`@@ -314,9 +314,9 @@ func (t *timerQueueActiveTaskExecutor) executeWorkflowTaskTimeoutTask(`
`314`	`314`	`if !ok {`
`315`	`315`	`return nil`
`316`	`316`	`}`
`317`		`- ok, err = verifyTaskVersion(t.shard, t.logger, namespace.ID(task.NamespaceID), workflowTask.Version, task.Version, task)`
`318`		`- if err != nil \|\| !ok {`
`319`		`- return err`
	`317`	`+ ok = VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), workflowTask.Version, task.Version, task)`
	`318`	`+ if !ok {`
	`319`	`+ return nil`
`320`	`320`	`}`
`321`	`321`
`322`	`322`	`if workflowTask.Attempt != task.ScheduleAttempt {`
`@@ -447,9 +447,9 @@ func (t *timerQueueActiveTaskExecutor) executeActivityRetryTimerTask(`
`447`	`447`	`}`
`448`	`448`	`return nil`
`449`	`449`	`}`
`450`		`- ok, err = verifyTaskVersion(t.shard, t.logger, namespace.ID(task.NamespaceID), activityInfo.Version, task.Version, task)`
`451`		`- if err != nil \|\| !ok {`
`452`		`- return err`
	`450`	`+ ok = VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), activityInfo.Version, task.Version, task)`
	`451`	`+ if !ok {`
	`452`	`+ return nil`
`453`	`453`	`}`
`454`	`454`
`455`	`455`	`targetNamespaceID := activityInfo.NamespaceId`
`@@ -509,9 +509,9 @@ func (t *timerQueueActiveTaskExecutor) executeWorkflowTimeoutTask(`
`509`	`509`	`if err != nil {`
`510`	`510`	`return err`
`511`	`511`	`}`
`512`		`- ok, err := verifyTaskVersion(t.shard, t.logger, namespace.ID(task.NamespaceID), startVersion, task.Version, task)`
`513`		`- if err != nil \|\| !ok {`
`514`		`- return err`
	`512`	`+ ok := VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), startVersion, task.Version, task)`
	`513`	`+ if !ok {`
	`514`	`+ return nil`
`515`	`515`	`}`
`516`	`516`
`517`	`517`	`eventBatchFirstEventID := mutableState.GetNextEventID()`
Original file line number	Diff line number	Diff line change
`@@ -291,9 +291,9 @@ func (t *timerQueueStandbyTaskExecutor) executeWorkflowTaskTimeoutTask(`
`291`	`291`	`return nil, nil`
`292`	`292`	`}`
`293`	`293`
`294`		`- ok, err := verifyTaskVersion(t.shard, t.logger, namespace.ID(timerTask.NamespaceID), workflowTask.Version, timerTask.Version, timerTask)`
`295`		`- if err != nil \|\| !ok {`
`296`		`- return nil, err`
	`294`	`+ ok := VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), workflowTask.Version, timerTask.Version, timerTask)`
	`295`	`+ if !ok {`
	`296`	`+ return nil, nil`
`297`	`297`	`}`
`298`	`298`
`299`	`299`	`return getHistoryResendInfo(mutableState)`
`@@ -369,9 +369,9 @@ func (t *timerQueueStandbyTaskExecutor) executeWorkflowTimeoutTask(`
`369`	`369`	`if err != nil {`
`370`	`370`	`return nil, err`
`371`	`371`	`}`
`372`		`- ok, err := verifyTaskVersion(t.shard, t.logger, namespace.ID(timerTask.NamespaceID), startVersion, timerTask.Version, timerTask)`
`373`		`- if err != nil \|\| !ok {`
`374`		`- return nil, err`
	`372`	`+ ok := VerifyTaskVersion(t.shard, t.logger, mutableState.GetNamespaceEntry(), startVersion, timerTask.Version, timerTask)`
	`373`	`+ if !ok {`
	`374`	`+ return nil, nil`
`375`	`375`	`}`
`376`	`376`
`377`	`377`	`return getHistoryResendInfo(mutableState)`