add TestVersionedWorkflows case

Author: arzonus

test/replaytests/replay_test.go

@@ -191,3 +191,104 @@ func TestContinueAsNew(t *testing.T) {
 	err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), "continue_as_new.json")
 	assert.ErrorContains(t, err, "missing replay decision for WorkflowExecutionContinuedAsNew")
 }
+
+// TestSafeDeploymentVersionedWorkflow verifies that versioned workflows can be executed
+// safely across different versions without causing non-deterministic errors.
+// There are 2 workflow executions:
+//
+// * VersionedWorkflowFoo - which is the first version of the workflow which version of change id is DefaultVersion
+//   - This workflow is supposed to execute FooActivity
+//
+// * VersionedWorkflowBar - which is the second version of the workflow which version of change id is 1
+//   - This workflow is supposed to execute BarActivity
+//
+// There are 4 versions of the workflow:
+//
+// * VersionedWorkflowV1 - which supports only DefaultVersion and executes FooActivity
+//   - This workflow is able to replay the history of only of VersionedWorkflowFoo
+//
+// * VersionedWorkflowV2 - which supports DefaultVersion and Version 1, and can execute FooActivity or BarActivity
+//   - This workflow is able to replay the history of both of VersionedWorkflowFoo and VersionedWorkflowBar
+//   - A first execution of this workflow will should execute FooActivity, because of usage workflow.ExecuteWithMinVersion(),
+//     but the test can't check it due to Replay
+//
+// * VersionedWorkflowV3 - which supports DefaultVersion and Version 1, and can execute FooActivity or BarActivity
+//   - This workflow is able to replay the history of both of VersionedWorkflowFoo and VersionedWorkflowBar
+//   - A first execution of this workflow will should execute BarActivity, but the test can't check it due to Replay
+//
+// * VersionedWorkflowV4 - which supports Version 1, and can execute BarActivity
+//   - This workflow is able to replay the history only of VersionedWorkflowBar
+//
+// So the test focusing workflows supports forward and backward compatibility of the workflows
+func TestVersionedWorkflows(t *testing.T) {
+	const (
+		versionedWorkflowFooHistoryFile = "versioned_workflow_foo.json"
+		versionedWorkflowBarHistoryFile = "versioned_workflow_bar.json"
+	)
+
+	t.Run("VersionedWorkflowV1", func(t *testing.T) {
+		replayer := worker.NewWorkflowReplayer()
+		replayer.RegisterWorkflowWithOptions(VersionedWorkflowV1, workflow.RegisterOptions{Name: versionedWorkflowName})
+		replayer.RegisterActivityWithOptions(FooActivity, activity.RegisterOptions{Name: fooActivityName})
+
+		t.Run("successfully replayed with VersionedWorkflowFoo", func(t *testing.T) {
+			err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), versionedWorkflowFooHistoryFile)
+			require.NoError(t, err, "Failed to replay VersionedWorkflowFoo history")
+		})
+
+		t.Run("fail to replay with VersionedWorkflowBar", func(t *testing.T) {
+			err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), versionedWorkflowBarHistoryFile)
+			require.Error(t, err, "Expected to fail replaying VersionedWorkflowBar history")
+		})
+	})
+
+	t.Run("VersionedWorkflowV2", func(t *testing.T) {
+		replayer := worker.NewWorkflowReplayer()
+		replayer.RegisterWorkflowWithOptions(VersionedWorkflowV2, workflow.RegisterOptions{Name: versionedWorkflowName})
+		replayer.RegisterActivityWithOptions(FooActivity, activity.RegisterOptions{Name: fooActivityName})
+		replayer.RegisterActivityWithOptions(BarActivity, activity.RegisterOptions{Name: barActivityName})
+
+		t.Run("successfully replayed with VersionedWorkflowFoo", func(t *testing.T) {
+			err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), versionedWorkflowFooHistoryFile)
+			require.NoError(t, err, "Failed to replay VersionedWorkflowFoo history")
+		})
+
+		t.Run("successfully replayed with VersionedWorkflowBar", func(t *testing.T) {
+			err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), versionedWorkflowBarHistoryFile)
+			require.NoError(t, err, "Failed to replay VersionedWorkflowBar history")
+		})
+	})
+
+	t.Run("VersionedWorkflowV3", func(t *testing.T) {
+		replayer := worker.NewWorkflowReplayer()
+		replayer.RegisterWorkflowWithOptions(VersionedWorkflowV3, workflow.RegisterOptions{Name: versionedWorkflowName})
+		replayer.RegisterActivityWithOptions(FooActivity, activity.RegisterOptions{Name: fooActivityName})
+		replayer.RegisterActivityWithOptions(BarActivity, activity.RegisterOptions{Name: barActivityName})
+
+		t.Run("successfully replayed with VersionedWorkflowFoo", func(t *testing.T) {
+			err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), versionedWorkflowFooHistoryFile)
+			require.NoError(t, err, "Failed to replay VersionedWorkflowFoo history")
+		})
+
+		t.Run("successfully replayed with VersionedWorkflowBar", func(t *testing.T) {
+			err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), versionedWorkflowBarHistoryFile)
+			require.NoError(t, err, "Failed to replay VersionedWorkflowBar history")
+		})
+	})
+
+	t.Run("VersionedWorkflowV4", func(t *testing.T) {
+		replayer := worker.NewWorkflowReplayer()
+		replayer.RegisterWorkflowWithOptions(VersionedWorkflowV4, workflow.RegisterOptions{Name: versionedWorkflowName})
+		replayer.RegisterActivityWithOptions(BarActivity, activity.RegisterOptions{Name: barActivityName})
+
+		t.Run("fail to replay with VersionedWorkflowFoo", func(t *testing.T) {
+			err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), versionedWorkflowFooHistoryFile)
+			require.Error(t, err, "Expected to fail replaying VersionedWorkflowFoo history")
+		})
+
+		t.Run("successfully replayed with VersionedWorkflowBar", func(t *testing.T) {
+			err := replayer.ReplayWorkflowHistoryFromJSONFile(zaptest.NewLogger(t), versionedWorkflowBarHistoryFile)
+			require.NoError(t, err, "Failed to replay VersionedWorkflowBar history")
+		})
+	})
+}

