Observables should not 'restart'

[jakearchibald]

I know you're going to hate me for this, but I'd like to re-litigate the concerns I had on blink-dev:

const ob = new Observable((subscriber) => {
  subscriber.next(1);

  setTimeout(() => {
    subscriber.next(2);
  }, 1000);

  setTimeout(() => {
    subscriber.next(3);
    subscriber.complete();
  }, 2000);
});

ob.toArray().then((vals) => {
  console.log(vals); // [1, 2, 3]
});

ob.toArray().then((vals) => {
  console.log(vals); // [2, 3]
});

setTimeout(() => {
  ob.toArray().then((vals) => {
    console.log(vals); // [3]
  });
}, 1500);

The model above appears to be:

• The SubscribeCallback is called synchronously the first time something subscribes to the observable.
• Additional subscriptions do not result in SubscribeCallback being called again
• Additional subscriptions may have missed values already emitted.

Even though this different to general observable behaviour, it makes sense to me. The fact that two synchronous subscriptions get different values is initially surprising, but it falls out of the synchronous nature of observables, and that's an essential part of the design.

However:

setTimeout(() => {
  ob.toArray().then((vals) => {
    console.log(vals); // [1, 2, 3]
  });
}, 2500);

At this point, the behaviour seems to flip. The SubscribeCallback will be called again, since it's no longer active.

This seems internally inconsistent, and unreliable from the outside. In observable terms, it's both 'hot' and 'cold'. IMO one of the following patterns should be chosen:

1. If values were emitted before you subscribed, tough luck, you missed them.
2. The SubscribeCallback will be called again so you get all values from the start.

…rather than it being 'it depends'. Option 1 seems the best, meaning in the above toArray() example, the fulfilled value should be [].

The current behaviour makes other things weird:

const ob = Observable.from([1, 2, 3]);

ob.toArray().then((vals) => {
  console.log(vals); // [1, 2, 3]
});

ob.toArray().then((vals) => {
  console.log(vals); // [1, 2, 3]
});

vs

const ob = Observable.from([1, 2, 3].values() /* ⬅️⬅️ */);

ob.toArray().then((vals) => {
  console.log(vals); // [1, 2, 3]
});

ob.toArray().then((vals) => {
  console.log(vals); // [] ⬅️⬅️
});

The first example exhibits the "get all the values from the start" behaviour. The second example tries to do this, but because the input is an iterator which doesn't support 'restarting', you get a seemingly inconsistent result.

When I raised this on blink-dev, folks said that my surprise at this was down to my inexperience with observables, and I accepted that. But, I've since talked to folks experienced in observables, and they seem surprised by this behaviour too.

Given that there's a single engine shipping this, and the above behaviour doesn't really come up when using element.when(…), I hope this can be changed without significant breakage.

@esprehn @bakkot @keithamus @acutmore what do you think? Am I way off here?

[mfp22]

Restarting is required for reactivity.

Observables are best for events, not states. Events happen whether anyone is listening or not.

However, event sources can be expensive. Websocket connections, for example. So you want to tear them down when nothing is subscribed, and reestablish them when a new subscriber arrives again.

In StateAdapt I use RxJS to drive state lifecycle. When components use a store, state is initialized and event sources are subscribed to. Work can be done at this point, like HTTP requests. When no more components are interested in the store, subscriptions to event sources are cleared, and HTTP requests are canceled. When new components subscribe again, fresh data is fetched. This has been an extremely convenient model. It prevents event sources from having to be imperatively controlled. (Imperative = opposite of reactive)

If you don't want observables to restart, you can always keep an extra subscription to hold their state. This is easy, but if observables behaved the other way, how would we make them reset without burdening every subscriber with this concern? A manual subscription next to the observable definition is just one very central command, whereas something above and beyond subscriptions to wake up the observable again would potentially require many commands scattered across code.

For observables to accurately declare event streams that can be appropriately reactive, in practice, efficiency requires knowing when subscriptions start, stop and then start again.

I'm surprised that people experienced with observables would be surprised by this behavior, and yet... most users of RxJS do not follow FRP closely. Not because they have tried it and don't like it, but because observables tend to be foisted on people by Angular or some other technology somebody chose for the team, and it's never done fully reactively. Most devs who have used RxJS have never seen proper FRP code. But if observables are for anything, they should support reactivity. What else is there? And all these promise APIs—I predict they will almost never be used, except by devs inexperienced with reactivity, trying to get back to familiar imperative territory. But observables should not be thought of as similar to promises at all. I really don't see the point of most of those APIs. That's kind of beside the point, but I do feel that it would be a tragedy to finally get the FRP primitive in JavaScript only to have major compromises to accommodate other mental models that already have plenty of APIs to support them. That would bite us especially down the road, because the overarching momentum in web dev has been towards more and more reactivity, and eventually a lot of devs will want something ootb that completely supports reactivity.

