fix(race): concurrent next calls with defer/stream (#2975)

* fix(race): concurrent next calls

* refactor test

* use invariant

* disable eslint error

* fix

Author: robrichard

src/execution/__tests__/stream-test.ts

@@ -244,6 +244,22 @@ async function complete(document: DocumentNode, rootValue: unknown = {}) {
   return result;
 }
 
+async function completeAsync(document: DocumentNode, numCalls: number) {
+  const schema = new GraphQLSchema({ query });
+
+  const result = await execute({ schema, document, rootValue: {} });
+
+  assert(isAsyncIterable(result));
+
+  const iterator = result[Symbol.asyncIterator]();
+
+  const promises = [];
+  for (let i = 0; i < numCalls; i++) {
+    promises.push(iterator.next());
+  }
+  return Promise.all(promises);
+}
+
 describe('Execute: stream directive', () => {
   it('Can stream a list field', async () => {
     const document = parse('{ scalarList @stream(initialCount: 1) }');
@@ -739,6 +755,64 @@ describe('Execute: stream directive', () => {
       },
     });
   });
+  it('Can handle concurrent calls to .next() without waiting', async () => {
+    const document = parse(`
+      query { 
+        asyncIterableList @stream(initialCount: 2) {
+          name
+          id
+        }
+      }
+    `);
+    const result = await completeAsync(document, 4);
+    expectJSON(result).toDeepEqual([
+      {
+        done: false,
+        value: {
+          data: {
+            asyncIterableList: [
+              {
+                name: 'Luke',
+                id: '1',
+              },
+              {
+                name: 'Han',
+                id: '2',
+              },
+            ],
+          },
+          hasNext: true,
+        },
+      },
+      {
+        done: false,
+        value: {
+          incremental: [
+            {
+              items: [
+                {
+                  name: 'Leia',
+                  id: '3',
+                },
+              ],
+              path: ['asyncIterableList', 2],
+            },
+          ],
+          hasNext: true,
+        },
+      },
+      {
+        done: false,
+        value: {
+          hasNext: false,
+        },
+      },
+      {
+        done: true,
+        value: undefined,
+      },
+    ]);
+  });
   it('Handles error thrown in async iterable before initialCount is reached', async () => {
     const document = parse(`
       query { 

src/execution/execute.ts

@@ -1821,7 +1821,16 @@ function yieldSubsequentPayloads(
       },
     );
 
+    if (exeContext.subsequentPayloads.length === 0) {
+      // a different call to next has exhausted all payloads
+      return { value: undefined, done: true };
+    }
     const index = exeContext.subsequentPayloads.indexOf(asyncPayloadRecord);
+    if (index === -1) {
+      // a different call to next has consumed this payload
+      return race();
+    }
+
     exeContext.subsequentPayloads.splice(index, 1);
 
     const incrementalResult: IncrementalResult = {};