Revert "history service should do event reordering making sure corres…

…ponding … (#601)" (#611)

This reverts commit 6164b73.

idl/github.com/uber/cadence/shared.thrift

@@ -85,10 +85,6 @@ enum TimeoutType {
   HEARTBEAT,
 }
 
-// whenever this list of decision is changed
-// do change the mutableStateBuilder.go
-// function shouldBufferEvent
-// to make sure wo do the correct event ordering
 enum DecisionType {
   ScheduleActivityTask,
   RequestCancelActivityTask,

service/history/historyBuilder_test.go

@@ -142,8 +142,6 @@ func (s *historyBuilderSuite) TestHistoryBuilderDynamicSuccess() {
 	s.Equal(int64(3), s.getPreviousDecisionStartedEventID())
 
 	activityStartedEvent := s.addActivityTaskStartedEvent(5, activityTaskList, identity)
-	s.validateActivityTaskStartedEvent(activityStartedEvent, bufferedEventID, 5, identity)
-	s.Nil(s.msBuilder.FlushBufferedEvents())
 	s.validateActivityTaskStartedEvent(activityStartedEvent, 7, 5, identity)
 	s.Equal(int64(8), s.getNextEventID())
 	ai3, activity1Running1 := s.msBuilder.GetActivityInfo(5)
@@ -152,9 +150,6 @@ func (s *historyBuilderSuite) TestHistoryBuilderDynamicSuccess() {
 	s.Equal(int64(3), s.getPreviousDecisionStartedEventID())
 
 	activityCompletedEvent := s.addActivityTaskCompletedEvent(5, 7, activity1Result, identity)
-	s.validateActivityTaskCompletedEvent(activityCompletedEvent, bufferedEventID, 5, 7, activity1Result,
-		identity)
-	s.Nil(s.msBuilder.FlushBufferedEvents())
 	s.validateActivityTaskCompletedEvent(activityCompletedEvent, 8, 5, 7, activity1Result,
 		identity)
 	s.Equal(int64(9), s.getNextEventID())
@@ -171,8 +166,6 @@ func (s *historyBuilderSuite) TestHistoryBuilderDynamicSuccess() {
 	s.Equal(int64(3), s.getPreviousDecisionStartedEventID())
 
 	activity2StartedEvent := s.addActivityTaskStartedEvent(6, activityTaskList, identity)
-	s.validateActivityTaskStartedEvent(activity2StartedEvent, bufferedEventID, 6, identity)
-	s.Nil(s.msBuilder.FlushBufferedEvents())
 	s.validateActivityTaskStartedEvent(activity2StartedEvent, 10, 6, identity)
 	s.Equal(int64(11), s.getNextEventID())
 	ai4, activity2Running1 := s.msBuilder.GetActivityInfo(6)
@@ -182,9 +175,6 @@ func (s *historyBuilderSuite) TestHistoryBuilderDynamicSuccess() {
 
 	activity2FailedEvent := s.addActivityTaskFailedEvent(6, 10, activity2Reason, activity2Details,
 		identity)
-	s.validateActivityTaskFailedEvent(activity2FailedEvent, bufferedEventID, 6, 10, activity2Reason,
-		activity2Details, identity)
-	s.Nil(s.msBuilder.FlushBufferedEvents())
 	s.validateActivityTaskFailedEvent(activity2FailedEvent, 11, 6, 10, activity2Reason,
 		activity2Details, identity)
 	s.Equal(int64(12), s.getNextEventID())
@@ -397,8 +387,6 @@ func (s *historyBuilderSuite) TestHistoryBuilderFlushBufferedEvents() {
 
 	// 7 activity1 started
 	activityStartedEvent := s.addActivityTaskStartedEvent(5, activityTaskList, identity)
-	s.validateActivityTaskStartedEvent(activityStartedEvent, bufferedEventID, 5, identity)
-	s.Nil(s.msBuilder.FlushBufferedEvents())
 	s.validateActivityTaskStartedEvent(activityStartedEvent, 7, 5, identity)
 	s.Equal(int64(8), s.getNextEventID())
 	ai3, activity1Running1 := s.msBuilder.GetActivityInfo(5)
@@ -408,9 +396,8 @@ func (s *historyBuilderSuite) TestHistoryBuilderFlushBufferedEvents() {
 
 	// 8 activity1 completed
 	activityCompletedEvent := s.addActivityTaskCompletedEvent(5, 7, activity1Result, identity)
-	s.validateActivityTaskCompletedEvent(activityCompletedEvent, bufferedEventID, 5, 7, activity1Result, identity)
-	s.Nil(s.msBuilder.FlushBufferedEvents())
-	s.validateActivityTaskCompletedEvent(activityCompletedEvent, 8, 5, 7, activity1Result, identity)
+	s.validateActivityTaskCompletedEvent(activityCompletedEvent, 8, 5, 7, activity1Result,
+		identity)
 	s.Equal(int64(9), s.getNextEventID())
 	_, activity1Running2 := s.msBuilder.GetActivityInfo(5)
 	s.False(activity1Running2)
@@ -545,9 +532,9 @@ func (s *historyBuilderSuite) TestHistoryBuilderWorkflowCancellationRequested()
 	cancellationRequestedEvent := s.addExternalWorkflowExecutionCancelRequested(
 		5, targetDomain, targetExecution.GetWorkflowId(), targetExecution.GetRunId(),
 	)
-	s.validateExternalWorkflowExecutionCancelRequested(cancellationRequestedEvent, bufferedEventID, 5, targetDomain, targetExecution)
-	s.Nil(s.msBuilder.FlushBufferedEvents())
-	s.validateExternalWorkflowExecutionCancelRequested(cancellationRequestedEvent, 6, 5, targetDomain, targetExecution)
+	s.validateExternalWorkflowExecutionCancelRequested(
+		cancellationRequestedEvent, 6, 5, targetDomain, targetExecution,
+	)
 	s.Equal(int64(7), s.getNextEventID())
 }
 
@@ -618,10 +605,6 @@ func (s *historyBuilderSuite) TestHistoryBuilderWorkflowCancellationFailed() {
 	cancellationRequestedEvent := s.addRequestCancelExternalWorkflowExecutionFailedEvent(
 		4, 5, targetDomain, targetExecution.GetWorkflowId(), targetExecution.GetRunId(), cancellationFailedCause,
 	)
-	s.validateRequestCancelExternalWorkflowExecutionFailedEvent(
-		cancellationRequestedEvent, bufferedEventID, 4, 5, targetDomain, targetExecution, cancellationFailedCause,
-	)
-	s.Nil(s.msBuilder.FlushBufferedEvents())
 	s.validateRequestCancelExternalWorkflowExecutionFailedEvent(
 		cancellationRequestedEvent, 6, 4, 5, targetDomain, targetExecution, cancellationFailedCause,
 	)

service/history/historyEngine_test.go

@@ -3502,7 +3502,6 @@ func addCompleteWorkflowEvent(builder *mutableStateBuilder, decisionCompletedEve
 }
 
 func createMutableState(builder *mutableStateBuilder) *persistence.WorkflowMutableState {
-	builder.FlushBufferedEvents()
 	info := copyWorkflowExecutionInfo(builder.executionInfo)
 	activityInfos := make(map[int64]*persistence.ActivityInfo)
 	for id, info := range builder.pendingActivityInfoIDs {

service/history/mutableStateBuilder.go

@@ -378,54 +378,26 @@ func (e *mutableStateBuilder) createNewHistoryEvent(eventType workflow.EventType
 }
 
 func (e *mutableStateBuilder) shouldBufferEvent(eventType workflow.EventType) bool {
+	if !e.HasInFlightDecisionTask() {
+		// do not buffer event if there is no in-flight decision
+		return false
+	}
+
 	switch eventType {
-	case // do not buffer for workflow state change
-		workflow.EventTypeWorkflowExecutionStarted,
+	case workflow.EventTypeDecisionTaskCompleted,
+		workflow.EventTypeDecisionTaskFailed,
+		workflow.EventTypeDecisionTaskTimedOut,
 		workflow.EventTypeWorkflowExecutionCompleted,
 		workflow.EventTypeWorkflowExecutionFailed,
 		workflow.EventTypeWorkflowExecutionTimedOut,
 		workflow.EventTypeWorkflowExecutionTerminated,
 		workflow.EventTypeWorkflowExecutionContinuedAsNew,
 		workflow.EventTypeWorkflowExecutionCanceled:
+		// do not buffer event if it is any type of close decision or close workflow
 		return false
-	case // decision event should not be buffered
-		workflow.EventTypeDecisionTaskScheduled,
-		workflow.EventTypeDecisionTaskStarted,
-		workflow.EventTypeDecisionTaskCompleted,
-		workflow.EventTypeDecisionTaskFailed,
-		workflow.EventTypeDecisionTaskTimedOut:
-		return false
-	case // events generated directly from decisions should not be buffered
-		// workflow complete, failed, cancelled and continue-as-new events are duplication of above
-		// just put is here for reference
-		// workflow.EventTypeWorkflowExecutionCompleted,
-		// workflow.EventTypeWorkflowExecutionFailed,
-		// workflow.EventTypeWorkflowExecutionCanceled,
-		// workflow.EventTypeWorkflowExecutionContinuedAsNew,
-		workflow.EventTypeActivityTaskScheduled,
-		workflow.EventTypeActivityTaskCancelRequested,
-		workflow.EventTypeTimerStarted,
-		// DecisionTypeCancelTimer is an excption. This decision will be mapped
-		// to either workflow.EventTypeTimerCanceled, or workflow.EventTypeCancelTimerFailed.
-		// So both should not be buffered. Ref: historyEngine, search for "workflow.DecisionTypeCancelTimer"
-		workflow.EventTypeTimerCanceled,
-		workflow.EventTypeCancelTimerFailed,
-		workflow.EventTypeRequestCancelExternalWorkflowExecutionInitiated,
-		workflow.EventTypeMarkerRecorded,
-		workflow.EventTypeStartChildWorkflowExecutionInitiated,
-		workflow.EventTypeSignalExternalWorkflowExecutionInitiated:
-		// do not buffer event if event is directly generated from a corresponding decision
-
-		// sanity check there is no decision on the fly
-		if e.HasInFlightDecisionTask() {
-			msg := fmt.Sprintf("history mutable state is processing event: %v while there is decision pending. "+
-				"domainID: %v, workflow ID: %v, run ID: %v.", eventType, e.executionInfo.DomainID, e.executionInfo.WorkflowID, e.executionInfo.RunID)
-			panic(msg)
-		}
-		return false
-	default:
-		return true
 	}
+
+	return true
 }
 
 func (e *mutableStateBuilder) createNewHistoryEventWithTimestamp(eventID int64, eventType workflow.EventType,

service/history/timerQueueProcessor_test.go

@@ -166,14 +166,13 @@ func (s *timerQueueProcessorSuite) addDecisionTimer(domainID string, we workflow
 	builder.Load(state)
 
 	di := addDecisionTaskScheduledEvent(builder)
-	startedEvent := addDecisionTaskStartedEvent(builder, di.ScheduleID, state.ExecutionInfo.TaskList, "identity")
+	addDecisionTaskStartedEvent(builder, di.ScheduleID, state.ExecutionInfo.TaskList, "identity")
 
 	timeOutTask := tb.AddDecisionTimoutTask(di.ScheduleID, di.Attempt, 1)
 	timerTasks := []persistence.Task{timeOutTask}
 
 	s.updateTimerSeqNumbers(timerTasks)
 
-	addDecisionTaskCompletedEvent(builder, di.ScheduleID, startedEvent.GetEventId(), nil, "identity")
 	err2 := s.UpdateWorkflowExecution(state.ExecutionInfo, nil, nil, condition, timerTasks, nil, nil, nil, nil, nil)
 	s.Nil(err2, "No error expected.")
 	return timerTasks