realm / realm-core / thomas.goyne_121

{

    REALM_ASSERT(m_state != State::Active);

    m_state = State::Active;

    // First time the session becomes active, register a notification on the sentinel subscription set to restart the

    // session and update to native FLX.

    if (m_migration_sentinel_query_version) {

        m_flx_subscription_store->get_by_version(*m_migration_sentinel_query_version)

            .get_state_change_notification(sync::SubscriptionSet::State::Complete)

            .get_async([=, weak_self = weak_from_this()](StatusWith<sync::SubscriptionSet::State> s) {

                if (!s.is_ok()) {

                REALM_ASSERT(s.get_value() == sync::SubscriptionSet::State::Complete);

                if (auto strong_self = weak_self.lock()) {

                    strong_self->m_migration_store->cancel_migration();

                    strong_self->restart_session();

                }

            });

        m_migration_sentinel_query_version.reset();

    }

    // when entering from the Dying state the session will still be bound

    if (!m_session) {

        create_sync_session();

        m_session->bind();

    }

    // Register all the pending wait-for-completion blocks. This can

    // potentially add a redundant callback if we're coming from the Dying

    // state, but that's okay (we won't call the user callbacks twice).

    SyncSession::CompletionCallbacks callbacks_to_register;

    std::swap(m_completion_callbacks, callbacks_to_register);

    for (auto& [id, callback_tuple] : callbacks_to_register) {

        add_completion_callback(std::move(callback_tuple.second), callback_tuple.first);

    }

}

{

    util::CheckedUniqueLock lock(m_state_mutex);

    do_restart_session(std::move(lock));

}

{

    REALM_ASSERT(m_state != State::Dying);

    m_state = State::Dying;

    // If we have no session, we cannot possibly upload anything.

    if (!m_session) {

    size_t current_death_count = ++m_death_count;

    m_session->async_wait_for_upload_completion([weak_session = weak_from_this(), current_death_count](Status) {

        if (auto session = weak_session.lock()) {

            util::CheckedUniqueLock lock(session->m_state_mutex);

            if (session->m_state == State::Dying && session->m_death_count == current_death_count) {

                session->become_inactive(std::move(lock));

            }

        }

    });

    m_state_mutex.unlock(lock);

}

{

    REALM_ASSERT(m_state != State::Inactive);

    m_state = State::Inactive;

    do_become_inactive(std::move(lock), status, cancel_subscription_notifications);

}

{

    REALM_ASSERT(m_state != State::Paused);

    auto old_state = m_state;

    m_state = State::Paused;

    // Nothing to do if we're already inactive besides update the state.

    if (old_state == State::Inactive) {

        m_state_mutex.unlock(lock);

        return;

    }

    do_become_inactive(std::move(lock), Status::OK(), true);

}

{

    // Nothing to do if the sync session is currently paused

    // It will be resumed when resume() is called

    if (m_state == State::Paused)

    // Go straight to inactive so the progress completion waiters will

    // continue to wait until the session restarts and completes the

    // upload/download sync

    m_state = State::Inactive;

    if (m_session) {

        m_session.reset();

    }

    // Create a new session and re-register the completion callbacks

    // The latest server path will be retrieved from sync_manager when

    // the new session is created by create_sync_session() in become

    // active.

    become_active();

}

{

    // Manually set the disconnected state. Sync would also do this, but

    // since the underlying SyncSession object already have been destroyed,

    // we are not able to get the callback.

    util::CheckedUniqueLock connection_state_lock(m_connection_state_mutex);

    auto old_state = m_connection_state;

    auto new_state = m_connection_state = SyncSession::ConnectionState::Disconnected;

    connection_state_lock.unlock();

    SyncSession::CompletionCallbacks waits;

    std::swap(waits, m_completion_callbacks);

    m_session = nullptr;

    if (m_sync_manager) {

        m_sync_manager->unregister_session(m_db->get_path());

    }

    auto subscription_store = m_flx_subscription_store;

    m_state_mutex.unlock(lock);

    // Send notifications after releasing the lock to prevent deadlocks in the callback.

    if (old_state != new_state) {

        m_connection_change_notifier.invoke_callbacks(old_state, connection_state());

    }

    if (status.is_ok())

        status = Status(ErrorCodes::OperationAborted, "Sync session became inactive");

    if (subscription_store && cancel_subscription_notifications) {

        subscription_store->notify_all_state_change_notifications(status);

    }

    // Inform any queued-up completion handlers that they were cancelled.

    for (auto& [id, callback] : waits)

        callback.second(status);

}

{

    REALM_ASSERT(m_state != State::WaitingForAccessToken);

    m_state = State::WaitingForAccessToken;

}

{

    // TODO: ideally this would write to the logs as well in case users didn't set up their error handler.

    {

        util::CheckedUniqueLock lock(m_state_mutex);

        cancel_pending_waits(std::move(lock), status);

    }

    if (user) {

        user->log_out();

    }

    if (auto error_handler = config(&SyncConfig::error_handler)) {

        auto user_facing_error = SyncError({ErrorCodes::AuthError, status.reason()}, true);

        error_handler(shared_from_this(), std::move(user_facing_error));

    }

}

{

    using namespace realm::sync;

    // Auth failure is returned as a 401 (unauthorized) or 403 (forbidden) response

    if (error.additional_status_code) {

        auto status_code = HTTPStatus(*error.additional_status_code);

        if (status_code == HTTPStatus::Unauthorized || status_code == HTTPStatus::Forbidden)

            return true;

    }

    return false;

}

{

    using namespace realm::sync;

    // Check for unhandled 301/308 permanent redirect response

    if (error.additional_status_code) {

        auto status_code = HTTPStatus(*error.additional_status_code);

        if (status_code == HTTPStatus::MovedPermanently || status_code == HTTPStatus::PermanentRedirect)

    }

    return false;

}

{

    return [session, restart_session](util::Optional<app::AppError> error) {

        auto session_user = session->user();

        if (!session_user) {

        else if (error) {

            if (error->code() == ErrorCodes::ClientAppDeallocated) {

            else if (ErrorCodes::error_categories(error->code()).test(ErrorCategory::client_error)) {

                // any other client errors other than app_deallocated are considered fatal because

                // there was a problem locally before even sending the request to the server

                // eg. ClientErrorCode::user_not_found, ClientErrorCode::user_not_logged_in,

                // ClientErrorCode::too_many_redirects

                session->handle_bad_auth(session_user, error->to_status());

            }

            else if (check_for_auth_failure(*error)) {

                // A 401 response on a refresh request means that the token cannot be refreshed and we should not

                // retry. This can be because an admin has revoked this user's sessions, the user has been disabled,

                // or the refresh token has expired according to the server's clock.

                session->handle_bad_auth(

                    session_user,

                    {error->code(), util::format("Unable to refresh the user access token: %1", error->reason())});

            }

            else if (check_for_redirect_response(*error)) {

            else {

                // A refresh request has failed. This is an unexpected non-fatal error and we would

                // like to retry but we shouldn't do this immediately in order to not swamp the

                // server with requests. Consider two scenarios:

                // 1) If this request was spawned from the proactive token check, or a user

                // initiated request, the token may actually be valid. Just advance to Active

                // from WaitingForAccessToken if needed and let the sync server tell us if the

                // token is valid or not. If this also fails we will end up in case 2 below.

                // 2) If the sync connection initiated the request because the server is

                // unavailable or the connection otherwise encounters an unexpected error, we want

                // to let the sync client attempt to reinitialize the connection using its own

                // internal backoff timer which will happen automatically so nothing needs to

                // happen here.

                util::CheckedUniqueLock lock(session->m_state_mutex);

                if (session->m_state == State::WaitingForAccessToken) {

                    session->become_active();

                }

            }

        }

        else {

            // If the session needs to be restarted, then restart the session now

            // The latest access token and server url will be pulled from the sync

            // manager when the new session is started.

            if (restart_session) {

                session->restart_session();

            }

            // Otherwise, update the access token and reconnect

            else {

                session->update_access_token(session_user->access_token());

            }

        }

    };

}

{

    REALM_ASSERT(m_config.sync_config);

    // we don't want the following configs enabled during a client reset

    m_config.scheduler = nullptr;

    m_config.audit_config = nullptr;

    // Adjust the sync_config if using PBS sync and already in the migrated or rollback state

    if (m_migration_store->is_migrated() || m_migration_store->is_rollback_in_progress()) {

        m_config.sync_config = sync::MigrationStore::convert_sync_config_to_flx(m_original_sync_config);

    }

    // If using FLX, set up m_flx_subscription_store and the history_write_validator

    if (m_config.sync_config->flx_sync_requested) {

        create_subscription_store();

        std::weak_ptr<sync::SubscriptionStore> weak_sub_mgr(m_flx_subscription_store);

        set_write_validator_factory(weak_sub_mgr);

    }

    // After a migration to FLX, if the user opens the realm with a flexible sync configuration, we need to first

    // upload any unsynced changes before updating to native FLX.

    // A subscription set is used as sentinel so we know when to stop uploading.

    // Note: Currently, a sentinel subscription set is always created even if there is nothing to upload.

    if (m_migration_store->is_migrated() && m_original_sync_config->flx_sync_requested) {

        m_migration_store->create_sentinel_subscription_set(*m_flx_subscription_store);

        m_migration_sentinel_query_version = m_migration_store->get_sentinel_subscription_set_version();

        REALM_ASSERT(m_migration_sentinel_query_version);

    }

}

{

    shutdown_and_wait();

    util::CheckedLockGuard lk(m_state_mutex);

    m_sync_manager = nullptr;

}

{

    util::CheckedLockGuard config_lock(m_config_mutex);

    // Add a SyncFileActionMetadata marking the Realm as needing to be deleted.

    std::string recovery_path;

    auto original_path = path();

    error.user_info[SyncError::c_original_file_path_key] = original_path;

    if (should_backup == ShouldBackup::yes) {

        recovery_path = util::reserve_unique_file_name(

            m_sync_manager->recovery_directory_path(m_config.sync_config->recovery_directory),

            util::create_timestamped_template("recovered_realm"));

        error.user_info[SyncError::c_recovery_file_path_key] = recovery_path;

    }

    using Action = SyncFileActionMetadata::Action;

    auto action = should_backup == ShouldBackup::yes ? Action::BackUpThenDeleteRealm : Action::DeleteRealm;

    m_sync_manager->perform_metadata_update([action, original_path = std::move(original_path),

                                             recovery_path = std::move(recovery_path),

                                             partition_value = m_config.sync_config->partition_value,

                                             identity = m_config.sync_config->user->identity()](const auto& manager) {

        manager.make_file_action_metadata(original_path, partition_value, identity, action, recovery_path);

    });

}

{

    // first check that recovery will not be prevented

    if (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);

            return;

        }

    }

    std::vector<char> encryption_key;

    {

        util::CheckedLockGuard lock(m_config_mutex);

        encryption_key = m_config.encryption_key;

    }

    DBOptions options;

    options.allow_file_format_upgrade = false;

    options.enable_async_writes = false;

    if (!encryption_key.empty())

        options.encryption_key = encryption_key.data();

    DBRef db;

    auto fresh_path = client_reset::get_fresh_path_for(m_db->get_path());

    try {

        // We want to attempt to use a pre-existing file to reduce the chance of

        // downloading the first part of the file only to then delete it over

        // and over, but if we fail to open it then we should just start over.

        try {

            db = DB::create(sync::make_client_replication(), fresh_path, options);

        }

        catch (...) {

            util::File::try_remove(fresh_path);

        }

        if (!db) {

            db = DB::create(sync::make_client_replication(), fresh_path, options);

        }

    }

    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);

        return;

    }

    util::CheckedLockGuard state_lock(m_state_mutex);

    if (m_state != State::Active) {

    std::shared_ptr<SyncSession> fresh_sync_session;

    {

        util::CheckedLockGuard config_lock(m_config_mutex);

        RealmConfig config = m_config;

        config.path = fresh_path;

        // in case of migrations use the migrated config

        auto fresh_config = m_migrated_sync_config ? *m_migrated_sync_config : *m_config.sync_config;

        // deep copy the sync config so we don't modify the live session's config

        config.sync_config = std::make_shared<SyncConfig>(fresh_config);

        config.sync_config->client_resync_mode = ClientResyncMode::Manual;

        fresh_sync_session = m_sync_manager->get_session(db, config);

        auto& history = static_cast<sync::ClientReplication&>(*db->get_replication());

        // the fresh Realm may apply writes to this db after it has outlived its sync session

        // the writes are used to generate a changeset for recovery, but are never committed

        history.set_write_validator_factory({});

    }

    fresh_sync_session->assert_mutex_unlocked();

    // The fresh realm uses flexible sync.

    if (auto fresh_sub_store = fresh_sync_session->get_flx_subscription_store()) {

        auto fresh_sub = fresh_sub_store->get_latest();

        // The local realm uses flexible sync as well so copy the active subscription set to the fresh realm.

        if (auto local_subs_store = m_flx_subscription_store) {

            sync::SubscriptionSet active = local_subs_store->get_active();

            auto fresh_mut_sub = fresh_sub.make_mutable_copy();

            fresh_mut_sub.import(active);

            fresh_sub = fresh_mut_sub.commit();

        }

        fresh_sub.get_state_change_notification(sync::SubscriptionSet::State::Complete)

            .then([=, weak_self = weak_from_this()](sync::SubscriptionSet::State state) {

                if (server_requests_action != sync::ProtocolErrorInfo::Action::MigrateToFLX) {

                    return util::Future<sync::SubscriptionSet::State>::make_ready(state);

                }

                auto strong_self = weak_self.lock();

                if (!strong_self || !strong_self->m_migration_store->is_migration_in_progress()) {

                    return util::Future<sync::SubscriptionSet::State>::make_ready(state);

                // fresh_sync_session is using a new realm file that doesn't have the migration_store info

                // so the query string from the local migration store will need to be provided

                auto query_string = strong_self->m_migration_store->get_query_string();

                REALM_ASSERT(query_string);

                // Create subscriptions in the fresh realm based on the schema instructions received in the bootstrap

                // message.

                fresh_sync_session->m_migration_store->create_subscriptions(*fresh_sub_store, *query_string);

                auto latest_subs = fresh_sub_store->get_latest();

                {

                    util::CheckedLockGuard lock(strong_self->m_state_mutex);

                    // Save a copy of the subscriptions so we add them to the local realm once the

                    // subscription store is created.

                    strong_self->m_active_subscriptions_after_migration = latest_subs;

                }

                return latest_subs.get_state_change_notification(sync::SubscriptionSet::State::Complete);

            })

            .get_async([=, weak_self = weak_from_this()](StatusWith<sync::SubscriptionSet::State> s) {

                // 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()) {

                    if (s.is_ok()) {

                        strong_self->handle_fresh_realm_downloaded(db, Status::OK(), server_requests_action);

                    }

                    else {

                        strong_self->handle_fresh_realm_downloaded(nullptr, s.get_status(), server_requests_action);

                    }

                }

            });

    }

    else { // pbs

        fresh_sync_session->wait_for_download_completion([=, weak_self = weak_from_this()](Status s) {

            // 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);

            }

        });

    }

    fresh_sync_session->revive_if_needed();

}

void SyncSession::handle_fresh_realm_downloaded(DBRef db, Status status,

                                                sync::ProtocolErrorInfo::Action server_requests_action)

{

    util::CheckedUniqueLock lock(m_state_mutex);

    if (m_state != State::Active) {

        return;

    }

    // The download can fail for many reasons. For example:

    // - 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) {

            return;

        }

        lock.unlock();

        sync::SessionErrorInfo synthetic(

            Status{ErrorCodes::AutoClientResetFailed,

                   util::format("A fatal error occurred during client reset: '%1'", status.reason())},

            sync::IsFatal{true});

        handle_error(synthetic);

        return;

    }

    // Performing a client reset requires tearing down our current

    // sync session and creating a new one with the relevant client reset config. This

    // will result in session completion handlers firing

    // when the old session is torn down, which we don't want as this

    // is supposed to be transparent to the user.

    //

    // To avoid this, we need to move the completion handlers aside temporarily so

    // 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;

        std::swap(m_completion_callbacks, callbacks);

        // always swap back, even if advance_state throws

        auto guard = util::make_scope_exit([&]() noexcept {

            util::CheckedUniqueLock lock(m_state_mutex);

            if (m_completion_callbacks.empty())

                std::swap(callbacks, m_completion_callbacks);

            else

                m_completion_callbacks.merge(std::move(callbacks));

        });

        // Do not cancel the notifications on subscriptions.

        bool cancel_subscription_notifications = false;

        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) {

            apply_sync_config_after_migration_or_rollback();

            auto flx_sync_requested = config(&SyncConfig::flx_sync_requested);

            update_subscription_store(flx_sync_requested);

        }

    }

    revive_if_needed();

}

