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

@@ -190,6 +190,22 @@ async function complete(
   return result;
 }
 
+async function completeAsync(document: DocumentNode, numCalls: number) {
+  const schema = new GraphQLSchema({ query, enableDeferStream: true });
+
+  const result = await execute({ schema, document, rootValue: {} });
+
+  invariant(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('Should ignore @stream if not enabled', async () => {
     const document = parse('{ scalarList @stream(initialCount: 1) }');
@@ -597,6 +613,58 @@ 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: {
+          data: {
+            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

@@ -1652,8 +1652,20 @@ export class Dispatcher {
 
           resolved = true;
 
+          if (this._subsequentPayloads.length === 0) {
+            // a different call to next has exhausted all payloads
+            resolve({ value: undefined, done: true });
+            return;
+          }
+
           const index = this._subsequentPayloads.indexOf(promise);
 
+          if (index === -1) {
+            // a different call to next has consumed this payload
+            resolve(this._race());
+            return;
+          }
+
           this._subsequentPayloads.splice(index, 1);
 
           const { value, done } = payload;