fix: hold a shared socket until the child adopts it, and free exec slots at the close

Windows CI, intermittently, on 3.14 and 3.15: a popen worker died before
its handshake with WSAENOTSOCK, which reached the coordinator as a reset
connection during makegateway.  socket.share() hands the child a *blob*,
and there is no socket on the other end until it calls fromshare() -- so
closing our copy the moment the spawn returns is a race the child loses.
The POSIX comment ("the child holds its own copy now") was true for
pass_fds and quietly wrong for share.  We now close after the handshake
there, which is the point where the child has provably adopted.

The server side has the same race and is not fixed: it cannot wait for a
handshake that goes to the coordinator, and holding the socket instead
would cost the coordinator its EOF when a worker dies.  Recorded with the
option it needs (a marker byte from the worker) in HANDOFF and the roadmap.

Second, from a flake in the exec-capacity test under -n 12, which turned
out to be the feature and not the test: the admission slot was released
when the exec *task* unwound, a moment after executetask had already sent
the channel close.  That close is exactly what tells a coordinator at
capacity it may send the next request, so waitclose() + remote_exec()
could be refused for a slot that was already free.  The release moved into
_close_finished, ahead of the close, and is idempotent so the task's
finally still covers whatever never got that far.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: RonnyPfannschmidt

CHANGELOG.rst

@@ -76,6 +76,10 @@ series, once the consumers that need them have released without them.
   one anyway -- capping exactly the concurrency the profile exists to
   provide. The wait is a ``trio.Event`` woken from the exec's own thread
   now; same for the main-thread exec under ``profile=thread``.
+* The exec slot is released just before the exec's channel close goes out,
+  not after its task unwinds: that close is what tells a coordinator at
+  capacity it may send the next request, so ``waitclose()`` followed by
+  ``remote_exec()`` must not be refused for a slot that is already free.
 * **An exec that finishes after its connection died no longer takes the
   worker down.** Closing the channel is how an exec reports it finished, and
   a connection that went away first makes that raise; the exception reached
@@ -107,6 +111,11 @@ series, once the consumers that need them have released without them.
   ``EXECNET_IGNORE_VERSION_SKEW=1`` in the worker's environment (reachable
   as ``env:EXECNET_IGNORE_VERSION_SKEW=1`` in a spec) downgrades it to the
   old warning.
+* Windows: a ``popen`` coordinator kept its copy of the shared socket open
+  until the worker's handshake. ``socket.share()`` hands over a blob, not a
+  socket -- the child only has one once it calls ``fromshare()`` -- so
+  closing at spawn time raced the child into ``WSAENOTSOCK`` and it died
+  before handshaking, which the coordinator saw as a reset connection.
 * A worker that dies abruptly reports ``EOFError`` on every transport. A
   killed peer *resets* a socket -- Windows reports ``WSAECONNRESET`` --
   where a pipe would simply reach EOF, so the same event used to surface

HANDOFF.md

@@ -123,8 +123,12 @@ built.  Admission is **bounded** for the thread-shaped strategies
 (`exec_capacity()`, half the trio thread limiter) and a request over the
 line is refused on its channel, not queued — reported as
 `remote_status().execcapacity`, `None` where execs are tasks or greenlets
-and cost no thread.  Nothing may wait for an exec by parking a pool
-thread: that spends the budget it is rationing.
+and cost no thread.  Two ordering rules hold it together: nothing may wait
+for an exec by parking a pool thread (that spends the budget it is
+rationing), and the slot is released *before* the exec's channel close
+goes out (that close is what tells a coordinator at capacity to send the
+next request, so `waitclose(); remote_exec()` must not be refused for a
+slot already freed).
 `AsyncGroup.makegateway` defaults workers to `thread` — the coordinator's
 shape does not dictate the worker's.
 
@@ -196,6 +200,20 @@ shape does not dictate the worker's.
 - **Hand a socket over as a socket, never as an fd.**  Rebuilding one with
   `socket.socket(fileno=fd)` re-derives family/type/proto by querying the
   handle, which PyPy on Windows fails with `WinError 10014`.
+- **A `share()` blob is not a socket yet.**  The child has one only once it
+  calls `fromshare()`, so the sharer must keep its own copy open until then
+  or the child gets `WSAENOTSOCK` (10038) and dies before the handshake —
+  a rare, timing-dependent Windows startup failure that looks like a reset
+  connection on the coordinator (seen on the 3.14/3.15 CI jobs, 2026-08-01).
+  `connect_popen_worker` therefore closes its copy *after* the handshake on
+  Windows and before the handshake everywhere else.  **The server side has
+  the same race and is not fixed**: `serve_socket_connection` closes the
+  accepted socket as soon as the spawn returns, and it cannot see the
+  handshake, which goes to the coordinator.  Closing late instead is not
+  the answer — a server holding the connection open means a dead worker
+  stops being an EOF for the coordinator.  It needs the worker to signal
+  adoption (a marker byte on its stdout, which costs the socket worker its
+  inherited stdout), so it is a design call rather than a patch.
 - Filling in a *missing* spec value is idempotent and fine; rewriting one
   the caller set is not.  xdist reuses one spec object and re-reads it.
 - Worker teardown ends in `os._exit(0)` because trio's `to_thread` cache

ROADMAP-3.0.md

@@ -366,6 +366,15 @@ protocol is the expensive version of this.
   a new reason.
 - **Unverified**: whether the socket/`installvia` path works on the Windows
   CI job at all.  Assume any platform CI has not exercised is broken.
