MB-68823: Merge trinity/9a76d5e99 into morpheus

* 9a76d5e MB-68823: Revert log to pre review.couchbase.org/c/kv_engine/+/234445
* 6d16ff8 MB-68823: Don't throw EPEngine exception to core in StreamReq path

Change-Id: I994d7defc72791a24d0c5f98f74b865dcda16dea

Author: paolococchi

daemon/protocol/mcbp/engine_wrapper.cc

@@ -845,9 +845,11 @@ cb::engine_errc dcpStreamReq(Cookie& cookie,
                                std::move(callback),
                                json);
     if (ret == cb::engine_errc::disconnect) {
-        LOG_WARNING_CTX("dcp.stream_req returned cb::engine_errc::disconnect",
-                        {"conn_id", connection.getId()},
-                        {"description", connection.getDescription()});
+        LOG_WARNING(
+                "{}: dcp.stream_req returned cb::engine_errc::disconnect - "
+                "Cookie {}",
+                connection.getId(),
+                cookie.to_json().dump());
         connection.setTerminationReason("Engine forced disconnect");
     }
     return ret;

engines/ep/src/ep_engine.cc

@@ -1424,6 +1424,10 @@ cb::engine_errc EventuallyPersistentEngine::stream_req(
                         {"error", e.what()},
                         {"status", e.engine_code()});
         return cb::engine_errc(e.code().value());
+    } catch (const std::exception& e) {
+        EP_LOG_ERR("EventuallyPersistentEngine::stream_req: Exception {}",
+                   e.what());
+        return cb::engine_errc::disconnect;
     }
 }
 

tests/testapp/testapp_bucket.cc

@@ -359,3 +359,72 @@ TEST_P(BucketTest, DeleteSelectedBucket) {
     adminConnection->selectBucket("bucket");
     deleteBucket(*adminConnection, "bucket", [](const std::string&) {});
 }
+
+class MemTrackingBucketTest : public BucketTest {
+public:
+    static void SetUpTestCase() {
+        // Note: Important to set the env BEFORE starting memcached,
+        // env vars wouldn't be passed to the memcached process otherwise
+        // (unless testapp is run in -e (embedded) mode)
+        ASSERT_FALSE(getenv("CB_ARENA_MALLOC_VERIFY_DEALLOC_CLIENT"));
+        ASSERT_EQ(0, setenv("CB_ARENA_MALLOC_VERIFY_DEALLOC_CLIENT", "1", 0));
+        BucketTest::SetUpTestCase();
+    }
+
+    static void TearDownTestCase() {
+        EXPECT_EQ(0, unsetenv("CB_ARENA_MALLOC_VERIFY_DEALLOC_CLIENT"));
+        BucketTest::TearDownTestCase();
+    }
+};
+
+INSTANTIATE_TEST_SUITE_P(TransportProtocols,
+                         MemTrackingBucketTest,
+                         ::testing::Values(TransportProtocols::McbpSsl),
+                         ::testing::PrintToStringParamName());
+
+TEST_P(MemTrackingBucketTest, MB_68823) {
+    // Note: Not using adminConnection as the connection in the test is
+    // forcibly disconnected and we need adminConnection at TearDown.
+    auto& conn = getConnection();
+    conn.authenticate("@admin");
+    conn.selectBucket(bucketName);
+    conn.setFeature(cb::mcbp::Feature::JSON, true);
+    conn.setFeature(cb::mcbp::Feature::Collections, true);
+    conn.dcpOpenProducer("dcp-conn_invalid-stream-req-filter");
+    conn.dcpControl("enable_noop", "true");
+
+    // Invalid StreamReq filter (with cid duplicate) throws in Filter::ctor.
+    // Before the fix the test fails by:
+    //
+    // ===ERROR===: JeArenaMalloc deallocation mismatch
+    //     Memory freed by client:100 domain:None which is assigned arena:0,
+    //     but memory was previously allocated from arena:2 (client-specific
+    //     arena).
+    //     Allocation address:0x10b1b1080 size:192
+    try {
+        conn.dcpStreamRequest(Vbid(0),
+                              cb::mcbp::DcpAddStreamFlag::None,
+                              0, // startSeq
+                              ~0, // endSeq,
+                              0, // vbUuid
+                              0, // snapStart
+                              0, // snapEnd
+                              R"({"collections":["0", "0"]})"_json); // filter
+    } catch (const std::exception&) {
+        const auto timeout =
+                std::chrono::steady_clock::now() + std::chrono::seconds{10};
+        const auto line =
+                "EventuallyPersistentEngine::stream_req: Exception GSL: "
+                "Precondition failure: 'emplaced'";
+        const auto expectedLogInstances = 1;
+        do {
+            if (mcd_env->verifyLogLine(line) == expectedLogInstances) {
+                return;
+            }
+            std::this_thread::sleep_for(std::chrono::milliseconds{100});
+        } while (std::chrono::steady_clock::now() < timeout);
+
+        FAIL() << "Timeout before the log line was dumped to the file";
+    }
+    FAIL() << "StreamRequest should have failed";
+}

tests/testapp/testapp_environment.cc

@@ -421,6 +421,17 @@ bool McdEnvironment::iterateAuditEvents(
     return false;
 }
 
+size_t McdEnvironment::verifyLogLine(std::string_view string) const {
+    size_t ret = 0;
+    mcd_env->iterateLogLines([&string, &ret](auto line) {
+        if (line.find(string) != std::string_view::npos) {
+            ++ret;
+        }
+        return true;
+    });
+    return ret;
+}
+
 std::string McdEnvironment::readConcurrentUpdatedFile(
         const cb::dek::Entity entity, const std::filesystem::path& path) const {
     std::string content;

tests/testapp/testapp_environment.h

@@ -275,6 +275,12 @@ class McdEnvironment {
     bool iterateAuditEvents(
             const std::function<bool(const nlohmann::json&)>& callback) const;
 
+    /**
+     * @param string to search
+     * @return The number of instances of string in the memcached log file
+     */
+    size_t verifyLogLine(std::string_view string) const;
+
     /// Do we have support for IPv4 addresses on the machine
     [[nodiscard]] bool haveIPv4() const {
         return !ipaddresses.first.empty();