Address false positives in replay mode

Author: taylanisikdemir

internal/internal_task_handlers.go

@@ -261,6 +261,28 @@ func isDecisionEvent(eventType s.EventType) bool {
 	}
 }
 
+// isDecisionEventForReplay is different from isDecisionEvent because during replays
+// we want to intentionally ignore workflow complete/fail/cancel/continueasnew events so that
+// decision tree replays matches with the workflow processing respond tasks
+func isDecisionEventForReplay(eventType s.EventType) bool {
+	switch eventType {
+	case
+		s.EventTypeActivityTaskScheduled,
+		s.EventTypeActivityTaskCancelRequested,
+		s.EventTypeTimerStarted,
+		s.EventTypeTimerCanceled,
+		s.EventTypeCancelTimerFailed,
+		s.EventTypeMarkerRecorded,
+		s.EventTypeStartChildWorkflowExecutionInitiated,
+		s.EventTypeRequestCancelExternalWorkflowExecutionInitiated,
+		s.EventTypeSignalExternalWorkflowExecutionInitiated,
+		s.EventTypeUpsertWorkflowSearchAttributes:
+		return true
+	default:
+		return false
+	}
+}
+
 // NextDecisionEvents returns events that there processed as new by the next decision.
 // TODO(maxim): Refactor to return a struct instead of multiple parameters
 func (eh *history) NextDecisionEvents() (result []*s.HistoryEvent, markers []*s.HistoryEvent, binaryChecksum *string, err error) {
@@ -840,6 +862,19 @@ process_Workflow_Loop:
 	return response, err
 }
 
