Fix handling of work limit in online compaction

The work limit could wrap around resulting in an unlimited run, which could
take a long time to finish.

CHANGELOG.md

@@ -16,6 +16,7 @@
 * Properties and types not present in the requested schema would be missing from the reported schema in several scenarios, such as if the Realm was being opened with a different schema version than the persisted one, and if the new tables or columns were added while the Realm instance did not have an active read transaction ([PR #6211](https://github.com/realm/realm-core/pull/6211), since v13.2.0).
 * If a client reset w/recovery or discard local is interrupted while the "fresh" realm is being downloaded, the sync client may crash with a MultpleSyncAgents exception ([#6217](https://github.com/realm/realm-core/issues/6217), since v11.13.0)
 * Changesets from the server sent during FLX bootstrapping that are larger than 16MB can cause the sync client to crash with a LogicError (PR [#6218](https://github.com/realm/realm-core/pull/6218), since v12.0.0)
+* Online compaction may cause a single commit to take a long time ([#6245](https://github.com/realm/realm-core/pull/6245), since v13.0.0)
 
 ### Breaking changes
 * `SyncSession::log_out()` has been renamed to `SyncSession::force_close()` to reflect what it actually does ([#6183](https://github.com/realm/realm-core/pull/6183))

src/realm/db.cpp

@@ -2225,8 +2225,8 @@ void DB::low_level_commit(uint_fast64_t new_version, Transaction& transaction, b
 
     if (auto limit = out.get_evacuation_limit()) {
         // Get a work limit based on the size of the transaction we're about to commit
-        // Assume at least 4K on top of that for the top arrays
-        size_t work_limit = 4 * 1024 + m_alloc.get_commit_size() / 2;
+        // Add 4k to ensure progress on small commits
+        size_t work_limit = m_alloc.get_commit_size() / 2 + out.get_free_list_size() + 0x1000;
         transaction.cow_outliers(out.get_evacuation_progress(), limit, work_limit);
     }
 

src/realm/group_writer.cpp

@@ -679,10 +679,8 @@ ref_type GroupWriter::write_group()
     // bigger databases the space required for free lists will be relatively less.
     max_free_list_size += 10;
 
-    // If current size is less than 128 MB, the database need not expand above 2 GB
-    // which means that the positions and sizes can still be in 32 bit.
-    int size_per_entry = (m_logical_size < 0x8000000 ? 8 : 16) + 8;
-    size_t max_free_space_needed = Array::get_max_byte_size(top.size()) + size_per_entry * max_free_list_size;
+    size_t max_free_space_needed =
+        Array::get_max_byte_size(top.size()) + size_per_free_list_entry() * max_free_list_size;
 
     ALLOC_DBG_COUT("  Allocating file space for freelists:" << std::endl);
     // Reserve space for remaining arrays. We ask for some extra bytes beyond the

src/realm/group_writer.hpp

@@ -116,6 +116,11 @@ class GroupWriter : public _impl::ArrayWriterBase {
         return m_backoff ? 0 : m_evacuation_limit;
     }
 
+    size_t get_free_list_size()
+    {
+        return m_free_positions.size() * size_per_free_list_entry();
+    }
+
     std::vector<size_t>& get_evacuation_progress()
     {
         return m_evacuation_progress;
@@ -181,6 +186,7 @@ class GroupWriter : public _impl::ArrayWriterBase {
     size_t m_evacuation_limit;
     int64_t m_backoff;
     size_t m_logical_size = 0;
+    bool m_final_step = false;
     Durability m_durability;
 
     //  m_free_in_file;
@@ -254,6 +260,13 @@ class GroupWriter : public _impl::ArrayWriterBase {
 
     /// Debug helper - extends the TopRefMap with list of reachable blocks
     void map_reachable();
+
+    size_t size_per_free_list_entry() const
+    {
+        // If current size is less than 128 MB, the database need not expand above 2 GB
+        // which means that the positions and sizes can still be in 32 bit.
+        return (m_logical_size < 0x8000000 ? 8 : 16) + 8;
+    }
 };
 
 

src/realm/transaction.cpp

@@ -933,9 +933,9 @@ void Transaction::set_transact_stage(DB::TransactStage stage) noexcept
 
 class NodeTree {
 public:
-    NodeTree(size_t evac_limit, size_t& work_limit)
+    NodeTree(size_t evac_limit, size_t work_limit)
         : m_evac_limit(evac_limit)
-        , m_work_limit(work_limit)
+        , m_work_limit(int64_t(work_limit))
         , m_moved(0)
     {
     }
@@ -955,18 +955,15 @@ class NodeTree {
     ///                   to point to the node we have just processed
     bool trv(Array& current_node, unsigned level, std::vector<size_t>& progress)
     {
+        if (m_work_limit < 0) {
+            return false;
+        }
         if (current_node.is_read_only()) {
-            auto byte_size = current_node.get_byte_size();
+            size_t byte_size = current_node.get_byte_size();
             if ((current_node.get_ref() + byte_size) > m_evac_limit) {
                 current_node.copy_on_write();
                 m_moved++;
-                if (m_work_limit > byte_size) {
-                    m_work_limit -= byte_size;
-                }
-                else {
-                    m_work_limit = 0;
-                    return false;
-                }
+                m_work_limit -= byte_size;
             }
         }
 
@@ -998,12 +995,12 @@ class NodeTree {
 
 private:
     size_t m_evac_limit;
-    size_t m_work_limit;
+    int64_t m_work_limit;
     size_t m_moved;
 };
 
 
-void Transaction::cow_outliers(std::vector<size_t>& progress, size_t evac_limit, size_t& work_limit)
+void Transaction::cow_outliers(std::vector<size_t>& progress, size_t evac_limit, size_t work_limit)
 {
     NodeTree node_tree(evac_limit, work_limit);
     if (progress.empty()) {

src/realm/transaction.hpp

@@ -188,7 +188,7 @@ class Transaction : public Group {
     void complete_async_commit();
     void acquire_write_lock() REQUIRES(!m_async_mutex);
 
-    void cow_outliers(std::vector<size_t>& progress, size_t evac_limit, size_t& work_limit);
+    void cow_outliers(std::vector<size_t>& progress, size_t evac_limit, size_t work_limit);
     void close_read_with_lock() REQUIRES(!m_async_mutex, db->m_mutex);
 
     DBRef db;