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

60.73

/src/realm/sync/transform.cpp

#include <realm/sync/transform.hpp>

#include <realm/sync/changeset_encoder.hpp>
#include <realm/sync/noinst/changeset_index.hpp>
#include <realm/sync/noinst/protocol_codec.hpp>

#if REALM_DEBUG
#include <sstream>
#include <iostream> // std::cerr used for debug tracing
#include <mutex>    // std::unique_lock used for debug tracing
#endif              // REALM_DEBUG

using namespace realm;
using namespace realm::sync;
using namespace realm::util;

namespace {

#if REALM_DEBUG
#if defined(_MSC_VER)
#define TERM_RED ""
#define TERM_YELLOW ""
#define TERM_CYAN ""
#define TERM_MAGENTA ""
#define TERM_GREEN ""
#define TERM_BOLD ""
#define TERM_RESET ""
#else
#define TERM_RED "\x1b[31;22m"
#define TERM_YELLOW "\x1b[33;22m"
#define TERM_CYAN "\x1b[36;22m"
#define TERM_MAGENTA "\x1b[35;22m"
#define TERM_GREEN "\x1b[32;22m"
#define TERM_BOLD "\x1b[1m"
#define TERM_RESET "\x1b[39;49;22m"
#endif
#endif

struct Discriminant {
    timestamp_type timestamp;
    file_ident_type client_file_ident;
    Discriminant(timestamp_type t, file_ident_type p)
        : timestamp(t)
        , client_file_ident(p)
    {
    }

    Discriminant(const Discriminant&) = default;
    Discriminant& operator=(const Discriminant&) = default;

    bool operator<(const Discriminant& other) const
    {
        return timestamp == other.timestamp ? (client_file_ident < other.client_file_ident)
                                            : timestamp < other.timestamp;
    }

    bool operator==(const Discriminant& other) const
    {
        return timestamp == other.timestamp && client_file_ident == other.client_file_ident;
    }
    bool operator!=(const Discriminant& other) const
    {
        return !((*this) == other);
    }
};

struct TransformerImpl;

struct Side {
    TransformerImpl& m_transformer;
    Changeset* m_changeset = nullptr;
    Discriminant m_discriminant;

    bool was_discarded = false;
    bool was_replaced = false;
    size_t m_path_len = 0;

    Side(TransformerImpl& transformer)
        : m_transformer(transformer)
        , m_discriminant(0, 0)
    {
    }

    virtual void skip_tombstones() noexcept = 0;
    virtual void next_instruction() noexcept = 0;
    virtual Instruction& get() noexcept = 0;

    void substitute(const Instruction& instr)
    {
        was_replaced = true;
        get() = instr;
    }

    StringData get_string(StringBufferRange range) const
    {
        // Relying on the transaction parser to only provide valid StringBufferRanges.
        return m_changeset->get_string(range);
    }

    StringData get_string(InternString intern_string) const
    {
        // Rely on the parser having checked the consistency of the interned strings
        return m_changeset->get_string(intern_string);
    }

    InternString intern_string(StringData data) const
    {
        return m_changeset->intern_string(data);
    }

    const Discriminant& timestamp() const
    {
        return m_discriminant;
    }

    InternString adopt_string(const Side& other_side, InternString other_string)
    {
        // FIXME: This needs to change if we choose to compare strings through a
        // mapping of InternStrings.
        StringData string = other_side.get_string(other_string);
        return intern_string(string);
    }

protected:
    void init_with_instruction(const Instruction& instr) noexcept
    {
        was_discarded = false;
        was_replaced = false;
        m_path_len = instr.path_length();
    }
};

struct MajorSide : Side {
    MajorSide(TransformerImpl& transformer)
        : Side(transformer)
    {
    }

    void set_next_changeset(Changeset* changeset) noexcept;
    void discard();
    void prepend(Instruction operation);
    template <class InputIterator>
    void prepend(InputIterator begin, InputIterator end);

    void init_with_instruction(Changeset::iterator position) noexcept
    {
        REALM_ASSERT(position >= m_changeset->begin());
        REALM_ASSERT(position != m_changeset->end());
        m_position = position;
        skip_tombstones();
        REALM_ASSERT(position != m_changeset->end());

        m_discriminant = Discriminant{m_changeset->origin_timestamp, m_changeset->origin_file_ident};

        Side::init_with_instruction(get());
    }

    void skip_tombstones() noexcept final
    {
        while (m_position != m_changeset->end() && !*m_position) {
            ++m_position;
        }
    }

    void next_instruction() noexcept final
    {
        REALM_ASSERT(m_position != m_changeset->end());
        do {
            ++m_position;
        } while (m_position != m_changeset->end() && !*m_position);
    }

    Instruction& get() noexcept final
    {
        return **m_position;
    }

    size_t get_object_ids_in_current_instruction(_impl::ChangesetIndex::GlobalID* ids, size_t max_ids)
    {
        return _impl::get_object_ids_in_instruction(*m_changeset, get(), ids, max_ids);
    }

    Changeset::iterator m_position;
};

struct MinorSide : Side {
    using Position = _impl::ChangesetIndex::RangeIterator;

    MinorSide(TransformerImpl& transformer)
        : Side(transformer)
    {
    }

    void discard();
    void prepend(Instruction operation);
    template <class InputIterator>
    void prepend(InputIterator begin, InputIterator end);

    void substitute(const Instruction& instr)
    {
        was_replaced = true;
        get() = instr;
    }

    Position begin() noexcept
    {
        return Position{m_conflict_ranges};
    }

    Position end() noexcept
    {
        return Position{m_conflict_ranges, Position::end_tag{}};
    }

    void update_changeset_pointer() noexcept
    {
        if (REALM_LIKELY(m_position != end())) {
            m_changeset = m_position.m_outer->first;
        }
        else {
            m_changeset = nullptr;
        }
    }

    void skip_tombstones() noexcept final
    {
        if (m_position != end() && *m_position)
            return;
        skip_tombstones_slow();
    }

    REALM_NOINLINE void skip_tombstones_slow() noexcept
    {
        while (m_position != end() && !*m_position) {
            ++m_position;
        }
        update_changeset_pointer();
    }

    void next_instruction() noexcept final
    {
        REALM_ASSERT(m_position != end());
        ++m_position;
        update_changeset_pointer();
        skip_tombstones();
    }

    Instruction& get() noexcept final
    {
        Instruction* instr = *m_position;
        REALM_ASSERT(instr != nullptr);
        return *instr;
    }

    void init_with_instruction(Position position)
    {
        // REALM_ASSERT(position >= Position(m_conflict_ranges));
        REALM_ASSERT(position != end());
        m_position = position;
        update_changeset_pointer();
        skip_tombstones();
        REALM_ASSERT(position != end());

        m_discriminant = Discriminant{m_changeset->origin_timestamp, m_changeset->origin_file_ident};

        Side::init_with_instruction(get());
    }

    _impl::ChangesetIndex::RangeIterator m_position;
    _impl::ChangesetIndex* m_changeset_index = nullptr;
    _impl::ChangesetIndex::Ranges* m_conflict_ranges = nullptr;
};

#if defined(REALM_DEBUG) // LCOV_EXCL_START Debug utilities

struct MergeTracer {
public:
    Side& m_minor;
    Side& m_major;
    const Changeset& m_minor_log;
    const Changeset& m_major_log;
    Instruction minor_before;
    Instruction major_before;

    // field => pair(original_value, change)
    struct Diff {
        std::map<std::string, std::pair<int64_t, int64_t>> numbers;
        std::map<std::string, std::pair<std::string, std::string>> strings;

        bool empty() const noexcept
        {
            return numbers.empty() && strings.empty();
        }
    };

    explicit MergeTracer(Side& minor, Side& major)
        : m_minor(minor)
        , m_major(major)
        , m_minor_log(*minor.m_changeset)
        , m_major_log(*major.m_changeset)
        , minor_before(minor.get())
        , major_before(major.get())
    {
    }

    struct FieldTracer : sync::Changeset::Reflector::Tracer {
        std::string m_name;
        std::map<std::string, std::string, std::less<>> m_fields;

        const Changeset* m_changeset = nullptr;

        void set_changeset(const Changeset* changeset) override
        {
            m_changeset = changeset;
        }

        StringData get_string(InternString str)
        {
            return m_changeset->get_string(str);
        }

        void name(StringData n) override
        {
            m_name = n;
        }

        void path(StringData n, InternString table, const Instruction::PrimaryKey& pk,
                  util::Optional<InternString> field, const Instruction::Path* path) override
        {
            std::stringstream ss;
            m_changeset->print_path(ss, table, pk, field, path);
            m_fields.emplace(n, ss.str());
        }

        void field(StringData n, InternString str) override
        {
            m_fields.emplace(n, get_string(str));
        }

        void field(StringData n, Instruction::Payload::Type type) override
        {
            m_fields.emplace(n, get_type_name(type));
        }

        void field(StringData n, Instruction::CollectionType type) override
        {
            m_fields.emplace(n, get_collection_type(type));
        }

        void field(StringData n, const Instruction::PrimaryKey& key) override
        {
            auto real_key = m_changeset->get_key(key);
            std::stringstream ss;
            ss << format_pk(real_key);
            m_fields.emplace(n, ss.str());
        }

        void field(StringData n, const Instruction::Payload& value) override
        {
            std::stringstream ss;
            m_changeset->print_value(ss, value);
            m_fields.emplace(n, ss.str());
        }

        void field(StringData n, const Instruction::Path& value) override
        {
            std::stringstream ss;
            m_changeset->print_path(ss, value);
            m_fields.emplace(n, ss.str());
        }

        void field(StringData n, uint32_t value) override
        {
            std::stringstream ss;
            ss << value;
            m_fields.emplace(n, ss.str());
        }
    };

    struct PrintDiffTracer : sync::Changeset::Reflector::Tracer {
        std::ostream& m_os;
        const FieldTracer& m_before;
        bool m_first = true;
        const Changeset* m_changeset = nullptr;

        PrintDiffTracer(std::ostream& os, const FieldTracer& before)
            : m_os(os)
            , m_before(before)
        {
        }

        void set_changeset(const Changeset* changeset) override
        {
            m_changeset = changeset;
        }

        StringData get_string(InternString str) const noexcept
        {
            return m_changeset->get_string(str);
        }

        void name(StringData n) override
        {
            m_os << std::left << std::setw(16) << std::string(n);
        }

        void path(StringData n, InternString table, const Instruction::PrimaryKey& pk,
                  util::Optional<InternString> field, const Instruction::Path* path) override
        {
            std::stringstream ss;
            m_changeset->print_path(ss, table, pk, field, path);
            diff_field(n, ss.str());
        }

        void field(StringData n, InternString str) override
        {
            diff_field(n, get_string(str));
        }

        void field(StringData n, Instruction::Payload::Type type) override
        {
            diff_field(n, get_type_name(type));
        }

        void field(StringData n, Instruction::CollectionType type) override
        {
            diff_field(n, get_collection_type(type));
        }

        void field(StringData n, const Instruction::PrimaryKey& value) override
        {
            std::stringstream ss;
            ss << format_pk(m_changeset->get_key(value));
            diff_field(n, ss.str());
        }

        void field(StringData n, const Instruction::Payload& value) override
        {
            std::stringstream ss;
            m_changeset->print_value(ss, value);
            diff_field(n, ss.str());
        }

        void field(StringData n, const Instruction::Path& value) override
        {
            std::stringstream ss;
            m_changeset->print_path(ss, value);
            diff_field(n, ss.str());
        }

        void field(StringData n, uint32_t value) override
        {
            std::stringstream ss;
            ss << value;
            diff_field(n, ss.str());
        }

        void diff_field(StringData name, std::string value)
        {
            std::stringstream ss;
            ss << name << "=";
            auto it = m_before.m_fields.find(name);
            if (it == m_before.m_fields.end() || it->second == value) {
                ss << value;
            }
            else {
                ss << it->second << "->" << value;
            }
            if (!m_first) {
                m_os << ", ";
            }
            m_os << ss.str();
            m_first = false;
        }
    };

    static void print_instr(std::ostream& os, const Instruction& instr, const Changeset& changeset)
    {
        Changeset::Printer printer{os};
        Changeset::Reflector reflector{printer, changeset};
        instr.visit(reflector);
    }

    bool print_diff(std::ostream& os, bool print_unmodified, const Instruction& before, const Changeset& before_log,
                    Side& side) const
    {
        if (side.was_discarded) {
            print_instr(os, before, before_log);
            os << " (DISCARDED)";
        }
        else if (side.was_replaced) {
            print_instr(os, before, before_log);
            os << " (REPLACED)";
        }
        else {
            Instruction after = side.get();
            if (print_unmodified || (before != after)) {
                FieldTracer before_tracer;
                before_tracer.set_changeset(&before_log);
                PrintDiffTracer after_tracer{os, before_tracer};
                Changeset::Reflector before_reflector{before_tracer, *side.m_changeset};
                Changeset::Reflector after_reflector{after_tracer, *side.m_changeset};
                before.visit(before_reflector);
                after.visit(after_reflector); // This prints the diff'ed instruction
            }
            else {
                os << "(=)";
            }
        }
        return true;
    }

    void print_diff(std::ostream& os, bool print_unmodified) const
    {
        bool must_print_minor = m_minor.was_discarded || m_minor.was_replaced;
        if (!must_print_minor) {
            Instruction minor_after = m_minor.get();
            must_print_minor = (minor_before != minor_after);
        }
        bool must_print_major = m_major.was_discarded || m_major.was_replaced;
        if (!must_print_major) {
            Instruction major_after = m_major.get();
            must_print_major = (major_before != major_after);
        }
        bool must_print = (print_unmodified || must_print_minor || must_print_major);
        if (must_print) {
            std::stringstream ss_minor;
            std::stringstream ss_major;

            print_diff(ss_minor, true, minor_before, m_minor_log, m_minor);
            print_diff(ss_major, print_unmodified, major_before, m_major_log, m_major);

            os << std::left << std::setw(80) << ss_minor.str();
            os << ss_major.str() << "\n";
        }
    }

    void pad_or_ellipsis(std::ostream& os, const std::string& str, int width) const
    {
        // FIXME: Does not work with UTF-8.
        if (str.size() > size_t(width)) {
            os << str.substr(0, width - 1) << "~";
        }
        else {
            os << std::left << std::setw(width) << str;
        }
    }
};
#endif // LCOV_EXCL_STOP REALM_DEBUG

struct TransformerImpl {
    MajorSide m_major_side;
    MinorSide m_minor_side;
    MinorSide::Position m_minor_end;
    bool m_trace;

    TransformerImpl(bool trace)
        : m_major_side{*this}
        , m_minor_side{*this}
        , m_trace{trace}
    {
    }

    void transform()
    {
        m_major_side.m_position = m_major_side.m_changeset->begin();
        m_major_side.skip_tombstones();

        while (m_major_side.m_position != m_major_side.m_changeset->end()) {
            m_major_side.init_with_instruction(m_major_side.m_position);

            set_conflict_ranges();
            m_minor_end = m_minor_side.end();
            m_minor_side.m_position = m_minor_side.begin();
            transform_major();

            if (!m_major_side.was_discarded)
                // Discarding the instruction moves to the next one.
                m_major_side.next_instruction();
            m_major_side.skip_tombstones();
        }
    }

    _impl::ChangesetIndex::Ranges* get_conflict_ranges_for_instruction(const Instruction& instr)
    {
        _impl::ChangesetIndex& index = *m_minor_side.m_changeset_index;

        if (_impl::is_schema_change(instr)) {
            ///
            /// CONFLICT GROUP: Everything touching that class
            ///
            auto ranges = index.get_everything();
            if (!ranges->empty()) {
#if REALM_DEBUG // LCOV_EXCL_START
                if (m_trace) {
                    std::cerr << TERM_RED << "Conflict group: Everything (due to schema change)\n" << TERM_RESET;
                }
#endif // REALM_DEBUG LCOV_EXCL_STOP
            }
            return ranges;
        }
        else {
            ///
            /// CONFLICT GROUP: Everything touching the involved objects,
            /// including schema changes.
            ///
            _impl::ChangesetIndex::GlobalID major_ids[2];
            size_t num_major_ids = m_major_side.get_object_ids_in_current_instruction(major_ids, 2);
            REALM_ASSERT(num_major_ids <= 2);
            REALM_ASSERT(num_major_ids >= 1);
#if REALM_DEBUG // LCOV_EXCL_START
            if (m_trace) {
                std::cerr << TERM_RED << "Conflict group: ";
                if (num_major_ids == 0) {
                    std::cerr << "(nothing - no object references)";
                }
                for (size_t i = 0; i < num_major_ids; ++i) {
                    std::cerr << major_ids[i].table_name << "[" << format_pk(major_ids[i].object_id) << "]";
                    if (i + 1 != num_major_ids)
                        std::cerr << ", ";
                }
                std::cerr << "\n" << TERM_RESET;
            }
#endif // REALM_DEBUG LCOV_EXCL_STOP
            auto ranges = index.get_modifications_for_object(major_ids[0]);
            if (num_major_ids == 2) {
                // Check that the index has correctly joined the ranges for the
                // two object IDs.
                REALM_ASSERT(ranges == index.get_modifications_for_object(major_ids[1]));
            }
            return ranges;
        }
    }

    void set_conflict_ranges()
    {
        const auto& major_instr = m_major_side.get();
        m_minor_side.m_conflict_ranges = get_conflict_ranges_for_instruction(major_instr);
        /* m_minor_side.m_changeset_index->verify(); */
    }

    void set_next_major_changeset(Changeset* changeset) noexcept
    {
        m_major_side.m_changeset = changeset;
        m_major_side.m_position = changeset->begin();
        m_major_side.skip_tombstones();
    }

    void discard_major()
    {
        m_major_side.m_position = m_major_side.m_changeset->erase_stable(m_major_side.m_position);
        m_major_side.was_discarded = true; // This terminates the loop in transform_major();
        m_major_side.m_changeset->set_dirty(true);
    }

    void discard_minor()
    {
        m_minor_side.was_discarded = true;
        m_minor_side.m_position = m_minor_side.m_changeset_index->erase_instruction(m_minor_side.m_position);
        m_minor_side.m_changeset->set_dirty(true);
        m_minor_side.update_changeset_pointer();
    }

    template <class InputIterator>
    void prepend_major(InputIterator instr_begin, InputIterator instr_end)
    {
        REALM_ASSERT(*m_major_side.m_position); // cannot prepend a tombstone
        auto insert_position = m_major_side.m_position;
        m_major_side.m_position = m_major_side.m_changeset->insert_stable(insert_position, instr_begin, instr_end);
        m_major_side.m_changeset->set_dirty(true);
        size_t num_prepended = instr_end - instr_begin;
        transform_prepended_major(num_prepended);
    }

    void prepend_major(Instruction instr)
    {
        const Instruction* begin = &instr;
        const Instruction* end = begin + 1;
        prepend_major(begin, end);
    }

    template <class InputIterator>
    void prepend_minor(InputIterator instr_begin, InputIterator instr_end)
    {
        REALM_ASSERT(*m_minor_side.m_position); // cannot prepend a tombstone
        auto insert_position = m_minor_side.m_position.m_pos;
        m_minor_side.m_position.m_pos =
            m_minor_side.m_changeset->insert_stable(insert_position, instr_begin, instr_end);
        m_minor_side.m_changeset->set_dirty(true);
        size_t num_prepended = instr_end - instr_begin;
        // Go back to the instruction that initiated this prepend
        for (size_t i = 0; i < num_prepended; ++i) {
            ++m_minor_side.m_position;
        }
        REALM_ASSERT(m_minor_end == m_minor_side.end());
    }

    void prepend_minor(Instruction instr)
    {
        const Instruction* begin = &instr;
        const Instruction* end = begin + 1;
        prepend_minor(begin, end);
    }

    void transform_prepended_major(size_t num_prepended)
    {
        auto orig_major_was_discarded = m_major_side.was_discarded;
        auto orig_major_path_len = m_major_side.m_path_len;

        // Reset 'was_discarded', as it should refer to the prepended
        // instructions in the below, not the instruction that instigated the
        // prepend.
        m_major_side.was_discarded = false;
        REALM_ASSERT(m_major_side.m_position != m_major_side.m_changeset->end());

#if defined(REALM_DEBUG) // LCOV_EXCL_START
        if (m_trace) {
            std::cerr << std::setw(80) << " ";
            MergeTracer::print_instr(std::cerr, m_major_side.get(), *m_major_side.m_changeset);
            std::cerr << " (PREPENDED " << num_prepended << ")\n";
        }
#endif // REALM_DEBUG LCOV_EXCL_STOP

        for (size_t i = 0; i < num_prepended; ++i) {
            auto orig_minor_index = m_minor_side.m_position;
            auto orig_minor_was_discarded = m_minor_side.was_discarded;
            auto orig_minor_was_replaced = m_minor_side.was_replaced;
            auto orig_minor_path_len = m_minor_side.m_path_len;

            // Skip the instruction that initiated this prepend.
            if (!m_minor_side.was_discarded) {
                // Discarding an instruction moves to the next.
                m_minor_side.next_instruction();
            }

            REALM_ASSERT(m_major_side.m_position != m_major_side.m_changeset->end());
            m_major_side.init_with_instruction(m_major_side.m_position);
            REALM_ASSERT(!m_major_side.was_discarded);
            REALM_ASSERT(m_major_side.m_position != m_major_side.m_changeset->end());
            transform_major();
            if (!m_major_side.was_discarded) {
                // Discarding an instruction moves to the next.
                m_major_side.next_instruction();
            }
            REALM_ASSERT(m_major_side.m_position != m_major_side.m_changeset->end());

            m_minor_side.m_position = orig_minor_index;
            m_minor_side.was_discarded = orig_minor_was_discarded;
            m_minor_side.was_replaced = orig_minor_was_replaced;
            m_minor_side.m_path_len = orig_minor_path_len;
            m_minor_side.update_changeset_pointer();
        }

#if defined(REALM_DEBUG) // LCOV_EXCL_START
        if (m_trace) {
            std::cerr << TERM_CYAN << "(end transform of prepended major)\n" << TERM_RESET;
        }
#endif // REALM_DEBUG LCOV_EXCL_STOP

        m_major_side.was_discarded = orig_major_was_discarded;
        m_major_side.m_path_len = orig_major_path_len;
    }

    void transform_major()
    {
        m_minor_side.skip_tombstones();

#if defined(REALM_DEBUG) // LCOV_EXCL_START Debug tracing
        const bool print_noop_merges = false;
        bool new_major = true; // print an instruction every time we go to the next major regardless
#endif                         // LCOV_EXCL_STOP REALM_DEBUG

        while (m_minor_side.m_position != m_minor_end) {
            m_minor_side.init_with_instruction(m_minor_side.m_position);

#if defined(REALM_DEBUG) // LCOV_EXCL_START Debug tracing
            if (m_trace) {
                MergeTracer tracer{m_minor_side, m_major_side};
                merge_instructions(m_major_side, m_minor_side);
                if (new_major)
                    std::cerr << TERM_CYAN << "\n(new major round)\n" << TERM_RESET;
                tracer.print_diff(std::cerr, new_major || print_noop_merges);
                new_major = false;
            }
            else {
#endif // LCOV_EXCL_STOP REALM_DEBUG
                merge_instructions(m_major_side, m_minor_side);
#if defined(REALM_DEBUG) // LCOV_EXCL_START
            }
#endif // LCOV_EXCL_STOP REALM_DEBUG

            if (m_major_side.was_discarded)
                break;
            if (!m_minor_side.was_discarded)
                // Discarding an instruction moves to the next one.
                m_minor_side.next_instruction();
            m_minor_side.skip_tombstones();
        }
    }

    void merge_instructions(MajorSide& left, MinorSide& right);
};

void MajorSide::set_next_changeset(Changeset* changeset) noexcept
{
    m_transformer.set_next_major_changeset(changeset);
}
void MajorSide::discard()
{
    m_transformer.discard_major();
}
void MajorSide::prepend(Instruction operation)
{
    m_transformer.prepend_major(std::move(operation));
}
template <class InputIterator>
void MajorSide::prepend(InputIterator begin, InputIterator end)
{
    m_transformer.prepend_major(std::move(begin), std::move(end));
}
void MinorSide::discard()
{
    m_transformer.discard_minor();
}
void MinorSide::prepend(Instruction operation)
{
    m_transformer.prepend_minor(std::move(operation));
}
template <class InputIterator>
void MinorSide::prepend(InputIterator begin, InputIterator end)
{
    m_transformer.prepend_minor(std::move(begin), std::move(end));
}

REALM_NORETURN void throw_bad_merge(std::string msg)
{
    throw sync::TransformError{std::move(msg)};
}

template <class... Params>
REALM_NORETURN void bad_merge(const char* msg, Params&&... params)
{
    throw_bad_merge(util::format(msg, std::forward<Params>(params)...));
}

REALM_NORETURN void bad_merge(Side& side, Instruction::PathInstruction instr, const std::string& msg)
{
    std::stringstream ss;
    side.m_changeset->print_path(ss, instr.table, instr.object, instr.field, &instr.path);
    bad_merge("%1 (instruction target: %2). Please contact support.", msg, ss.str());
}

template <class LeftInstruction, class RightInstruction, class Enable = void>
struct Merge;
template <class Outer>
struct MergeNested;

struct MergeUtils {
    MergeUtils(Side& left_side, Side& right_side)
        : m_left_side(left_side)
        , m_right_side(right_side)
    {
    }

    // CAUTION: All of these utility functions rely implicitly on the "left" and
    // "right" arguments corresponding to the left and right sides. If the
    // arguments are switched, mayhem ensues.

    bool same_string(InternString left, InternString right) const noexcept
    {
        // FIXME: Optimize string comparison by building a map of InternString values up front.
        return m_left_side.m_changeset->get_string(left) == m_right_side.m_changeset->get_string(right);
    }

    bool same_key(const Instruction::PrimaryKey& left, const Instruction::PrimaryKey& right) const noexcept
    {
        // FIXME: Once we can compare string by InternString map lookups,
        // compare the string components of the keys using that.
        PrimaryKey left_key = m_left_side.m_changeset->get_key(left);
        PrimaryKey right_key = m_right_side.m_changeset->get_key(right);
        return left_key == right_key;
    }

    bool same_payload(const Instruction::Payload& left, const Instruction::Payload& right)
    {
        using Type = Instruction::Payload::Type;

        if (left.type != right.type)
            return false;

        switch (left.type) {
            case Type::Null:
                return true;
            case Type::Erased:
                return true;
            case Type::Set:
                return true;
            case Type::List:
                return true;
            case Type::Dictionary:
                return true;
            case Type::ObjectValue:
                return true;
            case Type::GlobalKey:
                return left.data.key == right.data.key;
            case Type::Int:
                return left.data.integer == right.data.integer;
            case Type::Bool:
                return left.data.boolean == right.data.boolean;
            case Type::String:
                return m_left_side.get_string(left.data.str) == m_right_side.get_string(right.data.str);
            case Type::Binary:
                return m_left_side.get_string(left.data.binary) == m_right_side.get_string(right.data.binary);
            case Type::Timestamp:
                return left.data.timestamp == right.data.timestamp;
            case Type::Float:
                return left.data.fnum == right.data.fnum;
            case Type::Double:
                return left.data.dnum == right.data.dnum;
            case Type::Decimal:
                return left.data.decimal == right.data.decimal;
            case Type::Link: {
                if (!same_key(left.data.link.target, right.data.link.target)) {
                    return false;
                }
                auto left_target = m_left_side.get_string(left.data.link.target_table);
                auto right_target = m_right_side.get_string(right.data.link.target_table);
                return left_target == right_target;
            }
            case Type::ObjectId:
                return left.data.object_id == right.data.object_id;
            case Type::UUID:
                return left.data.uuid == right.data.uuid;
        }

        bad_merge("Invalid payload type in instruction");
    }

    bool same_path_element(const Instruction::Path::Element& left,
                           const Instruction::Path::Element& right) const noexcept
    {
        auto pred = util::overload{
            [](uint32_t lhs, uint32_t rhs) {
                return lhs == rhs;
            },
            [this](InternString lhs, InternString rhs) {
                return same_string(lhs, rhs);
            },
            // FIXME: Paths contain incompatible element types. Should we raise an error here?
            [](InternString, uint32_t) {
                return false;
            },
            [](uint32_t, InternString) {
                return false;
            },
        };
        return mpark::visit(pred, left, right);
    }

    bool same_path(const Instruction::Path& left, const Instruction::Path& right) const noexcept
    {
        if (left.size() == right.size()) {
            for (size_t i = 0; i < left.size(); ++i) {
                if (!same_path_element(left[i], right[i])) {
                    return false;
                }
            }
            return true;
        }
        return false;
    }

    bool same_table(const Instruction::TableInstruction& left,
                    const Instruction::TableInstruction& right) const noexcept
    {
        return same_string(left.table, right.table);
    }

    bool same_object(const Instruction::ObjectInstruction& left,
                     const Instruction::ObjectInstruction& right) const noexcept
    {
        return same_table(left, right) && same_key(left.object, right.object);
    }

    template <class Left, class Right>
    bool same_column(const Left& left, const Right& right) const noexcept
    {
        if constexpr (std::is_convertible_v<const Right&, const Instruction::PathInstruction&>) {
            const Instruction::PathInstruction& rhs = right;
            return same_table(left, right) && same_string(left.field, rhs.field);
        }
        else if constexpr (std::is_convertible_v<const Right&, const Instruction::ObjectInstruction&>) {
            // CreateObject/EraseObject do not have a column built in.
            return false;
        }
        else {
            return same_table(left, right) && same_string(left.field, right.field);
        }
    }

    bool same_field(const Instruction::PathInstruction& left,
                    const Instruction::PathInstruction& right) const noexcept
    {
        return same_object(left, right) && same_string(left.field, right.field);
    }

    bool same_path(const Instruction::PathInstruction& left, const Instruction::PathInstruction& right) const noexcept
    {
        return same_field(left, right) && same_path(left.path, right.path);
    }

    bool same_container(const Instruction::Path& left, const Instruction::Path& right) const noexcept
    {
        // The instructions refer to the same container if the paths have the
        // same length, and elements [0..n-1] are equal (so the last element is
        // disregarded). If the path length is 1, this counts as referring to
        // the same container.
        if (left.size() == right.size()) {
            if (left.size() == 0)
                return true;

            for (size_t i = 0; i < left.size() - 1; ++i) {
                if (!same_path_element(left[i], right[i])) {
                    return false;
                }
            }
            return true;
        }
        return false;
    }

    bool same_container(const Instruction::PathInstruction& left,
                        const Instruction::PathInstruction& right) const noexcept
    {
        return same_field(left, right) && same_container(left.path, right.path);
    }

    // NOTE: `is_prefix_of()` should only return true if the left is a strictly
    // shorter path than the right, and the entire left path is the initial
    // sequence of the right.

    bool is_prefix_of(const Instruction::ObjectInstruction&, const Instruction::TableInstruction&) const noexcept
    {
        // Right side is a schema instruction.
        return false;
    }

    bool is_prefix_of(const Instruction::ObjectInstruction& left,
                      const Instruction::PathInstruction& right) const noexcept
    {
        return same_object(left, right);
    }

    // Returns the next path element if the first path is a parent of the second path.
    // Example:
    //  * is_prefix_of(field1.123.field2, field1.123.field2.456) = 456
    //  * is_prefix_of(field1.123.field2, field1.123.field3.456) = {}

    std::optional<Instruction::Path::Element> is_prefix_of(const Instruction::PathInstruction&,
                                                           const Instruction::TableInstruction&) const noexcept
    {
        // Path instructions can never be full prefixes of table-level instructions. Note that this also covers
        // ObjectInstructions.
        return {};
    }

    std::optional<Instruction::Path::Element> is_prefix_of(const Instruction::PathInstruction& left,
                                                           const Instruction::PathInstruction& right) const noexcept
    {
        if (left.path.size() < right.path.size() && same_field(left, right)) {
            for (size_t i = 0; i < left.path.size(); ++i) {
                if (!same_path_element(left.path[i], right.path[i])) {
                    return {};
                }
            }
            return right.path[left.path.size()];
        }
        return {};
    }

    // True if the left side is an instruction that touches a container within
    // right's path. Equivalent to is_prefix_of, except the last element (the
    // index) is not considered.
    bool is_container_prefix_of(const Instruction::PathInstruction& left,
                                const Instruction::PathInstruction& right) const
    {
        if (left.path.size() != 0 && left.path.size() < right.path.size() && same_field(left, right)) {
            for (size_t i = 0; i < left.path.size() - 1; ++i) {
                if (!same_path_element(left.path[i], right.path[i])) {
                    return false;
                }
            }
            return true;
        }
        return false;
    }

    bool is_container_prefix_of(const Instruction::PathInstruction&, const Instruction::TableInstruction&) const
    {
        return false;
    }

