Skip to content

Commit

Permalink
Only track pending client resets done by the same core version
Browse files Browse the repository at this point in the history 
If the previous attempt at performing a client reset was done with a different
core version then we should retry the client reset as the new version may have
fixed a bug that made the previous attempt fail (or may be a downgrade to a
version before when the bug was introduced). This also simplifies the tracking
as it means that we don't need to be able to read trackers created by different
versions.

This also means that we can freely change the schema of the table, which this
takes advantage of to drop the unused primary key and make the error required,
as we never actually stored null and the code reading it would have crashed if
it encountered a null error.
  • Loading branch information
@tgoyne
tgoyne committed Aug 1, 2024
1 parent f35c76e commit 5a1a7b6
Show file tree
Hide file tree
Showing 14 changed files with 209 additions and 333 deletions.
1 change: 1 addition & 0 deletions src/realm/sync/CMakeLists.txt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,7 @@ set(NOINST_HEADERS
noinst/integer_codec.hpp
noinst/migration_store.hpp
noinst/pending_bootstrap_store.hpp
noinst/pending_reset_store.hpp
noinst/protocol_codec.hpp
noinst/root_certs.hpp
noinst/sync_metadata_schema.hpp
Expand Down
6 changes: 3 additions & 3 deletions src/realm/sync/client.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1655,7 +1655,7 @@ void SessionWrapper::handle_pending_client_reset_acknowledgement()
{
REALM_ASSERT(!m_finalized);

auto has_pending_reset = PendingResetStore::has_pending_reset(m_db->start_frozen());
auto has_pending_reset = PendingResetStore::has_pending_reset(*m_db->start_frozen());
if (!has_pending_reset) {
return; // nothing to do
}
Expand All @@ -1678,7 +1678,7 @@ void SessionWrapper::handle_pending_client_reset_acknowledgement()
logger.debug(util::LogCategory::reset, "Server has acknowledged %1", pending_reset);

auto tr = self->m_db->start_write();
auto cur_pending_reset = PendingResetStore::has_pending_reset(tr);
auto cur_pending_reset = PendingResetStore::has_pending_reset(*tr);
if (!cur_pending_reset) {
logger.debug(util::LogCategory::reset, "Client reset cycle detection tracker already removed.");
return;
Expand All @@ -1689,7 +1689,7 @@ void SessionWrapper::handle_pending_client_reset_acknowledgement()
else {
logger.info(util::LogCategory::reset, "Found new %1", cur_pending_reset);
}
PendingResetStore::clear_pending_reset(tr);
PendingResetStore::clear_pending_reset(*tr);
tr->commit();
});
}
Expand Down
15 changes: 7 additions & 8 deletions src/realm/sync/noinst/client_reset.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -410,18 +410,17 @@ void transfer_group(const Transaction& group_src, Transaction& group_dst, util::
}
}