[keithamus]

In the I2S @domfarolino stated:

there are more surprises/quirks with the non-ref-counted approach

Perhaps Dom you can talk to this point more? I suppose one potential surprise would be re-instantiating event listeners or network fetches.

[jakearchibald]

I don't see this as a choice between ref counted and not ref counted. If anything, it feels like we have something that's a mix of the two.

[jakearchibald]

@mfp22 I feel you're talking about "pausing", which you would do in cases where subscribers unsubscribe. I'm talking about "complete".

[domfarolino]

Perhaps Dom you can talk to this point more? I suppose one potential surprise would be re-instantiating event listeners or network fetches.

Yeah, restarting a possibly-expensive operation on every subscription is a complaint we heard from a lot of developers, and one of the things they would fix about the Observable model if they could. So we thought we'd be threading the needle by allowing them to tap into pre-existing subscriptions, but also making them maximally useful by restarting dead subscriptions. I can understand how it's surprising though.

I'm worried about the inconsistencies we'd be introducing by still letting a subscriber passively tap into a pre-existing subscription, but NOT restarting dead subscriptions. Now, if someone hands you an Observable and you subscribe to it a little late, it's OK, you'll still receive values. In the non-restarting model, if you subscribe a little late, it's OK, but if you subscribe a little too late, then it's actually not OK. You're competing with other consumers of the same Observable, and might be tempted to subscribe prematurely and queue values that you might not need, just so you can know whether the Observable has completed by the time you actually want to subscribe.

We'd need to figure out what should happen when you subscribe to a completed/errored Observable. I don't think we can "do nothing", because that's totally opaque to the consumer. It turns delays between values into a scary gap: is it just a long delay? or is it a forever delay, because I missed the boat?? We could queue the error or completed signal, and replay it to any new subscribers, but that introduces queueing which is a little weird, and also punishes consumers for subscribing too late, which was never a problem in the RxJS model, or our current model. I'm not sure this is a thing Observable-using developers would expect.

I don't feel super duper strongly myself, especially since I'm not an Observables daily driver, so I'd like to hear more from the community.

[acutmore]

I'm surprised that people experienced with observables would be surprised by this behavior,

@jakearchibald showed me this yesterday and I was surprised because all of my previous Observables experience was with RxJS 4 and 5 - where they are cold and not shared. The last time I had checked on the proposal that was still the case and was unaware there had been a shift. So in my mind this behavior could be provided by the platform/a-library as an opt in share() - as that was how I learnt and taught RxJS when I worked with them.

Observables are best for events, not states. Events happen whether anyone is listening or not.

I agree, though would phrase this differently - Observables are best for events that happen whether anyone is listening or not. Like you say, button clicks and external events that can be subscribed to via a websocket.

For example setTimeout is not a state, so is it an event - the timeout could be seen as an event. Though a hot let FiveSec$ = ObservableLib.timer(5000) would be bad because each consumer would get a different delay depending if the subscription is currently live. Developers should know that they would always create a fresh observable if the event is an effect of the observer, i.e. let FiveSec$ = () => ObservableLib.timer(5000).

This does make me wonder if Observable.from(iterable) is a good thing to include - iterables do not tend to represent events that are happening even if there is no observer. Does it encourage the wrong mental model?

[domfarolino]

For example setTimeout is not a state, so is it an event - the timeout could be seen as an event.

But you can't consider setTimeout() an "event that happens whether anyone is listening or not" as you framed it, because the RxJS that you're familiar with would fire the setTimeout for every single subscriber, so implicitly there was always a listener—it was the subscriber the triggered it. Our current spec tries to preserve this by not making you end up in a situation where when you subscribe a little too late compared to other consumers, you accidentally end up with nothing, just because of your subscription timing.

[acutmore]

But you can't consider setTimeout() an "event that happens whether anyone is listening or not" as you framed it,

I was not clear. That is what I was saying. I was saying it is an event but not the type of event that hot-shared Observables are a good match for.

Many posts describing Observables talk about state vs events. I wish more things would emphases the type of events that observables are best at modelling - as I think that will help avoid the confusing cases that @jakearchibald's comment demonstrated - they happened because it was the wrong type of event.

[mfp22]

I'm definitely talking about completing. HTTP requests happen once and complete, should be shared, and should re-trigger when new subscribers arrive after complete. Timers too. I have relied on this behavior for 8 years. It's definitely necessary for reactivity. Unfortunately to keep my last comment shorter, I deleted a paragraph going over this in a lot more detail.

