Finish async iterables in the non incremental delivery case

Replicates graphql/graphql-js@d9fc656

Author: Cito

src/graphql/execution/execute.py

@@ -941,82 +941,88 @@ async def complete_async_iterator_value(
         awaitable_indices: list[int] = []
         append_awaitable = awaitable_indices.append
         stream_usage = self.get_stream_usage(field_group, path)
+        try:
+            early_return = async_iterator.aclose  # type: ignore[attr-defined]
+        except AttributeError:
+            early_return = None
         index = 0
-        while True:
-            if stream_usage and index >= stream_usage.initial_count:
-                stream_item_queue = self.build_async_stream_item_queue(
-                    index,
-                    path,
-                    async_iterator,
-                    stream_usage.field_group,
-                    info,
-                    item_type,
-                )
-
-                try:
-                    early_return = async_iterator.aclose()  # type: ignore
-                except AttributeError:
-                    early_return = None
-                stream_record: StreamRecord
-
-                if early_return is None:
-                    stream_record = StreamRecord(
-                        stream_item_queue, path, stream_usage.label
-                    )
-                else:
-                    stream_record = CancellableStreamRecord(
-                        early_return,
-                        stream_item_queue,
+        try:
+            while True:
+                if stream_usage and index >= stream_usage.initial_count:
+                    stream_item_queue = self.build_async_stream_item_queue(
+                        index,
                         path,
-                        stream_usage.label,
+                        async_iterator,
+                        stream_usage.field_group,
+                        info,
+                        item_type,
                     )
-                    if self.cancellable_streams is None:  # pragma: no branch
-                        self.cancellable_streams = set()
-                    self.cancellable_streams.add(stream_record)
-                    self._canceled_iterators.add(async_iterator)
 
-                add_increment(stream_record)
-                break
+                    stream_record: StreamRecord
 
-            item_path = path.add_key(index, None)
-            try:
-                item = await anext(async_iterator)
-            except StopAsyncIteration:
-                break
-            except Exception as raw_error:
-                raise located_error(
-                    raw_error, to_nodes(field_group), path.as_list()
-                ) from raw_error
+                    if early_return is None:
+                        stream_record = StreamRecord(
+                            stream_item_queue, path, stream_usage.label
+                        )
+                    else:
+                        stream_record = CancellableStreamRecord(
+                            early_return(),
+                            stream_item_queue,
+                            path,
+                            stream_usage.label,
+                        )
+                        if self.cancellable_streams is None:  # pragma: no branch
+                            self.cancellable_streams = set()
+                        self.cancellable_streams.add(stream_record)
+                        self._canceled_iterators.add(async_iterator)
 
-            if is_awaitable(item):
-                append_completed(
-                    complete_awaitable_list_item_value(
-                        item,
-                        graphql_wrapped_result,
-                        item_type,
-                        field_group,
-                        info,
-                        item_path,
-                        incremental_context,
-                        defer_map,
-                    )
-                )
-                append_awaitable(index)
+                    add_increment(stream_record)
+                    break
 
-            elif complete_list_item_value(
-                item,
-                completed_results,
-                graphql_wrapped_result,
-                item_type,
-                field_group,
-                info,
-                item_path,
-                incremental_context,
-                defer_map,
-            ):
-                append_awaitable(index)
+                item_path = path.add_key(index, None)
+                try:
+                    item = await anext(async_iterator)
+                except StopAsyncIteration:
+                    break
+                except Exception as raw_error:
+                    raise located_error(
+                        raw_error, to_nodes(field_group), path.as_list()
+                    ) from raw_error
+
+                if is_awaitable(item):
+                    append_completed(
+                        complete_awaitable_list_item_value(
+                            item,
+                            graphql_wrapped_result,
+                            item_type,
+                            field_group,
+                            info,
+                            item_path,
+                            incremental_context,
+                            defer_map,
+                        )
+                    )
+                    append_awaitable(index)
 
-            index += 1
+                elif complete_list_item_value(
+                    item,
+                    completed_results,
+                    graphql_wrapped_result,
+                    item_type,
+                    field_group,
+                    info,
+                    item_path,
+                    incremental_context,
+                    defer_map,
+                ):
+                    append_awaitable(index)
+
+                index += 1
+        except Exception:
+            if early_return is not None:  # pragma: no branch
+                with suppress_exceptions:
+                    await early_return()
+            raise
 
         if not awaitable_indices:
             return graphql_wrapped_result

tests/execution/test_lists.py

@@ -356,6 +356,37 @@ async def list_field():
         assert await _complete(list_field(), "[Int!]") == ({"listField": None}, errors)
         assert await _complete(list_field(), "[Int!]!") == (None, errors)
 
+    async def calls_aclose_when_non_null_list_item_errors():
+        values = (1, None, 2)
+
+        class ListField:
+            def __init__(self) -> None:
+                self.index = 0
+                self.closed = False
+
+            def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                value = values[self.index]
+                self.index += 1
+                return value
+
+            async def aclose(self) -> None:
+                self.closed = True
+
+        list_field = ListField()
+        errors = [
+            {
+                "message": "Cannot return null for non-nullable field Query.listField.",
+                "locations": [(1, 3)],
+                "path": ["listField", 1],
+            }
+        ]
+
+        assert await _complete(list_field, "[Int!]") == ({"listField": None}, errors)
+        assert list_field.closed
+
 
 def describe_execute_handles_list_nullability():
     async def _complete(list_field: Any, as_type: str) -> ExecutionResult: