remove graphql

- Remove the compile time control
- Introduce the runtime control flag

New flag `--experimental-enabled=system-replication`

.github/workflows/diff.yaml

@@ -383,70 +383,6 @@ jobs:
             # multiple paths could be defined
             build/logs
 
-  experimental_build_mt:
-    name: "MultiTenancy replication build"
-    runs-on: [self-hosted, Linux, X64, Diff]
-    env:
-      THREADS: 24
-      MEMGRAPH_ENTERPRISE_LICENSE: ${{ secrets.MEMGRAPH_ENTERPRISE_LICENSE }}
-      MEMGRAPH_ORGANIZATION_NAME: ${{ secrets.MEMGRAPH_ORGANIZATION_NAME }}
-
-    steps:
-      - name: Set up repository
-        uses: actions/checkout@v4
-        with:
-          # Number of commits to fetch. `0` indicates all history for all
-          # branches and tags. (default: 1)
-          fetch-depth: 0
-
-
-      - name: Build release binaries
-        run: |
-          # Activate toolchain.
-          source /opt/toolchain-v4/activate
-
-          # Initialize dependencies.
-          ./init
-
-          # Build MT replication experimental binaries.
-          cd build
-          cmake -DCMAKE_BUILD_TYPE=Release -D MG_EXPERIMENTAL_REPLICATION_MULTITENANCY=ON ..
-          make -j$THREADS
-
-      - name: Run unit tests
-        run: |
-          # Activate toolchain.
-          source /opt/toolchain-v4/activate
-
-          # Run unit tests.
-          cd build
-          ctest -R memgraph__unit --output-on-failure -j$THREADS
-
-      - name: Run e2e tests
-        run: |
-          cd tests
-          ./setup.sh /opt/toolchain-v4/activate
-          source ve3/bin/activate_e2e
-          cd e2e
-
-          # Just the replication based e2e tests
-          ./run.sh "Replicate multitenancy"
-          ./run.sh "Show"
-          ./run.sh "Show while creating invalid state"
-          ./run.sh "Delete edge replication"
-          ./run.sh "Read-write benchmark"
-          ./run.sh "Index replication"
-          ./run.sh "Constraints"
-
-      - name: Save test data
-        uses: actions/upload-artifact@v4
-        if: always()
-        with:
-          name: "Test data(MultiTenancy replication build)"
-          path: |
-            # multiple paths could be defined
-            build/logs
-
   release_jepsen_test:
     name: "Release Jepsen Test"
     runs-on: [self-hosted, Linux, X64, Debian10, JepsenControl]

.github/workflows/release_debian10.yaml

@@ -1,4 +1,7 @@
 name: Release Debian 10
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref_name }}
+  cancel-in-progress: true
 
 on:
   workflow_dispatch:
@@ -10,7 +13,12 @@ on:
         options:
           - Release
           - RelWithDebInfo
-
+  push:
+    branches:
+      - "release/**"
+    tags:
+      - "v*.*.*-rc*"
+      - "v*.*-rc*"
   schedule:
     - cron: "0 22 * * *"
 

.github/workflows/release_ubuntu2004.yaml

@@ -1,4 +1,7 @@
 name: Release Ubuntu 20.04
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref_name }}
+  cancel-in-progress: true
 
 on:
   workflow_dispatch:
@@ -10,7 +13,12 @@ on:
         options:
           - Release
           - RelWithDebInfo
-
+  push:
+    branches:
+      - "release/**"
+    tags:
+      - "v*.*.*-rc*"
+      - "v*.*-rc*"
   schedule:
     - cron: "0 22 * * *"
 

.github/workflows/stress_test_large.yaml

@@ -1,4 +1,7 @@
 name: Stress test large
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref_name }}
+  cancel-in-progress: true
 
 on:
   workflow_dispatch:
@@ -10,7 +13,10 @@ on:
         options:
           - Release
           - RelWithDebInfo
-
+  push:
+    tags:
+      - "v*.*.*-rc*"
+      - "v*.*-rc*"
   schedule:
     - cron: "0 22 * * *"
 

CMakeLists.txt

@@ -291,16 +291,6 @@ option(TSAN "Build with Thread Sanitizer. To get a reasonable performance option
 option(UBSAN "Build with Undefined Behaviour Sanitizer" OFF)
 
 # Build feature flags
-option(MG_EXPERIMENTAL_REPLICATION_MULTITENANCY "Feature flag for experimental replicaition of multitenacy" OFF)
-
-if (NOT MG_ENTERPRISE AND MG_EXPERIMENTAL_REPLICATION_MULTITENANCY)
-    set(MG_EXPERIMENTAL_REPLICATION_MULTITENANCY OFF)
-    message(FATAL_ERROR "MG_EXPERIMENTAL_REPLICATION_MULTITENANCY with community edition build isn't possible")
-endif ()
-
-if (MG_EXPERIMENTAL_REPLICATION_MULTITENANCY)
-    add_compile_definitions(MG_EXPERIMENTAL_REPLICATION_MULTITENANCY)
-endif ()
 
 if (TEST_COVERAGE)
   string(TOLOWER ${CMAKE_BUILD_TYPE} lower_build_type)

src/coordination/CMakeLists.txt

@@ -10,7 +10,6 @@ target_sources(mg-coordination
         include/coordination/coordinator_exceptions.hpp
         include/coordination/coordinator_slk.hpp
         include/coordination/coordinator_instance.hpp
-        include/coordination/coordinator_cluster_config.hpp
         include/coordination/coordinator_handlers.hpp
         include/coordination/constants.hpp
         include/coordination/instance_status.hpp

src/coordination/coordinator_client.cpp

@@ -41,16 +41,21 @@ CoordinatorClient::CoordinatorClient(CoordinatorInstance *coord_instance, Coordi
 auto CoordinatorClient::InstanceName() const -> std::string { return config_.instance_name; }
 auto CoordinatorClient::SocketAddress() const -> std::string { return rpc_client_.Endpoint().SocketAddress(); }
 
+auto CoordinatorClient::InstanceDownTimeoutSec() const -> std::chrono::seconds {
+  return config_.instance_down_timeout_sec;
+}
+
 void CoordinatorClient::StartFrequentCheck() {
   if (instance_checker_.IsRunning()) {
     return;
   }
 
-  MG_ASSERT(config_.health_check_frequency_sec > std::chrono::seconds(0),
+  MG_ASSERT(config_.instance_health_check_frequency_sec > std::chrono::seconds(0),
             "Health check frequency must be greater than 0");
 
   instance_checker_.Run(
-      config_.instance_name, config_.health_check_frequency_sec, [this, instance_name = config_.instance_name] {
+      config_.instance_name, config_.instance_health_check_frequency_sec,
+      [this, instance_name = config_.instance_name] {
         try {
           spdlog::trace("Sending frequent heartbeat to machine {} on {}", instance_name,
                         rpc_client_.Endpoint().SocketAddress());
@@ -121,5 +126,33 @@ auto CoordinatorClient::SendSwapMainUUIDRpc(const utils::UUID &uuid) const -> bo
   return false;
 }
 
+auto CoordinatorClient::SendUnregisterReplicaRpc(std::string const &instance_name) const -> bool {
+  try {
+    auto stream{rpc_client_.Stream<UnregisterReplicaRpc>(instance_name)};
+    if (!stream.AwaitResponse().success) {
+      spdlog::error("Failed to receive successful RPC response for unregistering replica!");
+      return false;
+    }
+    return true;
+  } catch (rpc::RpcFailedException const &) {
+    spdlog::error("Failed to unregister replica!");
+  }
+  return false;
+}
+
+auto CoordinatorClient::SendEnableWritingOnMainRpc() const -> bool {
+  try {
+    auto stream{rpc_client_.Stream<EnableWritingOnMainRpc>()};
+    if (!stream.AwaitResponse().success) {
+      spdlog::error("Failed to receive successful RPC response for enabling writing on main!");
+      return false;
+    }
+    return true;
+  } catch (rpc::RpcFailedException const &) {
+    spdlog::error("Failed to enable writing on main!");
+  }
+  return false;
+}
+
 }  // namespace memgraph::coordination
 #endif

src/coordination/coordinator_handlers.cpp

@@ -39,6 +39,18 @@ void CoordinatorHandlers::Register(memgraph::coordination::CoordinatorServer &se
         spdlog::info("Received SwapMainUUIDRPC on coordinator server");
         CoordinatorHandlers::SwapMainUUIDHandler(replication_handler, req_reader, res_builder);
       });
+
+  server.Register<coordination::UnregisterReplicaRpc>(
+      [&replication_handler](slk::Reader *req_reader, slk::Builder *res_builder) -> void {
+        spdlog::info("Received UnregisterReplicaRpc on coordinator server");
+        CoordinatorHandlers::UnregisterReplicaHandler(replication_handler, req_reader, res_builder);
+      });
+
+  server.Register<coordination::EnableWritingOnMainRpc>(
+      [&replication_handler](slk::Reader *req_reader, slk::Builder *res_builder) -> void {
+        spdlog::info("Received EnableWritingOnMainRpc on coordinator server");
+        CoordinatorHandlers::EnableWritingOnMainHandler(replication_handler, req_reader, res_builder);
+      });
 }
 
 void CoordinatorHandlers::SwapMainUUIDHandler(replication::ReplicationHandler &replication_handler,
@@ -142,9 +154,58 @@ void CoordinatorHandlers::PromoteReplicaToMainHandler(replication::ReplicationHa
       }
     }
   }
-  spdlog::error(fmt::format("FICO : Promote replica to main was success {}", std::string(req.main_uuid_)));
+  spdlog::info("Promote replica to main was success {}", std::string(req.main_uuid_));
   slk::Save(coordination::PromoteReplicaToMainRes{true}, res_builder);
 }
 
+void CoordinatorHandlers::UnregisterReplicaHandler(replication::ReplicationHandler &replication_handler,
+                                                   slk::Reader *req_reader, slk::Builder *res_builder) {
+  if (!replication_handler.IsMain()) {
+    spdlog::error("Unregistering replica must be performed on main.");
+    slk::Save(coordination::UnregisterReplicaRes{false}, res_builder);
+    return;
+  }
+
+  coordination::UnregisterReplicaReq req;
+  slk::Load(&req, req_reader);
+
+  auto res = replication_handler.UnregisterReplica(req.instance_name);
+  switch (res) {
+    using enum memgraph::query::UnregisterReplicaResult;
+    case SUCCESS:
+      slk::Save(coordination::UnregisterReplicaRes{true}, res_builder);
+      break;
+    case NOT_MAIN:
+      spdlog::error("Unregistering replica must be performed on main.");
+      slk::Save(coordination::UnregisterReplicaRes{false}, res_builder);
+      break;
+    case CAN_NOT_UNREGISTER:
+      spdlog::error("Could not unregister replica.");
+      slk::Save(coordination::UnregisterReplicaRes{false}, res_builder);
+      break;
+    case COULD_NOT_BE_PERSISTED:
+      spdlog::error("Could not persist replica unregistration.");
+      slk::Save(coordination::UnregisterReplicaRes{false}, res_builder);
+      break;
+  }
+}
+
+void CoordinatorHandlers::EnableWritingOnMainHandler(replication::ReplicationHandler &replication_handler,
+                                                     slk::Reader * /*req_reader*/, slk::Builder *res_builder) {
+  if (!replication_handler.IsMain()) {
+    spdlog::error("Enable writing on main must be performed on main!");
+    slk::Save(coordination::EnableWritingOnMainRes{false}, res_builder);
+    return;
+  }
+
+  if (!replication_handler.GetReplState().EnableWritingOnMain()) {
+    spdlog::error("Enabling writing on main failed!");
+    slk::Save(coordination::EnableWritingOnMainRes{false}, res_builder);
+    return;
+  }
+
+  slk::Save(coordination::EnableWritingOnMainRes{true}, res_builder);
+}
+
 }  // namespace memgraph::dbms
 #endif

src/coordination/coordinator_instance.cpp

@@ -17,6 +17,7 @@
 #include "nuraft/coordinator_state_machine.hpp"
 #include "nuraft/coordinator_state_manager.hpp"
 #include "utils/counter.hpp"
+#include "utils/functional.hpp"
 
 #include <range/v3/view.hpp>
 #include <shared_mutex>
@@ -87,6 +88,11 @@ CoordinatorInstance::CoordinatorInstance()
 
     auto const curr_main_uuid = self->GetMainUUID();
     if (curr_main_uuid == repl_instance_uuid.value()) {
+      if (!repl_instance.EnableWritingOnMain()) {
+        spdlog::error("Failed to enable writing on main instance {}", repl_instance_name);
+        return;
+      }
+
       repl_instance.OnSuccessPing();
       return;
     }
@@ -122,17 +128,9 @@ CoordinatorInstance::CoordinatorInstance()
   };
 }
 
