RCORE-1872 Sync client should allow server bootstrapping at any time

CHANGELOG.md

@@ -2,7 +2,7 @@
 
 ### Enhancements
 * <New feature description> (PR [#????](https://github.com/realm/realm-core/pull/????))
-* None.
+* Add support for server initiated bootstraps. ([PR #7440](https://github.com/realm/realm-core/pull/7440))
 
 ### Fixed
 * <How do the end-user experience this issue? what was the impact?> ([#????](https://github.com/realm/realm-core/issues/????), since v?.?.?)

dependencies.yml

@@ -2,6 +2,7 @@ PACKAGE_NAME: realm-core
 VERSION: 14.6.2
 OPENSSL_VERSION: 3.2.0
 ZLIB_VERSION: 1.2.13
-# https://github.com/10gen/baas/commits
-# 5138f5c is 2024 Apr 19
-BAAS_VERSION: 5138f5c924f9f376d31cbc2b1819e1d934600277
+# TODO: Remove when switching to use Protocol version in RCORE-1972 and merging to master
+# 51d0e54 is BAAS-30091 - 2024 May 01
+BAAS_VERSION: 51d0e5411b9dab52cb24977d470c2c910d1b18c8
+BAAS_VERSION_TYPE: githash

evergreen/config.yml

@@ -234,8 +234,9 @@ functions:
         shell: bash
         env:
           BAASAAS_API_KEY: "${baasaas_api_key}"
+          # BAAS_VERSION and VERSION_TYPE are set by realm-core/dependencies.yml
           BAASAAS_REF_SPEC: "${BAAS_VERSION}"
-          BAASAAS_START_MODE: "githash"
+          BAASAAS_START_MODE: "${BAAS_VERSION_TYPE|githash}"
         script: |-
           set -o errexit
           set -o verbose

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

@@ -39,6 +39,8 @@
 #include <realm/sync/noinst/sync_schema_migration.hpp>
 #include <realm/sync/protocol.hpp>
 
+#include <ostream>
+
 using namespace realm;
 using namespace realm::_impl;
 
@@ -47,6 +49,23 @@ using SessionWaiterPointer = void (sync::Session::*)(util::UniqueFunction<void(s
 constexpr const char SyncError::c_original_file_path_key[];
 constexpr const char SyncError::c_recovery_file_path_key[];
 
+namespace realm {
+
+std::ostream& operator<<(std::ostream& out, const SyncSession::ConnectionState& val)
+{
+    switch (val) {
+        case SyncSession::ConnectionState::Disconnected:
+            return out << "DISCONNECTED";
+        case SyncSession::ConnectionState::Connecting:
+            return out << "CONNECTING";
+        case SyncSession::ConnectionState::Connected:
+            return out << "CONNECTED";
+    }
+    return out << "UNKNOWN";
+}
+
+} // namespace realm
+
 /// STATES:
 ///
 /// WAITING_FOR_ACCESS_TOKEN: a request has been initiated to ask

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

@@ -29,9 +29,10 @@
 #include <realm/util/future.hpp>
 #include <realm/version_id.hpp>
 
+#include <iosfwd>
+#include <map>
 #include <mutex>
 #include <unordered_map>
-#include <map>
 
 namespace realm {
 class DB;
@@ -536,6 +537,8 @@ class SyncSession : public std::enable_shared_from_this<SyncSession> {
     bool m_schema_migration_in_progress GUARDED_BY(m_state_mutex) = false;
 };
 
+std::ostream& operator<<(std::ostream& out, const SyncSession::ConnectionState& val);
+
 } // namespace realm
 
 #endif // REALM_OS_SYNC_SESSION_HPP

src/realm/sync/client.cpp

@@ -839,12 +839,8 @@ void SessionImpl::update_subscription_version_info()
 bool SessionImpl::process_flx_bootstrap_message(const SyncProgress& progress, DownloadBatchState batch_state,
                                                 int64_t query_version, const ReceivedChangesets& received_changesets)
 {
-    // Ignore the call if the session is not active
-    if (m_state != State::Active) {
-        return false;
-    }
-
-    if (is_steady_state_download_message(batch_state, query_version)) {
+    // Ignore the message if the session is not active or a steady state message
+    if (m_state != State::Active || batch_state == DownloadBatchState::SteadyState) {
         return false;
     }
 
@@ -1103,24 +1099,53 @@ SyncClientHookAction SessionImpl::call_debug_hook(SyncClientHookEvent event, con
     return call_debug_hook(data);
 }
 
-bool SessionImpl::is_steady_state_download_message(DownloadBatchState batch_state, int64_t query_version)
+bool SessionImpl::is_steady_state_download_message(DownloadBatchState batch_state, int64_t query_version,
+                                                   bool same_remote_versions)
 {
     // Should never be called if session is not active
     REALM_ASSERT_EX(m_state == State::Active, m_state);
-    if (batch_state == DownloadBatchState::SteadyState) {
-        return true;
+    // Query version should always be the same or increasing
+    REALM_ASSERT_3(query_version, >=, m_wrapper.m_flx_active_version);
+
+    // Return early if already steady state or PBS (doesn't use bootstraps)
+    if (batch_state == DownloadBatchState::SteadyState || !m_is_flx_sync_session) {
+        return true; // Steady state
     }
 
-    if (!m_is_flx_sync_session) {
-        return true;
+    // Bootstrap messages (i.e. non-steady state) are identified by:
+    // * DownloadBatchState of MoreToCome
+    // * DownloadBatchState of LastInBatch, and
+    //   * first LastInBatch=true after one or more MoreToCome messages
+    //   * query_version greater than the active query_version
+    //   * LastInBatch=true message has 2 or more changesets and all have the same
+    //     remote_version
+
+    // LastInBatch=False messages are always a bootstrap message
+    if (batch_state == DownloadBatchState::MoreToCome) {
+        return false; // Not steady state
     }
 
-    // If this is a steady state DOWNLOAD, no need for special handling.
-    if (batch_state == DownloadBatchState::LastInBatch && query_version == m_wrapper.m_flx_active_version) {
-        return true;
+    // Messages with query_version greater than the active version are always a
+    // bootstrap message
+    if (query_version > m_wrapper.m_flx_active_version) {
+        return false; // Not steady state
+    }
+
+    // If this is the first LastInBatch=true message after one or more
+    // LastInBatch=false messages, this is the end of a bootstrap
+    if (m_last_download_batch_state == DownloadBatchState::MoreToCome) {
+        return false; // Not steady state
+    }
+
+    // Otherwise, if this is a LastInBatch=true message whose query version matches
+    // the current active version and all the changesets in the message have the
+    // same remote version, then this is a server initiated single message bootstrap
+    if (same_remote_versions) {
+        return false; // Not steady state
     }
 
-    return false;
+    // If none of the previous checks were successful, then this is a steady state msg
+    return true; // Steady state
 }
 
 void SessionImpl::init_progress_handler()

src/realm/sync/noinst/client_impl_base.cpp

@@ -1891,7 +1891,7 @@ void Session::send_bind_message()
     session_ident_type session_ident = m_ident;
     bool need_client_file_ident = !have_client_file_ident();
     const bool is_subserver = false;
-
+    m_last_download_batch_state = DownloadBatchState::SteadyState;
 
     ClientProtocol& protocol = m_conn.get_client_protocol();
     int protocol_version = m_conn.get_negotiated_protocol_version();
@@ -2373,8 +2373,6 @@ Status Session::receive_download_message(const DownloadMessage& message)
     // If this is a PBS connection, then every download message is its own complete batch.
     bool last_in_batch = is_flx ? *message.last_in_batch : true;
     auto batch_state = last_in_batch ? sync::DownloadBatchState::LastInBatch : sync::DownloadBatchState::MoreToCome;
-    if (is_steady_state_download_message(batch_state, query_version))
-        batch_state = DownloadBatchState::SteadyState;
 
     auto&& progress = message.progress;
     if (is_flx) {
@@ -2414,21 +2412,39 @@ Status Session::receive_download_message(const DownloadMessage& message)
         return status;
     }
 
-    version_type server_version = m_progress.download.server_version;
+    // Start with the download server version from the last download message, since the changesets in the new
+    // download message must have a remote version greater than (or equal to for FLX) this value.
+    version_type last_remote_version = m_progress.download.server_version;
     version_type last_integrated_client_version = m_progress.download.last_integrated_client_version;
+    // If there are 2 or more changesets in this message, track whether or not the changesets all have the same
+    // download_server_version to help with determining if this is a single message server-initiated bootstrap or not.
+    bool same_remote_version = last_in_batch && message.changesets.size() > 1;
+    bool after_first_changeset = false;
+
     for (const RemoteChangeset& changeset : message.changesets) {
-        // Check that per-changeset server version is strictly increasing, except in FLX sync where the server
-        // version must be increasing, but can stay the same during bootstraps.
-        bool good_server_version = m_is_flx_sync_session ? (changeset.remote_version >= server_version)
-                                                         : (changeset.remote_version > server_version);
+        // Check that per-changeset server version is strictly increasing since the last download server version,
+        // except in FLX sync where the server version must be increasing, but can stay the same during bootstraps.
+        bool good_server_version = m_is_flx_sync_session ? (changeset.remote_version >= last_remote_version)
+                                                         : (changeset.remote_version > last_remote_version);
         // Each server version cannot be greater than the one in the header of the download message.
         good_server_version = good_server_version && (changeset.remote_version <= progress.download.server_version);
         if (!good_server_version) {
             return {ErrorCodes::SyncProtocolInvariantFailed,
                     util::format("Bad server version in changeset header (DOWNLOAD) (%1, %2, %3)",
-                                 changeset.remote_version, server_version, progress.download.server_version)};
+                                 changeset.remote_version, last_remote_version, progress.download.server_version)};
+        }
+        // Check to see if all the changesets in this LastInBatch=true message have the same remote_version
+        // If so, this is a server-initiated single message bootstrap
+        if (same_remote_version && after_first_changeset) {
+            // After the first changeset compare the previous changeset's server version to the current changeset
+            // server version. If they are different then this is definitely not a bootstrap message
+            same_remote_version = changeset.remote_version == last_remote_version;
         }
-        server_version = changeset.remote_version;
+        // Skip the first changeset, since `server_version` contains the incorrect value for this check
+        after_first_changeset = true;
+        // Save the remote version for the current changeset to compare against the next changeset
+        last_remote_version = changeset.remote_version;
+
         // Check that per-changeset last integrated client version is "weakly"
         // increasing.
         bool good_client_version =
@@ -2453,6 +2469,11 @@ Status Session::receive_download_message(const DownloadMessage& message)
         }
     }
 
+    if (is_steady_state_download_message(batch_state, query_version, same_remote_version)) {
+        batch_state = DownloadBatchState::SteadyState;
+    }
+    m_last_download_batch_state = batch_state;
+
     auto hook_action = call_debug_hook(SyncClientHookEvent::DownloadMessageReceived, progress, query_version,
                                        batch_state, message.changesets.size());
     if (hook_action == SyncClientHookAction::EarlyReturn) {

src/realm/sync/noinst/client_impl_base.hpp

@@ -1133,6 +1133,10 @@ class ClientImpl::Session {
     // the detection of download completion.
     request_ident_type m_last_triggering_download_mark = 0;
 
+    // Keep track of the DownloadBatchState for the last download message to help
+    // determine if the current LastInBatch came after a message with MoreToCome.
+    DownloadBatchState m_last_download_batch_state = DownloadBatchState::SteadyState;
+
     SessionWrapper& m_wrapper;
 
     request_ident_type m_last_pending_test_command_ident = 0;
@@ -1197,7 +1201,8 @@ class ClientImpl::Session {
     SyncClientHookAction call_debug_hook(SyncClientHookEvent event, const ProtocolErrorInfo&);
     SyncClientHookAction call_debug_hook(const SyncClientHookData& data);
 
-    bool is_steady_state_download_message(DownloadBatchState batch_state, int64_t query_version);
+    bool is_steady_state_download_message(DownloadBatchState batch_state, int64_t query_version,
+                                          bool same_remote_versions);
 
     void init_progress_handler();
     void enable_progress_notifications();

src/realm/sync/noinst/pending_bootstrap_store.cpp

@@ -169,6 +169,7 @@ void PendingBootstrapStore::add_batch(int64_t query_version, util::Optional<Sync
         BinaryData compressed_data(compressed_changesets[idx].data(), compressed_changesets[idx].size());
         cur_changeset.set(m_changeset_data, compressed_data);
     }
+    size_t total_changesets = changesets_list.size();
 
     tr->commit();
 
@@ -177,15 +178,18 @@ void PendingBootstrapStore::add_batch(int64_t query_version, util::Optional<Sync
     }
 
     if (did_create) {
-        m_logger.debug(util::LogCategory::changeset, "Created new pending bootstrap object for query version %1",
-                       query_version);
+        m_logger.debug(util::LogCategory::changeset,
+                       "Created new pending bootstrap object with %1 changesets for query version %2",
+                       total_changesets, query_version);
     }
     else {
-        m_logger.debug(util::LogCategory::changeset, "Added batch to pending bootstrap object for query version %1",
-                       query_version);
+        m_logger.debug(util::LogCategory::changeset,
+                       "Added batch of %1 changesets (%2 total) to pending bootstrap object for query version %3",
+                       changesets.size(), total_changesets, query_version);
     }
     if (progress) {
-        m_logger.debug(util::LogCategory::changeset, "Finalized pending bootstrap object for query version %1",
+        m_logger.debug(util::LogCategory::changeset,
+                       "Finalized pending bootstrap object with %1 changesets for query version %2", total_changesets,
                        query_version);
     }
     m_has_pending = true;

src/realm/sync/protocol.hpp

@@ -57,6 +57,11 @@ namespace sync {
 //      Server replaces 'downloadable_bytes' (which was always zero prior this version)
 //      with an estimated progress value (double from 0.0 to 1.0) for flx sessions
 //
+//   13 Support for syncing collections (lists and dictionaries) in Mixed columns
+//
+//   14 Support for server initiated bootstraps, including bootstraps for role/
+//      permissions changes instead of performing a client reset when changed.
+//
 //  XX Changes:
 //     - TBD
 //