Ct/find flakyness client reset

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

@@ -185,8 +185,12 @@ expect object RealmInterop {
     fun realm_config_set_in_memory(config: RealmConfigurationPointer, inMemory: Boolean)
     fun realm_schema_validate(schema: RealmSchemaPointer, mode: SchemaValidationMode): Boolean
 
-    fun realm_create_scheduler(): RealmSchedulerPointer
-    fun realm_create_scheduler(dispatcher: CoroutineDispatcher): RealmSchedulerPointer
+    fun realm_create_scheduler(name: String = "noname"): RealmSchedulerPointer
+    fun realm_create_scheduler(
+        dispatcher: CoroutineDispatcher,
+        name: String = "noname"
+    ): RealmSchedulerPointer
+
     /**
      * Open a realm on the current thread.
      *

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

@@ -210,11 +210,22 @@ actual object RealmInterop {
         realmc.realm_config_set_in_memory(config.cptr(), inMemory)
     }
 
-    actual fun realm_create_scheduler(): RealmSchedulerPointer =
-        LongPointerWrapper(realmc.realm_create_generic_scheduler())
+    actual fun realm_create_scheduler(name: String): RealmSchedulerPointer =
+        LongPointerWrapper(
+            realmc.realm_create_generic_scheduler().also {
+                println("Scheduler $name pointer : $it")
+            }
+        )
 
-    actual fun realm_create_scheduler(dispatcher: CoroutineDispatcher): RealmSchedulerPointer =
-        LongPointerWrapper(realmc.realm_create_scheduler(JVMScheduler(dispatcher)))
+    actual fun realm_create_scheduler(
+        dispatcher: CoroutineDispatcher,
+        name: String,
+    ): RealmSchedulerPointer =
+        LongPointerWrapper(
+            realmc.realm_create_scheduler(JVMScheduler(dispatcher)).also {
+                println("Scheduler $name pointer: $it")
+            }
+        )
 
     actual fun realm_open(
         config: RealmConfigurationPointer,
@@ -2144,6 +2155,7 @@ private class JVMScheduler(dispatcher: CoroutineDispatcher) {
     val scope: CoroutineScope = CoroutineScope(dispatcher)
 
     fun notifyCore(schedulerPointer: Long) {
+        println("Scheduler invoke: $schedulerPointer")
         scope.launch {
             realmc.invoke_core_notify_callback(schedulerPointer)
         }

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

@@ -541,14 +541,14 @@ actual object RealmInterop {
         return Pair(realmPtr, fileCreated.value)
     }
 
-    actual fun realm_create_scheduler(): RealmSchedulerPointer {
+    actual fun realm_create_scheduler(name: String): RealmSchedulerPointer {
         // If there is no notification dispatcher use the default scheduler.
         // Re-verify if this is actually needed when notification scheduler is fully in place.
         val scheduler = checkedPointerResult(realm_wrapper.realm_scheduler_make_default())
         return CPointerWrapper<RealmSchedulerT>(scheduler)
     }
 
-    actual fun realm_create_scheduler(dispatcher: CoroutineDispatcher): RealmSchedulerPointer {
+    actual fun realm_create_scheduler(dispatcher: CoroutineDispatcher, name: String): RealmSchedulerPointer {
         printlntid("createSingleThreadDispatcherScheduler")
         val scheduler = SingleThreadDispatcherScheduler(tid(), dispatcher)
 

packages/library-base/src/commonMain/kotlin/io/realm/kotlin/Realm.kt

@@ -120,7 +120,7 @@ public interface Realm : TypedRealm {
             if (!fileExists(configuration.path)) return false
             val config = (configuration as InternalConfiguration)
 
-            return RealmInterop.realm_create_scheduler()
+            return RealmInterop.realm_create_scheduler("compact-realm")
                 .use { scheduler ->
                     val (dbPointer, _) = RealmInterop.realm_open(
                         config = config.createNativeConfiguration(),

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

@@ -34,9 +34,9 @@ import io.realm.kotlin.internal.interop.LiveRealmPointer
 import io.realm.kotlin.internal.interop.MigrationCallback
 import io.realm.kotlin.internal.interop.RealmConfigurationPointer
 import io.realm.kotlin.internal.interop.RealmInterop
+import io.realm.kotlin.internal.interop.RealmSchedulerPointer
 import io.realm.kotlin.internal.interop.RealmSchemaPointer
 import io.realm.kotlin.internal.interop.SchemaMode
-import io.realm.kotlin.internal.interop.use
 import io.realm.kotlin.internal.platform.appFilesDirectory
 import io.realm.kotlin.internal.platform.prepareRealmFilePath
 import io.realm.kotlin.internal.platform.realmObjectCompanionOrThrow
@@ -107,16 +107,14 @@ public open class ConfigurationImpl(
         return configInitializer(nativeConfig)
     }
 
-    override suspend fun openRealm(realm: RealmImpl): Pair<FrozenRealmReference, Boolean> {
+    override suspend fun openRealm(realm: RealmImpl): Triple<FrozenRealmReference, Boolean, RealmSchedulerPointer> {
         val configPtr = realm.configuration.createNativeConfiguration()
-        return RealmInterop.realm_create_scheduler()
-            .use { scheduler ->
-                val (dbPointer, fileCreated) = RealmInterop.realm_open(configPtr, scheduler)
-                val liveRealmReference = LiveRealmReference(realm, dbPointer)
-                val frozenReference = liveRealmReference.snapshot(realm)
-                liveRealmReference.close()
-                frozenReference to fileCreated
-            }
+        val scheduler = RealmInterop.realm_create_scheduler("configurationImpl")
+        val (dbPointer, fileCreated) = RealmInterop.realm_open(configPtr, scheduler)
+        val liveRealmReference = LiveRealmReference(realm, dbPointer)
+        val frozenReference = liveRealmReference.snapshot(realm)
+        liveRealmReference.close()
+        return Triple(frozenReference, fileCreated, scheduler)
     }
 
     override suspend fun initializeRealmData(realm: RealmImpl, realmFileCreated: Boolean) {

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

@@ -18,6 +18,7 @@ package io.realm.kotlin.internal
 
 import io.realm.kotlin.Configuration
 import io.realm.kotlin.internal.interop.RealmConfigurationPointer
+import io.realm.kotlin.internal.interop.RealmSchedulerPointer
 import io.realm.kotlin.internal.interop.SchemaMode
 import io.realm.kotlin.internal.util.CoroutineDispatcherFactory
 import io.realm.kotlin.types.BaseRealmObject
@@ -59,7 +60,7 @@ public interface InternalConfiguration : Configuration {
      * @param realm instance of the Realm that is being created.
      * @returns a pair of (LiveRealmPointer, FileCreated)
      */
-    public suspend fun openRealm(realm: RealmImpl): Pair<FrozenRealmReference, Boolean>
+    public suspend fun openRealm(realm: RealmImpl): Triple<FrozenRealmReference, Boolean, RealmSchedulerPointer>
 
     /**
      * This function is a way `RealmImpl` can defer how the Realm is initialized once opened.

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

@@ -22,6 +22,7 @@ import io.realm.kotlin.Realm
 import io.realm.kotlin.dynamic.DynamicRealm
 import io.realm.kotlin.internal.dynamic.DynamicRealmImpl
 import io.realm.kotlin.internal.interop.RealmInterop
+import io.realm.kotlin.internal.interop.RealmSchedulerPointer
 import io.realm.kotlin.internal.interop.SynchronizableObject
 import io.realm.kotlin.internal.platform.copyAssetFile
 import io.realm.kotlin.internal.platform.fileExists
@@ -95,6 +96,13 @@ public class RealmImpl private constructor(
     override val realmReference: FrozenRealmReference
         get() = realmReference()
 
+    // TODO We have to keep this scheduler alive while the initial realm is open. We have no notification
+    // from core when we can release, to be safe we will tie its lifecycle to the user-facing Realm.
+    // Releasing this scheduler might incur in race-conditions, leading to core crashing.
+    // For example after a client reset it might try to notify with this scheduler that might have been
+    // released.
+    private lateinit var initialScheduler: RealmSchedulerPointer
+
     // TODO Bit of an overkill to have this as we are only catching the initial frozen version.
     //  Maybe we could just rely on the notifier to issue the initial frozen version, but that
     //  would require us to sync that up. Didn't address this as we already have a todo on fixing
@@ -129,7 +137,8 @@ public class RealmImpl private constructor(
                         }
                     }
                 }
-                val (frozenReference, fileCreated) = configuration.openRealm(this@RealmImpl)
+                val (frozenReference, fileCreated, scheduler) = configuration.openRealm(this@RealmImpl)
+                initialScheduler = scheduler
                 realmFileCreated = assetFileCopied || fileCreated
                 versionTracker.trackAndCloseExpiredReferences(frozenReference)
                 _realmReference.value = frozenReference
@@ -279,7 +288,6 @@ public class RealmImpl private constructor(
         if (!realmStateFlow.tryEmit(State.CLOSED)) {
             log.warn("Cannot signal internal close")
         }
-
         notificationScheduler.close()
         writeScheduler.close()
     }

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

@@ -42,10 +42,11 @@ public fun interface CoroutineDispatcherFactory {
         public fun managed(name: String, threads: Int = 1): CoroutineDispatcherFactory {
             return CoroutineDispatcherFactory {
                 ManagedDispatcherHolder(
-                    when (threads) {
+                    dispatcher = when (threads) {
                         1 -> singleThreadDispatcher(name)
                         else -> multiThreadDispatcher(threads)
-                    }
+                    },
+                    name = name
                 )
             }
         }
@@ -56,7 +57,7 @@ public fun interface CoroutineDispatcherFactory {
          */
         public fun unmanaged(dispatcher: CoroutineDispatcher): CoroutineDispatcherFactory {
             return CoroutineDispatcherFactory {
-                UnmanagedDispatcherHolder(dispatcher)
+                UnmanagedDispatcherHolder(dispatcher, "unmanaged")
             }
         }
     }