On types of events & cold observables: By default, I believe even cold observables should not be produced for every subscriber, and these events are not really different types. If 2 components activate, by default what they get from the same observable should be lazy and synchronized imo as if they were independent events (not produced for each subscriber). This is especially critical with HTTP requests, and the default behavior of RxJS has tripped up a lot of beginners (multiple requests sent). In StateAdapt, internally observables are shared and not subscribed to multiple times, effectively producing this behavior, and nobody has ever complained. But LOTS of people complain about multiple HTTP requests firing for RxJS.

And again, to get the opposite behavior, there's an escape hatch: You can wrap the resource in a factory function. This is acceptable for the rare scenarios it's needed, imo, but maybe there's a less disruptive possibility that has been discussed elsewhere - an anti-share operator or something.

[jakearchibald]

HTTP requests happen once and complete, should be shared, and should re-trigger when new subscribers arrive after complete

Yes to sharing, but absolutely no to re-triggering - that sets developers up for failure. In fact, until you mentioned fetching, I hadn't realised how bad this is in practice. Consider the following:

const ob = new Observable(async (subscriber) => {
  const controller = new AbortController();
  subscriber.addTeardown(() => controller.abort());

  for await (const item of fetchObjects(controller.signal)) {
    subscriber.next(item);
  }

  subscriber.complete();
});

// Sometime later (1)
ob.subscribe(…);

// Sometime later still (2)
setTimeout(() => {
  ob.subscribe(…);
}, 1000);

In a fast connection environment, such as a development server, both subscriptions will start from the start, because (1) completes before (2) subscribes. It fools the developer into thinking that this follows the 'cold' pattern.

The developer might even see the same behaviour after they've pushed the code live, because they're lucky enough to be on a speedy connection.

But, then they start getting weird bug reports, because, due to slower connectivity speeds of less fortunate users, (2) seemingly out of nowhere gets the 'hot' behaviour, and therefore doesn't get data from the start.

This is a footgun. We should not be shipping APIs that hide their complexity in dev environments.

I'm not proposing the 'cold' pattern. I'm saying we should go all-in on the 'hot' pattern. As in, if (2) subscribes after subscriber.complete() is called, their subscription is immediately ended.

With that pattern, developers will realise, in dev mode, that their code depends on a fresh fetch each time, or some sort of buffering.

@domfarolino do you find this compelling at all?

[bakkot]

I am extremely surprised to learn that Observables in this spec can implicitly transition from being in a terminated state to not being in a terminated state and I really hope that can be changed. That makes the state machine for an Observable very confusing. (Sorry for not noticing this earlier, and thanks @jakearchibald for the ping.)

I think it's reasonable to have the default RxJS behavior where Observables themselves are not ever in any particular state, and rather each individual subscription runs through a "not started -> running -> terminated" state machine. That's a very straightforward state machine.

And I think it would be reasonable to have instead the Observable keep track of state, such that it goes through a "not started -> running -> terminated" state machine, and subscribers simply tap in to whatever state it's in when they subscribe (with the caveat that the first one triggers the not started -> running transition, which is a little surprising the first time you use Observables). That again is very straightforward.

But I don't think it is reasonable to say that Observables are keeping track of state, but also that they can implicitly transition back from terminated to running by someone external adding a new subscriber after the observable has explicitly called .complete(). I don't think anyone would expect that transition in the state machine. I checked with a couple acquaintances who have some experience with Observables and they were also very surprised.

I think that if you want to have the Observable keep track of state, which is totally reasonable, then transitions back from terminated to running need to be explicit, not implicit. It would be sensible to have .restart method or something, which puts it back in the state where future subscribers run the SubscribeCallback. It's just not sensible to do that transition implicitly.

I'm worried about the inconsistencies we'd be introducing by still letting a subscriber passively tap into a pre-existing subscription, but NOT restarting dead subscriptions. Now, if someone hands you an Observable and you subscribe to it a little late, it's OK, you'll still receive values. In the non-restarting model, if you subscribe a little late, it's OK, but if you subscribe a little too late, then it's actually not OK. You're competing with other consumers of the same Observable, and might be tempted to subscribe prematurely and queue values that you might not need, just so you can know whether the Observable has completed by the time you actually want to subscribe.

This is a confusing paragraph to read from the "observables represent streams of events" point of view. If you subscribe after the stream of events has completed, then you should of course not expect to get values! If you want to see events as they happen, you need to be subscribed before those events happen.

[jakearchibald]

One consequence of not restarting:

const ob = element.when('pointermove');

// 1
ob.first().then((event) => {
  console.log('Next pointer move event:', event);
});

setTimeout(() => {
  // 2
  ob.first().then((event) => {
    console.log('Next pointer move event:', event);
  });
}, 1000);

One of two things will happen:

