From 25212bc46d50e4658d589688fe5a63295f4c126f Mon Sep 17 00:00:00 2001 From: Michael Wilkerson-Barker Date: Thu, 30 May 2024 22:09:55 -0400 Subject: [PATCH] RCORE-1386 Track client reset reason in table that detects client reset cycles (#7649) * Broke out the client reset error and action storage from PR #7542 * Removed client reset recovery_allowed flag and other updates from review * Updated pending_client_reset store to use the schema metadata tables * Fixed pausing a session does not hold the DB open test * Moved ownership of reset store to SessionWrapper * Fixed migration test crash - need to save client reset error in handle fresh realm downloaded * Updated PendingResetStore to be static functions instead of an initialized object; updates from review * Make ClientReset::error no longer optional; fixed subscriptions tests * updated changelog after release * updates from review --- CHANGELOG.md | 2 + Package.swift | 1 + src/realm/db.hpp | 16 + src/realm/object-store/sync/sync_session.cpp | 91 ++--- src/realm/object-store/sync/sync_session.hpp | 10 +- src/realm/sync/CMakeLists.txt | 1 + src/realm/sync/client.cpp | 71 ++-- src/realm/sync/client_base.hpp | 3 +- src/realm/sync/noinst/client_history_impl.cpp | 6 +- src/realm/sync/noinst/client_history_impl.hpp | 4 +- src/realm/sync/noinst/client_impl_base.cpp | 26 +- src/realm/sync/noinst/client_impl_base.hpp | 1 - src/realm/sync/noinst/client_reset.cpp | 154 ++------ src/realm/sync/noinst/client_reset.hpp | 25 +- .../sync/noinst/client_reset_operation.cpp | 23 +- .../sync/noinst/client_reset_operation.hpp | 8 +- .../sync/noinst/client_reset_recovery.cpp | 2 +- src/realm/sync/noinst/migration_store.cpp | 40 +-- .../sync/noinst/pending_bootstrap_store.cpp | 12 +- src/realm/sync/noinst/pending_reset_store.cpp | 313 ++++++++++++++++ src/realm/sync/noinst/pending_reset_store.hpp | 91 +++++ .../sync/noinst/sync_metadata_schema.cpp | 84 +++-- .../sync/noinst/sync_metadata_schema.hpp | 3 + src/realm/sync/subscriptions.cpp | 23 +- test/object-store/sync/app.cpp | 18 +- test/object-store/sync/client_reset.cpp | 26 +- .../util/sync/sync_test_utils.cpp | 18 +- test/test_client_reset.cpp | 339 +++++++++++++----- test/util/compare_groups.cpp | 6 +- 29 files changed, 981 insertions(+), 436 deletions(-) create mode 100644 src/realm/sync/noinst/pending_reset_store.cpp create mode 100644 src/realm/sync/noinst/pending_reset_store.hpp diff --git a/CHANGELOG.md b/CHANGELOG.md index e63d5b4570e..89700ac466a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ### Enhancements * (PR [#????](https://github.com/realm/realm-core/pull/????)) +* Report the originating error that caused a client reset to occur. ([#6154](https://github.com/realm/realm-core/issues/6154)) ### Fixed * ([#????](https://github.com/realm/realm-core/issues/????), since v?.?.?) @@ -20,6 +21,7 @@ * Work around a bug in VC++ that resulted in runtime errors when running the tests in a debug build (#[7741](https://github.com/realm/realm-core/issues/7741)). * Refactor `sync::Session` to eliminate the bind() step of session creation ([#7609](https://github.com/realm/realm-core/pull/7609)). * Add ScopeExitFail which only calls the handler if exiting the scope via an uncaught exception ([#7609](https://github.com/realm/realm-core/pull/7609)). +* Add the originating error and server requests action that caused a client reset to occur to the client reset tracking metadata storage. ([PR #7649](https://github.com/realm/realm-core/pull/7649)) ---------------------------------------------- diff --git a/Package.swift b/Package.swift index 2e3a3305ecf..956827d1e70 100644 --- a/Package.swift +++ b/Package.swift @@ -121,6 +121,7 @@ let notSyncServerSources: [String] = [ "realm/sync/noinst/compact_changesets.cpp", "realm/sync/noinst/migration_store.cpp", "realm/sync/noinst/pending_bootstrap_store.cpp", + "realm/sync/noinst/pending_reset_store.cpp", "realm/sync/noinst/protocol_codec.cpp", "realm/sync/noinst/sync_metadata_schema.cpp", "realm/sync/noinst/sync_schema_migration.cpp", diff --git a/src/realm/db.hpp b/src/realm/db.hpp index af8fc23633c..1a8a1b67461 100644 --- a/src/realm/db.hpp +++ b/src/realm/db.hpp @@ -697,6 +697,22 @@ inline int DB::get_file_format_version() const noexcept return m_file_format_version; } +inline std::ostream& operator<<(std::ostream& os, const DB::TransactStage& stage) +{ + switch (stage) { + case DB::TransactStage::transact_Ready: + return os << "transact_Ready"; + case DB::TransactStage::transact_Reading: + return os << "transact_Reading"; + case DB::TransactStage::transact_Frozen: + return os << "transact_Frozen"; + case DB::TransactStage::transact_Writing: + return os << "transact_Writing"; + } + REALM_UNREACHABLE(); +} + + } // namespace realm #endif // REALM_DB_HPP diff --git a/src/realm/object-store/sync/sync_session.cpp b/src/realm/object-store/sync/sync_session.cpp index 9cbbc5cf9f9..9fb0802fe0e 100644 --- a/src/realm/object-store/sync/sync_session.cpp +++ b/src/realm/object-store/sync/sync_session.cpp @@ -422,17 +422,15 @@ void SyncSession::update_error_and_mark_file_for_deletion(SyncError& error, Shou } } -void SyncSession::download_fresh_realm(sync::ProtocolErrorInfo::Action server_requests_action) +void SyncSession::download_fresh_realm(const sync::SessionErrorInfo& error_info) { // first check that recovery will not be prevented - if (server_requests_action == sync::ProtocolErrorInfo::Action::ClientResetNoRecovery) { + if (error_info.server_requests_action == sync::ProtocolErrorInfo::Action::ClientResetNoRecovery) { auto mode = config(&SyncConfig::client_resync_mode); if (mode == ClientResyncMode::Recover) { handle_fresh_realm_downloaded( - nullptr, - {ErrorCodes::RuntimeError, - "A client reset is required but the server does not permit recovery for this client"}, - server_requests_action); + nullptr, {ErrorCodes::RuntimeError, + "A client reset is required but the server does not permit recovery for this client"}); return; } } @@ -469,7 +467,7 @@ void SyncSession::download_fresh_realm(sync::ProtocolErrorInfo::Action server_re catch (...) { // Failed to open the fresh path after attempting to delete it, so we // just can't do automatic recovery. - handle_fresh_realm_downloaded(nullptr, exception_to_status(), server_requests_action); + handle_fresh_realm_downloaded(nullptr, exception_to_status()); return; } @@ -477,6 +475,7 @@ void SyncSession::download_fresh_realm(sync::ProtocolErrorInfo::Action server_re if (m_state != State::Active) { return; } + RealmConfig fresh_config; { util::CheckedLockGuard config_lock(m_config_mutex); @@ -511,7 +510,7 @@ void SyncSession::download_fresh_realm(sync::ProtocolErrorInfo::Action server_re using SubscriptionState = sync::SubscriptionSet::State; fresh_sub.get_state_change_notification(SubscriptionState::Complete) .then([=](SubscriptionState) -> util::Future { - if (server_requests_action != sync::ProtocolErrorInfo::Action::MigrateToFLX) { + if (error_info.server_requests_action != sync::ProtocolErrorInfo::Action::MigrateToFLX) { return fresh_sub; } if (!self->m_migration_store->is_migration_in_progress()) { @@ -527,7 +526,7 @@ void SyncSession::download_fresh_realm(sync::ProtocolErrorInfo::Action server_re fresh_sync_session->m_migration_store->create_subscriptions(*fresh_sub_store, *query_string); return fresh_sub_store->get_latest() .get_state_change_notification(SubscriptionState::Complete) - .then([=](SubscriptionState) { + .then([fresh_sub_store](SubscriptionState) { return fresh_sub_store->get_latest(); }); }) @@ -536,29 +535,32 @@ void SyncSession::download_fresh_realm(sync::ProtocolErrorInfo::Action server_re // it immediately fresh_sync_session->force_close(); if (subs.is_ok()) { - self->handle_fresh_realm_downloaded(db, Status::OK(), server_requests_action, - std::move(subs.get_value())); + self->handle_fresh_realm_downloaded(db, std::move(error_info), std::move(subs.get_value())); } else { - self->handle_fresh_realm_downloaded(nullptr, subs.get_status(), server_requests_action); + self->handle_fresh_realm_downloaded(nullptr, std::move(subs.get_status())); } }); } else { // pbs - fresh_sync_session->wait_for_download_completion([=, weak_self = weak_from_this()](Status s) { + fresh_sync_session->wait_for_download_completion([=, weak_self = weak_from_this()](Status status) { // Keep the sync session alive while it's downloading, but then close // it immediately fresh_sync_session->force_close(); if (auto strong_self = weak_self.lock()) { - strong_self->handle_fresh_realm_downloaded(db, s, server_requests_action); + if (status.is_ok()) { + strong_self->handle_fresh_realm_downloaded(db, std::move(error_info)); + } + else { + strong_self->handle_fresh_realm_downloaded(nullptr, std::move(status)); + } } }); } fresh_sync_session->revive_if_needed(); } -void SyncSession::handle_fresh_realm_downloaded(DBRef db, Status status, - sync::ProtocolErrorInfo::Action server_requests_action, +void SyncSession::handle_fresh_realm_downloaded(DBRef db, StatusWith error_info, std::optional new_subs) { util::CheckedUniqueLock lock(m_state_mutex); @@ -569,15 +571,15 @@ void SyncSession::handle_fresh_realm_downloaded(DBRef db, Status status, // - unable to write the fresh copy to the file system // - during download of the fresh copy, the fresh copy itself is reset // - in FLX mode there was a problem fulfilling the previously active subscription - if (!status.is_ok()) { - if (status == ErrorCodes::OperationAborted) { + if (!error_info.is_ok()) { + if (error_info.get_status() == ErrorCodes::OperationAborted) { return; } lock.unlock(); sync::SessionErrorInfo synthetic( Status{ErrorCodes::AutoClientResetFailed, - util::format("A fatal error occurred during client reset: '%1'", status.reason())}, + util::format("A fatal error occurred during client reset: '%1'", error_info.get_status())}, sync::IsFatal{true}); handle_error(synthetic); return; @@ -593,9 +595,12 @@ void SyncSession::handle_fresh_realm_downloaded(DBRef db, Status status, // that moving to the inactive state doesn't clear them - they will be // re-registered when the session becomes active again. { - m_server_requests_action = server_requests_action; m_client_reset_fresh_copy = db; CompletionCallbacks callbacks; + // Save the client reset error for when the original sync session is revived + REALM_ASSERT(error_info.is_ok()); // required if we get here + m_client_reset_error = std::move(error_info.get_value()); + std::swap(m_completion_callbacks, callbacks); // always swap back, even if advance_state throws auto guard = util::make_scope_exit([&]() noexcept { @@ -607,11 +612,13 @@ void SyncSession::handle_fresh_realm_downloaded(DBRef db, Status status, }); // Do not cancel the notifications on subscriptions. bool cancel_subscription_notifications = false; + bool is_migration = + m_client_reset_error->server_requests_action == sync::ProtocolErrorInfo::Action::MigrateToFLX || + m_client_reset_error->server_requests_action == sync::ProtocolErrorInfo::Action::RevertToPBS; become_inactive(std::move(lock), Status::OK(), cancel_subscription_notifications); // unlocks the lock // Once the session is inactive, update sync config and subscription store after migration. - if (server_requests_action == sync::ProtocolErrorInfo::Action::MigrateToFLX || - server_requests_action == sync::ProtocolErrorInfo::Action::RevertToPBS) { + if (is_migration) { apply_sync_config_after_migration_or_rollback(); auto flx_sync_requested = config(&SyncConfig::flx_sync_requested); update_subscription_store(flx_sync_requested, std::move(new_subs)); @@ -695,7 +702,7 @@ void SyncSession::handle_error(sync::SessionErrorInfo error) case ClientResyncMode::RecoverOrDiscard: [[fallthrough]]; case ClientResyncMode::Recover: - download_fresh_realm(error.server_requests_action); + download_fresh_realm(error); return; // do not propagate the error to the user at this point } break; @@ -707,7 +714,7 @@ void SyncSession::handle_error(sync::SessionErrorInfo error) m_migration_store->migrate_to_flx(*error.migration_query_string, m_original_sync_config->partition_value); save_sync_config_after_migration_or_rollback(); - download_fresh_realm(error.server_requests_action); + download_fresh_realm(error); return; case sync::ProtocolErrorInfo::Action::RevertToPBS: // If the client was updated to use FLX natively, but the server was rolled back to PBS, @@ -721,7 +728,7 @@ void SyncSession::handle_error(sync::SessionErrorInfo error) // Original config was PBS, rollback the migration m_migration_store->rollback_to_pbs(); save_sync_config_after_migration_or_rollback(); - download_fresh_realm(error.server_requests_action); + download_fresh_realm(error); return; case sync::ProtocolErrorInfo::Action::RefreshUser: if (auto u = user()) { @@ -824,17 +831,15 @@ void SyncSession::handle_progress_update(uint64_t downloaded, uint64_t downloada upload_estimate, query_version); } -static sync::Session::Config::ClientReset make_client_reset_config(const RealmConfig& base_config, - const std::shared_ptr& sync_config, - DBRef&& fresh_copy, bool recovery_is_allowed, - bool schema_migration_detected) + +static sync::Session::Config::ClientReset +make_client_reset_config(const RealmConfig& base_config, const std::shared_ptr& sync_config, + DBRef&& fresh_copy, sync::SessionErrorInfo&& error_info, bool schema_migration_detected) { REALM_ASSERT(sync_config->client_resync_mode != ClientResyncMode::Manual); - sync::Session::Config::ClientReset config; - config.mode = sync_config->client_resync_mode; - config.fresh_copy = std::move(fresh_copy); - config.recovery_is_allowed = recovery_is_allowed; + sync::Session::Config::ClientReset config{sync_config->client_resync_mode, std::move(fresh_copy), + std::move(error_info.status), error_info.server_requests_action}; // The conditions here are asymmetric because if we have *either* a before // or after callback we need to make sure to initialize the local schema @@ -940,17 +945,15 @@ void SyncSession::create_sync_session() } session_config.custom_http_headers = sync_config.custom_http_headers; - if (m_server_requests_action != sync::ProtocolErrorInfo::Action::NoAction) { - // Migrations are allowed to recover local data. - const bool allowed_to_recover = m_server_requests_action == sync::ProtocolErrorInfo::Action::ClientReset || - m_server_requests_action == sync::ProtocolErrorInfo::Action::MigrateToFLX || - m_server_requests_action == sync::ProtocolErrorInfo::Action::RevertToPBS; - // Use the original sync config, not the updated one from the migration store - session_config.client_reset_config = - make_client_reset_config(m_config, m_original_sync_config, std::move(m_client_reset_fresh_copy), - allowed_to_recover, m_previous_schema_version.has_value()); - session_config.schema_version = m_previous_schema_version.value_or(m_config.schema_version); - m_server_requests_action = sync::ProtocolErrorInfo::Action::NoAction; + if (m_client_reset_error) { + auto client_reset_error = std::exchange(m_client_reset_error, std::nullopt); + if (client_reset_error->server_requests_action != sync::ProtocolErrorInfo::Action::NoAction) { + // Use the original sync config, not the updated one from the migration store + session_config.client_reset_config = + make_client_reset_config(m_config, m_original_sync_config, std::move(m_client_reset_fresh_copy), + std::move(*client_reset_error), m_previous_schema_version.has_value()); + session_config.schema_version = m_previous_schema_version.value_or(m_config.schema_version); + } } session_config.progress_handler = [weak_self](uint_fast64_t downloaded, uint_fast64_t downloadable, diff --git a/src/realm/object-store/sync/sync_session.hpp b/src/realm/object-store/sync/sync_session.hpp index 0a148f3bdd1..83e35f3b8da 100644 --- a/src/realm/object-store/sync/sync_session.hpp +++ b/src/realm/object-store/sync/sync_session.hpp @@ -22,6 +22,7 @@ #include #include #include +#include #include #include @@ -40,7 +41,6 @@ class SyncUser; namespace sync { class Session; -struct SessionErrorInfo; class MigrationStore; } // namespace sync @@ -406,10 +406,9 @@ class SyncSession : public std::enable_shared_from_this { void apply_sync_config_after_migration_or_rollback() REQUIRES(!m_config_mutex, !m_state_mutex); void save_sync_config_after_migration_or_rollback() REQUIRES(!m_config_mutex); - void download_fresh_realm(sync::ProtocolErrorInfo::Action server_requests_action) + void download_fresh_realm(const sync::SessionErrorInfo& error_info) REQUIRES(!m_config_mutex, !m_state_mutex, !m_connection_state_mutex); - void handle_fresh_realm_downloaded(DBRef db, Status status, - sync::ProtocolErrorInfo::Action server_requests_action, + void handle_fresh_realm_downloaded(DBRef db, StatusWith error_info, std::optional new_subs = std::nullopt) REQUIRES(!m_state_mutex, !m_config_mutex, !m_connection_state_mutex); void handle_error(sync::SessionErrorInfo) REQUIRES(!m_state_mutex, !m_config_mutex, !m_connection_state_mutex); @@ -504,8 +503,7 @@ class SyncSession : public std::enable_shared_from_this { std::shared_ptr m_migrated_sync_config GUARDED_BY(m_config_mutex); const std::shared_ptr m_migration_store; std::optional m_migration_sentinel_query_version GUARDED_BY(m_state_mutex); - sync::ProtocolErrorInfo::Action - m_server_requests_action GUARDED_BY(m_state_mutex) = sync::ProtocolErrorInfo::Action::NoAction; + std::optional m_client_reset_error GUARDED_BY(m_state_mutex); DBRef m_client_reset_fresh_copy GUARDED_BY(m_state_mutex); _impl::SyncClient& m_client; SyncManager* m_sync_manager GUARDED_BY(m_state_mutex) = nullptr; diff --git a/src/realm/sync/CMakeLists.txt b/src/realm/sync/CMakeLists.txt index afa711d9e04..ac6b64f7b2e 100644 --- a/src/realm/sync/CMakeLists.txt +++ b/src/realm/sync/CMakeLists.txt @@ -9,6 +9,7 @@ set(SYNC_SOURCES noinst/compact_changesets.cpp noinst/migration_store.cpp noinst/pending_bootstrap_store.cpp + noinst/pending_reset_store.cpp noinst/protocol_codec.cpp noinst/sync_metadata_schema.cpp noinst/sync_schema_migration.cpp diff --git a/src/realm/sync/client.cpp b/src/realm/sync/client.cpp index b2786796105..da9057debf4 100644 --- a/src/realm/sync/client.cpp +++ b/src/realm/sync/client.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -1869,41 +1870,43 @@ void SessionWrapper::handle_pending_client_reset_acknowledgement() { REALM_ASSERT(!m_finalized); - auto pending_reset = _impl::client_reset::has_pending_reset(*m_db->start_frozen()); - REALM_ASSERT(pending_reset); - m_sess->logger.info("Tracking pending client reset of type \"%1\" from %2", pending_reset->type, - pending_reset->time); - async_wait_for(true, true, [self = util::bind_ptr(this), pending_reset = *pending_reset](Status status) { - if (status == ErrorCodes::OperationAborted) { - return; - } - auto& logger = self->m_sess->logger; - if (!status.is_ok()) { - logger.error("Error while tracking client reset acknowledgement: %1", status); - return; - } + auto has_pending_reset = PendingResetStore::has_pending_reset(m_db->start_frozen()); + if (!has_pending_reset) { + return; // nothing to do + } - auto wt = self->m_db->start_write(); - auto cur_pending_reset = _impl::client_reset::has_pending_reset(*wt); - if (!cur_pending_reset) { - logger.debug( - "Was going to remove client reset tracker for type \"%1\" from %2, but it was already removed", - pending_reset.type, pending_reset.time); - return; - } - else if (cur_pending_reset->type != pending_reset.type || cur_pending_reset->time != pending_reset.time) { - logger.debug( - "Was going to remove client reset tracker for type \"%1\" from %2, but found type \"%3\" from %4.", - pending_reset.type, pending_reset.time, cur_pending_reset->type, cur_pending_reset->time); - } - else { - logger.debug("Client reset of type \"%1\" from %2 has been acknowledged by the server. " - "Removing cycle detection tracker.", - pending_reset.type, pending_reset.time); - } - _impl::client_reset::remove_pending_client_resets(*wt); - wt->commit(); - }); + m_sess->logger.info(util::LogCategory::reset, "Tracking %1", *has_pending_reset); + + // Now that the client reset merge is complete, wait for the changes to synchronize with the server + async_wait_for( + true, true, [self = util::bind_ptr(this), pending_reset = std::move(*has_pending_reset)](Status status) { + if (status == ErrorCodes::OperationAborted) { + return; + } + auto& logger = self->m_sess->logger; + if (!status.is_ok()) { + logger.error(util::LogCategory::reset, "Error while tracking client reset acknowledgement: %1", + status); + return; + } + + logger.debug(util::LogCategory::reset, "Server has acknowledged %1", pending_reset); + + auto tr = self->m_db->start_write(); + auto cur_pending_reset = PendingResetStore::has_pending_reset(tr); + if (!cur_pending_reset) { + logger.debug(util::LogCategory::reset, "Client reset cycle detection tracker already removed."); + return; + } + if (*cur_pending_reset == pending_reset) { + logger.debug(util::LogCategory::reset, "Removing client reset cycle detection tracker."); + } + else { + logger.info(util::LogCategory::reset, "Found new %1", cur_pending_reset); + } + PendingResetStore::clear_pending_reset(tr); + tr->commit(); + }); } void SessionWrapper::update_subscription_version_info() diff --git a/src/realm/sync/client_base.hpp b/src/realm/sync/client_base.hpp index 71b9722ea18..8493503f92b 100644 --- a/src/realm/sync/client_base.hpp +++ b/src/realm/sync/client_base.hpp @@ -62,7 +62,8 @@ class SyncSocketProvider; struct ClientReset { realm::ClientResyncMode mode; DBRef fresh_copy; - bool recovery_is_allowed = true; + Status error; + sync::ProtocolErrorInfo::Action action = sync::ProtocolErrorInfo::Action::ClientReset; util::UniqueFunction notify_before_client_reset; util::UniqueFunction notify_after_client_reset; }; diff --git a/src/realm/sync/noinst/client_history_impl.cpp b/src/realm/sync/noinst/client_history_impl.cpp index b132fc2bea9..3a5e94bf4e4 100644 --- a/src/realm/sync/noinst/client_history_impl.cpp +++ b/src/realm/sync/noinst/client_history_impl.cpp @@ -273,7 +273,7 @@ util::UniqueFunction ClientReplication::make_wr } void ClientHistory::get_status(version_type& current_client_version, SaltedFileIdent& client_file_ident, - SyncProgress& progress, bool* has_pending_client_reset) const + SyncProgress& progress) const { TransactionRef rt = m_db->start_read(); // Throws version_type current_client_version_2 = rt->get_version(); @@ -306,10 +306,6 @@ void ClientHistory::get_status(version_type& current_client_version, SaltedFileI REALM_ASSERT(current_client_version >= s_initial_version + 0); if (current_client_version == s_initial_version + 0) current_client_version = 0; - - if (has_pending_client_reset) { - *has_pending_client_reset = _impl::client_reset::has_pending_reset(*rt).has_value(); - } } void ClientHistory::set_client_file_ident(SaltedFileIdent client_file_ident, bool fix_up_object_ids) diff --git a/src/realm/sync/noinst/client_history_impl.hpp b/src/realm/sync/noinst/client_history_impl.hpp index f3ea2bb5f5d..5a92e069c09 100644 --- a/src/realm/sync/noinst/client_history_impl.hpp +++ b/src/realm/sync/noinst/client_history_impl.hpp @@ -136,8 +136,8 @@ class ClientHistory final : public _impl::History, public TransformHistory { /// The returned SyncProgress is the one that was last stored by /// set_sync_progress(), or `SyncProgress{}` if set_sync_progress() has /// never been called. - void get_status(version_type& current_client_version, SaltedFileIdent& client_file_ident, SyncProgress& progress, - bool* has_pending_client_reset = nullptr) const; + void get_status(version_type& current_client_version, SaltedFileIdent& client_file_ident, + SyncProgress& progress) const; /// Stores the server assigned client file identifier in the associated /// Realm file, such that it is available via get_status() during future diff --git a/src/realm/sync/noinst/client_impl_base.cpp b/src/realm/sync/noinst/client_impl_base.cpp index 21e2a4c3453..0db329e4815 100644 --- a/src/realm/sync/noinst/client_impl_base.cpp +++ b/src/realm/sync/noinst/client_impl_base.cpp @@ -1700,15 +1700,13 @@ void Session::activate() logger.debug("Activating"); // Throws - bool has_pending_client_reset = false; if (REALM_LIKELY(!get_client().is_dry_run())) { 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); if (!m_performing_client_reset) { - get_history().get_status(m_last_version_available, m_client_file_ident, m_progress, - &has_pending_client_reset); // Throws + get_history().get_status(m_last_version_available, m_client_file_ident, m_progress); // Throws } } logger.debug("client_file_ident = %1, client_file_ident_salt = %2", m_client_file_ident.ident, @@ -1745,9 +1743,8 @@ void Session::activate() on_integration_failure(IntegrationException(exception_to_status())); } - if (has_pending_client_reset) { - handle_pending_client_reset_acknowledgement(); - } + // Checks if there is a pending client reset + handle_pending_client_reset_acknowledgement(); } @@ -2273,11 +2270,9 @@ bool Session::client_reset_if_needed() 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(), *client_reset_config->fresh_copy, client_reset_config->mode, - std::move(client_reset_config->notify_before_client_reset), - std::move(client_reset_config->notify_after_client_reset), m_client_file_ident, get_flx_subscription_store(), - on_flx_version_complete, client_reset_config->recovery_is_allowed); + 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); if (!did_reset) { return false; } @@ -2286,9 +2281,7 @@ bool Session::client_reset_if_needed() logger.debug("Client reset is completed, path=%1", get_realm_path()); // Throws SaltedFileIdent client_file_ident; - bool has_pending_client_reset = false; - get_history().get_status(m_last_version_available, client_file_ident, m_progress, - &has_pending_client_reset); // Throws + get_history().get_status(m_last_version_available, client_file_ident, m_progress); // 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, @@ -2305,9 +2298,8 @@ bool Session::client_reset_if_needed() m_allow_upload = true; REALM_ASSERT_EX(m_last_version_selected_for_upload == 0, m_last_version_selected_for_upload); - if (has_pending_client_reset) { - handle_pending_client_reset_acknowledgement(); - } + // Checks if there is a pending client reset + handle_pending_client_reset_acknowledgement(); update_subscription_version_info(); diff --git a/src/realm/sync/noinst/client_impl_base.hpp b/src/realm/sync/noinst/client_impl_base.hpp index f6a6d4a84a3..57555f2f297 100644 --- a/src/realm/sync/noinst/client_impl_base.hpp +++ b/src/realm/sync/noinst/client_impl_base.hpp @@ -10,7 +10,6 @@ #include #include #include -#include #include #include #include diff --git a/src/realm/sync/noinst/client_reset.cpp b/src/realm/sync/noinst/client_reset.cpp index 8a243ef2a84..8c2fe55f83b 100644 --- a/src/realm/sync/noinst/client_reset.cpp +++ b/src/realm/sync/noinst/client_reset.cpp @@ -28,6 +28,7 @@ #include #include #include +#include #include #include @@ -61,9 +62,7 @@ std::ostream& operator<<(std::ostream& os, const ClientResyncMode& mode) return os; } -} // namespace realm - -namespace realm::_impl::client_reset { +namespace _impl::client_reset { static inline bool should_skip_table(const Transaction& group, TableKey key) { @@ -411,129 +410,34 @@ void transfer_group(const Transaction& group_src, Transaction& group_dst, util:: } } -// A table without a "class_" prefix will not generate sync instructions. -constexpr static std::string_view s_meta_reset_table_name("client_reset_metadata"); -constexpr static std::string_view s_pk_col_name("id"); -constexpr static std::string_view s_version_column_name("version"); -constexpr static std::string_view s_timestamp_col_name("event_time"); -constexpr static std::string_view s_reset_type_col_name("type_of_reset"); -constexpr int64_t metadata_version = 1; - -void remove_pending_client_resets(Transaction& wt) +ClientResyncMode reset_precheck_guard(const TransactionRef& wt_local, ClientResyncMode mode, + PendingReset::Action action, const std::optional& error, + util::Logger& logger) { - if (auto table = wt.get_table(s_meta_reset_table_name); table && !table->is_empty()) { - table->clear(); - } -} - -util::Optional has_pending_reset(const Transaction& rt) -{ - ConstTableRef table = rt.get_table(s_meta_reset_table_name); - if (!table || table->size() == 0) { - return util::none; - } - ColKey timestamp_col = table->get_column_key(s_timestamp_col_name); - ColKey type_col = table->get_column_key(s_reset_type_col_name); - ColKey version_col = table->get_column_key(s_version_column_name); - REALM_ASSERT(timestamp_col); - REALM_ASSERT(type_col); - REALM_ASSERT(version_col); - if (table->size() > 1) { - // this may happen if a future version of this code changes the format and expectations around reset metadata. - throw ClientResetFailed( - util::format("Previous client resets detected (%1) but only one is expected.", table->size())); - } - Obj first = *table->begin(); - REALM_ASSERT(first); - PendingReset pending; - int64_t version = first.get(version_col); - pending.time = first.get(timestamp_col); - if (version > metadata_version) { - throw ClientResetFailed(util::format("Unsupported client reset metadata version: %1 vs %2, from %3", version, - metadata_version, pending.time)); - } - int64_t type = first.get(type_col); - if (type == 0) { - pending.type = ClientResyncMode::DiscardLocal; - } - else if (type == 1) { - pending.type = ClientResyncMode::Recover; - } - else { - throw ClientResetFailed( - util::format("Unsupported client reset metadata type: %1 from %2", type, pending.time)); - } - return pending; -} - -void track_reset(Transaction& wt, ClientResyncMode mode) -{ - REALM_ASSERT(mode != ClientResyncMode::Manual); - TableRef table = wt.get_table(s_meta_reset_table_name); - ColKey version_col, timestamp_col, type_col; - if (!table) { - table = wt.add_table_with_primary_key(s_meta_reset_table_name, type_ObjectId, s_pk_col_name); - REALM_ASSERT(table); - version_col = table->add_column(type_Int, s_version_column_name); - timestamp_col = table->add_column(type_Timestamp, s_timestamp_col_name); - type_col = table->add_column(type_Int, s_reset_type_col_name); - } - else { - version_col = table->get_column_key(s_version_column_name); - timestamp_col = table->get_column_key(s_timestamp_col_name); - type_col = table->get_column_key(s_reset_type_col_name); - } - REALM_ASSERT(version_col); - REALM_ASSERT(timestamp_col); - REALM_ASSERT(type_col); - int64_t mode_val = 0; // Discard - if (mode == ClientResyncMode::Recover || mode == ClientResyncMode::RecoverOrDiscard) { - mode_val = 1; // Recover - } - - if (table->size() > 1) { - // this may happen if a future version of this code changes the format and expectations around reset metadata. - throw ClientResetFailed( - util::format("Previous client resets detected (%1) but only one is expected.", table->size())); - } - table->create_object_with_primary_key(ObjectId::gen(), - {{version_col, metadata_version}, - {timestamp_col, Timestamp(std::chrono::system_clock::now())}, - {type_col, mode_val}}); - - // Ensure we save the tracker object even if we encounter an error and roll - // back the client reset later - wt.commit_and_continue_writing(); -} - -static ClientResyncMode reset_precheck_guard(Transaction& wt, ClientResyncMode mode, bool recovery_is_allowed, - util::Logger& logger) -{ - if (auto previous_reset = has_pending_reset(wt)) { - logger.info(util::LogCategory::reset, "A previous reset was detected of type: '%1' at: %2", - previous_reset->type, previous_reset->time); - switch (previous_reset->type) { + if (auto previous_reset = sync::PendingResetStore::has_pending_reset(wt_local)) { + logger.info(util::LogCategory::reset, "Found a previous %1", *previous_reset); + switch (previous_reset->mode) { case ClientResyncMode::Manual: REALM_UNREACHABLE(); case ClientResyncMode::DiscardLocal: throw ClientResetFailed(util::format("A previous '%1' mode reset from %2 did not succeed, " "giving up on '%3' mode to prevent a cycle", - previous_reset->type, previous_reset->time, mode)); + previous_reset->mode, previous_reset->time, mode)); case ClientResyncMode::Recover: switch (mode) { case ClientResyncMode::Recover: throw ClientResetFailed(util::format("A previous '%1' mode reset from %2 did not succeed, " "giving up on '%3' mode to prevent a cycle", - previous_reset->type, previous_reset->time, mode)); + previous_reset->mode, previous_reset->time, mode)); case ClientResyncMode::RecoverOrDiscard: mode = ClientResyncMode::DiscardLocal; logger.info(util::LogCategory::reset, "A previous '%1' mode reset from %2 downgrades this mode ('%3') to DiscardLocal", - previous_reset->type, previous_reset->time, mode); - remove_pending_client_resets(wt); + previous_reset->mode, previous_reset->time, mode); + sync::PendingResetStore::clear_pending_reset(wt_local); break; case ClientResyncMode::DiscardLocal: - remove_pending_client_resets(wt); + sync::PendingResetStore::clear_pending_reset(wt_local); // previous mode Recover and this mode is Discard, this is not a cycle yet break; case ClientResyncMode::Manual: @@ -543,10 +447,10 @@ static ClientResyncMode reset_precheck_guard(Transaction& wt, ClientResyncMode m case ClientResyncMode::RecoverOrDiscard: throw ClientResetFailed(util::format("Unexpected previous '%1' mode reset from %2 did not " "succeed, giving up on '%3' mode to prevent a cycle", - previous_reset->type, previous_reset->time, mode)); + previous_reset->mode, previous_reset->time, mode)); } } - if (!recovery_is_allowed) { + if (action == PendingReset::Action::ClientResetNoRecovery) { if (mode == ClientResyncMode::Recover) { throw ClientResetFailed( "Client reset mode is set to 'Recover' but the server does not allow recovery for this client"); @@ -558,27 +462,30 @@ static ClientResyncMode reset_precheck_guard(Transaction& wt, ClientResyncMode m mode = ClientResyncMode::DiscardLocal; } } - track_reset(wt, mode); + sync::PendingResetStore::track_reset(wt_local, mode, action, error); + // Ensure we save the tracker object even if we encounter an error and roll + // back the client reset later + wt_local->commit_and_continue_writing(); return mode; } -bool perform_client_reset_diff(DB& db_local, DB& db_remote, sync::SaltedFileIdent client_file_ident, - util::Logger& logger, ClientResyncMode mode, bool recovery_is_allowed, - sync::SubscriptionStore* sub_store, +bool perform_client_reset_diff(DB& db_local, sync::ClientReset& reset_config, sync::SaltedFileIdent client_file_ident, + util::Logger& logger, sync::SubscriptionStore* sub_store, util::FunctionRef on_flx_version_complete) { + DB& db_remote = *reset_config.fresh_copy; auto wt_local = db_local.start_write(); - auto actual_mode = reset_precheck_guard(*wt_local, mode, recovery_is_allowed, logger); + auto actual_mode = + reset_precheck_guard(wt_local, reset_config.mode, reset_config.action, reset_config.error, logger); 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, recovery_is_allowed = %6, " - "actual_mode = %7, will_recover = %8", - db_local.get_path(), client_file_ident.ident, client_file_ident.salt, db_remote.get_path(), mode, - recovery_is_allowed, actual_mode, recover_local_changes); + "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(*db_local.get_replication()); auto& history_local = repl_local.get_history(); @@ -642,4 +549,5 @@ bool perform_client_reset_diff(DB& db_local, DB& db_remote, sync::SaltedFileIden return recover_local_changes; } -} // namespace realm::_impl::client_reset +} // namespace _impl::client_reset +} // namespace realm diff --git a/src/realm/sync/noinst/client_reset.hpp b/src/realm/sync/noinst/client_reset.hpp index afd8307a297..4035072c2e2 100644 --- a/src/realm/sync/noinst/client_reset.hpp +++ b/src/realm/sync/noinst/client_reset.hpp @@ -20,11 +20,11 @@ #define REALM_NOINST_CLIENT_RESET_HPP #include -#include +#include #include #include -#include +#include namespace realm { @@ -32,9 +32,6 @@ std::ostream& operator<<(std::ostream& os, const ClientResyncMode& mode); namespace sync { class SubscriptionStore; -} - -namespace _impl::client_reset { // The reset fails if there seems to be conflict between the // instructions and state. @@ -50,6 +47,10 @@ struct ClientResetFailed : public std::runtime_error { using std::runtime_error::runtime_error; }; +} // namespace sync + +namespace _impl::client_reset { + // transfer_group() transfers all tables, columns, objects and values from the src // group to the dst group and deletes everything in the dst group that is absent in // the src group. An update is only performed when a comparison shows that a @@ -61,13 +62,9 @@ struct ClientResetFailed : public std::runtime_error { void transfer_group(const Transaction& tr_src, Transaction& tr_dst, util::Logger& logger, bool allow_schema_additions); -struct PendingReset { - ClientResyncMode type; - Timestamp time; -}; -void remove_pending_client_resets(Transaction& wt); -util::Optional has_pending_reset(const Transaction& wt); -void track_reset(Transaction& wt, ClientResyncMode mode); +ClientResyncMode reset_precheck_guard(const TransactionRef& wt_local, ClientResyncMode mode, + sync::ProtocolErrorInfo::Action action, const std::optional& error, + util::Logger& logger); // preform_client_reset_diff() takes the Realm performs a client reset on // the Realm in 'path_local' given the Realm 'path_fresh' as the source of truth. @@ -76,8 +73,8 @@ void track_reset(Transaction& wt, ClientResyncMode mode); // If the fresh path is provided, the local Realm is changed such that its state is equal // to the fresh Realm. Then the local Realm will have its client file ident set to // 'client_file_ident' -bool perform_client_reset_diff(DB& db, DB& db_remote, sync::SaltedFileIdent client_file_ident, util::Logger& logger, - ClientResyncMode mode, bool recovery_is_allowed, sync::SubscriptionStore* sub_store, +bool perform_client_reset_diff(DB& db, sync::ClientReset& reset_config, sync::SaltedFileIdent client_file_ident, + util::Logger& logger, sync::SubscriptionStore* sub_store, util::FunctionRef on_flx_version_complete); } // namespace _impl::client_reset diff --git a/src/realm/sync/noinst/client_reset_operation.cpp b/src/realm/sync/noinst/client_reset_operation.cpp index b9d96b8d7b7..1ff26c19cf8 100644 --- a/src/realm/sync/noinst/client_reset_operation.cpp +++ b/src/realm/sync/noinst/client_reset_operation.cpp @@ -51,20 +51,20 @@ bool is_fresh_path(const std::string& path) return path.substr(path.size() - suffix_len, suffix_len) == c_fresh_suffix; } -bool perform_client_reset(util::Logger& logger, DB& db, DB& fresh_db, ClientResyncMode mode, - CallbackBeforeType notify_before, CallbackAfterType notify_after, +bool perform_client_reset(util::Logger& logger, DB& db, sync::ClientReset&& reset_config, sync::SaltedFileIdent new_file_ident, sync::SubscriptionStore* sub_store, - util::FunctionRef on_flx_version, bool recovery_is_allowed) + util::FunctionRef on_flx_version) { - REALM_ASSERT(mode != ClientResyncMode::Manual); + REALM_ASSERT(reset_config.mode != ClientResyncMode::Manual); + REALM_ASSERT(reset_config.fresh_copy); logger.debug(util::LogCategory::reset, - "Possibly beginning client reset operation: realm_path = %1, mode = %2, recovery_allowed = %3", - db.get_path(), mode, recovery_is_allowed); + "Possibly beginning client reset operation: realm_path = %1, mode = %2, action = %3, error = %4", + db.get_path(), reset_config.mode, reset_config.action, reset_config.error); auto always_try_clean_up = util::make_scope_exit([&]() noexcept { - std::string path_to_clean = fresh_db.get_path(); + std::string path_to_clean = reset_config.fresh_copy->get_path(); try { - fresh_db.close(); + reset_config.fresh_copy->close(); constexpr bool delete_lockfile = true; DB::delete_files(path_to_clean, nullptr, delete_lockfile); } @@ -88,6 +88,9 @@ bool perform_client_reset(util::Logger& logger, DB& db, DB& fresh_db, ClientResy return false; } + auto notify_before = std::move(reset_config.notify_before_client_reset); + auto notify_after = std::move(reset_config.notify_after_client_reset); + VersionID frozen_before_state_version = notify_before ? notify_before() : latest_version; // If m_notify_after is set, pin the previous state to keep it around. @@ -95,8 +98,8 @@ bool perform_client_reset(util::Logger& logger, DB& db, DB& fresh_db, ClientResy if (notify_after) { previous_state = db.start_frozen(frozen_before_state_version); } - bool did_recover = client_reset::perform_client_reset_diff( - db, fresh_db, new_file_ident, logger, mode, recovery_is_allowed, sub_store, on_flx_version); // throws + bool did_recover = client_reset::perform_client_reset_diff(db, reset_config, new_file_ident, logger, sub_store, + on_flx_version); // throws if (notify_after) { notify_after(previous_state->get_version_of_current_transaction(), did_recover); diff --git a/src/realm/sync/noinst/client_reset_operation.hpp b/src/realm/sync/noinst/client_reset_operation.hpp index 5c2ce5bec7f..09a301b6b51 100644 --- a/src/realm/sync/noinst/client_reset_operation.hpp +++ b/src/realm/sync/noinst/client_reset_operation.hpp @@ -23,6 +23,7 @@ #include #include #include +#include #include #include @@ -37,10 +38,9 @@ using CallbackAfterType = util::UniqueFunction; std::string get_fresh_path_for(const std::string& realm_path); bool is_fresh_path(const std::string& realm_path); -bool perform_client_reset(util::Logger& logger, DB& target_db, DB& fresh_db, ClientResyncMode mode, - CallbackBeforeType notify_before, CallbackAfterType notify_after, - sync::SaltedFileIdent new_file_ident, sync::SubscriptionStore*, - util::FunctionRef on_flx_version, bool recovery_is_allowed); +bool perform_client_reset(util::Logger& logger, DB& db, sync::ClientReset&& reset_config, + sync::SaltedFileIdent new_file_ident, sync::SubscriptionStore* sub_store, + util::FunctionRef on_flx_version); } // namespace realm::_impl::client_reset diff --git a/src/realm/sync/noinst/client_reset_recovery.cpp b/src/realm/sync/noinst/client_reset_recovery.cpp index 4fbaa586f8f..80b5d0aad09 100644 --- a/src/realm/sync/noinst/client_reset_recovery.cpp +++ b/src/realm/sync/noinst/client_reset_recovery.cpp @@ -538,7 +538,7 @@ REALM_NORETURN void RecoverLocalChangesetsHandler::handle_error(const std::strin std::string full_message = util::format("Unable to automatically recover local changes during client reset: '%1'", message); m_logger.error(util::LogCategory::reset, full_message.c_str()); - throw realm::_impl::client_reset::ClientResetFailed(full_message); + throw realm::sync::ClientResetFailed(full_message); } util::AppendBuffer RecoverLocalChangesetsHandler::process_changeset(const ChunkedBinaryData& changeset) diff --git a/src/realm/sync/noinst/migration_store.cpp b/src/realm/sync/noinst/migration_store.cpp index ff6c631048d..db27e510141 100644 --- a/src/realm/sync/noinst/migration_store.cpp +++ b/src/realm/sync/noinst/migration_store.cpp @@ -51,36 +51,30 @@ bool MigrationStore::load_data(bool read_only) }}, }; - std::optional schema_version; auto tr = m_db->start_read(); - if (read_only) { - // Writing is disabled - SyncMetadataSchemaVersionsReader schema_versions(tr); - schema_version = schema_versions.get_version_for(tr, internal_schema_groups::c_flx_migration_store); - if (!schema_version) { - return false; // Either table is not initialized or version does not exist - } - } - else { // writable - SyncMetadataSchemaVersions schema_versions(tr); - schema_version = schema_versions.get_version_for(tr, internal_schema_groups::c_flx_migration_store); - // Create the version and metadata_schema if it doesn't exist - if (!schema_version) { - tr->promote_to_write(); - schema_versions.set_version_for(tr, internal_schema_groups::c_flx_migration_store, c_schema_version); - create_sync_metadata_schema(tr, &internal_tables); - tr->commit_and_continue_as_read(); - } - } - // Load the metadata schema unless it was just created - if (!m_migration_table) { + // Start with a reader so it doesn't try to write until we are ready + SyncMetadataSchemaVersionsReader schema_versions_reader(tr); + if (auto schema_version = + schema_versions_reader.get_version_for(tr, internal_schema_groups::c_flx_migration_store)) { if (*schema_version != c_schema_version) { throw RuntimeError(ErrorCodes::UnsupportedFileFormatVersion, "Invalid schema version for flexible sync migration store metadata"); } load_sync_metadata_schema(tr, &internal_tables); } - + else { + if (read_only) { + // Writing is disabled + return false; // Either table is not initialized or version does not exist + } + tr->promote_to_write(); + // Ensure the schema versions table is initialized (may add its own commit) + SyncMetadataSchemaVersions schema_versions(tr); + // Create the metadata schema and set the version (in the same commit) + schema_versions.set_version_for(tr, internal_schema_groups::c_flx_migration_store, c_schema_version); + create_sync_metadata_schema(tr, &internal_tables); + tr->commit_and_continue_as_read(); + } REALM_ASSERT(m_migration_table); // Read the migration object if exists, or default to not migrated diff --git a/src/realm/sync/noinst/pending_bootstrap_store.cpp b/src/realm/sync/noinst/pending_bootstrap_store.cpp index 03fb7f97231..d576762cf30 100644 --- a/src/realm/sync/noinst/pending_bootstrap_store.cpp +++ b/src/realm/sync/noinst/pending_bootstrap_store.cpp @@ -97,8 +97,10 @@ PendingBootstrapStore::PendingBootstrapStore(DBRef db, util::Logger& logger) }}}; auto tr = m_db->start_read(); - SyncMetadataSchemaVersions schema_versions(tr); - if (auto schema_version = schema_versions.get_version_for(tr, internal_schema_groups::c_pending_bootstraps)) { + // Start with a reader so it doesn't try to write until we are ready + SyncMetadataSchemaVersionsReader schema_versions_reader(tr); + if (auto schema_version = + schema_versions_reader.get_version_for(tr, internal_schema_groups::c_pending_bootstraps)) { if (*schema_version != c_schema_version) { throw RuntimeError(ErrorCodes::SchemaVersionMismatch, "Invalid schema version for FLX sync pending bootstrap table group"); @@ -107,10 +109,14 @@ PendingBootstrapStore::PendingBootstrapStore(DBRef db, util::Logger& logger) } else { tr->promote_to_write(); - create_sync_metadata_schema(tr, &internal_tables); + // Ensure the schema versions table is initialized (may add its own commit) + SyncMetadataSchemaVersions schema_versions(tr); + // Create the metadata schema and set the version (in the same commit) schema_versions.set_version_for(tr, internal_schema_groups::c_pending_bootstraps, c_schema_version); + create_sync_metadata_schema(tr, &internal_tables); tr->commit_and_continue_as_read(); } + REALM_ASSERT(m_table); if (auto bootstrap_table = tr->get_table(m_table); !bootstrap_table->is_empty()) { m_has_pending = true; diff --git a/src/realm/sync/noinst/pending_reset_store.cpp b/src/realm/sync/noinst/pending_reset_store.cpp new file mode 100644 index 00000000000..266dbb36d39 --- /dev/null +++ b/src/realm/sync/noinst/pending_reset_store.cpp @@ -0,0 +1,313 @@ +/////////////////////////////////////////////////////////////////////////// +// +// Copyright 2024 Realm Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +//////////////////////////////////////////////////////////////////////////// + +#include + +#include +#include + +#include +#include + +using namespace realm; +using namespace _impl; +using namespace sync; + +namespace realm::sync { + +std::ostream& operator<<(std::ostream& os, const sync::PendingReset& pr) +{ + if (pr.action == sync::ProtocolErrorInfo::Action::NoAction || pr.time.is_null()) { + os << "empty pending client reset"; + } + else if (pr.action != sync::ProtocolErrorInfo::Action::ClientReset) { + os << "pending '" << pr.action << "' client reset of type: '" << pr.mode << "' at: " << pr.time; + } + else { + os << "pending client reset of type: '" << pr.mode << "' at: " << pr.time; + } + if (pr.error) { + os << " for error: " << *pr.error; + } + return os; +} + +bool operator==(const sync::PendingReset& lhs, const sync::PendingReset& rhs) +{ + return (lhs.mode == rhs.mode && lhs.action == rhs.action && lhs.time == rhs.time); +} + +bool operator==(const sync::PendingReset& lhs, const PendingReset::Action& action) +{ + return lhs.action == action; +} + +// A table without a "class_" prefix will not generate sync instructions. +constexpr static std::string_view s_meta_reset_table_name("client_reset_metadata"); +constexpr static std::string_view s_pk_col_name("id"); +constexpr static std::string_view s_timestamp_col_name("reset_time"); +constexpr static std::string_view s_reset_recovery_mode_col_name("reset_mode"); +constexpr static std::string_view s_reset_action_col_name("reset_action"); +constexpr static std::string_view s_reset_error_code_col_name("reset_error_code"); +constexpr static std::string_view s_reset_error_msg_col_name("reset_error_msg"); +constexpr int64_t s_pending_reset_version = 2; + +void PendingResetStore::clear_pending_reset(const TransactionRef& wr_tr) +{ + // Write transaction required + REALM_ASSERT(wr_tr->get_transact_stage() == DB::TransactStage::transact_Writing); + auto reset_store = PendingResetStore::load_or_create_schema(wr_tr); + REALM_ASSERT(reset_store.m_pending_reset_table); + // Ensure the pending reset table is empty + if (auto table = wr_tr->get_table(reset_store.m_pending_reset_table); table && !table->is_empty()) { + table->clear(); + } + // Don't commit at the end - allow the caller to do it when they are ready +} + +std::optional PendingResetStore::has_pending_reset(const TransactionRef& rd_tr) +{ + // Make sure the schema has been loaded and try to read legacy data if it's not found + auto reset_store = PendingResetStore::load_schema(rd_tr); + if (!reset_store) { + return PendingResetStore::read_legacy_pending_reset(rd_tr); + } + // Otherwise, read the pending reset entry using the schema metadata + REALM_ASSERT(reset_store->m_pending_reset_table); + auto table = rd_tr->get_table(reset_store->m_pending_reset_table); + + if (!table || table->size() == 0) { + return std::nullopt; + } + if (table->size() > 1) { + // this may happen if a future version of this code changes the format and expectations around reset metadata. + throw ClientResetFailed( + util::format("Previous client resets detected (%1) but only one is expected.", table->size())); + } + auto reset_entry = *table->begin(); + PendingReset pending; + pending.time = reset_entry.get(reset_store->m_timestamp); + pending.mode = to_resync_mode(reset_entry.get(reset_store->m_recovery_mode)); + pending.action = to_reset_action(reset_entry.get(reset_store->m_action)); + auto error_code = reset_entry.get(reset_store->m_error_code); + if (error_code > 0) { + pending.error = Status(static_cast(error_code), + reset_entry.get(reset_store->m_error_message)); + } + return pending; +} + +void PendingResetStore::track_reset(const TransactionRef& wr_tr, ClientResyncMode mode, PendingReset::Action action, + const std::optional& error) +{ + REALM_ASSERT(mode != ClientResyncMode::Manual); + // Write transaction required + REALM_ASSERT(wr_tr->get_transact_stage() == DB::TransactStage::transact_Writing); + if (auto table = wr_tr->get_table(s_meta_reset_table_name); table && table->size() > 0) { + // this may happen if a future version of this code changes the format and expectations around reset + // metadata. + throw ClientResetFailed( + util::format("Previous client resets detected (%1) but only one is expected.", table->size())); + } + auto reset_store = PendingResetStore::load_or_create_schema(wr_tr); + + REALM_ASSERT(reset_store.m_pending_reset_table); + auto table = wr_tr->get_table(reset_store.m_pending_reset_table); + REALM_ASSERT(table); + // Create the new object + auto obj = table->create_object_with_primary_key( + ObjectId::gen(), { + {reset_store.m_timestamp, Timestamp(std::chrono::system_clock::now())}, + {reset_store.m_recovery_mode, from_resync_mode(mode)}, + {reset_store.m_action, from_reset_action(action)}, + }); + // Add the error, if provided + if (error) { + obj.set(reset_store.m_error_code, static_cast(error->code())); + obj.set(reset_store.m_error_message, error->reason()); + } + // Don't commit at the end - allow the caller to do it when they are ready +} + +PendingResetStore::PendingResetStore(const TransactionRef& rd_tr) + : m_internal_tables{ + {&m_pending_reset_table, + s_meta_reset_table_name, + {&m_id, s_pk_col_name, type_ObjectId}, + { + {&m_timestamp, s_timestamp_col_name, type_Timestamp}, + {&m_recovery_mode, s_reset_recovery_mode_col_name, type_Int}, + {&m_action, s_reset_action_col_name, type_Int}, + {&m_error_code, s_reset_error_code_col_name, type_Int, true}, + {&m_error_message, s_reset_error_msg_col_name, type_String, true}, + }}, + } +{ + // Works with read, write, and frozen transactions + SyncMetadataSchemaVersionsReader schema_versions(rd_tr); + auto schema_version = schema_versions.get_version_for(rd_tr, internal_schema_groups::c_pending_reset_store); + + // Load the metadata schema info if a schema version was found + if (schema_version) { + if (*schema_version != s_pending_reset_version) { + // Unsupported schema version + throw RuntimeError(ErrorCodes::UnsupportedFileFormatVersion, + "Found invalid schema version for existing client reset cycle tracking metadata"); + } + load_sync_metadata_schema(rd_tr, &m_internal_tables); + if (m_pending_reset_table) { + // If the schema info was read, then store the schema version + m_schema_version = schema_version; + } + } +} + +std::optional PendingResetStore::load_schema(const TransactionRef& rd_tr) +{ + PendingResetStore reset_store(rd_tr); + if (reset_store.m_schema_version) { + return reset_store; + } + return std::nullopt; +} + +PendingResetStore PendingResetStore::load_or_create_schema(const TransactionRef& wr_tr) +{ + PendingResetStore reset_store(wr_tr); + if (reset_store.m_schema_version) { + // If the schema metadata was found, return the initialized class + return reset_store; + } + // Otherwise, set it up from scratch - Make sure the transaction is set for writing + if (wr_tr->get_transact_stage() == DB::TransactStage::transact_Reading) { + wr_tr->promote_to_write(); + } + // Ensure writing - all other transaction stages are not allowed + REALM_ASSERT_EX(wr_tr->get_transact_stage() == DB::TransactStage::transact_Writing, wr_tr->get_transact_stage()); + + // Drop the old table and any stale pending resets + if (wr_tr->has_table(s_meta_reset_table_name)) { + wr_tr->remove_table(s_meta_reset_table_name); + } + + // Ensure the schema versions table is initialized (may add its own commit) + SyncMetadataSchemaVersions schema_versions(wr_tr); + // Create the metadata schema and set the version (in the same commit) + schema_versions.set_version_for(wr_tr, internal_schema_groups::c_pending_reset_store, s_pending_reset_version); + create_sync_metadata_schema(wr_tr, &reset_store.m_internal_tables); + REALM_ASSERT(reset_store.m_pending_reset_table); + reset_store.m_schema_version = s_pending_reset_version; + + // Don't commit yet + return reset_store; +} + +std::optional PendingResetStore::read_legacy_pending_reset(const TransactionRef& rd_tr) +{ + // Try to read the pending reset info from v1 of the schema + constexpr static std::string_view s_v1_version_column_name("version"); + constexpr static std::string_view s_v1_timestamp_col_name("event_time"); + constexpr static std::string_view s_v1_reset_mode_col_name("type_of_reset"); + + // Check for pending reset v1 - does not use schema version + TableRef table = rd_tr->get_table(s_meta_reset_table_name); + if (table && table->size() > 0) { + ColKey version_col = table->get_column_key(s_v1_version_column_name); + ColKey timestamp_col = table->get_column_key(s_v1_timestamp_col_name); + ColKey mode_col = table->get_column_key(s_v1_reset_mode_col_name); + Obj reset_entry = *table->begin(); + + if (version_col && reset_entry.get(version_col) == 1LL) { + REALM_ASSERT(timestamp_col); + REALM_ASSERT(mode_col); + PendingReset pending; + pending.time = reset_entry.get(timestamp_col); + pending.mode = to_resync_mode(reset_entry.get(mode_col)); + // Create a fake action depending on the resync mode + pending.action = pending.mode == ClientResyncMode::DiscardLocal + ? sync::ProtocolErrorInfo::Action::ClientResetNoRecovery + : sync::ProtocolErrorInfo::Action::ClientReset; + return pending; + } + } + // Add checking for future schema versions here + return std::nullopt; +} + +int64_t PendingResetStore::from_reset_action(PendingReset::Action action) +{ + switch (action) { + case PendingReset::Action::ClientReset: + return 1; + case PendingReset::Action::ClientResetNoRecovery: + return 2; + case PendingReset::Action::MigrateToFLX: + return 3; + case PendingReset::Action::RevertToPBS: + return 4; + default: + throw ClientResetFailed(util::format("Unsupported client reset action: %1 for pending reset", action)); + } +} + +PendingReset::Action PendingResetStore::to_reset_action(int64_t action) +{ + switch (action) { + case 1: + return PendingReset::Action::ClientReset; + case 2: + return PendingReset::Action::ClientResetNoRecovery; + case 3: + return PendingReset::Action::MigrateToFLX; + case 4: + return PendingReset::Action::RevertToPBS; + default: + return PendingReset::Action::NoAction; + } +} + +ClientResyncMode PendingResetStore::to_resync_mode(int64_t mode) +{ + // Retains compatibility with v1 + // RecoverOrDiscard is treated as Recover and is not stored + switch (mode) { + case 0: // DiscardLocal + return ClientResyncMode::DiscardLocal; + case 1: // Recover + return ClientResyncMode::Recover; + default: + throw ClientResetFailed(util::format("Unsupported client reset resync mode: %1 for pending reset", mode)); + } +} + +int64_t PendingResetStore::from_resync_mode(ClientResyncMode mode) +{ + // Retains compatibility with v1 + switch (mode) { + case ClientResyncMode::DiscardLocal: + return 0; // DiscardLocal + case ClientResyncMode::RecoverOrDiscard: + [[fallthrough]]; // RecoverOrDiscard is treated as Recover + case ClientResyncMode::Recover: + return 1; // Recover + default: + throw ClientResetFailed(util::format("Unsupported client reset resync mode: %1 for pending reset", mode)); + } +} + +} // namespace realm::sync diff --git a/src/realm/sync/noinst/pending_reset_store.hpp b/src/realm/sync/noinst/pending_reset_store.hpp new file mode 100644 index 00000000000..a6e0878d8b7 --- /dev/null +++ b/src/realm/sync/noinst/pending_reset_store.hpp @@ -0,0 +1,91 @@ +/////////////////////////////////////////////////////////////////////////// +// +// Copyright 2024 Realm Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +//////////////////////////////////////////////////////////////////////////// + +#ifndef REALM_NOINST_PENDING_RESET_STORE_HPP +#define REALM_NOINST_PENDING_RESET_STORE_HPP + +#include +#include +#include +#include +#include + +#include + +#include +#include + +namespace realm::sync { + +struct PendingReset { + using Action = sync::ProtocolErrorInfo::Action; + Timestamp time; + ClientResyncMode mode; + Action action = Action::NoAction; + std::optional error; +}; + +std::ostream& operator<<(std::ostream& os, const sync::PendingReset& pr); +bool operator==(const sync::PendingReset& lhs, const sync::PendingReset& rhs); +bool operator==(const sync::PendingReset& lhs, const PendingReset::Action& action); + +class PendingResetStore { +public: + // Store the pending reset tracking information - it is an error if the tracking info already + // exists in the store + // Requires a writable transaction and changes must be committed manually + static void track_reset(const TransactionRef& wr_tr, ClientResyncMode mode, PendingReset::Action action, + const std::optional& error = std::nullopt); + // Clear the pending reset tracking information, if it exists + // Requires a writable transaction and changes must be committed manually + static void clear_pending_reset(const TransactionRef& wr_tr); + static std::optional has_pending_reset(const TransactionRef& rd_tr); + + static int64_t from_reset_action(PendingReset::Action action); + static PendingReset::Action to_reset_action(int64_t action); + static ClientResyncMode to_resync_mode(int64_t mode); + static int64_t from_resync_mode(ClientResyncMode mode); + +private: + // The instantiated class is only used internally + PendingResetStore(const TransactionRef& rd_tr); + + std::vector m_internal_tables; + TableKey m_pending_reset_table; + ColKey m_id; + ColKey m_version; + ColKey m_timestamp; + ColKey m_recovery_mode; + ColKey m_action; + ColKey m_error_code; + ColKey m_error_message; + std::optional m_schema_version = std::nullopt; + + // Returns true if the schema was loaded + static std::optional load_schema(const TransactionRef& rd_tr); + // Loads the schema or creates it if it doesn't exist + // Requires a writable transaction and changes must be committed manually + static PendingResetStore load_or_create_schema(const TransactionRef& wr_tr); + + // Try to read the pending reset info from v1 of the schema + static std::optional read_legacy_pending_reset(const TransactionRef& rd_tr); +}; + +} // namespace realm::sync + +#endif // REALM_NOINST_PENDING_RESET_STORE_HPP diff --git a/src/realm/sync/noinst/sync_metadata_schema.cpp b/src/realm/sync/noinst/sync_metadata_schema.cpp index 97ac90ccb72..4b5f2c76617 100644 --- a/src/realm/sync/noinst/sync_metadata_schema.cpp +++ b/src/realm/sync/noinst/sync_metadata_schema.cpp @@ -165,17 +165,14 @@ void load_sync_metadata_schema(const TransactionRef& tr, std::vector legacy_table_def{ - {&legacy_table_key, c_flx_metadata_table, {{&legacy_version_key, c_meta_schema_version_field, type_Int}}}}; std::vector unified_schema_version_table_def{ {&m_table, c_sync_internal_schemas_table, {&m_schema_group_field, c_meta_schema_schema_group_field, type_String}, {{&m_version_field, c_meta_schema_version_field, type_Int}}}}; - REALM_ASSERT_3(tr->get_transact_stage(), ==, DB::transact_Reading); + // Any type of transaction is allowed, including frozen and write, as long as it supports reading + REALM_ASSERT_EX(tr->get_transact_stage() != DB::transact_Ready, tr->get_transact_stage()); // If the legacy_meta_table exists, then this table hasn't been converted and // the metadata schema versions information has not been upgraded/not accurate if (tr->has_table(c_flx_metadata_table)) { @@ -188,10 +185,39 @@ SyncMetadataSchemaVersionsReader::SyncMetadataSchemaVersionsReader(const Transac } } +std::optional SyncMetadataSchemaVersionsReader::get_legacy_version(const TransactionRef& tr) +{ + if (!tr->has_table(c_flx_metadata_table)) { + return std::nullopt; + } + + TableKey legacy_table_key; + ColKey legacy_version_key; + std::vector legacy_table_def{ + {&legacy_table_key, c_flx_metadata_table, {{&legacy_version_key, c_meta_schema_version_field, type_Int}}}}; + + // Convert the legacy table to the regular schema versions table if it exists + load_sync_metadata_schema(tr, &legacy_table_def); + + if (auto legacy_meta_table = tr->get_table(legacy_table_key); + legacy_meta_table && legacy_meta_table->size() > 0) { + auto legacy_obj = legacy_meta_table->get_object(0); + return legacy_obj.get(legacy_version_key); + } + + return std::nullopt; +} + std::optional SyncMetadataSchemaVersionsReader::get_version_for(const TransactionRef& tr, std::string_view schema_group_name) { if (!m_table) { + // The legacy version only applies to the subscription store, don't query otherwise + if (schema_group_name == internal_schema_groups::c_flx_subscription_store) { + if (auto legacy_version = get_legacy_version(tr)) { + return legacy_version; + } + } return util::none; } @@ -211,17 +237,17 @@ std::optional SyncMetadataSchemaVersionsReader::get_version_for(const T SyncMetadataSchemaVersions::SyncMetadataSchemaVersions(const TransactionRef& tr) : SyncMetadataSchemaVersionsReader(tr) { - TableKey legacy_table_key; - ColKey legacy_version_key; - std::vector legacy_table_def{ - {&legacy_table_key, c_flx_metadata_table, {{&legacy_version_key, c_meta_schema_version_field, type_Int}}}}; std::vector unified_schema_version_table_def{ {&m_table, c_sync_internal_schemas_table, {&m_schema_group_field, c_meta_schema_schema_group_field, type_String}, {{&m_version_field, c_meta_schema_version_field, type_Int}}}}; - REALM_ASSERT_3(tr->get_transact_stage(), ==, DB::transact_Reading); + DB::TransactStage orig = tr->get_transact_stage(); + bool modified = false; + + // Read and write transactions are allowed, but not frozen + REALM_ASSERT_EX((orig == DB::transact_Reading || orig == DB::transact_Writing), orig); // If the versions table exists, then m_table would have been initialized by the reader constructor // If the versions table doesn't exist, then initialize it now if (!m_table) { @@ -231,28 +257,34 @@ SyncMetadataSchemaVersions::SyncMetadataSchemaVersions(const TransactionRef& tr) load_sync_metadata_schema(tr, &unified_schema_version_table_def); } else { - tr->promote_to_write(); + // Only write the versions table if it doesn't exist + if (tr->get_transact_stage() != DB::transact_Writing) { + tr->promote_to_write(); + } create_sync_metadata_schema(tr, &unified_schema_version_table_def); - tr->commit_and_continue_as_read(); + modified = true; } } - if (!tr->has_table(c_flx_metadata_table)) { - return; + if (auto legacy_version = get_legacy_version(tr)) { + // Migrate from just having a subscription store metadata table to having multiple table groups with multiple + // versions. + if (tr->get_transact_stage() != DB::transact_Writing) { + tr->promote_to_write(); + } + // Only the flx subscription store can potentially have the legacy metadata table + set_version_for(tr, internal_schema_groups::c_flx_subscription_store, *legacy_version); + tr->remove_table(c_flx_metadata_table); + modified = true; } + if (!modified) + return; // nothing to commit - // Convert the legacy table to the regular schema versions table if it exists - load_sync_metadata_schema(tr, &legacy_table_def); - // Migrate from just having a subscription store metadata table to having multiple table groups with multiple - // versions. - tr->promote_to_write(); - auto legacy_meta_table = tr->get_table(legacy_table_key); - auto legacy_obj = legacy_meta_table->get_object(0); - // Only the flx subscription store can potentially have the legacy metadata table - set_version_for(tr, internal_schema_groups::c_flx_subscription_store, - legacy_obj.get(legacy_version_key)); - tr->remove_table(legacy_table_key); - tr->commit_and_continue_as_read(); + // Commit and revert to the original transact stage + if (orig == DB::transact_Reading) + tr->commit_and_continue_as_read(); + else + tr->commit_and_continue_writing(); } void SyncMetadataSchemaVersions::set_version_for(const TransactionRef& tr, std::string_view schema_group_name, diff --git a/src/realm/sync/noinst/sync_metadata_schema.hpp b/src/realm/sync/noinst/sync_metadata_schema.hpp index 2bd8efa2993..fcbaa026b99 100644 --- a/src/realm/sync/noinst/sync_metadata_schema.hpp +++ b/src/realm/sync/noinst/sync_metadata_schema.hpp @@ -36,6 +36,7 @@ namespace internal_schema_groups { constexpr static std::string_view c_flx_subscription_store("flx_subscription_store"); constexpr static std::string_view c_pending_bootstraps("pending_bootstraps"); constexpr static std::string_view c_flx_migration_store("flx_migration_store"); +constexpr static std::string_view c_pending_reset_store("pending_reset_store"); } // namespace internal_schema_groups /* @@ -129,6 +130,8 @@ class SyncMetadataSchemaVersionsReader { std::optional get_version_for(const TransactionRef& tr, std::string_view schema_group_name); + std::optional get_legacy_version(const TransactionRef& tr); + protected: TableKey m_table; ColKey m_version_field; diff --git a/src/realm/sync/subscriptions.cpp b/src/realm/sync/subscriptions.cpp index 9498d9418ca..8205d0370fa 100644 --- a/src/realm/sync/subscriptions.cpp +++ b/src/realm/sync/subscriptions.cpp @@ -630,22 +630,27 @@ SubscriptionStore::SubscriptionStore(Private, DBRef db) }; auto tr = m_db->start_read(); - SyncMetadataSchemaVersions schema_versions(tr); + // Start with a reader so it doesn't try to write until we are ready + SyncMetadataSchemaVersionsReader schema_versions_reader(tr); - if (auto schema_version = schema_versions.get_version_for(tr, internal_schema_groups::c_flx_subscription_store); - !schema_version) { - tr->promote_to_write(); - schema_versions.set_version_for(tr, internal_schema_groups::c_flx_subscription_store, c_flx_schema_version); - create_sync_metadata_schema(tr, &internal_tables); - tr->commit_and_continue_as_read(); - } - else { + if (auto schema_version = + schema_versions_reader.get_version_for(tr, internal_schema_groups::c_flx_subscription_store)) { if (*schema_version != c_flx_schema_version) { throw RuntimeError(ErrorCodes::UnsupportedFileFormatVersion, "Invalid schema version for flexible sync metadata"); } load_sync_metadata_schema(tr, &internal_tables); } + else { + tr->promote_to_write(); + // Ensure the schema versions table is initialized (may add its own commit) + SyncMetadataSchemaVersions schema_versions(tr); + // Create the metadata schema and set the version (in the same commit) + schema_versions.set_version_for(tr, internal_schema_groups::c_flx_subscription_store, c_flx_schema_version); + create_sync_metadata_schema(tr, &internal_tables); + tr->commit_and_continue_as_read(); + } + REALM_ASSERT(m_sub_set_table); // Make sure the subscription set table is properly initialized initialize_subscriptions_table(std::move(tr)); diff --git a/test/object-store/sync/app.cpp b/test/object-store/sync/app.cpp index e07a5b3bc17..98f3cbb703b 100644 --- a/test/object-store/sync/app.cpp +++ b/test/object-store/sync/app.cpp @@ -3167,6 +3167,7 @@ TEST_CASE("app: sync integration", "[sync][pbs][app][baas]") { } SECTION("pausing a session does not hold the DB open") { + auto logger = util::Logger::get_default_logger(); SyncTestFile config(app->current_user(), partition, schema); DBRef dbref; std::shared_ptr sync_sess_ext_ref; @@ -3179,29 +3180,38 @@ TEST_CASE("app: sync integration", "[sync][pbs][app][baas]") { sync_sess_ext_ref = realm->sync_session()->external_reference(); dbref = TestHelper::get_db(*realm); - // One ref each for the + // An active PBS realm should have one ref each for: // - RealmCoordinator // - SyncSession + // - MigrationStore // - SessionWrapper // - local dbref - REQUIRE(dbref.use_count() >= 4); + logger->trace("DBRef ACTIVE use count: %1", dbref.use_count()); + REQUIRE(dbref.use_count() >= 5); realm->sync_session()->pause(); state = realm->sync_session()->state(); REQUIRE(state == SyncSession::State::Paused); + logger->trace("DBRef PAUSING called use count: %1", dbref.use_count()); } - // Closing the realm should leave one ref for the SyncSession and one for the local dbref. + // Closing the realm should leave one ref each for: + // - SyncSession + // - MigrationStore + // - local dbref REQUIRE_THAT( [&] { + logger->trace("DBRef PAUSED use count: %1", dbref.use_count()); return dbref.use_count() < 4; }, ReturnsTrueWithinTimeLimit{}); - // Releasing the external reference should leave one ref (the local dbref) only. + // Releasing the external reference should leave one ref for: + // - local dbref sync_sess_ext_ref.reset(); REQUIRE_THAT( [&] { + logger->trace("DBRef TEARDOWN use count: %1", dbref.use_count()); return dbref.use_count() == 1; }, ReturnsTrueWithinTimeLimit{}); diff --git a/test/object-store/sync/client_reset.cpp b/test/object-store/sync/client_reset.cpp index 0f99eaca474..9014d8f8e35 100644 --- a/test/object-store/sync/client_reset.cpp +++ b/test/object-store/sync/client_reset.cpp @@ -37,6 +37,7 @@ #include #include #include +#include #include #include @@ -1666,21 +1667,26 @@ TEST_CASE("sync: client reset", "[sync][pbs][client reset][baas]") { } // end discard local section SECTION("cycle detection") { - auto has_reset_cycle_flag = [](SharedRealm realm) -> util::Optional<_impl::client_reset::PendingReset> { + auto has_reset_cycle_flag = [](SharedRealm realm) -> util::Optional { auto db = TestHelper::get_db(realm); - auto rt = db->start_read(); - return _impl::client_reset::has_pending_reset(*rt); + auto rd_tr = db->start_frozen(); + return sync::PendingResetStore::has_pending_reset(rd_tr); }; + auto logger = util::Logger::get_default_logger(); ThreadSafeSyncError err; local_config.sync_config->error_handler = [&](std::shared_ptr, SyncError error) { + logger->error("Detected cycle detection error: %1", error.status); err = error; }; - auto make_fake_previous_reset = [&local_config](ClientResyncMode type) { - local_config.sync_config->notify_before_client_reset = [previous_type = type](SharedRealm realm) { + auto make_fake_previous_reset = [&local_config](ClientResyncMode mode, + sync::ProtocolErrorInfo::Action action = + sync::ProtocolErrorInfo::Action::ClientReset) { + local_config.sync_config->notify_before_client_reset = [mode, action](SharedRealm realm) { auto db = TestHelper::get_db(realm); - auto wt = db->start_write(); - _impl::client_reset::track_reset(*wt, previous_type); - wt->commit(); + auto wr_tr = db->start_write(); + sync::PendingResetStore::track_reset( + wr_tr, mode, action, {{ErrorCodes::SyncClientResetRequired, "Bad client file ident"}}); + wr_tr->commit(); }; }; SECTION("a normal reset adds and removes a cycle detection flag") { @@ -1695,7 +1701,7 @@ TEST_CASE("sync: client reset", "[sync][pbs][client reset][baas]") { SharedRealm realm = Realm::get_shared_realm(std::move(realm_ref), util::Scheduler::make_default()); auto flag = has_reset_cycle_flag(realm); REQUIRE(bool(flag)); - REQUIRE(flag->type == ClientResyncMode::Recover); + REQUIRE(flag->mode == ClientResyncMode::Recover); REQUIRE(did_recover); std::lock_guard lock(mtx); ++after_callback_invocations; @@ -1723,7 +1729,7 @@ TEST_CASE("sync: client reset", "[sync][pbs][client reset][baas]") { auto realm = Realm::get_shared_realm(local_config); auto flag = has_reset_cycle_flag(realm); REQUIRE(flag); - CHECK(flag->type == ClientResyncMode::Recover); + CHECK(flag->mode == ClientResyncMode::Recover); } SECTION("In DiscardLocal mode: a previous failed discard reset is detected and generates an error") { diff --git a/test/object-store/util/sync/sync_test_utils.cpp b/test/object-store/util/sync/sync_test_utils.cpp index f65ea2e4302..1d1adbaddf5 100644 --- a/test/object-store/util/sync/sync_test_utils.cpp +++ b/test/object-store/util/sync/sync_test_utils.cpp @@ -18,6 +18,7 @@ #include +#include #include #include @@ -28,6 +29,8 @@ #include #include +#include +#include #include #include @@ -208,7 +211,7 @@ std::string get_compile_time_base_url() return unquote_string(REALM_QUOTE(REALM_MONGODB_ENDPOINT)); #else return {}; -#endif +#endif // REALM_MONGODB_ENDPOINT } std::string get_compile_time_admin_url() @@ -218,10 +221,9 @@ std::string get_compile_time_admin_url() return unquote_string(REALM_QUOTE(REALM_ADMIN_ENDPOINT)); #else return {}; -#endif +#endif // REALM_ADMIN_ENDPOINT } #endif // REALM_ENABLE_AUTH_TESTS -#endif // REALM_ENABLE_SYNC #if REALM_APP_SERVICES AutoVerifiedEmailCredentials::AutoVerifiedEmailCredentials() @@ -307,6 +309,8 @@ void async_open_realm(const Realm::Config& config, finish(std::move(tsr), err); } +#endif // REALM_ENABLE_SYNC + class TestHelper { public: static DBRef& get_db(SharedRealm const& shared_realm) @@ -424,13 +428,13 @@ struct FakeLocalClientReset : public TestClientReset { sync::SaltedFileIdent fake_ident{1, 123456789}; auto local_db = TestHelper::get_db(local_realm); - auto remote_db = TestHelper::get_db(remote_realm); auto logger = util::Logger::get_default_logger(); + sync::ClientReset reset_config{m_mode, + TestHelper::get_db(remote_realm), + {ErrorCodes::SyncClientResetRequired, "Bad client file ident"}}; using _impl::client_reset::perform_client_reset_diff; - constexpr bool recovery_is_allowed = true; - perform_client_reset_diff(*local_db, *remote_db, fake_ident, *logger, m_mode, recovery_is_allowed, - nullptr, [](int64_t) {}); + perform_client_reset_diff(*local_db, reset_config, fake_ident, *logger, nullptr, [](int64_t) {}); remote_realm->close(); if (m_on_post_reset) { diff --git a/test/test_client_reset.cpp b/test/test_client_reset.cpp index 846ec976115..3a7ecac0b65 100644 --- a/test/test_client_reset.cpp +++ b/test/test_client_reset.cpp @@ -3,6 +3,7 @@ #include #include #include +#include #include #include #include @@ -184,10 +185,11 @@ TEST(ClientReset_NoLocalChanges) Session::Config session_config; { - Session::Config::ClientReset client_reset_config; - client_reset_config.mode = ClientResyncMode::DiscardLocal; - client_reset_config.fresh_copy = std::move(sg_fresh); - session_config.client_reset_config = std::move(client_reset_config); + Session::Config::ClientReset cr_config{ + ClientResyncMode::DiscardLocal, + sg_fresh, + {ErrorCodes::SyncClientResetRequired, "Bad client file identifier (IDENT)"}}; + session_config.client_reset_config = std::move(cr_config); } Session session = fixture.make_session(sg, server_path, std::move(session_config)); session.wait_for_download_complete_or_client_stopped(); @@ -250,12 +252,11 @@ TEST(ClientReset_InitialLocalChanges) // Start a client reset. There is no need for a reset, but we can do it. Session::Config session_config_2; - { - Session::Config::ClientReset client_reset_config; - client_reset_config.mode = ClientResyncMode::DiscardLocal; - client_reset_config.fresh_copy = std::move(sg_fresh); - session_config_2.client_reset_config = std::move(client_reset_config); - } + Session::Config::ClientReset cr_config{ + ClientResyncMode::DiscardLocal, + sg_fresh, + {ErrorCodes::SyncClientResetRequired, "Bad client file identifier (IDENT)"}}; + session_config_2.client_reset_config = std::move(cr_config); Session session_2 = fixture.make_session(db_2, server_path, std::move(session_config_2)); session_2.wait_for_upload_complete_or_client_stopped(); session_2.wait_for_download_complete_or_client_stopped(); @@ -363,10 +364,14 @@ TEST_TYPES(ClientReset_LocalChangesWhenOffline, std::true_type, std::false_type) } DBRef sg_fresh1 = DB::create(make_client_replication(), path_fresh1); + Session::Config::ClientReset cr_config{ + recover ? ClientResyncMode::Recover : ClientResyncMode::DiscardLocal, + sg_fresh1, + {ErrorCodes::SyncClientResetRequired, "Bad client file identifier (IDENT)"}, + recover ? sync::ProtocolErrorInfo::Action::ClientReset + : sync::ProtocolErrorInfo::Action::ClientResetNoRecovery}; Session::Config session_config_3; - session_config_3.client_reset_config = Session::Config::ClientReset{}; - session_config_3.client_reset_config->mode = recover ? ClientResyncMode::Recover : ClientResyncMode::DiscardLocal; - session_config_3.client_reset_config->fresh_copy = std::move(sg_fresh1); + session_config_3.client_reset_config = std::move(cr_config); Session session_3 = fixture.make_session(sg, server_path, std::move(session_config_3)); session_3.wait_for_upload_complete_or_client_stopped(); session_3.wait_for_download_complete_or_client_stopped(); @@ -605,17 +610,19 @@ TEST(ClientReset_ThreeClients) { Session::Config session_config_1; { - Session::Config::ClientReset client_reset_config; - client_reset_config.mode = ClientResyncMode::DiscardLocal; - client_reset_config.fresh_copy = std::move(sg_fresh1); - session_config_1.client_reset_config = std::move(client_reset_config); + Session::Config::ClientReset cr_config{ + ClientResyncMode::DiscardLocal, + sg_fresh1, + {ErrorCodes::SyncClientResetRequired, "Bad client file identifier (IDENT)"}}; + session_config_1.client_reset_config = std::move(cr_config); } Session::Config session_config_2; { - Session::Config::ClientReset client_reset_config; - client_reset_config.mode = ClientResyncMode::DiscardLocal; - client_reset_config.fresh_copy = std::move(sg_fresh2); - session_config_2.client_reset_config = std::move(client_reset_config); + Session::Config::ClientReset cr_config{ + ClientResyncMode::DiscardLocal, + sg_fresh2, + {ErrorCodes::SyncClientResetRequired, "Bad client file identifier (IDENT)"}}; + session_config_2.client_reset_config = std::move(cr_config); } Session session_1 = fixture.make_session(path_1, server_path, std::move(session_config_1)); Session session_2 = fixture.make_session(path_2, server_path, std::move(session_config_2)); @@ -724,10 +731,12 @@ TEST(ClientReset_DoNotRecoverSchema) { Session::Config session_config; { - Session::Config::ClientReset client_reset_config; - client_reset_config.mode = ClientResyncMode::DiscardLocal; - client_reset_config.fresh_copy = std::move(sg_fresh1); - session_config.client_reset_config = std::move(client_reset_config); + Session::Config::ClientReset cr_config{ + ClientResyncMode::DiscardLocal, + sg_fresh1, + {ErrorCodes::SyncClientResetRequired, "Bad client file identifier (IDENT)"}, + sync::ProtocolErrorInfo::Action::ClientReset}; + session_config.client_reset_config = std::move(cr_config); } BowlOfStonesSemaphore bowl; @@ -752,9 +761,10 @@ TEST(ClientReset_DoNotRecoverSchema) CHECK(!compare_groups(rt_1, rt_2)); const Group& group = rt_1.get_group(); - CHECK_EQUAL(group.size(), 2); + CHECK_EQUAL(group.size(), 3); CHECK(group.get_table("class_table1")); CHECK(group.get_table("client_reset_metadata")); + CHECK(group.get_table("sync_internal_schemas")); CHECK_NOT(group.get_table("class_table2")); const Group& group2 = rt_2.get_group(); CHECK_EQUAL(group2.size(), 1); @@ -810,9 +820,11 @@ TEST(ClientReset_PinnedVersion) Session::Config session_config; { - session_config.client_reset_config = Session::Config::ClientReset{}; - session_config.client_reset_config->mode = ClientResyncMode::DiscardLocal; - session_config.client_reset_config->fresh_copy = std::move(sg_fresh); + Session::Config::ClientReset cr_config{ + ClientResyncMode::DiscardLocal, + sg_fresh, + {ErrorCodes::SyncClientResetRequired, "Bad client file identifier (IDENT)"}}; + session_config.client_reset_config = std::move(cr_config); } Session session = fixture.make_bound_session(sg, server_path_1, std::move(session_config)); @@ -836,62 +848,62 @@ void mark_as_synchronized(DB& db) history.set_client_file_ident({1, 0}, false); } -void expect_reset(unit_test::TestContext& test_context, DB& target, DB& fresh, ClientResyncMode mode, - bool allow_recovery = true) +void expect_reset(unit_test::TestContext& test_context, DBRef& target, DBRef& fresh, ClientResyncMode mode, + SubscriptionStore* sub_store = nullptr, bool allow_recovery = true) { - auto db_version = target.get_version_of_latest_snapshot(); - auto fresh_path = fresh.get_path(); - bool did_reset = _impl::client_reset::perform_client_reset( - *test_context.logger, target, fresh, mode, nullptr, nullptr, {100, 200}, nullptr, [](int64_t) {}, - allow_recovery); - CHECK(did_reset); - - // Should have closed and deleted the fresh realm - CHECK_NOT(fresh.is_attached()); - CHECK_NOT(util::File::exists(fresh_path)); - - // Should have performed exactly two writes on the target DB: one to track - // that we're attempting recovery, and one with the actual reset - CHECK_EQUAL(target.get_version_of_latest_snapshot(), db_version + 2); - - // Should have set the client file ident - CHECK_EQUAL(target.start_read()->get_sync_file_id(), 100); - - // Client resets aren't marked as complete until the server has acknowledged - // sync completion to avoid reset cycles + CHECK(target); + CHECK(fresh); + // Ensure the schema is initialized before starting the test { - auto wt = target.start_write(); - _impl::client_reset::remove_pending_client_resets(*wt); - wt->commit(); - } -} - -void expect_reset(unit_test::TestContext& test_context, DB& target, DB& fresh, ClientResyncMode mode, - SubscriptionStore* sub_store) -{ - auto db_version = target.get_version_of_latest_snapshot(); - auto fresh_path = fresh.get_path(); - bool did_reset = _impl::client_reset::perform_client_reset( - *test_context.logger, target, fresh, mode, nullptr, nullptr, {100, 200}, sub_store, [](int64_t) {}, true); + auto wr_tr = target->start_write(); + PendingResetStore::clear_pending_reset(wr_tr); + wr_tr->commit(); + } + + auto db_version = target->get_version_of_latest_snapshot(); + auto fresh_path = fresh->get_path(); + Status error{ErrorCodes::SyncClientResetRequired, "Bad client file identifier (IDENT)"}; + auto action = allow_recovery ? sync::ProtocolErrorInfo::Action::ClientReset + : sync::ProtocolErrorInfo::Action::ClientResetNoRecovery; + // Pending reset store doesn't save RecoverOrDiscard + auto expected_mode = [](ClientResyncMode mode, bool allow_recovery) { + if (mode != ClientResyncMode::RecoverOrDiscard) + return mode; + if (allow_recovery) + return ClientResyncMode::Recover; + return ClientResyncMode::DiscardLocal; + }(mode, allow_recovery); + + sync::ClientReset cr_config{mode, fresh, error, action}; + + bool did_reset = _impl::client_reset::perform_client_reset(*test_context.logger, *target, std::move(cr_config), + {100, 200}, sub_store, [](int64_t) {}); CHECK(did_reset); // Should have closed and deleted the fresh realm - CHECK_NOT(fresh.is_attached()); + CHECK_NOT(fresh->is_attached()); CHECK_NOT(util::File::exists(fresh_path)); // Should have performed exactly two writes on the target DB: one to track // that we're attempting recovery, and one with the actual reset - CHECK_EQUAL(target.get_version_of_latest_snapshot(), db_version + 2); + CHECK_EQUAL(target->get_version_of_latest_snapshot(), db_version + 2); // Should have set the client file ident - CHECK_EQUAL(target.start_read()->get_sync_file_id(), 100); + CHECK_EQUAL(target->start_read()->get_sync_file_id(), 100); // Client resets aren't marked as complete until the server has acknowledged // sync completion to avoid reset cycles { - auto wt = target.start_write(); - _impl::client_reset::remove_pending_client_resets(*wt); - wt->commit(); + auto tr = target->start_read(); + auto pending_reset = PendingResetStore::has_pending_reset(tr); + CHECK(pending_reset); + CHECK(pending_reset->action == action); + CHECK(pending_reset->mode == expected_mode); + CHECK(pending_reset->error == error); + tr->promote_to_write(); + PendingResetStore::clear_pending_reset(tr); + tr->commit_and_continue_as_read(); + CHECK_NOT(PendingResetStore::has_pending_reset(tr)); } } @@ -910,6 +922,140 @@ std::pair prepare_db(const std::string& path, const std::string& c return {db, db_2}; } +TEST(ClientReset_ConvertResyncMode) +{ + CHECK(PendingResetStore::to_resync_mode(0) == ClientResyncMode::DiscardLocal); + CHECK(PendingResetStore::to_resync_mode(1) == ClientResyncMode::Recover); + CHECK_THROW(PendingResetStore::to_resync_mode(2), sync::ClientResetFailed); + + CHECK(PendingResetStore::from_resync_mode(ClientResyncMode::DiscardLocal) == 0); + CHECK(PendingResetStore::from_resync_mode(ClientResyncMode::RecoverOrDiscard) == 1); + CHECK(PendingResetStore::from_resync_mode(ClientResyncMode::Recover) == 1); + CHECK_THROW(PendingResetStore::from_resync_mode(ClientResyncMode::Manual), sync::ClientResetFailed); +} + +TEST(ClientReset_ConvertResetAction) +{ + CHECK(PendingResetStore::to_reset_action(0) == sync::ProtocolErrorInfo::Action::NoAction); + CHECK(PendingResetStore::to_reset_action(1) == sync::ProtocolErrorInfo::Action::ClientReset); + CHECK(PendingResetStore::to_reset_action(2) == sync::ProtocolErrorInfo::Action::ClientResetNoRecovery); + CHECK(PendingResetStore::to_reset_action(3) == sync::ProtocolErrorInfo::Action::MigrateToFLX); + CHECK(PendingResetStore::to_reset_action(4) == sync::ProtocolErrorInfo::Action::RevertToPBS); + CHECK(PendingResetStore::to_reset_action(5) == sync::ProtocolErrorInfo::Action::NoAction); + + CHECK(PendingResetStore::from_reset_action(sync::ProtocolErrorInfo::Action::ClientReset) == 1); + CHECK(PendingResetStore::from_reset_action(sync::ProtocolErrorInfo::Action::ClientResetNoRecovery) == 2); + CHECK(PendingResetStore::from_reset_action(sync::ProtocolErrorInfo::Action::MigrateToFLX) == 3); + CHECK(PendingResetStore::from_reset_action(sync::ProtocolErrorInfo::Action::RevertToPBS) == 4); + CHECK_THROW(PendingResetStore::from_reset_action(sync::ProtocolErrorInfo::Action::MigrateSchema), + sync::ClientResetFailed); +} + +DBRef setup_metadata_table_v1(test_util::unit_test::TestContext& test_context, std::string path, Timestamp ts, + int64_t type) +{ + DBRef db = DB::create(make_client_replication(), path); + auto wt = db->start_write(); + auto table = wt->add_table_with_primary_key("client_reset_metadata", type_ObjectId, "id"); + CHECK(table); + auto version_col = table->add_column(type_Int, "version"); + auto timestamp_col = table->add_column(type_Timestamp, "event_time"); + auto type_col = table->add_column(type_Int, "type_of_reset"); + wt->commit_and_continue_writing(); + auto id = ObjectId::gen(); + table->create_object_with_primary_key(id, { + {version_col, 1}, + {timestamp_col, ts}, + {type_col, type}, + }); + wt->commit_and_continue_as_read(); + table = wt->get_table("client_reset_metadata"); + size_t table_size = table->size(); + CHECK(table_size == 1); + return db; +} + +TEST_TYPES(ClientReset_V1Table, std::integral_constant, + std::integral_constant) +{ + SHARED_GROUP_TEST_PATH(path_v1); + auto timestamp = Timestamp(std::chrono::system_clock::now()); + auto reset_type = PendingResetStore::from_resync_mode(TEST_TYPE::value); + DBRef db = setup_metadata_table_v1(test_context, path_v1, timestamp, reset_type); + auto rd_tr = db->start_read(); + auto reset = PendingResetStore::has_pending_reset(rd_tr); + CHECK(reset); + CHECK(reset->time == timestamp); + CHECK(reset->mode == TEST_TYPE::value); + if (TEST_TYPE::value == ClientResyncMode::DiscardLocal) { + CHECK(reset->action == sync::ProtocolErrorInfo::Action::ClientResetNoRecovery); + } + else { + CHECK(reset->action == sync::ProtocolErrorInfo::Action::ClientReset); + } +} + +TEST(ClientReset_TrackReset_V1_EntryExists) +{ + SHARED_GROUP_TEST_PATH(path_v1); + auto timestamp = Timestamp(std::chrono::system_clock::now()); + auto reset_type = PendingResetStore::from_resync_mode(ClientResyncMode::Recover); + // Create a previous v1 entry + DBRef db = setup_metadata_table_v1(test_context, path_v1, timestamp, reset_type); + auto wr_tr = db->start_write(); + // Should throw an exception, since the table isn't empty + CHECK_THROW(PendingResetStore::track_reset(wr_tr, ClientResyncMode::DiscardLocal, + sync::ProtocolErrorInfo::Action::RevertToPBS), + sync::ClientResetFailed); +} + +TEST(ClientReset_TrackReset_Existing_empty_V1_table) +{ + SHARED_GROUP_TEST_PATH(path_v1); + auto timestamp = Timestamp(std::chrono::system_clock::now()); + auto reset_type = PendingResetStore::from_resync_mode(ClientResyncMode::Recover); + Status error{ErrorCodes::SyncClientResetRequired, "Bad client file ident"}; + DBRef db = setup_metadata_table_v1(test_context, path_v1, timestamp, reset_type); + auto wr_tr = db->start_write(); + PendingResetStore::clear_pending_reset(wr_tr); + wr_tr->commit_and_continue_writing(); + PendingResetStore::track_reset(wr_tr, ClientResyncMode::DiscardLocal, + sync::ProtocolErrorInfo::Action::RevertToPBS, error); + wr_tr->commit_and_continue_as_read(); + auto reset = PendingResetStore::has_pending_reset(wr_tr); + CHECK(reset); + CHECK(reset->mode == ClientResyncMode::DiscardLocal); + CHECK(reset->action == sync::ProtocolErrorInfo::Action::RevertToPBS); + CHECK(reset->error == error); + timestamp = Timestamp(std::chrono::system_clock::now()); + // Verify timestamp is at least close to current time + CHECK(abs(reset->time.get_seconds() - timestamp.get_seconds()) < 5); +} + +TEST_TYPES( + ClientReset_TrackReset_v2, + std::integral_constant, + std::integral_constant, + std::integral_constant, + std::integral_constant) +{ + SHARED_GROUP_TEST_PATH(test_path); + DBRef db = DB::create(make_client_replication(), test_path); + Status error{ErrorCodes::SyncClientResetRequired, "Bad client file ident"}; + sync::ProtocolErrorInfo::Action reset_action = TEST_TYPE::value; + auto tr = db->start_write(); + PendingResetStore::track_reset(tr, ClientResyncMode::DiscardLocal, reset_action, error); + tr->commit_and_continue_as_read(); + auto reset = PendingResetStore::has_pending_reset(tr); + CHECK(reset); + CHECK(reset->mode == ClientResyncMode::DiscardLocal); + CHECK(reset->action == reset_action); + CHECK(reset->error == error); + auto timestamp = Timestamp(std::chrono::system_clock::now()); + // Verify timestamp is at least close to current time + CHECK((reset->time.get_seconds() - timestamp.get_seconds() < 5)); +} + TEST(ClientReset_UninitializedFile) { SHARED_GROUP_TEST_PATH(path_1); @@ -921,12 +1067,17 @@ TEST(ClientReset_UninitializedFile) }); auto db_empty = DB::create(make_client_replication(), path_3); + sync::ClientReset cr_config{ClientResyncMode::Recover, + db_fresh, + {ErrorCodes::SyncClientResetRequired, "Bad client file identifier (IDENT)"}}; + // Should not perform a client reset because the target file has never been // written to - bool did_reset = _impl::client_reset::perform_client_reset( - *test_context.logger, *db_empty, *db_fresh, ClientResyncMode::Recover, nullptr, nullptr, {100, 200}, nullptr, - [](int64_t) {}, true); + bool did_reset = _impl::client_reset::perform_client_reset(*test_context.logger, *db_empty, std::move(cr_config), + {100, 200}, nullptr, [](int64_t) {}); CHECK_NOT(did_reset); + auto rd_tr = db_empty->start_frozen(); + CHECK_NOT(PendingResetStore::has_pending_reset(rd_tr)); // Should still have closed and deleted the fresh realm CHECK_NOT(db_fresh->is_attached()); @@ -963,7 +1114,8 @@ TEST(ClientReset_NoChanges) // Perform a reset with a fresh Realm that exactly matches the current // one, which shouldn't result in any changes regardless of mode db->write_copy(path_fresh, nullptr); - expect_reset(test_context, *db, *DB::create(make_client_replication(), path_fresh), mode); + auto db_fresh = DB::create(make_client_replication(), path_fresh); + expect_reset(test_context, db, db_fresh, mode); // End state should exactly match the pre-reset state CHECK_OR_RETURN(compare_groups(*db->start_read(), *backup_db->start_read())); @@ -1005,7 +1157,7 @@ TEST(ClientReset_SimpleNonconflictingChanges) wt->commit(); } - expect_reset(test_context, *db, *db_fresh, mode, allow_recovery); + expect_reset(test_context, db, db_fresh, mode, nullptr, allow_recovery); if (allow_recovery) { // Should have both the objects created locally and from the reset realm @@ -1062,7 +1214,7 @@ TEST(ClientReset_SimpleConflictingWrites) wt->commit(); } - expect_reset(test_context, *db, *db_fresh, mode, allow_recovery); + expect_reset(test_context, db, db_fresh, mode, nullptr, allow_recovery); auto tr = db->start_read(); auto table = tr->get_table("class_table"); @@ -1088,11 +1240,16 @@ TEST(ClientReset_Recover_RecoveryDisabled) auto dbs = prepare_db(path_1, path_2, [](Transaction& tr) { tr.add_table_with_primary_key("class_table", type_Int, "pk"); }); - CHECK_THROW((_impl::client_reset::perform_client_reset( - *test_context.logger, *dbs.first, *dbs.second, ClientResyncMode::Recover, nullptr, nullptr, - {100, 200}, nullptr, [](int64_t) {}, false)), - _impl::client_reset::ClientResetFailed); - CHECK_NOT(_impl::client_reset::has_pending_reset(*dbs.first->start_read())); + sync::ClientReset cr_config{ClientResyncMode::Recover, + dbs.second, + {ErrorCodes::SyncClientResetRequired, "Bad client file identifier (IDENT)"}, + sync::ProtocolErrorInfo::Action::ClientResetNoRecovery}; + + CHECK_THROW((_impl::client_reset::perform_client_reset(*test_context.logger, *dbs.first, std::move(cr_config), + {100, 200}, nullptr, [](int64_t) {})), + sync::ClientResetFailed); + auto rd_tr = dbs.first->start_frozen(); + CHECK_NOT(PendingResetStore::has_pending_reset(rd_tr)); } TEST(ClientReset_Recover_ModificationsOnDeletedObject) @@ -1125,7 +1282,7 @@ TEST(ClientReset_Recover_ModificationsOnDeletedObject) wt->commit(); } - expect_reset(test_context, *db, *db_fresh, ClientResyncMode::Recover); + expect_reset(test_context, db, db_fresh, ClientResyncMode::Recover); auto tr = db->start_read(); auto table = tr->get_table("class_table"); @@ -1165,7 +1322,7 @@ TEST(ClientReset_DiscardLocal_DiscardsPendingSubscriptions) pending_sets.push_back(std::move(set)); } - expect_reset(test_context, *db, *db_fresh, ClientResyncMode::DiscardLocal, sub_store.get()); + expect_reset(test_context, db, db_fresh, ClientResyncMode::DiscardLocal, sub_store.get()); CHECK(sub_store->get_pending_subscriptions().empty()); auto subs = sub_store->get_latest(); @@ -1199,7 +1356,7 @@ TEST_TYPES(ClientReset_DiscardLocal_MakesAwaitingMarkActiveSubscriptionsComplete auto set = add_subscription(*sub_store, "complete", query, SubscriptionSet::State::AwaitingMark); auto future = set.get_state_change_notification(SubscriptionSet::State::Complete); - expect_reset(test_context, *db, *db_fresh, TEST_TYPE::value, sub_store.get()); + expect_reset(test_context, db, db_fresh, TEST_TYPE::value, sub_store.get()); CHECK_EQUAL(future.get(), SubscriptionSet::State::Complete); CHECK_EQUAL(set.state(), SubscriptionSet::State::AwaitingMark); @@ -1227,7 +1384,7 @@ TEST(ClientReset_Recover_DoesNotCompletePendingSubscriptions) futures.push_back(subs.get_state_change_notification(SubscriptionSet::State::Complete)); } - expect_reset(test_context, *db, *db_fresh, ClientResyncMode::Recover, sub_store.get()); + expect_reset(test_context, db, db_fresh, ClientResyncMode::Recover, sub_store.get()); for (auto& fut : futures) { CHECK_NOT(fut.is_ready()); @@ -1278,7 +1435,7 @@ TEST(ClientReset_Recover_UpdatesRemoteServerVersions) history.set_sync_progress(progress, nullptr, info_out); } - expect_reset(test_context, *db, *db_fresh, ClientResyncMode::Recover, nullptr); + expect_reset(test_context, db, db_fresh, ClientResyncMode::Recover, nullptr); auto& history = static_cast(db->get_replication())->get_history(); history.ensure_updated(db->get_version_of_latest_snapshot()); @@ -1338,7 +1495,7 @@ TEST(ClientReset_Recover_UploadableBytes) history.get_upload_download_bytes(db.get(), unused, unused, unused, pre_reset_uploadable_bytes, unused); CHECK_GREATER(pre_reset_uploadable_bytes, 0); - expect_reset(test_context, *db, *db_fresh, ClientResyncMode::Recover, nullptr); + expect_reset(test_context, db, db_fresh, ClientResyncMode::Recover, nullptr); uint_fast64_t post_reset_uploadable_bytes; history.get_upload_download_bytes(db.get(), unused, unused, unused, post_reset_uploadable_bytes, unused); @@ -1384,7 +1541,7 @@ TEST(ClientReset_Recover_ListsAreOnlyCopiedOnce) wt->commit(); } - expect_reset(test_context, *db, *db_fresh, ClientResyncMode::Recover, nullptr); + expect_reset(test_context, db, db_fresh, ClientResyncMode::Recover, nullptr); // List should match the pre-reset local state auto rt = db->start_read(); @@ -1441,7 +1598,7 @@ TEST(ClientReset_Recover_RecoverableChangesOnListsAfterUnrecoverableAreNotDuplic wt->commit(); } - expect_reset(test_context, *db, *db_fresh, ClientResyncMode::Recover, sub_store.get()); + expect_reset(test_context, db, db_fresh, ClientResyncMode::Recover, sub_store.get()); // List should match the pre-reset local state auto rt = db->start_read(); @@ -1537,7 +1694,7 @@ TEST(ClientReset_Recover_ReciprocalListChanges) // shouldn't modify the group. However, if it reapplied the original changesets // and not the reciprocal history, it'd result in the list being // [0, 1, 2, 11, 10, 21, 12, 31, 20, 41, 22, 30, 32, 40, 42] - expect_reset(test_context, *db, *db_fresh, ClientResyncMode::Recover, nullptr); + expect_reset(test_context, db, db_fresh, ClientResyncMode::Recover, nullptr); auto rt = db->start_read(); auto list = rt->get_table("class_table")->begin()->get_list("list"); @@ -1596,7 +1753,7 @@ TEST(ClientReset_Recover_UpdatesReciprocalHistory) // client reset will discard the recovered array insertion as the object // doesn't exist, but keep the object creation - expect_reset(test_context, *db, *db_fresh, ClientResyncMode::Recover, nullptr); + expect_reset(test_context, db, db_fresh, ClientResyncMode::Recover, nullptr); // Recreate the object and add a different value to the list { diff --git a/test/util/compare_groups.cpp b/test/util/compare_groups.cpp index 6ac669a268d..4b16cdb2fea 100644 --- a/test/util/compare_groups.cpp +++ b/test/util/compare_groups.cpp @@ -951,12 +951,16 @@ bool compare_groups(const Transaction& group_1, const Transaction& group_2) bool compare_groups(const Transaction& group_1, const Transaction& group_2, util::FunctionRef filter_func, util::Logger& logger) { + std::vector ignored_tables = {"pk", "metadata", "client_reset_metadata", "flx_metadata", + "sync_internal_schemas"}; + auto filter = [&](const Group& group, std::vector& tables) { auto table_keys = group.get_table_keys(); for (auto i : table_keys) { ConstTableRef table = group.get_table(i); StringData name = table->get_name(); - if (name != "pk" && name != "metadata" && name != "client_reset_metadata" && filter_func(name)) + if (std::find(ignored_tables.begin(), ignored_tables.end(), name) == ignored_tables.end() && + filter_func(name)) tables.push_back(name); } };