diff --git a/service/history/api/reapplyevents/api.go b/service/history/api/reapplyevents/api.go index 06d9d796250..6233a6eca28 100644 --- a/service/history/api/reapplyevents/api.go +++ b/service/history/api/reapplyevents/api.go @@ -6,10 +6,10 @@ import ( "github.com/google/uuid" historypb "go.temporal.io/api/history/v1" "go.temporal.io/api/serviceerror" + enumsspb "go.temporal.io/server/api/enums/v1" "go.temporal.io/server/common" "go.temporal.io/server/common/definition" "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/versionhistory" "go.temporal.io/server/service/history/api" @@ -86,21 +86,25 @@ func Invoke( baseRunID := mutableState.GetExecutionState().GetRunId() resetRunID := uuid.New() baseRebuildLastEventID := mutableState.GetLastCompletedWorkflowTaskStartedEventId() - - // TODO when https://github.com/uber/cadence/issues/2420 is finished, remove this block, - // since cannot reapply event to a finished workflow which had no workflow tasks started if baseRebuildLastEventID == common.EmptyEventID { - shardContext.GetLogger().Warn("cannot reapply event to a finished workflow with no workflow task", - tag.WorkflowNamespaceID(namespaceID.String()), - tag.WorkflowID(workflowID), - ) - metrics.EventReapplySkippedCount.With(shardContext.GetMetricsHandler()).Record( - 1, - metrics.OperationTag(metrics.HistoryReapplyEventsScope)) - return &api.UpdateWorkflowAction{ - Noop: true, - CreateWorkflowTask: false, - }, nil + // No completed workflow task. Pick the reset anchor by scenario: + // - real pending workflow task (scheduled/started): anchor at it. Resetting + // to a pending task is already supported by the resetter (it fails it). + // - transient (failing, attempt > 1) or speculative pending task: not a + // usable anchor - it has no persisted WorkflowTaskScheduled event (its + // ScheduledEventID is a not-yet-written NextEventID placeholder), so skip it. + // - no workflow task at all: nothing to anchor on. + // In the latter two, baseRebuildLastEventID stays EmptyEventID and the + // version-history lookup below fails the reapply. + if workflowTask := mutableState.GetPendingWorkflowTask(); workflowTask != nil && + !mutableState.IsTransientWorkflowTask() && + workflowTask.Type != enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE { + if workflowTask.StartedEventID != common.EmptyEventID { + baseRebuildLastEventID = workflowTask.StartedEventID + } else { + baseRebuildLastEventID = workflowTask.ScheduledEventID + } + } } baseVersionHistories := mutableState.GetExecutionInfo().GetVersionHistories() diff --git a/service/history/ndc/transaction_manager.go b/service/history/ndc/transaction_manager.go index 3df1c64bf4e..0e5b25fe301 100644 --- a/service/history/ndc/transaction_manager.go +++ b/service/history/ndc/transaction_manager.go @@ -9,7 +9,9 @@ import ( commonpb "go.temporal.io/api/common/v1" historypb "go.temporal.io/api/history/v1" "go.temporal.io/api/serviceerror" + enumsspb "go.temporal.io/server/api/enums/v1" "go.temporal.io/server/chasm" + "go.temporal.io/server/common" "go.temporal.io/server/common/cluster" "go.temporal.io/server/common/locks" "go.temporal.io/server/common/log" @@ -309,6 +311,25 @@ func (r *transactionMgrImpl) backfillWorkflowEventsReapply( baseRunID := baseMutableState.GetExecutionState().GetRunId() resetRunID := uuid.NewString() baseRebuildLastEventID := baseMutableState.GetLastCompletedWorkflowTaskStartedEventId() + if baseRebuildLastEventID == common.EmptyEventID { + // No completed workflow task. Pick the reset anchor by scenario: + // - real pending workflow task (scheduled/started): anchor at it. Resetting to a + // pending task is already supported by the resetter. + // - transient (failing, attempt > 1) or speculative pending task: not a usable + // anchor - it has no persisted WorkflowTaskScheduled event (its ScheduledEventID + // is a not-yet-written NextEventID placeholder), so skip it. + // - no workflow task at all: nothing to anchor on. + if workflowTask := baseMutableState.GetPendingWorkflowTask(); workflowTask != nil && + !baseMutableState.IsTransientWorkflowTask() && + workflowTask.Type != enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE { + if workflowTask.StartedEventID != common.EmptyEventID { + baseRebuildLastEventID = workflowTask.StartedEventID + } else { + baseRebuildLastEventID = workflowTask.ScheduledEventID + } + } + } + baseVersionHistories := baseMutableState.GetExecutionInfo().GetVersionHistories() baseCurrentVersionHistory, err := versionhistory.GetCurrentVersionHistory(baseVersionHistories) if err != nil { diff --git a/service/history/ndc/transaction_manager_test.go b/service/history/ndc/transaction_manager_test.go index e5fa38043de..74a593fc11c 100644 --- a/service/history/ndc/transaction_manager_test.go +++ b/service/history/ndc/transaction_manager_test.go @@ -11,9 +11,11 @@ import ( enumspb "go.temporal.io/api/enums/v1" historypb "go.temporal.io/api/history/v1" "go.temporal.io/api/serviceerror" + enumsspb "go.temporal.io/server/api/enums/v1" historyspb "go.temporal.io/server/api/history/v1" persistencespb "go.temporal.io/server/api/persistence/v1" "go.temporal.io/server/chasm" + "go.temporal.io/server/common" "go.temporal.io/server/common/cluster" "go.temporal.io/server/common/log" "go.temporal.io/server/common/metrics" @@ -334,6 +336,410 @@ func (s *transactionMgrSuite) TestBackfillWorkflow_CurrentWorkflow_Closed_ResetF s.True(releaseCalled) } +func (s *transactionMgrSuite) TestBackfillWorkflow_CurrentWorkflow_Closed_ResetError() { + // When ResetWorkflow fails with a non-InvalidArgument error, the reapply is not swallowed: + // BackfillWorkflow propagates the error so the replication task is retried. + ctx := context.Background() + + namespaceID := namespace.ID("some random namespace ID") + workflowID := "some random workflow ID" + runID := "some random run ID" + LastCompletedWorkflowTaskStartedEventID := int64(9999) + nextEventID := LastCompletedWorkflowTaskStartedEventID * 2 + lastWorkflowTaskStartedVersion := s.namespaceEntry.FailoverVersion(workflowID) + versionHistory := versionhistory.NewVersionHistory([]byte("branch token"), []*historyspb.VersionHistoryItem{ + {EventId: LastCompletedWorkflowTaskStartedEventID, Version: lastWorkflowTaskStartedVersion}, + }) + histories := versionhistory.NewVersionHistories(versionHistory) + + releaseCalled := false + + targetWorkflow := NewMockWorkflow(s.controller) + weContext := historyi.NewMockWorkflowContext(s.controller) + mutableState := historyi.NewMockMutableState(s.controller) + var releaseFn historyi.ReleaseWorkflowContextFunc = func(error) { releaseCalled = true } + + workflowEvents := &persistence.WorkflowEvents{} + historySize := rand.Int63() + + targetWorkflow.EXPECT().GetContext().Return(weContext).AnyTimes() + targetWorkflow.EXPECT().GetMutableState().Return(mutableState).AnyTimes() + targetWorkflow.EXPECT().GetReleaseFn().Return(releaseFn).AnyTimes() + + s.mockClusterMetadata.EXPECT().ClusterNameForFailoverVersion(s.namespaceEntry.IsGlobalNamespace(), s.namespaceEntry.FailoverVersion(workflowID)).Return(cluster.TestCurrentClusterName).AnyTimes() + s.mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes() + + mutableState.EXPECT().IsCurrentWorkflowGuaranteed().Return(false).AnyTimes() + mutableState.EXPECT().IsWorkflowExecutionRunning().Return(false).AnyTimes() + mutableState.EXPECT().GetNamespaceEntry().Return(s.namespaceEntry).AnyTimes() + mutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{ + NamespaceId: namespaceID.String(), + WorkflowId: workflowID, + VersionHistories: histories, + }).AnyTimes() + mutableState.EXPECT().GetExecutionState().Return(&persistencespb.WorkflowExecutionState{ + RunId: runID, + }).AnyTimes() + mutableState.EXPECT().GetNextEventID().Return(nextEventID).AnyTimes() + mutableState.EXPECT().GetLastCompletedWorkflowTaskStartedEventId().Return(LastCompletedWorkflowTaskStartedEventID) + mutableState.EXPECT().AddHistorySize(historySize) + + s.mockWorkflowResetter.EXPECT().ResetWorkflow( + ctx, + namespaceID, + workflowID, + runID, + versionHistory.GetBranchToken(), + LastCompletedWorkflowTaskStartedEventID, + lastWorkflowTaskStartedVersion, + nextEventID, + gomock.Any(), + targetWorkflow, + targetWorkflow, + EventsReapplicationResetWorkflowReason, + workflowEvents.Events, + nil, + false, // allowResetWithPendingChildren + nil, // post reset operations + ).Return(serviceerror.NewInternal("reset boom")) + + s.mockExecutionMgr.EXPECT().GetCurrentExecution(gomock.Any(), &persistence.GetCurrentExecutionRequest{ + ShardID: s.mockShard.GetShardID(), + NamespaceID: namespaceID.String(), + WorkflowID: workflowID, + ArchetypeID: chasm.WorkflowArchetypeID, + }).Return(&persistence.GetCurrentExecutionResponse{RunID: runID}, nil) + + // UpdateWorkflowExecutionWithNew must not be invoked: the reset error is propagated first. + weContext.EXPECT().PersistWorkflowEvents(gomock.Any(), s.mockShard, workflowEvents).Return(historySize, nil) + + err := s.transactionMgr.BackfillWorkflow(ctx, targetWorkflow, workflowEvents) + s.Error(err) + s.True(releaseCalled) +} + +func (s *transactionMgrSuite) TestBackfillWorkflow_CurrentWorkflow_Closed_NoWorkflowTaskBoundary() { + // A closed current workflow with no completed workflow task and no usable pending workflow + // task (never had one, or its only task was started then cleared on close) has no + // workflow-task boundary to anchor on. There is nothing to reset to, so the reapply errors + // out (the version-history lookup of EmptyEventID fails) and the replication task is retried. + ctx := context.Background() + + namespaceID := namespace.ID("some random namespace ID") + workflowID := "some random workflow ID" + runID := "some random run ID" + nextEventID := int64(2) + lastWorkflowTaskStartedVersion := s.namespaceEntry.FailoverVersion(workflowID) + versionHistory := versionhistory.NewVersionHistory([]byte("branch token"), []*historyspb.VersionHistoryItem{ + {EventId: common.FirstEventID, Version: lastWorkflowTaskStartedVersion}, + }) + histories := versionhistory.NewVersionHistories(versionHistory) + historySize := rand.Int63() + + releaseCalled := false + + targetWorkflow := NewMockWorkflow(s.controller) + weContext := historyi.NewMockWorkflowContext(s.controller) + mutableState := historyi.NewMockMutableState(s.controller) + var releaseFn historyi.ReleaseWorkflowContextFunc = func(error) { releaseCalled = true } + + workflowEvents := &persistence.WorkflowEvents{} + + targetWorkflow.EXPECT().GetContext().Return(weContext).AnyTimes() + targetWorkflow.EXPECT().GetMutableState().Return(mutableState).AnyTimes() + targetWorkflow.EXPECT().GetReleaseFn().Return(releaseFn).AnyTimes() + + s.mockClusterMetadata.EXPECT().ClusterNameForFailoverVersion(s.namespaceEntry.IsGlobalNamespace(), s.namespaceEntry.FailoverVersion(workflowID)).Return(cluster.TestCurrentClusterName).AnyTimes() + s.mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes() + + mutableState.EXPECT().IsCurrentWorkflowGuaranteed().Return(false).AnyTimes() + mutableState.EXPECT().IsWorkflowExecutionRunning().Return(false).AnyTimes() + mutableState.EXPECT().GetNamespaceEntry().Return(s.namespaceEntry).AnyTimes() + mutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{ + NamespaceId: namespaceID.String(), + WorkflowId: workflowID, + VersionHistories: histories, + }).AnyTimes() + mutableState.EXPECT().GetExecutionState().Return(&persistencespb.WorkflowExecutionState{ + RunId: runID, + }).AnyTimes() + mutableState.EXPECT().GetNextEventID().Return(nextEventID).AnyTimes() + // No completed workflow task and no pending one: no boundary to anchor on. + mutableState.EXPECT().GetLastCompletedWorkflowTaskStartedEventId().Return(common.EmptyEventID) + mutableState.EXPECT().GetPendingWorkflowTask().Return(nil).AnyTimes() + mutableState.EXPECT().AddHistorySize(historySize) + + // ResetWorkflow must not be invoked: the EmptyEventID version-history lookup errors out first. + s.mockExecutionMgr.EXPECT().GetCurrentExecution(gomock.Any(), &persistence.GetCurrentExecutionRequest{ + ShardID: s.mockShard.GetShardID(), + NamespaceID: namespaceID.String(), + WorkflowID: workflowID, + ArchetypeID: chasm.WorkflowArchetypeID, + }).Return(&persistence.GetCurrentExecutionResponse{RunID: runID}, nil) + + weContext.EXPECT().PersistWorkflowEvents(gomock.Any(), s.mockShard, workflowEvents).Return(historySize, nil) + + err := s.transactionMgr.BackfillWorkflow(ctx, targetWorkflow, workflowEvents) + s.Error(err) + s.True(releaseCalled) +} + +func (s *transactionMgrSuite) TestBackfillWorkflow_CurrentWorkflow_Closed_PendingWorkflowTask() { + // A closed current workflow that never completed a workflow task but still has a real, + // pending (scheduled, never started) workflow task is anchored at that scheduled task, so + // the resetter can rebuild to it and reapply the events onto a new run. + ctx := context.Background() + + namespaceID := namespace.ID("some random namespace ID") + workflowID := "some random workflow ID" + runID := "some random run ID" + scheduledEventID := int64(2) + nextEventID := int64(4) + scheduledEventVersion := s.namespaceEntry.FailoverVersion(workflowID) + versionHistory := versionhistory.NewVersionHistory([]byte("branch token"), []*historyspb.VersionHistoryItem{ + {EventId: scheduledEventID, Version: scheduledEventVersion}, + }) + histories := versionhistory.NewVersionHistories(versionHistory) + historySize := rand.Int63() + + releaseCalled := false + + targetWorkflow := NewMockWorkflow(s.controller) + weContext := historyi.NewMockWorkflowContext(s.controller) + mutableState := historyi.NewMockMutableState(s.controller) + var releaseFn historyi.ReleaseWorkflowContextFunc = func(error) { releaseCalled = true } + + workflowEvents := &persistence.WorkflowEvents{} + + targetWorkflow.EXPECT().GetContext().Return(weContext).AnyTimes() + targetWorkflow.EXPECT().GetMutableState().Return(mutableState).AnyTimes() + targetWorkflow.EXPECT().GetReleaseFn().Return(releaseFn).AnyTimes() + + s.mockClusterMetadata.EXPECT().ClusterNameForFailoverVersion(s.namespaceEntry.IsGlobalNamespace(), s.namespaceEntry.FailoverVersion(workflowID)).Return(cluster.TestCurrentClusterName).AnyTimes() + s.mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes() + + mutableState.EXPECT().IsCurrentWorkflowGuaranteed().Return(false).AnyTimes() + mutableState.EXPECT().IsWorkflowExecutionRunning().Return(false).AnyTimes() + mutableState.EXPECT().GetNamespaceEntry().Return(s.namespaceEntry).AnyTimes() + mutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{ + NamespaceId: namespaceID.String(), + WorkflowId: workflowID, + VersionHistories: histories, + }).AnyTimes() + mutableState.EXPECT().GetExecutionState().Return(&persistencespb.WorkflowExecutionState{ + RunId: runID, + }).AnyTimes() + mutableState.EXPECT().GetNextEventID().Return(nextEventID).AnyTimes() + // No completed workflow task, but a real scheduled (never started) one survives the close. + mutableState.EXPECT().GetLastCompletedWorkflowTaskStartedEventId().Return(common.EmptyEventID) + mutableState.EXPECT().IsTransientWorkflowTask().Return(false).AnyTimes() + mutableState.EXPECT().GetPendingWorkflowTask().Return(&historyi.WorkflowTaskInfo{ + ScheduledEventID: scheduledEventID, + StartedEventID: common.EmptyEventID, + Type: enumsspb.WORKFLOW_TASK_TYPE_NORMAL, + }).AnyTimes() + mutableState.EXPECT().AddHistorySize(historySize) + + // Reset must be anchored at the scheduled workflow task's event. + s.mockWorkflowResetter.EXPECT().ResetWorkflow( + ctx, + namespaceID, + workflowID, + runID, + versionHistory.GetBranchToken(), + scheduledEventID, + scheduledEventVersion, + nextEventID, + gomock.Any(), + targetWorkflow, + targetWorkflow, + EventsReapplicationResetWorkflowReason, + workflowEvents.Events, + nil, + false, // allowResetWithPendingChildren + nil, // post reset operations + ).Return(nil) + + s.mockExecutionMgr.EXPECT().GetCurrentExecution(gomock.Any(), &persistence.GetCurrentExecutionRequest{ + ShardID: s.mockShard.GetShardID(), + NamespaceID: namespaceID.String(), + WorkflowID: workflowID, + ArchetypeID: chasm.WorkflowArchetypeID, + }).Return(&persistence.GetCurrentExecutionResponse{RunID: runID}, nil) + + weContext.EXPECT().PersistWorkflowEvents(gomock.Any(), s.mockShard, workflowEvents).Return(historySize, nil) + weContext.EXPECT().UpdateWorkflowExecutionWithNew( + gomock.Any(), s.mockShard, persistence.UpdateWorkflowModeBypassCurrent, nil, nil, historyi.TransactionPolicyPassive, (*historyi.TransactionPolicy)(nil), + ).Return(nil) + + err := s.transactionMgr.BackfillWorkflow(ctx, targetWorkflow, workflowEvents) + s.NoError(err) + s.True(releaseCalled) +} + +func (s *transactionMgrSuite) TestBackfillWorkflow_CurrentWorkflow_Closed_StartedPendingWorkflowTask() { + // A closed current workflow that never completed a workflow task but has a real pending + // workflow task that was already started (StartedEventID set) is anchored at that started + // event, not its scheduled event, so the resetter rebuilds to the started boundary. + ctx := context.Background() + + namespaceID := namespace.ID("some random namespace ID") + workflowID := "some random workflow ID" + runID := "some random run ID" + scheduledEventID := int64(2) + startedEventID := int64(3) + nextEventID := int64(4) + startedEventVersion := s.namespaceEntry.FailoverVersion(workflowID) + versionHistory := versionhistory.NewVersionHistory([]byte("branch token"), []*historyspb.VersionHistoryItem{ + {EventId: startedEventID, Version: startedEventVersion}, + }) + histories := versionhistory.NewVersionHistories(versionHistory) + historySize := rand.Int63() + + releaseCalled := false + + targetWorkflow := NewMockWorkflow(s.controller) + weContext := historyi.NewMockWorkflowContext(s.controller) + mutableState := historyi.NewMockMutableState(s.controller) + var releaseFn historyi.ReleaseWorkflowContextFunc = func(error) { releaseCalled = true } + + workflowEvents := &persistence.WorkflowEvents{} + + targetWorkflow.EXPECT().GetContext().Return(weContext).AnyTimes() + targetWorkflow.EXPECT().GetMutableState().Return(mutableState).AnyTimes() + targetWorkflow.EXPECT().GetReleaseFn().Return(releaseFn).AnyTimes() + + s.mockClusterMetadata.EXPECT().ClusterNameForFailoverVersion(s.namespaceEntry.IsGlobalNamespace(), s.namespaceEntry.FailoverVersion(workflowID)).Return(cluster.TestCurrentClusterName).AnyTimes() + s.mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes() + + mutableState.EXPECT().IsCurrentWorkflowGuaranteed().Return(false).AnyTimes() + mutableState.EXPECT().IsWorkflowExecutionRunning().Return(false).AnyTimes() + mutableState.EXPECT().GetNamespaceEntry().Return(s.namespaceEntry).AnyTimes() + mutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{ + NamespaceId: namespaceID.String(), + WorkflowId: workflowID, + VersionHistories: histories, + }).AnyTimes() + mutableState.EXPECT().GetExecutionState().Return(&persistencespb.WorkflowExecutionState{ + RunId: runID, + }).AnyTimes() + mutableState.EXPECT().GetNextEventID().Return(nextEventID).AnyTimes() + // No completed workflow task, but a real pending task that was already started survives the close. + mutableState.EXPECT().GetLastCompletedWorkflowTaskStartedEventId().Return(common.EmptyEventID) + mutableState.EXPECT().IsTransientWorkflowTask().Return(false).AnyTimes() + mutableState.EXPECT().GetPendingWorkflowTask().Return(&historyi.WorkflowTaskInfo{ + ScheduledEventID: scheduledEventID, + StartedEventID: startedEventID, + Type: enumsspb.WORKFLOW_TASK_TYPE_NORMAL, + }).AnyTimes() + mutableState.EXPECT().AddHistorySize(historySize) + + // Reset must be anchored at the started workflow task's event, not the scheduled one. + s.mockWorkflowResetter.EXPECT().ResetWorkflow( + ctx, + namespaceID, + workflowID, + runID, + versionHistory.GetBranchToken(), + startedEventID, + startedEventVersion, + nextEventID, + gomock.Any(), + targetWorkflow, + targetWorkflow, + EventsReapplicationResetWorkflowReason, + workflowEvents.Events, + nil, + false, // allowResetWithPendingChildren + nil, // post reset operations + ).Return(nil) + + s.mockExecutionMgr.EXPECT().GetCurrentExecution(gomock.Any(), &persistence.GetCurrentExecutionRequest{ + ShardID: s.mockShard.GetShardID(), + NamespaceID: namespaceID.String(), + WorkflowID: workflowID, + ArchetypeID: chasm.WorkflowArchetypeID, + }).Return(&persistence.GetCurrentExecutionResponse{RunID: runID}, nil) + + weContext.EXPECT().PersistWorkflowEvents(gomock.Any(), s.mockShard, workflowEvents).Return(historySize, nil) + weContext.EXPECT().UpdateWorkflowExecutionWithNew( + gomock.Any(), s.mockShard, persistence.UpdateWorkflowModeBypassCurrent, nil, nil, historyi.TransactionPolicyPassive, (*historyi.TransactionPolicy)(nil), + ).Return(nil) + + err := s.transactionMgr.BackfillWorkflow(ctx, targetWorkflow, workflowEvents) + s.NoError(err) + s.True(releaseCalled) +} + +func (s *transactionMgrSuite) TestBackfillWorkflow_CurrentWorkflow_Closed_TransientWorkflowTaskNotUsable() { + // A transient (continuously failing, attempt > 1) pending task has no persisted + // WorkflowTaskScheduled event, so it must NOT be used as a reset anchor. ResetWorkflow must + // not be invoked; the EmptyEventID version-history lookup errors out instead. + ctx := context.Background() + + namespaceID := namespace.ID("some random namespace ID") + workflowID := "some random workflow ID" + runID := "some random run ID" + nextEventID := int64(5) + version := s.namespaceEntry.FailoverVersion(workflowID) + versionHistory := versionhistory.NewVersionHistory([]byte("branch token"), []*historyspb.VersionHistoryItem{ + {EventId: common.FirstEventID, Version: version}, + }) + histories := versionhistory.NewVersionHistories(versionHistory) + historySize := rand.Int63() + + releaseCalled := false + + targetWorkflow := NewMockWorkflow(s.controller) + weContext := historyi.NewMockWorkflowContext(s.controller) + mutableState := historyi.NewMockMutableState(s.controller) + var releaseFn historyi.ReleaseWorkflowContextFunc = func(error) { releaseCalled = true } + + workflowEvents := &persistence.WorkflowEvents{} + + targetWorkflow.EXPECT().GetContext().Return(weContext).AnyTimes() + targetWorkflow.EXPECT().GetMutableState().Return(mutableState).AnyTimes() + targetWorkflow.EXPECT().GetReleaseFn().Return(releaseFn).AnyTimes() + + s.mockClusterMetadata.EXPECT().ClusterNameForFailoverVersion(s.namespaceEntry.IsGlobalNamespace(), s.namespaceEntry.FailoverVersion(workflowID)).Return(cluster.TestCurrentClusterName).AnyTimes() + s.mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes() + + mutableState.EXPECT().IsCurrentWorkflowGuaranteed().Return(false).AnyTimes() + mutableState.EXPECT().IsWorkflowExecutionRunning().Return(false).AnyTimes() + mutableState.EXPECT().GetNamespaceEntry().Return(s.namespaceEntry).AnyTimes() + mutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{ + NamespaceId: namespaceID.String(), + WorkflowId: workflowID, + VersionHistories: histories, + }).AnyTimes() + mutableState.EXPECT().GetExecutionState().Return(&persistencespb.WorkflowExecutionState{ + RunId: runID, + }).AnyTimes() + mutableState.EXPECT().GetNextEventID().Return(nextEventID).AnyTimes() + mutableState.EXPECT().GetLastCompletedWorkflowTaskStartedEventId().Return(common.EmptyEventID) + mutableState.EXPECT().IsTransientWorkflowTask().Return(true).AnyTimes() + // Transient task: ScheduledEventID is a NextEventID placeholder with no backing event. + mutableState.EXPECT().GetPendingWorkflowTask().Return(&historyi.WorkflowTaskInfo{ + ScheduledEventID: nextEventID, + StartedEventID: common.EmptyEventID, + Type: enumsspb.WORKFLOW_TASK_TYPE_NORMAL, + }).AnyTimes() + mutableState.EXPECT().AddHistorySize(historySize) + + s.mockExecutionMgr.EXPECT().GetCurrentExecution(gomock.Any(), &persistence.GetCurrentExecutionRequest{ + ShardID: s.mockShard.GetShardID(), + NamespaceID: namespaceID.String(), + WorkflowID: workflowID, + ArchetypeID: chasm.WorkflowArchetypeID, + }).Return(&persistence.GetCurrentExecutionResponse{RunID: runID}, nil) + + weContext.EXPECT().PersistWorkflowEvents(gomock.Any(), s.mockShard, workflowEvents).Return(historySize, nil) + + err := s.transactionMgr.BackfillWorkflow(ctx, targetWorkflow, workflowEvents) + s.Error(err) + s.True(releaseCalled) +} + func (s *transactionMgrSuite) TestBackfillWorkflow_CurrentWorkflow_Passive_Open() { ctx := context.Background()