enable early abort from incremental delivery despite pending promises

without `Promise.race()`!

Author: yaacovCR

packages/executor/src/execution/IncrementalPublisher.ts

@@ -115,7 +115,7 @@ class IncrementalPublisher {
     let isDone = false;
 
     this._context.signal?.addEventListener('abort', () => {
-      isDone = true;
+      this._incrementalGraph.completedIncrementalData().return();
     });
 
     const _next = async (): Promise<
@@ -138,10 +138,6 @@ class IncrementalPublisher {
       const asyncIterator = completedIncrementalData[Symbol.asyncIterator]();
       let iteration = await asyncIterator.next();
       while (!iteration.done) {
-        if (this._context.signal?.aborted) {
-          throw this._context.signal.reason;
-        }
-
         for (const completedResult of iteration.value) {
           this._handleCompletedIncrementalData(completedResult, context);
         }
@@ -176,6 +172,10 @@ class IncrementalPublisher {
         iteration = await asyncIterator.next();
       }
 
+      if (this._context.signal?.aborted) {
+        throw this._context.signal.reason;
+      }
+
       await this._returnAsyncIteratorsIgnoringErrors();
       return { value: undefined, done: true };
     };

packages/executor/src/execution/__tests__/abort-signal.test.ts

@@ -469,6 +469,89 @@ describe('Abort Signal', () => {
     await expect(next$).rejects.toThrow('This operation was aborted');
     expect(bResolverGotInvoked).toBe(false);
   });
+  it('stops pending stream execution for never-returning incremental delivery (@defer)', async () => {
+    const aResolverGotInvokedD = createDeferred();
+    const requestGotCancelledD = createDeferred();
+    let bResolverGotInvoked = false;
+
+    const schema = makeExecutableSchema({
+      typeDefs: /* GraphQL */ `
+        type Query {
+          root: A!
+        }
+        type A {
+          a: B!
+        }
+        type B {
+          b: String
+        }
+      `,
+      resolvers: {
+        Query: {
+          async root() {
+            return {};
+          },
+        },
+        A: {
+          async a() {
+            aResolverGotInvokedD.resolve();
+            await requestGotCancelledD.promise;
+            return {};
+          },
+        },
+        B: {
+          b() {
+            bResolverGotInvoked = true;
+            return new Promise(() => {});
+          },
+        },
+      },
+    });
+    const controller = new AbortController();
+    const result = await normalizedExecutor({
+      schema,
+      document: parse(/* GraphQL */ `
+        query {
+          root {
+            ... @defer {
+              a {
+                b
+              }
+            }
+          }
+        }
+      `),
+      signal: controller.signal,
+    });
+
+    if (!isAsyncIterable(result)) {
+      throw new Error('Result is not an async iterable');
+    }
+
+    const iterator = result[Symbol.asyncIterator]();
+    const next = await iterator.next();
+    expect(next.value).toMatchInlineSnapshot(`
+{
+  "data": {
+    "root": {},
+  },
+  "hasNext": true,
+  "pending": [
+    {
+      "id": "0",
+      "path": [
+        "root",
+      ],
+    },
+  ],
+}
+`);
+    const next$ = iterator.next();
+    await aResolverGotInvokedD.promise;
+    controller.abort();
+    await expect(next$).rejects.toThrow('This operation was aborted');
+    expect(bResolverGotInvoked).toBe(false);
+  });
   it('stops promise execution', async () => {
     const controller = new AbortController();
     const d = createDeferred();