@@ -85,6 +86,8 @@ public sealed interface DispatcherHolder {
      */
     public val dispatcher: CoroutineDispatcher
 
+    public val name: String
+
     /**
      * Mark the dispatcher as no longer being used. For internal dispatchers, this will also
      * close them.
@@ -95,12 +98,14 @@ public sealed interface DispatcherHolder {
 @OptIn(ExperimentalCoroutinesApi::class)
 private class ManagedDispatcherHolder(
     override val dispatcher: CloseableCoroutineDispatcher,
+    override val name: String,
 ) : DispatcherHolder {
     override fun close(): Unit = dispatcher.close()
 }
 
 private class UnmanagedDispatcherHolder(
-    override val dispatcher: CoroutineDispatcher
+    override val dispatcher: CoroutineDispatcher,
+    override val name: String,
 ) : DispatcherHolder {
     override fun close(): Unit = Unit
 }
@@ -115,7 +120,7 @@ public class LiveRealmContext(
 
     public val scheduler: RealmSchedulerPointer = runBlocking {
         withContext(dispatcherHolder.dispatcher) {
-            RealmInterop.realm_create_scheduler(dispatcher)
+            RealmInterop.realm_create_scheduler(dispatcher, dispatcherHolder.name)
         }
     }
 

packages/library-sync/src/commonMain/kotlin/io/realm/kotlin/mongodb/internal/SyncConfigurationImpl.kt

@@ -28,6 +28,7 @@ import io.realm.kotlin.internal.interop.RealmAppPointer
 import io.realm.kotlin.internal.interop.RealmAsyncOpenTaskPointer
 import io.realm.kotlin.internal.interop.RealmConfigurationPointer
 import io.realm.kotlin.internal.interop.RealmInterop
+import io.realm.kotlin.internal.interop.RealmSchedulerPointer
 import io.realm.kotlin.internal.interop.RealmSyncConfigurationPointer
 import io.realm.kotlin.internal.interop.RealmSyncSessionPointer
 import io.realm.kotlin.internal.interop.SyncAfterClientResetHandler
@@ -69,7 +70,7 @@ internal class SyncConfigurationImpl(
     override val initialRemoteData: InitialRemoteDataConfiguration?
 ) : InternalConfiguration by configuration, SyncConfiguration {
 
-    override suspend fun openRealm(realm: RealmImpl): Pair<FrozenRealmReference, Boolean> {
+    override suspend fun openRealm(realm: RealmImpl): Triple<FrozenRealmReference, Boolean, RealmSchedulerPointer> {
         // Partition-based Realms with `waitForInitialRemoteData` enabled will use
         // async open first do download the server side Realm. This is much faster than
         // creating the Realm locally first and then downloading (and integrating) changes into
@@ -131,8 +132,8 @@ internal class SyncConfigurationImpl(
         // So there are two possibilities for the file to be created:
         // 1) .waitForInitialRemoteData caused async open to be used, which created the file.
         // 2) The synced Realm was opened locally first (without async open), which then created the file.
-        val result: Pair<FrozenRealmReference, Boolean> = configuration.openRealm(realm)
-        return Pair(result.first, result.second || asyncOpenCreatedRealmFile.value)
+        val result: Triple<FrozenRealmReference, Boolean, RealmSchedulerPointer> = configuration.openRealm(realm)
+        return Triple(result.first, result.second || asyncOpenCreatedRealmFile.value, result.third)
     }
 
     override suspend fun initializeRealmData(realm: RealmImpl, realmFileCreated: Boolean) {

packages/test-base/src/commonMain/kotlin/io/realm/kotlin/test/StandaloneDynamicMutableRealm.kt

@@ -45,7 +45,7 @@ internal class StandaloneDynamicMutableRealm private constructor(
     ) {
     constructor(configuration: InternalConfiguration) : this(
         configuration,
-        RealmInterop.realm_create_scheduler()
+        RealmInterop.realm_create_scheduler("dynamic-standalone")
     )
 
     override fun close() {