fix(sim): surface worker errors from the ParallelSimWrapper handshake

CHANGELOG.md

@@ -37,6 +37,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - `RLTaskEnv.step` publishes the *terminal* observation in `info["observations"]["raw"]["obs"]` instead of the episode's first one (off-policy truncation bootstraps in clean_rl SAC/TD3 and FastTD3 read it).
 - IsaacGym and PyBullet reported `joint_pos_target` in native DoF order while `joint_pos` is in sorted-name order; both now use `get_joint_names(sort=True)` (completes #12).
+- `ParallelSimWrapper`: a worker that died during handler construction or `launch` surfaced as a bare `EOFError`/`ConnectionResetError` from the handshake and left the other workers running; the handshake now raises the worker's own traceback, `close()` tolerates dead workers, and a failed constructor tears the pool down.
 - MuJoCo: `<size memory="512M">` is reserved by default; humanoid + mesh scenes no longer die with
   `mj_stackAlloc: out of memory` (get_started/10_mount_camera.py).
 - `hf_util`: a symlinked `roboverse_data` is no longer refused as path traversal; concurrent

metasim/sim/parallel.py

@@ -5,6 +5,7 @@
 import multiprocessing as mp
 import platform
 import sys
+import time
 import traceback
 from copy import deepcopy
 from functools import partial
@@ -151,11 +152,15 @@ def __init__(self, scenario: ScenarioCfg, extra_spec: dict[str, BaseQueryType] |
                 self.processes.append(process)
                 work_remote.close()
 
-            # To make sure environments are initialized in all workers
-            for remote in self.remotes:
-                remote.send(("handshake", (None,)))
-            for remote in self.remotes:
-                remote.recv()
+            # Every worker must reach its command loop before the wrapper is usable. A worker whose
+            # handler construction raised has already exited, so the handshake reads through
+            # ``_recv_or_surface`` and raises its real traceback (from ``error_queue``) instead of
+            # a bare ``EOFError``; the surviving workers are torn down so nothing is left behind.
+            try:
+                self._handshake()
+            except BaseException:
+                self._terminate_workers()
+                raise
 
         def _check_error(self):
             """Drain the error queue and detect dead workers.
@@ -195,6 +200,7 @@ def _recv_or_surface(self, remote_idx: int):
                 return self.remotes[remote_idx].recv()
             except (EOFError, ConnectionResetError, BrokenPipeError) as err:
                 # Worker died — _check_error will raise with the real message.
+                self._wait_for_error_report()
                 self._check_error()
                 raise RuntimeError(
                     f"Parallel worker {remote_idx} closed its pipe without reporting an error. "
@@ -203,23 +209,58 @@ def _recv_or_surface(self, remote_idx: int):
 
         def launch(self):
             for remote in self.remotes:
-                remote.send(("launch", (None,)))
+                self._send(remote, ("launch", (None,)))
             self.waiting = False
-            # ``launch`` sends no reply; handshake round-trips so every
-            # worker has finished launching before the wrapper returns.
-            for remote in self.remotes:
-                remote.send(("handshake", (None,)))
-            for remote in self.remotes:
-                remote.recv()
-            self._check_error()
+            # ``launch`` sends no reply; the handshake round-trips so every worker has finished
+            # launching before the wrapper returns, and surfaces a worker that died in ``launch``.
+            self._handshake()
 
         def close(self):
             if self.closed:
                 return
             for remote in self.remotes:
-                remote.send(("close", (None,)))
+                self._send(remote, ("close", (None,)))
+            self._terminate_workers()
+
+        @staticmethod
+        def _send(remote: Connection, msg) -> None:
+            """``send`` that tolerates a dead worker: the following ``recv`` / ``_check_error``
+            reports the worker's real error, which a ``BrokenPipeError`` here would only hide."""
+            try:
+                remote.send(msg)
+            except (BrokenPipeError, ConnectionResetError, EOFError, OSError):
+                pass
+
+        def _handshake(self) -> None:
+            """Round-trip every worker; raises the worker's own error if one has died."""
+            for remote in self.remotes:
+                self._send(remote, ("handshake", (None,)))
+            for idx in range(len(self.remotes)):
+                self._recv_or_surface(idx)
+            self._check_error()
+
+        def _wait_for_error_report(self, timeout: float = 2.0) -> None:
+            """Give a dying worker's queue feeder thread time to deliver its traceback.
+
+            ``mp.Queue.put`` hands the item to a background thread; the pipe end can close before
+            the report is readable on this side, which would turn a real traceback into
+            "died without reporting an error".
+            """
+            deadline = time.monotonic() + timeout
+            while self.error_queue.empty() and time.monotonic() < deadline:
+                if any(p.is_alive() for p in self.processes):
+                    time.sleep(0.01)
+                else:
+                    break
+
+        def _terminate_workers(self, join_timeout: float = 10.0) -> None:
+            """Join every worker, force-terminating any that does not exit in time."""
+            for process in self.processes:
+                process.join(timeout=join_timeout)
             for process in self.processes:
-                process.join()
+                if process.is_alive():
+                    process.terminate()
+                    process.join(timeout=join_timeout)
             self.closed = True
 
         def _set_states(self, states: list[DictEnvState], env_ids: list[int] | None = None) -> None:

metasim/test/test_parallel_error_handling_general.py

@@ -100,6 +100,12 @@ def __init__(self, alive: bool = True, exitcode: int | None = None):
     def is_alive(self) -> bool:
         return self._alive
 
+    def join(self, timeout: float | None = None) -> None:
+        pass
+
+    def terminate(self) -> None:
+        self._alive = False
+
 
 def _make_parallel_stub_handler(remotes, processes, error_queue=None) -> Any:
     """Build a ``ParallelHandler`` instance that bypasses ``__init__`` so
@@ -217,3 +223,53 @@ def test_set_seed_forwards_to_every_worker():
         cmd, payload = remote.sent[-1]
         assert cmd == "set_seed", f"worker {i} got {cmd!r}, expected 'set_seed'"
         assert payload == (123,), f"worker {i} got payload {payload!r}, expected (123,)"
+
+
+@pytest.mark.general
+def test_launch_surfaces_worker_error_instead_of_eof():
+    """A worker that dies inside ``launch`` used to surface as a bare ``EOFError`` from the
+    handshake ``recv``; the real traceback in ``error_queue`` must be what the caller sees."""
+    err_queue = _SyncQueue()
+    err_queue.put(("FileNotFoundError", "missing.usd", ["worker traceback\n"]))
+    handler = _make_parallel_stub_handler(
+        remotes=[_FakeRemote(recv_raises=EOFError)],
+        processes=[_FakeProcess(alive=False, exitcode=1)],
+        error_queue=err_queue,
+    )
+    with pytest.raises(RuntimeError, match=r"Parallel worker error \(FileNotFoundError\): missing\.usd"):
+        handler.launch()
+
+
+@pytest.mark.general
+def test_close_tolerates_dead_worker():
+    """``close`` must not raise ``BrokenPipeError`` on a worker that already died."""
+
+    class _DeadRemote(_FakeRemote):
+        def send(self, msg):
+            raise BrokenPipeError
+
+    handler = _make_parallel_stub_handler(
+        remotes=[_DeadRemote()],
+        processes=[_FakeProcess(alive=False, exitcode=1)],
+    )
+    handler.close()
+    assert handler.closed
+
+
+class _BoomHandler(_StubBaseHandler):
+    """Handler whose construction fails, as an asset-loading error inside a worker would."""
+
+    def __init__(self, *_a, **_k):
+        raise RuntimeError("asset missing: boom.urdf")
+
+
+@pytest.mark.general
+def test_constructor_reports_worker_construction_error():
+    """End-to-end with real worker processes: a handler that raises in ``__init__`` used to
+    surface as ``EOFError`` from the constructor handshake, with the traceback lost."""
+    from metasim.scenario.scenario import ScenarioCfg
+
+    scenario = ScenarioCfg(num_envs=2)
+    with pytest.raises(RuntimeError, match=r"asset missing: boom\.urdf") as excinfo:
+        ParallelSimWrapper(_BoomHandler)(scenario)
+    assert "EOFError" not in str(excinfo.value)