gh-156784: unignore error port unreachable in proactor

Lib/asyncio/proactor_events.py

@@ -534,6 +534,18 @@ def _loop_writing(self, fut=None):
                                                               addr=addr)
         except OSError as exc:
             self._protocol.error_received(exc)
+            if self._buffer:
+                # Re-arm the write loop so buffered data isn't stranded and
+                # a paused protocol is eventually resumed (gh-156698).
+                def resume_writing():
+                    # a sendto() may have armed a write in the meantime;
+                    # its own callback will drain the rest of the buffer.
+                    if self._write_fut is None:
+                        self._loop_writing()
+
+                self._loop.call_soon(resume_writing)
+            else:
+                self._maybe_resume_protocol()
         except Exception as exc:
             self._fatal_error(exc, 'Fatal write error on datagram transport')
         else:
@@ -582,6 +594,13 @@ def _loop_reading(self, fut=None):
                 self._loop.call_soon(self._loop_reading)
         except OSError as exc:
             self._protocol.error_received(exc)
+            if not self._closing and not self._conn_lost:
+                # Some errors are transient and recoverable, e.g. a
+                # ConnectionResetError raised synchronously by WSARecvFrom()
+                # from a stale ICMP port-unreachable notification on a UDP
+                # socket. Re-arm the read loop instead of leaving it dead
+                # (gh-127057).
+                self._loop.call_soon(self._loop_reading)
         except exceptions.CancelledError:
             if not self._closing:
                 raise

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, flags=0):
         self._register_with_iocp(conn)