void SyncSession::OnlyForTesting::handle_error(SyncSession& session, sync::SessionErrorInfo&& error)

{

    session.handle_error(std::move(error));

}

// This method should only be called from within the error handler callback registered upon the underlying

// `m_session`.

void SyncSession::handle_error(sync::SessionErrorInfo error)

{

    enum class NextStateAfterError { none, inactive, error };

    auto next_state = error.is_fatal ? NextStateAfterError::error : NextStateAfterError::none;

    util::Optional<ShouldBackup> delete_file;

    bool log_out_user = false;

    bool unrecognized_by_client = false;

    if (error.status == ErrorCodes::AutoClientResetFailed) {

        // At this point, automatic recovery has been attempted but it failed.

        // Fallback to a manual reset and let the user try to handle it.

        next_state = NextStateAfterError::inactive;

        delete_file = ShouldBackup::yes;

    }

    else if (error.server_requests_action != sync::ProtocolErrorInfo::Action::NoAction) {

        switch (error.server_requests_action) {

            case sync::ProtocolErrorInfo::Action::NoAction:

                REALM_UNREACHABLE(); // This is not sent by the MongoDB server

            case sync::ProtocolErrorInfo::Action::ApplicationBug:

                [[fallthrough]];

            case sync::ProtocolErrorInfo::Action::ProtocolViolation:

                break;

            case sync::ProtocolErrorInfo::Action::Warning:

                break; // not fatal, but should be bubbled up to the user below.

            case sync::ProtocolErrorInfo::Action::Transient:

                return;

                next_state = NextStateAfterError::inactive;

                delete_file = ShouldBackup::no;

                break;

            case sync::ProtocolErrorInfo::Action::ClientReset:

                [[fallthrough]];

            case sync::ProtocolErrorInfo::Action::ClientResetNoRecovery:

                switch (config(&SyncConfig::client_resync_mode)) {

                    case ClientResyncMode::Manual:

                        next_state = NextStateAfterError::inactive;

                        delete_file = ShouldBackup::yes;

                        break;

                    case ClientResyncMode::DiscardLocal:

                        [[fallthrough]];

                    case ClientResyncMode::RecoverOrDiscard:

                        [[fallthrough]];

                    case ClientResyncMode::Recover:

                        download_fresh_realm(error.server_requests_action);

                        return; // do not propgate the error to the user at this point

                }

                break;

            case sync::ProtocolErrorInfo::Action::MigrateToFLX:

                // Should not receive this error if original sync config is FLX

                REALM_ASSERT(!m_original_sync_config->flx_sync_requested);

                REALM_ASSERT(error.migration_query_string && !error.migration_query_string->empty());

                // Original config was PBS, migrating to FLX

                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);

                return;

            case sync::ProtocolErrorInfo::Action::RevertToPBS:

                // If the client was updated to use FLX natively, but the server was rolled back to PBS,

                if (m_original_sync_config->flx_sync_requested) {

                                     "Received 'RevertToPBS' from server after rollback while client is natively "

                                     "using FLX - expected 'SwitchToPBS'");

                }

                // 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);

                return;

            case sync::ProtocolErrorInfo::Action::RefreshUser:

                if (auto u = user()) {

                    u->refresh_custom_data(false, handle_refresh(shared_from_this(), false));

                    return;

                }

                break;

            case sync::ProtocolErrorInfo::Action::RefreshLocation:

                if (auto u = user()) {

                    u->refresh_custom_data(true, handle_refresh(shared_from_this(), true));

                    return;

                }

                next_state = NextStateAfterError::inactive;

                log_out_user = true;

                break;

        }

    }

    else {

        // Unrecognized error code.

        unrecognized_by_client = true;

    }

    util::CheckedUniqueLock lock(m_state_mutex);

    SyncError sync_error{error.status, error.is_fatal, error.log_url, std::move(error.compensating_writes)};

    // `action` is used over `shouldClientReset` and `isRecoveryModeDisabled`.

    sync_error.server_requests_action = error.server_requests_action;

    sync_error.is_unrecognized_by_client = unrecognized_by_client;

    if (delete_file)

        update_error_and_mark_file_for_deletion(sync_error, *delete_file);

    if (m_state == State::Dying && error.is_fatal) {

        become_inactive(std::move(lock), error.status);

        return;

    }

    if (m_state == State::Inactive || m_state == State::Paused) {

        return;

    }

    switch (next_state) {

        case NextStateAfterError::none:

            if (config(&SyncConfig::cancel_waits_on_nonfatal_error)) {

                cancel_pending_waits(std::move(lock), sync_error.status); // unlocks the mutex

            }

            break;

        case NextStateAfterError::inactive: {

            become_inactive(std::move(lock), sync_error.status);

            break;

        }

        case NextStateAfterError::error: {

            cancel_pending_waits(std::move(lock), sync_error.status);

            break;

        }

    }

    if (log_out_user) {

            u->log_out();

    }

    if (auto error_handler = config(&SyncConfig::error_handler)) {

        error_handler(shared_from_this(), std::move(sync_error));

    }

}

