fix and improve test coverage of replays, add explanation comments to not-obvious places

Author: taylanisikdemir

internal/error.go

@@ -508,6 +508,7 @@ func (e *NonDeterministicError) Error() string {
 	case "mismatch":
 		// historical text
 		return "nondeterministic workflow: " +
+			"mismatching history event and replay decision found. " +
 			"history event is " + e.HistoryEventText + ", " +
 			"replay decision is " + e.DecisionText
 	default:

internal/internal_activity.go

@@ -432,6 +432,11 @@ func deSerializeFunctionResult(f interface{}, result []byte, to interface{}, dat
 	}
 
 	// For everything we return result.
+	// TODO(remove comment):
+	// Code reaches here for 2 cases
+	// 1. activity is executed by name (not the func pointer) and it wasn't registered
+	// 2. activity is executed by func pointer and the signature indicates it doesn't/can't return data.
+	//		for example it only has one return parameter (which can only be be error).
 	return decodeArg(dataConverter, result, to)
 }
 

internal/internal_task_handlers.go

@@ -947,6 +947,9 @@ ProcessEvents:
 				return nil, err
 			}
 
+			// Break the event processing loop if the workflow is completed except in replay mode.
+			// In replay mode we check for nondeterminism cases and
+			// breaking the loop causes missing events in respondEvents which can cause false positives or false negatives.
 			if w.isWorkflowCompleted && !isInReplay {
 				break ProcessEvents
 			}

internal/internal_task_handlers_test.go

