Repository: graphql/graphql-js
Pull request #3725 (2022-09-04 16:41:52)

changelog: use top-level await to simplify code
Motivation: since we switched to ESM now we can use top-level await to simplify code

netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *compassionate-pike-271cb3* processing.


|  Name | Link |
|---------------------------------|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | 16745024a6e0e50875d7964da1f10fe36cbfae9f |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/compassionate-pike-271cb3/deploys/6314d551b02a910009a1a980 |
github-actions[bot]: Hi @IvanGoncharov, I'm @github-actions bot happy to help you with this PR 👋

<details>
<summary> Supported commands </summary>

Please post this commands in separate comments and only one per comment:
* `@github-actions run-benchmark` - Run benchmark comparing base and merge commits for this PR
* `@github-actions publish-pr-on-npm` - Build package from this PR and publish it on NPM
</details>


Patch:
diff --git a/resources/gen-changelog.ts b/resources/gen-changelog.ts
index 821f29b9..baae5d15 100644
--- a/resources/gen-changelog.ts
+++ b/resources/gen-changelog.ts
@@ -53,36 +53,23 @@ if (repoURLMatch?.groups == null) {
 }
 const { githubOrg, githubRepo } = repoURLMatch.groups;
 
-getChangeLog()
-  .then((changelog) => process.stdout.write(changelog))
-  .catch((error) => {
-    console.error(error);
-    process.exit(1);
-  });
+process.stdout.write(await genChangeLog());
 
