Derive upload completion entirely from the state of the history

Rather than tracking a bunch of derived state in-memory, check for upload
completion by checking if there are any unuploaded changesets. This is both
multiprocess-compatible and is more precise than the old checks, which had some
false-negatives.

Author: tgoyne

CHANGELOG.md

@@ -9,6 +9,7 @@
 * Fixed a change of mode from Strong to All when removing links from an embedded object that links to a tombstone. This affects sync apps that use embedded objects which have a `Lst<Mixed>` that contains a link to another top level object which has been deleted by another sync client (creating a tombstone locally). In this particular case, the switch would cause any remaining link removals to recursively delete the destination object if there were no other links to it. ([#7828](https://github.com/realm/realm-core/issues/7828), since 14.0.0-beta.0)
 * Fixed removing backlinks from the wrong objects if the link came from a nested list, nested dictionary, top-level dictionary, or list of mixed, and the source table had more than 256 objects. This could manifest as `array_backlink.cpp:112: Assertion failed: int64_t(value >> 1) == key.value` when removing an object. ([#7594](https://github.com/realm/realm-core/issues/7594), since v11 for dictionaries)
 * Fixed the collapse/rejoin of clusters which contained nested collections with links. This could manifest as `array.cpp:319: Array::move() Assertion failed: begin <= end [2, 1]` when removing an object. ([#7839](https://github.com/realm/realm-core/issues/7839), since the introduction of nested collections in v14.0.0-beta.0)
+* wait_for_upload_completion() was inconsistent in how it handled commits which did not produce any changesets to upload. Previously it would sometimes complete immediately if all commits waiting to be uploaded were empty, and at other times it would wait for a server roundtrip. It will now always complete immediately. ([PR #7796](https://github.com/realm/realm-core/pull/7796)).
 
 ### Breaking changes
 * None.
@@ -20,6 +21,7 @@
 
 ### Internals
 * Fixed `Table::remove_object_recursive` which wouldn't recursively follow links through a single `Mixed` property. This feature is exposed publicly on `Table` but no SDK currently uses it, so this is considered internal. ([#7829](https://github.com/realm/realm-core/issues/7829), likely since the introduction of Mixed)
+* Upload completion is now tracked in a multiprocess-compatible manner ([PR #7796](https://github.com/realm/realm-core/pull/7796)).
 
 ----------------------------------------------
 

src/realm/chunked_binary.cpp

@@ -45,6 +45,13 @@ bool ChunkedBinaryData::is_null() const
     return chunk.is_null();
 }
 
+bool ChunkedBinaryData::empty() const
+{
+    BinaryIterator copy = m_begin;
+    BinaryData chunk = copy.get_next();
+    return chunk.size() == 0;
+}
+
 char ChunkedBinaryData::operator[](size_t index) const
 {
     BinaryIterator copy = m_begin;

src/realm/chunked_binary.hpp

@@ -54,6 +54,9 @@ class ChunkedBinaryData {
     /// the first chunk points to the nullptr.
     bool is_null() const;
 
+    /// Equivalent to `size() == 0`, but O(1) rather than O(N).
+    bool empty() const;
+
     /// FIXME: O(n)
     char operator[](size_t index) const;
 

src/realm/sync/client.cpp

@@ -228,7 +228,8 @@ class SessionWrapper final : public util::AtomicRefCountBase, DB::CommitListener
     std::vector<WaitOperCompletionHandler> m_download_completion_handlers;
     std::vector<WaitOperCompletionHandler> m_sync_completion_handlers;
 
-    void on_upload_completion();
+    version_type m_upload_completion_requested_version = -1;
+
     void on_download_completion();
     void on_suspended(const SessionErrorInfo& error_info);
     void on_resumed();
@@ -238,7 +239,9 @@ class SessionWrapper final : public util::AtomicRefCountBase, DB::CommitListener
     void on_flx_sync_version_complete(int64_t version);
 
     void init_progress_handler();
-    void report_progress();
+    void check_progress();
+    void report_progress(ReportedProgress& p, DownloadableProgress downloadable);
+    void report_upload_completion(version_type);
 
     friend class SessionWrapperStack;
     friend class ClientImpl::Session;
@@ -736,23 +739,14 @@ void SessionImpl::initiate_integrate_changesets(std::uint_fast64_t downloadable_
             // Fake it for "dry run" mode
             client_version = m_last_version_available + 1;
         }
-        on_changesets_integrated(client_version, progress, !changesets.empty()); // Throws
+        on_changesets_integrated(client_version, progress); // Throws
     }
     catch (const IntegrationException& e) {
         on_integration_failure(e);
     }
 }
 
 
-void SessionImpl::on_upload_completion()
-{
-    // Ignore the call if the session is not active
-    if (m_state == State::Active) {
-        m_wrapper.on_upload_completion(); // Throws
-    }
-}
-
-
 void SessionImpl::on_download_completion()
 {
     // Ignore the call if the session is not active
@@ -842,7 +836,6 @@ bool SessionImpl::process_flx_bootstrap_message(const SyncProgress& progress, Do
     REALM_ASSERT_EX(hook_action == SyncClientHookAction::NoAction, hook_action);
 
     if (batch_state == DownloadBatchState::MoreToCome) {
-        notify_sync_progress();
         return true;
     }
 
@@ -935,7 +928,7 @@ void SessionImpl::process_pending_flx_bootstrap()
     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);
+    on_changesets_integrated(new_version.realm_version, progress);
     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.
@@ -1090,14 +1083,6 @@ void SessionImpl::enable_progress_notifications()
     m_wrapper.m_reliable_download_progress = true;
 }
 
-void SessionImpl::notify_sync_progress()
-{
-    if (m_state != State::Active)
-        return;
-
-    m_wrapper.report_progress();
-}
-
 util::Future<std::string> SessionImpl::send_test_command(std::string body)
 {
     if (m_state != State::Active) {
@@ -1301,7 +1286,7 @@ void SessionWrapper::on_commit(version_type new_version)
             return; // Already finalized
         SessionImpl& sess = *self->m_sess;
         sess.recognize_sync_version(new_version); // Throws
-        self->report_progress();                  // Throws
+        self->check_progress();                   // Throws
     });
 }
 
@@ -1343,6 +1328,7 @@ void SessionWrapper::async_wait_for(bool upload_completion, bool download_comple
             return;
         }
         if (upload_completion) {
+            self->m_upload_completion_requested_version = self->m_db->get_version_of_latest_snapshot();
             if (download_completion) {
                 // Wait for upload and download completion
                 self->m_sync_completion_handlers.push_back(std::move(handler)); // Throws
@@ -1358,7 +1344,7 @@ void SessionWrapper::async_wait_for(bool upload_completion, bool download_comple
         }
         SessionImpl& sess = *self->m_sess;
         if (upload_completion)
-            sess.request_upload_completion_notification(); // Throws
+            self->check_progress();
         if (download_completion)
             sess.request_download_completion_notification(); // Throws
     });                                                      // Throws
@@ -1484,7 +1470,7 @@ void SessionWrapper::actualize()
     }
 
     if (!m_client_reset_config)
-        report_progress(); // Throws
+        check_progress(); // Throws
 }
 
 void SessionWrapper::force_close()
@@ -1569,24 +1555,14 @@ inline void SessionWrapper::finalize_before_actualization() noexcept
     m_db = nullptr;
 }
 
-void SessionWrapper::on_upload_completion()
-{
-    REALM_ASSERT(!m_finalized);
-    while (!m_upload_completion_handlers.empty()) {
-        auto handler = std::move(m_upload_completion_handlers.back());
-        m_upload_completion_handlers.pop_back();
-        handler(Status::OK()); // Throws
-    }
-    while (!m_sync_completion_handlers.empty()) {
-        auto handler = std::move(m_sync_completion_handlers.back());
-        m_download_completion_handlers.push_back(std::move(handler)); // Throws
-        m_sync_completion_handlers.pop_back();
-    }
-}
-
-
 void SessionWrapper::on_download_completion()
 {
+    // Ensure that progress handlers get called before completion handlers. The
+    // download completing performed a commit and will trigger progress
+    // notifications asynchronously, but they would arrive after the download
+    // completion without this.
+    check_progress();
+
     while (!m_download_completion_handlers.empty()) {
         auto handler = std::move(m_download_completion_handlers.back());
         m_download_completion_handlers.pop_back();
@@ -1642,26 +1618,52 @@ void SessionWrapper::on_connection_state_changed(ConnectionState state,
 
 void SessionWrapper::init_progress_handler()
 {
-    ClientHistory::get_upload_download_bytes(m_db.get(), m_final_downloaded, m_final_uploaded);
+    ClientHistory::get_upload_download_state(m_db.get(), m_final_downloaded, m_final_uploaded);
 }
 
-void SessionWrapper::report_progress()
+void SessionWrapper::check_progress()
 {
     REALM_ASSERT(!m_finalized);
     REALM_ASSERT(m_sess);
 
+    if (!m_progress_handler && m_upload_completion_handlers.empty() && m_sync_completion_handlers.empty())
+        return;
+
+    version_type uploaded_version;
+    ReportedProgress p;
+    DownloadableProgress downloadable;
+    ClientHistory::get_upload_download_state(*m_db, p.downloaded, downloadable, p.uploaded, p.uploadable, p.snapshot,
+                                             uploaded_version);
+
+    report_progress(p, downloadable);
+    report_upload_completion(uploaded_version);
+}
+
+void SessionWrapper::report_upload_completion(version_type uploaded_version)
+{
+    if (uploaded_version < m_upload_completion_requested_version)
+        return;
+
+    std::move(m_sync_completion_handlers.begin(), m_sync_completion_handlers.end(),
+              std::back_inserter(m_download_completion_handlers));
+    m_sync_completion_handlers.clear();
+
+    while (!m_upload_completion_handlers.empty()) {
+        auto handler = std::move(m_upload_completion_handlers.back());
+        m_upload_completion_handlers.pop_back();
+        handler(Status::OK()); // Throws
+    }
+}
+
+void SessionWrapper::report_progress(ReportedProgress& p, DownloadableProgress downloadable)
+{
     if (!m_progress_handler)
         return;
 
     // Ignore progress messages from before we first receive a DOWNLOAD message
     if (!m_reliable_download_progress)
         return;
 
-    ReportedProgress p;
-    DownloadableProgress downloadable;
-    ClientHistory::get_upload_download_bytes(m_db.get(), p.downloaded, downloadable, p.uploaded, p.uploadable,
-                                             p.snapshot);
-
     auto calculate_progress = [](uint64_t transferred, uint64_t transferable, uint64_t final_transferred) {
         REALM_ASSERT_DEBUG_EX(final_transferred <= transferred, final_transferred, transferred, transferable);
         REALM_ASSERT_DEBUG_EX(transferred <= transferable, final_transferred, transferred, transferable);

src/realm/sync/noinst/client_history_impl.cpp

@@ -157,7 +157,7 @@ std::vector<ClientHistory::LocalChange> ClientHistory::get_local_changes(version
             // find_sync_history_entry() returns 0 to indicate not found and
             // otherwise adds 1 to the version, and then get_reciprocal_transform()
             // subtracts 1 from the version
-            if (auto changeset = get_reciprocal_transform(version + 1, compressed); changeset.size()) {
+            if (auto changeset = get_reciprocal_transform(version + 1, compressed); !changeset.empty()) {
                 changesets.push_back({version, changeset});
             }
         }
@@ -344,7 +344,7 @@ void ClientHistory::set_sync_progress(const SyncProgress& progress, Downloadable
     ensure_updated(local_version); // Throws
     prepare_for_write();           // Throws
 
-    update_sync_progress(progress, downloadable_bytes, wt); // Throws
+    update_sync_progress(progress, downloadable_bytes); // Throws
 
     // Note: This transaction produces an empty changeset. Empty changesets are
     // not uploaded to the server.
@@ -489,17 +489,17 @@ void ClientHistory::integrate_server_changesets(
         // During the bootstrap phase in flexible sync, the server sends multiple download messages with the same
         // synthetic server version that represents synthetic changesets generated from state on the server.
         if (batch_state == DownloadBatchState::LastInBatch && changesets_to_integrate.empty()) {
-            update_sync_progress(progress, downloadable_bytes, transact); // Throws
+            update_sync_progress(progress, downloadable_bytes); // Throws
         }
         // Always update progress for download messages from steady state.
         else if (batch_state == DownloadBatchState::SteadyState && !changesets_to_integrate.empty()) {
             auto partial_progress = progress;
             partial_progress.download.server_version = last_changeset.remote_version;
             partial_progress.download.last_integrated_client_version = last_changeset.last_integrated_local_version;
-            update_sync_progress(partial_progress, downloadable_bytes, transact); // Throws
+            update_sync_progress(partial_progress, downloadable_bytes); // Throws
         }
         else if (batch_state == DownloadBatchState::SteadyState && changesets_to_integrate.empty()) {
-            update_sync_progress(progress, downloadable_bytes, transact); // Throws
+            update_sync_progress(progress, downloadable_bytes); // Throws
         }
         if (run_in_write_tr) {
             run_in_write_tr(transact, changesets_for_cb);
@@ -610,33 +610,65 @@ size_t ClientHistory::transform_and_apply_server_changesets(util::Span<Changeset
 }
 
 
-void ClientHistory::get_upload_download_bytes(DB* db, std::uint_fast64_t& downloaded_bytes,
+void ClientHistory::get_upload_download_state(DB& db, std::uint_fast64_t& downloaded_bytes,
                                               DownloadableProgress& downloadable_bytes,
                                               std::uint_fast64_t& uploaded_bytes,
                                               std::uint_fast64_t& uploadable_bytes,
-                                              std::uint_fast64_t& snapshot_version)
+                                              std::uint_fast64_t& snapshot_version, version_type& uploaded_version)
 {
-    TransactionRef rt = db->start_read(); // Throws
+    TransactionRef rt = db.start_read(); // Throws
     version_type current_client_version = rt->get_version();
 
     downloaded_bytes = 0;
     downloadable_bytes = uint64_t(0);
     uploaded_bytes = 0;
     uploadable_bytes = 0;
     snapshot_version = current_client_version;
+    uploaded_version = 0;
 
     using gf = _impl::GroupFriend;
-    if (ref_type ref = gf::get_history_ref(*rt)) {
-        Array root(db->get_alloc());
-        root.init_from_ref(ref);
-        downloaded_bytes = root.get_as_ref_or_tagged(s_progress_downloaded_bytes_iip).get_as_int();
-        downloadable_bytes = root.get_as_ref_or_tagged(s_progress_downloadable_bytes_iip).get_as_int();
-        uploadable_bytes = root.get_as_ref_or_tagged(s_progress_uploadable_bytes_iip).get_as_int();
-        uploaded_bytes = root.get_as_ref_or_tagged(s_progress_uploaded_bytes_iip).get_as_int();
+    ref_type ref = gf::get_history_ref(*rt);
+    if (!ref)
+        return;
+
+    Array root(db.get_alloc());
+    root.init_from_ref(ref);
+    downloaded_bytes = root.get_as_ref_or_tagged(s_progress_downloaded_bytes_iip).get_as_int();
+    downloadable_bytes = root.get_as_ref_or_tagged(s_progress_downloadable_bytes_iip).get_as_int();
+    uploadable_bytes = root.get_as_ref_or_tagged(s_progress_uploadable_bytes_iip).get_as_int();
+    uploaded_bytes = root.get_as_ref_or_tagged(s_progress_uploaded_bytes_iip).get_as_int();
+
+    uploaded_version = root.get_as_ref_or_tagged(s_progress_upload_client_version_iip).get_as_int();
+    if (uploaded_version == current_client_version)
+        return;
+
+    BinaryColumn changesets(db.get_alloc());
+    changesets.init_from_ref(root.get_as_ref(s_changesets_iip));
+    IntegerBpTree origin_file_idents(db.get_alloc());
+    origin_file_idents.init_from_ref(root.get_as_ref(s_origin_file_idents_iip));
+
+    // `base_version` is the oldest version we have history for. If this is
+    // greater than uploaded_version, all of the versions in between the two had
+    // empty changesets and did not need to be uploaded. If this is less than
+    // uploaded_version, we have changesets which have been uploaded but the
+    // server has not yet told us we can delete and we may need to use for merging.
+    auto base_version = current_client_version - changesets.size();
+    if (uploaded_version < base_version) {
+        uploaded_version = base_version;
+    }
+
+    auto count = size_t(current_client_version - uploaded_version);
+    for (size_t i = changesets.size() - count; i < changesets.size(); ++i) {
+        if (origin_file_idents.get(i) == 0) {
+            size_t pos = 0;
+            if (changesets.get_at(i, pos).size() != 0)
+                break;
+        }
+        ++uploaded_version;
     }
 }
 
-void ClientHistory::get_upload_download_bytes(DB* db, std::uint_fast64_t& downloaded_bytes,
+void ClientHistory::get_upload_download_state(DB* db, std::uint_fast64_t& downloaded_bytes,
                                               std::uint_fast64_t& uploaded_bytes)
 {
     TransactionRef rt = db->start_read(); // Throws
@@ -711,7 +743,7 @@ auto ClientHistory::find_sync_history_entry(Arrays& arrays, version_type base_ve
         bool not_from_server = (origin_file_ident == 0);
         if (not_from_server) {
             ChunkedBinaryData chunked_changeset(arrays.changesets, offset + i);
-            if (chunked_changeset.size() > 0) {
+            if (!chunked_changeset.empty()) {
                 entry.origin_file_ident = file_ident_type(origin_file_ident);
                 entry.remote_version = last_integrated_server_version;
                 entry.origin_timestamp = timestamp_type(arrays.origin_timestamps.get(offset + i));
@@ -845,8 +877,7 @@ void ClientHistory::add_sync_history_entry(const HistoryEntry& entry)
 }
 
 
-void ClientHistory::update_sync_progress(const SyncProgress& progress, DownloadableProgress downloadable_bytes,
-                                         TransactionRef)
+void ClientHistory::update_sync_progress(const SyncProgress& progress, DownloadableProgress downloadable_bytes)
 {
     Array& root = m_arrays->root;