Skip to content

Commit 3f91c1c

Browse files
CitoCito
committed
fix(incremental): prevent lost or delayed successful shared results
Replicates graphql/graphql-js@36d96f2
1 parent 3d4a6c5 commit 3f91c1c

4 files changed

Lines changed: 418 additions & 22 deletions

File tree

‎src/graphql/execution/incremental/work_queue.py‎

Lines changed: 25 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -437,7 +437,8 @@ def _prune_empty_groups(
437437
for new_group in new_groups:
438438
new_group_node = group_nodes.get(new_group)
439439
if new_group_node:
440-
if new_group_node.pending:
440+
# Settled tasks can still hold values that have not been published.
441+
if new_group_node.tasks:
441442
non_empty_new_groups.append(new_group)
442443
else:
443444
del group_nodes[new_group]
@@ -448,14 +449,29 @@ def _prune_empty_groups(
448449

449450
def _start_new_work(
450451
self, new_groups: Sequence[Group], new_streams: Sequence[Stream]
451-
) -> None:
452-
"""Promote new groups and streams to root and start them."""
453-
for group in new_groups:
452+
) -> list[GroupValuesEvent | GroupSuccessEvent]:
453+
"""Promote new groups and streams to root and start them.
454+
455+
Returns the events for groups that could be finished right away.
456+
"""
457+
group_events: list[GroupValuesEvent | GroupSuccessEvent] = []
458+
# Completing a released group can append more groups and streams to start.
459+
groups_to_start = list(new_groups)
460+
streams_to_start = list(new_streams)
461+
for group in groups_to_start:
454462
self._root_groups[group] = None
455463
self._start_group(group)
456-
for stream in new_streams:
464+
group_node = self._group_nodes.get(group)
465+
# defensive guard mirroring the JS version; starting a group
466+
# never removes its node
467+
if group_node: # pragma: no branch
468+
self._maybe_finish_group(
469+
group, group_node, group_events, groups_to_start, streams_to_start
470+
)
471+
for stream in streams_to_start:
457472
self._root_streams[stream] = None
458473
self._start_stream(stream)
474+
return group_events
459475

460476
def _start_group(self, group: Group) -> None:
461477
"""Start all tasks of a group."""
@@ -559,8 +575,7 @@ def _task_success(
559575
group, group_node, group_events, new_groups, new_streams
560576
)
561577

562-
self._start_new_work(new_groups, new_streams)
563-
return group_events
578+
return [*group_events, *self._start_new_work(new_groups, new_streams)]
564579

565580
def _maybe_finish_group(
566581
self,
@@ -610,6 +625,9 @@ def _stream_items(
610625
for value, work in items:
611626
item_new_groups, item_new_streams = self._maybe_integrate_work(work)
612627
non_empty_new_groups = self._prune_empty_groups(item_new_groups)
628+
# These groups are newly introduced, so their task outcomes have not yet
629+
# been processed. Even synchronous task results are queued as graph
630+
# events. Therefore, no group events can be returned here.
613631
self._start_new_work(non_empty_new_groups, item_new_streams)
614632
values.append(value)
615633
new_groups.extend(non_empty_new_groups)

‎tests/execution/incremental/test_defer.py‎

Lines changed: 292 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2537,7 +2537,6 @@ async def reports_one_cached_shared_failure_when_a_nested_fragment_is_announced(
25372537
},
25382538
]
25392539

2540-
@pytest.mark.xfail(strict=True, reason="lost shared result")
25412540
@pytest.mark.parametrize("early_execution", [False, True])
25422541
async def delivers_a_buffered_shared_value_after_its_other_fragment_fails(
25432542
early_execution,
@@ -2612,9 +2611,300 @@ async def delivers_a_buffered_shared_value_after_its_other_fragment_fails(
26122611
"hasNext": False,
26132612
}
26142613

2615-
with pytest.raises(StopAsyncIteration): # pragma: no cover
2614+
with pytest.raises(StopAsyncIteration):
2615+
await anext(iterator)
2616+
2617+
async def delivers_a_shared_value_before_another_fragment_finishes_its_field():
2618+
"""Delivers a shared value before another fragment finishes its field
2619+
2620+
Delivers a shared value before another fragment finishes its separate field.
2621+
"""
2622+
document = parse(
2623+
"""
2624+
{
2625+
hero {
2626+
... @defer(label: "R") {
2627+
nonNullName
2628+
id
2629+
}
2630+
... @defer(label: "P") {
2631+
name
2632+
... @defer(label: "C") {
2633+
id
2634+
}
2635+
}
2636+
}
2637+
}
2638+
"""
2639+
)
2640+
name_future: Future[str] = Future()
2641+
non_null_name_future: Future[str] = Future()
2642+
2643+
result = experimental_execute_incrementally(
2644+
schema,
2645+
document,
2646+
{
2647+
"hero": {
2648+
"id": 1,
2649+
"name": name_future,
2650+
"nonNullName": non_null_name_future,
2651+
}
2652+
},
2653+
)
2654+
assert isinstance(result, ExperimentalIncrementalExecutionResults)
2655+
2656+
assert result.initial_result.formatted == {
2657+
"data": {"hero": {}},
2658+
"pending": [
2659+
{"id": "0", "path": ["hero"], "label": "R"},
2660+
{"id": "1", "path": ["hero"], "label": "P"},
2661+
],
2662+
"hasNext": True,
2663+
}
2664+
2665+
iterator = result.subsequent_results
2666+
2667+
name_future.set_result("Luke")
2668+
2669+
assert (await anext(iterator)).formatted == {
2670+
"pending": [{"id": "2", "path": ["hero"], "label": "C"}],
2671+
"incremental": [
2672+
{"data": {"name": "Luke"}, "id": "1"},
2673+
{"data": {"id": "1"}, "id": "2"},
2674+
],
2675+
"completed": [{"id": "1"}, {"id": "2"}],
2676+
"hasNext": True,
2677+
}
2678+
2679+
non_null_name_future.set_result("Luke")
2680+
2681+
assert (await anext(iterator)).formatted == {
2682+
"incremental": [{"data": {"nonNullName": "Luke"}, "id": "0"}],
2683+
"completed": [{"id": "0"}],
2684+
"hasNext": False,
2685+
}
2686+
2687+
with pytest.raises(StopAsyncIteration):
2688+
await anext(iterator)
2689+
2690+
async def delivers_shared_child_data_when_second_parent_finishes_first():
2691+
"""Delivers shared child data when second parent finishes before first child
2692+
2693+
Delivers shared child data when the second parent finishes before the first
2694+
child.
2695+
"""
2696+
document = parse(
2697+
"""
2698+
{
2699+
hero {
2700+
... @defer(label: "P1") {
2701+
name
2702+
... @defer(label: "C1") {
2703+
id
2704+
friends { name }
2705+
}
2706+
}
2707+
... @defer(label: "P2") {
2708+
nonNullName
2709+
... @defer(label: "C2") {
2710+
id
2711+
}
2712+
}
2713+
}
2714+
}
2715+
"""
2716+
)
2717+
name_future: Future[str] = Future()
2718+
friends_future: Future[list[Friend]] = Future()
2719+
2720+
result = experimental_execute_incrementally(
2721+
schema,
2722+
document,
2723+
{
2724+
"hero": {
2725+
"id": 1,
2726+
"name": "Luke",
2727+
"nonNullName": name_future,
2728+
"friends": friends_future,
2729+
}
2730+
},
2731+
)
2732+
assert isinstance(result, ExperimentalIncrementalExecutionResults)
2733+
2734+
assert result.initial_result.formatted == {
2735+
"data": {"hero": {}},
2736+
"pending": [
2737+
{"id": "0", "path": ["hero"], "label": "P1"},
2738+
{"id": "1", "path": ["hero"], "label": "P2"},
2739+
],
2740+
"hasNext": True,
2741+
}
2742+
2743+
iterator = result.subsequent_results
2744+
2745+
assert (await anext(iterator)).formatted == {
2746+
"pending": [{"id": "2", "path": ["hero"], "label": "C1"}],
2747+
"incremental": [{"data": {"name": "Luke"}, "id": "0"}],
2748+
"completed": [{"id": "0"}],
2749+
"hasNext": True,
2750+
}
2751+
2752+
name_future.set_result("Luke")
2753+
2754+
assert (await anext(iterator)).formatted == {
2755+
"pending": [{"id": "3", "path": ["hero"], "label": "C2"}],
2756+
"incremental": [
2757+
{"data": {"nonNullName": "Luke"}, "id": "1"},
2758+
{"data": {"id": "1"}, "id": "3"},
2759+
],
2760+
"completed": [{"id": "1"}, {"id": "3"}],
2761+
"hasNext": True,
2762+
}
2763+
2764+
friends_future.set_result(friends)
2765+
2766+
assert (await anext(iterator)).formatted == {
2767+
"incremental": [
2768+
{
2769+
"data": {
2770+
"friends": [
2771+
{"name": "Han"},
2772+
{"name": "Leia"},
2773+
{"name": "C-3PO"},
2774+
]
2775+
},
2776+
"id": "2",
2777+
},
2778+
],
2779+
"completed": [{"id": "2"}],
2780+
"hasNext": False,
2781+
}
2782+
2783+
with pytest.raises(StopAsyncIteration):
26162784
await anext(iterator)
26172785

2786+
async def finishes_a_later_shared_owner_without_repeating_delivered_data():
2787+
"""Finishes a later shared owner without repeating data
2788+
2789+
Finishes a later shared owner without repeating data delivered after
2790+
another owner failed.
2791+
"""
2792+
document = parse(
2793+
"""
2794+
{
2795+
hero {
2796+
... @defer(label: "R") {
2797+
nonNullName
2798+
id
2799+
}
2800+
... @defer(label: "P1") {
2801+
name
2802+
... @defer(label: "C1") {
2803+
id
2804+
}
2805+
}
2806+
... @defer(label: "P2") {
2807+
nestedObject { name }
2808+
... @defer(label: "C2") {
2809+
id
2810+
friends { name }
2811+
}
2812+
}
2813+
}
2814+
}
2815+
"""
2816+
)
2817+
name_future: Future[str] = Future()
2818+
object_future: Future[dict[str, str]] = Future()
2819+
id_calls = 0
2820+
2821+
def id_(_info):
2822+
nonlocal id_calls
2823+
id_calls += 1
2824+
return 1
2825+
2826+
result = experimental_execute_incrementally(
2827+
schema,
2828+
document,
2829+
{
2830+
"hero": {
2831+
"id": id_,
2832+
"name": name_future,
2833+
"nonNullName": None,
2834+
"nestedObject": object_future,
2835+
"friends": friends,
2836+
}
2837+
},
2838+
)
2839+
assert isinstance(result, ExperimentalIncrementalExecutionResults)
2840+
2841+
assert result.initial_result.formatted == {
2842+
"data": {"hero": {}},
2843+
"pending": [
2844+
{"id": "0", "path": ["hero"], "label": "R"},
2845+
{"id": "1", "path": ["hero"], "label": "P1"},
2846+
{"id": "2", "path": ["hero"], "label": "P2"},
2847+
],
2848+
"hasNext": True,
2849+
}
2850+
2851+
iterator = result.subsequent_results
2852+
2853+
assert (await anext(iterator)).formatted == {
2854+
"completed": [
2855+
{
2856+
"id": "0",
2857+
"errors": [
2858+
{
2859+
"message": "Cannot return null"
2860+
" for non-nullable field Hero.nonNullName.",
2861+
"locations": [{"line": 5, "column": 19}],
2862+
"path": ["hero", "nonNullName"],
2863+
},
2864+
],
2865+
},
2866+
],
2867+
"hasNext": True,
2868+
}
2869+
2870+
name_future.set_result("Luke")
2871+
2872+
assert (await anext(iterator)).formatted == {
2873+
"pending": [{"id": "3", "path": ["hero"], "label": "C1"}],
2874+
"incremental": [
2875+
{"data": {"name": "Luke"}, "id": "1"},
2876+
{"data": {"id": "1"}, "id": "3"},
2877+
],
2878+
"completed": [{"id": "1"}, {"id": "3"}],
2879+
"hasNext": True,
2880+
}
2881+
2882+
object_future.set_result({"name": "parent"})
2883+
2884+
assert (await anext(iterator)).formatted == {
2885+
"pending": [{"id": "4", "path": ["hero"], "label": "C2"}],
2886+
"incremental": [
2887+
{"data": {"nestedObject": {"name": "parent"}}, "id": "2"},
2888+
{
2889+
"data": {
2890+
"friends": [
2891+
{"name": "Han"},
2892+
{"name": "Leia"},
2893+
{"name": "C-3PO"},
2894+
]
2895+
},
2896+
"id": "4",
2897+
},
2898+
],
2899+
"completed": [{"id": "2"}, {"id": "4"}],
2900+
"hasNext": False,
2901+
}
2902+
2903+
with pytest.raises(StopAsyncIteration):
2904+
await anext(iterator)
2905+
2906+
assert id_calls == 1
2907+
26182908
async def handles_cancelling_child_deferred_fragments_if_parent_fragment_fails():
26192909
"""Handles cancelling child deferred fragments if parent fragment fails"""
26202910
document = parse(

0 commit comments

Comments
 (0)