[3.14] gh-158004: preserve asyncio.as_completed() items after anext cancellation (GH-158005) (#159878)

gh-158004: preserve `asyncio.as_completed()` items after anext cancellation (GH-158005)
(cherry picked from commit 4b33a9d)

Co-authored-by: Pengyu Lee <lipengyu@kylinos.cn>
Co-authored-by: Kumar Aditya <kumaraditya@python.org>

Author: 3 people

Lib/asyncio/tasks.py

@@ -619,7 +619,11 @@ async def _wait_for_one(self, resolve=False):
         # Wait for the next future to be done and return it unless resolve is
         # set, in which case return either the result of the future or raise
         # an exception.
-        f = await self._done.get()
+        try:
+            f = await self._done.get()
+        except:
+            self._todo_left += 1
+            raise
         if f is None:
             # Dummy value from _handle_timeout().
             raise exceptions.TimeoutError

Lib/test/test_asyncio/test_tasks.py

@@ -1702,6 +1702,46 @@ async def coro(i):
                     )
                     self.assertCountEqual(results, (0, 1, 2, 3))
 
+    def test_as_completed_async_iterator_cancelled_anext(self):
+        async def main():
+            loop = asyncio.get_running_loop()
+            a = loop.create_future()
+            b = loop.create_future()
+            iterator = asyncio.as_completed([a, b])
+
+            waiter = asyncio.create_task(anext(iterator))
+            await asyncio.sleep(0)
+            waiter.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await waiter
+
+            a.set_result('a')
+            b.set_result('b')
+            return [await f async for f in iterator]
+
+        results = self.loop.run_until_complete(main())
+        self.assertCountEqual(results, ['a', 'b'])
+
+    def test_as_completed_plain_iterator_cancelled_wait(self):
+        async def main():
+            loop = asyncio.get_running_loop()
+            a = loop.create_future()
+            b = loop.create_future()
+            iterator = asyncio.as_completed([a, b])
+
+            waiter = asyncio.create_task(next(iterator))
+            await asyncio.sleep(0)
+            waiter.cancel()
+            with self.assertRaises(asyncio.CancelledError):
+                await waiter
+
+            a.set_result('a')
+            b.set_result('b')
+            return [await c for c in iterator]
+
+        results = self.loop.run_until_complete(main())
+        self.assertCountEqual(results, ['a', 'b'])
+
     def test_as_completed_reverse_wait(self):
         # Tests the plain iterator style of as_completed iteration to
         # ensure that the first future awaited resolves to the first

Misc/NEWS.d/next/Library/2026-09-23-19-59-03.gh-issue-158004.p68swW.rst

@@ -0,0 +1,2 @@
+Fix :func:`asyncio.as_completed` to continue yielding every awaitable after
+a pending iteration step is cancelled.