-function getChangeLog(): Promise<string> {
+async function genChangeLog(): Promise<string> {
   const { version } = packageJSON;
 
   let tag: string | null = null;
   let commitsList = git().revList('--reverse', `v${version}..`);
-  if (commitsList === '') {
+  if (commitsList.length === 0) {
     const parentPackageJSON = git().catFile('blob', 'HEAD~1:package.json');
     const parentVersion = JSON.parse(parentPackageJSON).version;
     commitsList = git().revList('--reverse', `v${parentVersion}..HEAD~1`);
     tag = `v${version}`;
   }
 
+  const allPRs = await getPRsInfo(commitsList);
   const date = git().log('-1', '--format=%cd', '--date=short');
-  return getCommitsInfo(commitsList.split('\n'))
-    .then((commitsInfo) => getPRsInfo(commitsInfoToPRs(commitsInfo)))
-    .then((prsInfo) => genChangeLog(tag, date, prsInfo));
-}
 
-function genChangeLog(
-  tag: string | null,
-  date: string,
-  allPRs: ReadonlyArray<PRInfo>,
-): string {
   const byLabel: { [label: string]: Array<PRInfo> } = {};
   const committersByLogin: { [login: string]: AuthorInfo } = {};
 
@@ -181,9 +168,9 @@ interface CommitInfo {
   };
 }
 
-async function batchCommitInfo(
+async function batchCommitToPR(
   commits: ReadonlyArray<string>,
-): Promise<Array<CommitInfo>> {
+): Promise<ReadonlyArray<number>> {
   let commitsSubQuery = '';
   for (const oid of commits) {
     commitsSubQuery += `
@@ -212,11 +199,12 @@ async function batchCommitInfo(
     }
   `);
 
-  const commitsInfo = [];
+  const prNumbers = [];
   for (const oid of commits) {
-    commitsInfo.push(response.repository['commit_' + oid]);
+    const commitInfo: CommitInfo = response.repository['commit_' + oid];
+    prNumbers.push(commitInfoToPR(commitInfo));
   }
-  return commitsInfo;
+  return prNumbers;
 }
 
 interface AuthorInfo {
@@ -237,7 +225,9 @@ interface PRInfo {
   };
 }
 
-async function batchPRInfo(prNumbers: Array<number>): Promise<Array<PRInfo>> {
+async function batchPRInfo(
+  prNumbers: ReadonlyArray<number>,
+): Promise<Array<PRInfo>> {
   let prsSubQuery = '';
   for (const number of prNumbers) {
     prsSubQuery += `
@@ -276,56 +266,47 @@ async function batchPRInfo(prNumbers: Array<number>): Promise<Array<PRInfo>> {
   return prsInfo;
 }
 
-function commitsInfoToPRs(commits: ReadonlyArray<CommitInfo>): Array<number> {
-  const prNumbers = new Set<number>();
-  for (const commit of commits) {
-    const associatedPRs = commit.associatedPullRequests.nodes.filter(
-      (pr) => pr.repository.nameWithOwner === `${githubOrg}/${githubRepo}`,
-    );
-    if (associatedPRs.length === 0) {
-      const match = / \(#(?<prNumber>[0-9]+)\)$/m.exec(commit.message);
-      if (match?.groups?.prNumber != null) {
-        prNumbers.add(parseInt(match.groups.prNumber, 10));
-        continue;
-      }
-      throw new Error(
-        `Commit ${commit.oid} has no associated PR: ${commit.message}`,
-      );
-    }
-    if (associatedPRs.length > 1) {
-      throw new Error(
-        `Commit ${commit.oid} is associated with multiple PRs: ${commit.message}`,
-      );
+function commitInfoToPR(commit: CommitInfo): number {
+  const associatedPRs = commit.associatedPullRequests.nodes.filter(
+    (pr) => pr.repository.nameWithOwner === `${githubOrg}/${githubRepo}`,
+  );
+  if (associatedPRs.length === 0) {
+    const match = / \(#(?<prNumber>[0-9]+)\)$/m.exec(commit.message);
+    if (match?.groups?.prNumber != null) {
+      return parseInt(match.groups.prNumber, 10);
     }
-
-    prNumbers.add(associatedPRs[0].number);
+    throw new Error(
+      `Commit ${commit.oid} has no associated PR: ${commit.message}`,
+    );
+  }
+  if (associatedPRs.length > 1) {
+    throw new Error(
+      `Commit ${commit.oid} is associated with multiple PRs: ${commit.message}`,
+    );
   }
 
-  return [...prNumbers.values()];
+  return associatedPRs[0].number;
 }
 
 async function getPRsInfo(
-  prNumbers: ReadonlyArray<number>,
-): Promise<Array<PRInfo>> {
-  // Split pr into batches of 50 to prevent timeouts
-  const prInfoPromises = [];
-  for (let i = 0; i < prNumbers.length; i += 50) {
-    const batch = prNumbers.slice(i, i + 50);
-    prInfoPromises.push(batchPRInfo(batch));
-  }
+  commits: ReadonlyArray<string>,
+): Promise<ReadonlyArray<PRInfo>> {
+  let prNumbers = await splitBatches(commits, batchCommitToPR);
+  prNumbers = Array.from(new Set(prNumbers)); // Remove duplicates
 
-  return (await Promise.all(prInfoPromises)).flat();
+  return splitBatches(prNumbers, batchPRInfo);
 }
 
-async function getCommitsInfo(
-  commits: ReadonlyArray<string>,
-): Promise<Array<CommitInfo>> {
-  // Split commits into batches of 50 to prevent timeouts
-  const commitInfoPromises = [];
-  for (let i = 0; i < commits.length; i += 50) {
-    const batch = commits.slice(i, i + 50);
-    commitInfoPromises.push(batchCommitInfo(batch));
+// Split commits into batches of 50 to prevent timeouts
+async function splitBatches<I, R>(
+  array: ReadonlyArray<I>,
+  batchFn: (array: ReadonlyArray<I>) => Promise<ReadonlyArray<R>>,
+): Promise<ReadonlyArray<R>> {
+  const promises = [];
+  for (let i = 0; i < array.length; i += 50) {
+    const batchItems = array.slice(i, i + 50);
+    promises.push(batchFn(batchItems));
   }
 
-  return (await Promise.all(commitInfoPromises)).flat();
+  return (await Promise.all(promises)).flat();
 }
diff --git a/resources/utils.ts b/resources/utils.ts
index 701f63c2..d1ae8c0b 100644
--- a/resources/utils.ts
+++ b/resources/utils.ts
@@ -69,8 +69,9 @@ export function git(options?: GITOptions) {
     revParse(...args: ReadonlyArray<string>): string {
       return spawnOutput('git', ['rev-parse', ...cmdOptions, ...args], options);
     },
-    revList(...args: ReadonlyArray<string>): string {
-      return spawnOutput('git', ['rev-list', ...cmdOptions, ...args], options);
+    revList(...args: ReadonlyArray<string>): Array<string> {
+      const allArgs = ['rev-list', ...cmdOptions, ...args];
+      return spawnOutput('git', allArgs, options).split('\n');
     },
     catFile(...args: ReadonlyArray<string>): string {
       return spawnOutput('git', ['cat-file', ...cmdOptions, ...args], options);
diff --git a/src/execution/execute.ts b/src/execution/execute.ts
index ab2062c0..c1f025eb 100644
--- a/src/execution/execute.ts
+++ b/src/execution/execute.ts
@@ -151,19 +151,17 @@ export interface FormattedExecutionResult<
   extensions?: TExtensions;
 }
 
-export type ExperimentalExecuteIncrementallyResults<
+export interface ExperimentalIncrementalExecutionResults<
   TData = ObjMap<unknown>,
   TExtensions = ObjMap<unknown>,
-> =
-  | { singleResult: ExecutionResult<TData, TExtensions> }
-  | {
-      initialResult: InitialIncrementalExecutionResult<TData, TExtensions>;
-      subsequentResults: AsyncGenerator<
-        SubsequentIncrementalExecutionResult<TData, TExtensions>,
-        void,
-        void
-      >;
-    };
+> {
+  initialResult: InitialIncrementalExecutionResult<TData, TExtensions>;
+  subsequentResults: AsyncGenerator<
+    SubsequentIncrementalExecutionResult<TData, TExtensions>,
+    void,
+    void
+  >;
+}
 
 export interface InitialIncrementalExecutionResult<
   TData = ObjMap<unknown>,
@@ -287,19 +285,19 @@ const UNEXPECTED_MULTIPLE_PAYLOADS =
 export function execute(args: ExecutionArgs): PromiseOrValue<ExecutionResult> {
   const result = experimentalExecuteIncrementally(args);
   if (!isPromise(result)) {
-    if ('singleResult' in result) {
-      return result.singleResult;
+    if ('initialResult' in result) {
+      throw new Error(UNEXPECTED_MULTIPLE_PAYLOADS);
     }
-    throw new Error(UNEXPECTED_MULTIPLE_PAYLOADS);
+    return result;
   }
 
   return result.then((incrementalResult) => {
-    if ('singleResult' in incrementalResult) {
-      return incrementalResult.singleResult;
+    if ('initialResult' in incrementalResult) {
+      return {
+        errors: [new GraphQLError(UNEXPECTED_MULTIPLE_PAYLOADS)],
+      };
     }
-    return {
-      errors: [new GraphQLError(UNEXPECTED_MULTIPLE_PAYLOADS)],
-    };
+    return incrementalResult;
   });
 }
 
@@ -308,23 +306,23 @@ export function execute(args: ExecutionArgs): PromiseOrValue<ExecutionResult> {
  * including `@defer` and `@stream` as proposed in
  * https://github.com/graphql/graphql-spec/pull/742
  *
- * This function returns a Promise of an ExperimentalExecuteIncrementallyResults
- * object. This object either contains a single ExecutionResult as
- * `singleResult`, or an `initialResult` and a stream of `subsequentResults`.
+ * This function returns a Promise of an ExperimentalIncrementalExecutionResults
+ * object. This object either consists of a single ExecutionResult, or an
+ * object containing an `initialResult` and a stream of `subsequentResults`.
  *
  * If the arguments to this function do not result in a legal execution context,
  * a GraphQLError will be thrown immediately explaining the invalid input.
  */
 export function experimentalExecuteIncrementally(
   args: ExecutionArgs,
-): PromiseOrValue<ExperimentalExecuteIncrementallyResults> {
+): PromiseOrValue<ExecutionResult | ExperimentalIncrementalExecutionResults> {
   // If a valid execution context cannot be created due to incorrect arguments,
   // a "Response" with only errors is returned.
   const exeContext = buildExecutionContext(args);
 
   // Return early errors if execution context failed.
   if (!('schema' in exeContext)) {
-    return { singleResult: { errors: exeContext } };
+    return { errors: exeContext };
   }
 
   return executeImpl(exeContext);
@@ -332,7 +330,7 @@ export function experimentalExecuteIncrementally(
 
 function executeImpl(
   exeContext: ExecutionContext,
-): PromiseOrValue<ExperimentalExecuteIncrementallyResults> {
+): PromiseOrValue<ExecutionResult | ExperimentalIncrementalExecutionResults> {
   // Return a Promise that will eventually resolve to the data described by
   // The "Response" section of the GraphQL specification.
   //
@@ -359,11 +357,11 @@ function executeImpl(
               subsequentResults: yieldSubsequentPayloads(exeContext),
             };
           }
-          return { singleResult: initialResult };
+          return initialResult;
         },
         (error) => {
           exeContext.errors.push(error);
-          return { singleResult: buildResponse(null, exeContext.errors) };
+          return buildResponse(null, exeContext.errors);
         },
       );
     }
@@ -377,10 +375,10 @@ function executeImpl(
         subsequentResults: yieldSubsequentPayloads(exeContext),
       };
     }
-    return { singleResult: initialResult };
+    return initialResult;
   } catch (error) {
     exeContext.errors.push(error);
-    return { singleResult: buildResponse(null, exeContext.errors) };
+    return buildResponse(null, exeContext.errors);
   }
 }
 
@@ -397,7 +395,7 @@ export function executeSync(args: ExecutionArgs): ExecutionResult {
     throw new Error('GraphQL execution failed to complete synchronously.');
   }
 
-  return result.singleResult;
+  return result;
 }
 
 /**
@@ -735,13 +733,17 @@ function executeField(
       // to take a second callback for the error case.
       return completed.then(undefined, (rawError) => {
         const error = locatedError(rawError, fieldNodes, pathToArray(path));
-        return handleFieldError(error, returnType, errors);
+        const handledError = handleFieldError(error, returnType, errors);
+        filterSubsequentPayloads(exeContext, path);
+        return handledError;
       });
     }
     return completed;
   } catch (rawError) {
     const error = locatedError(rawError, fieldNodes, pathToArray(path));
-    return handleFieldError(error, returnType, errors);
+    const handledError = handleFieldError(error, returnType, errors);
+    filterSubsequentPayloads(exeContext, path);
+    return handledError;
   }
 }
 
@@ -1094,6 +1096,7 @@ function completeListValue(
         index >= stream.initialCount
       ) {
         previousAsyncPayloadRecord = executeStreamField(
+          path,
           itemPath,
           item,
           exeContext,
@@ -1141,7 +1144,9 @@ function completeListValue(
               fieldNodes,
               pathToArray(itemPath),
             );
-            return handleFieldError(error, itemType, errors);
+            const handledError = handleFieldError(error, itemType, errors);
+            filterSubsequentPayloads(exeContext, itemPath);
+            return handledError;
           }),
         );
       } else {
@@ -1149,7 +1154,9 @@ function completeListValue(
       }
     } catch (rawError) {
       const error = locatedError(rawError, fieldNodes, pathToArray(itemPath));
-      completedResults.push(handleFieldError(error, itemType, errors));
+      const handledError = handleFieldError(error, itemType, errors);
+      filterSubsequentPayloads(exeContext, itemPath);
+      completedResults.push(handledError);
     }
     index++;
   }
@@ -1577,7 +1584,9 @@ export function experimentalSubscribeIncrementally(
 }
 
 async function* ensureAsyncIterable(
-  someExecutionResult: ExperimentalExecuteIncrementallyResults,
+  someExecutionResult:
+    | ExecutionResult
+    | ExperimentalIncrementalExecutionResults,
 ): AsyncGenerator<
   | ExecutionResult
   | InitialIncrementalExecutionResult
@@ -1589,7 +1598,7 @@ async function* ensureAsyncIterable(
     yield someExecutionResult.initialResult;
     yield* someExecutionResult.subsequentResults;
   } else {
-    yield someExecutionResult.singleResult;
+    yield someExecutionResult;
   }
 }
 
@@ -1813,6 +1822,7 @@ function executeDeferredFragment(
 
 function executeStreamField(
   path: Path,
+  itemPath: Path,
   item: PromiseOrValue<unknown>,
   exeContext: ExecutionContext,
   fieldNodes: ReadonlyArray<FieldNode>,
@@ -1823,7 +1833,7 @@ function executeStreamField(
 ): AsyncPayloadRecord {
   const asyncPayloadRecord = new StreamRecord({
     label,
-    path,
+    path: itemPath,
     parentContext,
     exeContext,
   });
@@ -1838,7 +1848,7 @@ function executeStreamField(
             itemType,
             fieldNodes,
             info,
-            path,
+            itemPath,
             resolved,
             asyncPayloadRecord,
           ),
@@ -1849,7 +1859,7 @@ function executeStreamField(
           itemType,
           fieldNodes,
           info,
-          path,
+          itemPath,
           item,
           asyncPayloadRecord,
         );
@@ -1859,20 +1869,32 @@ function executeStreamField(
         // Note: we don't rely on a `catch` method, but we do expect "thenable"
         // to take a second callback for the error case.
         completedItem = completedItem.then(undefined, (rawError) => {
-          const error = locatedError(rawError, fieldNodes, pathToArray(path));
-          return handleFieldError(error, itemType, asyncPayloadRecord.errors);
+          const error = locatedError(
+            rawError,
+            fieldNodes,
+            pathToArray(itemPath),
+          );
+          const handledError = handleFieldError(
+            error,
+            itemType,
+            asyncPayloadRecord.errors,
+          );
+          filterSubsequentPayloads(exeContext, itemPath, asyncPayloadRecord);
+          return handledError;
         });
       }
     } catch (rawError) {
-      const error = locatedError(rawError, fieldNodes, pathToArray(path));
+      const error = locatedError(rawError, fieldNodes, pathToArray(itemPath));
       completedItems = handleFieldError(
         error,
         itemType,
         asyncPayloadRecord.errors,
       );
+      filterSubsequentPayloads(exeContext, itemPath, asyncPayloadRecord);
     }
   } catch (error) {
     asyncPayloadRecord.errors.push(error);
+    filterSubsequentPayloads(exeContext, path, asyncPayloadRecord);
     asyncPayloadRecord.addItems(null);
     return asyncPayloadRecord;
   }
@@ -1882,6 +1904,7 @@ function executeStreamField(
       (value) => [value],
       (error) => {
         asyncPayloadRecord.errors.push(error);
+        filterSubsequentPayloads(exeContext, path, asyncPayloadRecord);
         return null;
       },
     );
@@ -1913,6 +1936,7 @@ async function executeStreamIteratorItem(
   } catch (rawError) {
     const error = locatedError(rawError, fieldNodes, pathToArray(fieldPath));
     const value = handleFieldError(error, itemType, asyncPayloadRecord.errors);
+    filterSubsequentPayloads(exeContext, fieldPath, asyncPayloadRecord);
     // don't continue if iterator throws
     return { done: true, value };
   }
@@ -1935,13 +1959,20 @@ async function executeStreamIteratorItem(
           fieldNodes,
           pathToArray(fieldPath),
         );
-        return handleFieldError(error, itemType, asyncPayloadRecord.errors);
+        const handledError = handleFieldError(
+          error,
+          itemType,
+          asyncPayloadRecord.errors,
+        );
+        filterSubsequentPayloads(exeContext, fieldPath, asyncPayloadRecord);
+        return handledError;
       });
     }
     return { done: false, value: completedItem };
   } catch (rawError) {
     const error = locatedError(rawError, fieldNodes, pathToArray(fieldPath));
     const value = handleFieldError(error, itemType, asyncPayloadRecord.errors);
+    filterSubsequentPayloads(exeContext, fieldPath, asyncPayloadRecord);
     return { done: false, value };
   }
 }
@@ -1998,13 +2029,47 @@ async function executeStreamIterator(
         break;
       }
     } catch (err) {
-      // do nothing, error is already handled above
+      // entire stream has errored and bubbled upwards
+      filterSubsequentPayloads(exeContext, path, asyncPayloadRecord);
+      if (iterator?.return) {
+        iterator.return().catch(() => {
+          // ignore errors
+        });
+      }
+      return;
     }
     previousAsyncPayloadRecord = asyncPayloadRecord;
     index++;
   }
 }
 
+function filterSubsequentPayloads(
+  exeContext: ExecutionContext,
+  nullPath?: Path,
+  currentAsyncRecord?: AsyncPayloadRecord,
+): void {
+  const nullPathArray = pathToArray(nullPath);
+  exeContext.subsequentPayloads.forEach((asyncRecord) => {
+    if (asyncRecord === currentAsyncRecord) {
+      // don't remove payload from where error originates
+      return;
+    }
+    for (let i = 0; i < nullPathArray.length; i++) {
+      if (asyncRecord.path[i] !== nullPathArray[i]) {
+        // asyncRecord points to a path unaffected by this payload
+        return;
+      }
+    }
+    // asyncRecord path points to nulled error field
+    if (isStreamPayload(asyncRecord) && asyncRecord.iterator?.return) {
+      asyncRecord.iterator.return().catch(() => {
+        // ignore error
+      });
+    }
+    exeContext.subsequentPayloads.delete(asyncRecord);
+  });
+}
+
 function getCompletedIncrementalResults(
   exeContext: ExecutionContext,
 ): Array<IncrementalResult> {
@@ -2027,9 +2092,7 @@ function getCompletedIncrementalResults(
       (incrementalResult as IncrementalDeferResult).data = data ?? null;
     }
 
-    incrementalResult.path = asyncPayloadRecord.path
-      ? pathToArray(asyncPayloadRecord.path)
-      : [];
+    incrementalResult.path = asyncPayloadRecord.path;
     if (asyncPayloadRecord.label) {
       incrementalResult.label = asyncPayloadRecord.label;
     }
@@ -2118,7 +2181,7 @@ class DeferredFragmentRecord {
   type: 'defer';
   errors: Array<GraphQLError>;
   label: string | undefined;
-  path: Path | undefined;
+  path: Array<string | number>;
   promise: Promise<void>;
   data: ObjMap<unknown> | null;
   parentContext: AsyncPayloadRecord | undefined;
@@ -2133,7 +2196,7 @@ class DeferredFragmentRecord {
   }) {
     this.type = 'defer';
     this.label = opts.label;
-    this.path = opts.path;
+    this.path = pathToArray(opts.path);
     this.parentContext = opts.parentContext;
     this.errors = [];
     this._exeContext = opts.exeContext;
@@ -2164,7 +2227,7 @@ class StreamRecord {
   type: 'stream';
   errors: Array<GraphQLError>;
   label: string | undefined;
-  path: Path | undefined;
+  path: Array<string | number>;
   items: Array<unknown> | null;
   promise: Promise<void>;
   parentContext: AsyncPayloadRecord | undefined;
@@ -2183,7 +2246,7 @@ class StreamRecord {
     this.type = 'stream';
     this.items = null;
     this.label = opts.label;
-    this.path = opts.path;
+    this.path = pathToArray(opts.path);
     this.parentContext = opts.parentContext;
     this.iterator = opts.iterator;
     this.errors = [];
diff --git a/src/execution/index.ts b/src/execution/index.ts
index 2130c17f..46a4688d 100644
--- a/src/execution/index.ts
+++ b/src/execution/index.ts
@@ -14,7 +14,7 @@ export {
 export type {
   ExecutionArgs,
   ExecutionResult,
-  ExperimentalExecuteIncrementallyResults,
+  ExperimentalIncrementalExecutionResults,
   InitialIncrementalExecutionResult,
   SubsequentIncrementalExecutionResult,
   IncrementalDeferResult,
diff --git a/src/index.ts b/src/index.ts
index b60b15b2..5e05627f 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -335,7 +335,7 @@ export {
 export type {
   ExecutionArgs,
   ExecutionResult,
-  ExperimentalExecuteIncrementallyResults,
+  ExperimentalIncrementalExecutionResults,
   InitialIncrementalExecutionResult,
   SubsequentIncrementalExecutionResult,
   IncrementalDeferResult,
diff --git a/website/docs/tutorials/defer-stream.md b/website/docs/tutorials/defer-stream.md
index 5235cab4..7f1c9ceb 100644
--- a/website/docs/tutorials/defer-stream.md
+++ b/website/docs/tutorials/defer-stream.md
@@ -3,7 +3,7 @@ title: Enabling Defer & Stream
 sidebar_label: Enabling Defer & Stream
 ---
 
-The `@defer` and `@stream` directives are not enabled by default. In order to use these directives, you must add them to your GraphQL Schema.
+The `@defer` and `@stream` directives are not enabled by default. In order to use these directives, you must add them to your GraphQL Schema and use the `experimentalExecuteIncrementally` function instead of `execute`.
 
 ```js
 import {
@@ -21,6 +21,11 @@ const schema = new GraphQLSchema({
     GraphQLStreamDirective,
   ],
 });
+
+const result = experimentalExecuteIncrementally({
+  schema,
+  document,
+});
 ```
 
 If the `directives` option is passed to `GraphQLSchema`, the default directives will not be included. `specifiedDirectives` must be passed to ensure all standard directives are added in addition to `defer` & `stream`.


Test patch:
diff --git a/src/execution/__tests__/defer-test.ts b/src/execution/__tests__/defer-test.ts
index ea5f069b..95c54479 100644
--- a/src/execution/__tests__/defer-test.ts
+++ b/src/execution/__tests__/defer-test.ts
@@ -99,7 +99,7 @@ async function complete(document: DocumentNode) {
     }
     return results;
   }
-  return result.singleResult;
+  return result;
 }
 
 describe('Execute: defer directive', () => {
diff --git a/src/execution/__tests__/stream-test.ts b/src/execution/__tests__/stream-test.ts
index 94cc83a2..df25d6e8 100644
--- a/src/execution/__tests__/stream-test.ts
+++ b/src/execution/__tests__/stream-test.ts
@@ -58,7 +58,21 @@ const query = new GraphQLObjectType({
           scalarField: {
             type: GraphQLString,
           },
+          nonNullScalarField: {
+            type: new GraphQLNonNull(GraphQLString),
+          },
           nestedFriendList: { type: new GraphQLList(friendType) },
+          deeperNestedObject: {
+            type: new GraphQLObjectType({
+              name: 'DeeperNestedObject',
+              fields: {
+                nonNullScalarField: {
+                  type: new GraphQLNonNull(GraphQLString),
+                },
+                deeperNestedFriendList: { type: new GraphQLList(friendType) },
+              },
+            }),
+          },
         },
       }),
     },
@@ -84,7 +98,7 @@ async function complete(document: DocumentNode, rootValue: unknown = {}) {
     }
     return results;
   }
-  return result.singleResult;
+  return result;
 }
 
 async function completeAsync(
@@ -765,9 +779,15 @@ describe('Execute: stream directive', () => {
     `);
     const result = await complete(document, {
       async *nonNullFriendList() {
-        yield await Promise.resolve(friends[0]);
-        yield await Promise.resolve(null);
-        yield await Promise.resolve(friends[1]);
+        try {
+          yield await Promise.resolve(friends[0]);
+          yield await Promise.resolve(null); /* c8 ignore start */
+          // Not reachable, early return
+        } finally {
+          /* c8 ignore stop */
+          // eslint-disable-next-line no-unsafe-finally
+          throw new Error('Oops');
+        }
       },
     });
     expectJSON(result).toDeepEqual([
@@ -792,18 +812,6 @@ describe('Execute: stream directive', () => {
             ],
           },
         ],
-        hasNext: true,
-      },
-      {
-        incremental: [
-          {
-            items: [{ name: 'Han' }],
-            path: ['nonNullFriendList', 2],
-          },
-        ],
-        hasNext: true,
-      },
-      {
         hasNext: false,
       },
     ]);
@@ -886,15 +894,6 @@ describe('Execute: stream directive', () => {
             ],
           },
         ],
-        hasNext: true,
-      },
-      {
-        incremental: [
-          {
-            items: [{ nonNullName: 'Han' }],
-            path: ['nonNullFriendList', 2],
-          },
-        ],
         hasNext: false,
       },
     ]);
