Weird behavior of shareReplay with withLatestFrom

[begrs]

Hi, a sample code that produces my issue is here: https://stackblitz.com/edit/rxjs-mvugpe?file=index.ts

I have two observables as sources input1$ and input2$. Now when one of them is updated, some computation is triggered in the observable calculated$.
The calculated observable has two subscribers to its value, one of which is combined$. Combined$ also requires the input1$ and input2$ values for its computation (here just a console log).

Now one of the inputs is changed to 5. This causes calculated$ to be updated, which in turn updates secondSub and combined$.
What is very weird, is that in combined$ the OLD value of the changed input is used, not the one used for the calculation in calculated$.

However, if I move secondSub (line 12) after combined$, or remove it alltogether, the console long in combined$ actually produces the desired input with the CURRENT value of the changed input.


Removing the shareReplay(1) also fixes the issue - but that was kind of expected.

This behavior feels very weird to me, could someone explain why it behaves that way and what I could use instead, or confirm that it is an error?
(if the latter, I will open a new issue)

Thanks in advance

[voliva]

It has to do with the order of subscription.

withLatestFrom is meant to work with any observable, not only BehaviorSubjects and/or shareReplay()-ed observables. So the way it works is that it subscribes to the observables you give it as parameter, and then it will just append the latest value emited from them when a new value comes from source$.

On the first case, you subscribe to calculated$ before you subscribe to combined$. What happens here initially is calculated$ will subscribe to input1$ and input2$, and then you have secondSub subscribed to calculated$. The list of subscriptions is as follows:

// List of subscribers
input1$: [calculated$]
calculated$: [secondSub]

Then you subscribe to combined$. First withLatestFrom will subscribe to both inputs, then the source calculated$. Looking at the list of subscribers:

// List of subscribers
input1$: [calculated$, withLatestFrom]
calculated$: [secondSub, combined$]

So now when input1$ emits, it will synchronously call calculated$, which will synchronously call secondSub and combined$, which will hit withLatestFrom before it actually had the chance to receive the updated value from input1$, so it will synchronously emit the last value it got, which was the last. Then, input1$ will synchronously give the new value to withLatestFrom, but it's already late.

On the second case everything changes. First you subscribe to combined$, so withLatestFrom subscribes to both inputs before calculated$ makes its subscriptions:

// List of subscribers
input1$: [withLatestFrom, calculated$]
calculated$: [combined$]

And then you subscribe to calculated$, which in fact won't change the result now. Because now when input1$ emits, it will first call withLatestFrom, which will receive the new value, and then calculated$, which will emit a new value, and withLatestFrom will append the updated value.

Hopefully this made sense, sorry for the long post 😅 . I don't think it's to be considered an error on rxjs, but I think it's a general issue on rxjs that it's sometimes confusing, as you need to understand rxjs' internals (Observables and/or operators) to figure out these kind of things.

As for posible solutions, given that input1$ is a BehaviorSubject, you could just call .getValue() on them instead of using withLatestFrom().

If inputs weren't BehaviorSubject, then I think the alternative would be to change combined$ to be:

const combined$ = calculated$
  .pipe(
    observeOn(asyncScheduler),
    withLatestFrom(input1$, input2$),
    shareReplay(1)
  )

which basically delays the emissions from calculated$ to the next JS event loop, so that withLatestFrom can get those latest values.

[begrs]

@voliva Thanks so much for the detailed explanation, it was incredibly helpful.
I will try the proposed solution.

Hm I still would consider it a bug, but I get that changing something like that would probably break all existing code so nothing can be done about it. But nonetheless through your explanation I now feel comfortable opening an issue about it :)

Reasoning to open an issue:
For me the correct way of handling withLatestFrom and combineLatest would be to first subscribe to combineLatest observables, since even in other use cases, the combineLatest is kind of a combination of multiple observables as a source and should happen before any piped operator. In some of the use cases it does not make any difference, but would in my opinion be more consistent with how users think it will work.
The current behavior is as you said not obvious and can cause a lot of silent errors in a lot of places.
In my case the only reason I found it at all was a division by zero due to the old value.
The secondSub in my case is located in another typescript file and input1, input2 and calculated are part of a kind of module wide "store" to which all components have access and share their values by.
I know RX.js sometimes behaves that way, but it kind of breaks any separation of concerns approaches when the order of independent subscribers (secondSub and combined have nothing to do with each and have no direct interaction, they only use the same source value) causes different values in the value streams.

[gneu77]

Very good explanation by @voliva !
Here's how I would change the calculated$ pipe, to get the desired behavior:

const combined$ = calculated$
  .pipe(
    switchMap(c =>
      combineLatest([input1$, input2$]).pipe(map(([i1, i2]) => [c, i1, i2]))
    ),
    shareReplay(1)
  )

However, this would fail of course, as soon as calculated$ becomes async (e.g. just add a delay(1) to the pipe).
So the only generic way to make sure you always have the correct inputs for the result is to bundle it in the calculated$ pipe.