asyncio proactor udp transport stops responding after send to port that isn't listening due to ConnectionResetError from WSARecvFrom

[bugale]

Bug report

Bug description:

I face a similar issue to #91227 , although instead of the error being thrown from the GetOverlappedResult, I get a ConnectionResetError thrown directly from the WSARecvFrom call.

I am not quite sure what causes Windows to decide to throw the error from WSARecvFrom rather than from the respective GetOverlappedResult, but it appears to have something to do with timings and/or number of times an ICMP unreachable has been sent.

I haven't put a log of effort to investigate, but I am able to reproduce it 100% of the times on my machine using this code:

import asyncio
import time
import socket


class Protocol(asyncio.DatagramProtocol):
    def connection_made(self, transport):
        self.transport = transport

    def datagram_received(self, data, addr):
        print(f'Received {data}')
        time.sleep(1)
        self.transport.sendto(b'test', ('127.0.0.1', 51346))


def send_data():
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    for i in range(100):
        print(f'Sending {str(i)}')
        s.sendto(str(i).encode(), ('127.0.0.1', 1337))
        time.sleep(1)
    s.close()


async def main():
    loop = asyncio.get_running_loop()
    transport, _ = await loop.create_datagram_endpoint(Protocol, local_addr=('127.0.0.1', 1337))
    await asyncio.get_running_loop().run_in_executor(None, send_data)
    transport.close()


asyncio.run(main())

I usually get something like this output:

Sending 0
Received b'0'
Sending 1
Received b'1'
Sending 2
Received b'2'
Sending 3
Sending 4
Sending 5
Sending 6
Sending 7

(i.e. after ~3 successful receives, the server stops receiving data).

While debugging it, I was able to see that the call to ov.WSARecvFrom(conn.fileno(), nbytes, flags) in windows_event.py throws a ConnectionResetError at some point, which is caught in _loop_writing in proactor_events.py, causing it to basically stop its receiving loop.

An easy fix would be to retry the call to ov.WSARecvFrom(conn.fileno(), nbytes, flags) on ConnectionResetError. It seems to work after 2 retries, though I couldn't find a documentation that guarantees this behavior.
Something like this works for me:

    def recvfrom(self, conn, nbytes, flags=0):
        self._register_with_iocp(conn)
-        ov = _overlapped.Overlapped(NULL)
        try:
+            for _ in range(10):
+                try:
+                    ov = _overlapped.Overlapped(NULL)
-            ov.WSARecvFrom(conn.fileno(), nbytes, flags)
+                    ov.WSARecvFrom(conn.fileno(), nbytes, flags)
+                    break
+                except ConnectionResetError:
+                    pass
        except BrokenPipeError:
            return self._result((b'', None))

        return self._register(ov, conn, partial(self._finish_recvfrom,
                                                empty_result=b''))

I could open a PR with such a change, though I am not sure how it should be tested, as I wasn't able to reproduce the issue without ugly and non-deterministic sleeps.

CPython versions tested on:

3.13

Operating systems tested on:

Windows