+// ProcessWorkflowTask processes the given workflow which includes
+// - fetching, reordering and replaying historical decision events. (Decision events in this context is an umbrella term for workflow relevant events)
+// - state machine is incrementally built with every decision.
+// - state machine makes sure that when a workflow restarts for some reason same activities (or timers etc.) are not called again and previous result state is loaded into memory
+//
+// Note about Replay tests mode:
+//
+//	This mode works by replaying the historical decision events responses (as defined in isDecisionEventForReplay())
+//	and comparing these with the replays gotten from state machine
+//
+//	Compared to isDecisionEvent(), isDecisionEventForReplay() omits the following events even though they are workflow relevant respond events:
+//		complete/failed/cancel/continueasnew
+//	The reason is that state machine doesn't have a correspondong decision for these so they cause false positive non-determinism errors in Replay tests.
 func (w *workflowExecutionContextImpl) ProcessWorkflowTask(workflowTask *workflowTask) (interface{}, error) {
 	task := workflowTask.task
 	historyIterator := workflowTask.historyIterator
@@ -895,8 +930,15 @@ ProcessEvents:
 		for i, event := range reorderedEvents {
 			isInReplay := reorderedHistory.IsReplayEvent(event)
 			isLast := !isInReplay && i == len(reorderedEvents)-1
-			if !skipReplayCheck && isDecisionEvent(event.GetEventType()) {
-				respondEvents = append(respondEvents, event)
+			if !skipReplayCheck {
+				isDecisionEventFn := isDecisionEvent
+				if isInReplay {
+					isDecisionEventFn = isDecisionEventForReplay
+				}
+
+				if isDecisionEventFn(event.GetEventType()) {
+					respondEvents = append(respondEvents, event)
+				}
 			}
 
 			if isPreloadMarkerEvent(event) {
@@ -914,7 +956,8 @@ ProcessEvents:
 			if err != nil {
 				return nil, err
 			}
-			if w.isWorkflowCompleted {
+
+			if w.isWorkflowCompleted && !isInReplay {
 				break ProcessEvents
 			}
 		}
@@ -932,8 +975,7 @@ ProcessEvents:
 			}
 		}
 		isReplay := len(reorderedEvents) > 0 && reorderedHistory.IsReplayEvent(reorderedEvents[len(reorderedEvents)-1])
-		lastDecisionEventsForReplayTest := isReplayTest && !reorderedHistory.HasNextDecisionEvents()
-		if isReplay && !lastDecisionEventsForReplayTest {
+		if isReplay {
 			eventDecisions := eventHandler.decisionsHelper.getDecisions(true)
 			if len(eventDecisions) > 0 && !skipReplayCheck {
 				replayDecisions = append(replayDecisions, eventDecisions...)
@@ -980,7 +1022,6 @@ ProcessEvents:
 	}
 
 	if nonDeterministicErr != nil {
-
 		w.wth.metricsScope.GetTaggedScope(tagWorkflowType, task.WorkflowType.GetName()).Counter(metrics.NonDeterministicError).Inc(1)
 		w.wth.logger.Error("non-deterministic-error",
 			zap.String(tagWorkflowType, task.WorkflowType.GetName()),

test/replaytests/replay_test.go

@@ -73,6 +73,7 @@ func TestGreetingsWorkflow(t *testing.T) {
 }
 
 // Should have failed but passed. Maybe, because the result recorded in history still matches the return type of the workflow.
+// TODO(remove comment): Passes now
 func TestGreetingsWorkflow3(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
 	replayer.RegisterActivityWithOptions(getNameActivity3, activity.RegisterOptions{Name: "main.getNameActivity", DisableAlreadyRegisteredCheck: true})
@@ -121,14 +122,15 @@ func TestExclusiveChoiceWorkflowWithUnregisteredActivity(t *testing.T) {
 // that registered activity is different from executed activity.
 // The replayer relies on whatever is recorded in the History so as long as the main activity name in the options matched partially
 // it doesn't raise errors.
+// TODO(remove comment): Replayer now catches this
 func TestExclusiveChoiceWorkflowWithDifferentActvityCombo(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
 
 	replayer.RegisterWorkflowWithOptions(exclusiveChoiceWorkflow2, workflow.RegisterOptions{Name: "choice"})
 	replayer.RegisterActivityWithOptions(getAppleOrderActivity, activity.RegisterOptions{Name: "main.getOrderActivity"})
 	replayer.RegisterActivityWithOptions(orderAppleActivity, activity.RegisterOptions{Name: "testactivity"})
 	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "choice.json")
-	require.NoError(t, err)
+	assert.ErrorContains(t, err, "nondeterministic workflow")
 }
 
 func TestBranchWorkflow(t *testing.T) {
@@ -155,9 +157,10 @@ func TestBranchWorkflowWithExtraBranch(t *testing.T) {
 func TestSequentialStepsWorkflow(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
 
-	replayer.RegisterWorkflowWithOptions(sequantialStepsWorkflow, workflow.RegisterOptions{Name: "sequentialStepsWorkflow"})
+	replayer.RegisterWorkflowWithOptions(replayerHelloWorldWorkflow, workflow.RegisterOptions{Name: "fx.ReplayerHelloWorldWorkflow"})
+	replayer.RegisterActivityWithOptions(replayerHelloWorldActivity, activity.RegisterOptions{Name: "replayerhello"})
 
-	// branch.json file contains history of a run with 3 activity calls
+	// sequential.json file contains history of a run with 2 activity calls sequentially
 	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "sequential.json")
 	assert.ErrorContains(t, err, "nondeterministic workflow")
 }
@@ -173,11 +176,12 @@ func TestParallel(t *testing.T) {
 
 // Should have failed since the first go routine has only one branch whereas the history has two branches.
 // The replayer totally misses this change.
+// TODO(remove comment): Replayer now catches this
 func TestParallel2(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
 
 	replayer.RegisterWorkflowWithOptions(sampleParallelWorkflow2, workflow.RegisterOptions{Name: "branch2"})
 
 	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "branch2.json")
-	require.NoError(t, err)
+	assert.ErrorContains(t, err, "nondeterministic workflow")
 }