RCORE-2126 Clear incomplete bootstraps when the connection is established

CHANGELOG.md

@@ -6,7 +6,7 @@
 
 ### Fixed
 * <How do the end-user experience this issue? what was the impact?> ([#????](https://github.com/realm/realm-core/issues/????), since v?.?.?)
-* None.
+* If the client disconnects while a bootstrap is in progresss, stale data from the previous bootstrap may be included when the client reconnects and the bootstrap is restarted. This can lead to objects stored in the database that do not match the actual state of the server and potentially leading to compensating writes. ([#7707](https://github.com/realm/realm-core/issues/7707), since v12.0.0)
 
 ### Breaking changes
 * None.

src/realm/sync/client.cpp

@@ -827,15 +827,7 @@ bool SessionImpl::process_flx_bootstrap_message(const DownloadMessage& message)
         return true;
     }
 
-    try {
-        process_pending_flx_bootstrap(); // throws
-    }
-    catch (const IntegrationException& e) {
-        on_integration_failure(e);
-    }
-    catch (...) {
-        on_integration_failure(IntegrationException(exception_to_status()));
-    }
+    try_process_pending_flx_bootstrap();
 
     return true;
 }
@@ -847,6 +839,10 @@ void SessionImpl::process_pending_flx_bootstrap()
     if (!m_is_flx_sync_session || m_state != State::Active) {
         return;
     }
+    // Ignore the call if the client detects an error.
+    if (m_client_error) {
+        return;
+    }
     auto bootstrap_store = m_wrapper.get_flx_pending_bootstrap_store();
     if (!bootstrap_store->has_pending()) {
         return;
@@ -917,6 +913,19 @@ void SessionImpl::process_pending_flx_bootstrap()
     REALM_ASSERT_EX(action == SyncClientHookAction::NoAction || action == SyncClientHookAction::EarlyReturn, action);
 }
 
+void SessionImpl::try_process_pending_flx_bootstrap()
+{
+    try {
+        process_pending_flx_bootstrap(); // throws
+    }
+    catch (const IntegrationException& error) {
+        on_integration_failure(error);
+    }
+    catch (...) {
+        on_integration_failure(IntegrationException(exception_to_status()));
+    }
+}
+
 void SessionImpl::on_flx_sync_error(int64_t version, std::string_view err_msg)
 {
     // Ignore the call if the session is not active

src/realm/sync/noinst/client_impl_base.cpp

@@ -1296,7 +1296,7 @@ Session* Connection::find_and_validate_session(session_ident_type session_ident,
         logger.error("Bad session identifier in %1 message, session_ident = %2", message, session_ident);
         close_due_to_protocol_error(
             {ErrorCodes::SyncProtocolInvariantFailed,
-             util::format("Received message %1 for session iden %2 when that session never existed", message,
+             util::format("Received message %1 for session ident %2 when that session never existed", message,
                           session_ident)});
     }
     else {
@@ -1513,15 +1513,7 @@ void Session::cancel_resumption_delay()
     if (unbind_process_complete())
         initiate_rebind(); // Throws
 
-    try {
-        process_pending_flx_bootstrap(); // throws
-    }
-    catch (const IntegrationException& error) {
-        on_integration_failure(error);
-    }
-    catch (...) {
-        on_integration_failure(IntegrationException(exception_to_status()));
-    }
+    try_process_pending_flx_bootstrap();
 
     m_conn.one_more_active_unsuspended_session(); // Throws
     if (m_try_again_activation_timer) {
@@ -1709,15 +1701,7 @@ void Session::activate()
     REALM_ASSERT(!m_suspended);
     m_conn.one_more_active_unsuspended_session(); // Throws
 
-    try {
-        process_pending_flx_bootstrap(); // throws
-    }
-    catch (const IntegrationException& error) {
-        on_integration_failure(error);
-    }
-    catch (...) {
-        on_integration_failure(IntegrationException(exception_to_status()));
-    }
+    try_process_pending_flx_bootstrap();
 
     // Checks if there is a pending client reset
     handle_pending_client_reset_acknowledgement();

src/realm/sync/noinst/client_impl_base.hpp

@@ -932,6 +932,7 @@ class ClientImpl::Session {
 
     // Processes any pending FLX bootstraps, if one exists. Otherwise this is a noop.
     void process_pending_flx_bootstrap();
+    void try_process_pending_flx_bootstrap();
 
     bool client_reset_if_needed();
     void handle_pending_client_reset_acknowledgement();
@@ -941,7 +942,6 @@ class ClientImpl::Session {
     void begin_resumption_delay(const ProtocolErrorInfo& error_info);
     void clear_resumption_delay_state();
 
-private:
     Connection& m_conn;
     const session_ident_type m_ident;
 
@@ -1398,6 +1398,8 @@ inline void ClientImpl::Session::connection_established(bool fast_reconnect)
     // the bind messsage
     call_debug_hook(SyncClientHookEvent::SessionConnected);
 
+    try_process_pending_flx_bootstrap();
+
     if (!m_suspended) {
         // Ready to send BIND message
         enlist_to_send(); // Throws

test/object-store/sync/flx_sync.cpp

@@ -5174,6 +5174,44 @@ TEST_CASE("flx: no upload during bootstraps", "[sync][flx][bootstrap][baas]") {
     wait_for_download(*realm);
 }
 
+TEST_CASE("flx: bootstrap is properly applied when the connection is reestablished", "[sync][flx][bootstrap][baas]") {
+    FLXSyncTestHarness harness("flx_bootstrap", {g_large_array_schema, {"queryable_int_field"}});
+    auto& app_session = harness.session().app_session();
+    REQUIRE(app_session.admin_api.patch_app_settings(app_session.server_app_id,
+                                                     {{"sync", {{"num_objects_before_bootstrap_flush", 1}}}}));
+    REQUIRE(app_session.admin_api.patch_app_settings(
+        app_session.server_app_id, {{"sync", {{"qbs_download_changeset_soft_max_byte_size", 1000}}}}));
+
+    fill_large_array_schema(harness);
+    auto config = harness.make_test_file();
+    bool once = false;
+    config.sync_config->on_sync_client_event_hook = [&once](std::weak_ptr<SyncSession>,
+                                                            const SyncClientHookData& data) {
+        if (data.query_version != 1) {
+            return SyncClientHookAction::NoAction;
+        }
+        if (data.event == SyncClientHookEvent::BootstrapMessageProcessed && !once) {
+            once = true;
+            return SyncClientHookAction::TriggerReconnect;
+        }
+        // The batch of changesets added to the PendingBoostrapStore before disconnect
+        // were removed when the connection was reestablished.
+        // There are 5 changesets in 5 download messages and one additional download message
+        // with an empty changeset (as per server design).
+        if (data.event == SyncClientHookEvent::BootstrapProcessed) {
+            CHECK(data.num_changesets == 6);
+        }
+        return SyncClientHookAction::NoAction;
+    };
+
+    auto realm = Realm::get_shared_realm(config);
+    auto table = realm->read_group().get_table("class_TopLevel");
+    auto new_subs = realm->get_latest_subscription_set().make_mutable_copy();
+    new_subs.insert_or_assign(Query(table));
+    auto subs = new_subs.commit();
+    subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();
+}
+
 } // namespace realm::app
 
 #endif // REALM_ENABLE_AUTH_TESTS