gh-156784: do not ignore error port unreachable in proactor event loop implementation (#156788)

Author: graingert

Lib/asyncio/windows_events.py

@@ -8,7 +8,6 @@
 import _overlapped
 import _winapi
 import errno
-from functools import partial
 import math
 import msvcrt
 import socket
@@ -460,24 +459,20 @@ def finish_socket_func(trans, key, ov):
         try:
             return ov.getresult()
         except OSError as exc:
+            # ERROR_PORT_UNREACHABLE is reported by WSARecvFrom when the
+            # same socket was previously used to send to an address that
+            # isn't listening (gh-91227). Surface it as a
+            # ConnectionResetError, like the other recoverable codes here,
+            # so it propagates to the caller the same way a plain
+            # socket.recvfrom() already does on SelectorEventLoop, instead
+            # of being silently swallowed.
             if exc.winerror in (_overlapped.ERROR_NETNAME_DELETED,
-                                _overlapped.ERROR_OPERATION_ABORTED):
+                                _overlapped.ERROR_OPERATION_ABORTED,
+                                _overlapped.ERROR_PORT_UNREACHABLE):
                 raise ConnectionResetError(*exc.args)
             else:
                 raise
 
-    @classmethod
-    def _finish_recvfrom(cls, trans, key, ov, *, empty_result):
-        try:
-            return cls.finish_socket_func(trans, key, ov)
-        except OSError as exc:
-            # WSARecvFrom will report ERROR_PORT_UNREACHABLE when the same
-            # socket is used to send to an address that is not listening.
-            if exc.winerror == _overlapped.ERROR_PORT_UNREACHABLE:
-                return empty_result, None
-            else:
-                raise
-
     def recv(self, conn, nbytes, flags=0):
         self._register_with_iocp(conn)
         ov = _overlapped.Overlapped(NULL)
@@ -512,8 +507,7 @@ def recvfrom(self, conn, nbytes, flags=0):
         except BrokenPipeError:
             return self._result((b'', None))
 
-        return self._register(ov, conn, partial(self._finish_recvfrom,
-                                                empty_result=b''))
+        return self._register(ov, conn, self.finish_socket_func)
 
     def recvfrom_into(self, conn, buf, nbytes=0):
         self._register_with_iocp(conn)
@@ -523,8 +517,7 @@ def recvfrom_into(self, conn, buf, nbytes=0):
         except BrokenPipeError:
             return self._result((0, None))
 
-        return self._register(ov, conn, partial(self._finish_recvfrom,
-                                                empty_result=0))
+        return self._register(ov, conn, self.finish_socket_func)
 
     def sendto(self, conn, buf, flags=0, addr=None):
         self._register_with_iocp(conn)

Lib/test/test_asyncio/test_events.py

@@ -1523,9 +1523,18 @@ class Protocol(asyncio.DatagramProtocol):
 
             _received_datagram = None
 
+            def connection_made(self, transport):
+                self.errors = []
+                self.error_received_event = loop.create_future()
+
             def datagram_received(self, data, addr):
                 self._received_datagram.set_result(data)
 
+            def error_received(self, exc):
+                self.errors.append(exc)
+                if not self.error_received_event.done():
+                    self.error_received_event.set_result(None)
+
             async def wait_for_datagram_received(self):
                 self._received_datagram = loop.create_future()
                 result = await asyncio.wait_for(self._received_datagram, 10)
@@ -1580,6 +1589,15 @@ def create_socket():
             protocol_1.wait_for_datagram_received()
         ), b'd')
 
+        if sys.platform == 'win32':
+            # The bad send to addr_3 should be surfaced to the protocol
+            # via error_received() instead of being silently dropped,
+            # while transport_1 keeps working as shown above.
+            loop.run_until_complete(
+                asyncio.wait_for(protocol_1.error_received_event, 10))
+            self.assertTrue(protocol_1.errors)
+            self.assertIsInstance(protocol_1.errors[0], ConnectionResetError)
+
         transport_1.close()
         transport_2.close()
 

Lib/test/test_asyncio/test_sock_lowlevel.py

@@ -613,19 +613,10 @@ def test_create_connection_sock(self):
 
 if sys.platform == 'win32':
 
-    class SelectEventLoopTests(BaseSockTestsMixin,
-                               test_utils.TestCase):
-
-        def create_event_loop(self):
-            return asyncio.SelectorEventLoop()
-
-
-    class ProactorEventLoopTests(BaseSockTestsMixin,
-                                 test_utils.TestCase):
-
-        def create_event_loop(self):
-            return asyncio.ProactorEventLoop()
-
+    class _DatagramSendToNonListeningAddressMixin:
+        # Shared by SelectEventLoopTests and ProactorEventLoopTests so that
+        # sock_recvfrom()/sock_recvfrom_into() behave identically on both
+        # event loop implementations.
 
         async def _basetest_datagram_send_to_non_listening_address(self,
                                                                    recvfrom):
@@ -634,8 +625,9 @@ async def _basetest_datagram_send_to_non_listening_address(self,
             #   https://github.com/python/cpython/issues/88906
             #   https://bugs.python.org/issue47071
             #   https://bugs.python.org/issue44743
-            # The Proactor event loop would fail to receive datagram messages
-            # after sending a message to an address that wasn't listening.
+            # Sending a datagram to an address that isn't listening can
+            # surface as a ConnectionResetError on a later receive; the
+            # socket must still be usable afterwards.
 
             def create_socket():
                 sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
@@ -670,7 +662,8 @@ def create_socket():
 
             # this should send to an address that isn't listening
             await self.loop.sock_sendto(socket_1, b'c', addr_3)
-            self.assertEqual(await socket_1_recv_task, b'')
+            with self.assertRaises(ConnectionResetError):
+                await socket_1_recv_task
             socket_1_recv_task = self.loop.create_task(recvfrom(socket_1))
             await asyncio.sleep(0)
 
@@ -706,6 +699,22 @@ async def recvfrom_into(socket):
                 self._basetest_datagram_send_to_non_listening_address(
                     recvfrom_into))
 
+
+    class SelectEventLoopTests(_DatagramSendToNonListeningAddressMixin,
+                               BaseSockTestsMixin,
+                               test_utils.TestCase):
+
+        def create_event_loop(self):
+            return asyncio.SelectorEventLoop()
+
+
+    class ProactorEventLoopTests(_DatagramSendToNonListeningAddressMixin,
+                                 BaseSockTestsMixin,
+                                 test_utils.TestCase):
+
+        def create_event_loop(self):
+            return asyncio.ProactorEventLoop()
+
 else:
     import selectors
 

Misc/NEWS.d/next/Library/2026-09-01-00-00-00.gh-issue-156784.8gvtt4.rst

@@ -0,0 +1,3 @@
+:class:`asyncio.ProactorEventLoop` no longer silently ignores
+:exc:`ConnectionResetError` on UDP sockets, matching
+:class:`asyncio.SelectorEventLoop`.