Raise startup failure outside nursery

_trio_parallel_workers/__init__.py

@@ -61,11 +61,11 @@ def worker_behavior(recv_pipe, send_pipe, idle_timeout, init, retire):
     # between processes. (Trio will take charge via cancellation.)
     signal.signal(signal.SIGINT, signal.SIG_IGN)
     try:
-        if isinstance(init, bytes):  # true except on "fork"
-            # Signal successful startup to spawn/forkserver parents.
+        if sys.platform == "win32":
+            # Signal successful startup.
             send_pipe.send_bytes(ACK)
+        if isinstance(init, bytes):  # true except on "fork"
             init = loads(init)
-        if isinstance(retire, bytes):  # true except on "fork"
             retire = loads(retire)
         init()
         while safe_poll(recv_pipe, idle_timeout):

trio_parallel/_proc.py

@@ -69,23 +69,29 @@ async def start(self):
         self._child_send_pipe.close()
         self._child_recv_pipe.close()
 
-        # The following is mainly needed in the case of accidental recursive spawn
-        async def wait_then_fail():
-            await self.wait()
-            raise BrokenWorkerProcessError("Worker failed to start", self.proc)
+        if sys.platform != "win32":
+            return
 
-        async with trio.open_nursery() as nursery:
-            nursery.start_soon(wait_then_fail)
+        # Give a nice error on accidental recursive spawn instead of hanging
+        async def wait_for_ack():
             try:
                 code = await self._receive_chan.receive()
+                assert code == tp_workers.ACK
             except BaseException:
                 self.kill()
                 with trio.CancelScope(shield=True):
                     await self.wait()  # noqa: TRIO102
                 raise
-            assert code == tp_workers.ACK
             nursery.cancel_scope.cancel()
 
+        exitcode = None
+        async with trio.open_nursery() as nursery:
+            nursery.start_soon(wait_for_ack)
+            exitcode = await self.wait()
+            nursery.cancel_scope.cancel()
+        if exitcode is not None:
+            raise BrokenWorkerProcessError("Worker failed to start", self.proc)
+
     async def run_sync(self, sync_fn: Callable, *args) -> Optional[Outcome]:
         try:
             job = dumps((sync_fn, args), protocol=HIGHEST_PROTOCOL)

trio_parallel/_tests/test_defaults.py

@@ -214,6 +214,8 @@ def test_startup_failure_doesnt_hang(tmp_path):
     )
     assert not result.stdout
     assert b"An attempt has been made to start a new process" in result.stderr
+    assert b"ExceptionGroup" not in result.stderr
+    assert b"MultiError" not in result.stderr
     assert result.returncode