Merge remote-tracking branch 'couchbase/morpheus'

* couchbase/morpheus:
  CBD-6348: Fix compile warnings
  CBD-6348: [BP] Replace sprintf with snprintf
  CBD-6348: [BP] Use std::string_view instead of gsl::cstring_span
  MB-68916: Fix exception throw at EPE::initilize
  MB-68916: Fix ep_testsuite_dcp
  MB-68916: Remove continuous_dcp_ctx boilerplate from ep_testsuite_dcp

Change-Id: Ibe7c0b6c05f0c9c487c22113ac466d10c76b8c67

Author: trondn

engines/ep/src/ep_engine.cc

@@ -1892,6 +1892,24 @@ cb::engine_errc EventuallyPersistentEngine::initialize(
         const nlohmann::json& encryption,
         std::string_view chronicleAuthToken,
         const nlohmann::json& collectionManifest) {
+    auto handle = acquireEngine(this);
+    try {
+        return handle->initializeInner(
+                config, encryption, chronicleAuthToken, collectionManifest);
+    } catch (const cb::engine_error& e) {
+        NonBucketAllocationGuard guard;
+        throw cb::engine_error(e.engine_code(), e.what());
+    } catch (const std::exception& e) {
+        NonBucketAllocationGuard guard;
+        throw std::runtime_error(e.what());
+    }
+}
+
+cb::engine_errc EventuallyPersistentEngine::initializeInner(
+        std::string_view config,
+        const nlohmann::json& encryption,
+        std::string_view chronicleAuthToken,
+        const nlohmann::json& collectionManifest) {
     if (config.empty()) {
         return cb::engine_errc::invalid_arguments;
     }

engines/ep/src/ep_engine.h

@@ -1201,6 +1201,11 @@ class EventuallyPersistentEngine : public EngineIface, public DcpIface {
 protected:
     friend class EpEngineValueChangeListener;
 
+    cb::engine_errc initializeInner(std::string_view,
+                                    const nlohmann::json&,
+                                    std::string_view,
+                                    const nlohmann::json&);
+
     cb::engine_errc compactDatabaseInner(
             CookieIface& cookie,
             Vbid vbid,

engines/ep/tests/ep_testsuite_dcp.cc

@@ -237,6 +237,8 @@ class TestDcpConsumer {
         stream_ctxs.push_back(ctx);
     }
 
+    void openConnectionAndStreams();
+
     void run(bool openConn = true);
 
     // Stop the thread if it is running. This is safe to be called from
@@ -381,23 +383,23 @@ void TestDcpConsumer::deleteOrExpireCase(TestDcpConsumer::VBStats& stats,
     }
 }
 
-void TestDcpConsumer::run(bool openConn) {
+void TestDcpConsumer::openConnectionAndStreams() {
     checkle(size_t{1}, stream_ctxs.size(), "No dcp_stream arguments provided!");
-
-    /* Open the connection with the DCP producer */
-    if (openConn) {
-        openConnection();
-    }
-
+    openConnection();
     if (collectionFilter) {
         // Enable noop ops needed for collections
         checkeq(cb::engine_errc::success,
                 dcp->control(
                         *cookie, opaque, DcpControlKeys::EnableNoop, "true"),
                 "Failed to enable noop");
     }
-    /* Open streams in the above open connection */
     openStreams();
+}
+
+void TestDcpConsumer::run(bool openConn) {
+    if (openConn) {
+        openConnectionAndStreams();
+    }
 
     size_t num_stream_ends_received = 0;
     uint32_t bytes_read = 0;
@@ -1132,15 +1134,6 @@ struct writer_thread_ctx {
     Vbid vbid;
 };
 
-struct continuous_dcp_ctx {
-    EngineIface* h;
-    const void *cookie;
-    Vbid vbid;
-    const std::string &name;
-    uint64_t start_seqno;
-    std::unique_ptr<TestDcpConsumer> dcpConsumer;
-};
-
 //Forward declaration required for dcp_thread_func
 static uint32_t add_stream_for_consumer(EngineIface* h,
                                         CookieIface* cookie,
@@ -1263,21 +1256,6 @@ static void writer_thread(void* args) {
     }
 }
 
-static void continuous_dcp_thread(void* args) {
-    auto* cdc = static_cast<continuous_dcp_ctx*>(args);
-
-    DcpStreamCtx ctx;
-    ctx.vbucket = cdc->vbid;
-    std::string vbuuid_entry("vb_" + std::to_string(cdc->vbid.get()) + ":0:id");
-    ctx.vb_uuid = get_ull_stat(cdc->h, vbuuid_entry, "failovers");
-    ctx.seqno = {cdc->start_seqno, std::numeric_limits<uint64_t>::max()};
-    ctx.snapshot = {cdc->start_seqno, cdc->start_seqno};
-    ctx.skip_verification = true;
-
-    cdc->dcpConsumer->addStreamCtx(ctx);
-    cdc->dcpConsumer->run();
-}
-
 /* DCP step thread that keeps running till it reads upto 'exp_mutations'.
    Note: the exp_mutations is cumulative across all streams in the DCP
          connection */
@@ -2183,7 +2161,7 @@ static test_result testDcpProducerExpiredItemBackfill(
     }
 
     tdc.addStreamCtx(ctx);
-
+    tdc.openStreams();
     tdc.run(false);
 
     testHarness->destroy_cookie(cookie);
@@ -2447,6 +2425,7 @@ static enum test_result test_dcp_producer_stream_req_coldness(EngineIface* h) {
     ctx.flags |= cb::mcbp::DcpAddStreamFlag::DiskOnly;
 
     tdc.addStreamCtx(ctx);
+    tdc.openStreams();
     tdc.run(false);
 
     checkeq(tdc.getNruCounters()[1],
@@ -2581,27 +2560,28 @@ static enum test_result test_dcp_producer_stream_latest(EngineIface* h) {
 }
 
 static enum test_result test_dcp_producer_keep_stream_open(EngineIface* h) {
-    const std::string conn_name("unittest");
-    const int num_items = 2, vb = 0;
+    const int num_items = 2;
 
     write_items(h, num_items);
 
     wait_for_flusher_to_settle(h);
     verify_curr_items(h, num_items, "Wrong amount of items");
 
+    // We want to stream items till end and keep the stream open. Then we want
+    // to verify the stream is still open
     auto* cookie = testHarness->create_cookie(h);
-
-    /* We want to stream items till end and keep the stream open. Then we want
-       to verify the stream is still open */
-    struct continuous_dcp_ctx cdc = {
-            h,
-            cookie,
-            Vbid(0),
-            conn_name,
-            0,
-            std::make_unique<TestDcpConsumer>(conn_name, cookie, h)};
-    auto dcp_thread = create_thread([&cdc]() { continuous_dcp_thread(&cdc); },
-                                    "dcp_thread");
+    const std::string conn_name = "test-consumer";
+    auto consumer = std::make_unique<TestDcpConsumer>(conn_name, cookie, h);
+    DcpStreamCtx ctx;
+    ctx.vbucket = Vbid(0);
+    ctx.vb_uuid = get_ull_stat(h, "vb_0:0:id", "failovers");
+    ctx.seqno = {0, std::numeric_limits<uint64_t>::max()};
+    ctx.snapshot = {0, 0};
+    ctx.skip_verification = true;
+    consumer->addStreamCtx(ctx);
+    consumer->openConnectionAndStreams();
+    auto dcp_thread =
+            create_thread([&]() { consumer->run(false); }, "dcp_thread");
 
     /* Wait for producer to be created */
     wait_for_stat_to_be(h, "ep_dcp_producer_count", 1, "dcp");
@@ -2613,19 +2593,17 @@ static enum test_result test_dcp_producer_keep_stream_open(EngineIface* h) {
 
     /* Wait for the dcp test client to receive upto highest seqno we have */
     cb::RelaxedAtomic<uint64_t> exp_items(num_items);
-    wait_for_val_to_be("last_sent_seqno",
-                       cdc.dcpConsumer->producers.last_byseqno,
-                       exp_items);
+    wait_for_val_to_be(
+            "last_sent_seqno", consumer->producers.last_byseqno, exp_items);
 
-    /* Check if the stream is still open after sending out latest items */
-    std::string stat_stream_state("eq_dcpq:" + conn_name + ":stream_" +
-                             std::to_string(vb) + "_state");
+    // Check if the stream is still open after sending out latest items
+    std::string stat_stream_state("eq_dcpq:" + conn_name + ":stream_0_state");
     std::string state = get_str_stat(h, stat_stream_state, "dcp");
     checkeq(state.compare("in-memory"), 0, "Stream is not open");
 
-    /* Before closing the connection stop the thread that continuously polls
-       for dcp data */
-    cdc.dcpConsumer->stop();
+    // Before closing the connection stop the thread that continuously polls for
+    // dcp data
+    consumer->stop();
     cookie->notifyIoComplete(cb::engine_errc::success);
     dcp_thread.join();
     testHarness->destroy_cookie(cookie);
@@ -2724,16 +2702,18 @@ static enum test_result test_dcp_producer_keep_stream_open_replica(
     /* We want to stream items till end and keep the stream open. Then we want
        to verify the stream is still open */
     auto* cookie1 = testHarness->create_cookie(h);
-    const std::string conn_name1("unittest1");
-    struct continuous_dcp_ctx cdc = {
-            h,
-            cookie1,
-            Vbid(0),
-            conn_name1,
-            0,
-            std::make_unique<TestDcpConsumer>(conn_name1, cookie1, h)};
-    auto dcp_thread = create_thread([&cdc]() { continuous_dcp_thread(&cdc); },
-                                    "dcp_thread");
+    const std::string conn_name1 = "test-consumer";
+    auto consumer = std::make_unique<TestDcpConsumer>(conn_name1, cookie1, h);
+    DcpStreamCtx ctx;
+    ctx.vbucket = Vbid(0);
+    ctx.vb_uuid = get_ull_stat(h, "vb_0:0:id", "failovers");
+    ctx.seqno = {0, std::numeric_limits<uint64_t>::max()};
+    ctx.snapshot = {0, 0};
+    ctx.skip_verification = true;
+    consumer->addStreamCtx(ctx);
+    consumer->openConnectionAndStreams();
+    auto dcp_thread =
+            create_thread([&]() { consumer->run(false); }, "dcp_thread");
 
     /* Wait for producer to be created */
     wait_for_stat_to_be(h, "ep_dcp_producer_count", 1, "dcp");
@@ -2745,9 +2725,8 @@ static enum test_result test_dcp_producer_keep_stream_open_replica(
 
     /* Wait for the dcp test client to receive upto highest seqno we have */
     cb::RelaxedAtomic<uint64_t> exp_items(3 * num_items);
-    wait_for_val_to_be("last_sent_seqno",
-                       cdc.dcpConsumer->producers.last_byseqno,
-                       exp_items);
+    wait_for_val_to_be(
+            "last_sent_seqno", consumer->producers.last_byseqno, exp_items);
 
     /* Check if correct snap end seqno is sent */
     std::string stat_stream_last_sent_snap_end_seqno("eq_dcpq:" + conn_name1 +
@@ -2765,7 +2744,7 @@ static enum test_result test_dcp_producer_keep_stream_open_replica(
 
     /* Before closing the connection stop the thread that continuously polls
        for dcp data */
-    cdc.dcpConsumer->stop();
+    consumer->stop();
     cookie1->notifyIoComplete(cb::engine_errc::success);
     dcp_thread.join();
 
@@ -2777,7 +2756,6 @@ static enum test_result test_dcp_producer_keep_stream_open_replica(
 
 static enum test_result test_dcp_producer_stream_cursor_movement(
         EngineIface* h) {
-    const std::string conn_name("unittest");
     const int num_items = 30;
     for (int j = 0; j < num_items; ++j) {
         if (j % 10 == 0) {
@@ -2795,19 +2773,21 @@ static enum test_result test_dcp_producer_stream_cursor_movement(
     wait_for_flusher_to_settle(h);
     verify_curr_items(h, num_items, "Wrong amount of items");
 
+    // We want to stream items till end and keep the stream open. We want to
+    // verify if the DCP cursor has moved to new open checkpoint
     auto* cookie = testHarness->create_cookie(h);
-
-    /* We want to stream items till end and keep the stream open. We want to
-       verify if the DCP cursor has moved to new open checkpoint */
-    struct continuous_dcp_ctx cdc = {
-            h,
-            cookie,
-            Vbid(0),
-            conn_name,
-            20,
-            std::make_unique<TestDcpConsumer>(conn_name, cookie, h)};
-    auto dcp_thread = create_thread([&cdc]() { continuous_dcp_thread(&cdc); },
-                                    "dcp_thread");
+    const std::string conn_name = "test-consumer";
+    auto consumer = std::make_unique<TestDcpConsumer>(conn_name, cookie, h);
+    DcpStreamCtx ctx;
+    ctx.vbucket = Vbid(0);
+    ctx.vb_uuid = get_ull_stat(h, "vb_0:0:id", "failovers");
+    ctx.seqno = {20, std::numeric_limits<uint64_t>::max()};
+    ctx.snapshot = {20, 20};
+    ctx.skip_verification = true;
+    consumer->addStreamCtx(ctx);
+    consumer->openConnectionAndStreams();
+    auto dcp_thread =
+            create_thread([&]() { consumer->run(false); }, "dcp_thread");
 
     /* Wait for producer to be created */
     wait_for_stat_to_be(h, "ep_dcp_producer_count", 1, "dcp");
@@ -2819,9 +2799,8 @@ static enum test_result test_dcp_producer_stream_cursor_movement(
 
     /* Wait for the dcp test client to receive upto highest seqno we have */
     cb::RelaxedAtomic<uint64_t> exp_items(num_items);
-    wait_for_val_to_be("last_sent_seqno",
-                       cdc.dcpConsumer->producers.last_byseqno,
-                       exp_items);
+    wait_for_val_to_be(
+            "last_sent_seqno", consumer->producers.last_byseqno, exp_items);
 
     /* We want to make sure that no cursors are lingering on any of the previous
        checkpoints. For that we wait for checkpoint remover to remove all but
@@ -2830,7 +2809,7 @@ static enum test_result test_dcp_producer_stream_cursor_movement(
 
     /* Before closing the connection stop the thread that continuously polls
        for dcp data */
-    cdc.dcpConsumer->stop();
+    consumer->stop();
     cookie->notifyIoComplete(cb::engine_errc::success);
     dcp_thread.join();
     testHarness->destroy_cookie(cookie);
@@ -5793,6 +5772,7 @@ static test_result test_dcp_replica_stream_expiries(
                 "Failed to enable_expiry_opcode");
     }
     tdc.addStreamCtx(ctx);
+    tdc.openStreams();
     tdc.run(false);
 
     testHarness->destroy_cookie(cookie1);
@@ -5878,7 +5858,7 @@ static test_result test_stream_deleteWithMeta_expiration(
     }
 
     tdc.addStreamCtx(ctx);
-
+    tdc.openStreams();
     tdc.run(false);
 
     testHarness->destroy_cookie(cookie);
@@ -7084,19 +7064,21 @@ static enum test_result test_dcp_multiple_streams(EngineIface* h) {
 }
 
 static enum test_result test_dcp_on_vbucket_state_change(EngineIface* h) {
-    const std::string conn_name = "unittest";
+    // Set up a DcpTestConsumer that would remain in in-memory mode
     auto* cookie = testHarness->create_cookie(h);
 
-    // Set up a DcpTestConsumer that would remain in in-memory mode
-    struct continuous_dcp_ctx cdc = {
-            h,
-            cookie,
-            Vbid(0),
-            conn_name,
-            0,
-            std::make_unique<TestDcpConsumer>(conn_name, cookie, h)};
-    auto dcp_thread = create_thread([&cdc]() { continuous_dcp_thread(&cdc); },
-                                    "dcp_thread");
+    const std::string conn_name = "test-consumer";
+    auto consumer = std::make_unique<TestDcpConsumer>(conn_name, cookie, h);
+    DcpStreamCtx ctx;
+    ctx.vbucket = Vbid(0);
+    ctx.vb_uuid = get_ull_stat(h, "vb_0:0:id", "failovers");
+    ctx.seqno = {0, std::numeric_limits<uint64_t>::max()};
+    ctx.snapshot = {0, 0};
+    ctx.skip_verification = true;
+    consumer->addStreamCtx(ctx);
+    consumer->openConnectionAndStreams();
+    auto dcp_thread =
+            create_thread([&]() { consumer->run(false); }, "dcp_thread");
 
     // Wait for producer to be created
     wait_for_stat_to_be(h, "ep_dcp_producer_count", 1, "dcp");
@@ -7120,7 +7102,7 @@ static enum test_result test_dcp_on_vbucket_state_change(EngineIface* h) {
     // Expect producers->last_end_status to carry StateChanged as reason
     // for stream closure
     check_expression(cb::mcbp::DcpStreamEndStatus::StateChanged ==
-                             cdc.dcpConsumer->producers.last_end_status,
+                             consumer->producers.last_end_status,
                      "Last DCP flag not StateChanged");
 
     testHarness->destroy_cookie(cookie);
@@ -7933,6 +7915,7 @@ static enum test_result testDcpOsoBackfill(EngineIface* h) {
             "Failed control enable_out_of_order_snapshots");
 
     tdc.addStreamCtx(ctx);
+    tdc.openStreams();
     tdc.run(false);
 
     testHarness->destroy_cookie(cookie);