RCORE-2135 Make AsyncOpenTask wait until all pending subscriptions finish bootstrapping (#7723)

* Make AsyncOpenTask to wait until all pending subscriptions finish bootstrapping

* Don't use subscription machinery for pbs

* Update changelog

* Fix test due to scheduling

Author: danieltabacaru

CHANGELOG.md

@@ -7,6 +7,7 @@
 ### Fixed
 * <How do the end-user experience this issue? what was the impact?> ([#????](https://github.com/realm/realm-core/issues/????), since v?.?.?)
 * Valgrind could report a branch on an uninitialized read when opening something that is not an encrypted Realm file as an encrypted Realm file ([PR #7789](https://github.com/realm/realm-core/pull/7789), since v14.10.0).
+* Opening an FLX realm asynchronously may not wait to download all data ([#7720](https://github.com/realm/realm-core/issues/7720), since FLX sync was introduced).
 
 ### Breaking changes
 * None.
@@ -468,7 +469,7 @@
 * Fixed equality queries on a Mixed property with an index possibly returning the wrong result if values of different types happened to have the same StringIndex hash. ([6407](https://github.com/realm/realm-core/issues/6407) since v11.0.0-beta.5).
 * If you have more than 8388606 links pointing to one specific object, the program will crash. ([#6577](https://github.com/realm/realm-core/issues/6577), since v6.0.0)
 * Query for NULL value in Dictionary<Mixed> would give wrong results ([6748])(https://github.com/realm/realm-core/issues/6748), since v10.0.0)
-* A Realm generated on a non-apple ARM 64 device and copied to another platform (and vice-versa) were non-portable due to a sorting order difference. This impacts strings or binaries that have their first difference at a non-ascii character. These items may not be found in a set, or in an indexed column if the strings had a long common prefix (> 200 characters). ([PR #6670](https://github.com/realm/realm-core/pull/6670), since 2.0.0-rc7 for indexes, and since since the introduction of sets in v10.2.0)
+* A Realm generated on a non-apple ARM 64 device and copied to another platform (and vice-versa) were non-portable due to a sorting order difference. This impacts strings or binaries that have their first difference at a non-ascii character. These items may not be found in a set, or in an indexed column if the strings had a long common prefix (> 200 characters). ([PR #6670](https://github.com/realm/realm-core/pull/6670), since 2.0.0-rc7 for indexes, and since the introduction of sets in v10.2.0)
 
 ### Breaking changes
 * Support for upgrading from Realm files produced by RealmCore v5.23.9 or earlier is no longer supported.

src/realm/object-store/impl/realm_coordinator.cpp

@@ -386,7 +386,7 @@ void RealmCoordinator::do_get_realm(RealmConfig&& config, std::shared_ptr<Realm>
         const auto subscription_version = current_subscription.version();
         // in case we are hitting this check while during a normal open, we need to take in
         // consideration if the db was created during this call. Since this may be the first time
-        // we are actually creating a realm. For async open this does not apply, infact db_created
+        // we are actually creating a realm. For async open this does not apply, in fact db_created
         // will always be false.
         if (!first_time_open)
             first_time_open = db_created;

src/realm/object-store/sync/async_open_task.cpp

@@ -59,7 +59,7 @@ void AsyncOpenTask::start(AsyncOpenCallback callback)
             return;
         }
 
-        self->migrate_schema_or_complete(std::move(callback), coordinator, status);
+        self->migrate_schema_or_complete(std::move(callback), coordinator);
     });
     session->revive_if_needed();
 }
@@ -108,15 +108,21 @@ void AsyncOpenTask::unregister_download_progress_notifier(uint64_t token)
         m_session->unregister_progress_notifier(token);
 }
 
-void AsyncOpenTask::attach_to_subscription_initializer(AsyncOpenCallback&& callback,
-                                                       std::shared_ptr<_impl::RealmCoordinator> coordinator,
-                                                       bool rerun_on_launch)
+void AsyncOpenTask::wait_for_bootstrap_or_complete(AsyncOpenCallback&& callback,
+                                                   std::shared_ptr<_impl::RealmCoordinator> coordinator,
+                                                   Status status)
 {
-    // Attaching the subscription initializer to the latest subscription that was committed.
-    // This is going to be enough, for waiting that the subscription committed by init_subscription_initializer has
-    // been completed (either if it is the first time that the file is created or if rerun on launch was set to true).
-    // If the same Realm file is already opened, there is the possibility that this code may wait on a subscription
-    // that was not committed by init_subscription_initializer.
+    if (!status.is_ok()) {
+        async_open_complete(std::move(callback), coordinator, status);
+        return;
+    }
+
+    auto config = coordinator->get_config();
+    // FlX sync is not used so there is nothing to bootstrap.
+    if (!config.sync_config || !config.sync_config->flx_sync_requested) {
+        async_open_complete(std::move(callback), coordinator, status);
+        return;
+    }
 
     SharedRealm shared_realm;
     try {
@@ -126,13 +132,13 @@ void AsyncOpenTask::attach_to_subscription_initializer(AsyncOpenCallback&& callb
         async_open_complete(std::move(callback), coordinator, exception_to_status());
         return;
     }
-    const auto init_subscription = shared_realm->get_latest_subscription_set();
-    const auto sub_state = init_subscription.state();
+    const auto subscription_set = shared_realm->get_latest_subscription_set();
+    const auto sub_state = subscription_set.state();
 
-    if ((sub_state != sync::SubscriptionSet::State::Complete) || (m_db_first_open && rerun_on_launch)) {
+    if (sub_state != sync::SubscriptionSet::State::Complete) {
         // We need to wait until subscription initializer completes
         std::shared_ptr<AsyncOpenTask> self(shared_from_this());
-        init_subscription.get_state_change_notification(sync::SubscriptionSet::State::Complete)
+        subscription_set.get_state_change_notification(sync::SubscriptionSet::State::Complete)
             .get_async([self, coordinator, callback = std::move(callback)](
                            StatusWith<realm::sync::SubscriptionSet::State> state) mutable {
                 self->async_open_complete(std::move(callback), coordinator, state.get_status());
@@ -172,7 +178,7 @@ void AsyncOpenTask::async_open_complete(AsyncOpenCallback&& callback,
 }
 
 void AsyncOpenTask::migrate_schema_or_complete(AsyncOpenCallback&& callback,
-                                               std::shared_ptr<_impl::RealmCoordinator> coordinator, Status status)
+                                               std::shared_ptr<_impl::RealmCoordinator> coordinator)
 {
     util::CheckedUniqueLock lock(m_mutex);
     if (!m_session)
@@ -186,7 +192,7 @@ void AsyncOpenTask::migrate_schema_or_complete(AsyncOpenCallback&& callback,
     }();
 
     if (!pending_migration) {
-        wait_for_bootstrap_or_complete(std::move(callback), coordinator, status);
+        wait_for_bootstrap_or_complete(std::move(callback), coordinator, Status::OK());
         return;
     }
 
@@ -216,19 +222,4 @@ void AsyncOpenTask::migrate_schema_or_complete(AsyncOpenCallback&& callback,
         });
 }
 
-void AsyncOpenTask::wait_for_bootstrap_or_complete(AsyncOpenCallback&& callback,
-                                                   std::shared_ptr<_impl::RealmCoordinator> coordinator,
-                                                   Status status)
-{
-    auto config = coordinator->get_config();
-    if (config.sync_config && config.sync_config->flx_sync_requested &&
-        config.sync_config->subscription_initializer && status.is_ok()) {
-        const bool rerun_on_launch = config.sync_config->rerun_init_subscription_on_open;
-        attach_to_subscription_initializer(std::move(callback), coordinator, rerun_on_launch);
-    }
-    else {
-        async_open_complete(std::move(callback), coordinator, status);
-    }
-}
-
 } // namespace realm

src/realm/object-store/sync/async_open_task.hpp

@@ -68,10 +68,7 @@ class AsyncOpenTask : public std::enable_shared_from_this<AsyncOpenTask> {
 
     void async_open_complete(AsyncOpenCallback&&, std::shared_ptr<_impl::RealmCoordinator>, Status)
         REQUIRES(!m_mutex);
-    void attach_to_subscription_initializer(AsyncOpenCallback&&, std::shared_ptr<_impl::RealmCoordinator>, bool)
-        REQUIRES(!m_mutex);
-    void migrate_schema_or_complete(AsyncOpenCallback&&, std::shared_ptr<_impl::RealmCoordinator>, Status)
-        REQUIRES(!m_mutex);
+    void migrate_schema_or_complete(AsyncOpenCallback&&, std::shared_ptr<_impl::RealmCoordinator>) REQUIRES(!m_mutex);
     void wait_for_bootstrap_or_complete(AsyncOpenCallback&&, std::shared_ptr<_impl::RealmCoordinator>, Status)
         REQUIRES(!m_mutex);
 

test/object-store/sync/flx_sync.cpp

@@ -1380,9 +1380,10 @@ TEST_CASE("flx: client reset", "[sync][flx][client reset][baas]") {
                               .get_state_change_notification(sync::SubscriptionSet::State::Complete)
                               .get();
             CHECK(result == sync::SubscriptionSet::State::Complete);
-            wait_for_advance(*realm);
+            realm->sync_session()->shutdown_and_wait();
             realm->close();
         }
+        _impl::RealmCoordinator::assert_no_open_realms();
         {
             // ensure that an additional schema change after the successful reset is also accepted by the server
             changed_schema[0].persisted_properties.push_back(
@@ -1393,23 +1394,19 @@ TEST_CASE("flx: client reset", "[sync][flx][client reset][baas]") {
                                           {"str_field_2", PropertyType::String | PropertyType::Nullable},
                                       }});
             config_local.schema = changed_schema;
-            async_open_realm(config_local, [&](ThreadSafeReference&& ref, std::exception_ptr error) {
-                REQUIRE(ref);
-                REQUIRE_FALSE(error);
-                auto realm = Realm::get_shared_realm(std::move(ref));
-                auto table = realm->read_group().get_table("class_AddedClassSecond");
-                ColKey new_col = table->get_column_key("str_field_2");
-                REQUIRE(new_col);
-                auto new_subs = realm->get_latest_subscription_set().make_mutable_copy();
-                new_subs.insert_or_assign(Query(table).equal(new_col, "hello"));
-                auto subs = new_subs.commit();
-                realm->begin_transaction();
-                table->create_object_with_primary_key(Mixed{ObjectId::gen()}, {{new_col, "hello"}});
-                realm->commit_transaction();
-                subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();
-                wait_for_advance(*realm);
-                REQUIRE(table->size() == 1);
-            });
+            auto realm = Realm::get_shared_realm(config_local);
+            auto table = realm->read_group().get_table("class_AddedClassSecond");
+            ColKey new_col = table->get_column_key("str_field_2");
+            REQUIRE(new_col);
+            auto new_subs = realm->get_latest_subscription_set().make_mutable_copy();
+            new_subs.insert_or_assign(Query(table).equal(new_col, "hello"));
+            auto subs = new_subs.commit();
+            realm->begin_transaction();
+            table->create_object_with_primary_key(Mixed{ObjectId::gen()}, {{new_col, "hello"}});
+            realm->commit_transaction();
+            subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();
+            wait_for_advance(*realm);
+            REQUIRE(table->size() == 1);
         }
     }
 
@@ -4332,12 +4329,13 @@ TEST_CASE("flx: open realm + register subscription callback while bootstrapping"
           "[sync][flx][bootstrap][async open][baas]") {
     FLXSyncTestHarness harness("flx_bootstrap_and_subscribe");
     auto foo_obj_id = ObjectId::gen();
+    int64_t foo_obj_queryable_int = 5;
     harness.load_initial_data([&](SharedRealm realm) {
         CppContext c(realm);
         Object::create(c, realm, "TopLevel",
                        std::any(AnyDict{{"_id", foo_obj_id},
                                         {"queryable_str_field", "foo"s},
-                                        {"queryable_int_field", static_cast<int64_t>(5)},
+                                        {"queryable_int_field", foo_obj_queryable_int},
                                         {"non_queryable_field", "created as initial data seed"s}}));
     });
     SyncTestFile config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});
@@ -4594,6 +4592,62 @@ TEST_CASE("flx: open realm + register subscription callback while bootstrapping"
                 REQUIRE(r1_active <= 2);
             }
         }
+
+        SECTION("Wait to bootstrap all pending subscriptions even when subscription_initializer is not used") {
+            // Client 1
+            {
+                auto realm = Realm::get_shared_realm(config);
+                // Create subscription (version = 1) and bootstrap data.
+                subscribe_to_all_and_bootstrap(*realm);
+                realm->sync_session()->shutdown_and_wait();
+
+                // Create a new subscription (version = 2) while the session is closed.
+                // The new subscription does not match the object bootstrapped at version 1.
+                auto mutable_subscription = realm->get_latest_subscription_set().make_mutable_copy();
+                mutable_subscription.clear();
+                auto table = realm->read_group().get_table("class_TopLevel");
+                auto queryable_int_field = table->get_column_key("queryable_int_field");
+                mutable_subscription.insert_or_assign(
+                    Query(table).not_equal(queryable_int_field, foo_obj_queryable_int));
+                mutable_subscription.commit();
+
+                realm->close();
+            }
+
+            _impl::RealmCoordinator::assert_no_open_realms();
+
+            // Client 2 uploads data matching Client 1's subscription at version 1
+            harness.load_initial_data([&](SharedRealm realm) {
+                CppContext c(realm);
+                Object::create(c, realm, "TopLevel",
+                               std::any(AnyDict{{"_id", ObjectId::gen()},
+                                                {"queryable_str_field", "bar"s},
+                                                {"queryable_int_field", 2 * foo_obj_queryable_int},
+                                                {"non_queryable_field", "some data"s}}));
+            });
+
+            // Client 1 opens the realm asynchronously and expects the task to complete
+            // when the subscription at version 2 finishes bootstrapping.
+            auto async_open = Realm::get_synchronized_realm(config);
+            auto open_task_pf = util::make_promise_future<bool>();
+            int64_t latest_version, active_version;
+            auto open_realm_completed_callback =
+                [promise_holder = util::CopyablePromiseHolder(std::move(open_task_pf.promise)), &latest_version,
+                 &active_version](ThreadSafeReference ref, std::exception_ptr err) mutable {
+                    REQUIRE_FALSE(err);
+                    auto realm = Realm::get_shared_realm(std::move(ref));
+                    REQUIRE(realm);
+                    latest_version = realm->get_latest_subscription_set().version();
+                    active_version = realm->get_active_subscription_set().version();
+                    promise_holder.get_promise().emplace_value(true);
+                };
+            async_open->start(open_realm_completed_callback);
+            CHECK(open_task_pf.future.get());
+
+            // Check subscription at version 2 is marked Complete.
+            CHECK(active_version == 2);
+            CHECK(latest_version == active_version);
+        }
     }
 }
 TEST_CASE("flx sync: Client reset during async open", "[sync][flx][client reset][async open][baas]") {