[iris] Reject jobs with unsatisfiable routing constraints at submit time

Adds check_routing_feasibility() to the autoscaler, called from launch_job
for all jobs. Produces actionable diagnostics: device type/variant mismatch,
region/zone typo suggestions via fuzzy matching, and zone-vs-region confusion
detection (e.g. passing a zone value as a region constraint). Soft constraints
are excluded from the check so they don't prevent submission.

Fixes #4679

Author: github-actions[bot]

lib/iris/src/iris/cluster/controller/autoscaler/routing.py

@@ -7,16 +7,20 @@
 
 import difflib
 import math
+import re
+from collections.abc import Sequence
 from dataclasses import dataclass
 
 from iris.cluster.constraints import (
     ConstraintIndex,
     DeviceType,
+    extract_placement_requirements,
     get_device_type_enum,
     routing_constraints,
     soft_constraint_score,
     split_hard_soft,
 )
+from iris.rpc import job_pb2
 from iris.cluster.controller.autoscaler.models import (
     AdditiveReq,
     DemandEntry,
@@ -270,6 +274,77 @@ def _diagnose_no_matching_group(entry: DemandEntry, groups: list[ScalingGroup])
     )
 
 
+# GCP zones end with -{single letter}, e.g. us-central1-a.
+_ZONE_PATTERN = re.compile(r".+-[a-z]$")
+
+
+def _looks_like_zone(value: str) -> bool:
+    return bool(_ZONE_PATTERN.fullmatch(value))
+
+
+def diagnose_unsatisfiable_constraints(
+    constraints: Sequence[job_pb2.Constraint],
+    groups: list[ScalingGroup],
+) -> str:
+    """Produce a user-facing error when no scaling group can satisfy constraints.
+
+    Performs layered diagnosis (device, preemptible, zone, region) and
+    detects zone/region value confusion.
+    """
+    normalized = extract_placement_requirements(constraints)
+    device_type = normalized.device_type or DeviceType.CPU
+
+    device_matches = [g for g in groups if g.matches_device_requirement(device_type, normalized.device_variants)]
+    variants_str = _format_variants(normalized.device_variants)
+    if not device_matches:
+        available = ", ".join(g.name for g in groups)
+        return f"no scaling group provides device {device_type.value}:{variants_str} (available: {available})"
+
+    if normalized.preemptible is not None:
+        preempt_matches = [
+            g
+            for g in device_matches
+            if (g.config.resources.capacity_type == config_pb2.CAPACITY_TYPE_PREEMPTIBLE) == normalized.preemptible
+        ]
+        if not preempt_matches:
+            want = "preemptible" if normalized.preemptible else "non-preemptible"
+            return f"no {want} group provides device {device_type.value}:{variants_str}"
+        device_matches = preempt_matches
+
+    if normalized.required_zones:
+        available_zones = {g.zone for g in device_matches} - {None}
+        available_regions = {g.region for g in device_matches} - {None}
+        requested = sorted(normalized.required_zones)
+        parts = [f"no groups in zone {', '.join(requested)}"]
+        for z in requested:
+            # Prioritize zone/region confusion over fuzzy match
+            if not _looks_like_zone(z) and z in available_regions:
+                parts.append(f"'{z}' looks like a region, not a zone; use a region constraint instead")
+            else:
+                close = difflib.get_close_matches(z, available_zones, n=1, cutoff=0.7)
+                if close:
+                    parts.append(f"did you mean {close[0]}?")
+        return "; ".join(parts)
+
+    if normalized.required_regions:
+        available_regions = {g.region for g in device_matches} - {None}
+        available_zones = {g.zone for g in device_matches} - {None}
+        requested = sorted(normalized.required_regions)
+        parts = [f"no groups in region {', '.join(requested)}"]
+        for r in requested:
+            # Prioritize zone/region confusion over fuzzy match
+            if _looks_like_zone(r) and r in available_zones:
+                parts.append(f"'{r}' looks like a zone, not a region; use a zone constraint instead")
+            else:
+                close = difflib.get_close_matches(r, available_regions, n=1, cutoff=0.7)
+                if close:
+                    parts.append(f"did you mean {close[0]}?")
+        return "; ".join(parts)
+
+    available = ", ".join(g.name for g in groups)
+    return f"no scaling group matches constraints (available: {available})"
+
+
 def _diagnose_no_capacity(
     entry: DemandEntry,
     matching_groups: list[ScalingGroup],

lib/iris/src/iris/cluster/controller/autoscaler/runtime.py

@@ -30,7 +30,7 @@
     RemoteWorkerHandle,
     SliceHandle,
 )
