Wait for MARK message to mark a SubscriptionSet as Complete

evergreen/config_overrides.json

@@ -11,7 +11,6 @@
         }
     },
     "sync": {
-        "brokerRegistryCacheTTL": "0s",
         "allowSyncSessionTestCommands": true
     }
 }

src/realm.h

@@ -3565,10 +3565,11 @@ typedef struct realm_flx_sync_subscription_desc realm_flx_sync_subscription_desc
 typedef enum realm_flx_sync_subscription_set_state {
     RLM_SYNC_SUBSCRIPTION_UNCOMMITTED = 0,
     RLM_SYNC_SUBSCRIPTION_PENDING,
-    RLM_SYNC_BOOTSTRAPPING,
+    RLM_SYNC_SUBSCRIPTION_BOOTSTRAPPING,
     RLM_SYNC_SUBSCRIPTION_COMPLETE,
     RLM_SYNC_SUBSCRIPTION_ERROR,
     RLM_SYNC_SUBSCRIPTION_SUPERSEDED,
+    RLM_SYNC_SUBSCRIPTION_AWAITING_MARK,
 } realm_flx_sync_subscription_set_state_e;
 typedef void (*realm_sync_on_subscription_state_changed_t)(realm_userdata_t userdata,
                                                            realm_flx_sync_subscription_set_state_e state);

src/realm/object-store/c_api/sync.cpp

@@ -212,6 +212,49 @@ static_assert(realm_sync_error_action_e(ProtocolErrorInfo::Action::ClientReset)
               RLM_SYNC_ERROR_ACTION_CLIENT_RESET);
 static_assert(realm_sync_error_action_e(ProtocolErrorInfo::Action::ClientResetNoRecovery) ==
               RLM_SYNC_ERROR_ACTION_CLIENT_RESET_NO_RECOVERY);
+
+SubscriptionSet::State sub_state_from_c_enum(realm_flx_sync_subscription_set_state_e value)
+{
+    switch (static_cast<realm_flx_sync_subscription_set_state>(value)) {
+        case RLM_SYNC_SUBSCRIPTION_PENDING:
+            return SubscriptionSet::State::Pending;
+        case RLM_SYNC_SUBSCRIPTION_BOOTSTRAPPING:
+            return SubscriptionSet::State::Bootstrapping;
+        case RLM_SYNC_SUBSCRIPTION_AWAITING_MARK:
+            return SubscriptionSet::State::AwaitingMark;
+        case RLM_SYNC_SUBSCRIPTION_COMPLETE:
+            return SubscriptionSet::State::Complete;
+        case RLM_SYNC_SUBSCRIPTION_ERROR:
+            return SubscriptionSet::State::Error;
+        case RLM_SYNC_SUBSCRIPTION_SUPERSEDED:
+            return SubscriptionSet::State::Superseded;
+        case RLM_SYNC_SUBSCRIPTION_UNCOMMITTED:
+            return SubscriptionSet::State::Uncommitted;
+    }
+    REALM_UNREACHABLE();
+}
+
+realm_flx_sync_subscription_set_state_e sub_state_to_c_enum(SubscriptionSet::State state)
+{
+    switch (state) {
+        case SubscriptionSet::State::Pending:
+            return RLM_SYNC_SUBSCRIPTION_PENDING;
+        case SubscriptionSet::State::Bootstrapping:
+            return RLM_SYNC_SUBSCRIPTION_BOOTSTRAPPING;
+        case SubscriptionSet::State::AwaitingMark:
+            return RLM_SYNC_SUBSCRIPTION_AWAITING_MARK;
+        case SubscriptionSet::State::Complete:
+            return RLM_SYNC_SUBSCRIPTION_COMPLETE;
+        case SubscriptionSet::State::Error:
+            return RLM_SYNC_SUBSCRIPTION_ERROR;
+        case SubscriptionSet::State::Uncommitted:
+            return RLM_SYNC_SUBSCRIPTION_UNCOMMITTED;
+        case SubscriptionSet::State::Superseded:
+            return RLM_SYNC_SUBSCRIPTION_SUPERSEDED;
+    }
+    REALM_UNREACHABLE();
+}
+
 } // namespace
 
 static realm_sync_error_code_t to_capi(const std::error_code& error_code, std::string& message)
