MB-54516: Drop ModifyCollection events

If a collection is dropped, the modify event must be purged
along with any normal collection items. All of the dropKey
paths were setup to ignore any system events, this is no
longer true with the new event.

Change-Id: Ia89d4b7a56e9f3b5b23dd176e5d00da88bd83f3e
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/183265
Well-Formed: Restriction Checker
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: jimwwalker

engines/ep/src/collections/flush_accounting.cc

@@ -15,13 +15,6 @@
 
 namespace Collections::VB {
 
-// Returns two optionals
-// First is only initialised if the key is a collection system event.
-// Second is only initialised if there is a collection id present.
-// Any mutation will return nullopt,cid
-// A collection create/drop/modify will return event, cid
-// A scope event will return nullopt, nullopt
-
 /**
  * From a DocKey extract CollectionID and SystemEvent data.
  * The function returns two optionals because depending on the key there could

engines/ep/src/collections/scan_context.cc

@@ -12,6 +12,7 @@
 #include "collections/scan_context.h"
 
 #include "collections/kvstore.h"
+#include "systemevent_factory.h"
 
 namespace Collections::VB {
 
@@ -29,13 +30,23 @@ ScanContext::ScanContext(
 }
 
 bool ScanContext::isLogicallyDeleted(const DocKey& key, uint64_t seqno) const {
-    if (dropped.empty() || key.isInSystemCollection()) {
+    if (dropped.empty()) {
         return false;
     }
 
+    CollectionID cid;
+    if (key.isInSystemCollection()) {
+        auto modify = SystemEventFactory::isModifyCollection(key);
+        if (!modify) {
+            return false;
+        }
+        cid = modify.value();
+    } else {
+        cid = key.getCollectionID();
+    }
+
     // Is the key in a range which contains dropped collections and in the set?
-    return (seqno >= startSeqno && seqno <= endSeqno) &&
-           dropped.count(key.getCollectionID()) > 0;
+    return (seqno >= startSeqno && seqno <= endSeqno) && dropped.count(cid) > 0;
 }
 
 std::ostream& operator<<(std::ostream& os, const ScanContext& scanContext) {

engines/ep/src/ep_bucket.cc

@@ -1250,7 +1250,8 @@ void EPBucket::dropKey(VBucket& vb,
 
     auto docKey = diskKey.getDocKey();
     if (docKey.isInSystemCollection()) {
-        throw std::logic_error("EPBucket::dropKey called for a system key");
+        // SystemEvents aren't in memory so return.
+        return;
     }
 
     if (diskKey.isPrepared() && bySeqno > highCompletedSeqno) {

engines/ep/src/ephemeral_vb.cc

@@ -310,10 +310,16 @@ std::optional<SequenceList::RangeIterator> EphemeralVBucket::makeRangeIterator(
 
 bool EphemeralVBucket::isKeyLogicallyDeleted(const DocKey& key,
                                              int64_t bySeqno) const {
-    if (key.isInSystemCollection()) {
+    if (key.isInSystemCollection() &&
+        !SystemEventFactory::isModifyCollection(key)) {
         return false;
+        // else Modify events can be dropped
     }
-    auto cHandle = lockCollections(key);
+
+    // ReadLock the manifest and permit system keys - this ensures a Modify
+    // can lookup the correct collection
+    auto cHandle =
+            manifest->lock(key, Collections::VB::Manifest::AllowSystemKeys{});
     return !cHandle.valid() || cHandle.isLogicallyDeleted(bySeqno);
 }
 
@@ -929,11 +935,6 @@ size_t EphemeralVBucket::getNumPersistedDeletes() const {
 }
 
 void EphemeralVBucket::dropKey(const DocKey& key, int64_t bySeqno) {
-    // The system event doesn't get dropped here (tombstone purger will deal)
-    if (key.isInSystemCollection()) {
-        return;
-    }
-
     // if there is a pending item which will need removing from the DM,
     // store its seqno here for use after the HT lock has been released.
     int64_t prepareSeqno = 0;

engines/ep/src/kvstore/couch-kvstore/couch-kvstore.cc

@@ -935,7 +935,9 @@ static int time_purge_hook(Db& d,
 
             // Track nothing for the individual collection as the stats doc has
             // already been deleted.
-        } else {
+        } else if (!docKey.isInSystemCollection()) {
+            // item in a collection was dropped, track how many for item count
+            // adjustment.
             if (!info->deleted) {
                 ctx.stats.collectionsItemsPurged++;
             } else {

engines/ep/src/stored-value.cc

@@ -479,23 +479,30 @@ static std::string getSystemEventsValueFromStoredValue(const StoredValue& sv) {
     std::string_view itemValue{sv.getValue()->getData(),
                                sv.getValue()->getSize()};
 
-    if (systemEventType == SystemEvent::Scope) {
+    switch (systemEventType) {
+    case SystemEvent::Scope: {
         if (sv.isDeleted()) {
             auto eventData = Manifest::getDropScopeEventData(itemValue);
             return to_string(eventData);
         } else {
             auto eventData = Manifest::getCreateScopeEventData(itemValue);
             return to_string(eventData);
         }
-    } else if (systemEventType == SystemEvent::Collection) {
+        break;
+    }
+    case SystemEvent::Collection:
+    case SystemEvent::ModifyCollection: {
         if (sv.isDeleted()) {
             auto eventData = Manifest::getDropEventData(itemValue);
             return to_string(eventData);
         } else {
             auto eventData = Manifest::getCreateEventData(itemValue);
             return to_string(eventData);
         }
+        break;
     }
+    }
+
     throw std::invalid_argument(
             "getSystemEventsValueFromStoredValue(): StoredValue must be a "
             "SystemEvent for a collection or scope");

engines/ep/src/systemevent_factory.cc

@@ -85,10 +85,12 @@ StoredDocKey SystemEventFactory::makeCollectionEventKey(CollectionID cid,
 
 CollectionID SystemEventFactory::getCollectionIDFromKey(const DocKey& key) {
     // Input key is made up of a sequence of prefixes as per makeCollectionEvent
-    // or makeScopeEvent
+    // or makeModifyCollectionEvent
     // This function skips (1), checks (2) and returns 3
     auto se = getSystemEventType(key);
-    Expects(se.first == SystemEvent::Collection); // expected Collection
+    // expected Collection/ModifyCollection event
+    Expects(se.first == SystemEvent::Collection ||
+            se.first == SystemEvent::ModifyCollection);
     return cb::mcbp::unsigned_leb128<CollectionIDType>::decode(se.second).first;
 }
 
@@ -124,6 +126,15 @@ std::pair<SystemEvent, uint32_t> SystemEventFactory::getTypeAndID(
     return {type, cb::mcbp::unsigned_leb128<uint32_t>::decode(buffer).first};
 }
 
+std::optional<CollectionID> SystemEventFactory::isModifyCollection(
+        const DocKey& key) {
+    auto [event, id] = SystemEventFactory::getTypeAndID(key);
+    if (event != SystemEvent::ModifyCollection) {
+        return {};
+    }
+    return CollectionID(id);
+}
+
 std::unique_ptr<SystemEventProducerMessage> SystemEventProducerMessage::make(
         uint32_t opaque, queued_item& item, cb::mcbp::DcpStreamId sid) {
     // Always ensure decompressed as we are about to use the value

engines/ep/src/systemevent_factory.h

@@ -107,6 +107,13 @@ class SystemEventFactory {
      */
     static std::pair<SystemEvent, uint32_t> getTypeAndID(const DocKey& key);
 
+    /**
+     * Given a key which the caller knows isSystemCollection return an optional
+     * which stores the ID of the modified collection iff the key represent a
+     * ModifyCollection event.
+     */
+    static std::optional<CollectionID> isModifyCollection(const DocKey& key);
+
 private:
     /**
      * Make an Item representing the SystemEvent

engines/ep/tests/module_tests/checkpoint_test.cc

@@ -2694,6 +2694,9 @@ TEST_P(CheckpointTest, CheckpointItemToString) {
     auto event =
             SystemEventFactory::makeCollectionEvent(CollectionID(99), {}, {});
     EXPECT_EQ("cid:0x1:0x0:0x63:_collection", event->getKey().to_string());
+    event = SystemEventFactory::makeModifyCollectionEvent(
+            CollectionID(999), {}, {});
+    EXPECT_EQ("cid:0x1:0x2:0x3e7:_collection", event->getKey().to_string());
     event = SystemEventFactory::makeScopeEvent(ScopeID(99), {}, {});
     EXPECT_EQ("cid:0x1:0x1:0x63:_scope", event->getKey().to_string());
 }

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

@@ -2177,6 +2177,65 @@ TEST_P(CollectionsEraserTest, MB_50747_delete_and_drop) {
     EXPECT_EQ(0, vb->getNumItems());
 }
 
+TEST_P(CollectionsEraserTest, drop_after_modify) {
+    // add two collections
+    CollectionsManifest cm(CollectionEntry::dairy);
+    vb->updateFromManifest(makeManifest(cm.add(CollectionEntry::fruit)));
+
+    flushVBucketToDiskIfPersistent(vbid, 2 /* 2 x system */);
+
+    // add some items
+    store_item(
+            vbid, StoredDocKey{"dairy:milk", CollectionEntry::dairy}, "nice");
+    store_item(vbid,
+               StoredDocKey{"dairy:butter", CollectionEntry::dairy},
+               "lovely");
+    store_item(
+            vbid, StoredDocKey{"fruit:apple", CollectionEntry::fruit}, "nice");
+    store_item(vbid,
+               StoredDocKey{"fruit:apricot", CollectionEntry::fruit},
+               "lovely");
+
+    flushVBucketToDiskIfPersistent(vbid, 4);
+
+    EXPECT_EQ(4, vb->getNumItems());
+
+    // Modify history setting of dairy false->true
+    vb->updateFromManifest(makeManifest(
+            cm.update(CollectionEntry::fruit, cb::NoExpiryLimit, true)));
+    store_item(
+            vbid, StoredDocKey{"fruit:apple", CollectionEntry::fruit}, "nice");
+
+    flushVBucketToDiskIfPersistent(vbid, 2);
+
+    // delete the collections
+    vb->updateFromManifest(makeManifest(
+            cm.remove(CollectionEntry::dairy).remove(CollectionEntry::fruit)));
+
+    EXPECT_FALSE(vb->lockCollections().exists(CollectionEntry::dairy));
+    EXPECT_FALSE(vb->lockCollections().exists(CollectionEntry::fruit));
+
+    flushVBucketToDiskIfPersistent(vbid, 2 /* 2 x system */);
+
+    if (!isPersistent()) {
+        // 2x create collection, 1x modify
+        EXPECT_EQ(3, vb->ht.getNumSystemItems());
+    }
+
+    runCollectionsEraser(vbid);
+
+    EXPECT_EQ(0, vb->getNumItems());
+
+    EXPECT_FALSE(vb->lockCollections().exists(CollectionEntry::dairy));
+    EXPECT_FALSE(vb->lockCollections().exists(CollectionEntry::fruit));
+
+    if (!isPersistent()) {
+        // 2 system events remain (tombstones of dairy/fruit).
+        // modify was purged
+        EXPECT_EQ(2, vb->ht.getNumSystemItems());
+    }
+}
+
 // Test cases which run for persistent and ephemeral buckets
 INSTANTIATE_TEST_SUITE_P(CollectionsEraserTests,
                          CollectionsEraserTest,