Repository: graphql/graphql-js
Pull request #3720 (2022-08-31 19:27:08)

Filter subsequent payloads when parent field is null
Implementing what was discussed in https://github.com/robrichard/defer-stream-wg/discussions/45

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


|  Name | Link |
|---------------------------------|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | 370aa29cb6c5e357afdf707ed43594f43ecb02aa |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/compassionate-pike-271cb3/deploys/630fb60ddf5d0500085068fd |
github-actions[bot]: Hi @robrichard, 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>

benjie: I don't think this is the approach I would have taken. Instead:

1. When a nullable position becomes null because of an error, add this path to a new set on `exeContext` (let's call the new set `errorNulledPaths` since it's a list of the paths that were nulled out due to an error caught by a nullable type)
2. In `filterSubsequentPayloads`, find and cancel the subsequentPayloads whose path _starts with or matches_ one of the paths in `errorNulledPaths` - no need to traverse the response data

Assuming that the length of paths is negligible, I think this basic approach would scale at O(N*M) where N is the number of errors and M is the number of subsequent payloads, but I'm sure there are clever ways to optimize it better.
robrichard: @benjie I significantly rewrote this. Now, as discussed in the WG today, I've added the logic to filter subsequent payloads at the point where the bubbled null is ultimately set. This no longer requires inspecting the payload response and should not add any overhead in the non-error case.
benjie: Definitely looks better; thanks @robrichard!

Patch:
diff --git a/src/execution/execute.ts b/src/execution/execute.ts
index ab2062c0..f5a8c301 100644
--- a/src/execution/execute.ts
+++ b/src/execution/execute.ts
@@ -735,13 +735,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 +1098,7 @@ function completeListValue(
         index >= stream.initialCount
       ) {
         previousAsyncPayloadRecord = executeStreamField(
+          path,
           itemPath,
           item,
           exeContext,
@@ -1141,7 +1146,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 +1156,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++;
   }
@@ -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 = [];


Test patch:
diff --git a/src/execution/__tests__/stream-test.ts b/src/execution/__tests__/stream-test.ts
index 94cc83a2..d90508fb 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) },
+              },
+            }),
+          },
         },
       }),
     },
@@ -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 {