gh-156400: Close the socket or pipe when transport creation fails in asyncio datagram/pipe endpoints

Lib/asyncio/base_events.py

@@ -1497,7 +1497,12 @@ async def create_datagram_endpoint(self, protocol_factory,
             else:
                 raise exceptions[0]
 
-        protocol = protocol_factory()
+        try:
+            protocol = protocol_factory()
+        except:
+            # gh-156400: no transport owns the socket yet, so close it.
+            sock.close()
+            raise
         waiter = self.create_future()
         transport = self._make_datagram_transport(
             sock, protocol, r_addr, waiter)
@@ -1714,7 +1719,12 @@ async def connect_accepted_socket(
         return transport, protocol
 
     async def connect_read_pipe(self, protocol_factory, pipe):
-        protocol = protocol_factory()
+        try:
+            protocol = protocol_factory()
+        except:
+            # gh-156400: no transport owns the pipe yet, so close it.
+            pipe.close()
+            raise
         waiter = self.create_future()
         transport = self._make_read_pipe_transport(pipe, protocol, waiter)
 
@@ -1730,7 +1740,12 @@ async def connect_read_pipe(self, protocol_factory, pipe):
         return transport, protocol
 
     async def connect_write_pipe(self, protocol_factory, pipe):
-        protocol = protocol_factory()
+        try:
+            protocol = protocol_factory()
+        except:
+            # gh-156400: no transport owns the pipe yet, so close it.
+            pipe.close()
+            raise
         waiter = self.create_future()
         transport = self._make_write_pipe_transport(pipe, protocol, waiter)
 

Lib/test/test_asyncio/test_base_events.py

@@ -2041,6 +2041,43 @@ def test_create_datagram_endpoint_sock(self):
         self.loop.run_until_complete(protocol.done)
         self.assertEqual('CLOSED', protocol.state)
 
+    def test_create_datagram_endpoint_transport_error_closes_sock(self):
+        # gh-156400: the socket is closed if the transport is never created.
+        sock = mock.Mock()
+        sock.type = socket.SOCK_DGRAM
+
+        def factory():
+            raise ZeroDivisionError
+
+        coro = self.loop.create_datagram_endpoint(factory, sock=sock)
+        with self.assertRaises(ZeroDivisionError):
+            self.loop.run_until_complete(coro)
+        self.assertTrue(sock.close.called)
+
+    def test_connect_read_pipe_transport_error_closes_pipe(self):
+        # gh-156400: the pipe is closed if the transport is never created.
+        pipe = mock.Mock()
+
+        def factory():
+            raise ZeroDivisionError
+
+        coro = self.loop.connect_read_pipe(factory, pipe)
+        with self.assertRaises(ZeroDivisionError):
+            self.loop.run_until_complete(coro)
+        self.assertTrue(pipe.close.called)
+
+    def test_connect_write_pipe_transport_error_closes_pipe(self):
+        # gh-156400: the pipe is closed if the transport is never created.
+        pipe = mock.Mock()
+
+        def factory():
+            raise ZeroDivisionError
+
+        coro = self.loop.connect_write_pipe(factory, pipe)
+        with self.assertRaises(ZeroDivisionError):
+            self.loop.run_until_complete(coro)
+        self.assertTrue(pipe.close.called)
+
     @unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
     def test_create_datagram_endpoint_sock_unix(self):
         fut = self.loop.create_datagram_endpoint(

Misc/NEWS.d/next/Library/2026-08-26-13-58-01.gh-issue-156400.dGrmP1.rst

@@ -0,0 +1,6 @@
+Fix socket and pipe leaks in :mod:`asyncio` when ``protocol_factory()`` raises
+in :meth:`loop.create_datagram_endpoint
+<asyncio.loop.create_datagram_endpoint>`, :meth:`loop.connect_read_pipe
+<asyncio.loop.connect_read_pipe>`, and :meth:`loop.connect_write_pipe
+<asyncio.loop.connect_write_pipe>`. The socket or pipe is now closed instead
+of leaking until garbage collection.