asyncio: a plain-function client_connected_cb that raises leaves the server transport open

[danielKim614]

Bug report

Bug description:

Bug report

Bug description

asyncio.start_server() accepts both coroutine and plain-function callbacks. When a coroutine callback raises, gh-110894 (gh-111601) made StreamReaderProtocol.connection_made() report the error and close the transport. The plain-function branch was not covered. If the callback raises synchronously, the exception propagates out of connection_made() (Lib/asyncio/streams.py).

1. The transport is never closed. The client sees an established connection that never answers, and the server-side connection leaks until process exit (ResourceWarning: unclosed transport and unclosed socket).
2. Server.wait_closed() hangs forever, because the leaked connection never finishes. A failing callback breaks graceful shutdown.
3. The error is reported with the generic 'Exception in callback StreamReaderProtocol.connection_made()' message instead of the 'Unhandled exception in client_connected_cb' message used by the coroutine branch.

Related: gh-155934 covers the same failure class one stage earlier in the server pipeline, an error while creating the transport for an accepted connection in BaseSelectorEventLoop._accept_connection2().

Reproducer

import asyncio


async def main():
    loop = asyncio.get_running_loop()
    messages = []
    loop.set_exception_handler(lambda l, ctx: messages.append(ctx.get("message")))

    def cb(reader, writer):          # plain function, not async
        raise RuntimeError("boom in client_connected_cb")

    server = await asyncio.start_server(cb, "127.0.0.1", 0)
    port = server.sockets[0].getsockname()[1]

    rd, wr = await asyncio.open_connection("127.0.0.1", port)
    wr.write(b"hello")
    await wr.drain()
    try:
        data = await asyncio.wait_for(rd.read(), timeout=1.0)
        print("server closed the connection, read ->", data)
    except ConnectionResetError:
        print("server closed the connection (RST, unread data was pending)")
    except TimeoutError:
        print("TIMEOUT: server never closed the connection (transport left open)")
    wr.close()
    server.close()
    try:
        await asyncio.wait_for(server.wait_closed(), timeout=1.0)
        print("server.wait_closed() returned")
    except TimeoutError:
        print("server.wait_closed() HANGS (leaked connection keeps the server alive)")
    print("exception handler messages:", messages)


asyncio.run(main())

Output on current main:

TIMEOUT: server never closed the connection (transport left open)
server.wait_closed() HANGS (leaked connection keeps the server alive)
exception handler messages: ['Exception in callback StreamReaderProtocol.connection_made()']
<sys>:0: ResourceWarning: unclosed <socket.socket fd=8, ...>
ResourceWarning: unclosed transport <_SelectorSocketTransport fd=8>

Suggested fix

Wrap the synchronous call in connection_made() and mirror the coroutine branch on failure. Report through the loop exception handler with the same message and close the transport. With that change the reproducer prints:

server closed the connection (RST, unread data was pending)
server.wait_closed() returned
exception handler messages: ['Unhandled exception in client_connected_cb']

I'd like to work on this as part of the PyCon KR sprint. @corona10 @hugovk

Found while verifying entries from devdanzin's stdlib audit catalog n.26

CPython versions tested on:

CPython main branch

Operating systems tested on:

macOS

Linked PRs

• gh-155941: Close the transport when a plain client_connected_cb raises #155942