Do not send internal errors for workflow trigger due to tenant resource limits

api-contracts/openapi/paths/v1/workflow-runs/workflow_run.yaml

@@ -469,6 +469,12 @@ trigger:
             schema:
               $ref: "../../../components/schemas/_index.yaml#/APIErrors"
         description: Forbidden
+      "429":
+        content:
+          application/json:
+            schema:
+              $ref: "../../../components/schemas/_index.yaml#/APIErrors"
+        description: Resource limit exceeded
     summary: Create workflow run
     tags:
       - Workflow Runs

api/v1/server/handlers/v1/workflow-runs/trigger.go

@@ -79,7 +79,7 @@ func (t *V1WorkflowRunsService) V1WorkflowRunCreate(ctx echo.Context, request ge
 					apierrors.NewAPIErrors(e.Message()),
 				), nil
 			case codes.ResourceExhausted:
-				return gen.V1WorkflowRunCreate400JSONResponse(
+				return gen.V1WorkflowRunCreate429JSONResponse(
 					apierrors.NewAPIErrors(e.Message()),
 				), nil
 			}
@@ -118,6 +118,20 @@ func (t *V1WorkflowRunsService) V1WorkflowRunCreate(ctx echo.Context, request ge
 	}
 
 	if rawWorkflowRun == nil || rawWorkflowRun.WorkflowRun == nil {
+		// check if the tenant has hit its resource limit, which would explain why the run was not created
+		canCreate, trLimit, limitErr := t.config.V1.TenantLimit().CanCreate(
+			ctx.Request().Context(),
+			sqlcv1.LimitResourceTASKRUN,
+			tenantId,
+			1,
+		)
+
+		if limitErr == nil && !canCreate {
+			return gen.V1WorkflowRunCreate429JSONResponse(
+				apierrors.NewAPIErrors(fmt.Sprintf("tenant has reached %d%% of its task runs limit", trLimit)),
+			), nil
+		}
+
 		return nil, fmt.Errorf("rawWorkflowRun not populated, we are likely seeing high latency in creating tasks")
 	}