Fix max deployment count check to not block new versions

Author: ShahabT

service/matching/physical_task_queue_manager.go

@@ -566,10 +566,13 @@ func (c *physicalTaskQueueManagerImpl) ensureRegisteredInDeploymentVersion(
 		}
 		var errMaxTaskQueuesInVersion workerdeployment.ErrMaxTaskQueuesInVersion
 		var errMaxVersionsInDeployment workerdeployment.ErrMaxVersionsInDeployment
+		var errMaxDeploymentsInNamespace workerdeployment.ErrMaxDeploymentsInNamespace
 		if errors.As(err, &errMaxTaskQueuesInVersion) {
 			err = errMaxTaskQueuesInVersion
 		} else if errors.As(err, &errMaxVersionsInDeployment) {
 			err = errMaxVersionsInDeployment
+		} else if errors.As(err, &errMaxDeploymentsInNamespace) {
+			err = errMaxDeploymentsInNamespace
 		} else {
 			// Do not surface low level error to user
 			c.logger.Error("error while registering version", tag.Error(err))

service/worker/workerdeployment/client.go

@@ -179,6 +179,7 @@ type Client interface {
 
 type ErrMaxTaskQueuesInVersion struct{ error }
 type ErrMaxVersionsInDeployment struct{ error }
+type ErrMaxDeploymentsInNamespace struct{ error }
 type ErrRegister struct{ error }
 
 // ClientImpl implements Client
@@ -791,15 +792,6 @@ func (d *ClientImpl) StartWorkerDeployment(
 	//revive:disable-next-line:defer
 	defer d.record("StartWorkerDeployment", &retErr, namespaceEntry.Name(), deploymentName, identity)()
 
-	// TODO (Carly): either max page size or default page size is 1000, so if there are > 1000 this would not catch it
-	deps, _, err := d.ListWorkerDeployments(ctx, namespaceEntry, 0, nil)
-	if err != nil {
-		return err
-	}
-	if len(deps) >= d.maxDeployments(namespaceEntry.Name().String()) {
-		return serviceerror.NewFailedPrecondition("maximum deployments in namespace, delete manually to continue deploying.")
-	}
-
 	workflowID := worker_versioning.GenerateDeploymentWorkflowID(deploymentName)
 
 	input, err := sdk.PreferProtoDataConverter.ToPayloads(&deploymentspb.WorkerDeploymentWorkflowArgs{
@@ -1020,13 +1012,20 @@ func (d *ClientImpl) updateWithStartWorkerDeployment(
 
 	workflowID := worker_versioning.GenerateDeploymentWorkflowID(deploymentName)
 
-	// TODO (Carly): either max page size or default page size is 1000, so if there are > 1000 this would not catch it
-	deps, _, err := d.ListWorkerDeployments(ctx, namespaceEntry, 0, nil)
+	_, _, err = d.DescribeWorkerDeployment(ctx, namespaceEntry, deploymentName)
 	if err != nil {
-		return nil, err
-	}
-	if len(deps) >= d.maxDeployments(namespaceEntry.Name().String()) {
-		return nil, serviceerror.NewFailedPrecondition("maximum deployments in namespace, delete manually to continue deploying.")
+		var notFound *serviceerror.NotFound
+		if errors.As(err, &notFound) {
+			// New deployment, make sure we're not exceeding the limit
+			count, err := d.countWorkerDeployments(ctx, namespaceEntry)
+			if err != nil {
+				return nil, err
+			}
+			limit := d.maxDeployments(namespaceEntry.Name().String())
+			if count >= int64(limit) {
+				return nil, ErrMaxDeploymentsInNamespace{error: errors.New(fmt.Sprintf("reached maximum deployments in namespace (%d)", limit))}
+			}
+		}
 	}
 
 	input, err := sdk.PreferProtoDataConverter.ToPayloads(&deploymentspb.WorkerDeploymentWorkflowArgs{
@@ -1054,6 +1053,26 @@ func (d *ClientImpl) updateWithStartWorkerDeployment(
 	)
 }
 
+func (d *ClientImpl) countWorkerDeployments(
+	ctx context.Context,
+	namespaceEntry *namespace.Namespace,
+) (count int64, retError error) {
+	query := WorkerDeploymentVisibilityBaseListQuery
+
+	persistenceResp, err := d.visibilityManager.CountWorkflowExecutions(
+		ctx,
+		&manager.CountWorkflowExecutionsRequest{
+			NamespaceID: namespaceEntry.ID(),
+			Namespace:   namespaceEntry.Name(),
+			Query:       query,
+		},
+	)
+	if err != nil {
+		return 0, err
+	}
+	return persistenceResp.Count, nil
+}
+
 func (d *ClientImpl) updateWithStartWorkerDeploymentVersion(
 	ctx context.Context,
 	namespaceEntry *namespace.Namespace,

tests/worker_deployment_test.go

@@ -258,7 +258,7 @@ func (s *WorkerDeploymentSuite) TestNamespaceDeploymentsLimit() {
 	s.ensureCreateVersionInDeployment(tv)
 
 	// pollers of the second deployment version should be rejected
-	s.pollFromDeploymentExpectFail(ctx, tv.WithDeploymentSeriesNumber(2), "Namespace deployments limit reached")
+	s.pollFromDeploymentExpectFail(ctx, tv.WithDeploymentSeriesNumber(2), "reached maximum deployments in namespace (1)")
 }
 
 func (s *WorkerDeploymentSuite) TestDescribeWorkerDeployment_TwoVersions_Sorted() {