@@ -569,8 +612,8 @@ realm_sync_on_subscription_set_state_change_wait(const realm_flx_sync_subscripti
 {
     REALM_ASSERT(subscription_set != nullptr);
     SubscriptionSet::State state =
-        subscription_set->get_state_change_notification(SubscriptionSet::State{notify_when}).get();
-    return realm_flx_sync_subscription_set_state_e(static_cast<int>(state));
+        subscription_set->get_state_change_notification(sub_state_from_c_enum(notify_when)).get();
+    return sub_state_to_c_enum(state);
 }
 
 RLM_API bool
@@ -581,13 +624,12 @@ realm_sync_on_subscription_set_state_change_async(const realm_flx_sync_subscript
 {
     REALM_ASSERT(subscription_set != nullptr && callback != nullptr);
     return wrap_err([&]() {
-        auto future_state = subscription_set->get_state_change_notification(SubscriptionSet::State{notify_when});
+        auto future_state = subscription_set->get_state_change_notification(sub_state_from_c_enum(notify_when));
         std::move(future_state)
             .get_async([callback, userdata = SharedUserdata(userdata, FreeUserdata(userdata_free))](
                            const StatusWith<SubscriptionSet::State>& state) -> void {
                 if (state.is_ok())
-                    callback(userdata.get(),
-                             realm_flx_sync_subscription_set_state_e(static_cast<int>(state.get_value())));
+                    callback(userdata.get(), sub_state_to_c_enum(state.get_value()));
                 else
                     callback(userdata.get(), realm_flx_sync_subscription_set_state_e::RLM_SYNC_SUBSCRIPTION_ERROR);
             });

src/realm/object-store/sync/sync_session.cpp

@@ -401,28 +401,7 @@ void SyncSession::download_fresh_realm(sync::ProtocolErrorInfo::Action server_re
         std::move(fresh_mut_sub)
             .commit()
             .get_state_change_notification(sync::SubscriptionSet::State::Complete)
-            .then([=, weak_self = weak_from_this()](sync::SubscriptionSet::State) {
-                auto pf = util::make_promise_future<void>();
-                sync_session->wait_for_download_completion([=, promise = std::move(pf.promise)](
-                                                               std::error_code ec) mutable {
-                    auto strong_self = weak_self.lock();
-                    if (!strong_self) {
-                        return promise.set_error({ErrorCodes::RuntimeError,
-                                                  "SyncSession was destroyed before download could be completed"});
-                    }
-
-                    if (ec) {
-                        return promise.set_error(
-                            {ErrorCodes::RuntimeError,
-                             util::format("Error waiting for download completion for fresh realm (code: %1): %2",
-                                          ec.value(), ec.message())});
-                    }
-
-                    promise.emplace_value();
-                });
-                return std::move(pf.future);
-            })
-            .get_async([=, weak_self = weak_from_this()](Status s) {
+            .get_async([=, weak_self = weak_from_this()](StatusWith<sync::SubscriptionSet::State> s) {
                 // Keep the sync session alive while it's downloading, but then close
                 // it immediately
                 sync_session->close();
@@ -431,7 +410,8 @@ void SyncSession::download_fresh_realm(sync::ProtocolErrorInfo::Action server_re
                         strong_self->handle_fresh_realm_downloaded(db, none, server_requests_action);
                     }
                     else {
-                        strong_self->handle_fresh_realm_downloaded(nullptr, s.reason(), server_requests_action);
+                        strong_self->handle_fresh_realm_downloaded(nullptr, s.get_status().reason(),
+                                                                   server_requests_action);
                     }
                 }
             });

src/realm/sync/client.cpp

@@ -252,6 +252,7 @@ class SessionWrapper final : public util::AtomicRefCountBase, public SyncTransac
     int64_t m_flx_active_version = 0;
     int64_t m_flx_last_seen_version = 0;
     int64_t m_flx_latest_version = 0;
+    int64_t m_flx_pending_mark_version = 0;
     std::unique_ptr<PendingBootstrapStore> m_flx_pending_bootstrap_store;
 
     bool m_initiated = false;
@@ -785,6 +786,7 @@ void SessionImpl::process_pending_flx_bootstrap()
     VersionInfo new_version;
     SyncProgress progress;
     int64_t query_version = -1;
+    size_t changesets_processed = 0;
     while (bootstrap_store->has_pending()) {
         auto pending_batch = bootstrap_store->peek_pending(batch_size_in_bytes);
         if (!pending_batch.progress) {
@@ -803,7 +805,6 @@ void SessionImpl::process_pending_flx_bootstrap()
             throw IntegrationException(ClientError::bad_changeset, "simulated failure");
         }
 
-
         history.integrate_server_changesets(
             *pending_batch.progress, &downloadable_bytes, pending_batch.changesets, new_version, batch_state, logger,
             [&](const TransactionRef& tr, size_t count) {
@@ -812,9 +813,7 @@ void SessionImpl::process_pending_flx_bootstrap()
             },
             get_transact_reporter());
         progress = *pending_batch.progress;
-
-        REALM_ASSERT(call_debug_hook(SyncClientHookEvent::DownloadMessageIntegrated, progress, query_version,
-                                     batch_state, pending_batch.changesets.size()) == SyncClientHookAction::NoAction);
+        changesets_processed += pending_batch.changesets.size();
 
         logger.info("Integrated %1 changesets from pending bootstrap for query version %2, producing client version "
                     "%3. %4 changesets remaining in bootstrap",
@@ -826,6 +825,10 @@ void SessionImpl::process_pending_flx_bootstrap()
     REALM_ASSERT_3(query_version, !=, -1);
     m_wrapper.on_sync_progress();
     on_flx_sync_progress(query_version, DownloadBatchState::LastInBatch);
+
+    auto action = call_debug_hook(SyncClientHookEvent::BootstrapProcessed, progress, query_version,
+                                  DownloadBatchState::LastInBatch, changesets_processed);
+    REALM_ASSERT(action == SyncClientHookAction::NoAction);
 }
 
 void SessionImpl::on_new_flx_subscription_set(int64_t new_version)
@@ -950,8 +953,10 @@ SessionWrapper::SessionWrapper(ClientImpl& client, DBRef db, std::shared_ptr<Sub
     REALM_ASSERT(dynamic_cast<ClientReplication*>(m_db->get_replication()));
 
     if (m_flx_subscription_store) {
-        std::tie(m_flx_active_version, m_flx_latest_version) =
-            m_flx_subscription_store->get_active_and_latest_versions();
+        auto versions_info = m_flx_subscription_store->get_active_and_latest_versions();
+        m_flx_active_version = versions_info.active;
+        m_flx_latest_version = versions_info.latest;
+        m_flx_pending_mark_version = versions_info.pending_mark;
     }
 }
 
@@ -1032,7 +1037,13 @@ void SessionWrapper::on_flx_sync_progress(int64_t new_version, DownloadBatchStat
                 return;
             }
             on_flx_sync_version_complete(new_version);
-            new_state = SubscriptionSet::State::Complete;
+            if (new_version == 0) {
+                new_state = SubscriptionSet::State::Complete;
+            }
+            else {
+                new_state = SubscriptionSet::State::AwaitingMark;
+                m_flx_pending_mark_version = new_version;
+            }
             break;
         case DownloadBatchState::MoreToCome:
             if (m_flx_last_seen_version == new_version) {
@@ -1438,6 +1449,16 @@ void SessionWrapper::on_download_completion()
         m_upload_completion_handlers.push_back(std::move(handler)); // Throws
         m_sync_completion_handlers.pop_back();
     }
+
+    if (m_flx_subscription_store && m_flx_pending_mark_version != SubscriptionSet::EmptyVersion) {
+        m_sess->logger.debug("Marking query version %1 as complete after receiving MARK message",
+                             m_flx_pending_mark_version);
+        auto mutable_subs = m_flx_subscription_store->get_mutable_by_version(m_flx_pending_mark_version);
+        mutable_subs.update_state(SubscriptionSet::State::Complete);
+        std::move(mutable_subs).commit();
+        m_flx_pending_mark_version = SubscriptionSet::EmptyVersion;
+    }
+
     util::LockGuard lock{m_client.m_mutex};
     if (m_staged_download_mark > m_reached_download_mark) {
         m_reached_download_mark = m_staged_download_mark;
@@ -1458,6 +1479,11 @@ void SessionWrapper::on_suspended(const SessionErrorInfo& error_info)
 void SessionWrapper::on_resumed()
 {
     m_suspended = false;
+    if (m_flx_subscription_store && m_flx_pending_mark_version != SubscriptionSet::EmptyVersion) {
+        m_sess->logger.debug("Requesting download notification for query version %1 after resume",
+                             m_flx_pending_mark_version);
+        m_sess->request_download_completion_notification();
+    }
     if (m_connection_state_change_listener) {
         ClientImpl::Connection& conn = m_sess->get_connection();
         if (conn.get_state() != ConnectionState::disconnected) {

src/realm/sync/config.hpp

@@ -134,6 +134,7 @@ enum class SyncClientHookEvent {
     DownloadMessageReceived,
     DownloadMessageIntegrated,
     BootstrapMessageProcessed,
+    BootstrapProcessed,
 };
 
 enum class SyncClientHookAction {

src/realm/sync/noinst/client_impl_base.cpp

@@ -1778,7 +1778,7 @@ void Session::send_query_change_message()
 
     m_last_sent_flx_query_version = latest_sub_set.version();
 
-    enlist_to_send(); // throws
+    request_download_completion_notification();
 }
 
 void Session::send_upload_message()