multiple autoupgrade workflows should not bounce

Author: Shivs11

tests/versioning_3_test.go

@@ -3734,6 +3734,7 @@ func (s *Versioning3Suite) TestAutoUpgradeWorkflow_NoBouncingBetweenVersions() {
 	}, 10*time.Second, 100*time.Millisecond)
 
 	// Ensure that the workflow does not bounce back to v1.
+	//nolint:testifylint
 	go s.idlePollWorkflow(tv1, true, ver3MinPollTime, "workflow should not bounce back to v1")
 
 	// Poll and complete the workflow on v2.
@@ -3749,6 +3750,217 @@ func (s *Versioning3Suite) TestAutoUpgradeWorkflow_NoBouncingBetweenVersions() {
 	}, 10*time.Second, 100*time.Millisecond)
 }
 
+func (s *Versioning3Suite) TestAutoUpgradeWorkflows_NoBouncingBetweenVersions() {
+	if !s.useNewDeploymentData {
+		s.T().Skip("This test is only supported on new deployment data")
+	}
+	/*
+
+		Test plan:
+		- Use only one read and write partition.
+		- Update the userData by setting current version to v0.
+		- Start 10 workflows on v0. MS reads v0 for these workflows.
+		- Update the userData by setting the current version to v1.
+		- Complete workflow task on v1. MS reads v1 for these workflows now.
+		- *Rollback* userData to v0 on that single partition (call lower level API)
+		- See if any workflow task goes back to original v0 poller (should not)
+
+	*/
+
+	s.OverrideDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, 1)
+	s.OverrideDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, 1)
+	s.OverrideDynamicConfig(dynamicconfig.UseRevisionNumberForWorkerVersioning, true)
+
+	tv0 := testvars.New(s).WithBuildIDNumber(0)
+	tv1 := tv0.WithBuildIDNumber(1)
+
+	// Update the userData by setting the current version to v0
+	s.updateTaskQueueDeploymentDataWithRoutingConfig(tv0, &deploymentpb.RoutingConfig{
+		CurrentDeploymentVersion:  worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(tv0.DeploymentVersionString()),
+		CurrentVersionChangedTime: timestamp.TimePtr(time.Now()),
+		RevisionNumber:            1,
+	}, map[string]*deploymentspb.WorkerDeploymentVersionData{tv0.DeploymentVersion().GetBuildId(): &deploymentspb.WorkerDeploymentVersionData{
+		Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
+	}}, []string{}, tqTypeWf, tqTypeAct)
+
+	// Wait until all task queue partitions know that v0 is current.
+	s.waitForDeploymentDataPropagation(tv0, versionStatusCurrent, false, tqTypeWf, tqTypeAct)
+
+	// Make numWorkflows different workflow ID's so that we can verify that all workflows are running on v0.
+	numWorkflows := 10
+	wfVarsV0 := make([]*testvars.TestVars, numWorkflows)
+	wfVarsV1 := make([]*testvars.TestVars, numWorkflows)
+
+	for i := 0; i < numWorkflows; i++ {
+		wfVarsV0[i] = tv0.WithWorkflowIDNumber(i)
+		wfVarsV1[i] = tv1.WithWorkflowIDNumber(i)
+	}
+
+	// Start all different workflows on version v0.
+	for i := 0; i < numWorkflows; i++ {
+		s.startWorkflow(wfVarsV0[i], nil)
+	}
+
+	// Poll for workflows on v0.
+	channels := make([]chan struct{}, numWorkflows)
+	for i := 0; i < numWorkflows; i++ {
+		channels[i] = make(chan struct{})
+		s.pollWftAndHandle(wfVarsV0[i], false, channels[i],
+			func(task *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
+				s.NotNil(task)
+				resp := respondEmptyWft(wfVarsV0[i], false, vbUnpinned)
+				resp.ForceCreateNewWorkflowTask = true
+				return resp, nil
+			})
+	}
+	// Wait for channels to be closed
+	for i := 0; i < numWorkflows; i++ {
+		<-channels[i]
+	}
+
+	// Verify that all workflows are running on v0.
+	for i := 0; i < numWorkflows; i++ {
+		s.EventuallyWithT(func(t *assert.CollectT) {
+			s.verifyWorkflowVersioning(wfVarsV0[i], vbUnpinned, tv0.Deployment(), nil, nil)
+		}, 10*time.Second, 100*time.Millisecond)
+	}
+
+	// Update the userData by setting the current version to v1.
+	s.updateTaskQueueDeploymentDataWithRoutingConfig(tv1, &deploymentpb.RoutingConfig{
+		CurrentDeploymentVersion:  worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(tv1.DeploymentVersionString()),
+		CurrentVersionChangedTime: timestamp.TimePtr(time.Now()),
+		RevisionNumber:            2,
+	}, map[string]*deploymentspb.WorkerDeploymentVersionData{tv1.DeploymentVersion().GetBuildId(): &deploymentspb.WorkerDeploymentVersionData{
+		Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
+	}}, []string{}, tqTypeWf, tqTypeAct)
+
+	// Wait until all task queue partitions know that v1 is current.
+	// s.waitForDeploymentDataPropagation(tv1, versionStatusCurrent, false, tqTypeWf, tqTypeAct)
+
+	// Poll for workflows but this time the workflow task should be acted upon by a v1 worker.
+	channels = make([]chan struct{}, numWorkflows)
+	for i := 0; i < numWorkflows; i++ {
+		channels[i] = make(chan struct{})
+		s.pollWftAndHandle(wfVarsV1[i], false, channels[i],
+			func(task *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
+				s.NotNil(task)
+				resp := respondEmptyWft(wfVarsV1[i], false, vbUnpinned)
+				resp.ForceCreateNewWorkflowTask = true
+				return resp, nil
+			})
+	}
+	// Wait for channels to be closed
+	for i := 0; i < numWorkflows; i++ {
+		<-channels[i]
+	}
+	// Verify that all workflows are running on v1.
+	for i := 0; i < numWorkflows; i++ {
+		s.EventuallyWithT(func(t *assert.CollectT) {
+			s.verifyWorkflowVersioning(wfVarsV1[i], vbUnpinned, tv1.Deployment(), nil, nil)
+		}, 10*time.Second, 100*time.Millisecond)
+	}
+
+	// Rollback the userData to v0. Using the lower API here.
+
+	currentData, err := s.GetTestCluster().MatchingClient().GetTaskQueueUserData(context.Background(), &matchingservice.GetTaskQueueUserDataRequest{
+		NamespaceId:   s.NamespaceID().String(),
+		TaskQueue:     tv0.TaskQueue().GetName(),
+		TaskQueueType: tqTypeWf,
+	})
+	s.NoError(err)
+
+	_, err = s.GetTestCluster().MatchingClient().UpdateTaskQueueUserData(context.Background(), &matchingservice.UpdateTaskQueueUserDataRequest{
+		NamespaceId: s.NamespaceID().String(),
+		TaskQueue:   tv0.TaskQueue().GetName(),
+		UserData: &persistencespb.VersionedTaskQueueUserData{
+			Data: &persistencespb.TaskQueueUserData{
+				PerType: map[int32]*persistencespb.TaskQueueTypeUserData{
+					int32(tqTypeWf): {
+						DeploymentData: &persistencespb.DeploymentData{
+							DeploymentsData: map[string]*persistencespb.WorkerDeploymentData{
+								tv0.DeploymentVersion().GetDeploymentName(): {
+									RoutingConfig: &deploymentpb.RoutingConfig{
+										CurrentDeploymentVersion:  worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(tv0.DeploymentVersionString()),
+										CurrentVersionChangedTime: timestamp.TimePtr(time.Now().Add(-time.Second)),
+										RevisionNumber:            0,
+									},
+									Versions: map[string]*deploymentspb.WorkerDeploymentVersionData{
+										tv0.DeploymentVersion().GetBuildId(): {
+											Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
+										},
+									},
+								},
+							},
+						},
+					},
+					int32(tqTypeAct): {
+						DeploymentData: &persistencespb.DeploymentData{
+							DeploymentsData: map[string]*persistencespb.WorkerDeploymentData{
+								tv0.DeploymentVersion().GetDeploymentName(): {
+									RoutingConfig: &deploymentpb.RoutingConfig{
+										CurrentDeploymentVersion:  worker_versioning.ExternalWorkerDeploymentVersionFromStringV31(tv0.DeploymentVersionString()),
+										CurrentVersionChangedTime: timestamp.TimePtr(time.Now().Add(-time.Second)),
+										RevisionNumber:            0,
+									},
+									Versions: map[string]*deploymentspb.WorkerDeploymentVersionData{
+										tv0.DeploymentVersion().GetBuildId(): {
+											Status: enumspb.WORKER_DEPLOYMENT_VERSION_STATUS_CURRENT,
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			Version: currentData.GetUserData().GetVersion(),
+		},
+	})
+	s.NoError(err)
+
+	// Verify that the userData is rolled back to v0.
+	s.Eventually(func() bool {
+		ms, err := s.GetTestCluster().MatchingClient().GetTaskQueueUserData(context.Background(), &matchingservice.GetTaskQueueUserDataRequest{
+			NamespaceId:   s.NamespaceID().String(),
+			TaskQueue:     tv0.TaskQueue().GetName(),
+			TaskQueueType: tqTypeWf,
+		})
+		s.NoError(err)
+
+		// Find the current version for this task-queue specifically. It should be set to v0.
+		current, currentRevisionNumber, _, _, _, _, _, _ := worker_versioning.CalculateTaskQueueVersioningInfo(ms.GetUserData().GetData().GetPerType()[int32(tqTypeWf)].GetDeploymentData())
+		return current.GetBuildId() == tv0.DeploymentVersion().GetBuildId() && currentRevisionNumber == 0
+	}, 10*time.Second, 100*time.Millisecond)
+
+	// Even though the userData is rolled back to v0, the workflows should only continue to run on v1.
+	// Start idle pollers on v0 and ensure they never receive a task.
+	for i := 0; i < numWorkflows; i++ {
+		//nolint:testifylint
+		go s.idlePollWorkflow(wfVarsV0[i], true, ver3MinPollTime, "workflows should not go to the old deployment")
+	}
+
+	// Complete all workflows on v1.
+	channels = make([]chan struct{}, numWorkflows)
+	for i := 0; i < numWorkflows; i++ {
+		channels[i] = make(chan struct{})
+		s.pollWftAndHandle(wfVarsV1[i], false, channels[i],
+			func(task *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
+				s.NotNil(task)
+				return respondCompleteWorkflow(wfVarsV1[i], vbUnpinned), nil
+			})
+	}
+	// Wait for channels to be closed
+	for i := 0; i < numWorkflows; i++ {
+		<-channels[i]
+	}
+
+	// Verify that all workflows are completed on v1.
+	for i := 0; i < numWorkflows; i++ {
+		s.EventuallyWithT(func(t *assert.CollectT) {
+			s.verifyWorkflowVersioning(wfVarsV1[i], vbUnpinned, tv1.Deployment(), nil, nil)
+		}, 60*time.Second, 100*time.Millisecond)
+	}
+}
 func (s *Versioning3Suite) TestWorkflowTQLags_DependentActivityStartsTransition() {
 	if !s.useNewDeploymentData {
 		s.T().Skip("This test is only supported on new deployment data")