fix: survive a host that cannot hand a socket to a worker (PyPy on Windows)

16 of 17 CI jobs are green; this is the seventeenth.  PyPy on Windows has
`socket.share` as a name but the call does not work, so the capability
check passed and the failure surfaced three steps later as `EOFError: bad
socket handshake: b''` -- with the master gateway dead and no explanation
anywhere.

Three separate defects behind that one symptom:

* the capability is now settled by *doing* it once, against our own pid,
  rather than by `hasattr`.  A host that cannot hand over a socket refuses
  the request up front, where a reason can still reach the coordinator.
* a failed socket gateway no longer kills the gateway it was requested
  through.  It runs as a task on that worker's host, so asking for one
  unsupported sub-gateway cost the master too -- which is why the errors
  cascaded across 51 tests instead of failing one.
* `channel.send`/`receive` check for a foreign event loop before checking
  whether the channel is closed.  Both are caller bugs, but which one you
  were told about depended on whether the peer had closed yet, so the
  event-loop diagnostic lost a race on the slower interpreter.

The socket transport simply cannot work where neither `pass_fds` nor a
working `share()` exists, so those gateways skip there with that reason
rather than failing.  Removing the limitation means not handing the
socket over at all -- spawning the worker as the listener and reporting
its address back -- which is a bigger change than this one.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: RonnyPfannschmidt

CHANGELOG.rst

@@ -45,6 +45,21 @@
   by name. The launch command no longer needs ``head -c <N>`` byte
   accounting, a ``mktemp`` prelude, or ``exec`` to keep an fd alive, and the
   protocol stream never carries a payload.
+* ``channel.send()`` and ``channel.receive()`` check for a foreign event
+  loop before they check whether the channel is still open. Calling a
+  blocking API from inside a loop is a caller bug either way, and which of
+  the two errors you got depended on whether the peer had closed yet --
+  so the more useful message lost a race.
+* Whether a socket can be handed to a worker is now settled by *doing* it
+  once rather than by looking for ``socket.share``. An implementation with
+  the name but not a working call -- PyPy on Windows -- otherwise passed
+  the check and failed later, at the point where the only thing left to
+  tell the coordinator was a closed socket. Such a host now refuses the
+  request up front, and ``socket=``/``installvia=`` gateways are skipped
+  there rather than failing.
+* A socket gateway that fails to start no longer takes down the gateway it
+  was requested through. It ran as a task on that worker's host, so an
+  unsupported sub-gateway used to cost the master as well.
 * ``execnet server :0`` reported a port nothing was listening on. Binding a
   wildcard host with an ephemeral port gives *each* address family its own
   random port, and only the first was reported -- so a client dialling the

src/execnet/_channel.py

@@ -328,9 +328,12 @@ def send(self, item: object) -> None:
 
         OSError is raised if the write pipe was prematurely closed.
         """
+        # before the state check: calling a blocking API from inside an event
+        # loop is a bug in the caller either way, and whether the channel has
+        # closed yet is a race -- the diagnostic should not depend on it
+        self.gateway._check_event_loop("channel.send()")
         if self.isclosed():
             raise OSError(f"cannot send to {self!r}")
-        self.gateway._check_event_loop("channel.send()")
         self.gateway._send(Message.CHANNEL_DATA, self.id, dumps_internal(item))
 
     def receive(self, timeout: float | None = None) -> Any:
@@ -344,10 +347,10 @@ def receive(self, timeout: float | None = None) -> Any:
         reraised as channel.RemoteError exceptions containing
         a textual representation of the remote traceback.
         """
+        self.gateway._check_event_loop("channel.receive()")
         mailbox = self._mailbox
         if mailbox is None:
             raise OSError("cannot receive(), channel has receiver callback")
-        self.gateway._check_event_loop("channel.receive()")
         x = mailbox.get(timeout)
         if x is ENDMARKER:
             mailbox.put(x)  # for other receivers

src/execnet/_provision.py

