fix(node): datagrams straight through the socket on Windows' event loop

The ProactorEventLoop's datagram transport keeps one overlapped operation
per direction and starts the next from the previous one's completion, on a
later turn of the loop: one datagram sent and one received per socket and
per turn, however many wait. The busier the node, the slower every
connection, with cores to spare. Measured on loopback with the loop busy
0.5 ms a turn: 1,928 datagrams sent a second and 1,954 received, 87 % of
the rest dropped by the kernel. A browser's download from a node holding
its connections ran at 9 MB/s, the node at 45 % of a core, SCTP round trip
745 ms. Now a datagram goes out with a non-blocking sendto when nothing is
queued, and a completed read is followed by whatever the socket holds
before the next one is posted. Two-core machine, 400 MB: CLI download 27 ->
33 MB/s, upload 18 -> 22, browser over eight connections 9-14 -> 17-19.

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

4 files changed, 293 insertions, 2 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py
index d197044..07820b2 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py
@@ -15,6 +15,7 @@ from meshbay_common.protocol import MNP
 from meshbay_node import ops
 from meshbay_node import transfers as transfers_mod
 from meshbay_node.transport.webrtc import (
+    proactor_udp,  # noqa: F401  (patches asyncio's Windows loop on import)
     sctp_fix,  # noqa: F401  (patches aiortc on import)
     sctp_stats,  # noqa: F401  (patches aiortc on import)
     udp_buffer,  # noqa: F401  (patches aioice on import)
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/proactor_udp.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/proactor_udp.py
new file mode 100644
index 0000000..38cb97c
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/proactor_udp.py
@@ -0,0 +1,139 @@
+"""Datagrams moved straight through the socket on Windows' event loop, applied
+on import.
+
+The node runs on the ProactorEventLoop, the one that can start ffmpeg there
+(`platform.configure_event_loop`), and so do its connection processes. Its
+datagram transport keeps one overlapped operation in each direction and starts
+the next only from the completion callback of the one before, which runs on a
+later turn of the loop. So a socket sends **one datagram per turn of the loop,
+and receives one**, however many are waiting, and a turn lasts as long as
+everything else the loop has to do: the other connections, reading the file,
+the hub. The busier the node, the slower every connection, with cores to spare.
+
+Measured on loopback, 1200-byte datagrams, with the loop otherwise busy 0.5 ms
+a turn: 1,928 sent a second (2.3 MB/s), and 1,954 received, the kernel dropping
+87 % of what arrived; straight through the socket, 88,000 sent, and every
+datagram received at the 20,000 a second it was sent at. In the node holding the
+connections itself, a browser's download over eight connections ran at 9.1 MB/s
+with the node at 45 % of a core and an SCTP round trip of 745 ms on loopback:
+that is the queue in front of each socket. In a connection process serving the
+command-line client, 42 % of the time was `IocpProactor.sendto`. Linux never
+paid for the sending half: the selector loop's transport tries the socket first
+and queues only what it refuses.
+
+Sending: a datagram goes out with a non-blocking `sendto` when nothing is queued
+and no overlapped send is pending, so order is kept; anything else, the socket
+refusing included, goes through the proactor's own queue as before.
+
+Receiving: once an overlapped read completes, its datagram is delivered, then
+whatever else the socket holds is read with non-blocking `recvfrom`, and only
+then is the next overlapped read posted. Never while one is pending, which
+could hand a later datagram to the completion port ahead of the ones read here.
+The overlapped path turns the ICMP "port unreachable" a send to a silent
+address brings back into an empty read; read directly it is a
+ConnectionResetError, and it is skipped the same way.
+
+No effect off Windows, where the class is never used.
+"""
+
+import sys
+from asyncio import exceptions, proactor_events
+
+_Transport = proactor_events._ProactorDatagramTransport
+_proactor_sendto = _Transport.sendto
+
+# Datagrams read straight from the socket after one completion, at most: a
+# socket under a flood still lets the loop turn.
+DRAIN_MAX = 256
+
+
+def sendto(self, data, addr=None) -> None:
+    if (self._write_fut is None and not self._buffer and not self._conn_lost
+            and isinstance(data, (bytes, bytearray, memoryview)) and data
+            and (self._address is None or addr in (None, self._address))):
+        try:
+            if self._address is not None:
+                self._sock.send(data)
+            else:
+                self._sock.sendto(data, addr)
+            return
+        except (BlockingIOError, InterruptedError):
+            pass
+        except OSError as exc:
+            # What the overlapped path does with a failed send.
+            self._protocol.error_received(exc)
+            return
+    _proactor_sendto(self, data, addr)
+
+
+def _deliver(self, data, addr) -> None:
+    """To the protocol, and whatever it raises reported as a callback's would
+    be: in CPython's order the next read was posted first, and an exception
+    here must not end the reading either."""
+    try:
+        self._protocol.datagram_received(data, addr)
+    except (SystemExit, KeyboardInterrupt):
+        raise
+    except BaseException as exc:
+        self._loop.call_exception_handler({
+            "message": "Exception in datagram_received", "exception": exc,
+            "transport": self, "protocol": self._protocol})
+
+
+def _drain(self) -> None:
+    for _ in range(DRAIN_MAX):
+        if self._conn_lost or self._closing:
+            return
+        try:
+            if self._address is not None:
+                data, addr = self._sock.recv(self.max_size), self._address
+            else:
+                data, addr = self._sock.recvfrom(self.max_size)
+        except (BlockingIOError, InterruptedError):
+            return
+        except ConnectionResetError:
+            continue
+        except OSError as exc:
+            # The overlapped read posted next meets it again, and reports it
+            # the proactor's way.
+            self._protocol.error_received(exc)
+            return
+        if data:
+            _deliver(self, data, addr)
+
+
+def _loop_reading(self, fut=None) -> None:
+    """CPython's own (3.12), but a completed read is delivered and the socket
+    drained before the next overlapped read is posted, not after."""
+    try:
+        if self._conn_lost:
+            return
+        assert self._read_fut is fut or (self._read_fut is None and self._closing)
+        self._read_fut = None
+        if fut is not None:
+            res = fut.result()
+            if self._closing:
+                return
+            data, addr = (res, self._address) if self._address is not None else res
+            if data:
+                _deliver(self, data, addr)
+            _drain(self)
+        if self._conn_lost:
+            return
+        if self._address is not None:
+            self._read_fut = self._loop._proactor.recv(self._sock, self.max_size)
+        else:
+            self._read_fut = self._loop._proactor.recvfrom(self._sock, self.max_size)
+    except OSError as exc:
+        self._protocol.error_received(exc)
+    except exceptions.CancelledError:
+        if not self._closing:
+            raise
+    else:
+        if self._read_fut is not None:
+            self._read_fut.add_done_callback(self._loop_reading)
+
+
+if sys.platform == "win32":
+    _Transport.sendto = sendto
+    _Transport._loop_reading = _loop_reading
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py
index 832428a..c933aba 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/worker.py
@@ -224,9 +224,10 @@ async def _main() -> None:
     root.handlers[:] = [_Forward(writer.write)]
     root.setLevel(hello.get("level", logging.INFO))
 
-    # Patches aiortc on import, as the node does for the connections it holds
-    # itself.
+    # Patches aiortc, and asyncio's Windows loop, on import, as the node does
+    # for the connections it holds itself.
     from meshbay_node.transport.webrtc import (  # noqa: F401
+        proactor_udp,
         sctp_fix,
         sctp_stats,
         udp_buffer,
diff --git a/packages/meshbay-node/tests/test_proactor_udp.py b/packages/meshbay-node/tests/test_proactor_udp.py
new file mode 100644
index 0000000..0b22b4e
--- /dev/null
+++ b/packages/meshbay-node/tests/test_proactor_udp.py
@@ -0,0 +1,150 @@
+"""
+On Windows the node's datagrams go straight through the socket, not one
+overlapped operation per datagram and per turn of the loop (`proactor_udp.py`).
+
+Run on a real ProactorEventLoop, which only Windows has.
+"""
+
+import asyncio
+import socket
+import sys
+import time
+
+import pytest
+from meshbay_node.transport.webrtc import proactor_udp
+
+pytestmark = pytest.mark.skipif(sys.platform != "win32", reason="the proactor is Windows'")
+
+N = 2000
+
+
+class _Collect(asyncio.DatagramProtocol):
+    def __init__(self, raise_on: int = -1):
+        self.got: list[int] = []
+        self.raise_on = raise_on
+        self.all = asyncio.get_running_loop().create_future()
+
+    def datagram_received(self, data, addr):
+        self.got.append(int.from_bytes(data[:4], "big"))
+        if len(self.got) == N and not self.all.done():
+            self.all.set_result(None)
+        if len(self.got) == self.raise_on:
+            raise RuntimeError("a protocol's own bug")
+
+
+class _Refusing(socket.socket):
+    """A socket that refuses a send every so often, as a full buffer does, and
+    can answer one read with the reset an ICMP port unreachable leaves. The
+    overlapped path works by handle and never calls these."""
+
+    def __init__(self, every: int = 0, reset_once: bool = False):
+        super().__init__(socket.AF_INET, socket.SOCK_DGRAM)
+        self.every, self.calls, self.reset_once = every, 0, reset_once
+
+    def sendto(self, data, addr):
+        self.calls += 1
+        if self.every and self.calls % self.every == 0:
+            raise BlockingIOError
+        return super().sendto(data, addr)
+
+    def recvfrom(self, n):
+        if self.reset_once:
+            self.reset_once = False
+            raise ConnectionResetError(10054, "port unreachable")
+        return super().recvfrom(n)
+
+
+def _busy_turns(ms: float):
+    """A loop with other work to do: every turn takes `ms` longer."""
+    async def busy():
+        while True:
+            end = time.perf_counter() + ms / 1000
+            while time.perf_counter() < end:
+                pass
+            await asyncio.sleep(0)
+    return asyncio.ensure_future(busy())
+
+
+async def _exchange(*, refuse_every=0, reset_once=False, raise_on=-1, busy_ms=0.0):
+    """N datagrams from one proactor transport to another on loopback: what
+    arrived in what order, and how many overlapped sends and reads it took."""
+    loop = asyncio.get_running_loop()
+    rsock = _Refusing(reset_once=reset_once)
+    rsock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 8 << 20)
+    rsock.bind(("127.0.0.1", 0))
+    rtr, rx = await loop.create_datagram_endpoint(lambda: _Collect(raise_on), sock=rsock)
+    tsock = _Refusing(refuse_every)
+    tsock.bind(("127.0.0.1", 0))
+    tx, _ = await loop.create_datagram_endpoint(asyncio.DatagramProtocol, sock=tsock)
+    counts = {"sendto": 0, "recvfrom": 0}
+    originals = {name: getattr(loop._proactor, name) for name in counts}
+
+    def counting(name):
+        def call(*a, **kw):
+            counts[name] += 1
+            return originals[name](*a, **kw)
+        return call
+    for name in counts:
+        setattr(loop._proactor, name, counting(name))
+    errors = []
+    loop.set_exception_handler(lambda _loop, ctx: errors.append(ctx))
+    busy = _busy_turns(busy_ms) if busy_ms else None
+    try:
+        dest = rsock.getsockname()
+        for i in range(N):
+            tx.sendto(i.to_bytes(4, "big") * 50, dest)
+            if i % 50 == 49:
+                await asyncio.sleep(0)
+        await asyncio.wait_for(rx.all, 20)
+        return rx.got, counts, errors
+    finally:
+        if busy:
+            busy.cancel()
+        for name, fn in originals.items():
+            setattr(loop._proactor, name, fn)
+        tx.close()
+        rtr.close()
+
+
+def _run(**kw):
+    return asyncio.run(_exchange(**kw), loop_factory=asyncio.ProactorEventLoop)
+
+
+def test_the_transport_is_patched():
+    cls = asyncio.proactor_events._ProactorDatagramTransport
+    assert cls.sendto is proactor_udp.sendto
+    assert cls._loop_reading is proactor_udp._loop_reading
+
+
+def test_datagrams_go_straight_out_in_order():
+    got, counts, _ = _run()
+    assert got == list(range(N))
+    assert counts["sendto"] == 0
+
+
+def test_what_the_socket_refuses_is_queued_and_order_is_kept():
+    """A refusal sends that datagram the overlapped way, and the ones after it
+    wait behind it rather than overtake it."""
+    got, counts, _ = _run(refuse_every=97)
+    assert got == list(range(N))
+    assert counts["sendto"] >= N // 97
+
+
+def test_a_busy_loop_reads_everything_waiting_in_order():
+    """Unpatched, this loop read one datagram a turn: about 2,000 a second
+    whatever was waiting, and the kernel dropped the rest."""
+    got, counts, _ = _run(busy_ms=2)
+    assert got == list(range(N))
+    assert counts["recvfrom"] < N // 10
+
+
+def test_a_port_unreachable_on_a_direct_read_is_skipped():
+    got, _, errors = _run(reset_once=True, busy_ms=2)
+    assert got == list(range(N))
+    assert not errors
+
+
+def test_a_protocol_that_raises_does_not_end_the_reading():
+    got, _, errors = _run(raise_on=10)
+    assert got == list(range(N))
+    assert [str(e["exception"]) for e in errors] == ["a protocol's own bug"]