feat(router): bound the max_parallel_requests wait queue and return 429 on overflow

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/constants.py

@@ -30,8 +30,10 @@ RUNTIME_UPDATABLE_ROUTER_SETTINGS: Final[frozenset[str]] = frozenset(
         "enable_tag_filtering",
         "tag_routing_prefix",
         "optional_pre_call_checks",
+        "default_max_parallel_requests_queue_size",
     }
 )
+NULLABLE_RUNTIME_ROUTER_SETTINGS: Final[frozenset[str]] = frozenset({"default_max_parallel_requests_queue_size"})
 ROUTER_SETTINGS_MANAGED_OUTSIDE_CONFIG: Final[frozenset[str]] = frozenset(
     {
         "model_list",

litellm/proxy/proxy_server.py

@@ -70,6 +70,7 @@ from litellm.constants import (
     LITELLM_SETTINGS_SAFE_DB_OVERRIDES,
     LITELLM_UI_ALLOW_HEADERS,
     LITELLM_UI_SESSION_DURATION,
+    NULLABLE_RUNTIME_ROUTER_SETTINGS,
     RUNTIME_UPDATABLE_ROUTER_SETTINGS,
 )
 from litellm.litellm_core_utils.asyncify import asyncify
@@ -6900,13 +6901,20 @@ class ProxyConfig:
             ):
                 from litellm.utils import _update_dictionary
 
+                db_settings: Final = db_router_settings.param_value
                 db_overlay_deferring_empty_lists_to_config: Final = {
                     k: v
-                    for k, v in db_router_settings.param_value.items()
+                    for k, v in db_settings.items()
                     if not (k in config_router_settings and isinstance(v, list) and len(v) == 0)
                 }
-                combined_router_settings = _update_dictionary(
-                    config_router_settings, db_overlay_deferring_empty_lists_to_config
+                cleared_nullable_settings: Final = MappingProxyType(
+                    {k: None for k in NULLABLE_RUNTIME_ROUTER_SETTINGS if k in db_settings and db_settings[k] is None}
+                )
+                combined_router_settings = MappingProxyType(
+                    {
+                        **_update_dictionary(config_router_settings, db_overlay_deferring_empty_lists_to_config),
+                        **cleared_nullable_settings,
+                    }
                 )
             elif config_router_settings is not None and isinstance(config_router_settings, dict):
                 combined_router_settings = config_router_settings
@@ -17039,7 +17047,7 @@ async def update_config(
             raw_router_settings_without_none: Final = {
                 key: value
                 for key, value in raw_router_settings.items()
-                if key not in typed_router_settings and value is not None
+                if key not in typed_router_settings and (value is not None or key in NULLABLE_RUNTIME_ROUTER_SETTINGS)
             }
             router_settings_updates: Final = {**typed_router_settings, **raw_router_settings_without_none}
             new_router_settings: Final = {**existing, **router_settings_updates}

litellm/router.py

@@ -738,6 +738,7 @@ class Router:
         stream_timeout: float | None = None,
         default_litellm_params: dict | None = None,  # default params for Router.chat.completion.create
         default_max_parallel_requests: int | None = None,
+        default_max_parallel_requests_queue_size: int | None = None,
         set_verbose: bool = False,
         debug_level: Literal["DEBUG", "INFO"] = "INFO",
         default_fallbacks: list[str] | None = None,  # generic fallbacks, works across all deployments
@@ -935,6 +936,7 @@ class Router:
             None  # use this to track the users default deployment, when they want to use model = *
         )
         self.default_max_parallel_requests = default_max_parallel_requests
+        self._default_max_parallel_requests_queue_size = default_max_parallel_requests_queue_size
         self.provider_default_deployment_ids: list[str] = []
         self.pattern_router = PatternMatchRouter()
         self.team_pattern_routers: dict[str, PatternMatchRouter] = {}  # {"TEAM_ID": PatternMatchRouter}
@@ -3630,8 +3632,6 @@ class Router:
             input_kwargs.pop("silent_model", None)
             input_kwargs.pop("include_fallback_errors", None)
 
-            _response: Final = litellm.acompletion(**input_kwargs)
-
             logging_obj: Final[LiteLLMLogging | None] = kwargs.get("litellm_logging_obj", None)
 
             rpm_semaphore: Final = self._get_client(
@@ -3647,7 +3647,7 @@ class Router:
                     logging_obj=logging_obj,
                     parent_otel_span=parent_otel_span,
                 )
-                response = await _response
+                response = await litellm.acompletion(**input_kwargs)
 
                 ## CHECK CONTENT FILTER ERROR ##
                 if isinstance(response, ModelResponse):
@@ -4574,38 +4574,16 @@ class Router:
             )
 
             self.total_calls[model_name] += 1
-            response = litellm.aimage_generation(
-                **{
-                    **data,
-                    "prompt": prompt,
-                    "caching": self.cache_responses,
-                    "client": model_client,
-                    **kwargs,
-                }
-            )
-
-            ### CONCURRENCY-SAFE RPM CHECKS ###
-            rpm_semaphore: Final = self._get_client(
-                deployment=deployment,
-                kwargs=kwargs,
-                client_type="max_parallel_requests",
-            )
-
-            if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore):
-                async with rpm_semaphore:
-                    """
-                    - Check rpm limits before making the call
-                    - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe)
-                    """
-                    await self.async_routing_strategy_pre_call_checks(
-                        deployment=deployment, parent_otel_span=parent_otel_span
-                    )
-                    response = await response
-            else:
-                await self.async_routing_strategy_pre_call_checks(
-                    deployment=deployment, parent_otel_span=parent_otel_span
+            async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span):
+                response = await litellm.aimage_generation(
+                    **{
+                        **data,
+                        "prompt": prompt,
+                        "caching": self.cache_responses,
+                        "client": model_client,
+                        **kwargs,
+                    }
                 )
-                response = await response
 
             self.success_calls[model_name] += 1
             verbose_router_logger.info("litellm.aimage_generation(model=%s)\x1b[32m 200 OK\x1b[0m", model_name)
@@ -4679,38 +4657,16 @@ class Router:
             )
 
             self.total_calls[model_name] += 1
-            response = litellm.atranscription(
-                **{
-                    **data,
-                    "file": file,
-                    "caching": self.cache_responses,
-                    "client": model_client,
-                    **kwargs,
-                }
-            )
-
-            ### CONCURRENCY-SAFE RPM CHECKS ###
-            rpm_semaphore: Final = self._get_client(
-                deployment=deployment,
-                kwargs=kwargs,
-                client_type="max_parallel_requests",
-            )
-
-            if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore):
-                async with rpm_semaphore:
-                    """
-                    - Check rpm limits before making the call
-                    - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe)
-                    """
-                    await self.async_routing_strategy_pre_call_checks(
-                        deployment=deployment, parent_otel_span=parent_otel_span
-                    )
-                    response = await response
-            else:
-                await self.async_routing_strategy_pre_call_checks(
-                    deployment=deployment, parent_otel_span=parent_otel_span
+            async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span):
+                response = await litellm.atranscription(
+                    **{
+                        **data,
+                        "file": file,
+                        "caching": self.cache_responses,
+                        "client": model_client,
+                        **kwargs,
+                    }
                 )
