Streams should emit in a shared transaction when starting

[raquo]

Problem description

Typically when a stream starts, it does not emit any events immediately, this happens later. For example, parentStream.map(...) does not emit anything on start (unless parentStream does), FetchStream.get(...) needs to wait for the response, etc.

However, some streams do emit events when they're starting (i.e. when they are getting their first subscriber), and that can cause arguably surprising behaviour. I don't remember running into that, recently at least, but here is a trivial example in Laminar:

val container = dom.document.getElementById("app-container")
val stream = EventStream.fromValue(1)
render(
  container,
  div(
    child.text <-- stream,
    child.text <-- stream.map(_ * 10)
  )
)

(scribble)

You might expect the div to contain 1 and 10, but actually it will only contain 1. Why: because when the div is mounted, its subscriptions will be activated sequentially, one by one:

• First, child.text <-- stream will activate, this will start stream, and the stream will emit an event (1) in a new transaction. However, when this happened, we were not inside any other transaction, so the event has nothing to wait for, and will propagate immediately, being sent to child.text.

• Then, child.text <-- stream.map(_ * 10) will activate. The stream.map(_ * 10) stream will be started, and it will be added to the list of listeners on stream, but nothing else will happen, because stream is already started at this point, and the event that it emitted on start (1) has already been fully propagated – before this second subscription was activated. So, the stream stream.map(_ * 10) will not receive any events, and will not emit any events, and this second child.text will miss the event.

If stream was instead a signal, everything would have worked as you'd expect, because signals remember their current value, but streams by design have no concept of current value.

Importance

I think this is undesirable for a couple reasons:

1. It's unintuitive and not obvious. Looking at the code, you wouldn't expect it. You would get the expected behaviour if you changed val stream to def stream, creating two identical but independent streams, but that's not the kind of finagling I want Laminar users to need to do.

2. Running this same code inside a transaction (e.g. rendering this div inside child <-- someOtherStream.map(...)) will produce different results: in that case, since the 1 event is emitted in a new transaction, that transaction will wait for the currently ongoing transaction to finish, and by the time this happens, both of the child.text subscriptions will have be added, so both of them will process that event.

Scope

EventStream.fromValue(1) is just the most obvious case. Any stream that emits events immediately when it's starting is affected by this problem. Here's an exhaustive list of such streams, I think:

