[BP] MB-68753: Ensure new DcpConsumer is eligible for notifications

The DcpConsumer is contructed and leaving the parent member paused
as false. This means that ConnManager::run will ignore the connection
and that any local scheduleNotify will be a no-op.

This issue is behind the intermitent failure seen in MB-67544

This appears to work because we maybe relying upon step getting many
callbacks, and in the case where the connection uses its timeslice
no more callbacks until a notification is requested (which never
happens because of the paused=false issue).

Change-Id: I7357524b08d3e581b85c20b6ea76d0de4c58623f
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/234655
Tested-by: Jim Walker <[email protected]>
Well-Formed: Restriction Checker
Reviewed-by: Faizan Alam <[email protected]>

Author: jimwwalker

engines/ep/src/dcp/consumer.cc

@@ -212,6 +212,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() {
@@ -351,6 +355,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;
 }
 
@@ -1244,6 +1251,7 @@ void DcpConsumer::addStats(const AddStatFn& add_stat, const CookieIface* c) {
             c);
 
     addStat("synchronous_replication", isSyncReplicationEnabled(), add_stat, c);
+    addStat("pending_add_stream", pendingAddStream, add_stat, c);
 }
 
 void DcpConsumer::aggregateQueueStats(ConnCounter& aggregator) const {

engines/ep/src/dcp/consumer.h

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

engines/ep/src/dcp/passive_stream.cc

@@ -70,6 +70,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 {
@@ -496,6 +497,7 @@ void STDcpTest::processConsumerMutationsNearThreshold(bool beyondThreshold) {
     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 */
@@ -570,9 +572,6 @@ void STDcpTest::processConsumerMutationsNearThreshold(bool beyondThreshold) {
            task here */
         EXPECT_EQ(stop_processing, consumer->processBufferedItems());
 
-        /* Expect the connection to be notified */
-        EXPECT_FALSE(consumer->isPaused());
-
         /* Expect disconnect signal in Ephemeral with "fail_new_data" policy */
         EXPECT_EQ(cb::engine_errc::disconnect, consumer->step(producers));
     } else {

engines/ep/tests/module_tests/dcp_test.cc

@@ -2648,6 +2648,9 @@ TEST_P(FlowControlTest, NotifyConsumerOnlyIfFlowControlEnabled) {
     // 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,