Should race have a special case for synchronous observables?

[mhverbakel]

Potential Bug Report

Current Behavior
If a race observable is constructed with at least one synchronous observable, the order of events is confusing.

1. (Optionally) Subscribe to asynchronous observables that come before the first synchronous observable.
2. Subscribe to the first synchronous observable.
3. Copy the output from the synchronous observable.
4. Subscribe and immediately unsubscribe from all other observables.

Step 4 could be ommitted in my opinion. If this is intended behavior, maybe put a note about it in the documentation?

Reproduction
https://stackblitz.com/edit/rxjs-ec1mwd

import { race, EMPTY } from 'rxjs';
import { create } from 'rxjs-spy';
import { tag } from 'rxjs-spy/operators';
import { delay, map } from 'rxjs/operators';

create().log(/.*/);
const source = race(
  EMPTY.pipe(delay(100), tag("e0")),
  EMPTY.pipe(tag("e1")),
  EMPTY.pipe(delay(100), tag("e2")),
  EMPTY.pipe(delay(100), tag("e3")),
).pipe(
  tag("output"),
)

source.subscribe();

Tag = output; notification = subscribe; matching /.*/
== Step 1
Tag = e0; notification = subscribe; matching /.*/
== Step 2
Tag = e1; notification = subscribe; matching /.*/
Tag = e1; notification = complete; matching /.*/
== Step 3
Tag = output; notification = complete; matching /.*/
Tag = output; notification = unsubscribe; matching /.*/
Tag = e0; notification = unsubscribe; matching /.*/
Tag = e1; notification = unsubscribe; matching /.*/
== Step 4
Tag = e2; notification = subscribe; matching /.*/
Tag = e2; notification = unsubscribe; matching /.*/
Tag = e3; notification = subscribe; matching /.*/
Tag = e3; notification = unsubscribe; matching /.*/

[cartant]

Should race have a special case for synchronous observables?

No. In short, it is not possible to determine whether or not an observable is synchronous or not without subscribing to it. And race already has clearly defined behaviour.

[mhverbakel]

After diving into the code for race, I can now change this into an actual bug report.

Race is intended to stop subscribing as soon as the first Observable emits (see https://github.com/ReactiveX/rxjs/blob/master/src/internal/observable/race.ts#L124). However, completion is not setting the hasFirst flag, and therefore it continues subscribing and unsubscribing to the other observables.

I can show this with the same code example, except switching out EMPTY for of(1):
https://stackblitz.com/edit/rxjs-kuk8zx

import { race, of } from 'rxjs';
import { create } from 'rxjs-spy';
import { tag } from 'rxjs-spy/operators';
import { delay, map } from 'rxjs/operators';

create().log(/.*/);
const source = race(
  of(1).pipe(delay(100), tag("e0")),
  of(1).pipe(tag("e1")),
  of(1).pipe(delay(100), tag("e2")),
  of(1).pipe(delay(100), tag("e3")),
).pipe(
  tag("output"),
)

source.subscribe();

Tag = output; notification = subscribe; matching /.*/
== Step 1, as above
Tag = e0; notification = subscribe; matching /.*/
== Step 2, as above
Tag = e1; notification = subscribe; matching /.*/
Tag = e1; notification = next; matching /.*/; value = 1
== Unsubscribe of e0 happens before now
Tag = e0; notification = unsubscribe; matching /.*/
== Step 3
Tag = output; notification = next; matching /.*/; value =
1
Tag = e1; notification = complete; matching /.*/
Tag = output; notification = complete; matching /.*/
Tag = output; notification = unsubscribe; matching /.*/
Tag = e1; notification = unsubscribe; matching /.*/
== No step 4, e2 and e3 are never subscribed.

[mhverbakel]

I think fixing should be something along the lines of adding this:

  notifyComplete(innerSub: InnerSubscriber<T, T>): void {
    if (!this.hasFirst) {
      this.hasFirst = true;

      for (let i = 0; i < this.subscriptions.length; i++) {
        let subscription = this.subscriptions[i];

        subscription.unsubscribe();
        this.remove(subscription);
      }

      this.subscriptions = null;
    }

    this.destination.complete();
  }

[cartant]

To me, it looks like it's behaving as expected. You have contradictory statements in this comment, so it's difficult to understand what it is that you believe to be the problem.

[mhverbakel]

I'm trying to create a marbles test case for this, but it turns out that the cold() function never runs synchronously, although for example of() or EMPTY do in real world cases.

I've created a new example (https://stackblitz.com/edit/rxjs-w2fll9) and I will explain what happens and what strikes me.

In case A (3 asynchronous observables), all side effects are triggered.
In case B (2 asynchronous observables, one synchronous observable), the last side effect is not triggered.
In case C (same as B, but the winner is empty), the last side effect is suddenly triggered again.

In my initial report, I said that I think case C is wrong, because that is an unnecessary subscription (because it would immediately be unsubscribed, as B is already the winner). However, one could argue that B is the incorrect case here, as race should just always subscribe to all sources.

I don't care whether case B or C is incorrect, I only think that it should not make a difference whether one of the sources completes synchronously or asynchronously.

Sidenote: I've added a bonus test-case (D). If you change the concat to concat(A, B, C, D), you can see that side-effect C-3 is subscribed after all side-effects for D, which sounds like a plus for C being the invalid case.

[cartant]

There is a bug in C. The empty observable should win the race and no further subscriptions should be made, as race

...mirrors the first source Observable to emit a next, error or complete notification.

This is definitely a bug, but the title and some of the comments in this issue are very confusing. I'll create another issue that better describes and bug and will then close this one. Unless creating the other issue is something that you really want to do yourself.

[mhverbakel]

Absolutely fine with me. Close it once you created the new one.

[cartant]

Closed in favour of #4808