Close the resource only when protocol_factory() fails

On the proactor, _make_*_transport() takes ownership of the socket/pipe and
starts I/O before it can raise, so closing it in connect_*()/create_*()
double-manages it (broke test_windows_events). Close only when
protocol_factory() raises, before any transport exists.

Author: tonghuaroot

Lib/asyncio/base_events.py

@@ -1499,13 +1499,13 @@ async def create_datagram_endpoint(self, protocol_factory,
 
         try:
             protocol = protocol_factory()
-            waiter = self.create_future()
-            transport = self._make_datagram_transport(
-                sock, protocol, r_addr, waiter)
         except:
-            # gh-156400: close the socket if the transport is never created.
+            # 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)
         if self._debug:
             if local_addr:
                 logger.info("Datagram endpoint local_addr=%r remote_addr=%r "
@@ -1721,12 +1721,12 @@ async def connect_accepted_socket(
     async def connect_read_pipe(self, protocol_factory, pipe):
         try:
             protocol = protocol_factory()
-            waiter = self.create_future()
-            transport = self._make_read_pipe_transport(pipe, protocol, waiter)
         except:
-            # gh-156400: close the pipe if the transport is never created.
+            # 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)
 
         try:
             await waiter
@@ -1742,12 +1742,12 @@ async def connect_read_pipe(self, protocol_factory, pipe):
     async def connect_write_pipe(self, protocol_factory, pipe):
         try:
             protocol = protocol_factory()
-            waiter = self.create_future()
-            transport = self._make_write_pipe_transport(pipe, protocol, waiter)
         except:
-            # gh-156400: close the pipe if the transport is never created.
+            # 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)
 
         try:
             await waiter

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

@@ -1,5 +1,5 @@
-Fix socket and pipe leaks in :mod:`asyncio` when ``protocol_factory()`` or
-transport creation fails in :meth:`loop.create_datagram_endpoint
+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