Fix SyncSession crashing when accessed after receiving remote changes (…

…#1071)

CHANGELOG.md

@@ -10,7 +10,7 @@
 * [Sync] The sync variant `io.realm.kotlin:library-sync:1.4.0`, now support Apple Silicon targets, ie. `macosArm64()`, `iosArm64()` and `iosSimulatorArm64`.
 
 ### Fixed
-* None.
+* [Sync] Using the SyncSession after receiving changes from the server would sometimes crash. Issue [#1068](https://github.com/realm/realm-kotlin/issues/1068)
 
 ### Compatibility
 * This release is compatible with the following Kotlin releases:

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

@@ -37,13 +37,20 @@ internal class SyncedRealmContext<T : BaseRealm>(realm: T) {
     //  When we introduce a public DynamicRealm, this can also be a `DynamicRealmImpl`
     //  And we probably need to modify the SyncSessionImpl to take either of these two.
     private val baseRealm = realm as RealmImpl
-    private val dbPointer = baseRealm.realmReference.dbPointer
     internal val config: SyncConfiguration = baseRealm.configuration as SyncConfiguration
+    // Note: Session and Subscriptions only need a valid dbPointer when being created, after that, they
+    // have their own lifecycle and can be cached.
     internal val session: SyncSession by lazy {
-        SyncSessionImpl(baseRealm, RealmInterop.realm_sync_session_get(dbPointer))
+        SyncSessionImpl(
+            baseRealm,
+            RealmInterop.realm_sync_session_get(baseRealm.realmReference.dbPointer)
+        )
     }
     internal val subscriptions: SubscriptionSet<T> by lazy {
-        SubscriptionSetImpl<T>(realm, RealmInterop.realm_sync_get_latest_subscriptionset(dbPointer))
+        SubscriptionSetImpl(
+            realm,
+            RealmInterop.realm_sync_get_latest_subscriptionset(baseRealm.realmReference.dbPointer)
+        )
     }
 }
 

packages/test-sync/src/androidTest/kotlin/io/realm/kotlin/test/shared/SyncedRealmTests.kt

@@ -36,11 +36,15 @@ import io.realm.kotlin.mongodb.User
 import io.realm.kotlin.mongodb.exceptions.DownloadingRealmTimeOutException
 import io.realm.kotlin.mongodb.exceptions.SyncException
 import io.realm.kotlin.mongodb.exceptions.UnrecoverableSyncException
+import io.realm.kotlin.mongodb.sync.InitialSubscriptionsCallback
 import io.realm.kotlin.mongodb.sync.SyncConfiguration
 import io.realm.kotlin.mongodb.sync.SyncSession
 import io.realm.kotlin.mongodb.sync.SyncSession.ErrorHandler
 import io.realm.kotlin.mongodb.syncSession
+import io.realm.kotlin.notifications.InitialRealm
+import io.realm.kotlin.notifications.RealmChange
 import io.realm.kotlin.notifications.ResultsChange
+import io.realm.kotlin.notifications.UpdatedRealm
 import io.realm.kotlin.query.RealmResults
 import io.realm.kotlin.test.mongodb.TestApp
 import io.realm.kotlin.test.mongodb.asTestApp
@@ -55,10 +59,12 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.async
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.filter
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.takeWhile
+import kotlinx.coroutines.withTimeout
 import okio.FileSystem
 import okio.Path
 import okio.Path.Companion.toPath
@@ -202,6 +208,63 @@ class SyncedRealmTests {
         realm2.close()
     }
 
+    // Test for https://github.com/realm/realm-kotlin/issues/1070
+    @Ignore // Enable once #1070 is fixed
+    @Test
+    fun realmAsFlow_acrossSyncedChanges() = runBlocking {
+        val (email1, password1) = randomEmail() to "password1234"
+        val (email2, password2) = randomEmail() to "password1234"
+        val user1 = app.createUserAndLogIn(email1, password1)
+        val user2 = app.createUserAndLogIn(email2, password2)
+
+        val config1 = createSyncConfig(
+            user = user1,
+            name = "db1.realm",
+            partitionValue = partitionValue,
+            schema = setOf(SyncObjectWithAllTypes::class)
+        )
+        val realm1 = Realm.open(config1)
+        val c = Channel<RealmChange<Realm>>(1)
+        val observer = async {
+            realm1.asFlow().collect {
+                c.send(it)
+            }
+        }
+        val event: RealmChange<Realm> = c.receive()
+        assertTrue(event is InitialRealm)
+
+        // Write remote change
+        createSyncConfig(
+            user = user2,
+            name = "db2.realm",
+            partitionValue = partitionValue,
+            schema = setOf(SyncObjectWithAllTypes::class)
+        ).let { config ->
+            Realm.open(config).use { realm ->
+                realm.write {
+                    val id = "id-${Random.nextLong()}"
+                    val masterObject = SyncObjectWithAllTypes.createWithSampleData(id)
+                    copyToRealm(masterObject)
+                }
+                realm.syncSession.uploadAllLocalChanges()
+            }
+        }
+
+        // Wait for Realm.asFlow() to be updated based on remote change.
+        try {
+            withTimeout(timeout = 30.seconds) {
+                val updateEvent: RealmChange<Realm> = c.receive()
+                assertTrue(updateEvent is UpdatedRealm)
+                assertEquals(1, updateEvent.realm.query<SyncObjectWithAllTypes>().find().size)
+                assertEquals(1, realm.query<SyncObjectWithAllTypes>().find().size)
+            }
+        } finally {
+            realm1.close()
+            observer.cancel()
+            c.cancel()
+        }
+    }
+
     @Test
     fun canOpenWithRemoteSchema() {
         val (email, password) = randomEmail() to "password1234"
@@ -744,7 +807,12 @@ class SyncedRealmTests {
 
     @Test
     fun writeCopyTo_localToFlexibleSync_throws() = runBlocking {
-        val flexApp = TestApp(appName = io.realm.kotlin.test.mongodb.TEST_APP_FLEX)
+        val flexApp = TestApp(
+            appName = io.realm.kotlin.test.mongodb.TEST_APP_FLEX,
+            builder = {
+                it.syncRootDirectory(PlatformUtils.createTempDir("flx-sync-"))
+            }
+        )
         val (email1, password1) = randomEmail() to "password1234"
         val user1 = flexApp.createUserAndLogIn(email1, password1)
         val localConfig = createWriteCopyLocalConfig("local.realm")
@@ -768,6 +836,7 @@ class SyncedRealmTests {
                 localRealm.writeCopyTo(flexSyncConfig)
             }
         }
+        flexApp.close()
     }
 
     @Test
@@ -822,7 +891,12 @@ class SyncedRealmTests {
 
     @Test
     fun writeCopyTo_flexibleSyncToLocal() = runBlocking {
-        val flexApp = TestApp(appName = io.realm.kotlin.test.mongodb.TEST_APP_FLEX)
+        val flexApp = TestApp(
+            appName = io.realm.kotlin.test.mongodb.TEST_APP_FLEX,
+            builder = {
+                it.syncRootDirectory(PlatformUtils.createTempDir("flx-sync-"))
+            }
+        )
         val (email1, password1) = randomEmail() to "password1234"
         val user = flexApp.createUserAndLogIn(email1, password1)
         val localConfig = createWriteCopyLocalConfig("local.realm")
@@ -852,6 +926,7 @@ class SyncedRealmTests {
             assertEquals(1, localRealm.query<FlexParentObject>().count().find())
             assertEquals("local object", localRealm.query<FlexParentObject>().first().find()!!.name)
         }
+        flexApp.close()
     }
 
     @Test
@@ -969,6 +1044,77 @@ class SyncedRealmTests {
         }
     }
 
+    // Test for https://github.com/realm/realm-kotlin/issues/1068
+    // Note, this test is not 100% sure to surface the bug, but manual testing has shown that it
+    // works well enough. Also, even if it doesn't surface the bug, it will not the fail the test.
+    @Test
+    fun accessSessionAfterRemoteChange() = runBlocking {
+        val flexApp = TestApp(
+            appName = io.realm.kotlin.test.mongodb.TEST_APP_FLEX,
+            builder = {
+                it.syncRootDirectory(PlatformUtils.createTempDir("flx-sync-"))
+            }
+        )
+        val section = Random.nextInt()
+        val (email1, password1) = randomEmail() to "password1234"
+        val (email2, password2) = randomEmail() to "password1234"
+        val user1 = flexApp.createUserAndLogIn(email1, password1)
+        val user2 = flexApp.createUserAndLogIn(email2, password2)
+        val syncConfig1 = createFlexibleSyncConfig(
+            user = user1,
+            name = "sync1.realm",
+            initialSubscriptions = { realm: Realm ->
+                realm.query<FlexParentObject>("section = $0", section).subscribe()
+            }
+        )
+        val syncConfig2 = createFlexibleSyncConfig(
+            user = user2,
+            name = "sync2.realm",
+            initialSubscriptions = { realm: Realm ->
+                realm.query<FlexParentObject>("section = $0", section).subscribe()
+            }
+        )
+        val realm1 = Realm.open(syncConfig1)
+
+        Realm.open(syncConfig2).use { realm2 ->
+            realm2.write {
+                copyToRealm(FlexParentObject(section))
+            }
+            realm2.syncSession.uploadAllLocalChanges()
+        }
+
+        // Reading the object means we received it from the other Realm
+        withTimeout(30.seconds) {
+            val obj: FlexParentObject = realm1.query<FlexParentObject>("section = $0", section).asFlow()
+                .map { it.list }
+                .filter { it.isNotEmpty() }
+                .first().first()
+            assertEquals(section, obj.section)
+
+            // 1. Local write to work around https://github.com/realm/realm-kotlin/issues/1070
+            realm1.write { }
+
+            // 2. Trigger GC. This will GC the RealmReference JVM object, making the native reference
+            //    eligible for closing.
+            PlatformUtils.triggerGC()
+
+            // 3. On the next update of Realm, we run through the weak list of all previous
+            //    RealmReferences and close all native pointers with their JVM object GC'ed.
+            //    This should now include the object created in step 1.
+            realm1.write { }
+        }
+
+        // 4. With the original native dbPointer now being closed, accessing the syncSession for
+        //    the first time should still work.
+        try {
+            realm1.syncSession.pause()
+            assertEquals(SyncSession.State.INACTIVE, realm1.syncSession.state)
+        } finally {
+            realm1.close()
+            flexApp.close()
+        }
+    }
+
 //    @Test
 //    fun initialVersion() {
 //        assertEquals(INITIAL_VERSION, realm.version())
@@ -1336,13 +1482,19 @@ class SyncedRealmTests {
         encryptionKey: ByteArray? = null,
         log: LogConfiguration? = null,
         errorHandler: ErrorHandler? = null,
-        schema: Set<KClass<out BaseRealmObject>> = setOf(SyncObjectWithAllTypes::class),
+        schema: Set<KClass<out BaseRealmObject>> = setOf(
+            FlexParentObject::class,
+            FlexChildObject::class,
+            FlexEmbeddedObject::class
+        ),
+        initialSubscriptions: InitialSubscriptionsCallback? = null
     ): SyncConfiguration = SyncConfiguration.Builder(
         user = user,
         schema = schema
     ).name(name).also { builder ->
         if (encryptionKey != null) builder.encryptionKey(encryptionKey)
         if (errorHandler != null) builder.errorHandler(errorHandler)
         if (log != null) builder.log(log.level, log.loggers)
+        if (initialSubscriptions != null) builder.initialSubscriptions(false, initialSubscriptions)
     }.build()
 }