-auto CoordinatorInstance::ClusterHasAliveMain_() const -> bool {
-  auto const alive_main = [](ReplicationInstance const &instance) { return instance.IsMain() && instance.IsAlive(); };
-  return std::ranges::any_of(repl_instances_, alive_main);
-}
-
 auto CoordinatorInstance::ShowInstances() const -> std::vector<InstanceStatus> {
   auto const coord_instances = raft_state_.GetAllCoordinators();
 
-  std::vector<InstanceStatus> instances_status;
-  instances_status.reserve(repl_instances_.size() + coord_instances.size());
-
   auto const stringify_repl_role = [](ReplicationInstance const &instance) -> std::string {
     if (!instance.IsAlive()) return "unknown";
     if (instance.IsMain()) return "main";
@@ -154,8 +152,7 @@ auto CoordinatorInstance::ShowInstances() const -> std::vector<InstanceStatus> {
                                 // CoordinatorState to every instance, we can be smarter about this using our RPC.
   };
 
-  std::ranges::transform(coord_instances, std::back_inserter(instances_status), coord_instance_to_status);
-
+  auto instances_status = utils::fmap(coord_instance_to_status, coord_instances);
   {
     auto lock = std::shared_lock{coord_instance_lock_};
     std::ranges::transform(repl_instances_, std::back_inserter(instances_status), repl_instance_to_status);
@@ -194,6 +191,7 @@ auto CoordinatorInstance::TryFailover() -> void {
     }
   }
 
+  // TODO: (andi) fmap compliant
   ReplicationClientsInfo repl_clients_info;
   repl_clients_info.reserve(repl_instances_.size() - 1);
   std::ranges::transform(repl_instances_ | ranges::views::filter(is_not_new_main),
@@ -308,6 +306,35 @@ auto CoordinatorInstance::RegisterReplicationInstance(CoordinatorClientConfig co
   return RegisterInstanceCoordinatorStatus::SUCCESS;
 }
 
+auto CoordinatorInstance::UnregisterReplicationInstance(std::string instance_name)
+    -> UnregisterInstanceCoordinatorStatus {
+  auto lock = std::lock_guard{coord_instance_lock_};
+
+  auto const name_matches = [&instance_name](ReplicationInstance const &instance) {
+    return instance.InstanceName() == instance_name;
+  };
+
+  auto inst_to_remove = std::ranges::find_if(repl_instances_, name_matches);
+  if (inst_to_remove == repl_instances_.end()) {
+    return UnregisterInstanceCoordinatorStatus::NO_INSTANCE_WITH_NAME;
+  }
+
+  if (inst_to_remove->IsMain() && inst_to_remove->IsAlive()) {
+    return UnregisterInstanceCoordinatorStatus::IS_MAIN;
+  }
+
+  inst_to_remove->StopFrequentCheck();
+  auto curr_main = std::ranges::find_if(repl_instances_, &ReplicationInstance::IsMain);
+  MG_ASSERT(curr_main != repl_instances_.end(), "There must be a main instance when unregistering a replica");
+  if (!curr_main->SendUnregisterReplicaRpc(instance_name)) {
+    inst_to_remove->StartFrequentCheck();
+    return UnregisterInstanceCoordinatorStatus::RPC_FAILED;
+  }
+  std::erase_if(repl_instances_, name_matches);
+
+  return UnregisterInstanceCoordinatorStatus::SUCCESS;
+}
+
 auto CoordinatorInstance::AddCoordinatorInstance(uint32_t raft_server_id, uint32_t raft_port, std::string raft_address)
     -> void {
   raft_state_.AddCoordinatorInstance(raft_server_id, raft_port, std::move(raft_address));

src/coordination/coordinator_rpc.cpp

@@ -52,6 +52,34 @@ void DemoteMainToReplicaRes::Load(DemoteMainToReplicaRes *self, memgraph::slk::R
   memgraph::slk::Load(self, reader);
 }
 
+void UnregisterReplicaReq::Save(UnregisterReplicaReq const &self, memgraph::slk::Builder *builder) {
+  memgraph::slk::Save(self, builder);
+}
+
+void UnregisterReplicaReq::Load(UnregisterReplicaReq *self, memgraph::slk::Reader *reader) {
+  memgraph::slk::Load(self, reader);
+}
+
+void UnregisterReplicaRes::Save(UnregisterReplicaRes const &self, memgraph::slk::Builder *builder) {
+  memgraph::slk::Save(self, builder);
+}
+
+void UnregisterReplicaRes::Load(UnregisterReplicaRes *self, memgraph::slk::Reader *reader) {
+  memgraph::slk::Load(self, reader);
+}
+
+void EnableWritingOnMainRes::Save(EnableWritingOnMainRes const &self, memgraph::slk::Builder *builder) {
+  memgraph::slk::Save(self, builder);
+}
+
+void EnableWritingOnMainRes::Load(EnableWritingOnMainRes *self, memgraph::slk::Reader *reader) {
+  memgraph::slk::Load(self, reader);
+}
+
+void EnableWritingOnMainReq::Save(EnableWritingOnMainReq const &self, memgraph::slk::Builder *builder) {}
+
+void EnableWritingOnMainReq::Load(EnableWritingOnMainReq *self, memgraph::slk::Reader *reader) {}
+
 }  // namespace coordination
 
 constexpr utils::TypeInfo coordination::PromoteReplicaToMainReq::kType{utils::TypeId::COORD_FAILOVER_REQ,
@@ -64,8 +92,21 @@ constexpr utils::TypeInfo coordination::DemoteMainToReplicaReq::kType{utils::Typ
                                                                       "CoordDemoteToReplicaReq", nullptr};
 
 constexpr utils::TypeInfo coordination::DemoteMainToReplicaRes::kType{utils::TypeId::COORD_SET_REPL_MAIN_RES,
+
                                                                       "CoordDemoteToReplicaRes", nullptr};
 
+constexpr utils::TypeInfo coordination::UnregisterReplicaReq::kType{utils::TypeId::COORD_UNREGISTER_REPLICA_REQ,
+                                                                    "UnregisterReplicaReq", nullptr};
+
+constexpr utils::TypeInfo coordination::UnregisterReplicaRes::kType{utils::TypeId::COORD_UNREGISTER_REPLICA_RES,
+                                                                    "UnregisterReplicaRes", nullptr};
+
+constexpr utils::TypeInfo coordination::EnableWritingOnMainReq::kType{utils::TypeId::COORD_ENABLE_WRITING_ON_MAIN_REQ,
+                                                                      "CoordEnableWritingOnMainReq", nullptr};
+
+constexpr utils::TypeInfo coordination::EnableWritingOnMainRes::kType{utils::TypeId::COORD_ENABLE_WRITING_ON_MAIN_RES,
+                                                                      "CoordEnableWritingOnMainRes", nullptr};
+
 namespace slk {
 
 void Save(const memgraph::coordination::PromoteReplicaToMainRes &self, memgraph::slk::Builder *builder) {
@@ -102,6 +143,30 @@ void Load(memgraph::coordination::DemoteMainToReplicaRes *self, memgraph::slk::R
   memgraph::slk::Load(&self->success, reader);
 }
 
+void Save(memgraph::coordination::UnregisterReplicaReq const &self, memgraph::slk::Builder *builder) {
+  memgraph::slk::Save(self.instance_name, builder);
+}
+
+void Load(memgraph::coordination::UnregisterReplicaReq *self, memgraph::slk::Reader *reader) {
+  memgraph::slk::Load(&self->instance_name, reader);
+}
+
+void Save(memgraph::coordination::UnregisterReplicaRes const &self, memgraph::slk::Builder *builder) {
+  memgraph::slk::Save(self.success, builder);
+}
+
+void Load(memgraph::coordination::UnregisterReplicaRes *self, memgraph::slk::Reader *reader) {
+  memgraph::slk::Load(&self->success, reader);
+}
+
+void Save(memgraph::coordination::EnableWritingOnMainRes const &self, memgraph::slk::Builder *builder) {
+  memgraph::slk::Save(self.success, builder);
+}
+
+void Load(memgraph::coordination::EnableWritingOnMainRes *self, memgraph::slk::Reader *reader) {
+  memgraph::slk::Load(&self->success, reader);
+}
+
 }  // namespace slk
 
 }  // namespace memgraph

src/coordination/coordinator_state.cpp

@@ -56,6 +56,20 @@ auto CoordinatorState::RegisterReplicationInstance(CoordinatorClientConfig confi
       data_);
 }
 
+auto CoordinatorState::UnregisterReplicationInstance(std::string instance_name) -> UnregisterInstanceCoordinatorStatus {
+  MG_ASSERT(std::holds_alternative<CoordinatorInstance>(data_),
+            "Coordinator cannot unregister instance since variant holds wrong alternative");
+
+  return std::visit(
+      memgraph::utils::Overloaded{[](const CoordinatorMainReplicaData & /*coordinator_main_replica_data*/) {
+                                    return UnregisterInstanceCoordinatorStatus::NOT_COORDINATOR;
+                                  },
+                                  [&instance_name](CoordinatorInstance &coordinator_instance) {
+                                    return coordinator_instance.UnregisterReplicationInstance(instance_name);
+                                  }},
+      data_);
+}
+
 auto CoordinatorState::SetReplicationInstanceToMain(std::string instance_name) -> SetInstanceToMainCoordinatorStatus {
   MG_ASSERT(std::holds_alternative<CoordinatorInstance>(data_),
             "Coordinator cannot register replica since variant holds wrong alternative");

src/coordination/include/coordination/coordinator_client.hpp

@@ -11,12 +11,12 @@
 
 #pragma once
 
-#include "utils/uuid.hpp"
 #ifdef MG_ENTERPRISE
 
 #include "coordination/coordinator_config.hpp"
 #include "rpc/client.hpp"
 #include "utils/scheduler.hpp"
+#include "utils/uuid.hpp"
 
 namespace memgraph::coordination {
 
@@ -46,17 +46,24 @@ class CoordinatorClient {
   auto SocketAddress() const -> std::string;
 
   [[nodiscard]] auto DemoteToReplica() const -> bool;
+  // TODO: (andi) Consistent naming
   auto SendPromoteReplicaToMainRpc(const utils::UUID &uuid, ReplicationClientsInfo replication_clients_info) const
       -> bool;
 
   auto SendSwapMainUUIDRpc(const utils::UUID &uuid) const -> bool;
 
+  auto SendUnregisterReplicaRpc(std::string const &instance_name) const -> bool;
+
+  auto SendEnableWritingOnMainRpc() const -> bool;
+
   auto ReplicationClientInfo() const -> ReplClientInfo;
 
   auto SetCallbacks(HealthCheckCallback succ_cb, HealthCheckCallback fail_cb) -> void;
 
   auto RpcClient() -> rpc::Client & { return rpc_client_; }
 
+  auto InstanceDownTimeoutSec() const -> std::chrono::seconds;
+
   friend bool operator==(CoordinatorClient const &first, CoordinatorClient const &second) {
     return first.config_ == second.config_;
   }

src/coordination/include/coordination/coordinator_config.hpp

@@ -28,7 +28,8 @@ struct CoordinatorClientConfig {
   std::string instance_name;
   std::string ip_address;
   uint16_t port{};
-  std::chrono::seconds health_check_frequency_sec{1};
+  std::chrono::seconds instance_health_check_frequency_sec{1};
+  std::chrono::seconds instance_down_timeout_sec{5};
 
   auto SocketAddress() const -> std::string { return ip_address + ":" + std::to_string(port); }
 

src/coordination/include/coordination/coordinator_handlers.hpp

@@ -33,6 +33,11 @@ class CoordinatorHandlers {
                                          slk::Builder *res_builder);
   static void SwapMainUUIDHandler(replication::ReplicationHandler &replication_handler, slk::Reader *req_reader,
                                   slk::Builder *res_builder);
+
+  static void UnregisterReplicaHandler(replication::ReplicationHandler &replication_handler, slk::Reader *req_reader,
+                                       slk::Builder *res_builder);
+  static void EnableWritingOnMainHandler(replication::ReplicationHandler &replication_handler, slk::Reader *req_reader,
+                                         slk::Builder *res_builder);
 };
 
 }  // namespace memgraph::dbms

src/coordination/include/coordination/coordinator_instance.hpp

@@ -30,6 +30,7 @@ class CoordinatorInstance {
   CoordinatorInstance();
 
   [[nodiscard]] auto RegisterReplicationInstance(CoordinatorClientConfig config) -> RegisterInstanceCoordinatorStatus;
+  [[nodiscard]] auto UnregisterReplicationInstance(std::string instance_name) -> UnregisterInstanceCoordinatorStatus;
 
   [[nodiscard]] auto SetReplicationInstanceToMain(std::string instance_name) -> SetInstanceToMainCoordinatorStatus;
 
@@ -44,8 +45,6 @@ class CoordinatorInstance {
   auto SetMainUUID(utils::UUID new_uuid) -> void;
 
  private:
-  auto ClusterHasAliveMain_() const -> bool;
-
   HealthCheckCallback main_succ_cb_, main_fail_cb_, replica_succ_cb_, replica_fail_cb_;
 
   // NOTE: Must be std::list because we rely on pointer stability

src/coordination/include/coordination/coordinator_rpc.hpp

@@ -82,6 +82,60 @@ struct DemoteMainToReplicaRes {
 
 using DemoteMainToReplicaRpc = rpc::RequestResponse<DemoteMainToReplicaReq, DemoteMainToReplicaRes>;
 
+struct UnregisterReplicaReq {
+  static const utils::TypeInfo kType;
+  static const utils::TypeInfo &GetTypeInfo() { return kType; }
+
+  static void Load(UnregisterReplicaReq *self, memgraph::slk::Reader *reader);
+  static void Save(UnregisterReplicaReq const &self, memgraph::slk::Builder *builder);
+
+  explicit UnregisterReplicaReq(std::string instance_name) : instance_name(std::move(instance_name)) {}
+
+  UnregisterReplicaReq() = default;
+
+  std::string instance_name;
+};
+
+struct UnregisterReplicaRes {
+  static const utils::TypeInfo kType;
+  static const utils::TypeInfo &GetTypeInfo() { return kType; }
+
+  static void Load(UnregisterReplicaRes *self, memgraph::slk::Reader *reader);
+  static void Save(const UnregisterReplicaRes &self, memgraph::slk::Builder *builder);
+
+  explicit UnregisterReplicaRes(bool success) : success(success) {}
+  UnregisterReplicaRes() = default;
+
+  bool success;
+};
+
+using UnregisterReplicaRpc = rpc::RequestResponse<UnregisterReplicaReq, UnregisterReplicaRes>;
+
+struct EnableWritingOnMainReq {
+  static const utils::TypeInfo kType;
+  static const utils::TypeInfo &GetTypeInfo() { return kType; }
+
+  static void Load(EnableWritingOnMainReq *self, memgraph::slk::Reader *reader);
+  static void Save(EnableWritingOnMainReq const &self, memgraph::slk::Builder *builder);
+
+  EnableWritingOnMainReq() = default;
+};
+
+struct EnableWritingOnMainRes {
+  static const utils::TypeInfo kType;
+  static const utils::TypeInfo &GetTypeInfo() { return kType; }
+
+  static void Load(EnableWritingOnMainRes *self, memgraph::slk::Reader *reader);
+  static void Save(EnableWritingOnMainRes const &self, memgraph::slk::Builder *builder);
+
+  explicit EnableWritingOnMainRes(bool success) : success(success) {}
+  EnableWritingOnMainRes() = default;
+
+  bool success;
+};
+
+using EnableWritingOnMainRpc = rpc::RequestResponse<EnableWritingOnMainReq, EnableWritingOnMainRes>;
+
 }  // namespace memgraph::coordination
 
 // SLK serialization declarations
@@ -99,6 +153,14 @@ void Load(memgraph::coordination::DemoteMainToReplicaRes *self, memgraph::slk::R
 void Save(const memgraph::coordination::DemoteMainToReplicaReq &self, memgraph::slk::Builder *builder);
 void Load(memgraph::coordination::DemoteMainToReplicaReq *self, memgraph::slk::Reader *reader);
 
+// UnregisterReplicaRpc
+void Save(memgraph::coordination::UnregisterReplicaRes const &self, memgraph::slk::Builder *builder);
+void Load(memgraph::coordination::UnregisterReplicaRes *self, memgraph::slk::Reader *reader);
+void Save(memgraph::coordination::UnregisterReplicaReq const &self, memgraph::slk::Builder *builder);
+void Load(memgraph::coordination::UnregisterReplicaReq *self, memgraph::slk::Reader *reader);
+
+void Save(memgraph::coordination::EnableWritingOnMainRes const &self, memgraph::slk::Builder *builder);
+void Load(memgraph::coordination::EnableWritingOnMainRes *self, memgraph::slk::Reader *reader);
 
 }  // namespace memgraph::slk
 

src/coordination/include/coordination/coordinator_state.hpp

@@ -34,6 +34,7 @@ class CoordinatorState {
   CoordinatorState &operator=(CoordinatorState &&) noexcept = delete;
 
   [[nodiscard]] auto RegisterReplicationInstance(CoordinatorClientConfig config) -> RegisterInstanceCoordinatorStatus;
+  [[nodiscard]] auto UnregisterReplicationInstance(std::string instance_name) -> UnregisterInstanceCoordinatorStatus;
 
   [[nodiscard]] auto SetReplicationInstanceToMain(std::string instance_name) -> SetInstanceToMainCoordinatorStatus;
 

src/coordination/include/coordination/raft_state.hpp

@@ -15,8 +15,6 @@
 
 #include <flags/replication.hpp>
 
-#include <optional>
-
 #include <libnuraft/nuraft.hxx>
 
 namespace memgraph::coordination {

src/coordination/include/coordination/register_main_replica_coordinator_status.hpp

@@ -28,6 +28,15 @@ enum class RegisterInstanceCoordinatorStatus : uint8_t {
   SUCCESS
 };
 
+enum class UnregisterInstanceCoordinatorStatus : uint8_t {
+  NO_INSTANCE_WITH_NAME,
+  IS_MAIN,
+  NOT_COORDINATOR,
+  NOT_LEADER,
+  RPC_FAILED,
+  SUCCESS,
+};
+
 enum class SetInstanceToMainCoordinatorStatus : uint8_t {
   NO_INSTANCE_WITH_NAME,
   NOT_COORDINATOR,

src/coordination/include/coordination/replication_instance.hpp

@@ -14,7 +14,6 @@
 #ifdef MG_ENTERPRISE
 
 #include "coordination/coordinator_client.hpp"
-#include "coordination/coordinator_cluster_config.hpp"
 #include "coordination/coordinator_exceptions.hpp"
 #include "replication_coordination_glue/role.hpp"
 
@@ -59,12 +58,18 @@ class ReplicationInstance {
   auto ReplicationClientInfo() const -> ReplClientInfo;
 
   auto EnsureReplicaHasCorrectMainUUID(utils::UUID const &curr_main_uuid) -> bool;
+
   auto SendSwapAndUpdateUUID(const utils::UUID &new_main_uuid) -> bool;
+  auto SendUnregisterReplicaRpc(std::string const &instance_name) -> bool;
+
+  // TODO: (andi) Inconsistent API
   auto GetClient() -> CoordinatorClient &;
 
+  auto EnableWritingOnMain() -> bool;
+
   auto SetNewMainUUID(utils::UUID const &main_uuid) -> void;
   auto ResetMainUUID() -> void;
-  auto GetMainUUID() -> const std::optional<utils::UUID> &;
+  auto GetMainUUID() const -> const std::optional<utils::UUID> &;
 
  private:
   CoordinatorClient client_;

src/coordination/replication_instance.cpp

@@ -34,9 +34,8 @@ auto ReplicationInstance::OnSuccessPing() -> void {
 }
 
 auto ReplicationInstance::OnFailPing() -> bool {
-  is_alive_ =
-      std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now() - last_response_time_).count() <
-      CoordinatorClusterConfig::alive_response_time_difference_sec_;
+  auto elapsed_time = std::chrono::system_clock::now() - last_response_time_;
+  is_alive_ = elapsed_time < client_.InstanceDownTimeoutSec();
   return is_alive_;
 }
 
@@ -86,9 +85,10 @@ auto ReplicationInstance::ReplicationClientInfo() const -> CoordinatorClientConf
 }
 
 auto ReplicationInstance::GetClient() -> CoordinatorClient & { return client_; }
+
 auto ReplicationInstance::SetNewMainUUID(utils::UUID const &main_uuid) -> void { main_uuid_ = main_uuid; }
 auto ReplicationInstance::ResetMainUUID() -> void { main_uuid_ = std::nullopt; }
-auto ReplicationInstance::GetMainUUID() -> const std::optional<utils::UUID> & { return main_uuid_; }
+auto ReplicationInstance::GetMainUUID() const -> std::optional<utils::UUID> const & { return main_uuid_; }
 
 auto ReplicationInstance::EnsureReplicaHasCorrectMainUUID(utils::UUID const &curr_main_uuid) -> bool {
   if (!main_uuid_ || *main_uuid_ != curr_main_uuid) {
@@ -105,5 +105,11 @@ auto ReplicationInstance::SendSwapAndUpdateUUID(const utils::UUID &new_main_uuid
   return true;
 }
 
+auto ReplicationInstance::SendUnregisterReplicaRpc(std::string const &instance_name) -> bool {
+  return client_.SendUnregisterReplicaRpc(instance_name);
+}
+
+auto ReplicationInstance::EnableWritingOnMain() -> bool { return client_.SendEnableWritingOnMainRpc(); }
+
 }  // namespace memgraph::coordination
 #endif

src/dbms/constants.hpp

@@ -16,10 +16,4 @@ namespace memgraph::dbms {
 constexpr std::string_view kDefaultDB = "memgraph";        //!< Name of the default database
 constexpr std::string_view kMultiTenantDir = "databases";  //!< Name of the multi-tenant directory
 
-#ifdef MG_EXPERIMENTAL_REPLICATION_MULTITENANCY
-constexpr bool allow_mt_repl = true;
-#else
-constexpr bool allow_mt_repl = false;
-#endif
-
 }  // namespace memgraph::dbms

src/dbms/coordinator_handler.cpp

@@ -25,6 +25,11 @@ auto CoordinatorHandler::RegisterReplicationInstance(memgraph::coordination::Coo
   return coordinator_state_.RegisterReplicationInstance(config);
 }
 
+auto CoordinatorHandler::UnregisterReplicationInstance(std::string instance_name)
+    -> coordination::UnregisterInstanceCoordinatorStatus {
+  return coordinator_state_.UnregisterReplicationInstance(std::move(instance_name));
+}
+
 auto CoordinatorHandler::SetReplicationInstanceToMain(std::string instance_name)
     -> coordination::SetInstanceToMainCoordinatorStatus {
   return coordinator_state_.SetReplicationInstanceToMain(std::move(instance_name));

src/dbms/coordinator_handler.hpp

@@ -28,9 +28,13 @@ class CoordinatorHandler {
  public:
   explicit CoordinatorHandler(coordination::CoordinatorState &coordinator_state);
 
+  // TODO: (andi) When moving coordinator state on same instances, rename from RegisterReplicationInstance to
+  // RegisterInstance
   auto RegisterReplicationInstance(coordination::CoordinatorClientConfig config)
       -> coordination::RegisterInstanceCoordinatorStatus;
 
+  auto UnregisterReplicationInstance(std::string instance_name) -> coordination::UnregisterInstanceCoordinatorStatus;
+
   auto SetReplicationInstanceToMain(std::string instance_name) -> coordination::SetInstanceToMainCoordinatorStatus;
 
   auto ShowInstances() const -> std::vector<coordination::InstanceStatus>;

src/dbms/dbms_handler.cpp

@@ -16,6 +16,7 @@
 
 #include "dbms/constants.hpp"
 #include "dbms/global.hpp"
+#include "flags/experimental.hpp"
 #include "spdlog/spdlog.h"
 #include "system/include/system/system.hpp"
 #include "utils/exceptions.hpp"
@@ -158,9 +159,9 @@ struct Durability {
   }
 };
 
-DbmsHandler::DbmsHandler(storage::Config config, memgraph::system::System &system,
-                         replication::ReplicationState &repl_state, auth::SynchedAuth &auth, bool recovery_on_startup)
-    : default_config_{std::move(config)}, auth_{auth}, repl_state_{repl_state}, system_{&system} {
+DbmsHandler::DbmsHandler(storage::Config config, replication::ReplicationState &repl_state, auth::SynchedAuth &auth,
+                         bool recovery_on_startup)
+    : default_config_{std::move(config)}, auth_{auth}, repl_state_{repl_state} {
   // TODO: Decouple storage config from dbms config
   // TODO: Save individual db configs inside the kvstore and restore from there
 
@@ -419,7 +420,10 @@ void DbmsHandler::UpdateDurability(const storage::Config &config, std::optional<
 #endif
 
 void DbmsHandler::RecoverStorageReplication(DatabaseAccess db_acc, replication::RoleMainData &role_main_data) {
-  if (allow_mt_repl || db_acc->name() == dbms::kDefaultDB) {
+  using enum memgraph::flags::Experiments;
+  auto const is_enterprise = license::global_license_checker.IsEnterpriseValidFast();
+  auto experimental_system_replication = flags::AreExperimentsEnabled(SYSTEM_REPLICATION);
+  if ((is_enterprise && experimental_system_replication) || db_acc->name() == dbms::kDefaultDB) {
     // Handle global replication state
     spdlog::info("Replication configuration will be stored and will be automatically restored in case of a crash.");
     // RECOVER REPLICA CONNECTIONS

src/dbms/dbms_handler.hpp

@@ -107,8 +107,7 @@ class DbmsHandler {
    * @param auth pointer to the global authenticator
    * @param recovery_on_startup restore databases (and its content) and authentication data
    */
-  DbmsHandler(storage::Config config, memgraph::system::System &system, replication::ReplicationState &repl_state,
-              auth::SynchedAuth &auth,
+  DbmsHandler(storage::Config config, replication::ReplicationState &repl_state, auth::SynchedAuth &auth,
               bool recovery_on_startup);  // TODO If more arguments are added use a config struct
 #else
   /**
@@ -116,9 +115,8 @@ class DbmsHandler {
    *
    * @param configs storage configuration
    */
-  DbmsHandler(storage::Config config, memgraph::system::System &system, replication::ReplicationState &repl_state)
+  DbmsHandler(storage::Config config, replication::ReplicationState &repl_state)
       : repl_state_{repl_state},
-        system_{&system},
         db_gatekeeper_{[&] {
                          config.salient.name = kDefaultDB;
                          return std::move(config);
@@ -272,6 +270,20 @@ class DbmsHandler {
   // coordination::CoordinatorState &CoordinatorState() { return coordinator_state_; }
 #endif
 
+  /**
+   * @brief Return all active databases.
+   *
+   * @return std::vector<std::string>
+   */
+  auto Count() const -> std::size_t {
+#ifdef MG_ENTERPRISE
+    std::shared_lock<LockT> rd(lock_);
+    return db_handler_.size();
+#else
+    return 1;
+#endif
+  }
+
   /**
    * @brief Return the statistics all databases.
    *
@@ -587,9 +599,6 @@ class DbmsHandler {
   //       current replication role. TODO: make Database Access explicit about the role and remove this from
   //       dbms stuff
   replication::ReplicationState &repl_state_;  //!< Ref to global replication state
- public:
-  // TODO fix to be non public/remove from dbms....maybe
-  system::System *system_;
 
 #ifndef MG_ENTERPRISE
   mutable utils::Gatekeeper<Database> db_gatekeeper_;  //!< Single databases gatekeeper

src/dbms/handler.hpp

@@ -144,6 +144,8 @@ class Handler {
   auto cbegin() const { return items_.cbegin(); }
   auto cend() const { return items_.cend(); }
 
+  auto size() const { return items_.size(); }
+
   struct string_hash {
     using is_transparent = void;
     [[nodiscard]] size_t operator()(const char *s) const { return std::hash<std::string_view>{}(s); }

src/flags/CMakeLists.txt

@@ -1,12 +1,17 @@
-add_library(mg-flags STATIC audit.cpp
-                            bolt.cpp
-                            general.cpp
-                            isolation_level.cpp
-                            log_level.cpp
-                            memory_limit.cpp
-                            run_time_configurable.cpp
-                            storage_mode.cpp
-                            query.cpp
-                            replication.cpp)
-target_include_directories(mg-flags PUBLIC ${CMAKE_SOURCE_DIR}/include)
-target_link_libraries(mg-flags PUBLIC spdlog::spdlog mg-settings mg-utils)
+add_library(mg-flags STATIC
+        audit.cpp
+        bolt.cpp
+        general.cpp
+        isolation_level.cpp
+        log_level.cpp
+        memory_limit.cpp
+        run_time_configurable.cpp
+        storage_mode.cpp
+        query.cpp
+        replication.cpp
+        experimental.cpp
+        experimental.hpp)
+target_include_directories(mg-flags PUBLIC include)
+target_link_libraries(mg-flags
+        PUBLIC spdlog::spdlog mg-settings mg-utils
+        PRIVATE lib::rangev3)

src/flags/all.hpp

@@ -12,6 +12,7 @@
 
 #include "flags/audit.hpp"
 #include "flags/bolt.hpp"
+#include "flags/experimental.hpp"
 #include "flags/general.hpp"
 #include "flags/isolation_level.hpp"
 #include "flags/log_level.hpp"

src/flags/experimental.cpp

@@ -0,0 +1,67 @@
+// Copyright 2024 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#include "flags/experimental.hpp"
+#include "range/v3/all.hpp"
+#include "utils/string.hpp"
+
+#include <map>
+#include <string_view>
+
+// Bolt server flags.
+// NOLINTNEXTLINE (cppcoreguidelines-avoid-non-const-global-variables)
+DEFINE_string(experimental_enabled, "",
+              "Experimental features to be used, comma seperated. Options [system-replication]");
+
+using namespace std::string_view_literals;
+
+namespace memgraph::flags {
+
+auto const mapping = std::map{std::pair{"system-replication"sv, Experiments::SYSTEM_REPLICATION}};
+
+auto ExperimentsInstance() -> Experiments & {
+  static auto instance = Experiments{};
+  return instance;
+}
+
+bool AreExperimentsEnabled(Experiments experiments) {
+  using t = std::underlying_type_t<Experiments>;
+
+  auto actual = static_cast<t>(ExperimentsInstance());
+  auto check = static_cast<t>(experiments);
+
+  return (actual & check) == check;
+}
+
+void InitializeExperimental() {
+  namespace rv = ranges::views;
+
+  auto const connonicalize_string = [](auto &&rng) {
+    auto const is_space = [](auto c) { return c == ' '; };
+    auto const to_lower = [](unsigned char c) { return std::tolower(c); };
+
+    return rng | rv::drop_while(is_space) | rv::take_while(std::not_fn(is_space)) | rv::transform(to_lower) |
+           ranges::to<std::string>;
+  };
+
+  auto const mapping_end = mapping.cend();
+  using underlying_type = std::underlying_type_t<Experiments>;
+  auto to_set = underlying_type{};
+  for (auto &&experiment : FLAGS_experimental_enabled | rv::split(',') | rv::transform(connonicalize_string)) {
+    if (auto it = mapping.find(experiment); it != mapping_end) {
+      to_set |= static_cast<underlying_type>(it->second);
+    }
+  }
+
+  ExperimentsInstance() = static_cast<Experiments>(to_set);
+}
+
+}  // namespace memgraph::flags

src/flags/experimental.hpp

@@ -0,0 +1,32 @@
+// Copyright 2024 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#pragma once
+
+#include "gflags/gflags.h"
+
+// Short help flag.
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DECLARE_string(experimental_enabled);
+
+namespace memgraph::flags {
+
+// Each bit is an enabled experiment
+// old experiments can be reused once code cleanup has happened
+enum class Experiments : uint8_t {
+  SYSTEM_REPLICATION = 1 << 0,
+};
+
+bool AreExperimentsEnabled(Experiments experiments);
+
+void InitializeExperimental();
+
+}  // namespace memgraph::flags

src/flags/replication.cpp

@@ -18,6 +18,10 @@ DEFINE_uint32(coordinator_server_port, 0, "Port on which coordinator servers wil
 DEFINE_uint32(raft_server_port, 0, "Port on which raft servers will be started.");
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
 DEFINE_uint32(raft_server_id, 0, "Unique ID of the raft server.");
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DEFINE_uint32(instance_down_timeout_sec, 5, "Time duration after which an instance is considered down.");
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DEFINE_uint32(instance_health_check_frequency_sec, 1, "The time duration between two health checks/pings.");
 #endif
 
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)

src/flags/replication.hpp

@@ -20,6 +20,10 @@ DECLARE_uint32(coordinator_server_port);
 DECLARE_uint32(raft_server_port);
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
 DECLARE_uint32(raft_server_id);
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DECLARE_uint32(instance_down_timeout_sec);
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DECLARE_uint32(instance_health_check_frequency_sec);
 #endif
 
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)

src/memgraph.cpp

@@ -134,6 +134,7 @@ int main(int argc, char **argv) {
   }
 
   memgraph::flags::InitializeLogger();
+  memgraph::flags::InitializeExperimental();
 
   // Unhandled exception handler init.
   std::set_terminate(&memgraph::utils::TerminateHandler);
@@ -355,6 +356,10 @@ int main(int argc, char **argv) {
   memgraph::query::InterpreterConfig interp_config{
       .query = {.allow_load_csv = FLAGS_allow_load_csv},
       .replication_replica_check_frequency = std::chrono::seconds(FLAGS_replication_replica_check_frequency_sec),
+#ifdef MG_ENTERPRISE
+      .instance_down_timeout_sec = std::chrono::seconds(FLAGS_instance_down_timeout_sec),
+      .instance_health_check_frequency_sec = std::chrono::seconds(FLAGS_instance_health_check_frequency_sec),
+#endif
       .default_kafka_bootstrap_servers = FLAGS_kafka_bootstrap_servers,
       .default_pulsar_service_url = FLAGS_pulsar_service_url,
       .stream_transaction_conflict_retries = FLAGS_stream_transaction_conflict_retries,
@@ -396,7 +401,7 @@ int main(int argc, char **argv) {
   memgraph::coordination::CoordinatorState coordinator_state;
 #endif
 
-  memgraph::dbms::DbmsHandler dbms_handler(db_config, system, repl_state
+  memgraph::dbms::DbmsHandler dbms_handler(db_config, repl_state
 #ifdef MG_ENTERPRISE
                                            ,
                                            auth_, FLAGS_data_recovery_on_startup
@@ -409,7 +414,7 @@ int main(int argc, char **argv) {
   auto replication_handler = memgraph::replication::ReplicationHandler{repl_state, dbms_handler
 #ifdef MG_ENTERPRISE
                                                                        ,
-                                                                       &system, auth_
+                                                                       system, auth_
 #endif
   };
 

src/query/config.hpp

@@ -22,6 +22,9 @@ struct InterpreterConfig {
   // The same as \ref memgraph::replication::ReplicationClientConfig
   std::chrono::seconds replication_replica_check_frequency{1};
 
+  std::chrono::seconds instance_down_timeout_sec{5};
+  std::chrono::seconds instance_health_check_frequency_sec{1};
+
   std::string default_kafka_bootstrap_servers;
   std::string default_pulsar_service_url;
   uint32_t stream_transaction_conflict_retries;

src/query/frontend/ast/ast.hpp

@@ -3034,7 +3034,7 @@ class ReplicationQuery : public memgraph::query::Query {
 
   enum class SyncMode { SYNC, ASYNC };
 
-  enum class ReplicaState { READY, REPLICATING, RECOVERY, MAYBE_BEHIND };
+  enum class ReplicaState { READY, REPLICATING, RECOVERY, MAYBE_BEHIND, DIVERGED_FROM_MAIN };
 
   ReplicationQuery() = default;
 
@@ -3071,7 +3071,13 @@ class CoordinatorQuery : public memgraph::query::Query {
   static const utils::TypeInfo kType;
   const utils::TypeInfo &GetTypeInfo() const override { return kType; }
 
-  enum class Action { REGISTER_INSTANCE, SET_INSTANCE_TO_MAIN, SHOW_INSTANCES, ADD_COORDINATOR_INSTANCE };
+  enum class Action {
+    REGISTER_INSTANCE,
+    UNREGISTER_INSTANCE,
+    SET_INSTANCE_TO_MAIN,
+    SHOW_INSTANCES,
+    ADD_COORDINATOR_INSTANCE
+  };
 
   enum class SyncMode { SYNC, ASYNC };
 

src/query/frontend/ast/cypher_main_visitor.cpp

@@ -399,6 +399,14 @@ antlrcpp::Any CypherMainVisitor::visitRegisterInstanceOnCoordinator(
   return coordinator_query;
 }
 
+antlrcpp::Any CypherMainVisitor::visitUnregisterInstanceOnCoordinator(
+    MemgraphCypher::UnregisterInstanceOnCoordinatorContext *ctx) {
+  auto *coordinator_query = storage_->Create<CoordinatorQuery>();
+  coordinator_query->action_ = CoordinatorQuery::Action::UNREGISTER_INSTANCE;
+  coordinator_query->instance_name_ = std::any_cast<std::string>(ctx->instanceName()->symbolicName()->accept(this));
+  return coordinator_query;
+}
+
 antlrcpp::Any CypherMainVisitor::visitAddCoordinatorInstance(MemgraphCypher::AddCoordinatorInstanceContext *ctx) {
   auto *coordinator_query = storage_->Create<CoordinatorQuery>();
 

src/query/frontend/ast/cypher_main_visitor.hpp

@@ -243,6 +243,12 @@ class CypherMainVisitor : public antlropencypher::MemgraphCypherBaseVisitor {
    */
   antlrcpp::Any visitRegisterInstanceOnCoordinator(MemgraphCypher::RegisterInstanceOnCoordinatorContext *ctx) override;
 
+  /**
+   * @return CoordinatorQuery*
+   */
+  antlrcpp::Any visitUnregisterInstanceOnCoordinator(
+      MemgraphCypher::UnregisterInstanceOnCoordinatorContext *ctx) override;
+
   /**
    * @return CoordinatorQuery*
    */

src/query/frontend/opencypher/grammar/MemgraphCypher.g4

@@ -190,6 +190,7 @@ replicationQuery : setReplicationRole
                  ;
 
 coordinatorQuery : registerInstanceOnCoordinator
+                 | unregisterInstanceOnCoordinator
                  | setInstanceToMain
                  | showInstances
                  | addCoordinatorInstance
@@ -392,6 +393,8 @@ registerReplica : REGISTER REPLICA instanceName ( SYNC | ASYNC )
 
 registerInstanceOnCoordinator : REGISTER INSTANCE instanceName ON coordinatorSocketAddress ( AS ASYNC ) ? WITH replicationSocketAddress ;
 
+unregisterInstanceOnCoordinator : UNREGISTER INSTANCE instanceName ;
+
 setInstanceToMain : SET INSTANCE instanceName TO MAIN ;
 
 raftServerId : literal ;

src/query/frontend/opencypher/grammar/MemgraphCypherLexer.g4

@@ -141,6 +141,7 @@ TRIGGER                 : T R I G G E R ;
 TRIGGERS                : T R I G G E R S ;
 UNCOMMITTED             : U N C O M M I T T E D ;
 UNLOCK                  : U N L O C K ;
+UNREGISTER              : U N R E G I S T E R ;
 UPDATE                  : U P D A T E ;
 USE                     : U S E ;
 USER                    : U S E R ;

src/query/interpreter.cpp

@@ -93,6 +93,7 @@
 #include "utils/exceptions.hpp"
 #include "utils/file.hpp"
 #include "utils/flag_validation.hpp"
+#include "utils/functional.hpp"
 #include "utils/likely.hpp"
 #include "utils/logging.hpp"
 #include "utils/memory.hpp"
@@ -109,6 +110,7 @@
 
 #ifdef MG_ENTERPRISE
 #include "coordination/constants.hpp"
+#include "flags/experimental.hpp"
 #endif
 
 namespace memgraph::metrics {
@@ -437,6 +439,9 @@ class ReplQueryHandler {
         case storage::replication::ReplicaState::MAYBE_BEHIND:
           replica.state = ReplicationQuery::ReplicaState::MAYBE_BEHIND;
           break;
+        case storage::replication::ReplicaState::DIVERGED_FROM_MAIN:
+          replica.state = ReplicationQuery::ReplicaState::DIVERGED_FROM_MAIN;
+          break;
       }
 
       return replica;
@@ -457,11 +462,32 @@ class CoordQueryHandler final : public query::CoordinatorQueryHandler {
 
       : coordinator_handler_(coordinator_state) {}
 
-  /// @throw QueryRuntimeException if an error ocurred.
-  void RegisterReplicationInstance(const std::string &coordinator_socket_address,
-                                   const std::string &replication_socket_address,
-                                   const std::chrono::seconds instance_check_frequency,
-                                   const std::string &instance_name, CoordinatorQuery::SyncMode sync_mode) override {
+  void UnregisterInstance(std::string const &instance_name) override {
+    auto status = coordinator_handler_.UnregisterReplicationInstance(instance_name);
+    switch (status) {
+      using enum memgraph::coordination::UnregisterInstanceCoordinatorStatus;
+      case NO_INSTANCE_WITH_NAME:
+        throw QueryRuntimeException("No instance with such name!");
+      case IS_MAIN:
+        throw QueryRuntimeException(
+            "Alive main instance can't be unregistered! Shut it down to trigger failover and then unregister it!");
+      case NOT_COORDINATOR:
+        throw QueryRuntimeException("UNREGISTER INSTANCE query can only be run on a coordinator!");
+      case NOT_LEADER:
+        throw QueryRuntimeException("Couldn't unregister replica instance since coordinator is not a leader!");
+      case RPC_FAILED:
+        throw QueryRuntimeException(
+            "Couldn't unregister replica instance because current main instance couldn't unregister replica!");
+      case SUCCESS:
+        break;
+    }
+  }
+
+  void RegisterReplicationInstance(std::string const &coordinator_socket_address,
+                                   std::string const &replication_socket_address,
+                                   std::chrono::seconds const &instance_check_frequency,
+                                   std::chrono::seconds const &instance_down_timeout, std::string const &instance_name,
+                                   CoordinatorQuery::SyncMode sync_mode) override {
     const auto maybe_replication_ip_port =
         io::network::Endpoint::ParseSocketOrAddress(replication_socket_address, std::nullopt);
     if (!maybe_replication_ip_port) {
@@ -486,7 +512,8 @@ class CoordQueryHandler final : public query::CoordinatorQueryHandler {
         coordination::CoordinatorClientConfig{.instance_name = instance_name,
                                               .ip_address = coordinator_server_ip,
                                               .port = coordinator_server_port,
-                                              .health_check_frequency_sec = instance_check_frequency,
+                                              .instance_health_check_frequency_sec = instance_check_frequency,
+                                              .instance_down_timeout_sec = instance_down_timeout,
                                               .replication_client_info = repl_config,
                                               .ssl = std::nullopt};
 
@@ -1082,6 +1109,9 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
             case ReplicationQuery::ReplicaState::MAYBE_BEHIND:
               typed_replica.emplace_back("invalid");
               break;
+            case ReplicationQuery::ReplicaState::DIVERGED_FROM_MAIN:
+              typed_replica.emplace_back("diverged");
+              break;
           }
 
           typed_replicas.emplace_back(std::move(typed_replica));
@@ -1151,12 +1181,15 @@ Callback HandleCoordinatorQuery(CoordinatorQuery *coordinator_query, const Param
       auto coordinator_socket_address_tv = coordinator_query->coordinator_socket_address_->Accept(evaluator);
       auto replication_socket_address_tv = coordinator_query->replication_socket_address_->Accept(evaluator);
       callback.fn = [handler = CoordQueryHandler{*coordinator_state}, coordinator_socket_address_tv,
-                     replication_socket_address_tv, main_check_frequency = config.replication_replica_check_frequency,
+                     replication_socket_address_tv,
+                     instance_health_check_frequency_sec = config.instance_health_check_frequency_sec,
                      instance_name = coordinator_query->instance_name_,
+                     instance_down_timeout_sec = config.instance_down_timeout_sec,
                      sync_mode = coordinator_query->sync_mode_]() mutable {
         handler.RegisterReplicationInstance(std::string(coordinator_socket_address_tv.ValueString()),
                                             std::string(replication_socket_address_tv.ValueString()),
-                                            main_check_frequency, instance_name, sync_mode);
+                                            instance_health_check_frequency_sec, instance_down_timeout_sec,
+                                            instance_name, sync_mode);
         return std::vector<std::vector<TypedValue>>();
       };
 
@@ -1166,6 +1199,30 @@ Callback HandleCoordinatorQuery(CoordinatorQuery *coordinator_query, const Param
                       coordinator_socket_address_tv.ValueString(), coordinator_query->instance_name_));
       return callback;
     }
+    case CoordinatorQuery::Action::UNREGISTER_INSTANCE:
+      if (!license::global_license_checker.IsEnterpriseValidFast()) {
+        throw QueryException("Trying to use enterprise feature without a valid license.");
+      }
+
+      if constexpr (!coordination::allow_ha) {
+        throw QueryRuntimeException(
+            "High availability is experimental feature. Please set MG_EXPERIMENTAL_HIGH_AVAILABILITY compile flag to "
+            "be able to use this functionality.");
+      }
+      if (!FLAGS_raft_server_id) {
+        throw QueryRuntimeException("Only coordinator can register coordinator server!");
+      }
+      callback.fn = [handler = CoordQueryHandler{*coordinator_state},
+                     instance_name = coordinator_query->instance_name_]() mutable {
+        handler.UnregisterInstance(instance_name);
+        return std::vector<std::vector<TypedValue>>();
+      };
+      notifications->emplace_back(
+          SeverityLevel::INFO, NotificationCode::UNREGISTER_INSTANCE,
+          fmt::format("Coordinator has unregistered instance {}.", coordinator_query->instance_name_));
+
+      return callback;
+
     case CoordinatorQuery::Action::SET_INSTANCE_TO_MAIN: {
       if (!license::global_license_checker.IsEnterpriseValidFast()) {
         throw QueryException("Trying to use enterprise feature without a valid license.");
@@ -1208,17 +1265,13 @@ Callback HandleCoordinatorQuery(CoordinatorQuery *coordinator_query, const Param
       callback.fn = [handler = CoordQueryHandler{*coordinator_state},
                      replica_nfields = callback.header.size()]() mutable {
         auto const instances = handler.ShowInstances();
-        std::vector<std::vector<TypedValue>> result{};
-        result.reserve(result.size());
+        auto const converter = [](const auto &status) -> std::vector<TypedValue> {
+          return {TypedValue{status.instance_name}, TypedValue{status.raft_socket_address},
+                  TypedValue{status.coord_socket_address}, TypedValue{status.is_alive},
+                  TypedValue{status.cluster_role}};
+        };
 
-        std::ranges::transform(instances, std::back_inserter(result),
-                               [](const auto &status) -> std::vector<TypedValue> {
-                                 return {TypedValue{status.instance_name}, TypedValue{status.raft_socket_address},
-                                         TypedValue{status.coord_socket_address}, TypedValue{status.is_alive},
-                                         TypedValue{status.cluster_role}};
-                               });
-
-        return result;
+        return utils::fmap(converter, instances);
       };
       return callback;
     }
@@ -3881,7 +3934,9 @@ PreparedQuery PrepareMultiDatabaseQuery(ParsedQuery parsed_query, CurrentDB &cur
       if (current_db.in_explicit_db_) {
         throw QueryException("Database switching is prohibited if session explicitly defines the used database");
       }
-      if (!dbms::allow_mt_repl && is_replica) {
+
+      using enum memgraph::flags::Experiments;
+      if (!flags::AreExperimentsEnabled(SYSTEM_REPLICATION) && is_replica) {
         throw QueryException("Query forbidden on the replica!");
       }
       return PreparedQuery{{"STATUS"},
@@ -4346,9 +4401,19 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
 
     UpdateTypeCount(rw_type);
 
-    if (interpreter_context_->repl_state->IsReplica() && IsQueryWrite(rw_type)) {
-      query_execution = nullptr;
-      throw QueryException("Write query forbidden on the replica!");
+    bool const write_query = IsQueryWrite(rw_type);
+    if (write_query) {
+      if (interpreter_context_->repl_state->IsReplica()) {
+        query_execution = nullptr;
+        throw QueryException("Write query forbidden on the replica!");
+      }
+#ifdef MG_ENTERPRISE
+      if (FLAGS_coordinator_server_port && !interpreter_context_->repl_state->IsMainWriteable()) {
+        query_execution = nullptr;
+        throw QueryException(
+            "Write query forbidden on the main! Coordinator needs to enable writing on main by sending RPC message.");
+      }
+#endif
     }
 
     // Set the target db to the current db (some queries have different target from the current db)
@@ -4535,7 +4600,8 @@ void Interpreter::Commit() {
     auto const main_commit = [&](replication::RoleMainData &mainData) {
     // Only enterprise can do system replication
 #ifdef MG_ENTERPRISE
-      if (license::global_license_checker.IsEnterpriseValidFast()) {
+      using enum memgraph::flags::Experiments;
+      if (flags::AreExperimentsEnabled(SYSTEM_REPLICATION) && license::global_license_checker.IsEnterpriseValidFast()) {
         return system_transaction_->Commit(memgraph::system::DoReplication{mainData});
       }
 #endif

src/query/interpreter.hpp

@@ -105,10 +105,14 @@ class CoordinatorQueryHandler {
   };
 
   /// @throw QueryRuntimeException if an error ocurred.
-  virtual void RegisterReplicationInstance(const std::string &coordinator_socket_address,
-                                           const std::string &replication_socket_address,
-                                           const std::chrono::seconds instance_check_frequency,
-                                           const std::string &instance_name, CoordinatorQuery::SyncMode sync_mode) = 0;
+  virtual void RegisterReplicationInstance(std::string const &coordinator_socket_address,
+                                           std::string const &replication_socket_address,
+                                           std::chrono::seconds const &instance_health_check_frequency,
+                                           std::chrono::seconds const &instance_down_timeout,
+                                           std::string const &instance_name, CoordinatorQuery::SyncMode sync_mode) = 0;
+
+  /// @throw QueryRuntimeException if an error ocurred.
+  virtual void UnregisterInstance(std::string const &instance_name) = 0;
 
   /// @throw QueryRuntimeException if an error ocurred.
   virtual void SetReplicationInstanceToMain(const std::string &instance_name) = 0;

src/query/metadata.cpp

@@ -71,6 +71,8 @@ constexpr std::string_view GetCodeString(const NotificationCode code) {
       return "RegisterCoordinatorServer"sv;
     case NotificationCode::ADD_COORDINATOR_INSTANCE:
       return "AddCoordinatorInstance"sv;
+    case NotificationCode::UNREGISTER_INSTANCE:
+      return "UnregisterInstance"sv;
 #endif
     case NotificationCode::REPLICA_PORT_WARNING:
       return "ReplicaPortWarning"sv;

src/query/metadata.hpp

@@ -43,8 +43,9 @@ enum class NotificationCode : uint8_t {
   REPLICA_PORT_WARNING,
   REGISTER_REPLICA,
 #ifdef MG_ENTERPRISE
-  REGISTER_COORDINATOR_SERVER,
+  REGISTER_COORDINATOR_SERVER,  // TODO: (andi) What is this?
   ADD_COORDINATOR_INSTANCE,
+  UNREGISTER_INSTANCE,
 #endif
   SET_REPLICA,
   START_STREAM,

src/replication/include/replication/state.hpp

@@ -39,7 +39,8 @@ enum class RegisterReplicaError : uint8_t { NAME_EXISTS, ENDPOINT_EXISTS, COULD_
 
 struct RoleMainData {
   RoleMainData() = default;
-  explicit RoleMainData(ReplicationEpoch e, std::optional<utils::UUID> uuid = std::nullopt) : epoch_(std::move(e)) {
+  explicit RoleMainData(ReplicationEpoch e, bool writing_enabled, std::optional<utils::UUID> uuid = std::nullopt)
+      : epoch_(std::move(e)), writing_enabled_(writing_enabled) {
     if (uuid) {
       uuid_ = *uuid;
     }
@@ -54,6 +55,7 @@ struct RoleMainData {
   ReplicationEpoch epoch_;
   std::list<ReplicationClient> registered_replicas_{};  // TODO: data race issues
   utils::UUID uuid_;
+  bool writing_enabled_{false};
 };
 
 struct RoleReplicaData {
@@ -90,6 +92,21 @@ struct ReplicationState {
   bool IsMain() const { return GetRole() == replication_coordination_glue::ReplicationRole::MAIN; }
   bool IsReplica() const { return GetRole() == replication_coordination_glue::ReplicationRole::REPLICA; }
 
+  auto IsMainWriteable() const -> bool {
+    if (auto const *main = std::get_if<RoleMainData>(&replication_data_)) {
+      return main->writing_enabled_;
+    }
+    return false;
+  }
+
+  auto EnableWritingOnMain() -> bool {
+    if (auto *main = std::get_if<RoleMainData>(&replication_data_)) {
+      main->writing_enabled_ = true;
+      return true;
+    }
+    return false;
+  }
+
   bool HasDurability() const { return nullptr != durability_; }
 
   bool TryPersistRoleMain(std::string new_epoch, utils::UUID main_uuid);

src/replication/state.cpp

@@ -144,8 +144,8 @@ auto ReplicationState::FetchReplicationData() -> FetchReplicationResult_t {
     return std::visit(
         utils::Overloaded{
             [&](durability::MainRole &&r) -> FetchReplicationResult_t {
-              auto res =
-                  RoleMainData{std::move(r.epoch), r.main_uuid.has_value() ? r.main_uuid.value() : utils::UUID{}};
+              auto res = RoleMainData{std::move(r.epoch), false,
+                                      r.main_uuid.has_value() ? r.main_uuid.value() : utils::UUID{}};
               auto b = durability_->begin(durability::kReplicationReplicaPrefix);
               auto e = durability_->end(durability::kReplicationReplicaPrefix);
               for (; b != e; ++b) {
@@ -253,7 +253,7 @@ bool ReplicationState::SetReplicationRoleMain(const utils::UUID &main_uuid) {
     return false;
   }
 
-  replication_data_ = RoleMainData{ReplicationEpoch{new_epoch}, main_uuid};
+  replication_data_ = RoleMainData{ReplicationEpoch{new_epoch}, true, main_uuid};
 
   return true;
 }

src/replication_handler/include/replication_handler/replication_handler.hpp

@@ -12,6 +12,7 @@
 
 #include "auth/auth.hpp"
 #include "dbms/dbms_handler.hpp"
+#include "flags/experimental.hpp"
 #include "replication/include/replication/state.hpp"
 #include "replication_handler/system_rpc.hpp"
 #include "utils/result.hpp"
@@ -22,8 +23,8 @@ inline std::optional<query::RegisterReplicaError> HandleRegisterReplicaStatus(
     utils::BasicResult<replication::RegisterReplicaError, replication::ReplicationClient *> &instance_client);
 
 #ifdef MG_ENTERPRISE
-void StartReplicaClient(replication::ReplicationClient &client, dbms::DbmsHandler &dbms_handler, utils::UUID main_uuid,
-                        system::System *system, auth::SynchedAuth &auth);
+void StartReplicaClient(replication::ReplicationClient &client, system::System &system, dbms::DbmsHandler &dbms_handler,
+                        utils::UUID main_uuid, auth::SynchedAuth &auth);
 #else
 void StartReplicaClient(replication::ReplicationClient &client, dbms::DbmsHandler &dbms_handler, utils::UUID main_uuid);
 #endif
@@ -33,8 +34,8 @@ void StartReplicaClient(replication::ReplicationClient &client, dbms::DbmsHandle
 // When being called by interpreter no need to gain lock, it should already be under a system transaction
 // But concurrently the FrequentCheck is running and will need to lock before reading last_committed_system_timestamp_
 template <bool REQUIRE_LOCK = false>
-void SystemRestore(replication::ReplicationClient &client, dbms::DbmsHandler &dbms_handler,
-                   const utils::UUID &main_uuid, system::System *system, auth::SynchedAuth &auth) {
+void SystemRestore(replication::ReplicationClient &client, system::System &system, dbms::DbmsHandler &dbms_handler,
+                   const utils::UUID &main_uuid, auth::SynchedAuth &auth) {
   // Check if system is up to date
   if (client.state_.WithLock(
           [](auto &state) { return state == memgraph::replication::ReplicationClient::State::READY; }))
@@ -42,6 +43,10 @@ void SystemRestore(replication::ReplicationClient &client, dbms::DbmsHandler &db
 
   // Try to recover...
   {
+    using enum memgraph::flags::Experiments;
+    bool full_system_replication =
+        flags::AreExperimentsEnabled(SYSTEM_REPLICATION) && license::global_license_checker.IsEnterpriseValidFast();
+    // We still need to system replicate
     struct DbInfo {
       std::vector<storage::SalientConfig> configs;
       uint64_t last_committed_timestamp;
@@ -49,25 +54,25 @@ void SystemRestore(replication::ReplicationClient &client, dbms::DbmsHandler &db
     DbInfo db_info = std::invoke([&] {
       auto guard = std::invoke([&]() -> std::optional<memgraph::system::TransactionGuard> {
         if constexpr (REQUIRE_LOCK) {
-          return system->GenTransactionGuard();
+          return system.GenTransactionGuard();
         }
         return std::nullopt;
       });
 
-      if (license::global_license_checker.IsEnterpriseValidFast()) {
+      if (full_system_replication) {
         auto configs = std::vector<storage::SalientConfig>{};
         dbms_handler.ForEach([&configs](dbms::DatabaseAccess acc) { configs.emplace_back(acc->config().salient); });
         // TODO: This is `SystemRestore` maybe DbInfo is incorrect as it will need Auth also
-        return DbInfo{configs, system->LastCommittedSystemTimestamp()};
+        return DbInfo{configs, system.LastCommittedSystemTimestamp()};
       }
 
       // No license -> send only default config
-      return DbInfo{{dbms_handler.Get()->config().salient}, system->LastCommittedSystemTimestamp()};
+      return DbInfo{{dbms_handler.Get()->config().salient}, system.LastCommittedSystemTimestamp()};
     });
     try {
       auto stream = std::invoke([&]() {
         // Handle only default database is no license
-        if (!license::global_license_checker.IsEnterpriseValidFast()) {
+        if (!full_system_replication) {
           return client.rpc_client_.Stream<replication::SystemRecoveryRpc>(
               main_uuid, db_info.last_committed_timestamp, std::move(db_info.configs), auth::Auth::Config{},
               std::vector<auth::User>{}, std::vector<auth::Role>{});
@@ -98,7 +103,7 @@ void SystemRestore(replication::ReplicationClient &client, dbms::DbmsHandler &db
 struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
 #ifdef MG_ENTERPRISE
   explicit ReplicationHandler(memgraph::replication::ReplicationState &repl_state,
-                              memgraph::dbms::DbmsHandler &dbms_handler, memgraph::system::System *system,
+                              memgraph::dbms::DbmsHandler &dbms_handler, memgraph::system::System &system,
                               memgraph::auth::SynchedAuth &auth);
 #else
   explicit ReplicationHandler(memgraph::replication::ReplicationState &repl_state,
@@ -133,7 +138,7 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
   auto GetReplState() -> memgraph::replication::ReplicationState &;
 
  private:
-  template <bool HandleFailure>
+  template <bool AllowReplicaToDivergeFromMain>
   auto RegisterReplica_(const memgraph::replication::ReplicationClientConfig &config, bool send_swap_uuid)
       -> memgraph::utils::BasicResult<memgraph::query::RegisterReplicaError> {
     MG_ASSERT(repl_state_.IsMain(), "Only main instance can register a replica!");
@@ -155,7 +160,9 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
       }
     }
 
-    if (!memgraph::dbms::allow_mt_repl && dbms_handler_.All().size() > 1) {
+    using enum memgraph::flags::Experiments;
+    bool system_replication_enabled = flags::AreExperimentsEnabled(SYSTEM_REPLICATION);
+    if (!system_replication_enabled && dbms_handler_.Count() > 1) {
       spdlog::warn("Multi-tenant replication is currently not supported!");
     }
     const auto main_uuid =
@@ -170,7 +177,7 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
 
 #ifdef MG_ENTERPRISE
     // Update system before enabling individual storage <-> replica clients
-    SystemRestore(*maybe_client.GetValue(), dbms_handler_, main_uuid, system_, auth_);
+    SystemRestore(*maybe_client.GetValue(), system_, dbms_handler_, main_uuid, auth_);
 #endif
 
     const auto dbms_error = HandleRegisterReplicaStatus(maybe_client);
@@ -183,7 +190,7 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
     // Add database specific clients (NOTE Currently all databases are connected to each replica)
     dbms_handler_.ForEach([&](dbms::DatabaseAccess db_acc) {
       auto *storage = db_acc->storage();
-      if (!dbms::allow_mt_repl && storage->name() != dbms::kDefaultDB) {
+      if (!system_replication_enabled && storage->name() != dbms::kDefaultDB) {
         return;
       }
       // TODO: ATM only IN_MEMORY_TRANSACTIONAL, fix other modes
@@ -193,13 +200,15 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
           [storage, &instance_client_ptr, db_acc = std::move(db_acc),
            main_uuid](auto &storage_clients) mutable {  // NOLINT
             auto client = std::make_unique<storage::ReplicationStorageClient>(*instance_client_ptr, main_uuid);
-            // All good, start replica client
             client->Start(storage, std::move(db_acc));
-            // After start the storage <-> replica state should be READY or RECOVERING (if correctly started)
-            // MAYBE_BEHIND isn't a statement of the current state, this is the default value
-            // Failed to start due an error like branching of MAIN and REPLICA
-            const bool success = client->State() != storage::replication::ReplicaState::MAYBE_BEHIND;
-            if (HandleFailure || success) {
+            bool const success = std::invoke([state = client->State()]() {
+              if (state == storage::replication::ReplicaState::DIVERGED_FROM_MAIN) {
+                return AllowReplicaToDivergeFromMain;
+              }
+              return state != storage::replication::ReplicaState::MAYBE_BEHIND;
+            });
+
+            if (success) {
               storage_clients.push_back(std::move(client));
             }
             return success;
@@ -207,7 +216,7 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
     });
 
     // NOTE Currently if any databases fails, we revert back
-    if (!HandleFailure && !all_clients_good) {
+    if (!all_clients_good) {
       spdlog::error("Failed to register all databases on the REPLICA \"{}\"", config.name);
       UnregisterReplica(config.name);
       return memgraph::query::RegisterReplicaError::CONNECTION_FAILED;
@@ -215,7 +224,7 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
 
     // No client error, start instance level client
 #ifdef MG_ENTERPRISE
-    StartReplicaClient(*instance_client_ptr, dbms_handler_, main_uuid, system_, auth_);
+    StartReplicaClient(*instance_client_ptr, system_, dbms_handler_, main_uuid, auth_);
 #else
     StartReplicaClient(*instance_client_ptr, dbms_handler_, main_uuid);
 #endif
@@ -226,7 +235,7 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
   memgraph::dbms::DbmsHandler &dbms_handler_;
 
 #ifdef MG_ENTERPRISE
-  memgraph::system::System *system_;
+  memgraph::system::System &system_;
   memgraph::auth::SynchedAuth &auth_;
 #endif
 };

src/replication_handler/include/replication_handler/system_replication.hpp

@@ -27,11 +27,16 @@ inline void LogWrongMain(const std::optional<utils::UUID> &current_main_uuid, co
 #ifdef MG_ENTERPRISE
 void SystemHeartbeatHandler(uint64_t ts, const std::optional<utils::UUID> &current_main_uuid, slk::Reader *req_reader,
                             slk::Builder *res_builder);
+
 void SystemRecoveryHandler(memgraph::system::ReplicaHandlerAccessToState &system_state_access,
                            std::optional<utils::UUID> &current_main_uuid, dbms::DbmsHandler &dbms_handler,
                            auth::SynchedAuth &auth, slk::Reader *req_reader, slk::Builder *res_builder);
-void Register(replication::RoleReplicaData const &data, dbms::DbmsHandler &dbms_handler, auth::SynchedAuth &auth);
-bool StartRpcServer(dbms::DbmsHandler &dbms_handler, replication::RoleReplicaData &data, auth::SynchedAuth &auth);
+
+void Register(replication::RoleReplicaData const &data, system::System &system, dbms::DbmsHandler &dbms_handler,
+              auth::SynchedAuth &auth);
+
+bool StartRpcServer(dbms::DbmsHandler &dbms_handler, replication::RoleReplicaData &data, auth::SynchedAuth &auth,
+                    system::System &system);
 #else
 bool StartRpcServer(dbms::DbmsHandler &dbms_handler, replication::RoleReplicaData &data);
 #endif

src/replication_handler/replication_handler.cpp

@@ -17,25 +17,25 @@ namespace memgraph::replication {
 
 namespace {
 #ifdef MG_ENTERPRISE
-void RecoverReplication(memgraph::replication::ReplicationState &repl_state, memgraph::system::System *system,
+void RecoverReplication(memgraph::replication::ReplicationState &repl_state, memgraph::system::System &system,
                         memgraph::dbms::DbmsHandler &dbms_handler, memgraph::auth::SynchedAuth &auth) {
   /*
    * REPLICATION RECOVERY AND STARTUP
    */
 
   // Startup replication state (if recovered at startup)
-  auto replica = [&dbms_handler, &auth](memgraph::replication::RoleReplicaData &data) {
-    return StartRpcServer(dbms_handler, data, auth);
+  auto replica = [&dbms_handler, &auth, &system](memgraph::replication::RoleReplicaData &data) {
+    return memgraph::replication::StartRpcServer(dbms_handler, data, auth, system);
   };
 
   // Replication recovery and frequent check start
-  auto main = [system, &dbms_handler, &auth](memgraph::replication::RoleMainData &mainData) {
+  auto main = [&system, &dbms_handler, &auth](memgraph::replication::RoleMainData &mainData) {
     for (auto &client : mainData.registered_replicas_) {
       if (client.try_set_uuid &&
           replication_coordination_glue::SendSwapMainUUIDRpc(client.rpc_client_, mainData.uuid_)) {
         client.try_set_uuid = false;
       }
-      SystemRestore(client, dbms_handler, mainData.uuid_, system, auth);
+      SystemRestore(client, system, dbms_handler, mainData.uuid_, auth);
     }
     // DBMS here
     dbms_handler.ForEach([&mainData](memgraph::dbms::DatabaseAccess db_acc) {
@@ -43,7 +43,7 @@ void RecoverReplication(memgraph::replication::ReplicationState &repl_state, mem
     });
 
     for (auto &client : mainData.registered_replicas_) {
-      StartReplicaClient(client, dbms_handler, mainData.uuid_, system, auth);
+      StartReplicaClient(client, system, dbms_handler, mainData.uuid_, auth);
     }
 
     // Warning
@@ -120,8 +120,8 @@ inline std::optional<query::RegisterReplicaError> HandleRegisterReplicaStatus(
 }
 
 #ifdef MG_ENTERPRISE
-void StartReplicaClient(replication::ReplicationClient &client, dbms::DbmsHandler &dbms_handler, utils::UUID main_uuid,
-                        system::System *system, auth::SynchedAuth &auth) {
+void StartReplicaClient(replication::ReplicationClient &client, system::System &system, dbms::DbmsHandler &dbms_handler,
+                        utils::UUID main_uuid, auth::SynchedAuth &auth) {
 #else
 void StartReplicaClient(replication::ReplicationClient &client, dbms::DbmsHandler &dbms_handler,
                         utils::UUID main_uuid) {
@@ -129,12 +129,8 @@ void StartReplicaClient(replication::ReplicationClient &client, dbms::DbmsHandle
   // No client error, start instance level client
   auto const &endpoint = client.rpc_client_.Endpoint();
   spdlog::trace("Replication client started at: {}:{}", endpoint.address, endpoint.port);
-  client.StartFrequentCheck([&,
-#ifdef MG_ENTERPRISE
-                             system = system,
-#endif
-                             license = license::global_license_checker.IsEnterpriseValidFast(),
-                             main_uuid](bool reconnect, replication::ReplicationClient &client) mutable {
+  client.StartFrequentCheck([&, license = license::global_license_checker.IsEnterpriseValidFast(), main_uuid](
+                                bool reconnect, replication::ReplicationClient &client) mutable {
     if (client.try_set_uuid &&
         memgraph::replication_coordination_glue::SendSwapMainUUIDRpc(client.rpc_client_, main_uuid)) {
       client.try_set_uuid = false;
@@ -151,7 +147,7 @@ void StartReplicaClient(replication::ReplicationClient &client, dbms::DbmsHandle
       client.state_.WithLock([](auto &state) { state = memgraph::replication::ReplicationClient::State::BEHIND; });
     }
 #ifdef MG_ENTERPRISE
-    SystemRestore<true>(client, dbms_handler, main_uuid, system, auth);
+    SystemRestore<true>(client, system, dbms_handler, main_uuid, auth);
 #endif
     // Check if any database has been left behind
     dbms_handler.ForEach([&name = client.name_, reconnect](dbms::DatabaseAccess db_acc) {
@@ -168,7 +164,7 @@ void StartReplicaClient(replication::ReplicationClient &client, dbms::DbmsHandle
 
 #ifdef MG_ENTERPRISE
 ReplicationHandler::ReplicationHandler(memgraph::replication::ReplicationState &repl_state,
-                                       memgraph::dbms::DbmsHandler &dbms_handler, memgraph::system::System *system,
+                                       memgraph::dbms::DbmsHandler &dbms_handler, memgraph::system::System &system,
                                        memgraph::auth::SynchedAuth &auth)
     : repl_state_{repl_state}, dbms_handler_{dbms_handler}, system_{system}, auth_{auth} {
   RecoverReplication(repl_state_, system_, dbms_handler_, auth_);
@@ -216,18 +212,19 @@ bool ReplicationHandler::SetReplicationRoleReplica(const memgraph::replication::
   repl_state_.SetReplicationRoleReplica(config, main_uuid);
 
   // Start
-  const auto success = std::visit(memgraph::utils::Overloaded{[](memgraph::replication::RoleMainData &) {
-                                                                // ASSERT
-                                                                return false;
-                                                              },
-                                                              [this](memgraph::replication::RoleReplicaData &data) {
+  const auto success =
+      std::visit(memgraph::utils::Overloaded{[](memgraph::replication::RoleMainData &) {
+                                               // ASSERT
+                                               return false;
+                                             },
+                                             [this](memgraph::replication::RoleReplicaData &data) {
 #ifdef MG_ENTERPRISE
-                                                                return StartRpcServer(dbms_handler_, data, auth_);
+                                               return StartRpcServer(dbms_handler_, data, auth_, system_);
 #else
-                                                                return StartRpcServer(dbms_handler_, data);
+                                               return StartRpcServer(dbms_handler_, data);
 #endif
-                                                              }},
-                                  repl_state_.ReplicationData());
+                                             }},
+                 repl_state_.ReplicationData());
   // TODO Handle error (restore to main?)
   return success;
 }

src/replication_handler/system_replication.cpp

@@ -15,6 +15,7 @@
 
 #include "auth/replication_handlers.hpp"
 #include "dbms/replication_handlers.hpp"
+#include "flags/experimental.hpp"
 #include "license/license.hpp"
 #include "replication_handler/system_rpc.hpp"
 
@@ -56,10 +57,24 @@ void SystemRecoveryHandler(memgraph::system::ReplicaHandlerAccessToState &system
   memgraph::replication::SystemRecoveryReq req;
   memgraph::slk::Load(&req, req_reader);
 
+  using enum memgraph::flags::Experiments;
+  auto experimental_system_replication = flags::AreExperimentsEnabled(SYSTEM_REPLICATION);
+
+  // validate
   if (!current_main_uuid.has_value() || req.main_uuid != current_main_uuid) [[unlikely]] {
     LogWrongMain(current_main_uuid, req.main_uuid, SystemRecoveryReq::kType.name);
     return;
   }
+  if (!experimental_system_replication) {
+    if (req.database_configs.size() != 1 && req.database_configs[0].name != dbms::kDefaultDB) {
+      // a partial system recovery should be only be updating the default database uuid
+      return;  // Failure sent on exit
+    }
+    if (!req.users.empty() || !req.roles.empty()) {
+      // a partial system recovery should not be updating any users or roles
+      return;  // Failure sent on exit
+    }
+  }
 
   /*
    * DBMS
@@ -69,7 +84,9 @@ void SystemRecoveryHandler(memgraph::system::ReplicaHandlerAccessToState &system
   /*
    * AUTH
    */
-  if (!auth::SystemRecoveryHandler(auth, req.auth_config, req.users, req.roles)) return;  // Failure sent on exit
+  if (experimental_system_replication) {
+    if (!auth::SystemRecoveryHandler(auth, req.auth_config, req.users, req.roles)) return;  // Failure sent on exit
+  }
 
   /*
    * SUCCESSFUL RECOVERY
@@ -79,35 +96,44 @@ void SystemRecoveryHandler(memgraph::system::ReplicaHandlerAccessToState &system
   res = SystemRecoveryRes(SystemRecoveryRes::Result::SUCCESS);
 }
 
-void Register(replication::RoleReplicaData const &data, dbms::DbmsHandler &dbms_handler, auth::SynchedAuth &auth) {
+void Register(replication::RoleReplicaData const &data, system::System &system, dbms::DbmsHandler &dbms_handler,
+              auth::SynchedAuth &auth) {
   // NOTE: Register even without license as the user could add a license at run-time
-  // TODO: fix Register when system is removed from DbmsHandler
 
-  auto system_state_access = dbms_handler.system_->CreateSystemStateAccess();
+  auto system_state_access = system.CreateSystemStateAccess();
+
+  using enum memgraph::flags::Experiments;
+  auto experimental_system_replication = flags::AreExperimentsEnabled(SYSTEM_REPLICATION);
 
   // System
-  // TODO: remove, as this is not used
-  data.server->rpc_server_.Register<replication::SystemHeartbeatRpc>(
-      [&data, system_state_access](auto *req_reader, auto *res_builder) {
-        spdlog::debug("Received SystemHeartbeatRpc");
-        SystemHeartbeatHandler(system_state_access.LastCommitedTS(), data.uuid_, req_reader, res_builder);
-      });
+  if (experimental_system_replication) {
+    data.server->rpc_server_.Register<replication::SystemHeartbeatRpc>(
+        [&data, system_state_access](auto *req_reader, auto *res_builder) {
+          spdlog::debug("Received SystemHeartbeatRpc");
+          SystemHeartbeatHandler(system_state_access.LastCommitedTS(), data.uuid_, req_reader, res_builder);
+        });
+  }
+  // Needed even with experimental_system_replication=false becasue
+  // need to tell REPLICA the uuid to use for "memgraph" default database
   data.server->rpc_server_.Register<replication::SystemRecoveryRpc>(
       [&data, system_state_access, &dbms_handler, &auth](auto *req_reader, auto *res_builder) mutable {
         spdlog::debug("Received SystemRecoveryRpc");
         SystemRecoveryHandler(system_state_access, data.uuid_, dbms_handler, auth, req_reader, res_builder);
       });
 
-  // DBMS
-  dbms::Register(data, system_state_access, dbms_handler);
+  if (experimental_system_replication) {
+    // DBMS
+    dbms::Register(data, system_state_access, dbms_handler);
 
-  // Auth
-  auth::Register(data, system_state_access, auth);
+    // Auth
+    auth::Register(data, system_state_access, auth);
+  }
 }
 #endif
 
 #ifdef MG_ENTERPRISE
-bool StartRpcServer(dbms::DbmsHandler &dbms_handler, replication::RoleReplicaData &data, auth::SynchedAuth &auth) {
+bool StartRpcServer(dbms::DbmsHandler &dbms_handler, replication::RoleReplicaData &data, auth::SynchedAuth &auth,
+                    system::System &system) {
 #else
 bool StartRpcServer(dbms::DbmsHandler &dbms_handler, replication::RoleReplicaData &data) {
 #endif
@@ -115,7 +141,7 @@ bool StartRpcServer(dbms::DbmsHandler &dbms_handler, replication::RoleReplicaDat
   dbms::InMemoryReplicationHandlers::Register(&dbms_handler, data);
 #ifdef MG_ENTERPRISE
   // Register system handlers
-  Register(data, dbms_handler, auth);
+  Register(data, system, dbms_handler, auth);
 #endif
   // Start server
   if (!data.server->Start()) {

src/storage/v2/replication/enums.hpp

@@ -1,4 +1,4 @@
-// Copyright 2023 Memgraph Ltd.
+// Copyright 2024 Memgraph Ltd.
 //
 // Use of this software is governed by the Business Source License
 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -14,6 +14,6 @@
 
 namespace memgraph::storage::replication {
 
-enum class ReplicaState : std::uint8_t { READY, REPLICATING, RECOVERY, MAYBE_BEHIND };
+enum class ReplicaState : std::uint8_t { READY, REPLICATING, RECOVERY, MAYBE_BEHIND, DIVERGED_FROM_MAIN };
 
 }  // namespace memgraph::storage::replication

src/storage/v2/replication/replication_client.cpp

@@ -69,7 +69,7 @@ void ReplicationStorageClient::UpdateReplicaState(Storage *storage, DatabaseAcce
         "now hold unique data. Please resolve data conflicts and start the "
         "replication on a clean instance.",
         client_.name_, client_.name_, client_.name_);
-    // State not updated, hence in MAYBE_BEHIND state
+    replica_state_.WithLock([](auto &val) { val = replication::ReplicaState::DIVERGED_FROM_MAIN; });
     return;
   }
 
@@ -171,6 +171,10 @@ void ReplicationStorageClient::StartTransactionReplication(const uint64_t curren
           utils::MessageWithLink("Couldn't replicate data to {}.", client_.name_, "https://memgr.ph/replication"));
       TryCheckReplicaStateAsync(storage, std::move(db_acc));
       return;
+    case DIVERGED_FROM_MAIN:
+      spdlog::error(utils::MessageWithLink("Couldn't replicate data to {} since replica has diverged from main.",
+                                           client_.name_, "https://memgr.ph/replication"));
+      return;
     case READY:
       MG_ASSERT(!replica_stream_);
       try {

src/system/include/system/action.hpp

@@ -26,12 +26,14 @@ struct ISystemAction {
   /// Durability step which is defered until commit time
   virtual void DoDurability() = 0;
 
+#ifdef MG_ENTERPRISE
   /// Prepare the RPC payload that will be sent to all replicas clients
   virtual bool DoReplication(memgraph::replication::ReplicationClient &client, const utils::UUID &main_uuid,
                              memgraph::replication::ReplicationEpoch const &epoch,
                              Transaction const &system_tx) const = 0;
 
   virtual void PostReplication(memgraph::replication::RoleMainData &main_data) const = 0;
+#endif
 
   virtual ~ISystemAction() = default;
 };

src/system/include/system/transaction.hpp

@@ -57,11 +57,13 @@ struct Transaction {
       /// durability
       action->DoDurability();
 
-      /// replication prep
+#ifdef MG_ENTERPRISE
+      /// replication
       auto action_sync_status = handler.ApplyAction(*action, *this);
       if (action_sync_status != AllSyncReplicaStatus::AllCommitsConfirmed) {
         sync_status = AllSyncReplicaStatus::SomeCommitsUnconfirmed;
       }
+#endif
 
       actions_.pop_front();
     }
@@ -93,6 +95,7 @@ struct Transaction {
   std::list<std::unique_ptr<ISystemAction>> actions_;
 };
 
+#ifdef MG_ENTERPRISE
 struct DoReplication {
   explicit DoReplication(replication::RoleMainData &main_data) : main_data_{main_data} {}
   auto ApplyAction(ISystemAction const &action, Transaction const &system_tx) -> AllSyncReplicaStatus {
@@ -113,6 +116,7 @@ struct DoReplication {
   replication::RoleMainData &main_data_;
 };
 static_assert(ReplicationPolicy<DoReplication>);
+#endif
 
 struct DoNothing {
   auto ApplyAction(ISystemAction const & /*action*/, Transaction const & /*system_tx*/) -> AllSyncReplicaStatus {

src/utils/functional.hpp

@@ -11,12 +11,17 @@
 
 #pragma once
 
-#ifdef MG_ENTERPRISE
-namespace memgraph::coordination {
+#include <algorithm>
+#include <vector>
 
-struct CoordinatorClusterConfig {
-  static constexpr int alive_response_time_difference_sec_{5};
-};
+namespace memgraph::utils {
 
-}  // namespace memgraph::coordination
-#endif
+template <class F, class T, class R = typename std::result_of<F(T)>::type, class V = std::vector<R>>
+V fmap(F &&f, const std::vector<T> &v) {
+  V r;
+  r.reserve(v.size());
+  std::ranges::transform(v, std::back_inserter(r), std::forward<F>(f));
+  return r;
+}
+
+}  // namespace memgraph::utils

src/utils/typeinfo.hpp

@@ -107,6 +107,10 @@ enum class TypeId : uint64_t {
   COORD_SET_REPL_MAIN_RES,
   COORD_SWAP_UUID_REQ,
   COORD_SWAP_UUID_RES,
+  COORD_UNREGISTER_REPLICA_REQ,
+  COORD_UNREGISTER_REPLICA_RES,
+  COORD_ENABLE_WRITING_ON_MAIN_REQ,
+  COORD_ENABLE_WRITING_ON_MAIN_RES,
 
   // AST
   AST_LABELIX = 3000,

tests/e2e/CMakeLists.txt

@@ -56,7 +56,6 @@ add_subdirectory(python_query_modules_reloading)
 add_subdirectory(analyze_graph)
 add_subdirectory(transaction_queue)
 add_subdirectory(mock_api)
-#add_subdirectory(graphql)
 add_subdirectory(disk_storage)
 add_subdirectory(load_csv)
 add_subdirectory(init_file_flags)
@@ -81,10 +80,7 @@ if (MG_EXPERIMENTAL_HIGH_AVAILABILITY)
   add_subdirectory(high_availability_experimental)
 endif ()
 
-
-if (MG_EXPERIMENTAL_REPLICATION_MULTITENANCY)
-  add_subdirectory(replication_experimental)
-endif ()
+add_subdirectory(replication_experimental)
 
 copy_e2e_python_files(pytest_runner pytest_runner.sh "")
 copy_e2e_python_files(x x.sh "")

tests/e2e/configuration/default_config.py

@@ -69,6 +69,8 @@ startup_config_dict = {
     "coordinator_server_port": ("0", "0", "Port on which coordinator servers will be started."),
     "raft_server_port": ("0", "0", "Port on which raft servers will be started."),
     "raft_server_id": ("0", "0", "Unique ID of the raft server."),
+    "instance_down_timeout_sec": ("5", "5", "Time duration after which an instance is considered down."),
+    "instance_health_check_frequency_sec": ("1", "1", "The time duration between two health checks/pings."),
     "data_directory": ("mg_data", "mg_data", "Path to directory in which to save all permanent data."),
     "data_recovery_on_startup": (
         "false",
@@ -222,4 +224,9 @@ startup_config_dict = {
         "128",
         "The threshold for when to cache long delta chains. This is used for heavy read + write workloads where repeated processing of delta chains can become costly.",
     ),
+    "experimental_enabled": (
+        "",
+        "",
+        "Experimental features to be used, comma seperated. Options [system-replication]",
+    ),
 }

tests/e2e/high_availability_experimental/CMakeLists.txt

@@ -6,6 +6,7 @@ copy_e2e_python_files(ha_experimental coord_cluster_registration.py)
 copy_e2e_python_files(ha_experimental distributed_coords.py)
 copy_e2e_python_files(ha_experimental manual_setting_replicas.py)
 copy_e2e_python_files(ha_experimental not_replicate_from_old_main.py)
+copy_e2e_python_files(ha_experimental disable_writing_on_main_after_restart.py)
 copy_e2e_python_files(ha_experimental common.py)
 copy_e2e_python_files(ha_experimental workloads.yaml)
 

tests/e2e/high_availability_experimental/common.py

@@ -30,3 +30,14 @@ def safe_execute(function, *args):
         function(*args)
     except:
         pass
+
+
+# NOTE: Repeated execution because it can fail if Raft server is not up
+def add_coordinator(cursor, query):
+    for _ in range(10):
+        try:
+            execute_and_fetch_all(cursor, query)
+            return True
+        except Exception:
+            pass
+    return False

tests/e2e/high_availability_experimental/coord_cluster_registration.py

@@ -16,7 +16,7 @@ import tempfile
 
 import interactive_mg_runner
 import pytest
-from common import connect, execute_and_fetch_all, safe_execute
+from common import add_coordinator, connect, execute_and_fetch_all, safe_execute
 from mg_utils import mg_sleep_and_assert
 
 interactive_mg_runner.SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
@@ -104,17 +104,6 @@ MEMGRAPH_INSTANCES_DESCRIPTION = {
 }
 
 
-# NOTE: Repeated execution because it can fail if Raft server is not up
-def add_coordinator(cursor, query):
-    for _ in range(10):
-        try:
-            execute_and_fetch_all(cursor, query)
-            return True
-        except Exception:
-            pass
-    return False
-
-
 def test_register_repl_instances_then_coordinators():
     safe_execute(shutil.rmtree, TEMP_DIR)
     interactive_mg_runner.start_all(MEMGRAPH_INSTANCES_DESCRIPTION)
@@ -280,5 +269,148 @@ def test_coordinators_communication_with_restarts():
     mg_sleep_and_assert(expected_cluster_not_shared, check_coordinator2)
 
 
+# TODO: (andi) Test when dealing with distributed coordinators that you can register on one coordinator and unregister from any other coordinator
+@pytest.mark.parametrize(
+    "kill_instance",
+    [True, False],
+)
+def test_unregister_replicas(kill_instance):
+    safe_execute(shutil.rmtree, TEMP_DIR)
+    interactive_mg_runner.start_all(MEMGRAPH_INSTANCES_DESCRIPTION)
+
+    coordinator3_cursor = connect(host="localhost", port=7692).cursor()
+    execute_and_fetch_all(
+        coordinator3_cursor, "REGISTER INSTANCE instance_1 ON '127.0.0.1:10011' WITH '127.0.0.1:10001'"
+    )
+    execute_and_fetch_all(
+        coordinator3_cursor, "REGISTER INSTANCE instance_2 ON '127.0.0.1:10012' WITH '127.0.0.1:10002'"
+    )
+    execute_and_fetch_all(
+        coordinator3_cursor, "REGISTER INSTANCE instance_3 ON '127.0.0.1:10013' WITH '127.0.0.1:10003'"
+    )
+    execute_and_fetch_all(coordinator3_cursor, "SET INSTANCE instance_3 TO MAIN")
+
+    def check_coordinator3():
+        return sorted(list(execute_and_fetch_all(coordinator3_cursor, "SHOW INSTANCES")))
+
+    main_cursor = connect(host="localhost", port=7689).cursor()
+
+    def check_main():
+        return sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS")))
+
+    expected_cluster = [
+        ("coordinator_3", "127.0.0.1:10113", "", True, "coordinator"),
+        ("instance_1", "", "127.0.0.1:10011", True, "replica"),
+        ("instance_2", "", "127.0.0.1:10012", True, "replica"),
+        ("instance_3", "", "127.0.0.1:10013", True, "main"),
+    ]
+
+    expected_replicas = [
+        ("instance_1", "127.0.0.1:10001", "sync", 0, 0, "ready"),
+        ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
+    ]
+
+    mg_sleep_and_assert(expected_cluster, check_coordinator3)
+    mg_sleep_and_assert(expected_replicas, check_main)
+
+    if kill_instance:
+        interactive_mg_runner.kill(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_1")
+    execute_and_fetch_all(coordinator3_cursor, "UNREGISTER INSTANCE instance_1")
+
+    expected_cluster = [
+        ("coordinator_3", "127.0.0.1:10113", "", True, "coordinator"),
+        ("instance_2", "", "127.0.0.1:10012", True, "replica"),
+        ("instance_3", "", "127.0.0.1:10013", True, "main"),
+    ]
+
+    expected_replicas = [
+        ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
+    ]
+
+    mg_sleep_and_assert(expected_cluster, check_coordinator3)
+    mg_sleep_and_assert(expected_replicas, check_main)
+
+    if kill_instance:
+        interactive_mg_runner.kill(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_2")
+    execute_and_fetch_all(coordinator3_cursor, "UNREGISTER INSTANCE instance_2")
+
+    expected_cluster = [
+        ("coordinator_3", "127.0.0.1:10113", "", True, "coordinator"),
+        ("instance_3", "", "127.0.0.1:10013", True, "main"),
+    ]
+    expected_replicas = []
+
+    mg_sleep_and_assert(expected_cluster, check_coordinator3)
+    mg_sleep_and_assert(expected_replicas, check_main)
+
+
+def test_unregister_main():
+    safe_execute(shutil.rmtree, TEMP_DIR)
+    interactive_mg_runner.start_all(MEMGRAPH_INSTANCES_DESCRIPTION)
+
+    coordinator3_cursor = connect(host="localhost", port=7692).cursor()
+    execute_and_fetch_all(
+        coordinator3_cursor, "REGISTER INSTANCE instance_1 ON '127.0.0.1:10011' WITH '127.0.0.1:10001'"
+    )
+    execute_and_fetch_all(
+        coordinator3_cursor, "REGISTER INSTANCE instance_2 ON '127.0.0.1:10012' WITH '127.0.0.1:10002'"
+    )
+    execute_and_fetch_all(
+        coordinator3_cursor, "REGISTER INSTANCE instance_3 ON '127.0.0.1:10013' WITH '127.0.0.1:10003'"
+    )
+    execute_and_fetch_all(coordinator3_cursor, "SET INSTANCE instance_3 TO MAIN")
+
+    def check_coordinator3():
+        return sorted(list(execute_and_fetch_all(coordinator3_cursor, "SHOW INSTANCES")))
+
+    expected_cluster = [
+        ("coordinator_3", "127.0.0.1:10113", "", True, "coordinator"),
+        ("instance_1", "", "127.0.0.1:10011", True, "replica"),
+        ("instance_2", "", "127.0.0.1:10012", True, "replica"),
+        ("instance_3", "", "127.0.0.1:10013", True, "main"),
+    ]
+
+    mg_sleep_and_assert(expected_cluster, check_coordinator3)
+
+    try:
+        execute_and_fetch_all(coordinator3_cursor, "UNREGISTER INSTANCE instance_3")
+    except Exception as e:
+        assert (
+            str(e)
+            == "Alive main instance can't be unregistered! Shut it down to trigger failover and then unregister it!"
+        )
+
+    interactive_mg_runner.kill(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_3")
+
+    expected_cluster = [
+        ("coordinator_3", "127.0.0.1:10113", "", True, "coordinator"),
+        ("instance_1", "", "127.0.0.1:10011", True, "main"),
+        ("instance_2", "", "127.0.0.1:10012", True, "replica"),
+        ("instance_3", "", "127.0.0.1:10013", False, "unknown"),
+    ]
+
+    mg_sleep_and_assert(expected_cluster, check_coordinator3)
+
+    execute_and_fetch_all(coordinator3_cursor, "UNREGISTER INSTANCE instance_3")
+
+    expected_cluster = [
+        ("coordinator_3", "127.0.0.1:10113", "", True, "coordinator"),
+        ("instance_1", "", "127.0.0.1:10011", True, "main"),
+        ("instance_2", "", "127.0.0.1:10012", True, "replica"),
+    ]
+
+    expected_replicas = [
+        ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
+    ]
+
+    main_cursor = connect(host="localhost", port=7687).cursor()
+
+    def check_main():
+        return sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS")))
+
+    mg_sleep_and_assert(expected_cluster, check_coordinator3)
+    mg_sleep_and_assert(expected_replicas, check_main)
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/high_availability_experimental/disable_writing_on_main_after_restart.py

@@ -0,0 +1,181 @@
+# Copyright 2022 Memgraph Ltd.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+# License, and you may not use this file except in compliance with the Business Source License.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0, included in the file
+# licenses/APL.txt.
+
+import os
+import shutil
+import sys
+import tempfile
+
+import interactive_mg_runner
+import pytest
+from common import add_coordinator, connect, execute_and_fetch_all, safe_execute
+from mg_utils import mg_sleep_and_assert
+
+interactive_mg_runner.SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
+interactive_mg_runner.PROJECT_DIR = os.path.normpath(
+    os.path.join(interactive_mg_runner.SCRIPT_DIR, "..", "..", "..", "..")
+)
+interactive_mg_runner.BUILD_DIR = os.path.normpath(os.path.join(interactive_mg_runner.PROJECT_DIR, "build"))
+interactive_mg_runner.MEMGRAPH_BINARY = os.path.normpath(os.path.join(interactive_mg_runner.BUILD_DIR, "memgraph"))
+
+TEMP_DIR = tempfile.TemporaryDirectory().name
+
+MEMGRAPH_INSTANCES_DESCRIPTION = {
+    "instance_1": {
+        "args": [
+            "--bolt-port",
+            "7687",
+            "--log-level",
+            "TRACE",
+            "--coordinator-server-port",
+            "10011",
+            "--also-log-to-stderr",
+            "--instance-health-check-frequency-sec",
+            "1",
+            "--instance-down-timeout-sec",
+            "5",
+        ],
+        "log_file": "instance_1.log",
+        "data_directory": f"{TEMP_DIR}/instance_1",
+        "setup_queries": [],
+    },
+    "instance_2": {
+        "args": [
+            "--bolt-port",
+            "7688",
+            "--log-level",
+            "TRACE",
+            "--coordinator-server-port",
+            "10012",
+            "--also-log-to-stderr",
+            "--instance-health-check-frequency-sec",
+            "1",
+            "--instance-down-timeout-sec",
+            "5",
+        ],
+        "log_file": "instance_2.log",
+        "data_directory": f"{TEMP_DIR}/instance_2",
+        "setup_queries": [],
+    },
+    "instance_3": {
+        "args": [
+            "--bolt-port",
+            "7689",
+            "--log-level",
+            "TRACE",
+            "--coordinator-server-port",
+            "10013",
+            "--also-log-to-stderr",
+            "--instance-health-check-frequency-sec",
+            "5",
+            "--instance-down-timeout-sec",
+            "10",
+        ],
+        "log_file": "instance_3.log",
+        "data_directory": f"{TEMP_DIR}/instance_3",
+        "setup_queries": [],
+    },
+    "coordinator_1": {
+        "args": [
+            "--bolt-port",
+            "7690",
+            "--log-level=TRACE",
+            "--raft-server-id=1",
+            "--raft-server-port=10111",
+        ],
+        "log_file": "coordinator1.log",
+        "setup_queries": [],
+    },
+    "coordinator_2": {
+        "args": [
+            "--bolt-port",
+            "7691",
+            "--log-level=TRACE",
+            "--raft-server-id=2",
+            "--raft-server-port=10112",
+        ],
+        "log_file": "coordinator2.log",
+        "setup_queries": [],
+    },
+    "coordinator_3": {
+        "args": [
+            "--bolt-port",
+            "7692",
+            "--log-level=TRACE",
+            "--raft-server-id=3",
+            "--raft-server-port=10113",
+            "--also-log-to-stderr",
+        ],
+        "log_file": "coordinator3.log",
+        "setup_queries": [],
+    },
+}
+
+
+def test_writing_disabled_on_main_restart():
+    safe_execute(shutil.rmtree, TEMP_DIR)
+    interactive_mg_runner.start_all(MEMGRAPH_INSTANCES_DESCRIPTION)
+
+    coordinator3_cursor = connect(host="localhost", port=7692).cursor()
+
+    execute_and_fetch_all(
+        coordinator3_cursor, "REGISTER INSTANCE instance_3 ON '127.0.0.1:10013' WITH '127.0.0.1:10003'"
+    )
+    execute_and_fetch_all(coordinator3_cursor, "SET INSTANCE instance_3 TO MAIN")
+    assert add_coordinator(coordinator3_cursor, "ADD COORDINATOR 1 ON '127.0.0.1:10111'")
+    assert add_coordinator(coordinator3_cursor, "ADD COORDINATOR 2 ON '127.0.0.1:10112'")
+
+    def check_coordinator3():
+        return sorted(list(execute_and_fetch_all(coordinator3_cursor, "SHOW INSTANCES")))
+
+    expected_cluster_coord3 = [
+        ("coordinator_1", "127.0.0.1:10111", "", True, "coordinator"),
+        ("coordinator_2", "127.0.0.1:10112", "", True, "coordinator"),
+        ("coordinator_3", "127.0.0.1:10113", "", True, "coordinator"),
+        ("instance_3", "", "127.0.0.1:10013", True, "main"),
+    ]
+    mg_sleep_and_assert(expected_cluster_coord3, check_coordinator3)
+
+    interactive_mg_runner.kill(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_3")
+
+    expected_cluster_coord3 = [
+        ("coordinator_1", "127.0.0.1:10111", "", True, "coordinator"),
+        ("coordinator_2", "127.0.0.1:10112", "", True, "coordinator"),
+        ("coordinator_3", "127.0.0.1:10113", "", True, "coordinator"),
+        ("instance_3", "", "127.0.0.1:10013", False, "unknown"),
+    ]
+
+    mg_sleep_and_assert(expected_cluster_coord3, check_coordinator3)
+
+    interactive_mg_runner.start(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_3")
+
+    try:
+        instance3_cursor = connect(host="localhost", port=7689).cursor()
+        execute_and_fetch_all(instance3_cursor, "CREATE (n:Node {name: 'node'})")
+    except Exception as e:
+        assert (
+            str(e)
+            == "Write query forbidden on the main! Coordinator needs to enable writing on main by sending RPC message."
+        )
+
+    expected_cluster_coord3 = [
+        ("coordinator_1", "127.0.0.1:10111", "", True, "coordinator"),
+        ("coordinator_2", "127.0.0.1:10112", "", True, "coordinator"),
+        ("coordinator_3", "127.0.0.1:10113", "", True, "coordinator"),
+        ("instance_3", "", "127.0.0.1:10013", True, "main"),
+    ]
+
+    mg_sleep_and_assert(expected_cluster_coord3, check_coordinator3)
+    execute_and_fetch_all(instance3_cursor, "CREATE (n:Node {name: 'node'})")
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/high_availability_experimental/workloads.yaml

@@ -44,6 +44,10 @@ workloads:
     binary: "tests/e2e/pytest_runner.sh"
     args: ["high_availability_experimental/not_replicate_from_old_main.py"]
 
+  - name: "Disable writing on main after restart"
+    binary: "tests/e2e/pytest_runner.sh"
+    args: ["high_availability_experimental/disable_writing_on_main_after_restart.py"]
+
   - name: "Distributed coordinators"
     binary: "tests/e2e/pytest_runner.sh"
     args: ["high_availability_experimental/distributed_coords.py"]

tests/e2e/replication_experimental/auth.py

@@ -147,6 +147,7 @@ def test_auth_queries_on_replica(connection):
     MEMGRAPH_INSTANCES_DESCRIPTION_MANUAL = {
         "replica_1": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['replica_1']}",
                 "--log-level=TRACE",
@@ -160,6 +161,7 @@ def test_auth_queries_on_replica(connection):
         },
         "replica_2": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['replica_2']}",
                 "--log-level=TRACE",
@@ -173,6 +175,7 @@ def test_auth_queries_on_replica(connection):
         },
         "main": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['main']}",
                 "--log-level=TRACE",
@@ -213,6 +216,7 @@ def test_manual_users_recovery(connection):
     MEMGRAPH_INSTANCES_DESCRIPTION_MANUAL = {
         "replica_1": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['replica_1']}",
                 "--log-level=TRACE",
@@ -227,6 +231,7 @@ def test_manual_users_recovery(connection):
         },
         "replica_2": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['replica_2']}",
                 "--log-level=TRACE",
@@ -241,6 +246,7 @@ def test_manual_users_recovery(connection):
         },
         "main": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['main']}",
                 "--log-level=TRACE",
@@ -287,6 +293,7 @@ def test_env_users_recovery(connection):
     MEMGRAPH_INSTANCES_DESCRIPTION_MANUAL = {
         "replica_1": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['replica_1']}",
                 "--log-level=TRACE",
@@ -300,6 +307,7 @@ def test_env_users_recovery(connection):
         },
         "replica_2": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['replica_2']}",
                 "--log-level=TRACE",
@@ -315,6 +323,7 @@ def test_env_users_recovery(connection):
             "username": "user1",
             "password": "password",
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['main']}",
                 "--log-level=TRACE",
@@ -368,6 +377,7 @@ def test_manual_roles_recovery(connection):
     MEMGRAPH_INSTANCES_DESCRIPTION_MANUAL = {
         "replica_1": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['replica_1']}",
                 "--log-level=TRACE",
@@ -382,6 +392,7 @@ def test_manual_roles_recovery(connection):
         },
         "replica_2": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['replica_2']}",
                 "--log-level=TRACE",
@@ -398,6 +409,7 @@ def test_manual_roles_recovery(connection):
         },
         "main": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['main']}",
                 "--log-level=TRACE",
@@ -455,6 +467,7 @@ def test_auth_config_recovery(connection):
     MEMGRAPH_INSTANCES_DESCRIPTION_MANUAL = {
         "replica_1": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['replica_1']}",
                 "--log-level=TRACE",
@@ -475,6 +488,7 @@ def test_auth_config_recovery(connection):
         },
         "replica_2": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['replica_2']}",
                 "--log-level=TRACE",
@@ -497,6 +511,7 @@ def test_auth_config_recovery(connection):
         },
         "main": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['main']}",
                 "--log-level=TRACE",
@@ -550,6 +565,7 @@ def test_auth_replication(connection):
     MEMGRAPH_INSTANCES_DESCRIPTION_MANUAL = {
         "replica_1": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['replica_1']}",
                 "--log-level=TRACE",
@@ -563,6 +579,7 @@ def test_auth_replication(connection):
         },
         "replica_2": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['replica_2']}",
                 "--log-level=TRACE",
@@ -576,6 +593,7 @@ def test_auth_replication(connection):
         },
         "main": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['main']}",
                 "--log-level=TRACE",

tests/e2e/replication_experimental/multitenancy.py

@@ -39,6 +39,7 @@ def create_memgraph_instances_with_role_recovery(data_directory: Any) -> Dict[st
     return {
         "replica_1": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['replica_1']}",
                 "--log-level",
@@ -53,6 +54,7 @@ def create_memgraph_instances_with_role_recovery(data_directory: Any) -> Dict[st
         },
         "replica_2": {
             "args": [
+                "--experimental-enabled=system-replication",
                 "--bolt-port",
                 f"{BOLT_PORTS['replica_2']}",
                 "--log-level=TRACE",
@@ -65,7 +67,12 @@ def create_memgraph_instances_with_role_recovery(data_directory: Any) -> Dict[st
             "data_directory": f"{data_directory}/replica_2",
         },
         "main": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['main']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['main']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "main.log",
             "setup_queries": [],
         },
@@ -98,6 +105,7 @@ TEMP_DIR = tempfile.TemporaryDirectory().name
 MEMGRAPH_INSTANCES_DESCRIPTION_WITH_RECOVERY = {
     "replica_1": {
         "args": [
+            "--experimental-enabled=system-replication",
             "--bolt-port",
             f"{BOLT_PORTS['replica_1']}",
             "--log-level=TRACE",
@@ -109,6 +117,7 @@ MEMGRAPH_INSTANCES_DESCRIPTION_WITH_RECOVERY = {
     },
     "replica_2": {
         "args": [
+            "--experimental-enabled=system-replication",
             "--bolt-port",
             f"{BOLT_PORTS['replica_2']}",
             "--log-level=TRACE",
@@ -120,6 +129,7 @@ MEMGRAPH_INSTANCES_DESCRIPTION_WITH_RECOVERY = {
     },
     "main": {
         "args": [
+            "--experimental-enabled=system-replication",
             "--bolt-port",
             f"{BOLT_PORTS['main']}",
             "--log-level=TRACE",
@@ -206,7 +216,12 @@ def test_manual_databases_create_multitenancy_replication(connection):
 
     MEMGRAPH_INSTANCES_DESCRIPTION_MANUAL = {
         "replica_1": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['replica_1']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['replica_1']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "replica1.log",
             "setup_queries": [
                 "CREATE DATABASE A;",
@@ -215,7 +230,12 @@ def test_manual_databases_create_multitenancy_replication(connection):
             ],
         },
         "replica_2": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['replica_2']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['replica_2']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "replica2.log",
             "setup_queries": [
                 "CREATE DATABASE A;",
@@ -224,7 +244,12 @@ def test_manual_databases_create_multitenancy_replication(connection):
             ],
         },
         "main": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['main']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['main']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "main.log",
             "setup_queries": [
                 "CREATE DATABASE A;",
@@ -281,7 +306,12 @@ def test_manual_databases_create_multitenancy_replication_branching(connection):
 
     MEMGRAPH_INSTANCES_DESCRIPTION_MANUAL = {
         "replica_1": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['replica_1']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['replica_1']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "replica1.log",
             "setup_queries": [
                 "CREATE DATABASE A;",
@@ -294,7 +324,12 @@ def test_manual_databases_create_multitenancy_replication_branching(connection):
             ],
         },
         "replica_2": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['replica_2']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['replica_2']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "replica2.log",
             "setup_queries": [
                 "CREATE DATABASE A;",
@@ -307,7 +342,12 @@ def test_manual_databases_create_multitenancy_replication_branching(connection):
             ],
         },
         "main": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['main']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['main']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "main.log",
             "setup_queries": [
                 "CREATE DATABASE A;",
@@ -354,7 +394,12 @@ def test_manual_databases_create_multitenancy_replication_dirty_replica(connecti
 
     MEMGRAPH_INSTANCES_DESCRIPTION_MANUAL = {
         "replica_1": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['replica_1']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['replica_1']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "replica1.log",
             "setup_queries": [
                 "CREATE DATABASE A;",
@@ -364,7 +409,12 @@ def test_manual_databases_create_multitenancy_replication_dirty_replica(connecti
             ],
         },
         "replica_2": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['replica_2']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['replica_2']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "replica2.log",
             "setup_queries": [
                 "CREATE DATABASE A;",
@@ -374,7 +424,12 @@ def test_manual_databases_create_multitenancy_replication_dirty_replica(connecti
             ],
         },
         "main": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['main']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['main']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "main.log",
             "setup_queries": [
                 "CREATE DATABASE A;",
@@ -416,7 +471,12 @@ def test_manual_databases_create_multitenancy_replication_main_behind(connection
 
     MEMGRAPH_INSTANCES_DESCRIPTION_MANUAL = {
         "replica_1": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['replica_1']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['replica_1']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "replica1.log",
             "setup_queries": [
                 "CREATE DATABASE A;",
@@ -426,7 +486,12 @@ def test_manual_databases_create_multitenancy_replication_main_behind(connection
             ],
         },
         "replica_2": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['replica_2']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['replica_2']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "replica2.log",
             "setup_queries": [
                 "CREATE DATABASE A;",
@@ -436,7 +501,12 @@ def test_manual_databases_create_multitenancy_replication_main_behind(connection
             ],
         },
         "main": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['main']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['main']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "main.log",
             "setup_queries": [
                 f"REGISTER REPLICA replica_1 SYNC TO '127.0.0.1:{REPLICATION_PORTS['replica_1']}';",
@@ -532,14 +602,24 @@ def test_automatic_databases_multitenancy_replication_predefined(connection):
 
     MEMGRAPH_INSTANCES_DESCRIPTION_MANUAL = {
         "replica_1": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['replica_1']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['replica_1']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "replica1.log",
             "setup_queries": [
                 f"SET REPLICATION ROLE TO REPLICA WITH PORT {REPLICATION_PORTS['replica_1']};",
             ],
         },
         "main": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['main']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['main']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "main.log",
             "setup_queries": [
                 "CREATE DATABASE A;",
@@ -585,14 +665,24 @@ def test_automatic_databases_create_multitenancy_replication_dirty_main(connecti
 
     MEMGRAPH_INSTANCES_DESCRIPTION_MANUAL = {
         "replica_1": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['replica_1']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['replica_1']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "replica1.log",
             "setup_queries": [
                 f"SET REPLICATION ROLE TO REPLICA WITH PORT {REPLICATION_PORTS['replica_1']};",
             ],
         },
         "main": {
-            "args": ["--bolt-port", f"{BOLT_PORTS['main']}", "--log-level=TRACE"],
+            "args": [
+                "--experimental-enabled=system-replication",
+                "--bolt-port",
+                f"{BOLT_PORTS['main']}",
+                "--log-level=TRACE",
+            ],
             "log_file": "main.log",
             "setup_queries": [
                 "CREATE DATABASE A;",

tests/integration/telemetry/client.cpp

@@ -44,7 +44,7 @@ int main(int argc, char **argv) {
   memgraph::replication::ReplicationState repl_state(ReplicationStateRootPath(db_config));
 
   memgraph::system::System system_state;
-  memgraph::dbms::DbmsHandler dbms_handler(db_config, system_state, repl_state
+  memgraph::dbms::DbmsHandler dbms_handler(db_config, repl_state
 #ifdef MG_ENTERPRISE
                                            ,
                                            auth_, false

tests/unit/dbms_handler.cpp

@@ -71,8 +71,7 @@ class TestEnvironment : public ::testing::Environment {
                                                          memgraph::auth::Auth::Config{/* default */});
     system_state = std::make_unique<memgraph::system::System>();
     repl_state = std::make_unique<memgraph::replication::ReplicationState>(ReplicationStateRootPath(storage_conf));
-    ptr_ = std::make_unique<memgraph::dbms::DbmsHandler>(storage_conf, *system_state.get(), *repl_state.get(),
-                                                         *auth.get(), false);
+    ptr_ = std::make_unique<memgraph::dbms::DbmsHandler>(storage_conf, *repl_state.get(), *auth.get(), false);
   }
 
   void TearDown() override {

tests/unit/dbms_handler_community.cpp

@@ -55,7 +55,7 @@ class TestEnvironment : public ::testing::Environment {
                                                          memgraph::auth::Auth::Config{/* default */});
     system_state = std::make_unique<memgraph::system::System>();
     repl_state = std::make_unique<memgraph::replication::ReplicationState>(ReplicationStateRootPath(storage_conf));
-    ptr_ = std::make_unique<memgraph::dbms::DbmsHandler>(storage_conf, *system_state.get(), *repl_state.get());
+    ptr_ = std::make_unique<memgraph::dbms::DbmsHandler>(storage_conf, *repl_state.get());
   }
 
   void TearDown() override {

tests/unit/multi_tenancy.cpp

@@ -101,7 +101,7 @@ class MultiTenantTest : public ::testing::Test {
     explicit MinMemgraph(const memgraph::storage::Config &conf)
         : auth{conf.durability.storage_directory / "auth", memgraph::auth::Auth::Config{/* default */}},
           repl_state{ReplicationStateRootPath(conf)},
-          dbms{conf, system, repl_state, auth, true},
+          dbms{conf, repl_state, auth, true},
           interpreter_context{{},
                               &dbms,
                               &repl_state,

tests/unit/storage_v2_replication.cpp

@@ -113,7 +113,7 @@ struct MinMemgraph {
   MinMemgraph(const memgraph::storage::Config &conf)
       : auth{conf.durability.storage_directory / "auth", memgraph::auth::Auth::Config{/* default */}},
         repl_state{ReplicationStateRootPath(conf)},
-        dbms{conf, system_, repl_state
+        dbms{conf, repl_state
 #ifdef MG_ENTERPRISE
              ,
              auth, true
@@ -124,7 +124,7 @@ struct MinMemgraph {
         repl_handler(repl_state, dbms
 #ifdef MG_ENTERPRISE
                      ,
-                     &system_, auth
+                     system_, auth
 #endif
         ) {
   }