RCORE-2013 Allow non-streaming download callback on subsequent sync batches (#7561)

* Allow non-streaming download callback on subsequent sync batches and also on initial flx sync after query version 0
* Remove solved fixme and workaround in test for query version 0
* Resolve, refactor and consolidate check for SyncProgressNotifier after the fix
* Delay download progress notification also until progress for new data transfer is signaled with fresh progress value and transferred bytes

Author: Kirill Burtsev

CHANGELOG.md

@@ -13,6 +13,7 @@
 * Deleting the active user left the active user unset rather than selecting another logged-in user as the active user like logging out and removing users did. ([PR #7300](https://github.com/realm/realm-core/pull/7300)).
 * Fix compilation errors when using command-line `swift build` ([#7587](https://github.com/realm/realm-core/pull/7587), since v14.5.1).
 * Fixed crash when integrating removal of already removed dictionary key ([#7488](https://github.com/realm/realm-core/issues/7488), since v10.0.0).
+* Non-streaming download sync progress notification is fixed for flexible sync Realms where before it was sometimes stopping to emit values right after the registration of the callback (PR [#7561](https://github.com/realm/realm-core/issues/7561)).
 
 ### Breaking changes
 * The following things have been renamed or moved as part of moving all of the App Services functionality to the app namespace:
@@ -33,6 +34,7 @@
 * Sync user management has been removed from SyncManager. This functionality was already additionally available on App. ([PR #7300](https://github.com/realm/realm-core/pull/7300).
 * AuditConfig now has a base_file_path field which must be set by the SDK rather than inheriting it from the SyncManager. ([PR #7300](https://github.com/realm/realm-core/pull/7300).
 * App::switch_user() no longer returns a user. The return value was always exactly the passed-in user and any code which needs it can just use that. ([PR #7300](https://github.com/realm/realm-core/pull/7300).
+* Non-streaming download progress callback no longer stops reporting values immediately after the registration (if the progress update has happened earlier), but waits for the next batch of data to start syncing to report its progress, since the previous behaviour was not useful (PR [#7561](https://github.com/realm/realm-core/issues/7561)).
 
 ### Compatibility
 * Fileformat: Generates files with format v24. Reads and automatically upgrade from fileformat v10. If you want to upgrade from an earlier file format version you will have to use RealmCore v13.x.y or earlier.
@@ -298,7 +300,7 @@
 * Use `clonefile()` when possible in `File::copy()` on Apple platforms for faster copying. ([PR #7341](https://github.com/realm/realm-core/pull/7341)).
 
 ### Fixed
-* Fixed queries like `indexed_property == NONE {x}` which mistakenly matched on only x instead of not x. This only applies when an indexed property with equality (==, or IN) matches with `NONE` on a list of one item. If the constant list contained more than one value then it was working correctly. ([realm-js #7862](https://github.com/realm/realm-java/issues/7862), since v12.5.0) 
+* Fixed queries like `indexed_property == NONE {x}` which mistakenly matched on only x instead of not x. This only applies when an indexed property with equality (==, or IN) matches with `NONE` on a list of one item. If the constant list contained more than one value then it was working correctly. ([realm-js #7862](https://github.com/realm/realm-java/issues/7862), since v12.5.0)
 * 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)
 * Fixed invalid data in error reason string when registering a subscription change notification after the subscription has already failed. ([#6839](https://github.com/realm/realm-core/issues/6839), since v11.8.0)
 * Fixed crash in fulltext index using prefix search with no matches ([#7309](https://github.com/realm/realm-core/issues/7309), since v13.18.0)

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

@@ -291,10 +291,10 @@ static bool check_for_redirect_response(const app::AppError& error)
     return false;
 }
 
-util::UniqueFunction<void(util::Optional<app::AppError>)>
+util::UniqueFunction<void(std::optional<app::AppError>)>
 SyncSession::handle_refresh(const std::shared_ptr<SyncSession>& session, bool restart_session)
 {
-    return [session, restart_session](util::Optional<app::AppError> error) {
+    return [session, restart_session](std::optional<app::AppError> error) {
         auto session_user = session->user();
         if (!session_user) {
             util::CheckedUniqueLock lock(session->m_state_mutex);
@@ -651,7 +651,7 @@ void SyncSession::handle_error(sync::SessionErrorInfo error)
 {
     enum class NextStateAfterError { none, inactive, error };
     auto next_state = error.is_fatal ? NextStateAfterError::error : NextStateAfterError::none;
-    util::Optional<ShouldBackup> delete_file;
+    std::optional<ShouldBackup> delete_file;
     bool log_out_user = false;
     bool unrecognized_by_client = false;
 
@@ -966,7 +966,7 @@ void SyncSession::create_sync_session()
     // Sets up the connection state listener. This callback is used for both reporting errors as well as changes to
     // the connection state.
     m_session->set_connection_state_change_listener(
-        [weak_self](sync::ConnectionState state, util::Optional<sync::SessionErrorInfo> error) {
+        [weak_self](sync::ConnectionState state, std::optional<sync::SessionErrorInfo> error) {
             using cs = sync::ConnectionState;
             ConnectionState new_state = [&] {
                 switch (state) {
@@ -1520,15 +1520,15 @@ uint64_t SyncProgressNotifier::register_callback(std::function<ProgressNotifierC
     {
         std::lock_guard<std::mutex> lock(m_mutex);
         token_value = m_progress_notifier_token++;
-        NotifierPackage package{std::move(notifier), util::none, m_local_transaction_version, is_streaming,
+        NotifierPackage package{std::move(notifier), m_local_transaction_version, is_streaming,
                                 direction == NotifierType::download};
         if (!m_current_progress) {
             // Simply register the package, since we have no data yet.
             m_packages.emplace(token_value, std::move(package));
             return token_value;
         }
         bool skip_registration = false;
-        invocation = package.create_invocation(*m_current_progress, skip_registration);
+        invocation = package.create_invocation(*m_current_progress, skip_registration, true);
         if (skip_registration) {
             token_value = 0;
         }
@@ -1573,10 +1573,11 @@ void SyncProgressNotifier::set_local_version(uint64_t snapshot_version)
 }
 
 util::UniqueFunction<void()>
-SyncProgressNotifier::NotifierPackage::create_invocation(Progress const& current_progress, bool& is_expired)
+SyncProgressNotifier::NotifierPackage::create_invocation(Progress const& current_progress, bool& is_expired,
+                                                         bool initial_registration)
 {
     uint64_t transferred = is_download ? current_progress.downloaded : current_progress.uploaded;
-    uint64_t transferrable = is_download ? current_progress.downloadable : current_progress.uploadable;
+    uint64_t transferable = is_download ? current_progress.downloadable : current_progress.uploadable;
     double progress_estimate = is_download ? current_progress.download_estimate : current_progress.upload_estimate;
 
     // If the sync client has not yet processed all of the local
@@ -1585,21 +1586,36 @@ SyncProgressNotifier::NotifierPackage::create_invocation(Progress const& current
     if (!is_download && snapshot_version > current_progress.snapshot_version)
         return [] {};
 
-    if (!is_streaming) {
-        // The initial download size we get from the server is the uncompacted
-        // size, and so the download may complete before we actually receive
-        // that much data. When that happens, transferrable will drop and we
-        // need to use the new value instead of the captured one.
-        if (!captured_transferrable || *captured_transferrable > transferrable)
-            captured_transferrable = transferrable;
-        transferrable = *captured_transferrable;
+    // for download only invoke the callback on registration if is in active data transfer,
+    // otherwise delay notifying until an update with the new transfer signaled
+    if (is_download && !started_notifying && progress_estimate >= 1) {
+        if (initial_registration) {
+            initial_transferred = transferred;
+            return [] {};
+        }
+        else if (initial_transferred == transferred)
+            return [] {};
     }
 
-    // A notifier is expired if at least as many bytes have been transferred
-    // as were originally considered transferrable.
-    is_expired = !is_streaming && transferred >= transferrable;
+    started_notifying = true;
+
+    // only capture and adjust transferable bytes for upload non-streaming to provide
+    // the progress of upload for the callback registered right after the commit
+    if (!is_streaming && !is_download) {
+        if (!captured_transferable || *captured_transferable > transferable)
+            captured_transferable = transferable;
+        transferable = *captured_transferable;
+    }
+
+    // A notifier is expired for upload if at least as many bytes have been transferred
+    // as were originally considered transferable based on local committed version
+    // on callback registration, or when simply 1.0 progress is reached for download
+    // since the amount of bytes is not precisely known until the end
+    if (!is_streaming)
+        is_expired = is_download ? progress_estimate >= 1 : transferred >= transferable;
+
     return [=, notifier = notifier] {
-        notifier(transferred, transferrable, progress_estimate);
+        notifier(transferred, transferable, progress_estimate);
     };
 }
 

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

@@ -27,7 +27,6 @@
 
 #include <realm/util/checked_mutex.hpp>
 #include <realm/util/future.hpp>
-#include <realm/util/optional.hpp>
 #include <realm/version_id.hpp>
 
 #include <mutex>
@@ -80,12 +79,14 @@ class SyncProgressNotifier {
     // can register upon this session.
     struct NotifierPackage {
         std::function<ProgressNotifierCallback> notifier;
-        util::Optional<uint64_t> captured_transferrable;
         uint64_t snapshot_version;
         bool is_streaming;
         bool is_download;
-
-        util::UniqueFunction<void()> create_invocation(const Progress&, bool&);
+        bool started_notifying = false;
+        uint64_t initial_transferred = 0;
+        std::optional<uint64_t> captured_transferable;
+        util::UniqueFunction<void()> create_invocation(const Progress&, bool& is_expired,
+                                                       bool initial_registration = false);
     };
 
     // A counter used as a token to identify progress notifier callbacks registered on this session.
@@ -96,8 +97,10 @@ class SyncProgressNotifier {
     // Will be `none` until we've received the initial notification from sync.  Note that this
     // happens only once ever during the lifetime of a given `SyncSession`, since these values are
     // expected to semi-monotonically increase, and a lower-bounds estimate is still useful in the
-    // event more up-to-date information isn't yet available.
-    util::Optional<Progress> m_current_progress;
+    // event more up-to-date information isn't yet available.  FIXME: If we support transparent
+    // client reset in the future, we might need to reset the progress state variables if the Realm
+    // is rolled back.
+    std::optional<Progress> m_current_progress;
 
     std::unordered_map<uint64_t, NotifierPackage> m_packages;
 };
@@ -389,7 +392,9 @@ class SyncSession : public std::enable_shared_from_this<SyncSession> {
     }
     // }
 
-    static util::UniqueFunction<void(util::Optional<app::AppError>)>
+    std::shared_ptr<SyncManager> sync_manager() const REQUIRES(!m_state_mutex);
+
+    static util::UniqueFunction<void(std::optional<app::AppError>)>
     handle_refresh(const std::shared_ptr<SyncSession>&, bool);
 
     // Initialize or tear down the subscription store based on whether or not flx_sync_requested is true

src/realm/sync/client.cpp

@@ -1131,6 +1131,11 @@ void SessionImpl::init_progress_handler()
     m_wrapper.init_progress_handler();
 }
 
+void SessionImpl::enable_progress_notifications()
+{
+    m_wrapper.m_reliable_download_progress = true;
+}
+
 void SessionImpl::notify_upload_progress()
 {
     if (m_state != State::Active)
@@ -1751,20 +1756,12 @@ inline void SessionWrapper::finalize_before_actualization() noexcept
 inline void SessionWrapper::on_upload_progress(bool only_if_new_uploadable_data)
 {
     REALM_ASSERT(!m_finalized);
-
-    // don't set the flag in case of the progress change of local origin
-    // progress should be delayed until first DOWNLOAD message received
-    // since uploads are not allowed before that and can't progress
-    if (!only_if_new_uploadable_data)
-        m_reliable_download_progress = true;
-
     report_progress(/* is_download = */ false, only_if_new_uploadable_data); // Throws
 }
 
 inline void SessionWrapper::on_download_progress(const std::optional<uint64_t>& bootstrap_store_bytes)
 {
     REALM_ASSERT(!m_finalized);
-    m_reliable_download_progress = true;
     m_bootstrap_store_bytes = bootstrap_store_bytes;
     report_progress(/* is_download = */ true); // Throws
 }

src/realm/sync/noinst/client_impl_base.cpp

@@ -2367,6 +2367,9 @@ Status Session::receive_download_message(const DownloadMessage& message)
     bool is_flx = m_conn.is_flx_sync_connection();
     int64_t query_version = is_flx ? *message.query_version : 0;
 
+    if (!is_flx || query_version > 0)
+        enable_progress_notifications();
+
     // If this is a PBS connection, then every download message is its own complete batch.
     bool last_in_batch = is_flx ? *message.last_in_batch : true;
     auto batch_state = last_in_batch ? sync::DownloadBatchState::LastInBatch : sync::DownloadBatchState::MoreToCome;

src/realm/sync/noinst/client_impl_base.hpp

@@ -1200,6 +1200,7 @@ class ClientImpl::Session {
     bool is_steady_state_download_message(DownloadBatchState batch_state, int64_t query_version);
 
     void init_progress_handler();
+    void enable_progress_notifications();
     void notify_upload_progress();
     void update_download_estimate(double download_estimate);
     void notify_download_progress(const std::optional<uint64_t>& bootstrap_store_bytes = {});