fix(incrementalDelivery): fix null bubbling with async iterables

Replicates graphql/graphql-js@0b7daed

Author: Cito

src/graphql/execution/execute.py

@@ -1873,29 +1873,31 @@ async def execute_stream_iterator(
                 label, item_path, iterator, previous_async_payload_record, self
             )
 
-            awaitable_data = self.execute_stream_iterator_item(
-                iterator, field_modes, info, item_type, async_payload_record, item_path
-            )
-
             try:
-                data = await awaitable_data
+                data = await self.execute_stream_iterator_item(
+                    iterator,
+                    field_modes,
+                    info,
+                    item_type,
+                    async_payload_record,
+                    item_path,
+                )
             except StopAsyncIteration:
                 if async_payload_record.errors:
                     async_payload_record.add_items(None)  # pragma: no cover
                 else:
                     del self.subsequent_payloads[async_payload_record]
                 break
             except GraphQLError as error:
-                # entire stream has errored and bubbled upwards
+                async_payload_record.errors.append(error)
                 self.filter_subsequent_payloads(path, async_payload_record)
+                async_payload_record.add_items(None)
                 if iterator:  # pragma: no cover else
                     with suppress(Exception):
                         await iterator.aclose()  # type: ignore
                     # running generators cannot be closed since Python 3.8,
                     # so we need to remember that this iterator is already canceled
                     self._canceled_iterators.add(iterator)
-                async_payload_record.add_items(None)
-                async_payload_record.errors.append(error)
                 break
 
             async_payload_record.add_items([data])

tests/execution/test_executor.py

@@ -519,11 +519,7 @@ def handles_sync_errors_combined_with_async_ones():
 
         async def async_resolver(_obj, _info):
             nonlocal is_async_resolver_finished
-            sleep = asyncio.sleep
-            sleep(0)
-            sleep(0)
-            sleep(0)
-            is_async_resolver_finished = True
+            is_async_resolver_finished = True  # pragma: no cover
 
         schema = GraphQLSchema(
             GraphQLObjectType(

tests/execution/test_stream.py

@@ -1020,6 +1020,68 @@ async def scalar_list(_info):
 
     @pytest.mark.asyncio()
     async def handles_async_error_in_complete_value_after_initial_count_is_reached():
+        document = parse(
+            """
+            query {
+              friendList @stream(initialCount: 1) {
+                nonNullName
+              }
+            }
+            """
+        )
+
+        async def throw():
+            raise RuntimeError("Oops")
+
+        async def get_friend(i):
+            await sleep(0)
+            return {"nonNullName": throw() if i < 0 else friends[i].name}
+
+        def get_friends(_info):
+            return [get_friend(0), get_friend(-1), get_friend(1)]
+
+        result = await complete(
+            document,
+            {
+                "friendList": get_friends,
+            },
+        )
+        assert result == [
+            {
+                "data": {
+                    "friendList": [{"nonNullName": "Luke"}],
+                },
+                "hasNext": True,
+            },
+            {
+                "incremental": [
+                    {
+                        "items": [None],
+                        "path": ["friendList", 1],
+                        "errors": [
+                            {
+                                "message": "Oops",
+                                "locations": [{"line": 4, "column": 17}],
+                                "path": ["friendList", 1, "nonNullName"],
+                            },
+                        ],
+                    },
+                ],
+                "hasNext": True,
+            },
+            {
+                "incremental": [
+                    {
+                        "items": [{"nonNullName": "Han"}],
+                        "path": ["friendList", 2],
+                    },
+                ],
+                "hasNext": False,
+            },
+        ]
+
+    @pytest.mark.asyncio()
+    async def handles_async_error_in_complete_value_for_non_nullable_list():
         document = parse(
             """
             query {
@@ -1230,7 +1292,6 @@ async def friend_list(_info):
         }
 
     @pytest.mark.asyncio()
-    @pytest.mark.filterwarnings("ignore:.* was never awaited:RuntimeWarning")
     async def does_not_filter_payloads_when_null_error_is_in_a_different_path():
         document = parse(
             """
@@ -1298,7 +1359,6 @@ async def friend_list(_info):
         ]
 
     @pytest.mark.asyncio()
-    @pytest.mark.filterwarnings("ignore:.* was never awaited:RuntimeWarning")
     async def filters_stream_payloads_that_are_nulled_in_a_deferred_payload():
         document = parse(
             """