Fix RuntimeError "Invalid state while adding deferred fragment node"

`IncrementalGraph._add_deferred_fragment_node` raised an internal
RuntimeError while executing a spec-valid query when two sibling top-level
`@defer` fragments — each resolving a field that errors and each containing
its own nested `@defer` — were executed with async resolvers.

When a deferred field errors, that fragment's execution group fails and
the (parentless, root) deferred fragment is removed from the root nodes. A
sibling execution group completing afterwards then discovers its nested
`@defer` and, in `_add_incremental_data_records`, recurses up to the
now-removed parentless fragment, reaching the `# pragma: no cover` branch
and raising.

Return instead of raising when reaching a parentless fragment that is no
longer a root node: its subtree is no longer being delivered, so there is
nothing to attach it to. This mirrors the `future.cancelled()` guard added
to `_enqueue` (a race with a stopping/removed consumer).

Adds a regression test that reproduces the crash prior to this change.

Fixes #271

Author: ben-xo

src/graphql/execution/incremental_graph.py

@@ -257,9 +257,16 @@ def _add_deferred_fragment_node(
             return
         parent = deferred_fragment_record.parent
         if parent is None:
-            if initial_result_children is None:  # pragma: no cover
-                msg = "Invalid state while adding deferred fragment node."
-                raise RuntimeError(msg)
+            if initial_result_children is None:
+                # The fragment has already been removed from the root nodes
+                # (e.g. its execution group failed on a resolver error) while a
+                # sibling execution group that references its subtree is still
+                # completing. There is nothing to attach the record to and its
+                # subtree is no longer being delivered, so drop it rather than
+                # treating this as an invalid state. Mirrors the
+                # `future.cancelled()` guard in `_enqueue` — a race with a
+                # stopping/removed consumer.
+                return
             initial_result_children[deferred_fragment_record] = None
             return
         parent.children[deferred_fragment_record] = None

tests/execution/test_defer.py

@@ -3163,3 +3163,78 @@ async def author(_info):
         with pytest.raises(AbortError, match="This operation was aborted"):
             await anext_task
         assert items_source.aclose_finished
+
+
+def describe_defer_directive_with_errors_and_nested_defer():
+    """Regression tests for graphql-python/graphql-core#271."""
+
+    async def does_not_reach_invalid_state_when_sibling_fragments_error():
+        """Sibling erroring ``@defer`` fragments with nested ``@defer`` must not crash.
+
+        Two sibling top-level ``@defer`` fragments that each resolve a distinct
+        erroring field and each contain a nested ``@defer`` used to crash the
+        incremental graph with "Invalid state while adding deferred fragment
+        node": a failing
+        execution group removes its parentless (root) deferred fragment, and a
+        sibling execution group completing afterwards then recursed up into the
+        removed record and hit the invariant. See issue #271.
+        """
+
+        async def resolve_fast(_source, _info) -> str:
+            await sleep(0)
+            return "fast"
+
+        def resolve_error(_source, _info) -> str:
+            raise RuntimeError("bad")
+
+        async def resolve_obj(_source, _info) -> dict:
+            await sleep(0)
+            return {}
+
+        obj_type: GraphQLObjectType = GraphQLObjectType(
+            "Obj",
+            lambda: {
+                "fast": GraphQLField(GraphQLString, resolve=resolve_fast),
+                "boom": GraphQLField(GraphQLString, resolve=resolve_error),
+                "boomNonNull": GraphQLField(
+                    GraphQLNonNull(GraphQLString), resolve=resolve_error
+                ),
+                "child": GraphQLField(obj_type, resolve=resolve_obj),
+            },
+        )
+        local_schema = GraphQLSchema(
+            GraphQLObjectType(
+                "Query", {"obj": GraphQLField(obj_type, resolve=resolve_obj)}
+            )
+        )
+        # The two siblings must error on DISTINCT fields (`boom` vs `boomNonNull`):
+        # erroring on the same field in both does not reproduce the crash. The
+        # nested `@defer` selections may be identical.
+        document = parse(
+            """
+            query {
+              obj {
+                ... @defer(label: "a") {
+                  boom
+                  child { ... @defer(label: "b") { fast } }
+                }
+                ... @defer(label: "c") {
+                  boomNonNull
+                  child { ... @defer(label: "d") { fast } }
+                }
+              }
+            }
+            """
+        )
+
+        for early_execution in (False, True):
+            result = experimental_execute_incrementally(
+                local_schema, document, {}, enable_early_execution=early_execution
+            )
+            assert is_awaitable(result)
+            result = await result
+            assert isinstance(result, ExperimentalIncrementalExecutionResults)
+            # Draining the stream must terminate without raising
+            # "Invalid state while adding deferred fragment node".
+            async for _patch in result.subsequent_results:
+                pass