Receiving write_not_allowed error crashes the app (#7002)

* Make compensating_write_server_version in ProtocolErrorInfo optional

* Integration test for write_not_allowed(230) error

* Update changelog

* Add comment

Author: danieltabacaru

CHANGELOG.md

@@ -5,8 +5,8 @@
 * None.
 
 ### Fixed
-* <How do the end-user experience this issue? what was the impact?> ([#????](https://github.com/realm/realm-core/issues/????), since v?.?.?)
 * Fixed issue with double delete when using the CAPI for timers in platform networking ([#6993](https://github.com/realm/realm-core/issues/6993), since v13.3.0).
+* Receiving a write_not_allowed error from the server would have led to a crash. ([#6978](https://github.com/realm/realm-core/issues/6978), since v13.2.0)
 
 ### Breaking changes
 * Platform Networking CAPI has been updated to provide separate functions (instead of 1) for executing callback handlers depending on purpose ([PR #6994](https://github.com/realm/realm-core/pull/6994)).

src/realm/sync/noinst/client_impl_base.cpp

@@ -1501,15 +1501,17 @@ void Session::gather_pending_compensating_writes(util::Span<Changeset> changeset
     REALM_ASSERT_DEBUG(
         std::is_sorted(m_pending_compensating_write_errors.begin(), m_pending_compensating_write_errors.end(),
                        [](const ProtocolErrorInfo& lhs, const ProtocolErrorInfo& rhs) {
-                           return lhs.compensating_write_server_version < rhs.compensating_write_server_version;
+                           REALM_ASSERT_DEBUG(lhs.compensating_write_server_version.has_value());
+                           REALM_ASSERT_DEBUG(rhs.compensating_write_server_version.has_value());
+                           return *lhs.compensating_write_server_version < *rhs.compensating_write_server_version;
                        }));
 #endif
 
     while (!m_pending_compensating_write_errors.empty() &&
-           m_pending_compensating_write_errors.front().compensating_write_server_version <=
+           *m_pending_compensating_write_errors.front().compensating_write_server_version <=
                changesets.back().version) {
         auto& cur_error = m_pending_compensating_write_errors.front();
-        REALM_ASSERT_3(cur_error.compensating_write_server_version, >=, changesets.front().version);
+        REALM_ASSERT_3(*cur_error.compensating_write_server_version, >=, changesets.front().version);
         out->push_back(std::move(cur_error));
         m_pending_compensating_write_errors.pop_front();
     }
@@ -1552,7 +1554,7 @@ void Session::integrate_changesets(ClientReplication& repl, const SyncProgress&
     for (const auto& pending_error : pending_compensating_write_errors) {
         logger.info("Reporting compensating write for client version %1 in server version %2: %3",
                     pending_error.compensating_write_rejected_client_version,
-                    pending_error.compensating_write_server_version, pending_error.message);
+                    *pending_error.compensating_write_server_version, pending_error.message);
         try {
             on_connection_state_changed(
                 m_conn.get_state(),
@@ -2561,6 +2563,7 @@ Status Session::receive_error_message(const ProtocolErrorInfo& info)
         // If the client is not active, the compensating writes will not be processed now, but will be
         // sent again the next time the client connects
         if (m_state == Active) {
+            REALM_ASSERT(info.compensating_write_server_version.has_value());
             m_pending_compensating_write_errors.push_back(info);
         }
         return Status::OK();

src/realm/sync/noinst/protocol_codec.hpp

@@ -304,8 +304,12 @@ class ClientProtocol {
                             info.compensating_writes.push_back(std::move(cwei));
                         }
 
-                        info.compensating_write_server_version =
-                            json.at("compensatingWriteServerVersion").get<int64_t>();
+                        // Not provided when 'write_not_allowed' (230) error is received from the server.
+                        if (auto server_version = json.find("compensatingWriteServerVersion");
+                            server_version != json.end()) {
+                            info.compensating_write_server_version =
+                                std::make_optional<version_type>(server_version->get<int64_t>());
+                        }
                         info.compensating_write_rejected_client_version =
                             json.at("rejectedClientVersion").get<int64_t>();
                     }

src/realm/sync/protocol.hpp

@@ -289,7 +289,7 @@ struct ProtocolErrorInfo {
     bool client_reset_recovery_is_disabled = false;
     std::optional<bool> should_client_reset;
     std::optional<std::string> log_url;
-    version_type compensating_write_server_version = 0;
+    std::optional<version_type> compensating_write_server_version;
     version_type compensating_write_rejected_client_version = 0;
     std::vector<CompensatingWriteErrorInfo> compensating_writes;
     std::optional<ResumptionDelayInfo> resumption_delay_interval;

test/object-store/sync/flx_sync.cpp

@@ -3078,6 +3078,78 @@ TEST_CASE("flx: data ingest - dev mode", "[sync][flx][data ingest][baas]") {
         schema);
 }
 
+TEST_CASE("flx: data ingest - write not allowed", "[sync][flx][data ingest][baas]") {
+    AppCreateConfig::ServiceRole role;
+    role.name = "asymmetric_write_perms";
+
+    AppCreateConfig::ServiceRoleDocumentFilters doc_filters;
+    doc_filters.read = true;
+    doc_filters.write = false;
+    role.document_filters = doc_filters;
+
+    role.insert_filter = true;
+    role.delete_filter = true;
+    role.read = true;
+    role.write = true;
+
+    Schema schema({
+        {"Asymmetric",
+         ObjectSchema::ObjectType::TopLevelAsymmetric,
+         {
+             {"_id", PropertyType::ObjectId, Property::IsPrimary{true}},
+             {"location", PropertyType::String | PropertyType::Nullable},
+             {"embedded_obj", PropertyType::Object | PropertyType::Nullable, "Embedded"},
+         }},
+        {"Embedded",
+         ObjectSchema::ObjectType::Embedded,
+         {
+             {"value", PropertyType::String | PropertyType::Nullable},
+         }},
+    });
+    FLXSyncTestHarness::ServerSchema server_schema{schema, {}, {role}};
+    FLXSyncTestHarness harness("asymmetric_sync", server_schema);
+
+    auto error_received_pf = util::make_promise_future<void>();
+    SyncTestFile config(harness.app()->current_user(), harness.schema(), SyncConfig::FLXSyncEnabled{});
+    config.sync_config->on_sync_client_event_hook =
+        [promise = util::CopyablePromiseHolder(std::move(error_received_pf.promise))](
+            std::weak_ptr<SyncSession> weak_session, const SyncClientHookData& data) mutable {
+            if (data.event != SyncClientHookEvent::ErrorMessageReceived) {
+                return SyncClientHookAction::NoAction;
+            }
+            auto session = weak_session.lock();
+            REQUIRE(session);
+
+            auto error_code = sync::ProtocolError(data.error_info->raw_error_code);
+
+            if (error_code == sync::ProtocolError::initial_sync_not_completed) {
+                return SyncClientHookAction::NoAction;
+            }
+
+            REQUIRE(error_code == sync::ProtocolError::write_not_allowed);
+            REQUIRE_FALSE(data.error_info->compensating_write_server_version.has_value());
+            REQUIRE_FALSE(data.error_info->compensating_writes.empty());
+            promise.get_promise().emplace_value();
+
+            return SyncClientHookAction::EarlyReturn;
+        };
+
+    auto realm = Realm::get_shared_realm(config);
+
+    // Create an asymmetric object and upload it to the server.
+    {
+        realm->begin_transaction();
+        CppContext c(realm);
+        Object::create(c, realm, "Asymmetric",
+                       std::any(AnyDict{{"_id", ObjectId::gen()}, {"embedded_obj", AnyDict{{"value", "foo"s}}}}));
+        realm->commit_transaction();
+        wait_for_upload(*realm);
+    }
+
+    error_received_pf.future.get();
+    realm->close();
+}
+
 TEST_CASE("flx: send client error", "[sync][flx][baas]") {
     FLXSyncTestHarness harness("flx_client_error");
 
@@ -3607,7 +3679,7 @@ TEST_CASE("flx: compensating write errors get re-sent across sessions", "[sync][
         std::lock_guard<std::mutex> lk(errors_mutex);
         for (const auto& compensating_write : data.error_info->compensating_writes) {
             error_to_download_version.emplace_back(compensating_write.primary_key.get_object_id(),
-                                                   data.error_info->compensating_write_server_version);
+                                                   *data.error_info->compensating_write_server_version);
         }
 
         return SyncClientHookAction::NoAction;