gh-155941: Close the transport when client_connected_cb raises in asyncio (#155942)

Author: danielKim614

Lib/asyncio/streams.py

@@ -239,7 +239,17 @@ def connection_made(self, transport):
         self._over_ssl = transport.get_extra_info('sslcontext') is not None
         if self._client_connected_cb is not None:
             writer = StreamWriter(transport, self, reader, self._loop)
-            res = self._client_connected_cb(reader, writer)
+            try:
+                res = self._client_connected_cb(reader, writer)
+            except Exception as exc:
+                self._loop.call_exception_handler({
+                    'message': 'Unhandled exception in client_connected_cb',
+                    'exception': exc,
+                    'transport': transport,
+                })
+                transport.close()
+                self._strong_reader = None
+                return
             if coroutines.iscoroutine(res):
                 def callback(task):
                     if task.cancelled():

Lib/test/test_asyncio/test_streams.py

@@ -1267,6 +1267,38 @@ async def handle_echo(reader, writer):
         messages = self._basetest_unhandled_exceptions(handle_echo)
         self.assertEqual(messages, [])
 
+    def test_unhandled_exception_sync_callback(self):
+        # An exception raised by a plain-function client_connected_cb is
+        # reported like the coroutine case and the transport is closed.
+        port = socket_helper.find_unused_port()
+
+        messages = []
+        self.loop.set_exception_handler(lambda loop, ctx: messages.append(ctx))
+
+        async def client():
+            rd, wr = await asyncio.open_connection('localhost', port)
+            async with asyncio.timeout(60):
+                data = await rd.read()
+            self.assertEqual(data, b'')  # the server closed the connection
+            wr.close()
+            await wr.wait_closed()
+
+        async def main():
+            def handle_echo(reader, writer):
+                raise Exception('test')
+
+            server = await asyncio.start_server(
+                handle_echo, 'localhost', port)
+            await server.start_serving()
+            await client()
+            server.close()
+            await server.wait_closed()
+
+        self.loop.run_until_complete(main())
+
+        self.assertEqual(messages[0]['message'],
+                    'Unhandled exception in client_connected_cb')
+
     def test_open_connection_happy_eyeball_refcycles(self):
         port = socket_helper.find_unused_port()
         async def main():

Misc/NEWS.d/next/Library/2026-08-17-21-00-00.gh-issue-155941.strmCb.rst

@@ -0,0 +1,4 @@
+Fix :func:`asyncio.start_server` when a plain-function *client_connected_cb*
+raises: the error is now reported like the coroutine case and the transport
+is closed, instead of leaving the connection open forever (which also made
+:meth:`asyncio.Server.wait_closed` hang).