1787

Committed 28 Oct 2023 12:35PM UTC coverage: 91.591% (+0.009%) from 91.582%

Build # 1787

Build Type

push

Evergreen

Committed by

web-flow

Commit Message

Improve configurations for sanitized builds (#6911)

* Refactor sanitizer flags for different build types:

** Enable address sanitizer for msvc
** Allow to build with sanitizer for diffent optimized build (also Debug)
** Make RelASAN, RelTSAN, RelUSAN, RelUSAN just shortcuts for half-optimized builds

* Fix usage of moved object for fuzz tester
* Check asan/tsan on macos x64/arm64
* Check asan with msvc 2019
* Remove Jenkins sanitized builders replaced by evergreen configs
* Work-around stack-use-after-scope with msvc2019 and mpark
* Fix crash on check with staled ColKeys
* fix a buffer overrun in a test
* fix a race in async_open_realm test util
* Add some logger related test fixes
* Work around catch2 limmitation with not thread safe asserts and TSAN races
* Run multiprocesses tests under sanitizers
* add assert for an error reported by undefined sanitizer
* Workaround uv scheduler main thread only constraint for callbacks called from non main thread and requesting a realm

---------

Co-authored-by: James Stone <james.stone@mongodb.com>

Run Details

94356 of 173648 branches covered (0.0%)

54 of 63 new or added lines in 15 files covered. (85.71%)

2202 existing lines in 52 files now uncovered.

230692 of 251872 relevant lines covered (91.59%)

7167285.55 hits per line

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

96.17

/src/realm/sync/noinst/migration_store.cpp

#include <realm/sync/noinst/migration_store.hpp>

#include <realm/transaction.hpp>
#include <realm/sync/noinst/sync_metadata_schema.hpp>

namespace realm::sync {
namespace {
constexpr static int c_schema_version = 1;
constexpr static std::string_view c_flx_migration_table("flx_migration");
constexpr static std::string_view c_flx_migration_started_at("started_at");
constexpr static std::string_view c_flx_migration_completed_at("completed_at");
constexpr static std::string_view c_flx_migration_state("state");
constexpr static std::string_view c_flx_migration_query_string("query_string");
constexpr static std::string_view c_flx_migration_original_partition("original_partition");
constexpr static std::string_view
    c_flx_migration_sentinel_subscription_set_version("sentinel_subscription_set_version");
constexpr static std::string_view c_flx_subscription_name_prefix("flx_migrated_");

class MigrationStoreInit : public MigrationStore {
public:
    explicit MigrationStoreInit(DBRef db)
        : MigrationStore(std::move(db))
    {
    }
};

} // namespace

MigrationStoreRef MigrationStore::create(DBRef db)
{
    return std::make_shared<MigrationStoreInit>(std::move(db));
}

MigrationStore::MigrationStore(DBRef db)
    : m_db(std::move(db))
    , m_state(MigrationState::NotMigrated)
{
    load_data(true); // read_only, default to NotMigrated if table is not initialized
}

bool MigrationStore::load_data(bool read_only)
{
    std::unique_lock lock{m_mutex};

    if (m_migration_table) {
        return true; // already initialized
    }

    std::vector<SyncMetadataTable> internal_tables{
        {&m_migration_table,
         c_flx_migration_table,
         {
             {&m_migration_started_at, c_flx_migration_started_at, type_Timestamp},
             {&m_migration_completed_at, c_flx_migration_completed_at, type_Timestamp, true},
             {&m_migration_state, c_flx_migration_state, type_Int},
             {&m_migration_query_str, c_flx_migration_query_string, type_String},
             {&m_migration_partition, c_flx_migration_original_partition, type_String},
             {&m_sentinel_query_version, c_flx_migration_sentinel_subscription_set_version, type_Int, true},
         }},
    };

    std::optional<int64_t> schema_version;
    auto tr = m_db->start_read();
    if (read_only) {
        // Writing is disabled
        SyncMetadataSchemaVersionsReader schema_versions(tr);
        schema_version = schema_versions.get_version_for(tr, internal_schema_groups::c_flx_migration_store);
        if (!schema_version) {
            return false; // Either table is not initialized or version does not exist
        }
    }
    else { // writable
        SyncMetadataSchemaVersions schema_versions(tr);
        schema_version = schema_versions.get_version_for(tr, internal_schema_groups::c_flx_migration_store);
        // Create the version and metadata_schema if it doesn't exist
        if (!schema_version) {
            tr->promote_to_write();
            schema_versions.set_version_for(tr, internal_schema_groups::c_flx_migration_store, c_schema_version);
            create_sync_metadata_schema(tr, &internal_tables);
            tr->commit_and_continue_as_read();
        }
    }
    // Load the metadata schema unless it was just created
    if (!m_migration_table) {
        if (*schema_version != c_schema_version) {
            throw std::runtime_error("Invalid schema version for flexible sync migration store metadata");
        }
        load_sync_metadata_schema(tr, &internal_tables);
    }

    REALM_ASSERT(m_migration_table);

    // Read the migration object if exists, or default to not migrated
    if (auto migration_table = tr->get_table(m_migration_table); !migration_table->is_empty()) {
        auto migration_store_obj = migration_table->get_object(0);
        m_state = static_cast<MigrationState>(migration_store_obj.get<int64_t>(m_migration_state));
        m_query_string = migration_store_obj.get<String>(m_migration_query_str);
        m_migrated_partition = migration_store_obj.get<String>(m_migration_partition);
        m_sentinel_subscription_set_version =
            migration_store_obj.get<util::Optional<int64_t>>(m_sentinel_query_version);
    }
    else {
        m_state = MigrationState::NotMigrated;
        m_query_string.reset();
        m_migrated_partition.reset();
        m_sentinel_subscription_set_version.reset();
    }
    return true;
}

bool MigrationStore::is_migration_in_progress()
{
    std::lock_guard lock{m_mutex};
    return m_state == MigrationState::InProgress;
}

bool MigrationStore::is_migrated()
{
    std::lock_guard lock{m_mutex};
    return m_state == MigrationState::Migrated;
}

bool MigrationStore::is_rollback_in_progress()
{
    std::lock_guard lock{m_mutex};
    return m_state == MigrationState::RollbackInProgress;
}

void MigrationStore::complete_migration_or_rollback()
{
    // Ensure the migration table has been initialized
    bool loaded = load_data();
    REALM_ASSERT(loaded);

    std::unique_lock lock{m_mutex};
    if (m_state != MigrationState::InProgress && m_state != MigrationState::RollbackInProgress) {
        return;
    }

    // Complete rollback.
    if (m_state == MigrationState::RollbackInProgress) {
        clear(std::move(lock)); // releases the lock
        return;
    }

    // Complete migration.
    m_state = MigrationState::Migrated;

    auto tr = m_db->start_write();
    auto migration_table = tr->get_table(m_migration_table);
    REALM_ASSERT(!migration_table->is_empty());
    auto migration_store_obj = migration_table->get_object(0);
    migration_store_obj.set(m_migration_state, int64_t(m_state));
    migration_store_obj.set(m_migration_completed_at, Timestamp{std::chrono::system_clock::now()});
    tr->commit();
}

std::optional<std::string> MigrationStore::get_migrated_partition()
{
    std::lock_guard lock{m_mutex};
    // This will be valid if migration in progress or complete
    return m_migrated_partition;
}

std::optional<std::string> MigrationStore::get_query_string()
{
    std::lock_guard lock{m_mutex};
    // This will be valid if migration in progress or complete
    return m_query_string;
}

std::shared_ptr<realm::SyncConfig> MigrationStore::convert_sync_config(std::shared_ptr<realm::SyncConfig> config)
{
    REALM_ASSERT(config);
    // If load data failed in the constructor, m_state defaults to NotMigrated

    std::unique_lock lock{m_mutex};
    if (config->flx_sync_requested || m_state == MigrationState::NotMigrated ||
        m_state == MigrationState::RollbackInProgress) {
        return config;
    }

    // Once in the migrated state, the partition value cannot change for the same realm file
    if (m_state == MigrationState::Migrated && m_migrated_partition &&
        m_migrated_partition != config->partition_value) {
        throw LogicError(
            ErrorCodes::IllegalOperation,
            util::format("Partition value cannot be changed for migrated realms\n - original: %1\n -   config: %2",
                         m_migrated_partition, config->partition_value));
    }

    return convert_sync_config_to_flx(std::move(config));
}

std::shared_ptr<realm::SyncConfig>
MigrationStore::convert_sync_config_to_flx(std::shared_ptr<realm::SyncConfig> config)
{
    if (config->flx_sync_requested) {
        return config;
    }

    auto flx_config = std::make_shared<realm::SyncConfig>(*config); // deep copy
    flx_config->partition_value = "";
    flx_config->flx_sync_requested = true;

    return flx_config;
}

void MigrationStore::migrate_to_flx(std::string_view rql_query_string, std::string_view partition_value)
{
    REALM_ASSERT(!rql_query_string.empty());

    // Ensure the migration table has been initialized
    bool loaded = load_data();
    REALM_ASSERT(loaded);

    std::unique_lock lock{m_mutex};
    // Can call migrate_to_flx multiple times if migration has not completed.
    REALM_ASSERT(m_state != MigrationState::Migrated);
    m_state = MigrationState::InProgress;
    m_query_string.emplace(rql_query_string);
    m_migrated_partition.emplace(partition_value);

    auto tr = m_db->start_read();
    auto migration_table = tr->get_table(m_migration_table);
    // A migration object may exist if the migration was started in a previous session.
    if (migration_table->is_empty()) {
        tr->promote_to_write();
        auto migration_store_obj = migration_table->create_object();
        migration_store_obj.set(m_migration_query_str, *m_query_string);
        migration_store_obj.set(m_migration_state, int64_t(m_state));
        migration_store_obj.set(m_migration_partition, *m_migrated_partition);
        migration_store_obj.set(m_migration_started_at, Timestamp{std::chrono::system_clock::now()});
        tr->commit();
    }
    else {
        auto migration_store_obj = migration_table->get_object(0);
        auto state = static_cast<MigrationState>(migration_store_obj.get<int64_t>(m_migration_state));
        auto query_string = migration_store_obj.get<String>(m_migration_query_str);
        auto migrated_partition = migration_store_obj.get<String>(m_migration_partition);
        REALM_ASSERT(m_state == state);
        REALM_ASSERT(m_query_string == query_string);
        REALM_ASSERT(m_migrated_partition == migrated_partition);
    }
}

void MigrationStore::rollback_to_pbs()
{
    // Ensure the migration table has been initialized
    bool loaded = load_data();
    REALM_ASSERT(loaded);

    std::unique_lock lock{m_mutex};
    // Can call rollback_to_pbs multiple times if rollback has not completed.
    REALM_ASSERT(m_state != MigrationState::NotMigrated);
    m_state = MigrationState::RollbackInProgress;

    auto tr = m_db->start_write();
    auto migration_table = tr->get_table(m_migration_table);
    REALM_ASSERT(!migration_table->is_empty());
    auto migration_store_obj = migration_table->get_object(0);
    migration_store_obj.set(m_migration_state, int64_t(m_state));
    tr->commit();
}

void MigrationStore::cancel_migration()
{
    // Ensure the migration table has been initialized
    bool loaded = load_data();
    REALM_ASSERT(loaded);

    // Clear the migration state
    std::unique_lock lock{m_mutex};
    REALM_ASSERT(m_state == MigrationState::Migrated);
    clear(std::move(lock)); // releases the lock
}

void MigrationStore::clear(std::unique_lock<std::mutex>)
{
    // Make sure the migration table has been initialized before calling clear()
    REALM_ASSERT(m_migration_table);

    auto tr = m_db->start_read();
    auto migration_table = tr->get_table(m_migration_table);
    if (migration_table->is_empty()) {
        return; // already cleared
    }

    m_state = MigrationState::NotMigrated;
    m_query_string.reset();
    m_migrated_partition.reset();
    m_sentinel_subscription_set_version.reset();
    tr->promote_to_write();
    migration_table->clear();
    tr->commit();
}

Subscription MigrationStore::make_subscription(const std::string& object_class_name,
                                               const std::string& rql_query_string)
{
    REALM_ASSERT(!object_class_name.empty());

    std::string subscription_name = c_flx_subscription_name_prefix.data() + object_class_name;
    return Subscription{subscription_name, object_class_name, rql_query_string};
}

void MigrationStore::create_subscriptions(const SubscriptionStore& subs_store)
{
    std::unique_lock lock{m_mutex};
    if (m_state != MigrationState::Migrated) {
        return;
    }

    REALM_ASSERT(m_query_string);
    create_subscriptions(subs_store, *m_query_string);
}

void MigrationStore::create_subscriptions(const SubscriptionStore& subs_store, const std::string& rql_query_string)
{
    if (rql_query_string.empty()) {
        return;
    }

    auto mut_sub = subs_store.get_latest().make_mutable_copy();
    auto sub_count = mut_sub.size();

    auto tr = m_db->start_read();
    // List of tables covered by the latest subscription set.
    auto tables = subs_store.get_tables_for_latest(*tr);

    // List of tables in the realm.
    auto table_keys = tr->get_table_keys();
    for (const auto& key : table_keys) {
        if (!tr->table_is_public(key)) {
            continue;
        }
        auto table = tr->get_table(key);
        if (table->get_table_type() != Table::Type::TopLevel) {
            continue;
        }
        auto object_class_name = table->get_class_name();
        if (tables.find(object_class_name) == tables.end()) {
            auto sub = make_subscription(object_class_name, rql_query_string);
            mut_sub.insert_sub(sub);
        }
    }

    // No new subscription was added.
    if (mut_sub.size() == sub_count) {
        return;
    }

    // Commit new subscription set.
    mut_sub.commit();
}

void MigrationStore::create_sentinel_subscription_set(const SubscriptionStore& subs_store)
{
    std::lock_guard lock{m_mutex};
    if (m_state != MigrationState::Migrated) {
        return;
    }
    if (m_sentinel_subscription_set_version) {
        return;
    }
    auto mut_sub = subs_store.get_latest().make_mutable_copy();
    auto subscription_set_version = mut_sub.commit().version();
    m_sentinel_subscription_set_version.emplace(subscription_set_version);

    auto tr = m_db->start_write();
    auto migration_table = tr->get_table(m_migration_table);
    REALM_ASSERT(!migration_table->is_empty());
    auto migration_store_obj = migration_table->get_object(0);
    migration_store_obj.set(m_sentinel_query_version, *m_sentinel_subscription_set_version);
    tr->commit();
}

std::optional<int64_t> MigrationStore::get_sentinel_subscription_set_version()
{
    std::lock_guard lock{m_mutex};
    return m_sentinel_subscription_set_version;
}

} // namespace realm::sync

1	#include <realm/sync/noinst/migration_store.hpp>
2
3	#include <realm/transaction.hpp>
4	#include <realm/sync/noinst/sync_metadata_schema.hpp>
5
6	namespace realm::sync {
7	namespace {
8	constexpr static int c_schema_version = 1;
9	constexpr static std::string_view c_flx_migration_table("flx_migration");
10	constexpr static std::string_view c_flx_migration_started_at("started_at");
11	constexpr static std::string_view c_flx_migration_completed_at("completed_at");
12	constexpr static std::string_view c_flx_migration_state("state");
13	constexpr static std::string_view c_flx_migration_query_string("query_string");
14	constexpr static std::string_view c_flx_migration_original_partition("original_partition");
15	constexpr static std::string_view
16	c_flx_migration_sentinel_subscription_set_version("sentinel_subscription_set_version");
17	constexpr static std::string_view c_flx_subscription_name_prefix("flx_migrated_");
18
19	class MigrationStoreInit : public MigrationStore {
20	public:
21	explicit MigrationStoreInit(DBRef db)
22	: MigrationStore(std::move(db))
23	{	2,714✔
24	}	2,714✔
25	};
26
27	} // namespace
28
29	MigrationStoreRef MigrationStore::create(DBRef db)
30	{	2,714✔
31	return std::make_shared<MigrationStoreInit>(std::move(db));	2,714✔
32	}	2,714✔
33
34	MigrationStore::MigrationStore(DBRef db)
35	: m_db(std::move(db))
36	, m_state(MigrationState::NotMigrated)
37	{	2,714✔
38	load_data(true); // read_only, default to NotMigrated if table is not initialized	2,714✔
39	}	2,714✔
40
41	bool MigrationStore::load_data(bool read_only)
42	{	3,066✔
43	std::unique_lock lock{m_mutex};	3,066✔
44		1,360✔
45	if (m_migration_table) {	3,066✔
46	return true; // already initialized	120✔
47	}	120✔
48		1,300✔
49	std::vector<SyncMetadataTable> internal_tables{	2,946✔
50	{&m_migration_table,	2,946✔
51	c_flx_migration_table,	2,946✔
52	{	2,946✔
53	{&m_migration_started_at, c_flx_migration_started_at, type_Timestamp},	2,946✔
54	{&m_migration_completed_at, c_flx_migration_completed_at, type_Timestamp, true},	2,946✔
55	{&m_migration_state, c_flx_migration_state, type_Int},	2,946✔
56	{&m_migration_query_str, c_flx_migration_query_string, type_String},	2,946✔
57	{&m_migration_partition, c_flx_migration_original_partition, type_String},	2,946✔
58	{&m_sentinel_query_version, c_flx_migration_sentinel_subscription_set_version, type_Int, true},	2,946✔
59	}},	2,946✔
60	};	2,946✔
61		1,300✔
62	std::optional<int64_t> schema_version;	2,946✔
63	auto tr = m_db->start_read();	2,946✔
64	if (read_only) {	2,946✔
65	// Writing is disabled	1,184✔
66	SyncMetadataSchemaVersionsReader schema_versions(tr);	2,714✔
67	schema_version = schema_versions.get_version_for(tr, internal_schema_groups::c_flx_migration_store);	2,714✔
68	if (!schema_version) {	2,714✔
69	return false; // Either table is not initialized or version does not exist	2,690✔
70	}	2,690✔
71	}	232✔
72	else { // writable	232✔
73	SyncMetadataSchemaVersions schema_versions(tr);	232✔
74	schema_version = schema_versions.get_version_for(tr, internal_schema_groups::c_flx_migration_store);	232✔
75	// Create the version and metadata_schema if it doesn't exist	116✔
76	if (!schema_version) {	232✔
77	tr->promote_to_write();	232✔
78	schema_versions.set_version_for(tr, internal_schema_groups::c_flx_migration_store, c_schema_version);	232✔
79	create_sync_metadata_schema(tr, &internal_tables);	232✔
80	tr->commit_and_continue_as_read();	232✔
81	}	232✔
82	}	232✔
83	// Load the metadata schema unless it was just created	1,300✔
84	if (!m_migration_table) {	1,428✔
85	if (*schema_version != c_schema_version) {	24✔
UNCOV 86	throw std::runtime_error("Invalid schema version for flexible sync migration store metadata");	×
UNCOV 87	}	×
88	load_sync_metadata_schema(tr, &internal_tables);	24✔
89	}	24✔
90		128✔
91	REALM_ASSERT(m_migration_table);	256✔
92		128✔
93	// Read the migration object if exists, or default to not migrated	128✔
94	if (auto migration_table = tr->get_table(m_migration_table); !migration_table->is_empty()) {	256✔
95	auto migration_store_obj = migration_table->get_object(0);	16✔
96	m_state = static_cast<MigrationState>(migration_store_obj.get<int64_t>(m_migration_state));	16✔
97	m_query_string = migration_store_obj.get<String>(m_migration_query_str);	16✔
98	m_migrated_partition = migration_store_obj.get<String>(m_migration_partition);	16✔
99	m_sentinel_subscription_set_version =	16✔
100	migration_store_obj.get<util::Optional<int64_t>>(m_sentinel_query_version);	16✔
101	}	16✔
102	else {	240✔
103	m_state = MigrationState::NotMigrated;	240✔
104	m_query_string.reset();	240✔
105	m_migrated_partition.reset();	240✔
106	m_sentinel_subscription_set_version.reset();	240✔
107	}	240✔
108	return true;	256✔
109	}	256✔
110
111	bool MigrationStore::is_migration_in_progress()
112	{	96✔
113	std::lock_guard lock{m_mutex};	96✔
114	return m_state == MigrationState::InProgress;	96✔
115	}	96✔
116
117	bool MigrationStore::is_migrated()
118	{	5,448✔
119	std::lock_guard lock{m_mutex};	5,448✔
120	return m_state == MigrationState::Migrated;	5,448✔
121	}	5,448✔
122
123	bool MigrationStore::is_rollback_in_progress()
124	{	2,686✔
125	std::lock_guard lock{m_mutex};	2,686✔
126	return m_state == MigrationState::RollbackInProgress;	2,686✔
127	}	2,686✔
128
129	void MigrationStore::complete_migration_or_rollback()
130	{	260✔
131	// Ensure the migration table has been initialized	130✔
132	bool loaded = load_data();	260✔
133	REALM_ASSERT(loaded);	260✔
134		130✔
135	std::unique_lock lock{m_mutex};	260✔
136	if (m_state != MigrationState::InProgress && m_state != MigrationState::RollbackInProgress) {	260✔
137	return;	196✔
138	}	196✔
139		32✔
140	// Complete rollback.	32✔
141	if (m_state == MigrationState::RollbackInProgress) {	64✔
142	clear(std::move(lock)); // releases the lock	20✔
143	return;	20✔
144	}	20✔
145		22✔
146	// Complete migration.	22✔
147	m_state = MigrationState::Migrated;	44✔
148		22✔
149	auto tr = m_db->start_write();	44✔
150	auto migration_table = tr->get_table(m_migration_table);	44✔
151	REALM_ASSERT(!migration_table->is_empty());	44✔
152	auto migration_store_obj = migration_table->get_object(0);	44✔
153	migration_store_obj.set(m_migration_state, int64_t(m_state));	44✔
154	migration_store_obj.set(m_migration_completed_at, Timestamp{std::chrono::system_clock::now()});	44✔
155	tr->commit();	44✔
156	}	44✔
157
158	std::optional<std::string> MigrationStore::get_migrated_partition()
159	{	1,448✔
160	std::lock_guard lock{m_mutex};	1,448✔
161	// This will be valid if migration in progress or complete	738✔
162	return m_migrated_partition;	1,448✔
163	}	1,448✔
164
165	std::optional<std::string> MigrationStore::get_query_string()
166	{	140✔
167	std::lock_guard lock{m_mutex};	140✔
168	// This will be valid if migration in progress or complete	70✔
169	return m_query_string;	140✔
170	}	140✔
171
172	std::shared_ptr<realm::SyncConfig> MigrationStore::convert_sync_config(std::shared_ptr<realm::SyncConfig> config)
173	{	232✔
174	REALM_ASSERT(config);	232✔
175	// If load data failed in the constructor, m_state defaults to NotMigrated	116✔
176		116✔
177	std::unique_lock lock{m_mutex};	232✔
178	if (config->flx_sync_requested \|\| m_state == MigrationState::NotMigrated \|\|	232✔
179	m_state == MigrationState::RollbackInProgress) {	190✔
180	return config;	120✔
181	}	120✔
182		56✔
183	// Once in the migrated state, the partition value cannot change for the same realm file	56✔
184	if (m_state == MigrationState::Migrated && m_migrated_partition &&	112✔
185	m_migrated_partition != config->partition_value) {	74✔
186	throw LogicError(	12✔
187	ErrorCodes::IllegalOperation,	12✔
188	util::format("Partition value cannot be changed for migrated realms\n - original: %1\n - config: %2",	12✔
189	m_migrated_partition, config->partition_value));	12✔
190	}	12✔
191		50✔
192	return convert_sync_config_to_flx(std::move(config));	100✔
193	}	100✔
194
195	std::shared_ptr<realm::SyncConfig>
196	MigrationStore::convert_sync_config_to_flx(std::shared_ptr<realm::SyncConfig> config)
197	{	116✔
198	if (config->flx_sync_requested) {	116✔
199	return config;	4✔
200	}	4✔
201		56✔
202	auto flx_config = std::make_shared<realm::SyncConfig>(*config); // deep copy	112✔
203	flx_config->partition_value = "";	112✔
204	flx_config->flx_sync_requested = true;	112✔
205		56✔
206	return flx_config;	112✔
207	}	112✔
208
209	void MigrationStore::migrate_to_flx(std::string_view rql_query_string, std::string_view partition_value)
210	{	56✔
211	REALM_ASSERT(!rql_query_string.empty());	56✔
212		28✔
213	// Ensure the migration table has been initialized	28✔
214	bool loaded = load_data();	56✔
215	REALM_ASSERT(loaded);	56✔
216		28✔
217	std::unique_lock lock{m_mutex};	56✔
218	// Can call migrate_to_flx multiple times if migration has not completed.	28✔
219	REALM_ASSERT(m_state != MigrationState::Migrated);	56✔
220	m_state = MigrationState::InProgress;	56✔
221	m_query_string.emplace(rql_query_string);	56✔
222	m_migrated_partition.emplace(partition_value);	56✔
223		28✔
224	auto tr = m_db->start_read();	56✔
225	auto migration_table = tr->get_table(m_migration_table);	56✔
226	// A migration object may exist if the migration was started in a previous session.	28✔
227	if (migration_table->is_empty()) {	56✔
228	tr->promote_to_write();	44✔
229	auto migration_store_obj = migration_table->create_object();	44✔
230	migration_store_obj.set(m_migration_query_str, *m_query_string);	44✔
231	migration_store_obj.set(m_migration_state, int64_t(m_state));	44✔
232	migration_store_obj.set(m_migration_partition, *m_migrated_partition);	44✔
233	migration_store_obj.set(m_migration_started_at, Timestamp{std::chrono::system_clock::now()});	44✔
234	tr->commit();	44✔
235	}	44✔
236	else {	12✔
237	auto migration_store_obj = migration_table->get_object(0);	12✔
238	auto state = static_cast<MigrationState>(migration_store_obj.get<int64_t>(m_migration_state));	12✔
239	auto query_string = migration_store_obj.get<String>(m_migration_query_str);	12✔
240	auto migrated_partition = migration_store_obj.get<String>(m_migration_partition);	12✔
241	REALM_ASSERT(m_state == state);	12✔
242	REALM_ASSERT(m_query_string == query_string);	12✔
243	REALM_ASSERT(m_migrated_partition == migrated_partition);	12✔
244	}	12✔
245	}	56✔
246
247	void MigrationStore::rollback_to_pbs()
248	{	28✔
249	// Ensure the migration table has been initialized	14✔
250	bool loaded = load_data();	28✔
251	REALM_ASSERT(loaded);	28✔
252		14✔
253	std::unique_lock lock{m_mutex};	28✔
254	// Can call rollback_to_pbs multiple times if rollback has not completed.	14✔
255	REALM_ASSERT(m_state != MigrationState::NotMigrated);	28✔
256	m_state = MigrationState::RollbackInProgress;	28✔
257		14✔
258	auto tr = m_db->start_write();	28✔
259	auto migration_table = tr->get_table(m_migration_table);	28✔
260	REALM_ASSERT(!migration_table->is_empty());	28✔
261	auto migration_store_obj = migration_table->get_object(0);	28✔
262	migration_store_obj.set(m_migration_state, int64_t(m_state));	28✔
263	tr->commit();	28✔
264	}	28✔
265
266	void MigrationStore::cancel_migration()
267	{	8✔
268	// Ensure the migration table has been initialized	4✔
269	bool loaded = load_data();	8✔
270	REALM_ASSERT(loaded);	8✔
271		4✔
272	// Clear the migration state	4✔
273	std::unique_lock lock{m_mutex};	8✔
274	REALM_ASSERT(m_state == MigrationState::Migrated);	8✔
275	clear(std::move(lock)); // releases the lock	8✔
276	}	8✔
277
278	void MigrationStore::clear(std::unique_lock<std::mutex>)
279	{	28✔
280	// Make sure the migration table has been initialized before calling clear()	14✔
281	REALM_ASSERT(m_migration_table);	28✔
282		14✔
283	auto tr = m_db->start_read();	28✔
284	auto migration_table = tr->get_table(m_migration_table);	28✔
285	if (migration_table->is_empty()) {	28✔
UNCOV 286	return; // already cleared	×
UNCOV 287	}	×
288		14✔
289	m_state = MigrationState::NotMigrated;	28✔
290	m_query_string.reset();	28✔
291	m_migrated_partition.reset();	28✔
292	m_sentinel_subscription_set_version.reset();	28✔
293	tr->promote_to_write();	28✔
294	migration_table->clear();	28✔
295	tr->commit();	28✔
296	}	28✔
297
298	Subscription MigrationStore::make_subscription(const std::string& object_class_name,
299	const std::string& rql_query_string)
300	{	48✔
301	REALM_ASSERT(!object_class_name.empty());	48✔
302		24✔
303	std::string subscription_name = c_flx_subscription_name_prefix.data() + object_class_name;	48✔
304	return Subscription{subscription_name, object_class_name, rql_query_string};	48✔
305	}	48✔
306
307	void MigrationStore::create_subscriptions(const SubscriptionStore& subs_store)
308	{	2,438✔
309	std::unique_lock lock{m_mutex};	2,438✔
310	if (m_state != MigrationState::Migrated) {	2,438✔
311	return;	2,410✔
312	}	2,410✔
313		14✔
314	REALM_ASSERT(m_query_string);	28✔
315	create_subscriptions(subs_store, *m_query_string);	28✔
316	}	28✔
317
318	void MigrationStore::create_subscriptions(const SubscriptionStore& subs_store, const std::string& rql_query_string)
319	{	68✔
320	if (rql_query_string.empty()) {	68✔
UNCOV 321	return;	×
UNCOV 322	}	×
323		34✔
324	auto mut_sub = subs_store.get_latest().make_mutable_copy();	68✔
325	auto sub_count = mut_sub.size();	68✔
326		34✔
327	auto tr = m_db->start_read();	68✔
328	// List of tables covered by the latest subscription set.	34✔
329	auto tables = subs_store.get_tables_for_latest(*tr);	68✔
330		34✔
331	// List of tables in the realm.	34✔
332	auto table_keys = tr->get_table_keys();	68✔
333	for (const auto& key : table_keys) {	536✔
334	if (!tr->table_is_public(key)) {	536✔
335	continue;	444✔
336	}	444✔
337	auto table = tr->get_table(key);	92✔
338	if (table->get_table_type() != Table::Type::TopLevel) {	92✔
339	continue;	×
340	}	×
341	auto object_class_name = table->get_class_name();	92✔
342	if (tables.find(object_class_name) == tables.end()) {	92✔
343	auto sub = make_subscription(object_class_name, rql_query_string);	48✔
344	mut_sub.insert_sub(sub);	48✔
345	}	48✔
346	}	92✔
347		34✔
348	// No new subscription was added.	34✔
349	if (mut_sub.size() == sub_count) {	68✔
350	return;	24✔
351	}	24✔
352		22✔
353	// Commit new subscription set.	22✔
354	mut_sub.commit();	44✔
355	}	44✔
356
357	void MigrationStore::create_sentinel_subscription_set(const SubscriptionStore& subs_store)
358	{	4✔
359	std::lock_guard lock{m_mutex};	4✔
360	if (m_state != MigrationState::Migrated) {	4✔
UNCOV 361	return;	×
UNCOV 362	}	×
363	if (m_sentinel_subscription_set_version) {	4✔
UNCOV 364	return;	×
UNCOV 365	}	×
366	auto mut_sub = subs_store.get_latest().make_mutable_copy();	4✔
367	auto subscription_set_version = mut_sub.commit().version();	4✔
368	m_sentinel_subscription_set_version.emplace(subscription_set_version);	4✔
369		2✔
370	auto tr = m_db->start_write();	4✔
371	auto migration_table = tr->get_table(m_migration_table);	4✔
372	REALM_ASSERT(!migration_table->is_empty());	4✔
373	auto migration_store_obj = migration_table->get_object(0);	4✔
374	migration_store_obj.set(m_sentinel_query_version, *m_sentinel_subscription_set_version);	4✔
375	tr->commit();	4✔
376	}	4✔
377
378	std::optional<int64_t> MigrationStore::get_sentinel_subscription_set_version()
379	{	10,968✔
380	std::lock_guard lock{m_mutex};	10,968✔
381	return m_sentinel_subscription_set_version;	10,968✔
382	}	10,968✔
383
384	} // namespace realm::sync

realm / realm-core / 1787

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