test/replaytests/versioned_workflow.go

@@ -0,0 +1,117 @@
+package replaytests
+
+import (
+	"go.uber.org/cadence/workflow"
+	"time"
+)
+
+const (
+	// testChangeID is a constant used to identify the version change in the workflow.
+	testChangeID = "test-change"
+
+	// fooActivityName and barActivityName are the names of the activities used in the workflows.
+	fooActivityName = "FooActivity"
+	barActivityName = "BarActivity"
+
+	// versionedWorkflowName is the name of the versioned workflow.
+	versionedWorkflowName = "VersionedWorkflow"
+)
+
+var activityOptions = workflow.ActivityOptions{
+	ScheduleToStartTimeout: time.Minute,
+	StartToCloseTimeout:    time.Minute,
+	HeartbeatTimeout:       time.Second * 20,
+}
+
+// VersionedWorkflowV1 is the first version of the workflow, and it supports only DefaultVersion.
+// It supports workflow executions started by this version VersionedWorkflowV1
+// and VersionedWorkflowV2, as all of them will have the change ID set to DefaultVersion.
+func VersionedWorkflowV1(ctx workflow.Context, _ string) (string, error) {
+	ctx = workflow.WithActivityOptions(ctx, activityOptions)
+
+	var result string
+	err := workflow.ExecuteActivity(ctx, fooActivityName, "").Get(ctx, &result)
+	if err != nil {
+		return "", err
+	}
+
+	return result, nil
+}
+
+// VersionedWorkflowV2 is the second version of the workflow. It supports DefaultVersion and Version 1.
+// All workflows started by this version will have the change ID set to DefaultVersion.
+// It supports workflow executions started by VersionedWorkflowV1 and VersionedWorkflowV2,
+// as all of them will have the change ID set to DefaultVersion.
+// It also supports workflow executions started by VersionedWorkflowV3 and VersionedWorkflowV4
+// because the code supports execution of Version 1 of the workflow.
+func VersionedWorkflowV2(ctx workflow.Context, _ string) (string, error) {
+	ctx = workflow.WithActivityOptions(ctx, activityOptions)
+
+	var result string
+	var err error
+
+	version := workflow.GetVersion(ctx, testChangeID, workflow.DefaultVersion, 1, workflow.ExecuteWithMinVersion())
+	if version == workflow.DefaultVersion {
+		err = workflow.ExecuteActivity(ctx, fooActivityName, "").Get(ctx, &result)
+	} else {
+		err = workflow.ExecuteActivity(ctx, barActivityName, "").Get(ctx, &result)
+	}
+	if err != nil {
+		return "", err
+	}
+
+	return result, nil
+}
+
+// VersionedWorkflowV3 is the third version of the workflow. It supports DefaultVersion and Version 1 as well.
+// However, all workflows started by this version will have the change ID set to Version 1.
+// It supports workflow executions started by VersionedWorkflowV1 and VersionedWorkflowV2,
+// as all of them will have the change ID set to DefaultVersion, and it supports them.
+// It also supports workflow executions started by VersionedWorkflowV3 and VersionedWorkflowV4,
+// because the code supports execution of Version 1 of the workflow.
+func VersionedWorkflowV3(ctx workflow.Context, _ string) (string, error) {
+	ctx = workflow.WithActivityOptions(ctx, activityOptions)
+
+	var result string
+	var err error
+
+	version := workflow.GetVersion(ctx, testChangeID, workflow.DefaultVersion, 1)
+	if version == workflow.DefaultVersion {
+		err = workflow.ExecuteActivity(ctx, fooActivityName, "").Get(ctx, &result)
+	} else {
+		err = workflow.ExecuteActivity(ctx, barActivityName, "").Get(ctx, &result)
+	}
+	if err != nil {
+		return "", err
+	}
+
+	return result, nil
+}
+
+// VersionedWorkflowV4 is the fourth version of the workflow. It supports only Version 1.
+// All workflows started by this version will have the change ID set to Version 1.
+// It supports workflow executions started by VersionedWorkflowV3 and VersionedWorkflowV4,
+// as all of them will have the change ID set to Version 1.
+func VersionedWorkflowV4(ctx workflow.Context, _ string) (string, error) {
+	ctx = workflow.WithActivityOptions(ctx, activityOptions)
+
+	var result string
+
+	workflow.GetVersion(ctx, testChangeID, 1, 1)
+	err := workflow.ExecuteActivity(ctx, barActivityName, "").Get(ctx, &result)
+	if err != nil {
+		return "", err
+	}
+
+	return result, nil
+}
+
+// FooActivity returns "foo" as a result of the activity execution.
+func FooActivity(ctx workflow.Context, _ string) (string, error) {
+	return "foo", nil
+}
+
+// BarActivity returns "bar" as a result of the activity execution.
+func BarActivity(ctx workflow.Context, _ string) (string, error) {
+	return "bar", nil
+}