-from iris.cluster.constraints import ConstraintIndex, routing_constraints
+from iris.cluster.constraints import ConstraintIndex, routing_constraints, split_hard_soft
 from iris.cluster.controller.autoscaler.models import (
     DemandEntry,
     ScalingAction,
@@ -45,7 +45,7 @@
     load_autoscaler_checkpoint,
     restore_autoscaler_state,
 )
-from iris.cluster.controller.autoscaler.routing import route_demand
+from iris.cluster.controller.autoscaler.routing import diagnose_unsatisfiable_constraints, route_demand
 from iris.cluster.controller.autoscaler.scaling_group import ScalingGroup
 from iris.cluster.controller.autoscaler.status import routing_decision_to_proto
 from iris.cluster.controller.autoscaler.worker_registry import TrackedWorker, WorkerRegistry
@@ -530,6 +530,30 @@ def get_init_log(self, vm_id: str, tail: int | None = None) -> str:
         """Get bootstrap log for a VM by platform worker ID."""
         return self._worker_registry.init_log(vm_id, tail)
 
+    def check_routing_feasibility(
+        self,
+        constraints: list[job_pb2.Constraint],
+    ) -> str | None:
+        """Check if any scaling group can satisfy the job's hard routing constraints.
+
+        Returns None if at least one group matches, or a human-readable error
+        message explaining why no group can satisfy the constraints.
+        """
+        groups = list(self._groups.values())
+        if not groups:
+            return None
+
+        group_attrs = {g.name: g.to_attributes() for g in groups}
+        group_index = ConstraintIndex.build(group_attrs)
+        routing_cs = routing_constraints(constraints)
+        hard_cs, _ = split_hard_soft(routing_cs)
+        matching_names = group_index.matching_entities(hard_cs)
+
+        if matching_names:
+            return None
+
+        return diagnose_unsatisfiable_constraints(constraints, groups)
+
     def check_coscheduling_feasibility(
         self,
         replicas: int,
@@ -543,6 +567,10 @@ def check_coscheduling_feasibility(
         Returns None if feasible, or a human-readable error message if no scaling
         group can accommodate the replica count.
         """
+        routing_error = self.check_routing_feasibility(constraints)
+        if routing_error:
+            return routing_error
+
         groups = list(self._groups.values())
         if not groups:
             return None
@@ -553,9 +581,6 @@ def check_coscheduling_feasibility(
         matching_names = group_index.matching_entities(routing_cs)
         matching_groups = [g for g in groups if g.name in matching_names]
 
-        if not matching_groups:
-            return f"no scaling group matches the job constraints; " f"available groups: {[g.name for g in groups]}"
-
         if any(replicas % g.num_vms == 0 for g in matching_groups):
             return None
 

lib/iris/src/iris/cluster/controller/service.py

@@ -1154,21 +1154,25 @@ def launch_job(
         # device-variant, etc.) replace auto-generated ones.
         request = _inject_resource_constraints(request)
 
-        # Reject coscheduled jobs that can never be scheduled: if no scaling
-        # group has num_vms matching the replica count, the job would sit in
-        # the queue forever.
-        if request.HasField("coscheduling"):
-            autoscaler = self._controller.autoscaler
-            if autoscaler is not None:
+        # Reject jobs whose hard routing constraints cannot be satisfied by
+        # any scaling group.  Without this check the job sits in the queue
+        # forever.  For coscheduled jobs, also verify the replica count is
+        # compatible with a matching group's num_vms.
+        autoscaler = self._controller.autoscaler
+        if autoscaler is not None:
+            error = autoscaler.check_routing_feasibility(
+                constraints=list(request.constraints),
+            )
+            if not error and request.HasField("coscheduling"):
                 error = autoscaler.check_coscheduling_feasibility(
                     replicas=request.replicas,
                     constraints=list(request.constraints),
                 )
-                if error:
-                    raise ConnectError(
-                        Code.FAILED_PRECONDITION,
-                        f"Job is unschedulable: {error}",
-                    )
+            if error:
+                raise ConnectError(
+                    Code.FAILED_PRECONDITION,
+                    f"Job is unschedulable: {error}",
+                )
 
         self._transitions.submit_job(job_id, request, Timestamp.now())
         self._controller.wake()

lib/iris/src/iris/cluster/providers/local/cluster.py

@@ -333,6 +333,9 @@ def make_local_cluster_config(max_workers: int) -> config_pb2.IrisClusterConfig:
             device_type=config_pb2.ACCELERATOR_TYPE_CPU,
         ),
     )
+    # Match the region that the fake GCP provider sets on local workers so
+    # that inherited region constraints from child jobs are satisfiable.
+    sg.worker.attributes["region"] = "local"
     base_config.scale_groups["local-cpu"].CopyFrom(sg)
 
     return make_local_config(base_config)

lib/iris/tests/cluster/controller/test_demand_routing.py

@@ -17,9 +17,13 @@
 )
 from iris.cluster.controller.autoscaler.scaling_group import ScalingGroup
 from iris.cluster.constraints import (
+    Constraint,
+    ConstraintOp,
     DeviceType,
     PlacementRequirements,
     WellKnownAttribute,
+    region_constraint,
+    zone_constraint,
 )
 from iris.rpc import config_pb2
 from iris.rpc import job_pb2
@@ -1332,14 +1336,103 @@ def test_infeasible_no_group_matches_constraints(self):
         autoscaler = self._make_autoscaler({"gpu-group": ScalingGroup(config, make_mock_platform())})
         result = autoscaler.check_coscheduling_feasibility(8, self._make_constraints())
         assert result is not None
-        assert "no scaling group matches" in result
+        assert "no scaling group provides" in result
 
     def test_no_groups_returns_none(self):
         """Returns None when there are no groups (no validation possible)."""
         autoscaler = self._make_autoscaler({})
         assert autoscaler.check_coscheduling_feasibility(8, []) is None
 
 
+# ---------------------------------------------------------------------------
+# Routing feasibility (submit-time constraint validation)
+# ---------------------------------------------------------------------------
+
+
+class TestCheckRoutingFeasibility:
+    """Tests for Autoscaler.check_routing_feasibility()."""
+
+    def _make_autoscaler(self, groups):
+        from iris.cluster.controller.autoscaler import Autoscaler
+
+        return Autoscaler(
+            scale_groups=groups,
+            evaluation_interval=Duration.from_seconds(0.1),
+            platform=make_mock_platform(),
+        )
+
+    def _tpu_constraints(self, variant: str = "v5p-8") -> list[job_pb2.Constraint]:
+        return make_demand_entries(1, device_type=DeviceType.TPU, device_variant=variant)[0].constraints
+
+    def test_feasible_matching_group(self):
+        """Returns None when a group matches the constraints."""
+        config = make_scale_group_config(name="tpu-group", max_slices=5, num_vms=4)
+        autoscaler = self._make_autoscaler({"tpu-group": ScalingGroup(config, make_mock_platform())})
+        assert autoscaler.check_routing_feasibility(self._tpu_constraints()) is None
+
+    def test_infeasible_wrong_device_type(self):
+        """Rejects when no group has the requested device type."""
+        config = make_scale_group_config(
+            name="gpu-group", max_slices=5, num_vms=4, accelerator_type=config_pb2.ACCELERATOR_TYPE_GPU
+        )
+        autoscaler = self._make_autoscaler({"gpu-group": ScalingGroup(config, make_mock_platform())})
+        result = autoscaler.check_routing_feasibility(self._tpu_constraints())
+        assert result is not None
+        assert "tpu" in result
+
+    def test_infeasible_wrong_region(self):
+        """Rejects when no group is in the requested region."""
+        config = make_scale_group_config(name="tpu-group", max_slices=5, num_vms=4, zones=["us-central1-a"])
+        constraints = self._tpu_constraints()
+        constraints.append(region_constraint(["europe-west4"]).to_proto())
+        autoscaler = self._make_autoscaler({"tpu-group": ScalingGroup(config, make_mock_platform())})
+        result = autoscaler.check_routing_feasibility(constraints)
+        assert result is not None
+        assert "region" in result
+
+    def test_infeasible_zone_used_as_region(self):
+        """Detects when a zone value is specified as a region constraint."""
+        config = make_scale_group_config(name="tpu-group", max_slices=5, num_vms=4, zones=["us-central1-a"])
+        constraints = self._tpu_constraints()
+        # User mistakenly passes a zone value as a region constraint
+        constraints.append(region_constraint(["us-central1-a"]).to_proto())
+        autoscaler = self._make_autoscaler({"tpu-group": ScalingGroup(config, make_mock_platform())})
+        result = autoscaler.check_routing_feasibility(constraints)
+        assert result is not None
+        assert "looks like a zone" in result
+
+    def test_infeasible_region_used_as_zone(self):
+        """Detects when a region value is specified as a zone constraint."""
+        config = make_scale_group_config(name="tpu-group", max_slices=5, num_vms=4, zones=["us-central1-a"])
+        constraints = self._tpu_constraints()
+        # User mistakenly passes a region value as a zone constraint
+        constraints.append(zone_constraint("us-central1").to_proto())
+        autoscaler = self._make_autoscaler({"tpu-group": ScalingGroup(config, make_mock_platform())})
+        result = autoscaler.check_routing_feasibility(constraints)
+        assert result is not None
+        assert "looks like a region" in result
+
+    def test_soft_constraint_does_not_reject(self):
+        """Soft constraints that don't match any group should not cause rejection."""
+        config = make_scale_group_config(name="tpu-group", max_slices=5, num_vms=4, zones=["us-central1-a"])
+        constraints = self._tpu_constraints()
+        # Add a soft region constraint for a region that doesn't exist
+        soft = Constraint(
+            key=WellKnownAttribute.REGION,
+            op=ConstraintOp.EQ,
+            value="europe-west4",
+            mode=job_pb2.CONSTRAINT_MODE_PREFERRED,
+        )
+        constraints.append(soft.to_proto())
+        autoscaler = self._make_autoscaler({"tpu-group": ScalingGroup(config, make_mock_platform())})
+        assert autoscaler.check_routing_feasibility(constraints) is None
+
+    def test_no_groups_returns_none(self):
+        """Returns None when there are no groups (no validation possible)."""
+        autoscaler = self._make_autoscaler({})
+        assert autoscaler.check_routing_feasibility([]) is None
+
+
 # ---------------------------------------------------------------------------
 # Allocation tier blocking
 # ---------------------------------------------------------------------------

lib/iris/tests/e2e/test_smoke.py

@@ -366,10 +366,14 @@ def test_dashboard_task_logs(smoke_cluster, verbose_job, smoke_page, smoke_scree
 
 def test_dashboard_constraints(smoke_cluster, smoke_page, smoke_screenshot):
     """Constraint chips rendered on job detail."""
+    # Use soft constraints to avoid submit-time routing feasibility rejection;
+    # the test only checks that constraint chips render on the dashboard.
     constraints = [
-        Constraint(key="region", op=ConstraintOp.EQ, value="local"),
+        Constraint(key="region", op=ConstraintOp.EQ, value="local", mode=job_pb2.CONSTRAINT_MODE_PREFERRED),
         Constraint(key="env-tag", op=ConstraintOp.EXISTS),
-        Constraint(key="device-variant", op=ConstraintOp.IN, values=("v5p-8", "v6e-4")),
+        Constraint(
+            key="device-variant", op=ConstraintOp.IN, values=("v5p-8", "v6e-4"), mode=job_pb2.CONSTRAINT_MODE_PREFERRED
+        ),
     ]
     with smoke_cluster.launched_job(TestJobs.quick, "smoke-constraints", constraints=constraints) as job:
         time.sleep(3)