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

Add VBucket state checks to processAbort so that the abort is not
processed against an active vbucket.

Change-Id: I757e24e55ada716932bd49b984ec5d5b5f75c444
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/200506
Reviewed-by: Dave Rigby <[email protected]>
Tested-by: Jim Walker <[email protected]>
Well-Formed: Restriction Checker

Author: jimwwalker

engines/ep/src/dcp/passive_stream.cc

@@ -848,6 +848,11 @@ cb::engine_errc PassiveStream::processAbort(
         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;
+    }
+
     auto rv = vb->abort(abort.getKey(),
                         abort.getPreparedSeqno(),
                         abort.getAbortSeqno(),

engines/ep/tests/module_tests/dcp_durability_stream_test.cc

@@ -4557,7 +4557,8 @@ TEST_P(DurabilityPassiveStreamPersistentTest, BufferDcpAbort) {
 // 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.
-TEST_P(DurabilityPassiveStreamPersistentTest, ReplicaToActiveBufferedCommit) {
+void DurabilityPassiveStreamPersistentTest::replicaToActiveBufferedResolution(
+        bool resolutionIsCommit) {
     // Begin by pushing a mutation. Only to get away from seqno:0
     auto key = makeStoredDocKey("key");
     EXPECT_EQ(cb::engine_errc::success,
@@ -4584,11 +4585,16 @@ TEST_P(DurabilityPassiveStreamPersistentTest, ReplicaToActiveBufferedCommit) {
 
     auto vb = engine->getVBucket(vbid);
 
-    // Buffer the commit of the prepare
+    // Buffer the commit/abort of the prepare
     EXPECT_EQ(cb::engine_errc::success,
               snapshot(*consumer, stream->getOpaque(), 3, 3));
-    EXPECT_EQ(cb::engine_errc::success,
-              commit(*consumer, stream->getOpaque(), durableKey, 2, 3));
+    if (resolutionIsCommit) {
+        EXPECT_EQ(cb::engine_errc::success,
+                  commit(*consumer, stream->getOpaque(), durableKey, 2, 3));
+    } else {
+        EXPECT_EQ(cb::engine_errc::success,
+                  abort(*consumer, stream->getOpaque(), durableKey, 2, 3));
+    }
     EXPECT_EQ(2, vb->getHighSeqno());
 
     std::function<void()> hook = [this]() {
@@ -4601,16 +4607,16 @@ TEST_P(DurabilityPassiveStreamPersistentTest, ReplicaToActiveBufferedCommit) {
     };
     stream->setProcessBufferedMessages_postFront_Hook(hook);
 
-    // And process the buffered commit. This commit should be rejected. Prior
-    // to the fix it would be processed against the active.
+    // 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());
 
-    // Before fixing high-seqno would be at 3 (commit)
+    // Before fixing high-seqno would be at 3 (commit/abort)
     EXPECT_EQ(2, vb->getHighSeqno());
 
     setVBucketState(
@@ -4628,6 +4634,7 @@ TEST_P(DurabilityPassiveStreamPersistentTest, ReplicaToActiveBufferedCommit) {
 
     // And fail. With MB-59518 an error is logged about a failure to find the
     // prepare, this function would throw because the commit is not successful
+    // Note the abort case would fail here in the same way as commit
     try {
         vb->processResolvedSyncWrites();
     } catch (const std::exception& e) {
@@ -4637,6 +4644,14 @@ TEST_P(DurabilityPassiveStreamPersistentTest, ReplicaToActiveBufferedCommit) {
     flushVBucketToDiskIfPersistent(vbid);
 }
 
+TEST_P(DurabilityPassiveStreamPersistentTest, ReplicaToActiveBufferedCommit) {
+    replicaToActiveBufferedResolution(true);
+}
+
+TEST_P(DurabilityPassiveStreamPersistentTest, ReplicaToActiveBufferedAbort) {
+    replicaToActiveBufferedResolution(false);
+}
+
 void DurabilityPromotionStreamTest::SetUp() {
     // Set up as a replica
     DurabilityPassiveStreamTest::SetUp();

engines/ep/tests/module_tests/dcp_durability_stream_test.h

@@ -275,6 +275,9 @@ class DurabilityPassiveStreamPersistentTest
      * Read the highCompletedSeqno from disk.
      */
     uint64_t getPersistedHCS();
+
+    /// Test coverage for MB-59518
+    void replicaToActiveBufferedResolution(bool resolutionIsCommit);
 };
 
 class DurabilityPassiveStreamEphemeralTest

engines/ep/tests/module_tests/evp_store_single_threaded_test.cc

@@ -1127,6 +1127,14 @@ cb::engine_errc STParameterizedBucketTest::commit(DcpConsumer& consumer,
     return consumer.commit(opaque, vbid, key, prepareSeqno, seqno);
 }
 
+cb::engine_errc STParameterizedBucketTest::abort(DcpConsumer& consumer,
+                                                 uint32_t opaque,
+                                                 const DocKey& key,
+                                                 uint64_t prepareSeqno,
+                                                 uint64_t seqno) {
+    return consumer.abort(opaque, vbid, key, prepareSeqno, seqno);
+}
+
 /*
  * 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

@@ -846,6 +846,11 @@ class STParameterizedBucketTest
                            const DocKey& key,
                            uint64_t prepareSeqno,
                            uint64_t seqno);
+    cb::engine_errc abort(DcpConsumer& consumer,
+                          uint32_t opaque,
+                          const DocKey& key,
+                          uint64_t prepareSeqno,
+                          uint64_t seqno);
 
 protected:
     void SetUp() override;