Changing a MutableState inside a coroutine seemingly leads to skipped emissions

[julioromano]

I'm incrementing an initially 0 MutableState<Int> twice in a row:

1. If done outside of a coroutine it works as expected and myPresenter() emits [0,1,2].
2. If done inside of a newly launched coroutine it emits only [0,2] (skipping the "1" state) .

Is this a bug or working as intended?
And if it is WAI, how come? I naively thought that in both cases I should expect my presenter to emit [0,1,2].

(running Kotlin 1.8.21, kotlinx.coroutines 1.7.1, compose-runtime 1.4.3, molecule 0.9.0, turbine 0.13.0)

package example.test

import androidx.compose.runtime.Composable
import androidx.compose.runtime.MutableState
import androidx.compose.runtime.mutableStateOf
import androidx.compose.runtime.remember
import androidx.compose.runtime.rememberCoroutineScope
import app.cash.molecule.RecompositionClock
import app.cash.molecule.moleculeFlow
import app.cash.turbine.test
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.runTest
import org.junit.Test
import kotlin.test.assertEquals

data class MyState(
    val anInt: Int,
    val eventSink: (MyEvent) -> Unit,
)

sealed interface MyEvent {
    object Increment : MyEvent
    object IncrementSuspending : MyEvent
}

@Composable fun myPresenter(): MyState {
    val scope = rememberCoroutineScope()
    val anInt: MutableState<Int> = remember { mutableStateOf(0) }
    return MyState(anInt.value) {
        when (it) {
            MyEvent.Increment -> {
                anInt.value++
                anInt.value++
            }
            MyEvent.IncrementSuspending -> scope.launch {
                anInt.value++
                anInt.value++
            }
        }
    }
}

class MoleculeTestCase {
    @Test fun `process Increment event`() = runTest {
        moleculeFlow(RecompositionClock.Immediate) { myPresenter() }.test {
            awaitItem().apply {
                assertEquals(0, anInt)
                eventSink(MyEvent.Increment)
            }
            assertEquals(1, awaitItem().anInt)
            assertEquals(2, awaitItem().anInt)
        }
    }

    @Test fun `process IncrementSuspending event`() = runTest {
        moleculeFlow(RecompositionClock.Immediate) { myPresenter() }.test {
            awaitItem().apply {
                assertEquals(0, anInt)
                eventSink(MyEvent.IncrementSuspending)
            }
            assertEquals(2, awaitItem().anInt)
        }
    }
}

[ganfra]

@JakeWharton any idea about that?

[jingibus]

The immediate clock is a best-effort tool: we can trigger recomposition when the state is dirtied, but it is not possible to guarantee that a recomposition will occur each time the state is dirtied.

Honestly, that MyEvent.Increment causes two recompositions is more surprising to me here! I would prefer not to, but obviously unit testing Circuit-style presenters does expose you to it...

[julioromano]

Adding a delay(1) in IncrementSuspending like:

MyEvent.IncrementSuspending -> scope.launch {
    anInt.value++
    delay(1) // delay(0) or yield() won't do the trick!
    anInt.value++
}

Makes it emit 3 items too.

Are you confident it is a limitation of GatedFrameClock or are we hitting some opaque behaviors of coroutines?

[jingibus]

Are you confident it is a limitation of GatedFrameClock or are we hitting some opaque behaviors of coroutines?

Speaking as someone who wrote GatedFrameClock and maintains Turbine, this is specifically caused by Turbine's internal usage of the Unconfined dispatcher. If we got rid of that, two increments in a row without an intervening suspension would consistently produce only one item (which is what I would expect to be the case: coalescing multiple state changes into one emission is an important thing Compose does for us)