RCORE-2185 Sync client should steal file ident of fresh realm when performing client reset

CHANGELOG.md

@@ -2,7 +2,7 @@
 
 ### Enhancements
 * <New feature description> (PR [#????](https://github.com/realm/realm-core/pull/????))
-* None.
+* The local realm will assume the the client file ident of the fresh realm during a client reset. ([PR #7850](https://github.com/realm/realm-core/pull/7850))
 
 ### Fixed
 * <How do the end-user experience this issue? what was the impact?> ([#????](https://github.com/realm/realm-core/issues/????), since v?.?.?)

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

@@ -913,7 +913,7 @@ void SyncSession::create_sync_session()
     session_config.simulate_integration_error = sync_config.simulate_integration_error;
     session_config.flx_bootstrap_batch_size_bytes = sync_config.flx_bootstrap_batch_size_bytes;
     session_config.session_reason =
-        client_reset::is_fresh_path(m_config.path) ? sync::SessionReason::ClientReset : sync::SessionReason::Sync;
+        client_reset::is_fresh_path(m_config.path) ? sync::SessionReason::FreshRealm : sync::SessionReason::Sync;
     session_config.schema_version = m_config.schema_version;
 
     if (sync_config.on_sync_client_event_hook) {

src/realm/sync/client.cpp

@@ -1172,7 +1172,7 @@ SessionWrapper::SessionWrapper(ClientImpl& client, DBRef db, std::shared_ptr<Sub
     , m_progress_handler(std::move(config.progress_handler))
     , m_connection_state_change_listener(std::move(config.connection_state_change_listener))
     , m_debug_hook(std::move(config.on_sync_client_event_hook))
-    , m_session_reason(m_client_reset_config ? SessionReason::ClientReset : config.session_reason)
+    , m_session_reason(m_client_reset_config ? SessionReason::ClientResetDiff : config.session_reason)
     , m_schema_version(config.schema_version)
     , m_flx_subscription_store(std::move(flx_sub_store))
     , m_migration_store(std::move(migration_store))

src/realm/sync/client_base.hpp

@@ -285,12 +285,27 @@ inline std::ostream& operator<<(std::ostream& os, ConnectionState state)
 
 // The reason a synchronization session is used for.
 enum class SessionReason {
-    // Regular synchronization
+    // Regular synchronization => BIND message sessionReason = 0
     Sync = 0,
-    // Download a fresh realm
-    ClientReset,
+    // Download a fresh realm => BIND message sessionReason = 1
+    FreshRealm,
+    // Client reset diff session => BIND message sessionReason = 1
+    ClientResetDiff,
 };
 
+inline std::ostream& operator<<(std::ostream& os, SessionReason reason)
+{
+    switch (reason) {
+        case SessionReason::Sync:
+            return os << "Sync";
+        case SessionReason::FreshRealm:
+            return os << "FreshRealm";
+        case SessionReason::ClientResetDiff:
+            return os << "ClientResetDiff";
+    }
+    REALM_TERMINATE("Invalid SessionReason value");
+}
+
 } // namespace realm::sync
 
 #endif // REALM_SYNC_CLIENT_BASE_HPP

src/realm/sync/config.hpp

@@ -134,6 +134,7 @@ enum class SyncClientHookEvent {
     BindMessageSent,
     ClientResetMergeComplete,
     BootstrapBatchAboutToProcess,
+    UploadMessageSent,
 };
 
 enum class SyncClientHookAction {

src/realm/sync/noinst/client_impl_base.cpp

@@ -1704,7 +1704,8 @@ void Session::activate()
         bool file_exists = util::File::exists(get_realm_path());
         m_performing_client_reset = get_client_reset_config().has_value();
 
-        logger.info("client_reset_config = %1, Realm exists = %2 ", m_performing_client_reset, file_exists);
+        logger.info("client_reset_config = %1, Realm exists = %2, session_reason = %3 ", m_performing_client_reset,
+                    file_exists, get_session_reason());
         if (!m_performing_client_reset) {
             get_history().get_status(m_last_version_available, m_client_file_ident, m_progress); // Throws
         }
@@ -1717,7 +1718,7 @@ void Session::activate()
     REALM_ASSERT_3(m_last_version_available, >=, m_progress.upload.client_version);
     init_progress_handler();
 
-    logger.debug("last_version_available  = %1", m_last_version_available);                    // Throws
+    logger.debug("last_version_available = %1", m_last_version_available);                     // Throws
     logger.debug("progress_download_server_version = %1", m_progress.download.server_version); // Throws
     logger.debug("progress_download_client_version = %1",
                  m_progress.download.last_integrated_client_version);                                      // Throws
@@ -1887,14 +1888,20 @@ void Session::send_message()
             return false;
         }
 
-        return m_upload_progress.client_version >= m_pending_flx_sub_set->snapshot_version;
+        // Send QUERY messages when the upload progress client version reaches the snapshot version
+        // of a pending subscription, or if this is a fresh realm download session, since UPLOAD
+        // messages are not allowed and the upload progress will not be updated.
+        return m_upload_progress.client_version >= m_pending_flx_sub_set->snapshot_version ||
+               REALM_UNLIKELY(is_fresh_realm_download());
     };
 
     if (check_pending_flx_version()) {
         return send_query_change_message(); // throws
     }
 
-    if (m_allow_upload && (m_last_version_available > m_upload_progress.client_version)) {
+    // Don't allow UPLOAD messages for client reset fresh realm download sessions
+    if (REALM_LIKELY(!is_fresh_realm_download()) && m_allow_upload &&
+        (m_last_version_available > m_upload_progress.client_version)) {
         return send_upload_message(); // Throws
     }
 }
@@ -1905,7 +1912,7 @@ void Session::send_bind_message()
     REALM_ASSERT_EX(m_state == Active, m_state);
 
     session_ident_type session_ident = m_ident;
-    bool need_client_file_ident = !have_client_file_ident();
+    bool need_client_file_ident = !have_client_file_ident() && !m_performing_client_reset;
     const bool is_subserver = false;
 
 
@@ -1919,7 +1926,19 @@ void Session::send_bind_message()
         if (auto migrated_partition = get_migration_store()->get_migrated_partition()) {
             bind_json_data["migratedPartition"] = *migrated_partition;
         }
-        bind_json_data["sessionReason"] = static_cast<uint64_t>(get_session_reason());
+        auto xlate_session_reason = [](SessionReason reason) -> uint64_t {
+            switch (reason) {
+                case SessionReason::FreshRealm:
+                    [[fallthrough]];
+                case SessionReason::ClientResetDiff:
+                    return 1;
+                case SessionReason::Sync:
+                    [[fallthrough]];
+                default:
+                    return 0;
+            }
+        };
+        bind_json_data["sessionReason"] = xlate_session_reason(get_session_reason());
         auto schema_version = get_schema_version();
         // Send 0 if schema is not versioned.
         bind_json_data["schemaVersion"] = schema_version != uint64_t(-1) ? schema_version : 0;
@@ -2155,6 +2174,8 @@ void Session::send_upload_message()
                                                locked_server_version); // Throws
     m_conn.initiate_write_message(out, this);                          // Throws
 
+    call_debug_hook(SyncClientHookEvent::UploadMessageSent);
+
     // Other messages may be waiting to be sent
     enlist_to_send(); // Throws
 }
@@ -2265,29 +2286,56 @@ bool Session::client_reset_if_needed()
         return false;
     }
 
+    // Save a copy of the status and action in case an error/exception occurs
+    Status cr_status = client_reset_config->error;
+    ProtocolErrorInfo::Action cr_action = client_reset_config->action;
+
     auto on_flx_version_complete = [this](int64_t version) {
         this->on_flx_sync_version_complete(version);
     };
-    bool did_reset =
-        client_reset::perform_client_reset(logger, *get_db(), std::move(*client_reset_config), m_client_file_ident,
-                                           get_flx_subscription_store(), on_flx_version_complete);
-
-    call_debug_hook(SyncClientHookEvent::ClientResetMergeComplete);
-    if (!did_reset) {
+    try {
+        // Storage for the file ident from the fresh realm that was migrated to the local realm
+        SaltedFileIdent client_file_ident_out;
+        bool did_reset = client_reset::perform_client_reset(logger, *get_db(), std::move(*client_reset_config),
+                                                            client_file_ident_out, get_flx_subscription_store(),
+                                                            on_flx_version_complete);
+        m_client_file_ident = client_file_ident_out;
+
+        call_debug_hook(SyncClientHookEvent::ClientResetMergeComplete);
+        if (!did_reset) {
+            return false;
+        }
+    }
+    catch (const std::exception& e) {
+        auto err_msg = util::format("A fatal error occurred during '%1' client reset diff for %2: '%3'", cr_action,
+                                    cr_status, e.what());
+        logger.error(err_msg.c_str());
+        SessionErrorInfo err_info(Status{ErrorCodes::AutoClientResetFailed, err_msg}, IsFatal{true});
+        suspend(err_info);
         return false;
     }
 
     // The fresh Realm has been used to reset the state
-    logger.debug("Client reset is completed, path=%1", get_realm_path()); // Throws
+    logger.debug("Client reset is completed, path = %1", get_realm_path()); // Throws
+    logger.debug("client_file_ident = %1, client_file_ident_salt = %2", m_client_file_ident.ident,
+                 m_client_file_ident.salt); // Throws
 
     SaltedFileIdent client_file_ident;
     get_history().get_status(m_last_version_available, client_file_ident, m_progress); // Throws
+    // Print the version/progress information before performing the asserts
+    logger.debug("last_version_available = %1", m_last_version_available); // Throws
+    logger.debug("upload_progress_client_version = %1, upload_progress_server_version = %2",
+                 m_progress.upload.client_version,
+                 m_progress.upload.last_integrated_server_version); // Throws
+    logger.debug("download_progress_client_version = %1, download_progress_server_version = %2",
+                 m_progress.download.last_integrated_client_version,
+                 m_progress.download.server_version); // Throws
+
     REALM_ASSERT_3(m_client_file_ident.ident, ==, client_file_ident.ident);
     REALM_ASSERT_3(m_client_file_ident.salt, ==, client_file_ident.salt);
     REALM_ASSERT_EX(m_progress.download.last_integrated_client_version == 0,
                     m_progress.download.last_integrated_client_version);
     REALM_ASSERT_EX(m_progress.upload.client_version == 0, m_progress.upload.client_version);
-    logger.trace("last_version_available  = %1", m_last_version_available); // Throws
 
     m_upload_progress = m_progress.upload;
     m_download_progress = m_progress.download;
@@ -2325,13 +2373,24 @@ Status Session::receive_ident_message(SaltedFileIdent client_file_ident)
     bool legal_at_this_time = (m_bind_message_sent && !have_client_file_ident() && !m_error_message_received &&
                                !m_unbound_message_received);
     if (REALM_UNLIKELY(!legal_at_this_time)) {
-        return {ErrorCodes::SyncProtocolInvariantFailed, "Received IDENT message when it was not legal"};
+        std::string_view illegal_reason;
+        if (m_bind_message_sent)
+            illegal_reason = "BIND message not sent";
+        else if (have_client_file_ident())
+            illegal_reason = "file ident has already been assigned to session";
+        else if (m_error_message_received)
+            illegal_reason = "error message has been received from the server";
+        else if (m_unbound_message_received)
+            illegal_reason = "session has already been suspended";
+        return {ErrorCodes::SyncProtocolInvariantFailed,
+                util::format("Received IDENT message when it was not legal: %1", illegal_reason)};
     }
     if (REALM_UNLIKELY(client_file_ident.ident < 1)) {
-        return {ErrorCodes::SyncProtocolInvariantFailed, "Bad client file identifier in IDENT message"};
+        return {ErrorCodes::SyncProtocolInvariantFailed,
+                util::format("Bad client file identifier in IDENT message: %1", client_file_ident.ident)};
     }
     if (REALM_UNLIKELY(client_file_ident.salt == 0)) {
-        return {ErrorCodes::SyncProtocolInvariantFailed, "Bad client file identifier salt in IDENT message"};
+        return {ErrorCodes::SyncProtocolInvariantFailed, "Bad client file identifier salt (0) in IDENT message"};
     }
 
     m_client_file_ident = client_file_ident;
@@ -2342,36 +2401,11 @@ Status Session::receive_ident_message(SaltedFileIdent client_file_ident)
         return Status::OK();       // Success
     }
 
-    // if a client reset happens, it will take care of setting the file ident
-    // and if not, we do it here
-    bool did_client_reset = false;
-
-    // Save some of the client reset info for reporting to the client if an error occurs.
-    Status cr_status(Status::OK()); // Start with no client reset
-    ProtocolErrorInfo::Action cr_action = ProtocolErrorInfo::Action::NoAction;
-    if (auto& cr_config = get_client_reset_config()) {
-        cr_status = cr_config->error;
-        cr_action = cr_config->action;
-    }
-
-    try {
-        did_client_reset = client_reset_if_needed();
-    }
-    catch (const std::exception& e) {
-        auto err_msg = util::format("A fatal error occurred during '%1' client reset for %2: '%3'", cr_action,
-                                    cr_status, e.what());
-        logger.error(err_msg.c_str());
-        SessionErrorInfo err_info(Status{ErrorCodes::AutoClientResetFailed, err_msg}, IsFatal{true});
-        suspend(err_info);
-        return Status::OK();
-    }
-    if (!did_client_reset) {
-        get_history().set_client_file_ident(client_file_ident,
-                                            m_fix_up_object_ids); // Throws
-        m_progress.download.last_integrated_client_version = 0;
-        m_progress.upload.client_version = 0;
-        m_last_version_selected_for_upload = 0;
-    }
+    get_history().set_client_file_ident(client_file_ident,
+                                        m_fix_up_object_ids); // Throws
+    m_progress.download.last_integrated_client_version = 0;
+    m_progress.upload.client_version = 0;
+    m_last_version_selected_for_upload = 0;
 
     // Ready to send the IDENT message
     ensure_enlisted_to_send(); // Throws

src/realm/sync/noinst/client_impl_base.hpp

@@ -892,9 +892,19 @@ class ClientImpl::Session {
     util::Optional<ClientReset>& get_client_reset_config() noexcept;
 
     // Get the reason a synchronization session is used for (regular sync or client reset)
-    // - Client reset state means the session is going to be used to download a fresh realm.
+    // - Sync state means the session is a regular session
+    // - FreshRealm state means the session is going to be used to download a fresh realm.
+    // - ClientResetDiff state means the session is going to perform the client reset diff.
     SessionReason get_session_reason() noexcept;
 
+    // Returns true if the current session is being used to download a fresh realm, since
+    // UPLOAD messages are not allowed to be sent during the fresh realm download for a
+    // client reset, but QUERY and MARK messages are allowed.
+    inline bool is_fresh_realm_download() noexcept
+    {
+        return get_session_reason() == SessionReason::FreshRealm;
+    }
+
     /// Returns the schema version the synchronization session connects with to the server.
     uint64_t get_schema_version() noexcept;
 
@@ -1502,6 +1512,12 @@ inline void ClientImpl::Session::message_sent()
     // No message will be sent after the UNBIND message
     REALM_ASSERT(!m_unbind_message_send_complete);
 
+    // Is there a pending client reset diff to try to perform once the bind
+    // message has been sent?
+    if (m_bind_message_sent && m_performing_client_reset) {
+        client_reset_if_needed(); // resets m_performing_client_reset to false
+    }
+
     if (m_unbind_message_sent) {
         REALM_ASSERT(!m_enlisted_to_send);
 

src/realm/sync/noinst/client_reset.cpp

@@ -480,7 +480,7 @@ ClientResyncMode reset_precheck_guard(const TransactionRef& wt_local, ClientResy
     return mode;
 }
 
-bool perform_client_reset_diff(DB& db_local, sync::ClientReset& reset_config, sync::SaltedFileIdent client_file_ident,
+bool perform_client_reset_diff(DB& db_local, sync::ClientReset& reset_config, sync::SaltedFileIdent& file_ident_out,
                                util::Logger& logger, sync::SubscriptionStore* sub_store,
                                util::FunctionRef<void(int64_t)> on_flx_version_complete)
 {
@@ -491,13 +491,6 @@ bool perform_client_reset_diff(DB& db_local, sync::ClientReset& reset_config, sy
     bool recover_local_changes =
         actual_mode == ClientResyncMode::Recover || actual_mode == ClientResyncMode::RecoverOrDiscard;
 
-    logger.info(util::LogCategory::reset,
-                "Client reset: path_local = %1, client_file_ident = (ident: %2, salt: %3), "
-                "remote_path = %4, requested_mode = %5, action = %6, actual_mode = %7, will_recover = %8, "
-                "originating_error = %9",
-                db_local.get_path(), client_file_ident.ident, client_file_ident.salt, db_remote.get_path(),
-                reset_config.mode, reset_config.action, actual_mode, recover_local_changes, reset_config.error);
-
     auto& repl_local = dynamic_cast<ClientReplication&>(*db_local.get_replication());
     auto& history_local = repl_local.get_history();
     history_local.ensure_updated(wt_local->get_version());
@@ -507,14 +500,22 @@ bool perform_client_reset_diff(DB& db_local, sync::ClientReset& reset_config, sy
     auto& history_remote = repl_remote.get_history();
 
     sync::SaltedVersion fresh_server_version = {0, 0};
+    sync::SaltedFileIdent fresh_file_ident = {0, 0};
     {
         SyncProgress remote_progress;
         sync::version_type remote_version_unused;
-        SaltedFileIdent remote_ident_unused;
-        history_remote.get_status(remote_version_unused, remote_ident_unused, remote_progress);
+        history_remote.get_status(remote_version_unused, fresh_file_ident, remote_progress);
         fresh_server_version = remote_progress.latest_server_version;
     }
 
+    logger.info(util::LogCategory::reset,
+                "Client reset: path_local = %1, fresh_file_ident = (ident: %2, salt: %3), "
+                "fresh_server_version = (ident: %4, salt: %5), remote_path = %6, requested_mode = %7, action = %8, "
+                "actual_mode = %9, will_recover = %10, originating_error = %11",
+                db_local.get_path(), fresh_file_ident.ident, fresh_file_ident.salt, fresh_server_version.version,
+                fresh_server_version.salt, db_remote.get_path(), reset_config.mode, reset_config.action, actual_mode,
+                recover_local_changes, reset_config.error);
+
     TransactionRef tr_remote;
     std::vector<client_reset::RecoveredChange> recovered;
     if (recover_local_changes) {
@@ -534,7 +535,7 @@ bool perform_client_reset_diff(DB& db_local, sync::ClientReset& reset_config, sy
 
     // now that the state of the fresh and local Realms are identical,
     // reset the local sync history and steal the fresh Realm's ident
-    history_local.set_history_adjustments(logger, wt_local->get_version(), client_file_ident, fresh_server_version,
+    history_local.set_history_adjustments(logger, wt_local->get_version(), fresh_file_ident, fresh_server_version,
                                           recovered);
 
     int64_t subscription_version = 0;
@@ -549,6 +550,7 @@ bool perform_client_reset_diff(DB& db_local, sync::ClientReset& reset_config, sy
 
     wt_local->commit_and_continue_as_read();
     on_flx_version_complete(subscription_version);
+    file_ident_out = fresh_file_ident;
 
     VersionID new_version_local = wt_local->get_version_of_current_transaction();
     logger.info(util::LogCategory::reset,