@@ -43,6 +43,7 @@
 TRANSPORTS = ("socket", "stdio")
 
 
+@cache
 def socket_handoff_available() -> bool:
     """Whether we can hand a socket to a worker process we spawn ourselves.
 
@@ -51,11 +52,26 @@ def socket_handoff_available() -> bool:
     (``WSADuplicateSocket``) duplicates the socket into a named pid, and the
     resulting blob rides in the worker config -- see the ``share`` transport
     in ``_trio_worker``.
+
+    The Windows answer is settled by *doing* it once, against our own pid,
+    rather than by looking for the method.  An implementation that has the
+    name but not a working call -- PyPy on Windows -- would otherwise pass
+    the check and fail at the point where the only thing left to tell the
+    coordinator is a closed socket.
     """
     import socket as _socket
 
-    if sys.platform.startswith("win"):
-        return hasattr(_socket.socket, "share")
+    if not socket_share_required():
+        return True
+    try:
+        left, right = _socket.socketpair()
+        try:
+            left.share(os.getpid())  # type: ignore[attr-defined]  # Windows
+        finally:
+            left.close()
+            right.close()
+    except Exception:
+        return False
     return True
 
 

src/execnet/_trio_host.py

@@ -901,7 +901,14 @@ async def _start_socket_and_reply(
     stream = await listeners[0].accept()
     for listener in listeners:
         await listener.aclose()
-    await serve_socket_connection(stream, reap=True)
+    try:
+        await serve_socket_connection(stream, reap=True)
+    except Exception as exc:
+        # This runs as a task on *this worker's* host: letting it propagate
+        # tears the whole gateway down, so a coordinator asking for one
+        # unsupported sub-gateway would lose the master it asked through.
+        # The connection is already closed, so the coordinator gets its EOF.
+        trace(f"socket gateway for channel {channelid} failed: {exc!r}")
 
 
 def handle_start_socket(gateway: BaseGateway, channelid: int, data: bytes) -> None:

testing/conftest.py

@@ -12,6 +12,7 @@
 
 import execnet
 from execnet import Gateway
+from execnet import _provision
 from execnet._execmodel import ExecModel
 from execnet._execmodel import get_execmodel
 
@@ -188,6 +189,11 @@ def gw(
         if request.param == "popen":
             gw = group.makegateway("popen//id=popen//profile=%s" % profile)
         elif request.param == "socket":
+            if not _provision.socket_handoff_available():
+                # the server accepts the connection and must then give it to
+                # a worker process; where neither pass_fds nor a working
+                # socket.share() exists (PyPy on Windows) there is no way to
+                pytest.skip("this interpreter cannot hand a socket to a worker")
             pname = "sproxy1"
             if pname not in group:
                 proxygw = group.makegateway("popen//id=%s" % pname)

testing/test_socketserver_cli.py

@@ -14,12 +14,19 @@
 import pytest
 
 import execnet
+from execnet import _provision
 
 SERVER = shutil.which("execnet-socketserver")
 
-pytestmark = pytest.mark.skipif(
-    SERVER is None, reason="execnet-socketserver console script not installed"
-)
+pytestmark = [
+    pytest.mark.skipif(
+        SERVER is None, reason="execnet-socketserver console script not installed"
+    ),
+    pytest.mark.skipif(
+        not _provision.socket_handoff_available(),
+        reason="the server must hand the accepted socket to a worker process",
+    ),
+]
 
 
 @pytest.fixture

testing/test_xspec.py

@@ -253,6 +253,10 @@ def test_socket_second(
         assert rinfo.cwd == rinfo2.cwd
         assert rinfo.version_info == rinfo2.version_info
 
+    @pytest.mark.skipif(
+        not _provision.socket_handoff_available(),
+        reason="the server must hand the accepted socket to a worker process",
+    )
     def test_socket_installvia(self) -> None:
         group = execnet.Group()
         group.makegateway("popen//id=p1")