Refuse a static port that something else is already serving

docs/user-guide/training-backend.md

@@ -485,10 +485,6 @@ A router sits in front of the SGLang workers. Router-side flags take a `--router
 --router-balance-abs-threshold 0   # force uniform distribution (lowers prefix-cache hit rate)
 ```
 
-Miles always starts its own router per model; external router mode
-(`--sglang-router-ip`/`--sglang-router-port`) was removed and is expected to
-return with the k8s-native mode.
-
 ---
 
 ## Further reading

miles/utils/misc.py

@@ -93,6 +93,10 @@ def _get_node_ip() -> str:
     def _get_free_port_block(*, start_port: int, count: int) -> int:
         return get_free_port(start_port=start_port, consecutive=count)
 
+    @staticmethod
+    def _is_port_available(*, port: int) -> bool:
+        return is_port_available(port)
+
     @staticmethod
     def _get_gpu_uuids(gpu_ids: list[int]) -> list[str | None]:
         return get_gpu_uuids(gpu_ids)

miles/utils/test_utils/mock_sglang_engine.py

@@ -92,6 +92,10 @@ def _get_node_ip(self):
         self._record("_get_node_ip", (), {})
         return NodeProbeMixin._get_node_ip()
 
+    def _is_port_available(self, *, port: int) -> bool:
+        self._record("_is_port_available", (), {"port": port})
+        return True
+
     def _get_gpu_uuids(self, gpu_ids: list[int]):
         self._record("_get_gpu_uuids", (gpu_ids,), {})
         return [None] * len(gpu_ids)

miles/utils/workers/ray_worker_manager.py

@@ -148,15 +148,22 @@ async def alloc_ports(self) -> None:
 
         node_ip = await self.actor_handle._get_node_ip.remote()
         for port_info in self.spec.port_infos:
