Convert ResourceExhausted errors from versioning API's to Internal ones with nice error message (#7815)

## What changed?
- WISOTT

## Why?
- Better user experience

## How did you test it?
- [x] built
- [x] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [x] added new functional test(s)

## Potential risks
- None

Author: Shivs11

service/frontend/workflow_handler.go

@@ -87,11 +87,18 @@ var (
 	maxTime = time.Date(2100, 1, 1, 1, 0, 0, 0, time.UTC)
 
 	// Tail room for context deadline to bail out from retry for long poll.
-	longPollTailRoom = time.Second
-
+	longPollTailRoom  = time.Second
 	errWaitForRefresh = serviceerror.NewDeadlineExceeded("waiting for schedule to refresh status of completed workflows")
 )
 
+const (
+	errTooManySetCurrentVersionRequests = "Too many SetWorkerDeploymentCurrentVersion requests have been issued in rapid succession. Please throttle the request rate to avoid exceeding Worker Deployment resource limits."
+	errTooManySetRampingVersionRequests = "Too many SetWorkerDeploymentRampingVersion requests have been issued in rapid succession. Please throttle the request rate to avoid exceeding Worker Deployment resource limits."
+	errTooManyDeleteDeploymentRequests  = "Too many DeleteWorkerDeployment requests have been issued in rapid succession. Please throttle the request rate to avoid exceeding Worker Deployment resource limits."
+	errTooManyDeleteVersionRequests     = "Too many DeleteWorkerDeploymentVersion requests have been issued in rapid succession. Please throttle the request rate to avoid exceeding Worker Deployment resource limits."
+	errTooManyVersionMetadataRequests   = "Too many UpdateWorkerDeploymentVersionMetadata requests have been issued in rapid succession. Please throttle the request rate to avoid exceeding Worker Deployment resource limits."
+)
+
 type (
 	// WorkflowHandler - gRPC handler interface for workflowservice
 	WorkflowHandler struct {
@@ -3414,6 +3421,9 @@ func (wh *WorkflowHandler) SetWorkerDeploymentCurrentVersion(ctx context.Context
 
 	resp, err := wh.workerDeploymentClient.SetCurrentVersion(ctx, namespaceEntry, request.DeploymentName, versionStr, request.Identity, request.IgnoreMissingTaskQueues, request.GetConflictToken())
 	if err != nil {
+		if common.IsResourceExhausted(err) {
+			return nil, serviceerror.NewResourceExhaustedf(enumspb.RESOURCE_EXHAUSTED_CAUSE_BUSY_WORKFLOW, errTooManySetCurrentVersionRequests)
+		}
 		return nil, err
 	}
 
@@ -3470,6 +3480,9 @@ func (wh *WorkflowHandler) SetWorkerDeploymentRampingVersion(ctx context.Context
 
 	resp, err := wh.workerDeploymentClient.SetRampingVersion(ctx, namespaceEntry, request.DeploymentName, versionStr, request.GetPercentage(), request.GetIdentity(), request.IgnoreMissingTaskQueues, request.GetConflictToken())
 	if err != nil {
+		if common.IsResourceExhausted(err) {
+			return nil, serviceerror.NewResourceExhaustedf(enumspb.RESOURCE_EXHAUSTED_CAUSE_BUSY_WORKFLOW, errTooManySetRampingVersionRequests)
+		}
 		return nil, err
 	}
 
@@ -3574,6 +3587,9 @@ func (wh *WorkflowHandler) DeleteWorkerDeployment(ctx context.Context, request *
 
 	err = wh.workerDeploymentClient.DeleteWorkerDeployment(ctx, namespaceEntry, request.DeploymentName, request.Identity)
 	if err != nil {
+		if common.IsResourceExhausted(err) {
+			return nil, serviceerror.NewResourceExhaustedf(enumspb.RESOURCE_EXHAUSTED_CAUSE_BUSY_WORKFLOW, errTooManyDeleteDeploymentRequests)
+		}
 		return nil, err
 	}
 
@@ -3600,6 +3616,9 @@ func (wh *WorkflowHandler) DeleteWorkerDeploymentVersion(ctx context.Context, re
 
 	err = wh.workerDeploymentClient.DeleteWorkerDeploymentVersion(ctx, namespaceEntry, versionStr, request.SkipDrainage, request.Identity)
 	if err != nil {
+		if common.IsResourceExhausted(err) {
+			return nil, serviceerror.NewResourceExhaustedf(enumspb.RESOURCE_EXHAUSTED_CAUSE_BUSY_WORKFLOW, errTooManyDeleteVersionRequests)
+		}
 		return nil, err
 	}
 
@@ -3628,10 +3647,12 @@ func (wh *WorkflowHandler) UpdateWorkerDeploymentVersionMetadata(ctx context.Con
 		versionStr = worker_versioning.ExternalWorkerDeploymentVersionToString(request.GetDeploymentVersion())
 	}
 
-	// todo (Shivam): Should we get identity from the request?
 	identity := uuid.New()
 	updatedMetadata, err := wh.workerDeploymentClient.UpdateVersionMetadata(ctx, namespaceEntry, versionStr, request.UpsertEntries, request.RemoveEntries, identity)
 	if err != nil {
+		if common.IsResourceExhausted(err) {
+			return nil, serviceerror.NewResourceExhaustedf(enumspb.RESOURCE_EXHAUSTED_CAUSE_BUSY_WORKFLOW, errTooManyVersionMetadataRequests)
+		}
 		return nil, err
 	}
 

service/matching/physical_task_queue_manager.go

@@ -427,9 +427,6 @@ func (c *physicalTaskQueueManagerImpl) DispatchNexusTask(
 }
 
 func (c *physicalTaskQueueManagerImpl) UpdatePollerInfo(id pollerIdentity, pollMetadata *pollMetadata) {
-	if c.queue.Version().IsVersioned() {
-		fmt.Printf("poll info updated in %v\n", c.queue.Version())
-	}
 	c.pollerHistory.updatePollerInfo(id, pollMetadata)
 }
 

service/worker/workerdeployment/client.go

@@ -1572,10 +1572,3 @@ func (d *ClientImpl) getSyncBatchSize() int32 {
 	}
 	return syncBatchSize
 }
-
-// isFailedPrecondition checks if the error is a FailedPrecondition error. It also checks if the FailedPrecondition error is wrapped in an ApplicationError.
-func isFailedPrecondition(err error) bool {
-	var failedPreconditionError *serviceerror.FailedPrecondition
-	var applicationError *temporal.ApplicationError
-	return errors.As(err, &failedPreconditionError) || (errors.As(err, &applicationError) && applicationError.Type() == errFailedPrecondition)
-}

service/worker/workerdeployment/util.go

@@ -1,6 +1,7 @@
 package workerdeployment
 
 import (
+	"errors"
 	"fmt"
 	"strings"
 	"time"
@@ -145,3 +146,10 @@ func getSafeDurationConfig(ctx workflow.Context, id string, unsafeGetter func()
 func durationEq(a, b any) bool {
 	return a == b
 }
+
+// isFailedPrecondition checks if the error is a FailedPrecondition error. It also checks if the FailedPrecondition error is wrapped in an ApplicationError.
+func isFailedPrecondition(err error) bool {
+	var failedPreconditionError *serviceerror.FailedPrecondition
+	var applicationError *temporal.ApplicationError
+	return errors.As(err, &failedPreconditionError) || (errors.As(err, &applicationError) && applicationError.Type() == errFailedPrecondition)
+}

service/worker/workerdeployment/version_workflow.go

@@ -425,18 +425,6 @@ func (d *VersionWorkflowRunner) handleRegisterWorker(ctx workflow.Context, args
 		d.lock.Unlock()
 	}()
 
-	v := workflow.GetVersion(ctx, "RefactorRegisterWorker", workflow.DefaultVersion, 1)
-	if v == workflow.DefaultVersion {
-		// wait until deployment workflow started
-		err = workflow.Await(ctx, func() bool { return d.VersionState.StartedDeploymentWorkflow })
-		if err != nil {
-			d.logger.Error("Update canceled before deployment workflow started")
-			// TODO (Carly): This is likely due to too many deployments, but make sure we excluded other possible errors here and send a proper error message all the time.
-			// TODO (Carly): mention the limit in here or make sure matching does in the error returned to the poller
-			return temporal.NewApplicationError("failed to create deployment version, likely you are exceeding the limit of allowed deployments in a namespace", errTooManyDeployments)
-		}
-	}
-
 	// Add the task queue to the local state.
 	if d.VersionState.TaskQueueFamilies == nil {
 		d.VersionState.TaskQueueFamilies = make(map[string]*deploymentspb.VersionLocalState_TaskQueueFamilyData)
@@ -459,23 +447,6 @@ func (d *VersionWorkflowRunner) handleRegisterWorker(ctx workflow.Context, args
 	}
 
 	activityCtx := workflow.WithActivityOptions(ctx, defaultActivityOptions)
-
-	if v == workflow.DefaultVersion {
-		// First try to add version to worker-deployment workflow so it rejects in case we hit the limit
-		err = workflow.ExecuteActivity(activityCtx, d.a.AddVersionToWorkerDeployment, &deploymentspb.AddVersionToWorkerDeploymentRequest{
-			DeploymentName: d.VersionState.Version.GetDeploymentName(),
-			UpdateArgs: &deploymentspb.AddVersionUpdateArgs{
-				Version:    worker_versioning.WorkerDeploymentVersionToString(d.VersionState.Version),
-				CreateTime: d.VersionState.CreateTime,
-			},
-			RequestId: d.newUUID(ctx),
-		}).Get(ctx, nil)
-		if err != nil {
-			// TODO (carly): make sure the error message that goes to the user is informative and has the limit mentioned
-			return temporal.NewApplicationError("too many versions in this deployment", errTooManyVersions)
-		}
-	}
-
 	// sync to user data
 	var syncRes deploymentspb.SyncDeploymentVersionUserDataResponse
 	err = workflow.ExecuteActivity(activityCtx, d.a.SyncDeploymentVersionUserData, &deploymentspb.SyncDeploymentVersionUserDataRequest{
@@ -545,31 +516,46 @@ func (d *VersionWorkflowRunner) handleSyncState(ctx workflow.Context, args *depl
 		Version: state.GetVersion(),
 	}
 
-	v := workflow.GetVersion(ctx, "SyncStateWithBatching", workflow.DefaultVersion, 1)
-	if v == workflow.DefaultVersion {
-		for _, tqName := range workflow.DeterministicKeys(state.TaskQueueFamilies) {
-			byType := state.TaskQueueFamilies[tqName]
-			data := &deploymentspb.DeploymentVersionData{
-				Version:           d.VersionState.Version,
-				RoutingUpdateTime: args.RoutingUpdateTime,
-				CurrentSinceTime:  args.CurrentSinceTime,
-				RampingSinceTime:  args.RampingSinceTime,
-				RampPercentage:    args.RampPercentage,
-			}
-			var types []enumspb.TaskQueueType
-			for _, tqType := range workflow.DeterministicKeys(byType.TaskQueues) {
-				types = append(types, enumspb.TaskQueueType(tqType))
-			}
+	// send in the task-queue families in batches of syncBatchSize
+	batches := make([][]*deploymentspb.SyncDeploymentVersionUserDataRequest_SyncUserData, 0)
+	for _, tqName := range workflow.DeterministicKeys(state.TaskQueueFamilies) {
+		byType := state.TaskQueueFamilies[tqName]
+		data := &deploymentspb.DeploymentVersionData{
+			Version:           d.VersionState.Version,
+			RoutingUpdateTime: args.RoutingUpdateTime,
+			CurrentSinceTime:  args.CurrentSinceTime,
+			RampingSinceTime:  args.RampingSinceTime,
+			RampPercentage:    args.RampPercentage,
+		}
+		var types []enumspb.TaskQueueType
+		for _, tqType := range workflow.DeterministicKeys(byType.TaskQueues) {
+			types = append(types, enumspb.TaskQueueType(tqType))
+		}
 
-			syncReq.Sync = append(syncReq.Sync, &deploymentspb.SyncDeploymentVersionUserDataRequest_SyncUserData{
-				Name:  tqName,
-				Types: types,
-				Data:  data,
-			})
+		syncReq.Sync = append(syncReq.Sync, &deploymentspb.SyncDeploymentVersionUserDataRequest_SyncUserData{
+			Name:  tqName,
+			Types: types,
+			Data:  data,
+		})
+
+		if len(syncReq.Sync) == int(d.VersionState.SyncBatchSize) {
+			batches = append(batches, syncReq.Sync)
+			syncReq.Sync = make([]*deploymentspb.SyncDeploymentVersionUserDataRequest_SyncUserData, 0) // reset the syncReq.Sync slice for the next batch
 		}
+	}
+	if len(syncReq.Sync) > 0 {
+		batches = append(batches, syncReq.Sync)
+	}
+
+	// calling SyncDeploymentVersionUserData for each batch
+	for _, batch := range batches {
 		activityCtx := workflow.WithActivityOptions(ctx, defaultActivityOptions)
 		var syncRes deploymentspb.SyncDeploymentVersionUserDataResponse
-		err = workflow.ExecuteActivity(activityCtx, d.a.SyncDeploymentVersionUserData, syncReq).Get(ctx, &syncRes)
+
+		err = workflow.ExecuteActivity(activityCtx, d.a.SyncDeploymentVersionUserData, &deploymentspb.SyncDeploymentVersionUserDataRequest{
+			Version: state.GetVersion(),
+			Sync:    batch,
+		}).Get(ctx, &syncRes)
 		if err != nil {
 			// TODO (Shivam): Compensation functions required to roll back the local state + activity changes.
 			return nil, err
@@ -587,67 +573,6 @@ func (d *VersionWorkflowRunner) handleSyncState(ctx workflow.Context, args *depl
 				return nil, err
 			}
 		}
-	} else {
-
-		// send in the task-queue families in batches of syncBatchSize
-		batches := make([][]*deploymentspb.SyncDeploymentVersionUserDataRequest_SyncUserData, 0)
-
-		for _, tqName := range workflow.DeterministicKeys(state.TaskQueueFamilies) {
-			byType := state.TaskQueueFamilies[tqName]
-			data := &deploymentspb.DeploymentVersionData{
-				Version:           d.VersionState.Version,
-				RoutingUpdateTime: args.RoutingUpdateTime,
-				CurrentSinceTime:  args.CurrentSinceTime,
-				RampingSinceTime:  args.RampingSinceTime,
-				RampPercentage:    args.RampPercentage,
-			}
-			var types []enumspb.TaskQueueType
-			for _, tqType := range workflow.DeterministicKeys(byType.TaskQueues) {
-				types = append(types, enumspb.TaskQueueType(tqType))
-			}
-
-			syncReq.Sync = append(syncReq.Sync, &deploymentspb.SyncDeploymentVersionUserDataRequest_SyncUserData{
-				Name:  tqName,
-				Types: types,
-				Data:  data,
-			})
-
-			if len(syncReq.Sync) == int(d.VersionState.SyncBatchSize) {
-				batches = append(batches, syncReq.Sync)
-				syncReq.Sync = make([]*deploymentspb.SyncDeploymentVersionUserDataRequest_SyncUserData, 0) // reset the syncReq.Sync slice for the next batch
-			}
-		}
-		if len(syncReq.Sync) > 0 {
-			batches = append(batches, syncReq.Sync)
-		}
-
-		// calling SyncDeploymentVersionUserData for each batch
-		for _, batch := range batches {
-			activityCtx := workflow.WithActivityOptions(ctx, defaultActivityOptions)
-			var syncRes deploymentspb.SyncDeploymentVersionUserDataResponse
-
-			err = workflow.ExecuteActivity(activityCtx, d.a.SyncDeploymentVersionUserData, &deploymentspb.SyncDeploymentVersionUserDataRequest{
-				Version: state.GetVersion(),
-				Sync:    batch,
-			}).Get(ctx, &syncRes)
-			if err != nil {
-				// TODO (Shivam): Compensation functions required to roll back the local state + activity changes.
-				return nil, err
-			}
-			if len(syncRes.TaskQueueMaxVersions) > 0 {
-				// wait for propagation
-				err = workflow.ExecuteActivity(
-					activityCtx,
-					d.a.CheckWorkerDeploymentUserDataPropagation,
-					&deploymentspb.CheckWorkerDeploymentUserDataPropagationRequest{
-						TaskQueueMaxVersions: syncRes.TaskQueueMaxVersions,
-					}).Get(ctx, nil)
-				if err != nil {
-					// TODO (Shivam): Compensation functions required to roll back the local state + activity changes.
-					return nil, err
-				}
-			}
-		}
 	}
 
 	wasAcceptingNewWorkflows := state.GetCurrentSinceTime() != nil || state.GetRampingSinceTime() != nil