RCORE-2099 Restore progress notifier behavior when sync session is already caught up (#7681)

jbreams · web-flow · commit c280bdb17522 · 2024-05-20T16:01:27.000-04:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -5,8 +5,7 @@
 * None.
 
 ### Fixed
-* <How do the end-user experience this issue? what was the impact?> ([#????](https://github.com/realm/realm-core/issues/????), since v?.?.?)
-* None.
+* A non-streaming progress notifier would not immediately call its callback after registration. Instead you would have to wait for a download message to be received to get your first update - if you were already caught up when you registered the notifier you could end up waiting a long time for the server to deliver a download that would call/expire your notifier ([#7627](https://github.com/realm/realm-core/issues/7627), since v14.6.0).
 
 ### Breaking changes
 * None.
diff --git a/src/realm/object-store/sync/sync_session.cpp b/src/realm/object-store/sync/sync_session.cpp
@@ -821,10 +821,10 @@ void SyncSession::cancel_pending_waits(util::CheckedUniqueLock lock, Status erro
 
 void SyncSession::handle_progress_update(uint64_t downloaded, uint64_t downloadable, uint64_t uploaded,
                                          uint64_t uploadable, uint64_t snapshot_version, double download_estimate,
-                                         double upload_estimate)
+                                         double upload_estimate, int64_t query_version)
 {
     m_progress_notifier.update(downloaded, downloadable, uploaded, uploadable, snapshot_version, download_estimate,
-                               upload_estimate);
+                               upload_estimate, query_version);
 }
 
 static sync::Session::Config::ClientReset make_client_reset_config(const RealmConfig& base_config,
@@ -962,10 +962,10 @@ void SyncSession::create_sync_session()
     m_session->set_progress_handler([weak_self](uint_fast64_t downloaded, uint_fast64_t downloadable,
                                                 uint_fast64_t uploaded, uint_fast64_t uploadable,
                                                 uint_fast64_t snapshot_version, double download_estimate,
-                                                double upload_estimate) {
+                                                double upload_estimate, int64_t query_version) {
         if (auto self = weak_self.lock()) {
             self->handle_progress_update(downloaded, downloadable, uploaded, uploadable, snapshot_version,
-                                         download_estimate, upload_estimate);
+                                         download_estimate, upload_estimate, query_version);
         }
     });
 
@@ -1267,7 +1267,11 @@ void SyncSession::wait_for_download_completion(util::UniqueFunction<void(Status)
 uint64_t SyncSession::register_progress_notifier(std::function<ProgressNotifierCallback>&& notifier,
                                                  ProgressDirection direction, bool is_streaming)
 {
-    return m_progress_notifier.register_callback(std::move(notifier), direction, is_streaming);
+    int64_t pending_query_version = 0;
+    if (auto sub_store = get_flx_subscription_store()) {
+        pending_query_version = sub_store->get_version_info().latest;
+    }
+    return m_progress_notifier.register_callback(std::move(notifier), direction, is_streaming, pending_query_version);
 }
 
 void SyncSession::unregister_progress_notifier(uint64_t token)
@@ -1519,22 +1523,23 @@ void SyncSession::did_drop_external_reference()
 }
 
 uint64_t SyncProgressNotifier::register_callback(std::function<ProgressNotifierCallback> notifier,
-                                                 NotifierType direction, bool is_streaming)
+                                                 NotifierType direction, bool is_streaming,
+                                                 int64_t pending_query_version)
 {
     util::UniqueFunction<void()> invocation;
     uint64_t token_value = 0;
     {
         std::lock_guard<std::mutex> lock(m_mutex);
         token_value = m_progress_notifier_token++;
         NotifierPackage package{std::move(notifier), m_local_transaction_version, is_streaming,
-                                direction == NotifierType::download};
+                                direction == NotifierType::download, pending_query_version};
         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, true);
+        invocation = package.create_invocation(*m_current_progress, skip_registration);
         if (skip_registration) {
             token_value = 0;
         }
@@ -1553,13 +1558,14 @@ void SyncProgressNotifier::unregister_callback(uint64_t token)
 }
 
 void SyncProgressNotifier::update(uint64_t downloaded, uint64_t downloadable, uint64_t uploaded, uint64_t uploadable,
-                                  uint64_t snapshot_version, double download_estimate, double upload_estimate)
+                                  uint64_t snapshot_version, double download_estimate, double upload_estimate,
+                                  int64_t query_version)
 {
     std::vector<util::UniqueFunction<void()>> invocations;
     {
         std::lock_guard<std::mutex> lock(m_mutex);
-        m_current_progress = Progress{uploadable,      downloadable,      uploaded,        downloaded,
-                                      upload_estimate, download_estimate, snapshot_version};
+        m_current_progress = Progress{uploadable,      downloadable,      uploaded,         downloaded,
+                                      upload_estimate, download_estimate, snapshot_version, query_version};
 
         for (auto it = m_packages.begin(); it != m_packages.end();) {
             bool should_delete = false;
@@ -1579,49 +1585,50 @@ void SyncProgressNotifier::set_local_version(uint64_t snapshot_version)
 }
 
 util::UniqueFunction<void()>
-SyncProgressNotifier::NotifierPackage::create_invocation(Progress const& current_progress, bool& is_expired,
-                                                         bool initial_registration)
+SyncProgressNotifier::NotifierPackage::create_invocation(Progress const& current_progress, bool& is_expired)
 {
-    uint64_t transferred = is_download ? current_progress.downloaded : current_progress.uploaded;
+    uint64_t transfered = is_download ? current_progress.downloaded : current_progress.uploaded;
     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
-    // transactions then the uploadable data is incorrect and we should
-    // not invoke the callback
-    if (!is_download && snapshot_version > current_progress.snapshot_version)
-        return [] {};
-
-    // 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)
+    double estimate = is_download ? current_progress.download_estimate : current_progress.upload_estimate;
+
+    if (!is_streaming) {
+        // If the sync client has not yet processed all of the local
+        // transactions then the uploadable data is incorrect and we should
+        // not invoke the callback
+        if (!is_download && snapshot_version > current_progress.snapshot_version)
             return [] {};
-    }
 
-    started_notifying = true;
+        // If this is a non-streaming download progress update and this notifier was
+        // created for a later query version (e.g. we're currently downloading
+        // subscription set version zero, but subscription set version 1 existed
+        // when the notifier was registered), then we want to skip this callback.
+        if (is_download && current_progress.query_version < pending_query_version) {
+            return [] {};
+        }
 
-    // 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) {
+        // 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_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;
+        // Since we can adjust the transferrable downwards the estimate for uploads
+        // won't be correct since the sync client's view of the estimate is based on
+        // the total number of uploadable bytes available rather than the number of
+        // bytes this NotifierPackage was waiting to upload.
+        if (!is_download) {
+            estimate = transferable > 0 ? std::min(transfered / double(transferable), 1.0) : 0.0;
+        }
+    }
 
+    // A notifier is expired if at least as many bytes have been transferred
+    // as were originally considered transferrable.
+    is_expired =
+        !is_streaming && (transfered >= transferable && (!is_download || !pending_query_version || estimate >= 1.0));
     return [=, notifier = notifier] {
-        notifier(transferred, transferable, progress_estimate);
+        notifier(transfered, transferable, estimate);
     };
 }
 
diff --git a/src/realm/object-store/sync/sync_session.hpp b/src/realm/object-store/sync/sync_session.hpp
@@ -54,12 +54,13 @@ class SyncProgressNotifier {
     using ProgressNotifierCallback = void(uint64_t transferred_bytes, uint64_t transferrable_bytes,
                                           double progress_estimate);
 
-    uint64_t register_callback(std::function<ProgressNotifierCallback>, NotifierType direction, bool is_streaming);
+    uint64_t register_callback(std::function<ProgressNotifierCallback>, NotifierType direction, bool is_streaming,
+                               int64_t pending_query_version);
     void unregister_callback(uint64_t);
 
     void set_local_version(uint64_t);
     void update(uint64_t downloaded, uint64_t downloadable, uint64_t uploaded, uint64_t uploadable,
-                uint64_t snapshot_version, double download_estimate = 1.0, double upload_estimate = 1.0);
+                uint64_t snapshot_version, double download_estimate, double upload_estimate, int64_t query_version);
 
 private:
     mutable std::mutex m_mutex;
@@ -73,6 +74,7 @@ class SyncProgressNotifier {
         double upload_estimate;
         double download_estimate;
         uint64_t snapshot_version;
+        int64_t query_version;
     };
 
     // A PODS encapsulating some information for progress notifier callbacks a binding
@@ -82,11 +84,9 @@ class SyncProgressNotifier {
         uint64_t snapshot_version;
         bool is_streaming;
         bool is_download;
-        bool started_notifying = false;
-        uint64_t initial_transferred = 0;
+        int64_t pending_query_version = 0;
         std::optional<uint64_t> captured_transferable;
-        util::UniqueFunction<void()> create_invocation(const Progress&, bool& is_expired,
-                                                       bool initial_registration = false);
+        util::UniqueFunction<void()> create_invocation(const Progress&, bool& is_expired);
     };
 
     // A counter used as a token to identify progress notifier callbacks registered on this session.
@@ -174,7 +174,7 @@ class SyncSession : public std::enable_shared_from_this<SyncSession> {
     // Note that bindings should dispatch the callback onto a separate thread or queue
     // in order to avoid blocking the sync client.
     uint64_t register_progress_notifier(std::function<ProgressNotifierCallback>&&, ProgressDirection,
-                                        bool is_streaming);
+                                        bool is_streaming) REQUIRES(!m_state_mutex);
 
     // Unregister a previously registered notifier. If the token is invalid,
     // this method does nothing.
@@ -422,7 +422,7 @@ class SyncSession : public std::enable_shared_from_this<SyncSession> {
     void cancel_pending_waits(util::CheckedUniqueLock, Status) RELEASE(m_state_mutex);
     enum class ShouldBackup { yes, no };
     void update_error_and_mark_file_for_deletion(SyncError&, ShouldBackup) REQUIRES(m_state_mutex, !m_config_mutex);
-    void handle_progress_update(uint64_t, uint64_t, uint64_t, uint64_t, uint64_t, double, double);
+    void handle_progress_update(uint64_t, uint64_t, uint64_t, uint64_t, uint64_t, double, double, int64_t);
     void handle_new_flx_sync_query(int64_t version);
 
     void nonsync_transact_notify(VersionID::version_type) REQUIRES(!m_state_mutex);
diff --git a/src/realm/sync/client.cpp b/src/realm/sync/client.cpp
@@ -980,11 +980,11 @@ void SessionImpl::process_pending_flx_bootstrap()
                     std::chrono::duration_cast<std::chrono::milliseconds>(duration).count(),
                     pending_batch.remaining_changesets);
     }
-    on_changesets_integrated(new_version.realm_version, progress, changesets_processed > 0);
 
     REALM_ASSERT_3(query_version, !=, -1);
     on_flx_sync_progress(query_version, DownloadBatchState::LastInBatch);
 
+    on_changesets_integrated(new_version.realm_version, progress, changesets_processed > 0);
     auto action = call_debug_hook(SyncClientHookEvent::BootstrapProcessed, progress, query_version,
                                   DownloadBatchState::LastInBatch, changesets_processed);
     // NoAction/EarlyReturn are both valid no-op actions to take here.
@@ -1958,14 +1958,15 @@ void SessionWrapper::report_progress(bool is_download, bool only_if_new_uploadab
             ss << std::fixed << std::setprecision(4) << d;
             return ss.str();
         };
-        m_sess->logger.debug("Progress handler called, downloaded = %1, downloadable = %2, estimate = %3, "
-                             "uploaded = %4, uploadable = %5, estimate = %6, snapshot version = %7",
-                             p.downloaded, p.downloadable, to_str(download_estimate), p.uploaded, p.uploadable,
-                             to_str(upload_estimate), p.snapshot);
+        m_sess->logger.debug(
+            "Progress handler called, downloaded = %1, downloadable = %2, estimate = %3, "
+            "uploaded = %4, uploadable = %5, estimate = %6, snapshot version = %7, query_version = %8",
+            p.downloaded, p.downloadable, to_str(download_estimate), p.uploaded, p.uploadable,
+            to_str(upload_estimate), p.snapshot, m_flx_active_version);
     }
 
     m_progress_handler(p.downloaded, p.downloadable, p.uploaded, p.uploadable, p.snapshot, download_estimate,
-                       upload_estimate);
+                       upload_estimate, m_flx_last_seen_version);
 }
 
 util::Future<std::string> SessionWrapper::send_test_command(std::string body)
diff --git a/src/realm/sync/client.hpp b/src/realm/sync/client.hpp
@@ -162,7 +162,7 @@ class Session {
     using ProgressHandler = void(std::uint_fast64_t downloaded_bytes, std::uint_fast64_t downloadable_bytes,
                                  std::uint_fast64_t uploaded_bytes, std::uint_fast64_t uploadable_bytes,
                                  std::uint_fast64_t snapshot_version, double download_estimate,
-                                 double upload_estimate);
+                                 double upload_estimate, int64_t query_version);
     using WaitOperCompletionHandler = util::UniqueFunction<void(Status)>;
     using SSLVerifyCallback = bool(const std::string& server_address, port_type server_port, const char* pem_data,
                                    size_t pem_size, int preverify_ok, int depth);
diff --git a/test/object-store/sync/session/progress_notifications.cpp b/test/object-store/sync/session/progress_notifications.cpp
diff --git a/test/test_sync.cpp b/test/test_sync.cpp
