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

* fix(race): concurrent next calls * refactor test * use invariant * disable eslint error * fix
graphql · Mar 7, 2022 · 562d787 · 562d787
1 parent 5a794b8
commit 562d787
Show file tree

Hide file tree

Showing 2 changed files with 79 additions and 0 deletions.
diff --git a/src/execution/__tests__/stream-test.ts b/src/execution/__tests__/stream-test.ts
@@ -193,6 +193,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: {} });
+
+  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('Can stream a list field', async () => {
     const document = parse('{ scalarList @stream(initialCount: 1) }');
@@ -613,6 +629,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 { 

diff --git a/src/execution/execute.ts b/src/execution/execute.ts
@@ -1596,7 +1596,18 @@ function yieldSubsequentPayloads(
 
     const data = await asyncPayloadRecord.data;
 
+    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);
 
     if (asyncPayloadRecord.isCompletedIterator) {