Sync schema migration

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

@@ -38,6 +38,8 @@ object CoreErrorConverter {
         return userError ?: when {
             ErrorCode.RLM_ERR_INDEX_OUT_OF_BOUNDS == errorCode ->
                 IndexOutOfBoundsException(message)
+            ErrorCode.RLM_ERR_INVALID_SCHEMA_VERSION == errorCode ->
+                InvalidSchemaVersionException(message)
             ErrorCategory.RLM_ERR_CAT_INVALID_ARG in categories && ErrorCategory.RLM_ERR_CAT_SYNC_ERROR !in categories -> {
                 // Some sync errors flagged as both logical and illegal. In our case, we consider those
                 // IllegalState, so discard them them here and let them fall through to the bottom case

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

@@ -0,0 +1,22 @@
+/*
+ * Copyright 2024 Realm Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.realm.kotlin.internal.interop
+
+/**
+ * Exception thrown when there is a mismatch between the schema version defined in the configuration
+ * and the persisted one.
+ */
+class InvalidSchemaVersionException(override val message: String?) : IllegalStateException()

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

@@ -227,6 +227,8 @@ expect object RealmInterop {
     // dispatcher. The realm itself must also be opened on the same thread
     fun realm_open(config: RealmConfigurationPointer, scheduler: RealmSchedulerPointer): Pair<LiveRealmPointer, Boolean>
 
+    fun realm_open(config: RealmConfigurationPointer): LiveRealmPointer
+
     // Opening a Realm asynchronously. Only supported for synchronized realms.
     fun realm_open_synchronized(config: RealmConfigurationPointer): RealmAsyncOpenTaskPointer
     fun realm_async_open_task_start(task: RealmAsyncOpenTaskPointer, callback: AsyncOpenCallback)
@@ -647,6 +649,10 @@ expect object RealmInterop {
     fun realm_sync_client_config_set_pong_keepalive_timeout(syncClientConfig: RealmSyncClientConfigurationPointer, timeoutMs: ULong)
     fun realm_sync_client_config_set_fast_reconnect_limit(syncClientConfig: RealmSyncClientConfigurationPointer, timeoutMs: ULong)
 
+    fun realm_get_persisted_schema_version(
+        config: RealmConfigurationPointer
+    ): Long
+
     fun realm_sync_config_new(
         user: RealmUserPointer,
         partition: String

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

@@ -241,6 +241,13 @@ actual object RealmInterop {
         return Pair(realmPtr, fileCreated)
     }
 
+    actual fun realm_open(
+        config: RealmConfigurationPointer,
+    ): LiveRealmPointer {
+        val realmPtr = LongPointerWrapper<LiveRealmT>(realmc.realm_open(config.cptr()))
+        return realmPtr
+    }
+
     actual fun realm_open_synchronized(config: RealmConfigurationPointer): RealmAsyncOpenTaskPointer {
         return LongPointerWrapper(realmc.realm_open_synchronized(config.cptr()))
     }
@@ -1438,6 +1445,10 @@ actual object RealmInterop {
         realmc.realm_sync_client_config_set_fast_reconnect_limit(syncClientConfig.cptr(), timeoutMs.toLong())
     }
 
+    actual fun realm_get_persisted_schema_version(
+        config: RealmConfigurationPointer
+    ): Long = realmc.realm_get_persisted_schema_version(config.cptr())
+
     actual fun realm_network_transport_new(networkTransport: NetworkTransport): RealmNetworkTransportPointer {
         return LongPointerWrapper(realmc.realm_network_transport_new(networkTransport))
     }

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

@@ -192,7 +192,7 @@ actual enum class ErrorCode(
 
     actual companion object {
         actual fun of(nativeValue: Int): ErrorCode? =
-            values().firstOrNull { value ->
+            entries.firstOrNull { value ->
                 value.nativeValue == nativeValue
             }
     }

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

@@ -556,6 +556,11 @@ actual object RealmInterop {
         return Pair(realmPtr, fileCreated.value)
     }
 
+    actual fun realm_open(config: RealmConfigurationPointer): LiveRealmPointer {
+        val realmPtr = CPointerWrapper<LiveRealmT>(realm_wrapper.realm_open(config.cptr()))
+        return realmPtr
+    }
+
     actual fun realm_create_scheduler(): 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.
@@ -2607,6 +2612,10 @@ actual object RealmInterop {
         realm_wrapper.realm_sync_client_config_set_fast_reconnect_limit(syncClientConfig.cptr(), timeoutMs)
     }
 
+    actual fun realm_get_persisted_schema_version(
+        config: RealmConfigurationPointer
+    ): Long = realm_wrapper.realm_get_persisted_schema_version(config.cptr()).toLong()
+
     actual fun realm_sync_config_set_error_handler(
         syncConfig: RealmSyncConfigurationPointer,
         errorHandler: SyncErrorCallback

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

@@ -283,18 +283,6 @@ public interface Configuration {
             this.writeDispatcher = dispatcher
         } as S
 
-        /**
-         * Sets the schema version of the Realm. This must be equal to or higher than the schema
-         * version of the existing Realm file, if any. If the schema version is higher than the
-         * already existing Realm, a migration is needed.
-         */
-        public fun schemaVersion(schemaVersion: Long): S {
-            if (schemaVersion < 0) {
-                throw IllegalArgumentException("Realm schema version numbers must be 0 (zero) or higher. Yours was: $schemaVersion")
-            }
-            return apply { this.schemaVersion = schemaVersion } as S
-        }
-
         /**
          * Sets the 64 byte key used to encrypt and decrypt the Realm file. If no key is provided
          * the Realm file will be unencrypted.

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

@@ -132,6 +132,18 @@ public interface RealmConfiguration : Configuration {
                 this.automaticEmbeddedObjectConstraintsResolution = resolveEmbeddedObjectConstraints
             }
 
+        /**
+         * Sets the schema version of the Realm. This must be equal to or higher than the schema
+         * version of the existing Realm file, if any. If the schema version is higher than the
+         * already existing Realm, a migration is needed.
+         */
+        public fun schemaVersion(schemaVersion: Long): Builder {
+            if (schemaVersion < 0) {
+                throw IllegalArgumentException("Realm schema version numbers must be 0 (zero) or higher. Yours was: $schemaVersion")
+            }
+            return apply { this.schemaVersion = schemaVersion }
+        }
+
         override fun name(name: String): Builder = apply {
             checkName(name)
             this.name = name

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

@@ -98,6 +98,7 @@ internal class SubscriptionSetImpl<T : BaseRealm>(
                                 channel.trySend(false)
                             }
                             else -> {
+                                println("STATE $state")
                                 // Ignore all other states, wait for either complete or error.
                             }
                         }

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

@@ -47,6 +47,7 @@ import io.realm.kotlin.mongodb.sync.RecoverOrDiscardUnsyncedChangesStrategy
 import io.realm.kotlin.mongodb.sync.RecoverUnsyncedChangesStrategy
 import io.realm.kotlin.mongodb.sync.SyncClientResetStrategy
 import io.realm.kotlin.mongodb.sync.SyncConfiguration
+import io.realm.kotlin.mongodb.sync.SyncMigrationRemoteDataConfiguration
 import io.realm.kotlin.mongodb.sync.SyncMode
 import io.realm.kotlin.mongodb.sync.SyncSession
 import kotlinx.atomicfu.AtomicBoolean
@@ -69,7 +70,8 @@ internal class SyncConfigurationImpl(
     override val errorHandler: SyncSession.ErrorHandler,
     override val syncClientResetStrategy: SyncClientResetStrategy,
     override val initialSubscriptions: InitialSubscriptionsConfiguration?,
-    override val initialRemoteData: InitialRemoteDataConfiguration?
+    override val initialRemoteData: InitialRemoteDataConfiguration?,
+    override val schemaMigrationRemoteData: SyncMigrationRemoteDataConfiguration?,
 ) : InternalConfiguration by configuration, SyncConfiguration {
 
     override suspend fun openRealm(realm: RealmImpl): Pair<FrozenRealmReference, Boolean> {
@@ -84,14 +86,28 @@ internal class SyncConfigurationImpl(
         // unnecessary pressure on the server.
         val fileExists: Boolean = fileExists(configuration.path)
         val asyncOpenCreatedRealmFile: AtomicBoolean = atomic(false)
-        if (initialRemoteData != null && !fileExists) {
+
+        val configPtr = createNativeConfiguration()
+
+        if (
+            (!fileExists && initialRemoteData != null) ||
+            (fileExists && isSyncMigrationPending(configPtr))
+        ) {
+            // There are two different timeout:
+            // - initial remote data timeout, when it is the first time we open the Realm.
+            // - schema migration timeout, when a sync schema migration is required.
+            val timeout = if (fileExists)
+                schemaMigrationRemoteData!!.timeout
+            else
+                initialRemoteData!!.timeout
+
             // Channel to work around not being able to use `suspendCoroutine` to wrap the callback, as
             // that results in the `Continuation` being frozen, which breaks it.
             val channel = Channel<Any>(1)
             val taskPointer: AtomicRef<RealmAsyncOpenTaskPointer?> = atomic(null)
             try {
-                val result: Any = withTimeout(initialRemoteData.timeout.inWholeMilliseconds) {
-                    withContext(realm.notificationScheduler.dispatcher) {
+                val result: Any = withTimeout(timeout.inWholeMilliseconds) {
+                    withContext(realm.writeScheduler.dispatcher) {
                         val callback = AsyncOpenCallback { error: Throwable? ->
                             if (error != null) {
                                 channel.trySend(error)
@@ -100,7 +116,6 @@ internal class SyncConfigurationImpl(
                             }
                         }
 
-                        val configPtr = createNativeConfiguration()
                         taskPointer.value = RealmInterop.realm_open_synchronized(configPtr)
                         RealmInterop.realm_async_open_task_start(taskPointer.value!!, callback)
                         channel.receive()
@@ -111,6 +126,7 @@ internal class SyncConfigurationImpl(
                         // Track whether or not async open created the file.
                         asyncOpenCreatedRealmFile.value = true
                     }
+
                     is Throwable -> throw result
                     else -> throw IllegalStateException("Unexpected value: $result")
                 }
@@ -138,6 +154,17 @@ internal class SyncConfigurationImpl(
         return Pair(result.first, result.second || asyncOpenCreatedRealmFile.value)
     }
 
+    /**
+     * Checks whether a sync Realm requires a migration, this happens when the schema version provided in
+     * the config differs from the persisted one.
+     */
+    internal fun isSyncMigrationPending(configPtr: RealmConfigurationPointer): Boolean =
+        if (fileExists(configuration.path)) {
+            RealmInterop.realm_get_persisted_schema_version(configPtr) != configuration.schemaVersion
+        } else {
+            false
+        }
+
     override suspend fun initializeRealmData(realm: RealmImpl, realmFileCreated: Boolean) {
         // Create or update subscriptions for Flexible Sync realms as needed.
         initialSubscriptions?.let { initialSubscriptionsConfig ->
@@ -173,7 +200,7 @@ internal class SyncConfigurationImpl(
         return syncInitializer(ptr)
     }
 
-    private val syncInitializer: (RealmConfigurationPointer) -> RealmConfigurationPointer
+    private var syncInitializer: (RealmConfigurationPointer) -> RealmConfigurationPointer
 
     init {
         // We need to freeze `errorHandler` reference on initial thread
@@ -184,12 +211,16 @@ internal class SyncConfigurationImpl(
         val initializerHelper = when (resetStrategy) {
             is DiscardUnsyncedChangesStrategy ->
                 DiscardUnsyncedChangesHelper(resetStrategy, configuration)
+
             is ManuallyRecoverUnsyncedChangesStrategy ->
                 ManuallyRecoverUnsyncedChangesHelper(resetStrategy)
+
             is RecoverUnsyncedChangesStrategy ->
                 RecoverUnsyncedChangesHelper(resetStrategy, configuration)
+
             is RecoverOrDiscardUnsyncedChangesStrategy ->
                 RecoverOrDiscardUnsyncedChangesHelper(resetStrategy, configuration)
+
             else -> throw IllegalArgumentException("Unsupported client reset strategy: $resetStrategy")
         }
 
@@ -263,7 +294,7 @@ private interface ClientResetStrategyHelper {
 
 private abstract class OnBeforeOnAfterHelper<T : SyncClientResetStrategy> constructor(
     val strategy: T,
-    val configuration: InternalConfiguration
+    val configuration: InternalConfiguration,
 ) : ClientResetStrategyHelper {
 
     abstract fun getResyncMode(): SyncSessionResyncMode
@@ -285,7 +316,7 @@ private abstract class OnBeforeOnAfterHelper<T : SyncClientResetStrategy> constr
 
 private class RecoverOrDiscardUnsyncedChangesHelper constructor(
     strategy: RecoverOrDiscardUnsyncedChangesStrategy,
-    configuration: InternalConfiguration
+    configuration: InternalConfiguration,
 ) : OnBeforeOnAfterHelper<RecoverOrDiscardUnsyncedChangesStrategy>(strategy, configuration) {
 
     override fun getResyncMode(): SyncSessionResyncMode =
@@ -303,7 +334,7 @@ private class RecoverOrDiscardUnsyncedChangesHelper constructor(
             override fun onAfterReset(
                 realmBefore: FrozenRealmPointer,
                 realmAfter: LiveRealmPointer,
-                didRecover: Boolean
+                didRecover: Boolean,
             ) {
                 // Needed to allow writes on the Mutable after Realm
                 RealmInterop.realm_begin_write(realmAfter)
@@ -338,7 +369,7 @@ private class RecoverOrDiscardUnsyncedChangesHelper constructor(
     override fun onSyncError(
         session: SyncSession,
         appPointer: RealmAppPointer,
-        error: SyncError
+        error: SyncError,
     ) {
         // If there is a user exception we appoint it as the cause of the client reset
         strategy.onManualResetFallback(
@@ -350,7 +381,7 @@ private class RecoverOrDiscardUnsyncedChangesHelper constructor(
 
 private class RecoverUnsyncedChangesHelper constructor(
     strategy: RecoverUnsyncedChangesStrategy,
-    configuration: InternalConfiguration
+    configuration: InternalConfiguration,
 ) : OnBeforeOnAfterHelper<RecoverUnsyncedChangesStrategy>(strategy, configuration) {
 
     override fun getResyncMode(): SyncSessionResyncMode =
@@ -368,7 +399,7 @@ private class RecoverUnsyncedChangesHelper constructor(
             override fun onAfterReset(
                 realmBefore: FrozenRealmPointer,
                 realmAfter: LiveRealmPointer,
-                didRecover: Boolean
+                didRecover: Boolean,
             ) {
                 // Needed to allow writes on the Mutable after Realm
                 RealmInterop.realm_begin_write(realmAfter)
@@ -400,7 +431,7 @@ private class RecoverUnsyncedChangesHelper constructor(
     override fun onSyncError(
         session: SyncSession,
         appPointer: RealmAppPointer,
-        error: SyncError
+        error: SyncError,
     ) {
         // If there is a user exception we appoint it as the cause of the client reset
         strategy.onManualResetFallback(
@@ -412,7 +443,7 @@ private class RecoverUnsyncedChangesHelper constructor(
 
 private class DiscardUnsyncedChangesHelper constructor(
     strategy: DiscardUnsyncedChangesStrategy,
-    configuration: InternalConfiguration
+    configuration: InternalConfiguration,
 ) : OnBeforeOnAfterHelper<DiscardUnsyncedChangesStrategy>(strategy, configuration) {
 
     override fun getResyncMode(): SyncSessionResyncMode =
@@ -430,7 +461,7 @@ private class DiscardUnsyncedChangesHelper constructor(
             override fun onAfterReset(
                 realmBefore: FrozenRealmPointer,
                 realmAfter: LiveRealmPointer,
-                didRecover: Boolean
+                didRecover: Boolean,
             ) {
                 // Needed to allow writes on the Mutable after Realm
                 RealmInterop.realm_begin_write(realmAfter)
@@ -462,7 +493,7 @@ private class DiscardUnsyncedChangesHelper constructor(
     override fun onSyncError(
         session: SyncSession,
         appPointer: RealmAppPointer,
-        error: SyncError
+        error: SyncError,
     ) {
         strategy.onManualResetFallback(
             session,
@@ -472,7 +503,7 @@ private class DiscardUnsyncedChangesHelper constructor(
 }
 
 private class ManuallyRecoverUnsyncedChangesHelper(
-    val strategy: ManuallyRecoverUnsyncedChangesStrategy
+    val strategy: ManuallyRecoverUnsyncedChangesStrategy,
 ) : ClientResetStrategyHelper {
 
     override fun initialize(nativeSyncConfig: RealmSyncConfigurationPointer) {
@@ -485,7 +516,7 @@ private class ManuallyRecoverUnsyncedChangesHelper(
     override fun onSyncError(
         session: SyncSession,
         appPointer: RealmAppPointer,
-        error: SyncError
+        error: SyncError,
     ) {
         strategy.onClientReset(
             session,