gh-145541: Fix InvalidStateError in BaseSubprocessTransport._call_connection_lost()

Change `if not waiter.cancelled()` to `if not waiter.done()` in both
_try_finish() and _call_connection_lost() so that waiters whose result
was already set by _try_finish() are not set again by
_call_connection_lost(), which would raise InvalidStateError.

When _connect_pipes is cancelled (e.g. by SIGINT during subprocess
creation), _pipes_connected stays False. If the process then exits,
_try_finish() sets the result on exit waiters because _pipes_connected
is False, and then schedules _call_connection_lost() because all pipes
are disconnected. _call_connection_lost() must skip waiters that are
already done.

Author: daandemeyer

Lib/asyncio/base_subprocess.py

@@ -265,7 +265,7 @@ def _try_finish(self):
             # to avoid hanging forever in self._wait as otherwise _exit_waiters
             # would never be woken up, we wake them up here.
             for waiter in self._exit_waiters:
-                if not waiter.cancelled():
+                if not waiter.done():
                     waiter.set_result(self._returncode)
         if all(p is not None and p.disconnected
                for p in self._pipes.values()):
@@ -278,7 +278,7 @@ def _call_connection_lost(self, exc):
         finally:
             # wake up futures waiting for wait()
             for waiter in self._exit_waiters:
-                if not waiter.cancelled():
+                if not waiter.done():
                     waiter.set_result(self._returncode)
             self._exit_waiters = None
             self._loop = None

Lib/test/test_asyncio/test_subprocess.py

@@ -111,6 +111,33 @@ def test_subprocess_repr(self):
         )
         transport.close()
 
+    def test_proc_exited_no_invalid_state_error_on_exit_waiters(self):
+        # gh-145541: when _connect_pipes hasn't completed (so
+        # _pipes_connected is False) and the process exits, _try_finish()
+        # sets the result on exit waiters. Then _call_connection_lost() must
+        # not call set_result() again on the same waiters.
+        waiter = self.loop.create_future()
+        transport, protocol = self.create_transport(waiter)
+
+        # Simulate a waiter registered via _wait() before the process exits.
+        exit_waiter = self.loop.create_future()
+        transport._exit_waiters.append(exit_waiter)
+
+        # _connect_pipes hasn't completed, so _pipes_connected is False.
+        self.assertFalse(transport._pipes_connected)
+
+        # Simulate process exit. _try_finish() will set the result on
+        # exit_waiter because _pipes_connected is False, and then schedule
+        # _call_connection_lost() because _pipes is empty (vacuously all
+        # disconnected). _call_connection_lost() must skip exit_waiter
+        # because it's already done.
+        transport._process_exited(6)
+        self.loop.run_until_complete(waiter)
+
+        self.assertEqual(exit_waiter.result(), 6)
+
+        transport.close()
+
 
 class SubprocessMixin:
 

Misc/NEWS.d/next/Library/2026-03-05-19-01-28.gh-issue-145551.gItPRl.rst

@@ -0,0 +1 @@
+Fix InvalidStateError when cancelling process created by :func:`asyncio.create_subprocess_exec` or :func:`asyncio.create_subprocess_shell`. Patch by Daan De Meyer.