The fix in gh-91227 is incorrect, the SelectorEventLoop has the correct behaviour calling error_received and continuing to run the Transport

[graingert]

The fix in gh-91227 is incorrect, the SelectorEventLoop has the correct behaviour calling error_received and continuing to run the Transport

Originally posted by @graingert in #32011 (comment)

Linked PRs

• gh-156784: unignore error port unreachable in proactor  #156788

[graingert]

"""Demonstrates that sending a UDP datagram to a port that isn't
listening now surfaces a recoverable error consistently on both
SelectorEventLoop and ProactorEventLoop (see gh-91227, gh-127057):

  - DatagramProtocol.error_received() is called, and the transport
    keeps working afterwards instead of the read loop dying silently.

Windows only. Run with: python snippet.py
"""
import asyncio
import socket


class EchoProtocol(asyncio.DatagramProtocol):
    def connection_made(self, transport):
        self.transport = transport
        self.received = []
        self.errors = []

    def datagram_received(self, data, addr):
        self.received.append(data)

    def error_received(self, exc):
        self.errors.append(exc)


def closed_socket_address():
    """Bind and immediately close a UDP socket to get an address that
    is guaranteed not to be listening."""
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind(('127.0.0.1', 0))
    addr = sock.getsockname()
    sock.close()
    return addr


async def demo_transport(loop):
    print('--- DatagramProtocol.error_received() ---')
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setblocking(False)
    sock.bind(('127.0.0.1', 0))
    own_addr = sock.getsockname()

    transport, protocol = await loop.create_datagram_endpoint(
        EchoProtocol, sock=sock)

    transport.sendto(b'ping', closed_socket_address())

    for _ in range(100):
        await asyncio.sleep(0.05)
        if protocol.errors:
            break

    if protocol.errors:
        print(f'error_received() fired: {protocol.errors[0]!r}')
    else:
        print('error_received() did NOT fire (timed out waiting)')

    # The transport must still work afterwards.
    transport.sendto(b'still alive', own_addr)
    for _ in range(100):
        await asyncio.sleep(0.05)
        if protocol.received:
            break
    print(f'transport still usable afterwards: {protocol.received!r}')

    transport.close()


async def main(loop):
    await demo_transport(loop)


def run(name, loop):
    print(f'\n=== {name} ===')
    try:
        loop.run_until_complete(main(loop))
    finally:
        loop.close()


if __name__ == '__main__':
    run('SelectorEventLoop', asyncio.SelectorEventLoop())
    run('ProactorEventLoop', asyncio.ProactorEventLoop())

on 3.14 it reports:

PS C:\Users\tagra\projects\cpython> py -3.14 .\snippet.py

=== SelectorEventLoop ===
--- DatagramProtocol.error_received() ---
error_received() fired: ConnectionResetError(10054, 'An existing connection was forcibly closed by the remote host', None, 10054, None)
transport still usable afterwards: [b'still alive']

=== ProactorEventLoop ===
--- DatagramProtocol.error_received() ---
error_received() did NOT fire (timed out waiting)
transport still usable afterwards: [b'still alive']

on my patched branch I get:

PS C:\Users\tagra\projects\cpython> ./PCbuild/amd64/python_d.exe snippet.py

=== SelectorEventLoop ===
--- DatagramProtocol.error_received() ---
error_received() fired: ConnectionResetError(10054, 'An existing connection was forcibly closed by the remote host', None, 10054, None)
transport still usable afterwards: [b'still alive']

=== ProactorEventLoop ===
--- DatagramProtocol.error_received() ---
error_received() fired: ConnectionResetError(22, 'No service is operating at the destination network endpoint on the remote system', None, 1234, None)
transport still usable afterwards: [b'still alive']