Fix various dispatcher issues

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

@@ -3603,18 +3603,41 @@ actual object RealmInterop {
         }
 
         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)
+            try {
+                scope.launch {
+                    try {
+                        printlntid("on dispatcher")
+                        lock.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()
+                }
+            } catch (ex: IllegalStateException) {
+                // 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 while Core is in the process
+                // of sending notifications. If this happens we might end up in this
+                // `notify` method with the dispatcher 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
+                // it. So as a last resort we are catching the error here and ignore it.
+                //
+                // This should be safe as we are only closing the dispatcher when closing the
+                // Realm anyway.
+                //
+                // Note, JVM and Native behave differently on this. See this issue for more
+                // details: https://github.com/Kotlin/kotlinx.coroutines/issues/3993
+                if (ex.message?.contains("was closed, attempted to schedule") == false) {
+                    throw ex
+                } else {
+                    ex.printStackTrace()
                 }
             }
         }

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 closingDispatchersThrowIllegalState() {
+        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)
+        }
+    }
 }