Task queue config for whole-queue concurrency limit

openapi/openapiv2.json

@@ -11373,16 +11373,29 @@
       "default": "RUNTIME_TYPE_UNSPECIFIED",
       "description": " - RUNTIME_TYPE_UNSPECIFIED: Should never actually be set, exists to follow convention of having a default.\nSDKs should just leave `runtimes` empty if none can be determined.\n - RUNTIME_TYPE_GRAAL_AOT: Report when the application is compiled ahead-of-time as a GraalVM native image.\n - RUNTIME_TYPE_KOTLIN: Report when any registered Workflow or Activity implementation is a Kotlin class."
     },
+    "UpdateTaskQueueConfigRequestConcurrencyLimitUpdate": {
+      "type": "object",
+      "properties": {
+        "concurrencyLimit": {
+          "$ref": "#/definitions/v1ConcurrencyLimit",
+          "title": "Limit to be updated"
+        },
+        "reason": {
+          "type": "string",
+          "title": "Reason why the limit was updated"
+        }
+      }
+    },
     "UpdateTaskQueueConfigRequestRateLimitUpdate": {
       "type": "object",
       "properties": {
         "rateLimit": {
           "$ref": "#/definitions/v1RateLimit",
-          "title": "Rate Limit to be updated"
+          "title": "Rate limit to be updated"
         },
         "reason": {
           "type": "string",
-          "description": "Reason for why the rate limit was set."
+          "title": "Reason why the rate limit was updated"
         }
       }
     },
@@ -13281,6 +13294,10 @@
             "type": "string"
           },
           "description": "If set, removes any existing fairness weight overrides for each specified fairness key.\nFairness weights for corresponding keys fall back to the values set during task creation (if any),\nor to the default weight of 1.0."
+        },
+        "updateQueueConcurrencyLimit": {
+          "$ref": "#/definitions/UpdateTaskQueueConfigRequestConcurrencyLimitUpdate",
+          "description": "Update queue-wide concurrency limit.\nIf not set, this configuration is unchanged.\nIf the `concurrency_limit` field in the `ConcurrencyLimitUpdate` is missing, remove the existing limit."
         }
       }
     },
@@ -15237,6 +15254,27 @@
       },
       "description": "ComputeStatus represents compute-related configuration and health for a Worker Deployment Version."
     },
+    "v1ConcurrencyLimit": {
+      "type": "object",
+      "properties": {
+        "concurrentTasks": {
+          "type": "integer",
+          "format": "int32",
+          "description": "Maximum number of concurrent tasks that may be running at a time across the whole task\nqueue. Note that in situations like network partitions, the server's idea of what is\nrunning on workers may be out of date, and this limit could be exceeded from the point\nof view of workers."
+        }
+      }
+    },
+    "v1ConcurrencyLimitConfig": {
+      "type": "object",
+      "properties": {
+        "concurrencyLimit": {
+          "$ref": "#/definitions/v1ConcurrencyLimit"
+        },
+        "metadata": {
+          "$ref": "#/definitions/v1ConfigMetadata"
+        }
+      }
+    },
     "v1ConfigMetadata": {
       "type": "object",
       "properties": {
@@ -20353,6 +20391,10 @@
             "format": "float"
           },
           "description": "If set, overrides the fairness weights for the corresponding fairness keys."
+        },
+        "queueConcurrencyLimit": {
+          "$ref": "#/definitions/v1ConcurrencyLimitConfig",
+          "description": "Concurrency limit for the whole queue."
         }
       }
     },
@@ -20407,7 +20449,10 @@
         },
         "rateLimitingActive": {
           "type": "boolean",
-          "description": "Whether rate limiting blocked any dispatches within the recent observation window (approximately\n30 seconds). When true, adding more workers will not increase throughput — the bottleneck is the\nrate limit, not worker count. This field is useful for auto-scaling systems to avoid unnecessary\nscale-up."
+          "description": "Whether rate limiting or flow control (concurrency limiting, circuit breaker, etc.) blocked any\ndispatches within the recent observation window (approximately 30 seconds). When true, adding more\nworkers will not increase throughput — the bottleneck is the rate limit, not worker count. This\nfield is useful for auto-scaling systems to avoid unnecessary scale-up."
+        },
+        "flowControlActive": {
+          "type": "boolean"
         }
       },
       "description": "TaskQueueStats contains statistics about task queue backlog and activity.\n\nFor workflow task queue type, this result is partial because tasks sent to sticky queues are not included. Read\ncomments above each metric to understand the impact of sticky queue exclusion on that metric accuracy."

