gh-127057: Re-arm proactor datagram read loop after a recoverable OSError

Lib/asyncio/proactor_events.py

@@ -573,6 +573,13 @@ def _loop_reading(self, fut=None):
                                                                self.max_size)
         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/test/test_asyncio/test_events.py

@@ -1583,6 +1583,71 @@ def create_socket():
         transport_1.close()
         transport_2.close()
 
+    def test_datagram_recvfrom_connection_reset_recovers(self):
+        # gh-127057: on Windows, a UDP socket that previously sent a
+        # datagram to an address that wasn't listening can raise
+        # ConnectionResetError (WSAECONNRESET) on a later receive
+        # attempt: synchronously from WSARecvFrom() on ProactorEventLoop
+        # (instead of via the completion result already handled in
+        # _finish_recvfrom() for gh-91227), or from a plain recvfrom()
+        # on SelectorEventLoop. Either way the transport must keep
+        # working afterwards instead of the read loop dying silently.
+        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 a real ICMP port-unreachable now, before the socket is
+        # wrapped in a transport and before any read is armed for it --
+        # on Windows this is what makes a Proactor's first WSARecvFrom()
+        # call raise synchronously.
+        sock.sendto(b'x', closed_addr)
+
+        transport, protocol = loop.run_until_complete(
+            loop.create_datagram_endpoint(Protocol, sock=sock))
+
+        # The transport must still be able to receive afterwards -- this
+        # is the actual regression check, and must hold regardless of
+        # whether this platform surfaced an error for the bad send above.
+        transport.sendto(b'ping', addr)
+        loop.run_until_complete(
+            asyncio.wait_for(protocol.datagram_received_event, 10))
+        self.assertEqual(protocol.received, [b'ping'])
+
+        if sys.platform == 'win32':
+            # Windows reliably reports the bad send via a later recvfrom();
+            # other platforms generally don't deliver ICMP errors to a
+            # plain recv() 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,6 @@
+Fix :class:`asyncio.ProactorEventLoop` UDP transports so a
+:exc:`ConnectionResetError` raised synchronously from ``WSARecvFrom``
+(reported when the same socket was previously used to send to an
+address that isn't listening) no longer breaks the read loop. This
+complements the existing handling of the equivalent asynchronous
+``ERROR_PORT_UNREACHABLE`` completion result.