refactor(incremental): close stream iterator only on abnormal stop (#4641)

Author: yaacovCR

src/execution/incremental/IncrementalExecutor.ts

@@ -751,8 +751,8 @@ export class IncrementalExecutor<
             cancelStreamItems.forEach((cancelStreamItem) =>
               cancelStreamItem(reason),
             );
+            returnIteratorCatchingErrors(iterator);
           }
-          returnIteratorCatchingErrors(iterator);
         });
         await (enableEarlyExecution ? Promise.resolve() : started);
         if (stopRequested) {

src/execution/incremental/__tests__/stream-test.ts

@@ -2784,10 +2784,12 @@ describe('Execute: stream directive', () => {
     const iterable = {
       [Symbol.asyncIterator]: () => ({
         next: () => {
-          const friend = friends[index++];
-          if (friend == null) {
-            return Promise.resolve({ done: true, value: undefined });
+          if (index > 1) {
+            return new Promise(() => {
+              // never resolves
+            });
           }
+          const friend = friends[index++];
           return Promise.resolve({ done: false, value: friend });
         },
         return: () => {

src/execution/legacyIncremental/__tests__/legacy-stream-test.ts

@@ -2703,10 +2703,12 @@ describe('Execute: stream directive (legacy)', () => {
     const iterable = {
       [Symbol.asyncIterator]: () => ({
         next: () => {
-          const friend = friends[index++];
-          if (friend == null) {
-            return Promise.resolve({ done: true, value: undefined });
+          if (index > 1) {
+            return new Promise(() => {
+              // never resolves
+            });
           }
+          const friend = friends[index++];
           return Promise.resolve({ done: false, value: friend });
         },
         return: () => {