    // When the left side has a shorter path, get the path component on the
    // right that corresponds to the last component on the left.
    //
    // Note that this will only be used in the context of array indices, because
    // those are the only path components that are modified during OT.
    uint32_t& corresponding_index_in_path(const Instruction::PathInstruction& left,
                                          Instruction::PathInstruction& right) const
    {
        REALM_ASSERT(left.path.size() != 0);
        REALM_ASSERT(left.path.size() < right.path.size());
        REALM_ASSERT(mpark::holds_alternative<uint32_t>(left.path.back()));
        size_t index = left.path.size() - 1;
        if (!mpark::holds_alternative<uint32_t>(right.path[index])) {
            bad_merge("Inconsistent paths");
        }
        return mpark::get<uint32_t>(right.path[index]);
    }

    uint32_t& corresponding_index_in_path(const Instruction::PathInstruction&,
                                          const Instruction::TableInstruction&) const
    {
        // A path instruction can never have a shorter path than something that
        // isn't a PathInstruction.
        REALM_UNREACHABLE();
    }

    void merge_get_vs_move(uint32_t& get_ndx, const uint32_t& move_from_ndx,
                           const uint32_t& move_to_ndx) const noexcept
    {
        if (get_ndx == move_from_ndx) {
            // CONFLICT: Update of a moved element.
            //
            // RESOLUTION: On the left side, use the MOVE operation to transform the
            // UPDATE operation received from the right side.
            get_ndx = move_to_ndx; // --->
        }
        else {
            // Right update vs left remove
            if (get_ndx > move_from_ndx) {
                get_ndx -= 1; // --->
            }
            // Right update vs left insert
            if (get_ndx >= move_to_ndx) {
                get_ndx += 1; // --->
            }
        }
    }

protected:
    Side& m_left_side;
    Side& m_right_side;
};

template <class LeftInstruction, class RightInstruction>
struct MergeBase : MergeUtils {
    static const Instruction::Type A = Instruction::GetInstructionType<LeftInstruction>::value;
    static const Instruction::Type B = Instruction::GetInstructionType<RightInstruction>::value;
    static_assert(A >= B, "A < B. Please reverse the order of instruction types. :-)");

