MB-41719: Make test_producer_stream_end_on_client_close_stream ST

Change-Id: Ia997c31b34a6a7d3090eb71e94bee0e3c26a9f8d
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/137423
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: BenHuddleston

engines/ep/tests/mock/mock_dcp_conn_map.cc

@@ -43,3 +43,9 @@ bool MockDcpConnMap::removeConn(const void* cookie) {
 bool MockDcpConnMap::doesVbConnExist(Vbid vbid, const std::string& name) {
     return connStore->doesVbConnExist(vbid, name);
 }
+
+bool MockDcpConnMap::doesConnHandlerExist(const std::string& name) {
+    return connStore->getCookieToConnectionMapHandle()
+            ->findConnHandlerByName(name)
+            .get();
+}

engines/ep/tests/mock/mock_dcp_conn_map.h

@@ -53,6 +53,8 @@ class MockDcpConnMap : public DcpConnMap {
     /// structure
     bool doesVbConnExist(Vbid vbid, const std::string& name);
 
+    bool doesConnHandlerExist(const std::string& name);
+
 protected:
     /**
      * @param engine The engine

engines/ep/tests/module_tests/dcp_single_threaded_test.cc

@@ -31,6 +31,47 @@
 
 class STDcpTest : public STParameterizedBucketTest {
 protected:
+    struct StreamRequestResult {
+        ENGINE_ERROR_CODE status;
+        uint64_t rollbackSeqno;
+    };
+
+    // callbackCount needs to be static as its used inside of the static
+    // function fakeDcpAddFailoverLog.
+    // static int callbackCount;
+
+    /*
+     * Fake callback emulating dcp_add_failover_log
+     */
+    static ENGINE_ERROR_CODE fakeDcpAddFailoverLog(
+            vbucket_failover_t* entry,
+            size_t nentries,
+            gsl::not_null<const void*> cookie) {
+        // callbackCount++;
+        return ENGINE_SUCCESS;
+    }
+
+    StreamRequestResult doStreamRequest(DcpProducer& producer,
+                                        uint64_t startSeqno = 0,
+                                        uint64_t endSeqno = ~0,
+                                        uint64_t snapStart = 0,
+                                        uint64_t snapEnd = ~0,
+                                        uint64_t vbUUID = 0) {
+        StreamRequestResult result;
+        result.status = producer.streamRequest(/*flags*/ 0,
+                                               /*opaque*/ 0,
+                                               vbid,
+                                               startSeqno,
+                                               endSeqno,
+                                               vbUUID,
+                                               snapStart,
+                                               snapEnd,
+                                               &result.rollbackSeqno,
+                                               fakeDcpAddFailoverLog,
+                                               {});
+        return result;
+    }
+
     /**
      * @param producerState Are we simulating a negotiation against a Producer
      *  that enables IncludeDeletedUserXattrs?
@@ -565,6 +606,63 @@ TEST_P(STDcpTest,
     processConsumerMutationsNearThreshold(false);
 }
 
+/* Checks that the DCP producer does an async stream close when the DCP client
+   expects "DCP_STREAM_END" msg. */
+TEST_P(STDcpTest, test_producer_stream_end_on_client_close_stream) {
+    setVBucketStateAndRunPersistTask(vbid, vbucket_state_active);
+
+    const void* cookie = create_mock_cookie(engine.get());
+    auto& mockConnMap = static_cast<MockDcpConnMap&>(engine->getDcpConnMap());
+
+    /* Create a new Dcp producer */
+    auto producer = std::make_shared<MockDcpProducer>(*engine,
+                                                      cookie,
+                                                      "test_producer",
+                                                      /*flags*/ 0);
+    mockConnMap.addConn(cookie, producer);
+    EXPECT_TRUE(mockConnMap.doesConnHandlerExist("test_producer"));
+
+    /* Send a control message to the producer indicating that the DCP client
+       expects a "DCP_STREAM_END" upon stream close */
+    const std::string sendStreamEndOnClientStreamCloseCtrlMsg(
+            "send_stream_end_on_client_close_stream");
+    const std::string sendStreamEndOnClientStreamCloseCtrlValue("true");
+    EXPECT_EQ(ENGINE_SUCCESS,
+              producer->control(0,
+                                sendStreamEndOnClientStreamCloseCtrlMsg,
+                                sendStreamEndOnClientStreamCloseCtrlValue));
+
+    /* Open stream */
+    EXPECT_EQ(ENGINE_SUCCESS, doStreamRequest(*producer).status);
+    EXPECT_TRUE(mockConnMap.doesVbConnExist(vbid, "test_producer"));
+
+    /* Close stream */
+    EXPECT_EQ(ENGINE_SUCCESS, producer->closeStream(0, vbid));
+
+    /* Expect a stream end message */
+    MockDcpMessageProducers producers(engine.get());
+    EXPECT_EQ(ENGINE_SUCCESS, producer->step(&producers));
+    EXPECT_EQ(cb::mcbp::ClientOpcode::DcpStreamEnd, producers.last_op);
+    EXPECT_EQ(cb::mcbp::DcpStreamEndStatus::Closed, producers.last_end_status);
+
+    /* Re-open stream for the same vbucket on the conn */
+    EXPECT_EQ(ENGINE_SUCCESS, doStreamRequest(*producer).status);
+
+    /* Check that the new stream is opened properly */
+    auto stream = producer->findStream(vbid);
+    auto* as = static_cast<ActiveStream*>(stream.get());
+    EXPECT_TRUE(as->isInMemory());
+
+    // MB-27769: Prior to the fix, this would fail here because we would skip
+    // adding the connhandler into the connmap vbConns vector, causing the
+    // stream to never get notified.
+    EXPECT_TRUE(mockConnMap.doesVbConnExist(vbid, "test_producer"));
+
+    mockConnMap.disconnect(cookie);
+    EXPECT_FALSE(mockConnMap.doesVbConnExist(vbid, "test_producer"));
+    mockConnMap.manageConnections();
+}
+
 INSTANTIATE_TEST_SUITE_P(PersistentAndEphemeral,
                          STDcpTest,
                          STParameterizedBucketTest::allConfigValues());

engines/ep/tests/module_tests/dcp_test.cc

@@ -1161,72 +1161,6 @@ TEST_P(ConnectionTest, test_mb17042_duplicate_cookie_producer_connections) {
         << "Dead connections still remain";
 }
 
-/* Checks that the DCP producer does an async stream close when the DCP client
-   expects "DCP_STREAM_END" msg. */
-TEST_P(ConnectionTest, test_producer_stream_end_on_client_close_stream) {
-#ifdef UNDEFINED_SANITIZER
-    // See below MB-28739 comment for why this is skipped.
-    std::cerr << "MB-28739[UBsan] skipping test\n";
-    return;
-#endif
-    const void* cookie = create_mock_cookie(engine);
-    /* Create a new Dcp producer */
-    producer = std::make_shared<MockDcpProducer>(*engine,
-                                                 cookie,
-                                                 "test_producer",
-                                                 /*flags*/ 0);
-
-    /* Send a control message to the producer indicating that the DCP client
-       expects a "DCP_STREAM_END" upon stream close */
-    const std::string sendStreamEndOnClientStreamCloseCtrlMsg(
-            "send_stream_end_on_client_close_stream");
-    const std::string sendStreamEndOnClientStreamCloseCtrlValue("true");
-    EXPECT_EQ(ENGINE_SUCCESS,
-              producer->control(0,
-                                sendStreamEndOnClientStreamCloseCtrlMsg,
-                                sendStreamEndOnClientStreamCloseCtrlValue));
-
-    /* Open stream */
-    EXPECT_EQ(ENGINE_SUCCESS, doStreamRequest(*producer).status);
-
-    // MB-28739[UBSan]: The following cast is undefined behaviour - the DCP
-    // connection map object is of type DcpConnMap; so it's undefined to cast
-    // to MockDcpConnMap.
-    // However, in this instance MockDcpConnMap has identical member variables
-    // to DcpConnMap - the mock just exposes normally private data - and so
-    // this /seems/ ok.
-    // As such allow it in general, but skip this test under UBSan.
-    auto& mockConnMap =
-            static_cast<MockDcpConnMap&>(engine->getDcpConnMap());
-    mockConnMap.addConn(cookie, producer);
-    EXPECT_TRUE(mockConnMap.doesVbConnExist(vbid, "test_producer"));
-
-    /* Close stream */
-    EXPECT_EQ(ENGINE_SUCCESS, producer->closeStream(0, vbid));
-
-    /* Expect a stream end message */
-    MockDcpMessageProducers producers(handle);
-    EXPECT_EQ(ENGINE_SUCCESS, producer->step(&producers));
-    EXPECT_EQ(cb::mcbp::ClientOpcode::DcpStreamEnd, producers.last_op);
-    EXPECT_EQ(cb::mcbp::DcpStreamEndStatus::Closed, producers.last_end_status);
-
-    /* Re-open stream for the same vbucket on the conn */
-    EXPECT_EQ(ENGINE_SUCCESS, doStreamRequest(*producer).status);
-
-    /* Check that the new stream is opened properly */
-    auto stream = producer->findStream(vbid);
-    auto* as = static_cast<ActiveStream*>(stream.get());
-    EXPECT_TRUE(as->isInMemory());
-
-    // MB-27769: Prior to the fix, this would fail here because we would skip
-    // adding the connhandler into the connmap vbConns vector, causing the
-    // stream to never get notified.
-    EXPECT_TRUE(mockConnMap.doesVbConnExist(vbid, "test_producer"));
-
-    mockConnMap.disconnect(cookie);
-    EXPECT_FALSE(mockConnMap.doesVbConnExist(vbid, "test_producer"));
-    mockConnMap.manageConnections();
-}
 
 /* Checks that the DCP producer does a synchronous stream close when the DCP
    client does not expect "DCP_STREAM_END" msg. */