Reference implementation of defer and stream spec (#3659)

Co-authored-by: Liliana Matos <liliana@1stdibs.com>
Co-authored-by: David Glasser <glasser@davidglasser.net>

src/execution/__tests__/flattenAsyncIterable-test.ts

@@ -0,0 +1,149 @@
+import { expect } from 'chai';
+import { describe, it } from 'mocha';
+
+import { flattenAsyncIterable } from '../flattenAsyncIterable';
+
+describe('flattenAsyncIterable', () => {
+  it('flatten nested async generators', async () => {
+    async function* source() {
+      yield await Promise.resolve(
+        (async function* nested(): AsyncGenerator<number, void, void> {
+          yield await Promise.resolve(1.1);
+          yield await Promise.resolve(1.2);
+        })(),
+      );
+      yield await Promise.resolve(
+        (async function* nested(): AsyncGenerator<number, void, void> {
+          yield await Promise.resolve(2.1);
+          yield await Promise.resolve(2.2);
+        })(),
+      );
+    }
+
+    const doubles = flattenAsyncIterable(source());
+
+    const result = [];
+    for await (const x of doubles) {
+      result.push(x);
+    }
+    expect(result).to.deep.equal([1.1, 1.2, 2.1, 2.2]);
+  });
+
+  it('allows returning early from a nested async generator', async () => {
+    async function* source() {
+      yield await Promise.resolve(
+        (async function* nested(): AsyncGenerator<number, void, void> {
+          yield await Promise.resolve(1.1);
+          yield await Promise.resolve(1.2);
+        })(),
+      );
+      yield await Promise.resolve(
+        (async function* nested(): AsyncGenerator<number, void, void> {
+          yield await Promise.resolve(2.1); /* c8 ignore start */
+          // Not reachable, early return
+          yield await Promise.resolve(2.2);
+        })(),
+      );
+      // Not reachable, early return
+      yield await Promise.resolve(
+        (async function* nested(): AsyncGenerator<number, void, void> {
+          yield await Promise.resolve(3.1);
+          yield await Promise.resolve(3.2);
+        })(),
+      );
+    }
+    /* c8 ignore stop */
+
+    const doubles = flattenAsyncIterable(source());
+
+    expect(await doubles.next()).to.deep.equal({ value: 1.1, done: false });
+    expect(await doubles.next()).to.deep.equal({ value: 1.2, done: false });
+    expect(await doubles.next()).to.deep.equal({ value: 2.1, done: false });
+
+    // Early return
+    expect(await doubles.return()).to.deep.equal({
+      value: undefined,
+      done: true,
+    });
+
+    // Subsequent next calls
+    expect(await doubles.next()).to.deep.equal({
+      value: undefined,
+      done: true,
+    });
+    expect(await doubles.next()).to.deep.equal({
+      value: undefined,
+      done: true,
+    });
+  });
+
+  it('allows throwing errors from a nested async generator', async () => {
+    async function* source() {
+      yield await Promise.resolve(
+        (async function* nested(): AsyncGenerator<number, void, void> {
+          yield await Promise.resolve(1.1);
+          yield await Promise.resolve(1.2);
+        })(),
+      );
+      yield await Promise.resolve(
+        (async function* nested(): AsyncGenerator<number, void, void> {
+          yield await Promise.resolve(2.1); /* c8 ignore start */
+          // Not reachable, early return
+          yield await Promise.resolve(2.2);
+        })(),
+      );
+      // Not reachable, early return
+      yield await Promise.resolve(
+        (async function* nested(): AsyncGenerator<number, void, void> {
+          yield await Promise.resolve(3.1);
+          yield await Promise.resolve(3.2);
+        })(),
+      );
+    }
+    /* c8 ignore stop */
+
+    const doubles = flattenAsyncIterable(source());
+
+    expect(await doubles.next()).to.deep.equal({ value: 1.1, done: false });
+    expect(await doubles.next()).to.deep.equal({ value: 1.2, done: false });
+    expect(await doubles.next()).to.deep.equal({ value: 2.1, done: false });
+
+    // Throw error
+    let caughtError;
+    try {
+      await doubles.throw('ouch'); /* c8 ignore start */
+    } catch (e) {
+      caughtError = e;
+    }
+    expect(caughtError).to.equal('ouch');
+  });
+  it('completely yields sub-iterables even when next() called in parallel', async () => {
+    async function* source() {
+      yield await Promise.resolve(
+        (async function* nested(): AsyncGenerator<number, void, void> {
+          yield await Promise.resolve(1.1);
+          yield await Promise.resolve(1.2);
+        })(),
+      );
+      yield await Promise.resolve(
+        (async function* nested(): AsyncGenerator<number, void, void> {
+          yield await Promise.resolve(2.1);
+          yield await Promise.resolve(2.2);
+        })(),
+      );
+    }
+
+    const result = flattenAsyncIterable(source());
+
+    const promise1 = result.next();
+    const promise2 = result.next();
+    expect(await promise1).to.deep.equal({ value: 1.1, done: false });
+    expect(await promise2).to.deep.equal({ value: 1.2, done: false });
+    expect(await result.next()).to.deep.equal({ value: 2.1, done: false });
+    expect(await result.next()).to.deep.equal({ value: 2.2, done: false });
+    expect(await result.next()).to.deep.equal({
+      value: undefined,
+      done: true,
+    });
+  });
+});

src/execution/__tests__/mutations-test.ts

@@ -1,4 +1,4 @@
-import { expect } from 'chai';
+import { assert, expect } from 'chai';
 import { describe, it } from 'mocha';
 
 import { expectJSON } from '../../__testUtils__/expectJSON';
@@ -10,7 +10,11 @@ import { GraphQLObjectType } from '../../type/definition';
 import { GraphQLInt } from '../../type/scalars';
 import { GraphQLSchema } from '../../type/schema';
 
-import { execute, executeSync } from '../execute';
+import {
+  execute,
+  executeSync,
+  experimentalExecuteIncrementally,
+} from '../execute';
 
 class NumberHolder {
   theNumber: number;
@@ -50,6 +54,13 @@ class Root {
 const numberHolderType = new GraphQLObjectType({
   fields: {
     theNumber: { type: GraphQLInt },
+    promiseToGetTheNumber: {
+      type: GraphQLInt,
+      resolve: async (root) => {
+        await new Promise((resolve) => setTimeout(resolve, 0));
+        return root.theNumber;
+      },
+    },
   },
   name: 'NumberHolder',
 });
@@ -191,4 +202,132 @@ describe('Execute: Handles mutation execution ordering', () => {
       ],
     });
   });
