[3.14] gh-127057: reschedule proactor datagram read loop after ConnectionResetError (GH-156726) (#156977)

gh-127057: reschedule proactor datagram read loop after ConnectionResetError  (GH-156726)
(cherry picked from commit 0ea2971)

Co-authored-by: Thomas Grainger <tagrain@gmail.com>

Author: miss-islington

Lib/asyncio/proactor_events.py

@@ -571,6 +571,15 @@ def _loop_reading(self, fut=None):
             else:
                 self._read_fut = self._loop._proactor.recvfrom(self._sock,
                                                                self.max_size)
+        except ConnectionResetError as exc:
+            # WSARecvFrom() reports a stale ICMP port unreachable
+            # notification as a synchronous ConnectionResetError when the
+            # same socket was used to send to an address that is not
+            # listening.  This is transient, so reschedule the read loop
+            # instead of leaving it dead.
+            self._protocol.error_received(exc)
+            if not self._closing:
+                self._loop.call_soon(self._loop_reading)
         except OSError as exc:
             self._protocol.error_received(exc)
         except exceptions.CancelledError:

Lib/test/test_asyncio/test_events.py

@@ -1583,6 +1583,60 @@ def create_socket():
         transport_1.close()
         transport_2.close()
 
+    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
+        # receive.  The transport must keep working afterwards.
+        loop = self.loop
+
+        class Protocol(asyncio.DatagramProtocol):
+            def connection_made(self, transport):
+                self.transport = transport
+                self.errors = []
+                self.received = []
+                self.datagram_received_event = loop.create_future()
+
+            def error_received(self, exc):
+                self.errors.append(exc)
+
+            def datagram_received(self, data, addr):
+                self.received.append(data)
+                if not self.datagram_received_event.done():
+                    self.datagram_received_event.set_result(None)
+
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        sock.setblocking(False)
+        sock.bind(('127.0.0.1', 0))
+        addr = sock.getsockname()
+
+        # Bind and immediately close a second socket to get an address
+        # that is guaranteed not to be listening.
+        closed = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        closed.bind(('127.0.0.1', 0))
+        closed_addr = closed.getsockname()
+        closed.close()
+
+        # Trigger the error before the socket is wrapped in a transport,
+        # so that the first read raises synchronously.
+        sock.sendto(b'x', closed_addr)
+
+        transport, protocol = loop.run_until_complete(
+            loop.create_datagram_endpoint(Protocol, sock=sock))
+
+        transport.sendto(b'ping', addr)
+        loop.run_until_complete(asyncio.wait_for(
+            protocol.datagram_received_event, support.SHORT_TIMEOUT))
+        self.assertEqual(protocol.received, [b'ping'])
+
+        if sys.platform == 'win32':
+            # Other platforms don't report ICMP errors on an
+            # unconnected UDP socket.
+            self.assertEqual(len(protocol.errors), 1)
+            self.assertIsInstance(protocol.errors[0], ConnectionResetError)
+
+        transport.close()
+        test_utils.run_briefly(loop)
+
     def test_internal_fds(self):
         loop = self.create_event_loop()
         if not isinstance(loop, selector_events.BaseSelectorEventLoop):

Misc/NEWS.d/next/Library/2026-08-31-00-00-00.gh-issue-127057.N9yg62.rst

@@ -0,0 +1,3 @@
+Fix :class:`asyncio.ProactorEventLoop` UDP transports so that a
+:exc:`ConnectionResetError` raised by ``WSARecvFrom`` no longer stops the
+read loop.