fix(forEach): properly unsubs after error in next handler (#6677)

fixes #6676

Author: benlesh

spec/Observable-spec.ts

@@ -153,9 +153,9 @@ describe('Observable', () => {
           },
           (err) => {
             results.push(err);
-            // Since the consuming code can no longer interfere with the synchronous
-            // producer, the remaining results are nexted.
-            expect(results).to.deep.equal([1, 2, 3, 4, expected]);
+            // The error should unsubscribe from the source, meaning we 
+            // should not see the number 4.
+            expect(results).to.deep.equal([1, 2, 3, expected]);
           }
         );
     });

src/internal/Observable.ts

@@ -313,21 +313,19 @@ export class Observable<T> implements Subscribable<T> {
     promiseCtor = getPromiseCtor(promiseCtor);
 
     return new promiseCtor<void>((resolve, reject) => {
-      // Must be declared in a separate statement to avoid a ReferenceError when
-      // accessing subscription below in the closure due to Temporal Dead Zone.
-      let subscription: Subscription;
-      subscription = this.subscribe(
-        (value) => {
+      const subscriber = new SafeSubscriber<T>({
+        next: (value) => {
           try {
             next(value);
           } catch (err) {
             reject(err);
-            subscription?.unsubscribe();
+            subscriber.unsubscribe();
           }
         },
-        reject,
-        resolve
-      );
+        error: reject,
+        complete: resolve,
+      });
+      this.subscribe(subscriber);
     }) as Promise<void>;
   }