Fix race in send_upload_message

CHANGELOG.md

@@ -9,6 +9,8 @@
 * Potential stack-use-after-scope issue on changesets integration with msvc-2019 and mpack code ([PR #6911](/~https://github.com/realm/realm-core/pull/6911))
 * Fix compilation with non-beta Xcode 15. Building for visionOS now requires explicitly specifying `-DCMAKE_XCODE_ATTRIBUTE_SDKROOT=xros` (PR [#7055](/~https://github.com/realm/realm-core/pull/7055)).
 * Fixed FLX subscriptions not being sent to the server if the session was interrupted during bootstrapping. ([#7077](/~https://github.com/realm/realm-core/issues/7077), since v11.8.0)
+* Fixed FLX subscriptions not being sent to the server if an upload message was sent immediately after a subscription was committed but before the sync client checks for new subscriptions via `SubscriptionStore::get_next_pending_version()`. ([#7076](/~https://github.com/realm/realm-core/issues/7076), since v13.23.1)
+* Fixed application crash with 'KeyNotFound' exception when subscriptions are marked complete after a client reset. ([#7090](/~https://github.com/realm/realm-core/issues/7090), since v12.3.0)
 
 ### Breaking changes
 * None.

src/realm/sync/client.cpp

@@ -123,6 +123,8 @@ class SessionWrapper final : public util::AtomicRefCountBase, DB::CommitListener
 
     void handle_pending_client_reset_acknowledgement();
 
+    void update_subscription_version_info();
+
     std::string get_appservices_connection_id();
 
 private:
@@ -784,6 +786,13 @@ void SessionImpl::handle_pending_client_reset_acknowledgement()
     }
 }
 
+void SessionImpl::update_subscription_version_info()
+{
+    // Ignore the call if the session is not active
+    if (m_state == State::Active) {
+        m_wrapper.update_subscription_version_info();
+    }
+}
 
 bool SessionImpl::process_flx_bootstrap_message(const SyncProgress& progress, DownloadBatchState batch_state,
                                                 int64_t query_version, const ReceivedChangesets& received_changesets)
@@ -1129,11 +1138,7 @@ SessionWrapper::SessionWrapper(ClientImpl& client, DBRef db, std::shared_ptr<Sub
     REALM_ASSERT(m_db->get_replication());
     REALM_ASSERT(dynamic_cast<ClientReplication*>(m_db->get_replication()));
 
-    if (m_flx_subscription_store) {
-        auto versions_info = m_flx_subscription_store->get_version_info();
-        m_flx_active_version = versions_info.active;
-        m_flx_pending_mark_version = versions_info.pending_mark;
-    }
+    update_subscription_version_info();
 }
 
 SessionWrapper::~SessionWrapper() noexcept
@@ -1821,6 +1826,15 @@ void SessionWrapper::handle_pending_client_reset_acknowledgement()
     });
 }
 
+void SessionWrapper::update_subscription_version_info()
+{
+    if (!m_flx_subscription_store)
+        return;
+    auto versions_info = m_flx_subscription_store->get_version_info();
+    m_flx_active_version = versions_info.active;
+    m_flx_pending_mark_version = versions_info.pending_mark;
+}
+
 std::string SessionWrapper::get_appservices_connection_id()
 {
     auto pf = util::make_promise_future<std::string>();

src/realm/sync/noinst/client_impl_base.cpp

@@ -2000,14 +2000,11 @@ void Session::send_upload_message()
     if (REALM_UNLIKELY(get_client().is_dry_run()))
         return;
 
-    version_type target_upload_version = get_db()->get_version_of_latest_snapshot();
+    version_type target_upload_version = m_last_version_available;
     if (m_pending_flx_sub_set) {
         REALM_ASSERT(m_is_flx_sync_session);
         target_upload_version = m_pending_flx_sub_set->snapshot_version;
     }
-    if (target_upload_version > m_last_version_available) {
-        m_last_version_available = target_upload_version;
-    }
 
     const ClientReplication& repl = access_realm(); // Throws
 
@@ -2305,6 +2302,8 @@ Status Session::receive_ident_message(SaltedFileIdent client_file_ident)
             handle_pending_client_reset_acknowledgement();
         }
 
+        update_subscription_version_info();
+
         // If a migration or rollback is in progress, mark it complete when client reset is completed.
         if (auto migration_store = get_migration_store()) {
             migration_store->complete_migration_or_rollback();

src/realm/sync/noinst/client_impl_base.hpp

@@ -1013,6 +1013,8 @@ class ClientImpl::Session {
 
     void handle_pending_client_reset_acknowledgement();
 
+    void update_subscription_version_info();
+
     void gather_pending_compensating_writes(util::Span<Changeset> changesets, std::vector<ProtocolErrorInfo>* out);
 
     void begin_resumption_delay(const ProtocolErrorInfo& error_info);

src/realm/sync/noinst/client_reset.cpp

@@ -575,8 +575,9 @@ LocalVersionIDs perform_client_reset_diff(DBRef db_local, DBRef db_remote, sync:
         if (!sub_store) {
             return;
         }
-        auto mut_subs = sub_store->get_active().make_mutable_copy();
-        int64_t before_version = mut_subs.version();
+        auto subs = sub_store->get_active();
+        int64_t before_version = subs.version();
+        auto mut_subs = subs.make_mutable_copy();
         mut_subs.update_state(sync::SubscriptionSet::State::Complete);
         auto sub = std::move(mut_subs).commit();
         if (on_flx_version_complete) {