Incremental delivery: experimental_execute_incrementally hangs (never terminates) on some invalid @stream queries

Hi there,

Whilst investigating a production bug (see #271), the fuzzing harness found a second bug. Not as critical as 271 because it only triggers on an invalid graphql shape, but it is a reproducible hang that can be reached in production. Found with the help of Claude code.

Hope this helps - unfortunately I don't have a suggested fix for this one as the root cause was less obvious.

- Ben (Mixcloud)

---

## Summary

`experimental_execute_incrementally` **never terminates** for certain queries combining `@stream` with nested `@defer`/child selections — the async iterator over `subsequent_results` blocks forever, with no final `hasNext: false`. Reproduced reliably on **graphql-core 3.3.0rc0** (≈13/15 attempts hang at a 3 s timeout in isolation, ≈9/10 at 5 s).

**Scope, stated up front:** every deadlocking query I found is **spec-invalid** — it fails `graphql.validate` (a streamed list field with no subselection, and/or a field-merge conflict between a streamed and a non-streamed selection of the same field). A caller that validates before executing rejects these first. Despite extensive search I could **not** find a spec-*valid* query that hangs — making the query valid removes the hang. So this is a robustness issue rather than a correctness issue on valid input: execution should fail fast / terminate rather than hang, since `experimental_execute_incrementally` can be (and is, for performance) called without a separate validation pass, and a hang on any input is a DoS risk.

## Environment

- graphql-core `3.3.0rc0`
- Python 3.14

## Query (spec-invalid; hangs instead of erroring)

```graphql
query {
obj {
... {
child {
child { ... @defer(label: "L4") { items @stream(initialCount: 1) } }
... @defer { child { fast items } }
}
}
}
}
```

`validate()` reports: _"Fields 'child' conflict because subfields 'items' conflict because they have differing stream directives"_ and _"Field 'items' … must have a selection of subfields"_. `experimental_execute_incrementally` neither raises nor completes.

## What is / isn't required (isolation testing)

| Query | Valid? | Hangs? |
| --- | --- | --- |
| The query above | invalid | **yes (~13/15)** |
| `items @stream(initialCount: 1)` with no subselection, alone | invalid | no |
| Streamed + non-streamed `items` merge conflict, alone | invalid | no |
| The same structure made valid (aliased, subselections added) | valid | **no** |
| ~15 other hand-crafted valid `@stream`/`@defer`/nesting shapes | valid | no |

So neither invalid feature *alone* hangs; the deadlock needs the invalid feature embedded in the nested `@defer` + `@stream` + child structure — and making that structure valid stops it.

## Notes

- Timing-sensitive (needs out-of-order async resolver completion), so the repro loops; not 100%/run but high (≈85–90%).
- Likely the same incremental-publisher/graph subsystem as the `_enqueue` "race with a stopping consumer" guard and the separate `_add_deferred_fragment_node` invariant crash (filed as a separate issue): the graph doesn't converge to a terminal state for these inputs.

### Self-contained reproduction (graphql-core only)

```python
import asyncio
import random

from graphql import (
GraphQLDeferDirective,
GraphQLField,
GraphQLList,
GraphQLObjectType,
GraphQLSchema,
GraphQLStreamDirective,
GraphQLString,
parse,
specified_directives,
validate,
)
from graphql.execution import (
ExperimentalIncrementalExecutionResults,
experimental_execute_incrementally,
)
from graphql.pyutils import is_awaitable

ATTEMPTS = 15
TIMEOUT_S = 3.0

async def r_fast(_s, _i) -> str:
await asyncio.sleep(random.uniform(0, 0.002))
return "fast"

async def r_obj(_s, _i) -> dict:
await asyncio.sleep(random.uniform(0, 0.001))
return {}

async def r_list(_s, _i) -> list:
await asyncio.sleep(random.uniform(0, 0.001))
return [{}, {}]

OBJ = GraphQLObjectType(
"Obj",
lambda: {
"fast": GraphQLField(GraphQLString, resolve=r_fast),
"child": GraphQLField(OBJ, resolve=r_obj),
"items": GraphQLField(GraphQLList(OBJ), resolve=r_list),
},
)
schema = GraphQLSchema(
GraphQLObjectType("Query", {"obj": GraphQLField(OBJ, resolve=r_obj)}),
directives=[*specified_directives, GraphQLDeferDirective, GraphQLStreamDirective],
)

QUERY = """
query {
obj {
... {
child {
child { ... @defer(label: "L4") { items @stream(initialCount: 1) } }
... @defer { child { fast items } }
}
}
}
}
"""

async def run_once() -> None:
result = experimental_execute_incrementally(schema, parse(QUERY), root_value={})
if is_awaitable(result):
result = await result
if isinstance(result, ExperimentalIncrementalExecutionResults):
async for _patch in result.subsequent_results:
pass

async def main() -> None:
import graphql

print("graphql-core:", graphql.__version__)
errs = validate(schema, parse(QUERY))
print("query is spec-" + ("INVALID" if errs else "VALID"))
hangs = 0
for i in range(1, ATTEMPTS + 1):
try:
await asyncio.wait_for(run_once(), TIMEOUT_S)
except TimeoutError:
hangs += 1
print(f"attempt {i}: HUNG (no termination within {TIMEOUT_S}s)")
print(f"{hangs}/{ATTEMPTS} attempts hung (never terminated).")

asyncio.run(main())
```

---

_This non-termination, and the finding that valid queries do not reproduce it, were established by fuzzing `@defer`/`@stream` query shapes._

Start with the self-contained reproduction and run it repeatedly under graphql-core 3.3.0rc0 to observe the timeout. Trace experimental_execute_incrementally through the incremental-publisher/graph subsystem, especially _enqueue and _add_deferred_fragment_node. Done means the nested invalid @stream/@defer query terminates with an error or final hasNext: false without changing the behavior of the valid-query cases.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.