Update server version on all entries in the sync history during client reset with recovery (#7291)

* Update server version on all entries in the sync history during client reset with recovery

* Fix test

* Changes after code review

* Update changelog

Author: danieltabacaru

CHANGELOG.md

@@ -5,8 +5,7 @@
 * Allow the query builder to construct >, >=, <, <= queries for string constants. This is a case sensitive lexicographical comparison. Improved performance of RQL (parsed) queries on a non-linked string property using: >, >=, <, <=, operators and fixed behaviour that a null string should be evaulated as less than everything, previously nulls were not matched. ([#3939](https://github.com/realm/realm-core/issues/3939), this is a prerequisite for https://github.com/realm/realm-swift/issues/8008).
 
 ### Fixed
-* <How do the end-user experience this issue? what was the impact?> ([#????](https://github.com/realm/realm-core/issues/????), since v?.?.?)
-* None.
+* Uploading the changesets recovered during an automatic client reset recovery may lead to 'Bad server version' errors and a new client reset. ([#7279](https://github.com/realm/realm-core/issues/7279), since v13.24.1)
 
 ### Breaking changes
 * None.

src/realm/sync/noinst/client_history_impl.cpp

@@ -66,9 +66,9 @@ void ClientHistory::set_client_reset_adjustments(
         do_trim_sync_history(discard_count);
 
         if (logger.would_log(util::Logger::Level::debug)) {
-            logger.debug("Client reset adjustments: trimming %1 history entries and updating %2 of %3 remaining "
-                         "history entries:",
-                         discard_count, recovered_changesets.size(), sync_history_size());
+            logger.debug("Client reset adjustments: trimming %1 history entries and updating the remaining history "
+                         "entries (%2)",
+                         discard_count, sync_history_size());
             for (size_t i = 0, size = m_arrays->changesets.size(); i < size; ++i) {
                 logger.debug("- %1: ident(%2) changeset_size(%3) remote_version(%4)", i,
                              m_arrays->origin_file_idents.get(i), m_arrays->changesets.get(i).size(),
@@ -83,10 +83,15 @@ void ClientHistory::set_client_reset_adjustments(
             auto i = size_t(version - m_sync_history_base_version);
             util::compression::allocate_and_compress_nonportable(arena, changeset, compressed);
             m_arrays->changesets.set(i, BinaryData{compressed.data(), compressed.size()}); // Throws
-            m_arrays->remote_versions.set(i, server_version.version);
             m_arrays->reciprocal_transforms.set(i, BinaryData());
-            logger.debug("Updating %1: client_version(%2) changeset_size(%3) server_version(%4)", i, version,
-                         compressed.size(), server_version.version);
+        }
+        // Server version is updated for *every* entry in the sync history to ensure that server versions don't
+        // decrease.
+        for (size_t i = 0, size = m_arrays->remote_versions.size(); i < size; ++i) {
+            m_arrays->remote_versions.set(i, server_version.version);
+            version_type version = m_sync_history_base_version + i;
+            logger.debug("Updating %1: client_version(%2) changeset_size(%3) server_version(%4)", i, version + 1,
+                         m_arrays->changesets.get(i).size(), server_version.version);
         }
     }
     logger.debug("New uploadable bytes after client reset adjustment: %1", uploadable_bytes);

test/object-store/sync/flx_sync.cpp

@@ -582,6 +582,7 @@ TEST_CASE("flx: client reset", "[sync][flx][client reset][baas]") {
     config_remote.path += ".remote";
     const std::string str_field_value = "foo";
     const int64_t local_added_int = 100;
+    const int64_t local_added_int2 = 150;
     const int64_t remote_added_int = 200;
     size_t before_reset_count = 0;
     size_t after_reset_count = 0;
@@ -745,6 +746,63 @@ TEST_CASE("flx: client reset", "[sync][flx][client reset][baas]") {
             ->run();
     }
 
+    SECTION("Recover: offline writes interleaved with subscriptions and empty writes") {
+        config_local.sync_config->client_resync_mode = ClientResyncMode::Recover;
+        auto&& [reset_future, reset_handler] = make_client_reset_handler();
+        config_local.sync_config->notify_after_client_reset = reset_handler;
+        auto test_reset = reset_utils::make_baas_flx_client_reset(config_local, config_remote, harness.session());
+        test_reset
+            ->make_local_changes([&](SharedRealm local_realm) {
+                // The sequence of events bellow generates five changesets:
+                //  1. create sub1 => empty changeset
+                //  2. create local_added_int object
+                //  3. create empty changeset
+                //  4. create sub2 => empty changeset
+                //  5. create local_added_int2 object
+                //
+                // Before sending 'sub2' to the server, an UPLOAD message is sent first.
+                // The upload message contains changeset 2. (local_added_int) with the cursor
+                // of changeset 3. (empty changeset).
+
+                add_subscription_for_new_object(local_realm, str_field_value, local_added_int);
+                // Commit empty changeset.
+                local_realm->begin_transaction();
+                local_realm->commit_transaction();
+                add_subscription_for_new_object(local_realm, str_field_value, local_added_int2);
+            })
+            ->make_remote_changes([&](SharedRealm remote_realm) {
+                add_subscription_for_new_object(remote_realm, str_field_value, remote_added_int);
+                sync::SubscriptionSet::State actual =
+                    remote_realm->get_latest_subscription_set()
+                        .get_state_change_notification(sync::SubscriptionSet::State::Complete)
+                        .get();
+                REQUIRE(actual == sync::SubscriptionSet::State::Complete);
+            })
+            ->on_post_reset([&, client_reset_future = std::move(reset_future)](SharedRealm local_realm) {
+                ClientResyncMode mode = client_reset_future.get();
+                REQUIRE(mode == ClientResyncMode::Recover);
+                auto subs = local_realm->get_latest_subscription_set();
+                subs.get_state_change_notification(sync::SubscriptionSet::State::Complete).get();
+                // make sure that the subscription for "foo" survived the reset
+                size_t count_of_foo = count_queries_with_str(subs, util::format("\"%1\"", str_field_value));
+                subs.refresh();
+                REQUIRE(subs.state() == sync::SubscriptionSet::State::Complete);
+                REQUIRE(count_of_foo == 1);
+                local_realm->refresh();
+                auto table = local_realm->read_group().get_table("class_TopLevel");
+                auto str_col = table->get_column_key("queryable_str_field");
+                auto int_col = table->get_column_key("queryable_int_field");
+                auto tv = table->where().equal(str_col, StringData(str_field_value)).find_all();
+                tv.sort(int_col);
+                // the objects we created while offline was recovered, and the remote object was downloaded
+                REQUIRE(tv.size() == 3);
+                CHECK(tv.get_object(0).get<Int>(int_col) == local_added_int);
+                CHECK(tv.get_object(1).get<Int>(int_col) == local_added_int2);
+                CHECK(tv.get_object(2).get<Int>(int_col) == remote_added_int);
+            })
+            ->run();
+    }
+
     auto validate_integrity_of_arrays = [](TableRef table) -> size_t {
         auto sum_col = table->get_column_key("sum_of_list_field");
         auto array_col = table->get_column_key("list_of_ints_field");