Merge "MB-68753: Merge commit '2bf70e3a2' into 'couchbase/trinity'" into trinity

Author: Gerrit Code Review

engines/ep/src/dcp/consumer.cc

@@ -206,6 +206,10 @@ DcpConsumer::DcpConsumer(EventuallyPersistentEngine& engine,
     // Enable ChangeStreams on this connection
     changeStreamsNegotiation.state =
             BlockingDcpControlNegotiation::State::PendingRequest;
+
+    // MB-68753: Pause the consumer so subsequent scheduleNotify will wake the
+    // consumer and the connection will get callbacks from ConnManager::run
+    pause(PausedReason::ReadyListEmpty);
 }
 
 DcpConsumer::~DcpConsumer() {
@@ -345,6 +349,9 @@ cb::engine_errc DcpConsumer::addStream(uint32_t opaque,
     opaqueMap_[new_opaque] = std::make_pair(opaque, vbucket);
     pendingAddStream = false;
 
+    // A DcpStreamRequest should be ready for transmission
+    scheduleNotify();
+
     return cb::engine_errc::success;
 }
 
@@ -1276,6 +1283,7 @@ void DcpConsumer::addStats(const AddStatFn& add_stat, CookieIface& c) {
             c);
 
     addStat("synchronous_replication", isSyncReplicationEnabled(), add_stat, c);
+    addStat("pending_add_stream", pendingAddStream, add_stat, c);
 }
 
 void DcpConsumer::addStreamStats(const AddStatFn& add_stat,

engines/ep/src/dcp/consumer.h

@@ -570,7 +570,7 @@ class DcpConsumer : public ConnHandler,
     std::chrono::duration<float> dcpNoopTxInterval;
 
     // Step can't start sending packets until we've received add stream
-    bool pendingAddStream = true;
+    cb::RelaxedAtomic<bool> pendingAddStream = true;
 
     bool pendingEnableNoop;
     NoopIntervalNegotiation noopIntervalNegotiation;

engines/ep/src/dcp/passive_stream.cc

@@ -68,6 +68,9 @@ PassiveStream::PassiveStream(EventuallyPersistentEngine* e,
     std::lock_guard<std::mutex> lh(streamMutex);
     streamRequest_UNLOCKED(vb_uuid);
     itemsReady.store(true);
+    // Cannot call scheduleNotify here as we're not in the readyStreamsVBQueue.
+    // The addStream path will call scheduleNotify as at that point this new
+    // object will be in the readyStreamsVBQueue and can be picked up by step.
 }
 
 PassiveStream::~PassiveStream() {

engines/ep/tests/module_tests/dcp_single_threaded_test.cc

@@ -25,6 +25,7 @@
 #include "vbucket.h"
 
 #include <programs/engine_testapp/mock_cookie.h>
+#include <programs/engine_testapp/mock_server.h>
 #include <spdlog/spdlog.h>
 
 class STDcpTest : public STParameterizedBucketTest {
@@ -584,6 +585,7 @@ TEST_P(STDcpTest, ProcessUnackedBytesAtReplicationOOM) {
     auto& mockConnMap = static_cast<MockDcpConnMap&>(connMap);
     auto consumer =
             std::make_shared<MockDcpConsumer>(*engine, cookie, "test_consumer");
+    EXPECT_TRUE(consumer->isPaused());
     mockConnMap.addConn(cookie, consumer);
 
     // Replica vbucket
@@ -642,9 +644,6 @@ TEST_P(STDcpTest, ProcessUnackedBytesAtReplicationOOM) {
         EXPECT_EQ(0, vb.getHighSeqno());
         EXPECT_EQ(0, unackedBytes);
 
-        // Expect the connection to be notified
-        EXPECT_FALSE(consumer->isPaused());
-
         // Simulate the DcpConsumerTask - Nothing to process, we are just
         // disconnecting
         EXPECT_EQ(all_processed, consumer->processUnackedBytes());

engines/ep/tests/module_tests/dcp_test.cc

@@ -2776,6 +2776,9 @@ void FlowControlTestBase::testNotifyConsumerOnlyIfFlowControlEnabled(
     // Setup the stream
     ASSERT_EQ(cb::engine_errc::success,
               consumer->addStream(opaque, vbid, 0 /*flags*/));
+    // Test for addStream notify because this queues an outbound message
+    ASSERT_TRUE(mock_cookie_notified(cookie_to_mock_cookie(cookie)));
+    mock_waitfor_cookie(cookie_to_mock_cookie(cookie));
     opaque += 1;
     ASSERT_EQ(cb::engine_errc::success,
               consumer->snapshotMarker(opaque,