MB-35226: Do not process seqno ack after stream is dead

We should not ack after a stream is dead because logically we
should not process any other messages. This previously could have
manifested in a where we receive a seqno ack for a replica that does
not exist in the replication topology and would have been added to
the queuedSeqnoAcks after a stream was dead. Setting the stream to
dead removes the queuedSeqnoAck so we would keep a queuedSeqnoAck
for a stream that no longer exists. If this replica were then to
reconnect and stream up to and ack the same sequence number, then
we would fire a monotonic invariant exception.

Change-Id: I976b4a1dedde58cde351ea543aca94e0f6370957
Reviewed-on: http://review.couchbase.org/112714
Reviewed-by: Dave Rigby <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: BenHuddleston

engines/ep/src/dcp/active_stream.cc

@@ -1746,3 +1746,32 @@ bool ActiveStream::removeCheckpointCursor() {
     }
     return false;
 }
+
+ENGINE_ERROR_CODE ActiveStream::seqnoAck(const std::string& consumerName,
+                                         uint64_t preparedSeqno) {
+    VBucketPtr vb = engine->getVBucket(vb_);
+    if (!vb) {
+        return ENGINE_NOT_MY_VBUCKET;
+    }
+
+    // Take the vb state lock so that we don't change the state of
+    // this vb. Done before the streamMutex is acquired to prevent a lock order
+    // inversion.
+    {
+        folly::SharedMutex::ReadHolder vbStateLh(vb->getStateLock());
+
+        // Locked with the streamMutex to ensure that we cannot race with a
+        // stream end
+        {
+            LockHolder lh(streamMutex);
+
+            // We cannot ack something on a dead stream.
+            if (!isActive()) {
+                return ENGINE_SUCCESS;
+            }
+
+            return vb->seqnoAcknowledged(
+                    vbStateLh, consumerName, preparedSeqno);
+        } // end stream mutex lock scope
+    } // end vb state lock scope
+}

engines/ep/src/dcp/active_stream.h

@@ -19,6 +19,7 @@
 
 #include "collections/vbucket_filter.h"
 #include "dcp/stream.h"
+#include <memcached/engine_error.h>
 #include <spdlog/common.h>
 
 class CheckpointManager;
@@ -199,6 +200,15 @@ class ActiveStream : public Stream,
         std::vector<queued_item> items;
     };
 
+    /**
+     * Process a seqno ack against this stream.
+     *
+     * @param consumerName the name of the consumer acking
+     * @param preparedSeqno the seqno that the consumer is acking
+     */
+    ENGINE_ERROR_CODE seqnoAck(const std::string& consumerName,
+                               uint64_t preparedSeqno);
+
 protected:
     /**
      * @param vb reference to the associated vbucket

engines/ep/src/dcp/producer.cc

@@ -1044,29 +1044,25 @@ ENGINE_ERROR_CODE DcpProducer::seqno_acknowledged(uint32_t opaque,
                 " but we don't have a StreamContainer for that vb");
     }
 
-    uint64_t lastSentSeqno;
+    std::shared_ptr<ActiveStream> stream;
     {
         auto handle = rv->second->rlock();
 
         // Producer for replication should only have one stream.
         Expects(handle.size() == 1);
 
-        auto* s = dynamic_cast<ActiveStream*>(handle.get().get());
-
-        // We should not have received a seqno ack if the stream
-        // is not an ActiveStream
-        Expects(s);
-        lastSentSeqno = s->getLastSentSeqno();
+        stream = dynamic_pointer_cast<ActiveStream>(handle.get());
+        Expects(stream.get());
     }
 
-    if (prepared_seqno > lastSentSeqno) {
+    if (prepared_seqno > stream->getLastSentSeqno()) {
         throw std::logic_error(
                 "Replica acked seqno:" + std::to_string(prepared_seqno) +
                 " greater than last sent seqno:" +
-                std::to_string(lastSentSeqno));
+                std::to_string(stream->getLastSentSeqno()));
     }
 
-    return vb->seqnoAcknowledged(consumerName, prepared_seqno);
+    return stream->seqnoAck(consumerName, prepared_seqno);
 }
 
 bool DcpProducer::handleResponse(const protocol_binary_response_header* resp) {

engines/ep/src/durability/active_durability_monitor.cc

@@ -860,6 +860,7 @@ void ActiveDurabilityMonitor::State::updateNodeAck(const std::string& node,
         // track the ack for this node in case we are about to get a topology
         // change in which this node will exist.
         queuedSeqnoAcks[node] = seqno;
+        queuedSeqnoAcks[node].setLabel("queuedSeqnoAck: " + node);
     }
 }
 

engines/ep/src/vbucket.cc

@@ -3820,12 +3820,10 @@ bool VBucket::isLogicallyNonExistent(
            cHandle.isLogicallyDeleted(v.getBySeqno());
 }
 
-ENGINE_ERROR_CODE VBucket::seqnoAcknowledged(const std::string& replicaId,
-                                             uint64_t preparedSeqno) {
-    // Take the state lock to ensure that we don't change state whilst
-    // processing a seqno ack.
-    folly::SharedMutex::ReadHolder rlh(getStateLock());
-
+ENGINE_ERROR_CODE VBucket::seqnoAcknowledged(
+        const folly::SharedMutex::ReadHolder& vbStateLock,
+        const std::string& replicaId,
+        uint64_t preparedSeqno) {
     // We may receive an ack after we have set a vBucket to dead during a
     // takeover; just ignore it.
     if (getState() == vbucket_state_dead) {

engines/ep/src/vbucket.h

@@ -1668,11 +1668,14 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
      * Inform the vBucket that sequence number(s) have been acknowledged by
      * a replica node.
      *
+     * @param vbStateLock read lock on the vBucket state.
      * @param replicaId The replica node which has acknowledged.
      * @param preparedSeqno The sequence number the replica has prepared up to.
      */
