MB-41321: 2/4 Clean-up if collection exists in old and new drop containers

Collections::Flush was erasing a collection from the new drop 'map'
when it was found to be in both 'old' and 'new' - this is so the
flatbuffer output (which is a vector) only had the dropped collection
once.

However this leads to code executing later in the flush to not see
the collection when iterating through the 'droppecCollections' map.
For example notifyManifestOfAnyDroppedCollections would not get the
VB::Manifest to clean-up (effectively leaking memory).

This fix ensures we still generate the collection once in the
flatbuffers output, but also keep it in the list and now clean-up.

Change-Id: I3f9f91ee34b82a748e8145b70a4ce47089ad3687
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/137698
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: jimwwalker

engines/ep/src/collections/flush.cc

@@ -475,29 +475,44 @@ flatbuffers::DetachedBuffer Flush::encodeOpenCollections(
 
 flatbuffers::DetachedBuffer Flush::encodeDroppedCollections(
         std::vector<Collections::KVStore::DroppedCollection>& existingDropped) {
+    flatbuffers::FlatBufferBuilder builder;
+    std::vector<flatbuffers::Offset<Collections::KVStore::Dropped>> output;
+
+    // Add Collection's to this set only if they are in both old and new
+    // sets of dropped collections - this ensures we generate it once into
+    // the new output.
+    std::unordered_set<CollectionID> skip;
+
     // Iterate through the existing dropped collections and look each up in the
     // commit metadata. If the collection is in both lists, we will just update
     // the existing data (adjusting the endSeqno) and then erase the collection
     // from the commit meta's dropped collections.
     for (auto& collection : existingDropped) {
         if (auto itr = droppedCollections.find(collection.collectionId);
             itr != droppedCollections.end()) {
+            // Collection is in both old and new 'sets' of dropped collections
+            // we only want it in the output once - update the end-seqno here
+            // and add to skip set
             collection.endSeqno = itr->second.endSeqno;
-
-            // Now kick the collection out of collectionsMeta, its contribution
-            // to the final output is complete
-            droppedCollections.erase(itr);
+            skip.emplace(collection.collectionId);
         }
+        auto newEntry = Collections::KVStore::CreateDropped(
+                builder,
+                collection.startSeqno,
+                collection.endSeqno,
+                uint32_t(collection.collectionId));
+        output.push_back(newEntry);
     }
 
-    flatbuffers::FlatBufferBuilder builder;
-    std::vector<flatbuffers::Offset<Collections::KVStore::Dropped>> output;
-
-    // Now merge, first the newly dropped collections
-    // Iterate  through the list collections dropped in the commit batch and
+    // Now add the newly dropped collections
+    // Iterate through the set of collections dropped in the commit batch and
     // and create flatbuffer versions of each one
     for (const auto& [cid, dropped] : droppedCollections) {
         (void)cid;
+        if (skip.count(cid) > 0) {
+            // This collection is already in output
+            continue;
+        }
         auto newEntry = Collections::KVStore::CreateDropped(
                 builder,
                 dropped.startSeqno,
@@ -506,16 +521,6 @@ flatbuffers::DetachedBuffer Flush::encodeDroppedCollections(
         output.push_back(newEntry);
     }
 
-    // and now copy across the existing dropped collections
-    for (const auto& entry : existingDropped) {
-        auto newEntry = Collections::KVStore::CreateDropped(
-                builder,
-                entry.startSeqno,
-                entry.endSeqno,
-                uint32_t(entry.collectionId));
-        output.push_back(newEntry);
-    }
-
     auto vector = builder.CreateVector(output);
     auto final =
             Collections::KVStore::CreateDroppedCollections(builder, vector);

engines/ep/tests/module_tests/collections/evp_store_collections_test.cc

@@ -2858,6 +2858,31 @@ TEST_P(CollectionsParameterizedTest, OneScopeStatsByIdParsing) {
     EXPECT_EQ(cm.getUid(), result.getManifestId());
 }
 
+// Test a specific issue spotted, in the case when a collection exists
+// as dropped in storage and is newly dropped - we clean-up
+TEST_P(CollectionsPersistentParameterizedTest, FlushDropCreateDropCleansUp) {
+    CollectionsManifest cm;
+    cm.add(CollectionEntry::fruit); // seq:1
+    setCollections(cookie, std::string{cm});
+    flushVBucketToDiskIfPersistent(vbid, 1);
+    cm.remove(CollectionEntry::fruit); // seq:2
+    setCollections(cookie, std::string{cm});
+    flushVBucketToDiskIfPersistent(vbid, 1);
+    cm.add(CollectionEntry::fruit); // seq:3
+    setCollections(cookie, std::string{cm});
+    flushVBucketToDiskIfPersistent(vbid, 1);
+    cm.remove(CollectionEntry::fruit); // seq:4
+    setCollections(cookie, std::string{cm});
+    flushVBucketToDiskIfPersistent(vbid, 1);
+
+    VBucketPtr vb = store->getVBucket(vbid);
+
+    // Final flush should of removed making this invalid and we get an exception
+    EXPECT_THROW(vb->getManifest().lock().getStatsForFlush(
+                         CollectionEntry::fruit, 4),
+                 std::logic_error);
+}
+
 INSTANTIATE_TEST_SUITE_P(CollectionsExpiryLimitTests,
                          CollectionsExpiryLimitTest,
                          ::testing::Bool(),