void SyncSession::cancel_pending_waits(util::CheckedUniqueLock lock, Status error)

{

    CompletionCallbacks callbacks;

    std::swap(callbacks, m_completion_callbacks);

    // Inform any waiters on pending subscription states that they were cancelled

    if (m_flx_subscription_store) {

        auto subscription_store = m_flx_subscription_store;

        m_state_mutex.unlock(lock);

        subscription_store->notify_all_state_change_notifications(error);

    }

    else {

        m_state_mutex.unlock(lock);

    }

    // Inform any queued-up completion handlers that they were cancelled.

    for (auto& [id, callback] : callbacks)

        callback.second(error);

}

void SyncSession::handle_progress_update(uint64_t downloaded, uint64_t downloadable, uint64_t uploaded,

                                         uint64_t uploadable, uint64_t download_version, uint64_t snapshot_version)

{

    m_progress_notifier.update(downloaded, downloadable, uploaded, uploadable, download_version, snapshot_version);

}

static sync::Session::Config::ClientReset make_client_reset_config(const RealmConfig& base_config,

                                                                   const std::shared_ptr<SyncConfig>& sync_config,

                                                                   DBRef&& fresh_copy, bool recovery_is_allowed)

{

    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;

    // 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

    // before the client reset happens.

    if (!sync_config->notify_before_client_reset && !sync_config->notify_after_client_reset)

        return config;

    RealmConfig realm_config = base_config;

    realm_config.sync_config = std::make_shared<SyncConfig>(*sync_config); // deep copy

    realm_config.scheduler = util::Scheduler::make_dummy();

    if (sync_config->notify_after_client_reset) {

        config.notify_after_client_reset = [realm_config](VersionID previous_version, bool did_recover) {

            auto coordinator = _impl::RealmCoordinator::get_coordinator(realm_config);

            ThreadSafeReference active_after = coordinator->get_unbound_realm();

            SharedRealm frozen_before = coordinator->get_realm(realm_config, previous_version);

            REALM_ASSERT(frozen_before);

            REALM_ASSERT(frozen_before->is_frozen());

            realm_config.sync_config->notify_after_client_reset(std::move(frozen_before), std::move(active_after),

                                                                did_recover);

        };

    }

    config.notify_before_client_reset = [config = std::move(realm_config)]() -> VersionID {

        // Opening the Realm live here may make a write if the schema is different

        // than what exists on disk. It is necessary to pass a fully usable Realm

        // to the user here. Note that the schema changes made here will be considered

        // an "offline write" to be recovered if this is recovery mode.

        auto before = Realm::get_shared_realm(config);

        before->read_group();

        if (auto& notify_before = config.sync_config->notify_before_client_reset) {

            notify_before(config.sync_config->freeze_before_reset_realm ? before->freeze() : before);

        }

        // Note that if the SDK requested a live Realm this may be a different

        // version than what we had before calling the callback.

        return before->read_transaction_version();

    };

    return config;

}

