Closing a subscription leaves a distinct source iterator open #283

On current main (d913a61), map_async_iterable registers cleanup on the async iterable object, while async iteration consumes the object returned by its aiter method. A valid wrapper can return a separate async generator. Closing the mapped stream then leaves that generator open, so its finally block does not run. The same behavior is reachable through subscribe().

Minimal reproducer:

import asyncio
from graphql.execution.async_iterables import map_async_iterable

async def main():
    closed = False

    async def events():
        nonlocal closed
        try:
            yield 1
            yield 2
        finally:
            closed = True

    iterator = events()

    class Events:
        def __aiter__(self):
            return iterator

    async def identity(value):
        return value

    mapped = map_async_iterable(Events(), identity)
    try:
        assert await anext(mapped) == 1
        await mapped.aclose()
        assert closed  # Fails: the source iterator is still open.
    finally:
        await iterator.aclose()

asyncio.run(main())

Expected: closing the mapped stream awaits cleanup of the iterator actually supplying events. Actual: cleanup is attempted on the wrapper; the generator remains open and can yield the second event.

A narrow local fix acquires the iterator before registering cleanup (aclosing(aiter(iterable))). Four new regressions fail on the base and pass with that change, including one through the public subscription API. The full local suites pass on Python 3.10 and 3.14. This differs from the broader AnyIO/cancellation work in #165 and #166; it does not change pending-next cancellation behavior.

Does closing the consumed iterator match the intended port contract? I can link the focused patch after discussion, following the PR template. AI assistance was used to investigate and validate this report.