-                response = await response
 
             self.success_calls[model_name] += 1
             verbose_router_logger.info("litellm.atranscription(model=%s)\x1b[32m 200 OK\x1b[0m", model_name)
@@ -4794,38 +4750,16 @@ class Router:
             )
 
             self.total_calls[model_name] += 1
-            response = litellm.aspeech(
-                **{
-                    **data,
-                    "input": input,
-                    "voice": data.get("voice") if voice is None else voice,
-                    "client": model_client,
-                    **kwargs,
-                }
-            )
-
-            ### CONCURRENCY-SAFE RPM CHECKS ###
-            rpm_semaphore: Final = self._get_client(
-                deployment=deployment,
-                kwargs=kwargs,
-                client_type="max_parallel_requests",
-            )
-
-            if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore):
-                async with rpm_semaphore:
-                    """
-                    - Check rpm limits before making the call
-                    - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe)
-                    """
-                    await self.async_routing_strategy_pre_call_checks(
-                        deployment=deployment, parent_otel_span=parent_otel_span
-                    )
-                    response = await response
-            else:
-                await self.async_routing_strategy_pre_call_checks(
-                    deployment=deployment, parent_otel_span=parent_otel_span
+            async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span):
+                response = await litellm.aspeech(
+                    **{
+                        **data,
+                        "input": input,
+                        "voice": data.get("voice") if voice is None else voice,
+                        "client": model_client,
+                        **kwargs,
+                    }
                 )
-                response = await response
 
             self.success_calls[model_name] += 1
             verbose_router_logger.info("litellm.aspeech(model=%s)\x1b[32m 200 OK\x1b[0m", model_name)
@@ -4990,37 +4924,16 @@ class Router:
             )
             self.total_calls[model_name] += 1
 
-            response = litellm.atext_completion(
-                **{
-                    **data,
-                    "prompt": prompt,
-                    "caching": self.cache_responses,
-                    "client": model_client,
-                    **kwargs,
-                }
-            )
-
-            rpm_semaphore: Final = self._get_client(
-                deployment=deployment,
-                kwargs=kwargs,
-                client_type="max_parallel_requests",
-            )
-
-            if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore):
-                async with rpm_semaphore:
-                    """
-                    - Check rpm limits before making the call
-                    - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe)
-                    """
-                    await self.async_routing_strategy_pre_call_checks(
-                        deployment=deployment, parent_otel_span=parent_otel_span
-                    )
-                    response = await response
-            else:
-                await self.async_routing_strategy_pre_call_checks(
-                    deployment=deployment, parent_otel_span=parent_otel_span
+            async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span):
+                response = await litellm.atext_completion(
+                    **{
+                        **data,
+                        "prompt": prompt,
+                        "caching": self.cache_responses,
+                        "client": model_client,
+                        **kwargs,
+                    }
                 )
-                response = await response
 
             self.success_calls[model_name] += 1
             verbose_router_logger.info("litellm.atext_completion(model=%s)\x1b[32m 200 OK\x1b[0m", model_name)
@@ -5081,37 +4994,16 @@ class Router:
             )
             self.total_calls[model_name] += 1
 
-            response = litellm.aadapter_completion(
-                **{
-                    **data,
-                    "adapter_id": adapter_id,
-                    "caching": self.cache_responses,
-                    "client": model_client,
-                    **kwargs,
-                }
-            )
-
-            rpm_semaphore: Final = self._get_client(
-                deployment=deployment,
-                kwargs=kwargs,
-                client_type="max_parallel_requests",
-            )
-
-            if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore):
-                async with rpm_semaphore:
-                    """
-                    - Check rpm limits before making the call
-                    - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe)
-                    """
-                    await self.async_routing_strategy_pre_call_checks(
-                        deployment=deployment, parent_otel_span=parent_otel_span
-                    )
-                    response = await response
-            else:
-                await self.async_routing_strategy_pre_call_checks(
-                    deployment=deployment, parent_otel_span=parent_otel_span
+            async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span):
+                response = await litellm.aadapter_completion(
+                    **{
+                        **data,
+                        "adapter_id": adapter_id,
+                        "caching": self.cache_responses,
+                        "client": model_client,
+                        **kwargs,
+                    }
                 )
-                response = await response
 
             self.success_calls[model_name] += 1
             verbose_router_logger.info("litellm.aadapter_completion(model=%s)\x1b[32m 200 OK\x1b[0m", model_name)
@@ -5341,29 +5233,8 @@ class Router:
             if custom_llm_provider is not None:
                 response_kwargs["custom_llm_provider"] = custom_llm_provider
 
-            response = original_generic_function(**response_kwargs)
-
-            rpm_semaphore: Final = self._get_client(
-                deployment=deployment,
-                kwargs=kwargs,
-                client_type="max_parallel_requests",
-            )
-
-            if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore):
-                async with rpm_semaphore:
-                    """
-                    - Check rpm limits before making the call
-                    - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe)
-                    """
-                    await self.async_routing_strategy_pre_call_checks(
-                        deployment=deployment, parent_otel_span=parent_otel_span
-                    )
-                    response = await response
-            else:
-                await self.async_routing_strategy_pre_call_checks(
-                    deployment=deployment, parent_otel_span=parent_otel_span
-                )
-                response = await response
+            async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span):
+                response = await original_generic_function(**response_kwargs)
 
             if self._should_raise_anthropic_refusal_error(
                 model=model,
@@ -5971,38 +5842,16 @@ class Router:
             )
 
             self.total_calls[model_name] += 1
-            response = litellm.aembedding(
-                **{
-                    **data,
-                    "input": input,
-                    "caching": self.cache_responses,
-                    "client": model_client,
-                    **kwargs,
-                }
-            )
-
-            ### CONCURRENCY-SAFE RPM CHECKS ###
-            rpm_semaphore: Final = self._get_client(
-                deployment=deployment,
-                kwargs=kwargs,
-                client_type="max_parallel_requests",
-            )
-
-            if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore):
-                async with rpm_semaphore:
-                    """
-                    - Check rpm limits before making the call
-                    - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe)
-                    """
-                    await self.async_routing_strategy_pre_call_checks(
-                        deployment=deployment, parent_otel_span=parent_otel_span
-                    )
-                    response = await response
-            else:
-                await self.async_routing_strategy_pre_call_checks(
-                    deployment=deployment, parent_otel_span=parent_otel_span
+            async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span):
+                response = await litellm.aembedding(
+                    **{
+                        **data,
+                        "input": input,
+                        "caching": self.cache_responses,
+                        "client": model_client,
+                        **kwargs,
+                    }
                 )