+- **The server-side `share()` handoff still races** (the popen one is
+  fixed; see the invariant in `HANDOFF.md`).  `serve_socket_connection`
+  closes the accepted socket when the spawn returns, which can beat the
+  worker's `fromshare()` — and it cannot wait for the handshake, which goes
+  to the coordinator, nor simply close late, which would keep a dead
+  worker's connection open and cost the coordinator its EOF.  The fix is a
+  marker byte from the worker once it has adopted; the price is that a
+  socket worker's stdout becomes a pipe to its server rather than the
+  user's.  Decide before claiming Windows `socket=` works.
 
 ## Suggested order
 

src/execnet/_gateway_base.py

@@ -270,7 +270,14 @@ def _close_finished(self, channel: Channel, error: str | None = None) -> None:
         ordinary teardown race (a killed worker, a terminate that outran the
         exec), and there is no longer anyone to raise at.  Letting the OSError
         out lands it in the exec task, whose nursery is the worker's root one.
+
+        The exec's admission slot goes back *first*: this close is also what
+        tells a coordinator at capacity that it may send the next request,
+        and it must not be able to arrive before the slot it frees.
         """
+        execpool = self._execpool
+        if execpool is not None:
+            execpool.release_slot(channel.id)
         try:
             channel.close(error)
         except OSError as exc:

src/execnet/_trio_gateway.py

@@ -1124,15 +1124,22 @@ async def connect_popen_worker(spec: Any) -> tuple[ByteStream, trio.Process]:
 
     import socket as _socket
 
+    # Windows hands the socket over by *sharing* it, and a share() blob is
+    # not a socket yet: the child only has one once it calls fromshare().
+    # Closing our copy before that races the child into WSAENOTSOCK, which
+    # it reports by dying without a handshake -- so hold it until the
+    # handshake byte says the child has adopted.  POSIX inherited the fd
+    # itself and needs the opposite: close now, or the pair never ends.
+    shared = _provision.socket_share_required()
     ours, theirs = _socket.socketpair()
     try:
         process = await _spawn_with_socket(spec, theirs)
     except BaseException:
         ours.close()
         theirs.close()
         raise
-    # The child holds its own copy now; ours would keep the pair open.
-    theirs.close()
+    if not shared:
+        theirs.close()
     stream = trio.SocketStream(trio.socket.from_stdlib_socket(ours))
     try:
         await read_handshake_ack(stream, "bootstrap")
@@ -1143,6 +1150,9 @@ async def connect_popen_worker(spec: Any) -> tuple[ByteStream, trio.Process]:
                 await process.wait()
             await stream.aclose()
         raise
+    finally:
+        if shared:
+            theirs.close()
     return stream, process
 
 

src/execnet/_trio_worker.py

@@ -388,6 +388,9 @@ def __init__(
         #: admitted and not yet finished -- the number STATUS reports, and
         #: what admission is capped on
         self._running = 0
+        #: channel ids currently holding one of those slots, so releasing is
+        #: idempotent (it happens at the close, and again when the task ends)
+        self._holding: set[int] = set()
         #: resolved on the loop at the first request (see exec_capacity)
         self._capacity: int | None = None
         self._shutting_down = False
@@ -423,8 +426,22 @@ def capacity(self) -> int | None:
             self._capacity = exec_capacity()
         return self._capacity
 
-    def _track_finish(self) -> None:
+    def release_slot(self, channelid: int) -> None:
+        """Give back the admission slot ``channelid`` holds, once.
+
+        Called twice on the ordinary path, and the *early* call is the one
+        that matters: from ``_close_finished``, just before the exec's
+        channel close goes out.  That close is how a coordinator learns it
+        may send the next request, so it must not be able to arrive before
+        the slot it frees -- otherwise ``waitclose(); remote_exec()`` on a
+        worker at capacity is refused for a slot that was already gone.
+        The task's own ``finally`` then covers everything that never got as
+        far as closing.
+        """
         with self._lock:
+            if channelid not in self._holding:
+                return
+            self._holding.discard(channelid)
             self._running -= 1
             if self._running == 0:
                 self._idle.set()
@@ -445,6 +462,7 @@ def schedule(self, channel: Channel, sourcetask: bytes) -> None:
                 full = True
             else:
                 full = False
+                self._holding.add(channel.id)
                 self._running += 1
                 self._idle.clear()
         if full:
@@ -490,7 +508,7 @@ async def _run_exec(self, channel: Channel, item: ExecItem) -> None:
         except BaseException as exc:
             trace(f"exec task for channel {channel.id} failed: {exc!r}")
         finally:
-            self._track_finish()
+            self.release_slot(channel.id)
 
     def integrate_as_primary_thread(self) -> None:
         self.strategy.integrate_as_primary_thread()

testing/test_gateway.py

@@ -719,7 +719,9 @@ def test_one_exec_too_many_is_refused_not_hung(
             over = gw.remote_exec("channel.send('running')")
             with pytest.raises(RemoteError, match="concurrency limit"):
                 over.receive(TESTTIMEOUT)
-            # and the worker still works once a slot actually frees
+            # and a slot is free the moment its close is observable: the
+            # release happens before that close goes out, so this sequence
+            # cannot be refused for a slot that is already gone
             freed = channels.pop()
             freed.send(None)
             freed.waitclose(TESTTIMEOUT)
@@ -762,6 +764,7 @@ async def main() -> int:
             gateway=None,  # type: ignore[arg-type]
             strategy=BoomStrategy(),
         )
+        pump._holding.add(DeadChannel.id)
         pump._running = 1
         pump._idle.clear()
         await pump._run_exec(DeadChannel(), ())  # type: ignore[arg-type]