package tests import ( "context" "errors" "fmt" "slices" "strings" "sync" "testing" "time" "github.com/google/uuid" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" batchpb "go.temporal.io/api/batch/v1" commonpb "go.temporal.io/api/common/v1" computepb "go.temporal.io/api/compute/v1" deploymentpb "go.temporal.io/api/deployment/v1" enumspb "go.temporal.io/api/enums/v1" "go.temporal.io/api/serviceerror" workflowpb "go.temporal.io/api/workflow/v1" "go.temporal.io/api/workflowservice/v1" computeprovider "go.temporal.io/auto-scaled-workers/wci/workflow/compute_provider" sdkclient "go.temporal.io/sdk/client" "go.temporal.io/sdk/worker" "go.temporal.io/sdk/workflow" deploymentspb "go.temporal.io/server/api/deployment/v1" "go.temporal.io/server/api/matchingservice/v1" persistencespb "go.temporal.io/server/api/persistence/v1" "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/testing/parallelsuite" "go.temporal.io/server/common/testing/testhooks" "go.temporal.io/server/common/testing/testvars" "go.temporal.io/server/common/tqid" "go.temporal.io/server/common/worker_versioning" "go.temporal.io/server/service/worker/workerdeployment" "go.temporal.io/server/tests/testcore" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/fieldmaskpb" "google.golang.org/protobuf/types/known/timestamppb" ) const ( maxConcurrentBatchOperations = 3 testVersionDrainageRefreshInterval = 3 * time.Second testVersionDrainageVisibilityGracePeriod = 3 * time.Second testLongVersionDrainageRefreshInterval = 10 * time.Second testLongVersionDrainageVisibilityGracePeriod = 10 * time.Second testMaxVersionsInDeployment = 4 ) type ( DeploymentVersionSuite struct { parallelsuite.Suite[*DeploymentVersionSuite] } ) // TODO: this is always true. cleanup code const useV32 = true var ( testRandomMetadataValue = []byte("random metadata value") ) func TestDeploymentVersionSuite(t *testing.T) { testcore.UseSuiteScopedCluster(t) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. parallelsuite.RunLegacySequential(t, &DeploymentVersionSuite{}) //nolint:staticcheck // SA1019: suite reuses one worker-service cluster to avoid per-test cluster churn. } // newTestEnv creates a TestEnv with the dynamic config and test variables this suite needs. // Additional per-test options may be passed in opts. func (s *DeploymentVersionSuite) newTestEnv(opts ...testcore.TestOption) *testcore.TestEnv { baseOpts := []testcore.TestOption{ testcore.WithDynamicConfig(dynamicconfig.MatchingDeploymentWorkflowVersion, int(workerdeployment.VersionDataRevisionNumber)), // Make sure we don't hit the rate limiter in tests testcore.WithDynamicConfig(dynamicconfig.FrontendGlobalNamespaceNamespaceReplicationInducingAPIsRPS, 1000), testcore.WithDynamicConfig(dynamicconfig.FrontendMaxNamespaceNamespaceReplicationInducingAPIsBurstRatioPerInstance, 1), testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, 1), testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, 1), // Reduce the chance of hitting max batch job limit in tests testcore.WithDynamicConfig(dynamicconfig.FrontendMaxConcurrentBatchOperationPerNamespace, maxConcurrentBatchOperations), testcore.WithDynamicConfig(dynamicconfig.VersionDrainageStatusRefreshInterval, testVersionDrainageRefreshInterval), testcore.WithDynamicConfig(dynamicconfig.VersionDrainageStatusVisibilityGracePeriod, testVersionDrainageVisibilityGracePeriod), // Keep deployment versions short because worker-deployment system workflow IDs must fit into 255 characters (database constraint). testcore.WithTestVars(func(tv *testvars.TestVars) *testvars.TestVars { return tv.WithDeploymentSeries("wd").WithBuildID("b") }), } return testcore.NewEnv(s.T(), append(baseOpts, opts...)...) } // pollFromDeployment calls PollWorkflowTaskQueue to start deployment related workflows func (s *DeploymentVersionSuite) pollFromDeployment(ctx context.Context, env *testcore.TestEnv, tv *testvars.TestVars) { _, _ = env.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{ Namespace: env.Namespace().String(), TaskQueue: tv.TaskQueue(), Identity: uuid.NewString(), DeploymentOptions: tv.WorkerDeploymentOptions(true), }) } func (s *DeploymentVersionSuite) pollActivityFromDeployment(ctx context.Context, env *testcore.TestEnv, tv *testvars.TestVars) { _, _ = env.FrontendClient().PollActivityTaskQueue(ctx, &workflowservice.PollActivityTaskQueueRequest{ Namespace: env.Namespace().String(), TaskQueue: tv.TaskQueue(), Identity: uuid.NewString(), DeploymentOptions: tv.WorkerDeploymentOptions(true), }) } func (s *DeploymentVersionSuite) describeVersion(env *testcore.TestEnv, tv *testvars.TestVars) (*workflowservice.DescribeWorkerDeploymentVersionResponse, error) { ctx, cancel := context.WithTimeout(s.Context(), 10*time.Second) defer cancel() req := &workflowservice.DescribeWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), } if useV32 { req.DeploymentVersion = tv.ExternalDeploymentVersion() } else { req.Version = tv.DeploymentVersionString() //nolint:staticcheck // SA1019: worker versioning v0.31 } return env.FrontendClient().DescribeWorkerDeploymentVersion(ctx, req) } func (s *DeploymentVersionSuite) updateMetadata(env *testcore.TestEnv, tv *testvars.TestVars, upsertEntries map[string]*commonpb.Payload, removeEntries []string) (*workflowservice.UpdateWorkerDeploymentVersionMetadataResponse, error) { ctx, cancel := context.WithTimeout(s.Context(), 10*time.Second) defer cancel() req := &workflowservice.UpdateWorkerDeploymentVersionMetadataRequest{ Namespace: env.Namespace().String(), UpsertEntries: upsertEntries, RemoveEntries: removeEntries, } if useV32 { req.DeploymentVersion = tv.ExternalDeploymentVersion() } else { req.Version = tv.DeploymentVersionString() //nolint:staticcheck // SA1019: worker versioning v0.31 } return env.FrontendClient().UpdateWorkerDeploymentVersionMetadata(ctx, req) } func (s *DeploymentVersionSuite) startVersionWorkflow(ctx context.Context, env *testcore.TestEnv, tv *testvars.TestVars) { go s.pollFromDeployment(ctx, env, tv) s.EventuallyWithT(func(t *assert.CollectT) { a := assert.New(t) resp, err := s.describeVersion(env, tv) if !a.NoError(err) { return } // regardless of s.useV32, we want to read both version formats a.Equal(tv.DeploymentVersionString(), resp.GetWorkerDeploymentVersionInfo().GetVersion()) a.Equal(tv.ExternalDeploymentVersion().GetDeploymentName(), resp.GetWorkerDeploymentVersionInfo().GetDeploymentVersion().GetDeploymentName()) a.Equal(tv.ExternalDeploymentVersion().GetBuildId(), resp.GetWorkerDeploymentVersionInfo().GetDeploymentVersion().GetBuildId()) a.Equal(enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, resp.GetWorkerDeploymentVersionInfo().GetStatus()) newResp, err := env.FrontendClient().DescribeWorkerDeployment(ctx, &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: tv.DeploymentSeries(), }) if !a.NoError(err) { return } var versionSummaryNames []string var versionSummaryVersions []*deploymentpb.WorkerDeploymentVersion for _, versionSummary := range newResp.GetWorkerDeploymentInfo().GetVersionSummaries() { versionSummaryNames = append(versionSummaryNames, versionSummary.GetVersion()) versionSummaryVersions = append(versionSummaryVersions, versionSummary.GetDeploymentVersion()) } a.Contains(versionSummaryNames, tv.DeploymentVersionString()) contains := slices.ContainsFunc(versionSummaryVersions, func(v *deploymentpb.WorkerDeploymentVersion) bool { return v.GetDeploymentName() == tv.ExternalDeploymentVersion().GetDeploymentName() && v.GetBuildId() == tv.ExternalDeploymentVersion().GetBuildId() }) a.True(contains) }, time.Second*5, time.Millisecond*200) } func (s *DeploymentVersionSuite) startVersionWorkflowExpectFailAddVersion(ctx context.Context, env *testcore.TestEnv, tv *testvars.TestVars) { _, err := env.FrontendClient().PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{ Namespace: env.Namespace().String(), TaskQueue: tv.TaskQueue(), Identity: "random", DeploymentOptions: tv.WorkerDeploymentOptions(true), }) var resourceExhausted *serviceerror.ResourceExhausted s.ErrorAs(err, &resourceExhausted) s.Contains(resourceExhausted.Message, "maximum number of versions") } func (s *DeploymentVersionSuite) TestForceCAN_NoOpenWFS() { env := s.newTestEnv() // Start a version workflow s.startVersionWorkflow(s.Context(), env, env.Tv()) // Set the version as current err := s.setCurrent(env, env.Tv(), false) s.NoError(err) // ForceCAN versionWorkflowID := workerdeployment.GenerateVersionWorkflowID(env.Tv().DeploymentSeries(), env.Tv().BuildID()) workflowExecution := &commonpb.WorkflowExecution{ WorkflowId: versionWorkflowID, } err = env.SendSignal(env.Namespace().String(), workflowExecution, workerdeployment.ForceCANSignalName, nil, env.Tv().ClientIdentity()) s.NoError(err) // verifying we see our registered workers in the version deployment even after a CAN s.EventuallyWithT(func(t *assert.CollectT) { a := assert.New(t) resp, err := s.describeVersion(env, env.Tv()) if !a.NoError(err) { return } a.Equal(env.Tv().DeploymentVersionString(), resp.GetWorkerDeploymentVersionInfo().GetVersion()) //nolint:staticcheck // SA1019: worker versioning v0.31 a.Equal(env.Tv().ExternalDeploymentVersion().GetDeploymentName(), resp.GetWorkerDeploymentVersionInfo().GetDeploymentVersion().GetDeploymentName()) a.Equal(env.Tv().ExternalDeploymentVersion().GetBuildId(), resp.GetWorkerDeploymentVersionInfo().GetDeploymentVersion().GetBuildId()) a.Len(resp.GetVersionTaskQueues(), 1) a.Len(resp.GetWorkerDeploymentVersionInfo().GetTaskQueueInfos(), 1) // verify that the version state is intact even after a CAN a.Equal(env.Tv().TaskQueue().GetName(), resp.GetVersionTaskQueues()[0].Name) a.Equal(env.Tv().TaskQueue().GetName(), resp.GetWorkerDeploymentVersionInfo().GetTaskQueueInfos()[0].Name) a.NotNil(resp.GetWorkerDeploymentVersionInfo().GetCurrentSinceTime()) a.NotNil(resp.GetWorkerDeploymentVersionInfo().GetRoutingChangedTime()) a.NotNil(resp.GetWorkerDeploymentVersionInfo().GetCurrentSinceTime()) a.Nil(resp.GetWorkerDeploymentVersionInfo().GetDrainageInfo()) a.Equal(enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT, resp.GetWorkerDeploymentVersionInfo().GetStatus()) }, time.Second*10, time.Millisecond*1000) } func (s *DeploymentVersionSuite) TestForceCAN_WithOverrideState() { env := s.newTestEnv() // Start a version workflow s.startVersionWorkflow(s.Context(), env, env.Tv()) // Create a modified state with metadata to verify override works overrideState := &deploymentspb.VersionLocalState{ Version: env.Tv().DeploymentVersion(), CreateTime: timestamppb.New(time.Now()), Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, Metadata: &deploymentpb.VersionMetadata{ Entries: map[string]*commonpb.Payload{ "override-key": {Data: []byte("override-value")}, }, }, TaskQueueFamilies: map[string]*deploymentspb.VersionLocalState_TaskQueueFamilyData{ env.Tv().TaskQueue().GetName(): { TaskQueues: map[int32]*deploymentspb.TaskQueueVersionData{ int32(enumspb.TASK_QUEUE_TYPE_WORKFLOW): {}, }, }, }, } // Create signal args with the override state signalArgs := &deploymentspb.ForceCANVersionSignalArgs{ OverrideState: overrideState, } marshaledData, err := proto.Marshal(signalArgs) s.NoError(err) signalPayload := &commonpb.Payloads{ Payloads: []*commonpb.Payload{ { Metadata: map[string][]byte{ "encoding": []byte("binary/protobuf"), }, Data: marshaledData, }, }, } // Send ForceCAN signal with override state versionWorkflowID := workerdeployment.GenerateVersionWorkflowID(env.Tv().DeploymentSeries(), env.Tv().BuildID()) workflowExecution := &commonpb.WorkflowExecution{ WorkflowId: versionWorkflowID, } err = env.SendSignal(env.Namespace().String(), workflowExecution, workerdeployment.ForceCANSignalName, signalPayload, env.Tv().ClientIdentity()) s.NoError(err) // Verify that the override state is used after CAN (metadata should be present) s.EventuallyWithT(func(t *assert.CollectT) { a := assert.New(t) resp, err := s.describeVersion(env, env.Tv()) if !a.NoError(err) { return } // Verify the metadata from override state is present entries := resp.GetWorkerDeploymentVersionInfo().GetMetadata().GetEntries() if !a.Len(entries, 1) { return } a.Equal([]byte("override-value"), entries["override-key"].Data) }, time.Second*10, time.Millisecond*1000) } func (s *DeploymentVersionSuite) TestDescribeVersion_RegisterTaskQueue() { env := s.newTestEnv() numberOfDeployments := 1 // Starting a deployment workflow go s.pollFromDeployment(s.Context(), env, env.Tv()) // Querying the Deployment s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) resp, err := s.describeVersion(env, env.Tv()) a.NoError(err) a.Equal(env.Tv().DeploymentVersionString(), resp.GetWorkerDeploymentVersionInfo().GetVersion()) //nolint:staticcheck // SA1019: worker versioning v0.31 a.Equal(env.Tv().ExternalDeploymentVersion().GetDeploymentName(), resp.GetWorkerDeploymentVersionInfo().GetDeploymentVersion().GetDeploymentName()) a.Equal(env.Tv().ExternalDeploymentVersion().GetBuildId(), resp.GetWorkerDeploymentVersionInfo().GetDeploymentVersion().GetBuildId()) a.Equal(enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, resp.GetWorkerDeploymentVersionInfo().GetStatus()) a.Len(resp.GetWorkerDeploymentVersionInfo().GetTaskQueueInfos(), numberOfDeployments) a.Equal(env.Tv().TaskQueue().GetName(), resp.GetWorkerDeploymentVersionInfo().GetTaskQueueInfos()[0].Name) a.Len(resp.GetVersionTaskQueues(), numberOfDeployments) a.Equal(env.Tv().TaskQueue().GetName(), resp.GetVersionTaskQueues()[0].Name) }, time.Second*5, time.Millisecond*200) } func (s *DeploymentVersionSuite) TestDescribeVersion_RegisterTaskQueue_ConcurrentPollers() { env := s.newTestEnv() root, err := tqid.PartitionFromProto(env.Tv().TaskQueue(), env.Namespace().String(), enumspb.TASK_QUEUE_TYPE_WORKFLOW) s.NoError(err) // Making concurrent polls to 4 partitions, 3 polls to each for p := range 4 { tv2 := env.Tv().WithTaskQueue(root.TaskQueue().NormalPartition(p).RpcName()) for range 3 { go s.pollFromDeployment(s.Context(), env, tv2) go s.pollActivityFromDeployment(s.Context(), env, tv2) } } // Querying the Worker Deployment Version s.EventuallyWithT(func(t *assert.CollectT) { a := assert.New(t) resp, err := s.describeVersion(env, env.Tv()) if !a.NoError(err) { return } a.Equal(env.Tv().DeploymentVersionString(), resp.GetWorkerDeploymentVersionInfo().GetVersion()) //nolint:staticcheck // SA1019: worker versioning v0.31 a.Equal(env.Tv().ExternalDeploymentVersion().GetDeploymentName(), resp.GetWorkerDeploymentVersionInfo().GetDeploymentVersion().GetDeploymentName()) a.Equal(env.Tv().ExternalDeploymentVersion().GetBuildId(), resp.GetWorkerDeploymentVersionInfo().GetDeploymentVersion().GetBuildId()) a.Equal(enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, resp.GetWorkerDeploymentVersionInfo().GetStatus()) a.Len(resp.GetWorkerDeploymentVersionInfo().GetTaskQueueInfos(), 2) a.Equal(env.Tv().TaskQueue().GetName(), resp.GetWorkerDeploymentVersionInfo().GetTaskQueueInfos()[0].Name) a.Len(resp.GetVersionTaskQueues(), 2) a.Equal(env.Tv().TaskQueue().GetName(), resp.GetVersionTaskQueues()[0].Name) }, time.Second*10, time.Millisecond*1000) } //nolint:forbidigo func (s *DeploymentVersionSuite) TestDrainageStatus_SetCurrentVersion_NoOpenWFs() { env := s.newTestEnv() tv1 := env.Tv().WithBuildIDNumber(1) tv2 := env.Tv().WithBuildIDNumber(2) // Start deployment workflow 1 and wait for the deployment version to exist s.startVersionWorkflow(s.Context(), env, tv1) // Start deployment workflow 2 and wait for the deployment version to exist s.startVersionWorkflow(s.Context(), env, tv2) // non-current deployments have never been used and have no drainage info s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{}, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, 0) s.checkVersionDrainageAndVersionStatus(env, tv2, &deploymentpb.VersionDrainageInfo{}, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, 0) // SetCurrent tv1 err := s.setCurrent(env, tv1, true) s.NoError(err) // Both versions have no drainage info and tv1 has it's status updated to current s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{}, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT, 0) s.checkVersionDrainageAndVersionStatus(env, tv2, &deploymentpb.VersionDrainageInfo{}, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, 0) baseTime := time.Now() // SetCurrent tv2 --> tv1 starts the child drainage workflow err = s.setCurrent(env, tv2, true) s.NoError(err) changed1, checked1 := s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{ Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING, }, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING, 0) s.Greater(changed1, baseTime) s.GreaterOrEqual(checked1, changed1) // tv1 should now be "drained" changed2, checked2 := s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{ Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINED, }, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINED, testVersionDrainageVisibilityGracePeriod) s.Greater(changed2, changed1) s.GreaterOrEqual(checked2, changed2) } func (s *DeploymentVersionSuite) TestDrainageStatus_SetCurrentVersion_YesOpenWFs() { env := s.newTestEnv() tv1 := env.Tv().WithBuildIDNumber(1) tv2 := env.Tv().WithBuildIDNumber(2) // start deployment workflow 1 and wait for the deployment version to exist s.startVersionWorkflow(s.Context(), env, tv1) // start deployment workflow 2 and wait for the deployment version to exist s.startVersionWorkflow(s.Context(), env, tv2) // non-current deployments have never been used and have no drainage info s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{}, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, 0) s.checkVersionDrainageAndVersionStatus(env, tv2, &deploymentpb.VersionDrainageInfo{}, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, 0) // SetCurrent tv1 err := s.setCurrent(env, tv1, true) s.NoError(err) // both versions have no drainage info and tv1 has it's status updated to current s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{}, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT, 0) s.checkVersionDrainageAndVersionStatus(env, tv2, &deploymentpb.VersionDrainageInfo{}, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, 0) // start a pinned workflow on v1 run := s.startPinnedWorkflow(s.Context(), env, tv1) baseTime := time.Now() // SetCurrent tv2 --> tv1 starts the child drainage workflow err = s.setCurrent(env, tv2, true) s.NoError(err) // tv1 should now be "draining" for visibilityGracePeriod duration changed1, checked1 := s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{ Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING, }, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING, 0) s.Greater(changed1, baseTime) s.GreaterOrEqual(checked1, changed1) // tv1 should still be "draining" for visibilityGracePeriod duration changed2, checked2 := s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{ Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING, }, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING, testVersionDrainageVisibilityGracePeriod) s.Equal(changed2, changed1) s.Greater(checked2, checked1) // tv1 should still be "draining" after a refresh intervals changed3, checked3 := s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{ Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING, }, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING, testVersionDrainageRefreshInterval) s.Equal(changed3, changed1) s.Greater(checked3, checked2) // terminate workflow _, err = env.FrontendClient().TerminateWorkflowExecution(s.Context(), &workflowservice.TerminateWorkflowExecutionRequest{ Namespace: env.Namespace().String(), WorkflowExecution: &commonpb.WorkflowExecution{ WorkflowId: run.GetID(), RunId: run.GetRunID(), }, Reason: "test", Identity: tv1.ClientIdentity(), }) s.NoError(err) // tv1 should now be "drained" changed4, checked4 := s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{ Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINED, }, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINED, 0) s.Greater(changed4, changed3) s.GreaterOrEqual(checked4, changed4) } func (s *DeploymentVersionSuite) startVersionedWorkflow(ctx context.Context, env *testcore.TestEnv, tv *testvars.TestVars, behavior workflow.VersioningBehavior) sdkclient.WorkflowRun { started := make(chan struct{}, 1) wf := func(ctx workflow.Context) (string, error) { started <- struct{}{} workflow.GetSignalChannel(ctx, "wait").Receive(ctx, nil) if workflow.GetInfo(ctx).Attempt == 1 { return "", errors.New("try again") //nolint:err113 } panic("oops") } wId := testcore.RandomizeStr("id") w := worker.New(env.SdkClient(), tv.TaskQueue().String(), worker.Options{ DeploymentOptions: worker.DeploymentOptions{ Version: tv.SDKDeploymentVersion(), UseVersioning: true, }, Identity: wId, }) w.RegisterWorkflowWithOptions(wf, workflow.RegisterOptions{VersioningBehavior: behavior}) s.NoError(w.Start()) defer w.Stop() run, err := env.SdkClient().ExecuteWorkflow(ctx, sdkclient.StartWorkflowOptions{TaskQueue: tv.TaskQueue().String()}, wf) s.NoError(err) env.WaitForChannel(started) return run } func (s *DeploymentVersionSuite) startPinnedWorkflow(ctx context.Context, env *testcore.TestEnv, tv *testvars.TestVars) sdkclient.WorkflowRun { return s.startVersionedWorkflow(ctx, env, tv, workflow.VersioningBehaviorPinned) } func (s *DeploymentVersionSuite) startUnpinnedWorkflow(ctx context.Context, env *testcore.TestEnv, tv *testvars.TestVars) sdkclient.WorkflowRun { return s.startVersionedWorkflow(ctx, env, tv, workflow.VersioningBehaviorAutoUpgrade) } func (s *DeploymentVersionSuite) TestVersionIgnoresDrainageSignalWhenCurrentOrRamping() { env := s.newTestEnv() tv1 := env.Tv().WithBuildIDNumber(1) // Start deployment workflow 1 and wait for the deployment version to exist s.startVersionWorkflow(s.Context(), env, tv1) // Make it current err := s.setCurrent(env, tv1, false) s.NoError(err) // Signal it to be drained. Only do this in tests. versionWorkflowID := workerdeployment.GenerateVersionWorkflowID(tv1.DeploymentSeries(), tv1.BuildID()) workflowExecution := &commonpb.WorkflowExecution{ WorkflowId: versionWorkflowID, } input := &deploymentpb.VersionDrainageInfo{ Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINED, LastChangedTime: timestamppb.New(time.Now()), LastCheckedTime: timestamppb.New(time.Now()), } marshaledData, err := input.Marshal() s.NoError(err) signalPayload := &commonpb.Payloads{ Payloads: []*commonpb.Payload{ { Metadata: map[string][]byte{ "encoding": []byte("binary/protobuf"), }, Data: marshaledData, }, }, } err = env.SendSignal(env.Namespace().String(), workflowExecution, workerdeployment.SyncDrainageSignalName, signalPayload, tv1.ClientIdentity()) s.NoError(err) // describe version and confirm that it is not drained // add a 3s time requirement so that it does not succeed immediately sentSignal := time.Now() s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) a.Greater(time.Since(sentSignal), 2*time.Second) resp, err := s.describeVersion(env, tv1) a.NoError(err) a.NotEqual(enumspb.VERSION_DRAINAGE_STATUS_DRAINED, resp.GetWorkerDeploymentVersionInfo().GetDrainageInfo().GetStatus()) }, time.Second*10, time.Millisecond*1000) } // Testing DeleteVersion func (s *DeploymentVersionSuite) TestDeleteVersion_DeleteCurrentVersion() { // Override the dynamic config so that we can verify we don't get any unexpected masked errors. env := s.newTestEnv(testcore.WithDynamicConfig(dynamicconfig.FrontendMaskInternalErrorDetails, true)) tv1 := env.Tv().WithBuildIDNumber(1) // Create a deployment version s.startVersionWorkflow(s.Context(), env, tv1) // Set version as current err := s.setCurrent(env, tv1, false) s.NoError(err) // Deleting this version should fail since the version is current s.tryDeleteVersion(env, tv1, fmt.Sprintf(workerdeployment.ErrVersionIsCurrentOrRamping, tv1.DeploymentVersionStringV32()), false) // Verifying workflow is not in a locked state after an invalid delete request such as the one above. If the workflow were in a locked // state, the passed context would have timed out making the following operation fail. s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) resp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: tv1.DeploymentSeries(), }) a.NoError(err) a.Equal(tv1.DeploymentVersionString(), resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetCurrentVersion()) //nolint:staticcheck // SA1019: worker versioning v0.31 a.Equal(tv1.ExternalDeploymentVersion(), resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetCurrentDeploymentVersion()) }, time.Second*5, time.Millisecond*200) } func (s *DeploymentVersionSuite) TestDeleteVersion_DeleteRampedVersion() { env := s.newTestEnv() tv1 := env.Tv().WithBuildIDNumber(1) // Create a deployment version s.startVersionWorkflow(s.Context(), env, tv1) // Set version as ramping err := s.setRamping(env, tv1, 0) s.NoError(err) // Deleting this version should fail since the version is ramping s.tryDeleteVersion(env, tv1, fmt.Sprintf(workerdeployment.ErrVersionIsCurrentOrRamping, tv1.DeploymentVersionStringV32()), false) // Verifying workflow is not in a locked state after an invalid delete request such as the one above. If the workflow were in a locked // state, the passed context would have timed out making the following operation fail. s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) resp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: tv1.DeploymentSeries(), }) a.NoError(err) a.Equal(tv1.DeploymentVersionString(), resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetRampingVersion()) //nolint:staticcheck // SA1019: worker versioning v0.31 a.Equal(tv1.ExternalDeploymentVersion(), resp.GetWorkerDeploymentInfo().GetRoutingConfig().GetRampingDeploymentVersion()) }, time.Second*5, time.Millisecond*200) } func (s *DeploymentVersionSuite) TestDeleteVersion_NoWfs() { env := s.newTestEnv(testcore.WithDynamicConfig(dynamicconfig.PollerHistoryTTL, 500*time.Millisecond)) tv1 := env.Tv().WithBuildIDNumber(1) // Create a deployment version s.startVersionWorkflow(s.Context(), env, tv1) //nolint:forbidigo time.Sleep(2 * time.Second) // todo (Shivam): remove this after the above skip is removed // delete should succeed s.tryDeleteVersion(env, tv1, "", false) // deployment version does not exist in the deployment list s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) resp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: tv1.DeploymentSeries(), }) a.NoError(err) if resp != nil { for _, vs := range resp.GetWorkerDeploymentInfo().GetVersionSummaries() { a.NotEqual(tv1.DeploymentVersionString(), vs.Version) //nolint:staticcheck // SA1019: worker versioning v0.31 a.NotEqual(tv1.ExternalDeploymentVersion().GetDeploymentName(), vs.GetDeploymentVersion().GetDeploymentName()) a.NotEqual(tv1.ExternalDeploymentVersion().GetBuildId(), vs.GetDeploymentVersion().GetBuildId()) } } }, time.Second*5, time.Millisecond*200) } func (s *DeploymentVersionSuite) TestDeleteVersion_DrainingVersion() { env := s.newTestEnv() tv1 := env.Tv().WithBuildIDNumber(1) // Start deployment workflow 1 and wait for the deployment version to exist s.startVersionWorkflow(s.Context(), env, tv1) // Make the version current err := s.setCurrent(env, tv1, false) s.NoError(err) // Start another version workflow tv2 := env.Tv().WithBuildIDNumber(2) s.startVersionWorkflow(s.Context(), env, tv2) // Setting this version to current should start the drainage workflow for version1 and make it draining err = s.setCurrent(env, tv2, true) s.NoError(err) // Version should be draining s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{ Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING, LastChangedTime: nil, // don't test this now LastCheckedTime: nil, // don't test this now }, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING, 0) // delete should fail s.tryDeleteVersion(env, tv1, fmt.Sprintf(workerdeployment.ErrVersionIsDraining, tv1.DeploymentVersionStringV32()), false) } func (s *DeploymentVersionSuite) TestDeleteVersion_Drained_But_Pollers_Exist() { env := s.newTestEnv() tv1 := env.Tv().WithBuildIDNumber(1) // Start deployment workflow 1 and wait for the deployment version to exist s.startVersionWorkflow(s.Context(), env, tv1) // Make the version current err := s.setCurrent(env, tv1, false) s.NoError(err) // Start another version workflow tv2 := env.Tv().WithBuildIDNumber(2) s.startVersionWorkflow(s.Context(), env, tv2) // Setting this version to current should start the drainage workflow for version1 err = s.setCurrent(env, tv2, true) s.NoError(err) // Signal the first version to be drained. Only do this in tests. s.signalAndWaitForDrained(env, tv1) // Version will bypass "drained" check but delete should still fail since we have active pollers. s.tryDeleteVersion(env, tv1, fmt.Sprintf(workerdeployment.ErrVersionHasPollers, tv1.DeploymentVersionStringV32()), false) } func (s *DeploymentVersionSuite) signalAndWaitForDrained(env *testcore.TestEnv, tv *testvars.TestVars) { versionWorkflowID := workerdeployment.GenerateVersionWorkflowID(tv.DeploymentSeries(), tv.BuildID()) workflowExecution := &commonpb.WorkflowExecution{ WorkflowId: versionWorkflowID, } input := &deploymentpb.VersionDrainageInfo{ Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINED, LastChangedTime: timestamppb.New(time.Now()), LastCheckedTime: timestamppb.New(time.Now()), } marshaledData, err := input.Marshal() s.NoError(err) signalPayload := &commonpb.Payloads{ Payloads: []*commonpb.Payload{ { Metadata: map[string][]byte{ "encoding": []byte("binary/protobuf"), }, Data: marshaledData, }, }, } err = env.SendSignal(env.Namespace().String(), workflowExecution, workerdeployment.SyncDrainageSignalName, signalPayload, tv.ClientIdentity()) s.NoError(err) // wait for drained s.EventuallyWithT(func(t *assert.CollectT) { resp, err := s.describeVersion(env, tv) assert.NoError(t, err) assert.Equal(t, enumspb.VERSION_DRAINAGE_STATUS_DRAINED, resp.GetWorkerDeploymentVersionInfo().GetDrainageInfo().GetStatus()) }, 10*time.Second, time.Second) } func (s *DeploymentVersionSuite) waitForPollers(env *testcore.TestEnv, tv *testvars.TestVars, moreExpectedVersions ...*testvars.TestVars) { expectedVersionsStr := []string{tv.DeploymentVersionStringV32()} for _, tv2 := range moreExpectedVersions { if !tv2.ExternalDeploymentVersion().Equal(tv.ExternalDeploymentVersion()) { expectedVersionsStr = append(expectedVersionsStr, tv2.DeploymentVersionStringV32()) } } s.EventuallyWithT(func(t *assert.CollectT) { resp, err := env.FrontendClient().DescribeTaskQueue(s.Context(), &workflowservice.DescribeTaskQueueRequest{ Namespace: env.Namespace().String(), TaskQueue: tv.TaskQueue(), TaskQueueType: enumspb.TASK_QUEUE_TYPE_WORKFLOW, }) require.NoError(t, err) versionsSeen := 0 for _, poller := range resp.Pollers { pollerV := worker_versioning.WorkerDeploymentVersionToStringV32(worker_versioning.DeploymentVersionFromOptions(poller.GetDeploymentOptions())) for _, v := range expectedVersionsStr { if pollerV == v { versionsSeen++ } } } require.Equal(t, len(expectedVersionsStr), versionsSeen) }, 10*time.Second, 100*time.Millisecond) } func (s *DeploymentVersionSuite) waitForNoPollers(env *testcore.TestEnv, tv *testvars.TestVars, moreUnexpectedVersions ...*testvars.TestVars) { unexpectedVersionsStr := []string{tv.DeploymentVersionStringV32()} for _, tv2 := range moreUnexpectedVersions { if !tv2.ExternalDeploymentVersion().Equal(tv.ExternalDeploymentVersion()) { unexpectedVersionsStr = append(unexpectedVersionsStr, tv2.DeploymentVersionStringV32()) } } s.EventuallyWithT(func(t *assert.CollectT) { resp, err := env.FrontendClient().DescribeTaskQueue(s.Context(), &workflowservice.DescribeTaskQueueRequest{ Namespace: env.Namespace().String(), TaskQueue: tv.TaskQueue(), TaskQueueType: enumspb.TASK_QUEUE_TYPE_WORKFLOW, }) require.NoError(t, err) versionsSeen := 0 fmt.Printf("Pollers: %+v\n", resp.Pollers) for _, poller := range resp.Pollers { pollerV := worker_versioning.WorkerDeploymentVersionToStringV32(worker_versioning.DeploymentVersionFromOptions(poller.GetDeploymentOptions())) for _, v := range unexpectedVersionsStr { if pollerV == v { versionsSeen++ } } } require.Equal(t, 0, versionsSeen) }, 10*time.Second, 100*time.Millisecond) } func (s *DeploymentVersionSuite) TestVersionScavenger_DeleteOnAdd() { env := s.newTestEnv( testcore.WithDynamicConfig(dynamicconfig.PollerHistoryTTL, 3*time.Second), testcore.WithDynamicConfig(dynamicconfig.MatchingMaxVersionsInDeployment, testMaxVersionsInDeployment), // we don't want the version to drain in this test testcore.WithDynamicConfig(dynamicconfig.VersionDrainageStatusVisibilityGracePeriod, 60*time.Second), testcore.WithDynamicConfig(dynamicconfig.TaskQueueInfoByBuildIdTTL, 0), ) // Set deployment register error backoff to zero so to speed up the test. env.InjectHook(testhooks.NewHook(testhooks.MatchingDeploymentRegisterErrorBackoff, 0*time.Second)) tvs := make([]*testvars.TestVars, testMaxVersionsInDeployment) // max out the versions for i := range testMaxVersionsInDeployment { tvs[i] = env.Tv().WithBuildIDNumber(i) s.startVersionWorkflow(s.Context(), env, tvs[i]) } // Make tvs[0] current err := s.setCurrent(env, tvs[0], false) s.NoError(err) // Make tvs[1] current, hence tvs[0] should go to draining err = s.setCurrent(env, tvs[1], false) s.NoError(err) // CI can be slow, keep sending fresh polls to ensure that auto deletion logic sees them when we want to add tvMax so it can't add. pollContext, cancelPolls := context.WithTimeout(context.Background(), 3*time.Second) go func() { for i := range testMaxVersionsInDeployment { go s.pollFromDeployment(pollContext, env, tvs[i]) } t := time.NewTicker(time.Second) for { select { case <-pollContext.Done(): return case <-t.C: for i := range testMaxVersionsInDeployment { go s.pollFromDeployment(pollContext, env, tvs[i]) } } } }() s.waitForPollers(env, tvs[0], tvs...) tvMax := env.Tv().WithBuildIDNumber(9999) cancelPolls() // try to add a version and it fails because none of the versions can be deleted s.startVersionWorkflowExpectFailAddVersion(s.Context(), env, tvMax) // this waits for no pollers from any of original versions (tvMax pollers should be fine) s.waitForNoPollers(env, tvs[0], tvs...) // try to add the version again, and it succeeds, after deleting tvs[2] version but not tvs[3] (both are eligible) s.startVersionWorkflow(s.Context(), env, tvMax) // tvs[0] is draining so can't be deleted. tvs[1] is current, so tvs[2] should be deleted. s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) resp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: tvMax.DeploymentSeries(), }) a.NoError(err) var versions []string for _, vs := range resp.GetWorkerDeploymentInfo().GetVersionSummaries() { versions = append(versions, vs.Version) //nolint:staticcheck // SA1019: worker versioning v0.31 } a.NotContains(versions, tvs[2].DeploymentVersionString()) a.Contains(versions, tvs[0].DeploymentVersionString()) a.Contains(versions, tvs[1].DeploymentVersionString()) a.Contains(versions, tvs[3].DeploymentVersionString()) }, time.Second*5, time.Millisecond*200) } func (s *DeploymentVersionSuite) TestDeleteVersion_ValidDelete() { env := s.newTestEnv(testcore.WithDynamicConfig(dynamicconfig.PollerHistoryTTL, 500*time.Millisecond)) tv1 := env.Tv().WithBuildIDNumber(1) // Start deployment workflow 1 and wait for the deployment version to exist s.startVersionWorkflow(s.Context(), env, tv1) // Signal the first version to be drained. Only do this in tests. s.signalAndWaitForDrained(env, tv1) // Wait for pollers going away s.waitForNoPollers(env, tv1, tv1) // delete succeeds s.tryDeleteVersion(env, tv1, "", false) // deployment version does not exist in the deployment list s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) resp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: tv1.DeploymentSeries(), }) a.NoError(err) if resp != nil { for _, vs := range resp.GetWorkerDeploymentInfo().GetVersionSummaries() { a.NotEqual(tv1.DeploymentVersionString(), vs.Version) //nolint:staticcheck // SA1019: worker versioning v0.31 a.NotEqual(tv1.ExternalDeploymentVersion().GetDeploymentName(), vs.GetDeploymentVersion().GetDeploymentName()) a.NotEqual(tv1.ExternalDeploymentVersion().GetBuildId(), vs.GetDeploymentVersion().GetBuildId()) } } }, time.Second*5, time.Millisecond*200) // idempotency check: deleting the same version again should succeed s.tryDeleteVersion(env, tv1, "", false) } func (s *DeploymentVersionSuite) skipBeforeVersion(version workerdeployment.DeploymentWorkflowVersion) { if workerdeployment.VersionDataRevisionNumber < version { s.T().Skipf("test supports version %v and newer", version) } } func (s *DeploymentVersionSuite) TestDeleteVersion_ValidDelete_SkipDrainage() { env := s.newTestEnv(testcore.WithDynamicConfig(dynamicconfig.PollerHistoryTTL, 500*time.Millisecond)) tv1 := env.Tv().WithBuildIDNumber(1) // Start deployment workflow 1 and wait for the deployment version to exist s.startVersionWorkflow(s.Context(), env, tv1) // Wait for pollers going away s.EventuallyWithT(func(t *assert.CollectT) { resp, err := env.FrontendClient().DescribeTaskQueue(s.Context(), &workflowservice.DescribeTaskQueueRequest{ Namespace: env.Namespace().String(), TaskQueue: tv1.TaskQueue(), TaskQueueType: enumspb.TASK_QUEUE_TYPE_WORKFLOW, }) require.NoError(t, err) require.Empty(t, resp.Pollers) }, 5*time.Second, time.Second) // skipDrainage=true will make delete succeed s.tryDeleteVersion(env, tv1, "", false) // deployment version does not exist in the deployment list s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) resp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: tv1.DeploymentSeries(), }) a.NoError(err) if resp != nil { for _, vs := range resp.GetWorkerDeploymentInfo().GetVersionSummaries() { a.NotEqual(tv1.DeploymentVersionString(), vs.Version) //nolint:staticcheck // SA1019: worker versioning v0.31 a.NotEqual(tv1.ExternalDeploymentVersion().GetDeploymentName(), vs.GetDeploymentVersion().GetDeploymentName()) a.NotEqual(tv1.ExternalDeploymentVersion().GetBuildId(), vs.GetDeploymentVersion().GetBuildId()) } } }, time.Second*5, time.Millisecond*200) // idempotency check: deleting the same version again should succeed s.tryDeleteVersion(env, tv1, "", false) // Describe Worker Deployment should give not found // describe deployment version gives not found error s.EventuallyWithT(func(t *assert.CollectT) { a := assert.New(t) _, err := s.describeVersion(env, tv1) a.Error(err) var nfe *serviceerror.NotFound a.ErrorAs(err, &nfe) }, time.Second*5, time.Millisecond*200) } func (s *DeploymentVersionSuite) TestDeleteVersion_ConcurrentDeleteVersion() { env := s.newTestEnv(testcore.WithDynamicConfig(dynamicconfig.PollerHistoryTTL, 500*time.Millisecond)) tv1 := env.Tv().WithBuildIDNumber(1) // Start deployment workflow 1 and wait for the deployment version to exist s.startVersionWorkflow(s.Context(), env, tv1) // Wait for pollers going away s.EventuallyWithT(func(t *assert.CollectT) { resp, err := env.FrontendClient().DescribeTaskQueue(s.Context(), &workflowservice.DescribeTaskQueueRequest{ Namespace: env.Namespace().String(), TaskQueue: tv1.TaskQueue(), TaskQueueType: enumspb.TASK_QUEUE_TYPE_WORKFLOW, }) require.NoError(t, err) require.Empty(t, resp.Pollers) }, 10*time.Second, time.Second) // concurrent delete version requests should not break the system. var wg sync.WaitGroup wg.Add(2) go func() { defer wg.Done() s.tryDeleteVersion(env, tv1, "", false) //nolint:testifylint // concurrent delete requests are expected to all succeed }() go func() { defer wg.Done() s.tryDeleteVersion(env, tv1, "", false) //nolint:testifylint // concurrent delete requests are expected to all succeed }() wg.Wait() // deployment version does not exist in the deployment list s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) resp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: tv1.DeploymentSeries(), }) a.NoError(err) if resp != nil { for _, vs := range resp.GetWorkerDeploymentInfo().GetVersionSummaries() { a.NotEqual(tv1.DeploymentVersionString(), vs.Version) //nolint:staticcheck // SA1019: worker versioning v0.31 a.NotEqual(tv1.ExternalDeploymentVersion().GetDeploymentName(), vs.GetDeploymentVersion().GetDeploymentName()) a.NotEqual(tv1.ExternalDeploymentVersion().GetBuildId(), vs.GetDeploymentVersion().GetBuildId()) } } }, time.Second*10, time.Millisecond*200) } // VersionMissingTaskQueues func (s *DeploymentVersionSuite) TestVersionMissingTaskQueues_InvalidSetCurrentVersion() { // Override the dynamic config to verify we don't get any unexpected masked errors. env := s.newTestEnv(testcore.WithDynamicConfig(dynamicconfig.FrontendMaskInternalErrorDetails, true)) tv1 := env.Tv().WithBuildIDNumber(1).WithTaskQueue(env.Tv().Any().String()) // Start deployment workflow 1 and wait for the deployment version to exist pollerCtx1, pollerCancel1 := context.WithCancel(s.Context()) s.startVersionWorkflow(pollerCtx1, env, tv1) // SetCurrent so that the task queue puts the version in its versions info err := s.setCurrent(env, tv1, false) s.NoError(err) // new version with a different registered task-queue tv2 := env.Tv().WithBuildIDNumber(2).WithTaskQueue(env.Tv().Any().String()) s.startVersionWorkflow(s.Context(), env, tv2) // Cancel pollers on task_queue_1 to increase the backlog of tasks pollerCancel1() // Start a workflow on task_queue_1 to increase the add rate s.startWorkflow(env, tv1, tv1.VersioningOverridePinned()) // SetCurrent tv2 err = s.setCurrent(env, tv2, false) // SetCurrent should fail since task_queue_1 does not have a current version than the deployment's existing current version // and it either has a backlog of tasks being present or an add rate > 0. s.EqualError(err, fmt.Sprintf(workerdeployment.ErrCurrentVersionDoesNotHaveAllTaskQueues, tv2.DeploymentVersionStringV32())) } func (s *DeploymentVersionSuite) TestVersionMissingTaskQueues_ValidSetCurrentVersion() { env := s.newTestEnv() tv1 := env.Tv().WithBuildIDNumber(1).WithTaskQueue(env.Tv().Any().String()) s.startVersionWorkflow(s.Context(), env, tv1) // SetCurrent so that the task queue puts the version in its versions info err := s.setCurrent(env, tv1, false) s.NoError(err) // new version with a different registered task-queue tv2 := env.Tv().WithBuildIDNumber(2).WithTaskQueue(env.Tv().Any().String()) s.startVersionWorkflow(s.Context(), env, tv2) // SetCurrent tv2 err = s.setCurrent(env, tv2, false) // SetCurrent tv2 should succeed as task_queue_1, despite missing from the new current version, has no backlogged tasks/add-rate > 0 s.NoError(err) } func (s *DeploymentVersionSuite) TestVersionMissingTaskQueues_InvalidSetRampingVersion() { // Override the dynamic config to verify we don't get any unexpected masked errors. env := s.newTestEnv(testcore.WithDynamicConfig(dynamicconfig.FrontendMaskInternalErrorDetails, true)) tv1 := env.Tv().WithBuildIDNumber(1).WithTaskQueue(env.Tv().Any().String()) // Start deployment workflow 1 and wait for the deployment version to exist pollerCtx1, pollerCancel1 := context.WithCancel(s.Context()) s.startVersionWorkflow(pollerCtx1, env, tv1) // SetCurrent so that the task queue puts the version in its versions info err := s.setCurrent(env, tv1, false) s.NoError(err) // new version with a different registered task-queue tv2 := env.Tv().WithBuildIDNumber(2).WithTaskQueue(env.Tv().Any().String()) s.startVersionWorkflow(s.Context(), env, tv2) // Cancel pollers on task_queue_1 to increase the backlog of tasks pollerCancel1() // Start a workflow on task_queue_1 to increase the add rate s.startWorkflow(env, tv1, tv1.VersioningOverridePinned()) // SetRampingVersion to tv2 err = s.setRamping(env, tv2, 0) // SetRampingVersion should fail since task_queue_1 does not have a current version than the deployment's existing current version // and it either has a backlog of tasks being present or an add rate > 0. s.EqualError(err, fmt.Sprintf(workerdeployment.ErrRampingVersionDoesNotHaveAllTaskQueues, tv2.DeploymentVersionStringV32())) } func (s *DeploymentVersionSuite) TestVersionMissingTaskQueues_ValidSetRampingVersion() { env := s.newTestEnv() tv1 := env.Tv().WithBuildIDNumber(1).WithTaskQueue(env.Tv().Any().String()) // Start deployment workflow 1 and wait for the deployment version to exist s.startVersionWorkflow(s.Context(), env, tv1) // SetCurrent so that the task queue puts the version in its versions info err := s.setCurrent(env, tv1, false) s.NoError(err) // new version with a different registered task-queue tv2 := env.Tv().WithBuildIDNumber(2).WithTaskQueue(env.Tv().Any().String()) s.startVersionWorkflow(s.Context(), env, tv2) // SetRampingVersion to tv2 err = s.setRamping(env, tv2, 0) // SetRampingVersion to tv2 should succeed as task_queue_1, despite missing from the new current version, has no backlogged tasks/add-rate > 0 s.NoError(err) } func (s *DeploymentVersionSuite) TestUpdateVersionMetadata() { env := s.newTestEnv() tv1 := env.Tv().WithBuildIDNumber(1) // Start deployment workflow 1 and wait for the deployment version to exist s.startVersionWorkflow(s.Context(), env, tv1) metadata := map[string]*commonpb.Payload{ "key1": {Data: testRandomMetadataValue}, "key2": {Data: testRandomMetadataValue}, } _, err := s.updateMetadata(env, tv1, metadata, nil) s.NoError(err) resp, err := s.describeVersion(env, tv1) s.NoError(err) // validating the metadata entries := resp.GetWorkerDeploymentVersionInfo().GetMetadata().GetEntries() s.Len(entries, 2) s.Equal(testRandomMetadataValue, entries["key1"].Data) s.Equal(testRandomMetadataValue, entries["key2"].Data) // Remove all the entries _, err = s.updateMetadata(env, tv1, nil, []string{"key1", "key2"}) s.NoError(err) resp, err = s.describeVersion(env, tv1) s.NoError(err) entries = resp.GetWorkerDeploymentVersionInfo().GetMetadata().GetEntries() s.Empty(entries) // update metadata for the second time with an explicit identity metadataIdentity := tv1.Any().String() metadataReq := &workflowservice.UpdateWorkerDeploymentVersionMetadataRequest{ Namespace: env.Namespace().String(), UpsertEntries: metadata, Identity: metadataIdentity, } if useV32 { metadataReq.DeploymentVersion = tv1.ExternalDeploymentVersion() } else { metadataReq.Version = tv1.DeploymentVersionString() //nolint:staticcheck // SA1019: worker versioning v0.31 } _, err = env.FrontendClient().UpdateWorkerDeploymentVersionMetadata(s.Context(), metadataReq) s.NoError(err) resp, err = s.describeVersion(env, tv1) s.NoError(err) // validating the metadata entries = resp.GetWorkerDeploymentVersionInfo().GetMetadata().GetEntries() s.Len(entries, 2) s.Equal(testRandomMetadataValue, entries["key1"].Data) s.Equal(testRandomMetadataValue, entries["key2"].Data) // LastModifierIdentity should match the identity provided in the metadata update s.Equal(metadataIdentity, resp.GetWorkerDeploymentVersionInfo().GetLastModifierIdentity()) } // testInvokeScaler returns the scaler that scaling groups backed by the test-invoke compute // provider must declare. func testInvokeScaler() *computepb.ComputeScaler { return &computepb.ComputeScaler{Type: "no-sync"} } func (s *DeploymentVersionSuite) createDeploymentAndVersion( env *testcore.TestEnv, tv *testvars.TestVars, identity string, computeConfig *computepb.ComputeConfig, ) { s.T().Helper() _, err := env.FrontendClient().CreateWorkerDeployment(s.Context(), &workflowservice.CreateWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: tv.DeploymentSeries(), RequestId: tv.Any().String(), }) s.NoError(err) _, err = env.FrontendClient().CreateWorkerDeploymentVersion(s.Context(), &workflowservice.CreateWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), DeploymentVersion: tv.ExternalDeploymentVersion(), Identity: identity, RequestId: tv.Any().String(), ComputeConfig: computeConfig, }) s.NoError(err) // Wait for version to be created. s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) descResp, err := s.describeVersion(env, tv) a.NoError(err) a.NotNil(descResp.GetWorkerDeploymentVersionInfo()) }, 10*time.Second, 500*time.Millisecond) } func (s *DeploymentVersionSuite) TestUpdateComputeConfig_Success() { env := s.newTestEnv() createIdentity := env.Tv().Any().String() validProvider := computeprovider.TestInvokeComputeProviderValidComputeProvider() s.createDeploymentAndVersion(env, env.Tv(), createIdentity, &computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg1": {Provider: validProvider, Scaler: testInvokeScaler()}, }, }) // Update compute config with a different identity and a new scaling group. updateIdentity := env.Tv().Any().String() _, err := env.FrontendClient().UpdateWorkerDeploymentVersionComputeConfig(s.Context(), &workflowservice.UpdateWorkerDeploymentVersionComputeConfigRequest{ Namespace: env.Namespace().String(), DeploymentVersion: env.Tv().ExternalDeploymentVersion(), ComputeConfigScalingGroups: map[string]*computepb.ComputeConfigScalingGroupUpdate{ "sg2": { ScalingGroup: &computepb.ComputeConfigScalingGroup{ TaskQueueTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_ACTIVITY}, Provider: validProvider, Scaler: testInvokeScaler(), }, }, }, Identity: updateIdentity, RequestId: env.Tv().Any().String(), }) s.NoError(err) // Verify both scaling groups exist and LastModifierIdentity is updated. s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) descResp, err := s.describeVersion(env, env.Tv()) a.NoError(err) info := descResp.GetWorkerDeploymentVersionInfo() a.Equal(updateIdentity, info.GetLastModifierIdentity()) a.True(proto.Equal(&computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg1": {Provider: validProvider, Scaler: testInvokeScaler()}, "sg2": { TaskQueueTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_ACTIVITY}, Provider: validProvider, Scaler: testInvokeScaler(), }, }, }, info.GetComputeConfig())) }, 10*time.Second, 500*time.Millisecond) // Verify the compute config summary is reflected in DescribeWorkerDeployment version summaries. s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) descDeployResp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: env.Tv().DeploymentSeries(), }) a.NoError(err) var versionSummary *deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary for _, vs := range descDeployResp.GetWorkerDeploymentInfo().GetVersionSummaries() { if vs.GetVersion() == env.Tv().DeploymentVersionString() { //nolint:staticcheck // SA1019: worker versioning v0.31 versionSummary = vs break } } a.NotNil(versionSummary, "version %s not found in DescribeWorkerDeployment", env.Tv().DeploymentVersionString()) a.True(proto.Equal(&computepb.ComputeConfigSummary{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroupSummary{ "sg1": { ProviderType: validProvider.GetType(), }, "sg2": { TaskQueueTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_ACTIVITY}, ProviderType: validProvider.GetType(), }, }, }, versionSummary.GetComputeConfig())) }, 10*time.Second, 500*time.Millisecond) // Verify the compute config summary is reflected in ListWorkerDeployments latest version summary. s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) listResp, err := env.FrontendClient().ListWorkerDeployments(s.Context(), &workflowservice.ListWorkerDeploymentsRequest{ Namespace: env.Namespace().String(), }) a.NoError(err) var found *workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary for _, d := range listResp.GetWorkerDeployments() { if d.GetName() == env.Tv().DeploymentSeries() { found = d break } } a.NotNil(found, "deployment %s not found in ListWorkerDeployments", env.Tv().DeploymentSeries()) a.True(proto.Equal(&computepb.ComputeConfigSummary{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroupSummary{ "sg1": { ProviderType: validProvider.GetType(), }, "sg2": { TaskQueueTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_ACTIVITY}, ProviderType: validProvider.GetType(), }, }, }, found.GetLatestVersionSummary().GetComputeConfig())) }, 60*time.Second, 500*time.Millisecond) } func (s *DeploymentVersionSuite) TestUpdateComputeConfig_UpdateExistingGroup() { env := s.newTestEnv() validProvider := computeprovider.TestInvokeComputeProviderValidComputeProvider() s.createDeploymentAndVersion(env, env.Tv(), env.Tv().Any().String(), &computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg1": { TaskQueueTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_WORKFLOW}, Provider: validProvider, Scaler: testInvokeScaler(), }, }, }) // Partially update sg1's task queue types via field mask. _, err := env.FrontendClient().UpdateWorkerDeploymentVersionComputeConfig(s.Context(), &workflowservice.UpdateWorkerDeploymentVersionComputeConfigRequest{ Namespace: env.Namespace().String(), DeploymentVersion: env.Tv().ExternalDeploymentVersion(), ComputeConfigScalingGroups: map[string]*computepb.ComputeConfigScalingGroupUpdate{ "sg1": { ScalingGroup: &computepb.ComputeConfigScalingGroup{ TaskQueueTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_ACTIVITY}, }, UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"task_queue_types"}}, }, }, Identity: env.Tv().Any().String(), RequestId: env.Tv().Any().String(), }) s.NoError(err) // Verify task queue types changed but provider is preserved. s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) descResp, err := s.describeVersion(env, env.Tv()) a.NoError(err) a.True(proto.Equal(&computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg1": { TaskQueueTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_ACTIVITY}, Provider: validProvider, Scaler: testInvokeScaler(), }, }, }, descResp.GetWorkerDeploymentVersionInfo().GetComputeConfig())) }, 10*time.Second, 500*time.Millisecond) } func (s *DeploymentVersionSuite) TestUpdateComputeConfig_RemoveScalingGroup() { env := s.newTestEnv() validProvider := computeprovider.TestInvokeComputeProviderValidComputeProvider() s.createDeploymentAndVersion(env, env.Tv(), env.Tv().Any().String(), &computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg1": {Provider: validProvider, Scaler: testInvokeScaler()}, "sg2": { TaskQueueTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_ACTIVITY}, Provider: validProvider, Scaler: testInvokeScaler(), }, }, }) // Remove sg1. _, err := env.FrontendClient().UpdateWorkerDeploymentVersionComputeConfig(s.Context(), &workflowservice.UpdateWorkerDeploymentVersionComputeConfigRequest{ Namespace: env.Namespace().String(), DeploymentVersion: env.Tv().ExternalDeploymentVersion(), RemoveComputeConfigScalingGroups: []string{"sg1"}, Identity: env.Tv().Any().String(), RequestId: env.Tv().Any().String(), }) s.NoError(err) // Verify only sg2 remains. s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) descResp, err := s.describeVersion(env, env.Tv()) a.NoError(err) a.True(proto.Equal(&computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg2": { TaskQueueTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_ACTIVITY}, Provider: validProvider, Scaler: testInvokeScaler(), }, }, }, descResp.GetWorkerDeploymentVersionInfo().GetComputeConfig())) }, 10*time.Second, 500*time.Millisecond) } func (s *DeploymentVersionSuite) TestUpdateComputeConfig_VersionNotFound() { env := s.newTestEnv() // Create deployment but no version. _, err := env.FrontendClient().CreateWorkerDeployment(s.Context(), &workflowservice.CreateWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: env.Tv().DeploymentSeries(), RequestId: env.Tv().Any().String(), }) s.NoError(err) _, err = env.FrontendClient().UpdateWorkerDeploymentVersionComputeConfig(s.Context(), &workflowservice.UpdateWorkerDeploymentVersionComputeConfigRequest{ Namespace: env.Namespace().String(), DeploymentVersion: env.Tv().ExternalDeploymentVersion(), ComputeConfigScalingGroups: map[string]*computepb.ComputeConfigScalingGroupUpdate{ "sg1": { ScalingGroup: &computepb.ComputeConfigScalingGroup{ Provider: computeprovider.TestInvokeComputeProviderValidComputeProvider(), }, }, }, Identity: env.Tv().Any().String(), RequestId: env.Tv().Any().String(), }) s.Error(err) var notFound *serviceerror.NotFound s.ErrorAs(err, ¬Found) } func (s *DeploymentVersionSuite) TestUpdateComputeConfig_InvalidProvider() { env := s.newTestEnv() s.createDeploymentAndVersion(env, env.Tv(), env.Tv().Any().String(), &computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg1": {Provider: computeprovider.TestInvokeComputeProviderValidComputeProvider(), Scaler: testInvokeScaler()}, }, }) _, err := env.FrontendClient().UpdateWorkerDeploymentVersionComputeConfig(s.Context(), &workflowservice.UpdateWorkerDeploymentVersionComputeConfigRequest{ Namespace: env.Namespace().String(), DeploymentVersion: env.Tv().ExternalDeploymentVersion(), ComputeConfigScalingGroups: map[string]*computepb.ComputeConfigScalingGroupUpdate{ "sg2": { ScalingGroup: &computepb.ComputeConfigScalingGroup{ TaskQueueTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_ACTIVITY}, Provider: &computepb.ComputeProvider{Type: "invalid-provider"}, }, }, }, Identity: env.Tv().Any().String(), RequestId: env.Tv().Any().String(), }) s.Error(err) var invalidArg *serviceerror.InvalidArgument s.ErrorAs(err, &invalidArg) s.Contains(invalidArg.Message, "invalid compute provider type") // Verify compute config summary is unchanged — the failed update should not have added sg2. s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) descDeployResp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: env.Tv().DeploymentSeries(), }) a.NoError(err) var versionSummary *deploymentpb.WorkerDeploymentInfo_WorkerDeploymentVersionSummary for _, vs := range descDeployResp.GetWorkerDeploymentInfo().GetVersionSummaries() { if vs.GetVersion() == env.Tv().DeploymentVersionString() { //nolint:staticcheck // SA1019: worker versioning v0.31 versionSummary = vs break } } a.NotNil(versionSummary, "version %s not found in deployment summaries", env.Tv().DeploymentVersionString()) a.True(proto.Equal(&computepb.ComputeConfigSummary{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroupSummary{ "sg1": { ProviderType: computeprovider.TestInvokeComputeProviderValidComputeProvider().GetType(), }, }, }, versionSummary.GetComputeConfig())) }, 10*time.Second, 500*time.Millisecond) } func (s *DeploymentVersionSuite) TestUpdateComputeConfig_DeletedVersion() { env := s.newTestEnv(testcore.WithDynamicConfig(dynamicconfig.PollerHistoryTTL, 500*time.Millisecond)) s.createDeploymentAndVersion(env, env.Tv(), env.Tv().Any().String(), nil) // Delete the version (skip drainage, no pollers since we created via CreateWorkerDeploymentVersion). s.tryDeleteVersion(env, env.Tv(), "", true) // Try to update compute config on the deleted version. _, err := env.FrontendClient().UpdateWorkerDeploymentVersionComputeConfig(s.Context(), &workflowservice.UpdateWorkerDeploymentVersionComputeConfigRequest{ Namespace: env.Namespace().String(), DeploymentVersion: env.Tv().ExternalDeploymentVersion(), ComputeConfigScalingGroups: map[string]*computepb.ComputeConfigScalingGroupUpdate{ "sg1": { ScalingGroup: &computepb.ComputeConfigScalingGroup{ Provider: computeprovider.TestInvokeComputeProviderValidComputeProvider(), }, }, }, Identity: env.Tv().Any().String(), RequestId: env.Tv().Any().String(), }) s.Error(err) } func (s *DeploymentVersionSuite) TestValidateComputeConfig_Valid() { env := s.newTestEnv() s.createDeploymentAndVersion(env, env.Tv(), env.Tv().Any().String(), nil) _, err := env.FrontendClient().ValidateWorkerDeploymentVersionComputeConfig(s.Context(), &workflowservice.ValidateWorkerDeploymentVersionComputeConfigRequest{ Namespace: env.Namespace().String(), DeploymentVersion: env.Tv().ExternalDeploymentVersion(), ComputeConfigScalingGroups: map[string]*computepb.ComputeConfigScalingGroupUpdate{ "sg1": { ScalingGroup: &computepb.ComputeConfigScalingGroup{ Provider: computeprovider.TestInvokeComputeProviderValidComputeProvider(), Scaler: testInvokeScaler(), }, }, }, Identity: env.Tv().Any().String(), }) s.NoError(err) // Verify the validation had no side effects — no compute config should exist. descResp, err := s.describeVersion(env, env.Tv()) s.NoError(err) s.Nil(descResp.GetWorkerDeploymentVersionInfo().GetComputeConfig()) } func (s *DeploymentVersionSuite) TestValidateComputeConfig_InvalidProvider() { env := s.newTestEnv() s.createDeploymentAndVersion(env, env.Tv(), env.Tv().Any().String(), nil) _, err := env.FrontendClient().ValidateWorkerDeploymentVersionComputeConfig(s.Context(), &workflowservice.ValidateWorkerDeploymentVersionComputeConfigRequest{ Namespace: env.Namespace().String(), DeploymentVersion: env.Tv().ExternalDeploymentVersion(), ComputeConfigScalingGroups: map[string]*computepb.ComputeConfigScalingGroupUpdate{ "sg1": { ScalingGroup: &computepb.ComputeConfigScalingGroup{ Provider: &computepb.ComputeProvider{Type: "invalid-provider"}, }, }, }, Identity: env.Tv().Any().String(), }) s.Error(err) var invalidArg *serviceerror.InvalidArgument s.ErrorAs(err, &invalidArg) s.Contains(invalidArg.Message, "invalid compute provider type") } func (s *DeploymentVersionSuite) TestValidateComputeConfig_VersionNotFound() { env := s.newTestEnv() // No deployment or version created — validate should still work since // the WCI validates independently. _, err := env.FrontendClient().CreateWorkerDeployment(s.Context(), &workflowservice.CreateWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: env.Tv().DeploymentSeries(), RequestId: env.Tv().Any().String(), }) s.NoError(err) _, err = env.FrontendClient().ValidateWorkerDeploymentVersionComputeConfig(s.Context(), &workflowservice.ValidateWorkerDeploymentVersionComputeConfigRequest{ Namespace: env.Namespace().String(), DeploymentVersion: env.Tv().ExternalDeploymentVersion(), ComputeConfigScalingGroups: map[string]*computepb.ComputeConfigScalingGroupUpdate{ "sg1": { ScalingGroup: &computepb.ComputeConfigScalingGroup{ Provider: computeprovider.TestInvokeComputeProviderValidComputeProvider(), Scaler: testInvokeScaler(), }, }, }, Identity: env.Tv().Any().String(), }) s.NoError(err) } func (s *DeploymentVersionSuite) checkVersionDrainageAndVersionStatus( env *testcore.TestEnv, tv *testvars.TestVars, expectedDrainageInfo *deploymentpb.VersionDrainageInfo, expectedStatus enumspb.WorkerDeploymentVersionStatus, waitFor time.Duration, ) (changedTime, checkedTime time.Time) { if waitFor > 0 { // wait for the requested duration before looking at the result ( +1 sec for system latency) time.Sleep(waitFor + 1*time.Second) //nolint:forbidigo } s.EventuallyWithT(func(t *assert.CollectT) { a := assert.New(t) resp, err := s.describeVersion(env, tv) a.NoError(err) dInfo := resp.GetWorkerDeploymentVersionInfo().GetDrainageInfo() a.Equal(expectedDrainageInfo.Status, dInfo.GetStatus()) if expectedDrainageInfo.LastCheckedTime != nil { a.Equal(expectedDrainageInfo.LastCheckedTime, dInfo.GetLastCheckedTime()) } if expectedDrainageInfo.LastChangedTime != nil { a.Equal(expectedDrainageInfo.LastChangedTime, dInfo.GetLastChangedTime()) } a.Equal(expectedStatus, resp.GetWorkerDeploymentVersionInfo().GetStatus()) changedTime = dInfo.GetLastChangedTime().AsTime() checkedTime = dInfo.GetLastCheckedTime().AsTime() }, 15*time.Second, time.Second) return changedTime, checkedTime } func (s *DeploymentVersionSuite) checkVersionStatusInDeployment( env *testcore.TestEnv, tv *testvars.TestVars, expectedStatus enumspb.WorkerDeploymentVersionStatus, ) { s.EventuallyWithT(func(t *assert.CollectT) { a := assert.New(t) resp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: tv.DeploymentSeries(), }) a.NoError(err) found := false for _, versionSummary := range resp.GetWorkerDeploymentInfo().GetVersionSummaries() { if versionSummary.GetVersion() == tv.DeploymentVersionString() { //nolint:staticcheck // SA1019: worker versioning v0.31 a.Equal(expectedStatus, versionSummary.GetStatus(), "DescribeWorkerDeployment should show version %s as %s", tv.DeploymentVersionString(), expectedStatus) found = true break } } a.True(found, "Version %s should be found in DescribeWorkerDeployment response", tv.DeploymentVersionString()) }, 10*time.Second, 500*time.Millisecond) } func (s *DeploymentVersionSuite) checkDescribeWorkflowAfterOverride( env *testcore.TestEnv, wf *commonpb.WorkflowExecution, expectedOverride *workflowpb.VersioningOverride, ) { s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) resp, err := env.FrontendClient().DescribeWorkflowExecution(s.Context(), &workflowservice.DescribeWorkflowExecutionRequest{ Namespace: env.Namespace().String(), Execution: wf, }) a.NoError(err) a.NotNil(resp) a.NotNil(resp.GetWorkflowExecutionInfo()) actualOverride := resp.GetWorkflowExecutionInfo().GetVersioningInfo().GetVersioningOverride() if useV32 { // v0.32 override a.Equal(expectedOverride.GetAutoUpgrade(), actualOverride.GetAutoUpgrade()) a.Equalf(expectedOverride.GetPinned().GetVersion().GetBuildId(), actualOverride.GetPinned().GetVersion().GetBuildId(), "expected pinned version build id %v, got %v", expectedOverride.GetPinned().GetVersion().GetBuildId(), actualOverride.GetPinned().GetVersion().GetBuildId()) a.Equalf(expectedOverride.GetPinned().GetVersion().GetDeploymentName(), actualOverride.GetPinned().GetVersion().GetDeploymentName(), "expected pinned version deployment name %v, got %v", expectedOverride.GetPinned().GetVersion().GetDeploymentName(), actualOverride.GetPinned().GetVersion().GetDeploymentName()) a.Equalf(expectedOverride.GetPinned().GetBehavior(), actualOverride.GetPinned().GetBehavior(), "expected pinned override behavior %v, got %v", expectedOverride.GetPinned().GetBehavior(), actualOverride.GetPinned().GetBehavior()) if worker_versioning.OverrideIsPinned(expectedOverride) { a.Equal(expectedOverride.GetPinned().GetVersion().GetDeploymentName(), resp.GetWorkflowExecutionInfo().GetWorkerDeploymentName()) } } else { // v0.31 override a.Equal(expectedOverride.GetBehavior().String(), actualOverride.GetBehavior().String()) //nolint:staticcheck // SA1019: worker versioning v0.31 if actualOverrideDeployment := actualOverride.GetPinnedVersion(); expectedOverride.GetPinnedVersion() != actualOverrideDeployment { //nolint:staticcheck // SA1019: worker versioning v0.31 a.Fail(fmt.Sprintf("pinned override mismatch. expected: {%s}, actual: {%s}", expectedOverride.GetPinnedVersion(), //nolint:staticcheck // SA1019: worker versioning v0.31 actualOverrideDeployment, )) } if worker_versioning.OverrideIsPinned(expectedOverride) { d, _ := worker_versioning.WorkerDeploymentVersionFromStringV31(expectedOverride.GetPinnedVersion()) //nolint:staticcheck // SA1019: worker versioning v0.31 a.Equal(d.GetDeploymentName(), resp.GetWorkflowExecutionInfo().GetWorkerDeploymentName()) } } }, 10*time.Second, 50*time.Millisecond) } func (s *DeploymentVersionSuite) checkWorkflowUpdateOptionsEventIdentity( ctx context.Context, env *testcore.TestEnv, wf *commonpb.WorkflowExecution, expectedIdentity string, ) { s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) resp, err := env.FrontendClient().GetWorkflowExecutionHistory(ctx, &workflowservice.GetWorkflowExecutionHistoryRequest{ Namespace: env.Namespace().String(), Execution: wf, }) a.NoError(err) a.NotNil(resp) events := resp.GetHistory().GetEvents() for resp.NextPageToken != nil { // probably there won't ever be more than one page of events in these tests resp, err = env.FrontendClient().GetWorkflowExecutionHistory(ctx, &workflowservice.GetWorkflowExecutionHistoryRequest{ Namespace: env.Namespace().String(), Execution: wf, NextPageToken: resp.NextPageToken, }) a.NoError(err) a.NotNil(resp) events = append(events, resp.GetHistory().GetEvents()...) } for _, event := range events { if event.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED { a.Equal(expectedIdentity, event.GetWorkflowExecutionOptionsUpdatedEventAttributes().GetIdentity()) } } }, 10*time.Second, 50*time.Millisecond) } func (s *DeploymentVersionSuite) checkVersionIsCurrent(ctx context.Context, env *testcore.TestEnv, tv *testvars.TestVars) { // Querying the Deployment Version s.EventuallyWithT(func(t *assert.CollectT) { a := assert.New(t) resp, err := s.describeVersion(env, tv) if !a.NoError(err) { return } a.Equal(tv.DeploymentVersionString(), resp.GetWorkerDeploymentVersionInfo().GetVersion()) //nolint:staticcheck // SA1019: worker versioning v0.31 a.Equal(tv.ExternalDeploymentVersion().GetDeploymentName(), resp.GetWorkerDeploymentVersionInfo().GetDeploymentVersion().GetDeploymentName()) a.Equal(tv.ExternalDeploymentVersion().GetBuildId(), resp.GetWorkerDeploymentVersionInfo().GetDeploymentVersion().GetBuildId()) a.NotNil(resp.GetWorkerDeploymentVersionInfo().GetCurrentSinceTime()) a.Equal(enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT, resp.GetWorkerDeploymentVersionInfo().GetStatus()) }, time.Second*10, time.Millisecond*1000) } func (s *DeploymentVersionSuite) checkVersionIsRamping(ctx context.Context, env *testcore.TestEnv, tv *testvars.TestVars) { // Querying the Deployment Version s.EventuallyWithT(func(t *assert.CollectT) { a := assert.New(t) resp, err := s.describeVersion(env, tv) if !a.NoError(err) { return } a.Equal(tv.DeploymentVersionString(), resp.GetWorkerDeploymentVersionInfo().GetVersion()) //nolint:staticcheck // SA1019: worker versioning v0.31 a.Equal(tv.ExternalDeploymentVersion().GetDeploymentName(), resp.GetWorkerDeploymentVersionInfo().GetDeploymentVersion().GetDeploymentName()) a.Equal(tv.ExternalDeploymentVersion().GetBuildId(), resp.GetWorkerDeploymentVersionInfo().GetDeploymentVersion().GetBuildId()) a.NotNil(resp.GetWorkerDeploymentVersionInfo().GetRampingSinceTime()) a.Equal(enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_RAMPING, resp.GetWorkerDeploymentVersionInfo().GetStatus()) }, time.Second*10, time.Millisecond*1000) } func (s *DeploymentVersionSuite) setCurrent(env *testcore.TestEnv, tv *testvars.TestVars, ignoreMissingTQs bool) error { ctx, cancel := context.WithTimeout(s.Context(), 10*time.Second) defer cancel() req := &workflowservice.SetWorkerDeploymentCurrentVersionRequest{ Namespace: env.Namespace().String(), DeploymentName: tv.DeploymentSeries(), IgnoreMissingTaskQueues: ignoreMissingTQs, Identity: tv.ClientIdentity(), } if useV32 { req.BuildId = tv.BuildID() } else { req.Version = tv.DeploymentVersionString() //nolint:staticcheck // SA1019: worker versioning v0.31 } _, err := env.FrontendClient().SetWorkerDeploymentCurrentVersion(ctx, req) if err == nil { s.checkVersionIsCurrent(ctx, env, tv) } return err } func (s *DeploymentVersionSuite) setRamping( env *testcore.TestEnv, tv *testvars.TestVars, percentage float32, ) error { ctx, cancel := context.WithTimeout(s.Context(), 20*time.Second) defer cancel() v := tv.DeploymentVersionString() bid := tv.BuildID() req := &workflowservice.SetWorkerDeploymentRampingVersionRequest{ Namespace: env.Namespace().String(), DeploymentName: tv.DeploymentSeries(), Percentage: percentage, Identity: tv.ClientIdentity(), } if useV32 { req.BuildId = bid } else { req.Version = v //nolint:staticcheck // SA1019: worker versioning v0.31 } _, err := env.FrontendClient().SetWorkerDeploymentRampingVersion(ctx, req) if err == nil { s.checkVersionIsRamping(ctx, env, tv) } return err } func (s *DeploymentVersionSuite) startWorkflow( env *testcore.TestEnv, tv *testvars.TestVars, override *workflowpb.VersioningOverride, ) string { request := &workflowservice.StartWorkflowExecutionRequest{ RequestId: tv.Any().String(), Namespace: env.Namespace().String(), WorkflowId: tv.WorkflowID(), WorkflowType: tv.WorkflowType(), TaskQueue: tv.TaskQueue(), Identity: tv.WorkerIdentity(), VersioningOverride: override, } we, err0 := env.FrontendClient().StartWorkflowExecution(s.Context(), request) s.NoError(err0) return we.GetRunId() } func (s *DeploymentVersionSuite) tryDeleteVersion( env *testcore.TestEnv, tv *testvars.TestVars, expectedError string, skipDrainage bool, ) { req := &workflowservice.DeleteWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), SkipDrainage: skipDrainage, } if useV32 { req.DeploymentVersion = tv.ExternalDeploymentVersion() } else { req.Version = tv.DeploymentVersionString() //nolint:staticcheck // SA1019: worker versioning v0.31 } _, err := env.FrontendClient().DeleteWorkerDeploymentVersion(s.Context(), req) if expectedError == "" { s.NoError(err) } else { s.EqualErrorf(err, expectedError, err.Error()) } } func (s *DeploymentVersionSuite) setAndCheckOverride(env *testcore.TestEnv, tv *testvars.TestVars, override *workflowpb.VersioningOverride) { s.setAndCheckOverrideWithExpectedOutput(env, tv, override, override) } func (s *DeploymentVersionSuite) setAndCheckOverrideWithExpectedOutput(env *testcore.TestEnv, tv *testvars.TestVars, inputOverride, expectedOutputOverride *workflowpb.VersioningOverride) { optsIn := &workflowpb.WorkflowExecutionOptions{VersioningOverride: inputOverride} optsOut := &workflowpb.WorkflowExecutionOptions{VersioningOverride: expectedOutputOverride} // Set input override --> describe workflow shows the expected output override updateResp, err := env.FrontendClient().UpdateWorkflowExecutionOptions(s.Context(), &workflowservice.UpdateWorkflowExecutionOptionsRequest{ Namespace: env.Namespace().String(), WorkflowExecution: tv.WorkflowExecution(), WorkflowExecutionOptions: optsIn, UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"versioning_override"}}, Identity: tv.ClientIdentity(), }) s.NoError(err) s.True(proto.Equal(updateResp.GetWorkflowExecutionOptions(), optsOut)) s.checkDescribeWorkflowAfterOverride(env, tv.WorkflowExecution(), expectedOutputOverride) s.checkWorkflowUpdateOptionsEventIdentity(s.Context(), env, tv.WorkflowExecution(), tv.ClientIdentity()) } // The following tests test the VersioningOverride functionality when passed via the UpdateWorkflowExecutionOptions API. func (s *DeploymentVersionSuite) TestUpdateWorkflowExecutionOptions_SetPinned_CacheMissAndHits() { env := s.newTestEnv( // TODO: remove WithWorkerService once legacy suite-scoped cluster behavior is removed. testcore.WithWorkerService("worker-deployment version membership cache test"), testcore.WithDynamicConfig(dynamicconfig.VersionMembershipCacheTTL, 5*time.Second), ) // start an unversioned workflow s.startWorkflow(env, env.Tv(), nil) opts := &workflowpb.WorkflowExecutionOptions{VersioningOverride: s.makePinnedOverride(env.Tv())} // Setting a pinned override should fail since the version does not exist resp, err := env.FrontendClient().UpdateWorkflowExecutionOptions(s.Context(), &workflowservice.UpdateWorkflowExecutionOptionsRequest{ Namespace: env.Namespace().String(), WorkflowExecution: env.Tv().WorkflowExecution(), WorkflowExecutionOptions: opts, UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"versioning_override"}}, Identity: env.Tv().ClientIdentity(), }) s.Error(err) s.Nil(resp) // Start a versioned poller which shall create a version; however, the cache TTL is not expired yet. This would result in a cache hit which would return // a stale value for the version presence in the task queue. s.startVersionWorkflow(s.Context(), env, env.Tv()) // Setting a pinned override should fail since the stale cache entry is returned. resp, err = env.FrontendClient().UpdateWorkflowExecutionOptions(s.Context(), &workflowservice.UpdateWorkflowExecutionOptionsRequest{ Namespace: env.Namespace().String(), WorkflowExecution: env.Tv().WorkflowExecution(), WorkflowExecutionOptions: opts, UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"versioning_override"}}, Identity: env.Tv().ClientIdentity(), }) s.Error(err) s.Nil(resp) // Wait for the cache TTL to expire s.Eventually(func() bool { _, err := env.FrontendClient().UpdateWorkflowExecutionOptions(s.Context(), &workflowservice.UpdateWorkflowExecutionOptionsRequest{ Namespace: env.Namespace().String(), WorkflowExecution: env.Tv().WorkflowExecution(), WorkflowExecutionOptions: opts, UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"versioning_override"}}, Identity: env.Tv().ClientIdentity(), }) return err == nil }, 10*time.Second, 500*time.Millisecond) // The Pinned Override should have now succeeded with no error. Verify that the // the workflow shows the override. s.checkDescribeWorkflowAfterOverride(env, env.Tv().WorkflowExecution(), opts.VersioningOverride) } func (s *DeploymentVersionSuite) TestUpdateWorkflowExecutionOptions_SetUnpinnedThenUnset() { env := s.newTestEnv() // start an unversioned workflow s.startWorkflow(env, env.Tv(), nil) // 1. Set unpinned override --> describe workflow shows the override s.setAndCheckOverride(env, env.Tv(), s.makeAutoUpgradeOverride()) // 2. Unset using empty update opts with mutation mask --> describe workflow shows no more override s.setAndCheckOverride(env, env.Tv(), nil) } func (s *DeploymentVersionSuite) TestUpdateWorkflowExecutionOptions_SetPinnedThenUnset() { env := s.newTestEnv() // Start a versioned poller which shall create a version; the version must be present before it can be set as an override. s.startVersionWorkflow(s.Context(), env, env.Tv()) // start an unversioned workflow s.startWorkflow(env, env.Tv(), nil) // 1. Set pinned override on our new unversioned workflow --> describe workflow shows the override s.setAndCheckOverride(env, env.Tv(), s.makePinnedOverride(env.Tv())) // 2. Unset using empty update opts with mutation mask --> describe workflow shows no more override s.setAndCheckOverride(env, env.Tv(), nil) } func (s *DeploymentVersionSuite) TestUpdateWorkflowExecutionOptions_EmptyFields() { env := s.newTestEnv() // Start a versioned poller which shall create a version; the version must be present before it can be set as an override. s.startVersionWorkflow(s.Context(), env, env.Tv()) // start an unversioned workflow s.startWorkflow(env, env.Tv(), nil) // 1. Pinned update with empty mask --> describe workflow shows no change updateResp, err := env.FrontendClient().UpdateWorkflowExecutionOptions(s.Context(), &workflowservice.UpdateWorkflowExecutionOptionsRequest{ Namespace: env.Namespace().String(), WorkflowExecution: env.Tv().WorkflowExecution(), WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{ VersioningOverride: s.makePinnedOverride(env.Tv()), }, UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{}}, }) s.NoError(err) s.True(proto.Equal(updateResp.GetWorkflowExecutionOptions(), &workflowpb.WorkflowExecutionOptions{})) s.checkDescribeWorkflowAfterOverride(env, env.Tv().WorkflowExecution(), nil) } func (s *DeploymentVersionSuite) TestUpdateWorkflowExecutionOptions_SetPinnedSetPinned() { env := s.newTestEnv() tv1 := env.Tv().WithBuildIDNumber(1) tv2 := env.Tv().WithBuildIDNumber(2) // Start a versioned poller which shall create the two versions; the versions must be present before they can be set as overrides. s.startVersionWorkflow(s.Context(), env, tv1) s.startVersionWorkflow(s.Context(), env, tv2) // start an unversioned workflow s.startWorkflow(env, env.Tv(), nil) // 1. Set pinned override 1 --> describe workflow shows the override s.setAndCheckOverride(env, env.Tv(), s.makePinnedOverride(tv1)) // 3. Set pinned override 2 --> describe workflow shows the override s.setAndCheckOverride(env, env.Tv(), s.makePinnedOverride(tv2)) } func (s *DeploymentVersionSuite) TestUpdateWorkflowExecutionOptions_SetImpliedPinnedSuccess() { env := s.newTestEnv() if !useV32 { s.T().Skip("Implied pinned overrides are only supported in v3.2+") } tv1 := env.Tv().WithBuildIDNumber(1) // Start a versioned poller which shall create the two versions; the versions must be present before they can be set as overrides. s.startVersionWorkflow(s.Context(), env, tv1) // Set tv1 to current, so that the test workflow will be naturally pinned to tv1. err := s.setCurrent(env, tv1, true) s.NoError(err) // Start a workflow pinned to tv1. run := s.startPinnedWorkflow(s.Context(), env, tv1) noVersionPinnedOverride := &workflowpb.VersioningOverride{Override: &workflowpb.VersioningOverride_Pinned{ Pinned: &workflowpb.VersioningOverride_PinnedOverride{ Behavior: workflowpb.VersioningOverride_PINNED_OVERRIDE_BEHAVIOR_PINNED, }, }} yesVersionPinnedOverride := &workflowpb.VersioningOverride{Override: &workflowpb.VersioningOverride_Pinned{ Pinned: &workflowpb.VersioningOverride_PinnedOverride{ Behavior: workflowpb.VersioningOverride_PINNED_OVERRIDE_BEHAVIOR_PINNED, Version: tv1.ExternalDeploymentVersion(), }, }} // 1. Set pinned override without a version --> describe workflow shows the override with pinned override version set to tv1. s.setAndCheckOverrideWithExpectedOutput(env, tv1.WithWorkflowID(run.GetID()), noVersionPinnedOverride, yesVersionPinnedOverride) } func (s *DeploymentVersionSuite) TestUpdateWorkflowExecutionOptions_SetImpliedPinnedError() { env := s.newTestEnv() if !useV32 { s.T().Skip("Implied pinned overrides are only supported in v3.2+") } tv1 := env.Tv().WithBuildIDNumber(1) // Start a versioned poller which shall create the two versions; the versions must be present before they can be set as overrides. s.startVersionWorkflow(s.Context(), env, tv1) // Set tv1 to current, so that the test workflow will run on v1. err := s.setCurrent(env, tv1, true) s.NoError(err) // Start an auto-upgrade workflow. run := s.startUnpinnedWorkflow(s.Context(), env, tv1) noVersionPinnedOverride := &workflowpb.VersioningOverride{Override: &workflowpb.VersioningOverride_Pinned{ Pinned: &workflowpb.VersioningOverride_PinnedOverride{ Behavior: workflowpb.VersioningOverride_PINNED_OVERRIDE_BEHAVIOR_PINNED, }, }} // 1. Set pinned override without a version --> errors because workflow is not already pinned to a version optsIn := &workflowpb.WorkflowExecutionOptions{VersioningOverride: noVersionPinnedOverride} // Set input override --> describe workflow shows the expected output override _, err = env.FrontendClient().UpdateWorkflowExecutionOptions(s.Context(), &workflowservice.UpdateWorkflowExecutionOptionsRequest{ Namespace: env.Namespace().String(), WorkflowExecution: tv1.WithWorkflowID(run.GetID()).WorkflowExecution(), WorkflowExecutionOptions: optsIn, UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"versioning_override"}}, Identity: tv1.ClientIdentity(), }) s.Error(err) s.Contains(err.Error(), fmt.Sprintf("must specify a specific pinned override version because workflow with id %v has behavior %s and is not yet pinned to any version", run.GetID(), enumspb.VERSIONING_BEHAVIOR_AUTO_UPGRADE.String()), ) } func (s *DeploymentVersionSuite) TestUpdateWorkflowExecutionOptions_SetUnpinnedSetUnpinned() { env := s.newTestEnv() // start an unversioned workflow s.startWorkflow(env, env.Tv(), nil) // 1. Set unpinned override --> describe workflow shows the override s.setAndCheckOverride(env, env.Tv(), s.makeAutoUpgradeOverride()) // 2. Set unpinned override --> describe workflow shows the override s.setAndCheckOverride(env, env.Tv(), s.makeAutoUpgradeOverride()) } func (s *DeploymentVersionSuite) TestUpdateWorkflowExecutionOptions_SetUnpinnedSetPinned() { env := s.newTestEnv() // Start a versioned poller which shall create a version; the version must be present before it can be set as an override. s.startVersionWorkflow(s.Context(), env, env.Tv()) // start an unversioned workflow s.startWorkflow(env, env.Tv(), nil) // 1. Set unpinned override --> describe workflow shows the override s.setAndCheckOverride(env, env.Tv(), s.makeAutoUpgradeOverride()) // 2. Set pinned override 1 --> describe workflow shows the override s.setAndCheckOverride(env, env.Tv(), s.makePinnedOverride(env.Tv())) } func (s *DeploymentVersionSuite) TestUpdateWorkflowExecutionOptions_SetPinnedSetUnpinned() { env := s.newTestEnv() // Start a versioned poller which shall create a version; the version must be present before it can be set as an override. s.startVersionWorkflow(s.Context(), env, env.Tv()) // start an unversioned workflow s.startWorkflow(env, env.Tv(), nil) // 1. Set pinned override 1 --> describe workflow shows the override s.setAndCheckOverride(env, env.Tv(), s.makePinnedOverride(env.Tv())) // 2. Set unpinned override --> describe workflow shows the override s.setAndCheckOverride(env, env.Tv(), s.makeAutoUpgradeOverride()) } func (s *DeploymentVersionSuite) TestUpdateWorkflowExecutionOptions_ReactivateVersionOnPinned() { env := s.newTestEnv() tv1 := env.Tv().WithBuildIDNumber(1) // Pinned target (INACTIVE) tv2 := env.Tv().WithBuildIDNumber(2) // Current version // v1 starts INACTIVE (never set as current). The reactivation signal handler in // version_workflow.go treats DRAINED and INACTIVE identically — both flip to DRAINING. s.startVersionWorkflow(s.Context(), env, tv1) // v2 becomes the current version so the initial (non-pinned) workflow has a target. s.startVersionWorkflow(s.Context(), env, tv2) err := s.setCurrent(env, tv2, true) s.NoError(err) wf := func(version string) func(ctx workflow.Context) (string, error) { return func(ctx workflow.Context) (string, error) { workflow.GetSignalChannel(ctx, "complete").Receive(ctx, nil) return "done from " + version, nil } } // Register a worker for version 1 (INACTIVE) so it can accept workflows when // UpdateWorkflowExecutionOptions is called to pin the workflow to version 1. w1 := worker.New(env.SdkClient(), tv1.TaskQueue().String(), worker.Options{ DeploymentOptions: worker.DeploymentOptions{ Version: tv1.SDKDeploymentVersion(), UseVersioning: true, }, }) w1.RegisterWorkflowWithOptions(wf("v1"), workflow.RegisterOptions{ Name: "waitingWorkflow", VersioningBehavior: workflow.VersioningBehaviorAutoUpgrade, }) s.NoError(w1.Start()) defer w1.Stop() // Register and start worker for version 2 on THE SAME task queue as version 1 w2 := worker.New(env.SdkClient(), tv1.TaskQueue().String(), worker.Options{ DeploymentOptions: worker.DeploymentOptions{ Version: tv2.SDKDeploymentVersion(), UseVersioning: true, }, }) w2.RegisterWorkflowWithOptions(wf("v2"), workflow.RegisterOptions{ Name: "waitingWorkflow", VersioningBehavior: workflow.VersioningBehaviorAutoUpgrade, }) s.NoError(w2.Start()) defer w2.Stop() s.waitForPollers(env, tv1, tv2) // Start the workflow. The workflow shall start on version 2, by default, since it is the current version. wfTV := env.Tv() run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{ TaskQueue: tv1.TaskQueue().String(), ID: wfTV.WorkflowID(), }, "waitingWorkflow") s.NoError(err) // Pin the running workflow to version 1 (INACTIVE) using UpdateWorkflowExecutionOptions. pinnedOverride := &workflowpb.VersioningOverride{ Override: &workflowpb.VersioningOverride_Pinned{ Pinned: &workflowpb.VersioningOverride_PinnedOverride{ Behavior: workflowpb.VersioningOverride_PINNED_OVERRIDE_BEHAVIOR_PINNED, Version: tv1.ExternalDeploymentVersion(), }, }, } // Pin the workflow to version 1 (both versions are on the same task queue). // Use Eventually to bypass version membership cache checks. s.Eventually(func() bool { _, err = env.FrontendClient().UpdateWorkflowExecutionOptions(s.Context(), &workflowservice.UpdateWorkflowExecutionOptionsRequest{ Namespace: env.Namespace().String(), WorkflowExecution: &commonpb.WorkflowExecution{ WorkflowId: wfTV.WorkflowID(), RunId: run.GetRunID(), }, WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{ VersioningOverride: pinnedOverride, }, UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"versioning_override"}}, }) return err == nil }, 10*time.Second, 500*time.Millisecond) // Verify workflow has the pinned override s.checkDescribeWorkflowAfterOverride(env, &commonpb.WorkflowExecution{ WorkflowId: wfTV.WorkflowID(), RunId: run.GetRunID(), }, pinnedOverride) // Wait for version 1 to show up as DRAINING s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{ Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING, }, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING, 0) // Verify via DescribeWorkerDeployment that the version status is updated s.checkVersionStatusInDeployment(env, tv1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING) // Signal workflow to complete s.NoError(env.SdkClient().SignalWorkflow(s.Context(), wfTV.WorkflowID(), run.GetRunID(), "complete", nil)) // Wait for workflow to complete and verify it ran on version 1 var result string s.NoError(run.Get(s.Context(), &result)) s.Equal("done from v1", result, "Workflow should have completed on version 1") } func (s *DeploymentVersionSuite) TestStartWorkflowExecution_ReactivateVersionOnPinned() { env := s.newTestEnv() tv1 := env.Tv() // v1 starts INACTIVE (never set as current). The reactivation signal handler in // version_workflow.go treats DRAINED and INACTIVE identically — both flip to DRAINING. s.startVersionWorkflow(s.Context(), env, tv1) wf := func(version string) func(ctx workflow.Context) (string, error) { return func(ctx workflow.Context) (string, error) { workflow.GetSignalChannel(ctx, "complete").Receive(ctx, nil) return "done from " + version, nil } } // Register a worker for version 1 (INACTIVE) so it can accept workflows when // StartWorkflowExecution is called with a pinned override to version 1. w1 := worker.New(env.SdkClient(), tv1.TaskQueue().String(), worker.Options{ DeploymentOptions: worker.DeploymentOptions{ Version: tv1.SDKDeploymentVersion(), UseVersioning: true, }, }) w1.RegisterWorkflowWithOptions(wf("v1"), workflow.RegisterOptions{ Name: "waitingWorkflow", VersioningBehavior: workflow.VersioningBehaviorAutoUpgrade, }) s.NoError(w1.Start()) defer w1.Stop() // Start a new workflow with the pinned override pointing to version 1 (INACTIVE). wfTV := env.Tv() var run sdkclient.WorkflowRun s.Eventually(func() bool { var startErr error run, startErr = env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{ TaskQueue: tv1.TaskQueue().String(), ID: wfTV.WorkflowID(), VersioningOverride: &sdkclient.PinnedVersioningOverride{ Version: tv1.SDKDeploymentVersion(), }, }, "waitingWorkflow") return startErr == nil }, 10*time.Second, 500*time.Millisecond) // Verify workflow has the pinned override s.checkDescribeWorkflowAfterOverride(env, &commonpb.WorkflowExecution{ WorkflowId: wfTV.WorkflowID(), RunId: run.GetRunID(), }, &workflowpb.VersioningOverride{ Override: &workflowpb.VersioningOverride_Pinned{ Pinned: &workflowpb.VersioningOverride_PinnedOverride{ Behavior: workflowpb.VersioningOverride_PINNED_OVERRIDE_BEHAVIOR_PINNED, Version: tv1.ExternalDeploymentVersion(), }, }, }) // Wait for version 1 to show up as DRAINING (reactivated from INACTIVE) s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{ Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING, }, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING, 0) // Verify via DescribeWorkerDeployment that the version status is updated s.checkVersionStatusInDeployment(env, tv1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING) // Signal workflow to complete s.NoError(env.SdkClient().SignalWorkflow(s.Context(), wfTV.WorkflowID(), run.GetRunID(), "complete", nil)) // Wait for workflow to complete and verify it ran on version 1 var result string s.NoError(run.Get(s.Context(), &result)) s.Equal("done from v1", result, "Workflow should have completed on version 1") } func (s *DeploymentVersionSuite) TestStartWorkflowExecution_ReactivateVersionOnPinned_WithConflictPolicy() { env := s.newTestEnv() tv1 := env.Tv().WithBuildIDNumber(1) tv2 := env.Tv().WithBuildIDNumber(2) // v1 starts INACTIVE (never set as current). The reactivation signal handler in // version_workflow.go treats DRAINED and INACTIVE identically — both flip to DRAINING. s.startVersionWorkflow(s.Context(), env, tv1) // v2 becomes the current version so the initial (non-pinned) workflow has a target. s.startVersionWorkflow(s.Context(), env, tv2) err := s.setCurrent(env, tv2, true) s.NoError(err) wf := func(version string) func(ctx workflow.Context) (string, error) { return func(ctx workflow.Context) (string, error) { workflow.GetSignalChannel(ctx, "complete").Receive(ctx, nil) return "done from " + version, nil } } // Register a worker for version 1 (INACTIVE) so it can accept workflows w1 := worker.New(env.SdkClient(), tv1.TaskQueue().String(), worker.Options{ DeploymentOptions: worker.DeploymentOptions{ Version: tv1.SDKDeploymentVersion(), UseVersioning: true, }, }) w1.RegisterWorkflowWithOptions(wf("v1"), workflow.RegisterOptions{ Name: "waitingWorkflow", VersioningBehavior: workflow.VersioningBehaviorAutoUpgrade, }) s.NoError(w1.Start()) defer w1.Stop() // Start a first workflow (no pinning, uses current version v2) to create a running execution // with a specific workflow ID that we will terminate via conflict policy. wfTV := env.Tv() w2 := worker.New(env.SdkClient(), tv2.TaskQueue().String(), worker.Options{ DeploymentOptions: worker.DeploymentOptions{ Version: tv2.SDKDeploymentVersion(), UseVersioning: true, }, }) w2.RegisterWorkflowWithOptions(wf("v2"), workflow.RegisterOptions{ Name: "waitingWorkflow", VersioningBehavior: workflow.VersioningBehaviorAutoUpgrade, }) s.NoError(w2.Start()) defer w2.Stop() s.Eventually(func() bool { _, startErr := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{ TaskQueue: tv2.TaskQueue().String(), ID: wfTV.WorkflowID(), }, "waitingWorkflow") return startErr == nil }, 10*time.Second, 500*time.Millisecond) // Now start a second workflow with the SAME workflow ID, pinned to v1 (INACTIVE), // using TERMINATE_EXISTING conflict policy. This goes through the handleConflict method in api.go. var run sdkclient.WorkflowRun s.Eventually(func() bool { var startErr error run, startErr = env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{ TaskQueue: tv1.TaskQueue().String(), ID: wfTV.WorkflowID(), VersioningOverride: &sdkclient.PinnedVersioningOverride{ Version: tv1.SDKDeploymentVersion(), }, WorkflowIDConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_TERMINATE_EXISTING, }, "waitingWorkflow") return startErr == nil }, 10*time.Second, 500*time.Millisecond) // Verify workflow has the pinned override s.checkDescribeWorkflowAfterOverride(env, &commonpb.WorkflowExecution{ WorkflowId: wfTV.WorkflowID(), RunId: run.GetRunID(), }, &workflowpb.VersioningOverride{ Override: &workflowpb.VersioningOverride_Pinned{ Pinned: &workflowpb.VersioningOverride_PinnedOverride{ Behavior: workflowpb.VersioningOverride_PINNED_OVERRIDE_BEHAVIOR_PINNED, Version: tv1.ExternalDeploymentVersion(), }, }, }) // Wait for version 1 to show up as DRAINING (reactivated from INACTIVE) s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{ Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING, }, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING, 0) // Verify via DescribeWorkerDeployment that the version status is updated s.checkVersionStatusInDeployment(env, tv1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING) // Signal workflow to complete s.NoError(env.SdkClient().SignalWorkflow(s.Context(), wfTV.WorkflowID(), run.GetRunID(), "complete", nil)) // Wait for workflow to complete and verify it ran on version 1 var result string s.NoError(run.Get(s.Context(), &result)) s.Equal("done from v1", result, "Workflow should have completed on version 1") } func (s *DeploymentVersionSuite) TestSignalWithStartWorkflowExecution_ReactivateVersionOnPinned() { env := s.newTestEnv() tv1 := env.Tv() // v1 starts INACTIVE (never set as current). The reactivation signal handler in // version_workflow.go treats DRAINED and INACTIVE identically — both flip to DRAINING. s.startVersionWorkflow(s.Context(), env, tv1) wf := func(version string) func(ctx workflow.Context) (string, error) { return func(ctx workflow.Context) (string, error) { workflow.GetSignalChannel(ctx, "complete").Receive(ctx, nil) return "done from " + version, nil } } // Register a worker for version 1 (INACTIVE) so it can accept workflows when // SignalWithStartWorkflowExecution is called with a pinned override to version 1. w1 := worker.New(env.SdkClient(), tv1.TaskQueue().String(), worker.Options{ DeploymentOptions: worker.DeploymentOptions{ Version: tv1.SDKDeploymentVersion(), UseVersioning: true, }, }) w1.RegisterWorkflowWithOptions(wf("v1"), workflow.RegisterOptions{ Name: "waitingWorkflow", VersioningBehavior: workflow.VersioningBehaviorAutoUpgrade, }) s.NoError(w1.Start()) defer w1.Stop() // Use SignalWithStart with the pinned override pointing to version 1 (INACTIVE). // This should START a new workflow (not signal an existing one) since no workflow exists yet. wfTV := env.Tv() var run sdkclient.WorkflowRun s.Eventually(func() bool { var startErr error run, startErr = env.SdkClient().SignalWithStartWorkflow(s.Context(), wfTV.WorkflowID(), "start-signal", // signal name nil, // signal arg sdkclient.StartWorkflowOptions{ TaskQueue: tv1.TaskQueue().String(), VersioningOverride: &sdkclient.PinnedVersioningOverride{ Version: tv1.SDKDeploymentVersion(), }, }, "waitingWorkflow", ) return startErr == nil }, 10*time.Second, 500*time.Millisecond) // Verify workflow has the pinned override s.checkDescribeWorkflowAfterOverride(env, &commonpb.WorkflowExecution{ WorkflowId: wfTV.WorkflowID(), RunId: run.GetRunID(), }, &workflowpb.VersioningOverride{ Override: &workflowpb.VersioningOverride_Pinned{ Pinned: &workflowpb.VersioningOverride_PinnedOverride{ Behavior: workflowpb.VersioningOverride_PINNED_OVERRIDE_BEHAVIOR_PINNED, Version: tv1.ExternalDeploymentVersion(), }, }, }) // Wait for version 1 to show up as DRAINING (reactivated from INACTIVE) s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{ Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING, }, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING, 0) // Verify via DescribeWorkerDeployment that the version status is updated s.checkVersionStatusInDeployment(env, tv1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING) // Signal workflow to complete s.NoError(env.SdkClient().SignalWorkflow(s.Context(), wfTV.WorkflowID(), run.GetRunID(), "complete", nil)) // Wait for workflow to complete and verify it ran on version 1 var result string s.NoError(run.Get(s.Context(), &result)) s.Equal("done from v1", result, "Workflow should have completed on version 1") } func (s *DeploymentVersionSuite) TestResetWorkflowExecution_ReactivateVersionOnPinned() { env := s.newTestEnv() tv1 := env.Tv().WithBuildIDNumber(1) // Pinned target (INACTIVE) tv2 := env.Tv().WithBuildIDNumber(2) // Current version // v1 starts INACTIVE (never set as current). The reactivation signal handler in // version_workflow.go treats DRAINED and INACTIVE identically — both flip to DRAINING. s.startVersionWorkflow(s.Context(), env, tv1) // v2 becomes the current version so the initial (non-pinned) workflow has a target. s.startVersionWorkflow(s.Context(), env, tv2) err := s.setCurrent(env, tv2, true) s.NoError(err) // Workflow that waits for a signal, used for both versions. // Returns a string indicating which version completed it. wf := func(version string) func(ctx workflow.Context) (string, error) { return func(ctx workflow.Context) (string, error) { workflow.GetSignalChannel(ctx, "complete").Receive(ctx, nil) return "done from " + version, nil } } // Register a worker for version 1 (INACTIVE) so it can accept workflows when // ResetWorkflowExecution is called with a pinned override to version 1. w1 := worker.New(env.SdkClient(), tv1.TaskQueue().String(), worker.Options{ DeploymentOptions: worker.DeploymentOptions{ Version: tv1.SDKDeploymentVersion(), UseVersioning: true, }, }) w1.RegisterWorkflowWithOptions(wf("v1"), workflow.RegisterOptions{ Name: "waitingWorkflow", VersioningBehavior: workflow.VersioningBehaviorAutoUpgrade, }) s.NoError(w1.Start()) defer w1.Stop() // Register and start worker for version 2 on THE SAME task queue as version 1 w2 := worker.New(env.SdkClient(), tv1.TaskQueue().String(), worker.Options{ DeploymentOptions: worker.DeploymentOptions{ Version: tv2.SDKDeploymentVersion(), UseVersioning: true, }, }) w2.RegisterWorkflowWithOptions(wf("v2"), workflow.RegisterOptions{ Name: "waitingWorkflow", VersioningBehavior: workflow.VersioningBehaviorAutoUpgrade, }) s.NoError(w2.Start()) defer w2.Stop() s.waitForPollers(env, tv1, tv2) // Start a workflow on the current version (v2) wfTV := env.Tv() run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{ TaskQueue: tv1.TaskQueue().String(), ID: wfTV.WorkflowID(), }, "waitingWorkflow") s.NoError(err) // Wait for the workflow to start and complete its first workflow task (creates a reset point) s.Eventually(func() bool { hist := env.SdkClient().GetWorkflowHistory(s.Context(), wfTV.WorkflowID(), run.GetRunID(), false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT) for hist.HasNext() { event, err := hist.Next() if err != nil { return false } if event.EventType == enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED { return true } } return false }, 10*time.Second, 200*time.Millisecond, "Workflow should have completed its first workflow task") // Find the first workflow task complete event ID for the reset point var resetEventID int64 hist := env.SdkClient().GetWorkflowHistory(s.Context(), wfTV.WorkflowID(), run.GetRunID(), false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT) for hist.HasNext() { event, err := hist.Next() s.NoError(err) if event.EventType == enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED { resetEventID = event.EventId break } } s.Positive(resetEventID, "Should have found a workflow task complete event") // Reset the workflow with PostResetOperations containing a versioning override pinned to v1 (which is currently INACTIVE) var resetResp *workflowservice.ResetWorkflowExecutionResponse s.Eventually(func() bool { var resetErr error resetResp, resetErr = env.FrontendClient().ResetWorkflowExecution(s.Context(), &workflowservice.ResetWorkflowExecutionRequest{ Namespace: env.Namespace().String(), WorkflowExecution: &commonpb.WorkflowExecution{ WorkflowId: wfTV.WorkflowID(), RunId: run.GetRunID(), }, Reason: "testing-reset-reactivation", RequestId: uuid.NewString(), WorkflowTaskFinishEventId: resetEventID, PostResetOperations: []*workflowpb.PostResetOperation{ { Variant: &workflowpb.PostResetOperation_UpdateWorkflowOptions_{ UpdateWorkflowOptions: &workflowpb.PostResetOperation_UpdateWorkflowOptions{ WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{ VersioningOverride: &workflowpb.VersioningOverride{ Override: &workflowpb.VersioningOverride_Pinned{ Pinned: &workflowpb.VersioningOverride_PinnedOverride{ Behavior: workflowpb.VersioningOverride_PINNED_OVERRIDE_BEHAVIOR_PINNED, Version: tv1.ExternalDeploymentVersion(), }, }, }, }, UpdateMask: &fieldmaskpb.FieldMask{ Paths: []string{"versioning_override"}, }, }, }, }, }, }) return resetErr == nil }, 10*time.Second, 500*time.Millisecond) newRunID := resetResp.RunId // Verify the reset workflow has the pinned override s.checkDescribeWorkflowAfterOverride(env, &commonpb.WorkflowExecution{ WorkflowId: wfTV.WorkflowID(), RunId: newRunID, }, &workflowpb.VersioningOverride{ Override: &workflowpb.VersioningOverride_Pinned{ Pinned: &workflowpb.VersioningOverride_PinnedOverride{ Behavior: workflowpb.VersioningOverride_PINNED_OVERRIDE_BEHAVIOR_PINNED, Version: tv1.ExternalDeploymentVersion(), }, }, }) // Wait for version 1 to show up as DRAINING (reactivated from INACTIVE) s.checkVersionDrainageAndVersionStatus(env, tv1, &deploymentpb.VersionDrainageInfo{ Status: enumspb.VERSION_DRAINAGE_STATUS_DRAINING, }, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING, 0) // Verify via DescribeWorkerDeployment that the version status is DRAINING s.checkVersionStatusInDeployment(env, tv1, enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_DRAINING) // Signal the reset workflow to complete s.NoError(env.SdkClient().SignalWorkflow(s.Context(), wfTV.WorkflowID(), newRunID, "complete", nil)) // Wait for the reset workflow to complete and verify it ran on version 1 resetRun := env.SdkClient().GetWorkflow(s.Context(), wfTV.WorkflowID(), newRunID) var result string s.NoError(resetRun.Get(s.Context(), &result)) s.Equal("done from v1", result, "Reset workflow should have completed on version 1") } // The following tests test the VersioningOverride functionality when passed via the BatchUpdateWorkflowExecutionOptions API. func (s *DeploymentVersionSuite) TestBatchUpdateWorkflowExecutionOptions_SetPinned_VersionDoesNotExist() { s.runBatchUpdateWorkflowExecutionOptionsTest(false) } func (s *DeploymentVersionSuite) TestBatchUpdateWorkflowExecutionOptions_SetPinnedThenUnset() { s.runBatchUpdateWorkflowExecutionOptionsTest(true) } func (s *DeploymentVersionSuite) runBatchUpdateWorkflowExecutionOptionsTest(createVersionFirst bool) { env := s.newTestEnv() // start some unversioned workflows workflowType := "UpdateOptionsBatchTestFunc" workflows := make([]*commonpb.WorkflowExecution, 0) for range 5 { run, err := env.SdkClient().ExecuteWorkflow(s.Context(), sdkclient.StartWorkflowOptions{TaskQueue: env.Tv().TaskQueue().Name}, workflowType) s.NoError(err) workflows = append(workflows, &commonpb.WorkflowExecution{ WorkflowId: run.GetID(), RunId: run.GetRunID(), }) } pinnedOverride := s.makePinnedOverride(env.Tv()) batchJobID := uuid.NewString() if createVersionFirst { // Start a versioned poller which shall create a version s.startVersionWorkflow(s.Context(), env, env.Tv()) } // start batch update-options operation _, err := env.SdkClient().WorkflowService().StartBatchOperation(context.Background(), &workflowservice.StartBatchOperationRequest{ Namespace: env.Namespace().String(), Operation: &workflowservice.StartBatchOperationRequest_UpdateWorkflowOptionsOperation{ UpdateWorkflowOptionsOperation: &batchpb.BatchOperationUpdateWorkflowExecutionOptions{ Identity: env.Tv().ClientIdentity(), WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{VersioningOverride: pinnedOverride}, UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"versioning_override"}}, }, }, Executions: workflows, JobId: batchJobID, Reason: "test", }) s.NoError(err) if !createVersionFirst { s.checkBatchOperationFails(env, batchJobID, len(workflows)) for _, wf := range workflows { s.checkDescribeWorkflowAfterOverride(env, wf, nil) } return } // wait til batch completes successfully s.checkListAndWaitForBatchCompletion(env, batchJobID) // check all the workflows for _, wf := range workflows { s.checkDescribeWorkflowAfterOverride(env, wf, pinnedOverride) s.checkWorkflowUpdateOptionsEventIdentity(s.Context(), env, wf, env.Tv().ClientIdentity()) } // unset with empty update opts with mutation mask batchJobID = uuid.NewString() err = s.startBatchJobWithinConcurrentJobLimit(env, &workflowservice.StartBatchOperationRequest{ Namespace: env.Namespace().String(), JobId: batchJobID, Reason: "test", Executions: workflows, Operation: &workflowservice.StartBatchOperationRequest_UpdateWorkflowOptionsOperation{ UpdateWorkflowOptionsOperation: &batchpb.BatchOperationUpdateWorkflowExecutionOptions{ Identity: env.Tv().ClientIdentity(), WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{}, UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"versioning_override"}}, }, }, }) s.NoError(err) // wait til batch completes s.checkListAndWaitForBatchCompletion(env, batchJobID) // check all the workflows for _, wf := range workflows { s.checkDescribeWorkflowAfterOverride(env, wf, nil) s.checkWorkflowUpdateOptionsEventIdentity(s.Context(), env, wf, env.Tv().ClientIdentity()) } } func (s *DeploymentVersionSuite) startBatchJobWithinConcurrentJobLimit(env *testcore.TestEnv, req *workflowservice.StartBatchOperationRequest) error { var err error s.Eventually(func() bool { _, err = env.FrontendClient().StartBatchOperation(s.Context(), req) if err == nil { return true } else if strings.Contains(err.Error(), "Max concurrent batch operations is reached") { return false // retry } return true }, 5*time.Second, 500*time.Millisecond) return err } func (s *DeploymentVersionSuite) checkListAndWaitForBatchCompletion(env *testcore.TestEnv, jobID string) { s.EventuallyWithT(func(t *assert.CollectT) { a := assert.New(t) listResp, err := env.FrontendClient().ListBatchOperations(s.Context(), &workflowservice.ListBatchOperationsRequest{ Namespace: env.Namespace().String(), }) a.NoError(err) a.NotEmpty(listResp.GetOperationInfo()) if len(listResp.GetOperationInfo()) > 0 { a.Equal(jobID, listResp.GetOperationInfo()[0].GetJobId()) } }, 10*time.Second, 50*time.Millisecond) s.EventuallyWithT(func(t *assert.CollectT) { a := assert.New(t) descResp, err := env.FrontendClient().DescribeBatchOperation(s.Context(), &workflowservice.DescribeBatchOperationRequest{ Namespace: env.Namespace().String(), JobId: jobID, }) a.NoError(err) a.NotEqual(enumspb.BATCH_OPERATION_STATE_FAILED, descResp.GetState(), "batch operation failed. description: %+v", descResp) a.Equal(enumspb.BATCH_OPERATION_STATE_COMPLETED, descResp.GetState()) }, 10*time.Second, 50*time.Millisecond) } func (s *DeploymentVersionSuite) checkBatchOperationFails(env *testcore.TestEnv, jobID string, numWorkflows int) { s.EventuallyWithT(func(t *assert.CollectT) { a := assert.New(t) descResp, err := env.FrontendClient().DescribeBatchOperation(s.Context(), &workflowservice.DescribeBatchOperationRequest{ Namespace: env.Namespace().String(), JobId: jobID, }) a.NoError(err) // All workflows should have failed validation a.Equal(int64(numWorkflows), descResp.GetFailureOperationCount(), "expected all operations to fail") }, 30*time.Second, 500*time.Millisecond) } func (s *DeploymentVersionSuite) makePinnedOverride(tv *testvars.TestVars) *workflowpb.VersioningOverride { if useV32 { return &workflowpb.VersioningOverride{Override: &workflowpb.VersioningOverride_Pinned{ Pinned: &workflowpb.VersioningOverride_PinnedOverride{ Behavior: workflowpb.VersioningOverride_PINNED_OVERRIDE_BEHAVIOR_PINNED, Version: tv.ExternalDeploymentVersion(), }, }} } return &workflowpb.VersioningOverride{ PinnedVersion: tv.DeploymentVersionString(), //nolint:staticcheck // SA1019: worker versioning v0.31 Behavior: enumspb.VERSIONING_BEHAVIOR_PINNED, //nolint:staticcheck // SA1019: worker versioning v0.31 } } func (s *DeploymentVersionSuite) makeAutoUpgradeOverride() *workflowpb.VersioningOverride { if useV32 { return &workflowpb.VersioningOverride{Override: &workflowpb.VersioningOverride_AutoUpgrade{AutoUpgrade: true}} } return &workflowpb.VersioningOverride{Behavior: enumspb.VERSIONING_BEHAVIOR_AUTO_UPGRADE} //nolint:staticcheck // SA1019: worker versioning v0.31 } func (s *DeploymentVersionSuite) TestStartWorkflowExecution_WithPinnedOverride_CacheMissAndHits() { env := s.newTestEnv( // TODO: remove WithWorkerService once legacy suite-scoped cluster behavior is removed. testcore.WithWorkerService("worker-deployment version membership cache test"), testcore.WithDynamicConfig(dynamicconfig.VersionMembershipCacheTTL, 5*time.Second), ) override := s.makePinnedOverride(env.Tv()) request := &workflowservice.StartWorkflowExecutionRequest{ RequestId: env.Tv().Any().String(), Namespace: env.Namespace().String(), WorkflowId: env.Tv().WorkflowID(), WorkflowType: env.Tv().WorkflowType(), TaskQueue: env.Tv().TaskQueue(), Identity: env.Tv().WorkerIdentity(), VersioningOverride: override, } // First call should fail since the version to override is not present in the task queue. _, err0 := env.FrontendClient().StartWorkflowExecution(s.Context(), request) s.Error(err0) // Start a versioned poller which shall create the version; the version must be present before it can be set as an override. s.startVersionWorkflow(s.Context(), env, env.Tv()) // Wait for the cache TTL to expire; On expiry of the cache TTL, it would result in a fresh RPC which would verify the version presence, // eventually leading to the StartWorkflowExecution call succeeding. var resp *workflowservice.StartWorkflowExecutionResponse s.Eventually(func() bool { var err error resp, err = env.FrontendClient().StartWorkflowExecution(s.Context(), request) return err == nil }, 10*time.Second, 500*time.Millisecond) // The StartWorkflowExecution should now succeed with no error. Verify that the workflow shows the override. s.checkDescribeWorkflowAfterOverride(env, &commonpb.WorkflowExecution{ WorkflowId: env.Tv().WorkflowID(), RunId: resp.GetRunId(), }, override) } func (s *DeploymentVersionSuite) TestStartWorkflowExecution_WithUnpinnedOverride() { env := s.newTestEnv() override := s.makeAutoUpgradeOverride() wf := &commonpb.WorkflowExecution{ WorkflowId: env.Tv().WorkflowID(), RunId: s.startWorkflow(env, env.Tv(), override), } s.checkDescribeWorkflowAfterOverride(env, wf, override) } func (s *DeploymentVersionSuite) TestSignalWithStartWorkflowExecution_WithPinnedOverride_CacheMissAndHits() { env := s.newTestEnv( // TODO: remove WithWorkerService once legacy suite-scoped cluster behavior is removed. testcore.WithWorkerService("worker-deployment version membership cache test"), testcore.WithDynamicConfig(dynamicconfig.VersionMembershipCacheTTL, 5*time.Second), ) override := s.makePinnedOverride(env.Tv()) request := &workflowservice.SignalWithStartWorkflowExecutionRequest{ Namespace: env.Namespace().String(), WorkflowId: env.Tv().WorkflowID(), WorkflowType: env.Tv().WorkflowType(), TaskQueue: env.Tv().TaskQueue(), Identity: env.Tv().ClientIdentity(), RequestId: env.Tv().RequestID(), SignalName: "test-signal", SignalInput: nil, VersioningOverride: override, } // Since the version to override is not present in the task queue, the call should fail. _, err := env.FrontendClient().SignalWithStartWorkflowExecution(s.Context(), request) s.Error(err) // Start a versioned poller which shall create the version; the version must be present before it can be set as an override. s.startVersionWorkflow(s.Context(), env, env.Tv()) // Wait for the cache TTL to expire; On expiry of the cache TTL, it would result in a fresh RPC which would verify the version presence, // eventually leading to the SignalWithStartWorkflowExecution call succeeding. var resp *workflowservice.SignalWithStartWorkflowExecutionResponse s.Eventually(func() bool { var err error resp, err = env.FrontendClient().SignalWithStartWorkflowExecution(s.Context(), request) return err == nil && resp.GetStarted() }, 10*time.Second, 500*time.Millisecond) wf := &commonpb.WorkflowExecution{ WorkflowId: env.Tv().WorkflowID(), RunId: resp.GetRunId(), } s.checkDescribeWorkflowAfterOverride(env, wf, override) } func (s *DeploymentVersionSuite) TestSignalWithStartWorkflowExecution_WithUnpinnedOverride() { env := s.newTestEnv() override := s.makeAutoUpgradeOverride() resp, err := env.FrontendClient().SignalWithStartWorkflowExecution(s.Context(), &workflowservice.SignalWithStartWorkflowExecutionRequest{ Namespace: env.Namespace().String(), WorkflowId: env.Tv().WorkflowID(), WorkflowType: env.Tv().WorkflowType(), TaskQueue: env.Tv().TaskQueue(), Identity: env.Tv().ClientIdentity(), RequestId: env.Tv().RequestID(), SignalName: "test-signal", SignalInput: nil, VersioningOverride: override, }) s.NoError(err) s.True(resp.GetStarted()) wf := &commonpb.WorkflowExecution{ WorkflowId: env.Tv().WorkflowID(), RunId: resp.GetRunId(), } s.checkDescribeWorkflowAfterOverride(env, wf, override) } func (s *DeploymentVersionSuite) TestDeleteVersion_ThenRecreateByPolling() { s.skipBeforeVersion(workerdeployment.VersionDataRevisionNumber) env := s.newTestEnv(testcore.WithDynamicConfig(dynamicconfig.PollerHistoryTTL, 500*time.Millisecond)) tv1 := env.Tv().WithBuildIDNumber(1) s.startVersionWorkflow(s.Context(), env, tv1) vd := s.getTaskQueueVersionData(env, tv1, enumspb.TASK_QUEUE_TYPE_WORKFLOW, tv1.ExternalDeploymentVersion()) s.Equal(int64(0), vd.GetRevisionNumber()) s.False(vd.GetDeleted()) // Wait for pollers to go away s.EventuallyWithT(func(t *assert.CollectT) { resp, err := env.FrontendClient().DescribeTaskQueue(s.Context(), &workflowservice.DescribeTaskQueueRequest{ Namespace: env.Namespace().String(), TaskQueue: tv1.TaskQueue(), TaskQueueType: enumspb.TASK_QUEUE_TYPE_WORKFLOW, }) require.NoError(t, err) require.Empty(t, resp.Pollers) }, 5*time.Second, time.Second) // Delete the version s.tryDeleteVersion(env, tv1, "", false) // Verify the version is gone from the task queue s.EventuallyWithT(func(t *assert.CollectT) { vd = s.getTaskQueueVersionData(env, tv1, enumspb.TASK_QUEUE_TYPE_WORKFLOW, tv1.ExternalDeploymentVersion()) require.New(t).Nil(vd) }, time.Second*5, time.Millisecond*200) // Verify the version is gone from the deployment s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) resp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: tv1.DeploymentSeries(), }) a.NoError(err) for _, vs := range resp.GetWorkerDeploymentInfo().GetVersionSummaries() { //nolint:staticcheck // SA1019 deprecated Version will clean up later a.NotEqual(tv1.DeploymentVersionString(), vs.Version) } }, time.Second*5, time.Millisecond*200) // Poll again to recreate the version s.startVersionWorkflow(s.Context(), env, tv1) // Verify the version is back (undeleted) in the deployment s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) resp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: tv1.DeploymentSeries(), }) a.NoError(err) found := false for _, vs := range resp.GetWorkerDeploymentInfo().GetVersionSummaries() { //nolint:staticcheck // SA1019 deprecated Version will clean up later if vs.Version == tv1.DeploymentVersionString() { found = true } } a.True(found, "version should be recreated after polling") }, time.Second*5, time.Millisecond*200) // Ensure the version data revived properly in the task queue vd = s.getTaskQueueVersionData(env, tv1, enumspb.TASK_QUEUE_TYPE_WORKFLOW, tv1.ExternalDeploymentVersion()) s.Equal(int64(0), vd.GetRevisionNumber()) s.False(vd.GetDeleted()) } // getTaskQueueDeploymentData gets the deployment data for a given TQ type. The data is always // returned from the WF type root partition, so no need to wait for propagation before calling this // function. func (s *DeploymentVersionSuite) getTaskQueueDeploymentData( env *testcore.TestEnv, tv *testvars.TestVars, tqType enumspb.TaskQueueType, ) *persistencespb.DeploymentData { ctx, cancel := context.WithTimeout(s.Context(), time.Second*5) defer cancel() resp, err := env.GetTestCluster().MatchingClient().GetTaskQueueUserData( ctx, &matchingservice.GetTaskQueueUserDataRequest{ NamespaceId: env.NamespaceID().String(), TaskQueue: tv.TaskQueue().GetName(), TaskQueueType: tqTypeWf, }) s.NoError(err) return resp.GetUserData().GetData().GetPerType()[int32(tqType)].GetDeploymentData() } func (s *DeploymentVersionSuite) getTaskQueueVersionData( env *testcore.TestEnv, tv *testvars.TestVars, tqType enumspb.TaskQueueType, version *deploymentpb.WorkerDeploymentVersion, ) *deploymentspb.WorkerDeploymentVersionData { data := s.getTaskQueueDeploymentData(env, tv, tqType) return data.GetDeploymentsData()[version.GetDeploymentName()].GetVersions()[version.GetBuildId()] } func (s *DeploymentVersionSuite) TestCreateWorkerDeploymentVersion_Success() { env := s.newTestEnv() buildID := env.Tv().BuildID() requestID := env.Tv().Any().String() identity := env.Tv().Any().String() // First create the deployment _, err := env.FrontendClient().CreateWorkerDeployment(s.Context(), &workflowservice.CreateWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: env.Tv().DeploymentSeries(), RequestId: env.Tv().Any().String(), }) s.NoError(err) computeConfig := &computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg1": { Provider: computeprovider.TestInvokeComputeProviderValidComputeProvider(), Scaler: testInvokeScaler(), }, }, } // Create a version in the deployment resp, err := env.FrontendClient().CreateWorkerDeploymentVersion(s.Context(), &workflowservice.CreateWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), DeploymentVersion: &deploymentpb.WorkerDeploymentVersion{ DeploymentName: env.Tv().DeploymentSeries(), BuildId: buildID, }, Identity: identity, RequestId: requestID, ComputeConfig: computeConfig, }) s.NoError(err) s.NotNil(resp) // Verify the version exists via DescribeWorkerDeploymentVersion s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) descResp, err := env.FrontendClient().DescribeWorkerDeploymentVersion(s.Context(), &workflowservice.DescribeWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), Version: env.Tv().DeploymentVersionString(), }) a.NoError(err) a.NotNil(descResp.GetWorkerDeploymentVersionInfo()) a.Equal(env.Tv().DeploymentVersionStringV32(), worker_versioning.ExternalWorkerDeploymentVersionToString(descResp.GetWorkerDeploymentVersionInfo().GetDeploymentVersion())) a.NotNil(descResp.GetWorkerDeploymentVersionInfo().GetCreateTime()) a.True(proto.Equal(computeConfig, descResp.GetWorkerDeploymentVersionInfo().GetComputeConfig())) a.Equal(enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CREATED, descResp.GetWorkerDeploymentVersionInfo().GetStatus()) a.Equal(identity, descResp.GetWorkerDeploymentVersionInfo().GetLastModifierIdentity()) }, 10*time.Second, 500*time.Millisecond) // Verify the version shows up in deployment's version summaries with CREATED status and correct compute config summary. s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) descDeployResp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: env.Tv().DeploymentSeries(), }) a.NoError(err) a.Len(descDeployResp.GetWorkerDeploymentInfo().GetVersionSummaries(), 1) versionSummary := descDeployResp.GetWorkerDeploymentInfo().GetVersionSummaries()[0] a.Equal(env.Tv().DeploymentVersionStringV32(), worker_versioning.ExternalWorkerDeploymentVersionToString(versionSummary.GetDeploymentVersion())) a.Equal(enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CREATED, versionSummary.GetStatus()) a.True(proto.Equal(&computepb.ComputeConfigSummary{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroupSummary{ "sg1": { ProviderType: computeConfig.GetScalingGroups()["sg1"].GetProvider().GetType(), }, }, }, versionSummary.GetComputeConfig())) }, 10*time.Second, 500*time.Millisecond) // Verify the compute config summary is reflected in ListWorkerDeployments latest version summary. s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) listResp, err := env.FrontendClient().ListWorkerDeployments(s.Context(), &workflowservice.ListWorkerDeploymentsRequest{ Namespace: env.Namespace().String(), }) a.NoError(err) var found *workflowservice.ListWorkerDeploymentsResponse_WorkerDeploymentSummary for _, d := range listResp.GetWorkerDeployments() { if d.GetName() == env.Tv().DeploymentSeries() { found = d break } } a.NotNil(found, "deployment %s not found in ListWorkerDeployments", env.Tv().DeploymentSeries()) a.True(proto.Equal(&computepb.ComputeConfigSummary{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroupSummary{ "sg1": { ProviderType: computeConfig.GetScalingGroups()["sg1"].GetProvider().GetType(), }, }, }, found.GetLatestVersionSummary().GetComputeConfig())) }, 10*time.Second, 500*time.Millisecond) } func (s *DeploymentVersionSuite) TestCreateWorkerDeploymentVersion_ThenPoll_TaskQueueInVersionInfo() { env := s.newTestEnv() buildID := env.Tv().BuildID() // Create the deployment _, err := env.FrontendClient().CreateWorkerDeployment(s.Context(), &workflowservice.CreateWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: env.Tv().DeploymentSeries(), RequestId: env.Tv().Any().String(), }) s.NoError(err) // Create the version explicitly _, err = env.FrontendClient().CreateWorkerDeploymentVersion(s.Context(), &workflowservice.CreateWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), DeploymentVersion: &deploymentpb.WorkerDeploymentVersion{ DeploymentName: env.Tv().DeploymentSeries(), BuildId: buildID, }, RequestId: env.Tv().Any().String(), }) s.NoError(err) // Verify the version starts with CREATED status s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) descResp, err := env.FrontendClient().DescribeWorkerDeploymentVersion(s.Context(), &workflowservice.DescribeWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), Version: env.Tv().DeploymentVersionString(), }) a.NoError(err) a.Equal(enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CREATED, descResp.GetWorkerDeploymentVersionInfo().GetStatus()) }, 10*time.Second, 500*time.Millisecond) // Poll from the version to register a task queue go s.pollFromDeployment(s.Context(), env, env.Tv()) // Verify the task queue shows up and status transitions to INACTIVE s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) descResp, err := env.FrontendClient().DescribeWorkerDeploymentVersion(s.Context(), &workflowservice.DescribeWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), Version: env.Tv().DeploymentVersionString(), }) a.NoError(err) tqInfos := descResp.GetWorkerDeploymentVersionInfo().GetTaskQueueInfos() a.GreaterOrEqual(len(tqInfos), 1) found := false for _, tqInfo := range tqInfos { if tqInfo.GetName() == env.Tv().TaskQueue().GetName() { found = true break } } a.True(found, "expected task queue %q in version info, got %v", env.Tv().TaskQueue().GetName(), tqInfos) a.Equal(enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, descResp.GetWorkerDeploymentVersionInfo().GetStatus()) }, 30*time.Second, 500*time.Millisecond) // Verify the version shows up in deployment's version summaries with INACTIVE status s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) descDeployResp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: env.Tv().DeploymentSeries(), }) a.NoError(err) a.Len(descDeployResp.GetWorkerDeploymentInfo().GetVersionSummaries(), 1) a.Equal(env.Tv().DeploymentVersionStringV32(), worker_versioning.ExternalWorkerDeploymentVersionToString(descDeployResp.GetWorkerDeploymentInfo().GetVersionSummaries()[0].GetDeploymentVersion())) a.Equal(enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_INACTIVE, descDeployResp.GetWorkerDeploymentInfo().GetVersionSummaries()[0].GetStatus()) }, 10*time.Second, 500*time.Millisecond) } func (s *DeploymentVersionSuite) TestCreateWorkerDeploymentVersion_Idempotent() { env := s.newTestEnv() buildID := env.Tv().BuildID() requestID := env.Tv().Any().String() // First create the deployment _, err := env.FrontendClient().CreateWorkerDeployment(s.Context(), &workflowservice.CreateWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: env.Tv().DeploymentSeries(), RequestId: env.Tv().Any().String(), }) s.NoError(err) // Create a version _, err = env.FrontendClient().CreateWorkerDeploymentVersion(s.Context(), &workflowservice.CreateWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), DeploymentVersion: &deploymentpb.WorkerDeploymentVersion{ DeploymentName: env.Tv().DeploymentSeries(), BuildId: buildID, }, RequestId: requestID, }) s.NoError(err) // Create the same version again with same request ID - should be idempotent _, err = env.FrontendClient().CreateWorkerDeploymentVersion(s.Context(), &workflowservice.CreateWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), DeploymentVersion: &deploymentpb.WorkerDeploymentVersion{ DeploymentName: env.Tv().DeploymentSeries(), BuildId: buildID, }, RequestId: requestID, }) s.NoError(err) } func (s *DeploymentVersionSuite) TestCreateWorkerDeploymentVersion_AlreadyExists_DifferentRequestID() { env := s.newTestEnv() buildID := env.Tv().BuildID() requestID1 := env.Tv().Any().String() requestID2 := env.Tv().Any().String() // First create the deployment _, err := env.FrontendClient().CreateWorkerDeployment(s.Context(), &workflowservice.CreateWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: env.Tv().DeploymentSeries(), RequestId: env.Tv().Any().String(), }) s.NoError(err) // Create a version _, err = env.FrontendClient().CreateWorkerDeploymentVersion(s.Context(), &workflowservice.CreateWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), DeploymentVersion: &deploymentpb.WorkerDeploymentVersion{ DeploymentName: env.Tv().DeploymentSeries(), BuildId: buildID, }, RequestId: requestID1, }) s.NoError(err) // Try to create the same version with different request ID - should fail _, err = env.FrontendClient().CreateWorkerDeploymentVersion(s.Context(), &workflowservice.CreateWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), DeploymentVersion: &deploymentpb.WorkerDeploymentVersion{ DeploymentName: env.Tv().DeploymentSeries(), BuildId: buildID, }, RequestId: requestID2, }) s.Error(err) var alreadyExists *serviceerror.AlreadyExists s.ErrorAs(err, &alreadyExists) } func (s *DeploymentVersionSuite) TestCreateWorkerDeploymentVersion_DeploymentNotFound() { env := s.newTestEnv() // Try to create a version for a deployment that doesn't exist _, err := env.FrontendClient().CreateWorkerDeploymentVersion(s.Context(), &workflowservice.CreateWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), DeploymentVersion: &deploymentpb.WorkerDeploymentVersion{ DeploymentName: env.Tv().DeploymentSeries(), BuildId: env.Tv().BuildID(), }, RequestId: env.Tv().Any().String(), }) s.Error(err) var notFound *serviceerror.NotFound s.ErrorAs(err, ¬Found) } func (s *DeploymentVersionSuite) TestCreateWorkerDeploymentVersion_InvalidArgs() { testCases := []struct { name string deploymentName string buildID string expectedError string }{ { name: "empty deployment name", deploymentName: "", buildID: "build-1", expectedError: "deployment name cannot be empty", }, { name: "empty build ID", deploymentName: "my-deployment", buildID: "", expectedError: "build ID cannot be empty", }, } for _, tc := range testCases { s.Run(tc.name, func(s *DeploymentVersionSuite) { env := s.newTestEnv() _, err := env.FrontendClient().CreateWorkerDeploymentVersion(s.Context(), &workflowservice.CreateWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), DeploymentVersion: &deploymentpb.WorkerDeploymentVersion{ DeploymentName: tc.deploymentName, BuildId: tc.buildID, }, RequestId: env.Tv().Any().String(), }) s.Error(err) var invalidArg *serviceerror.InvalidArgument s.ErrorAs(err, &invalidArg) s.Contains(invalidArg.Message, tc.expectedError) }) } } func (s *DeploymentVersionSuite) TestCreateWorkerDeploymentVersion_AutoCreatedByPoller_ConflictWithExplicitCreate() { env := s.newTestEnv() // Create version via polling (auto-creates deployment and version) s.startVersionWorkflow(s.Context(), env, env.Tv()) // Try to explicitly create the same version with a different request ID _, err := env.FrontendClient().CreateWorkerDeploymentVersion(s.Context(), &workflowservice.CreateWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), DeploymentVersion: &deploymentpb.WorkerDeploymentVersion{ DeploymentName: env.Tv().DeploymentSeries(), BuildId: env.Tv().BuildID(), }, RequestId: env.Tv().Any().String(), }) s.Error(err) var alreadyExists *serviceerror.AlreadyExists s.ErrorAs(err, &alreadyExists) } func (s *DeploymentVersionSuite) TestCreateWorkerDeploymentVersion_MultipleVersions() { env := s.newTestEnv() // First create the deployment _, err := env.FrontendClient().CreateWorkerDeployment(s.Context(), &workflowservice.CreateWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: env.Tv().DeploymentSeries(), RequestId: env.Tv().Any().String(), }) s.NoError(err) computeConfig1 := &computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg1": { Provider: computeprovider.TestInvokeComputeProviderValidComputeProvider(), Scaler: testInvokeScaler(), }, }, } computeConfig2 := &computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg2": { Provider: computeprovider.TestInvokeComputeProviderValidComputeProvider(), Scaler: testInvokeScaler(), }, }, } // Create first version tv1 := env.Tv().WithBuildIDNumber(1) _, err = env.FrontendClient().CreateWorkerDeploymentVersion(s.Context(), &workflowservice.CreateWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), DeploymentVersion: &deploymentpb.WorkerDeploymentVersion{ DeploymentName: env.Tv().DeploymentSeries(), BuildId: tv1.BuildID(), }, RequestId: env.Tv().Any().String(), ComputeConfig: computeConfig1, }) s.NoError(err) // Create second version tv2 := env.Tv().WithBuildIDNumber(2) _, err = env.FrontendClient().CreateWorkerDeploymentVersion(s.Context(), &workflowservice.CreateWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), DeploymentVersion: &deploymentpb.WorkerDeploymentVersion{ DeploymentName: env.Tv().DeploymentSeries(), BuildId: tv2.BuildID(), }, RequestId: env.Tv().Any().String(), ComputeConfig: computeConfig2, }) s.NoError(err) // Verify both versions show up in deployment's version summaries s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) descResp, err := env.FrontendClient().DescribeWorkerDeployment(s.Context(), &workflowservice.DescribeWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: env.Tv().DeploymentSeries(), }) a.NoError(err) a.Len(descResp.GetWorkerDeploymentInfo().GetVersionSummaries(), 2) }, 10*time.Second, 500*time.Millisecond) // Verify compute configs via DescribeWorkerDeploymentVersion s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) descV1, err := env.FrontendClient().DescribeWorkerDeploymentVersion(s.Context(), &workflowservice.DescribeWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), Version: tv1.DeploymentVersionString(), }) a.NoError(err) a.True(proto.Equal(computeConfig1, descV1.GetWorkerDeploymentVersionInfo().GetComputeConfig())) }, 10*time.Second, 500*time.Millisecond) s.EventuallyWithT(func(t *assert.CollectT) { a := require.New(t) descV2, err := env.FrontendClient().DescribeWorkerDeploymentVersion(s.Context(), &workflowservice.DescribeWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), Version: tv2.DeploymentVersionString(), }) a.NoError(err) a.True(proto.Equal(computeConfig2, descV2.GetWorkerDeploymentVersionInfo().GetComputeConfig())) }, 10*time.Second, 500*time.Millisecond) } // TestCreateWorkerDeploymentVersion_InvalidComputeConfig_ReturnsPromptly verifies that // creating a version with invalid compute config returns an InvalidArgument error promptly, // even when the deployment workflow is busy (e.g., processing a recently polled version). // This is a regression test for a bug where the activity error was retryable, causing the // deployment workflow to stay busy and the client to time out with "too many requests". func (s *DeploymentVersionSuite) TestCreateWorkerDeploymentVersion_InvalidComputeConfig_ReturnsPromptly() { env := s.newTestEnv() // Create a version via polling, which auto-creates the deployment and keeps // the deployment workflow active with lazy-creation processing. s.startVersionWorkflow(s.Context(), env, env.Tv()) // Immediately attempt to create another version on the same deployment with // invalid compute config. This should return an InvalidArgument error // promptly (not time out with "too many requests"). tv2 := env.Tv().WithBuildIDNumber(2) invalidProvider := computeprovider.TestInvokeComputeProviderInvalidComputeProvider() start := time.Now() _, err := env.FrontendClient().CreateWorkerDeploymentVersion(s.Context(), &workflowservice.CreateWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), DeploymentVersion: &deploymentpb.WorkerDeploymentVersion{ DeploymentName: env.Tv().DeploymentSeries(), BuildId: tv2.BuildID(), }, RequestId: tv2.Any().String(), ComputeConfig: &computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg1": {Provider: invalidProvider}, }, }, }) elapsed := time.Since(start) s.Error(err) var invalidArg *serviceerror.InvalidArgument s.ErrorAs(err, &invalidArg) s.Contains(invalidArg.Message, "illegal_field found in config") // The error should return well before the deployment client's 1-minute // retry timeout. If it takes longer than 30 seconds, the activity error // is likely still retryable (the bug we're testing for). s.Less(elapsed, 30*time.Second, "create-version with invalid compute config should fail promptly, not time out") } func (s *DeploymentVersionSuite) TestCreateWorkerDeploymentVersion_InvalidScalingGroups() { validProvider := computeprovider.TestInvokeComputeProviderValidComputeProvider() testCases := []struct { name string computeConfig *computepb.ComputeConfig expectedError string }{ { name: "invalid compute provider type", computeConfig: &computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg1": {TaskQueueTypes: nil, Provider: &computepb.ComputeProvider{Type: "invalid-provider"}}, }, }, expectedError: "invalid compute provider type", }, { name: "invalid compute provider details", computeConfig: &computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg1": {TaskQueueTypes: nil, Provider: computeprovider.TestInvokeComputeProviderInvalidComputeProvider()}, }, }, expectedError: "illegal_field found in config", }, { name: "two catch-all scaling groups", computeConfig: &computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg1": {TaskQueueTypes: nil, Provider: validProvider}, "sg2": {TaskQueueTypes: nil, Provider: validProvider}, }, }, expectedError: "only one scaling group can have no task types defined", }, { name: "overlapping workflow task queue type", computeConfig: &computepb.ComputeConfig{ ScalingGroups: map[string]*computepb.ComputeConfigScalingGroup{ "sg1": {TaskQueueTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_WORKFLOW}, Provider: validProvider}, "sg2": {TaskQueueTypes: []enumspb.TaskQueueType{enumspb.TASK_QUEUE_TYPE_WORKFLOW, enumspb.TASK_QUEUE_TYPE_ACTIVITY}, Provider: validProvider}, }, }, expectedError: "task type Workflow appears in more than one entry", }, } for _, tc := range testCases { s.Run(tc.name, func(s *DeploymentVersionSuite) { env := s.newTestEnv() // Create the deployment first _, err := env.FrontendClient().CreateWorkerDeployment(s.Context(), &workflowservice.CreateWorkerDeploymentRequest{ Namespace: env.Namespace().String(), DeploymentName: env.Tv().DeploymentSeries(), RequestId: env.Tv().Any().String(), }) s.NoError(err) _, err = env.FrontendClient().CreateWorkerDeploymentVersion(s.Context(), &workflowservice.CreateWorkerDeploymentVersionRequest{ Namespace: env.Namespace().String(), DeploymentVersion: &deploymentpb.WorkerDeploymentVersion{ DeploymentName: env.Tv().DeploymentSeries(), BuildId: env.Tv().BuildID(), }, RequestId: env.Tv().Any().String(), ComputeConfig: tc.computeConfig, }) s.Error(err) var invalidArg *serviceerror.InvalidArgument s.ErrorAs(err, &invalidArg) s.Contains(invalidArg.Message, tc.expectedError) }) } }