    MergeBase(Side& left_side, Side& right_side)
        : MergeUtils(left_side, right_side)
    {
    }
    MergeBase(MergeBase&&) = delete;
};

#define DEFINE_MERGE(A, B)                                                                                           \
    template <>                                                                                                      \
    struct Merge<A, B> {                                                                                             \
        template <class LeftSide, class RightSide>                                                                   \
        struct DoMerge : MergeBase<A, B> {                                                                           \
            A& left;                                                                                                 \
            B& right;                                                                                                \
            LeftSide& left_side;                                                                                     \
            RightSide& right_side;                                                                                   \
            DoMerge(A& left, B& right, LeftSide& left_side, RightSide& right_side)                                   \
                : MergeBase<A, B>(left_side, right_side)                                                             \
                , left(left)                                                                                         \
                , right(right)                                                                                       \
                , left_side(left_side)                                                                               \
                , right_side(right_side)                                                                             \
            {                                                                                                        \
            }                                                                                                        \
            void do_merge();                                                                                         \
        };                                                                                                           \
        template <class LeftSide, class RightSide>                                                                   \
        static inline void merge(A& left, B& right, LeftSide& left_side, RightSide& right_side)                      \
        {                                                                                                            \
            DoMerge<LeftSide, RightSide> do_merge{left, right, left_side, right_side};                               \
            do_merge.do_merge();                                                                                     \
        }                                                                                                            \
    };                                                                                                               \
    template <class LeftSide, class RightSide>                                                                       \
    void Merge<A, B>::DoMerge<LeftSide, RightSide>::do_merge()

#define DEFINE_MERGE_NOOP(A, B)                                                                                      \
    template <>                                                                                                      \
    struct Merge<A, B> {                                                                                             \
        static const Instruction::Type left_type = Instruction::GetInstructionType<A>::value;                        \
        static const Instruction::Type right_type = Instruction::GetInstructionType<B>::value;                       \
        static_assert(left_type >= right_type,                                                                       \
                      "left_type < right_type. Please reverse the order of instruction types. :-)");                 \
        template <class LeftSide, class RightSide>                                                                   \
        static inline void merge(A&, B&, LeftSide&, RightSide&)                                                      \
        { /* Do nothing */                                                                                           \
        }                                                                                                            \
    }


#define DEFINE_NESTED_MERGE(A)                                                                                       \
    template <>                                                                                                      \
    struct MergeNested<A> {                                                                                          \
        template <class B, class OuterSide, class InnerSide>                                                         \
        struct DoMerge : MergeUtils {                                                                                \
            A& outer;                                                                                                \
            B& inner;                                                                                                \
            OuterSide& outer_side;                                                                                   \
            InnerSide& inner_side;                                                                                   \
            DoMerge(A& outer, B& inner, OuterSide& outer_side, InnerSide& inner_side)                                \
                : MergeUtils(outer_side, inner_side)                                                                 \
                , outer(outer)                                                                                       \
                , inner(inner)                                                                                       \
                , outer_side(outer_side)                                                                             \
                , inner_side(inner_side)                                                                             \
            {                                                                                                        \
            }                                                                                                        \
            void do_merge();                                                                                         \
        };                                                                                                           \
        template <class B, class OuterSide, class InnerSide>                                                         \
        static inline void merge(A& outer, B& inner, OuterSide& outer_side, InnerSide& inner_side)                   \
        {                                                                                                            \
            DoMerge<B, OuterSide, InnerSide> do_merge{outer, inner, outer_side, inner_side};                         \
            do_merge.do_merge();                                                                                     \
        }                                                                                                            \
    };                                                                                                               \
    template <class B, class OuterSide, class InnerSide>                                                             \
    void MergeNested<A>::DoMerge<B, OuterSide, InnerSide>::do_merge()

#define DEFINE_NESTED_MERGE_NOOP(A)                                                                                  \
    template <>                                                                                                      \
    struct MergeNested<A> {                                                                                          \
        template <class B, class OuterSide, class InnerSide>                                                         \
        static inline void merge(const A&, const B&, const OuterSide&, const InnerSide&)                             \
        { /* Do nothing */                                                                                           \
        }                                                                                                            \
    }

// Implementation that reverses order.
template <class A, class B>
struct Merge<A, B,
             typename std::enable_if<(Instruction::GetInstructionType<A>::value <
                                      Instruction::GetInstructionType<B>::value)>::type> {
    template <class LeftSide, class RightSide>
    static void merge(A& left, B& right, LeftSide& left_side, RightSide& right_side)
    {
        Merge<B, A>::merge(right, left, right_side, left_side);
    }
};

///
///  GET READY!
///
///  Realm supports 14 instructions at the time of this writing. Each
///  instruction type needs one rule for each other instruction type. We only
///  define one rule to handle each combination (A vs B and B vs A are handle by
///  a single rule).
///
///  This gives (14 * (14 + 1)) / 2 = 105 merge rules below.
///
///  Merge rules are ordered such that the second instruction type is always of
///  a lower enum value than the first.
///
///  Nested merge rules apply when one instruction has a strictly longer path
///  than another. All instructions that have a path of the same length will
///  meet each other through regular merge rules, regardless of whether they
///  share a prefix.
///


/// AddTable rules

DEFINE_NESTED_MERGE_NOOP(Instruction::AddTable);

DEFINE_MERGE(Instruction::AddTable, Instruction::AddTable)
{
    if (same_table(left, right)) {
        StringData left_name = left_side.get_string(left.table);
        if (auto left_spec = mpark::get_if<Instruction::AddTable::TopLevelTable>(&left.type)) {
            if (auto right_spec = mpark::get_if<Instruction::AddTable::TopLevelTable>(&right.type)) {
                StringData left_pk_name = left_side.get_string(left_spec->pk_field);
                StringData right_pk_name = right_side.get_string(right_spec->pk_field);
                if (left_pk_name != right_pk_name) {
                    bad_merge(
                        "Schema mismatch: '%1' has primary key '%2' on one side, but primary key '%3' on the other.",
                        left_name, left_pk_name, right_pk_name);
                }

                if (left_spec->pk_type != right_spec->pk_type) {
                    bad_merge("Schema mismatch: '%1' has primary key '%2', which is of type %3 on one side and type "
                              "%4 on the other.",
                              left_name, left_pk_name, get_type_name(left_spec->pk_type),
                              get_type_name(right_spec->pk_type));
                }

                if (left_spec->pk_nullable != right_spec->pk_nullable) {
                    bad_merge("Schema mismatch: '%1' has primary key '%2', which is nullable on one side, but not "
                              "the other.",
                              left_name, left_pk_name);
                }

                if (left_spec->is_asymmetric != right_spec->is_asymmetric) {
                    bad_merge("Schema mismatch: '%1' is asymmetric on one side, but not on the other.", left_name);
                }
            }
            else {
                bad_merge("Schema mismatch: '%1' has a primary key on one side, but not on the other.", left_name);
            }
        }
        else if (mpark::get_if<Instruction::AddTable::EmbeddedTable>(&left.type)) {
            if (!mpark::get_if<Instruction::AddTable::EmbeddedTable>(&right.type)) {
                bad_merge("Schema mismatch: '%1' is an embedded table on one side, but not the other.", left_name);
            }
        }

        // Names are the same, PK presence is the same, and if there is a primary
        // key, its name, type, and nullability are the same. Discard both sides.
        left_side.discard();
        right_side.discard();
        return;
    }
}

DEFINE_MERGE(Instruction::EraseTable, Instruction::AddTable)
{
    if (same_table(left, right)) {
        right_side.discard();
    }
}

DEFINE_MERGE_NOOP(Instruction::CreateObject, Instruction::AddTable);
DEFINE_MERGE_NOOP(Instruction::EraseObject, Instruction::AddTable);
DEFINE_MERGE_NOOP(Instruction::Update, Instruction::AddTable);
DEFINE_MERGE_NOOP(Instruction::AddInteger, Instruction::AddTable);
DEFINE_MERGE_NOOP(Instruction::AddColumn, Instruction::AddTable);
DEFINE_MERGE_NOOP(Instruction::EraseColumn, Instruction::AddTable);
DEFINE_MERGE_NOOP(Instruction::ArrayInsert, Instruction::AddTable);
DEFINE_MERGE_NOOP(Instruction::ArrayMove, Instruction::AddTable);
DEFINE_MERGE_NOOP(Instruction::ArrayErase, Instruction::AddTable);
DEFINE_MERGE_NOOP(Instruction::Clear, Instruction::AddTable);
DEFINE_MERGE_NOOP(Instruction::SetInsert, Instruction::AddTable);
DEFINE_MERGE_NOOP(Instruction::SetErase, Instruction::AddTable);

/// EraseTable rules

DEFINE_NESTED_MERGE(Instruction::EraseTable)
{
    if (same_table(outer, inner)) {
        inner_side.discard();
    }
}

DEFINE_MERGE(Instruction::EraseTable, Instruction::EraseTable)
{
    if (same_table(left, right)) {
        left_side.discard();
        right_side.discard();
    }
}

// Handled by nesting rule.
DEFINE_MERGE_NOOP(Instruction::CreateObject, Instruction::EraseTable);
DEFINE_MERGE_NOOP(Instruction::EraseObject, Instruction::EraseTable);
DEFINE_MERGE_NOOP(Instruction::Update, Instruction::EraseTable);
DEFINE_MERGE_NOOP(Instruction::AddInteger, Instruction::EraseTable);

DEFINE_MERGE(Instruction::AddColumn, Instruction::EraseTable)
{
    // AddColumn on an erased table handled by nesting.

    if (left.type == Instruction::Payload::Type::Link && same_string(left.link_target_table, right.table)) {
        // Erase of a table where the left side adds a link column targeting it.
        Instruction::EraseColumn erase_column;
        erase_column.table = right_side.adopt_string(left_side, left.table);
        erase_column.field = right_side.adopt_string(left_side, left.field);
        right_side.prepend(erase_column);
        left_side.discard();
    }
}

// Handled by nested rule.
DEFINE_MERGE_NOOP(Instruction::EraseColumn, Instruction::EraseTable);
DEFINE_MERGE_NOOP(Instruction::ArrayInsert, Instruction::EraseTable);
DEFINE_MERGE_NOOP(Instruction::ArrayMove, Instruction::EraseTable);
DEFINE_MERGE_NOOP(Instruction::ArrayErase, Instruction::EraseTable);
DEFINE_MERGE_NOOP(Instruction::Clear, Instruction::EraseTable);
DEFINE_MERGE_NOOP(Instruction::SetInsert, Instruction::EraseTable);
DEFINE_MERGE_NOOP(Instruction::SetErase, Instruction::EraseTable);


/// CreateObject rules

// CreateObject cannot interfere with instructions that have a longer path.
DEFINE_NESTED_MERGE_NOOP(Instruction::CreateObject);

// CreateObject is idempotent.
DEFINE_MERGE_NOOP(Instruction::CreateObject, Instruction::CreateObject);

DEFINE_MERGE(Instruction::EraseObject, Instruction::CreateObject)
{
    if (same_object(left, right)) {
        // CONFLICT: Create and Erase of the same object.
        //
        // RESOLUTION: Erase always wins.
        right_side.discard();
    }
}

DEFINE_MERGE_NOOP(Instruction::Update, Instruction::CreateObject);
DEFINE_MERGE_NOOP(Instruction::AddInteger, Instruction::CreateObject);
DEFINE_MERGE_NOOP(Instruction::AddColumn, Instruction::CreateObject);
DEFINE_MERGE_NOOP(Instruction::EraseColumn, Instruction::CreateObject);
DEFINE_MERGE_NOOP(Instruction::ArrayInsert, Instruction::CreateObject);
DEFINE_MERGE_NOOP(Instruction::ArrayMove, Instruction::CreateObject);
DEFINE_MERGE_NOOP(Instruction::ArrayErase, Instruction::CreateObject);
DEFINE_MERGE_NOOP(Instruction::Clear, Instruction::CreateObject);
DEFINE_MERGE_NOOP(Instruction::SetInsert, Instruction::CreateObject);
DEFINE_MERGE_NOOP(Instruction::SetErase, Instruction::CreateObject);


/// Erase rules

DEFINE_NESTED_MERGE(Instruction::EraseObject)
{
    if (is_prefix_of(outer, inner)) {
        // Erase always wins.
        inner_side.discard();
    }
}

DEFINE_MERGE(Instruction::EraseObject, Instruction::EraseObject)
{
    if (same_object(left, right)) {
        // We keep the most recent erase. This prevents the situation where a
        // high number of EraseObject instructions in the past trumps a
        // Erase-Create pair in the future.
        if (right_side.timestamp() < left_side.timestamp()) {
            right_side.discard();
        }
        else {
            left_side.discard();
        }
    }
}

// Handled by nested merge.
DEFINE_MERGE_NOOP(Instruction::Update, Instruction::EraseObject);
DEFINE_MERGE_NOOP(Instruction::AddInteger, Instruction::EraseObject);
DEFINE_MERGE_NOOP(Instruction::AddColumn, Instruction::EraseObject);
DEFINE_MERGE_NOOP(Instruction::EraseColumn, Instruction::EraseObject);
DEFINE_MERGE_NOOP(Instruction::ArrayInsert, Instruction::EraseObject);
DEFINE_MERGE_NOOP(Instruction::ArrayMove, Instruction::EraseObject);
DEFINE_MERGE_NOOP(Instruction::ArrayErase, Instruction::EraseObject);
DEFINE_MERGE_NOOP(Instruction::Clear, Instruction::EraseObject);
DEFINE_MERGE_NOOP(Instruction::SetInsert, Instruction::EraseObject);
DEFINE_MERGE_NOOP(Instruction::SetErase, Instruction::EraseObject);


/// Update rules

DEFINE_NESTED_MERGE(Instruction::Update)
{
    using Type = Instruction::Payload::Type;

    if (outer.value.type == Type::ObjectValue) {
        // Creating an embedded object is an idempotent operation, and should
        // not eliminate updates to the subtree.
        return;
    }

    // Setting a value higher up in the hierarchy overwrites any modification to
    // the inner value, regardless of when this happened.
    if (auto next_element = is_prefix_of(outer, inner)) {
        //  If this is a collection in mixed, we will allow the inner instruction
        //  to pass so long as it references the proper type (list or dictionary).
        if (outer.value.type == Type::List && mpark::holds_alternative<uint32_t>(*next_element)) {
            return;
        }
        else if (outer.value.type == Type::Dictionary && mpark::holds_alternative<InternString>(*next_element)) {
            return;
        }
        inner_side.discard();
    }
}

DEFINE_MERGE(Instruction::Update, Instruction::Update)
{
    // The two instructions are at the same level of nesting.

    using Type = Instruction::Payload::Type;

    if (same_path(left, right)) {
        bool left_is_default = false;
        bool right_is_default = false;
        if (!(left.is_array_update() == right.is_array_update())) {
            bad_merge(left_side, left, "Merge error: left.is_array_update() == right.is_array_update()");
        }

        if (!left.is_array_update()) {
            if (right.is_array_update()) {
                bad_merge(right_side, right, "Merge error: !right.is_array_update()");
            }
            left_is_default = left.is_default;
            right_is_default = right.is_default;
        }
        else if (!(left.prior_size == right.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.prior_size == right.prior_size");
        }

        if (left.value.type != right.value.type) {
            // Embedded object creation should always lose to an Update(value),
            // because these structures are nested, and we need to discard any
            // update inside the structure.
            if (left.value.type == Type::ObjectValue) {
                left_side.discard();
                return;
            }
            else if (right.value.type == Type::ObjectValue) {
                right_side.discard();
                return;
            }
        }

        // Updates to List or Dictionary are idempotent. If both sides are setting to the same value,
        // let them both pass through. It is important that the instruction application rules reflect this.
        // If it is not two lists or dictionaries, then the normal last-writer-wins rules will take effect below.
        if (left.value.type == right.value.type &&
            (left.value.type == Type::List || left.value.type == Type::Dictionary)) {
            return;
        }

        // CONFLICT: Two updates of the same element.
        //
        // RESOLUTION: Suppress the effect of the UPDATE operation with the lower
        // timestamp. Note that the timestamps can never be equal. This is
        // achieved on both sides by discarding the received UPDATE operation if
        // it has a lower timestamp than the previously applied UPDATE operation.
        if (left_is_default == right_is_default) {
            if (left_side.timestamp() < right_side.timestamp()) {
                left_side.discard(); // --->
            }
            else {
                right_side.discard(); // <---
            }
        }
        else {
            if (left_is_default) {
                left_side.discard();
            }
            else {
                right_side.discard();
            }
        }
    }
}

DEFINE_MERGE(Instruction::AddInteger, Instruction::Update)
{
    // The two instructions are at the same level of nesting.

    if (same_path(left, right)) {
        // CONFLICT: Add vs Set of the same element.
        //
        // RESOLUTION: If the Add was later than the Set, add its value to
        // the payload of the Set instruction. Otherwise, discard it.

        bool right_is_default = !right.is_array_update() && right.is_default;

        // Five Cases Here:
        // 1. AddInteger is after Update and Update is of a non-integer type
        //     - Discard the AddInteger; AddInteger to a mixed field is a no-op
        // 2: AddInteger is after the Update and the Update instruction contains an integer payload:
        //     - We increment the Update instruction payload
        // 3: AddInteger is after Update and Update is null:
        //     - No conflict
        // 4: Update is after AddInteger and Update.default is false
        //     - Discard the AddInteger
        // 5: Update is after AddInteger and Update.default is true
        //     - Treat the Update as if it were before the AddInteger instruction

        // Note: AddInteger survives SetDefault, regardless of timestamp.
        if (right_side.timestamp() < left_side.timestamp() || right_is_default) {
            if (right.value.is_null()) {
                // The AddInteger happened "after" the Update(null). This becomes a
                // no-op, but if the server later integrates a Update(int) that
                // came-before the AddInteger, it will be taken into account again.
                return;
            }

            // The AddInteger happened after an Update with a non int type
            // This must be operating on a mixed field. Discard the AddInteger
            if (right.value.type != Instruction::Payload::Type::Int) {
                left_side.discard();
                return;
            }

            // Wrapping add
            uint64_t ua = uint64_t(right.value.data.integer);
            uint64_t ub = uint64_t(left.value);
            right.value.data.integer = int64_t(ua + ub);
        }
        else {
            left_side.discard();
        }
    }
}

DEFINE_MERGE_NOOP(Instruction::AddColumn, Instruction::Update);

DEFINE_MERGE(Instruction::EraseColumn, Instruction::Update)
{
    if (same_column(left, right)) {
        right_side.discard();
    }
}

DEFINE_MERGE(Instruction::ArrayInsert, Instruction::Update)
{
    if (same_container(left, right)) {
        REALM_ASSERT(right.is_array_update());
        if (!(left.prior_size == right.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.prior_size == right.prior_size");
        }
        if (!(left.index() <= left.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.index() <= left.prior_size");
        }
        if (!(right.index() < right.prior_size)) {
            bad_merge(right_side, right, "Merge error: right.index() < right.prior_size");
        }
        right.prior_size += 1;
        if (right.index() >= left.index()) {
            right.index() += 1; // --->
        }
    }
}

DEFINE_MERGE(Instruction::ArrayMove, Instruction::Update)
{
    if (same_container(left, right)) {
        REALM_ASSERT(right.is_array_update());

        if (!(left.index() < left.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.index() < left.prior_size");
        }
        if (!(right.index() < right.prior_size)) {
            bad_merge(right_side, right, "Merge error: right.index() < right.prior_size");
        }

        // FIXME: This marks both sides as dirty, even when they are unmodified.
        merge_get_vs_move(right.index(), left.index(), left.ndx_2);
    }
}

DEFINE_MERGE(Instruction::ArrayErase, Instruction::Update)
{
    if (same_container(left, right)) {
        REALM_ASSERT(right.is_array_update());
        if (!(left.prior_size == right.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.prior_size == right.prior_size");
        }
        if (!(left.index() < left.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.index() < left.prior_size");
        }
        if (!(right.index() < right.prior_size)) {
            bad_merge(right_side, right, "Merge error: right.index() < right.prior_size");
        }

        // CONFLICT: Update of a removed element.
        //
        // RESOLUTION: Discard the UPDATE operation received on the right side.
        right.prior_size -= 1;

        if (left.index() == right.index()) {
            // CONFLICT: Update of a removed element.
            //
            // RESOLUTION: Discard the UPDATE operation received on the right side.
            right_side.discard();
        }
        else if (right.index() > left.index()) {
            right.index() -= 1;
        }
    }
}

DEFINE_MERGE(Instruction::Clear, Instruction::Update)
{
    using Type = Instruction::Payload::Type;
    using CollectionType = Instruction::CollectionType;

    // The two instructions are at the same level of nesting.
    if (same_path(left, right)) {
        REALM_ASSERT(right.value.type != Type::Set);
        // If both sides are setting/operating on the same type, let them both pass through.
        // It is important that the instruction application rules reflect this.
        // If it is not two lists or dictionaries, then the normal last-writer-wins rules will take effect below.
        if (left.collection_type == CollectionType::List && right.value.type == Type::List) {
            return;
        }
        if (left.collection_type == CollectionType::Dictionary && right.value.type == Type::Dictionary) {
            return;
        }

        // CONFLICT: Clear and Update of the same element.
        //
        // RESOLUTION: Discard the instruction with the lower timestamp. This has the
        // effect of preserving insertions that came after the clear (if it has the
        // higher timestamp), or preserve additional updates (and potential insertions)
        // that came after the update.
        if (left_side.timestamp() < right_side.timestamp()) {
            left_side.discard();
        }
        else {
            right_side.discard();
        }
    }
}

// Handled by nested rule
DEFINE_MERGE_NOOP(Instruction::SetInsert, Instruction::Update);
DEFINE_MERGE_NOOP(Instruction::SetErase, Instruction::Update);


/// AddInteger rules

DEFINE_NESTED_MERGE(Instruction::AddInteger)
{
    if (is_prefix_of(outer, inner)) {
        inner_side.discard();
    }
}

DEFINE_MERGE(Instruction::Clear, Instruction::AddInteger)
{
    // The two instructions are at the same level of nesting.
    if (same_path(left, right)) {
        right_side.discard();
    }
}

DEFINE_MERGE_NOOP(Instruction::AddInteger, Instruction::AddInteger);
DEFINE_MERGE_NOOP(Instruction::AddColumn, Instruction::AddInteger);
DEFINE_MERGE_NOOP(Instruction::EraseColumn, Instruction::AddInteger);
DEFINE_MERGE_NOOP(Instruction::ArrayInsert, Instruction::AddInteger);
DEFINE_MERGE_NOOP(Instruction::ArrayMove, Instruction::AddInteger);
DEFINE_MERGE_NOOP(Instruction::ArrayErase, Instruction::AddInteger);
DEFINE_MERGE_NOOP(Instruction::SetInsert, Instruction::AddInteger);
DEFINE_MERGE_NOOP(Instruction::SetErase, Instruction::AddInteger);


/// AddColumn rules

DEFINE_NESTED_MERGE_NOOP(Instruction::AddColumn);

DEFINE_MERGE(Instruction::AddColumn, Instruction::AddColumn)
{
    if (same_column(left, right)) {
        StringData left_name = left_side.get_string(left.field);
        if (left.type != right.type) {
            bad_merge(
                "Schema mismatch: Property '%1' in class '%2' is of type %3 on one side and type %4 on the other.",
                left_name, left_side.get_string(left.table), get_type_name(left.type), get_type_name(right.type));
        }

        if (left.nullable != right.nullable) {
            bad_merge("Schema mismatch: Property '%1' in class '%2' is nullable on one side and not on the other.",
                      left_name, left_side.get_string(left.table));
        }

        if (left.collection_type != right.collection_type) {
            auto collection_type_name = [](Instruction::CollectionType type) -> const char* {
                switch (type) {
                    case Instruction::CollectionType::Single:
                        return "single value";
                    case Instruction::CollectionType::List:
                        return "list";
                    case Instruction::CollectionType::Dictionary:
                        return "dictionary";
                    case Instruction::CollectionType::Set:
                        return "set";
                }
                REALM_TERMINATE("");
            };

            std::stringstream ss;
            const char* left_type = collection_type_name(left.collection_type);
            const char* right_type = collection_type_name(right.collection_type);
            bad_merge("Schema mismatch: Property '%1' in class '%2' is a %3 on one side, and a %4 on the other.",
                      left_name, left_side.get_string(left.table), left_type, right_type);
        }

        if (left.type == Instruction::Payload::Type::Link) {
            StringData left_target = left_side.get_string(left.link_target_table);
            StringData right_target = right_side.get_string(right.link_target_table);
            if (left_target != right_target) {
                bad_merge("Schema mismatch: Link property '%1' in class '%2' points to class '%3' on one side and to "
                          "'%4' on the other.",
                          left_name, left_side.get_string(left.table), left_target, right_target);
            }
        }

        // Name, type, nullability and link targets match -- discard both
        // sides and proceed.
        left_side.discard();
        right_side.discard();
    }
}

DEFINE_MERGE(Instruction::EraseColumn, Instruction::AddColumn)
{
    if (same_column(left, right)) {
        right_side.discard();
    }
}

DEFINE_MERGE_NOOP(Instruction::ArrayInsert, Instruction::AddColumn);
DEFINE_MERGE_NOOP(Instruction::ArrayMove, Instruction::AddColumn);
DEFINE_MERGE_NOOP(Instruction::ArrayErase, Instruction::AddColumn);
DEFINE_MERGE_NOOP(Instruction::Clear, Instruction::AddColumn);
DEFINE_MERGE_NOOP(Instruction::SetInsert, Instruction::AddColumn);
DEFINE_MERGE_NOOP(Instruction::SetErase, Instruction::AddColumn);


/// EraseColumn rules

DEFINE_NESTED_MERGE_NOOP(Instruction::EraseColumn);

DEFINE_MERGE(Instruction::EraseColumn, Instruction::EraseColumn)
{
    if (same_column(left, right)) {
        left_side.discard();
        right_side.discard();
    }
}

DEFINE_MERGE_NOOP(Instruction::ArrayInsert, Instruction::EraseColumn);
DEFINE_MERGE_NOOP(Instruction::ArrayMove, Instruction::EraseColumn);
DEFINE_MERGE_NOOP(Instruction::ArrayErase, Instruction::EraseColumn);
DEFINE_MERGE_NOOP(Instruction::Clear, Instruction::EraseColumn);
DEFINE_MERGE_NOOP(Instruction::SetInsert, Instruction::EraseColumn);
DEFINE_MERGE_NOOP(Instruction::SetErase, Instruction::EraseColumn);

/// ArrayInsert rules

DEFINE_NESTED_MERGE(Instruction::ArrayInsert)
{
    if (is_container_prefix_of(outer, inner)) {
        auto& index = corresponding_index_in_path(outer, inner);
        if (index >= outer.index()) {
            index += 1;
        }
    }
}

DEFINE_MERGE(Instruction::ArrayInsert, Instruction::ArrayInsert)
{
    if (same_container(left, right)) {
        if (!(left.prior_size == right.prior_size)) {
            bad_merge(right_side, right, "Exception: left.prior_size == right.prior_size");
        }
        left.prior_size++;
        right.prior_size++;

        if (left.index() > right.index()) {
            left.index() += 1; // --->
        }
        else if (left.index() < right.index()) {
            right.index() += 1; // <---
        }
        else { // left index == right index
            // CONFLICT: Two element insertions at the same position.
            //
            // Resolution: Place the inserted elements in order of increasing
            // timestamp. Note that the timestamps can never be equal.
            if (left_side.timestamp() < right_side.timestamp()) {
                right.index() += 1;
            }
            else {
                left.index() += 1;
            }
        }
    }
}

DEFINE_MERGE(Instruction::ArrayMove, Instruction::ArrayInsert)
{
    if (same_container(left, right)) {
        left.prior_size += 1;

        // Left insertion vs right removal
        if (right.index() > left.index()) {
            right.index() -= 1; // --->
        }
        else {
            left.index() += 1; // <---
        }

        // Left insertion vs left insertion
        if (right.index() < left.ndx_2) {
            left.ndx_2 += 1; // <---
        }
        else if (right.index() > left.ndx_2) {
            right.index() += 1; // --->
        }
        else { // right.index() == left.ndx_2
            // CONFLICT: Insertion and movement to same position.
            //
            // RESOLUTION: Place the two elements in order of increasing
            // timestamp. Note that the timestamps can never be equal.
            if (left_side.timestamp() < right_side.timestamp()) {
                left.ndx_2 += 1; // <---
            }
            else {
                right.index() += 1; // --->
            }
        }
    }
}

DEFINE_MERGE(Instruction::ArrayErase, Instruction::ArrayInsert)
{
    if (same_container(left, right)) {
        if (!(left.prior_size == right.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.prior_size == right.prior_size");
        }
        if (!(left.index() < left.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.index() < left.prior_size");
        }
        if (!(right.index() <= right.prior_size)) {
            bad_merge(left_side, left, "Merge error: right.index() <= right.prior_size");
        }

        left.prior_size++;
        right.prior_size--;
        if (right.index() <= left.index()) {
            left.index() += 1; // --->
        }
        else {
            right.index() -= 1; // <---
        }
    }
}

// Handled by nested rules
DEFINE_MERGE_NOOP(Instruction::Clear, Instruction::ArrayInsert);
DEFINE_MERGE_NOOP(Instruction::SetInsert, Instruction::ArrayInsert);
DEFINE_MERGE_NOOP(Instruction::SetErase, Instruction::ArrayInsert);


/// ArrayMove rules

DEFINE_NESTED_MERGE(Instruction::ArrayMove)
{
    if (is_container_prefix_of(outer, inner)) {
        auto& index = corresponding_index_in_path(outer, inner);
        merge_get_vs_move(outer.index(), index, outer.ndx_2);
    }
}

DEFINE_MERGE(Instruction::ArrayMove, Instruction::ArrayMove)
{
    if (same_container(left, right)) {
        if (!(left.prior_size == right.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.prior_size == right.prior_size");
        }
        if (!(left.index() < left.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.index() < left.prior_size");
        }
        if (!(right.index() < right.prior_size)) {
            bad_merge(right_side, right, "Merge error: right.index() < right.prior_size");
        }
        if (!(left.ndx_2 < left.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.ndx_2 < left.prior_size");
        }
        if (!(right.ndx_2 < right.prior_size)) {
            bad_merge(right_side, right, "Merge error: right.ndx_2 < right.prior_size");
        }

        if (left.index() < right.index()) {
            right.index() -= 1; // <---
        }
        else if (left.index() > right.index()) {
            left.index() -= 1; // --->
        }
        else {
            // CONFLICT: Two movements of same element.
            //
            // RESOLUTION: Respect the MOVE operation associated with the higher
            // timestamp. If the previously applied MOVE operation has the higher
            // timestamp, discard the received MOVE operation, otherwise use the
            // previously applied MOVE operation to transform the received MOVE
            // operation. Note that the timestamps are never equal.
            if (left_side.timestamp() < right_side.timestamp()) {
                right.index() = left.ndx_2; // <---
                left_side.discard();        // --->
                if (right.index() == right.ndx_2) {
                    right_side.discard(); // <---
                }
            }
            else {
                left.index() = right.ndx_2; // --->
                if (left.index() == left.ndx_2) {
                    left_side.discard(); // --->
                }
                right_side.discard(); // <---
            }
            return;
        }

        // Left insertion vs right removal
        if (left.ndx_2 > right.index()) {
            left.ndx_2 -= 1; // --->
        }
        else {
            right.index() += 1; // <---
        }

        // Left removal vs right insertion
        if (left.index() < right.ndx_2) {
            right.ndx_2 -= 1; // <---
        }
        else {
            left.index() += 1; // --->
        }

        // Left insertion vs right insertion
        if (left.ndx_2 < right.ndx_2) {
            right.ndx_2 += 1; // <---
        }
        else if (left.ndx_2 > right.ndx_2) {
            left.ndx_2 += 1; // --->
        }
        else { // left.ndx_2 == right.ndx_2
            // CONFLICT: Two elements moved to the same position
            //
            // RESOLUTION: Place the moved elements in order of increasing
            // timestamp. Note that the timestamps can never be equal.
            if (left_side.timestamp() < right_side.timestamp()) {
                right.ndx_2 += 1; // <---
            }
            else {
                left.ndx_2 += 1; // --->
            }
        }

        if (left.index() == left.ndx_2) {
            left_side.discard(); // --->
        }
        if (right.index() == right.ndx_2) {
            right_side.discard(); // <---
        }
    }
}

DEFINE_MERGE(Instruction::ArrayErase, Instruction::ArrayMove)
{
    if (same_container(left, right)) {
        if (!(left.prior_size == right.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.prior_size == right.prior_size");
        }
        if (!(left.index() < left.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.index() < left.prior_size");
        }
        if (!(right.index() < right.prior_size)) {
            bad_merge(right_side, right, "Merge error: right.index() < right.prior_size");
        }

        right.prior_size -= 1;

        if (left.index() == right.index()) {
            // CONFLICT: Removal of a moved element.
            //
            // RESOLUTION: Discard the received MOVE operation on the left side, and
            // use the previously applied MOVE operation to transform the received
            // REMOVE operation on the right side.
            left.index() = right.ndx_2; // --->
            right_side.discard();       // <---
        }
        else {
            // Left removal vs right removal
            if (left.index() > right.index()) {
                left.index() -= 1; // --->
            }
            else {                  // left.index() < right.index()
                right.index() -= 1; // <---
            }
            // Left removal vs right insertion
            if (left.index() >= right.ndx_2) {
                left.index() += 1; // --->
            }
            else {
                right.ndx_2 -= 1; // <---
            }

            if (right.index() == right.ndx_2) {
                right_side.discard(); // <---
            }
        }
    }
}

// Handled by nested rule.
DEFINE_MERGE_NOOP(Instruction::Clear, Instruction::ArrayMove);
DEFINE_MERGE_NOOP(Instruction::SetInsert, Instruction::ArrayMove);
DEFINE_MERGE_NOOP(Instruction::SetErase, Instruction::ArrayMove);


/// ArrayErase rules

DEFINE_NESTED_MERGE(Instruction::ArrayErase)
{
    if (is_prefix_of(outer, inner)) {
        // Erase of subtree - inner instruction touches the subtree.
        inner_side.discard();
    }
    else if (is_container_prefix_of(outer, inner)) {
        // Erase of a sibling element in the container - adjust the path.
        auto& index = corresponding_index_in_path(outer, inner);
        if (outer.index() < index) {
            index -= 1;
        }
        else {
            REALM_ASSERT(index != outer.index());
        }
    }
}

DEFINE_MERGE(Instruction::ArrayErase, Instruction::ArrayErase)
{
    if (same_container(left, right)) {
        if (!(left.prior_size == right.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.prior_size == right.prior_size");
        }
        if (!(left.index() < left.prior_size)) {
            bad_merge(left_side, left, "Merge error: left.index() < left.prior_size");
        }
        if (!(right.index() < right.prior_size)) {
            bad_merge(right_side, right, "Merge error: right.index() < right.prior_size");
        }

        left.prior_size -= 1;
        right.prior_size -= 1;

        if (left.index() > right.index()) {
            left.index() -= 1; // --->
        }
        else if (left.index() < right.index()) {
            right.index() -= 1; // <---
        }
        else { // left.index() == right.index()
            // CONFLICT: Two removals of the same element.
            //
            // RESOLUTION: On each side, discard the received REMOVE operation.
            left_side.discard();  // --->
            right_side.discard(); // <---
        }
    }
}

// Handled by nested rules.
DEFINE_MERGE_NOOP(Instruction::Clear, Instruction::ArrayErase);
DEFINE_MERGE_NOOP(Instruction::SetInsert, Instruction::ArrayErase);
DEFINE_MERGE_NOOP(Instruction::SetErase, Instruction::ArrayErase);


/// Clear rules

DEFINE_NESTED_MERGE(Instruction::Clear)
{
    // Note: Clear instructions do not have an index in their path.
    if (is_prefix_of(outer, inner)) {
        inner_side.discard();
    }
}

DEFINE_MERGE(Instruction::Clear, Instruction::Clear)
{
    if (same_path(left, right)) {
        // CONFLICT: Two clears of the same container.
        //
        // RESOLUTION: Discard the clear with the lower timestamp. This has the
        // effect of preserving insertions that came after the clear from the
        // side that has the higher timestamp.
        if (left_side.timestamp() < right_side.timestamp()) {
            left_side.discard();
        }
        else {
            right_side.discard();
        }
    }
}

DEFINE_MERGE(Instruction::SetInsert, Instruction::Clear)
{
    if (same_path(left, right)) {
        left_side.discard();
    }
}

DEFINE_MERGE(Instruction::SetErase, Instruction::Clear)
{
    if (same_path(left, right)) {
        left_side.discard();
    }
}


/// SetInsert rules

DEFINE_NESTED_MERGE_NOOP(Instruction::SetInsert);

DEFINE_MERGE(Instruction::SetInsert, Instruction::SetInsert)
{
    if (same_path(left, right)) {
        // CONFLICT: Two inserts into the same set.
        //
        // RESOLUTION: If the values are the same, discard the insertion with the lower timestamp. Otherwise,
        // do nothing.
        //
        // NOTE: Set insertion is idempotent. Keeping the instruction with the higher timestamp is necessary
        // because we want to maintain associativity in the case where intermittent erases (as ordered by
        // timestamp) arrive at a later point in time.
        if (same_payload(left.value, right.value)) {
            if (left_side.timestamp() < right_side.timestamp()) {
                left_side.discard();
            }
            else {
                right_side.discard();
            }
        }
    }
}

DEFINE_MERGE(Instruction::SetErase, Instruction::SetInsert)
{
    if (same_path(left, right)) {
        // CONFLICT: Insertion and erase in the same set.
        //
        // RESOLUTION: If the inserted/erased values are the same, discard the instruction with the lower
        // timestamp. Otherwise, do nothing.
        //
        // Note: Set insertion and erase are both idempotent. Keeping the instruction with the higher
        // timestamp is necessary because we want to maintain associativity.
        if (same_payload(left.value, right.value)) {
            if (left_side.timestamp() < right_side.timestamp()) {
                left_side.discard();
            }
            else {
                right_side.discard();
            }
        }
    }
}


/// SetErase rules.

DEFINE_NESTED_MERGE_NOOP(Instruction::SetErase);

DEFINE_MERGE(Instruction::SetErase, Instruction::SetErase)
{
    if (same_path(left, right)) {
        // CONFLICT: Two erases in the same set.
        //
        // RESOLUTION: If the values are the same, discard the instruction with the lower timestamp.
        // Otherwise, do nothing.
        if (left.value == right.value) {
            if (left_side.timestamp() < right_side.timestamp()) {
                left_side.discard();
            }
            else {
                right_side.discard();
            }
        }
    }
}


///
/// END OF MERGE RULES!
///

template <class Left, class Right>
void merge_instructions_2(Left& left, Right& right, MajorSide& left_side, MinorSide& right_side)
{
    Merge<Left, Right>::merge(left, right, left_side, right_side);
}

template <class Outer, class Inner, class OuterSide, class InnerSide>
void merge_nested_2(Outer& outer, Inner& inner, OuterSide& outer_side, InnerSide& inner_side)
{
    MergeNested<Outer>::merge(outer, inner, outer_side, inner_side);
}

template <class OuterSide, class InnerSide>
void merge_nested(OuterSide& outer_side, InnerSide& inner_side)
{
    outer_side.get().visit([&](auto& outer) {
        inner_side.get().visit([&](auto& inner) {
            merge_nested_2(outer, inner, outer_side, inner_side);
        });
    });
}

void TransformerImpl::merge_instructions(MajorSide& their_side, MinorSide& our_side)
{
    // FIXME: Find a way to avoid heap-copies of the path.
    Instruction their_before = their_side.get();
    Instruction our_before = our_side.get();

    if (their_side.get().get_if<Instruction::Update>()) {
        REALM_ASSERT(their_side.m_path_len > 2);
    }
    if (our_side.get().get_if<Instruction::Update>()) {
        REALM_ASSERT(our_side.m_path_len > 2);
    }
    if (their_side.get().get_if<Instruction::EraseObject>()) {
        REALM_ASSERT(their_side.m_path_len == 2);
    }
    if (our_side.get().get_if<Instruction::EraseObject>()) {
        REALM_ASSERT(our_side.m_path_len == 2);
    }

    // Update selections on the major side (outer loop) according to events on
    // the minor side (inner loop). The selection may only be impacted if the
    // instruction level is lower (i.e. at a higher point in the hierarchy).
    if (our_side.m_path_len < their_side.m_path_len) {
        merge_nested(our_side, their_side);
        if (their_side.was_discarded)
            return;
    }
    else if (our_side.m_path_len > their_side.m_path_len) {
        merge_nested(their_side, our_side);
        if (our_side.was_discarded)
            return;
    }

    if (!their_side.was_discarded && !our_side.was_discarded) {
        // Even if the instructions were nested, we must still perform a regular
        // merge, because link-related instructions contain information from higher
        // levels (both rows, columns, and tables).
        //
        // FIXME: This condition goes away when dangling links are implemented.
        their_side.get().visit([&](auto& their_instruction) {
            our_side.get().visit([&](auto& our_instruction) {
                merge_instructions_2(their_instruction, our_instruction, their_side, our_side);
            });
        });
    }

    // Note: `left` and/or `right` may be dangling at this point due to
    // discard/prepend. However, if they were not discarded, their iterators are
    // required to point to an instruction of the same type.
    if (!their_side.was_discarded && !their_side.was_replaced) {
        const auto& their_after = their_side.get();
        if (!(their_after == their_before)) {
            their_side.m_changeset->set_dirty(true);
        }
    }

    if (!our_side.was_discarded && !our_side.was_replaced) {
        const auto& our_after = our_side.get();
        if (!(our_after == our_before)) {
            our_side.m_changeset->set_dirty(true);
        }
    }
}

} // anonymous namespace

namespace realm::sync {
void Transformer::merge_changesets(file_ident_type local_file_ident, util::Span<Changeset> their_changesets,
                                   util::Span<Changeset*> our_changesets, util::Logger& logger)
{
    REALM_ASSERT(our_changesets.size() != 0);
    REALM_ASSERT(their_changesets.size() != 0);
    bool trace = false;
#if REALM_DEBUG && !REALM_UWP
    // FIXME: Not thread-safe (use config parameter instead and confine environment reading to test/test_all.cpp)
    const char* trace_p = ::getenv("UNITTEST_TRACE_TRANSFORM");
    trace = (trace_p && StringData{trace_p} != "no");
    static std::mutex trace_mutex;
    util::Optional<std::unique_lock<std::mutex>> l;
    if (trace) {
        l = std::unique_lock<std::mutex>{trace_mutex};
    }
#endif
    TransformerImpl transformer{trace};

    _impl::ChangesetIndex their_index;
    size_t their_num_instructions = 0;
    size_t our_num_instructions = 0;

    // Loop through all instructions on both sides and build conflict groups.
    // This causes the index to merge ranges that are connected by instructions
    // on the left side, but which aren't connected on the right side.
    // FIXME: The conflict groups can be persisted as part of the changeset to
    // skip this step in the future.
    for (size_t i = 0; i < their_changesets.size(); ++i) {
        size_t num_instructions = their_changesets[i].size();
        their_num_instructions += num_instructions;
        logger.trace("Scanning incoming changeset [%1/%2] (%3 instructions)", i + 1, their_changesets.size(),
                     num_instructions);

        their_index.scan_changeset(their_changesets[i]);
    }
    for (size_t i = 0; i < our_changesets.size(); ++i) {
        Changeset& our_changeset = *our_changesets[i];
        size_t num_instructions = our_changeset.size();
        our_num_instructions += num_instructions;
        logger.trace(util::LogCategory::changeset, "Scanning local changeset [%1/%2] (%3 instructions)", i + 1,
                     our_changesets.size(), num_instructions);

        their_index.scan_changeset(our_changeset);
    }

    // Build the index.
    for (size_t i = 0; i < their_changesets.size(); ++i) {
        logger.trace(util::LogCategory::changeset, "Indexing incoming changeset [%1/%2] (%3 instructions)", i + 1,
                     their_changesets.size(), their_changesets[i].size());
        their_index.add_changeset(their_changesets[i]);
    }

    logger.debug(util::LogCategory::changeset,
                 "Finished changeset indexing (incoming: %1 changeset(s) / %2 instructions, local: %3 "
                 "changeset(s) / %4 instructions, conflict group(s): %5)",
                 their_changesets.size(), their_num_instructions, our_changesets.size(), our_num_instructions,
                 their_index.get_num_conflict_groups());

#if REALM_DEBUG // LCOV_EXCL_START
    if (trace) {
        std::cerr << TERM_YELLOW << "\n=> PEER " << std::hex << local_file_ident
                  << " merging "
                     "changeset(s)/from peer(s):\n";
        for (size_t i = 0; i < their_changesets.size(); ++i) {
            std::cerr << "Changeset version " << std::dec << their_changesets[i].version << " from peer "
                      << their_changesets[i].origin_file_ident << " at timestamp "
                      << their_changesets[i].origin_timestamp << "\n";
        }
        std::cerr << "Transforming through local changeset(s):\n";
        for (size_t i = 0; i < our_changesets.size(); ++i) {
            std::cerr << "Changeset version " << our_changesets[i]->version << " from peer "
                      << our_changesets[i]->origin_file_ident << " at timestamp "
                      << our_changesets[i]->origin_timestamp << "\n";
        }

        for (size_t i = 0; i < our_changesets.size(); ++i) {
            std::cerr << TERM_RED << "\nLOCAL (RECIPROCAL) CHANGESET BEFORE MERGE:\n" << TERM_RESET;
            our_changesets[i]->print(std::cerr);
        }

        for (size_t i = 0; i < their_changesets.size(); ++i) {
            std::cerr << TERM_RED << "\nINCOMING CHANGESET BEFORE MERGE:\n" << TERM_RESET;
            their_changesets[i].print(std::cerr);
        }

        std::cerr << TERM_MAGENTA << "\nINCOMING CHANGESET INDEX:\n" << TERM_RESET;
        their_index.print(std::cerr);
        std::cerr << '\n';
        their_index.verify();

        std::cerr << TERM_YELLOW << std::setw(80) << std::left << "MERGE TRACE (incoming):"
                  << "MERGE TRACE (local):\n"
                  << TERM_RESET;
    }
#else
    static_cast<void>(local_file_ident);
#endif // REALM_DEBUG LCOV_EXCL_STOP

    for (size_t i = 0; i < our_changesets.size(); ++i) {
        logger.trace(
            util::LogCategory::changeset,
            "Transforming local changeset [%1/%2] through %3 incoming changeset(s) with %4 conflict group(s)", i + 1,
            our_changesets.size(), their_changesets.size(), their_index.get_num_conflict_groups());
        Changeset* our_changeset = our_changesets[i];

        transformer.m_major_side.set_next_changeset(our_changeset);
        // MinorSide uses the index to find the Changeset.
        transformer.m_minor_side.m_changeset_index = &their_index;
        transformer.transform(); // Throws
    }

    logger.debug(util::LogCategory::changeset,
                 "Finished transforming %1 local changesets through %2 incoming changesets (%3 vs %4 "
                 "instructions, in %5 conflict groups)",
                 our_changesets.size(), their_changesets.size(), our_num_instructions, their_num_instructions,
                 their_index.get_num_conflict_groups());

#if REALM_DEBUG // LCOV_EXCL_START
    // Check that the index is still valid after transformation.
    their_index.verify();
#endif // REALM_DEBUG LCOV_EXCL_STOP

#if REALM_DEBUG // LCOV_EXCL_START
    if (trace) {
        for (size_t i = 0; i < our_changesets.size(); ++i) {
            std::cerr << TERM_CYAN << "\nRECIPROCAL CHANGESET AFTER MERGE:\n" << TERM_RESET;
            our_changesets[i]->print(std::cerr);
            std::cerr << '\n';
        }
        for (size_t i = 0; i < their_changesets.size(); ++i) {
            std::cerr << TERM_CYAN << "INCOMING CHANGESET AFTER MERGE:\n" << TERM_RESET;
            their_changesets[i].print(std::cerr);
            std::cerr << '\n';
        }
    }
#endif // LCOV_EXCL_STOP REALM_DEBUG
}

size_t Transformer::transform_remote_changesets(TransformHistory& history, file_ident_type local_file_ident,
                                                version_type current_local_version,
                                                util::Span<Changeset> parsed_changesets,
                                                util::FunctionRef<bool(const Changeset*)> changeset_applier,
                                                util::Logger& logger)
{
    REALM_ASSERT(local_file_ident != 0);

    std::vector<Changeset*> our_changesets;

    // p points to the beginning of a range of changesets that share the same
    // "base", i.e. are based on the same local version.
    auto p = parsed_changesets.begin();
    auto parsed_changesets_end = parsed_changesets.end();

    while (p != parsed_changesets_end) {
        // Find the range of incoming changesets that share the same
        // last_integrated_local_version, which means we can merge them in one go.
        auto same_base_range_end = std::find_if(p + 1, parsed_changesets_end, [&](auto& changeset) {
            return p->last_integrated_remote_version != changeset.last_integrated_remote_version;
        });

        version_type begin_version = p->last_integrated_remote_version;
        version_type end_version = current_local_version;
        for (;;) {
            HistoryEntry history_entry;
            version_type version = history.find_history_entry(begin_version, end_version, history_entry);
            if (version == 0)
                break; // No more local changesets

            Changeset& our_changeset = get_reciprocal_transform(history, local_file_ident, version,
                                                                history_entry); // Throws
            our_changesets.push_back(&our_changeset);

            begin_version = version;
        }

        bool must_apply_all = false;

        if (!our_changesets.empty()) {
            merge_changesets(local_file_ident, {&*p, same_base_range_end}, our_changesets, logger); // Throws
            // We need to apply all transformed changesets if at least one reciprocal changeset was modified
            // during OT.
            must_apply_all = std::any_of(our_changesets.begin(), our_changesets.end(), [](const Changeset* c) {
                return c->is_dirty();
            });
        }

        auto continue_applying = true;
        for (; p != same_base_range_end && continue_applying; ++p) {
            // It is safe to stop applying the changesets if:
            //      1. There are no reciprocal changesets
            //      2. No reciprocal changeset was modified
            continue_applying = changeset_applier(p) || must_apply_all;
        }
        if (!continue_applying) {
            break;
        }

        our_changesets.clear(); // deliberately not releasing memory
    }

    // NOTE: Any exception thrown during flushing *MUST* lead to rollback of
    // the current transaction.
    flush_reciprocal_transform_cache(history); // Throws

    return p - parsed_changesets.begin();
}


Changeset& Transformer::get_reciprocal_transform(TransformHistory& history, file_ident_type local_file_ident,
                                                 version_type version, const HistoryEntry& history_entry)
{
    auto& changeset = m_reciprocal_transform_cache[version]; // Throws
    if (changeset.empty()) {
        bool is_compressed = false;
        ChunkedBinaryData data = history.get_reciprocal_transform(version, is_compressed);
        ChunkedBinaryInputStream in{data};
        if (is_compressed) {
            size_t total_size;
            auto decompressed = util::compression::decompress_nonportable_input_stream(in, total_size);
            REALM_ASSERT(decompressed);
            sync::parse_changeset(*decompressed, changeset); // Throws
        }
        else {
            sync::parse_changeset(in, changeset); // Throws
        }

        changeset.version = version;
        changeset.last_integrated_remote_version = history_entry.remote_version;
        changeset.origin_timestamp = history_entry.origin_timestamp;
        file_ident_type origin_file_ident = history_entry.origin_file_ident;
        if (origin_file_ident == 0)
            origin_file_ident = local_file_ident;
        changeset.origin_file_ident = origin_file_ident;
    }
    return changeset;
}


void Transformer::flush_reciprocal_transform_cache(TransformHistory& history)
{
    auto changesets = std::move(m_reciprocal_transform_cache);
    m_reciprocal_transform_cache.clear();
    ChangesetEncoder::Buffer output_buffer;
    for (const auto& [version, changeset] : changesets) {
        if (changeset.is_dirty()) {
            encode_changeset(changeset, output_buffer); // Throws
            BinaryData data{output_buffer.data(), output_buffer.size()};
            history.set_reciprocal_transform(version, data); // Throws
            output_buffer.clear();
        }
    }
}

void parse_remote_changeset(const RemoteChangeset& remote_changeset, Changeset& parsed_changeset)
{
    // origin_file_ident = 0 is currently used to indicate an entry of local
    // origin.
    REALM_ASSERT(remote_changeset.origin_file_ident != 0);
    REALM_ASSERT(remote_changeset.remote_version != 0);

    ChunkedBinaryInputStream remote_in{remote_changeset.data};
    parse_changeset(remote_in, parsed_changeset); // Throws

    parsed_changeset.version = remote_changeset.remote_version;
    parsed_changeset.last_integrated_remote_version = remote_changeset.last_integrated_local_version;
    parsed_changeset.origin_timestamp = remote_changeset.origin_timestamp;
    parsed_changeset.origin_file_ident = remote_changeset.origin_file_ident;
    parsed_changeset.original_changeset_size = remote_changeset.original_changeset_size;
}

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