Skip to content

Commit

Permalink
fix(race): concurrent next calls with defer/stream (#2975)
Browse files Browse the repository at this point in the history
* fix(race): concurrent next calls

* refactor test

* use invariant

* disable eslint error

* fix
  • Loading branch information
robrichard committed Aug 15, 2022
1 parent 33e23b3 commit 37b7867
Show file tree
Hide file tree
Showing 2 changed files with 83 additions and 0 deletions.
74 changes: 74 additions & 0 deletions src/execution/__tests__/stream-test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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) }');
Expand Down Expand Up @@ -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 {
Expand Down
9 changes: 9 additions & 0 deletions src/execution/execute.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 = {};
Expand Down

0 comments on commit 37b7867

Please sign in to comment.