Fix Flow.timeout swallowing the channel closure exception (#4072)

Fixes #4071

Author: dkhalanskyjb

kotlinx-coroutines-core/common/src/flow/operators/Delay.kt

@@ -394,6 +394,7 @@ private fun <T> Flow<T>.timeoutInternal(
             value.onSuccess {
                 downStream.emit(it)
             }.onClosed {
+                it?.let { throw it }
                 return@onReceiveCatching false
             }
             return@onReceiveCatching true

kotlinx-coroutines-core/common/test/flow/operators/TimeoutTest.kt

@@ -237,6 +237,17 @@ class TimeoutTest : TestBase() {
         testImmediateTimeout(-1.seconds)
     }
 
+    @Test
+    fun testClosing() = runTest {
+        assertFailsWith<TestException> {
+            channelFlow<Int> { close(TestException()) }
+                .timeout(Duration.INFINITE)
+                .collect {
+                    expectUnreached()
+                }
+        }
+    }
+
     private fun testImmediateTimeout(timeout: Duration) {
         expect(1)
         val flow = emptyFlow<Int>().timeout(timeout)