add cancellation support to async iterable iteration

GraphQL-core has no PromiseCanceller class:
cancellableIterable becomes cancellable_iterable,
wrapping each __anext__ in the existing with_abort_signal helper,
used in complete_list_value and map_source_to_response.
No mapAsyncIterable onDone callback is needed since
with_abort_signal cancels its own per-step wait.

Replicates graphql/graphql-js@46857e2

Author: Cito

src/graphql/execution/execute.py

@@ -964,6 +964,34 @@ async def with_abort_signal(self, awaitable: Awaitable[T]) -> T:
         msg = f"Unexpected error value: {inspect(reason)}"
         raise TypeError(msg)
 
+    def cancellable_iterable(self, iterable: AsyncIterable[T]) -> AsyncIterable[T]:
+        """Wrap an async iterable so pending iteration is cancelled on abort.
+
+        When the abort signal is triggered, any pending ``__anext__`` call returns
+        immediately by raising the abort reason. This mirrors the JavaScript
+        ``PromiseCanceller.cancellableIterable``; GraphQL-Core needs no
+        ``PromiseCanceller`` class since :meth:`with_abort_signal` already provides
+        the cancellation mechanism.
+        """
+        if self.abort_signal is None:
+            return iterable
+        with_abort_signal = self.with_abort_signal
+        iterator = iterable.__aiter__()
+
+        class CancellableAsyncIterator:
+            def __aiter__(self) -> AsyncIterator[T]:
+                return self
+
+            def __anext__(self) -> Awaitable[T]:
+                return with_abort_signal(iterator.__anext__())
+
+            async def aclose(self) -> None:
+                aclose = getattr(iterator, "aclose", None)
+                if aclose is not None:
+                    await aclose()
+
+        return CancellableAsyncIterator()
+
     async def complete_awaitable_value(
         self,
         return_type: GraphQLOutputType,
@@ -1189,7 +1217,7 @@ def complete_list_value(
         item_type = return_type.of_type
 
         if self.is_async_iterable(result):
-            async_iterator = result.__aiter__()
+            async_iterator = self.cancellable_iterable(result).__aiter__()
 
             return self.complete_async_iterator_value(
                 item_type,
@@ -1724,7 +1752,7 @@ async def callback(payload: Any) -> ExecutionResult:
                 else cast("ExecutionResult", result)
             )
 
-        return map_async_iterable(result_or_stream, callback)
+        return map_async_iterable(self.cancellable_iterable(result_or_stream), callback)
 
     def collect_execution_groups(
         self,

tests/execution/test_abort_signal.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from asyncio import Event, Future, ensure_future, sleep
-from collections.abc import Awaitable
+from collections.abc import AsyncIterator, Awaitable
 
 import pytest
 
@@ -415,6 +415,102 @@ def todo(_info):
             ],
         )
 
+    async def stops_the_execution_when_aborted_despite_a_hanging_async_item():
+        abort_controller = AbortController()
+        document = parse(
+            """
+      query {
+        todo {
+          id
+          items
+        }
+      }
+    """
+        )
+
+        async def items(_info):
+            # never reached: the iterator is cancelled before its body runs
+            yield await Future()  # will never resolve  # pragma: no cover
+
+        def todo(_info):
+            return {"id": "1", "items": items}
+
+        awaitable_result = execute(
+            schema,
+            document,
+            abort_signal=abort_controller.signal,
+            root_value={"todo": todo},
+        )
+        assert isinstance(awaitable_result, Awaitable)
+
+        abort_controller.abort()
+
+        result = await awaitable_result
+
+        assert result.errors is not None
+        assert isinstance(result.errors[0].original_error, AbortError)
+        assert result == (
+            {"todo": {"id": "1", "items": None}},
+            [
+                {
+                    "message": "This operation was aborted",
+                    "locations": [(5, 11)],
+                    "path": ["todo", "items"],
+                }
+            ],
+        )
+
+    async def stops_the_execution_when_aborted_despite_a_hanging_iterator_no_close():
+        # Like the test above, but the async iterator has no aclose() method, so
+        # the cancellable wrapper has nothing to forward the close to.
+        abort_controller = AbortController()
+        document = parse(
+            """
+      query {
+        todo {
+          id
+          items
+        }
+      }
+    """
+        )
+
+        class Items:
+            def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                # never reached: the iterator is cancelled before its body runs
+                return await Future()  # will never resolve  # pragma: no cover
+
+        def todo(_info):
+            return {"id": "1", "items": Items()}
+
+        awaitable_result = execute(
+            schema,
+            document,
+            abort_signal=abort_controller.signal,
+            root_value={"todo": todo},
+        )
+        assert isinstance(awaitable_result, Awaitable)
+
+        abort_controller.abort()
+
+        result = await awaitable_result
+
+        assert result.errors is not None
+        assert isinstance(result.errors[0].original_error, AbortError)
+        assert result == (
+            {"todo": {"id": "1", "items": None}},
+            [
+                {
+                    "message": "This operation was aborted",
+                    "locations": [(5, 11)],
+                    "path": ["todo", "items"],
+                }
+            ],
+        )
+
     async def stops_the_execution_when_aborted_with_proper_null_bubbling():
         abort_controller = AbortController()
         document = parse(
@@ -532,7 +628,7 @@ async def stops_the_execution_when_aborted_pre_execute():
 
         assert result == (None, [{"message": "This operation was aborted"}])
 
-    async def stops_the_execution_when_aborted_during_subscription():
+    async def stops_the_execution_when_aborted_prior_to_return_of_subscription():
         abort_controller = AbortController()
         document = parse(
             """
@@ -545,17 +641,17 @@ async def stops_the_execution_when_aborted_during_subscription():
         def foo(_info):
             return Future()  # will never resolve
 
-        awaitable_result = subscribe(
+        subscription_promise = subscribe(
             schema,
             document,
             abort_signal=abort_controller.signal,
             root_value={"foo": foo},
         )
-        assert isinstance(awaitable_result, Awaitable)
+        assert isinstance(subscription_promise, Awaitable)
 
         abort_controller.abort()
 
-        result = await awaitable_result
+        result = await subscription_promise
 
         assert result == (
             None,
@@ -567,3 +663,98 @@ def foo(_info):
                 }
             ],
         )
+
+    async def successfully_wraps_the_subscription():
+        abort_controller = AbortController()
+        document = parse(
+            """
+      subscription {
+        foo
+      }
+    """
+        )
+
+        async def foo():
+            yield {"foo": "foo"}
+
+        async def resolve_foo(_info):
+            return foo()
+
+        subscription_promise = subscribe(
+            schema,
+            document,
+            abort_signal=abort_controller.signal,
+            root_value={"foo": resolve_foo},
+        )
+        assert isinstance(subscription_promise, Awaitable)
+        subscription = await subscription_promise
+
+        assert isinstance(subscription, AsyncIterator)
+
+        assert await anext(subscription) == ({"foo": "foo"}, None)
+
+        with pytest.raises(StopAsyncIteration):
+            await anext(subscription)
+
+    async def stops_the_execution_when_aborted_during_subscription():
+        abort_controller = AbortController()
+        document = parse(
+            """
+      subscription {
+        foo
+      }
+    """
+        )
+
+        async def foo():
+            yield {"foo": "foo"}
+
+        subscription = subscribe(
+            schema,
+            document,
+            abort_signal=abort_controller.signal,
+            root_value={"foo": foo()},
+        )
+
+        assert isinstance(subscription, AsyncIterator)
+
+        assert await anext(subscription) == ({"foo": "foo"}, None)
+
+        abort_controller.abort()
+
+        with pytest.raises(AbortError, match="This operation was aborted"):
+            await anext(subscription)
+
+    async def stops_the_execution_when_aborted_during_async_returned_subscription():
+        abort_controller = AbortController()
+        document = parse(
+            """
+      subscription {
+        foo
+      }
+    """
+        )
+
+        async def foo():
+            yield {"foo": "foo"}
+
+        async def resolve_foo(_info):
+            return foo()
+
+        subscription_promise = subscribe(
+            schema,
+            document,
+            abort_signal=abort_controller.signal,
+            root_value={"foo": resolve_foo},
+        )
+        assert isinstance(subscription_promise, Awaitable)
+        subscription = await subscription_promise
+
+        assert isinstance(subscription, AsyncIterator)
+
+        assert await anext(subscription) == ({"foo": "foo"}, None)
+
+        abort_controller.abort()
+
+        with pytest.raises(AbortError, match="This operation was aborted"):
+            await anext(subscription)