execution: reduce parent executor retention in incremental callbacks (#4665)

Author: yaacovCR

src/execution/Executor.ts

@@ -375,18 +375,24 @@ export class Executor<
     this.aborted = true;
   }
 
-  finishSharedExecution(): void {
-    this.resolverAbortController?.abort();
+  getFinishSharedExecution(): () => void {
+    const resolverAbortController = this.resolverAbortController;
     const asyncWorkFinishedHook =
       this.validatedExecutionArgs.hooks?.asyncWorkFinished;
     if (asyncWorkFinishedHook === undefined) {
-      return;
+      return () => resolverAbortController?.abort();
     }
-    runAsyncWorkFinishedHook(
-      this.validatedExecutionArgs,
-      this.sharedExecutionContext,
-      asyncWorkFinishedHook,
-    );
+
+    const validatedExecutionArgs = this.validatedExecutionArgs;
+    const sharedExecutionContext = this.sharedExecutionContext;
+    return () => {
+      resolverAbortController?.abort();
+      runAsyncWorkFinishedHook(
+        validatedExecutionArgs,
+        sharedExecutionContext,
+        asyncWorkFinishedHook,
+      );
+    };
   }
 
   /**
@@ -396,7 +402,7 @@ export class Executor<
   buildResponse(
     data: ObjMap<unknown> | null,
   ): ExecutionResult | TAlternativeInitialResponse {
-    this.finishSharedExecution();
+    this.getFinishSharedExecution()();
     this.finish();
     const errors = this.collectedErrors.errors;
     const result = errors.length ? { errors, data } : { data };

src/execution/incremental/IncrementalExecutor.ts

@@ -310,14 +310,18 @@ export class IncrementalExecutor<
     this.streams = [];
   }
 
-  createSubExecutor(
+  getCreateSubExecutor(): (
     deferUsageSet?: DeferUsageSet,
-  ): IncrementalExecutor<TExperimental> {
-    return new IncrementalExecutor(
-      this.validatedExecutionArgs,
-      this.sharedExecutionContext,
-      deferUsageSet,
-    );
+  ) => IncrementalExecutor<TExperimental> {
+    const validatedExecutionArgs = this.validatedExecutionArgs;
+    const sharedExecutionContext = this.sharedExecutionContext;
+
+    return (deferUsageSet?: DeferUsageSet) =>
+      new IncrementalExecutor<TExperimental>(
+        validatedExecutionArgs,
+        sharedExecutionContext,
+        deferUsageSet,
+      );
   }
 
   override abort(reason?: unknown): void {
@@ -354,7 +358,7 @@ export class IncrementalExecutor<
       errors,
       work,
       this.validatedExecutionArgs.externalAbortSignal,
-      () => this.finishSharedExecution(),
+      this.getFinishSharedExecution(),
     ) as TExperimental;
   }
 
@@ -490,10 +494,11 @@ export class IncrementalExecutor<
     newGroupedFieldSets: Map<DeferUsageSet, GroupedFieldSet>,
     deliveryGroupMap: ReadonlyMap<DeferUsage, DeliveryGroup>,
   ): void {
+    const createSubExecutor = this.getCreateSubExecutor();
     for (const [deferUsageSet, groupedFieldSet] of newGroupedFieldSets) {
       const deliveryGroups = getDeliveryGroups(deferUsageSet, deliveryGroupMap);
 
-      const executor = this.createSubExecutor(deferUsageSet);
+      const executor = createSubExecutor(deferUsageSet);
 
       const executionGroup: ExecutionGroup = {
         groups: deliveryGroups,
@@ -721,7 +726,9 @@ export class IncrementalExecutor<
     itemType: GraphQLOutputType,
     isAsync: boolean | undefined,
   ): Queue<StreamItemResult> {
+    const createSubExecutor = this.getCreateSubExecutor();
     const { enableEarlyExecution } = this.validatedExecutionArgs;
+    const sharedExecutionContext = this.sharedExecutionContext;
     const queue = new Queue<StreamItemResult>(
       async ({ push, stop, onStop, started }) => {
         const abortStreamItems = new Set<(reason?: unknown) => void>();
@@ -735,13 +742,13 @@ export class IncrementalExecutor<
               abortStreamItem(reason);
             }
             if (isAsync) {
-              this.sharedExecutionContext.asyncWorkTracker.add(
+              sharedExecutionContext.asyncWorkTracker.add(
                 returnIteratorCatchingErrors(
                   iterator as AsyncIterator<unknown>,
                 ),
               );
             } else {
-              this.sharedExecutionContext.asyncWorkTracker.addValues(
+              sharedExecutionContext.asyncWorkTracker.addValues(
                 collectIteratorPromises(iterator as Iterator<unknown>),
               );
             }
@@ -785,7 +792,7 @@ export class IncrementalExecutor<
 
           const itemPath = addPath(streamPath, index, undefined);
 
-          const executor = this.createSubExecutor();
+          const executor = createSubExecutor();
 
           let streamItemResult = executor.completeStreamItem(
             itemPath,

src/execution/legacyIncremental/BranchingIncrementalExecutor.ts

@@ -174,14 +174,18 @@ export interface FormattedLegacyIncrementalStreamResult<
 }
 /** @internal */
 export class BranchingIncrementalExecutor extends IncrementalExecutor<LegacyExperimentalIncrementalExecutionResults> {
-  override createSubExecutor(
+  override getCreateSubExecutor(): (
     deferUsageSet?: DeferUsageSet,
-  ): IncrementalExecutor<LegacyExperimentalIncrementalExecutionResults> {
-    return new BranchingIncrementalExecutor(
-      this.validatedExecutionArgs,
-      this.sharedExecutionContext,
-      deferUsageSet,
-    );
+  ) => BranchingIncrementalExecutor {
+    const validatedExecutionArgs = this.validatedExecutionArgs;
+    const sharedExecutionContext = this.sharedExecutionContext;
+
+    return (deferUsageSet?: DeferUsageSet) =>
+      new BranchingIncrementalExecutor(
+        validatedExecutionArgs,
+        sharedExecutionContext,
+        deferUsageSet,
+      );
   }
 
   override buildResponse(
@@ -202,7 +206,7 @@ export class BranchingIncrementalExecutor extends IncrementalExecutor<LegacyExpe
       errors,
       work,
       this.validatedExecutionArgs.externalAbortSignal,
-      () => this.finishSharedExecution(),
+      this.getFinishSharedExecution(),
     );
   }