-                response = await response
 
             self.success_calls[model_name] += 1
             verbose_router_logger.info("litellm.aembedding(model=%s)\x1b[32m 200 OK\x1b[0m", model_name)
@@ -6111,37 +5960,18 @@ class Router:
                     "gcs_bucket_name" in data
                 ):  # TODO: Remove this once we have a better way to handle GCS bucket name:  Problem is that we need to pass the gcs_bucket_name to the router for the create_file call but it doesn't show up there
                     kwargs_copy.setdefault("litellm_metadata", {})["gcs_bucket_name"] = data["gcs_bucket_name"]
-                response = litellm.acreate_file(
-                    **{
-                        **data,
-                        "custom_llm_provider": custom_llm_provider,
-                        "caching": self.cache_responses,
-                        "client": model_client,
-                        **kwargs_copy,
-                    }
-                )
-
-                rpm_semaphore: Final = self._get_client(
-                    deployment=deployment,
-                    kwargs=kwargs_copy,
-                    client_type="max_parallel_requests",
-                )
-
-                if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore):
-                    async with rpm_semaphore:
-                        """
-                        - Check rpm limits before making the call
-                        - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe)
-                        """
-                        await self.async_routing_strategy_pre_call_checks(
-                            deployment=deployment, parent_otel_span=parent_otel_span
-                        )
-                        response = await response
-                else:
-                    await self.async_routing_strategy_pre_call_checks(
-                        deployment=deployment, parent_otel_span=parent_otel_span
+                async with self._deployment_slot(
+                    deployment=deployment, kwargs=kwargs_copy, parent_otel_span=parent_otel_span
+                ):
+                    response = await litellm.acreate_file(
+                        **{
+                            **data,
+                            "custom_llm_provider": custom_llm_provider,
+                            "caching": self.cache_responses,
+                            "client": model_client,
+                            **kwargs_copy,
+                        }
                     )
-                    response = await response
 
                 self.success_calls[model_name] += 1
                 verbose_router_logger.info("litellm.acreate_file(model=%s)\x1b[32m 200 OK\x1b[0m", model_name)
@@ -6231,33 +6061,16 @@ class Router:
             )
             custom_llm_provider = custom_llm_provider or inferred_custom_llm_provider
 
-            response = avector_store_create_sdk(
-                **{
-                    **data,
-                    "custom_llm_provider": custom_llm_provider,
-                    "caching": self.cache_responses,
-                    "client": model_client,
-                    **kwargs,
-                }
-            )
-
-            rpm_semaphore: Final = self._get_client(
-                deployment=deployment,
-                kwargs=kwargs,
-                client_type="max_parallel_requests",
-            )
-
-            if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore):
-                async with rpm_semaphore:
-                    await self.async_routing_strategy_pre_call_checks(
-                        deployment=deployment, parent_otel_span=parent_otel_span
-                    )
-                    response = await response
-            else:
-                await self.async_routing_strategy_pre_call_checks(
-                    deployment=deployment, parent_otel_span=parent_otel_span
+            async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span):
+                response = await avector_store_create_sdk(
+                    **{
+                        **data,
+                        "custom_llm_provider": custom_llm_provider,
+                        "caching": self.cache_responses,
+                        "client": model_client,
+                        **kwargs,
+                    }
                 )
-                response = await response
 
             self.success_calls[model_name] += 1
             verbose_router_logger.info("litellm.avector_store_create(model=%s)\x1b[32m 200 OK\x1b[0m", model_name)
@@ -6343,37 +6156,16 @@ class Router:
             )
             custom_llm_provider = custom_llm_provider or inferred_custom_llm_provider
 
-            response = litellm.acreate_batch(
-                **{
-                    **data,
-                    "custom_llm_provider": custom_llm_provider,
-                    "caching": self.cache_responses,
-                    "client": model_client,
-                    **kwargs,
-                }
-            )
-
-            rpm_semaphore: Final = self._get_client(
-                deployment=deployment,
-                kwargs=kwargs,
-                client_type="max_parallel_requests",
-            )
-
-            if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore):
-                async with rpm_semaphore:
-                    """
-                    - Check rpm limits before making the call
-                    - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe)
-                    """
-                    await self.async_routing_strategy_pre_call_checks(
-                        deployment=deployment, parent_otel_span=parent_otel_span
-                    )
-                    response = await response
-            else:
-                await self.async_routing_strategy_pre_call_checks(
-                    deployment=deployment, parent_otel_span=parent_otel_span
+            async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span):
+                response = await litellm.acreate_batch(
+                    **{
+                        **data,
+                        "custom_llm_provider": custom_llm_provider,
+                        "caching": self.cache_responses,
+                        "client": model_client,
+                        **kwargs,
+                    }
                 )
-                response = await response
 
             self.success_calls[model_name] += 1
             verbose_router_logger.info("litellm.acreate_batch(model=%s)\x1b[32m 200 OK\x1b[0m", model_name)
@@ -6564,37 +6356,16 @@ class Router:
             )
             custom_llm_provider = custom_llm_provider or inferred_custom_llm_provider
 
-            response = litellm.acancel_batch(
-                **{
-                    **data,
-                    "custom_llm_provider": custom_llm_provider,
-                    "caching": self.cache_responses,
-                    "client": model_client,
-                    **kwargs,
-                }
-            )
-
-            rpm_semaphore: Final = self._get_client(
-                deployment=deployment,
-                kwargs=kwargs,
-                client_type="max_parallel_requests",
-            )
-
-            if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore):
-                async with rpm_semaphore:
-                    """
-                    - Check rpm limits before making the call
-                    - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe)
-                    """
-                    await self.async_routing_strategy_pre_call_checks(
-                        deployment=deployment, parent_otel_span=parent_otel_span
-                    )
-                    response = await response
-            else:
-                await self.async_routing_strategy_pre_call_checks(
-                    deployment=deployment, parent_otel_span=parent_otel_span
+            async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span):
+                response = await litellm.acancel_batch(
+                    **{
+                        **data,
+                        "custom_llm_provider": custom_llm_provider,
+                        "caching": self.cache_responses,
+                        "client": model_client,
+                        **kwargs,
+                    }
                 )
-                response = await response
 
             self.success_calls[model_name] += 1
             verbose_router_logger.info("litellm.acancel_batch(model=%s)\x1b[32m 200 OK\x1b[0m", model_name)
@@ -8729,6 +8500,23 @@ class Router:
                         )
                     raise e
 
