diff --git a/src/execution/IncrementalGraph.ts b/src/execution/IncrementalGraph.ts index df41505204..cf5e95c285 100644 --- a/src/execution/IncrementalGraph.ts +++ b/src/execution/IncrementalGraph.ts @@ -1,3 +1,4 @@ +import { BoxedPromiseOrValue } from '../jsutils/BoxedPromiseOrValue.js'; import { isPromise } from '../jsutils/isPromise.js'; import { promiseWithResolvers } from '../jsutils/promiseWithResolvers.js'; @@ -120,7 +121,12 @@ export class IncrementalGraph { incrementalDataRecord.streamItemQueue, ); } else { - const result = incrementalDataRecord.result.value; + const deferredGroupedFieldSetResult = incrementalDataRecord.result; + const result = + deferredGroupedFieldSetResult instanceof BoxedPromiseOrValue + ? deferredGroupedFieldSetResult.value + : deferredGroupedFieldSetResult().value; + if (isPromise(result)) { // eslint-disable-next-line @typescript-eslint/no-floating-promises result.then((resolved) => this._enqueue(resolved)); @@ -299,7 +305,10 @@ export class IncrementalGraph { let incrementalDataRecords: Array = []; let streamItemRecord: StreamItemRecord | undefined; while ((streamItemRecord = streamItemQueue.shift()) !== undefined) { - let result = streamItemRecord.value; + let result = + streamItemRecord instanceof BoxedPromiseOrValue + ? streamItemRecord.value + : streamItemRecord().value; if (isPromise(result)) { if (items.length > 0) { this._enqueue({ diff --git a/src/execution/__tests__/defer-test.ts b/src/execution/__tests__/defer-test.ts index 537f875d37..e74aebb9ae 100644 --- a/src/execution/__tests__/defer-test.ts +++ b/src/execution/__tests__/defer-test.ts @@ -135,11 +135,16 @@ const query = new GraphQLObjectType({ const schema = new GraphQLSchema({ query }); -async function complete(document: DocumentNode, rootValue: unknown = { hero }) { +async function complete( + document: DocumentNode, + rootValue: unknown = { hero }, + enableEarlyExecution = false, +) { const result = await experimentalExecuteIncrementally({ schema, document, rootValue, + enableEarlyExecution, }); if ('initialResult' in result) { @@ -247,6 +252,118 @@ describe('Execute: defer directive', () => { }, ]); }); + it('Does not execute deferred fragments early when not specified', async () => { + const document = parse(` + query HeroNameQuery { + hero { + id + ...NameFragment @defer + } + } + fragment NameFragment on Hero { + name + } + `); + const order: Array = []; + const result = await complete(document, { + hero: { + ...hero, + id: async () => { + await resolveOnNextTick(); + await resolveOnNextTick(); + order.push('slow-id'); + return hero.id; + }, + name: () => { + order.push('fast-name'); + return hero.name; + }, + }, + }); + + expectJSON(result).toDeepEqual([ + { + data: { + hero: { + id: '1', + }, + }, + pending: [{ id: '0', path: ['hero'] }], + hasNext: true, + }, + { + incremental: [ + { + data: { + name: 'Luke', + }, + id: '0', + }, + ], + completed: [{ id: '0' }], + hasNext: false, + }, + ]); + expect(order).to.deep.equal(['slow-id', 'fast-name']); + }); + it('Does execute deferred fragments early when specified', async () => { + const document = parse(` + query HeroNameQuery { + hero { + id + ...NameFragment @defer + } + } + fragment NameFragment on Hero { + name + } + `); + const order: Array = []; + const result = await complete( + document, + { + hero: { + ...hero, + id: async () => { + await resolveOnNextTick(); + await resolveOnNextTick(); + order.push('slow-id'); + return hero.id; + }, + name: () => { + order.push('fast-name'); + return hero.name; + }, + }, + }, + true, + ); + + expectJSON(result).toDeepEqual([ + { + data: { + hero: { + id: '1', + }, + }, + pending: [{ id: '0', path: ['hero'] }], + hasNext: true, + }, + { + incremental: [ + { + data: { + name: 'Luke', + }, + id: '0', + }, + ], + completed: [{ id: '0' }], + hasNext: false, + }, + ]); + expect(order).to.deep.equal(['fast-name', 'slow-id']); + }); it('Can defer fragments on the top level Query field', async () => { const document = parse(` query HeroNameQuery { @@ -1492,20 +1609,24 @@ describe('Execute: defer directive', () => { } } `); - const result = await complete(document, { - a: { - b: { - c: { - d: 'd', - nonNullErrorField: async () => { - await resolveOnNextTick(); - return null; + const result = await complete( + document, + { + a: { + b: { + c: { + d: 'd', + nonNullErrorField: async () => { + await resolveOnNextTick(); + return null; + }, }, }, + someField: 'someField', }, - someField: 'someField', }, - }); + true, + ); expectJSON(result).toDeepEqual([ { data: { @@ -1564,12 +1685,16 @@ describe('Execute: defer directive', () => { } } `); - const result = await complete(document, { - hero: { - ...hero, - nonNullName: () => null, + const result = await complete( + document, + { + hero: { + ...hero, + nonNullName: () => null, + }, }, - }); + true, + ); expectJSON(result).toDeepEqual({ data: { hero: null, @@ -1596,12 +1721,16 @@ describe('Execute: defer directive', () => { } } `); - const result = await complete(document, { - hero: { - ...hero, - nonNullName: () => null, + const result = await complete( + document, + { + hero: { + ...hero, + nonNullName: () => null, + }, }, - }); + true, + ); expectJSON(result).toDeepEqual([ { data: {}, diff --git a/src/execution/__tests__/stream-test.ts b/src/execution/__tests__/stream-test.ts index d3d8749d7e..15ad4028a5 100644 --- a/src/execution/__tests__/stream-test.ts +++ b/src/execution/__tests__/stream-test.ts @@ -84,11 +84,16 @@ const query = new GraphQLObjectType({ const schema = new GraphQLSchema({ query }); -async function complete(document: DocumentNode, rootValue: unknown = {}) { +async function complete( + document: DocumentNode, + rootValue: unknown = {}, + enableEarlyExecution = false, +) { const result = await experimentalExecuteIncrementally({ schema, document, rootValue, + enableEarlyExecution, }); if ('initialResult' in result) { @@ -354,11 +359,134 @@ describe('Execute: stream directive', () => { { incremental: [ { - items: [ - { name: 'Luke', id: '1' }, - { name: 'Han', id: '2' }, - { name: 'Leia', id: '3' }, - ], + items: [{ name: 'Luke', id: '1' }], + id: '0', + }, + ], + hasNext: true, + }, + { + incremental: [ + { + items: [{ name: 'Han', id: '2' }], + id: '0', + }, + ], + hasNext: true, + }, + { + incremental: [ + { + items: [{ name: 'Leia', id: '3' }], + id: '0', + }, + ], + completed: [{ id: '0' }], + hasNext: false, + }, + ]); + }); + it('Does not execute early if not specified', async () => { + const document = parse(` + query { + friendList @stream(initialCount: 0) { + id + } + } + `); + const order: Array = []; + const result = await complete(document, { + friendList: () => + friends.map((f, i) => ({ + id: async () => { + const slowness = 3 - i; + for (let j = 0; j < slowness; j++) { + // eslint-disable-next-line no-await-in-loop + await resolveOnNextTick(); + } + order.push(i); + return f.id; + }, + })), + }); + expectJSON(result).toDeepEqual([ + { + data: { + friendList: [], + }, + pending: [{ id: '0', path: ['friendList'] }], + hasNext: true, + }, + { + incremental: [ + { + items: [{ id: '1' }], + id: '0', + }, + ], + hasNext: true, + }, + { + incremental: [ + { + items: [{ id: '2' }], + id: '0', + }, + ], + hasNext: true, + }, + { + incremental: [ + { + items: [{ id: '3' }], + id: '0', + }, + ], + completed: [{ id: '0' }], + hasNext: false, + }, + ]); + expect(order).to.deep.equal([0, 1, 2]); + }); + it('Executes early if specified', async () => { + const document = parse(` + query { + friendList @stream(initialCount: 0) { + id + } + } + `); + const order: Array = []; + const result = await complete( + document, + { + friendList: () => + friends.map((f, i) => ({ + id: async () => { + const slowness = 3 - i; + for (let j = 0; j < slowness; j++) { + // eslint-disable-next-line no-await-in-loop + await resolveOnNextTick(); + } + order.push(i); + return f.id; + }, + })), + }, + true, + ); + expectJSON(result).toDeepEqual([ + { + data: { + friendList: [], + }, + pending: [{ id: '0', path: ['friendList'] }], + hasNext: true, + }, + { + incremental: [ + { + items: [{ id: '1' }, { id: '2' }, { id: '3' }], id: '0', }, ], @@ -366,6 +494,7 @@ describe('Execute: stream directive', () => { hasNext: false, }, ]); + expect(order).to.deep.equal([2, 1, 0]); }); it('Can stream a field that returns a list with nested promises', async () => { const document = parse(` @@ -491,7 +620,7 @@ describe('Execute: stream directive', () => { { incremental: [ { - items: [null, { name: 'Leia', id: '3' }], + items: [null], id: '0', errors: [ { @@ -502,6 +631,15 @@ describe('Execute: stream directive', () => { ], }, ], + hasNext: true, + }, + { + incremental: [ + { + items: [{ name: 'Leia', id: '3' }], + id: '0', + }, + ], completed: [{ id: '0' }], hasNext: false, }, @@ -556,6 +694,9 @@ describe('Execute: stream directive', () => { id: '0', }, ], + hasNext: true, + }, + { completed: [{ id: '0' }], hasNext: false, }, @@ -595,6 +736,9 @@ describe('Execute: stream directive', () => { id: '0', }, ], + hasNext: true, + }, + { completed: [{ id: '0' }], hasNext: false, }, @@ -626,6 +770,128 @@ describe('Execute: stream directive', () => { }, }); }); + it('Does not execute early if not specified, when streaming from an async iterable', async () => { + const document = parse(` + query { + friendList @stream(initialCount: 0) { + id + } + } + `); + const order: Array = []; + // eslint-disable-next-line @typescript-eslint/require-await + const slowFriend = async (n: number) => ({ + id: async () => { + const slowness = (3 - n) * 10; + for (let j = 0; j < slowness; j++) { + // eslint-disable-next-line no-await-in-loop + await resolveOnNextTick(); + } + order.push(n); + return friends[n].id; + }, + }); + const result = await complete(document, { + async *friendList() { + yield await Promise.resolve(slowFriend(0)); + yield await Promise.resolve(slowFriend(1)); + yield await Promise.resolve(slowFriend(2)); + }, + }); + expectJSON(result).toDeepEqual([ + { + data: { + friendList: [], + }, + pending: [{ id: '0', path: ['friendList'] }], + hasNext: true, + }, + { + incremental: [ + { + items: [{ id: '1' }], + id: '0', + }, + ], + hasNext: true, + }, + { + incremental: [ + { + items: [{ id: '2' }], + id: '0', + }, + ], + hasNext: true, + }, + { + incremental: [ + { + items: [{ id: '3' }], + id: '0', + }, + ], + hasNext: true, + }, + { + completed: [{ id: '0' }], + hasNext: false, + }, + ]); + expect(order).to.deep.equal([0, 1, 2]); + }); + it('Executes early if specified when streaming from an async iterable', async () => { + const document = parse(` + query { + friendList @stream(initialCount: 0) { + id + } + } + `); + const order: Array = []; + const slowFriend = (n: number) => ({ + id: async () => { + const slowness = (3 - n) * 10; + for (let j = 0; j < slowness; j++) { + // eslint-disable-next-line no-await-in-loop + await resolveOnNextTick(); + } + order.push(n); + return friends[n].id; + }, + }); + const result = await complete( + document, + { + async *friendList() { + yield await Promise.resolve(slowFriend(0)); + yield await Promise.resolve(slowFriend(1)); + yield await Promise.resolve(slowFriend(2)); + }, + }, + true, + ); + expectJSON(result).toDeepEqual([ + { + data: { + friendList: [], + }, + pending: [{ id: '0', path: ['friendList'] }], + hasNext: true, + }, + { + incremental: [ + { + items: [{ id: '1' }, { id: '2' }, { id: '3' }], + id: '0', + }, + ], + completed: [{ id: '0' }], + hasNext: false, + }, + ]); + expect(order).to.deep.equal([2, 1, 0]); + }); it('Can handle concurrent calls to .next() without waiting', async () => { const document = parse(` query { @@ -895,7 +1161,7 @@ describe('Execute: stream directive', () => { { incremental: [ { - items: [null, { nonNullName: 'Han' }], + items: [null], id: '0', errors: [ { @@ -906,6 +1172,15 @@ describe('Execute: stream directive', () => { ], }, ], + hasNext: true, + }, + { + incremental: [ + { + items: [{ nonNullName: 'Han' }], + id: '0', + }, + ], completed: [{ id: '0' }], hasNext: false, }, @@ -937,7 +1212,7 @@ describe('Execute: stream directive', () => { { incremental: [ { - items: [null, { nonNullName: 'Han' }], + items: [null], id: '0', errors: [ { @@ -948,6 +1223,15 @@ describe('Execute: stream directive', () => { ], }, ], + hasNext: true, + }, + { + incremental: [ + { + items: [{ nonNullName: 'Han' }], + id: '0', + }, + ], completed: [{ id: '0' }], hasNext: false, }, @@ -1063,7 +1347,7 @@ describe('Execute: stream directive', () => { { incremental: [ { - items: [null, { nonNullName: 'Han' }], + items: [null], id: '0', errors: [ { @@ -1074,6 +1358,18 @@ describe('Execute: stream directive', () => { ], }, ], + hasNext: true, + }, + { + incremental: [ + { + items: [{ nonNullName: 'Han' }], + id: '0', + }, + ], + hasNext: true, + }, + { completed: [{ id: '0' }], hasNext: false, }, @@ -1092,9 +1388,6 @@ describe('Execute: stream directive', () => { yield await Promise.resolve({ nonNullName: friends[0].name }); yield await Promise.resolve({ nonNullName: () => Promise.reject(new Error('Oops')), - }); - yield await Promise.resolve({ - nonNullName: friends[1].name, }); /* c8 ignore start */ } /* c8 ignore stop */, }); @@ -1149,17 +1442,12 @@ describe('Execute: stream directive', () => { nonNullName: () => Promise.reject(new Error('Oops')), }, }); - case 2: - return Promise.resolve({ - done: false, - value: { nonNullName: friends[1].name }, - }); // Not reached /* c8 ignore next 5 */ - case 3: + case 2: return Promise.resolve({ done: false, - value: { nonNullName: friends[2].name }, + value: { nonNullName: friends[1].name }, }); } }, @@ -1222,17 +1510,12 @@ describe('Execute: stream directive', () => { nonNullName: () => Promise.reject(new Error('Oops')), }, }); - case 2: - return Promise.resolve({ - done: false, - value: { nonNullName: friends[1].name }, - }); // Not reached /* c8 ignore next 5 */ - case 3: + case 2: return Promise.resolve({ done: false, - value: { nonNullName: friends[2].name }, + value: { nonNullName: friends[1].name }, }); } }, @@ -1373,6 +1656,10 @@ describe('Execute: stream directive', () => { }, { incremental: [ + { + items: [{ name: 'Luke' }], + id: '1', + }, { data: { scalarField: null }, id: '0', @@ -1384,12 +1671,12 @@ describe('Execute: stream directive', () => { }, ], }, - { - items: [{ name: 'Luke' }], - id: '1', - }, ], - completed: [{ id: '0' }, { id: '1' }], + completed: [{ id: '0' }], + hasNext: true, + }, + { + completed: [{ id: '1' }], hasNext: false, }, ]); @@ -1495,6 +1782,9 @@ describe('Execute: stream directive', () => { ], }, ], + hasNext: true, + }, + { completed: [{ id: '0' }], hasNext: false, }, @@ -1556,6 +1846,7 @@ describe('Execute: stream directive', () => { }, }, }, + enableEarlyExecution: true, }); assert('initialResult' in executeResult); const iterator = executeResult.subsequentResults[Symbol.asyncIterator](); @@ -1646,6 +1937,9 @@ describe('Execute: stream directive', () => { id: '0', }, ], + hasNext: true, + }, + { completed: [{ id: '0' }], hasNext: false, }, @@ -1690,13 +1984,22 @@ describe('Execute: stream directive', () => { { incremental: [ { - items: [ - { id: '1', name: 'Luke' }, - { id: '2', name: 'Han' }, - ], + items: [{ id: '1', name: 'Luke' }], id: '0', }, ], + hasNext: true, + }, + { + incremental: [ + { + items: [{ id: '2', name: 'Han' }], + id: '0', + }, + ], + hasNext: true, + }, + { completed: [{ id: '0' }], hasNext: false, }, @@ -1754,18 +2057,51 @@ describe('Execute: stream directive', () => { data: { scalarField: 'slow', nestedFriendList: [] }, id: '0', }, + ], + completed: [{ id: '0' }], + hasNext: true, + }, + done: false, + }); + + const result3 = await iterator.next(); + expectJSON(result3).toDeepEqual({ + value: { + incremental: [ + { + items: [{ name: 'Luke' }], + id: '1', + }, + ], + hasNext: true, + }, + done: false, + }); + + const result4 = await iterator.next(); + expectJSON(result4).toDeepEqual({ + value: { + incremental: [ { - items: [{ name: 'Luke' }, { name: 'Han' }], + items: [{ name: 'Han' }], id: '1', }, ], - completed: [{ id: '0' }, { id: '1' }], + hasNext: true, + }, + done: false, + }); + + const result5 = await iterator.next(); + expectJSON(result5).toDeepEqual({ + value: { + completed: [{ id: '1' }], hasNext: false, }, done: false, }); - const result3 = await iterator.next(); - expectJSON(result3).toDeepEqual({ + const result6 = await iterator.next(); + expectJSON(result6).toDeepEqual({ value: undefined, done: true, }); @@ -1824,16 +2160,11 @@ describe('Execute: stream directive', () => { const result2 = await result2Promise; expectJSON(result2).toDeepEqual({ value: { - pending: [{ id: '2', path: ['friendList', 1], label: 'DeferName' }], incremental: [ { data: { name: 'Luke' }, id: '0', }, - { - items: [{ id: '2' }], - id: '1', - }, ], completed: [{ id: '0' }], hasNext: true, @@ -1846,13 +2177,27 @@ describe('Execute: stream directive', () => { const result3 = await result3Promise; expectJSON(result3).toDeepEqual({ value: { - completed: [{ id: '1' }], + pending: [{ id: '2', path: ['friendList', 1], label: 'DeferName' }], + incremental: [ + { + items: [{ id: '2' }], + id: '1', + }, + ], hasNext: true, }, done: false, }); const result4 = await iterator.next(); expectJSON(result4).toDeepEqual({ + value: { + completed: [{ id: '1' }], + hasNext: true, + }, + done: false, + }); + const result5 = await iterator.next(); + expectJSON(result5).toDeepEqual({ value: { incremental: [ { @@ -1865,8 +2210,8 @@ describe('Execute: stream directive', () => { }, done: false, }); - const result5 = await iterator.next(); - expectJSON(result5).toDeepEqual({ + const result6 = await iterator.next(); + expectJSON(result6).toDeepEqual({ value: undefined, done: true, }); @@ -1925,25 +2270,35 @@ describe('Execute: stream directive', () => { const result2 = await result2Promise; expectJSON(result2).toDeepEqual({ value: { - pending: [{ id: '2', path: ['friendList', 1], label: 'DeferName' }], incremental: [ { data: { name: 'Luke' }, id: '0', }, + ], + completed: [{ id: '0' }], + hasNext: true, + }, + done: false, + }); + + const result3 = await iterator.next(); + expectJSON(result3).toDeepEqual({ + value: { + pending: [{ id: '2', path: ['friendList', 1], label: 'DeferName' }], + incremental: [ { items: [{ id: '2' }], id: '1', }, ], - completed: [{ id: '0' }], hasNext: true, }, done: false, }); - const result3 = await iterator.next(); - expectJSON(result3).toDeepEqual({ + const result4 = await iterator.next(); + expectJSON(result4).toDeepEqual({ value: { incremental: [ { @@ -1957,10 +2312,10 @@ describe('Execute: stream directive', () => { done: false, }); - const result4Promise = iterator.next(); + const result5Promise = iterator.next(); resolveIterableCompletion(null); - const result4 = await result4Promise; - expectJSON(result4).toDeepEqual({ + const result5 = await result5Promise; + expectJSON(result5).toDeepEqual({ value: { completed: [{ id: '1' }], hasNext: false, @@ -1968,8 +2323,8 @@ describe('Execute: stream directive', () => { done: false, }); - const result5 = await iterator.next(); - expectJSON(result5).toDeepEqual({ + const result6 = await iterator.next(); + expectJSON(result6).toDeepEqual({ value: undefined, done: true, }); diff --git a/src/execution/execute.ts b/src/execution/execute.ts index 7b87d25060..d01b6ee768 100644 --- a/src/execution/execute.ts +++ b/src/execution/execute.ts @@ -140,6 +140,7 @@ export interface ExecutionContext { fieldResolver: GraphQLFieldResolver; typeResolver: GraphQLTypeResolver; subscribeFieldResolver: GraphQLFieldResolver; + enableEarlyExecution: boolean; errors: Array | undefined; cancellableStreams: Set | undefined; } @@ -159,6 +160,7 @@ export interface ExecutionArgs { fieldResolver?: Maybe>; typeResolver?: Maybe>; subscribeFieldResolver?: Maybe>; + enableEarlyExecution?: Maybe; } export interface StreamUsage { @@ -437,6 +439,7 @@ export function buildExecutionContext( fieldResolver, typeResolver, subscribeFieldResolver, + enableEarlyExecution, } = args; // If the schema used for execution is invalid, throw an error. @@ -500,6 +503,7 @@ export function buildExecutionContext( fieldResolver: fieldResolver ?? defaultFieldResolver, typeResolver: typeResolver ?? defaultTypeResolver, subscribeFieldResolver: subscribeFieldResolver ?? defaultFieldResolver, + enableEarlyExecution: enableEarlyExecution === true, errors: undefined, cancellableStreams: undefined, }; @@ -2110,12 +2114,17 @@ function executeDeferredGroupedFieldSets( deferMap, ); - deferredGroupedFieldSetRecord.result = new BoxedPromiseOrValue( - shouldDefer(parentDeferUsages, deferUsageSet) - ? Promise.resolve().then(executor) - : executor(), + const shouldDeferThisDeferUsageSet = shouldDefer( + parentDeferUsages, + deferUsageSet, ); + deferredGroupedFieldSetRecord.result = shouldDeferThisDeferUsageSet + ? exeContext.enableEarlyExecution + ? new BoxedPromiseOrValue(Promise.resolve().then(executor)) + : () => new BoxedPromiseOrValue(executor()) + : new BoxedPromiseOrValue(executor()); + newDeferredGroupedFieldSetRecords.push(deferredGroupedFieldSetRecord); } @@ -2226,57 +2235,74 @@ function buildSyncStreamItemQueue( info: GraphQLResolveInfo, itemType: GraphQLOutputType, ): Array { - const streamItemQueue: Array = [ - new BoxedPromiseOrValue( - Promise.resolve().then(() => { - const initialPath = addPath(streamPath, initialIndex, undefined); - const firstStreamItem = new BoxedPromiseOrValue( - completeStreamItem( - initialPath, - initialItem, - exeContext, - { errors: undefined }, - fieldGroup, - info, - itemType, - ), - ); - let iteration = iterator.next(); - let currentIndex = initialIndex + 1; - let currentStreamItem = firstStreamItem; - while (!iteration.done) { - // TODO: add test case for early sync termination - /* c8 ignore next 4 */ - const result = currentStreamItem.value; - if (!isPromise(result) && result.errors !== undefined) { - break; - } + const streamItemQueue: Array = []; - const itemPath = addPath(streamPath, currentIndex, undefined); + const enableEarlyExecution = exeContext.enableEarlyExecution; - currentStreamItem = new BoxedPromiseOrValue( - completeStreamItem( - itemPath, - iteration.value, - exeContext, - { errors: undefined }, - fieldGroup, - info, - itemType, - ), - ); - streamItemQueue.push(currentStreamItem); + const firstExecutor = () => { + const initialPath = addPath(streamPath, initialIndex, undefined); + const firstStreamItem = new BoxedPromiseOrValue( + completeStreamItem( + initialPath, + initialItem, + exeContext, + { errors: undefined }, + fieldGroup, + info, + itemType, + ), + ); - iteration = iterator.next(); - currentIndex = initialIndex + 1; + let iteration = iterator.next(); + let currentIndex = initialIndex + 1; + let currentStreamItem: + | BoxedPromiseOrValue + | (() => BoxedPromiseOrValue) = firstStreamItem; + while (!iteration.done) { + // TODO: add test case for early sync termination + /* c8 ignore next 6 */ + if (currentStreamItem instanceof BoxedPromiseOrValue) { + const result = currentStreamItem.value; + if (!isPromise(result) && result.errors !== undefined) { + break; } + } - streamItemQueue.push(new BoxedPromiseOrValue({})); + const itemPath = addPath(streamPath, currentIndex, undefined); - return firstStreamItem.value; - }), - ), - ]; + const value = iteration.value; + + const currentExecutor = () => + completeStreamItem( + itemPath, + value, + exeContext, + { errors: undefined }, + fieldGroup, + info, + itemType, + ); + + currentStreamItem = enableEarlyExecution + ? new BoxedPromiseOrValue(currentExecutor()) + : () => new BoxedPromiseOrValue(currentExecutor()); + + streamItemQueue.push(currentStreamItem); + + iteration = iterator.next(); + currentIndex = initialIndex + 1; + } + + streamItemQueue.push(new BoxedPromiseOrValue({})); + + return firstStreamItem.value; + }; + + streamItemQueue.push( + enableEarlyExecution + ? new BoxedPromiseOrValue(Promise.resolve().then(firstExecutor)) + : () => new BoxedPromiseOrValue(firstExecutor()), + ); return streamItemQueue; } @@ -2291,20 +2317,24 @@ function buildAsyncStreamItemQueue( itemType: GraphQLOutputType, ): Array { const streamItemQueue: Array = []; + const executor = () => + getNextAsyncStreamItemResult( + streamItemQueue, + streamPath, + initialIndex, + asyncIterator, + exeContext, + fieldGroup, + info, + itemType, + ); + streamItemQueue.push( - new BoxedPromiseOrValue( - getNextAsyncStreamItemResult( - streamItemQueue, - streamPath, - initialIndex, - asyncIterator, - exeContext, - fieldGroup, - info, - itemType, - ), - ), + exeContext.enableEarlyExecution + ? new BoxedPromiseOrValue(executor()) + : () => new BoxedPromiseOrValue(executor()), ); + return streamItemQueue; } @@ -2345,19 +2375,22 @@ async function getNextAsyncStreamItemResult( itemType, ); + const executor = () => + getNextAsyncStreamItemResult( + streamItemQueue, + streamPath, + index, + asyncIterator, + exeContext, + fieldGroup, + info, + itemType, + ); + streamItemQueue.push( - new BoxedPromiseOrValue( - getNextAsyncStreamItemResult( - streamItemQueue, - streamPath, - index, - asyncIterator, - exeContext, - fieldGroup, - info, - itemType, - ), - ), + exeContext.enableEarlyExecution + ? new BoxedPromiseOrValue(executor()) + : () => new BoxedPromiseOrValue(executor()), ); return result; diff --git a/src/execution/types.ts b/src/execution/types.ts index 9340ab1b85..50f9a083f8 100644 --- a/src/execution/types.ts +++ b/src/execution/types.ts @@ -203,9 +203,13 @@ export function isNonReconcilableDeferredGroupedFieldSetResult( return deferredGroupedFieldSetResult.errors !== undefined; } +type ThunkIncrementalResult = + | BoxedPromiseOrValue + | (() => BoxedPromiseOrValue); + export interface DeferredGroupedFieldSetRecord { deferredFragmentRecords: ReadonlyArray; - result: BoxedPromiseOrValue; + result: ThunkIncrementalResult; } export type SubsequentResultRecord = DeferredFragmentRecord | StreamRecord; @@ -223,7 +227,7 @@ export interface StreamItemResult { errors?: ReadonlyArray | undefined; } -export type StreamItemRecord = BoxedPromiseOrValue; +export type StreamItemRecord = ThunkIncrementalResult; export interface StreamRecord { path: Path;