@@ -523,8 +517,7 @@ def recvfrom_into(self, conn, buf, flags=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,9 +1589,147 @@ 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()
 
+    def _test_datagram_write_error_resumes_paused_protocol(self, first, second):
+        # See https://github.com/python/cpython/issues/156698: a
+        # datagram write error must not strand data left in the write
+        # buffer, nor leave a paused protocol paused forever.
+        loop = self.loop
+
+        class Protocol(asyncio.DatagramProtocol):
+            def connection_made(self, transport):
+                self.transport = transport
+                self.paused = False
+                self.resumed = False
+                self.errors = []
+                self.error_received_event = loop.create_future()
+
+            def pause_writing(self):
+                self.paused = True
+
+            def resume_writing(self):
+                self.resumed = True
+
+            def error_received(self, exc):
+                self.errors.append(exc)
+                if not self.error_received_event.done():
+                    self.error_received_event.set_result(None)
+
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        sock.setblocking(False)
+        sock.bind(('127.0.0.1', 0))
+        transport, protocol = loop.run_until_complete(
+            loop.create_datagram_endpoint(Protocol, sock=sock))
+        addr = sock.getsockname()
+
+        # A high water mark of 0 makes pausing deterministic whenever
+        # anything is left in the write buffer.
+        transport.set_write_buffer_limits(0)
+
+        # The first sendto() may arm an in-flight write, so the second
+        # one can end up queued behind it; queuing is what trips
+        # pause_writing() at a high water mark of 0.
+        transport.sendto(first, addr)
+        transport.sendto(second, addr)
+
+        loop.run_until_complete(
+            asyncio.wait_for(protocol.error_received_event, 10))
+        self.assertTrue(protocol.errors)
+        self.assertIsInstance(protocol.errors[0], OSError)
+
+        # The write buffer must not be left stranded.
+        test_utils.run_until(
+            loop, lambda: transport.get_write_buffer_size() == 0)
+
+        # A protocol that got paused must eventually be resumed too --
+        # without requiring an unsolicited extra sendto() to un-stick it.
+        if protocol.paused:
+            test_utils.run_until(loop, lambda: protocol.resumed)
+
+        transport.close()
+        test_utils.run_briefly(loop)
+
+    def test_datagram_write_error_resumes_paused_protocol_in_flight(self):
+        # oversized datagram fails while in flight; a normal datagram
+        # queued right behind it must not be stranded.
+        oversized = b'\x00' * 70000
+        self._test_datagram_write_error_resumes_paused_protocol(
+            oversized, b'queued')
+
+    def test_datagram_write_error_resumes_paused_protocol_from_callback(self):
+        # oversized datagram fails once it reaches the front of the
+        # buffer; the protocol must not stay paused forever.
+        oversized = b'\x00' * 70000
+        self._test_datagram_write_error_resumes_paused_protocol(
+            b'ok', oversized)
+
+    def test_datagram_write_error_reentrant_sendto(self):
+        # See https://github.com/python/cpython/issues/156698: an
+        # error_received() callback that sends more data synchronously
+        # can itself arm a new write. The write-loop restart scheduled
+        # for the failed write must notice that and not try to start a
+        # second, conflicting one.
+        loop = self.loop
+        unhandled = []
+        loop.set_exception_handler(lambda loop, context: unhandled.append(context))
+
+        class Protocol(asyncio.DatagramProtocol):
+            def connection_made(self, transport):
+                self.transport = transport
+                self.sent_extra = False
+                self.errors = []
+                self.done = loop.create_future()
+
+            def datagram_received(self, data, addr):
+                if not self.done.done():
+                    self.done.set_result(None)
+
+            def error_received(self, exc):
+                self.errors.append(exc)
+                if not self.sent_extra:
+                    # Reentrantly kicks off another write while the
+                    # failing one is still unwinding on the stack.
+                    self.sent_extra = True
+                    self.transport.sendto(b'extra', self.addr)
+
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        sock.setblocking(False)
+        sock.bind(('127.0.0.1', 0))
+        transport, protocol = loop.run_until_complete(
+            loop.create_datagram_endpoint(Protocol, sock=sock))
+        protocol.addr = addr = sock.getsockname()
+
+        oversized = b'\x00' * 70000
+        transport.sendto(oversized, addr)
+        transport.sendto(b'queued', addr)
+
+        # The 'extra' datagram sent from error_received() is delivered
+        # back to the same socket; waiting for it proves the write loop
+        # kept running instead of wedging or crashing.
+        loop.run_until_complete(asyncio.wait_for(protocol.done, 10))
+
+        test_utils.run_until(
+            loop, lambda: transport.get_write_buffer_size() == 0)
+
+        transport.close()
+        test_utils.run_briefly(loop)
+
+        self.assertTrue(protocol.errors)
+        self.assertFalse(
+            unhandled,
+            f'unhandled exception in the write loop: {unhandled}')
+
     def test_datagram_recvfrom_connection_reset_recovers(self):
         # gh-127057: a UDP socket that sent a datagram to an address that
         # wasn't listening can raise ConnectionResetError on a later

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-08-31-00-00-00.gh-issue-156698.gyDxUe.rst

@@ -0,0 +1,4 @@
+Fix :class:`asyncio.ProactorEventLoop` UDP transports so that a write
+error no longer strands a paused protocol: the write loop is now
+rescheduled when data remains buffered, and the protocol is resumed
+when the buffer has drained.

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

@@ -0,0 +1,11 @@
+Fix :class:`asyncio.ProactorEventLoop` UDP transports and sockets so
+that ``ERROR_PORT_UNREACHABLE``/``WSAECONNRESET`` (reported when the
+same socket was previously used to send to an address that isn't
+listening) is surfaced as a :exc:`ConnectionResetError` instead of
+being silently swallowed. Datagram transports now deliver it to the
+protocol via :meth:`~asyncio.DatagramProtocol.error_received` and
+re-arm the read loop, and :meth:`~asyncio.loop.sock_recvfrom` /
+:meth:`~asyncio.loop.sock_recvfrom_into` now raise it, matching the
+behaviour :class:`asyncio.SelectorEventLoop` already had. This
+corrects the fix applied in gh-91227, which papered over the issue by
+discarding the error.