MB-59518: Stop PassiveStream from processing ops against active VB

Add VBucket state checks to processSystemEvent and update the
setWithMeta and deleteWithMeta calls to ensure only a replica or
pending vbucket can be operated on by the PassiveStream.

Change-Id: I783b5e2fc646dc9c191c1e341541c04bbbafafac
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/200507
Tested-by: Jim Walker <[email protected]>
Well-Formed: Restriction Checker
Reviewed-by: Paolo Cocchi <[email protected]>

Author: jimwwalker

engines/ep/src/dcp/passive_stream.cc

@@ -667,9 +667,7 @@ cb::engine_errc PassiveStream::processMessage(MutationConsumerMessage* message,
                                                  0,
                                                  nullptr,
                                                  consumer->getCookie(),
-                                                 {vbucket_state_active,
-                                                  vbucket_state_replica,
-                                                  vbucket_state_pending},
+                                                 permittedVBStates,
                                                  CheckConflicts::No,
                                                  true,
                                                  GenerateBySeqno::No,
@@ -703,9 +701,7 @@ cb::engine_errc PassiveStream::processMessage(MutationConsumerMessage* message,
                 nullptr,
                 message->getVBucket(),
                 consumer->getCookie(),
-                {vbucket_state_active,
-                 vbucket_state_replica,
-                 vbucket_state_pending},
+                permittedVBStates,
                 CheckConflicts::No,
                 meta,
                 GenerateBySeqno::No,
@@ -847,7 +843,6 @@ cb::engine_errc PassiveStream::processAbort(
     if (!vb) {
         return cb::engine_errc::not_my_vbucket;
     }
-
     folly::SharedMutex::ReadHolder rlh(vb->getStateLock());
     if (!permittedVBStates.test(vb->getState())) {
         return cb::engine_errc::not_my_vbucket;
@@ -878,6 +873,10 @@ cb::engine_errc PassiveStream::processSystemEvent(
     if (!vb) {
         return cb::engine_errc::not_my_vbucket;
     }
+    folly::SharedMutex::ReadHolder rlh(vb->getStateLock());
+    if (!permittedVBStates.test(vb->getState())) {
+        return cb::engine_errc::not_my_vbucket;
+    }
 
     cb::engine_errc rv = cb::engine_errc::success;
 

engines/ep/tests/module_tests/dcp_stream_test.cc

@@ -6816,6 +6816,87 @@ TEST_P(CDCPassiveStreamTest, TouchedByExpelCheckpointNotReused) {
                                  0));
 }
 
+// MB-59518 is an issue where a DCP passive-stream processed a commit against
+// an active vbucket resulting in a duplicate commit (and failure). The issue
+// occurred because it was possible to reach the VBucket::commit function
+// during a replica->active state change provided that operations are buffered.
+// The fix was to put strong VBucket state checks in the PassiveStream commit
+// path.
+void SingleThreadedPassiveStreamTest::replicaToActiveBufferedRejected(
+        DcpResponse::Event event) {
+    consumer->disableFlatBuffersSystemEvents();
+
+    // Force consumer buffering.
+    auto& stats = engine->getEpStats();
+    stats.replicationThrottleThreshold = 0;
+    const size_t size = stats.getMaxDataSize();
+    engine->setMaxDataSize(1);
+    ASSERT_EQ(ReplicationThrottle::Status::Pause,
+              engine->getReplicationThrottle().getStatus());
+
+    auto vb = engine->getVBucket(vbid);
+
+    // Buffer the requested operation
+    snapshot(*consumer, stream->getOpaque(), 1, 1);
+
+    if (event == DcpResponse::Event::Mutation) {
+        EXPECT_EQ(cb::engine_errc::success,
+                  mutation(*consumer,
+                           stream->getOpaque(),
+                           makeStoredDocKey("key"),
+                           1));
+    } else if (event == DcpResponse::Event::Deletion) {
+        EXPECT_EQ(cb::engine_errc::success,
+                  deletion(*consumer,
+                           stream->getOpaque(),
+                           makeStoredDocKey("key"),
+                           1));
+    } else if (event == DcpResponse::Event::SystemEvent) {
+        EXPECT_EQ(cb::engine_errc::success,
+                  createCollection(*consumer, stream->getOpaque(), 1));
+    } else {
+        FAIL() << "Unexpected event";
+    }
+
+    // Was buffered and not applied.
+    EXPECT_EQ(0, vb->getHighSeqno());
+
+    std::function<void()> hook = [this]() {
+        // Change the vbucket state using the hook. This will happen whilst the
+        // consumer buffering "thread" has a stream pointer. Important or not
+        // is that ns_server uses "null" in the chain when the failover occurs
+        setVBucketState(
+                vbid,
+                vbucket_state_active,
+                nlohmann::json::parse(R"({"topology":[["active",null]]})"),
+                TransferVB::No);
+    };
+    stream->setProcessBufferedMessages_postFront_Hook(hook);
+
+    // And process the buffered commit. This commit should be rejected!
+    stats.replicationThrottleThreshold = 99;
+    engine->setMaxDataSize(size);
+    ASSERT_EQ(ReplicationThrottle::Status::Process,
+              engine->getReplicationThrottle().getStatus());
+    EXPECT_EQ(more_to_process, consumer->processBufferedItems());
+    EXPECT_EQ(all_processed, consumer->processBufferedItems());
+
+    // Expect no change, operation was rejected
+    EXPECT_EQ(0, vb->getHighSeqno());
+}
+
+TEST_P(SingleThreadedPassiveStreamTest, ReplicaToActiveBufferedMutation) {
+    replicaToActiveBufferedRejected(DcpResponse::Event::Mutation);
+}
+
+TEST_P(SingleThreadedPassiveStreamTest, ReplicaToActiveBufferedDeletion) {
+    replicaToActiveBufferedRejected(DcpResponse::Event::Deletion);
+}
+
+TEST_P(SingleThreadedPassiveStreamTest, ReplicaToActiveBufferedSystemEvent) {
+    replicaToActiveBufferedRejected(DcpResponse::Event::SystemEvent);
+}
+
 INSTANTIATE_TEST_SUITE_P(Persistent,
                          CDCPassiveStreamTest,
                          STParameterizedBucketTest::magmaConfigValues(),

engines/ep/tests/module_tests/dcp_stream_test.h

@@ -11,6 +11,7 @@
 
 #pragma once
 
+#include "dcp/response.h"
 #include "dcp_test.h"
 #include "hash_table.h"
 #include "test_manifest.h"
@@ -182,6 +183,9 @@ class SingleThreadedPassiveStreamTest
             const std::optional<cb::durability::Requirements>& durReqs,
             bool compressed = false);
 
+    // coverage for MB-59518
+    void replicaToActiveBufferedRejected(DcpResponse::Event event);
+
 protected:
     // Should the DcpConsumer have SyncReplication enabled when created in
     // SetUp()?

engines/ep/tests/module_tests/evp_store_single_threaded_test.cc

@@ -27,6 +27,7 @@
 #include "checkpoint_manager.h"
 #include "checkpoint_remover.h"
 #include "checkpoint_utils.h"
+#include "collections/events_generated.h"
 #include "collections/vbucket_manifest_handles.h"
 #include "dcp/active_stream_checkpoint_processor_task.h"
 #include "dcp/backfill-manager.h"
@@ -1093,6 +1094,22 @@ cb::engine_errc STParameterizedBucketTest::mutation(DcpConsumer& consumer,
                              /*nru*/ 0);
 }
 
+cb::engine_errc STParameterizedBucketTest::deletion(DcpConsumer& consumer,
+                                                    uint32_t opaque,
+                                                    const DocKey& key,
+                                                    uint64_t seqno) {
+    return consumer.deletion(opaque,
+                             key,
+                             cb::const_byte_buffer(),
+                             /*priv_bytes*/ 0,
+                             PROTOCOL_BINARY_DATATYPE_JSON,
+                             /*cas*/ 1,
+                             vbid,
+                             /*bySeqno*/ seqno,
+                             /*revSeqno*/ 0,
+                             /*meta*/ cb::const_byte_buffer());
+}
+
 cb::engine_errc STParameterizedBucketTest::prepare(DcpConsumer& consumer,
                                                    uint32_t opaque,
                                                    const DocKey& key,
@@ -1130,6 +1147,32 @@ cb::engine_errc STParameterizedBucketTest::abort(DcpConsumer& consumer,
     return consumer.abort(opaque, vbid, key, prepareSeqno, seqno);
 }
 
+cb::engine_errc STParameterizedBucketTest::createCollection(
+        DcpConsumer& consumer, uint32_t opaque, uint64_t seqno) {
+    // Create just the fruit collection @ seqno using flatbuffer message
+    flatbuffers::FlatBufferBuilder fb;
+    const auto collection = CollectionEntry::fruit;
+    auto fbPayload = Collections::VB::CreateCollection(
+            fb,
+            1,
+            uint32_t(ScopeEntry::defaultS.getId()),
+            uint32_t(collection.getId()),
+            false, /*max-ttl*/
+            0,
+            fb.CreateString(collection.name.data(), collection.name.size()),
+            false /*history*/);
+    fb.Finish(fbPayload);
+    return consumer.systemEvent(
+            opaque,
+            vbid,
+            mcbp::systemevent::id::CreateCollection,
+            seqno,
+            mcbp::systemevent::version::version2,
+            {reinterpret_cast<const uint8_t*>(collection.name.data()),
+             collection.name.size()},
+            {fb.GetBufferPointer(), fb.GetSize()});
+}
+
 /*
  * MB-31175
  * The following test checks to see that when we call handleSlowStream in an

engines/ep/tests/module_tests/evp_store_single_threaded_test.h

@@ -839,6 +839,10 @@ class STParameterizedBucketTest
                              uint32_t opaque,
                              const DocKey& key,
                              uint64_t seqno);
+    cb::engine_errc deletion(DcpConsumer& consumer,
+                             uint32_t opaque,
+                             const DocKey& key,
+                             uint64_t seqno);
     cb::engine_errc prepare(DcpConsumer& consumer,
                             uint32_t opaque,
                             const DocKey& key,
@@ -853,6 +857,9 @@ class STParameterizedBucketTest
                           const DocKey& key,
                           uint64_t prepareSeqno,
                           uint64_t seqno);
+    cb::engine_errc createCollection(DcpConsumer& consumer,
+                                     uint32_t opaque,
+                                     uint64_t seqno);
 
 protected:
     void SetUp() override;