@@ -953,6 +952,302 @@ describe('Execute: stream directive', () => {
       },
     ]);
   });
+  it('Filters payloads that are nulled', async () => {
+    const document = parse(`
+      query { 
+        nestedObject {
+          nonNullScalarField
+          nestedFriendList @stream(initialCount: 0) {
+            name
+          }
+        }
+      }
+    `);
+    const result = await complete(document, {
+      nestedObject: {
+        nonNullScalarField: () => Promise.resolve(null),
+        async *nestedFriendList() {
+          yield await Promise.resolve(friends[0]);
+        },
+      },
+    });
+    expectJSON(result).toDeepEqual({
+      errors: [
+        {
+          message:
+            'Cannot return null for non-nullable field NestedObject.nonNullScalarField.',
+          locations: [{ line: 4, column: 11 }],
+          path: ['nestedObject', 'nonNullScalarField'],
+        },
+      ],
+      data: {
+        nestedObject: null,
+      },
+    });
+  });
+  it('Does not filter payloads when null error is in a different path', async () => {
+    const document = parse(`
+      query { 
+        otherNestedObject: nestedObject {
+          ... @defer {
+            scalarField
+          }
+        }
+        nestedObject {
+          nestedFriendList @stream(initialCount: 0) {
+            name
+          }
+        }
+      }
+    `);
+    const result = await complete(document, {
+      nestedObject: {
+        scalarField: () => Promise.reject(new Error('Oops')),
+        async *nestedFriendList() {
+          yield await Promise.resolve(friends[0]);
+        },
+      },
+    });
+    expectJSON(result).toDeepEqual([
+      {
+        data: {
+          otherNestedObject: {},
+          nestedObject: { nestedFriendList: [] },
+        },
+        hasNext: true,
+      },
+      {
+        incremental: [
+          {
+            data: { scalarField: null },
+            path: ['otherNestedObject'],
+            errors: [
+              {
+                message: 'Oops',
+                locations: [{ line: 5, column: 13 }],
+                path: ['otherNestedObject', 'scalarField'],
+              },
+            ],
+          },
+          {
+            items: [{ name: 'Luke' }],
+            path: ['nestedObject', 'nestedFriendList', 0],
+          },
+        ],
+        hasNext: true,
+      },
+      {
+        hasNext: false,
+      },
+    ]);
+  });
+  it('Filters stream payloads that are nulled in a deferred payload', async () => {
+    const document = parse(`
+      query { 
+        nestedObject {
+          ... @defer {
+            deeperNestedObject {
+              nonNullScalarField
+              deeperNestedFriendList @stream(initialCount: 0) {
+                name
+              }
+            }
+          }
+        }
+      }
+    `);
+    const result = await complete(document, {
+      nestedObject: {
+        deeperNestedObject: {
+          nonNullScalarField: () => Promise.resolve(null),
+          async *deeperNestedFriendList() {
+            yield await Promise.resolve(friends[0]);
+          },
+        },
+      },
+    });
+    expectJSON(result).toDeepEqual([
+      {
+        data: {
+          nestedObject: {},
+        },
+        hasNext: true,
+      },
+      {
+        incremental: [
+          {
+            data: {
+              deeperNestedObject: null,
+            },
+            path: ['nestedObject'],
+            errors: [
+              {
+                message:
+                  'Cannot return null for non-nullable field DeeperNestedObject.nonNullScalarField.',
+                locations: [{ line: 6, column: 15 }],
+                path: [
+                  'nestedObject',
+                  'deeperNestedObject',
+                  'nonNullScalarField',
+                ],
+              },
+            ],
+          },
+        ],
+        hasNext: false,
+      },
+    ]);
+  });
+  it('Filters defer payloads that are nulled in a stream response', async () => {
+    const document = parse(`
+    query { 
+      friendList @stream(initialCount: 0) {
+        nonNullName
+        ... @defer {
+          name
+        }
+      }
+    }
+  `);
+    const result = await complete(document, {
+      async *friendList() {
+        yield await Promise.resolve({
+          name: friends[0].name,
+          nonNullName: () => Promise.resolve(null),
+        });
+      },
+    });
+    expectJSON(result).toDeepEqual([
+      {
+        data: {
+          friendList: [],
+        },
+        hasNext: true,
+      },
+      {
+        incremental: [
+          {
+            items: [null],
+            path: ['friendList', 0],
+            errors: [
+              {
+                message:
+                  'Cannot return null for non-nullable field Friend.nonNullName.',
+                locations: [{ line: 4, column: 9 }],
+                path: ['friendList', 0, 'nonNullName'],
+              },
+            ],
+          },
+        ],
+        hasNext: true,
+      },
+      {
+        hasNext: false,
+      },
+    ]);
+  });
+
+  it('Returns iterator and ignores errors when stream payloads are filtered', async () => {
+    let returned = false;
+    let index = 0;
+    const iterable = {
+      [Symbol.asyncIterator]: () => ({
+        next: () => {
+          const friend = friends[index++];
+          if (!friend) {
+            return Promise.resolve({ done: true, value: undefined });
+          }
+          return Promise.resolve({
+            done: false,
+            value: {
+              name: friend.name,
+              nonNullName: null,
+            },
+          });
+        },
+        return: () => {
+          returned = true;
+          return Promise.reject(new Error('Oops'));
+        },
+      }),
+    };
+
+    const document = parse(`
+      query { 
+        nestedObject {
+          ... @defer {
+            deeperNestedObject {
+              nonNullScalarField
+              deeperNestedFriendList @stream(initialCount: 0) {
+                name
+              }
+            }
+          }
+        }
+      }
+    `);
+
+    const executeResult = await experimentalExecuteIncrementally({
+      schema,
+      document,
+      rootValue: {
+        nestedObject: {
+          deeperNestedObject: {
+            nonNullScalarField: () => Promise.resolve(null),
+            deeperNestedFriendList: iterable,
+          },
+        },
+      },
+    });
+    assert('initialResult' in executeResult);
+    const iterator = executeResult.subsequentResults[Symbol.asyncIterator]();
+
+    const result1 = executeResult.initialResult;
+    expectJSON(result1).toDeepEqual({
+      data: {
+        nestedObject: {},
+      },
+      hasNext: true,
+    });
+
+    const result2 = await iterator.next();
+    expectJSON(result2).toDeepEqual({
+      done: false,
+      value: {
+        incremental: [
+          {
+            data: {
+              deeperNestedObject: null,
+            },
+            path: ['nestedObject'],
+            errors: [
+              {
+                message:
+                  'Cannot return null for non-nullable field DeeperNestedObject.nonNullScalarField.',
+                locations: [{ line: 6, column: 15 }],
+                path: [
+                  'nestedObject',
+                  'deeperNestedObject',
+                  'nonNullScalarField',
+                ],
+              },
+            ],
+          },
+        ],
+        hasNext: true,
+      },
+    });
+    const result3 = await iterator.next();
+    expectJSON(result3).toDeepEqual({
+      done: false,
+      value: { hasNext: false },
+    });
+
+    const result4 = await iterator.next();
+    expectJSON(result4).toDeepEqual({ done: true, value: undefined });
+
+    assert(returned);
+  });
   it('Handles promises returned by completeValue after initialCount is reached', async () => {
     const document = parse(`
       query {