Clear incomplete bootstraps when the connection is established

Author: danieltabacaru

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/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 {
@@ -1709,16 +1709,6 @@ 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()));
-    }
-
     // Checks if there is a pending client reset
     handle_pending_client_reset_acknowledgement();
 }

src/realm/sync/noinst/client_impl_base.hpp

@@ -1398,6 +1398,16 @@ inline void ClientImpl::Session::connection_established(bool fast_reconnect)
     // the bind messsage
     call_debug_hook(SyncClientHookEvent::SessionConnected);
 
+    try {
+        process_pending_flx_bootstrap(); // throws
+    }
+    catch (const IntegrationException& error) {
+        on_integration_failure(error);
+    }
+    catch (...) {
+        on_integration_failure(IntegrationException(exception_to_status()));
+    }
+
     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