MB-68823: Don't throw EPEngine exception to core in StreamReq path

The exception object would be allocated in the EPE domain and then
wrongly released in the NoBucketDomain otherwise.

All frontend code paths into EPEngine suffer by that potentially (if
any path throws). A wider refactor for addressing that is in progress
on the master branch.

Note that the extra logging introduced in this patch replaces the
connection-details logging (provided at core level) that we lose by
catching the exception in EPE.

Change-Id: Ifacb6fb9970575c2177bbd3d26d85df226dee3e3
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/234445
Well-Formed: Restriction Checker
Reviewed-by: Trond Norbye <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: paolococchi

daemon/protocol/mcbp/engine_wrapper.cc

@@ -903,9 +903,10 @@ cb::engine_errc dcpStreamReq(Cookie& cookie,
                                json);
     if (ret == cb::engine_errc::disconnect) {
         LOG_WARNING(
-                "{}: {} dcp.stream_req returned cb::engine_errc::disconnect",
+                "{}: dcp.stream_req returned cb::engine_errc::disconnect - "
+                "Connection {}",
                 connection.getId(),
-                connection.getDescription());
+                connection.to_json().dump());
         connection.setTerminationReason("Engine forced disconnect");
     }
     return ret;

engines/ep/src/ep_engine.cc

@@ -1341,6 +1341,10 @@ cb::engine_errc EventuallyPersistentEngine::stream_req(
     } catch (const cb::engine_error& e) {
         EP_LOG_INFO("stream_req engine_error {}", e.what());
         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

@@ -420,3 +420,72 @@ TEST_P(BucketTest, MB58598_Connection_ID_clobbered) {
         FAIL() << "Timed out waiting for log messages to appear";
     }
 }
+
+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),
+                              0, // flags
+                              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

@@ -633,6 +633,17 @@ class McdEnvironmentImpl : public McdEnvironment {
         }
     }
 
+    size_t verifyLogLine(std::string_view string) const override {
+        size_t ret = 0;
+        mcd_env->iterateLogLines([&string, &ret](auto line) {
+            if (line.find(string) != std::string_view::npos) {
+                ++ret;
+            }
+            return true;
+        });
+        return ret;
+    }
+
 private:
     const std::filesystem::path test_directory;
     const std::filesystem::path isasl_file_name;

tests/testapp/testapp_environment.h

@@ -235,6 +235,12 @@ class McdEnvironment {
             const std::function<bool(std::string_view line)>& callback)
             const = 0;
 
+    /**
+     * @param string to search
+     * @return The number of instances of string in the memcached log file
+     */
+    virtual size_t verifyLogLine(std::string_view string) const = 0;
+
     virtual std::string getLogFilePattern() const = 0;
 
     /// Do we have support for IPv4 addresses on the machine