MB-35429: Ack correct PassiveStream

The seqno ack code incorrectly assumes that we have only one
Consumer in the vbConns "map". This may not be the case if we have
a Consumer with dead PassiveStream and another Consumer with an
alive one.

Instead of attempting to ack for the first Consumer found, ack for
all Consumers found but only if the PassiveStream is alive.

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

Author: BenHuddleston

engines/ep/src/dcp/dcpconnmap.cc

@@ -431,39 +431,43 @@ void DcpConnMap::notifyVBConnections(Vbid vbid, uint64_t bySeqno) {
 }
 
 void DcpConnMap::seqnoAckVBPassiveStream(Vbid vbid, int64_t seqno) {
-    std::shared_ptr<DcpConsumer> conn;
+    std::vector<std::shared_ptr<DcpConsumer>> conns;
     { // Locking scope for vbConnsLocks[index]
         size_t index = vbid.get() % vbConnLockNum;
         std::lock_guard<std::mutex> lg(vbConnLocks[index]);
 
-        // Note: logically we can have only one Consumer per VBucket, but I keep
-        // using the existing vbConns mapping for now (originally added for
-        // tracking only Producers).
+        // Note: logically we should only have one Consumer per vBucket but
+        // we may keep around old Consumers with either no PassiveStream for
+        // this vBucket or a dead PassiveStream. We need to search the list of
+        // ConnHandlers for the Consumer with the alive PassiveStream for this
+        // vBucket.
         for (auto& weakPtr : vbConns[vbid.get()]) {
             auto connection = weakPtr.lock();
             if (!connection) {
                 continue;
             }
             auto consumer = dynamic_pointer_cast<DcpConsumer>(connection);
             if (consumer) {
-                // MB-34437
-                // We should only have one consumer per vb (and we only care
-                // about consumers in this function) so we can now unlock the
-                // vbConnsLock to prevent a lock order inversion with
-                // PassiveStreamMap RWLock as ownership of the actual object is
-                // maintained by the shared_ptr.
-                conn = consumer;
-                break;
+                // We need to search for the alive stream in the Consumers but
+                // can't do so inside the vbConnLock due to lock order inversion
+                // so store the Consumers for later processing.
+                conns.push_back(std::move(consumer));
             }
         }
     }
 
-    // Note: Sync Repl enabled at Consumer only if Producer supports it.
-    //     This is to prevent that 6.5 Consumers send DCP_SEQNO_ACK to
-    //     pre-6.5 Producers (e.g., topology change in a 6.5 cluster
-    //     where a new pre-6.5 Active is elected).
-    if (conn && conn->isSyncReplicationEnabled()) {
-        conn->seqnoAckStream(vbid, seqno);
+    for (auto& conn : conns) {
+        // Note: Sync Repl enabled at Consumer only if Producer supports it.
+        //     This is to prevent that 6.5 Consumers send DCP_SEQNO_ACK to
+        //     pre-6.5 Producers (e.g., topology change in a 6.5 cluster
+        //     where a new pre-6.5 Active is elected).
+        if (conn->isSyncReplicationEnabled() && conn->isStreamPresent(vbid)) {
+            // Ideally, we would verify here that we only ack a single consumer,
+            // however, that is not possible without adding additional locking
+            // to ensure that no consumer can add new streams whilst this loop
+            // runs.
+            conn->seqnoAckStream(vbid, seqno);
+        }
     }
 }
 

engines/ep/src/dcp/passive_stream.cc

@@ -711,6 +711,10 @@ ENGINE_ERROR_CODE PassiveStream::processPrepare(
 void PassiveStream::seqnoAck(int64_t seqno) {
     {
         LockHolder lh(streamMutex);
+        if (!isActive()) {
+            return;
+        }
+
         pushToReadyQ(
                 std::make_unique<SeqnoAcknowledgement>(opaque_, vb_, seqno));
     }

engines/ep/tests/module_tests/dcp_test.cc

@@ -1548,6 +1548,65 @@ TEST_P(ConnectionTest, ConsumerWithConsumerNameEnablesSyncRepl) {
     destroy_mock_cookie(cookie);
 }
 
+TEST_P(ConnectionTest, AckCorrectPassiveStream) {
+    ASSERT_EQ(ENGINE_SUCCESS, set_vb_state(vbid, vbucket_state_replica));
+
+    const void* cookie1 = create_mock_cookie();
+    Vbid vbid = Vbid(0);
+
+    // Create our first consumer and enable SyncReplication so that we can ack
+    auto consumer1 =
+            std::make_shared<MockDcpConsumer>(*engine, cookie1, "consumer1");
+    consumer1->enableSyncReplication();
+
+    // Add our stream and accept to mimic real scenario and ensure it is alive
+    ASSERT_EQ(ENGINE_SUCCESS,
+              consumer1->addStream(0 /*opaque*/, vbid, 0 /*flags*/));
+    MockPassiveStream* stream1 = static_cast<MockPassiveStream*>(
+            (consumer1->getVbucketStream(vbid)).get());
+    stream1->acceptStream(cb::mcbp::Status::Success, 0 /*opaque*/);
+    ASSERT_TRUE(stream1->isActive());
+
+    ASSERT_EQ(2, stream1->public_readyQ().size());
+    auto resp = stream1->public_popFromReadyQ();
+    EXPECT_EQ(DcpResponse::Event::StreamReq, resp->getEvent());
+    resp = stream1->public_popFromReadyQ();
+    EXPECT_EQ(DcpResponse::Event::AddStream, resp->getEvent());
+    EXPECT_EQ(0, stream1->public_readyQ().size());
+
+    // Now close the stream to transition to dead
+    consumer1->closeStream(0 /*opaque*/, vbid);
+    ASSERT_FALSE(stream1->isActive());
+
+    // Add a new consumer and new PassiveStream
+    const void* cookie2 = create_mock_cookie();
+    auto consumer2 =
+            std::make_shared<MockDcpConsumer>(*engine, cookie2, "consumer2");
+    consumer2->enableSyncReplication();
+    ASSERT_EQ(ENGINE_SUCCESS,
+              consumer2->addStream(1 /*opaque*/, vbid, 0 /*flags*/));
+    MockPassiveStream* stream2 = static_cast<MockPassiveStream*>(
+            (consumer2->getVbucketStream(vbid)).get());
+    stream2->acceptStream(cb::mcbp::Status::Success, 1 /*opaque*/);
+    ASSERT_TRUE(stream2->isActive());
+
+    EXPECT_EQ(2, stream2->public_readyQ().size());
+    resp = stream2->public_popFromReadyQ();
+    EXPECT_EQ(DcpResponse::Event::StreamReq, resp->getEvent());
+    resp = stream2->public_popFromReadyQ();
+    EXPECT_EQ(DcpResponse::Event::AddStream, resp->getEvent());
+    EXPECT_EQ(0, stream2->public_readyQ().size());
+
+    // When we ack we should hit the active stream (stream2)
+    engine->getDcpConnMap().seqnoAckVBPassiveStream(vbid, 1);
+    EXPECT_EQ(0, stream1->public_readyQ().size());
+    EXPECT_EQ(1, stream2->public_readyQ().size());
+
+    ASSERT_EQ(ENGINE_SUCCESS, consumer2->closeStream(1 /*opaque*/, vbid));
+    destroy_mock_cookie(cookie1);
+    destroy_mock_cookie(cookie2);
+}
+
 /*
  * The following tests that when the disk_backfill_queue configuration is
  * set to false on receiving a snapshot marker it does not move into the