[BP] MB-58893: Extract collection size logic from tryAndScheduleOSOBackfill

And move it to the Collections::VB::Filter class, to make it easier to
write a unit test against.

Change-Id: I8b9cdb3e229713e7f76ab102d02e5f08617f0acc
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/200209
Reviewed-by: Vesko Karaganev <[email protected]>
Tested-by: Jim Walker <[email protected]>
Well-Formed: Restriction Checker

Author: veselink1

engines/ep/src/collections/vbucket_filter.cc

@@ -498,6 +498,19 @@ cb::engine_errc Filter::checkPrivileges(
     return cb::engine_errc::success;
 }
 
+Filter::CollectionSizeStats Filter::getSizeStats(
+        const Manifest& manifest) const {
+    size_t colItemCount = 0, colDiskSize = 0;
+    // For each collection, obtain the item count and disk size
+    for (auto [cid, sid] : filter) {
+        (void)sid;
+        const auto stats = manifest.lock(cid);
+        colItemCount += stats.getItemCount();
+        colDiskSize += stats.getDiskSize();
+    }
+    return {colItemCount, colDiskSize};
+}
+
 void Filter::dump() const {
     std::cerr << *this << std::endl;
 }

engines/ep/src/collections/vbucket_filter.h

@@ -218,6 +218,22 @@ class Filter {
         return !isPassThroughFilter() && !isLegacyFilter();
     }
 
+    struct CollectionSizeStats {
+        /**
+         * The number of items stored in the collections in the filter.
+         */
+        size_t itemCount;
+        /**
+         * The tracked size (in bytes) of the collections in the filter on disk.
+         */
+        size_t diskSize;
+    };
+
+    /**
+     * Get aggregated collection stats for all collections in the filter.
+     */
+    CollectionSizeStats getSizeStats(const Manifest& manifest) const;
+
     /**
      * Dump this to std::cerr
      */

engines/ep/src/dcp/active_stream.cc

@@ -1922,12 +1922,9 @@ bool ActiveStream::tryAndScheduleOSOBackfill(DcpProducer& producer,
             return false;
         }
         if (osoBackfill == "auto") {
-            // Retrieve collection stats from manifest; minimising the scope
-            // of manifest lock.
-            const auto [colItemCount, colDiskSize] = [&vb, cid] {
-                const auto stats = vb.getManifest().lock(cid);
-                return std::pair{stats.getItemCount(), stats.getDiskSize()};
-            }();
+            auto [colItemCount, colDiskSize] =
+                    filter.getSizeStats(vb.getManifest());
+
             const auto vbItemCount = vb.getNumItems();
             if (!isOSOPreferredForCollectionBackfill(
                         config, colItemCount, colDiskSize, vbItemCount)) {