@@ -887,10 +887,9 @@ func (t *TaskHandlersTestSuite) TestWorkflowTask_NondeterministicLogNonexistingI
 	require.NotNil(t.T(), replayErrorField)
 	require.Equal(t.T(), zapcore.ErrorType, replayErrorField.Type)
 	require.ErrorContains(t.T(), replayErrorField.Interface.(error),
-		"nondeterministic workflow: "+
+		"nondeterministic workflow: mismatching history event and replay decision found. "+
 			"history event is ActivityTaskScheduled: (ActivityId:NotAnActivityID, ActivityType:(Name:pkg.Greeter_Activity), TaskList:(Name:taskList), Input:[]), "+
-			"replay decision is ScheduleActivityTask: (ActivityId:0, ActivityType:(Name:Greeter_Activity), TaskList:(Name:taskList)")
-
+			"replay decision is ScheduleActivityTask: (ActivityId:0, ActivityType:(Name:Greeter_Activity), TaskList:(Name:taskList), Input:[], ScheduleToCloseTimeoutSeconds:120, ScheduleToStartTimeoutSeconds:60, StartToCloseTimeoutSeconds:60, HeartbeatTimeoutSeconds:20, Header:(Fields:map[]))")
 }
 
 func (t *TaskHandlersTestSuite) TestWorkflowTask_WorkflowReturnsPanicError() {

test/replaytests/choice.json

@@ -176,7 +176,34 @@
   },
   {
     "eventId": 12,
-    "timestamp": 1679427717321911295,
+    "timestamp": 1679427717321780254,
+    "eventType": "ActivityTaskStarted",
+    "version": 0,
+    "taskId": 5243011,
+    "activityTaskStartedEventAttributes": {
+      "scheduledEventId": 11,
+      "identity": "82203@agautam-NV709R969P@choiceGroup@41d230ae-253a-4d01-9079-322ef05c09fb",
+      "requestId": "ae2aad96-6588-4359-807b-a39a16f0896a",
+      "attempt": 0,
+      "lastFailureReason": ""
+    }
+  },
+  {
+    "eventId": 13,
+    "timestamp": 1679427717321780255,
+    "eventType": "ActivityTaskCompleted",
+    "version": 0,
+    "taskId": 5243000,
+    "activityTaskCompletedEventAttributes": {
+      "result": "ImJhbmFuYSIK",
+      "scheduledEventId": 11,
+      "startedEventId": 12,
+      "identity": "82203@agautam-NV709R969P@choiceGroup@41d230ae-253a-4d01-9079-322ef05c09fb"
+    }
+  },
+  {
+    "eventId": 14,
+    "timestamp": 1679427717321780256,
     "eventType": "WorkflowExecutionCompleted",
     "version": 0,
     "taskId": 5243011,

test/replaytests/exclusive_choice_workflow.go

@@ -39,7 +39,7 @@ const (
 	orderChoiceCherry = "cherry"
 )
 
-// exclusiveChoiceWorkflow Workflow Decider. This workflow executes Cherry order.
+// exclusiveChoiceWorkflow executes main.getOrderActivity and executes either cherry or banana activity depends on what main.getOrderActivity returns
 func exclusiveChoiceWorkflow(ctx workflow.Context) error {
 	// Get order.
 	ao := workflow.ActivityOptions{
@@ -50,7 +50,7 @@ func exclusiveChoiceWorkflow(ctx workflow.Context) error {
 	ctx = workflow.WithActivityOptions(ctx, ao)
 
 	var orderChoice string
-	err := workflow.ExecuteActivity(ctx, getOrderActivity).Get(ctx, &orderChoice)
+	err := workflow.ExecuteActivity(ctx, "main.getOrderActivity").Get(ctx, &orderChoice)
 	if err != nil {
 		return err
 	}
@@ -60,9 +60,9 @@ func exclusiveChoiceWorkflow(ctx workflow.Context) error {
 	// choose next activity based on order result
 	switch orderChoice {
 	case orderChoiceBanana:
-		workflow.ExecuteActivity(ctx, orderBananaActivity, orderChoice)
+		workflow.ExecuteActivity(ctx, "main.orderBananaActivity", orderChoice)
 	case orderChoiceCherry:
-		workflow.ExecuteActivity(ctx, orderCherryActivity, orderChoice)
+		workflow.ExecuteActivity(ctx, "main.orderCherryActivity", orderChoice)
 	default:
 		logger.Error("Unexpected order", zap.String("Choice", orderChoice))
 	}
@@ -71,8 +71,8 @@ func exclusiveChoiceWorkflow(ctx workflow.Context) error {
 	return nil
 }
 
-// This workflow explicitly executes Apple Activity received from the getorderActivity.
-func exclusiveChoiceWorkflow2(ctx workflow.Context) error {
+// exclusiveChoiceWorkflow executes main.getOrderActivity and executes either cherry or banana activity depends on what main.getOrderActivity returns
+func exclusiveChoiceWorkflowAlwaysCherry(ctx workflow.Context) error {
 	// Get order.
 	ao := workflow.ActivityOptions{
 		ScheduleToStartTimeout: time.Minute,
@@ -82,40 +82,25 @@ func exclusiveChoiceWorkflow2(ctx workflow.Context) error {
 	ctx = workflow.WithActivityOptions(ctx, ao)
 
 	var orderChoice string
-	err := workflow.ExecuteActivity(ctx, getAppleOrderActivity).Get(ctx, &orderChoice)
+	err := workflow.ExecuteActivity(ctx, "main.getOrderActivity").Get(ctx, &orderChoice)
 	if err != nil {
 		return err
 	}
 
 	logger := workflow.GetLogger(ctx)
+	logger.Sugar().Infof("Got order for %s but will ignore and order cherry!!", orderChoice)
 
-	// choose next activity based on order result. It's apple in this case.
-	switch orderChoice {
-	case orderChoiceApple:
-		workflow.ExecuteActivity(ctx, orderAppleActivity, orderChoice)
-	default:
-		logger.Error("Unexpected order", zap.String("Choice", orderChoice))
-	}
+	workflow.ExecuteActivity(ctx, "main.orderCherryActivity", orderChoice)
 
 	logger.Info("Workflow completed.")
 	return nil
 }
 
-func getOrderActivity() (string, error) {
-	fmt.Printf("Order is for Cherry")
-	return "cherry", nil
-}
-
-func getAppleOrderActivity() (string, error) {
+func getBananaOrderActivity() (string, error) {
 	fmt.Printf("Order is for Apple")
 	return "apple", nil
 }
 
-func orderAppleActivity(choice string) error {
-	fmt.Printf("Order choice: %v\n", choice)
-	return nil
-}
-
 func orderBananaActivity(choice string) error {
 	fmt.Printf("Order choice: %v\n", choice)
 	return nil

test/replaytests/replay_test.go

@@ -35,6 +35,7 @@ import (
 	"go.uber.org/cadence/workflow"
 )
 
+// Basic happy paths
 func TestReplayWorkflowHistoryFromFile(t *testing.T) {
 	for _, testFile := range []string{"basic.json", "basic_new.json", "version.json", "version_new.json"} {
 		t.Run("replay_"+strings.Split(testFile, ".")[0], func(t *testing.T) {
@@ -48,6 +49,7 @@ func TestReplayWorkflowHistoryFromFile(t *testing.T) {
 	}
 }
 
+// Child workflow happy path
 func TestReplayChildWorkflowBugBackport(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
 	replayer.RegisterWorkflowWithOptions(childWorkflow, workflow.RegisterOptions{Name: "child"})
@@ -62,18 +64,18 @@ func TestGreetingsWorkflowforActivity(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
 	replayer.RegisterWorkflowWithOptions(greetingsWorkflowActivity, workflow.RegisterOptions{Name: "greetings"})
 	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "greetings.json")
-	require.Error(t, err)
+	assert.ErrorContains(t, err, "nondeterministic workflow: mismatching history event and replay decision found")
 }
 
+// Simple greeting workflow with 3 activities executed sequentially: getGreetingsActivity, getNameActivity, sayGreetingsActivity
 func TestGreetingsWorkflow(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
 	replayer.RegisterWorkflowWithOptions(greetingsWorkflow, workflow.RegisterOptions{Name: "greetings"})
 	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "greetings.json")
 	require.NoError(t, err)
 }
 
-// Should have failed but passed. Maybe, because the result recorded in history still matches the return type of the workflow.
-// TODO(remove comment): Debug why is this still missed
+// Return types of activity change is not considered non-determinism (at least for now) so this test doesn't find non-determinism error
 func TestGreetingsWorkflow3(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
 	replayer.RegisterActivityWithOptions(getNameActivity3, activity.RegisterOptions{Name: "main.getNameActivity", DisableAlreadyRegisteredCheck: true})
@@ -82,107 +84,100 @@ func TestGreetingsWorkflow3(t *testing.T) {
 	require.NoError(t, err)
 }
 
-// Fails because the expected signature was different from history.
-func TestGreetingsWorkflow4(t *testing.T) {
+// The recorded history has following activities in this order:  main.getOrderActivity, main.orderBananaActivity
+// This test runs a version of choice workflow which does the exact same thing so no errors expected.
+func TestExclusiveChoiceWorkflowSuccess(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
-	replayer.RegisterActivityWithOptions(getNameActivity4, activity.RegisterOptions{Name: "main.getNameActivity", DisableAlreadyRegisteredCheck: true})
-	replayer.RegisterWorkflowWithOptions(greetingsWorkflow4, workflow.RegisterOptions{Name: "greetings"})
-	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "greetings.json")
-	require.Error(t, err)
+	replayer.RegisterWorkflowWithOptions(exclusiveChoiceWorkflow, workflow.RegisterOptions{Name: "choice"})
+	replayer.RegisterActivityWithOptions(getBananaOrderActivity, activity.RegisterOptions{Name: "main.getOrderActivity"})
+	replayer.RegisterActivityWithOptions(orderBananaActivity, activity.RegisterOptions{Name: "main.orderBananaActivity"})
+	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "choice.json")
+	require.NoError(t, err)
 }
 
-// Panic with failed to register activity. This passes in cadence_samples because it's registered in Helper.
-// To test it on cadence_samples change the https://github.com/uber-common/cadence-samples/blob/master/cmd/samples/recipes/greetings/greetings_workflow.go
-// to include the extra return types in getNameActivity.
-func TestGreetingsWorkflow2(t *testing.T) {
-
-	t.Skip("Panic with failed to register activity. Here the activity returns incompatible arguments so the test should fail")
+// The recorded history has following activities in this order:  main.getOrderActivity, main.orderBananaActivity
+// This test runs a version of choice workflow which does the exact same thing but the activities are not registered.
+// It doesn't matter for replayer so no exceptions expected.
+// The reason is that activity result decoding logic just passes the result back to the given pointer
+func TestExclusiveChoiceWorkflowActivitiyRegistrationMissing(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
-	replayer.RegisterActivityWithOptions(getNameActivity2, activity.RegisterOptions{Name: "main.getNameActivity", DisableAlreadyRegisteredCheck: true})
-	replayer.RegisterWorkflowWithOptions(greetingsWorkflow2, workflow.RegisterOptions{Name: "greetings"})
-	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "greetings.json")
-	require.Error(t, err)
+	replayer.RegisterWorkflowWithOptions(exclusiveChoiceWorkflow, workflow.RegisterOptions{Name: "choice"})
+	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "choice.json")
+	require.NoError(t, err)
 }
 
-// Ideally replayer doesn't concern itself with the change in the activity content until it matches the expected output type.
-// History has recorded the output of banana activity instead. The replayer should have failed because we have not registered any
-// activity here in the test.
-// The replayer still runs whatever it found in the history and passes.
-func TestExclusiveChoiceWorkflowWithUnregisteredActivity(t *testing.T) {
+// The recorded history has following activities in this order:  main.getOrderActivity, main.orderBananaActivity
+// This test runs a version of choice workflow which registers a single return parameter function for main.getOrderActivity
+// - Original main.getOrderActivity signature: func() (string, error)
+// - New main.getOrderActivity signature: func() error
+//
+// In this case result of main.getOrderActivity from history is not passed back to the given pointer by the workflow.
+// Compared to the activity registration missing scenario (above case) this is a little bit weird behavior.
+// The workflow code continues with orderChoice="" instead of "banana". Therefore it doesn't invoke 2nd activity main.getOrderActivity.
+// This means history has more events then replay decisions which causes non-determinism error
+func TestExclusiveChoiceWorkflowWithActivitySignatureChange(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
-
 	replayer.RegisterWorkflowWithOptions(exclusiveChoiceWorkflow, workflow.RegisterOptions{Name: "choice"})
+	replayer.RegisterActivityWithOptions(func() error { return nil }, activity.RegisterOptions{Name: "main.getOrderActivity"})
+	replayer.RegisterActivityWithOptions(orderBananaActivity, activity.RegisterOptions{Name: "main.orderBananaActivity"})
 	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "choice.json")
-	require.NoError(t, err)
+	assert.ErrorContains(t, err, "nondeterministic workflow: missing replay decision")
 }
 
-// This test registers Cherry Activity as the activity but calls Apple activity in the workflow code. Infact, Cherry and Banana
-// activities are not even a part of the workflow code in question.
-// History has recorded the output of banana activity. Here, The workflow is not waiting for the activity so it doesn't notice
-// 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) {
+// The recorded history has following activities in this order:   main.getOrderActivity, main.orderBananaActivity
+// This test runs a version of choice workflow which calls main.getOrderActivity and then calls the main.orderCherryActivity.
+// The replayer will find non-determinism because of mismatch between replay decision and history (banana vs cherry)
+func TestExclusiveChoiceWorkflowWithMismatchingActivity(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"})
+	replayer.RegisterWorkflowWithOptions(exclusiveChoiceWorkflowAlwaysCherry, workflow.RegisterOptions{Name: "choice"})
+	replayer.RegisterActivityWithOptions(getBananaOrderActivity, activity.RegisterOptions{Name: "main.getOrderActivity"})
+	replayer.RegisterActivityWithOptions(orderCherryActivity, activity.RegisterOptions{Name: "main.orderCherryActivity"})
 	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "choice.json")
-	assert.ErrorContains(t, err, "nondeterministic workflow")
+	assert.ErrorContains(t, err, "nondeterministic workflow: mismatching history event and replay decision found")
 }
 
+// Branch workflow happy case.
+// It branches out to 3 open activities and then they complete.
 func TestBranchWorkflow(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
-
 	replayer.RegisterWorkflowWithOptions(sampleBranchWorkflow, workflow.RegisterOptions{Name: "branch"})
-
 	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "branch.json")
 	require.NoError(t, err)
 }
 
-// Fails with a non deterministic error because there was an additional unexpected branch. Decreasing the number of branches will
-// also fail the test because the history expects the same number of branches executing the activity.
+// Branch workflow normal history file is replayed against modified workflow code which
+// has 2 branches only.  This causes nondetereministic error.
 func TestBranchWorkflowWithExtraBranch(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
-
 	replayer.RegisterWorkflowWithOptions(sampleBranchWorkflow2, workflow.RegisterOptions{Name: "branch"})
-
 	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "branch.json")
-	assert.ErrorContains(t, err, "nondeterministic workflow")
+	assert.ErrorContains(t, err, "nondeterministic workflow: missing replay decision")
 }
 
-// TestSequentialStepsWorkflow replays a history with 2 sequential activity calls and runs it against new version of the workflow code which only calls 1 activity.
-// This should be considered as non-determinism error.
+// TestSequentialStepsWorkflow replays a history with 2 sequential non overlapping activity calls (one completes before the other is scheduled)
+// and runs it against new version of the workflow code which only calls 1 activity.
+// This is considered as non-determinism error.
 func TestSequentialStepsWorkflow(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
-
 	replayer.RegisterWorkflowWithOptions(replayerHelloWorldWorkflow, workflow.RegisterOptions{Name: "fx.ReplayerHelloWorldWorkflow"})
 	replayer.RegisterActivityWithOptions(replayerHelloWorldActivity, activity.RegisterOptions{Name: "replayerhello"})
-
-	// 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")
+	assert.ErrorContains(t, err, "nondeterministic workflow: missing replay decision")
 }
 
-func TestParallel(t *testing.T) {
+// Runs simpleParallelWorkflow which starts two workflow.Go routines that executes 1 and 2 activities respectively.
+func TestSimpleParallelWorkflow(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
-
 	replayer.RegisterWorkflowWithOptions(sampleParallelWorkflow, workflow.RegisterOptions{Name: "branch2"})
-
 	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "branch2.json")
 	require.NoError(t, err)
 }
 
-// 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) {
+// Runs modified version of simpleParallelWorkflow which starts 1 less activity in the second workflow-gouroutine.
+// This is considered as non-determinism error.
+func TestSimpleParallelWorkflowWithMissingActivityCall(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
-
 	replayer.RegisterWorkflowWithOptions(sampleParallelWorkflow2, workflow.RegisterOptions{Name: "branch2"})
-
 	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "branch2.json")
-	assert.ErrorContains(t, err, "nondeterministic workflow")
+	assert.ErrorContains(t, err, "nondeterministic workflow: missing replay decision")
 }

test/replaytests/sequential_workflow.go

@@ -20,7 +20,7 @@ type replayerSampleMessage struct {
 // Corresponding unit test covers the scenario that new workflow's history records is subset of previous version's history.
 //
 //	v1: wf started -> call activity -> call activity -> wf complete
-//	v2:  wf started -> call activity -> wf complete
+//	v2: wf started -> call activity -> wf complete
 //
 // The v2 clearly has determinism issues and should be considered as non-determism error for replay tests.
 func replayerHelloWorldWorkflow(ctx workflow.Context, inputMsg *replayerSampleMessage) error {