+  it('Mutation fields with @defer do not block next mutation', async () => {
+    const document = parse(`
+      mutation M {
+        first: promiseToChangeTheNumber(newNumber: 1) {
+          ...DeferFragment @defer(label: "defer-label")
+        },
+        second: immediatelyChangeTheNumber(newNumber: 2) {
+          theNumber
+        }
+      }
+      fragment DeferFragment on NumberHolder {
+        promiseToGetTheNumber
+      }
+    `);
+
+    const rootValue = new Root(6);
+    const mutationResult = await experimentalExecuteIncrementally({
+      schema,
+      document,
+      rootValue,
+    });
+    const patches = [];
+
+    assert('initialResult' in mutationResult);
+    patches.push(mutationResult.initialResult);
+    for await (const patch of mutationResult.subsequentResults) {
+      patches.push(patch);
+    }
+
+    expect(patches).to.deep.equal([
+      {
+        data: {
+          first: {},
+          second: { theNumber: 2 },
+        },
+        hasNext: true,
+      },
+      {
+        incremental: [
+          {
+            label: 'defer-label',
+            path: ['first'],
+            data: {
+              promiseToGetTheNumber: 2,
+            },
+          },
+        ],
+        hasNext: false,
+      },
+    ]);
+  });
+  it('Mutation inside of a fragment', async () => {
+    const document = parse(`
+      mutation M {
+        ...MutationFragment
+        second: immediatelyChangeTheNumber(newNumber: 2) {
+          theNumber
+        }
+      }
+      fragment MutationFragment on Mutation {
+        first: promiseToChangeTheNumber(newNumber: 1) {
+          theNumber
+        },
+      }
+    `);
+
+    const rootValue = new Root(6);
+    const mutationResult = await execute({ schema, document, rootValue });
+
+    expect(mutationResult).to.deep.equal({
+      data: {
+        first: { theNumber: 1 },
+        second: { theNumber: 2 },
+      },
+    });
+  });
+  it('Mutation with @defer is not executed serially', async () => {
+    const document = parse(`
+      mutation M {
+        ...MutationFragment @defer(label: "defer-label")
+        second: immediatelyChangeTheNumber(newNumber: 2) {
+          theNumber
+        }
+      }
+      fragment MutationFragment on Mutation {
+        first: promiseToChangeTheNumber(newNumber: 1) {
+          theNumber
+        },
+      }
+    `);
+
+    const rootValue = new Root(6);
+    const mutationResult = await experimentalExecuteIncrementally({
+      schema,
+      document,
+      rootValue,
+    });
+    const patches = [];
+
+    assert('initialResult' in mutationResult);
+    patches.push(mutationResult.initialResult);
+    for await (const patch of mutationResult.subsequentResults) {
+      patches.push(patch);
+    }
+
+    expect(patches).to.deep.equal([
+      {
+        data: {
+          second: { theNumber: 2 },
+        },
+        hasNext: true,
+      },
+      {
+        incremental: [
+          {
+            label: 'defer-label',
+            path: [],
+            data: {
+              first: {
+                theNumber: 1,
+              },
+            },
+          },
+        ],
+        hasNext: false,
+      },
+    ]);
+  });
 });

src/execution/__tests__/nonnull-test.ts

@@ -3,6 +3,8 @@ import { describe, it } from 'mocha';
 
 import { expectJSON } from '../../__testUtils__/expectJSON';
 
+import type { PromiseOrValue } from '../../jsutils/PromiseOrValue';
+
 import { parse } from '../../language/parser';
 
 import { GraphQLNonNull, GraphQLObjectType } from '../../type/definition';
@@ -109,7 +111,7 @@ const schema = buildSchema(`
 function executeQuery(
   query: string,
   rootValue: unknown,
-): ExecutionResult | Promise<ExecutionResult> {
+): PromiseOrValue<ExecutionResult> {
   return execute({ schema, document: parse(query), rootValue });
 }