Strange catchError behavior inside of concat

[benlesh]

RxJS version: 6.2.0

Code to reproduce:

Try it out on this Stackblitz

import { Observable, of, concat, throwError } from 'rxjs';
import { map, catchError } from 'rxjs/operators';

concat(
  new Observable<string>(o => {
    o.next('first item');
    throw 'lol error';
  }).pipe(
    catchError(err => {
      console.log('caught', err);
      return of('blah');
    })
  ),
  of('this should be played'),
)
.subscribe({
  next(x) { console.log(x); },
  error(err) { console.log(err); },
})

Expected behavior:

Logs

first item
caught
blah
this should be played

Actual behavior:
Logs

first item
lol error

Additional information:

Appears to have to do with the difference in behaviors between when an Observable is lifted, vs when it's not WRT syncErrorThrowable.

[benlesh]

Thanks, @hansl for pointing this one out.

[cartant]

This seems related to #3560. See this comment.

[cartant]

I've had a quick look at this. It seems to be related to the syncErrorThrowable property and the internal SafeSubscriber in Subscriber.

In Observable.prototype.subscribe, _trySubscribe is called depends upon whether there is a source and the value of syncErrorThrowable.

The problem is that those conditions are met only for the subscribe call that's made to the outermost observable - the observable returned from the concat call.

It works like this:

• concat is implemented by applying the concatAll operator to an of observable.
• concatAll is implemented using mergeMap (with a concurrency of one) and mergeMap has an InnerSubscriber.
• InnerSubscriber does not copy its parent's syncErrorThrowable value.
• Essentially, that means that when mergeMap subscribes to the observable returned by catchError, to does so by calling _subscribe - and not _trySubscribe.
• The means when an error is thrown in the new Observable, the stack unwinds to the _trySubscribe call that's made in the of observable and the catchError implementation doesn't even get a look in.

Note that the of('this should be played') only matters because without it, concat is short-circuited to return the of observable - which sees the catchError operator returned and subscribed to directly - which means catchError gets a SafeSubscriber, etc.

A possible fix - which fixes the repro - could be something like this, but it needs more thought:

export class InnerSubscriber<T, R> extends Subscriber<R> {
  private index = 0;

  constructor(private parent: OuterSubscriber<T, R>, public outerValue: T, public outerIndex: number) {
    super();
    this.syncErrorThrowable = parent.syncErrorThrowable;
  }

  // ...
}

[benlesh]

@cartant, at first blush, that seems like the correct solution. The only thing is we'd need to make sure that the InnerSubscriber's syncErrorThrowable property was updated at the appropriate time to be set back to false (meaning it would need to be linked to the parent's syncErrorThrowable property). So perhaps a getter propertly on the InnerSubscriber that was backed by parent.syncErrorThrowable would be better?

(I haven't verified any of this, just thinking this might be the better way to approach this)

[cartant]

Yeah, using a getter would be the way to go.

I'm pretty sure that syncErrorThrowable is only assigned to during or immediately after the subscriber's construction, but "pretty sure" isn't really good enough. Using a getter would be less rash.

It's also assigned here, so it definitely has to be a getter.

[cartant]

Actually, I suppose it should really have getters for each of the syncError-related properties in the parent:

/** @internal */ syncErrorValue: any = null;
/** @internal */ syncErrorThrown: boolean = false;
/** @internal */ syncErrorThrowable: boolean = false;

Need to think about this.

No, that's not necessary. It doesn't even make sense, as the outer (the parent) subscribes to the inner.

[cartant]

As evident in the previous stream-of-consciousness comment, I've been messing with the above-mentioned fix.

I'm inclined to change my mind about it, as it's really only related to the deprecated sync-error handling and that doesn't even work with the fix applied.

If the inner subscriber's syncErrorThrowable uses the value of the parent's property - via assignment or a getter - it still fails, as the error will be rethrown - see #3560. That is, the catchError operator will catch lol error and will emit blah into the stream, but it then re-throws the error which is passed to the outermost error callback and then re-thrown again.

Given that this behaviour is still broken and that #3560 was closed, I think a better fix might be to use a different test for deciding between a _subscribe call and a _trySubscribe call.

Instead of this:

sink.add(this.source || !sink.syncErrorThrowable ? this._subscribe(sink) : this._trySubscribe(sink));

It could be something like this:

const noTry = config.useDeprecatedSynchronousErrorHandling ?
  this.source || !sink.syncErrorThrowable :
  this.source;
sink.add(noTry ? this._subscribe(sink) : this._trySubscribe(sink));

This solves the problem - for non-deprecated usage - in a less complicated way.

[benlesh]

Your solution looks solid, @cartant... proceed with the fix if you're so inclined.

[cartant]

@benlesh It's done in #3749. It does not include the getter business. To do anything with the deprecated syncError handling would require a lot more thought and I don't think it could be easily fixed.