MB-54516: Modify history out of sync with many vbuckets

The initial addition of history mistakenly added the flag to the
"shared metadata". When >1 active vbucket exists, the first one
to be processed updated collection.history, but sets the shared
metadata and then generates a Modify. The second vbucket sees
no difference in history and does not generate a Modify.

The replicas get out of sync, with vb0 correctly deduplicating
on active and replica, but vb1 only deduplicating on the active.

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

Author: jimwwalker

engines/ep/src/collections/collections_types.cc

@@ -103,22 +103,13 @@ std::string to_string(ManifestUpdateStatus status) {
 }
 
 CollectionSharedMetaDataView::CollectionSharedMetaDataView(
-        std::string_view name,
-        ScopeID scope,
-        cb::ExpiryLimit maxTtl,
-        CanDeduplicate canDeduplicate)
-    : name(name),
-      scope(scope),
-      maxTtl(std::move(maxTtl)),
-      canDeduplicate(canDeduplicate) {
+        std::string_view name, ScopeID scope, cb::ExpiryLimit maxTtl)
+    : name(name), scope(scope), maxTtl(std::move(maxTtl)) {
 }
 
 CollectionSharedMetaDataView::CollectionSharedMetaDataView(
         const CollectionSharedMetaData& meta)
-    : name(meta.name),
-      scope(meta.scope),
-      maxTtl(meta.maxTtl),
-      canDeduplicate(meta.canDeduplicate) {
+    : name(meta.name), scope(meta.scope), maxTtl(meta.maxTtl) {
 }
 
 std::string CollectionSharedMetaDataView::to_string() const {
@@ -127,33 +118,23 @@ std::string CollectionSharedMetaDataView::to_string() const {
     if (maxTtl) {
         rv += ", maxTtl:" + std::to_string(maxTtl.value().count());
     }
-    rv += ", " + ::to_string(canDeduplicate);
     return rv;
 }
 
-CollectionSharedMetaData::CollectionSharedMetaData(
-        std::string_view name,
-        ScopeID scope,
-        cb::ExpiryLimit maxTtl,
-        CanDeduplicate canDeduplicate)
-    : name(name),
-      scope(scope),
-      maxTtl(std::move(maxTtl)),
-      canDeduplicate(canDeduplicate) {
+CollectionSharedMetaData::CollectionSharedMetaData(std::string_view name,
+                                                   ScopeID scope,
+                                                   cb::ExpiryLimit maxTtl)
+    : name(name), scope(scope), maxTtl(std::move(maxTtl)) {
 }
 
 CollectionSharedMetaData::CollectionSharedMetaData(
         const CollectionSharedMetaDataView& view)
-    : name(view.name),
-      scope(view.scope),
-      maxTtl(view.maxTtl),
-      canDeduplicate(view.canDeduplicate) {
+    : name(view.name), scope(view.scope), maxTtl(view.maxTtl) {
 }
 
 bool CollectionSharedMetaData::operator==(
         const CollectionSharedMetaDataView& view) const {
-    return name == view.name && scope == view.scope && maxTtl == view.maxTtl &&
-           canDeduplicate == view.canDeduplicate;
+    return name == view.name && scope == view.scope && maxTtl == view.maxTtl;
 }
 
 bool CollectionSharedMetaData::operator==(
@@ -167,7 +148,7 @@ std::ostream& operator<<(std::ostream& os,
     if (meta.maxTtl) {
         os << ", maxTtl:" << meta.maxTtl.value().count();
     }
-    return os << ", " << to_string(meta.canDeduplicate);
+    return os;
 }
 
 ScopeSharedMetaDataView::ScopeSharedMetaDataView(

engines/ep/src/collections/collections_types.h

@@ -297,23 +297,20 @@ class CollectionSharedMetaDataView {
 public:
     CollectionSharedMetaDataView(std::string_view name,
                                  ScopeID scope,
-                                 cb::ExpiryLimit maxTtl,
-                                 CanDeduplicate canDeduplicate);
+                                 cb::ExpiryLimit maxTtl);
     CollectionSharedMetaDataView(const CollectionSharedMetaData&);
     std::string to_string() const;
     std::string_view name;
     const ScopeID scope;
     const cb::ExpiryLimit maxTtl;
-    const CanDeduplicate canDeduplicate;
 };
 
 // The type stored by the Manager SharedMetaDataTable
 class CollectionSharedMetaData : public RCValue {
 public:
     CollectionSharedMetaData(std::string_view name,
                              ScopeID scope,
-                             cb::ExpiryLimit maxTtl,
-                             CanDeduplicate canDeduplicate);
+                             cb::ExpiryLimit maxTtl);
     CollectionSharedMetaData(const CollectionSharedMetaDataView& view);
     bool operator==(const CollectionSharedMetaDataView& view) const;
     bool operator!=(const CollectionSharedMetaDataView& view) const {
@@ -327,13 +324,8 @@ class CollectionSharedMetaData : public RCValue {
     const std::string name;
     const ScopeID scope;
     const cb::ExpiryLimit maxTtl;
-
-    // atomic: can be changed by any thread
-    // mutable: the VB::Manifest "handle" provides a const interface (the map of
-    // collections cannot be changed), but allows some changes to be made to
-    // the values in that map.
-    mutable std::atomic<CanDeduplicate> canDeduplicate;
 };
+
 std::ostream& operator<<(std::ostream& os,
                          const CollectionSharedMetaData& meta);
 

engines/ep/src/collections/vbucket_manifest.cc

@@ -83,14 +83,13 @@ Manifest::Manifest(Manifest& other) : manager(other.manager) {
 
     // Collection/Scope maps are not trivially copyable, so we do it manually
     for (auto& [cid, entry] : other.map) {
-        CollectionSharedMetaDataView meta{entry.getName(),
-                                          entry.getScopeID(),
-                                          entry.getMaxTtl(),
-                                          entry.getCanDeduplicate()};
+        CollectionSharedMetaDataView meta{
+                entry.getName(), entry.getScopeID(), entry.getMaxTtl()};
         auto [itr, inserted] =
                 map.try_emplace(cid,
                                 other.manager->createOrReferenceMeta(cid, meta),
-                                entry.getStartSeqno());
+                                entry.getStartSeqno(),
+                                entry.getCanDeduplicate());
         Expects(inserted);
         itr->second.setItemCount(entry.getItemCount());
         itr->second.setDiskSize(entry.getDiskSize());
@@ -516,11 +515,12 @@ ManifestEntry& Manifest::addNewCollectionEntry(ScopeCollectionPair identifiers,
                                                CanDeduplicate canDeduplicate,
                                                int64_t startSeqno) {
     CollectionSharedMetaDataView meta{
-            collectionName, identifiers.first, maxTtl, canDeduplicate};
+            collectionName, identifiers.first, maxTtl};
     auto [itr, inserted] = map.try_emplace(
             identifiers.second,
             manager->createOrReferenceMeta(identifiers.second, meta),
-            startSeqno);
+            startSeqno,
+            canDeduplicate);
 
     if (!inserted) {
         throwException<std::logic_error>(

engines/ep/src/collections/vbucket_manifest_entry.cc

@@ -23,7 +23,8 @@ bool Collections::VB::ManifestEntry::operator==(
            itemCount == other.itemCount && diskSize == other.diskSize &&
            persistedHighSeqno == other.persistedHighSeqno &&
            numOpsGet == other.numOpsGet && numOpsDelete == other.numOpsDelete &&
-           numOpsStore == other.numOpsStore && meta == other.meta;
+           numOpsStore == other.numOpsStore && meta == other.meta &&
+           canDeduplicate == other.canDeduplicate;
 }
 
 std::string Collections::VB::ManifestEntry::getExceptionString(
@@ -87,5 +88,7 @@ std::ostream& Collections::VB::operator<<(
     if (manifestEntry.getMaxTtl()) {
         os << ", maxTtl:" << manifestEntry.getMaxTtl().value().count();
     }
+
+    os << ", " << manifestEntry.getCanDeduplicate();
     return os;
 }

engines/ep/src/collections/vbucket_manifest_entry.h

@@ -36,12 +36,14 @@ namespace VB {
 class ManifestEntry {
 public:
     ManifestEntry(SingleThreadedRCPtr<const VB::CollectionSharedMetaData> meta,
-                  uint64_t startSeqno)
+                  uint64_t startSeqno,
+                  CanDeduplicate canDeduplicate)
         : startSeqno(startSeqno),
           itemCount(0),
           highSeqno(startSeqno),
           persistedHighSeqno(startSeqno),
-          meta(std::move(meta)) {
+          meta(std::move(meta)),
+          canDeduplicate(canDeduplicate) {
     }
 
     // Mark copy and move as deleted as it simplifies the lifetime of
@@ -203,11 +205,11 @@ class ManifestEntry {
 
     /// @return Yes/No if this collection removes duplicate keys when possible
     CanDeduplicate getCanDeduplicate() const {
-        return meta->canDeduplicate;
+        return canDeduplicate;
     }
 
     void setCanDeduplicate(CanDeduplicate value) {
-        meta->canDeduplicate.store(value);
+        canDeduplicate.store(value);
     }
 
 private:
@@ -302,6 +304,8 @@ class ManifestEntry {
      * a reference to the data, which is stored inside the Manager.
      */
     SingleThreadedRCPtr<const CollectionSharedMetaData> meta;
+
+    std::atomic<CanDeduplicate> canDeduplicate;
 };
 
 std::ostream& operator<<(std::ostream& os, const ManifestEntry& manifestEntry);

engines/ep/tests/mock/mock_dcp.cc

@@ -561,4 +561,5 @@ void MockDcpMessageProducers::clear_dcp_data() {
     last_end_status = cb::mcbp::DcpStreamEndStatus::Ok;
     last_high_completed_seqno = std::nullopt;
     last_max_visible_seqno = std::nullopt;
+    last_can_deduplicate = CanDeduplicate::Yes;
 }

engines/ep/tests/mock/mock_dcp.h

@@ -12,6 +12,7 @@
 #pragma once
 
 #include <dcp/dcp-types.h>
+#include <ep_types.h>
 #include <mcbp/protocol/opcode.h>
 #include <memcached/dcp.h>
 #include <memcached/engine.h>
@@ -196,6 +197,7 @@ class MockDcpMessageProducers : public DcpMessageProducersIface {
     std::optional<uint64_t> last_high_completed_seqno;
     std::optional<uint64_t> last_max_visible_seqno;
     bool isCollectionsSupported = false;
+    CanDeduplicate last_can_deduplicate = CanDeduplicate::Yes;
 
 protected:
     /// Helper method for deletion / deletion_v2 / expiration

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

@@ -381,6 +381,8 @@ cb::engine_errc CollectionsDcpTestProducers::systemEventVersion2(
         last_collection_manifest_uid = collection->uid();
         last_key.assign(reinterpret_cast<const char*>(key.data()), key.size());
         EXPECT_NE(0, key.size());
+        last_can_deduplicate =
+                getCanDeduplicateFromHistory(collection->history());
         break;
     }
     case mcbp::systemevent::id::DeleteCollection: {

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

@@ -4486,6 +4486,74 @@ TEST_P(CollectionsDcpPersistentOnly, ModifyCollectionNotReplicated) {
     EXPECT_EQ(store->getVBucket(vbid)->getHighSeqno(), producers->last_byseqno);
 }
 
+TEST_P(CollectionsDcpPersistentOnly, ModifyCollectionTwoVbuckets) {
+    using namespace cb::mcbp;
+    using namespace mcbp::systemevent;
+    using namespace CollectionEntry;
+
+    // Create a second producer to simplify the step/expect. 1 VB per producer
+    auto* cookieP2 = create_mock_cookie(engine.get());
+
+    auto producer2 = SingleThreadedKVBucketTest::createDcpProducer(
+            cookieP2, IncludeDeleteTime::Yes, true /*FlatBuffers*/);
+    auto& mockConnMap = static_cast<MockDcpConnMap&>(engine->getDcpConnMap());
+    mockConnMap.addConn(cookieP2, producer);
+
+    // No transfer to consumer in this test - just check the data passed in
+    // each step
+    producers->consumer = nullptr;
+
+    // test uses 2 active VBuckets
+    store->setVBucketState(replicaVB, vbucket_state_active);
+
+    uint64_t rollbackSeqno = 0;
+    ASSERT_EQ(cb::engine_errc::success,
+              producer2->streamRequest(0, // flags
+                                       1, // opaque
+                                       replicaVB,
+                                       0,
+                                       ~0ull, // end_seqno
+                                       0,
+                                       0,
+                                       0,
+                                       &rollbackSeqno,
+                                       &CollectionsDcpTest::dcpAddFailoverLog,
+                                       {{nullptr, 0}}));
+
+    // Create with history and then disable
+    CollectionsManifest cm;
+    cm.add(CollectionEntry::vegetable, cb::NoExpiryLimit, true);
+    setCollections(cookie, cm);
+    cm.update(vegetable, cb::NoExpiryLimit);
+    setCollections(cookie, cm);
+
+    // expect 4 events
+    // create, modify on two vbuckets.
+    std::array<std::shared_ptr<MockDcpProducer>, 2> dcpProducers{
+            {producer, producer2}};
+    for (auto& dcp : dcpProducers) {
+        SingleThreadedKVBucketTest::notifyAndStepToCheckpoint(
+                *dcp,
+                *producers,
+                ClientOpcode::DcpSnapshotMarker,
+                true /*memory*/);
+        dcp->stepAndExpect(*producers, ClientOpcode::DcpSystemEvent);
+        EXPECT_EQ(producers->last_system_event, id::CreateCollection);
+        EXPECT_EQ(producers->last_collection_id, vegetable.getId());
+        EXPECT_EQ(producers->last_can_deduplicate, CanDeduplicate::No);
+
+        // checkpoint split...
+        dcp->stepAndExpect(*producers, ClientOpcode::DcpSnapshotMarker);
+        dcp->stepAndExpect(*producers, ClientOpcode::DcpSystemEvent);
+        EXPECT_EQ(producers->last_system_event, id::ModifyCollection);
+        EXPECT_EQ(producers->last_collection_id, vegetable.getId());
+        EXPECT_EQ(producers->last_can_deduplicate, CanDeduplicate::Yes);
+    }
+    producer2->closeAllStreams();
+    mockConnMap.removeConn(cookieP2);
+    destroy_mock_cookie(cookieP2);
+}
+
 // Test cases which run for persistent and ephemeral buckets
 INSTANTIATE_TEST_SUITE_P(CollectionsDcpEphemeralOrPersistent,
                          CollectionsDcpParameterizedTest,

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

@@ -72,20 +72,18 @@ class CollectionsVBFilterTest : public CollectionsTest {
     // Tests need entry data which appear in default and 'shop1' scopes
     std::unique_ptr<Collections::VB::ManifestEntry> entryInDefaultScope =
             std::make_unique<Collections::VB::ManifestEntry>(
-                    manager->createOrReferenceMeta(CollectionID::Default,
-                                                   {"name1",
-                                                    ScopeID::Default,
-                                                    {},
-                                                    CanDeduplicate::Yes}),
-                    0);
+                    manager->createOrReferenceMeta(
+                            CollectionID::Default,
+                            {"name1", ScopeID::Default, {}}),
+                    0,
+                    CanDeduplicate::Yes);
     std::unique_ptr<Collections::VB::ManifestEntry> entryInShop1Scope =
             std::make_unique<Collections::VB::ManifestEntry>(
-                    manager->createOrReferenceMeta(CollectionID::Default,
-                                                   {"name2",
-                                                    ScopeUid::shop1,
-                                                    {},
-                                                    CanDeduplicate::Yes}),
-                    0);
+                    manager->createOrReferenceMeta(
+                            CollectionID::Default,
+                            {"name2", ScopeUid::shop1, {}}),
+                    0,
+                    CanDeduplicate::Yes);
 
     CookieIface* cookie = nullptr;
 };