Fix NDC resetter persistence bugs (#3500)

cadence-workflow · Sep 11, 2020 · 3b56ad7 · 3b56ad7
1 parent 2fc957e
commit 3b56ad7
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 13 deletions.
diff --git a/common/persistence/sql/sqlExecutionManager.go b/common/persistence/sql/sqlExecutionManager.go
@@ -121,14 +121,6 @@ func (m *sqlExecutionManager) createWorkflowExecutionTx(
 		return nil, err
 	}
 
-	switch request.Mode {
-	case p.CreateWorkflowModeContinueAsNew:
-		// cannot create workflow with continue as new mode
-		return nil, &workflow.InternalServiceError{
-			Message: "CreateWorkflowExecution: operation failed, encounter invalid CreateWorkflowModeContinueAsNew",
-		}
-	}
-
 	var err error
 	var row *sqlplugin.CurrentExecutionsRow
 	if row, err = lockCurrentExecutionIfExists(tx, m.shardID, domainID, workflowID); err != nil {
@@ -179,6 +171,12 @@ func (m *sqlExecutionManager) createWorkflowExecutionTx(
 				return nil, err
 			}
 
+		case p.CreateWorkflowModeContinueAsNew:
+			// continueAsNew mode expects a current run exists
+			if err := assertRunIDMismatch(sqlplugin.MustParseUUID(executionInfo.RunID), row.RunID); err != nil {
+				return nil, err
+			}
+
 		default:
 			return nil, &workflow.InternalServiceError{
 				Message: fmt.Sprintf(

diff --git a/service/history/reset/resetter.go b/service/history/reset/resetter.go
@@ -285,7 +285,12 @@ func (r *workflowResetterImpl) persistToDB(
 	if err != nil {
 		return err
 	}
-	resetHistorySize, err := resetWorkflow.GetContext().PersistFirstWorkflowEvents(resetWorkflowEventsSeq[0])
+	if len(resetWorkflowEventsSeq) != 1{
+		return &shared.InternalServiceError{
+			Message: "there should be EXACTLY one batch of events for reset",
+		}
+	}
+	resetHistorySize, err := resetWorkflow.GetContext().PersistNonFirstWorkflowEvents(resetWorkflowEventsSeq[0])
 	if err != nil {
 		return err
 	}
@@ -312,7 +317,7 @@ func (r *workflowResetterImpl) replayResetWorkflow(
 	resetRequestID string,
 ) (execution.Workflow, error) {
 
-	resetBranchToken, err := r.generateBranchToken(
+	resetBranchToken, err := r.forkAndGenerateBranchToken(
 		domainID,
 		workflowID,
 		baseBranchToken,
@@ -399,7 +404,7 @@ func (r *workflowResetterImpl) failInflightActivity(
 	return nil
 }
 
-func (r *workflowResetterImpl) generateBranchToken(
+func (r *workflowResetterImpl) forkAndGenerateBranchToken(
 	domainID string,
 	workflowID string,
 	forkBranchToken []byte,

diff --git a/service/history/reset/resetter_test.go b/service/history/reset/resetter_test.go
@@ -191,7 +191,7 @@ func (s *workflowResetterSuite) TestPersistToDB_CurrentNotTerminated() {
 		gomock.Any(),
 		execution.TransactionPolicyActive,
 	).Return(resetSnapshot, resetEventsSeq, nil).Times(1)
-	resetContext.EXPECT().PersistFirstWorkflowEvents(resetEventsSeq[0]).Return(resetEventsSize, nil).Times(1)
+	resetContext.EXPECT().PersistNonFirstWorkflowEvents(resetEventsSeq[0]).Return(resetEventsSize, nil).Times(1)
 	resetContext.EXPECT().CreateWorkflowExecution(
 		resetSnapshot,
 		resetEventsSize,
@@ -312,7 +312,7 @@ func (s *workflowResetterSuite) TestGenerateBranchToken() {
 		ShardID:         common.IntPtr(s.mockShard.GetShardID()),
 	}).Return(&persistence.ForkHistoryBranchResponse{NewBranchToken: resetBranchToken}, nil).Times(1)
 
-	newBranchToken, err := s.workflowResetter.generateBranchToken(
+	newBranchToken, err := s.workflowResetter.forkAndGenerateBranchToken(
 		s.domainID, s.workflowID, baseBranchToken, baseNodeID, s.resetRunID,
 	)
 	s.NoError(err)