fix(incremental): emit only single completion when multiple deferred grouped field sets error

Replicates graphql/graphql-js@03cc5dc

Author: Cito

src/graphql/execution/incremental_graph.py

@@ -254,17 +254,18 @@ def complete_deferred_fragment(
     def remove_deferred_fragment(
         self,
         deferred_fragment_record: DeferredFragmentRecord,
-    ) -> None:
-        """Remove a deferred fragment."""
+    ) -> bool:
+        """Check if deferred fragment exists and remove it in that case."""
+        deferred_fragment_nodes = self._deferred_fragment_nodes
         try:
-            deferred_fragment_node = self._deferred_fragment_nodes[
-                deferred_fragment_record
-            ]
+            deferred_fragment_node = deferred_fragment_nodes[deferred_fragment_record]
         except KeyError:  # pragma: no cover
-            return
+            return False
         self._remove_pending(deferred_fragment_node)
+        del deferred_fragment_nodes[deferred_fragment_record]
         for child in deferred_fragment_node.children:  # pragma: no cover
             self.remove_deferred_fragment(child.deferred_fragment_record)
+        return True
 
     def remove_stream(self, stream_record: StreamRecord) -> None:
         """Remove a stream record as no longer pending."""

src/graphql/execution/incremental_publisher.py

@@ -221,12 +221,17 @@ def _handle_completed_deferred_grouped_field_set(
         ):
             remove_deferred = self._incremental_graph.remove_deferred_fragment
             for deferred_fragment_record in record.deferred_fragment_records:
+                if not remove_deferred(deferred_fragment_record):
+                    # multiple deferred grouped field sets could error for a fragment
+                    continue
                 id_ = deferred_fragment_record.id
-                if id_ is not None:  # pragma: no branch
-                    append_completed(
-                        CompletedResult(id_, deferred_grouped_field_set_result.errors)
-                    )
-                    remove_deferred(deferred_fragment_record)
+                if id_ is None:  # pragma: no cover
+                    msg = "Missing deferred fragment record identifier."
+                    raise RuntimeError(msg)
+                append_completed(
+                    CompletedResult(id_, deferred_grouped_field_set_result.errors)
+                )
+                remove_deferred(deferred_fragment_record)
             return
 
         deferred_grouped_field_set_result = cast(

tests/execution/test_defer.py

@@ -1580,6 +1580,154 @@ async def nulls_cross_defer_boundaries_null_first():
             },
         ]
 
+    async def handles_multiple_erroring_deferred_grouped_field_sets():
+        """Handles multiple erroring deferred grouped field sets"""
+        document = parse(
+            """
+            query {
+              ... @defer {
+                a {
+                  b {
+                    c {
+                      someError: nonNullErrorField
+                    }
+                  }
+                }
+              }
+              ... @defer {
+                a {
+                  b {
+                    c {
+                      anotherError: nonNullErrorField
+                    }
+                  }
+                }
+              }
+            }
+            """
+        )
+        result = await complete(
+            document,
+            {"a": {"b": {"c": {"nonNullErrorField": None}}}},
+        )
+        assert result == [
+            {
+                "data": {},
+                "pending": [
+                    {"id": "0", "path": []},
+                    {"id": "1", "path": []},
+                ],
+                "hasNext": True,
+            },
+            {
+                "completed": [
+                    {
+                        "id": "0",
+                        "errors": [
+                            {
+                                "message": "Cannot return null"
+                                " for non-nullable field c.nonNullErrorField.",
+                                "locations": [{"line": 7, "column": 23}],
+                                "path": ["a", "b", "c", "someError"],
+                            },
+                        ],
+                    },
+                    {
+                        "id": "1",
+                        "errors": [
+                            {
+                                "message": "Cannot return null"
+                                " for non-nullable field c.nonNullErrorField.",
+                                "locations": [{"line": 16, "column": 23}],
+                                "path": ["a", "b", "c", "anotherError"],
+                            },
+                        ],
+                    },
+                ],
+                "hasNext": False,
+            },
+        ]
+
+    async def handles_multiple_erroring_deferred_grouped_field_sets_for_same_fragment():
+        """Handles multiple erroring deferred grouped field sets for same fragment"""
+        document = parse(
+            """
+            query {
+              ... @defer {
+                a {
+                  b {
+                    someC: c {
+                      d: d
+                    }
+                    anotherC: c {
+                      d: d
+                    }
+                  }
+                }
+              }
+              ... @defer {
+                a {
+                  b {
+                    someC: c {
+                      someError: nonNullErrorField
+                    }
+                    anotherC: c {
+                      anotherError: nonNullErrorField
+                    }
+                  }
+                }
+              }
+            }
+            """
+        )
+        result = await complete(
+            document,
+            {"a": {"b": {"c": {"d": "d", "nonNullErrorField": None}}}},
+        )
+        assert result == [
+            {
+                "data": {},
+                "pending": [
+                    {"id": "0", "path": []},
+                    {"id": "1", "path": []},
+                ],
+                "hasNext": True,
+            },
+            {
+                "incremental": [
+                    {
+                        "data": {"a": {"b": {"someC": {}, "anotherC": {}}}},
+                        "id": "0",
+                    },
+                    {
+                        "data": {"d": "d"},
+                        "id": "0",
+                        "subPath": ["a", "b", "someC"],
+                    },
+                    {
+                        "data": {"d": "d"},
+                        "id": "0",
+                        "subPath": ["a", "b", "anotherC"],
+                    },
+                ],
+                "completed": [
+                    {
+                        "id": "1",
+                        "errors": [
+                            {
+                                "message": "Cannot return null"
+                                " for non-nullable field c.nonNullErrorField.",
+                                "locations": [{"line": 19, "column": 23}],
+                                "path": ["a", "b", "someC", "someError"],
+                            },
+                        ],
+                    },
+                    {"id": "0"},
+                ],
+                "hasNext": False,
+            },
+        ]
+
     async def nulls_cross_defer_boundaries_value_first():
         """Nulls cross defer boundaries, value first"""
         document = parse(