thomas.goyne_420

Committed 20 Jun 2024 08:37PM UTC coverage: 90.964% (-0.002%) from 90.966%

Build # thomas.goyne_420

Build Type

Pull #7796

Evergreen

Committed by

tgoyne

Commit Message

Derive upload completion entirely from the state of the history

Rather than tracking a bunch of derived state in-memory, check for upload
completion by checking if there are any unuploaded changesets. This is both
multiprocess-compatible and is more precise than the old checks, which had some
false-negatives.

Pull Request Pull Request #7796: RCORE-2160 Make upload completion reporting multiprocess-compatible

Run Details

102120 of 180324 branches covered (56.63%)

112 of 117 new or added lines in 8 files covered. (95.73%)

46 existing lines in 16 files now uncovered.

214645 of 235966 relevant lines covered (90.96%)

5898967.27 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.03

/src/realm/sync/noinst/client_history_impl.cpp

///////////////////////////////////////////////////////////////////////////
//
// Copyright 2022 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 <realm/sync/noinst/client_history_impl.hpp>

#include <realm/sync/changeset.hpp>
#include <realm/sync/changeset_parser.hpp>
#include <realm/sync/instruction_applier.hpp>
#include <realm/sync/instruction_replication.hpp>
#include <realm/sync/noinst/client_reset.hpp>
#include <realm/sync/noinst/client_reset_recovery.hpp>
#include <realm/transaction.hpp>
#include <realm/util/compression.hpp>
#include <realm/util/features.h>
#include <realm/util/functional.hpp>
#include <realm/util/scope_exit.hpp>
#include <realm/version.hpp>

#include <algorithm>
#include <ctime>
#include <cstring>
#include <utility>

