Weird Race condition with Channels and ViewModel/LiveData

dependencies.gradle

ext.kotlin_version = '1.3.41'
dependencies {
    implementation fileTree(dir: 'libs', include: ['*.jar'])
    implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk7:$kotlin_version"
    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.2.2'
    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.2.2'

    implementation 'androidx.core:core-ktx:1.1.0-rc02'
    implementation 'androidx.appcompat:appcompat:1.1.0-rc01'
    implementation 'androidx.activity:activity-ktx:1.1.0-alpha01'

    implementation 'androidx.lifecycle:lifecycle-livedata-ktx:2.1.0-rc01'
    implementation 'androidx.lifecycle:lifecycle-viewmodel-ktx:2.1.0-rc01'
    implementation 'androidx.lifecycle:lifecycle-extensions:2.1.0-rc01'
}

MainActivity.kt

class MainActivity : AppCompatActivity(R.layout.activity_main) {

    private val viewModel by viewModels<MainViewModel>()


    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)

        viewModel.liveData.observe(this, Observer {
            Log.d("MAIN", "Display $it in view")
            textView.text = it
        })

        button.setOnClickListener { viewModel.onClick() }
    }
}

class MainViewModel : ViewModel() {

    private val chan = ConflatedBroadcastChannel(Counter(0, "loaded"))

    val liveData = chan.toLiveData(viewModelScope) {
        Log.d("MAIN", "sending $it to view")
        "counter: $it"
    }

    init {
        viewModelScope.launch(Dispatchers.IO) {
            var lastCounterCount: Int? = null
            var job: Job? = null
            chan.consumeEach {
                Log.d("MAIN", "New counter in worker $it")
                if (it.counter != lastCounterCount) {

                    lastCounterCount = it.counter
                    job?.cancel()

                    Log.d("MAIN", "new count, new job")

                    job = viewModelScope.launch(Dispatchers.IO) {
                        Log.d("MAIN", "loading for ${it.counter}")

                        chan.send(chan.value.copy(state = "loading"))
                        delay(2000)

                        Log.d("MAIN", "finish loading for ${it.counter}")

                        chan.send(chan.value.copy(state = "loaded"))
                    }
                    job?.invokeOnCompletion { Log.d("MAIN", "job cancelled $it") }
                } else {
                    Log.d("MAIN", "Same count")
                }
            }
        }
    }

    fun onClick() {
        viewModelScope.launch(Dispatchers.IO) {
            val currentCounter = chan.value
            chan.send(currentCounter.copy(counter = currentCounter.counter + 1))
        }
    }
}

data class Counter(val counter: Int, val state: String)

fun <I, O> ConflatedBroadcastChannel<I>.toLiveData(
    scope: CoroutineScope,
    block: suspend (I) -> O
): LiveData<O> {
    var currentSubscription: ReceiveChannel<I>? = null
    return object : MutableLiveData<O>() {
        override fun onActive() {
            currentSubscription = openSubscription()
            scope.launch(Dispatchers.IO) {
                currentSubscription?.consumeEach {
                    postValue(block(it))
                }
            }.invokeOnCompletion { currentSubscription?.cancel() }
        }

        override fun onInactive() {
            currentSubscription?.cancel()
            currentSubscription = null
        }
    }
}