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

CHANGELOG.md

@@ -4,6 +4,7 @@
 * <New feature description> (PR [#????](https://github.com/realm/realm-core/pull/????))
 * Include the originating client reset error in AutoClientResetFailure errors. ([#7761](https://github.com/realm/realm-core/pull/7761))
 * Reduce the size of the local transaction log produced by creating objects, improving the performance of insertion-heavy transactions ([PR #7734](https://github.com/realm/realm-core/pull/7734)).
+* 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

@@ -3,5 +3,6 @@ VERSION: 14.9.0
 OPENSSL_VERSION: 3.2.0
 ZLIB_VERSION: 1.2.13
 # https://github.com/10gen/baas/commits
-# 9d1b4d6 is 2024 May 8
-BAAS_VERSION: 9d1b4d628babadfb606ebcadb93b1e5cae3c9565
+# 010c03e is 2024 June 5
+BAAS_VERSION: 010c03e0c200c59d36df0b7925bfcb1a7df23721
+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.hpp

@@ -30,9 +30,9 @@
 #include <realm/util/future.hpp>
 #include <realm/version_id.hpp>
 
+#include <map>
 #include <mutex>
 #include <unordered_map>
-#include <map>
 
 namespace realm {
 class DB;
@@ -345,6 +345,8 @@ class SyncSession : public std::enable_shared_from_this<SyncSession> {
             return session.get_appservices_connection_id();
         }
 
+        // Supported commands can be found in `handleTestCommandMessage()`
+        // in baas/devicesync/server/qbs_client_handler_functions.go
         static util::Future<std::string> send_test_command(SyncSession& session, std::string request)
         {
             return session.send_test_command(std::move(request));

src/realm/sync/client.cpp

@@ -803,12 +803,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;
     }
 
@@ -904,7 +900,7 @@ void SessionImpl::process_pending_flx_bootstrap()
         auto pending_batch = bootstrap_store->peek_pending(m_wrapper.m_flx_bootstrap_batch_size_bytes);
         if (!pending_batch.progress) {
             logger.info("Incomplete pending bootstrap found for query version %1", pending_batch.query_version);
-            // Close the write transation before clearing the bootstrap store to avoid a deadlock because the
+            // Close the write transaction before clearing the bootstrap store to avoid a deadlock because the
             // bootstrap store requires a write transaction itself.
             transact->close();
             bootstrap_store->clear();
@@ -1047,7 +1043,7 @@ SyncClientHookAction SessionImpl::call_debug_hook(SyncClientHookEvent event, con
     return call_debug_hook(data);
 }
 
-SyncClientHookAction SessionImpl::call_debug_hook(SyncClientHookEvent event, const ProtocolErrorInfo& error_info)
+SyncClientHookAction SessionImpl::call_debug_hook(SyncClientHookEvent event, const ProtocolErrorInfo* error_info)
 {
     if (REALM_LIKELY(!m_wrapper.m_debug_hook)) {
         return SyncClientHookAction::NoAction;
@@ -1061,37 +1057,12 @@ SyncClientHookAction SessionImpl::call_debug_hook(SyncClientHookEvent event, con
     data.batch_state = DownloadBatchState::SteadyState;
     data.progress = m_progress;
     data.num_changesets = 0;
-    data.query_version = 0;
-    data.error_info = &error_info;
+    data.query_version = m_last_sent_flx_query_version;
+    data.error_info = error_info;
 
     return call_debug_hook(data);
 }
 
-SyncClientHookAction SessionImpl::call_debug_hook(SyncClientHookEvent event)
-{
-    return call_debug_hook(event, m_progress, m_last_sent_flx_query_version, DownloadBatchState::SteadyState, 0);
-}
-
-bool SessionImpl::is_steady_state_download_message(DownloadBatchState batch_state, int64_t query_version)
-{
-    // 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;
-    }
-
-    if (!m_is_flx_sync_session) {
-        return true;
-    }
-
-    // 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;
-    }
-
-    return false;
-}
-
 void SessionImpl::init_progress_handler()
 {
     if (m_state != State::Unactivated && m_state != State::Active)

src/realm/sync/config.hpp

@@ -131,7 +131,10 @@ enum class SyncClientHookEvent {
     ErrorMessageReceived,
     SessionActivating,
     SessionSuspended,
+    SessionConnected,
+    SessionResumed,
     BindMessageSent,
+    IdentMessageSent,
     ClientResetMergeComplete,
     BootstrapBatchAboutToProcess,
 };

src/realm/sync/noinst/client_impl_base.cpp

@@ -1824,12 +1824,7 @@ void Session::send_message()
     if (!m_bind_message_sent)
         return send_bind_message(); // Throws
 
-    if (!m_ident_message_sent) {
-        if (have_client_file_ident())
-            send_ident_message(); // Throws
-        return;
-    }
-
+    // Pending test commands can be sent any time after the BIND message is sent
     const auto has_pending_test_command = std::any_of(m_pending_test_commands.begin(), m_pending_test_commands.end(),
                                                       [](const PendingTestCommand& command) {
                                                           return command.pending;
@@ -1838,6 +1833,12 @@ void Session::send_message()
         return send_test_command_message();
     }
 
+    if (!m_ident_message_sent) {
+        if (have_client_file_ident())
+            send_ident_message(); // Throws
+        return;
+    }
+
     if (m_error_to_send)
         return send_json_error_message(); // Throws
 
@@ -1908,7 +1909,6 @@ void Session::send_bind_message()
     bool need_client_file_ident = !have_client_file_ident();
     const bool is_subserver = false;
 
-
     ClientProtocol& protocol = m_conn.get_client_protocol();
     int protocol_version = m_conn.get_negotiated_protocol_version();
     OutputBuffer& out = m_conn.get_output_buffer();
@@ -1992,6 +1992,7 @@ void Session::send_ident_message()
     m_conn.initiate_write_message(out, this); // Throws
 
     m_ident_message_sent = true;
+    call_debug_hook(SyncClientHookEvent::IdentMessageSent);
 
     // Other messages may be waiting to be sent
     enlist_to_send(); // Throws
@@ -2392,22 +2393,17 @@ Status Session::receive_download_message(const DownloadMessage& message)
     if (!is_flx || query_version > 0)
         enable_progress_notifications();
 
-    // 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) {
         logger.debug("Received: DOWNLOAD(download_server_version=%1, download_client_version=%2, "
                      "latest_server_version=%3, latest_server_version_salt=%4, "
                      "upload_client_version=%5, upload_server_version=%6, progress_estimate=%7, "
-                     "last_in_batch=%8, query_version=%9, num_changesets=%10, ...)",
+                     "batch_state=%8, query_version=%9, num_changesets=%10, ...)",
                      progress.download.server_version, progress.download.last_integrated_client_version,
                      progress.latest_server_version.version, progress.latest_server_version.salt,
                      progress.upload.client_version, progress.upload.last_integrated_server_version,
-                     message.progress_estimate, last_in_batch, query_version, message.changesets.size()); // Throws
+                     message.progress_estimate, message.batch_state, query_version,
+                     message.changesets.size()); // Throws
     }
     else {
         logger.debug("Received: DOWNLOAD(download_server_version=%1, download_client_version=%2, "
@@ -2451,6 +2447,7 @@ Status Session::receive_download_message(const DownloadMessage& message)
                                  changeset.remote_version, server_version, progress.download.server_version)};
         }
         server_version = changeset.remote_version;
+
         // Check that per-changeset last integrated client version is "weakly"
         // increasing.
         bool good_client_version =
@@ -2476,7 +2473,7 @@ Status Session::receive_download_message(const DownloadMessage& message)
     }
 
     auto hook_action = call_debug_hook(SyncClientHookEvent::DownloadMessageReceived, progress, query_version,
-                                       batch_state, message.changesets.size());
+                                       message.batch_state, message.changesets.size());
     if (hook_action == SyncClientHookAction::EarlyReturn) {
         return Status::OK();
     }
@@ -2485,16 +2482,16 @@ Status Session::receive_download_message(const DownloadMessage& message)
     if (is_flx)
         update_download_estimate(message.progress_estimate);
 
-    if (process_flx_bootstrap_message(progress, batch_state, query_version, message.changesets)) {
+    if (process_flx_bootstrap_message(progress, message.batch_state, query_version, message.changesets)) {
         clear_resumption_delay_state();
         return Status::OK();
     }
 
     uint64_t downloadable_bytes = is_flx ? 0 : message.downloadable_bytes;
-    initiate_integrate_changesets(downloadable_bytes, batch_state, progress, message.changesets); // Throws
+    initiate_integrate_changesets(downloadable_bytes, message.batch_state, progress, message.changesets); // Throws
 
     hook_action = call_debug_hook(SyncClientHookEvent::DownloadMessageIntegrated, progress, query_version,
-                                  batch_state, message.changesets.size());
+                                  message.batch_state, message.changesets.size());
     if (hook_action == SyncClientHookAction::EarlyReturn) {
         return Status::OK();
     }
@@ -2604,7 +2601,7 @@ Status Session::receive_error_message(const ProtocolErrorInfo& info)
     // Can't process debug hook actions once the Session is undergoing deactivation, since
     // the SessionWrapper may not be available
     if (m_state == Active) {
-        auto debug_action = call_debug_hook(SyncClientHookEvent::ErrorMessageReceived, info);
+        auto debug_action = call_debug_hook(SyncClientHookEvent::ErrorMessageReceived, &info);
         if (debug_action == SyncClientHookAction::EarlyReturn) {
             return Status::OK();
         }
@@ -2664,7 +2661,7 @@ void Session::suspend(const SessionErrorInfo& info)
     // Notify the application of the suspension of the session if the session is
     // still in the Active state
     if (m_state == Active) {
-        call_debug_hook(SyncClientHookEvent::SessionSuspended, info);
+        call_debug_hook(SyncClientHookEvent::SessionSuspended, &info);
         m_conn.one_less_active_unsuspended_session(); // Throws
         on_suspended(info);                           // Throws
     }

src/realm/sync/noinst/client_impl_base.hpp

@@ -827,7 +827,7 @@ class ClientImpl::Session {
     /// To be used in connection with implementations of
     /// initiate_integrate_changesets().
     void integrate_changesets(const SyncProgress&, std::uint_fast64_t downloadable_bytes, const ReceivedChangesets&,
-                              VersionInfo&, DownloadBatchState last_in_batch);
+                              VersionInfo&, DownloadBatchState batch_state);
 
     /// To be used in connection with implementations of
     /// initiate_integrate_changesets().
@@ -1179,11 +1179,8 @@ class ClientImpl::Session {
 
     SyncClientHookAction call_debug_hook(SyncClientHookEvent event, const SyncProgress&, int64_t, DownloadBatchState,
                                          size_t);
-    SyncClientHookAction call_debug_hook(SyncClientHookEvent event, const ProtocolErrorInfo&);
+    SyncClientHookAction call_debug_hook(SyncClientHookEvent event, const ProtocolErrorInfo* = nullptr);
     SyncClientHookAction call_debug_hook(const SyncClientHookData& data);
-    SyncClientHookAction call_debug_hook(SyncClientHookEvent event);
-
-    bool is_steady_state_download_message(DownloadBatchState batch_state, int64_t query_version);
 
     void init_progress_handler();
     void enable_progress_notifications();
@@ -1470,6 +1467,10 @@ inline void ClientImpl::Session::connection_established(bool fast_reconnect)
         ++m_target_download_mark;
     }
 
+    // Call SessionResumed before sending the BIND Message to
+    // allow adding a test command between BIND and IDENT messages
+    call_debug_hook(SyncClientHookEvent::SessionConnected);
+
     if (!m_suspended) {
         // Ready to send BIND message
         enlist_to_send(); // Throws
@@ -1540,6 +1541,10 @@ inline void ClientImpl::Session::initiate_rebind()
 
     reset_protocol_state();
 
+    // Call SessionResumed before sending the BIND Message to
+    // allow adding a test command between BIND and IDENT messages
+    call_debug_hook(SyncClientHookEvent::SessionResumed);
+
     // Ready to send BIND message
     enlist_to_send(); // Throws
 }

src/realm/sync/noinst/pending_bootstrap_store.cpp

@@ -175,6 +175,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();
 
@@ -183,15 +184,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/noinst/protocol_codec.hpp

@@ -403,8 +403,8 @@ class ClientProtocol {
 
     struct DownloadMessage {
         SyncProgress progress;
-        std::optional<int64_t> query_version;
-        std::optional<bool> last_in_batch;
+        std::optional<int64_t> query_version; // FLX sync only
+        sync::DownloadBatchState batch_state = sync::DownloadBatchState::SteadyState;
         union {
             uint64_t downloadable_bytes = 0;
             double progress_estimate;
@@ -439,10 +439,15 @@ class ClientProtocol {
         if (is_flx) {
             message.query_version = msg.read_next<int64_t>();
             if (message.query_version < 0)
-                return report_error(ErrorCodes::SyncProtocolInvariantFailed, "Bad query version",
+                return report_error(ErrorCodes::SyncProtocolInvariantFailed, "Bad query version: %1",
                                     message.query_version);
-
-            message.last_in_batch = msg.read_next<bool>();
+            int batch_state = msg.read_next<int>();
+            if (batch_state != static_cast<int>(sync::DownloadBatchState::MoreToCome) &&
+                batch_state != static_cast<int>(sync::DownloadBatchState::LastInBatch) &&
+                batch_state != static_cast<int>(sync::DownloadBatchState::SteadyState)) {
+                return report_error(ErrorCodes::SyncProtocolInvariantFailed, "Bad batch state: %1", batch_state);
+            }
+            message.batch_state = static_cast<sync::DownloadBatchState>(batch_state);
 
             message.progress_estimate = msg.read_next<double>();
             if (message.progress_estimate < 0 || message.progress_estimate > 1)