openapi/openapiv3.yaml

@@ -11384,6 +11384,24 @@ components:
       description: |-
         ProviderValidationStatus represents the result of the most recent
          connectivity check between Temporal and a customer's compute provider.
+    ConcurrencyLimit:
+      type: object
+      properties:
+        concurrentTasks:
+          type: integer
+          description: |-
+            Maximum number of concurrent tasks that may be running at a time across the whole task
+             queue. Note that in situations like network partitions, the server's idea of what is
+             running on workers may be out of date, and this limit could be exceeded from the point
+             of view of workers.
+          format: int32
+    ConcurrencyLimitConfig:
+      type: object
+      properties:
+        concurrencyLimit:
+          $ref: '#/components/schemas/ConcurrencyLimit'
+        metadata:
+          $ref: '#/components/schemas/ConfigMetadata'
     ConfigMetadata:
       type: object
       properties:
@@ -18023,6 +18041,10 @@ components:
             type: number
             format: float
           description: If set, overrides the fairness weights for the corresponding fairness keys.
+        queueConcurrencyLimit:
+          allOf:
+            - $ref: '#/components/schemas/ConcurrencyLimitConfig'
+          description: Concurrency limit for the whole queue.
     TaskQueueReachability:
       type: object
       properties:
@@ -18105,10 +18127,12 @@ components:
         rateLimitingActive:
           type: boolean
           description: |-
-            Whether rate limiting blocked any dispatches within the recent observation window (approximately
-             30 seconds). When true, adding more workers will not increase throughput — the bottleneck is the
-             rate limit, not worker count. This field is useful for auto-scaling systems to avoid unnecessary
-             scale-up.
+            Whether rate limiting or flow control (concurrency limiting, circuit breaker, etc.) blocked any
+             dispatches within the recent observation window (approximately 30 seconds). When true, adding more
+             workers will not increase throughput — the bottleneck is the rate limit, not worker count. This
+             field is useful for auto-scaling systems to avoid unnecessary scale-up.
+        flowControlActive:
+          type: boolean
       description: |-
         TaskQueueStats contains statistics about task queue backlog and activity.
 
@@ -18907,16 +18931,33 @@ components:
             If set, removes any existing fairness weight overrides for each specified fairness key.
              Fairness weights for corresponding keys fall back to the values set during task creation (if any),
              or to the default weight of 1.0.
+        updateQueueConcurrencyLimit:
+          allOf:
+            - $ref: '#/components/schemas/UpdateTaskQueueConfigRequest_ConcurrencyLimitUpdate'
+          description: |-
+            Update queue-wide concurrency limit.
+             If not set, this configuration is unchanged.
+             If the `concurrency_limit` field in the `ConcurrencyLimitUpdate` is missing, remove the existing limit.
+    UpdateTaskQueueConfigRequest_ConcurrencyLimitUpdate:
+      type: object
+      properties:
+        concurrencyLimit:
+          allOf:
+            - $ref: '#/components/schemas/ConcurrencyLimit'
+          description: Limit to be updated
+        reason:
+          type: string
+          description: Reason why the limit was updated
     UpdateTaskQueueConfigRequest_RateLimitUpdate:
       type: object
       properties:
         rateLimit:
           allOf:
             - $ref: '#/components/schemas/RateLimit'
-          description: Rate Limit to be updated
+          description: Rate limit to be updated
         reason:
           type: string
-          description: Reason for why the rate limit was set.
+          description: Reason why the rate limit was updated
     UpdateTaskQueueConfigResponse:
       type: object
       properties:

temporal/api/taskqueue/v1/message.proto

@@ -144,11 +144,12 @@ message TaskQueueStats {
     //   worker instance.
     float tasks_dispatch_rate = 4;
 
-    // Whether rate limiting blocked any dispatches within the recent observation window (approximately
-    // 30 seconds). When true, adding more workers will not increase throughput — the bottleneck is the
-    // rate limit, not worker count. This field is useful for auto-scaling systems to avoid unnecessary
-    // scale-up.
+    // Whether rate limiting or flow control (concurrency limiting, circuit breaker, etc.) blocked any
+    // dispatches within the recent observation window (approximately 30 seconds). When true, adding more
+    // workers will not increase throughput — the bottleneck is the rate limit, not worker count. This
+    // field is useful for auto-scaling systems to avoid unnecessary scale-up.
     bool rate_limiting_active = 5;
+    bool flow_control_active = 6;
 }
 
 // Deprecated. Use `InternalTaskQueueStatus`. This is kept until `DescribeTaskQueue` supports legacy behavior.
@@ -337,25 +338,38 @@ message PollerScalingDecision {
   int32 poll_request_delta_suggestion = 1;
 }
 
-message RateLimit { 
+message RateLimit {
     // Zero is a valid rate limit.
     float requests_per_second = 1;
 }
 
+message ConcurrencyLimit {
+    // Maximum number of concurrent tasks that may be running at a time across the whole task
+    // queue. Note that in situations like network partitions, the server's idea of what is
+    // running on workers may be out of date, and this limit could be exceeded from the point
+    // of view of workers.
+    int32 concurrent_tasks = 1;
+}
+
 message ConfigMetadata {
     // Reason for why the config was set.
     string reason = 1;
 
     // Identity of the last updater.
     // Set by the request's identity field.
     string update_identity = 2;
-    
+
     // Time of the last update.
     google.protobuf.Timestamp update_time = 3;
 }
 
-message RateLimitConfig { 
-    RateLimit rate_limit = 1;    
+message RateLimitConfig {
+    RateLimit rate_limit = 1;
+    ConfigMetadata metadata = 2;
+}
+
+message ConcurrencyLimitConfig {
+    ConcurrencyLimit concurrency_limit = 1;
     ConfigMetadata metadata = 2;
 }
 
@@ -366,4 +380,6 @@ message TaskQueueConfig {
     RateLimitConfig fairness_keys_rate_limit_default = 2;
     // If set, overrides the fairness weights for the corresponding fairness keys.
     map<string, float> fairness_weight_overrides = 3;
+    // Concurrency limit for the whole queue.
+    ConcurrencyLimitConfig queue_concurrency_limit = 4;
 }

temporal/api/workflowservice/v1/request_response.proto

@@ -3058,9 +3058,16 @@ message ListWorkersResponse {
 
 message UpdateTaskQueueConfigRequest {
     message RateLimitUpdate {
-        // Rate Limit to be updated
+        // Rate limit to be updated
         temporal.api.taskqueue.v1.RateLimit rate_limit = 1;
-        // Reason for why the rate limit was set.
+        // Reason why the rate limit was updated
+        string reason = 2;
+    }
+
+    message ConcurrencyLimitUpdate {
+        // Limit to be updated
+        temporal.api.taskqueue.v1.ConcurrencyLimit concurrency_limit = 1;
+        // Reason why the limit was updated
         string reason = 2;
     }
 
@@ -3085,6 +3092,10 @@ message UpdateTaskQueueConfigRequest {
     // Fairness weights for corresponding keys fall back to the values set during task creation (if any),
     // or to the default weight of 1.0.
     repeated string unset_fairness_weight_overrides = 8;
+    // Update queue-wide concurrency limit.
+    // If not set, this configuration is unchanged.
+    // If the `concurrency_limit` field in the `ConcurrencyLimitUpdate` is missing, remove the existing limit.
+    ConcurrencyLimitUpdate update_queue_concurrency_limit = 9;
 }
 
 message UpdateTaskQueueConfigResponse {