-            port = (
-                self.manager_ref.port_allocator.alloc(
+            if port_info.allow_dynamic:
+                port = self.manager_ref.port_allocator.alloc(
                     self.actor_handle, node_ip=node_ip, consecutive=port_info.num_consecutive
                 )
-                if port_info.allow_dynamic
-                else port_info.static_port + (self.cell_index if port_info.offset_by_cell else 0)
-            )
+            else:
+                port = port_info.static_port + (self.cell_index if port_info.offset_by_cell else 0)
+                await self._assert_static_port_is_free(port=port, port_name=port_info.name, node_ip=node_ip)
             self.self_addrs[port_info.name] = HostAndPort(host=_wrap_ipv6(node_ip), port=port)
 
+    async def _assert_static_port_is_free(self, *, port: int, port_name: str, node_ip: str) -> None:
+        free = await self.actor_handle._is_port_available.remote(port=port)
+        assert free, (
+            f"Port {port} on {node_ip} is already in use, so {self.name} cannot serve its {port_name!r} "
+            f"endpoint there; a stale process from an earlier run is the usual cause"
+        )
+
     async def post_setup(self) -> None:
         ctx = LaunchCommandContext(
             cell_index=self.cell_index,

tests/fast/utils/test_utils/test_mock_sglang_engine.py

@@ -15,7 +15,7 @@ def _public_methods(cls) -> set[str]:
     """Public methods (and known semi-private helpers) of ``cls``."""
     if hasattr(cls, "__ray_actor_class__"):
         cls = cls.__ray_actor_class__  # unwrap @ray.remote
-    keep_underscored = {"_get_free_port_block", "_get_node_ip", "_get_gpu_uuids", "_collect_env_report"}
+    keep_underscored = {name for name in vars(NodeProbeMixin) if not name.startswith("__")}
     return {
         name
         for name, _ in inspect.getmembers(cls, predicate=inspect.isfunction)
@@ -95,6 +95,16 @@ def test_signature_compat_for_every_shared_method(self) -> None:
         )
 
 
+class TestNodeProbeHelpersAreCovered:
+    def test_every_node_probe_helper_takes_part_in_the_contract_comparison(self) -> None:
+        """A probe helper left out of the comparison set lets the mock silently lose a stub the real actor has."""
+        probe_helpers = {name for name in vars(NodeProbeMixin) if not name.startswith("__")}
+
+        assert probe_helpers
+        assert probe_helpers <= _public_methods(CommandActor)
+        assert probe_helpers <= _public_methods(MockSGLangEngine)
+
+
 # ----------------------------- real Ray smoke tests -----------------------------
 
 
@@ -139,6 +149,15 @@ def test_fault_injection_round_trips_through_ray(self, ray_local_mode):
             ray.kill(actor)
 
 
+class TestPortProbe:
+    def test_the_mock_reports_every_port_as_free_and_records_the_probe(self) -> None:
+        """A mock-driven launch must never be refused a pinned port by the real machine's port state."""
+        engine = MockSGLangEngine.__ray_actor_class__()
+
+        assert engine._is_port_available(port=30001) is True
+        assert engine.get_calls() == [("_is_port_available", (), {"port": 30001})]
+
+
 class TestNodeAddress:
     def test_the_mock_reports_the_node_ip_instead_of_loopback(self) -> None:
         """A mock engine placed on another node must publish an address its peers can reach."""

tests/fast/utils/workers/fake_ray.py

@@ -6,6 +6,7 @@
 
 _ASYNC_METHOD_NODE_IP = "_get_node_ip"
 _ASYNC_METHOD_FREE_PORT_BLOCK = "_get_free_port_block"
+_ASYNC_METHOD_IS_PORT_AVAILABLE = "_is_port_available"
 
 EVENT_CREATE = "create"
 EVENT_KILL = "kill"
@@ -171,6 +172,8 @@ def last_event_index(self, event: str) -> int:
     def _compute_value(self, *, handle: FakeRayActorHandle, method: str, kwargs: dict[str, Any]) -> Any:
         if method == _ASYNC_METHOD_NODE_IP:
             return handle.node_ip
+        if method == _ASYNC_METHOD_IS_PORT_AVAILABLE:
+            return kwargs["port"] not in self._used_ports.get(handle.node_ip, set())
         if method == _ASYNC_METHOD_FREE_PORT_BLOCK:
             return self._alloc_port_block(
                 node_ip=handle.node_ip, start_port=kwargs["start_port"], count=kwargs["count"]

tests/fast/utils/workers/test_ray_worker_manager.py

@@ -176,7 +176,7 @@ async def test_a_consecutive_port_block_is_reserved_as_a_whole(self, fake_ray_cl
         assert [call.kwargs["count"] for call in fake_ray_cluster.calls_of("_get_free_port_block")] == [5, 5]
 
     async def test_static_ports_bypass_the_allocator(self, fake_ray_cluster: FakeRayCluster):
-        """A port the worker cannot choose is taken from the spec verbatim, without probing the node."""
+        """A port the worker cannot choose is taken from the spec verbatim, without asking the allocator."""
         spec = _make_spec(
             "router",
             port_infos=[
@@ -188,6 +188,15 @@ async def test_static_ports_bypass_the_allocator(self, fake_ray_cluster: FakeRay
 
         assert manager.get_worker_addr("router-0-0").port == 7777
         assert len(fake_ray_cluster.calls_of("_get_free_port_block")) == 1
+        assert [call.kwargs["port"] for call in fake_ray_cluster.calls_of("_is_port_available")] == [7777]
+
+    async def test_a_static_port_already_taken_on_the_node_fails_the_launch(self, fake_ray_cluster: FakeRayCluster):
+        """A stale listener on the pinned port would otherwise be mistaken for our own worker."""
+        fake_ray_cluster.occupy_ports("10.0.0.1", 7777)
+        spec = _make_spec("router", port_infos=[PortInfo(name="primary", static_port=7777, allow_dynamic=False)])
+
+        with pytest.raises(AssertionError, match="7777 on 10.0.0.1 is already in use"):
+            await _launch([spec])
 
     async def test_ports_are_tracked_per_node(self, fake_ray_cluster: FakeRayCluster):
         """Workers on different nodes may reuse the same port number."""
@@ -214,6 +223,119 @@ async def test_ipv6_hosts_are_bracketed(self, fake_ray_cluster: FakeRayCluster):
         assert manager.get_worker_addr("router-0-0").host == "[2001:db8::7]"
 
 
+class TestStaticPortsAreProbedBeforeUse:
+    async def test_the_probe_asks_for_the_cell_offset_port_rather_than_the_spec_base(
+        self, fake_ray_cluster: FakeRayCluster
+    ):
+        """Probing the spec's base port would leave every cell but the first checking an address it never binds."""
+        spec = _make_spec(
+            "session-server",
+            num_cells=3,
+            port_infos=[PortInfo(name="primary", static_port=7000, allow_dynamic=False, offset_by_cell=True)],
+        )
+        await _launch([spec])
+
+        assert [call.kwargs["port"] for call in fake_ray_cluster.calls_of("_is_port_available")] == [7000, 7001, 7002]
+
+    async def test_an_occupied_cell_offset_port_refuses_the_launch(self, fake_ray_cluster: FakeRayCluster):
+        """A stale listener on a later cell's offset port must be caught too, not only one on the base port."""
+        fake_ray_cluster.occupy_ports("10.0.0.1", 7002)
+        spec = _make_spec(
+            "session-server",
+            num_cells=3,
+            port_infos=[PortInfo(name="primary", static_port=7000, allow_dynamic=False, offset_by_cell=True)],
+        )
+
+        with pytest.raises(AssertionError, match="Port 7002 on 10.0.0.1 is already in use"):
+            await _launch([spec])
+
+    async def test_the_probe_runs_on_the_node_the_worker_landed_on(self, fake_ray_cluster: FakeRayCluster):
+        """A port is only contended on the node that binds it, so the probe must go through the worker's own actor."""
+        fake_ray_cluster.use_node_ips("10.0.0.1", "10.0.0.2")
+        fake_ray_cluster.occupy_ports("10.0.0.2", 7777)
+        spec = _make_spec(
+            "router",
+            num_workers_per_cell=2,
+            port_infos=[PortInfo(name="primary", static_port=7777, allow_dynamic=False)],
+        )
+
+        with pytest.raises(AssertionError, match="Port 7777 on 10.0.0.2 is already in use"):
+            await _launch([spec])
+
+        assert [call.handle.node_ip for call in fake_ray_cluster.calls_of("_is_port_available")] == [
+            "10.0.0.1",
+            "10.0.0.2",
+        ]
+
+    async def test_a_port_occupied_on_an_unrelated_node_does_not_block_the_launch(
+        self, fake_ray_cluster: FakeRayCluster
+    ):
+        """Refusing a port because another machine uses it would make every run hostage to unrelated nodes."""
+        fake_ray_cluster.occupy_ports("10.0.0.2", 7777)
+        spec = _make_spec("router", port_infos=[PortInfo(name="primary", static_port=7777, allow_dynamic=False)])
+
+        manager = await _launch([spec])
+
+        assert manager.get_worker_addrs("router-0-0")["primary"].port == 7777
+
+    async def test_a_master_static_port_is_probed_only_by_the_worker_that_reserves_it(
+        self, fake_ray_cluster: FakeRayCluster
+    ):
+        """Peers that never bind the cell's master port would otherwise refuse a port their own worker 0 is serving."""
+        spec = _make_spec(
+            "engine",
+            num_cells=2,
+            num_workers_per_cell=3,
+            port_infos=[
+                PortInfo(name="primary", static_port=8000, allow_dynamic=True),
+                PortInfo(name="dist_init", static_port=9000, mode="master", allow_dynamic=False),
+            ],
+        )
+        await _launch([spec])
+
+        assert [call.kwargs["port"] for call in fake_ray_cluster.calls_of("_is_port_available")] == [9000, 9000]
+
+    async def test_dynamic_ports_are_never_probed_by_the_static_check(self, fake_ray_cluster: FakeRayCluster):
+        """The allocator already hands out a free port, and re-probing its own reservation would refuse it."""
+        spec = _make_spec(
+            "engine",
+            num_cells=2,
+            num_workers_per_cell=2,
+            port_infos=[
+                PortInfo(name="primary", static_port=8000, allow_dynamic=True),
+                PortInfo(name="nccl", static_port=10000, allow_dynamic=True),
+            ],
+        )
+        await _launch([spec])
+
+        assert fake_ray_cluster.calls_of("_is_port_available") == []
+
+    async def test_the_refusal_names_the_worker_and_the_endpoint_it_could_not_serve(
+        self, fake_ray_cluster: FakeRayCluster
+    ):
+        """An operator hunting the stale process needs to know which worker wanted which endpoint."""
+        fake_ray_cluster.occupy_ports("10.0.0.1", 7777)
+        spec = _make_spec("router", port_infos=[PortInfo(name="prometheus", static_port=7777, allow_dynamic=False)])
+
+        with pytest.raises(AssertionError, match="router-0-0 cannot serve its 'prometheus' endpoint"):
+            await _launch([spec])
+
+    async def test_a_refused_port_rolls_the_cell_back_before_any_command_runs(self, fake_ray_cluster: FakeRayCluster):
+        """A refusal that let the commands start would add our own processes to the stale ones already there."""
+        fake_ray_cluster.occupy_ports("10.0.0.1", 7777)
+        spec = _make_spec(
+            "router",
+            num_workers_per_cell=2,
+            port_infos=[PortInfo(name="primary", static_port=7777, allow_dynamic=False)],
+        )
+
+        with pytest.raises(AssertionError):
+            await _launch([spec])
+
+        assert fake_ray_cluster.calls_of("run") == []
+        assert [handle.killed for handle in fake_ray_cluster.handles] == [True, True]
+
+
 class TestPortAllocationDetails:
     async def test_the_allocator_probes_the_workers_own_actor_on_its_own_node(self, fake_ray_cluster: FakeRayCluster):
         """Ports must be probed on the node that will bind them, through that worker's own actor."""