Start dynamic worker ports above Ray's worker range

miles/utils/workers/addr_allocator.py

@@ -5,6 +5,8 @@
 
 logger = logging.getLogger(__name__)
 
+_DYNAMIC_PORT_START = 20000
+
 
 @dataclass
 class PortAllocator:
@@ -13,7 +15,7 @@ class PortAllocator:
     def alloc(self, actor, *, node_ip: str, consecutive: int = 1) -> int:
         # use small ports to prevent ephemeral port between 32768 and 65536.
         # also, ray uses port 10002-19999, thus we avoid near-10002 to avoid racing condition
-        start_port = self._next_port_of_ip.get(node_ip, 15000)
+        start_port = self._next_port_of_ip.get(node_ip, _DYNAMIC_PORT_START)
         port: int = ray.get(
             actor._get_free_port_block.remote(
                 start_port=start_port,

tests/fast/utils/workers/conftest.py

@@ -8,3 +8,15 @@ def patch_ray_get(monkeypatch):
     import miles.utils.workers.addr_allocator as mod
 
     monkeypatch.setattr(mod.ray, "get", lambda x: x)
+
+
+@pytest.fixture
+def patch_ray_get_failure(monkeypatch):
+    """Make ``ray.get(...)`` raise, mimicking a probe that is submitted
+    successfully but fails while its result is retrieved."""
+    import miles.utils.workers.addr_allocator as mod
+
+    def _raise(_object_ref):
+        raise RuntimeError("free port probe failed")
+
+    monkeypatch.setattr(mod.ray, "get", _raise)

tests/fast/utils/workers/test_addr_allocator.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import pytest
+
 from tests.fast.ray.rollout.conftest import fake_engine
 
 from miles.utils.workers.addr_allocator import PortAllocator
@@ -24,7 +26,7 @@ def test_alloc_starts_from_the_base_port_on_an_unseen_node(self, patch_ray_get):
         """A node with no cursor yet starts at the base port, away from ray's range."""
         cursors = PortAllocator()
         engine = fake_engine(host="10.0.0.1", port_seed=0)
-        assert cursors.alloc(engine, node_ip="10.0.0.1") == 15000
+        assert cursors.alloc(engine, node_ip="10.0.0.1") == 20000
 
     def test_alloc_consecutive_reserves_a_whole_block(self, patch_ray_get):
         """A consecutive=N allocation must move this node's cursor past the entire block."""
@@ -33,6 +35,31 @@ def test_alloc_consecutive_reserves_a_whole_block(self, patch_ray_get):
         first = cursors.alloc(engine, node_ip="10.0.0.1", consecutive=5)
         assert cursors._next_port_of_ip["10.0.0.1"] == first + 5
 
+    def test_a_skipped_candidate_range_advances_from_the_returned_block(self, patch_ray_get):
+        """When the actor skips occupied candidates, the cursor must advance from the returned port plus the block size."""
+        cursors = PortAllocator()
+        engine = fake_engine(host="10.0.0.1", port_seed=25000)
+        first = cursors.alloc(engine, node_ip="10.0.0.1", consecutive=4)
+        assert first == 25000
+        assert cursors._next_port_of_ip["10.0.0.1"] == 25004
+
+    def test_a_failed_probe_does_not_advance_the_node_cursor(self, patch_ray_get_failure):
+        """A probe that fails on result retrieval propagates its error and leaves this node's cursor untouched."""
+        cursors = PortAllocator()
+        cursors._next_port_of_ip["10.0.0.1"] = 20005
+        engine = fake_engine(host="10.0.0.1", port_seed=0)
+        with pytest.raises(RuntimeError, match="free port probe failed"):
+            cursors.alloc(engine, node_ip="10.0.0.1", consecutive=3)
+        assert cursors._next_port_of_ip == {"10.0.0.1": 20005}
+
+    def test_a_failed_probe_does_not_create_a_cursor_for_an_unseen_node(self, patch_ray_get_failure):
+        """A probe that fails on result retrieval leaves a never-seen node without any cursor."""
+        cursors = PortAllocator()
+        engine = fake_engine(host="10.0.0.1", port_seed=0)
+        with pytest.raises(RuntimeError, match="free port probe failed"):
+            cursors.alloc(engine, node_ip="10.0.0.1", consecutive=3)
+        assert cursors._next_port_of_ip == {}
+
     def test_alloc_tracks_nodes_independently(self, patch_ray_get):
         """Each node ip owns its own cursor."""
         cursors = PortAllocator()