add support of GetVersionOption, ExecuteWithVersion and ExecuteWithMinVersion

Author: arzonus

internal/interceptors.go

@@ -63,7 +63,7 @@ type WorkflowInterceptor interface {
 	GetSignalChannel(ctx Context, signalName string) Channel
 	SideEffect(ctx Context, f func(ctx Context) interface{}) Value
 	MutableSideEffect(ctx Context, id string, f func(ctx Context) interface{}, equals func(a, b interface{}) bool) Value
-	GetVersion(ctx Context, changeID string, minSupported, maxSupported Version) Version
+	GetVersion(ctx Context, changeID string, minSupported, maxSupported Version, opts ...GetVersionOption) Version
 	SetQueryHandler(ctx Context, queryType string, handler interface{}) error
 	IsReplaying(ctx Context) bool
 	HasLastCompletionResult(ctx Context) bool
@@ -158,8 +158,8 @@ func (t *WorkflowInterceptorBase) MutableSideEffect(ctx Context, id string, f fu
 }
 
 // GetVersion forwards to t.Next
-func (t *WorkflowInterceptorBase) GetVersion(ctx Context, changeID string, minSupported, maxSupported Version) Version {
-	return t.Next.GetVersion(ctx, changeID, minSupported, maxSupported)
+func (t *WorkflowInterceptorBase) GetVersion(ctx Context, changeID string, minSupported, maxSupported Version, opts ...GetVersionOption) Version {
+	return t.Next.GetVersion(ctx, changeID, minSupported, maxSupported, opts...)
 }
 
 // SetQueryHandler forwards to t.Next

internal/internal_event_handlers.go

@@ -602,29 +602,59 @@ func validateVersion(changeID string, version, minSupported, maxSupported Versio
 	}
 }
 
-func (wc *workflowEnvironmentImpl) GetVersion(changeID string, minSupported, maxSupported Version) Version {
+func (wc *workflowEnvironmentImpl) GetVersion(changeID string, minSupported, maxSupported Version, opts ...GetVersionOption) Version {
+	// Check if the changeID already has a version assigned
+	// If it does, validate the version against the min and max supported versions
 	if version, ok := wc.changeVersions[changeID]; ok {
 		validateVersion(changeID, version, minSupported, maxSupported)
 		return version
 	}
 
+	// Apply options to determine which version to use
+	options := &getVersionOptions{}
+	for _, opt := range opts {
+		opt(options)
+	}
+
 	var version Version
-	if wc.isReplay {
-		// GetVersion for changeID is called first time in replay mode, use DefaultVersion
+	switch {
+
+	// GetVersion for changeID is called first time in replay mode, use DefaultVersion
+	case wc.isReplay:
 		version = DefaultVersion
-	} else {
-		// GetVersion for changeID is called first time (non-replay mode), generate a marker decision for it.
-		// Also upsert search attributes to enable ability to search by changeVersion.
+
+	// If ExecuteWithVersion option is used, use the custom version provided
+	case options.customVersion != nil:
+		version = *options.customVersion
+
+	// If ExecuteWithMinVersion option is set, use the minimum supported version
+	case options.useMinVersion:
+		version = minSupported
+
+	// Otherwise, use the maximum supported version
+	default:
 		version = maxSupported
+	}
+
+	// Validate the version against the min and max supported versions
+	// ensuring it is within the acceptable range
+	validateVersion(changeID, version, minSupported, maxSupported)
+
+	// If the version is not the DefaultVersion, and it's not a replay, record it and update search attributes
+	// Keeping the DefaultVersion as a special case where no version marker is recorded
+	if !wc.isReplay && version != DefaultVersion {
+		// Record the version marker and update search attributes
 		wc.decisionsHelper.recordVersionMarker(changeID, version, wc.GetDataConverter())
 		err := wc.UpsertSearchAttributes(createSearchAttributesForChangeVersion(changeID, version, wc.changeVersions))
 		if err != nil {
 			wc.logger.Warn("UpsertSearchAttributes failed", zap.Error(err))
 		}
 	}
 
-	validateVersion(changeID, version, minSupported, maxSupported)
+	// Store the version in the changeVersions
+	// ensuring that it can be retrieved later
 	wc.changeVersions[changeID] = version
+
 	return version
 }
 

internal/internal_worker_base.go

@@ -76,7 +76,7 @@ type (
 		localActivityClient
 		workflowTimerClient
 		SideEffect(f func() ([]byte, error), callback resultHandler)
-		GetVersion(changeID string, minSupported, maxSupported Version) Version
+		GetVersion(changeID string, minSupported, maxSupported Version, opts ...GetVersionOption) Version
 		WorkflowInfo() *WorkflowInfo
 		Complete(result []byte, err error)
 		RegisterCancelHandler(handler func())

internal/internal_workflow_testsuite.go

@@ -1928,7 +1928,7 @@ func (env *testWorkflowEnvironmentImpl) SideEffect(f func() ([]byte, error), cal
 	callback(f())
 }
 
-func (env *testWorkflowEnvironmentImpl) GetVersion(changeID string, minSupported, maxSupported Version) (retVersion Version) {
+func (env *testWorkflowEnvironmentImpl) GetVersion(changeID string, minSupported, maxSupported Version, opts ...GetVersionOption) (retVersion Version) {
 	if mockVersion, ok := env.getMockedVersion(changeID, changeID, minSupported, maxSupported); ok {
 		// GetVersion for changeID is mocked
 		env.UpsertSearchAttributes(createSearchAttributesForChangeVersion(changeID, mockVersion, env.changeVersions))

internal/workflow.go

@@ -1567,11 +1567,96 @@ const DefaultVersion Version = -1
 // CadenceChangeVersion is used as search attributes key to find workflows with specific change version.
 const CadenceChangeVersion = "CadenceChangeVersion"
 
+// GetVersionOption represents a function that configures GetVersion behavior
+type GetVersionOption func(*getVersionOptions)
+
+type getVersionOptions struct {
+	// customVersion is used to force GetVersion to return a specific version
+	// instead of maxSupported version. Set up via ExecuteWithVersion option.
+	customVersion *Version
+
+	// useMinVersion is used to force GetVersion to return minSupported version
+	// instead of maxSupported version. Set up via ExecuteWithMinVersion option.
+	useMinVersion bool
+}
+
+// ExecuteWithVersion returns a GetVersionOption that forces a specific version to be returned
+// when executed for the first time, instead of returning maxSupported version.
+//
+// This option can be used when you want to separate the versioning of the workflow code and
+// activation of the new logic in the workflow code, to ensure that your changes can be safely rolled back
+// if needed. For example, initially a workflow has the following code:
+//
+//	err = workflow.ExecuteActivity(ctx, foo).Get(ctx, nil)
+//
+// It should be updated to:
+//
+//	err = workflow.ExecuteActivity(ctx, bar).Get(ctx, nil)
+//
+// Step 1
+// To roll out your changes safely, both versions of your workflow code should be compatible with each other.
+// To achieve that, you can use GetVersion with ExecuteWithVersion option.
+// When GetVersion is executed for the first time, it will return DefaultVersion instead of maxSupported version:
+//
+//	v := GetVersion(ctx, "fooChange", DefaultVersion, 1, ExecuteWithVersion(DefaultVersion))
+//	if v == DefaultVersion {
+//	    err = workflow.ExecuteActivity(ctx, foo).Get(ctx, nil)
+//	} else {
+//	    err = workflow.ExecuteActivity(ctx, bar).Get(ctx, nil)
+//	}
+//
+// At this step, the previous version of the code supports only DefaultVersion, however new version of the code
+// supports both DefaultVersion and 1. At the same time, the new version of the code is not yet activated,
+// so the workflow started on the new version of the code will still execute foo activity - previous version of the code.
+// This makes it possible to safely roll back your changes if needed, as the previous code supports DefaultVersion.
+//
+// Step 2
+// When the previous version of the code is no longer running, there is no need to start new workflow executions
+// with DefaultVersion anymore, and you can the maxSupported version to activate the new code. To achieve that you can
+// remove the usage of ExecuteWithVersion option. When GetVersion is executed for the first time, it will return maxSupported version:
+//
+//	v := GetVersion(ctx, "fooChange", DefaultVersion, 1)
+//	if v == DefaultVersion {
+//	    err = workflow.ExecuteActivity(ctx, foo).Get(ctx, nil)
+//	} else {
+//	    err = workflow.ExecuteActivity(ctx, bar).Get(ctx, nil)
+//	}
+//
+// At this step, the previous and old versions of the code support both versions DefaultVersion and 1,
+// however the new version of the code is activated, so the workflow started on the new version of the code
+// will execute bar activity - new version of the code. This makes it possible to safely roll back your changes if needed,
+// because both versions of the code support both versions DefaultVersion and 1.
+//
+// Step 3
+// When there are no running previous version of the code and there are no workflow executions
+// running DefaultVersion the correspondent branch can be removed:
+//
+//	GetVersion(ctx, "fooChange", 1, 1)
+//	err = workflow.ExecuteActivity(ctx, bar).Get(ctx, nil)
+//
+// ExecuteWithVersion option is useful when you want to ensure that your changes can be safely rolled back if needed, as
+// both versions of the workflow code are compatible with each other.
+func ExecuteWithVersion(version Version) GetVersionOption {
+	return func(o *getVersionOptions) {
+		o.customVersion = &version
+	}
+}
+
+// ExecuteWithMinVersion returns a GetVersionOption that makes GetVersion return minSupported version
+// when executed for the first time, instead of returning maxSupported version.
+// To see how this option can be used, see the ExecuteWithVersion option
+func ExecuteWithMinVersion() GetVersionOption {
+	return func(o *getVersionOptions) {
+		o.useMinVersion = true
+	}
+}
+
 // GetVersion is used to safely perform backwards incompatible changes to workflow definitions.
 // It is not allowed to update workflow code while there are workflows running as it is going to break
 // determinism. The solution is to have both old code that is used to replay existing workflows
 // as well as the new one that is used when it is executed for the first time.
-// GetVersion returns maxSupported version when is executed for the first time. This version is recorded into the
+// GetVersion returns maxSupported version (to return another version, check GetVersionOption),
+// when is executed for the first time. This version is recorded into the
 // workflow history as a marker event. Even if maxSupported version is changed the version that was recorded is
 // returned on replay. DefaultVersion constant contains version of code that wasn't versioned before.
 // For example initially workflow has the following code:
@@ -1632,13 +1717,13 @@ const CadenceChangeVersion = "CadenceChangeVersion"
 //	} else {
 //	  err = workflow.ExecuteActivity(ctx, qux, data).Get(ctx, nil)
 //	}
-func GetVersion(ctx Context, changeID string, minSupported, maxSupported Version) Version {
+func GetVersion(ctx Context, changeID string, minSupported, maxSupported Version, opts ...GetVersionOption) Version {
 	i := getWorkflowInterceptor(ctx)
-	return i.GetVersion(ctx, changeID, minSupported, maxSupported)
+	return i.GetVersion(ctx, changeID, minSupported, maxSupported, opts...)
 }
 
-func (wc *workflowEnvironmentInterceptor) GetVersion(ctx Context, changeID string, minSupported, maxSupported Version) Version {
-	return wc.env.GetVersion(changeID, minSupported, maxSupported)
+func (wc *workflowEnvironmentInterceptor) GetVersion(ctx Context, changeID string, minSupported, maxSupported Version, opts ...GetVersionOption) Version {
+	return wc.env.GetVersion(changeID, minSupported, maxSupported, opts...)
 }
 
 // SetQueryHandler sets the query handler to handle workflow query. The queryType specify which query type this handler