namespace realm::sync {

void ClientHistory::set_history_adjustments(
    util::Logger& logger, version_type current_version, SaltedFileIdent client_file_ident,
    SaltedVersion server_version, const std::vector<_impl::client_reset::RecoveredChange>& recovered_changesets)
{
    ensure_updated(current_version); // Throws
    prepare_for_write();             // Throws

    version_type client_version = m_sync_history_base_version + sync_history_size();
    REALM_ASSERT(client_version == current_version); // For now
    Array& root = m_arrays->root;
    m_group->set_sync_file_id(client_file_ident.ident); // Throws

    size_t uploadable_bytes = 0;
    if (recovered_changesets.empty()) {
        // Either we had nothing to upload or we're discarding the unsynced changes
        logger.debug("History adjustments: discarding %1 history entries", sync_history_size());
        do_trim_sync_history(sync_history_size()); // Throws
    }
    else {
        // Discard all sync history before the first recovered changeset. This is
        // required because we are going to discard our progress information and
        // so won't know which history entries have been uploaded already.
        auto first_version = recovered_changesets.front().version;
        REALM_ASSERT(first_version >= m_sync_history_base_version);
        auto discard_count = std::size_t(first_version - m_sync_history_base_version);
        do_trim_sync_history(discard_count);

        if (logger.would_log(util::Logger::Level::debug)) {
            logger.debug("History adjustments: trimming %1 history entries and updating the remaining history "
                         "entries (%2)",
                         discard_count, sync_history_size());
            for (size_t i = 0, size = m_arrays->changesets.size(); i < size; ++i) {
                logger.debug("- %1: ident(%2) changeset_size(%3) remote_version(%4)", i,
                             m_arrays->origin_file_idents.get(i), m_arrays->changesets.get(i).size(),
                             m_arrays->remote_versions.get(i));
            }
        }

        util::compression::CompressMemoryArena arena;
        util::AppendBuffer<char> compressed;
        for (auto& [changeset, version] : recovered_changesets) {
            uploadable_bytes += changeset.size();
            auto i = size_t(version - m_sync_history_base_version);
            util::compression::allocate_and_compress_nonportable(arena, changeset, compressed);
            m_arrays->changesets.set(i, BinaryData{compressed.data(), compressed.size()}); // Throws
            m_arrays->reciprocal_transforms.set(i, BinaryData());
        }
        // Server version is updated for *every* entry in the sync history to ensure that server versions don't
        // decrease.
        for (size_t i = 0, size = m_arrays->remote_versions.size(); i < size; ++i) {
            m_arrays->remote_versions.set(i, server_version.version);
            version_type version = m_sync_history_base_version + i;
            logger.debug("Updating %1: client_version(%2) changeset_size(%3) server_version(%4)", i, version + 1,
                         m_arrays->changesets.get(i).size(), server_version.version);
        }
    }
    logger.debug("New uploadable bytes after history adjustment: %1", uploadable_bytes);

    // Client progress versions are set to 0 to signal to the server that we've
    // reset our versioning. If we send the actual values, the server would
    // complain that the versions (probably) don't correspond to the ones sent
    // when downloading the fresh realm.
    root.set(s_progress_download_client_version_iip,
             RefOrTagged::make_tagged(0)); // Throws
    root.set(s_progress_upload_client_version_iip,
             RefOrTagged::make_tagged(0)); // Throws

    root.set(s_client_file_ident_salt_iip,
             RefOrTagged::make_tagged(client_file_ident.salt)); // Throws
    root.set(s_progress_download_server_version_iip,
             RefOrTagged::make_tagged(server_version.version)); // Throws
    root.set(s_progress_latest_server_version_iip,
             RefOrTagged::make_tagged(server_version.version)); // Throws
    root.set(s_progress_latest_server_version_salt_iip,
             RefOrTagged::make_tagged(server_version.salt)); // Throws
    root.set(s_progress_upload_server_version_iip,
             RefOrTagged::make_tagged(server_version.version)); // Throws
    root.set(s_progress_downloaded_bytes_iip,
             RefOrTagged::make_tagged(0)); // Throws
    root.set(s_progress_downloadable_bytes_iip,
             RefOrTagged::make_tagged(0)); // Throws
    root.set(s_progress_uploaded_bytes_iip,
             RefOrTagged::make_tagged(0)); // Throws
    root.set(s_progress_uploadable_bytes_iip,
             RefOrTagged::make_tagged(uploadable_bytes)); // Throws

    m_progress_download = {server_version.version, 0};
    m_applying_client_reset = true;
}

std::vector<ClientHistory::LocalChange> ClientHistory::get_local_changes(version_type current_version) const
{
    ensure_updated(current_version); // Throws
    std::vector<ClientHistory::LocalChange> changesets;
    if (!m_arrays || m_arrays->changesets.is_empty())
        return changesets;

    sync::version_type begin_version = 0;
    {
        sync::version_type local_version;
        SaltedFileIdent local_ident;
        SyncProgress local_progress;
        get_status(local_version, local_ident, local_progress);
        begin_version = local_progress.upload.client_version;
    }

    version_type end_version = m_sync_history_base_version + sync_history_size();
    if (begin_version < m_sync_history_base_version)
        begin_version = m_sync_history_base_version;

    for (version_type version = begin_version; version < end_version; ++version) {
        std::size_t ndx = std::size_t(version - m_sync_history_base_version);
        std::int_fast64_t origin_file_ident = m_arrays->origin_file_idents.get(ndx);
        bool not_from_server = (origin_file_ident == 0);
        if (not_from_server) {
            bool compressed = false;
            // find_sync_history_entry() returns 0 to indicate not found and
            // otherwise adds 1 to the version, and then get_reciprocal_transform()
            // subtracts 1 from the version
            if (auto changeset = get_reciprocal_transform(version + 1, compressed); !changeset.empty()) {
                changesets.push_back({version, changeset});
            }
        }
    }
    return changesets;
}

void ClientHistory::set_local_origin_timestamp_source(util::UniqueFunction<timestamp_type()> source_fn)
{
    m_local_origin_timestamp_source = std::move(source_fn);
}

// Overriding member function in realm::Replication
void ClientReplication::initialize(DB& sg)
{
    SyncReplication::initialize(sg); // Throws
    m_history.initialize(sg);
}


// Overriding member function in realm::Replication
auto ClientReplication::get_history_type() const noexcept -> HistoryType
{
    return hist_SyncClient;
}


// Overriding member function in realm::Replication
int ClientReplication::get_history_schema_version() const noexcept
{
    return get_client_history_schema_version();
}


// Overriding member function in realm::Replication
bool ClientReplication::is_upgradable_history_schema(int stored_schema_version) const noexcept
{
    if (stored_schema_version == 11) {
        return true;
    }
    return false;
}


// Overriding member function in realm::Replication
void ClientReplication::upgrade_history_schema(int stored_schema_version)
{
    // upgrade_history_schema() is called only when there is a need to upgrade
    // (`stored_schema_version < get_server_history_schema_version()`), and only
    // when is_upgradable_history_schema() returned true (`stored_schema_version
    // >= 1`).
    REALM_ASSERT(stored_schema_version < get_client_history_schema_version());
    REALM_ASSERT(stored_schema_version >= 11);
    int orig_schema_version = stored_schema_version;
    int schema_version = orig_schema_version;

    if (schema_version < 12) {
        m_history.compress_stored_changesets();
        schema_version = 12;
    }

    // NOTE: Future migration steps go here.

    REALM_ASSERT(schema_version == get_client_history_schema_version());

    // Record migration event
    m_history.record_current_schema_version(); // Throws
}

void ClientHistory::compress_stored_changesets()
{
    using gf = _impl::GroupFriend;
    Allocator& alloc = gf::get_alloc(*m_group);
    auto ref = gf::get_history_ref(*m_group);
    Arrays arrays{alloc, m_group, ref};

    util::AppendBuffer<char> compressed_buffer;
    util::AppendBuffer<char> decompressed_buffer;
    util::compression::CompressMemoryArena arena;
    auto columns = {&arrays.reciprocal_transforms, &arrays.changesets};
    for (auto column : columns) {
        for (size_t i = 0; i < column->size(); ++i) {
            ChunkedBinaryData data(*column, i);
            if (data.is_null())
                continue;
            data.copy_to(compressed_buffer);
            util::compression::allocate_and_compress_nonportable(arena, compressed_buffer, decompressed_buffer);
            column->set(i, BinaryData{decompressed_buffer.data(), decompressed_buffer.size()}); // Throws
        }
    }
}

// Overriding member function in realm::Replication
auto ClientReplication::prepare_changeset(const char* data, size_t size, version_type orig_version) -> version_type
{
    m_history.ensure_updated(orig_version);
    m_history.prepare_for_write(); // Throws

    BinaryData ct_changeset{data, size};
    auto& buffer = get_instruction_encoder().buffer();
    BinaryData sync_changeset(buffer.data(), buffer.size());

    return m_history.add_changeset(ct_changeset, sync_changeset); // Throws
}

util::UniqueFunction<SyncReplication::WriteValidator> ClientReplication::make_write_validator(Transaction& tr)
{
    if (!m_write_validator_factory) {
        return {};
    }

    return m_write_validator_factory(tr);
}

void ClientHistory::get_status(version_type& current_client_version, SaltedFileIdent& client_file_ident,
                               SyncProgress& progress) const
{
    TransactionRef rt = m_db->start_read(); // Throws
    version_type current_client_version_2 = rt->get_version();

    SaltedFileIdent client_file_ident_2{rt->get_sync_file_id(), 0};
    SyncProgress progress_2;
    using gf = _impl::GroupFriend;
    if (ref_type ref = gf::get_history_ref(*rt)) {
        Array root(m_db->get_alloc());
        root.init_from_ref(ref);
        client_file_ident_2.salt = salt_type(root.get_as_ref_or_tagged(s_client_file_ident_salt_iip).get_as_int());
        progress_2.latest_server_version.version =
            version_type(root.get_as_ref_or_tagged(s_progress_latest_server_version_iip).get_as_int());
        progress_2.latest_server_version.salt =
            version_type(root.get_as_ref_or_tagged(s_progress_latest_server_version_salt_iip).get_as_int());
        progress_2.download.server_version =
            version_type(root.get_as_ref_or_tagged(s_progress_download_server_version_iip).get_as_int());
        progress_2.download.last_integrated_client_version =
            version_type(root.get_as_ref_or_tagged(s_progress_download_client_version_iip).get_as_int());
        progress_2.upload.client_version =
            version_type(root.get_as_ref_or_tagged(s_progress_upload_client_version_iip).get_as_int());
        progress_2.upload.last_integrated_server_version =
            version_type(root.get_as_ref_or_tagged(s_progress_upload_server_version_iip).get_as_int());
    }

    current_client_version = current_client_version_2;
    client_file_ident = client_file_ident_2;
    progress = progress_2;

    REALM_ASSERT(current_client_version >= s_initial_version + 0);
    if (current_client_version == s_initial_version + 0)
        current_client_version = 0;
}

void ClientHistory::set_client_file_ident(SaltedFileIdent client_file_ident, bool fix_up_object_ids)
{
    REALM_ASSERT(client_file_ident.ident != 0);

    TransactionRef wt = m_db->start_write(); // Throws
    version_type local_version = wt->get_version();
    ensure_updated(local_version); // Throws
    prepare_for_write();           // Throws

    Array& root = m_arrays->root;
    REALM_ASSERT(wt->get_sync_file_id() == 0);
    wt->set_sync_file_id(client_file_ident.ident);
    root.set(s_client_file_ident_salt_iip,
             RefOrTagged::make_tagged(client_file_ident.salt)); // Throws
    root.set(s_progress_download_client_version_iip, RefOrTagged::make_tagged(0));
    root.set(s_progress_upload_client_version_iip, RefOrTagged::make_tagged(0));

    if (fix_up_object_ids) {
        fix_up_client_file_ident_in_stored_changesets(*wt, client_file_ident.ident); // Throws
    }

    // Note: This transaction produces an empty changeset. Empty changesets are
    // not uploaded to the server.
    wt->commit(); // Throws
}


// Overriding member function in realm::sync::ClientHistoryBase
void ClientHistory::set_sync_progress(const SyncProgress& progress, DownloadableProgress downloadable_bytes,
                                      VersionInfo& version_info)
{
    TransactionRef wt = m_db->start_write(); // Throws
    version_type local_version = wt->get_version();
    ensure_updated(local_version); // Throws
    prepare_for_write();           // Throws

    update_sync_progress(progress, downloadable_bytes); // Throws

    // Note: This transaction produces an empty changeset. Empty changesets are
    // not uploaded to the server.
    version_type new_version = wt->commit(); // Throws
    version_info.realm_version = new_version;
    version_info.sync_version = {new_version, 0};
}

void ClientHistory::find_uploadable_changesets(UploadCursor& upload_progress, version_type end_version,
                                               std::vector<UploadChangeset>& uploadable_changesets,
                                               version_type& locked_server_version) const
{
    TransactionRef rt = m_db->start_read(); // Throws
    auto& alloc = m_db->get_alloc();
    using gf = _impl::GroupFriend;
    ref_type ref = gf::get_history_ref(*rt);
    REALM_ASSERT(ref);

    Arrays arrays(alloc, rt.get(), ref);
    const auto sync_history_size = arrays.changesets.size();
    const auto sync_history_base_version = rt->get_version() - sync_history_size;

    std::size_t accum_byte_size_soft_limit = 131072;   // 128 KB
    std::size_t accum_byte_size_hard_limit = 16777216; // server-imposed limit
    std::size_t accum_byte_size = 0;

    version_type begin_version_2 = std::max(upload_progress.client_version, sync_history_base_version);
    version_type end_version_2 = std::max(end_version, sync_history_base_version);
    version_type last_integrated_upstream_version = upload_progress.last_integrated_server_version;

    while (accum_byte_size < accum_byte_size_soft_limit) {
        HistoryEntry entry;
        version_type last_integrated_upstream_version_2 = last_integrated_upstream_version;
        version_type version = find_sync_history_entry(arrays, sync_history_base_version, begin_version_2,
                                                       end_version_2, entry, last_integrated_upstream_version_2);

        if (version == 0) {
            begin_version_2 = end_version_2;
            last_integrated_upstream_version = last_integrated_upstream_version_2;
            break;
        }

        ChunkedBinaryInputStream is(entry.changeset);
        size_t size = util::compression::get_uncompressed_size_from_header(is);
        if (accum_byte_size + size >= accum_byte_size_hard_limit && !uploadable_changesets.empty())
            break;
        accum_byte_size += size;
        last_integrated_upstream_version = last_integrated_upstream_version_2;
        begin_version_2 = version;

        UploadChangeset uc;
        util::AppendBuffer<char> decompressed;
        ChunkedBinaryInputStream is_2(entry.changeset);
        auto ec = util::compression::decompress_nonportable(is_2, decompressed);
        if (ec == util::compression::error::decompress_unsupported) {
            REALM_TERMINATE(
                "Synchronized Realm files with unuploaded local changes cannot be copied between platforms.");
        }
        REALM_ASSERT_3(ec, ==, std::error_code{});

        uc.origin_timestamp = entry.origin_timestamp;
        uc.origin_file_ident = entry.origin_file_ident;
        uc.progress = UploadCursor{version, entry.remote_version};
        uc.changeset = BinaryData{decompressed.data(), decompressed.size()};
        uc.buffer = decompressed.release().release();
        uploadable_changesets.push_back(std::move(uc)); // Throws
    }

    upload_progress = {std::min(begin_version_2, end_version), last_integrated_upstream_version};

    locked_server_version = arrays.root.get_as_ref_or_tagged(s_progress_download_server_version_iip).get_as_int();
}


void ClientHistory::integrate_server_changesets(
    const SyncProgress& progress, DownloadableProgress downloadable_bytes,
    util::Span<const RemoteChangeset> incoming_changesets, VersionInfo& version_info, DownloadBatchState batch_state,
    util::Logger& logger, const TransactionRef& transact,
    util::UniqueFunction<void(const TransactionRef&, util::Span<Changeset>)> run_in_write_tr)
{
    REALM_ASSERT(incoming_changesets.size() != 0);
    REALM_ASSERT(
        (transact->get_transact_stage() == DB::transact_Writing && batch_state != DownloadBatchState::SteadyState) ||
        (transact->get_transact_stage() == DB::transact_Reading && batch_state == DownloadBatchState::SteadyState));
    std::vector<Changeset> changesets;
    changesets.resize(incoming_changesets.size()); // Throws

    // Parse incoming changesets without holding the write lock unless 'transact' is specified.
    try {
        for (std::size_t i = 0; i < incoming_changesets.size(); ++i) {
            const RemoteChangeset& changeset = incoming_changesets[i];
            parse_remote_changeset(changeset, changesets[i]); // Throws
            changesets[i].transform_sequence = i;
        }
    }
    catch (const BadChangesetError& e) {
        throw IntegrationException(ErrorCodes::BadChangeset,
                                   util::format("Failed to parse received changeset: %1", e.what()),
                                   ProtocolError::bad_changeset);
    }

    VersionID new_version{0, 0};
    auto num_changesets = incoming_changesets.size();
    util::Span<Changeset> changesets_to_integrate(changesets);
    const bool allow_lock_release = batch_state == DownloadBatchState::SteadyState;

    // Ideally, this loop runs only once, but it can run up to `incoming_changesets.size()` times, depending on the
    // number of times the sync client yields the write lock to allow the user to commit their changes.
    // In each iteration, at least one changeset is transformed and committed.
    // In FLX, all changesets are committed at once in the bootstrap phase (i.e, in one iteration).
    while (!changesets_to_integrate.empty()) {
        if (transact->get_transact_stage() == DB::transact_Reading) {
            transact->promote_to_write(); // Throws
        }
        VersionID old_version = transact->get_version_of_current_transaction();
        version_type local_version = old_version.version;
        auto sync_file_id = transact->get_sync_file_id();
        REALM_ASSERT(sync_file_id != 0);

        ensure_updated(local_version); // Throws
        prepare_for_write();           // Throws

        std::uint64_t downloaded_bytes_in_transaction = 0;
        auto changesets_transformed_count = transform_and_apply_server_changesets(
            changesets_to_integrate, transact, logger, downloaded_bytes_in_transaction, allow_lock_release);

        // downloaded_bytes always contains the total number of downloaded bytes
        // from the Realm. downloaded_bytes must be persisted in the Realm, since
        // the downloaded changesets are trimmed after use, and since it would be
        // expensive to traverse the entire history.
        Array& root = m_arrays->root;
        auto downloaded_bytes =
            std::uint64_t(root.get_as_ref_or_tagged(s_progress_downloaded_bytes_iip).get_as_int());
        downloaded_bytes += downloaded_bytes_in_transaction;
        root.set(s_progress_downloaded_bytes_iip, RefOrTagged::make_tagged(downloaded_bytes)); // Throws

        const RemoteChangeset& last_changeset = incoming_changesets[changesets_transformed_count - 1];
        auto changesets_for_cb = changesets_to_integrate.first(changesets_transformed_count);
        changesets_to_integrate = changesets_to_integrate.sub_span(changesets_transformed_count);
        incoming_changesets = incoming_changesets.sub_span(changesets_transformed_count);

        // During the bootstrap phase in flexible sync, the server sends multiple download messages with the same
        // synthetic server version that represents synthetic changesets generated from state on the server.
        if (batch_state == DownloadBatchState::LastInBatch && changesets_to_integrate.empty()) {
            update_sync_progress(progress, downloadable_bytes); // Throws
        }
        // Always update progress for download messages from steady state.
        else if (batch_state == DownloadBatchState::SteadyState && !changesets_to_integrate.empty()) {
            auto partial_progress = progress;
            partial_progress.download.server_version = last_changeset.remote_version;
            partial_progress.download.last_integrated_client_version = last_changeset.last_integrated_local_version;
            update_sync_progress(partial_progress, downloadable_bytes); // Throws
        }
        else if (batch_state == DownloadBatchState::SteadyState && changesets_to_integrate.empty()) {
            update_sync_progress(progress, downloadable_bytes); // Throws
        }
        if (run_in_write_tr) {
            run_in_write_tr(transact, changesets_for_cb);
        }

        // The reason we can use the `origin_timestamp`, and the `origin_file_ident`
        // from the last transformed changeset, and ignore all the other changesets, is
        // that these values are actually irrelevant for changesets of remote origin
        // stored in the client-side history (for now), except that
        // `origin_file_ident` is required to be nonzero, to mark it as having been
        // received from the server.
        HistoryEntry entry;
        entry.origin_timestamp = last_changeset.origin_timestamp;
        entry.origin_file_ident = last_changeset.origin_file_ident;
        entry.remote_version = last_changeset.remote_version;
        add_sync_history_entry(entry); // Throws

        // Tell prepare_commit()/add_changeset() not to write a history entry for
        // this transaction as we already did it.
        REALM_ASSERT(!m_applying_server_changeset);
        m_applying_server_changeset = true;
        // Commit and continue to write if in bootstrap phase and there are still changes to integrate.
        if (batch_state == DownloadBatchState::MoreToCome ||
            (batch_state == DownloadBatchState::LastInBatch && !changesets_to_integrate.empty())) {
            new_version = transact->commit_and_continue_writing(); // Throws
        }
        else {
            new_version = transact->commit_and_continue_as_read(); // Throws
        }

        logger.debug(util::LogCategory::changeset, "Integrated %1 changesets out of %2", changesets_transformed_count,
                     num_changesets);
    }

    REALM_ASSERT(new_version.version > 0);
    REALM_ASSERT(
        (batch_state == DownloadBatchState::MoreToCome && transact->get_transact_stage() == DB::transact_Writing) ||
        (batch_state != DownloadBatchState::MoreToCome && transact->get_transact_stage() == DB::transact_Reading));
    version_info.realm_version = new_version.version;
    version_info.sync_version = {new_version.version, 0};
}


size_t ClientHistory::transform_and_apply_server_changesets(util::Span<Changeset> changesets_to_integrate,
                                                            TransactionRef transact, util::Logger& logger,
                                                            std::uint64_t& downloaded_bytes, bool allow_lock_release)
{
    REALM_ASSERT(transact->get_transact_stage() == DB::transact_Writing);

    if (!m_replication.apply_server_changes()) {
        std::for_each(changesets_to_integrate.begin(), changesets_to_integrate.end(), [&](const Changeset& c) {
            downloaded_bytes += c.original_changeset_size;
        });
        // Skip over all changesets if they don't need to be transformed and applied.
        return changesets_to_integrate.size();
    }

    version_type local_version = transact->get_version_of_current_transaction().version;
    auto sync_file_id = transact->get_sync_file_id();

    try {
        for (auto& changeset : changesets_to_integrate) {
            REALM_ASSERT(changeset.last_integrated_remote_version <= local_version);
            REALM_ASSERT(changeset.origin_file_ident > 0 && changeset.origin_file_ident != sync_file_id);

            // It is possible that the synchronization history has been trimmed
            // to a point where a prefix of the merge window is no longer
            // available, but this can only happen if that prefix consisted
            // entirely of upload skippable entries. Since such entries (those
            // that are empty or of remote origin) will be skipped by the
            // transformer anyway, we can simply clamp the beginning of the
            // merge window to the beginning of the synchronization history,
            // when this situation occurs.
            //
            // See trim_sync_history() for further details.
            if (changeset.last_integrated_remote_version < m_sync_history_base_version)
                changeset.last_integrated_remote_version = m_sync_history_base_version;
        }

        constexpr std::size_t commit_byte_size_limit = 102400; // 100 KB

        auto changeset_applier = [&](const Changeset* transformed_changeset) -> bool {
            InstructionApplier applier{*transact};
            {
                TempShortCircuitReplication tscr{m_replication};
                applier.apply(*transformed_changeset); // Throws
            }
            downloaded_bytes += transformed_changeset->original_changeset_size;

            return !(allow_lock_release && m_db->other_writers_waiting_for_lock() &&
                     transact->get_commit_size() >= commit_byte_size_limit);
        };
        sync::Transformer transformer;
        auto changesets_transformed_count = transformer.transform_remote_changesets(
            *this, sync_file_id, local_version, changesets_to_integrate, changeset_applier, logger); // Throws
        return changesets_transformed_count;
    }
    catch (const BadChangesetError& e) {
        throw IntegrationException(ErrorCodes::BadChangeset,
                                   util::format("Failed to apply received changeset: %1", e.what()),
                                   ProtocolError::bad_changeset);
    }
    catch (const TransformError& e) {
        throw IntegrationException(ErrorCodes::BadChangeset,
                                   util::format("Failed to transform received changeset: %1", e.what()),
                                   ProtocolError::bad_changeset);
    }
}


void ClientHistory::get_upload_download_state(DB& db, std::uint_fast64_t& downloaded_bytes,
                                              DownloadableProgress& downloadable_bytes,
                                              std::uint_fast64_t& uploaded_bytes,
                                              std::uint_fast64_t& uploadable_bytes,
                                              std::uint_fast64_t& snapshot_version, version_type& uploaded_version)
{
    TransactionRef rt = db.start_read(); // Throws
    version_type current_client_version = rt->get_version();

    downloaded_bytes = 0;
    downloadable_bytes = uint64_t(0);
    uploaded_bytes = 0;
    uploadable_bytes = 0;
    snapshot_version = current_client_version;
    uploaded_version = 0;

    using gf = _impl::GroupFriend;
    ref_type ref = gf::get_history_ref(*rt);
    if (!ref)
        return;

    Array root(db.get_alloc());
    root.init_from_ref(ref);
    downloaded_bytes = root.get_as_ref_or_tagged(s_progress_downloaded_bytes_iip).get_as_int();
    downloadable_bytes = root.get_as_ref_or_tagged(s_progress_downloadable_bytes_iip).get_as_int();
    uploadable_bytes = root.get_as_ref_or_tagged(s_progress_uploadable_bytes_iip).get_as_int();
    uploaded_bytes = root.get_as_ref_or_tagged(s_progress_uploaded_bytes_iip).get_as_int();

    uploaded_version = root.get_as_ref_or_tagged(s_progress_upload_client_version_iip).get_as_int();
    if (uploaded_version == current_client_version)
        return;

    BinaryColumn changesets(db.get_alloc());
    changesets.init_from_ref(root.get_as_ref(s_changesets_iip));
    IntegerBpTree origin_file_idents(db.get_alloc());
    origin_file_idents.init_from_ref(root.get_as_ref(s_origin_file_idents_iip));

    // `base_version` is the oldest version we have history for. If this is
    // greater than uploaded_version, all of the versions in between the two had
    // empty changesets and did not need to be uploaded. If this is less than
    // uploaded_version, we have changesets which have been uploaded but the
    // server has not yet told us we can delete and we may need to use for merging.
    auto base_version = current_client_version - changesets.size();
    if (uploaded_version < base_version) {
        uploaded_version = base_version;
    }

    auto count = size_t(current_client_version - uploaded_version);
    for (size_t i = changesets.size() - count; i < changesets.size(); ++i) {
        if (origin_file_idents.get(i) == 0) {
            size_t pos = 0;
            if (changesets.get_at(i, pos).size() != 0)
                break;
        }
        ++uploaded_version;
    }
}

void ClientHistory::get_upload_download_state(DB* db, std::uint_fast64_t& downloaded_bytes,
                                              std::uint_fast64_t& uploaded_bytes)
{
    TransactionRef rt = db->start_read(); // Throws
    downloaded_bytes = 0;
    uploaded_bytes = 0;

    using gf = _impl::GroupFriend;
    if (ref_type ref = gf::get_history_ref(*rt)) {
        Array root(db->get_alloc());
        root.init_from_ref(ref);
        downloaded_bytes = root.get_as_ref_or_tagged(s_progress_downloaded_bytes_iip).get_as_int();
        uploaded_bytes = root.get_as_ref_or_tagged(s_progress_uploaded_bytes_iip).get_as_int();
    }
}

auto ClientHistory::find_history_entry(version_type begin_version, version_type end_version,
                                       HistoryEntry& entry) const noexcept -> version_type
{
    version_type last_integrated_server_version;
    return find_sync_history_entry(*m_arrays, m_sync_history_base_version, begin_version, end_version, entry,
                                   last_integrated_server_version);
}


ChunkedBinaryData ClientHistory::get_reciprocal_transform(version_type version, bool& is_compressed) const
{
    is_compressed = true;
    REALM_ASSERT(version > m_sync_history_base_version);

    std::size_t index = to_size_t(version - m_sync_history_base_version) - 1;
    REALM_ASSERT(index < sync_history_size());

    ChunkedBinaryData reciprocal{m_arrays->reciprocal_transforms, index};
    if (!reciprocal.is_null())
        return reciprocal;
    return ChunkedBinaryData{m_arrays->changesets, index};
}


void ClientHistory::set_reciprocal_transform(version_type version, BinaryData data)
{
    REALM_ASSERT(version > m_sync_history_base_version);

    std::size_t index = size_t(version - m_sync_history_base_version) - 1;
    REALM_ASSERT(index < sync_history_size());

    if (data.is_null()) {
        m_arrays->reciprocal_transforms.set(index, BinaryData{"", 0}); // Throws
        return;
    }

    auto compressed = util::compression::allocate_and_compress_nonportable(data);
    m_arrays->reciprocal_transforms.set(index, BinaryData{compressed.data(), compressed.size()}); // Throws
}


auto ClientHistory::find_sync_history_entry(Arrays& arrays, version_type base_version, version_type begin_version,
                                            version_type end_version, HistoryEntry& entry,
                                            version_type& last_integrated_server_version) noexcept -> version_type
{
    if (begin_version == 0)
        begin_version = s_initial_version + 0;

    REALM_ASSERT(begin_version <= end_version);
    REALM_ASSERT(begin_version >= base_version);
    REALM_ASSERT(end_version <= base_version + arrays.changesets.size());
    std::size_t n = to_size_t(end_version - begin_version);
    std::size_t offset = to_size_t(begin_version - base_version);
    for (std::size_t i = 0; i < n; ++i) {
        std::int_fast64_t origin_file_ident = arrays.origin_file_idents.get(offset + i);
        last_integrated_server_version = version_type(arrays.remote_versions.get(offset + i));
        bool not_from_server = (origin_file_ident == 0);
        if (not_from_server) {
            ChunkedBinaryData chunked_changeset(arrays.changesets, offset + i);
            if (!chunked_changeset.empty()) {
                entry.origin_file_ident = file_ident_type(origin_file_ident);
                entry.remote_version = last_integrated_server_version;
                entry.origin_timestamp = timestamp_type(arrays.origin_timestamps.get(offset + i));
                entry.changeset = chunked_changeset;
                return begin_version + i + 1;
            }
        }
    }
    return 0;
}

// sum_of_history_entry_sizes calculates the sum of the changeset sizes of the
// local history entries that produced a version that succeeds `begin_version`
// and precedes `end_version`.
std::uint_fast64_t ClientHistory::sum_of_history_entry_sizes(version_type begin_version,
                                                             version_type end_version) const noexcept
{
    if (begin_version >= end_version)
        return 0;

    REALM_ASSERT(m_arrays->changesets.is_attached());
    REALM_ASSERT(m_arrays->origin_file_idents.is_attached());
    REALM_ASSERT(end_version <= m_sync_history_base_version + sync_history_size());

    version_type begin_version_2 = begin_version;
    version_type end_version_2 = end_version;
    clamp_sync_version_range(begin_version_2, end_version_2);

    std::uint_fast64_t sum_of_sizes = 0;

    std::size_t n = to_size_t(end_version_2 - begin_version_2);
    std::size_t offset = to_size_t(begin_version_2 - m_sync_history_base_version);
    for (std::size_t i = 0; i < n; ++i) {

        // Only local changesets are considered
        if (m_arrays->origin_file_idents.get(offset + i) != 0)
            continue;

        ChunkedBinaryData changeset(m_arrays->changesets, offset + i);
        ChunkedBinaryInputStream in{changeset};
        sum_of_sizes += util::compression::get_uncompressed_size_from_header(in);
    }

    return sum_of_sizes;
}

void ClientHistory::prepare_for_write()
{
    if (m_arrays) {
        REALM_ASSERT(m_arrays->root.size() == s_root_size);
        return;
    }

    m_arrays.emplace(*m_db, *m_group);
}


Replication::version_type ClientHistory::add_changeset(BinaryData ct_changeset, BinaryData sync_changeset)
{
    // FIXME: BinaryColumn::set() currently interprets BinaryData(0,0) as
    // null. It should probably be changed such that BinaryData(0,0) is always
    // interpreted as the empty string. For the purpose of setting null values,
    // BinaryColumn::set() should accept values of type Optional<BinaryData>().
    if (ct_changeset.is_null())
        ct_changeset = BinaryData("", 0);
    m_arrays->ct_history.add(ct_changeset); // Throws

    REALM_ASSERT(!m_applying_server_changeset || !m_applying_client_reset);

    // If we're applying a changeset from the server then we should have already
    // added the history entry and don't need to do so here
    if (m_applying_server_changeset) {
        // We need to unset this before committing the write, as it's guarded
        // by the write lock
        m_applying_server_changeset = false;
        REALM_ASSERT(m_ct_history_base_version + ct_history_size() ==
                     m_sync_history_base_version + sync_history_size());
        REALM_ASSERT(sync_changeset.size() == 0);
        return m_ct_history_base_version + ct_history_size();
    }

    // We don't generate a changeset for any of the changes made as part of
    // applying a client reset as those changes are just bringing us into
    // alignment with the new server state
    if (m_applying_client_reset) {
        m_applying_client_reset = false;
        sync_changeset = {};
    }

    HistoryEntry entry;
    entry.origin_timestamp = m_local_origin_timestamp_source();
    entry.origin_file_ident = 0; // Of local origin
    entry.remote_version = m_progress_download.server_version;
    entry.changeset = sync_changeset;
    add_sync_history_entry(entry); // Throws

    // uploadable_bytes is updated at every local Realm change. The total
    // number of uploadable bytes must be persisted in the Realm, since the
    // synchronization history is trimmed. Even if the synchronization
    // history wasn't trimmed, it would be expensive to traverse the entire
    // history at every access to uploadable bytes.
    Array& root = m_arrays->root;
    std::uint_fast64_t uploadable_bytes = root.get_as_ref_or_tagged(s_progress_uploadable_bytes_iip).get_as_int();
    uploadable_bytes += entry.changeset.size();
    root.set(s_progress_uploadable_bytes_iip, RefOrTagged::make_tagged(uploadable_bytes));

    return m_ct_history_base_version + ct_history_size();
}

void ClientHistory::add_sync_history_entry(const HistoryEntry& entry)
{
    REALM_ASSERT(m_arrays->changesets.size() == sync_history_size());
    REALM_ASSERT(m_arrays->reciprocal_transforms.size() == sync_history_size());
    REALM_ASSERT(m_arrays->remote_versions.size() == sync_history_size());
    REALM_ASSERT(m_arrays->origin_file_idents.size() == sync_history_size());
    REALM_ASSERT(m_arrays->origin_timestamps.size() == sync_history_size());

    if (!entry.changeset.is_null()) {
        auto changeset = entry.changeset.get_first_chunk();
        auto compressed = util::compression::allocate_and_compress_nonportable(changeset);
        m_arrays->changesets.add(BinaryData{compressed.data(), compressed.size()}); // Throws
    }
    else {
        m_arrays->changesets.add(BinaryData("", 0)); // Throws
    }

    m_arrays->reciprocal_transforms.add(BinaryData{});                                            // Throws
    m_arrays->remote_versions.insert(realm::npos, std::int_fast64_t(entry.remote_version));       // Throws
    m_arrays->origin_file_idents.insert(realm::npos, std::int_fast64_t(entry.origin_file_ident)); // Throws
    m_arrays->origin_timestamps.insert(realm::npos, std::int_fast64_t(entry.origin_timestamp));   // Throws
}


void ClientHistory::update_sync_progress(const SyncProgress& progress, DownloadableProgress downloadable_bytes)
{
    Array& root = m_arrays->root;

    // Progress must never decrease
    if (auto current = version_type(root.get_as_ref_or_tagged(s_progress_latest_server_version_iip).get_as_int());
        progress.latest_server_version.version < current) {
        throw IntegrationException(ErrorCodes::SyncProtocolInvariantFailed,
                                   util::format("latest server version cannot decrease (current: %1, new: %2)",
                                                current, progress.latest_server_version.version),
                                   ProtocolError::bad_progress);
    }
    if (auto current = version_type(root.get_as_ref_or_tagged(s_progress_download_server_version_iip).get_as_int());
        progress.download.server_version < current) {
        throw IntegrationException(
            ErrorCodes::SyncProtocolInvariantFailed,
            util::format("server version of download cursor cannot decrease (current: %1, new: %2)", current,
                         progress.download.server_version),
            ProtocolError::bad_progress);
    }
    if (auto current = version_type(root.get_as_ref_or_tagged(s_progress_download_client_version_iip).get_as_int());
        progress.download.last_integrated_client_version < current) {
        throw IntegrationException(
            ErrorCodes::SyncProtocolInvariantFailed,
            util::format("last integrated client version of download cursor cannot decrease (current: %1, new: %2)",
                         current, progress.download.last_integrated_client_version),
            ProtocolError::bad_progress);
    }
    if (auto current = version_type(root.get_as_ref_or_tagged(s_progress_upload_client_version_iip).get_as_int());
        progress.upload.client_version < current) {
        throw IntegrationException(
            ErrorCodes::SyncProtocolInvariantFailed,
            util::format("client version of upload cursor cannot decrease (current: %1, new: %2)", current,
                         progress.upload.client_version),
            ProtocolError::bad_progress);
    }
    const auto last_integrated_server_version = progress.upload.last_integrated_server_version;
    if (auto current = version_type(root.get_as_ref_or_tagged(s_progress_upload_server_version_iip).get_as_int());
        last_integrated_server_version > 0 && last_integrated_server_version < current) {
        throw IntegrationException(
            ErrorCodes::SyncProtocolInvariantFailed,
            util::format("last integrated server version of upload cursor cannot decrease (current: %1, new: %2)",
                         current, last_integrated_server_version),
            ProtocolError::bad_progress);
    }

    auto uploaded_bytes = std::uint_fast64_t(root.get_as_ref_or_tagged(s_progress_uploaded_bytes_iip).get_as_int());
    auto previous_upload_client_version =
        version_type(root.get_as_ref_or_tagged(s_progress_upload_client_version_iip).get_as_int());
    uploaded_bytes += sum_of_history_entry_sizes(previous_upload_client_version, progress.upload.client_version);

    root.set(s_progress_download_server_version_iip,
             RefOrTagged::make_tagged(progress.download.server_version)); // Throws
    root.set(s_progress_download_client_version_iip,
             RefOrTagged::make_tagged(progress.download.last_integrated_client_version)); // Throws
    root.set(s_progress_latest_server_version_iip,
             RefOrTagged::make_tagged(progress.latest_server_version.version)); // Throws
    root.set(s_progress_latest_server_version_salt_iip,
             RefOrTagged::make_tagged(progress.latest_server_version.salt)); // Throws
    root.set(s_progress_upload_client_version_iip,
             RefOrTagged::make_tagged(progress.upload.client_version)); // Throws
    if (progress.upload.last_integrated_server_version > 0) {
        root.set(s_progress_upload_server_version_iip,
                 RefOrTagged::make_tagged(progress.upload.last_integrated_server_version)); // Throws
    }

    root.set(s_progress_downloadable_bytes_iip,
             RefOrTagged::make_tagged(downloadable_bytes.as_bytes())); // Throws
    root.set(s_progress_uploaded_bytes_iip,
             RefOrTagged::make_tagged(uploaded_bytes)); // Throws

    m_progress_download = progress.download;

    trim_sync_history(); // Throws
}


void ClientHistory::trim_ct_history()
{
    version_type begin = m_ct_history_base_version;
    version_type end = m_version_of_oldest_bound_snapshot;
    REALM_ASSERT(end >= begin);

    std::size_t n = std::size_t(end - begin);
    if (n == 0)
        return;

    // The new changeset is always added before set_oldest_bound_version()
    // is called. Therefore, the trimming operation can never leave the
    // history empty.
    REALM_ASSERT(n < ct_history_size());

    for (std::size_t i = 0; i < n; ++i) {
        std::size_t j = (n - 1) - i;
        m_arrays->ct_history.erase(j);
    }

    m_ct_history_base_version += n;

    REALM_ASSERT(m_ct_history_base_version + ct_history_size() == m_sync_history_base_version + sync_history_size());
}


// Trimming rules for synchronization history:
//
// Let C be the latest client version that was integrated on the server prior to
// the latest server version currently integrated by the client
// (`m_progress_download.last_integrated_client_version`).
//
// Definition: An *upload skippable history entry* is one whose changeset is
// either empty, or of remote origin.
//
// Then, a history entry, E, can be trimmed away if it precedes C, or E is
// upload skippable, and there are no upload nonskippable entries between C and
// E.
//
// Since the history representation is contiguous, it is necessary that the
// trimming rule upholds the following invariant:
//
// > If a changeset can be trimmed, then any earlier changeset can also be
// > trimmed.
//
// Note that C corresponds to the earliest possible beginning of the merge
// window for the next incoming changeset from the server.
void ClientHistory::trim_sync_history()
{
    version_type begin = m_sync_history_base_version;
    version_type end = std::max(m_progress_download.last_integrated_client_version, s_initial_version + 0);
    // Note: At this point, `end` corresponds to C in the description above.

    // `end` (`m_progress_download.last_integrated_client_version`) will precede
    // the beginning of the history, if we trimmed beyond
    // `m_progress_download.last_integrated_client_version` during the previous
    // trimming session. Since new entries, that have now become eligible for
    // scanning, may also be upload skippable, we need to continue the scan from
    // the beginning of the history in that case.
    if (end < begin)
        end = begin;

    // FIXME: It seems like in some cases, a particular history entry that
    // terminates the scan may get examined over and over every time
    // trim_history() is called. For this reason, it seems like it would be
    // worth considering to cache the outcome.

    // FIXME: It seems like there is a significant overlap between what is going
    // on here and in a place like find_uploadable_changesets(). Maybe there is
    // grounds for some refactoring to take that into account, especially, to
    // avoid scanning the same parts of the history for the same information
    // multiple times.

    {
        std::size_t offset = std::size_t(end - begin);
        std::size_t n = std::size_t(sync_history_size() - offset);
        std::size_t i = 0;
        while (i < n) {
            std::int_fast64_t origin_file_ident = m_arrays->origin_file_idents.get(offset + i);
            bool of_local_origin = (origin_file_ident == 0);
            if (of_local_origin) {
                std::size_t pos = 0;
                BinaryData chunk = m_arrays->changesets.get_at(offset + i, pos);
                bool nonempty = (chunk.size() > 0);
                if (nonempty)
                    break; // Not upload skippable
            }
            ++i;
        }
        end += i;
    }

    std::size_t n = std::size_t(end - begin);
    do_trim_sync_history(n); // Throws
}

bool ClientHistory::no_pending_local_changes(version_type version) const
{
    ensure_updated(version);
    for (size_t i = 0; i < sync_history_size(); i++) {
        if (m_arrays->origin_file_idents.get(i) == 0) {
            std::size_t pos = 0;
            BinaryData chunk = m_arrays->changesets.get_at(i, pos);
            if (chunk.size() > 0)
                return false;
        }
    }
    return true;
}

void ClientHistory::do_trim_sync_history(std::size_t n)
{
    REALM_ASSERT(m_arrays->changesets.size() == sync_history_size());
    REALM_ASSERT(m_arrays->reciprocal_transforms.size() == sync_history_size());
    REALM_ASSERT(m_arrays->remote_versions.size() == sync_history_size());
    REALM_ASSERT(m_arrays->origin_file_idents.size() == sync_history_size());
    REALM_ASSERT(m_arrays->origin_timestamps.size() == sync_history_size());
    REALM_ASSERT(n <= sync_history_size());

    if (n == sync_history_size()) {
        m_arrays->changesets.clear();
        m_arrays->reciprocal_transforms.clear();
        m_arrays->remote_versions.clear();
        m_arrays->origin_file_idents.clear();
        m_arrays->origin_timestamps.clear();
    }
    else if (n > 0) {
        for (std::size_t i = 0; i < n; ++i) {
            std::size_t j = (n - 1) - i;
            m_arrays->changesets.erase(j); // Throws
        }
        for (std::size_t i = 0; i < n; ++i) {
            std::size_t j = (n - 1) - i;
            m_arrays->reciprocal_transforms.erase(j); // Throws
        }
        for (std::size_t i = 0; i < n; ++i) {
            std::size_t j = (n - 1) - i;
            m_arrays->remote_versions.erase(j); // Throws
        }
        for (std::size_t i = 0; i < n; ++i) {
            std::size_t j = (n - 1) - i;
            m_arrays->origin_file_idents.erase(j); // Throws
        }
        for (std::size_t i = 0; i < n; ++i) {
            std::size_t j = (n - 1) - i;
            m_arrays->origin_timestamps.erase(j); // Throws
        }
    }

    m_sync_history_base_version += n;
}

void ClientHistory::fix_up_client_file_ident_in_stored_changesets(Transaction& group,
                                                                  file_ident_type client_file_ident)
{
    // Must be in write transaction!

    REALM_ASSERT(client_file_ident != 0);
    auto promote_global_key = [client_file_ident](GlobalKey* oid) {
        if (oid->hi() == 0) {
            // client_file_ident == 0
            *oid = GlobalKey{uint64_t(client_file_ident), oid->lo()};
            return true;
        }
        return false;
    };

    Group::TableNameBuffer buffer;
    auto get_table_for_class = [&](StringData class_name) -> ConstTableRef {
        return group.get_table(Group::class_name_to_table_name(class_name, buffer));
    };

    util::compression::CompressMemoryArena arena;
    util::AppendBuffer<char> compressed;

    // Fix up changesets.
    Array& root = m_arrays->root;
    uint64_t uploadable_bytes = root.get_as_ref_or_tagged(s_progress_uploadable_bytes_iip).get_as_int();
    for (size_t i = 0; i < sync_history_size(); ++i) {
        // We could have opened a pre-provisioned realm file. In this case we can skip the entries downloaded
        // from the server.
        if (m_arrays->origin_file_idents.get(i) != 0)
            continue;

        ChunkedBinaryData changeset{m_arrays->changesets, i};
        ChunkedBinaryInputStream is{changeset};
        size_t decompressed_size;
        auto decompressed = util::compression::decompress_nonportable_input_stream(is, decompressed_size);
        if (!decompressed)
            continue;
        Changeset log;
        parse_changeset(*decompressed, log);

        bool did_modify = false;
        auto last_class_name = InternString::npos;
        ConstTableRef selected_table;
        for (auto instr : log) {
            if (!instr)
                continue;

            if (auto obj_instr = instr->get_if<Instruction::ObjectInstruction>()) {
                // Cache the TableRef
                if (obj_instr->table != last_class_name) {
                    StringData class_name = log.get_string(obj_instr->table);
                    last_class_name = obj_instr->table;
                    selected_table = get_table_for_class(class_name);
                }

                // Fix up instructions using GlobalKey to identify objects.
                if (auto global_key = mpark::get_if<GlobalKey>(&obj_instr->object)) {
                    did_modify = promote_global_key(global_key);
                }

                // Fix up the payload for Set and ArrayInsert.
                Instruction::Payload* payload = nullptr;
                if (auto set_instr = instr->get_if<Instruction::Update>()) {
                    payload = &set_instr->value;
                }
                else if (auto list_insert_instr = instr->get_if<Instruction::ArrayInsert>()) {
                    payload = &list_insert_instr->value;
                }

                if (payload && payload->type == Instruction::Payload::Type::Link) {
                    if (auto global_key = mpark::get_if<GlobalKey>(&payload->data.link.target)) {
                        did_modify = promote_global_key(global_key);
                    }
                }
            }
        }

        if (did_modify) {
            ChangesetEncoder::Buffer modified;
            encode_changeset(log, modified);
            util::compression::allocate_and_compress_nonportable(arena, modified, compressed);
            m_arrays->changesets.set(i, BinaryData{compressed.data(), compressed.size()}); // Throws

            uploadable_bytes += modified.size() - decompressed_size;
        }
    }

    root.set(s_progress_uploadable_bytes_iip, RefOrTagged::make_tagged(uploadable_bytes));
}

void ClientHistory::set_group(Group* group, bool updated)
{
    _impl::History::set_group(group, updated);
    if (m_arrays)
        _impl::GroupFriend::set_history_parent(*m_group, m_arrays->root);
}

void ClientHistory::record_current_schema_version()
{
    using gf = _impl::GroupFriend;
    Allocator& alloc = gf::get_alloc(*m_group);
    auto ref = gf::get_history_ref(*m_group);
    REALM_ASSERT(ref != 0);
    Array root{alloc};
    gf::set_history_parent(*m_group, root);
    root.init_from_ref(ref);
    Array schema_versions{alloc};
    schema_versions.set_parent(&root, s_schema_versions_iip);
    schema_versions.init_from_parent();
    version_type snapshot_version = m_db->get_version_of_latest_snapshot();
    record_current_schema_version(schema_versions, snapshot_version); // Throws
}


void ClientHistory::record_current_schema_version(Array& schema_versions, version_type snapshot_version)
{
    static_assert(s_schema_versions_size == 4, "");
    REALM_ASSERT(schema_versions.size() == s_schema_versions_size);

    Allocator& alloc = schema_versions.get_alloc();
    {
        Array sv_schema_versions{alloc};
        sv_schema_versions.set_parent(&schema_versions, s_sv_schema_versions_iip);
        sv_schema_versions.init_from_parent();
        int schema_version = get_client_history_schema_version();
        sv_schema_versions.add(schema_version); // Throws
    }
    {
        Array sv_library_versions{alloc};
        sv_library_versions.set_parent(&schema_versions, s_sv_library_versions_iip);
        sv_library_versions.init_from_parent();
        const char* library_version = REALM_VERSION_STRING;
        std::size_t size = std::strlen(library_version);
        Array value{alloc};
        bool context_flag = false;
        value.create(Array::type_Normal, context_flag, size); // Throws
        _impl::ShallowArrayDestroyGuard adg{&value};
        using uchar = unsigned char;
        for (std::size_t i = 0; i < size; ++i)
            value.set(i, std::int_fast64_t(uchar(library_version[i]))); // Throws
        sv_library_versions.add(std::int_fast64_t(value.get_ref()));    // Throws
        adg.release();                                                  // Ownership transferred to parent array
    }
    {
        Array sv_snapshot_versions{alloc};
        sv_snapshot_versions.set_parent(&schema_versions, s_sv_snapshot_versions_iip);
        sv_snapshot_versions.init_from_parent();
        sv_snapshot_versions.add(std::int_fast64_t(snapshot_version)); // Throws
    }
    {
        Array sv_timestamps{alloc};
        sv_timestamps.set_parent(&schema_versions, s_sv_timestamps_iip);
        sv_timestamps.init_from_parent();
        std::time_t timestamp = std::time(nullptr);
        sv_timestamps.add(std::int_fast64_t(timestamp)); // Throws
    }
}

// Overriding member function in realm::_impl::History
void ClientHistory::update_from_ref_and_version(ref_type ref, version_type version)
{
    if (ref == 0) {
        // No history
        m_ct_history_base_version = version;
        m_sync_history_base_version = version;
        m_arrays.reset();
        m_progress_download = {0, 0};
        return;
    }
    if (REALM_LIKELY(m_arrays)) {
        m_arrays->init_from_ref(ref);
    }
    else {
        m_arrays.emplace(m_db->get_alloc(), m_group, ref);
    }

    m_ct_history_base_version = version - ct_history_size();
    m_sync_history_base_version = version - sync_history_size();
    REALM_ASSERT(m_arrays->changesets.size() == sync_history_size());
    REALM_ASSERT(m_arrays->reciprocal_transforms.size() == sync_history_size());
    REALM_ASSERT(m_arrays->remote_versions.size() == sync_history_size());
    REALM_ASSERT(m_arrays->origin_file_idents.size() == sync_history_size());
    REALM_ASSERT(m_arrays->origin_timestamps.size() == sync_history_size());

    const Array& root = m_arrays->root;
    m_progress_download.server_version =
        version_type(root.get_as_ref_or_tagged(s_progress_download_server_version_iip).get_as_int());
    m_progress_download.last_integrated_client_version =
        version_type(root.get_as_ref_or_tagged(s_progress_download_client_version_iip).get_as_int());
}


// Overriding member function in realm::_impl::History
void ClientHistory::update_from_parent(version_type current_version)
{
    using gf = _impl::GroupFriend;
    ref_type ref = gf::get_history_ref(*m_group);
    update_from_ref_and_version(ref, current_version); // Throws
}


// Overriding member function in realm::_impl::History
void ClientHistory::get_changesets(version_type begin_version, version_type end_version,
                                   BinaryIterator* iterators) const noexcept
{
    REALM_ASSERT(begin_version <= end_version);
    REALM_ASSERT(begin_version >= m_ct_history_base_version);
    REALM_ASSERT(end_version <= m_ct_history_base_version + ct_history_size());
    std::size_t n = to_size_t(end_version - begin_version);
    REALM_ASSERT(n == 0 || m_arrays);
    std::size_t offset = to_size_t(begin_version - m_ct_history_base_version);
    for (std::size_t i = 0; i < n; ++i)
        iterators[i] = BinaryIterator(&m_arrays->ct_history, offset + i);
}


// Overriding member function in realm::_impl::History
void ClientHistory::set_oldest_bound_version(version_type version)
{
    REALM_ASSERT(version >= m_version_of_oldest_bound_snapshot);
    if (version > m_version_of_oldest_bound_snapshot) {
        m_version_of_oldest_bound_snapshot = version;
        trim_ct_history(); // Throws
    }
}

// Overriding member function in realm::_impl::History
void ClientHistory::verify() const
{
#ifdef REALM_DEBUG
    // The size of the continuous transactions history can only be zero when the
    // Realm is in the initial empty state where top-ref is null.
    REALM_ASSERT(ct_history_size() != 0 || m_ct_history_base_version == s_initial_version + 0);

    if (!m_arrays) {
        REALM_ASSERT(m_progress_download.server_version == 0);
        REALM_ASSERT(m_progress_download.last_integrated_client_version == 0);
        return;
    }
    m_arrays->verify();

    auto& root = m_arrays->root;
    version_type progress_download_server_version =
        version_type(root.get_as_ref_or_tagged(s_progress_download_server_version_iip).get_as_int());
    version_type progress_download_client_version =
        version_type(root.get_as_ref_or_tagged(s_progress_download_client_version_iip).get_as_int());
    REALM_ASSERT(progress_download_server_version == m_progress_download.server_version);
    REALM_ASSERT(progress_download_client_version == m_progress_download.last_integrated_client_version);
    REALM_ASSERT(progress_download_client_version <= m_sync_history_base_version + sync_history_size());
    version_type remote_version_of_last_entry = 0;
    if (auto size = sync_history_size())
        remote_version_of_last_entry = m_arrays->remote_versions.get(size - 1);
    REALM_ASSERT(progress_download_server_version >= remote_version_of_last_entry);

    // Verify that there is no cooked history.
    Array cooked_history{m_db->get_alloc()};
    cooked_history.set_parent(&root, s_cooked_history_iip);
    REALM_ASSERT(cooked_history.get_ref_from_parent() == 0);
#endif // REALM_DEBUG
}

ClientHistory::Arrays::Arrays(Allocator& alloc) noexcept
    : root(alloc)
    , ct_history(alloc)
    , changesets(alloc)
    , reciprocal_transforms(alloc)
    , remote_versions(alloc)
    , origin_file_idents(alloc)
    , origin_timestamps(alloc)
{
}

ClientHistory::Arrays::Arrays(DB& db, Group& group)
    : Arrays(db.get_alloc())
{
    auto& alloc = db.get_alloc();
    {
        bool context_flag = false;
        std::size_t size = s_root_size;
        root.create(Array::type_HasRefs, context_flag, size); // Throws
    }
    _impl::DeepArrayDestroyGuard dg{&root};

    ct_history.set_parent(&root, s_ct_history_iip);
    ct_history.create(); // Throws
    changesets.set_parent(&root, s_changesets_iip);
    changesets.create(); // Throws
    reciprocal_transforms.set_parent(&root, s_reciprocal_transforms_iip);
    reciprocal_transforms.create(); // Throws
    remote_versions.set_parent(&root, s_remote_versions_iip);
    remote_versions.create(); // Throws
    origin_file_idents.set_parent(&root, s_origin_file_idents_iip);
    origin_file_idents.create(); // Throws
    origin_timestamps.set_parent(&root, s_origin_timestamps_iip);
    origin_timestamps.create(); // Throws

    { // `schema_versions` table
        Array schema_versions{alloc};
        bool context_flag = false;
        std::size_t size = s_schema_versions_size;
        schema_versions.create(Array::type_HasRefs, context_flag, size); // Throws
        _impl::DeepArrayDestroyGuard adg{&schema_versions};

        auto create_array = [&](NodeHeader::Type type, int ndx_in_parent) {
            MemRef mem = Array::create_empty_array(type, context_flag, alloc);
            ref_type ref = mem.get_ref();
            _impl::DeepArrayRefDestroyGuard ardg{ref, alloc};
            schema_versions.set_as_ref(ndx_in_parent, ref); // Throws
            ardg.release();                                 // Ownership transferred to parent array
        };
        create_array(Array::type_Normal, s_sv_schema_versions_iip);
        create_array(Array::type_HasRefs, s_sv_library_versions_iip);
        create_array(Array::type_Normal, s_sv_snapshot_versions_iip);
        create_array(Array::type_Normal, s_sv_timestamps_iip);

        version_type snapshot_version = db.get_version_of_latest_snapshot();
        record_current_schema_version(schema_versions, snapshot_version);  // Throws
        root.set_as_ref(s_schema_versions_iip, schema_versions.get_ref()); // Throws
        adg.release();                                                     // Ownership transferred to parent array
    }
    _impl::GroupFriend::prepare_history_parent(group, root, Replication::hist_SyncClient,
                                               get_client_history_schema_version(), 0); // Throws
    // Note: gf::prepare_history_parent() also ensures the the root array has a
    // slot for the history ref.
    root.update_parent(); // Throws
    dg.release();
}

ClientHistory::Arrays::Arrays(Allocator& alloc, Group* parent, ref_type ref)
    : Arrays(alloc)
{
    using gf = _impl::GroupFriend;
    root.init_from_ref(ref);
    if (parent)
        gf::set_history_parent(*parent, root);

    ct_history.set_parent(&root, s_ct_history_iip);
    changesets.set_parent(&root, s_changesets_iip);
    reciprocal_transforms.set_parent(&root, s_reciprocal_transforms_iip);
    remote_versions.set_parent(&root, s_remote_versions_iip);
    origin_file_idents.set_parent(&root, s_origin_file_idents_iip);
    origin_timestamps.set_parent(&root, s_origin_timestamps_iip);

    init_from_ref(ref); // Throws

    Array cooked_history{alloc};
    cooked_history.set_parent(&root, s_cooked_history_iip);
    // We should have no cooked history in existing Realms.
    REALM_ASSERT(cooked_history.get_ref_from_parent() == 0);
}

void ClientHistory::Arrays::Arrays::init_from_ref(ref_type ref)
{
    root.init_from_ref(ref);
    REALM_ASSERT(root.size() == s_root_size);
    {
        ref_type ref_2 = root.get_as_ref(s_ct_history_iip);
        ct_history.init_from_ref(ref_2); // Throws
    }
    {
        ref_type ref_2 = root.get_as_ref(s_changesets_iip);
        changesets.init_from_ref(ref_2); // Throws
    }
    {
        ref_type ref_2 = root.get_as_ref(s_reciprocal_transforms_iip);
        reciprocal_transforms.init_from_ref(ref_2); // Throws
    }
    remote_versions.init_from_parent();    // Throws
    origin_file_idents.init_from_parent(); // Throws
    origin_timestamps.init_from_parent();  // Throws
}

void ClientHistory::Arrays::verify() const
{
#ifdef REALM_DEBUG
    root.verify();
    ct_history.verify();
    changesets.verify();
    reciprocal_transforms.verify();
    remote_versions.verify();
    origin_file_idents.verify();
    origin_timestamps.verify();
    REALM_ASSERT(root.size() == s_root_size);
    REALM_ASSERT(reciprocal_transforms.size() == changesets.size());
    REALM_ASSERT(remote_versions.size() == changesets.size());
    REALM_ASSERT(origin_file_idents.size() == changesets.size());
    REALM_ASSERT(origin_timestamps.size() == changesets.size());
#endif // REALM_DEBUG
}

} // namespace realm::sync

realm / realm-core / thomas.goyne_420

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous