MB-66479: Run cluster_test with magma backend and fix hang

The snapshot test for instance will use a different code path
to collect the files in use which revealed a bug in GetFileFragment
causing it to hang forever when trying to fetch multiple files
as the state machinery didn't clear the EWB state from "too much
data"

Change-Id: Ib8fe47d9a3f416721f170f6018a125be1d701bb2
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/226984
Reviewed-by: Vesko Karaganev <vesko.karaganev@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

Author: trondn

cluster_framework/cluster.cc

@@ -24,6 +24,20 @@
 
 namespace cb::test {
 
+std::string format_as(BucketPersistenceBackend backend) {
+    switch (backend) {
+    case BucketPersistenceBackend::Couchstore:
+        return "couchdb";
+    case BucketPersistenceBackend::Magma:
+        return "magma";
+    }
+    Expects(false && "Unexpected backend");
+}
+
+void to_json(nlohmann::json& j, const BucketPersistenceBackend backend) {
+    j = format_as(backend);
+}
+
 Cluster::~Cluster() = default;
 
 class ClusterImpl : public Cluster {
@@ -33,6 +47,13 @@ class ClusterImpl : public Cluster {
     ClusterImpl(std::vector<std::unique_ptr<Node>>& nod,
                 std::filesystem::path dir);
     ~ClusterImpl() override;
+    void setBucketPersistenceBackend(
+            BucketPersistenceBackend backend_) override {
+        backend = backend_;
+    }
+    BucketPersistenceBackend getBucketPersistenceBackend() const override {
+        return backend;
+    }
     [[nodiscard]] std::shared_ptr<Bucket> createBucket(
             const std::string& name,
             const nlohmann::json& attributes,
@@ -76,6 +97,7 @@ class ClusterImpl : public Cluster {
     const std::string uuid;
     std::atomic<int64_t> revno{1};
     static constexpr int64_t epoch = 1;
+    BucketPersistenceBackend backend = BucketPersistenceBackend::Couchstore;
 };
 
 ClusterImpl::ClusterImpl(std::vector<std::unique_ptr<Node>>& nod,
@@ -239,7 +261,7 @@ std::shared_ptr<Bucket> ClusterImpl::createBucket(
     size_t replicas = std::min<size_t>(size() - 1, 3);
 
     nlohmann::json json = {{"max_size", 67108864},
-                           {"backend", "couchdb"},
+                           {"backend", backend},
                            {"couch_bucket", name},
                            {"max_vbuckets", vbuckets},
                            {"data_traffic_enabled", false},

cluster_framework/cluster.h

@@ -23,6 +23,10 @@ class MemcachedConnection;
 
 namespace cb::test {
 
+enum class BucketPersistenceBackend { Couchstore, Magma };
+std::string format_as(BucketPersistenceBackend backend);
+void to_json(nlohmann::json& j, const BucketPersistenceBackend backend);
+
 class Node;
 class Bucket;
 class AuthProviderService;
@@ -37,6 +41,11 @@ class Cluster {
 public:
     virtual ~Cluster();
 
+    /// Set the backend to use for buckets
+    virtual void setBucketPersistenceBackend(
+            BucketPersistenceBackend backend) = 0;
+    virtual BucketPersistenceBackend getBucketPersistenceBackend() const = 0;
+
     /**
      * Create a Couchbase bucket
      *

daemon/protocol/mcbp/get_file_fragment_context.cc

@@ -83,6 +83,7 @@ cb::engine_errc GetFileFragmentContext::step() {
             ret = transfer_with_sendfile();
             break;
         case State::Done:
+            cookie.clearEwouldblock();
             return cb::engine_errc::success;
         }
     }
@@ -272,6 +273,11 @@ cb::engine_errc GetFileFragmentContext::chain_file_chunk() {
                           iob->length()};
     connection.chainDataToOutputStream(
             std::make_unique<IOBufSendBuffer>(std::move(iob), view));
-    state = State::ReadFileChunk;
+    if (length) {
+        // We need to read another chunk
+        state = State::ReadFileChunk;
+    } else {
+        state = State::Done;
+    }
     return cb::engine_errc::too_much_data_in_output_buffer;
 }

tests/testapp_cluster/CMakeLists.txt

@@ -28,3 +28,15 @@ add_test(NAME cluster_test
 # Don't run it concurrently with any other tests.
 set_tests_properties(cluster_test
         PROPERTIES RUN_SERIAL TRUE)
+
+if (BUILD_ENTERPRISE AND NOT WIN32)
+    # Magma emits a lot of critical warnings and fails on
+    # win32 due to failing to open files.
+    # Only run on other platforms for now
+    add_test(NAME cluster_test_magma
+        WORKING_DIRECTORY ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}
+        COMMAND cluster_test --backend=magma)
+    # The cluster_test spins up 4 memcached instances and is relatively heavyweight.
+    # Don't run it concurrently with any other tests.
+    set_tests_properties(cluster_test_magma PROPERTIES RUN_SERIAL TRUE)
+endif()

tests/testapp_cluster/clustertest.cc

@@ -26,7 +26,7 @@ std::ostream& cb::test::operator<<(std::ostream& os,
               << " MiB, upper: " << stats.upper / 1024.0 / 1024.0 << " MiB}";
 }
 
-void cb::test::ClusterTest::StartCluster() {
+void cb::test::ClusterTest::StartCluster(BucketPersistenceBackend backend) {
     MemcachedConnection::setLookupUserPasswordFunction(
             [](const std::string& user) -> std::string {
                 if (user == "@admin") {
@@ -45,6 +45,7 @@ void cb::test::ClusterTest::StartCluster() {
             });
 
     cluster = Cluster::create(4);
+    cluster->setBucketPersistenceBackend(backend);
     if (!cluster) {
         std::cerr << "Failed to create the cluster" << std::endl;
         std::exit(EXIT_FAILURE);

tests/testapp_cluster/clustertest.h

@@ -20,6 +20,7 @@ namespace cb::test {
 
 class Bucket;
 class Cluster;
+enum class BucketPersistenceBackend;
 
 struct MemStats {
     size_t current{0};
@@ -52,7 +53,7 @@ class ClusterTest : public ::testing::Test {
     static void TearDownTestCase();
 
     /// Start the cluster with 4 nodes.
-    static void StartCluster();
+    static void StartCluster(BucketPersistenceBackend backend);
 
     /// Shutdown the cluster
     static void ShutdownCluster();

tests/testapp_cluster/main.cc

@@ -8,17 +8,21 @@
  *   the file licenses/APL2.txt.
  */
 
+#include "cluster_framework/cluster.h"
 #include "clustertest.h"
-
 #include <event2/thread.h>
 #include <platform/cbassert.h>
+#include <platform/command_line_options_parser.h>
 #include <platform/dirutils.h>
 #include <platform/platform_socket.h>
 #include <array>
 #include <csignal>
 #include <cstdlib>
 #include <string>
 
+using cb::test::BucketPersistenceBackend;
+using namespace std::string_view_literals;
+
 int main(int argc, char** argv) {
     setupWindowsDebugCRTAssertHandling();
     cb::net::initialize();
@@ -61,8 +65,38 @@ int main(int argc, char** argv) {
     }
 #endif
     ::testing::InitGoogleTest(&argc, argv);
+    auto backend = BucketPersistenceBackend::Couchstore;
+
+    cb::getopt::CommandLineOptionsParser parser;
+    using cb::getopt::Argument;
+    parser.addOption(
+            {[&backend](auto value) {
+                 if (value == "magma"sv) {
+                     backend = BucketPersistenceBackend::Magma;
+                 } else if (value == "couchstore"sv) {
+                     backend = BucketPersistenceBackend::Couchstore;
+                 } else {
+                     std::cerr << "backend must be 'couchstore' or 'magma'"
+                               << std::endl;
+                     exit(EXIT_FAILURE);
+                 }
+             },
+             "backend",
+             Argument::Required,
+             "magma",
+             "The name of the backend"});
+
+    parser.addOption({[&parser](auto) {
+                          std::cout << std::endl
+                                    << std::endl
+                                    << parser << std::endl;
+                      },
+                      "help",
+                      "This help text"});
+
+    parser.parse(argc, argv, [&parser]() { std::exit(EXIT_FAILURE); });
 
-    cb::test::ClusterTest::StartCluster();
+    cb::test::ClusterTest::StartCluster(backend);
     const auto ret = RUN_ALL_TESTS();
     cb::test::ClusterTest::ShutdownCluster();
 

tests/testapp_cluster/misc_tests.cc

@@ -458,14 +458,10 @@ TEST_F(BasicClusterTest, AllStatGroups) {
             // These require a vbucket argument.
             key += " 0";
             break;
-        case StatGroupId::Fusion: {
-            if (conn->statsMap("")["ep_backend"] != "magma") {
-                return;
-            }
-            // These require a existing subcmd and a vbucket argument.
-            key += " sync_info 0";
-            break;
-        }
+        case StatGroupId::Fusion:
+            // Fusion stats isn't implemented currently in the cluster
+            // testsuite
+            return;
         default:
             break;
         }