-    ENGINE_ERROR_CODE seqnoAcknowledged(const std::string& replicaId,
-                                        uint64_t preparedSeqno);
+    ENGINE_ERROR_CODE seqnoAcknowledged(
+            const folly::SharedMutex::ReadHolder& vbStateLock,
+            const std::string& replicaId,
+            uint64_t preparedSeqno);
 
     /**
      * Notify the DurabilityMonitor that the Flusher has persisted all the

engines/ep/tests/module_tests/dcp_durability_stream_test.cc

@@ -198,7 +198,7 @@ void DurabilityActiveStreamTest::testSendCompleteSyncWrite(Resolution res) {
         // Majority Prepare), simulating a SeqnoAck received from replica
         // satisfies Durability Requirements and triggers Commit. So, the
         // following indirectly calls VBucket::commit
-        vb->seqnoAcknowledged(replica, prepareSeqno);
+        stream->seqnoAck(replica, prepareSeqno);
         // Note: At FE we have an exact item count only at persistence.
         auto evictionType = std::get<1>(GetParam());
         if (evictionType == "value_only" || !persistent()) {
@@ -474,14 +474,21 @@ TEST_P(DurabilityActiveStreamTest, RemoveUnknownSeqnoAckAtDestruction) {
     // Our topology gives replica name as "replica" an our producer/stream has
     // name "test_producer". Simulate a seqno ack by calling the vBucket level
     // function.
-    vb->seqnoAcknowledged("test_producer", 1);
+    stream->seqnoAck("test_producer", 1);
 
     // An unknown seqno ack should not have committed the item
     EXPECT_EQ(0, vb->getNumItems());
 
     // Disconnect the ActiveStream
     stream->setDead(END_STREAM_DISCONNECTED);
 
+    // Attempt to ack the seqno again. The stream is dead so we should not
+    // process the ack although we return SUCCESS to avoid tearing down any
+    // connections. We verify that the seqno ack does not exist in the map
+    // by performing the topology change that would commit the prepare if it
+    // did.
+    EXPECT_EQ(ENGINE_SUCCESS, stream->seqnoAck("test_producer", 1));
+
     // If the seqno ack still existed in the queuedSeqnoAcks map then it would
     // result in a commit on topology change
     setVBucketStateAndRunPersistTask(

engines/ep/tests/module_tests/evp_store_durability_test.cc

@@ -1259,7 +1259,10 @@ TEST_P(DurabilityEPBucketTest, ActiveLocalNotifyPersistedSeqno) {
 
     // Replica acks disk-seqno
     EXPECT_EQ(ENGINE_SUCCESS,
-              vb->seqnoAcknowledged("replica", 3 /*preparedSeqno*/));
+              vb->seqnoAcknowledged(
+                      folly::SharedMutex::ReadHolder(vb->getStateLock()),
+                      "replica",
+                      3 /*preparedSeqno*/));
     // Active has not persisted, so Durability Requirements not satisfied yet
     checkPending();
 

engines/ep/tests/module_tests/evp_store_warmup_test.cc

@@ -657,7 +657,10 @@ void DurabilityWarmupTest::testCommittedSyncWrite(vbucket_state_t vbState,
     { // scoping vb - is invalid once resetEngineAndWarmup() is called.
         auto vb = engine->getVBucket(vbid);
         EXPECT_EQ(ENGINE_SUCCESS,
-                  vb->seqnoAcknowledged("replica", vb->getHighPreparedSeqno()));
+                  vb->seqnoAcknowledged(
+                          folly::SharedMutex::ReadHolder(vb->getStateLock()),
+                          "replica",
+                          vb->getHighPreparedSeqno()));
 
         flush_vbucket_to_disk(vbid, 1);
     }
