Use a single write transaction for DiscardLocal client resets on FLX realms (#7110)

Updating the subscription store in a separate write transaction from the
recovery means that we temporarily commit an invalid state. If the application
crashes between committing the client reset diff and updating the subscription
store, the next launch of the application would try to use the now-invalid
pending subscriptions that should have been discarded.

Author: tgoyne

CHANGELOG.md

@@ -6,7 +6,7 @@
 
 ### Fixed
 * <How do the end-user experience this issue? what was the impact?> ([#????](https://github.com/realm/realm-core/issues/????), since v?.?.?)
-* None.
+* A crash at a very specific time during a DiscardLocal client reset on a FLX Realm could leave subscriptions in an invalid state ([#7110](https://github.com/realm/realm-core/pull/7110), since v12.3.0).
 
 ### Breaking changes
 * None.

src/realm/sync/client.cpp

@@ -1785,10 +1785,7 @@ void SessionWrapper::handle_pending_client_reset_acknowledgement()
 {
     REALM_ASSERT(!m_finalized);
 
-    auto pending_reset = [&] {
-        auto ft = m_db->start_frozen();
-        return _impl::client_reset::has_pending_reset(ft);
-    }();
+    auto pending_reset = _impl::client_reset::has_pending_reset(*m_db->start_frozen());
     REALM_ASSERT(pending_reset);
     m_sess->logger.info("Tracking pending client reset of type \"%1\" from %2", pending_reset->type,
                         pending_reset->time);
@@ -1804,7 +1801,7 @@ void SessionWrapper::handle_pending_client_reset_acknowledgement()
         }
 
         auto wt = self->m_db->start_write();
-        auto cur_pending_reset = _impl::client_reset::has_pending_reset(wt);
+        auto cur_pending_reset = _impl::client_reset::has_pending_reset(*wt);
         if (!cur_pending_reset) {
             logger.debug(
                 "Was going to remove client reset tracker for type \"%1\" from %2, but it was already removed",
@@ -1821,7 +1818,7 @@ void SessionWrapper::handle_pending_client_reset_acknowledgement()
                          "Removing cycle detection tracker.",
                          pending_reset.type, pending_reset.time);
         }
-        _impl::client_reset::remove_pending_client_resets(wt);
+        _impl::client_reset::remove_pending_client_resets(*wt);
         wt->commit();
     });
 }

src/realm/sync/noinst/client_history_impl.cpp

@@ -265,10 +265,23 @@ void ClientHistory::get_status(version_type& current_client_version, SaltedFileI
         current_client_version = 0;
 
     if (has_pending_client_reset) {
-        *has_pending_client_reset = _impl::client_reset::has_pending_reset(rt).has_value();
+        *has_pending_client_reset = _impl::client_reset::has_pending_reset(*rt).has_value();
     }
 }
 
+SaltedFileIdent ClientHistory::get_client_file_ident(const Transaction& rt) const
+{
+    SaltedFileIdent client_file_ident{rt.get_sync_file_id(), 0};
+
+    using gf = _impl::GroupFriend;
+    if (ref_type ref = gf::get_history_ref(rt)) {
+        Array root(m_db->get_alloc());
+        root.init_from_ref(ref);
+        client_file_ident.salt = salt_type(root.get_as_ref_or_tagged(s_client_file_ident_salt_iip).get_as_int());
+    }
+
+    return client_file_ident;
+}
 
 void ClientHistory::set_client_file_ident(SaltedFileIdent client_file_ident, bool fix_up_object_ids)
 {

src/realm/sync/noinst/client_history_impl.hpp

@@ -152,6 +152,9 @@ class ClientHistory final : public _impl::History, public TransformHistory {
     /// when engaging in future synchronization sessions.
     void set_client_file_ident(SaltedFileIdent client_file_ident, bool fix_up_object_ids);
 
+    /// Gets the client file ident set with `set_client_file_ident`, or `{0, 0}` if it has never been set.
+    SaltedFileIdent get_client_file_ident(const Transaction& tr) const;
+
     /// Stores the synchronization progress in the associated Realm file in a
     /// way that makes it available via get_status() during future
     /// synchronization sessions. Progress is reported by the server in the

src/realm/sync/noinst/client_reset.cpp

@@ -415,20 +415,16 @@ constexpr static std::string_view s_timestamp_col_name("event_time");
 constexpr static std::string_view s_reset_type_col_name("type_of_reset");
 constexpr int64_t metadata_version = 1;
 
-void remove_pending_client_resets(TransactionRef wt)
+void remove_pending_client_resets(Transaction& wt)
 {
-    REALM_ASSERT(wt);
-    if (auto table = wt->get_table(s_meta_reset_table_name)) {
-        if (table->size()) {
-            table->clear();
-        }
+    if (auto table = wt.get_table(s_meta_reset_table_name); table && !table->is_empty()) {
+        table->clear();
     }
 }
 
-util::Optional<PendingReset> has_pending_reset(TransactionRef rt)
+util::Optional<PendingReset> has_pending_reset(const Transaction& rt)
 {
-    REALM_ASSERT(rt);
-    ConstTableRef table = rt->get_table(s_meta_reset_table_name);
+    ConstTableRef table = rt.get_table(s_meta_reset_table_name);
     if (!table || table->size() == 0) {
         return util::none;
     }
@@ -466,14 +462,13 @@ util::Optional<PendingReset> has_pending_reset(TransactionRef rt)
     return pending;
 }
 
-void track_reset(TransactionRef wt, ClientResyncMode mode)
+void track_reset(Transaction& wt, ClientResyncMode mode)
 {
-    REALM_ASSERT(wt);
     REALM_ASSERT(mode != ClientResyncMode::Manual);
-    TableRef table = wt->get_table(s_meta_reset_table_name);
+    TableRef table = wt.get_table(s_meta_reset_table_name);
     ColKey version_col, timestamp_col, type_col;
     if (!table) {
-        table = wt->add_table_with_primary_key(s_meta_reset_table_name, type_ObjectId, s_pk_col_name);
+        table = wt.add_table_with_primary_key(s_meta_reset_table_name, type_ObjectId, s_pk_col_name);
         REALM_ASSERT(table);
         version_col = table->add_column(type_Int, s_version_column_name);
         timestamp_col = table->add_column(type_Timestamp, s_timestamp_col_name);
@@ -503,10 +498,9 @@ void track_reset(TransactionRef wt, ClientResyncMode mode)
                                            {type_col, mode_val}});
 }
 
-static ClientResyncMode reset_precheck_guard(TransactionRef wt, ClientResyncMode mode, bool recovery_is_allowed,
+static ClientResyncMode reset_precheck_guard(Transaction& wt, ClientResyncMode mode, bool recovery_is_allowed,
                                              util::Logger& logger)
 {
-    REALM_ASSERT(wt);
     if (auto previous_reset = has_pending_reset(wt)) {
         logger.info("A previous reset was detected of type: '%1' at: %2", previous_reset->type, previous_reset->time);
         switch (previous_reset->type) {
@@ -557,19 +551,71 @@ static ClientResyncMode reset_precheck_guard(TransactionRef wt, ClientResyncMode
     return mode;
 }
 
-LocalVersionIDs perform_client_reset_diff(DBRef db_local, DBRef db_remote, sync::SaltedFileIdent client_file_ident,
+LocalVersionIDs perform_client_reset_diff(DB& db_local, DB& db_remote, sync::SaltedFileIdent client_file_ident,
                                           util::Logger& logger, ClientResyncMode mode, bool recovery_is_allowed,
                                           bool* did_recover_out, sync::SubscriptionStore* sub_store,
                                           util::UniqueFunction<void(int64_t)> on_flx_version_complete)
 {
-    REALM_ASSERT(db_local);
-    REALM_ASSERT(db_remote);
-    logger.info("Client reset, path_local = %1, "
-                "client_file_ident.ident = %2, "
-                "client_file_ident.salt = %3,"
-                "remote = %4, mode = %5, recovery_is_allowed = %6",
-                db_local->get_path(), client_file_ident.ident, client_file_ident.salt, db_remote->get_path(), mode,
-                recovery_is_allowed);
+    auto wt_local = db_local.start_write();
+    auto actual_mode = reset_precheck_guard(*wt_local, mode, recovery_is_allowed, logger);
+    bool recover_local_changes =
+        actual_mode == ClientResyncMode::Recover || actual_mode == ClientResyncMode::RecoverOrDiscard;
+
+    logger.info("Client reset: path_local = %1, "
+                "client_file_ident = (ident: %2, salt: %3), "
+                "remote_path = %4, requested_mode = %5, recovery_is_allowed = %6, "
+                "actual_mode = %7, will_recover = %8",
+                db_local.get_path(), client_file_ident.ident, client_file_ident.salt, db_remote.get_path(), mode,
+                recovery_is_allowed, actual_mode, recover_local_changes);
+
+    auto& repl_local = dynamic_cast<ClientReplication&>(*db_local.get_replication());
+    auto& history_local = repl_local.get_history();
+    history_local.ensure_updated(wt_local->get_version());
+    SaltedFileIdent orig_file_ident = history_local.get_client_file_ident(*wt_local);
+    VersionID old_version_local = wt_local->get_version_of_current_transaction();
+
+    auto& repl_remote = dynamic_cast<ClientReplication&>(*db_remote.get_replication());
+    auto& history_remote = repl_remote.get_history();
+
+    sync::SaltedVersion fresh_server_version = {0, 0};
+    {
+        SyncProgress remote_progress;
+        sync::version_type remote_version_unused;
+        SaltedFileIdent remote_ident_unused;
+        history_remote.get_status(remote_version_unused, remote_ident_unused, remote_progress);
+        fresh_server_version = remote_progress.latest_server_version;
+    }
+
+    if (!recover_local_changes) {
+        auto rt_remote = db_remote.start_read();
+        // transform the local Realm such that all public tables become identical to the remote Realm
+        transfer_group(*rt_remote, *wt_local, logger, false);
+
+        // now that the state of the fresh and local Realms are identical,
+        // reset the local sync history and steal the fresh Realm's ident
+        history_local.set_client_reset_adjustments(wt_local->get_version(), client_file_ident, fresh_server_version,
+                                                   BinaryData());
+
+        int64_t subscription_version = 0;
+        if (sub_store) {
+            subscription_version = sub_store->set_active_as_latest(*wt_local);
+        }
+
+        wt_local->commit_and_continue_as_read();
+        if (did_recover_out) {
+            *did_recover_out = false;
+        }
+        if (on_flx_version_complete) {
+            on_flx_version_complete(subscription_version);
+        }
+
+        VersionID new_version_local = wt_local->get_version_of_current_transaction();
+        logger.info("perform_client_reset_diff is done: old_version = (version: %1, index: %2), "
+                    "new_version = (version: %3, index: %4)",
+                    old_version_local.version, old_version_local.index, new_version_local.version,
+                    new_version_local.index);
+        return LocalVersionIDs{old_version_local, new_version_local};
+    }
 
     auto remake_active_subscription = [&]() {
         if (!sub_store) {
@@ -587,44 +633,16 @@ LocalVersionIDs perform_client_reset_diff(DBRef db_local, DBRef db_remote, sync:
                     sub.version());
     };
 
-    auto frozen_pre_local_state = db_local->start_frozen();
-    auto wt_local = db_local->start_write();
-    auto history_local = dynamic_cast<ClientHistory*>(wt_local->get_replication()->_get_history_write());
-    REALM_ASSERT(history_local);
-    VersionID old_version_local = wt_local->get_version_of_current_transaction();
-    wt_local->get_history()->ensure_updated(old_version_local.version);
-    SaltedFileIdent orig_file_ident;
-    {
-        sync::version_type old_version_unused;
-        SyncProgress old_progress_unused;
-        history_local->get_status(old_version_unused, orig_file_ident, old_progress_unused);
-    }
-    std::vector<ClientHistory::LocalChange> local_changes;
-
-    mode = reset_precheck_guard(wt_local, mode, recovery_is_allowed, logger);
-    bool recover_local_changes = (mode == ClientResyncMode::Recover || mode == ClientResyncMode::RecoverOrDiscard);
+    auto frozen_pre_local_state = db_local.start_frozen();
+    auto local_changes = history_local.get_local_changes(wt_local->get_version());
+    logger.info("Local changesets to recover: %1", local_changes.size());
 
-    if (recover_local_changes) {
-        local_changes = history_local->get_local_changes(wt_local->get_version());
-        logger.info("Local changesets to recover: %1", local_changes.size());
-    }
+    auto wt_remote = db_remote.start_write();
 
-    sync::SaltedVersion fresh_server_version = {0, 0};
-    auto wt_remote = db_remote->start_write();
-    auto history_remote = dynamic_cast<ClientHistory*>(wt_remote->get_replication()->_get_history_write());
-    REALM_ASSERT(history_remote);
-
-    SyncProgress remote_progress;
-    {
-        sync::version_type remote_version_unused;
-        SaltedFileIdent remote_ident_unused;
-        history_remote->get_status(remote_version_unused, remote_ident_unused, remote_progress);
-    }
-    fresh_server_version = remote_progress.latest_server_version;
     BinaryData recovered_changeset;
 
     // FLX with recovery has to be done in multiple commits, which is significantly different than other modes
-    if (recover_local_changes && sub_store) {
+    if (sub_store) {
         // In FLX recovery, save a copy of the pending subscriptions for later. This
         // needs to be done before they are wiped out by remake_active_subscription()
         std::vector<SubscriptionSet> pending_subscriptions = sub_store->get_pending_subscriptions();
@@ -633,8 +651,8 @@ LocalVersionIDs perform_client_reset_diff(DBRef db_local, DBRef db_remote, sync:
         // now that the state of the fresh and local Realms are identical,
         // reset the local sync history.
         // Note that we do not set the new file ident yet! This is done in the last commit.
-        history_local->set_client_reset_adjustments(wt_local->get_version(), orig_file_ident, fresh_server_version,
-                                                    recovered_changeset);
+        history_local.set_client_reset_adjustments(wt_local->get_version(), orig_file_ident, fresh_server_version,
+                                                   recovered_changeset);
         // The local Realm is committed. There are no changes to the remote Realm.
         wt_remote->rollback_and_continue_as_read();
         wt_local->commit_and_continue_as_read();
@@ -645,9 +663,8 @@ LocalVersionIDs perform_client_reset_diff(DBRef db_local, DBRef db_remote, sync:
         // as needed. This has the consequence that there may be extra notifications along
         // the way to the final state, but since separate commits are necessary, this is
         // unavoidable.
-        wt_local = db_local->start_write();
-        RecoverLocalChangesetsHandler handler{*wt_local, *frozen_pre_local_state, logger,
-                                              db_local->get_replication()};
+        wt_local = db_local.start_write();
+        RecoverLocalChangesetsHandler handler{*wt_local, *frozen_pre_local_state, logger, db_local.get_replication()};
         handler.process_changesets(local_changes, std::move(pending_subscriptions)); // throws on error
         // The new file ident is set as part of the final commit. This is to ensure that if
         // there are any exceptions during recovery, or the process is killed for some
@@ -656,45 +673,35 @@ LocalVersionIDs perform_client_reset_diff(DBRef db_local, DBRef db_remote, sync:
         // partially recovered, but interrupted may continue sync the next time it is
         // opened with only partially recovered state while having lost the history of any
         // offline modifications.
-        history_local->set_client_file_ident_in_wt(wt_local->get_version(), client_file_ident);
+        history_local.set_client_file_ident_in_wt(wt_local->get_version(), client_file_ident);
         wt_local->commit_and_continue_as_read();
     }
     else {
-        if (recover_local_changes) {
-            // In PBS recovery, the strategy is to apply all local changes to the remote realm first,
-            // and then transfer the modified state all at once to the local Realm. This creates a
-            // nice side effect for notifications because only the minimal state change is made.
-            RecoverLocalChangesetsHandler handler{*wt_remote, *frozen_pre_local_state, logger,
-                                                  db_remote->get_replication()};
-            handler.process_changesets(local_changes, {}); // throws on error
-            ClientReplication* client_repl = dynamic_cast<ClientReplication*>(wt_remote->get_replication());
-            REALM_ASSERT_RELEASE(client_repl);
-            ChangesetEncoder& encoder = client_repl->get_instruction_encoder();
-            const sync::ChangesetEncoder::Buffer& buffer = encoder.buffer();
-            recovered_changeset = {buffer.data(), buffer.size()};
-        }
+        // In PBS recovery, the strategy is to apply all local changes to the remote realm first,
+        // and then transfer the modified state all at once to the local Realm. This creates a
+        // nice side effect for notifications because only the minimal state change is made.
+        RecoverLocalChangesetsHandler handler{*wt_remote, *frozen_pre_local_state, logger,
+                                              db_remote.get_replication()};
+        handler.process_changesets(local_changes, {}); // throws on error
+        ChangesetEncoder& encoder = repl_remote.get_instruction_encoder();
+        const sync::ChangesetEncoder::Buffer& buffer = encoder.buffer();
+        recovered_changeset = {buffer.data(), buffer.size()};
 
         // transform the local Realm such that all public tables become identical to the remote Realm
         transfer_group(*wt_remote, *wt_local, logger, recover_local_changes);
 
         // now that the state of the fresh and local Realms are identical,
         // reset the local sync history and steal the fresh Realm's ident
-        history_local->set_client_reset_adjustments(wt_local->get_version(), client_file_ident, fresh_server_version,
-                                                    recovered_changeset);
+        history_local.set_client_reset_adjustments(wt_local->get_version(), client_file_ident, fresh_server_version,
+                                                   recovered_changeset);
 
         // Finally, the local Realm is committed. The changes to the remote Realm are discarded.
         wt_remote->rollback_and_continue_as_read();
         wt_local->commit_and_continue_as_read();
-
-        // If in FLX mode, make a copy of the active subscription set and mark it as
-        // complete. This will cause all other subscription sets to become superceded.
-        // In DiscardLocal mode, only the active subscription set is preserved, so we
-        // are done.
-        remake_active_subscription();
     }
 
     if (did_recover_out) {
-        *did_recover_out = recover_local_changes;
+        *did_recover_out = true;
     }
     VersionID new_version_local = wt_local->get_version_of_current_transaction();
     logger.info("perform_client_reset_diff is done, old_version.version = %1, "

src/realm/sync/noinst/client_reset.hpp

@@ -67,9 +67,9 @@ struct PendingReset {
     ClientResyncMode type;
     Timestamp time;
 };
-void remove_pending_client_resets(TransactionRef wt);
-util::Optional<PendingReset> has_pending_reset(TransactionRef wt);
-void track_reset(TransactionRef wt, ClientResyncMode mode);
+void remove_pending_client_resets(Transaction& wt);
+util::Optional<PendingReset> has_pending_reset(const Transaction& wt);
+void track_reset(Transaction& wt, ClientResyncMode mode);
 
 // 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.
@@ -86,7 +86,7 @@ struct LocalVersionIDs {
     realm::VersionID new_version;
 };
 
-LocalVersionIDs perform_client_reset_diff(DBRef db, DBRef db_remote, sync::SaltedFileIdent client_file_ident,
+LocalVersionIDs perform_client_reset_diff(DB& db, DB& db_remote, sync::SaltedFileIdent client_file_ident,
                                           util::Logger& logger, ClientResyncMode mode, bool recovery_is_allowed,
                                           bool* did_recover_out, sync::SubscriptionStore* sub_store,
                                           util::UniqueFunction<void(int64_t)> on_flx_version_complete);

src/realm/sync/noinst/client_reset_operation.cpp

@@ -99,7 +99,7 @@ bool ClientResetOperation::finalize(sync::SaltedFileIdent salted_file_ident, syn
     }
     bool did_recover_out = false;
     local_version_ids = client_reset::perform_client_reset_diff(
-        m_db, m_db_fresh, m_salted_file_ident, m_logger, m_mode, m_recovery_is_allowed, &did_recover_out, sub_store,
+        *m_db, *m_db_fresh, m_salted_file_ident, m_logger, m_mode, m_recovery_is_allowed, &did_recover_out, sub_store,
         std::move(on_flx_version_complete)); // throws
 
     if (m_notify_after) {