Non-conflating subscription count in MutableStateFlow and MutableSharedFlow

[qwwdfsad]

Problem statement

Currently, the infrastructure of Flow sharing is written with complete asynchrony and implementation unification in mind. While it benefits its usages in general, the implementation of sharing leaks StateFlow conflation into seemingly orthogonal concepts such as sharing strategies.

It can be demonstrated by the example from #2488:

val flow = flow {
    println("Flow is started") // Never printed
    emit(1)
}.stateIn(this, SharingStarted.Lazily, initialValue = 0)

assertEquals(expected = 0, actual = flow.first()) // Passes
yield()
assertNotEquals(illegal = 0, actual = flow.first()) // Fails

The root cause is sharing implementation -- any sharing strategy subscribes to the flow subscriptionCount and observes the subscriptions. When the collector subscribes and immediately (whether in the same dispatcher or in a different one, but the one that is lagging behind) unsubscribed, the original value is conflated to zero.

A more complex example that involves timings can be found in #2863.

Proposed solution

During the investigation, it was found that local changes cannot solve the asynchrony problem. While the targeted tweaks can address originally reported issues, the problem will be still unresolved and it will be possible to observe it in less trivial scenarios. It doesn't matter whether the sharing happens in the same thread (e.g. Dispatchers.Main) or in a separate one,
the timings and conflations can still be observed in a way that breaks the mental model of sharing.

Due to compatibility reasons, it's close to impossible to change the type of subscriptionCount and it's not the goal to replace trivially-readable (subscriptionCount.value) primitive with a brand new type.

Instead, we decided to make subscriptionCount non-conflating -- any subscriber will be able to observe all changes in subscribers number. While it is not aligned with how the rest of the state flows behave, it is the smallest potentially-breaking change that only affects implementors of sharing strategies and introduces small inconsistency, but provides clear mental model to the library users and does not introduce new concepts to the library

[bubenheimer]

@qwwdfsad: the kdoc documentation of this change currently appears misleading (kotlinx.coroutines release 1.7.3):

Implementation note: the resulting flow does not conflate subscription count.

This sounds as if the absence of conflation was not part of the contract for subscriptionCount, and just happened to be implemented this way at the present time and subject to change without notice.

Rather, my reading of this issue here suggests that non-conflation is intended to be part of the contract. Otherwise a consumer of this API would not validly be able to rely on this behavior.