gh-155934: Fix socket leak and silent error for asyncio accepted connections

Author: danielKim614

Lib/asyncio/selector_events.py

@@ -253,7 +253,9 @@ async def _accept_connection2(
         except (SystemExit, KeyboardInterrupt):
             raise
         except BaseException as exc:
-            if self._debug:
+            if transport is None:
+                conn.close()
+            if transport is None or self._debug:
                 context = {
                     'message':
                         'Error on transport creation for incoming connection',

Lib/test/test_asyncio/test_selector_events.py

@@ -421,6 +421,60 @@ def test_accept_connection_reschedules_once_on_resource_error(self):
         self.assertEqual(self.loop.call_exception_handler.call_count, 1)
         self.assertEqual(self.loop.call_later.call_count, 1)
 
+    def test_accept_connection2_factory_error_closes_conn(self):
+        # gh-155934: if the transport was never created, the accepted
+        # socket is closed and the error is reported even when debug
+        # mode is disabled.
+        self.loop.set_debug(False)
+        conn = mock.Mock()
+
+        def factory():
+            raise RuntimeError("protocol_factory failed")
+
+        self.loop.call_exception_handler = mock.Mock()
+        self.loop.run_until_complete(
+            self.loop._accept_connection2(factory, conn, {}))
+
+        self.assertTrue(conn.close.called)
+        self.loop.call_exception_handler.assert_called_once()
+
+    def test_accept_connection2_transport_error_closes_conn(self):
+        # gh-155934: same when the transport creation itself fails.
+        self.loop.set_debug(False)
+        conn = mock.Mock()
+        self.loop._make_socket_transport = mock.Mock(
+            side_effect=ZeroDivisionError)
+        self.loop.call_exception_handler = mock.Mock()
+
+        self.loop.run_until_complete(
+            self.loop._accept_connection2(mock.Mock(), conn, {}))
+
+        self.assertTrue(conn.close.called)
+        self.loop.call_exception_handler.assert_called_once()
+
+    def test_accept_connection2_waiter_error_stays_debug_only(self):
+        # Once the transport exists it owns the socket: waiter failures
+        # (e.g. SSL handshake errors) close the transport and stay
+        # debug-only, and the accepted socket is not closed directly.
+        self.loop.set_debug(False)
+        conn = mock.Mock()
+        transport = mock.Mock()
+
+        def make_transport(conn, protocol, waiter=None, **kwargs):
+            waiter.set_exception(OSError("handshake failed"))
+            return transport
+
+        self.loop._make_socket_transport = make_transport
+        self.loop.call_exception_handler = mock.Mock()
+
+        self.loop.run_until_complete(
+            self.loop._accept_connection2(mock.Mock(), conn, {}))
+
+        self.assertTrue(transport.close.called)
+        self.assertFalse(conn.close.called)
+        self.assertFalse(self.loop.call_exception_handler.called)
+
+
 class SelectorTransportTests(test_utils.TestCase):
 
     def setUp(self):

Misc/NEWS.d/next/Library/2026-08-17-18-00-00.gh-issue-155934.acCept.rst

@@ -0,0 +1,3 @@
+Fix a socket leak in :mod:`asyncio` when transport creation fails for a
+connection accepted by a server, and report the error via the loop exception
+handler even when debug mode is disabled.