[Serve] Reject gang-unaligned autoscaling bounds

doc/source/serve/advanced-guides/gang-scheduling.md

@@ -28,7 +28,7 @@ Configure gang scheduling by passing a `GangSchedulingConfig` to the `@serve.dep
 :language: python
 ```
 
-This creates 2 gangs of 4 replicas each resulting in a total of 8 replicas. Partial gang isn't allowed, and therefore `num_replicas` must be a multiple of `gang_size`.
+This creates 2 gangs of 4 replicas each resulting in a total of 8 replicas. Partial gang isn't allowed, and therefore `num_replicas` must be a multiple of `gang_size`. With autoscaling, the autoscaling bounds (`min_replicas`, `max_replicas`, and `initial_replicas`) must also be multiples of `gang_size`; see [Autoscaling](#autoscaling).
 
 ### How resources are reserved within a gang
 

python/ray/serve/_private/config.py

@@ -671,6 +671,46 @@ def handle_num_replicas_auto(
     return max_ongoing_requests, autoscaling_config
 
 
+_GANG_ALIGNED_AUTOSCALING_BOUNDS = (
+    "min_replicas",
+    "max_replicas",
+    "initial_replicas",
+)
+
+
+def validate_gang_scheduling_bounds(
+    gang_size: int,
+    autoscaling_config: Optional[Union[Dict, AutoscalingConfig]],
+) -> None:
+    """Validate that explicitly set autoscaling bounds are multiples of gang_size.
+
+    Bounds left at their defaults are not checked: gang scheduling rounds the
+    autoscaler's decisions up to whole gangs, so the defaults are safe.
+    """
+    if autoscaling_config is None:
+        return
+
+    if isinstance(autoscaling_config, AutoscalingConfig):
+        bounds = {
+            name: getattr(autoscaling_config, name)
+            for name in autoscaling_config.model_fields_set
+            if name in _GANG_ALIGNED_AUTOSCALING_BOUNDS
+        }
+    else:
+        bounds = {
+            name: autoscaling_config[name]
+            for name in _GANG_ALIGNED_AUTOSCALING_BOUNDS
+            if name in autoscaling_config
+        }
+
+    for field_name, value in bounds.items():
+        if value is not None and value % gang_size != 0:
+            raise ValueError(
+                f"autoscaling_config.{field_name} ({value}) must be a multiple of "
+                f"gang_size ({gang_size})."
+            )
+
+
 class ReplicaConfig:
     """Internal datastructure wrapping config options for a deployment's replicas.
 

python/ray/serve/api.py

@@ -17,6 +17,7 @@
     DeploymentConfig,
     ReplicaConfig,
     handle_num_replicas_auto,
+    validate_gang_scheduling_bounds,
 )
 from ray.serve._private.constants import (
     RAY_SERVE_FORCE_LOCAL_TESTING_MODE,
@@ -747,6 +748,19 @@ class MyDeployment:
             "gang_scheduling_config is provided. Use "
             "gang_scheduling_config.gang_placement_strategy instead."
         )
+    # Validate the user-provided bounds before handle_num_replicas_auto()
+    # merges in defaults, so only explicitly set bounds are checked.
+    if gang_scheduling_config not in [
+        DEFAULT.VALUE,
+        None,
+    ] and autoscaling_config not in [DEFAULT.VALUE, None]:
+        if isinstance(gang_scheduling_config, dict):
+            gang_size = gang_scheduling_config.get("gang_size")
+        else:
+            gang_size = gang_scheduling_config.gang_size
+        if gang_size is not None:
+            validate_gang_scheduling_bounds(gang_size, autoscaling_config)
+
     if num_replicas == "auto":
         num_replicas = None
         max_ongoing_requests, autoscaling_config = handle_num_replicas_auto(

python/ray/serve/deployment.py

@@ -6,6 +6,7 @@
     ReplicaConfig,
     RequestRouterConfig,
     handle_num_replicas_auto,
+    validate_gang_scheduling_bounds,
 )
 from ray.serve._private.usage import ServeUsageTag
 from ray.serve._private.utils import DEFAULT, Default
@@ -337,6 +338,24 @@ def options(
         if max_ongoing_requests is None:
             raise ValueError("`max_ongoing_requests` must be non-null, got None.")
 
+        # Validate the user-provided bounds before handle_num_replicas_auto()
+        # merges in defaults, so only explicitly set bounds are checked.
+        gang_config_for_validation = (
+            gang_scheduling_config
+            if gang_scheduling_config is not DEFAULT.VALUE
+            else self._deployment_config.gang_scheduling_config
+        )
+        if gang_config_for_validation is not None and autoscaling_config not in [
+            DEFAULT.VALUE,
+            None,
+        ]:
+            if isinstance(gang_config_for_validation, dict):
+                gang_size = gang_config_for_validation.get("gang_size")
+            else:
+                gang_size = gang_config_for_validation.gang_size
+            if gang_size is not None:
+                validate_gang_scheduling_bounds(gang_size, autoscaling_config)
+
         if num_replicas == "auto":
             max_ongoing_requests, autoscaling_config = handle_num_replicas_auto(
                 max_ongoing_requests, autoscaling_config

python/ray/serve/tests/unit/test_config.py

@@ -1140,6 +1140,73 @@ def f():
         assert f2._deployment_config.autoscaling_config.min_replicas == 4
         assert f2._deployment_config.autoscaling_config.max_replicas == 8
 
+    @pytest.mark.parametrize(
+        "bounds",
+        [
+            {"min_replicas": 3},
+            {"max_replicas": 6},
+            {"initial_replicas": 3},
+            {"min_replicas": 0, "max_replicas": 6},
+        ],
+    )
+    def test_gang_scheduling_config_rejects_unaligned_autoscaling_bounds(self, bounds):
+        with pytest.raises(
+            ValueError, match="autoscaling_config.*must be a multiple of gang_size"
+        ):
+
+            @serve.deployment(
+                num_replicas="auto",
+                gang_scheduling_config=GangSchedulingConfig(gang_size=4),
+                autoscaling_config=bounds,
+            )
+            def f():
+                return "test"
+
+    def test_gang_scheduling_config_rejects_unaligned_autoscaling_config_instance(
+        self,
+    ):
+        with pytest.raises(
+            ValueError, match="autoscaling_config.*must be a multiple of gang_size"
+        ):
+
+            @serve.deployment(
+                num_replicas="auto",
+                gang_scheduling_config=GangSchedulingConfig(gang_size=2),
+                autoscaling_config=AutoscalingConfig(min_replicas=3, max_replicas=8),
+            )
+            def f():
+                return "test"
+
+    def test_gang_scheduling_config_allows_default_autoscaling_bounds(self):
+        """Unset bounds keep their defaults and don't trip the gang check."""
+
+        @serve.deployment(
+            num_replicas="auto",
+            gang_scheduling_config=GangSchedulingConfig(gang_size=4),
+            autoscaling_config={"target_ongoing_requests": 2.0},
+        )
+        def f():
+            return "test"
+
+        assert f._deployment_config.autoscaling_config.min_replicas == 1
+        assert f._deployment_config.autoscaling_config.max_replicas == 100
+
+    def test_gang_scheduling_config_rejects_unaligned_bounds_via_options(self):
+        @serve.deployment(
+            num_replicas=4,
+            gang_scheduling_config=GangSchedulingConfig(gang_size=4),
+        )
+        def f():
+            return "test"
+
+        with pytest.raises(
+            ValueError, match="autoscaling_config.*must be a multiple of gang_size"
+        ):
+            f.options(
+                num_replicas="auto",
+                autoscaling_config={"min_replicas": 3, "max_replicas": 8},
+            )
+
     def test_gang_scheduling_config_proto_roundtrip(self):
         """Test roundtrip serialization of GangSchedulingConfig through protobuf."""