{

    if (m_session)

        return;

    util::CheckedLockGuard config_lock(m_config_mutex);

    REALM_ASSERT(m_config.sync_config);

    SyncConfig& sync_config = *m_config.sync_config;

    REALM_ASSERT(sync_config.user);

    sync::Session::Config session_config;

    session_config.signed_user_token = sync_config.user->access_token();

    session_config.user_id = sync_config.user->identity();

    session_config.realm_identifier = sync_config.partition_value;

    session_config.verify_servers_ssl_certificate = sync_config.client_validate_ssl;

    session_config.ssl_trust_certificate_path = sync_config.ssl_trust_certificate_path;

    session_config.ssl_verify_callback = sync_config.ssl_verify_callback;

    session_config.proxy_config = sync_config.proxy_config;

    session_config.simulate_integration_error = sync_config.simulate_integration_error;

    session_config.flx_bootstrap_batch_size_bytes = sync_config.flx_bootstrap_batch_size_bytes;

    session_config.session_reason =

        client_reset::is_fresh_path(m_config.path) ? sync::SessionReason::ClientReset : sync::SessionReason::Sync;

    if (sync_config.on_sync_client_event_hook) {

        session_config.on_sync_client_event_hook = [hook = sync_config.on_sync_client_event_hook,

                                                    anchor = weak_from_this()](const SyncClientHookData& data) {

            return hook(anchor, data);

        };

    }

    {

        std::string sync_route = m_sync_manager->sync_route();

        if (!m_client.decompose_server_url(sync_route, session_config.protocol_envelope,

                                           session_config.server_address, session_config.server_port,

                                           session_config.service_identifier)) {

            throw sync::BadServerUrl(sync_route);

        }

        // FIXME: Java needs the fully resolved URL for proxy support, but we also need it before

        // the session is created. How to resolve this?

        m_server_url = sync_route;

    }

    if (sync_config.authorization_header_name) {

        session_config.authorization_header_name = *sync_config.authorization_header_name;

    }

    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_server_requests_action = sync::ProtocolErrorInfo::Action::NoAction;

    }

    m_session = m_client.make_session(m_db, m_flx_subscription_store, m_migration_store, std::move(session_config));

    std::weak_ptr<SyncSession> weak_self = weak_from_this();

    // Set up the wrapped progress handler callback

    m_session->set_progress_handler([weak_self](uint_fast64_t downloaded, uint_fast64_t downloadable,

                                                uint_fast64_t uploaded, uint_fast64_t uploadable,

                                                uint_fast64_t progress_version, uint_fast64_t snapshot_version) {

        if (auto self = weak_self.lock()) {