fix(multi-host): dist init port collision and lost failure signal

- DIST_INIT_PORT 30000 -> 10011 to match the dist-init port published
  by the headless service and avoid collision with the HTTP server
  default (any suite leaving ModelRunConfig.port at 30000 would clash).

- run_model_run now wraps the full server lifecycle in one try/finally
  so rank 0 publishes a terminal state on every path, including
  popen_launch_server failures that previously left workers polling
  /status until wait_for_done's 60-min timeout.

- _publish_state writes exit_code before done so a worker that observes
  done=True is guaranteed to read the matching exit_code in the same
  poll, eliminating a race where success was reported despite rank 0
  failing.

- _reset_state at run entry prevents state from one run leaking into
  the next.

Co-Authored-By: Claude Opus 4.7 <[email protected]>

Author: zhengkezhou1

test/srt/multi_host/run_suite.py

@@ -29,13 +29,30 @@
     dry_run_suite,
 )
 
-DIST_INIT_PORT = 30000
+DIST_INIT_PORT = 10011
 SERVER_PORT = 30000
 CONTROL_PORT = 18080
 
 _control_state = {"done": False, "exit_code": 0}
 
 
+def _publish_state(exit_code: int) -> None:
+    """Set terminal state visible to worker ranks via /status.
+
+    Writes exit_code BEFORE done so any worker that observes ``done == True``
+    is guaranteed to read the matching exit_code in the same poll. Without
+    this ordering a worker can race the two assignments and report success
+    while rank 0 actually failed.
+    """
+    _control_state["exit_code"] = exit_code
+    _control_state["done"] = True
+
+
+def _reset_state() -> None:
+    _control_state["done"] = False
+    _control_state["exit_code"] = 0
+
+
 def _log(message: str) -> None:
     print(f"[multi-host-suite] {message}", flush=True)
 
@@ -223,44 +240,49 @@ def run_model_run(model_run: ModelRun, runtime_cfg: RuntimeConfig) -> int:
     _log(
         f"Launching model run={model_run.name}, rank={runtime_cfg.node_rank}, port={runtime_cfg.port}"
     )
-    control_server = start_control_server() if runtime_cfg.node_rank == 0 else None
-    base_url = f"http://{runtime_cfg.host}:{runtime_cfg.port}"
-    server_process = popen_launch_server(
-        model=model_run.model.model_path,
-        base_url=base_url,
-        timeout=1800,
-        other_args=build_other_server_args(model_run.model, runtime_cfg),
-    )
+    is_rank0 = runtime_cfg.node_rank == 0
+    _reset_state()
+    control_server = start_control_server() if is_rank0 else None
+    server_process = None
     exit_code = 0
 
     try:
-        if runtime_cfg.node_rank == 0:
+        base_url = f"http://{runtime_cfg.host}:{runtime_cfg.port}"
+        server_process = popen_launch_server(
+            model=model_run.model.model_path,
+            base_url=base_url,
+            timeout=1800,
+            other_args=build_other_server_args(model_run.model, runtime_cfg),
+        )
+
+        if is_rank0:
             for case in model_run.cases:
                 run_case(case, model_run.model.model_path, runtime_cfg.port)
-            _control_state["done"] = True
-            _control_state["exit_code"] = 0
         else:
             workload_name = _get_env("WORKLOAD_NAME")
             headless_service_name = _get_env("HEADLESS_SERVICE_NAME")
             control_url = f"http://{workload_name}-0.{headless_service_name}:{CONTROL_PORT}/status"
             exit_code = wait_for_done(control_url, server_process)
     except Exception:
         exit_code = 1
-        if runtime_cfg.node_rank == 0:
-            _control_state["done"] = True
-            _control_state["exit_code"] = exit_code
         raise
     finally:
-        if runtime_cfg.node_rank == 0:
+        if is_rank0:
+            # Always publish — covers success, case failure, and popen_launch_server
+            # failure (where server_process never got assigned). Without this,
+            # workers spin on /status until wait_for_done's 60-min timeout when
+            # rank 0 dies during launch.
+            _publish_state(exit_code)
             _log("Keeping control server alive for worker ranks")
             time.sleep(30)
-        _log("Stopping server process")
-        kill_process_tree(server_process.pid)
-        try:
-            server_process.wait(timeout=5)
-        except subprocess.TimeoutExpired:
-            server_process.kill()
-            server_process.wait()
+        if server_process is not None:
+            _log("Stopping server process")
+            kill_process_tree(server_process.pid)
+            try:
+                server_process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                server_process.kill()
+                server_process.wait()
         if control_server is not None:
             control_server.shutdown()