Fix crash when opening FLX realm after client reset failure

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?.?.?)
 * Fix the query parser needs to copy list of arguments and own the memory. ([#6674](https://github.com/realm/realm-core/pull/6674), since v12.5.0)
+* Fixed a potential crash when opening the realm after failing to download a fresh FLX realm during an automatic client reset ([#6494](https://github.com/realm/realm-core/issues/6494), since v12.3.0)
 
 
 ### Breaking changes

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

@@ -577,16 +577,7 @@ void SyncSession::handle_fresh_realm_downloaded(DBRef db, Status status,
             return;
         }
         lock.unlock();
-        if (auto local_subs_store = get_flx_subscription_store()) {
-            // In DiscardLocal mode, only the active subscription set is preserved
-            // this means that we have to remove all other subscriptions including later
-            // versioned ones.
-            auto mut_sub = local_subs_store->get_active().make_mutable_copy();
-            local_subs_store->supercede_all_except(mut_sub);
-            mut_sub.update_state(sync::SubscriptionSet::State::Error,
-                                 util::make_optional<std::string_view>(status.reason()));
-            std::move(mut_sub).commit();
-        }
+
         const bool try_again = false;
         sync::SessionErrorInfo synthetic(
             make_error_code(sync::Client::Error::auto_client_reset_failure),

src/realm/sync/subscriptions.cpp

@@ -901,35 +901,6 @@ void SubscriptionStore::supercede_prior_to(TransactionRef tr, int64_t version_id
     remove_query.remove();
 }
 
-void SubscriptionStore::supercede_all_except(MutableSubscriptionSet& mut_sub) const
-{
-    auto version_to_keep = mut_sub.version();
-    supercede_prior_to(mut_sub.m_tr, version_to_keep);
-
-    std::list<SubscriptionStore::NotificationRequest> to_finish;
-    std::unique_lock<std::mutex> lk(m_pending_notifications_mutex);
-    m_pending_notifications_cv.wait(lk, [&] {
-        return m_outstanding_requests == 0;
-    });
-    for (auto it = m_pending_notifications.begin(); it != m_pending_notifications.end();) {
-        if (it->version != version_to_keep) {
-            to_finish.splice(to_finish.end(), m_pending_notifications, it++);
-        }
-        else {
-            ++it;
-        }
-    }
-
-    REALM_ASSERT_EX(version_to_keep >= m_min_outstanding_version, version_to_keep, m_min_outstanding_version);
-    m_min_outstanding_version = version_to_keep;
-
-    lk.unlock();
-
-    for (auto& req : to_finish) {
-        req.promise.emplace_value(SubscriptionSet::State::Superseded);
-    }
-}
-
 MutableSubscriptionSet SubscriptionStore::make_mutable_copy(const SubscriptionSet& set) const
 {
     auto new_tr = m_db->start_write();

src/realm/sync/subscriptions.hpp

@@ -330,10 +330,6 @@ class SubscriptionStore : public std::enable_shared_from_this<SubscriptionStore>
     // version ID. If there is no SubscriptionSet with that version ID, this throws KeyNotFound.
     SubscriptionSet get_by_version(int64_t version_id) const;
 
-    // Fulfill all previous subscriptions by superceding them. This does not
-    // affect the mutable subscription identified by the parameter.
-    void supercede_all_except(MutableSubscriptionSet& mut_sub) const;
-
     // Returns true if there have been commits to the DB since the given version
     bool would_refresh(DB::version_type version) const noexcept;
 

test/object-store/sync/flx_sync.cpp

@@ -40,6 +40,7 @@
 #include "realm/sync/client_base.hpp"
 #include "realm/sync/config.hpp"
 #include "realm/sync/noinst/client_history_impl.hpp"
+#include <realm/sync/noinst/client_reset_operation.hpp>
 #include "realm/sync/noinst/pending_bootstrap_store.hpp"
 #include "realm/sync/noinst/server/access_token.hpp"
 #include "realm/sync/protocol.hpp"
@@ -282,6 +283,11 @@ TEST_CASE("flx: client reset", "[sync][flx][app][client reset]") {
              {"list_of_ints_field", PropertyType::Int | PropertyType::Array},
              {"sum_of_list_field", PropertyType::Int},
          }},
+        {"TopLevel2",
+         {
+             {"_id", PropertyType::ObjectId, Property::IsPrimary{true}},
+             {"queryable_str_field", PropertyType::String | PropertyType::Nullable},
+         }},
     };
 
     // some of these tests make additive schema changes which is only allowed in dev mode
@@ -432,6 +438,58 @@ TEST_CASE("flx: client reset", "[sync][flx][app][client reset]") {
             ->run();
     }
 
+    SECTION("Recover: subscription and offline writes after client reset failure") {
+        config_local.sync_config->client_resync_mode = ClientResyncMode::Recover;
+        auto&& [error_future, error_handler] = make_error_handler();
+        config_local.sync_config->error_handler = error_handler;
+
+        std::string fresh_path = realm::_impl::ClientResetOperation::get_fresh_path_for(config_local.path);
+        // create a non-empty directory that we'll fail to delete
+        util::make_dir(fresh_path);
+        util::File(util::File::resolve("file", fresh_path), util::File::mode_Write);
+
+        auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());
+        test_reset
+            ->make_local_changes([&](SharedRealm local_realm) {
+                auto mut_sub = local_realm->get_latest_subscription_set().make_mutable_copy();
+                auto table = local_realm->read_group().get_table("class_TopLevel2");
+                mut_sub.insert_or_assign(Query(table));
+                mut_sub.commit();
+
+                CppContext c(local_realm);
+                local_realm->begin_transaction();
+                Object::create(c, local_realm, "TopLevel2",
+                               std::any(AnyDict{{"_id"s, ObjectId::gen()}, {"queryable_str_field"s, "foo"s}}));
+                local_realm->commit_transaction();
+            })
+            ->on_post_reset([](SharedRealm local_realm) {
+                // Verify offline subscription was not removed.
+                auto subs = local_realm->get_latest_subscription_set();
+                auto table = local_realm->read_group().get_table("class_TopLevel2");
+                REQUIRE(subs.find(Query(table)));
+            })
+            ->run();
+
+        // Remove the folder preventing the completion of a client reset.
+        util::try_remove_dir_recursive(fresh_path);
+
+        auto config_copy = config_local;
+        config_local.sync_config->error_handler = nullptr;
+        auto&& [reset_future, reset_handler] = make_client_reset_handler();
+        config_copy.sync_config->notify_after_client_reset = reset_handler;
+
+        // Attempt to open the realm again.
+        // This time the client reset succeeds and the offline subscription and writes are recovered.
+        auto realm = Realm::get_shared_realm(config_copy);
+        ClientResyncMode mode = reset_future.get();
+        REQUIRE(mode == ClientResyncMode::Recover);
+
+        auto table = realm->read_group().get_table("class_TopLevel2");
+        auto str_col = table->get_column_key("queryable_str_field");
+        REQUIRE(table->size() == 1);
+        REQUIRE(table->get_object(0).get<String>(str_col) == "foo");
+    }
+
     SECTION("Recover: offline writes and subscriptions (multiple subscriptions)") {
         config_local.sync_config->client_resync_mode = ClientResyncMode::Recover;
         auto&& [reset_future, reset_handler] = make_client_reset_handler();
@@ -811,6 +869,36 @@ TEST_CASE("flx: client reset", "[sync][flx][app][client reset]") {
             })
             ->run();
     }
+
+    SECTION("DiscardLocal: open realm after client reset failure") {
+        config_local.sync_config->client_resync_mode = ClientResyncMode::DiscardLocal;
+        auto&& [error_future, error_handler] = make_error_handler();
+        config_local.sync_config->error_handler = error_handler;
+
+        std::string fresh_path = realm::_impl::ClientResetOperation::get_fresh_path_for(config_local.path);
+        // create a non-empty directory that we'll fail to delete
+        util::make_dir(fresh_path);
+        util::File(util::File::resolve("file", fresh_path), util::File::mode_Write);
+
+        auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());
+        test_reset->run();
+
+        // Client reset fails due to sync client not being able to create the fresh realm.
+        auto sync_error = wait_for_future(std::move(error_future)).get();
+        CHECK(sync_error.get_system_error() == sync::make_error_code(sync::ClientError::auto_client_reset_failure));
+
+        // Open the realm again. This should not crash.
+        {
+            auto config_copy = config_local;
+            auto&& [err_future, err_handler] = make_error_handler();
+            config_local.sync_config->error_handler = err_handler;
+
+            auto realm_post_reset = Realm::get_shared_realm(config_copy);
+            auto sync_error = wait_for_future(std::move(err_future)).get();
+            CHECK(sync_error.get_system_error() ==
+                  sync::make_error_code(sync::ClientError::auto_client_reset_failure));
+        }
+    }
 }
 
 TEST_CASE("flx: creating an object on a class with no subscription throws", "[sync][flx][app]") {