`asyncio.as_completed()` drops an awaitable after a pending `anext()` call is cancelled

[lpyu001]

Bug description:

Summary

Cancelling a pending asynchronous iteration step from asyncio.as_completed() consumes one of the iterator's remaining slots
without yielding an awaitable. If iteration is resumed after handling the cancellation, it stops one item early and one completed future is never yielded.

This can occur when a caller applies asyncio.timeout() to each iteration step and continues iterating after a timeout.

Reproduction Code

import asyncio


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()
    try:
        await waiter
    except asyncio.CancelledError:
        pass

    a.set_result("a")
    b.set_result("b")
    results = [await future async for future in iterator]
    print(results)


asyncio.run(main())

Actual Behavior

The program prints:

['a']

The iterator ends normally after yielding only one of the two futures.

Expected Behavior

The cancelled anext() call did not yield an awaitable, so resuming the iterator should still yield both futures. The program should print:

['a', 'b']

CPython versions tested on:

CPython main branch

Operating systems tested on:

Linux

Linked PRs

• gh-158004: Preserve as_completed items after anext cancellation #158005