fix: Resolve issues with deprecated synchronous error handling and chained operators

spec/Observable-spec.ts

@@ -608,6 +608,27 @@ describe('Observable', () => {
         }).to.throw('error!');
       });
 
+
+      // From issue: https://github.com/ReactiveX/rxjs/issues/5979
+      it('should still rethrow synchronous errors from next handlers on synchronous observables', () => {
+        expect(() => {
+          of('test').pipe(
+            // Any operators here
+            map(x => x + '!!!'),
+            map(x => x + x),
+            map(x => x + x),
+            map(x => x + x),
+          ).subscribe({
+            next: () => {
+              throw new Error(
+                'hi there!'
+              )
+            }
+          })
+        }).to.throw();
+      });
+
+
       afterEach(() => {
         config.useDeprecatedSynchronousErrorHandling = false;
       });

src/internal/Observable.ts

@@ -211,7 +211,7 @@ export class Observable<T> implements Subscribable<T> {
     // If we have an operator, it's the result of a lift, and we let the lift
     // mechanism do the subscription for us in the operator call. Otherwise,
     // if we have a source, it's a trusted observable we own, and we can call
-    // the _subscribe without wrapping it in a try/catch. If we are supposed to
+    // the `_subscribe` without wrapping it in a try/catch. If we are supposed to
     // use the deprecated sync error handling, then we don't need the try/catch either
     // otherwise, it may be from a user-made observable instance, and we want to
     // wrap it in a try/catch so we can handle errors appropriately.
@@ -224,6 +224,20 @@ export class Observable<T> implements Subscribable<T> {
         : this._trySubscribe(subscriber)
     );
 
+    if (config.useDeprecatedSynchronousErrorHandling) {
+      // In the case of the deprecated sync error handling,
+      // we need to crawl forward through our subscriber chain and
+      // look to see if there's any synchronously thrown errors.
+      // Does this suck for perf? Yes. So stop using the deprecated sync
+      // error handling already. We're removing this in v8.
+      let dest: any = subscriber;
+      while (dest) {
+        if (dest.__syncError) {
+          throw dest.__syncError;
+        }
+        dest = dest.destination;
+      }
+    }
     return subscriber;
   }
 
@@ -232,9 +246,9 @@ export class Observable<T> implements Subscribable<T> {
     try {
       return this._subscribe(sink);
     } catch (err) {
-      if (config.useDeprecatedSynchronousErrorHandling) {
-        throw err;
-      }
+      // We don't need to return anything in this case,
+      // because it's just going to try to `add()` to a subscription
+      // above.
       sink.error(err);
     }
   }

src/internal/Subscriber.ts

@@ -167,14 +167,40 @@ export class SafeSubscriber<T> extends Subscriber<T> {
       // Once we set the destination, the superclass `Subscriber` will
       // do it's magic in the `_next`, `_error`, and `_complete` methods.
       this.destination = {
-        next: next || noop,
-        error: error || defaultErrorHandler,
-        complete: complete || noop,
+        next: next ? maybeWrapForDeprecatedSyncErrorHandling(next, this) : noop,
+        error: error ? maybeWrapForDeprecatedSyncErrorHandling(error, this) : defaultErrorHandler,
+        complete: complete ? maybeWrapForDeprecatedSyncErrorHandling(complete, this) : noop,
       };
     }
   }
 }
 
+/**
+ * Checks to see if the user has chosen to use the super gross deprecated error handling that
+ * no one should ever use, ever. If they did choose that path, we need to catch their error
+ * so we can stick it on a super-secret property and check it after the subscription is done
+ * in the `Observable` subscribe call.
+ *
+ * We have to do this, because if we simply rethrow the error, it will be caught by any upstream
+ * try/catch blocks and send back down again, basically playing ping-pong with the error until the
+ * downstream runs out of chances to rethrow and it gives up.
+ *
+ * In the general case, for non-crazy people, this just returns the handler directly.
+ *
+ * @param handler The handler to wrap
+ * @param instance The SafeSubscriber instance we're going to mark if there's an error.
+ */
+function maybeWrapForDeprecatedSyncErrorHandling(handler: (arg?: any) => void, instance: SafeSubscriber<any>) {
+  return config.useDeprecatedSynchronousErrorHandling
+    ? (arg?: any) => {
+        try {
+          handler(arg);
+        } catch (err) {
+          (instance as any).__syncError = err;
+        }
+      }
+    : handler;
+}
 /**
  * An error handler used when no error handler was supplied
  * to the SafeSubscriber -- meaning no error handler was supplied

src/internal/config.ts

@@ -45,7 +45,7 @@ export const config = {
    * in v6 and higher. This behavior enables bad patterns like wrapping a subscribe
    * call in a try/catch block. It also enables producer interference, a nasty bug
    * where a multicast can be broken for all observers by a downstream consumer with
-   * an unhandled error. DO NOT USE THIS FLAG UNLESS IT'S NEEDED TO BY TIME
+   * an unhandled error. DO NOT USE THIS FLAG UNLESS IT'S NEEDED TO BUY TIME
    * FOR MIGRATION REASONS.
    *
    * @deprecated remove in v8. As of version 8, RxJS will no longer support synchronous throwing