fix decision state machine (#532)

yiminc · web-flow · commit 4be6b261f85a · 2018-07-30T11:25:01.000-07:00
diff --git a/internal/error_test.go b/internal/error_test.go
@@ -98,10 +98,11 @@ func Test_TimeoutError(t *testing.T) {
 		decisionsHelper: newDecisionsHelper(),
 		dataConverter:   newDefaultDataConverter(),
 	}
+	h := newDecisionsHelper()
 	var actualErr error
 	activityID := "activityID"
 	context.decisionsHelper.scheduledEventIDToActivityID[5] = activityID
-	di := newActivityDecisionStateMachine(
+	di := h.newActivityDecisionStateMachine(
 		&shared.ScheduleActivityTaskDecisionAttributes{ActivityId: common.StringPtr(activityID)})
 	di.state = decisionStateInitiated
 	di.setData(&scheduledActivity{
diff --git a/internal/internal_decision_state_machine.go b/internal/internal_decision_state_machine.go
@@ -65,6 +65,7 @@ type (
 		state   decisionState
 		history []string
 		data    interface{}
+		helper  *decisionsHelper
 	}
 
 	activityDecisionStateMachine struct {
@@ -207,67 +208,68 @@ func makeDecisionID(decisionType decisionType, id string) decisionID {
 	return decisionID{decisionType: decisionType, id: id}
 }
 
-func newDecisionStateMachineBase(decisionType decisionType, id string) *decisionStateMachineBase {
+func (h *decisionsHelper) newDecisionStateMachineBase(decisionType decisionType, id string) *decisionStateMachineBase {
 	return &decisionStateMachineBase{
 		id:      makeDecisionID(decisionType, id),
 		state:   decisionStateCreated,
 		history: []string{decisionStateCreated.String()},
+		helper:  h,
 	}
 }
 
-func newActivityDecisionStateMachine(attributes *s.ScheduleActivityTaskDecisionAttributes) *activityDecisionStateMachine {
-	base := newDecisionStateMachineBase(decisionTypeActivity, attributes.GetActivityId())
+func (h *decisionsHelper) newActivityDecisionStateMachine(attributes *s.ScheduleActivityTaskDecisionAttributes) *activityDecisionStateMachine {
+	base := h.newDecisionStateMachineBase(decisionTypeActivity, attributes.GetActivityId())
 	return &activityDecisionStateMachine{
 		decisionStateMachineBase: base,
 		attributes:               attributes,
 	}
 }
 
-func newTimerDecisionStateMachine(attributes *s.StartTimerDecisionAttributes) *timerDecisionStateMachine {
-	base := newDecisionStateMachineBase(decisionTypeTimer, attributes.GetTimerId())
+func (h *decisionsHelper) newTimerDecisionStateMachine(attributes *s.StartTimerDecisionAttributes) *timerDecisionStateMachine {
+	base := h.newDecisionStateMachineBase(decisionTypeTimer, attributes.GetTimerId())
 	return &timerDecisionStateMachine{
 		decisionStateMachineBase: base,
 		attributes:               attributes,
 	}
 }
 
-func newChildWorkflowDecisionStateMachine(attributes *s.StartChildWorkflowExecutionDecisionAttributes) *childWorkflowDecisionStateMachine {
-	base := newDecisionStateMachineBase(decisionTypeChildWorkflow, attributes.GetWorkflowId())
+func (h *decisionsHelper) newChildWorkflowDecisionStateMachine(attributes *s.StartChildWorkflowExecutionDecisionAttributes) *childWorkflowDecisionStateMachine {
+	base := h.newDecisionStateMachineBase(decisionTypeChildWorkflow, attributes.GetWorkflowId())
 	return &childWorkflowDecisionStateMachine{
 		decisionStateMachineBase: base,
 		attributes:               attributes,
 	}
 }
 
-func newNaiveDecisionStateMachine(decisionType decisionType, id string, decision *s.Decision) *naiveDecisionStateMachine {
-	base := newDecisionStateMachineBase(decisionType, id)
+func (h *decisionsHelper) newNaiveDecisionStateMachine(decisionType decisionType, id string, decision *s.Decision) *naiveDecisionStateMachine {
+	base := h.newDecisionStateMachineBase(decisionType, id)
 	return &naiveDecisionStateMachine{
 		decisionStateMachineBase: base,
 		decision:                 decision,
 	}
 }
 
-func newMarkerDecisionStateMachine(id string, attributes *s.RecordMarkerDecisionAttributes) *markerDecisionStateMachine {
+func (h *decisionsHelper) newMarkerDecisionStateMachine(id string, attributes *s.RecordMarkerDecisionAttributes) *markerDecisionStateMachine {
 	d := createNewDecision(s.DecisionTypeRecordMarker)
 	d.RecordMarkerDecisionAttributes = attributes
 	return &markerDecisionStateMachine{
-		naiveDecisionStateMachine: newNaiveDecisionStateMachine(decisionTypeMarker, id, d),
+		naiveDecisionStateMachine: h.newNaiveDecisionStateMachine(decisionTypeMarker, id, d),
 	}
 }
 
-func newCancelExternalWorkflowStateMachine(attributes *s.RequestCancelExternalWorkflowExecutionDecisionAttributes, cancellationID string) *cancelExternalWorkflowDecisionStateMachine {
+func (h *decisionsHelper) newCancelExternalWorkflowStateMachine(attributes *s.RequestCancelExternalWorkflowExecutionDecisionAttributes, cancellationID string) *cancelExternalWorkflowDecisionStateMachine {
 	d := createNewDecision(s.DecisionTypeRequestCancelExternalWorkflowExecution)
 	d.RequestCancelExternalWorkflowExecutionDecisionAttributes = attributes
 	return &cancelExternalWorkflowDecisionStateMachine{
-		naiveDecisionStateMachine: newNaiveDecisionStateMachine(decisionTypeCancellation, cancellationID, d),
+		naiveDecisionStateMachine: h.newNaiveDecisionStateMachine(decisionTypeCancellation, cancellationID, d),
 	}
 }
 
-func newSignalExternalWorkflowStateMachine(attributes *s.SignalExternalWorkflowExecutionDecisionAttributes, signalID string) *signalExternalWorkflowDecisionStateMachine {
+func (h *decisionsHelper) newSignalExternalWorkflowStateMachine(attributes *s.SignalExternalWorkflowExecutionDecisionAttributes, signalID string) *signalExternalWorkflowDecisionStateMachine {
 	d := createNewDecision(s.DecisionTypeSignalExternalWorkflowExecution)
 	d.SignalExternalWorkflowExecutionDecisionAttributes = attributes
 	return &signalExternalWorkflowDecisionStateMachine{
-		naiveDecisionStateMachine: newNaiveDecisionStateMachine(decisionTypeSignal, signalID, d),
+		naiveDecisionStateMachine: h.newNaiveDecisionStateMachine(decisionTypeSignal, signalID, d),
 	}
 }
 
@@ -295,6 +297,13 @@ func (d *decisionStateMachineBase) moveState(newState decisionState, event strin
 	d.history = append(d.history, event)
 	d.state = newState
 	d.history = append(d.history, newState.String())
+
+	if newState == decisionStateCompleted {
+		if elem, ok := d.helper.decisions[d.getID()]; ok {
+			d.helper.orderedDecisions.Remove(elem)
+			delete(d.helper.decisions, d.getID())
+		}
+	}
 }
 
 func (d *decisionStateMachineBase) failStateTransition(event string) {
@@ -660,12 +669,16 @@ func (h *decisionsHelper) getDecision(id decisionID) decisionStateMachine {
 }
 
 func (h *decisionsHelper) addDecision(decision decisionStateMachine) {
+	if _, ok := h.decisions[decision.getID()]; ok {
+		panicMsg := fmt.Sprintf("adding duplicate decision %v", decision)
+		panic(panicMsg)
+	}
 	element := h.orderedDecisions.PushBack(decision)
 	h.decisions[decision.getID()] = element
 }
 
 func (h *decisionsHelper) scheduleActivityTask(attributes *s.ScheduleActivityTaskDecisionAttributes) decisionStateMachine {
-	decision := newActivityDecisionStateMachine(attributes)
+	decision := h.newActivityDecisionStateMachine(attributes)
 	h.addDecision(decision)
 	return decision
 }
@@ -739,7 +752,7 @@ func (h *decisionsHelper) recordVersionMarker(changeID string, version Version,
 		Details:    details, // Keep
 	}
 
-	decision := newMarkerDecisionStateMachine(markerID, recordMarker)
+	decision := h.newMarkerDecisionStateMachine(markerID, recordMarker)
 	h.addDecision(decision)
 	return decision
 }
@@ -750,7 +763,7 @@ func (h *decisionsHelper) recordSideEffectMarker(sideEffectID int32, data []byte
 		MarkerName: common.StringPtr(sideEffectMarkerName),
 		Details:    data,
 	}
-	decision := newMarkerDecisionStateMachine(markerID, attributes)
+	decision := h.newMarkerDecisionStateMachine(markerID, attributes)
 	h.addDecision(decision)
 	return decision
 }
@@ -761,7 +774,7 @@ func (h *decisionsHelper) recordLocalActivityMarker(activityID string, result []
 		MarkerName: common.StringPtr(localActivityMarkerName),
 		Details:    result,
 	}
-	decision := newMarkerDecisionStateMachine(markerID, attributes)
+	decision := h.newMarkerDecisionStateMachine(markerID, attributes)
 	h.addDecision(decision)
 	return decision
 }
@@ -772,13 +785,13 @@ func (h *decisionsHelper) recordMutableSideEffectMarker(mutableSideEffectID stri
 		MarkerName: common.StringPtr(mutableSideEffectMarkerName),
 		Details:    data,
 	}
-	decision := newMarkerDecisionStateMachine(markerID, attributes)
+	decision := h.newMarkerDecisionStateMachine(markerID, attributes)
 	h.addDecision(decision)
 	return decision
 }
 
 func (h *decisionsHelper) startChildWorkflowExecution(attributes *s.StartChildWorkflowExecutionDecisionAttributes) decisionStateMachine {
-	decision := newChildWorkflowDecisionStateMachine(attributes)
+	decision := h.newChildWorkflowDecisionStateMachine(attributes)
 	h.addDecision(decision)
 	return decision
 }
@@ -833,7 +846,7 @@ func (h *decisionsHelper) requestCancelExternalWorkflowExecution(domain, workflo
 		Control:           []byte(cancellationID),
 		ChildWorkflowOnly: common.BoolPtr(false),
 	}
-	decision := newCancelExternalWorkflowStateMachine(attributes, cancellationID)
+	decision := h.newCancelExternalWorkflowStateMachine(attributes, cancellationID)
 	h.addDecision(decision)
 
 	return decision
@@ -893,7 +906,7 @@ func (h *decisionsHelper) signalExternalWorkflowExecution(domain, workflowID, ru
 		Control:           []byte(signalID),
 		ChildWorkflowOnly: common.BoolPtr(childWorkflowOnly),
 	}
-	decision := newSignalExternalWorkflowStateMachine(attributes, signalID)
+	decision := h.newSignalExternalWorkflowStateMachine(attributes, signalID)
 	h.addDecision(decision)
 	return decision
 }
@@ -925,7 +938,7 @@ func (h *decisionsHelper) getSignalID(initiatedEventID int64) string {
 }
 
 func (h *decisionsHelper) startTimer(attributes *s.StartTimerDecisionAttributes) decisionStateMachine {
-	decision := newTimerDecisionStateMachine(attributes)
+	decision := h.newTimerDecisionStateMachine(attributes)
 	h.addDecision(decision)
 	return decision
 }
diff --git a/internal/internal_event_handlers.go b/internal/internal_event_handlers.go
@@ -593,9 +593,9 @@ func (weh *workflowExecutionEventHandlerImpl) ProcessEvent(
 	event *m.HistoryEvent,
 	isReplay bool,
 	isLast bool,
-) (result []*m.Decision, err error) {
+) (err error) {
 	if event == nil {
-		return nil, errors.New("nil event provided")
+		return errors.New("nil event provided")
 	}
 	defer func() {
 		if p := recover(); p != nil {
@@ -746,7 +746,7 @@ func (weh *workflowExecutionEventHandlerImpl) ProcessEvent(
 	}
 
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	// When replaying histories to get stack trace or current state the last event might be not
@@ -756,7 +756,7 @@ func (weh *workflowExecutionEventHandlerImpl) ProcessEvent(
 		weh.workflowDefinition.OnDecisionTaskStarted()
 	}
 
-	return weh.decisionsHelper.getDecisions(true), nil
+	return nil
 }
 
 func (weh *workflowExecutionEventHandlerImpl) ProcessQuery(queryType string, queryArgs []byte) ([]byte, error) {
@@ -928,7 +928,7 @@ func (weh *workflowExecutionEventHandlerImpl) handleLocalActivityMarker(markerDa
 	return nil
 }
 
-func (weh *workflowExecutionEventHandlerImpl) ProcessLocalActivityResult(lar *localActivityResult) ([]*m.Decision, error) {
+func (weh *workflowExecutionEventHandlerImpl) ProcessLocalActivityResult(lar *localActivityResult) error {
 	// convert local activity result and error to marker data
 	lamd := localActivityMarkerData{
 		ActivityID: lar.task.activityID,
@@ -945,7 +945,7 @@ func (weh *workflowExecutionEventHandlerImpl) ProcessLocalActivityResult(lar *lo
 	// encode marker data
 	markerData, err := weh.encodeArg(lamd)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	// create marker event for local activity result
diff --git a/internal/internal_task_handlers.go b/internal/internal_task_handlers.go
@@ -54,7 +54,7 @@ type (
 	workflowExecutionEventHandler interface {
 		// Process a single event and return the assosciated decisions.
 		// Return List of decisions made, any error.
-		ProcessEvent(event *s.HistoryEvent, isReplay bool, isLast bool) ([]*s.Decision, error)
+		ProcessEvent(event *s.HistoryEvent, isReplay bool, isLast bool) error
 		// ProcessQuery process a query request.
 		ProcessQuery(queryType string, queryArgs []byte) ([]byte, error)
 		StackTrace() string
@@ -293,8 +293,7 @@ OrderEvents:
 }
 
 func isPreloadMarkerEvent(event *s.HistoryEvent) bool {
-	return event.GetEventType() == s.EventTypeMarkerRecorded &&
-		event.MarkerRecordedEventAttributes.GetMarkerName() != localActivityMarkerName
+	return event.GetEventType() == s.EventTypeMarkerRecorded
 }
 
 // newWorkflowTaskHandler returns an implementation of workflow task handler.
@@ -653,9 +652,12 @@ ProcessEvents:
 		}
 		// Markers are from the events that are produced from the current decision
 		for _, m := range markers {
-			_, err := eventHandler.ProcessEvent(m, true, false)
-			if err != nil {
-				return nil, err
+			if m.MarkerRecordedEventAttributes.GetMarkerName() != localActivityMarkerName {
+				// local activity marker needs to be applied after decision task started event
+				err := eventHandler.ProcessEvent(m, true, false)
+				if err != nil {
+					return nil, err
+				}
 			}
 		}
 
@@ -672,23 +674,26 @@ ProcessEvents:
 				return nil, err
 			}
 
-			eventDecisions, err := eventHandler.ProcessEvent(event, isInReplay, isLast)
+			err = eventHandler.ProcessEvent(event, isInReplay, isLast)
 			if err != nil {
 				return nil, err
 			}
+		}
 
-			if eventDecisions != nil {
-				if !isInReplay {
-					w.newDecisions = append(w.newDecisions, eventDecisions...)
-				} else if !skipReplayCheck {
-					replayDecisions = append(replayDecisions, eventDecisions...)
+		// now apply local activity markers
+		for _, m := range markers {
+			if m.MarkerRecordedEventAttributes.GetMarkerName() == localActivityMarkerName {
+				err := eventHandler.ProcessEvent(m, true, false)
+				if err != nil {
+					return nil, err
 				}
 			}
-
-			if w.isWorkflowCompleted {
-				// If workflow is already completed then we can break from processing
-				// further decisions.
-				break ProcessEvents
+		}
+		isReplay := len(reorderedEvents) > 0 && reorderedHistory.IsReplayEvent(reorderedEvents[len(reorderedEvents)-1])
+		if isReplay {
+			eventDecisions := eventHandler.decisionsHelper.getDecisions(true)
+			if len(eventDecisions) > 0 && !skipReplayCheck {
+				replayDecisions = append(replayDecisions, eventDecisions...)
 			}
 		}
 	}
@@ -728,15 +733,11 @@ ProcessEvents:
 }
 
 func (w *workflowExecutionContextImpl) ProcessLocalActivityResult(lar *localActivityResult) (interface{}, error) {
-	eventDecisions, err := w.eventHandler.ProcessLocalActivityResult(lar)
+	err := w.eventHandler.ProcessLocalActivityResult(lar)
 	if err != nil {
 		return nil, err
 	}
 
-	if eventDecisions != nil {
-		w.newDecisions = append(w.newDecisions, eventDecisions...)
-	}
-
 	return w.CompleteDecisionTask(true), nil
 }
 
@@ -760,6 +761,11 @@ func (w *workflowExecutionContextImpl) CompleteDecisionTask(waitLocalActivities
 		}
 	}
 
+	eventDecisions := w.eventHandler.decisionsHelper.getDecisions(true)
+	if len(eventDecisions) > 0 {
+		w.newDecisions = append(w.newDecisions, eventDecisions...)
+	}
+
 	completeRequest := w.wth.completeWorkflow(w.eventHandler, w.currentDecisionTask, w, w.newDecisions, !waitLocalActivities)
 	w.clearCurrentTask()
 

Original file line number	Diff line number	Diff line change
`@@ -65,6 +65,7 @@ type (`
`65`	`65`	`state decisionState`
`66`	`66`	`history []string`
`67`	`67`	`data interface{}`
	`68`	`+ helper *decisionsHelper`
`68`	`69`	`}`
`69`	`70`
`70`	`71`	`activityDecisionStateMachine struct {`
`@@ -207,67 +208,68 @@ func makeDecisionID(decisionType decisionType, id string) decisionID {`
`207`	`208`	`return decisionID{decisionType: decisionType, id: id}`
`208`	`209`	`}`
`209`	`210`
`210`		`-func newDecisionStateMachineBase(decisionType decisionType, id string) *decisionStateMachineBase {`
	`211`	`+func (h decisionsHelper) newDecisionStateMachineBase(decisionType decisionType, id string) decisionStateMachineBase {`
`211`	`212`	`return &decisionStateMachineBase{`
`212`	`213`	`id: makeDecisionID(decisionType, id),`
`213`	`214`	`state: decisionStateCreated,`
`214`	`215`	`history: []string{decisionStateCreated.String()},`
	`216`	`+ helper: h,`
`215`	`217`	`}`
`216`	`218`	`}`
`217`	`219`
`218`		`-func newActivityDecisionStateMachine(attributes s.ScheduleActivityTaskDecisionAttributes) activityDecisionStateMachine {`
`219`		`- base := newDecisionStateMachineBase(decisionTypeActivity, attributes.GetActivityId())`
	`220`	`+func (h decisionsHelper) newActivityDecisionStateMachine(attributes s.ScheduleActivityTaskDecisionAttributes) *activityDecisionStateMachine {`
	`221`	`+ base := h.newDecisionStateMachineBase(decisionTypeActivity, attributes.GetActivityId())`
`220`	`222`	`return &activityDecisionStateMachine{`
`221`	`223`	`decisionStateMachineBase: base,`
`222`	`224`	`attributes: attributes,`
`223`	`225`	`}`
`224`	`226`	`}`
`225`	`227`
`226`		`-func newTimerDecisionStateMachine(attributes s.StartTimerDecisionAttributes) timerDecisionStateMachine {`
`227`		`- base := newDecisionStateMachineBase(decisionTypeTimer, attributes.GetTimerId())`
	`228`	`+func (h decisionsHelper) newTimerDecisionStateMachine(attributes s.StartTimerDecisionAttributes) *timerDecisionStateMachine {`
	`229`	`+ base := h.newDecisionStateMachineBase(decisionTypeTimer, attributes.GetTimerId())`
`228`	`230`	`return &timerDecisionStateMachine{`
`229`	`231`	`decisionStateMachineBase: base,`
`230`	`232`	`attributes: attributes,`
`231`	`233`	`}`
`232`	`234`	`}`
`233`	`235`
`234`		`-func newChildWorkflowDecisionStateMachine(attributes s.StartChildWorkflowExecutionDecisionAttributes) childWorkflowDecisionStateMachine {`
`235`		`- base := newDecisionStateMachineBase(decisionTypeChildWorkflow, attributes.GetWorkflowId())`
	`236`	`+func (h decisionsHelper) newChildWorkflowDecisionStateMachine(attributes s.StartChildWorkflowExecutionDecisionAttributes) *childWorkflowDecisionStateMachine {`
	`237`	`+ base := h.newDecisionStateMachineBase(decisionTypeChildWorkflow, attributes.GetWorkflowId())`
`236`	`238`	`return &childWorkflowDecisionStateMachine{`
`237`	`239`	`decisionStateMachineBase: base,`
`238`	`240`	`attributes: attributes,`
`239`	`241`	`}`
`240`	`242`	`}`
`241`	`243`
`242`		`-func newNaiveDecisionStateMachine(decisionType decisionType, id string, decision s.Decision) naiveDecisionStateMachine {`
`243`		`- base := newDecisionStateMachineBase(decisionType, id)`
	`244`	`+func (h decisionsHelper) newNaiveDecisionStateMachine(decisionType decisionType, id string, decision s.Decision) *naiveDecisionStateMachine {`
	`245`	`+ base := h.newDecisionStateMachineBase(decisionType, id)`
`244`	`246`	`return &naiveDecisionStateMachine{`
`245`	`247`	`decisionStateMachineBase: base,`
`246`	`248`	`decision: decision,`
`247`	`249`	`}`
`248`	`250`	`}`
`249`	`251`
`250`		`-func newMarkerDecisionStateMachine(id string, attributes s.RecordMarkerDecisionAttributes) markerDecisionStateMachine {`
	`252`	`+func (h decisionsHelper) newMarkerDecisionStateMachine(id string, attributes s.RecordMarkerDecisionAttributes) *markerDecisionStateMachine {`
`251`	`253`	`d := createNewDecision(s.DecisionTypeRecordMarker)`
`252`	`254`	`d.RecordMarkerDecisionAttributes = attributes`
`253`	`255`	`return &markerDecisionStateMachine{`
`254`		`- naiveDecisionStateMachine: newNaiveDecisionStateMachine(decisionTypeMarker, id, d),`
	`256`	`+ naiveDecisionStateMachine: h.newNaiveDecisionStateMachine(decisionTypeMarker, id, d),`
`255`	`257`	`}`
`256`	`258`	`}`
`257`	`259`
`258`		`-func newCancelExternalWorkflowStateMachine(attributes s.RequestCancelExternalWorkflowExecutionDecisionAttributes, cancellationID string) cancelExternalWorkflowDecisionStateMachine {`
	`260`	`+func (h decisionsHelper) newCancelExternalWorkflowStateMachine(attributes s.RequestCancelExternalWorkflowExecutionDecisionAttributes, cancellationID string) *cancelExternalWorkflowDecisionStateMachine {`
`259`	`261`	`d := createNewDecision(s.DecisionTypeRequestCancelExternalWorkflowExecution)`
`260`	`262`	`d.RequestCancelExternalWorkflowExecutionDecisionAttributes = attributes`
`261`	`263`	`return &cancelExternalWorkflowDecisionStateMachine{`
`262`		`- naiveDecisionStateMachine: newNaiveDecisionStateMachine(decisionTypeCancellation, cancellationID, d),`
	`264`	`+ naiveDecisionStateMachine: h.newNaiveDecisionStateMachine(decisionTypeCancellation, cancellationID, d),`
`263`	`265`	`}`
`264`	`266`	`}`
`265`	`267`
`266`		`-func newSignalExternalWorkflowStateMachine(attributes s.SignalExternalWorkflowExecutionDecisionAttributes, signalID string) signalExternalWorkflowDecisionStateMachine {`
	`268`	`+func (h decisionsHelper) newSignalExternalWorkflowStateMachine(attributes s.SignalExternalWorkflowExecutionDecisionAttributes, signalID string) *signalExternalWorkflowDecisionStateMachine {`
`267`	`269`	`d := createNewDecision(s.DecisionTypeSignalExternalWorkflowExecution)`
`268`	`270`	`d.SignalExternalWorkflowExecutionDecisionAttributes = attributes`
`269`	`271`	`return &signalExternalWorkflowDecisionStateMachine{`
`270`		`- naiveDecisionStateMachine: newNaiveDecisionStateMachine(decisionTypeSignal, signalID, d),`
	`272`	`+ naiveDecisionStateMachine: h.newNaiveDecisionStateMachine(decisionTypeSignal, signalID, d),`
`271`	`273`	`}`
`272`	`274`	`}`
`273`	`275`
`@@ -295,6 +297,13 @@ func (d *decisionStateMachineBase) moveState(newState decisionState, event strin`
`295`	`297`	`d.history = append(d.history, event)`
`296`	`298`	`d.state = newState`
`297`	`299`	`d.history = append(d.history, newState.String())`
	`300`	`+`
	`301`	`+ if newState == decisionStateCompleted {`
	`302`	`+ if elem, ok := d.helper.decisions[d.getID()]; ok {`
	`303`	`+ d.helper.orderedDecisions.Remove(elem)`
	`304`	`+ delete(d.helper.decisions, d.getID())`
	`305`	`+ }`
	`306`	`+ }`
`298`	`307`	`}`
`299`	`308`
`300`	`309`	`func (d *decisionStateMachineBase) failStateTransition(event string) {`
`@@ -660,12 +669,16 @@ func (h *decisionsHelper) getDecision(id decisionID) decisionStateMachine {`
`660`	`669`	`}`
`661`	`670`
`662`	`671`	`func (h *decisionsHelper) addDecision(decision decisionStateMachine) {`
	`672`	`+ if _, ok := h.decisions[decision.getID()]; ok {`
	`673`	`+ panicMsg := fmt.Sprintf("adding duplicate decision %v", decision)`
	`674`	`+ panic(panicMsg)`
	`675`	`+ }`
`663`	`676`	`element := h.orderedDecisions.PushBack(decision)`
`664`	`677`	`h.decisions[decision.getID()] = element`
`665`	`678`	`}`
`666`	`679`
`667`	`680`	`func (h decisionsHelper) scheduleActivityTask(attributes s.ScheduleActivityTaskDecisionAttributes) decisionStateMachine {`
`668`		`- decision := newActivityDecisionStateMachine(attributes)`
	`681`	`+ decision := h.newActivityDecisionStateMachine(attributes)`
`669`	`682`	`h.addDecision(decision)`
`670`	`683`	`return decision`
`671`	`684`	`}`
`@@ -739,7 +752,7 @@ func (h *decisionsHelper) recordVersionMarker(changeID string, version Version,`
`739`	`752`	`Details: details, // Keep`
`740`	`753`	`}`
`741`	`754`
`742`		`- decision := newMarkerDecisionStateMachine(markerID, recordMarker)`
	`755`	`+ decision := h.newMarkerDecisionStateMachine(markerID, recordMarker)`
`743`	`756`	`h.addDecision(decision)`
`744`	`757`	`return decision`
`745`	`758`	`}`
`@@ -750,7 +763,7 @@ func (h *decisionsHelper) recordSideEffectMarker(sideEffectID int32, data []byte`
`750`	`763`	`MarkerName: common.StringPtr(sideEffectMarkerName),`
`751`	`764`	`Details: data,`
`752`	`765`	`}`
`753`		`- decision := newMarkerDecisionStateMachine(markerID, attributes)`
	`766`	`+ decision := h.newMarkerDecisionStateMachine(markerID, attributes)`
`754`	`767`	`h.addDecision(decision)`
`755`	`768`	`return decision`
`756`	`769`	`}`
`@@ -761,7 +774,7 @@ func (h *decisionsHelper) recordLocalActivityMarker(activityID string, result []`
`761`	`774`	`MarkerName: common.StringPtr(localActivityMarkerName),`
`762`	`775`	`Details: result,`
`763`	`776`	`}`
`764`		`- decision := newMarkerDecisionStateMachine(markerID, attributes)`
	`777`	`+ decision := h.newMarkerDecisionStateMachine(markerID, attributes)`
`765`	`778`	`h.addDecision(decision)`
`766`	`779`	`return decision`
`767`	`780`	`}`
`@@ -772,13 +785,13 @@ func (h *decisionsHelper) recordMutableSideEffectMarker(mutableSideEffectID stri`
`772`	`785`	`MarkerName: common.StringPtr(mutableSideEffectMarkerName),`
`773`	`786`	`Details: data,`
`774`	`787`	`}`
`775`		`- decision := newMarkerDecisionStateMachine(markerID, attributes)`
	`788`	`+ decision := h.newMarkerDecisionStateMachine(markerID, attributes)`
`776`	`789`	`h.addDecision(decision)`
`777`	`790`	`return decision`
`778`	`791`	`}`
`779`	`792`
`780`	`793`	`func (h decisionsHelper) startChildWorkflowExecution(attributes s.StartChildWorkflowExecutionDecisionAttributes) decisionStateMachine {`
`781`		`- decision := newChildWorkflowDecisionStateMachine(attributes)`
	`794`	`+ decision := h.newChildWorkflowDecisionStateMachine(attributes)`
`782`	`795`	`h.addDecision(decision)`
`783`	`796`	`return decision`
`784`	`797`	`}`
`@@ -833,7 +846,7 @@ func (h *decisionsHelper) requestCancelExternalWorkflowExecution(domain, workflo`
`833`	`846`	`Control: []byte(cancellationID),`
`834`	`847`	`ChildWorkflowOnly: common.BoolPtr(false),`
`835`	`848`	`}`
`836`		`- decision := newCancelExternalWorkflowStateMachine(attributes, cancellationID)`
	`849`	`+ decision := h.newCancelExternalWorkflowStateMachine(attributes, cancellationID)`
`837`	`850`	`h.addDecision(decision)`
`838`	`851`
`839`	`852`	`return decision`
`@@ -893,7 +906,7 @@ func (h *decisionsHelper) signalExternalWorkflowExecution(domain, workflowID, ru`
`893`	`906`	`Control: []byte(signalID),`
`894`	`907`	`ChildWorkflowOnly: common.BoolPtr(childWorkflowOnly),`
`895`	`908`	`}`
`896`		`- decision := newSignalExternalWorkflowStateMachine(attributes, signalID)`
	`909`	`+ decision := h.newSignalExternalWorkflowStateMachine(attributes, signalID)`
`897`	`910`	`h.addDecision(decision)`
`898`	`911`	`return decision`
`899`	`912`	`}`
`@@ -925,7 +938,7 @@ func (h *decisionsHelper) getSignalID(initiatedEventID int64) string {`
`925`	`938`	`}`
`926`	`939`
`927`	`940`	`func (h decisionsHelper) startTimer(attributes s.StartTimerDecisionAttributes) decisionStateMachine {`
`928`		`- decision := newTimerDecisionStateMachine(attributes)`
	`941`	`+ decision := h.newTimerDecisionStateMachine(attributes)`
`929`	`942`	`h.addDecision(decision)`
`930`	`943`	`return decision`
`931`	`944`	`}`
Original file line number	Diff line number	Diff line change
`@@ -593,9 +593,9 @@ func (weh *workflowExecutionEventHandlerImpl) ProcessEvent(`
`593`	`593`	`event *m.HistoryEvent,`
`594`	`594`	`isReplay bool,`
`595`	`595`	`isLast bool,`
`596`		`-) (result []*m.Decision, err error) {`
	`596`	`+) (err error) {`
`597`	`597`	`if event == nil {`
`598`		`- return nil, errors.New("nil event provided")`
	`598`	`+ return errors.New("nil event provided")`
`599`	`599`	`}`
`600`	`600`	`defer func() {`
`601`	`601`	`if p := recover(); p != nil {`
`@@ -746,7 +746,7 @@ func (weh *workflowExecutionEventHandlerImpl) ProcessEvent(`
`746`	`746`	`}`
`747`	`747`
`748`	`748`	`if err != nil {`
`749`		`- return nil, err`
	`749`	`+ return err`
`750`	`750`	`}`
`751`	`751`
`752`	`752`	`// When replaying histories to get stack trace or current state the last event might be not`
`@@ -756,7 +756,7 @@ func (weh *workflowExecutionEventHandlerImpl) ProcessEvent(`
`756`	`756`	`weh.workflowDefinition.OnDecisionTaskStarted()`
`757`	`757`	`}`
`758`	`758`
`759`		`- return weh.decisionsHelper.getDecisions(true), nil`
	`759`	`+ return nil`
`760`	`760`	`}`
`761`	`761`
`762`	`762`	`func (weh *workflowExecutionEventHandlerImpl) ProcessQuery(queryType string, queryArgs []byte) ([]byte, error) {`
`@@ -928,7 +928,7 @@ func (weh *workflowExecutionEventHandlerImpl) handleLocalActivityMarker(markerDa`
`928`	`928`	`return nil`
`929`	`929`	`}`
`930`	`930`
`931`		`-func (weh workflowExecutionEventHandlerImpl) ProcessLocalActivityResult(lar localActivityResult) ([]*m.Decision, error) {`
	`931`	`+func (weh workflowExecutionEventHandlerImpl) ProcessLocalActivityResult(lar localActivityResult) error {`
`932`	`932`	`// convert local activity result and error to marker data`
`933`	`933`	`lamd := localActivityMarkerData{`
`934`	`934`	`ActivityID: lar.task.activityID,`
`@@ -945,7 +945,7 @@ func (weh workflowExecutionEventHandlerImpl) ProcessLocalActivityResult(lar lo`
`945`	`945`	`// encode marker data`
`946`	`946`	`markerData, err := weh.encodeArg(lamd)`
`947`	`947`	`if err != nil {`
`948`		`- return nil, err`
	`948`	`+ return err`
`949`	`949`	`}`
`950`	`950`
`951`	`951`	`// create marker event for local activity result`
Original file line number	Diff line number	Diff line change
`@@ -54,7 +54,7 @@ type (`
`54`	`54`	`workflowExecutionEventHandler interface {`
`55`	`55`	`// Process a single event and return the assosciated decisions.`
`56`	`56`	`// Return List of decisions made, any error.`
`57`		`- ProcessEvent(event s.HistoryEvent, isReplay bool, isLast bool) ([]s.Decision, error)`
	`57`	`+ ProcessEvent(event *s.HistoryEvent, isReplay bool, isLast bool) error`
`58`	`58`	`// ProcessQuery process a query request.`
`59`	`59`	`ProcessQuery(queryType string, queryArgs []byte) ([]byte, error)`
`60`	`60`	`StackTrace() string`
`@@ -293,8 +293,7 @@ OrderEvents:`
`293`	`293`	`}`
`294`	`294`
`295`	`295`	`func isPreloadMarkerEvent(event *s.HistoryEvent) bool {`
`296`		`- return event.GetEventType() == s.EventTypeMarkerRecorded &&`
`297`		`- event.MarkerRecordedEventAttributes.GetMarkerName() != localActivityMarkerName`
	`296`	`+ return event.GetEventType() == s.EventTypeMarkerRecorded`
`298`	`297`	`}`
`299`	`298`
`300`	`299`	`// newWorkflowTaskHandler returns an implementation of workflow task handler.`
`@@ -653,9 +652,12 @@ ProcessEvents:`
`653`	`652`	`}`
`654`	`653`	`// Markers are from the events that are produced from the current decision`
`655`	`654`	`for _, m := range markers {`
`656`		`- _, err := eventHandler.ProcessEvent(m, true, false)`
`657`		`- if err != nil {`
`658`		`- return nil, err`
	`655`	`+ if m.MarkerRecordedEventAttributes.GetMarkerName() != localActivityMarkerName {`
	`656`	`+ // local activity marker needs to be applied after decision task started event`
	`657`	`+ err := eventHandler.ProcessEvent(m, true, false)`
	`658`	`+ if err != nil {`
	`659`	`+ return nil, err`
	`660`	`+ }`
`659`	`661`	`}`
`660`	`662`	`}`
`661`	`663`
`@@ -672,23 +674,26 @@ ProcessEvents:`
`672`	`674`	`return nil, err`
`673`	`675`	`}`
`674`	`676`
`675`		`- eventDecisions, err := eventHandler.ProcessEvent(event, isInReplay, isLast)`
	`677`	`+ err = eventHandler.ProcessEvent(event, isInReplay, isLast)`
`676`	`678`	`if err != nil {`
`677`	`679`	`return nil, err`
`678`	`680`	`}`
	`681`	`+ }`
`679`	`682`
`680`		`- if eventDecisions != nil {`
`681`		`- if !isInReplay {`
`682`		`- w.newDecisions = append(w.newDecisions, eventDecisions...)`
`683`		`- } else if !skipReplayCheck {`
`684`		`- replayDecisions = append(replayDecisions, eventDecisions...)`
	`683`	`+ // now apply local activity markers`
	`684`	`+ for _, m := range markers {`
	`685`	`+ if m.MarkerRecordedEventAttributes.GetMarkerName() == localActivityMarkerName {`
	`686`	`+ err := eventHandler.ProcessEvent(m, true, false)`
	`687`	`+ if err != nil {`
	`688`	`+ return nil, err`
`685`	`689`	`}`
`686`	`690`	`}`
`687`		`-`
`688`		`- if w.isWorkflowCompleted {`
`689`		`- // If workflow is already completed then we can break from processing`
`690`		`- // further decisions.`
`691`		`- break ProcessEvents`
	`691`	`+ }`
	`692`	`+ isReplay := len(reorderedEvents) > 0 && reorderedHistory.IsReplayEvent(reorderedEvents[len(reorderedEvents)-1])`
	`693`	`+ if isReplay {`
	`694`	`+ eventDecisions := eventHandler.decisionsHelper.getDecisions(true)`
	`695`	`+ if len(eventDecisions) > 0 && !skipReplayCheck {`
	`696`	`+ replayDecisions = append(replayDecisions, eventDecisions...)`
`692`	`697`	`}`
`693`	`698`	`}`
`694`	`699`	`}`
`@@ -728,15 +733,11 @@ ProcessEvents:`
`728`	`733`	`}`
`729`	`734`
`730`	`735`	`func (w workflowExecutionContextImpl) ProcessLocalActivityResult(lar localActivityResult) (interface{}, error) {`
`731`		`- eventDecisions, err := w.eventHandler.ProcessLocalActivityResult(lar)`
	`736`	`+ err := w.eventHandler.ProcessLocalActivityResult(lar)`
`732`	`737`	`if err != nil {`
`733`	`738`	`return nil, err`
`734`	`739`	`}`
`735`	`740`
`736`		`- if eventDecisions != nil {`
`737`		`- w.newDecisions = append(w.newDecisions, eventDecisions...)`
`738`		`- }`
`739`		`-`
`740`	`741`	`return w.CompleteDecisionTask(true), nil`
`741`	`742`	`}`
`742`	`743`
`@@ -760,6 +761,11 @@ func (w *workflowExecutionContextImpl) CompleteDecisionTask(waitLocalActivities`
`760`	`761`	`}`
`761`	`762`	`}`
`762`	`763`
	`764`	`+ eventDecisions := w.eventHandler.decisionsHelper.getDecisions(true)`
	`765`	`+ if len(eventDecisions) > 0 {`
	`766`	`+ w.newDecisions = append(w.newDecisions, eventDecisions...)`
	`767`	`+ }`
	`768`	`+`
`763`	`769`	`completeRequest := w.wth.completeWorkflow(w.eventHandler, w.currentDecisionTask, w, w.newDecisions, !waitLocalActivities)`
`764`	`770`	`w.clearCurrentTask()`
`765`	`771`