+    @contextlib.asynccontextmanager
+    async def _deployment_slot(
+        self, deployment: dict, kwargs: Mapping[str, object], parent_otel_span: Span | None
+    ) -> AsyncGenerator[None, None]:
+        """Holds the deployment's max_parallel_requests slot, if it has one, around the provider call. Routing
+        strategy pre-call checks run inside the slot so their rpm accounting stays concurrency-safe."""
+        rpm_semaphore: Final = self._get_client(
+            deployment=deployment,
+            kwargs=kwargs,
+            client_type="max_parallel_requests",
+        )
+        async with contextlib.AsyncExitStack() as slot:
+            if isinstance(rpm_semaphore, asyncio.Semaphore):
+                await slot.enter_async_context(rpm_semaphore)
+            await self.async_routing_strategy_pre_call_checks(deployment=deployment, parent_otel_span=parent_otel_span)
+            yield
+
     async def async_callback_filter_deployments(
         self,
         model: str,
@@ -12055,8 +11843,20 @@ class Router:
                 _settings_to_return[var] = self.lowestlatency_logger.routing_args.json()
 
         _settings_to_return["routing_groups"] = [group.model_dump() for group in self._routing_groups.values()]
+        _settings_to_return["default_max_parallel_requests_queue_size"] = self.default_max_parallel_requests_queue_size
         return _settings_to_return
 
+    @property
+    def default_max_parallel_requests_queue_size(self) -> int | None:
+        return self._default_max_parallel_requests_queue_size
+
+    @default_max_parallel_requests_queue_size.setter
+    def default_max_parallel_requests_queue_size(self, queue_size: int | None) -> None:
+        self._default_max_parallel_requests_queue_size = None if queue_size is None else int(queue_size)
+        InitalizeCachedClient.apply_default_max_parallel_requests_queue_size(
+            litellm_router_instance=self, queue_size=self._default_max_parallel_requests_queue_size
+        )
+
     def update_settings(self, **kwargs):
         """
         Update the router settings.

litellm/router_utils/client_initalization_utils.py

@@ -1,6 +1,10 @@
 import asyncio
+import time
 from typing import TYPE_CHECKING, Any, Final
 
+from litellm._logging import verbose_router_logger
+from litellm.exceptions import RateLimitError, RateLimitErrorCategory, RateLimitType
+from litellm.types.router import RouterErrors
 from litellm.utils import calculate_max_parallel_requests
 
 if TYPE_CHECKING:
@@ -11,6 +15,59 @@ else:
     LitellmRouter = Any
 
 
+class DeploymentSemaphore(asyncio.Semaphore):
+    """A deployment's max_parallel_requests slots. ``queue_size=None`` parks callers without bound, like a plain
+    ``asyncio.Semaphore``; otherwise a caller arriving while all slots are busy and ``queue_size`` callers already
+    wait gets a 429 instead of being parked."""
+
+    def __init__(self, max_parallel_requests: int, model_id: str, model_group: str, queue_size: int | None) -> None:
+        super().__init__(max_parallel_requests)
+        self.max_parallel_requests = max_parallel_requests
+        self.model_id = model_id
+        self.model_group = model_group
+        self.queue_size = queue_size
+        self.waiting = 0
+
+    async def acquire(self) -> bool:
+        if not self.locked():
+            return await super().acquire()
+        if self.queue_size is not None and self.waiting >= self.queue_size:
+            raise RateLimitError(
+                message=(
+                    f"{RouterErrors.max_parallel_requests_queue_full.value} Deployment model_group={self.model_group}, "
+                    f"id={self.model_id} has all max_parallel_requests={self.max_parallel_requests} slots in use and "
+                    f"{self.waiting} requests already waiting, which is its max_parallel_requests_queue_size="
+                    f"{self.queue_size}. Raise max_parallel_requests or max_parallel_requests_queue_size for this "
+                    "deployment, or unset max_parallel_requests_queue_size to queue without a bound"
+                ),
+                llm_provider="",
+                model=self.model_group,
+                category=RateLimitErrorCategory.LITELLM_RATE_LIMIT,
+                rate_limit_type=RateLimitType.CONCURRENT_REQUESTS,
+            )
+        self.waiting += 1
+        queued_at: Final = time.perf_counter()
+        verbose_router_logger.debug(
+            "Deployment model_group=%s, id=%s has all max_parallel_requests=%s slots in use, request queued "
+            "(waiting=%s, max_parallel_requests_queue_size=%s)",
+            self.model_group,
+            self.model_id,
+            self.max_parallel_requests,
+            self.waiting,
+            self.queue_size,
+        )
+        try:
+            return await super().acquire()
+        finally:
+            self.waiting -= 1
+            verbose_router_logger.debug(
+                "Deployment model_group=%s, id=%s request left the max_parallel_requests queue after %.1f ms",
+                self.model_group,
+                self.model_id,
+                (time.perf_counter() - queued_at) * 1000,
+            )
+
+
 class InitalizeCachedClient:
     @staticmethod
     def set_max_parallel_requests_client(litellm_router_instance: LitellmRouter, model: dict):
@@ -26,10 +83,35 @@ class InitalizeCachedClient:
             default_max_parallel_requests=litellm_router_instance.default_max_parallel_requests,
         )
         if calculated_max_parallel_requests:
-            semaphore: Final = asyncio.Semaphore(calculated_max_parallel_requests)
+            deployment_queue_size: Final = litellm_params.get("max_parallel_requests_queue_size", None)
+            semaphore: Final = DeploymentSemaphore(
+                max_parallel_requests=calculated_max_parallel_requests,
+                model_id=model_id,
+                model_group=model.get("model_name", ""),
+                queue_size=(
+                    deployment_queue_size
+                    if deployment_queue_size is not None
+                    else litellm_router_instance.default_max_parallel_requests_queue_size
+                ),
+            )
             cache_key: Final = f"{model_id}_max_parallel_requests_client"
             litellm_router_instance.cache.set_cache(
                 key=cache_key,
                 value=semaphore,
                 local_only=True,
             )
+
+    @staticmethod
+    def apply_default_max_parallel_requests_queue_size(
+        litellm_router_instance: LitellmRouter, queue_size: int | None
+    ) -> None:
+        inheriting_semaphores: Final = (
+            litellm_router_instance.cache.get_cache(
+                key=f"{model['model_info']['id']}_max_parallel_requests_client", local_only=True
+            )
+            for model in litellm_router_instance.model_list
+            if model["litellm_params"].get("max_parallel_requests_queue_size") is None
+        )
+        for semaphore in inheriting_semaphores:
+            if isinstance(semaphore, DeploymentSemaphore):
+                semaphore.queue_size = queue_size

litellm/types/management_endpoints/router_settings_endpoints.py

@@ -244,6 +244,17 @@ ROUTER_SETTINGS_FIELDS: Final[list[RouterSettingsField]] = [
         field_default=None,
         ui_field_name="Max Parallel Requests",
     ),
+    RouterSettingsField(
+        field_name="default_max_parallel_requests_queue_size",
+        field_type="Integer",
+        field_value=None,
+        field_description=(
+            "Default cap on how many requests may wait for a deployment's max_parallel_requests slot before "
+            "further requests get a 429. Unset queues without a bound"
+        ),
+        field_default=None,
+        ui_field_name="Max Parallel Requests Queue Size",
+    ),
     RouterSettingsField(
         field_name="enable_tag_filtering",
         field_type="Boolean",

litellm/types/router.py

@@ -497,6 +497,7 @@ class LiteLLMParamsTypedDict(TypedDict, total=False):
     order: int | None
     weight: int | None
     max_parallel_requests: int | None
+    max_parallel_requests_queue_size: ReadOnly[int | None]
     api_key: str | None
     api_base: str | None
     api_version: str | None
@@ -647,6 +648,7 @@ class RouterErrors(enum.Enum):
     """
 
     user_defined_ratelimit_error = "Deployment over user-defined ratelimit."
+    max_parallel_requests_queue_full = "Deployment max_parallel_requests queue is full."
     no_deployments_available = "No deployments available for selected model"
     all_deployments_in_cooldown = "All deployments for selected model are in cooldown"
     no_deployments_with_tag_routing = "Not allowed to access model due to tags configuration"

litellm/types/utils.py

@@ -3841,6 +3841,7 @@ all_litellm_params = (
         "itpm",
         "otpm",
         "max_parallel_requests",
+        "max_parallel_requests_queue_size",
         "input_cost_per_token",
         "output_cost_per_token",
         "input_cost_per_second",

tests/code_coverage_tests/router_code_coverage.py

@@ -88,6 +88,7 @@ ignored_function_names = [
     "_resolve_claude_code_session_router",  # Tested through Claude Code session routing in test_router.py
     "_get_claude_code_session_router_binding",  # Tested through the two-worker session routing test in test_router.py
     "_apply_updated_routing_strategy_args",  # Tested via update_settings in test_lowest_latency.py (file lacks "router" in name)
+    "default_max_parallel_requests_queue_size",  # Property, so its reads and assignments in test_router.py are never an ast.Call
 ]
 
 

tests/test_litellm/proxy/test_proxy_server.py

@@ -5051,6 +5051,39 @@ async def test_add_router_settings_from_db_config_empty_db_list_still_clears_unc
     assert combined_settings["num_retries"] == 1
 
 
+@pytest.mark.asyncio
+async def test_add_router_settings_from_db_config_null_queue_size_reaches_router():
+    """A cleared Admin UI field is stored as null. The reload must hand that None to the
+    router so a config.yaml bound is lifted, while an unrelated null still falls back to
+    the config value."""
+    from unittest.mock import AsyncMock, MagicMock
+
+    from litellm.proxy.proxy_server import ProxyConfig
+
+    proxy_config = ProxyConfig()
+    mock_router = MagicMock()
+    mock_router.update_settings = MagicMock()
+
+    config_data = {"router_settings": {"default_max_parallel_requests_queue_size": 2, "num_retries": 1}}
+
+    mock_db_config = MagicMock()
+    mock_db_config.param_value = {"default_max_parallel_requests_queue_size": None, "num_retries": None}
+
+    mock_prisma_client = MagicMock()
+    mock_prisma_client.db.litellm_config.find_first = AsyncMock(return_value=mock_db_config)
+
+    await proxy_config._add_router_settings_from_db_config(
+        config_data=config_data,
+        llm_router=mock_router,
+        prisma_client=mock_prisma_client,
+    )
+
+    combined_settings = mock_router.update_settings.call_args.kwargs
+    assert "default_max_parallel_requests_queue_size" in combined_settings
+    assert combined_settings["default_max_parallel_requests_queue_size"] is None
+    assert combined_settings["num_retries"] == 1
+
+
 @pytest.mark.asyncio
 async def test_add_router_settings_from_db_config_edge_cases():
     """
@@ -9334,6 +9367,32 @@ def test_update_config_litellm_settings_request_wins_for_non_callback_keys(
         restore()
 
 
+def test_update_config_router_settings_null_clears_max_parallel_requests_queue_size(
+    _update_config_setup,
+):
+    """Clearing the Admin UI field sends null. The stored row must hold null so the
+    reload hands None to the router and queueing becomes unbounded again, while an
+    unrelated null is still dropped rather than persisted."""
+    client, prisma, restore = _update_config_setup(
+        initial_rows={
+            "router_settings": {"default_max_parallel_requests_queue_size": 3, "num_retries": 2},
+        }
+    )
+    try:
+        resp = client.post(
+            "/config/update",
+            json={"router_settings": {"default_max_parallel_requests_queue_size": None, "timeout": None}},
+        )
+        assert resp.status_code == 200
+        stored = prisma.db.litellm_config.rows["router_settings"]
+        assert "default_max_parallel_requests_queue_size" in stored
+        assert stored["default_max_parallel_requests_queue_size"] is None
+        assert stored["num_retries"] == 2
+        assert "timeout" not in stored
+    finally:
+        restore()
+
+
 def test_update_config_success_callback_normalizes_existing_mixed_case(
     _update_config_setup,
 ):

tests/test_litellm/router_utils/test_client_initalization_utils.py

@@ -0,0 +1,189 @@
+import asyncio
+from typing import Final
+
+import pytest
+
+import litellm
+from litellm import Router
+from litellm.router_utils.client_initalization_utils import DeploymentSemaphore
+
+
+def _semaphore(queue_size: int | None, max_parallel_requests: int = 1) -> DeploymentSemaphore:
+    return DeploymentSemaphore(
+        max_parallel_requests=max_parallel_requests,
+        model_id="deployment-1",
+        model_group="gpt-5.6",
+        queue_size=queue_size,
+    )
+
+
+async def _hold(semaphore: DeploymentSemaphore, release: asyncio.Event) -> str:
+    async with semaphore:
+        await release.wait()
+        return "ok"
+
+
+async def _expect_rejection(semaphore: DeploymentSemaphore) -> litellm.RateLimitError:
+    with pytest.raises(litellm.RateLimitError) as excinfo:
+        await asyncio.wait_for(semaphore.acquire(), timeout=1)
+    return excinfo.value
+
+
+@pytest.mark.asyncio
+async def test_queue_full_rejects_new_caller_while_queued_callers_still_complete():
+    semaphore: Final = _semaphore(queue_size=2)
+    release: Final = asyncio.Event()
+    holder: Final = asyncio.create_task(_hold(semaphore, release))
+    await asyncio.sleep(0)
+    queued: Final = [asyncio.create_task(_hold(semaphore, release)) for _ in range(2)]
+    await asyncio.sleep(0)
+    assert semaphore.locked() and semaphore.waiting == 2
+
+    rejection: Final = await _expect_rejection(semaphore)
+
+    assert rejection.status_code == 429
+    assert "deployment-1" in rejection.message
+    assert "gpt-5.6" in rejection.message
+    assert "max_parallel_requests=1" in rejection.message
+    assert "max_parallel_requests_queue_size=2" in rejection.message
+    assert semaphore.waiting == 2
+
+    release.set()
+    assert await asyncio.wait_for(asyncio.gather(holder, *queued), timeout=2) == ["ok", "ok", "ok"]
+    assert semaphore.waiting == 0
+    assert not semaphore.locked()
+
+
+@pytest.mark.asyncio
+async def test_zero_queue_size_rejects_as_soon_as_every_slot_is_busy():
+    semaphore: Final = _semaphore(queue_size=0, max_parallel_requests=2)
+    release: Final = asyncio.Event()
+    holders: Final = [asyncio.create_task(_hold(semaphore, release)) for _ in range(2)]
+    await asyncio.sleep(0)
+
+    await _expect_rejection(semaphore)
+    assert semaphore.waiting == 0
+
+    release.set()
+    assert await asyncio.wait_for(asyncio.gather(*holders), timeout=2) == ["ok", "ok"]
+
+
+@pytest.mark.asyncio
+async def test_unset_queue_size_parks_every_caller_until_a_slot_frees():
+    semaphore: Final = _semaphore(queue_size=None)
+    release: Final = asyncio.Event()
+    callers: Final = [asyncio.create_task(_hold(semaphore, release)) for _ in range(50)]
+    await asyncio.sleep(0)
+    assert semaphore.waiting == 49
+
+    release.set()
+    assert await asyncio.wait_for(asyncio.gather(*callers), timeout=2) == ["ok"] * 50
+    assert semaphore.waiting == 0
+
+
+@pytest.mark.asyncio
+async def test_cancelled_waiter_gives_its_queue_slot_back():
+    semaphore: Final = _semaphore(queue_size=1)
+    release: Final = asyncio.Event()
+    holder: Final = asyncio.create_task(_hold(semaphore, release))
+    await asyncio.sleep(0)
+    cancelled: Final = asyncio.create_task(_hold(semaphore, release))
+    await asyncio.sleep(0)
+    assert semaphore.waiting == 1
+
+    cancelled.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await cancelled
+    assert semaphore.waiting == 0
+
+    replacement: Final = asyncio.create_task(_hold(semaphore, release))
+    await asyncio.sleep(0)
+    assert semaphore.waiting == 1
+    release.set()
+    assert await asyncio.wait_for(asyncio.gather(holder, replacement), timeout=2) == ["ok", "ok"]
+
+
+def _router_semaphore(router: Router, model_name: str) -> DeploymentSemaphore:
+    deployment: Final = router.get_deployment_by_model_group_name(model_group_name=model_name)
+    assert deployment is not None
+    client: Final = router._get_client(deployment=deployment.model_dump(), kwargs={}, client_type="max_parallel_requests")
+    assert isinstance(client, DeploymentSemaphore)
+    return client
+
+
+@pytest.mark.asyncio
+async def test_deployment_queue_size_overrides_router_default_and_zero_is_honored():
+    router: Final = Router(
+        model_list=[
+            {"model_name": "inherits-default", "litellm_params": {"model": "openai/gpt-5.6", "rpm": 1}},
+            {
+                "model_name": "no-queue",
+                "litellm_params": {"model": "openai/gpt-5.6", "tpm": 100, "max_parallel_requests_queue_size": 0},
+            },
+        ],
+        default_max_parallel_requests_queue_size=1,
+    )
+    release: Final = asyncio.Event()
+
+    inherits: Final = _router_semaphore(router, "inherits-default")
+    inherits_holder: Final = asyncio.create_task(_hold(inherits, release))
+    await asyncio.sleep(0)
+    inherits_waiter: Final = asyncio.create_task(_hold(inherits, release))
+    await asyncio.sleep(0)
+    assert "max_parallel_requests_queue_size=1" in (await _expect_rejection(inherits)).message
+
+    no_queue: Final = _router_semaphore(router, "no-queue")
+    no_queue_holder: Final = asyncio.create_task(_hold(no_queue, release))
+    await asyncio.sleep(0)
+    assert "max_parallel_requests_queue_size=0" in (await _expect_rejection(no_queue)).message
+
+    release.set()
+    await asyncio.wait_for(asyncio.gather(inherits_holder, inherits_waiter, no_queue_holder), timeout=2)
+
+
+@pytest.mark.asyncio
+async def test_router_without_queue_size_keeps_unbounded_queueing():
+    router: Final = Router(
+        model_list=[{"model_name": "gpt-5.6", "litellm_params": {"model": "openai/gpt-5.6", "max_parallel_requests": 1}}]
+    )
+    semaphore: Final = _router_semaphore(router, "gpt-5.6")
+    release: Final = asyncio.Event()
+    callers: Final = [asyncio.create_task(_hold(semaphore, release)) for _ in range(20)]
+    await asyncio.sleep(0)
+    assert semaphore.waiting == 19
+    release.set()
+    assert await asyncio.wait_for(asyncio.gather(*callers), timeout=2) == ["ok"] * 20
+
+
+@pytest.mark.asyncio
+async def test_update_settings_applies_default_queue_size_to_live_semaphores_without_an_override():
+    router: Final = Router(
+        model_list=[
+            {"model_name": "inherits-default", "litellm_params": {"model": "openai/gpt-5.6", "rpm": 1}},
+            {
+                "model_name": "pinned",
+                "litellm_params": {"model": "openai/gpt-5.6", "rpm": 1, "max_parallel_requests_queue_size": 5},
+            },
+        ],
+    )
+    inherits: Final = _router_semaphore(router, "inherits-default")
+    pinned: Final = _router_semaphore(router, "pinned")
+    assert router.get_settings()["default_max_parallel_requests_queue_size"] is None
+
+    router.update_settings(default_max_parallel_requests_queue_size="0")
+    assert router.get_settings()["default_max_parallel_requests_queue_size"] == 0
+    assert (inherits.queue_size, pinned.queue_size) == (0, 5)
+
+    release: Final = asyncio.Event()
+    holder: Final = asyncio.create_task(_hold(inherits, release))
+    await asyncio.sleep(0)
+    assert "max_parallel_requests_queue_size=0" in (await _expect_rejection(inherits)).message
+
+    router.update_settings(default_max_parallel_requests_queue_size=None)
+    assert (inherits.queue_size, pinned.queue_size) == (None, 5)
+    waiter: Final = asyncio.create_task(_hold(inherits, release))
+    await asyncio.sleep(0)
+    assert inherits.waiting == 1
+
+    release.set()
+    assert await asyncio.wait_for(asyncio.gather(holder, waiter), timeout=2) == ["ok", "ok"]

tests/test_litellm/test_router.py

@@ -1,11 +1,13 @@
 import asyncio
 import copy
 import functools
+import gc
 import json
 import logging
 import os
 import sys
 import threading
+import warnings
 from collections.abc import Awaitable, Callable, Mapping
 from datetime import datetime, timedelta
 from types import SimpleNamespace
@@ -45,6 +47,7 @@ from litellm.router import (
     _is_retriable_anthropic_status,
 )
 from litellm.router_strategy import simple_shuffle
+from litellm.router_utils.cooldown_handlers import _async_get_cooldown_deployments
 from litellm.types.llms.openai import ChatCompletionRequest
 from litellm.types.router import Deployment, DeploymentTypedDict, LiteLLM_Params, ModelInfo, PreRoutingHookResponse, RetryPolicy
 
@@ -16038,6 +16041,224 @@ async def test_router_max_parallel_requests_slot_released_when_stream_closed_ear
     assert tracker.current == 0
 
 
+@pytest.mark.asyncio
+async def test_router_max_parallel_requests_queue_size_turns_overflow_into_429(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    router: Final = Router(
+        model_list=[
+            {
+                "model_name": "gpt-5.6",
+                "litellm_params": {
+                    "model": "openai/gpt-5.6",
+                    "api_key": "sk-fake",
+                    "api_base": "https://max-parallel.local/v1",
+                    "max_parallel_requests": 1,
+                    "max_parallel_requests_queue_size": 1,
+                },
+                "model_info": {"id": "queue-bounded-deployment"},
+            },
+            {
+                "model_name": "gpt-5.6",
+                "litellm_params": {
+                    "model": "openai/gpt-5.6",
+                    "api_key": "sk-fake",
+                    "api_base": "https://max-parallel-sibling.local/v1",
+                },
+                "model_info": {"id": "queue-sibling-deployment"},
+            },
+        ],
+        num_retries=0,
+    )
+
+    async def upstream(request: httpx.Request) -> httpx.Response:
+        await asyncio.sleep(0.2)
+        return httpx.Response(
+            200,
+            json={
+                "id": "c",
+                "object": "chat.completion",
+                "created": 1,
+                "model": "gpt-5.6",
+                "choices": [{"index": 0, "message": {"role": "assistant", "content": "x"}, "finish_reason": "stop"}],
+            },
+        )
+
+    with respx.mock(assert_all_called=False) as respx_mock:
+        route: Final = respx_mock.post("https://max-parallel.local/v1/chat/completions").mock(side_effect=upstream)
+        sibling_route: Final = respx_mock.post("https://max-parallel-sibling.local/v1/chat/completions").mock(
+            side_effect=upstream
+        )
+        results: Final = await asyncio.wait_for(
+            asyncio.gather(
+                *(
+                    router.acompletion(model="queue-bounded-deployment", messages=[{"role": "user", "content": "hi"}])
+                    for _ in range(3)
+                ),
+                return_exceptions=True,
+            ),
+            timeout=10,
+        )
+
+    rejected: Final = [r for r in results if isinstance(r, BaseException)]
+    assert len(rejected) == 1 and len(results) == 3
+    assert isinstance(rejected[0], litellm.RateLimitError)
+    assert rejected[0].status_code == 429
+    assert "queue-bounded-deployment" in rejected[0].message
+    assert "max_parallel_requests_queue_size=1" in rejected[0].message
+    assert route.call_count == 2
+    assert sibling_route.call_count == 0
+    assert all("max_parallel_requests_queue_size" not in call.request.content.decode() for call in route.calls)
+    assert await _async_get_cooldown_deployments(litellm_router_instance=router, parent_otel_span=None) == []
+
+
+@pytest.mark.asyncio
+async def test_router_embedding_path_honors_max_parallel_requests_queue_size(monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    router: Final = Router(
+        model_list=[
+            {
+                "model_name": "embed",
+                "litellm_params": {
+                    "model": "openai/text-embedding-3-small",
+                    "api_key": "sk-fake",
+                    "api_base": "https://max-parallel-embed.local/v1",
+                    "max_parallel_requests": 1,
+                },
+                "model_info": {"id": "embed-bounded-deployment"},
+            }
+        ],
+        default_max_parallel_requests_queue_size=1,
+        num_retries=0,
+    )
+
+    async def upstream(request: httpx.Request) -> httpx.Response:
+        await asyncio.sleep(0.2)
+        return httpx.Response(
+            200,
+            json={
+                "object": "list",
+                "data": [{"object": "embedding", "index": 0, "embedding": [0.1, 0.2]}],
+                "model": "text-embedding-3-small",
+                "usage": {"prompt_tokens": 1, "total_tokens": 1},
+            },
+        )
+
+    with respx.mock() as respx_mock, warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        route: Final = respx_mock.post("https://max-parallel-embed.local/v1/embeddings").mock(side_effect=upstream)
+        results: Final = await asyncio.wait_for(
+            asyncio.gather(
+                *(router.aembedding(model="embed", input=["hi"]) for _ in range(3)),
+                return_exceptions=True,
+            ),
+            timeout=10,
+        )
+        gc.collect()
+
+    rejected: Final = [r for r in results if isinstance(r, BaseException)]
+    assert len(rejected) == 1 and len(results) == 3
+    assert isinstance(rejected[0], litellm.RateLimitError) and rejected[0].status_code == 429
+    assert "embed-bounded-deployment" in rejected[0].message
+    assert route.call_count == 2
+    assert [str(w.message) for w in caught if "never awaited" in str(w.message)] == []
+
+
+@pytest.mark.asyncio
+async def test_router_max_parallel_requests_queue_overflow_takes_the_ordinary_429_fallback_path(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    router: Final = Router(
+        model_list=[
+            {
+                "model_name": "gpt-5.6",
+                "litellm_params": {
+                    "model": "openai/gpt-5.6",
+                    "api_key": "sk-fake",
+                    "api_base": "https://max-parallel-primary.local/v1",
+                    "max_parallel_requests": 1,
+                    "max_parallel_requests_queue_size": 0,
+                },
+                "model_info": {"id": "queue-primary-deployment"},
+            },
+            {
+                "model_name": "gpt-5.6-fallback",
+                "litellm_params": {
+                    "model": "openai/gpt-5.6",
+                    "api_key": "sk-fake",
+                    "api_base": "https://max-parallel-fallback.local/v1",
+                },
+                "model_info": {"id": "queue-fallback-deployment"},
+            },
+        ],
+        fallbacks=[{"gpt-5.6": ["gpt-5.6-fallback"]}],
+        num_retries=0,
+    )
+
+    async def upstream(request: httpx.Request) -> httpx.Response:
+        await asyncio.sleep(0.2)
+        return httpx.Response(
+            200,
+            json={
+                "id": "c",
+                "object": "chat.completion",
+                "created": 1,
+                "model": "gpt-5.6",
+                "choices": [{"index": 0, "message": {"role": "assistant", "content": "x"}, "finish_reason": "stop"}],
+            },
+        )
+
+    with respx.mock() as respx_mock:
+        primary: Final = respx_mock.post("https://max-parallel-primary.local/v1/chat/completions").mock(
+            side_effect=upstream
+        )
+        fallback: Final = respx_mock.post("https://max-parallel-fallback.local/v1/chat/completions").mock(
+            side_effect=upstream
+        )
+        results: Final = await asyncio.wait_for(
+            asyncio.gather(
+                *(router.acompletion(model="gpt-5.6", messages=[{"role": "user", "content": "hi"}]) for _ in range(3))
+            ),
+            timeout=10,
+        )
+
+    assert len(results) == 3
+    assert primary.call_count == 1
+    assert fallback.call_count == 2
+    assert await _async_get_cooldown_deployments(litellm_router_instance=router, parent_otel_span=None) == []
+
+
+@pytest.mark.asyncio
+async def test_router_deployment_slot_rejects_once_queue_is_full_and_frees_slot_on_exit():
+    router: Final = Router(
+        model_list=[
+            {
+                "model_name": "gpt-5.6",
+                "litellm_params": {
+                    "model": "openai/gpt-5.6",
+                    "api_key": "sk-fake",
+                    "max_parallel_requests": 1,
+                    "max_parallel_requests_queue_size": 0,
+                },
+                "model_info": {"id": "slot-deployment"},
+            }
+        ]
+    )
+    deployment: Final = router.get_deployment(model_id="slot-deployment")
+    assert deployment is not None
+    kwargs: Final = {"model": "gpt-5.6"}
+
+    async with router._deployment_slot(deployment=deployment.model_dump(), kwargs=kwargs, parent_otel_span=None):
+        with pytest.raises(litellm.RateLimitError) as overflow:
+            async with router._deployment_slot(deployment=deployment.model_dump(), kwargs=kwargs, parent_otel_span=None):
+                pass
+    assert overflow.value.status_code == 429
+    assert "slot-deployment" in overflow.value.message
+
+    async with router._deployment_slot(deployment=deployment.model_dump(), kwargs=kwargs, parent_otel_span=None):
+        pass
+
+
 @pytest.mark.asyncio
 async def test_router_deployment_drop_params_string_true_is_honored(monkeypatch):
     from litellm import Router

tests/test_litellm/test_utils.py

@@ -60,6 +60,7 @@ from litellm.utils import (
     _snapshot_exception_for_hook,
     async_post_call_failure_deployment_hook,
     async_post_call_success_deployment_hook,
+    calculate_max_parallel_requests,
     client,
     get_non_default_completion_params,
     get_optional_params_image_gen,
@@ -6213,3 +6214,32 @@ def test_provider_prefixed_lookup_never_outranks_an_existing_row(local_model_cos
         ("openrouter/openai/gpt-4o", "openrouter", "openrouter/openai/gpt-4o"),
     ):
         assert litellm.get_model_info(model=model, custom_llm_provider=provider)["key"] == expected_key
+
+
+@pytest.mark.parametrize(
+    ("max_parallel_requests", "rpm", "tpm", "default_max_parallel_requests", "expected"),
+    [
+        (3, 100, 100_000, 7, 3),
+        (None, 100, 100_000, 7, 100),
+        (None, None, 100_000, 7, 600),
+        (None, None, 50, 7, 1),
+        (None, None, None, 7, 7),
+        (None, None, None, None, None),
+    ],
+)
+def test_calculate_max_parallel_requests_precedence(
+    max_parallel_requests: int | None,
+    rpm: int | None,
+    tpm: int | None,
+    default_max_parallel_requests: int | None,
+    expected: int | None,
+) -> None:
+    assert (
+        calculate_max_parallel_requests(
+            max_parallel_requests=max_parallel_requests,
+            rpm=rpm,
+            tpm=tpm,
+            default_max_parallel_requests=default_max_parallel_requests,
+        )
+        == expected
+    )

ui/litellm-dashboard/src/components/router_settings/index.test.tsx

@@ -137,6 +137,41 @@ describe("RouterSettings", () => {
     );
   });
 
+  it("should save default_max_parallel_requests_queue_size as a number and an empty field as null", async () => {
+    vi.mocked(getCallbacksCall).mockResolvedValue({
+      router_settings: { ...mockCallbacksResponse.router_settings, default_max_parallel_requests_queue_size: null },
+    });
+    const user = userEvent.setup();
+    renderWithProviders(<RouterSettings {...defaultProps} />);
+
+    await findStrategySelect();
+
+    const queueSize = await screen.findByRole("textbox", { name: /default_max_parallel_requests_queue_size/i });
+    fireEvent.change(queueSize, { target: { value: "4" } });
+    await user.click(screen.getByRole("button", { name: /save changes/i }));
+
+    await waitFor(() =>
+      expect(setCallbacksCall).toHaveBeenLastCalledWith(
+        "test-token",
+        expect.objectContaining({
+          router_settings: expect.objectContaining({ default_max_parallel_requests_queue_size: 4 }),
+        }),
+      ),
+    );
+
+    fireEvent.change(queueSize, { target: { value: "" } });
+    await user.click(screen.getByRole("button", { name: /save changes/i }));
+
+    await waitFor(() =>
+      expect(setCallbacksCall).toHaveBeenLastCalledWith(
+        "test-token",
+        expect.objectContaining({
+          router_settings: expect.objectContaining({ default_max_parallel_requests_queue_size: null }),
+        }),
+      ),
+    );
+  });
+
   it("should show a success notification after saving", async () => {
     const user = userEvent.setup();
     renderWithProviders(<RouterSettings {...defaultProps} />);

ui/litellm-dashboard/src/components/router_settings/index.tsx

@@ -86,7 +86,15 @@ const RouterSettings: React.FC<RouterSettingsProps> = ({ accessToken, userRole,
 
     const router_settings = formValue.routerSettings;
 
-    const numberKeys = new Set(["allowed_fails", "cooldown_time", "num_retries", "timeout", "retry_after"]);
+    const numberKeys = new Set([
+      "allowed_fails",
+      "cooldown_time",
+      "num_retries",
+      "timeout",
+      "retry_after",
+      "default_max_parallel_requests_queue_size",
+    ]);
+    const unsettableNumberKeys = new Set(["default_max_parallel_requests_queue_size"]);
     const jsonKeys = new Set(["model_group_alias"]);
     // retry_policy and model_group_retry_policy are owned by the Model Retry Settings tab;
     // routing_groups is owned by the Routing Groups tab. This page must not read or write them.
@@ -100,6 +108,7 @@ const RouterSettings: React.FC<RouterSettingsProps> = ({ accessToken, userRole,
       if (v.toLowerCase() === "null") return null;
 
       if (numberKeys.has(key)) {
+        if (v === "" && unsettableNumberKeys.has(key)) return null;
         const n = Number(v);
         return Number.isNaN(n) ? fallback : n;
       }