Keep syncing User Data as long as version changes (temporalio#7320)

## What changed?
<!-- Describe what has changed in this PR -->

UserDataManger in child partitions constantly fetches data from the
parent partition. There is a safely mechanism to prevent busy loop in
case the parent returns result immediately instead of only when the data
is updated.
The safety mechanism did not work well with Versioning 3 because many
updates are involved in a single tests. With this update we relax the
safety check to only apply when the user data version remains the same
and continue to fetch the data asap as long as each time it gets new
data.

Here, also I'm relaxing the "requested task queue user data for version
greater than known version" errors to be thrown only for the root
partition, as they can normally happen for child partitions.

## Why?
<!-- Tell your future self why have you made these changes -->
So that back-to-back Versioning 3 APIs do not have to wait for a long
time.

## How did you test it?
<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
Existing tests.

## Potential risks
<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->

## Documentation
<!-- Have you made sure this change doesn't falsify anything currently
stated in `docs/`? If significant
new behavior is added, have you described that in `docs/`? -->

## Is hotfix candidate?
<!-- Is this PR a hotfix candidate or does it require a notification to
be sent to the broader community? (Yes/No) -->
Yes.

Author: ShahabT

config/dynamicconfig/development-cass.yaml

@@ -40,6 +40,10 @@ system.enableDeployments:
   - value: true
 system.enableDeploymentVersions:
   - value: true
+frontend.workerVersioningRuleAPIs:
+  - value: true
+frontend.enableExecuteMultiOperation:
+  - value: true
 system.enableNexus:
   - value: true
 component.nexusoperations.callback.endpoint.template:

config/dynamicconfig/development-sql.yaml

@@ -39,6 +39,14 @@ frontend.workerVersioningDataAPIs:
   - value: true
 frontend.workerVersioningWorkflowAPIs:
   - value: true
+frontend.workerVersioningRuleAPIs:
+  - value: true
+system.enableDeploymentVersions:
+  - value: true
+system.enableDeployments:
+  - value: true
+frontend.enableExecuteMultiOperation:
+  - value: true
 system.enableNexus:
   - value: true
 component.nexusoperations.callback.endpoint.template:

service/matching/user_data_manager.go

@@ -58,7 +58,7 @@ const (
 	userDataClosed
 )
 
-const maxFastUserDataFetches = 10
+const maxFastUserDataFetches = 5
 
 type (
 	userDataManager interface {
@@ -288,9 +288,11 @@ func (m *userDataManagerImpl) fetchUserData(ctx context.Context) error {
 	// hasFetchedUserData is true if we have gotten a successful reply to GetTaskQueueUserData.
 	// It's used to control whether we do a long poll or a simple get.
 	hasFetchedUserData := false
+	userDataVersionChanged := false
 
 	op := func(ctx context.Context) error {
 		knownUserData, _, _ := m.GetUserData()
+		userDataVersionChanged = false
 
 		callCtx, cancel := context.WithTimeout(ctx, m.config.GetUserDataLongPollTimeout())
 		defer cancel()
@@ -323,6 +325,7 @@ func (m *userDataManagerImpl) fetchUserData(ctx context.Context) error {
 		// nil inner fields.
 		if res.GetUserData() != nil {
 			m.setUserDataForNonOwningPartition(res.GetUserData())
+			userDataVersionChanged = res.GetUserData().GetVersion() != knownUserData.GetVersion()
 			m.logNewUserData("fetched user data from parent", res.GetUserData())
 		} else {
 			m.logger.Debug("fetched user data from parent, no change")
@@ -344,7 +347,8 @@ func (m *userDataManagerImpl) fetchUserData(ctx context.Context) error {
 		// one. But if the remote is broken and returns success immediately, we might end up
 		// spinning. So enforce a minimum wait time that increases as long as we keep getting
 		// very fast replies.
-		if elapsed < m.config.GetUserDataMinWaitTime {
+		// If the user data version changed it means new data was received so we skip this check.
+		if !userDataVersionChanged && elapsed < m.config.GetUserDataMinWaitTime {
 			if fastResponseCounter >= maxFastUserDataFetches {
 				// maxFastUserDataFetches or more consecutive fast responses, let's throttle!
 				util.InterruptibleSleep(ctx, minWaitTime-elapsed)
@@ -575,7 +579,28 @@ func (m *userDataManagerImpl) HandleGetUserDataRequest(
 		} else if err != nil {
 			return nil, err
 		}
-		if req.WaitNewData && userData.GetVersion() == version {
+		if userData.GetVersion() > version {
+			resp.UserData = userData
+			m.logger.Info("returning user data",
+				tag.NewBoolTag("long-poll", req.WaitNewData),
+				tag.NewInt64("request-known-version", version),
+				tag.UserDataVersion(userData.Version),
+			)
+		} else if userData != nil && userData.Version < version && m.store != nil {
+			// When m.store == nil it means this is a non-owner partition, so it is possible
+			// for the requested version to be greater than the known version if there are
+			// concurrent user data updates in flight. We do not log an error in that case.
+
+			// This is highly unlikely to happen in the owner/root partition but may happen
+			// due to an edge case in during ownership transfer.
+			// We rely on client retries in this case to let the system eventually self-heal.
+			m.logger.Error("requested task queue user data for version greater than known version",
+				tag.NewInt64("request-known-version", version),
+				tag.UserDataVersion(userData.Version),
+			)
+			return nil, errRequestedVersionTooLarge
+		}
+		if req.WaitNewData && userData.GetVersion() <= version {
 			// long-poll: wait for data to change/appear
 			select {
 			case <-ctx.Done():
@@ -590,31 +615,7 @@ func (m *userDataManagerImpl) HandleGetUserDataRequest(
 				continue
 			}
 		}
-		if userData != nil {
-			if userData.Version > version {
-				resp.UserData = userData
-				m.logger.Info("returning user data",
-					tag.NewBoolTag("long-poll", req.WaitNewData),
-					tag.NewInt64("request-known-version", version),
-					tag.UserDataVersion(userData.Version),
-				)
-			} else if userData.Version < version {
-				if m.store != nil {
-					// When m.store == nil it means this is a non-owner partition, so it is possible
-					// for the requested version to be greater than the known version if there are
-					// concurrent user data updates in flight. We do not log an error in that case.
-
-					// This is highly unlikely to happen in the owner/root partition but may happen
-					// due to an edge case in during ownership transfer.
-					// We rely on client retries in this case to let the system eventually self-heal.
-					m.logger.Error("requested task queue user data for version greater than known version",
-						tag.NewInt64("request-known-version", version),
-						tag.UserDataVersion(userData.Version),
-					)
-				}
-				return nil, errRequestedVersionTooLarge
-			}
-		} else {
+		if userData == nil {
 			m.logger.Debug("returning empty user data (no data)", tag.NewBoolTag("long-poll", req.WaitNewData))
 		}
 		return resp, nil

service/matching/user_data_manager_test.go

@@ -331,7 +331,7 @@ func TestUserData_FetchesOnInit(t *testing.T) {
 		}).
 		Return(&matchingservice.GetTaskQueueUserDataResponse{
 			UserData: data1,
-		}, nil).MaxTimes(maxFastUserDataFetches)
+		}, nil).MaxTimes(maxFastUserDataFetches + 1)
 
 	m := createUserDataManager(t, controller, tqCfg)
 	m.config.GetUserDataMinWaitTime = 10 * time.Second // only one fetch
@@ -468,7 +468,9 @@ func TestUserData_RetriesFetchOnUnavailable(t *testing.T) {
 		}).
 		Return(&matchingservice.GetTaskQueueUserDataResponse{
 			UserData: data1,
-		}, nil).MaxTimes(maxFastUserDataFetches)
+		}, nil).
+		// +3 because the counter resets when version changes so the calls with error do not count
+		MaxTimes(maxFastUserDataFetches + 3)
 
 	m := createUserDataManager(t, controller, tqCfg)
 	m.config.GetUserDataMinWaitTime = 10 * time.Second // wait on success
@@ -552,7 +554,9 @@ func TestUserData_RetriesFetchOnUnImplemented(t *testing.T) {
 		}).
 		Return(&matchingservice.GetTaskQueueUserDataResponse{
 			UserData: data1,
-		}, nil).MaxTimes(maxFastUserDataFetches)
+		}, nil).
+		// +3 because the counter resets when version changes so the calls with error do not count
+		MaxTimes(maxFastUserDataFetches + 3)
 
 	m := createUserDataManager(t, controller, tqCfg)
 	m.config.GetUserDataMinWaitTime = 10 * time.Second // wait on success
@@ -621,7 +625,7 @@ func TestUserData_FetchesUpTree(t *testing.T) {
 		}).
 		Return(&matchingservice.GetTaskQueueUserDataResponse{
 			UserData: data1,
-		}, nil).MaxTimes(maxFastUserDataFetches)
+		}, nil).MaxTimes(maxFastUserDataFetches + 1)
 
 	m := createUserDataManager(t, controller, tqCfg)
 	m.config.GetUserDataMinWaitTime = 10 * time.Second // wait on success
@@ -672,7 +676,7 @@ func TestUserData_FetchesActivityToWorkflow(t *testing.T) {
 		}).
 		Return(&matchingservice.GetTaskQueueUserDataResponse{
 			UserData: data1,
-		}, nil).MaxTimes(maxFastUserDataFetches)
+		}, nil).MaxTimes(maxFastUserDataFetches + 1)
 
 	m := createUserDataManager(t, controller, tqCfg)
 	m.config.GetUserDataMinWaitTime = 10 * time.Second // wait on success
@@ -727,7 +731,7 @@ func TestUserData_FetchesStickyToNormal(t *testing.T) {
 		}).
 		Return(&matchingservice.GetTaskQueueUserDataResponse{
 			UserData: data1,
-		}, nil).MaxTimes(maxFastUserDataFetches)
+		}, nil).MaxTimes(maxFastUserDataFetches + 1)
 
 	m := createUserDataManager(t, controller, tqCfg)
 	m.config.GetUserDataMinWaitTime = 10 * time.Second // wait on success