Fix various dispatcher issues

packages/cinterop/src/nativeDarwin/kotlin/io/realm/kotlin/internal/interop/RealmInterop.kt

@@ -3595,32 +3595,59 @@ actual object RealmInterop {
         private val scope: CoroutineScope = CoroutineScope(dispatcher)
         val ref: CPointer<out CPointed> = StableRef.create(this).asCPointer()
         private lateinit var scheduler: CPointer<realm_scheduler_t>
-        private val lock = SynchronizableObject()
+        private val schedulerLock = SynchronizableObject()
+        private val dispatcherLock = SynchronizableObject()
         private var cancelled = false
+        private var dispatcherClosing = false
 
         fun setScheduler(scheduler: CPointer<realm_scheduler_t>) {
             this.scheduler = scheduler
         }
 
         override fun notify(work_queue: CPointer<realm_work_queue_t>?) {
-            scope.launch {
-                try {
-                    printlntid("on dispatcher")
-                    lock.withLock {
-                        if (!cancelled) {
-                            realm_wrapper.realm_scheduler_perform_work(work_queue)
+            // Use a lock as a work-around for https://github.com/realm/realm-kotlin/issues/1608
+            //
+            // As the Core listeners are separated from Coroutines, there is a chance
+            // that we have closed the Kotlin dispatcher and scheduler while Core is in the
+            // process of sending notifications. If this happens we might end up in this
+            // `notify` method with the dispatcher and scheduler being closed.
+            //
+            // As the ClosableDispatcher does not expose a `isClosed` state, it means
+            // there is no way for us to detect if it is safe to launch a task using
+            // the current coroutine APIs.
+            //
+            // Ass a work-around we use the `canceled` flag that is being set when the Scheduler
+            // is being released. This should be safe as we are only closing the dispatcher when
+            // releasing the scheduler. See [io.realm.kotlin.internal.util.LiveRealmContext] for
+            // the logic around this.
+            //
+            // Note, JVM and Native behave differently on this. See this issue for more
+            // details: https://github.com/Kotlin/kotlinx.coroutines/issues/3993
+            dispatcherLock.withLock {
+                if (!dispatcherClosing) {
+                    scope.launch {
+                        try {
+                            printlntid("on dispatcher")
+                            schedulerLock.withLock {
+                                if (!cancelled) {
+                                    realm_wrapper.realm_scheduler_perform_work(work_queue)
+                                }
+                            }
+                        } catch (e: Exception) {
+                            // Should never happen, but is included for development to get some indicators
+                            // on errors instead of silent crashes.
+                            e.printStackTrace()
                         }
                     }
-                } catch (e: Exception) {
-                    // Should never happen, but is included for development to get some indicators
-                    // on errors instead of silent crashes.
-                    e.printStackTrace()
                 }
             }
         }
 
         fun cancel() {
-            lock.withLock {
+            dispatcherLock.withLock {
+                dispatcherClosing = true
+            }
+            schedulerLock.withLock {
                 cancelled = true
             }
         }

packages/library-base/src/commonMain/kotlin/io/realm/kotlin/internal/util/CoroutineDispatcherFactory.kt

@@ -120,6 +120,10 @@ public class LiveRealmContext(
     }
 
     override fun close() {
+        // Warning: It is important to release the scheduler before closing the
+        // dispatcher. Failing to do it this way might create race conditions on Darwin.
+        // See SingleThreadDispatcherScheduler in RealmInterop for the Darwin
+        // source set.
         scheduler.release()
         dispatcherHolder.close()
     }

packages/test-base/src/commonTest/kotlin/io/realm/kotlin/test/common/notifications/RealmNotificationsTests.kt

@@ -108,16 +108,16 @@ class RealmNotificationsTests : FlowableTests {
         realm.write { /* Do nothing */ }
         val nextVersion = realm.version()
 
+        c1.receiveOrFail(message = "Failed to receive update event on Channel 1").let { realmChange ->
+            assertIs<UpdatedRealm<Realm>>(realmChange)
+            assertEquals(nextVersion, realmChange.realm.version())
+        }
+
         val observer2 = async {
             realm.asFlow().collect {
                 c2.send(it)
             }
         }
-
-        c1.receiveOrFail(message = "Failed to receive update event on Channel 1").let { realmChange ->
-            assertIs<UpdatedRealm<Realm>>(realmChange)
-            assertEquals(nextVersion, realmChange.realm.version())
-        }
         c2.receiveOrFail(message = "Failed to receive initial event on Channel 2").let { realmChange ->
             assertIs<InitialRealm<Realm>>(realmChange)
             assertEquals(nextVersion, realmChange.realm.version())

packages/test-base/src/jvmTest/kotlin/io/realm/kotlin/test/jvm/RealmTests.kt

@@ -29,11 +29,15 @@ import kotlinx.coroutines.CloseableCoroutineDispatcher
 import kotlinx.coroutines.async
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withTimeout
+import kotlin.random.Random
+import kotlin.random.nextUInt
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertTrue
+import kotlin.test.fail
 import kotlin.time.Duration.Companion.seconds
 
 fun totalThreadCount() = Thread.getAllStackTraces().size
@@ -43,6 +47,18 @@ fun totalThreadCount() = Thread.getAllStackTraces().size
  */
 class RealmTests {
 
+    // Test for https://github.com/Kotlin/kotlinx.coroutines/issues/3993
+    @Test
+    fun submittingToClosedDispatcherIsANoop() {
+        val dispatcher = singleThreadDispatcher("test-${Random.nextUInt()}")
+        dispatcher.close()
+        runBlocking {
+            launch(dispatcher) {
+                fail("Dispatcher was running")
+            }
+        }
+    }
+
     @Test
     fun cleanupDispatcherThreadsOnClose() = runBlocking {
         val tmpDir = PlatformUtils.createTempDir()

packages/test-base/src/nativeDarwinTest/kotlin/io/realm/kotlin/test/darwin/CoroutineTests.kt

@@ -20,20 +20,27 @@ import io.realm.kotlin.internal.platform.singleThreadDispatcher
 import io.realm.kotlin.test.platform.NsQueueDispatcher
 import io.realm.kotlin.test.platform.PlatformUtils
 import io.realm.kotlin.test.util.Utils.printlntid
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.async
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.newSingleThreadContext
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.test.runTest
 import platform.CoreFoundation.CFRunLoopRun
 import platform.Foundation.NSNumber
 import platform.darwin.DISPATCH_QUEUE_PRIORITY_BACKGROUND
 import platform.darwin.dispatch_get_global_queue
+import kotlin.random.Random
+import kotlin.random.nextUInt
 import kotlin.test.Ignore
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+import kotlin.test.fail
 
 /**
  * Various coroutine tests to track if basic dispatching, etc. works.
@@ -89,4 +96,22 @@ class CoroutineTests {
             assertEquals(tid, PlatformUtils.threadId())
         }
     }
+
+    // Test for https://github.com/Kotlin/kotlinx.coroutines/issues/3993
+    @Test
+    fun closingDispatchersThrowIllegalState() {
+        val dispatcher = singleThreadDispatcher("test-${Random.nextUInt()}")
+        dispatcher.close()
+        try {
+            runBlocking {
+                launch(dispatcher) {
+                    fail("Dispatcher was running")
+                }
+            }
+            fail("No error was thrown")
+        } catch (ex: IllegalStateException) {
+            assertFalse(ex is CancellationException, "Was: ${ex::class}")
+            assertTrue(ex.message!!.contains("was closed, attempted to schedule"), ex.message)
+        }
+    }
 }