Add deployment version features

dockerfiles/dynamicconfig/docker.yaml

@@ -7,9 +7,18 @@ system.enableActivityEagerExecution:
 system.enableEagerWorkflowStart:
   - value: true
     constraints: {}
+system.enableDeployments:
+  - value: true
+    constraints: {}
+system.enableDeploymentVersions:
+  - value: true
+    constraints: {}    
 frontend.enableUpdateWorkflowExecution:
   - value: true
     constraints: {}
 frontend.enableUpdateWorkflowExecutionAsyncAccepted:
   - value: true
     constraints: {}
+frontend.workerVersioningWorkflowAPIs:
+  - value: true
+    constraints: {}

features/deployment_versioning/common.go

@@ -0,0 +1,155 @@
+package deployment_versioning
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"github.com/temporalio/features/harness/go/harness"
+	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/worker"
+	"go.temporal.io/sdk/workflow"
+)
+
+func WaitForSignalOne(ctx workflow.Context) (string, error) {
+	var value string
+	workflow.GetSignalChannel(ctx, "start-signal").Receive(ctx, &value)
+	return value + "_v1", nil
+}
+
+func WaitForSignalTwo(ctx workflow.Context) (string, error) {
+	var value string
+	workflow.GetSignalChannel(ctx, "start-signal").Receive(ctx, &value)
+	return value + "_v2", nil
+}
+
+func StartWorker(ctx context.Context, r *harness.Runner, version string, versioningBehavior workflow.VersioningBehavior) worker.Worker {
+	w := worker.New(r.Client, r.TaskQueue, worker.Options{
+		DeploymentOptions: worker.DeploymentOptions{
+			UseVersioning:             true,
+			Version:                   version,
+			DefaultVersioningBehavior: versioningBehavior,
+		},
+	})
+	if strings.HasSuffix(version, "1.0") {
+		w.RegisterWorkflowWithOptions(WaitForSignalOne, workflow.RegisterOptions{
+			Name: "WaitForSignal",
+		})
+	} else {
+		w.RegisterWorkflowWithOptions(WaitForSignalTwo, workflow.RegisterOptions{
+			Name: "WaitForSignal",
+		})
+	}
+	return w
+}
+
+func WaitForDeploymentVersion(r *harness.Runner, ctx context.Context, dHandle client.WorkerDeploymentHandle, version string) error {
+	return r.DoUntilEventually(ctx, 300*time.Millisecond, 10*time.Second,
+		func() bool {
+			d, err := dHandle.Describe(ctx, client.WorkerDeploymentDescribeOptions{})
+			if err != nil {
+				return false
+			}
+			for _, v := range d.Info.VersionSummaries {
+				if v.Version == version {
+					return true
+				}
+			}
+			return false
+		})
+}
+
+func WaitForDeployment(r *harness.Runner, ctx context.Context, dHandle client.WorkerDeploymentHandle) error {
+	return r.DoUntilEventually(ctx, 300*time.Millisecond, 10*time.Second,
+		func() bool {
+			_, err := dHandle.Describe(ctx, client.WorkerDeploymentDescribeOptions{})
+			return err == nil
+		})
+}
+
+func WaitForWorkflowRunning(r *harness.Runner, ctx context.Context, handle client.WorkflowRun) error {
+	return r.DoUntilEventually(ctx, 300*time.Millisecond, 10*time.Second,
+		func() bool {
+			describeResp, err := r.Client.DescribeWorkflowExecution(ctx, handle.GetID(), handle.GetRunID())
+			if err != nil {
+				return false
+			}
+			status := describeResp.WorkflowExecutionInfo.Status
+			return enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING == status
+		})
+}
+
+func SignalAll(r *harness.Runner, ctx context.Context, handles []client.WorkflowRun) error {
+	for _, handle := range handles {
+		if err := r.Client.SignalWorkflow(ctx, handle.GetID(), handle.GetRunID(), "start-signal", "prefix"); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func SetCurrent(r *harness.Runner, ctx context.Context, deploymentName string, version string) error {
+	dHandle := r.Client.WorkerDeploymentClient().GetHandle(deploymentName)
+
+	if err := WaitForDeployment(r, ctx, dHandle); err != nil {
+		return err
+	}
+
+	response1, err := dHandle.Describe(ctx, client.WorkerDeploymentDescribeOptions{})
+	if err != nil {
+		return err
+	}
+
+	if err := WaitForDeploymentVersion(r, ctx, dHandle, version); err != nil {
+		return err
+	}
+
+	_, err = dHandle.SetCurrentVersion(ctx, client.WorkerDeploymentSetCurrentVersionOptions{
+		Version:       version,
+		ConflictToken: response1.ConflictToken,
+	})
+
+	return err
+}
+
+func SetRamp(r *harness.Runner, ctx context.Context, deploymentName string, version string, percentage float32) error {
+	dHandle := r.Client.WorkerDeploymentClient().GetHandle(deploymentName)
+
+	if err := WaitForDeployment(r, ctx, dHandle); err != nil {
+		return err
+	}
+
+	response1, err := dHandle.Describe(ctx, client.WorkerDeploymentDescribeOptions{})
+	if err != nil {
+		return err
+	}
+
+	if err := WaitForDeploymentVersion(r, ctx, dHandle, version); err != nil {
+		return err
+	}
+
+	_, err = dHandle.SetRampingVersion(ctx, client.WorkerDeploymentSetRampingVersionOptions{
+		Version:       version,
+		ConflictToken: response1.ConflictToken,
+		Percentage:    float32(100.0),
+	})
+
+	return err
+}
+
+func ServerSupportsDeployments(ctx context.Context, r *harness.Runner) bool {
+	// No system capability, only dynamic config in namespace, need to just try...
+	iter, err := r.Client.WorkerDeploymentClient().List(ctx, client.WorkerDeploymentListOptions{})
+	if err != nil {
+		return false
+	}
+	// Need to call `HasNext` to contact the server
+	for iter.HasNext() {
+		_, err := iter.Next()
+		if err != nil {
+			return false
+		}
+	}
+	return true
+}

features/deployment_versioning/routing_auto_upgrade/README.md

@@ -0,0 +1,14 @@
+# Deployment Versioning: Routing AutoUpgrade
+
+When the Current Version for a Deployment changes, already started workflows
+will transition to the new version if they are `AutoUpgrade`.
+
+# Detailed spec
+
+* Create a random deployment name `deployment_name`
+* Start a `deployment_name.1-0` worker, register workflow type `WaitForSignal` as `AutoUpgrade`, the implementation of that workflow should end returning `prefix_v1`.
+* Start a `deployment_name.2-0` worker, register workflow type `WaitForSignal` as `AutoUpgrade`, the implementation of that workflow should end returning `prefix_v2`.
+* Set Current version for `deployment_name` to `deployment_name.1-0`
+* Start `workflow_1` of type `WaitForSignal`, it should start AutoUpgrade and with version `deployment_name.1-0`
+* Set Current version for `deployment_name` to `deployment_name.2-0`
+* Signal workflow. The workflow (pinned) should exit returning `prefix_v2`. 

features/deployment_versioning/routing_auto_upgrade/feature.go

@@ -0,0 +1,90 @@
+package routing_auto_upgrade
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/temporalio/features/features/deployment_versioning"
+	"github.com/temporalio/features/harness/go/harness"
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/worker"
+	"go.temporal.io/sdk/workflow"
+)
+
+var deploymentName = uuid.NewString()
+
+// Wrap for correct Feature.Path
+func WaitForSignalOne(ctx workflow.Context) (string, error) {
+	return deployment_versioning.WaitForSignalOne(ctx)
+}
+
+var Feature = harness.Feature{
+	Workflows: []interface{}{
+		harness.WorkflowWithOptions{
+			Workflow: WaitForSignalOne,
+			Options: workflow.RegisterOptions{
+				Name:               "WaitForSignal",
+				VersioningBehavior: workflow.VersioningBehaviorAutoUpgrade,
+			},
+		},
+	},
+	Execute: Execute,
+	WorkerOptions: worker.Options{
+		DeploymentOptions: worker.DeploymentOptions{
+			UseVersioning: true,
+			Version:       deploymentName + ".1.0",
+		},
+	},
+	CheckHistory:    CheckHistory,
+	ExpectRunResult: "prefix_v2",
+}
+
+var worker2 worker.Worker
+
+func Execute(ctx context.Context, r *harness.Runner) (client.WorkflowRun, error) {
+	if supported := deployment_versioning.ServerSupportsDeployments(ctx, r); !supported {
+		return nil, r.Skip(fmt.Sprintf("server does not support deployment versioning"))
+	}
+
+	worker2 = deployment_versioning.StartWorker(ctx, r, deploymentName+".2.0",
+		workflow.VersioningBehaviorAutoUpgrade)
+	if err := worker2.Start(); err != nil {
+		return nil, err
+	}
+
+	if err := deployment_versioning.SetCurrent(r, ctx, deploymentName, deploymentName+".1.0"); err != nil {
+		return nil, err
+	}
+
+	run, err := r.Client.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
+		TaskQueue:                r.TaskQueue,
+		ID:                       "workflow_1",
+		WorkflowExecutionTimeout: 1 * time.Minute,
+	}, "WaitForSignal")
+
+	if err != nil {
+		return nil, err
+	}
+
+	if err := deployment_versioning.WaitForWorkflowRunning(r, ctx, run); err != nil {
+		return nil, err
+	}
+
+	if err := deployment_versioning.SetCurrent(r, ctx, deploymentName, deploymentName+".2.0"); err != nil {
+		return nil, err
+	}
+
+	if err := deployment_versioning.SignalAll(r, ctx, []client.WorkflowRun{run}); err != nil {
+		return nil, err
+	}
+
+	return run, nil
+}
+
+func CheckHistory(ctx context.Context, r *harness.Runner, run client.WorkflowRun) error {
+	// Shut down the 2.0 worker
+	worker2.Stop()
+	return r.CheckHistoryDefault(ctx, run)
+}