• CustomSource: any user-defined streams using this API can emit on start if the user chose so (these streams can also emit an error on start if the user's onStart function throws)
• EventStream.fromValue(1) and various aliases to it: by design
• PeriodicStream: calls tick() on start, which emits its current value
• x.flatten or x.flatMap Flattened / Flatmapped streams:
  • When using ConcurrentStreamStrategy manually, the stream emits an error on start when its parent is a signal in an error state
  • When flattening an EventStream[Signal[A]], the resulting SwitchSignalStream emits the signal's updated value on start IF its value was updated while the stream was stopped. Much like signal.changes, except... well, more on that below.
• Signal.changes: emits a new value if the signal has changed its value while the stream was stopped (As of Airstream v15)

Proposed solution

I mentioned signal.changes, but that one is different than the others, because in Airstream v15 I added special logic to it to handle something that is very similar to this problem, but is about re-starting those streams, not starting them for the first time. See the docs explaining that problem in this section. Relevant quote:

To avoid [the problem I mentioned], we have a special mechanism [for signal.changes] that lets us batch simultaneous events in a new transaction when restarting observables. Currently, it's only used to restart signal.changes. Basically, to avoid this restarting glitch, you want to wrap all your simultaneous observer additions into Transaction.onStart.shared { /* code here */ }, so any signal.changes you restart within that block will all emit in the same transaction. This is also not a perfect match for "normal" Airstream behaviour, and could potentially cause the other kind of FRP glitch where an intermediary event that you do expect to happen is swallowed by the system instead. However, of the two evils I think this is a lesser one, because it's much less common for that to be a problem. Ideally I would like to find a more robust mechanism for this edge case, but currently I lack the time required to do the research.

Airstream's DynamicOwner uses this onStart.shared mechanism when activating all of its subscriptions, and all Laminar's methods like Tag.apply and amend do the same when applying multiple modifiers at a time, so you really shouldn't ever need to use onStart.shared manually, unless you're an advanced user creating your own custom modifiers or ownership primitives.

So, I think all of those streams listed above, that emit events when staring, need the same kind of treatment that I gave to signal.changes in Airstream v15 – the code in their onStart methods that creates a new transaction immediately, needs to instead schedule the new transaction to be executed at the end of the current onStart.shared block.

Side effects

The proposed solution will result in our example executing as I would expect, the div containing both 1 and 10, that's good.

But it will also bring a more fundamental change: essentially, ALL of the events that are emitted like this – on start, or in Laminar, on mount – will now be treated as having had happened "at the same time", i.e. in the same transaction. On the surface, it seems like the right way to think about those events, since they all indeed had only one underlying event triggering them – the mounting – but the potential problem is, you normally wouldn't expect something like stream and stream.flatMap(...) to emit in the same transaction, regardless of what's inside ..., but now, they could. It's a similar problem to what signal.changes, but that one is a bit more niche, because it only applies to **re-**starting that stream, whereas this will now apply to more types of streams, and also when starting them for the first time, not just re-starting.

As I work on this, I will come up with a couple test cases demonstrating the side effect, it's a bit too involved to do right now. I hope that this will not be a significant problem.

What now

I'd like to know if you have encountered the problematic behaviour, and how surprising / annoying it was. For myself, I don't remember running into it, as I tend to use signals for anything that remotely resembles state, keeping streams only for things like user clicks for the most part, and those types of events just don't tend to happen on start / on mount.

I haven't started implementing this change yet, but I have a good idea of what needs to be done. If it actually works the way I think it will, I think the fix will be binary compatible, and I will release Airstream v16.0.1-M1, which you'll hopefully be able to try without updating any other dependencies. I don't want to publish a non-milestone 16.0.1 version because I don't want Scala Steward or other tooling suggesting it as a trivial update. I think "M1" should be enough to dissuade people anyway? If all is well then I'll eventually bring these changes into v17 a few months from now, but I do want a "preview" release that has nothing except these changes, to fish out any issues early on.

[yurique]

I don't remember having problems related to this, either.
But I'm usually only using .fromValue(..) to "pre-start" some other stream of events (like in EventStream.merge( fromValue( ... ), buttonClicksStream )).

[raquo]

I think I fixed this. The fix is pretty simple, but I had to arrive at it the hard way, after trying out increasingly complicated mechanisms that did not work.

The suffering

Initially I was operating under an assumption that I want or need to fire events that are emitted onStart by all of the streams listed in the Scope section in a shared transaction, similarly to how we handle events emitted onStart by signal.changes. This turned out to be both undesirable and seemingly impossible to get right.

It's hard to get right due to weird cases like EventStrean.fromSeq that emit multiple events onStart. The second and subsequent events can not be emitted in the shared transaction, because an observable can only emit once in any given transaction. However, I wasn't able to figure out how to schedule the execution of those subsequent transactions in a universally desirable and predictable order. Interactions with other features like flattening, and SyncObservable-s make me think that this is in fact not possible to do safely in the general sense, and that fromSeq could be only a private case of a bigger, problem that is unsolvable in principle, for the same reasons that we need transactions in the first place.

But this is all moot, because the behaviour I was trying to achieve is actually undesirable. I realized that signal.changes is different from all other streams in a rather fundamental way – its output does not create events from thin air, it gets them from a parent observable, and does not normally create its own transactions. All other streams in the Scope list are the opposite – they create their own events [1], and always fire them in new transactions. Because of that, there is no reason to shove all those events (that are normally fired in separate transactions) into one shared transaction – all we need to do is to delay the execution of those transactions until the Transaction.onStart.shared callback is finished, i.e. (in practical terms as it relates to this ticket's stated problem) until all subscribers have been added.

[1] Except the flatten ones, I will have to think about those again.

The actual solution

So, that's what I did. Now, any transaction created inside Transaction.onStart.shared is added to a separate postStartTransactions queue. After everything else is done, but before the shared block relinquishes control, each of those transactions is scheduled for execution after the current (shared) transaction, in the same order.

I also fixed a bug where multiple signal.changes would fire in the wrong order in the shared onStart transaction under certain conditions, and a bug where signal.changes wouldn't fire at all on start (not sure if that one was possible to encounter with typical Laminar usage, but it was possible in manual Airstream usage).

Try it out

I published Airstream 17.0.0-M1 with this fix. It should be binary compatible with Laminar & Airstream 16.0.0. If you are using Laminar 16.0.0, please add "com.raquo" %%% "airstream" % "17.0.0-M1" to your dependencies, and see if anything breaks (primary scope: elements with streams mentioned in Scope getting mounted). Everything should work fine, unless you're implicitly relying on this bug, which is unlikely.

That said, there could be slight differences in the order of events fired when mounting elements. For example, now signal.changes will emit before all other transactions such as EventStream.fromSeq, whereas previously it would depend on their relative order. (Remember, this is only when starting / re-starting multiple observables at the same time, e.g. when mounting an element that contains several subscriptions).

If you try it out, please subscribe to this issue to get notified of any problems reported by other users.