@@ -942,7 +945,11 @@ void DurabilityWarmupTest::testHCSPersistedAndLoadedIntoVBState() {
     // Complete the Prepare
     vb = store->getVBucket(vbid);
     ASSERT_TRUE(vb);
-    EXPECT_EQ(ENGINE_SUCCESS, vb->seqnoAcknowledged("replica", preparedSeqno));
+    EXPECT_EQ(ENGINE_SUCCESS,
+              vb->seqnoAcknowledged(
+                      folly::SharedMutex::ReadHolder(vb->getStateLock()),
+                      "replica",
+                      preparedSeqno));
     sv = vb->ht.findForRead(key).storedValue;
     ASSERT_TRUE(sv);
     ASSERT_TRUE(sv->isCommitted());
@@ -1048,13 +1055,19 @@ TEST_P(DurabilityWarmupTest, CommittedWithAckAfterWarmup) {
     flush_vbucket_to_disk(vbid);
     {
         auto vb = engine->getVBucket(vbid);
-        vb->seqnoAcknowledged("replica", 1);
+        vb->seqnoAcknowledged(
+                folly::SharedMutex::ReadHolder(vb->getStateLock()),
+                "replica",
+                1);
         flush_vbucket_to_disk(vbid, 1);
     }
     resetEngineAndWarmup();
     {
         auto vb = engine->getVBucket(vbid);
-        vb->seqnoAcknowledged("replica", 1);
+        vb->seqnoAcknowledged(
+                folly::SharedMutex::ReadHolder(vb->getStateLock()),
+                "replica",
+                1);
     }
 }
 
@@ -1075,14 +1088,20 @@ TEST_P(DurabilityWarmupTest, WarmUpHPSAndHCSWithNonSeqnoSortedItems) {
     flush_vbucket_to_disk(vbid, 2);
     {
         auto vb = engine->getVBucket(vbid);
-        vb->seqnoAcknowledged("replica", 2);
+        vb->seqnoAcknowledged(
+                folly::SharedMutex::ReadHolder(vb->getStateLock()),
+                "replica",
+                2);
         flush_vbucket_to_disk(vbid, 2);
     }
     SCOPED_TRACE("B");
     resetEngineAndWarmup();
     {
         auto vb = engine->getVBucket(vbid);
-        vb->seqnoAcknowledged("replica", 2);
+        vb->seqnoAcknowledged(
+                folly::SharedMutex::ReadHolder(vb->getStateLock()),
+                "replica",
+                2);
     }
 }
 

engines/ep/tests/module_tests/vbucket_durability_test.cc

@@ -148,7 +148,10 @@ void VBucketDurabilityTest::testAddPrepareAndCommit(
     ckptMgr->clear(*vbucket, ckptMgr->getHighSeqno());
 
     // Simulate replica and active seqno-ack
-    vbucket->seqnoAcknowledged(replica1, writes.back().seqno);
+    vbucket->seqnoAcknowledged(
+            folly::SharedMutex::ReadHolder(vbucket->getStateLock()),
+            replica1,
+            writes.back().seqno);
     simulateLocalAck(writes.back().seqno);
 
     int i = 0;
@@ -582,13 +585,19 @@ TEST_P(VBucketDurabilityTest, Active_Commit_MultipleReplicas) {
     checkPending();
 
     // replica2 acks, Durability Requirements not satisfied yet
-    vbucket->seqnoAcknowledged(replica2, preparedSeqno);
+    vbucket->seqnoAcknowledged(
+            folly::SharedMutex::ReadHolder(vbucket->getStateLock()),
+            replica2,
+            preparedSeqno);
     checkPending();
 
     // replica3 acks, Durability Requirements satisfied
     // Note: ensure 1 Ckpt in CM, easier to inspect the CkptList after Commit
     ckptMgr->clear(*vbucket, ckptMgr->getHighSeqno());
-    vbucket->seqnoAcknowledged(replica3, preparedSeqno);
+    vbucket->seqnoAcknowledged(
+            folly::SharedMutex::ReadHolder(vbucket->getStateLock()),
+            replica3,
+            preparedSeqno);
     checkCommitted();
 }
 
@@ -666,7 +675,10 @@ TEST_P(VBucketDurabilityTest, Active_PendingSkippedAtEjectionAndCommit) {
               swCompleteTrace);
 
     // Simulate replica and active seqno-ack
-    vbucket->seqnoAcknowledged(replica1, preparedSeqno);
+    vbucket->seqnoAcknowledged(
+            folly::SharedMutex::ReadHolder(vbucket->getStateLock()),
+            replica1,
+            preparedSeqno);
     simulateLocalAck(preparedSeqno);
 
     // Commit notified
@@ -1947,7 +1959,11 @@ TEST_P(VBucketDurabilityTest, IgnoreAckAtTakeoverDead) {
     // VBState transitions from Replica to Active
     vbucket->setState(vbucket_state_dead, nlohmann::json{});
 
-    EXPECT_EQ(ENGINE_SUCCESS, vbucket->seqnoAcknowledged("replica1", 3));
+    EXPECT_EQ(ENGINE_SUCCESS,
+              vbucket->seqnoAcknowledged(
+                      folly::SharedMutex::ReadHolder(vbucket->getStateLock()),
+                      "replica1",
+                      3));
     // We won't have crashed and we will have ignored the ack
     EXPECT_EQ(seqnos.size(), adm.getNumTracked());
 }