Flow builder does not cooperate for cancellation

[pashko]

I am seeing some odd behaviour when canceling Flow collection. In the following example:

runBlocking {
    val flow = flow {
        while (isActive) {
            emit(Unit)
        }
    }
    val job = launch(Dispatchers.Default) {
        flow.collect {
            if (isActive) {
                println("active")
            } else {
                println("not active")
            }
        }
    }
    delay(100)
    job.cancelAndJoin()
}

I expect the collection as well as the flow builder to be canceled, but actually the builder continues producing values and the collector keeps accepting them (printing "not active"), and isActive in the builder returns true.
Now to make things weirder, if I remove the dispatcher from launch, the collector continues to print "active" even after cancelation, and if after that I add .flowOn(Dispatchers.Default) to the flow, everything works as expected.
Is this a bug or am I missing something in how Flow works?

[zach-klippenstein]

I believe this is a duplicate of #1265.

[elizarov]

It is not really a duplicate, because #1261 uses flowOn and that's going to be fixed soon. The code here runs into another problem. Look at this line:

val flow = flow {
        while (isActive) { // where's `isActive` coming from? 
            emit(Unit)
        }
    }

The answer is that it coming not from the flow { ... }, but from the surrounding runBlocking which is active. We have the following plan for the inspection to flag issues like that: https://youtrack.jetbrains.com/issue/KT-38033