ClientResyncMode reset_precheck_guard(const TransactionRef& wt_local, ClientResyncMode mode,
PendingReset::Action action, const std::optional<Status>& error,
util::Logger& logger)
static ClientResyncMode reset_precheck_guard(const TransactionRef& wt_local, ClientResyncMode mode,
PendingReset::Action action, const Status& error, util::Logger& logger)
{
if (auto previous_reset = sync::PendingResetStore::has_pending_reset(wt_local)) {
if (auto previous_reset = sync::PendingResetStore::has_pending_reset(*wt_local)) {
logger.info(util::LogCategory::reset, "Found a previous %1", *previous_reset);
if (action != previous_reset->action) {
// IF a different client reset is being performed, cler the pending client reset and start over.
logger.info(util::LogCategory::reset,
"New '%1' client reset of type: '%2' is incompatible - clearing previous reset", action,
mode);
sync::PendingResetStore::clear_pending_reset(wt_local);
sync::PendingResetStore::clear_pending_reset(*wt_local);
}
else {
switch (previous_reset->mode) {
Expand All @@ -444,10 +443,10 @@ ClientResyncMode reset_precheck_guard(const TransactionRef& wt_local, ClientResy
util::LogCategory::reset,
"A previous '%1' mode reset from %2 downgrades this mode ('%3') to DiscardLocal",
previous_reset->mode, previous_reset->time, mode);
sync::PendingResetStore::clear_pending_reset(wt_local);
sync::PendingResetStore::clear_pending_reset(*wt_local);
break;
case ClientResyncMode::DiscardLocal:
sync::PendingResetStore::clear_pending_reset(wt_local);
sync::PendingResetStore::clear_pending_reset(*wt_local);
// previous mode Recover and this mode is Discard, this is not a cycle yet
break;
case ClientResyncMode::Manual:
Expand All @@ -473,7 +472,7 @@ ClientResyncMode reset_precheck_guard(const TransactionRef& wt_local, ClientResy
mode = ClientResyncMode::DiscardLocal;
}
}
sync::PendingResetStore::track_reset(wt_local, mode, action, error);
sync::PendingResetStore::track_reset(*wt_local, mode, action, error);
// Ensure we save the tracker object even if we encounter an error and roll
// back the client reset later
wt_local->commit_and_continue_writing();
Expand Down
4 changes: 0 additions & 4 deletions src/realm/sync/noinst/client_reset.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -62,10 +62,6 @@ namespace _impl::client_reset {
void transfer_group(const Transaction& tr_src, Transaction& tr_dst, util::Logger& logger,
bool allow_schema_additions);

ClientResyncMode reset_precheck_guard(const TransactionRef& wt_local, ClientResyncMode mode,
sync::ProtocolErrorInfo::Action action, const std::optional<Status>& error,
util::Logger& logger);

// preform_client_reset_diff() takes the Realm performs a client reset on
// the Realm in 'path_local' given the Realm 'path_fresh' as the source of truth.
// If the fresh path is not provided, discard mode is assumed and all data in the local
Expand Down
4 changes: 2 additions & 2 deletions src/realm/sync/noinst/migration_store.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ bool MigrationStore::load_data(bool read_only)
throw RuntimeError(ErrorCodes::UnsupportedFileFormatVersion,
"Invalid schema version for flexible sync migration store metadata");
}
load_sync_metadata_schema(tr, &internal_tables);
load_sync_metadata_schema(*tr, &internal_tables);
}
else {
if (read_only) {
Expand All @@ -72,7 +72,7 @@ bool MigrationStore::load_data(bool read_only)
SyncMetadataSchemaVersions schema_versions(tr);
// Create the metadata schema and set the version (in the same commit)
schema_versions.set_version_for(tr, internal_schema_groups::c_flx_migration_store, c_schema_version);
create_sync_metadata_schema(tr, &internal_tables);
create_sync_metadata_schema(*tr, &internal_tables);
tr->commit_and_continue_as_read();
}
REALM_ASSERT(m_migration_table);
Expand Down
4 changes: 2 additions & 2 deletions src/realm/sync/noinst/pending_bootstrap_store.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -109,15 +109,15 @@ PendingBootstrapStore::PendingBootstrapStore(DBRef db, util::Logger& logger,
throw RuntimeError(ErrorCodes::SchemaVersionMismatch,
"Invalid schema version for FLX sync pending bootstrap table group");
}
load_sync_metadata_schema(tr, &internal_tables);
load_sync_metadata_schema(*tr, &internal_tables);
}
else {
tr->promote_to_write();
// Ensure the schema versions table is initialized (may add its own commit)
SyncMetadataSchemaVersions schema_versions(tr);
// Create the metadata schema and set the version (in the same commit)
schema_versions.set_version_for(tr, internal_schema_groups::c_pending_bootstraps, c_schema_version);
create_sync_metadata_schema(tr, &internal_tables);
create_sync_metadata_schema(*tr, &internal_tables);
tr->commit_and_continue_as_read();
}
REALM_ASSERT(m_table);
Expand Down
189 changes: 51 additions & 138 deletions src/realm/sync/noinst/pending_reset_store.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@

using namespace realm;
using namespace _impl;
using namespace sync;

namespace realm::sync {

Expand All @@ -41,9 +40,7 @@ std::ostream& operator<<(std::ostream& os, const sync::PendingReset& pr)
else {
os << "pending client reset of type: '" << pr.mode << "' at: " << pr.time;
}
if (pr.error) {
os << " for error: " << *pr.error;
}
os << " for error: " << pr.error;
return os;
}

Expand All @@ -59,196 +56,112 @@ bool operator==(const sync::PendingReset& lhs, const PendingReset::Action& actio

// A table without a "class_" prefix will not generate sync instructions.
constexpr static std::string_view s_meta_reset_table_name("client_reset_metadata");
constexpr static std::string_view s_pk_col_name("id");
constexpr static std::string_view s_timestamp_col_name("reset_time");
constexpr static std::string_view s_reset_recovery_mode_col_name("reset_mode");
constexpr static std::string_view s_reset_action_col_name("reset_action");
constexpr static std::string_view s_reset_error_code_col_name("reset_error_code");
constexpr static std::string_view s_reset_error_msg_col_name("reset_error_msg");
constexpr int64_t s_pending_reset_version = 2;

void PendingResetStore::clear_pending_reset(const TransactionRef& wr_tr)
constexpr static std::string_view s_version_col_name("core_version");
constexpr static std::string_view s_timestamp_col_name("time");
constexpr static std::string_view s_reset_recovery_mode_col_name("mode");
constexpr static std::string_view s_reset_action_col_name("action");
constexpr static std::string_view s_reset_error_code_col_name("error_code");
constexpr static std::string_view s_reset_error_msg_col_name("error_msg");

void PendingResetStore::clear_pending_reset(Group& group)
{
// Write transaction required
REALM_ASSERT(wr_tr->get_transact_stage() == DB::TransactStage::transact_Writing);
auto reset_store = PendingResetStore::load_or_create_schema(wr_tr);
REALM_ASSERT(reset_store.m_pending_reset_table);
// Ensure the pending reset table is empty
if (auto table = wr_tr->get_table(reset_store.m_pending_reset_table); table && !table->is_empty()) {
if (auto table = group.get_table(s_meta_reset_table_name); table && !table->is_empty()) {
table->clear();
}
// Don't commit at the end - allow the caller to do it when they are ready
}

std::optional<PendingReset> PendingResetStore::has_pending_reset(const TransactionRef& rd_tr)
std::optional<PendingReset> PendingResetStore::has_pending_reset(const Group& group)
{
// Make sure the schema has been loaded and try to read legacy data if it's not found
auto reset_store = PendingResetStore::load_schema(rd_tr);
auto reset_store = PendingResetStore::load_schema(group);
if (!reset_store) {
return PendingResetStore::read_legacy_pending_reset(rd_tr);
// Table hasn't been created yet (or has the wrong schema)
return std::nullopt;
}
// Otherwise, read the pending reset entry using the schema metadata
REALM_ASSERT(reset_store->m_pending_reset_table);
auto table = rd_tr->get_table(reset_store->m_pending_reset_table);
auto table = group.get_table(reset_store->m_pending_reset_table);

if (!table || table->size() == 0) {
if (!table || table->size() != 1) {
return std::nullopt;
}
if (table->size() > 1) {
// this may happen if a future version of this code changes the format and expectations around reset metadata.
throw ClientResetFailed(
util::format("Previous client resets detected (%1) but only one is expected.", table->size()));
}
auto reset_entry = *table->begin();
if (reset_entry.get<String>(reset_store->m_version) != REALM_VERSION_STRING) {
// Previous pending reset was written by a different version, so ignore it
return std::nullopt;
}

PendingReset pending;
pending.time = reset_entry.get<Timestamp>(reset_store->m_timestamp);
pending.mode = to_resync_mode(reset_entry.get<int64_t>(reset_store->m_recovery_mode));
pending.action = to_reset_action(reset_entry.get<int64_t>(reset_store->m_action));
auto error_code = reset_entry.get<int64_t>(reset_store->m_error_code);
if (error_code > 0) {
if (error_code != 0) {
pending.error = Status(static_cast<ErrorCodes::Error>(error_code),
reset_entry.get<StringData>(reset_store->m_error_message));
}
return pending;
}

void PendingResetStore::track_reset(const TransactionRef& wr_tr, ClientResyncMode mode, PendingReset::Action action,
const std::optional<Status>& error)
void PendingResetStore::track_reset(Group& group, ClientResyncMode mode, PendingReset::Action action, Status error)
{
REALM_ASSERT(mode != ClientResyncMode::Manual);
// Write transaction required
REALM_ASSERT(wr_tr->get_transact_stage() == DB::TransactStage::transact_Writing);
if (auto table = wr_tr->get_table(s_meta_reset_table_name); table && table->size() > 0) {
// this may happen if a future version of this code changes the format and expectations around reset
// metadata.
throw ClientResetFailed(
util::format("Previous client resets detected (%1) but only one is expected.", table->size()));
}
auto reset_store = PendingResetStore::load_or_create_schema(wr_tr);
auto reset_store = PendingResetStore::load_or_create_schema(group);

REALM_ASSERT(reset_store.m_pending_reset_table);
auto table = wr_tr->get_table(reset_store.m_pending_reset_table);
auto table = group.get_table(reset_store.m_pending_reset_table);
REALM_ASSERT(table);
// Create the new object
auto obj = table->create_object_with_primary_key(
ObjectId::gen(), {
{reset_store.m_timestamp, Timestamp(std::chrono::system_clock::now())},
{reset_store.m_recovery_mode, from_resync_mode(mode)},
{reset_store.m_action, from_reset_action(action)},
});
// Add the error, if provided
if (error) {
obj.set(reset_store.m_error_code, static_cast<int64_t>(error->code()));
obj.set(reset_store.m_error_message, error->reason());
}
// Don't commit at the end - allow the caller to do it when they are ready
table->clear();
table->create_object(null_key, {
{reset_store.m_version, Mixed(REALM_VERSION_STRING)},
{reset_store.m_timestamp, Timestamp(std::chrono::system_clock::now())},
{reset_store.m_recovery_mode, from_resync_mode(mode)},
{reset_store.m_action, from_reset_action(action)},
{reset_store.m_error_code, static_cast<int64_t>(error.code())},
{reset_store.m_error_message, error.reason()},
});
}

PendingResetStore::PendingResetStore(const TransactionRef& rd_tr)
PendingResetStore::PendingResetStore(const Group& g)
: m_internal_tables{
{&m_pending_reset_table,
s_meta_reset_table_name,
{&m_id, s_pk_col_name, type_ObjectId},
{
{&m_version, s_version_col_name, type_String},
{&m_timestamp, s_timestamp_col_name, type_Timestamp},
{&m_recovery_mode, s_reset_recovery_mode_col_name, type_Int},
{&m_action, s_reset_action_col_name, type_Int},
{&m_error_code, s_reset_error_code_col_name, type_Int, true},
{&m_error_message, s_reset_error_msg_col_name, type_String, true},
{&m_error_code, s_reset_error_code_col_name, type_Int},
{&m_error_message, s_reset_error_msg_col_name, type_String},
}},
}
{
// Works with read, write, and frozen transactions
SyncMetadataSchemaVersionsReader schema_versions(rd_tr);
auto schema_version = schema_versions.get_version_for(rd_tr, internal_schema_groups::c_pending_reset_store);

// Load the metadata schema info if a schema version was found
if (schema_version) {
if (*schema_version != s_pending_reset_version) {
// Unsupported schema version
throw RuntimeError(ErrorCodes::UnsupportedFileFormatVersion,
"Found invalid schema version for existing client reset cycle tracking metadata");
}
load_sync_metadata_schema(rd_tr, &m_internal_tables);
if (m_pending_reset_table) {
// If the schema info was read, then store the schema version
m_schema_version = schema_version;
}
if (!try_load_sync_metadata_schema(g, &m_internal_tables).is_ok()) {
m_pending_reset_table = {};
}
}

std::optional<PendingResetStore> PendingResetStore::load_schema(const TransactionRef& rd_tr)
std::optional<PendingResetStore> PendingResetStore::load_schema(const Group& group)
{
PendingResetStore reset_store(rd_tr);
if (reset_store.m_schema_version) {
if (PendingResetStore reset_store(group); reset_store.m_pending_reset_table) {
return reset_store;
}
return std::nullopt;
}

PendingResetStore PendingResetStore::load_or_create_schema(const TransactionRef& wr_tr)
PendingResetStore PendingResetStore::load_or_create_schema(Group& group)
{
PendingResetStore reset_store(wr_tr);
if (reset_store.m_schema_version) {
// If the schema metadata was found, return the initialized class
return reset_store;
}
// Otherwise, set it up from scratch - Make sure the transaction is set for writing
if (wr_tr->get_transact_stage() == DB::TransactStage::transact_Reading) {
wr_tr->promote_to_write();
}
// Ensure writing - all other transaction stages are not allowed
REALM_ASSERT_EX(wr_tr->get_transact_stage() == DB::TransactStage::transact_Writing, wr_tr->get_transact_stage());
PendingResetStore reset_store(group);
if (!reset_store.m_pending_reset_table) {
// If the table exists but has the wrong schema just drop it
if (group.has_table(s_meta_reset_table_name)) {
group.remove_table(s_meta_reset_table_name);
}

// Drop the old table and any stale pending resets
if (wr_tr->has_table(s_meta_reset_table_name)) {
wr_tr->remove_table(s_meta_reset_table_name);
// Create the table with the correct schema
create_sync_metadata_schema(group, &reset_store.m_internal_tables);
}

// Ensure the schema versions table is initialized (may add its own commit)
SyncMetadataSchemaVersions schema_versions(wr_tr);
// Create the metadata schema and set the version (in the same commit)
schema_versions.set_version_for(wr_tr, internal_schema_groups::c_pending_reset_store, s_pending_reset_version);
create_sync_metadata_schema(wr_tr, &reset_store.m_internal_tables);
REALM_ASSERT(reset_store.m_pending_reset_table);
reset_store.m_schema_version = s_pending_reset_version;

// Don't commit yet
return reset_store;
}

std::optional<PendingReset> PendingResetStore::read_legacy_pending_reset(const TransactionRef& rd_tr)
{
// Try to read the pending reset info from v1 of the schema
constexpr static std::string_view s_v1_version_column_name("version");
constexpr static std::string_view s_v1_timestamp_col_name("event_time");
constexpr static std::string_view s_v1_reset_mode_col_name("type_of_reset");

// Check for pending reset v1 - does not use schema version
TableRef table = rd_tr->get_table(s_meta_reset_table_name);
if (table && table->size() > 0) {
ColKey version_col = table->get_column_key(s_v1_version_column_name);
ColKey timestamp_col = table->get_column_key(s_v1_timestamp_col_name);
ColKey mode_col = table->get_column_key(s_v1_reset_mode_col_name);
Obj reset_entry = *table->begin();

if (version_col && reset_entry.get<int64_t>(version_col) == 1LL) {
REALM_ASSERT(timestamp_col);
REALM_ASSERT(mode_col);
PendingReset pending;
pending.time = reset_entry.get<Timestamp>(timestamp_col);
pending.mode = to_resync_mode(reset_entry.get<int64_t>(mode_col));
// Create a fake action depending on the resync mode
pending.action = pending.mode == ClientResyncMode::DiscardLocal
? sync::ProtocolErrorInfo::Action::ClientResetNoRecovery
: sync::ProtocolErrorInfo::Action::ClientReset;
return pending;
}
}
// Add checking for future schema versions here
return std::nullopt;
}

int64_t PendingResetStore::from_reset_action(PendingReset::Action action)
{
switch (action) {
Expand Down
Loading

0 comments on commit 5a1a7b6

Please sign in to comment.