• If there's a pointermove event before 2: You get one log for the event, and the second either resolves with undefined, or rejects, depending on the behaviour of subscribing to a 'complete' observer.
• Otherwise: You get two logs for the same event.

This isn't great either. I think this is because the current design treats "no more subscribers" and "complete" as the same thing, but they really aren't.

Going from "no more subscribers" to "new subscribers" depends on being able to pause and resume. Going from "complete" to "restart" is a different thing.

Maybe we could have:

const ob = new Observable(async (subscriber) => {
  subscriber.addPauseResume(pauseCallback, resumeCallback);
});

This would allow observables that represent fetches to abort the fetch, but maintain state so the fetch can be restarted with a range request. For events, it would be:

const ob = new Observable((subscriber) => {
  let controller;

  const addListener = () => {
    controller = new AbortController();
    element.addEventListener('pointermove', (event) => {
      subscriber.next(event);
    });
  }

  subscriber.addPauseResume(
    () => controller.abort(),
    addListener
  );

  addListener();
});

For "complete" to "restart", @bakkot's restart() method sounds good.

---

An alternative pause/resume design:

const ob = new Observable(async (subscriber, pausedState) => {
  // Set up your observable here.
  // Use pausedState to 'resume' the observable.

  // Then:
  subscriber.addTeardown(() => {
    // Clean up your observable here.
    // Then, return a non-undefined value to indicate that this observable can be resumed.
    // If there's a future subscription,
    // the returned value will be passed to the next call of SubscribeCallback as pausedState.
    return {…};
  });
});

Note that even with this, 'complete' observables cannot be restart without some clear opt-in, like @bakkot's restart() method.

This nice(?) thing about the above API, is it keeps things simple for cases where re-running the SubscribeCallback is enough for resuming. The listener example from before would become:

const ob = new Observable((subscriber) => {
  const controller = new AbortController();

  element.addEventListener('pointermove', (event) => {
    subscriber.next(event);
  });

  subscriber.addTeardown(() => {
    controller.abort();
    // To mark this as resumable:
    return true;
  });
});

true is used simply because it's not-undefined. pausedState isn't needed in this particular case.

[jridgewell]

I'm worried about the inconsistencies we'd be introducing by still letting a subscriber passively tap into a pre-existing subscription, but NOT restarting dead subscriptions.

I think this creates inconsistencies, it doesn't solve them. Jake's examples are excellent, it's utterly surprising that callbacks are rerun a second time. Whether we get the hot/cold behavior depending on if there is any current subscription is more surprising than if there has ever been a subscription.

It's made even better by the Iterable example, which cannot be restarted. An Iterable is always hot, there's no way to for us to restart or rewind.

We'd need to figure out what should happen when you subscribe to a completed/errored Observable

Just make this explicit in the callback. Allow the setup to have a pause/resume that makes the resumable code obvious. If we need to close/reestablish a connection to the websocket because it's too expensive to keep open, then that can be done easily.

The intermediate observers from .map/.filter/etc calls can handle this automatically for developers. Same with the DOM APIs. But for the developer written callback setups, it'll make it much easier to reason about.

[mfp22]

@jakearchibald

Okay, your example is good and there actually is something important here.

Please excuse my passion—I have been in the trenches with observables for a long time.

It turns out I have similar concerns, but went with a different solution. I have implemented it twice. The I way I do this is I take an RxJS HTTP observable and merge it with NEVER. That way it never completes on its own. It only completes when subscribers are all gone. That's in both StateAdapt and in this custom operator.

Of course this would break concat and any of the promise APIs.

But I only merge with NEVER when I'm converting from an event stream to a state. It's really the complete callback I want to avoid, because yes when all subscribers unsubscribe I love canceled HTTP requests, but also finalize/finally runs to clean up the state (and any other resources for when it was in-demand).

This is fundamentally a different use case than an observable notifying me when it has nothing left for me.

And yet, when all subscribers leave and come back, it is critical for reactivity for it to restart automatically, for the reasons mentioned in previous comments.

@bakkot

It would not be reasonable for it to have a restart method. That requires imperative code to manage. That makes reactivity impossible, which is the fundamental reason for observables to exist at all.

If you subscribe after the stream of events has completed, then you should of course not expect to get values!

If you want to declare/represent a data source, it will be stale and require imperative code unless it can use its subscription information to efficiently produce events.

@jakearchibald

This isn't great either. I think this is because the current design treats "no more subscribers" and "complete" as the same thing, but they really aren't.

THIS!!!!!!!!!

I would love a separate operator for handling when an observable self-completes vs when it becomes unused. The fact that these have been conflated and handled with the same mechanism is because they should be handled by the same type of mechanism. Automatic restart is essential for reactivity!