MB-59518: Add test coverage for a buffered prepare

Prepare already had correct state checks as KVBucket::prepare locks
the vbucket and checks for replica or pending.

A unit test is added to cover the case of a buffered prepare

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

Author: jimwwalker

engines/ep/tests/module_tests/dcp_durability_stream_test.cc

@@ -4659,6 +4659,62 @@ TEST_P(DurabilityPassiveStreamPersistentTest, ReplicaToActiveBufferedAbort) {
     replicaToActiveBufferedResolution(false);
 }
 
+TEST_P(DurabilityPassiveStreamPersistentTest, ReplicaToActiveBufferedPrepare) {
+    // Begin by pushing a mutation. Only to get away from seqno:0
+    auto key = makeStoredDocKey("key");
+    EXPECT_EQ(cb::engine_errc::success,
+              snapshot(*consumer, stream->getOpaque(), 1, 1));
+    EXPECT_EQ(cb::engine_errc::success,
+              mutation(*consumer, stream->getOpaque(), key, 1));
+    flushVBucketToDiskIfPersistent(vbid);
+
+    EXPECT_EQ(cb::engine_errc::success,
+              snapshot(*consumer, stream->getOpaque(), 2, 2));
+
+    // Now begin 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 prepare.
+    // Note to detect a bug when processing a buffered snapshot marker, if we
+    // set the type as DISK it would hit an exception when calling
+    // setDuplicatePrepareWindow.
+    // Now push a prepare onto the replica.
+    auto durableKey = makeStoredDocKey("durable");
+
+    EXPECT_EQ(cb::engine_errc::success,
+              prepare(*consumer, stream->getOpaque(), durableKey, 2));
+    EXPECT_EQ(1, 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.
+        setVBucketState(vbid,
+                        vbucket_state_active,
+                        nlohmann::json::parse(R"({"topology":[["active"]]})"),
+                        TransferVB::No);
+    };
+    stream->setProcessBufferedMessages_postFront_Hook(hook);
+
+    // And process the buffered resolution. This operation should be rejected.
+    // Prior to the fix it would be processed against the active.
+    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());
+
+    // Prepare must be rejected, no change to high-seq
+    EXPECT_EQ(1, vb->getHighSeqno());
+}
+
 void DurabilityPromotionStreamTest::SetUp() {
     // Set up as a replica
     DurabilityPassiveStreamTest::SetUp();