removed current workflow CAS from persistence used by 2dc only

common/persistence/cassandra/cassandraPersistence.go

@@ -1560,31 +1560,7 @@ func (d *cassandraPersistence) ConflictResolveWorkflowExecution(request *p.Inter
 			return serviceerror.NewInternal(fmt.Sprintf("ConflictResolveWorkflowExecution operation failed. Error: %v", err))
 		}
 
-		if request.CurrentWorkflowCAS != nil {
-			prevRunID = request.CurrentWorkflowCAS.PrevRunID
-			prevLastWriteVersion := request.CurrentWorkflowCAS.PrevLastWriteVersion
-			prevState := request.CurrentWorkflowCAS.PrevState
-
-			batch.Query(templateUpdateCurrentWorkflowExecutionForNewQuery,
-				runID,
-				executionStateDatablob.Data,
-				executionStateDatablob.Encoding.String(),
-				replicationVersions.Data,
-				replicationVersions.Encoding.String(),
-				lastWriteVersion,
-				state,
-				shardID,
-				rowTypeExecution,
-				namespaceID,
-				workflowID,
-				permanentRunID,
-				defaultVisibilityTimestamp,
-				rowTypeExecutionTaskID,
-				prevRunID,
-				prevLastWriteVersion,
-				prevState,
-			)
-		} else if currentWorkflow != nil {
+		if currentWorkflow != nil {
 			prevRunID = currentWorkflow.ExecutionInfo.ExecutionState.RunId
 
 			batch.Query(templateUpdateCurrentWorkflowExecutionQuery,

common/persistence/dataInterfaces.go

@@ -583,18 +583,6 @@ type (
 
 		// current workflow
 		CurrentWorkflowMutation *WorkflowMutation
-
-		// TODO deprecate this once nDC migration is completed
-		//  basically should use CurrentWorkflowMutation instead
-		CurrentWorkflowCAS *CurrentWorkflowCAS
-	}
-
-	// CurrentWorkflowCAS represent a compare and swap on current record
-	// TODO deprecate this once nDC migration is completed
-	CurrentWorkflowCAS struct {
-		PrevRunID            string
-		PrevLastWriteVersion int64
-		PrevState            enumsspb.WorkflowExecutionState
 	}
 
 	// ResetWorkflowExecutionRequest is used to reset workflow execution state for current run and create new run

common/persistence/executionStore.go

@@ -310,7 +310,7 @@ func (m *executionManagerImpl) ConflictResolveWorkflowExecution(
 		}
 	}
 
-	if request.CurrentWorkflowMutation != nil && request.CurrentWorkflowCAS != nil {
+	if request.CurrentWorkflowMutation != nil {
 		return serviceerror.NewInternal("ConflictResolveWorkflowExecution: current workflow & current workflow CAS both set")
 	}
 
@@ -324,10 +324,6 @@ func (m *executionManagerImpl) ConflictResolveWorkflowExecution(
 		NewWorkflowSnapshot: serializedNewWorkflowMutation,
 
 		CurrentWorkflowMutation: serializedCurrentWorkflowMutation,
-
-		// TODO deprecate this once nDC migration is completed
-		//  basically should use CurrentWorkflowMutation instead
-		CurrentWorkflowCAS: request.CurrentWorkflowCAS,
 	}
 	return m.persistence.ConflictResolveWorkflowExecution(newRequest)
 }

common/persistence/persistence-tests/persistenceTestBase.go

@@ -876,11 +876,6 @@ func (s *TestBase) ConflictResolveWorkflowExecution(prevRunID string, prevLastWr
 	requestCancelInfos []*persistenceblobs.RequestCancelInfo, signalInfos []*persistenceblobs.SignalInfo, ids []string) error {
 	return s.ExecutionManager.ConflictResolveWorkflowExecution(&p.ConflictResolveWorkflowExecutionRequest{
 		RangeID: s.ShardInfo.GetRangeId(),
-		CurrentWorkflowCAS: &p.CurrentWorkflowCAS{
-			PrevRunID:            prevRunID,
-			PrevLastWriteVersion: prevLastWriteVersion,
-			PrevState:            prevState,
-		},
 		ResetWorkflowSnapshot: p.WorkflowSnapshot{
 			ExecutionInfo:       info,
 			ExecutionStats:      stats,

common/persistence/persistenceInterface.go

@@ -241,10 +241,6 @@ type (
 
 		// current workflow
 		CurrentWorkflowMutation *InternalWorkflowMutation
-
-		// TODO deprecate this once nDC migration is completed
-		//  basically should use CurrentWorkflowMutation instead
-		CurrentWorkflowCAS *CurrentWorkflowCAS
 	}
 
 	// InternalResetWorkflowExecutionRequest is used to reset workflow execution state for Persistence Interface

common/persistence/sql/sqlExecutionManager.go

@@ -569,27 +569,7 @@ func (m *sqlExecutionManager) conflictResolveWorkflowExecutionTx(
 		state := executionInfo.GetExecutionState().State
 		status := executionInfo.ExecutionState.Status
 
-		if request.CurrentWorkflowCAS != nil {
-			prevRunID := primitives.MustParseUUID(request.CurrentWorkflowCAS.PrevRunID)
-			prevLastWriteVersion := request.CurrentWorkflowCAS.PrevLastWriteVersion
-			prevState := request.CurrentWorkflowCAS.PrevState
-
-			if err := assertAndUpdateCurrentExecution(tx,
-				m.shardID,
-				namespaceID,
-				workflowID,
-				runID,
-				prevRunID,
-				prevLastWriteVersion,
-				prevState,
-				createRequestID,
-				state,
-				status,
-				startVersion,
-				lastWriteVersion); err != nil {
-				return serviceerror.NewInternal(fmt.Sprintf("ConflictResolveWorkflowExecution. Failed to comare and swap the current record. Error: %v", err))
-			}
-		} else if currentWorkflow != nil {
+		if currentWorkflow != nil {
 			prevRunID := primitives.MustParseUUID(currentWorkflow.ExecutionInfo.ExecutionState.RunId)
 
 			if err := assertRunIDAndUpdateCurrentExecution(tx,

common/persistence/sql/sqlExecutionManagerUtil.go

@@ -968,53 +968,6 @@ func assertRunIDAndUpdateCurrentExecution(
 	return updateCurrentExecution(tx, shardID, namespaceID, workflowID, newRunID, createRequestID, state, status, startVersion, lastWriteVersion)
 }
 
-func assertAndUpdateCurrentExecution(
-	tx sqlplugin.Tx,
-	shardID int,
-	namespaceID primitives.UUID,
-	workflowID string,
-	newRunID primitives.UUID,
-	previousRunID primitives.UUID,
-	previousLastWriteVersion int64,
-	previousState enumsspb.WorkflowExecutionState,
-	createRequestID string,
-	state enumsspb.WorkflowExecutionState,
-	status enumspb.WorkflowExecutionStatus,
-	startVersion int64,
-	lastWriteVersion int64,
-) error {
-
-	assertFn := func(currentRow *sqlplugin.CurrentExecutionsRow) error {
-		if !bytes.Equal(currentRow.RunID, previousRunID) {
-			return &p.ConditionFailedError{Msg: fmt.Sprintf(
-				"assertAndUpdateCurrentExecution failed. Current run ID was %v, expected %v",
-				currentRow.RunID,
-				previousRunID,
-			)}
-		}
-		if currentRow.LastWriteVersion != previousLastWriteVersion {
-			return &p.ConditionFailedError{Msg: fmt.Sprintf(
-				"assertAndUpdateCurrentExecution failed. Current last write version was %v, expected %v",
-				currentRow.LastWriteVersion,
-				previousLastWriteVersion,
-			)}
-		}
-		if currentRow.State != previousState {
-			return &p.ConditionFailedError{Msg: fmt.Sprintf(
-				"assertAndUpdateCurrentExecution failed. Current state %v, expected %v",
-				currentRow.State,
-				previousState,
-			)}
-		}
-		return nil
-	}
-	if err := assertCurrentExecution(tx, shardID, namespaceID, workflowID, assertFn); err != nil {
-		return err
-	}
-
-	return updateCurrentExecution(tx, shardID, namespaceID, workflowID, newRunID, createRequestID, state, status, startVersion, lastWriteVersion)
-}
-
 func assertCurrentExecution(
 	tx sqlplugin.Tx,
 	shardID int,

service/history/nDCTransactionMgrForExistingWorkflow.go

@@ -355,7 +355,6 @@ func (r *nDCTransactionMgrForExistingWorkflowImpl) suppressCurrentAndUpdateAsCur
 		currentWorkflow.getContext(),
 		currentWorkflow.getMutableState(),
 		currentWorkflowPolicy.ptr(),
-		nil,
 	)
 }
 
@@ -382,7 +381,6 @@ func (r *nDCTransactionMgrForExistingWorkflowImpl) conflictResolveAsCurrent(
 		nil,
 		nil,
 		nil,
-		nil,
 	)
 }
 
@@ -454,7 +452,6 @@ func (r *nDCTransactionMgrForExistingWorkflowImpl) conflictResolveAsZombie(
 		nil,
 		nil,
 		nil,
-		nil,
 	)
 }
 

service/history/nDCTransactionMgrForExistingWorkflow_test.go

@@ -198,7 +198,6 @@ func (s *nDCTransactionMgrForExistingWorkflowSuite) TestDispatchForExistingWorkf
 		currentContext,
 		currentMutableState,
 		currentWorkflowPolicy.ptr(),
-		(*persistence.CurrentWorkflowCAS)(nil),
 	).Return(nil).Times(1)
 
 	err := s.updateMgr.dispatchForExistingWorkflow(ctx, now, isWorkflowRebuilt, targetWorkflow, newWorkflow)
@@ -272,7 +271,6 @@ func (s *nDCTransactionMgrForExistingWorkflowSuite) TestDispatchForExistingWorkf
 		currentContext,
 		currentMutableState,
 		currentWorkflowPolicy.ptr(),
-		(*persistence.CurrentWorkflowCAS)(nil),
 	).Return(nil).Times(1)
 
 	err := s.updateMgr.dispatchForExistingWorkflow(ctx, now, isWorkflowRebuilt, targetWorkflow, newWorkflow)
@@ -469,7 +467,6 @@ func (s *nDCTransactionMgrForExistingWorkflowSuite) TestDispatchForExistingWorkf
 		(workflowExecutionContext)(nil),
 		(mutableState)(nil),
 		(*transactionPolicy)(nil),
-		(*persistence.CurrentWorkflowCAS)(nil),
 	).Return(nil).Times(1)
 
 	err := s.updateMgr.dispatchForExistingWorkflow(ctx, now, isWorkflowRebuilt, targetWorkflow, newWorkflow)
@@ -541,7 +538,6 @@ func (s *nDCTransactionMgrForExistingWorkflowSuite) TestDispatchForExistingWorkf
 		currentContext,
 		currentMutableState,
 		currentWorkflowPolicy.ptr(),
-		(*persistence.CurrentWorkflowCAS)(nil),
 	).Return(nil).Times(1)
 
 	err := s.updateMgr.dispatchForExistingWorkflow(ctx, now, isWorkflowRebuilt, targetWorkflow, newWorkflow)
@@ -614,7 +610,6 @@ func (s *nDCTransactionMgrForExistingWorkflowSuite) TestDispatchForExistingWorkf
 		(workflowExecutionContext)(nil),
 		(mutableState)(nil),
 		(*transactionPolicy)(nil),
-		(*persistence.CurrentWorkflowCAS)(nil),
 	).Return(nil).Times(1)
 
 	err := s.updateMgr.dispatchForExistingWorkflow(ctx, now, isWorkflowRebuilt, targetWorkflow, newWorkflow)
@@ -687,7 +682,6 @@ func (s *nDCTransactionMgrForExistingWorkflowSuite) TestDispatchForExistingWorkf
 		(workflowExecutionContext)(nil),
 		(mutableState)(nil),
 		(*transactionPolicy)(nil),
-		(*persistence.CurrentWorkflowCAS)(nil),
 	).Return(nil).Times(1)
 
 	err := s.updateMgr.dispatchForExistingWorkflow(ctx, now, isWorkflowRebuilt, targetWorkflow, newWorkflow)

service/history/workflowExecutionContext.go

@@ -99,7 +99,6 @@ type (
 			currentContext workflowExecutionContext,
 			currentMutableState mutableState,
 			currentTransactionPolicy *transactionPolicy,
-			workflowCAS *persistence.CurrentWorkflowCAS,
 		) error
 		updateWorkflowExecutionAsActive(
 			now time.Time,
@@ -414,7 +413,6 @@ func (c *workflowExecutionContextImpl) conflictResolveWorkflowExecution(
 	currentContext workflowExecutionContext,
 	currentMutableState mutableState,
 	currentTransactionPolicy *transactionPolicy,
-	workflowCAS *persistence.CurrentWorkflowCAS,
 ) (retError error) {
 
 	defer func() {
@@ -523,7 +521,6 @@ func (c *workflowExecutionContextImpl) conflictResolveWorkflowExecution(
 
 		CurrentWorkflowMutation: currentWorkflow,
 
-		CurrentWorkflowCAS: workflowCAS,
 		// Encoding, this is set by shard context
 	}); err != nil {
 		return err