Run wall-clock timeout in GlobalScope

This ensures its use of the Default dispatcher does not affect the use of a TestScheduler and its fake time.

src/commonMain/kotlin/app/cash/turbine/channel.kt

@@ -20,8 +20,10 @@ import kotlin.time.Duration
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.async
 import kotlinx.coroutines.channels.ChannelResult
@@ -115,7 +117,11 @@ private suspend fun <T> withWallclockTimeout(
   block: suspend CoroutineScope.() -> T,
 ): T = coroutineScope {
   val blockDeferred = async(start = CoroutineStart.UNDISPATCHED, block = block)
-  val timeoutJob = launch(Dispatchers.Default) { delay(timeout) }
+
+  // Run the timeout on a scope separate from the caller. This ensures that the use of the
+  // Default dispatcher does not affect the use of a TestScheduler and its fake time.
+  @OptIn(DelicateCoroutinesApi::class)
+  val timeoutJob = GlobalScope.launch(Dispatchers.Default) { delay(timeout) }
 
   select {
     blockDeferred.onAwait { result ->

src/commonTest/kotlin/app/cash/turbine/FlowTest.kt

@@ -44,6 +44,8 @@ import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onCompletion
 import kotlinx.coroutines.flow.onStart
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runCurrent
 import kotlinx.coroutines.test.runTest
 import kotlinx.coroutines.withContext
 import kotlinx.coroutines.yield
@@ -656,4 +658,30 @@ class FlowTest {
       assertEquals("Expected 3 items for two item channel but got 2 items and Complete", message)
     }
   }
+
+  @Test
+  fun virtualTimeCanBeControlled() = runTest {
+    flow {
+      delay(5000)
+      emit("1")
+      delay(5000)
+      emit("2")
+    }.test {
+      expectNoEvents()
+
+      advanceTimeBy(5000)
+      expectNoEvents()
+
+      runCurrent()
+      assertEquals("1", awaitItem())
+
+      advanceTimeBy(5000)
+      expectNoEvents()
+
+      runCurrent()
+      assertEquals("2", awaitItem())
+
+      awaitComplete()
+    }
+  }
 }