features/deployment_versioning/routing_pinned/README.md

@@ -0,0 +1,14 @@
+# Deployment Versioning: Version Routing Pinned
+
+When the Current Version for a Deployment changes, already started workflows
+will never transition to the new version if they are `Pinned`.
+
+# Detailed spec
+
+* Create a random deployment name `deployment_name`
+* Start a `deployment_name.1-0` worker, register workflow type `WaitForSignal` as `Pinned`, the implementation of that workflow should end returning `prefix_v1`.
+* Start a `deployment_name.2-0` worker, register workflow type `WaitForSignal` as `AutoUpgrade`, the implementation of that workflow should end returning `prefix_v2`.
+* Set Current version for `deployment_name` to `deployment_name.1-0`
+* Start `workflow_1` of type `WaitForSignal`, it should start pinned and with version `deployment_name.1-0`
+* Set Current version for `deployment_name` to `deployment_name.2-0`
+* Signal workflow. The workflow (pinned) should exit returning `prefix_v1`. 

features/deployment_versioning/routing_pinned/feature.go

@@ -0,0 +1,89 @@
+package routing_pinned
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/temporalio/features/features/deployment_versioning"
+	"github.com/temporalio/features/harness/go/harness"
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/worker"
+	"go.temporal.io/sdk/workflow"
+)
+
+var deploymentName = uuid.NewString()
+
+// Wrap for correct Feature.Path
+func WaitForSignalOne(ctx workflow.Context) (string, error) {
+	return deployment_versioning.WaitForSignalOne(ctx)
+}
+
+var Feature = harness.Feature{
+	Workflows: []interface{}{
+		harness.WorkflowWithOptions{
+			Workflow: WaitForSignalOne,
+			Options: workflow.RegisterOptions{
+				Name:               "WaitForSignal",
+				VersioningBehavior: workflow.VersioningBehaviorPinned,
+			},
+		},
+	},
+	Execute: Execute,
+	WorkerOptions: worker.Options{
+		DeploymentOptions: worker.DeploymentOptions{
+			UseVersioning: true,
+			Version:       deploymentName + ".1.0",
+		},
+	},
+	CheckHistory:    CheckHistory,
+	ExpectRunResult: "prefix_v1",
+}
+var worker2 worker.Worker
+
+func Execute(ctx context.Context, r *harness.Runner) (client.WorkflowRun, error) {
+	if supported := deployment_versioning.ServerSupportsDeployments(ctx, r); !supported {
+		return nil, r.Skip(fmt.Sprintf("server does not support deployment versioning"))
+	}
+
+	worker2 = deployment_versioning.StartWorker(ctx, r, deploymentName+".2.0",
+		workflow.VersioningBehaviorAutoUpgrade)
+	if err := worker2.Start(); err != nil {
+		return nil, err
+	}
+
+	if err := deployment_versioning.SetCurrent(r, ctx, deploymentName, deploymentName+".1.0"); err != nil {
+		return nil, err
+	}
+
+	run, err := r.Client.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
+		TaskQueue:                r.TaskQueue,
+		ID:                       "workflow_1",
+		WorkflowExecutionTimeout: 1 * time.Minute,
+	}, "WaitForSignal")
+
+	if err != nil {
+		return nil, err
+	}
+
+	if err := deployment_versioning.WaitForWorkflowRunning(r, ctx, run); err != nil {
+		return nil, err
+	}
+
+	if err := deployment_versioning.SetCurrent(r, ctx, deploymentName, deploymentName+".2.0"); err != nil {
+		return nil, err
+	}
+
+	if err := deployment_versioning.SignalAll(r, ctx, []client.WorkflowRun{run}); err != nil {
+		return nil, err
+	}
+
+	return run, nil
+}
+
+func CheckHistory(ctx context.Context, r *harness.Runner, run client.WorkflowRun) error {
+	// Shut down the 2.0 worker
+	worker2.Stop()
+	return r.CheckHistoryDefault(ctx, run)
+}