Fix a data race on write_validator_factory

PendingBootstrapStore is created early in session initialization, which
performs a write and thus reads the write validator. This can race with setting
the write validator on the thread which created the sync session, and tsan
complains about it.

Moving the initialization to SyncSession's constructor ensures that it's set
before we create the `sync::Session`.

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 ### Fixed
 * Fixed a segfault in sync compiled by MSVC 2022. ([#5557](https://github.com/realm/realm-core/pull/5557), since 12.1.0)
+* Fix a data race when openg a flexible sync Realm (since v12.1.0).
 
 ### Breaking changes
 * None.

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

@@ -39,7 +39,6 @@
 #include <realm/object-store/sync/sync_user.hpp>
 #include <realm/sync/history.hpp>
 #include <realm/sync/noinst/client_history_impl.hpp>
-#include "realm/sync/subscriptions.hpp"
 #endif
 
 #include <realm/db.hpp>
@@ -103,30 +102,6 @@ void RealmCoordinator::create_sync_session()
         return;
 
     m_sync_session = m_config.sync_config->user->sync_manager()->get_session(m_db, *m_config.sync_config);
-    if (m_config.sync_config && m_config.sync_config->flx_sync_requested) {
-        std::weak_ptr<sync::SubscriptionStore> weak_sub_mgr(m_sync_session->get_flx_subscription_store());
-        auto& history = static_cast<sync::ClientReplication&>(*m_db->get_replication());
-        history.set_write_validator_factory(
-            [weak_sub_mgr](Transaction& tr) -> util::UniqueFunction<sync::SyncReplication::WriteValidator> {
-                auto sub_mgr = weak_sub_mgr.lock();
-                if (!sub_mgr) {
-                    throw std::runtime_error("Subscription store was destroyed while user writes were on-going");
-                }
-
-                auto latest_sub_tables = sub_mgr->get_tables_for_latest(tr);
-                return [tables = std::move(latest_sub_tables)](const Table& table) {
-                    if (table.get_table_type() != Table::Type::TopLevel) {
-                        return;
-                    }
-                    auto object_class_name = Group::table_name_to_class_name(table.get_name());
-                    if (tables.find(object_class_name) == tables.end()) {
-                        throw NoSubscriptionForWrite(util::format(
-                            "Cannot write to class %1 when no flexible sync subscription has been created.",
-                            object_class_name));
-                    }
-                };
-            });
-    }
 
     std::weak_ptr<RealmCoordinator> weak_self = shared_from_this();
     SyncSession::Internal::set_sync_transact_callback(*m_sync_session, [weak_self](VersionID, VersionID) {

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

@@ -256,6 +256,26 @@ SyncSession::SyncSession(SyncClient& client, std::shared_ptr<DB> db, SyncConfig
     , m_client(client)
     , m_sync_manager(sync_manager)
 {
+    if (m_config.flx_sync_requested) {
+        auto& history = static_cast<sync::ClientReplication&>(*m_db->get_replication());
+        history.set_write_validator_factory([weak_sub_mgr = std::weak_ptr{m_flx_subscription_store}](Transaction& tr)
+                                                -> util::UniqueFunction<sync::SyncReplication::WriteValidator> {
+            auto sub_mgr = weak_sub_mgr.lock();
+            REALM_ASSERT_RELEASE(sub_mgr);
+            auto latest_sub_tables = sub_mgr->get_tables_for_latest(tr);
+            return [tables = std::move(latest_sub_tables)](const Table& table) {
+                if (table.get_table_type() != Table::Type::TopLevel) {
+                    return;
+                }
+                auto object_class_name = Group::table_name_to_class_name(table.get_name());
+                if (tables.find(object_class_name) == tables.end()) {
+                    throw NoSubscriptionForWrite(
+                        util::format("Cannot write to class %1 when no flexible sync subscription has been created.",
+                                     object_class_name));
+                }
+            };
+        });
+    }
 }
 
 std::shared_ptr<SyncManager> SyncSession::sync_manager() const