Override memgraph version to 2.15.0

- Remove the e2e that did concurrent mgp_* calls on the same transaction
  (ATM this is unsupported)
- Fix up the concurrent mgp_global_alloc test to be testing it more precisely
- Reduce the memory limit on detach delete test due to recent memory
  optimizations around deltas.
- No longer throw from hook, through jemalloc C, to our C++ on other
  side. This cause mutex unlocks to not happen.
- No longer allocate error messages while inside the hook. This caused
  recursive entry back inside jamalloc which would try to relock a
  non-recursive mutex.

.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 * * *"
 
@@ -321,7 +329,6 @@ jobs:
             --no-strict
 
   release_e2e_test:
-    if: false
     name: "Release End-to-end Test"
     runs-on: [self-hosted, Linux, X64, Debian10]
     timeout-minutes: 60

.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 * * *"
 
@@ -317,7 +325,6 @@ jobs:
             --no-strict
 
   release_e2e_test:
-    if: false
     name: "Release End-to-end Test"
     runs-on: [self-hosted, Linux, X64, Ubuntu20.04]
     timeout-minutes: 60

.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

@@ -64,7 +64,7 @@ option(MG_ENTERPRISE "Build Memgraph Enterprise Edition" ON)
 # Set the current version here to override the automatic version detection. The
 # version must be specified as `X.Y.Z`. Primarily used when building new patch
 # versions.
-set(MEMGRAPH_OVERRIDE_VERSION "")
+set(MEMGRAPH_OVERRIDE_VERSION "2.15.0")
 
 # Custom suffix that this version should have. The suffix can be any arbitrary
 # string. Primarily used when building a version for a specific customer.

licenses/BSL.txt

@@ -36,7 +36,7 @@ ADDITIONAL USE GRANT:   You may use the Licensed Work in accordance with the
                         3. using the Licensed Work to create a work or solution
                         which competes (or might reasonably be expected to
                         compete) with the Licensed Work.
-CHANGE DATE:            2028-21-01
+CHANGE DATE:            2028-28-02
 CHANGE LICENSE:         Apache License, Version 2.0
 
 For information about alternative licensing arrangements, please visit: https://memgraph.com/legal.

src/coordination/CMakeLists.txt

@@ -15,6 +15,7 @@ target_sources(mg-coordination
         include/coordination/instance_status.hpp
         include/coordination/replication_instance.hpp
         include/coordination/raft_state.hpp
+        include/coordination/rpc_errors.hpp
 
         include/nuraft/coordinator_log_store.hpp
         include/nuraft/coordinator_state_machine.hpp

src/coordination/coordinator_client.cpp

@@ -17,6 +17,7 @@
 #include "coordination/coordinator_config.hpp"
 #include "coordination/coordinator_rpc.hpp"
 #include "replication_coordination_glue/messages.hpp"
+#include "utils/result.hpp"
 
 namespace memgraph::coordination {
 
@@ -45,6 +46,10 @@ auto CoordinatorClient::InstanceDownTimeoutSec() const -> std::chrono::seconds {
   return config_.instance_down_timeout_sec;
 }
 
+auto CoordinatorClient::InstanceGetUUIDFrequencySec() const -> std::chrono::seconds {
+  return config_.instance_get_uuid_frequency_sec;
+}
+
 void CoordinatorClient::StartFrequentCheck() {
   if (instance_checker_.IsRunning()) {
     return;
@@ -140,5 +145,31 @@ auto CoordinatorClient::SendUnregisterReplicaRpc(std::string const &instance_nam
   return false;
 }
 
+auto CoordinatorClient::SendGetInstanceUUIDRpc() const
+    -> utils::BasicResult<GetInstanceUUIDError, std::optional<utils::UUID>> {
+  try {
+    auto stream{rpc_client_.Stream<GetInstanceUUIDRpc>()};
+    auto res = stream.AwaitResponse();
+    return res.uuid;
+  } catch (const rpc::RpcFailedException &) {
+    spdlog::error("RPC error occured while sending GetInstance UUID RPC");
+    return GetInstanceUUIDError::RPC_EXCEPTION;
+  }
+}
+
+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

@@ -45,6 +45,18 @@ void CoordinatorHandlers::Register(memgraph::coordination::CoordinatorServer &se
         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);
+      });
+
+  server.Register<coordination::GetInstanceUUIDRpc>(
+      [&replication_handler](slk::Reader *req_reader, slk::Builder *res_builder) -> void {
+        spdlog::info("Received GetInstanceUUIDRpc on coordinator server");
+        CoordinatorHandlers::GetInstanceUUIDHandler(replication_handler, req_reader, res_builder);
+      });
 }
 
 void CoordinatorHandlers::SwapMainUUIDHandler(replication::ReplicationHandler &replication_handler,
@@ -68,12 +80,6 @@ void CoordinatorHandlers::DemoteMainToReplicaHandler(replication::ReplicationHan
                                                      slk::Reader *req_reader, slk::Builder *res_builder) {
   spdlog::info("Executing DemoteMainToReplicaHandler");
 
-  if (!replication_handler.IsMain()) {
-    spdlog::error("Setting to replica must be performed on main.");
-    slk::Save(coordination::DemoteMainToReplicaRes{false}, res_builder);
-    return;
-  }
-
   coordination::DemoteMainToReplicaReq req;
   slk::Load(&req, req_reader);
 
@@ -83,11 +89,18 @@ void CoordinatorHandlers::DemoteMainToReplicaHandler(replication::ReplicationHan
 
   if (!replication_handler.SetReplicationRoleReplica(clients_config, std::nullopt)) {
     spdlog::error("Demoting main to replica failed!");
-    slk::Save(coordination::PromoteReplicaToMainRes{false}, res_builder);
+    slk::Save(coordination::DemoteMainToReplicaRes{false}, res_builder);
     return;
   }
 
-  slk::Save(coordination::PromoteReplicaToMainRes{true}, res_builder);
+  slk::Save(coordination::DemoteMainToReplicaRes{true}, res_builder);
+}
+
+void CoordinatorHandlers::GetInstanceUUIDHandler(replication::ReplicationHandler &replication_handler,
+                                                 slk::Reader * /*req_reader*/, slk::Builder *res_builder) {
+  spdlog::info("Executing GetInstanceUUIDHandler");
+
+  slk::Save(coordination::GetInstanceUUIDRes{replication_handler.GetReplicaUUID()}, res_builder);
 }
 
 void CoordinatorHandlers::PromoteReplicaToMainHandler(replication::ReplicationHandler &replication_handler,
@@ -148,7 +161,7 @@ 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);
 }
 
@@ -184,5 +197,22 @@ void CoordinatorHandlers::UnregisterReplicaHandler(replication::ReplicationHandl
   }
 }
 
+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>
@@ -47,9 +48,12 @@ CoordinatorInstance::CoordinatorInstance()
     spdlog::trace("Instance {} performing replica successful callback", repl_instance_name);
     auto &repl_instance = find_repl_instance(self, repl_instance_name);
 
+    // We need to get replicas UUID from time to time to ensure replica is listening to correct main
+    // and that it didn't go down for less time than we could notice
+    // We need to get id of main replica is listening to
+    // and swap if necessary
     if (!repl_instance.EnsureReplicaHasCorrectMainUUID(self->GetMainUUID())) {
-      spdlog::error(
-          fmt::format("Failed to swap uuid for replica instance {} which is alive", repl_instance.InstanceName()));
+      spdlog::error("Failed to swap uuid for replica instance {} which is alive", repl_instance.InstanceName());
       return;
     }
 
@@ -61,14 +65,6 @@ CoordinatorInstance::CoordinatorInstance()
     spdlog::trace("Instance {} performing replica failure callback", repl_instance_name);
     auto &repl_instance = find_repl_instance(self, repl_instance_name);
     repl_instance.OnFailPing();
-    // We need to restart main uuid from instance since it was "down" at least a second
-    // There is slight delay, if we choose to use isAlive, instance can be down and back up in less than
-    // our isAlive time difference, which would lead to instance setting UUID to nullopt and stopping accepting any
-    // incoming RPCs from valid main
-    // TODO(antoniofilipovic) this needs here more complex logic
-    // We need to get id of main replica is listening to on successful ping
-    // and swap it to correct uuid if it failed
-    repl_instance.ResetMainUUID();
   };
 
   main_succ_cb_ = [find_repl_instance](CoordinatorInstance *self, std::string_view repl_instance_name) -> void {
@@ -87,6 +83,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;
     }
@@ -125,9 +126,6 @@ CoordinatorInstance::CoordinatorInstance()
 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";
@@ -149,8 +147,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);
@@ -189,6 +186,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),
@@ -208,6 +206,10 @@ auto CoordinatorInstance::SetReplicationInstanceToMain(std::string instance_name
     -> SetInstanceToMainCoordinatorStatus {
   auto lock = std::lock_guard{coord_instance_lock_};
 
+  if (std::ranges::any_of(repl_instances_, &ReplicationInstance::IsMain)) {
+    return SetInstanceToMainCoordinatorStatus::MAIN_ALREADY_EXISTS;
+  }
+
   auto const is_new_main = [&instance_name](ReplicationInstance const &instance) {
     return instance.InstanceName() == instance_name;
   };

src/coordination/coordinator_rpc.cpp

@@ -68,6 +68,35 @@ void UnregisterReplicaRes::Load(UnregisterReplicaRes *self, memgraph::slk::Reade
   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) {}
+
+// GetInstanceUUID
+void GetInstanceUUIDReq::Save(const GetInstanceUUIDReq &self, memgraph::slk::Builder *builder) {
+  memgraph::slk::Save(self, builder);
+}
+
+void GetInstanceUUIDReq::Load(GetInstanceUUIDReq *self, memgraph::slk::Reader *reader) {
+  memgraph::slk::Load(self, reader);
+}
+
+void GetInstanceUUIDRes::Save(const GetInstanceUUIDRes &self, memgraph::slk::Builder *builder) {
+  memgraph::slk::Save(self, builder);
+}
+
+void GetInstanceUUIDRes::Load(GetInstanceUUIDRes *self, memgraph::slk::Reader *reader) {
+  memgraph::slk::Load(self, reader);
+}
+
 }  // namespace coordination
 
 constexpr utils::TypeInfo coordination::PromoteReplicaToMainReq::kType{utils::TypeId::COORD_FAILOVER_REQ,
@@ -89,8 +118,22 @@ constexpr utils::TypeInfo coordination::UnregisterReplicaReq::kType{utils::TypeI
 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};
+
+constexpr utils::TypeInfo coordination::GetInstanceUUIDReq::kType{utils::TypeId::COORD_GET_UUID_REQ, "CoordGetUUIDReq",
+                                                                  nullptr};
+
+constexpr utils::TypeInfo coordination::GetInstanceUUIDRes::kType{utils::TypeId::COORD_GET_UUID_RES, "CoordGetUUIDRes",
+                                                                  nullptr};
+
 namespace slk {
 
+// PromoteReplicaToMainRpc
+
 void Save(const memgraph::coordination::PromoteReplicaToMainRes &self, memgraph::slk::Builder *builder) {
   memgraph::slk::Save(self.success, builder);
 }
@@ -109,6 +152,7 @@ void Load(memgraph::coordination::PromoteReplicaToMainReq *self, memgraph::slk::
   memgraph::slk::Load(&self->replication_clients_info, reader);
 }
 
+// DemoteMainToReplicaRpc
 void Save(const memgraph::coordination::DemoteMainToReplicaReq &self, memgraph::slk::Builder *builder) {
   memgraph::slk::Save(self.replication_client_info, builder);
 }
@@ -125,6 +169,8 @@ void Load(memgraph::coordination::DemoteMainToReplicaRes *self, memgraph::slk::R
   memgraph::slk::Load(&self->success, reader);
 }
 
+// UnregisterReplicaRpc
+
 void Save(memgraph::coordination::UnregisterReplicaReq const &self, memgraph::slk::Builder *builder) {
   memgraph::slk::Save(self.instance_name, builder);
 }
@@ -141,6 +187,32 @@ void Load(memgraph::coordination::UnregisterReplicaRes *self, memgraph::slk::Rea
   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);
+}
+
+// GetInstanceUUIDRpc
+
+void Save(const memgraph::coordination::GetInstanceUUIDReq & /*self*/, memgraph::slk::Builder * /*builder*/) {
+  /* nothing to serialize*/
+}
+
+void Load(memgraph::coordination::GetInstanceUUIDReq * /*self*/, memgraph::slk::Reader * /*reader*/) {
+  /* nothing to serialize*/
+}
+
+void Save(const memgraph::coordination::GetInstanceUUIDRes &self, memgraph::slk::Builder *builder) {
+  memgraph::slk::Save(self.uuid, builder);
+}
+
+void Load(memgraph::coordination::GetInstanceUUIDRes *self, memgraph::slk::Reader *reader) {
+  memgraph::slk::Load(&self->uuid, reader);
+}
+
 }  // namespace slk
 
 }  // namespace memgraph

src/coordination/include/coordination/coordinator_client.hpp

@@ -11,12 +11,14 @@
 
 #pragma once
 
-#include "utils/uuid.hpp"
 #ifdef MG_ENTERPRISE
 
 #include "coordination/coordinator_config.hpp"
 #include "rpc/client.hpp"
+#include "rpc_errors.hpp"
+#include "utils/result.hpp"
 #include "utils/scheduler.hpp"
+#include "utils/uuid.hpp"
 
 namespace memgraph::coordination {
 
@@ -46,7 +48,7 @@ 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;
 
@@ -54,6 +56,10 @@ class CoordinatorClient {
 
   auto SendUnregisterReplicaRpc(std::string const &instance_name) const -> bool;
 
+  auto SendEnableWritingOnMainRpc() const -> bool;
+
+  auto SendGetInstanceUUIDRpc() const -> memgraph::utils::BasicResult<GetInstanceUUIDError, std::optional<utils::UUID>>;
+
   auto ReplicationClientInfo() const -> ReplClientInfo;
 
   auto SetCallbacks(HealthCheckCallback succ_cb, HealthCheckCallback fail_cb) -> void;
@@ -62,6 +68,8 @@ class CoordinatorClient {
 
   auto InstanceDownTimeoutSec() const -> std::chrono::seconds;
 
+  auto InstanceGetUUIDFrequencySec() const -> std::chrono::seconds;
+
   friend bool operator==(CoordinatorClient const &first, CoordinatorClient const &second) {
     return first.config_ == second.config_;
   }
@@ -69,7 +77,6 @@ class CoordinatorClient {
  private:
   utils::Scheduler instance_checker_;
 
-  // TODO: (andi) Pimpl?
   communication::ClientContext rpc_context_;
   mutable rpc::Client rpc_client_;
 

src/coordination/include/coordination/coordinator_config.hpp

@@ -30,6 +30,7 @@ struct CoordinatorClientConfig {
   uint16_t port{};
   std::chrono::seconds instance_health_check_frequency_sec{1};
   std::chrono::seconds instance_down_timeout_sec{5};
+  std::chrono::seconds instance_get_uuid_frequency_sec{10};
 
   auto SocketAddress() const -> std::string { return ip_address + ":" + std::to_string(port); }
 

src/coordination/include/coordination/coordinator_handlers.hpp

@@ -36,6 +36,11 @@ class CoordinatorHandlers {
 
   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);
+
+  static void GetInstanceUUIDHandler(replication::ReplicationHandler &replication_handler, slk::Reader *req_reader,
+                                     slk::Builder *res_builder);
 };
 
 }  // namespace memgraph::dbms

src/coordination/include/coordination/coordinator_rpc.hpp

@@ -111,6 +111,56 @@ struct UnregisterReplicaRes {
 
 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>;
+
+struct GetInstanceUUIDReq {
+  static const utils::TypeInfo kType;
+  static const utils::TypeInfo &GetTypeInfo() { return kType; }
+
+  static void Load(GetInstanceUUIDReq *self, memgraph::slk::Reader *reader);
+  static void Save(const GetInstanceUUIDReq &self, memgraph::slk::Builder *builder);
+
+  GetInstanceUUIDReq() = default;
+};
+
+struct GetInstanceUUIDRes {
+  static const utils::TypeInfo kType;
+  static const utils::TypeInfo &GetTypeInfo() { return kType; }
+
+  static void Load(GetInstanceUUIDRes *self, memgraph::slk::Reader *reader);
+  static void Save(const GetInstanceUUIDRes &self, memgraph::slk::Builder *builder);
+
+  explicit GetInstanceUUIDRes(std::optional<utils::UUID> uuid) : uuid(uuid) {}
+  GetInstanceUUIDRes() = default;
+
+  std::optional<utils::UUID> uuid;
+};
+
+using GetInstanceUUIDRpc = rpc::RequestResponse<GetInstanceUUIDReq, GetInstanceUUIDRes>;
+
 }  // namespace memgraph::coordination
 
 // SLK serialization declarations
@@ -128,12 +178,20 @@ 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);
 
+// GetInstanceUUIDRpc
+void Save(const memgraph::coordination::GetInstanceUUIDReq &self, memgraph::slk::Builder *builder);
+void Load(memgraph::coordination::GetInstanceUUIDReq *self, memgraph::slk::Reader *reader);
+void Save(const memgraph::coordination::GetInstanceUUIDRes &self, memgraph::slk::Builder *builder);
+void Load(memgraph::coordination::GetInstanceUUIDRes *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
 
 #endif

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

@@ -39,6 +39,7 @@ enum class UnregisterInstanceCoordinatorStatus : uint8_t {
 
 enum class SetInstanceToMainCoordinatorStatus : uint8_t {
   NO_INSTANCE_WITH_NAME,
+  MAIN_ALREADY_EXISTS,
   NOT_COORDINATOR,
   SUCCESS,
   COULD_NOT_PROMOTE_TO_MAIN,

src/coordination/include/coordination/replication_instance.hpp

@@ -18,6 +18,7 @@
 #include "replication_coordination_glue/role.hpp"
 
 #include <libnuraft/nuraft.hxx>
+#include "utils/result.hpp"
 #include "utils/uuid.hpp"
 
 namespace memgraph::coordination {
@@ -37,6 +38,9 @@ class ReplicationInstance {
 
   auto OnSuccessPing() -> void;
   auto OnFailPing() -> bool;
+  auto IsReadyForUUIDPing() -> bool;
+
+  void UpdateReplicaLastResponseUUID();
 
   auto IsAlive() const -> bool;
 
@@ -62,9 +66,12 @@ class ReplicationInstance {
   auto SendSwapAndUpdateUUID(const utils::UUID &new_main_uuid) -> bool;
   auto SendUnregisterReplicaRpc(std::string const &instance_name) -> bool;
 
-  // TODO: (andi) Inconsistent API
+
+  auto SendGetInstanceUUID() -> utils::BasicResult<coordination::GetInstanceUUIDError, std::optional<utils::UUID>>;
   auto GetClient() -> CoordinatorClient &;
 
+  auto EnableWritingOnMain() -> bool;
+
   auto SetNewMainUUID(utils::UUID const &main_uuid) -> void;
   auto ResetMainUUID() -> void;
   auto GetMainUUID() const -> const std::optional<utils::UUID> &;
@@ -74,6 +81,7 @@ class ReplicationInstance {
   replication_coordination_glue::ReplicationRole replication_role_;
   std::chrono::system_clock::time_point last_response_time_{};
   bool is_alive_{false};
+  std::chrono::system_clock::time_point last_check_of_uuid_{};
 
   // for replica this is main uuid of current main
   // for "main" main this same as in CoordinatorData

src/coordination/include/coordination/rpc_errors.hpp

@@ -0,0 +1,14 @@
+// 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.
+
+namespace memgraph::coordination {
+enum class GetInstanceUUIDError { NO_RESPONSE, RPC_EXCEPTION };
+}  // namespace memgraph::coordination

src/coordination/replication_instance.cpp

@@ -14,6 +14,7 @@
 #include "coordination/replication_instance.hpp"
 
 #include "replication_coordination_glue/handler.hpp"
+#include "utils/result.hpp"
 
 namespace memgraph::coordination {
 
@@ -39,6 +40,11 @@ auto ReplicationInstance::OnFailPing() -> bool {
   return is_alive_;
 }
 
+auto ReplicationInstance::IsReadyForUUIDPing() -> bool {
+  return std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now() - last_check_of_uuid_) >
+         client_.InstanceGetUUIDFrequencySec();
+}
+
 auto ReplicationInstance::InstanceName() const -> std::string { return client_.InstanceName(); }
 auto ReplicationInstance::SocketAddress() const -> std::string { return client_.SocketAddress(); }
 auto ReplicationInstance::IsAlive() const -> bool { return is_alive_; }
@@ -91,10 +97,20 @@ auto ReplicationInstance::ResetMainUUID() -> void { main_uuid_ = std::nullopt; }
 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) {
-    return SendSwapAndUpdateUUID(curr_main_uuid);
+  if (!IsReadyForUUIDPing()) {
+    return true;
   }
-  return true;
+  auto res = SendGetInstanceUUID();
+  if (res.HasError()) {
+    return false;
+  }
+  UpdateReplicaLastResponseUUID();
+
+  if (res.GetValue().has_value() && res.GetValue().value() == curr_main_uuid) {
+    return true;
+  }
+
+  return SendSwapAndUpdateUUID(curr_main_uuid);
 }
 
 auto ReplicationInstance::SendSwapAndUpdateUUID(const utils::UUID &new_main_uuid) -> bool {
@@ -109,5 +125,14 @@ auto ReplicationInstance::SendUnregisterReplicaRpc(std::string const &instance_n
   return client_.SendUnregisterReplicaRpc(instance_name);
 }
 
+auto ReplicationInstance::EnableWritingOnMain() -> bool { return client_.SendEnableWritingOnMainRpc(); }
+
+auto ReplicationInstance::SendGetInstanceUUID()
+    -> utils::BasicResult<coordination::GetInstanceUUIDError, std::optional<utils::UUID>> {
+  return client_.SendGetInstanceUUIDRpc();
+}
+
+void ReplicationInstance::UpdateReplicaLastResponseUUID() { last_check_of_uuid_ = std::chrono::system_clock::now(); }
+
 }  // namespace memgraph::coordination
 #endif

src/flags/replication.cpp

@@ -22,6 +22,8 @@ DEFINE_uint32(raft_server_id, 0, "Unique ID of the raft server.");
 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.");
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DEFINE_uint32(instance_get_uuid_frequency_sec, 10, "The time duration between two instance uuid checks.");
 #endif
 
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)

src/flags/replication.hpp

@@ -24,6 +24,8 @@ DECLARE_uint32(raft_server_id);
 DECLARE_uint32(instance_down_timeout_sec);
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
 DECLARE_uint32(instance_health_check_frequency_sec);
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DECLARE_uint32(instance_get_uuid_frequency_sec);
 #endif
 
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)

src/memgraph.cpp

@@ -359,6 +359,7 @@ int main(int argc, char **argv) {
 #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),
+      .instance_get_uuid_frequency_sec = std::chrono::seconds(FLAGS_instance_get_uuid_frequency_sec),
 #endif
       .default_kafka_bootstrap_servers = FLAGS_kafka_bootstrap_servers,
       .default_pulsar_service_url = FLAGS_pulsar_service_url,

src/memory/global_memory_control.cpp

@@ -61,10 +61,12 @@ void *my_alloc(extent_hooks_t *extent_hooks, void *new_addr, size_t size, size_t
   // This needs to be before, to throw exception in case of too big alloc
   if (*commit) [[likely]] {
     if (GetQueriesMemoryControl().IsThreadTracked()) [[unlikely]] {
-      GetQueriesMemoryControl().TrackAllocOnCurrentThread(size);
+      bool ok = GetQueriesMemoryControl().TrackAllocOnCurrentThread(size);
+      if (!ok) return nullptr;
     }
     // This needs to be here so it doesn't get incremented in case the first TrackAlloc throws an exception
-    memgraph::utils::total_memory_tracker.Alloc(static_cast<int64_t>(size));
+    bool ok = memgraph::utils::total_memory_tracker.Alloc(static_cast<int64_t>(size));
+    if (!ok) return nullptr;
   }
 
   auto *ptr = old_hooks->alloc(extent_hooks, new_addr, size, alignment, zero, commit, arena_ind);
@@ -118,10 +120,14 @@ static bool my_commit(extent_hooks_t *extent_hooks, void *addr, size_t size, siz
     return err;
   }
 
+  [[maybe_unused]] auto blocker = memgraph::utils::MemoryTracker::OutOfMemoryExceptionBlocker{};
   if (GetQueriesMemoryControl().IsThreadTracked()) [[unlikely]] {
-    GetQueriesMemoryControl().TrackAllocOnCurrentThread(length);
+    bool ok = GetQueriesMemoryControl().TrackAllocOnCurrentThread(length);
+    DMG_ASSERT(ok);
   }
-  memgraph::utils::total_memory_tracker.Alloc(static_cast<int64_t>(length));
+
+  auto ok = memgraph::utils::total_memory_tracker.Alloc(static_cast<int64_t>(length));
+  DMG_ASSERT(ok);
 
   return false;
 }

src/memory/new_delete.cpp

@@ -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
@@ -28,6 +28,12 @@ void *newImpl(const std::size_t size) {
     return ptr;
   }
 
+  [[maybe_unused]] auto blocker = memgraph::utils::MemoryTracker::OutOfMemoryExceptionBlocker{};
+  auto maybe_msg = memgraph::utils::MemoryErrorStatus().msg();
+  if (maybe_msg) {
+    throw memgraph::utils::OutOfMemoryException{std::move(*maybe_msg)};
+  }
+
   throw std::bad_alloc{};
 }
 
@@ -37,11 +43,21 @@ void *newImpl(const std::size_t size, const std::align_val_t align) {
     return ptr;
   }
 
+  [[maybe_unused]] auto blocker = memgraph::utils::MemoryTracker::OutOfMemoryExceptionBlocker{};
+  auto maybe_msg = memgraph::utils::MemoryErrorStatus().msg();
+  if (maybe_msg) {
+    throw memgraph::utils::OutOfMemoryException{std::move(*maybe_msg)};
+  }
+
   throw std::bad_alloc{};
 }
 
-void *newNoExcept(const std::size_t size) noexcept { return malloc(size); }
+void *newNoExcept(const std::size_t size) noexcept {
+  [[maybe_unused]] auto blocker = memgraph::utils::MemoryTracker::OutOfMemoryExceptionBlocker{};
+  return malloc(size);
+}
 void *newNoExcept(const std::size_t size, const std::align_val_t align) noexcept {
+  [[maybe_unused]] auto blocker = memgraph::utils::MemoryTracker::OutOfMemoryExceptionBlocker{};
   return aligned_alloc(size, static_cast<std::size_t>(align));
 }
 

src/memory/query_memory_control.cpp

@@ -54,14 +54,14 @@ void QueriesMemoryControl::EraseThreadToTransactionId(const std::thread::id &thr
   }
 }
 
-void QueriesMemoryControl::TrackAllocOnCurrentThread(size_t size) {
+bool QueriesMemoryControl::TrackAllocOnCurrentThread(size_t size) {
   auto thread_id_to_transaction_id_accessor = thread_id_to_transaction_id.access();
 
   // we might be just constructing mapping between thread id and transaction id
   // so we miss this allocation
   auto thread_id_to_transaction_id_elem = thread_id_to_transaction_id_accessor.find(std::this_thread::get_id());
   if (thread_id_to_transaction_id_elem == thread_id_to_transaction_id_accessor.end()) {
-    return;
+    return true;
   }
 
   auto transaction_id_to_tracker_accessor = transaction_id_to_tracker.access();
@@ -71,10 +71,10 @@ void QueriesMemoryControl::TrackAllocOnCurrentThread(size_t size) {
   // It can happen that some allocation happens between mapping thread to
   // transaction id, so we miss this allocation
   if (transaction_id_to_tracker == transaction_id_to_tracker_accessor.end()) [[unlikely]] {
-    return;
+    return true;
   }
   auto &query_tracker = transaction_id_to_tracker->tracker;
-  query_tracker.TrackAlloc(size);
+  return query_tracker.TrackAlloc(size);
 }
 
 void QueriesMemoryControl::TrackFreeOnCurrentThread(size_t size) {

src/memory/query_memory_control.hpp

@@ -62,7 +62,7 @@ class QueriesMemoryControl {
   // Find tracker for current thread if exists, track
   // query allocation and procedure allocation if
   // necessary
-  void TrackAllocOnCurrentThread(size_t size);
+  bool TrackAllocOnCurrentThread(size_t size);
 
   // Find tracker for current thread if exists, track
   // query allocation and procedure allocation if

src/query/config.hpp

@@ -24,6 +24,7 @@ struct InterpreterConfig {
 
   std::chrono::seconds instance_down_timeout_sec{5};
   std::chrono::seconds instance_health_check_frequency_sec{1};
+  std::chrono::seconds instance_get_uuid_frequency_sec{10};
 
   std::string default_kafka_bootstrap_servers;
   std::string default_pulsar_service_url;

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"
@@ -328,7 +329,7 @@ class ReplQueryHandler {
           .port = static_cast<uint16_t>(*port),
       };
 
-      if (!handler_->SetReplicationRoleReplica(config, std::nullopt)) {
+      if (!handler_->TrySetReplicationRoleReplica(config, std::nullopt)) {
         throw QueryRuntimeException("Couldn't set role to replica!");
       }
     }
@@ -485,8 +486,9 @@ class CoordQueryHandler final : public query::CoordinatorQueryHandler {
   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 {
+                                   std::chrono::seconds const &instance_down_timeout,
+                                   std::chrono::seconds const &instance_get_uuid_frequency,
+                                   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) {
@@ -513,6 +515,7 @@ class CoordQueryHandler final : public query::CoordinatorQueryHandler {
                                               .port = coordinator_server_port,
                                               .instance_health_check_frequency_sec = instance_check_frequency,
                                               .instance_down_timeout_sec = instance_down_timeout,
+                                              .instance_get_uuid_frequency_sec = instance_get_uuid_frequency,
                                               .replication_client_info = repl_config,
                                               .ssl = std::nullopt};
 
@@ -560,6 +563,8 @@ class CoordQueryHandler final : public query::CoordinatorQueryHandler {
       using enum memgraph::coordination::SetInstanceToMainCoordinatorStatus;
       case NO_INSTANCE_WITH_NAME:
         throw QueryRuntimeException("No instance with such name!");
+      case MAIN_ALREADY_EXISTS:
+        throw QueryRuntimeException("Couldn't set instance to main since there is already a main instance in cluster!");
       case NOT_COORDINATOR:
         throw QueryRuntimeException("SET INSTANCE TO MAIN query can only be run on a coordinator!");
       case COULD_NOT_PROMOTE_TO_MAIN:
@@ -1184,11 +1189,12 @@ Callback HandleCoordinatorQuery(CoordinatorQuery *coordinator_query, const Param
                      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,
+                     instance_get_uuid_frequency_sec = config.instance_get_uuid_frequency_sec,
                      sync_mode = coordinator_query->sync_mode_]() mutable {
         handler.RegisterReplicationInstance(std::string(coordinator_socket_address_tv.ValueString()),
                                             std::string(replication_socket_address_tv.ValueString()),
                                             instance_health_check_frequency_sec, instance_down_timeout_sec,
-                                            instance_name, sync_mode);
+                                            instance_get_uuid_frequency_sec, instance_name, sync_mode);
         return std::vector<std::vector<TypedValue>>();
       };
 
@@ -1264,17 +1270,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;
     }
@@ -4404,9 +4406,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)

src/query/interpreter.hpp

@@ -109,6 +109,7 @@ class CoordinatorQueryHandler {
                                            std::string const &replication_socket_address,
                                            std::chrono::seconds const &instance_health_check_frequency,
                                            std::chrono::seconds const &instance_down_timeout,
+                                           std::chrono::seconds const &instance_get_uuid_frequency,
                                            std::string const &instance_name, CoordinatorQuery::SyncMode sync_mode) = 0;
 
   /// @throw QueryRuntimeException if an error ocurred.

src/query/procedure/mg_procedure_impl.cpp

@@ -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
@@ -187,6 +187,7 @@ template <typename TFunc, typename... Args>
     spdlog::error("Memory allocation error during mg API call: {}", bae.what());
     return mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE;
   } catch (const memgraph::utils::OutOfMemoryException &oome) {
+    [[maybe_unused]] auto blocker = memgraph::utils::MemoryTracker::OutOfMemoryExceptionBlocker{};
     spdlog::error("Memory limit exceeded during mg API call: {}", oome.what());
     return mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE;
   } catch (const std::out_of_range &oore) {
@@ -198,12 +199,12 @@ template <typename TFunc, typename... Args>
   } catch (const std::logic_error &lee) {
     spdlog::error("Logic error during mg API call: {}", lee.what());
     return mgp_error::MGP_ERROR_LOGIC_ERROR;
-  } catch (const std::exception &e) {
-    spdlog::error("Unexpected error during mg API call: {}", e.what());
-    return mgp_error::MGP_ERROR_UNKNOWN_ERROR;
   } catch (const memgraph::utils::temporal::InvalidArgumentException &e) {
     spdlog::error("Invalid argument was sent to an mg API call for temporal types: {}", e.what());
     return mgp_error::MGP_ERROR_INVALID_ARGUMENT;
+  } catch (const std::exception &e) {
+    spdlog::error("Unexpected error during mg API call: {}", e.what());
+    return mgp_error::MGP_ERROR_UNKNOWN_ERROR;
   } catch (...) {
     spdlog::error("Unexpected error during mg API call");
     return mgp_error::MGP_ERROR_UNKNOWN_ERROR;

src/query/replication_query_handler.hpp

@@ -49,6 +49,9 @@ struct ReplicationQueryHandler {
   virtual bool SetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config,
                                          const std::optional<utils::UUID> &main_uuid) = 0;
 
+  virtual bool TrySetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config,
+                                            const std::optional<utils::UUID> &main_uuid) = 0;
+
   // as MAIN, define and connect to REPLICAs
   virtual auto TryRegisterReplica(const memgraph::replication::ReplicationClientConfig &config, bool send_swap_uuid)
       -> utils::BasicResult<RegisterReplicaError> = 0;

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

@@ -62,8 +62,9 @@ ReplicationState::ReplicationState(std::optional<std::filesystem::path> durabili
   }
 #endif
   if (std::holds_alternative<RoleReplicaData>(replication_data)) {
-    spdlog::trace("Recovered main's uuid for replica {}",
-                  std::string(std::get<RoleReplicaData>(replication_data).uuid_.value()));
+    auto &replica_uuid = std::get<RoleReplicaData>(replication_data).uuid_;
+    std::string uuid = replica_uuid.has_value() ? std::string(replica_uuid.value()) : "";
+    spdlog::trace("Recovered main's uuid for replica {}", uuid);
   } else {
     spdlog::trace("Recovered uuid for main {}", std::string(std::get<RoleMainData>(replication_data).uuid_));
   }
@@ -144,8 +145,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 +254,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

@@ -14,6 +14,7 @@
 #include "dbms/dbms_handler.hpp"
 #include "flags/experimental.hpp"
 #include "replication/include/replication/state.hpp"
+#include "replication_handler/system_replication.hpp"
 #include "replication_handler/system_rpc.hpp"
 #include "utils/result.hpp"
 
@@ -113,10 +114,14 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
   // as REPLICA, become MAIN
   bool SetReplicationRoleMain() override;
 
-  // as MAIN, become REPLICA
+  // as MAIN, become REPLICA, can be called on MAIN and REPLICA
   bool SetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config,
                                  const std::optional<utils::UUID> &main_uuid) override;
 
+  // as MAIN, become REPLICA, can be called only on MAIN
+  bool TrySetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config,
+                                    const std::optional<utils::UUID> &main_uuid) override;
+
   // as MAIN, define and connect to REPLICAs
   auto TryRegisterReplica(const memgraph::replication::ReplicationClientConfig &config, bool send_swap_uuid)
       -> memgraph::utils::BasicResult<memgraph::query::RegisterReplicaError> override;
@@ -137,12 +142,13 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
   auto GetReplState() const -> const memgraph::replication::ReplicationState &;
   auto GetReplState() -> memgraph::replication::ReplicationState &;
 
+  auto GetReplicaUUID() -> std::optional<utils::UUID>;
+
  private:
-  template <bool AllowReplicaToDivergeFromMain>
+  template <bool AllowRPCFailure>
   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!");
-
     auto maybe_client = repl_state_.RegisterReplica(config);
     if (maybe_client.HasError()) {
       switch (maybe_client.GetError()) {
@@ -159,7 +165,6 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
           break;
       }
     }
-
     using enum memgraph::flags::Experiments;
     bool system_replication_enabled = flags::AreExperimentsEnabled(SYSTEM_REPLICATION);
     if (!system_replication_enabled && dbms_handler_.Count() > 1) {
@@ -167,25 +172,21 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
     }
     const auto main_uuid =
         std::get<memgraph::replication::RoleMainData>(dbms_handler_.ReplicationState().ReplicationData()).uuid_;
-
     if (send_swap_uuid) {
       if (!memgraph::replication_coordination_glue::SendSwapMainUUIDRpc(maybe_client.GetValue()->rpc_client_,
                                                                         main_uuid)) {
         return memgraph::query::RegisterReplicaError::ERROR_ACCEPTING_MAIN;
       }
     }
-
 #ifdef MG_ENTERPRISE
     // Update system before enabling individual storage <-> replica clients
     SystemRestore(*maybe_client.GetValue(), system_, dbms_handler_, main_uuid, auth_);
 #endif
-
     const auto dbms_error = HandleRegisterReplicaStatus(maybe_client);
     if (dbms_error.has_value()) {
       return *dbms_error;
     }
     auto &instance_client_ptr = maybe_client.GetValue();
-
     bool all_clients_good = true;
     // Add database specific clients (NOTE Currently all databases are connected to each replica)
     dbms_handler_.ForEach([&](dbms::DatabaseAccess db_acc) {
@@ -195,7 +196,6 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
       }
       // TODO: ATM only IN_MEMORY_TRANSACTIONAL, fix other modes
       if (storage->storage_mode_ != storage::StorageMode::IN_MEMORY_TRANSACTIONAL) return;
-
       all_clients_good &= storage->repl_storage_state_.replication_clients_.WithLock(
           [storage, &instance_client_ptr, db_acc = std::move(db_acc),
            main_uuid](auto &storage_clients) mutable {  // NOLINT
@@ -203,9 +203,12 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
             client->Start(storage, std::move(db_acc));
             bool const success = std::invoke([state = client->State()]() {
               if (state == storage::replication::ReplicaState::DIVERGED_FROM_MAIN) {
-                return AllowReplicaToDivergeFromMain;
+                return false;
               }
-              return state != storage::replication::ReplicaState::MAYBE_BEHIND;
+              if (state == storage::replication::ReplicaState::MAYBE_BEHIND) {
+                return AllowRPCFailure;
+              }
+              return true;
             });
 
             if (success) {
@@ -214,14 +217,12 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
             return success;
           });
     });
-
     // NOTE Currently if any databases fails, we revert back
     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;
     }
-
     // No client error, start instance level client
 #ifdef MG_ENTERPRISE
     StartReplicaClient(*instance_client_ptr, system_, dbms_handler_, main_uuid, auth_);
@@ -231,6 +232,57 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
     return {};
   }
 
+  template <bool AllowIdempotency>
+  bool SetReplicationRoleReplica_(const memgraph::replication::ReplicationServerConfig &config,
+                                  const std::optional<utils::UUID> &main_uuid) {
+    if (repl_state_.IsReplica()) {
+      if (!AllowIdempotency) {
+        return false;
+      }
+      // We don't want to restart the server if we're already a REPLICA with correct config
+      auto &replica_data = std::get<memgraph::replication::RoleReplicaData>(repl_state_.ReplicationData());
+      if (replica_data.config == config) {
+        return true;
+      }
+      repl_state_.SetReplicationRoleReplica(config, main_uuid);
+#ifdef MG_ENTERPRISE
+      return StartRpcServer(dbms_handler_, replica_data, auth_, system_);
+#else
+      return StartRpcServer(dbms_handler_, replica_data);
+#endif
+    }
+
+    // TODO StorageState needs to be synched. Could have a dangling reference if someone adds a database as we are
+    //      deleting the replica.
+    // Remove database specific clients
+    dbms_handler_.ForEach([&](memgraph::dbms::DatabaseAccess db_acc) {
+      auto *storage = db_acc->storage();
+      storage->repl_storage_state_.replication_clients_.WithLock([](auto &clients) { clients.clear(); });
+    });
+    // Remove instance level clients
+    std::get<memgraph::replication::RoleMainData>(repl_state_.ReplicationData()).registered_replicas_.clear();
+
+    // Creates the server
+    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) {
+#ifdef MG_ENTERPRISE
+                                                 return StartRpcServer(dbms_handler_, data, auth_, system_);
+#else
+                                                 return StartRpcServer(dbms_handler_, data);
+#endif
+                                               }},
+                   repl_state_.ReplicationData());
+    // TODO Handle error (restore to main?)
+    return success;
+  }
+
   memgraph::replication::ReplicationState &repl_state_;
   memgraph::dbms::DbmsHandler &dbms_handler_;
 

src/replication_handler/replication_handler.cpp

@@ -192,41 +192,12 @@ bool ReplicationHandler::SetReplicationRoleMain() {
 
 bool ReplicationHandler::SetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config,
                                                    const std::optional<utils::UUID> &main_uuid) {
-  // We don't want to restart the server if we're already a REPLICA
-  if (repl_state_.IsReplica()) {
-    spdlog::trace("Instance has already has replica role.");
-    return false;
-  }
+  return SetReplicationRoleReplica_<false>(config, main_uuid);
+}
 
-  // TODO StorageState needs to be synched. Could have a dangling reference if someone adds a database as we are
-  //      deleting the replica.
-  // Remove database specific clients
-  dbms_handler_.ForEach([&](memgraph::dbms::DatabaseAccess db_acc) {
-    auto *storage = db_acc->storage();
-    storage->repl_storage_state_.replication_clients_.WithLock([](auto &clients) { clients.clear(); });
-  });
-  // Remove instance level clients
-  std::get<memgraph::replication::RoleMainData>(repl_state_.ReplicationData()).registered_replicas_.clear();
-
-  // Creates the server
-  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) {
-#ifdef MG_ENTERPRISE
-                                               return StartRpcServer(dbms_handler_, data, auth_, system_);
-#else
-                                               return StartRpcServer(dbms_handler_, data);
-#endif
-                                             }},
-                 repl_state_.ReplicationData());
-  // TODO Handle error (restore to main?)
-  return success;
+bool ReplicationHandler::TrySetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config,
+                                                      const std::optional<utils::UUID> &main_uuid) {
+  return SetReplicationRoleReplica_<true>(config, main_uuid);
 }
 
 bool ReplicationHandler::DoReplicaToMainPromotion(const utils::UUID &main_uuid) {
@@ -258,13 +229,13 @@ bool ReplicationHandler::DoReplicaToMainPromotion(const utils::UUID &main_uuid)
 auto ReplicationHandler::TryRegisterReplica(const memgraph::replication::ReplicationClientConfig &config,
                                             bool send_swap_uuid)
     -> memgraph::utils::BasicResult<memgraph::query::RegisterReplicaError> {
-  return RegisterReplica_<false>(config, send_swap_uuid);
+  return RegisterReplica_<true>(config, send_swap_uuid);
 }
 
 auto ReplicationHandler::RegisterReplica(const memgraph::replication::ReplicationClientConfig &config,
                                          bool send_swap_uuid)
     -> memgraph::utils::BasicResult<memgraph::query::RegisterReplicaError> {
-  return RegisterReplica_<true>(config, send_swap_uuid);
+  return RegisterReplica_<false>(config, send_swap_uuid);
 }
 
 auto ReplicationHandler::UnregisterReplica(std::string_view name) -> memgraph::query::UnregisterReplicaResult {
@@ -297,6 +268,11 @@ auto ReplicationHandler::GetRole() const -> memgraph::replication_coordination_g
   return repl_state_.GetRole();
 }
 
+auto ReplicationHandler::GetReplicaUUID() -> std::optional<utils::UUID> {
+  MG_ASSERT(repl_state_.IsReplica());
+  return std::get<RoleReplicaData>(repl_state_.ReplicationData()).uuid_;
+}
+
 auto ReplicationHandler::GetReplState() const -> const memgraph::replication::ReplicationState & { return repl_state_; }
 
 auto ReplicationHandler::GetReplState() -> memgraph::replication::ReplicationState & { return repl_state_; }

src/utils/exceptions.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
@@ -42,12 +42,26 @@ namespace memgraph::utils {
 class BasicException : public std::exception {
  public:
   /**
-   * @brief Constructor (C++ STL strings).
+   * @brief Constructor (C++ STL strings_view).
    *
    * @param message The error message.
    */
   explicit BasicException(std::string_view message) noexcept : msg_(message) {}
 
+  /**
+   * @brief Constructor (string literal).
+   *
+   * @param message The error message.
+   */
+  explicit BasicException(const char *message) noexcept : msg_(message) {}
+
+  /**
+   * @brief Constructor (C++ STL strings).
+   *
+   * @param message The error message.
+   */
+  explicit BasicException(std::string message) noexcept : msg_(std::move(message)) {}
+
   /**
    * @brief Constructor with format string (C++ STL strings).
    *

src/utils/functional.hpp

@@ -0,0 +1,27 @@
+// 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 <algorithm>
+#include <vector>
+
+namespace memgraph::utils {
+
+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/memory_tracker.cpp

@@ -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
@@ -104,7 +104,7 @@ void MemoryTracker::SetMaximumHardLimit(const int64_t limit) {
   maximum_hard_limit_ = limit;
 }
 
-void MemoryTracker::Alloc(const int64_t size) {
+bool MemoryTracker::Alloc(int64_t const size) {
   MG_ASSERT(size >= 0, "Negative size passed to the MemoryTracker.");
 
   const int64_t will_be = size + amount_.fetch_add(size, std::memory_order_relaxed);
@@ -116,12 +116,13 @@ void MemoryTracker::Alloc(const int64_t size) {
 
     amount_.fetch_sub(size, std::memory_order_relaxed);
 
-    throw OutOfMemoryException(
-        fmt::format("Memory limit exceeded! Attempting to allocate a chunk of {} which would put the current "
-                    "use to {}, while the maximum allowed size for allocation is set to {}.",
-                    GetReadableSize(size), GetReadableSize(will_be), GetReadableSize(current_hard_limit)));
+    // register our error data, we will pick this up on the other side of jemalloc
+    MemoryErrorStatus().set({size, will_be, current_hard_limit});
+
+    return false;
   }
   UpdatePeak(will_be);
+  return true;
 }
 
 void MemoryTracker::DoCheck() {
@@ -139,4 +140,23 @@ void MemoryTracker::DoCheck() {
 
 void MemoryTracker::Free(const int64_t size) { amount_.fetch_sub(size, std::memory_order_relaxed); }
 
+// DEVNOTE: important that this is allocated at thread construction time
+//          otherwise subtle bug where jemalloc will try to lock an non-recursive mutex
+//          that it already owns
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+thread_local MemoryTrackerStatus status;
+auto MemoryErrorStatus() -> MemoryTrackerStatus & { return status; }
+
+auto MemoryTrackerStatus::msg() -> std::optional<std::string> {
+  if (!data_) return std::nullopt;
+
+  auto [size, will_be, hard_limit] = *data_;
+  data_.reset();
+  return fmt::format(
+      "Memory limit exceeded! Attempting to allocate a chunk of {} which would put the current "
+      "use to {}, while the maximum allowed size for allocation is set to {}.",
+      // NOLINTNEXTLINE(bugprone-narrowing-conversions,cppcoreguidelines-narrowing-conversions)
+      GetReadableSize(size), GetReadableSize(will_be), GetReadableSize(hard_limit));
+}
+
 }  // namespace memgraph::utils

src/utils/memory_tracker.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
@@ -12,15 +12,35 @@
 #pragma once
 
 #include <atomic>
+#include <optional>
+#include <string>
 #include <type_traits>
 
 #include "utils/exceptions.hpp"
 
 namespace memgraph::utils {
 
+struct MemoryTrackerStatus {
+  struct data {
+    int64_t size;
+    int64_t will_be;
+    int64_t hard_limit;
+  };
+
+  // DEVNOTE: Do not call from within allocator, will cause another allocation
+  auto msg() -> std::optional<std::string>;
+
+  void set(data d) { data_ = d; }
+
+ private:
+  std::optional<data> data_;
+};
+
+auto MemoryErrorStatus() -> MemoryTrackerStatus &;
+
 class OutOfMemoryException : public utils::BasicException {
  public:
-  explicit OutOfMemoryException(const std::string &msg) : utils::BasicException(msg) {}
+  explicit OutOfMemoryException(std::string msg) : utils::BasicException(std::move(msg)) {}
   SPECIALIZE_GET_EXCEPTION_NAME(OutOfMemoryException)
 };
 
@@ -47,7 +67,7 @@ class MemoryTracker final {
 
   MemoryTracker &operator=(MemoryTracker &&) = delete;
 
-  void Alloc(int64_t size);
+  bool Alloc(int64_t size);
   void Free(int64_t size);
   void DoCheck();
 

src/utils/query_memory_tracker.cpp

@@ -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
@@ -17,18 +17,19 @@
 
 namespace memgraph::utils {
 
-void QueryMemoryTracker::TrackAlloc(size_t size) {
+bool QueryMemoryTracker::TrackAlloc(size_t size) {
   if (query_tracker_.has_value()) [[likely]] {
-    query_tracker_->Alloc(static_cast<int64_t>(size));
+    bool ok = query_tracker_->Alloc(static_cast<int64_t>(size));
+    if (!ok) return false;
   }
 
   auto *proc_tracker = GetActiveProc();
 
   if (proc_tracker == nullptr) {
-    return;
+    return true;
   }
 
-  proc_tracker->Alloc(static_cast<int64_t>(size));
+  return proc_tracker->Alloc(static_cast<int64_t>(size));
 }
 void QueryMemoryTracker::TrackFree(size_t size) {
   if (query_tracker_.has_value()) [[likely]] {

src/utils/query_memory_tracker.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
@@ -44,7 +44,7 @@ class QueryMemoryTracker {
   ~QueryMemoryTracker() = default;
 
   // Track allocation on query and procedure if active
-  void TrackAlloc(size_t);
+  bool TrackAlloc(size_t size);
 
   // Track Free on query and procedure if active
   void TrackFree(size_t);

src/utils/typeinfo.hpp

@@ -109,6 +109,11 @@ enum class TypeId : uint64_t {
   COORD_SWAP_UUID_RES,
   COORD_UNREGISTER_REPLICA_REQ,
   COORD_UNREGISTER_REPLICA_RES,
+  COORD_ENABLE_WRITING_ON_MAIN_REQ,
+  COORD_ENABLE_WRITING_ON_MAIN_RES,
+
+  COORD_GET_UUID_REQ,
+  COORD_GET_UUID_RES,
 
   // AST
   AST_LABELIX = 3000,

tests/e2e/CMakeLists.txt

@@ -40,7 +40,7 @@ endfunction()
 add_subdirectory(fine_grained_access)
 add_subdirectory(server)
 add_subdirectory(replication)
-#add_subdirectory(memory)
+add_subdirectory(memory)
 add_subdirectory(triggers)
 add_subdirectory(isolation_levels)
 add_subdirectory(streams)

tests/e2e/configuration/default_config.py

@@ -71,6 +71,7 @@ startup_config_dict = {
     "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."),
+    "instance_get_uuid_frequency_sec": ("10", "10", "The time duration between two instance uuid checks."),
     "data_directory": ("mg_data", "mg_data", "Path to directory in which to save all permanent data."),
     "data_recovery_on_startup": (
         "false",

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)

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/not_replicate_from_old_main.py

@@ -10,11 +10,13 @@
 # licenses/APL.txt.
 
 import os
+import shutil
 import sys
+import tempfile
 
 import interactive_mg_runner
 import pytest
-from common import execute_and_fetch_all
+from common import 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__))
@@ -38,7 +40,7 @@ MEMGRAPH_FIRST_CLUSTER_DESCRIPTION = {
 }
 
 
-MEMGRAPH_INSTANCES_DESCRIPTION = {
+MEMGRAPH_SECOND_CLUSTER_DESCRIPTION = {
     "replica": {
         "args": ["--bolt-port", "7689", "--log-level", "TRACE"],
         "log_file": "replica.log",
@@ -71,7 +73,7 @@ def test_replication_works_on_failover(connection):
     assert actual_data_on_main == expected_data_on_main
 
     # 3
-    interactive_mg_runner.start_all_keep_others(MEMGRAPH_INSTANCES_DESCRIPTION)
+    interactive_mg_runner.start_all_keep_others(MEMGRAPH_SECOND_CLUSTER_DESCRIPTION)
 
     # 4
     new_main_cursor = connection(7690, "main_2").cursor()
@@ -113,5 +115,144 @@ def test_replication_works_on_failover(connection):
     interactive_mg_runner.stop_all()
 
 
+def test_not_replicate_old_main_register_new_cluster(connection):
+    # Goal of this test is to check that although replica is registered in one cluster
+    # it can be re-registered to new cluster
+    # This flow checks if Registering replica is idempotent and that old main cannot talk to replica
+    # 1. We start all replicas and main in one cluster
+    # 2. Main from first cluster can see all replicas
+    # 3. We start all replicas and main in second cluster, by reusing one replica from first cluster
+    # 4. New main should see replica. Registration should pass (idempotent registration)
+    # 5. Old main should not talk to new replica
+    # 6. New main should talk to replica
+
+    TEMP_DIR = tempfile.TemporaryDirectory().name
+    MEMGRAPH_FISRT_COORD_CLUSTER_DESCRIPTION = {
+        "shared_instance": {
+            "args": [
+                "--bolt-port",
+                "7688",
+                "--log-level",
+                "TRACE",
+                "--coordinator-server-port",
+                "10011",
+            ],
+            "log_file": "instance_1.log",
+            "data_directory": f"{TEMP_DIR}/shared_instance",
+            "setup_queries": [],
+        },
+        "instance_2": {
+            "args": [
+                "--bolt-port",
+                "7689",
+                "--log-level",
+                "TRACE",
+                "--coordinator-server-port",
+                "10012",
+            ],
+            "log_file": "instance_2.log",
+            "data_directory": f"{TEMP_DIR}/instance_2",
+            "setup_queries": [],
+        },
+        "coordinator_1": {
+            "args": ["--bolt-port", "7690", "--log-level=TRACE", "--raft-server-id=1", "--raft-server-port=10111"],
+            "log_file": "coordinator.log",
+            "setup_queries": [
+                "REGISTER INSTANCE shared_instance ON '127.0.0.1:10011' WITH '127.0.0.1:10001';",
+                "REGISTER INSTANCE instance_2 ON '127.0.0.1:10012' WITH '127.0.0.1:10002';",
+                "SET INSTANCE instance_2 TO MAIN",
+            ],
+        },
+    }
+
+    # 1
+    interactive_mg_runner.start_all_keep_others(MEMGRAPH_FISRT_COORD_CLUSTER_DESCRIPTION)
+
+    # 2
+
+    first_cluster_coord_cursor = connection(7690, "coord_1").cursor()
+
+    def show_repl_cluster():
+        return sorted(list(execute_and_fetch_all(first_cluster_coord_cursor, "SHOW INSTANCES;")))
+
+    expected_data_up_first_cluster = [
+        ("coordinator_1", "127.0.0.1:10111", "", True, "coordinator"),
+        ("instance_2", "", "127.0.0.1:10012", True, "main"),
+        ("shared_instance", "", "127.0.0.1:10011", True, "replica"),
+    ]
+
+    mg_sleep_and_assert(expected_data_up_first_cluster, show_repl_cluster)
+
+    # 3
+
+    MEMGRAPH_SECOND_COORD_CLUSTER_DESCRIPTION = {
+        "instance_3": {
+            "args": [
+                "--bolt-port",
+                "7687",
+                "--log-level",
+                "TRACE",
+                "--coordinator-server-port",
+                "10013",
+            ],
+            "log_file": "instance_3.log",
+            "data_directory": f"{TEMP_DIR}/instance_3",
+            "setup_queries": [],
+        },
+        "coordinator_2": {
+            "args": ["--bolt-port", "7691", "--log-level=TRACE", "--raft-server-id=1", "--raft-server-port=10112"],
+            "log_file": "coordinator.log",
+            "setup_queries": [],
+        },
+    }
+
+    interactive_mg_runner.start_all_keep_others(MEMGRAPH_SECOND_COORD_CLUSTER_DESCRIPTION)
+    second_cluster_coord_cursor = connection(7691, "coord_2").cursor()
+    execute_and_fetch_all(
+        second_cluster_coord_cursor, "REGISTER INSTANCE shared_instance ON '127.0.0.1:10011' WITH '127.0.0.1:10001';"
+    )
+    execute_and_fetch_all(
+        second_cluster_coord_cursor, "REGISTER INSTANCE instance_3 ON '127.0.0.1:10013' WITH '127.0.0.1:10003';"
+    )
+    execute_and_fetch_all(second_cluster_coord_cursor, "SET INSTANCE instance_3 TO MAIN")
+
+    # 4
+
+    def show_repl_cluster():
+        return sorted(list(execute_and_fetch_all(second_cluster_coord_cursor, "SHOW INSTANCES;")))
+
+    expected_data_up_second_cluster = [
+        ("coordinator_1", "127.0.0.1:10112", "", True, "coordinator"),
+        ("instance_3", "", "127.0.0.1:10013", True, "main"),
+        ("shared_instance", "", "127.0.0.1:10011", True, "replica"),
+    ]
+
+    mg_sleep_and_assert(expected_data_up_second_cluster, show_repl_cluster)
+
+    # 5
+    main_1_cursor = connection(7689, "main_1").cursor()
+    with pytest.raises(Exception) as e:
+        execute_and_fetch_all(main_1_cursor, "CREATE ();")
+    assert (
+        str(e.value)
+        == "Replication Exception: At least one SYNC replica has not confirmed committing last transaction. Check the status of the replicas using 'SHOW REPLICAS' query."
+    )
+
+    shared_replica_cursor = connection(7688, "shared_replica").cursor()
+    res = execute_and_fetch_all(shared_replica_cursor, "MATCH (n) RETURN count(n);")[0][0]
+    assert res == 0, "Old main should not replicate to 'shared' replica"
+
+    # 6
+    main_2_cursor = connection(7687, "main_2").cursor()
+
+    execute_and_fetch_all(main_2_cursor, "CREATE ();")
+
+    shared_replica_cursor = connection(7688, "shared_replica").cursor()
+    res = execute_and_fetch_all(shared_replica_cursor, "MATCH (n) RETURN count(n);")[0][0]
+    assert res == 1, "New main should replicate to 'shared' replica"
+
+    interactive_mg_runner.stop_all()
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/high_availability_experimental/single_coordinator.py

@@ -147,6 +147,105 @@ def test_replication_works_on_failover():
     interactive_mg_runner.stop_all(MEMGRAPH_INSTANCES_DESCRIPTION)
 
 
+def test_replication_works_on_replica_instance_restart():
+    # Goal of this test is to check the replication works after replica goes down and restarts
+    # 1. We start all replicas, main and coordinator manually: we want to be able to kill them ourselves without relying on external tooling to kill processes.
+    # 2. We check that main has correct state
+    # 3. We kill replica
+    # 4. We check that main cannot replicate to replica
+    # 5. We bring replica back up
+    # 6. We check that replica gets data
+    safe_execute(shutil.rmtree, TEMP_DIR)
+
+    # 1
+    interactive_mg_runner.start_all(MEMGRAPH_INSTANCES_DESCRIPTION)
+
+    # 2
+    main_cursor = connect(host="localhost", port=7687).cursor()
+    expected_data_on_main = [
+        ("instance_1", "127.0.0.1:10001", "sync", 0, 0, "ready"),
+        ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "ready"),
+    ]
+    actual_data_on_main = sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;")))
+    assert actual_data_on_main == expected_data_on_main
+
+    # 3
+    coord_cursor = connect(host="localhost", port=7690).cursor()
+
+    interactive_mg_runner.kill(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_2")
+
+    def retrieve_data_show_repl_cluster():
+        return sorted(list(execute_and_fetch_all(coord_cursor, "SHOW INSTANCES;")))
+
+    expected_data_on_coord = [
+        ("coordinator_1", "127.0.0.1:10111", "", True, "coordinator"),
+        ("instance_1", "", "127.0.0.1:10011", True, "replica"),
+        ("instance_2", "", "127.0.0.1:10012", False, "unknown"),
+        ("instance_3", "", "127.0.0.1:10013", True, "main"),
+    ]
+    mg_sleep_and_assert(expected_data_on_coord, retrieve_data_show_repl_cluster)
+
+    def retrieve_data_show_replicas():
+        return sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;")))
+
+    expected_data_on_main = [
+        ("instance_1", "127.0.0.1:10001", "sync", 0, 0, "ready"),
+        ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "invalid"),
+    ]
+    mg_sleep_and_assert(expected_data_on_main, retrieve_data_show_replicas)
+
+    # 4
+    instance_1_cursor = connect(host="localhost", port=7688).cursor()
+    with pytest.raises(Exception) as e:
+        execute_and_fetch_all(main_cursor, "CREATE ();")
+    assert (
+        str(e.value)
+        == "Replication Exception: At least one SYNC replica has not confirmed committing last transaction. Check the status of the replicas using 'SHOW REPLICAS' query."
+    )
+
+    res_instance_1 = execute_and_fetch_all(instance_1_cursor, "MATCH (n) RETURN count(n)")[0][0]
+    assert res_instance_1 == 1
+
+    def retrieve_data_show_replicas():
+        return sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;")))
+
+    expected_data_on_main = [
+        ("instance_1", "127.0.0.1:10001", "sync", 2, 0, "ready"),
+        ("instance_2", "127.0.0.1:10002", "sync", 0, 0, "invalid"),
+    ]
+    mg_sleep_and_assert(expected_data_on_main, retrieve_data_show_replicas)
+
+    # 5.
+
+    interactive_mg_runner.start(MEMGRAPH_INSTANCES_DESCRIPTION, "instance_2")
+
+    def retrieve_data_show_repl_cluster():
+        return sorted(list(execute_and_fetch_all(coord_cursor, "SHOW INSTANCES;")))
+
+    expected_data_on_coord = [
+        ("coordinator_1", "127.0.0.1:10111", "", 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_data_on_coord, retrieve_data_show_repl_cluster)
+
+    def retrieve_data_show_replicas():
+        return sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;")))
+
+    expected_data_on_main = [
+        ("instance_1", "127.0.0.1:10001", "sync", 2, 0, "ready"),
+        ("instance_2", "127.0.0.1:10002", "sync", 2, 0, "ready"),
+    ]
+    mg_sleep_and_assert(expected_data_on_main, retrieve_data_show_replicas)
+
+    # 6.
+    instance_2_cursor = connect(port=7689, host="localhost").cursor()
+    execute_and_fetch_all(main_cursor, "CREATE ();")
+    res_instance_2 = execute_and_fetch_all(instance_2_cursor, "MATCH (n) RETURN count(n)")[0][0]
+    assert res_instance_2 == 2
+
+
 def test_show_instances():
     safe_execute(shutil.rmtree, TEMP_DIR)
     interactive_mg_runner.start_all(MEMGRAPH_INSTANCES_DESCRIPTION)
@@ -416,5 +515,20 @@ def test_automatic_failover_main_back_as_main():
     mg_sleep_and_assert([("main",)], retrieve_data_show_repl_role_instance3)
 
 
+def test_disable_multiple_mains():
+    safe_execute(shutil.rmtree, TEMP_DIR)
+    interactive_mg_runner.start_all(MEMGRAPH_INSTANCES_DESCRIPTION)
+
+    coord_cursor = connect(host="localhost", port=7690).cursor()
+
+    try:
+        execute_and_fetch_all(
+            coord_cursor,
+            "SET INSTANCE instance_1 TO MAIN;",
+        )
+    except Exception as e:
+        assert str(e) == "Couldn't set instance to main since there is already a main instance in cluster!"
+
+
 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/memory/CMakeLists.txt

@@ -22,9 +22,6 @@ target_link_libraries(memgraph__e2e__memory__limit_accumulation gflags mgclient
 add_executable(memgraph__e2e__memory__limit_edge_create memory_limit_edge_create.cpp)
 target_link_libraries(memgraph__e2e__memory__limit_edge_create gflags mgclient mg-utils mg-io)
 
-add_executable(memgraph__e2e__memory_limit_global_multi_thread_proc_create memory_limit_global_multi_thread_proc_create.cpp)
-target_link_libraries(memgraph__e2e__memory_limit_global_multi_thread_proc_create gflags mgclient mg-utils mg-io)
-
 add_executable(memgraph__e2e__memory_limit_global_thread_alloc_proc memory_limit_global_thread_alloc_proc.cpp)
 target_link_libraries(memgraph__e2e__memory_limit_global_thread_alloc_proc gflags mgclient mg-utils mg-io)
 

tests/e2e/memory/memory_control.cpp

@@ -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
@@ -44,7 +44,7 @@ int main(int argc, char **argv) {
     client->DiscardAll();
   }
 
-  const auto *create_query = "UNWIND range(1, 50) as u CREATE (n {string: \"Some longer string\"}) RETURN n;";
+  const auto *create_query = "UNWIND range(1, 100) as u CREATE (n {string: \"Some longer string\"}) RETURN n;";
 
   memgraph::utils::Timer timer;
   while (true) {

tests/e2e/memory/memory_limit_global_multi_thread_proc_create.cpp

@@ -1,67 +0,0 @@
-// Copyright 2023 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 <gflags/gflags.h>
-#include <algorithm>
-#include <exception>
-#include <ios>
-#include <iostream>
-#include <mgclient.hpp>
-
-#include "utils/logging.hpp"
-#include "utils/timer.hpp"
-
-DEFINE_uint64(bolt_port, 7687, "Bolt port");
-DEFINE_uint64(timeout, 120, "Timeout seconds");
-DEFINE_bool(multi_db, false, "Run test in multi db environment");
-
-int main(int argc, char **argv) {
-  google::SetUsageMessage("Memgraph E2E Global Memory Limit In Multi-Thread Create For Local Allocators");
-  gflags::ParseCommandLineFlags(&argc, &argv, true);
-  memgraph::logging::RedirectToStderr();
-
-  mg::Client::Init();
-
-  auto client =
-      mg::Client::Connect({.host = "127.0.0.1", .port = static_cast<uint16_t>(FLAGS_bolt_port), .use_ssl = false});
-  if (!client) {
-    LOG_FATAL("Failed to connect!");
-  }
-
-  if (FLAGS_multi_db) {
-    client->Execute("CREATE DATABASE clean;");
-    client->DiscardAll();
-    client->Execute("USE DATABASE clean;");
-    client->DiscardAll();
-    client->Execute("MATCH (n) DETACH DELETE n;");
-    client->DiscardAll();
-  }
-
-  bool error{false};
-  try {
-    client->Execute(
-        "CALL libglobal_memory_limit_multi_thread_create_proc.multi_create() PROCEDURE MEMORY UNLIMITED YIELD "
-        "allocated_all RETURN allocated_all "
-        "QUERY MEMORY LIMIT 50MB;");
-    auto result_rows = client->FetchAll();
-    if (result_rows) {
-      auto row = *result_rows->begin();
-      error = !row[0].ValueBool();
-    }
-
-  } catch (const std::exception &e) {
-    error = true;
-  }
-
-  MG_ASSERT(error, "Error should have happend");
-
-  return 0;
-}

tests/e2e/memory/procedures/CMakeLists.txt

@@ -6,7 +6,7 @@ target_include_directories(global_memory_limit_proc PRIVATE ${CMAKE_SOURCE_DIR}/
 
 add_library(query_memory_limit_proc_multi_thread SHARED query_memory_limit_proc_multi_thread.cpp)
 target_include_directories(query_memory_limit_proc_multi_thread PRIVATE ${CMAKE_SOURCE_DIR}/include)
-target_link_libraries(query_memory_limit_proc_multi_thread mg-utils)
+target_link_libraries(query_memory_limit_proc_multi_thread mg-utils )
 
 add_library(query_memory_limit_proc SHARED query_memory_limit_proc.cpp)
 target_include_directories(query_memory_limit_proc PRIVATE ${CMAKE_SOURCE_DIR}/include)
@@ -16,10 +16,6 @@ add_library(global_memory_limit_thread_proc SHARED global_memory_limit_thread_pr
 target_include_directories(global_memory_limit_thread_proc PRIVATE ${CMAKE_SOURCE_DIR}/include)
 target_link_libraries(global_memory_limit_thread_proc mg-utils)
 
-add_library(global_memory_limit_multi_thread_create_proc SHARED global_memory_limit_multi_thread_create_proc.cpp)
-target_include_directories(global_memory_limit_multi_thread_create_proc PRIVATE ${CMAKE_SOURCE_DIR}/include)
-target_link_libraries(global_memory_limit_multi_thread_create_proc mg-utils)
-
 add_library(proc_memory_limit SHARED proc_memory_limit.cpp)
 target_include_directories(proc_memory_limit PRIVATE ${CMAKE_SOURCE_DIR}/include)
 target_link_libraries(proc_memory_limit mg-utils)

tests/e2e/memory/procedures/global_memory_limit_multi_thread_create_proc.cpp

@@ -1,95 +0,0 @@
-// Copyright 2023 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 <atomic>
-#include <cassert>
-#include <exception>
-#include <functional>
-
-#include <mutex>
-#include <sstream>
-#include <string>
-#include <thread>
-#include <utility>
-#include <vector>
-
-#include "mg_procedure.h"
-#include "mgp.hpp"
-#include "utils/on_scope_exit.hpp"
-
-// change communication between threads with feature and promise
-std::atomic<int> created_vertices{0};
-constexpr int num_vertices_per_thread{100'000};
-constexpr int num_threads{2};
-
-void CallCreate(mgp_graph *graph, mgp_memory *memory) {
-  [[maybe_unused]] const enum mgp_error tracking_error = mgp_track_current_thread_allocations(graph);
-  for (int i = 0; i < num_vertices_per_thread; i++) {
-    struct mgp_vertex *vertex{nullptr};
-    auto enum_error = mgp_graph_create_vertex(graph, memory, &vertex);
-    if (enum_error != mgp_error::MGP_ERROR_NO_ERROR) {
-      break;
-    }
-    created_vertices.fetch_add(1, std::memory_order_acq_rel);
-  }
-  [[maybe_unused]] const enum mgp_error untracking_error = mgp_untrack_current_thread_allocations(graph);
-}
-
-void AllocFunc(mgp_graph *graph, mgp_memory *memory) {
-  try {
-    CallCreate(graph, memory);
-  } catch (const std::exception &e) {
-    return;
-  }
-}
-
-void MultiCreate(mgp_list *args, mgp_graph *memgraph_graph, mgp_result *result, mgp_memory *memory) {
-  mgp::MemoryDispatcherGuard guard{memory};
-  const auto arguments = mgp::List(args);
-  const auto record_factory = mgp::RecordFactory(result);
-  try {
-    std::vector<std::thread> threads;
-
-    for (int i = 0; i < 2; i++) {
-      threads.emplace_back(AllocFunc, memgraph_graph, memory);
-    }
-
-    for (int i = 0; i < num_threads; i++) {
-      threads[i].join();
-    }
-    if (created_vertices.load(std::memory_order_acquire) != num_vertices_per_thread * num_threads) {
-      record_factory.SetErrorMessage("Unable to allocate");
-      return;
-    }
-
-    auto new_record = record_factory.NewRecord();
-    new_record.Insert("allocated_all",
-                      created_vertices.load(std::memory_order_acquire) == num_vertices_per_thread * num_threads);
-  } catch (std::exception &e) {
-    record_factory.SetErrorMessage(e.what());
-  }
-}
-
-extern "C" int mgp_init_module(struct mgp_module *module, struct mgp_memory *memory) {
-  try {
-    mgp::MemoryDispatcherGuard guard{memory};
-
-    AddProcedure(MultiCreate, std::string("multi_create").c_str(), mgp::ProcedureType::Write, {},
-                 {mgp::Return(std::string("allocated_all").c_str(), mgp::Type::Bool)}, module, memory);
-
-  } catch (const std::exception &e) {
-    return 1;
-  }
-
-  return 0;
-}
-
-extern "C" int mgp_shutdown_module() { return 0; }

tests/e2e/memory/procedures/query_memory_limit_proc_multi_thread.cpp

@@ -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
@@ -13,73 +13,122 @@
 #include <cassert>
 #include <exception>
 #include <functional>
+#include <future>
 
+#include <exception>
+#include <latch>
+#include <list>
+#include <memory>
 #include <mutex>
-#include <sstream>
 #include <string>
 #include <thread>
-#include <utility>
 #include <vector>
 
 #include "mg_procedure.h"
 #include "mgp.hpp"
+#include "utils/memory_tracker.hpp"
 #include "utils/on_scope_exit.hpp"
 
-enum mgp_error Alloc(void *ptr) {
-  const size_t mb_size_268 = 1 << 28;
+using safe_ptr = std::unique_ptr<void, decltype([](void *p) { mgp_global_free(p); })>;
+enum class AllocFuncRes { NoIssues, UnableToAlloc, Unexpected };
+using result_t = std::pair<AllocFuncRes, std::list<safe_ptr>>;
 
-  return mgp_global_alloc(mb_size_268, (void **)(&ptr));
-}
+constexpr auto N_THREADS = 2;
+static_assert(N_THREADS > 0 && (N_THREADS & (N_THREADS - 1)) == 0);
 
-// change communication between threads with feature and promise
-std::atomic<int> num_allocations{0};
-std::vector<void *> ptrs_;
+constexpr auto mb_size_512 = 1 << 29;
+constexpr auto mb_size_16 = 1 << 24;
 
-void AllocFunc(mgp_graph *graph) {
+static_assert(mb_size_512 % N_THREADS == 0);
+static_assert(mb_size_16 % N_THREADS == 0);
+static_assert(mb_size_512 % mb_size_16 == 0);
+
+void AllocFunc(std::latch &start_latch, std::promise<result_t> promise, mgp_graph *graph) {
   [[maybe_unused]] const enum mgp_error tracking_error = mgp_track_current_thread_allocations(graph);
-  void *ptr = nullptr;
+  auto on_exit = memgraph::utils::OnScopeExit{[&]() {
+    [[maybe_unused]] const enum mgp_error untracking_error = mgp_untrack_current_thread_allocations(graph);
+  }};
+
+  std::list<safe_ptr> ptrs;
+
+  // Ensure test would concurrently run these allocations, wait until both are ready
+  start_latch.arrive_and_wait();
 
-  ptrs_.emplace_back(ptr);
   try {
-    enum mgp_error alloc_err { mgp_error::MGP_ERROR_NO_ERROR };
-    alloc_err = Alloc(ptr);
-    if (alloc_err != mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
-      num_allocations.fetch_add(1, std::memory_order_relaxed);
-    }
-    if (alloc_err != mgp_error::MGP_ERROR_NO_ERROR) {
-      assert(false);
+    constexpr auto allocation_limit = mb_size_512 / N_THREADS;
+    // many allocation to increase chance of seeing any concurent issues
+    for (auto total = 0; total < allocation_limit; total += mb_size_16) {
+      void *ptr = nullptr;
+      auto alloc_err = mgp_global_alloc(mb_size_16, &ptr);
+      if (alloc_err != mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE && ptr != nullptr) {
+        ptrs.emplace_back(ptr);
+      } else if (alloc_err == mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) {
+        // this is expected, the test checks N threads allocating to a limit of 512MB
+        promise.set_value({AllocFuncRes::UnableToAlloc, std::move(ptrs)});
+        return;
+      } else {
+        promise.set_value({AllocFuncRes::Unexpected, std::move(ptrs)});
+        return;
+      }
     }
   } catch (const std::exception &e) {
-    assert(false);
+    promise.set_value({AllocFuncRes::Unexpected, std::move(ptrs)});
+    return;
   }
-
-  [[maybe_unused]] const enum mgp_error untracking_error = mgp_untrack_current_thread_allocations(graph);
+  promise.set_value({AllocFuncRes::NoIssues, std::move(ptrs)});
+  return;
 }
 
 void DualThread(mgp_list *args, mgp_graph *memgraph_graph, mgp_result *result, mgp_memory *memory) {
   mgp::MemoryDispatcherGuard guard{memory};
   const auto arguments = mgp::List(args);
   const auto record_factory = mgp::RecordFactory(result);
-  num_allocations.store(0, std::memory_order_relaxed);
+
+  // 1 byte allocation to
+  auto ptr = std::invoke([&] {
+    void *ptr;
+    [[maybe_unused]] auto alloc_err = mgp_global_alloc(1, &ptr);
+    return safe_ptr{ptr};
+  });
+
   try {
-    std::vector<std::thread> threads;
-
-    for (int i = 0; i < 2; i++) {
-      threads.emplace_back(AllocFunc, memgraph_graph);
-    }
-
-    for (int i = 0; i < 2; i++) {
-      threads[i].join();
-    }
-    for (void *ptr : ptrs_) {
-      if (ptr != nullptr) {
-        mgp_global_free(ptr);
+    auto futures = std::vector<std::future<result_t>>{};
+    futures.reserve(N_THREADS);
+    std::latch start_latch{N_THREADS};
+    {
+      auto threads = std::vector<std::jthread>{};
+      threads.reserve(N_THREADS);
+      for (int i = 0; i < N_THREADS; i++) {
+        auto promise = std::promise<result_t>{};
+        futures.emplace_back(promise.get_future());
+        threads.emplace_back([&, promise = std::move(promise)]() mutable {
+          AllocFunc(start_latch, std::move(promise), memgraph_graph);
+        });
       }
+    }  // ~jthread will join
+
+    int alloc_errors = 0;
+    int unexpected_errors = 0;
+    for (auto &x : futures) {
+      auto [res, ptrs] = x.get();
+      alloc_errors += (res == AllocFuncRes::UnableToAlloc);
+      unexpected_errors += (res == AllocFuncRes::Unexpected);
+      // regardless of outcome, we want this thread to do the deallocation
+      ptrs.clear();
+    }
+
+    if (unexpected_errors != 0) {
+      record_factory.SetErrorMessage("Unanticipated error happened");
+      return;
+    }
+
+    if (alloc_errors < 1) {
+      record_factory.SetErrorMessage("Didn't hit the QUERY MEMORY LIMIT we expected");
+      return;
     }
 
     auto new_record = record_factory.NewRecord();
-
-    new_record.Insert("allocated_all", num_allocations.load(std::memory_order_relaxed) == 2);
+    new_record.Insert("test_passed", true);
   } catch (std::exception &e) {
     record_factory.SetErrorMessage(e.what());
   }
@@ -90,7 +139,7 @@ extern "C" int mgp_init_module(struct mgp_module *module, struct mgp_memory *mem
     mgp::memory = memory;
 
     AddProcedure(DualThread, std::string("dual_thread").c_str(), mgp::ProcedureType::Read, {},
-                 {mgp::Return(std::string("allocated_all").c_str(), mgp::Type::Bool)}, module, memory);
+                 {mgp::Return(std::string("test_passed").c_str(), mgp::Type::Bool)}, module, memory);
 
   } catch (const std::exception &e) {
     return 1;

tests/e2e/memory/query_memory_limit_proc_multi_thread.cpp

@@ -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
@@ -46,21 +46,21 @@ int main(int argc, char **argv) {
   }
 
   MG_ASSERT(
-      client->Execute("CALL libquery_memory_limit_proc_multi_thread.dual_thread() YIELD allocated_all RETURN "
-                      "allocated_all QUERY MEMORY LIMIT 500MB"));
+      client->Execute("CALL libquery_memory_limit_proc_multi_thread.dual_thread() YIELD test_passed RETURN "
+                      "test_passed QUERY MEMORY LIMIT 500MB"));
   bool error{false};
   try {
     auto result_rows = client->FetchAll();
     if (result_rows) {
       auto row = *result_rows->begin();
-      error = !row[0].ValueBool();
+      MG_ASSERT(row[0].ValueBool(), "Execpected the procedure to pass");
+    } else {
+      MG_ASSERT(false, "Expected at least one row");
     }
 
   } catch (const std::exception &e) {
-    error = true;
+    MG_ASSERT(error, "This error should not have happend {}", e.what());
   }
 
-  MG_ASSERT(error, "Error should have happend");
-
   return 0;
 }

tests/e2e/memory/workloads.yaml

@@ -6,7 +6,22 @@ args: &args
   - "--storage-gc-cycle-sec=180"
   - "--log-level=TRACE"
 
-in_memory_cluster: &in_memory_cluster
+args_150_MiB_limit: &args_150_MiB_limit
+  - "--bolt-port"
+  - *bolt_port
+  - "--memory-limit=150"
+  - "--storage-gc-cycle-sec=180"
+  - "--log-level=TRACE"
+
+in_memory_150_MiB_limit_cluster: &in_memory_150_MiB_limit_cluster
+  cluster:
+    main:
+      args: *args_150_MiB_limit
+      log_file: "memory-e2e.log"
+      setup_queries: []
+      validation_queries: []
+
+in_memory_1024_MiB_limit_cluster: &in_memory_1024_MiB_limit_cluster
   cluster:
     main:
       args: *args
@@ -61,6 +76,30 @@ disk_450_MiB_limit_cluster: &disk_450_MiB_limit_cluster
       setup_queries: []
       validation_queries: []
 
+args_300_MiB_limit: &args_300_MiB_limit
+  - "--bolt-port"
+  - *bolt_port
+  - "--memory-limit=300"
+  - "--storage-gc-cycle-sec=180"
+  - "--log-level=INFO"
+
+in_memory_300_MiB_limit_cluster: &in_memory_300_MiB_limit_cluster
+  cluster:
+    main:
+      args: *args_300_MiB_limit
+      log_file: "memory-e2e.log"
+      setup_queries: []
+      validation_queries: []
+
+
+disk_300_MiB_limit_cluster: &disk_300_MiB_limit_cluster
+  cluster:
+    main:
+      args: *args_300_MiB_limit
+      log_file: "memory-e2e.log"
+      setup_queries: []
+      validation_queries: []
+
 
 args_global_limit_1024_MiB: &args_global_limit_1024_MiB
   - "--bolt-port"
@@ -80,36 +119,36 @@ workloads:
   - name: "Memory control"
     binary: "tests/e2e/memory/memgraph__e2e__memory__control"
     args: ["--bolt-port", *bolt_port, "--timeout", "180"]
-    <<: *in_memory_cluster
+    <<: *in_memory_150_MiB_limit_cluster
 
   - name: "Memory control multi database"
     binary: "tests/e2e/memory/memgraph__e2e__memory__control"
     args: ["--bolt-port", *bolt_port, "--timeout", "180", "--multi-db", "true"]
-    <<: *in_memory_cluster
+    <<: *in_memory_150_MiB_limit_cluster
 
   - name: "Memory limit for modules upon loading"
     binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc"
     args: ["--bolt-port", *bolt_port, "--timeout", "180"]
     proc: "tests/e2e/memory/procedures/"
-    <<: *in_memory_cluster
+    <<: *in_memory_1024_MiB_limit_cluster
 
   - name: "Memory limit for modules upon loading multi database"
     binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc"
     args: ["--bolt-port", *bolt_port, "--timeout", "180", "--multi-db", "true"]
     proc: "tests/e2e/memory/procedures/"
-    <<: *in_memory_cluster
+    <<: *in_memory_1024_MiB_limit_cluster
 
   - name: "Memory limit for modules inside a procedure"
     binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc_proc"
     args: ["--bolt-port", *bolt_port, "--timeout", "180"]
     proc: "tests/e2e/memory/procedures/"
-    <<: *in_memory_cluster
+    <<: *in_memory_1024_MiB_limit_cluster
 
   - name: "Memory limit for modules inside a procedure multi database"
     binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc_proc"
     args: ["--bolt-port", *bolt_port, "--timeout", "180", "--multi-db", "true"]
     proc: "tests/e2e/memory/procedures/"
-    <<: *in_memory_cluster
+    <<: *in_memory_1024_MiB_limit_cluster
 
   - name: "Memory limit for modules upon loading for on-disk storage"
     binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc"
@@ -143,12 +182,12 @@ workloads:
   - name: "Memory control for detach delete"
     binary: "tests/e2e/memory/memgraph__e2e__memory__limit_delete"
     args: ["--bolt-port", *bolt_port]
-    <<: *in_memory_450_MiB_limit_cluster
+    <<: *in_memory_300_MiB_limit_cluster
 
   - name: "Memory control for detach delete on disk storage"
     binary: "tests/e2e/memory/memgraph__e2e__memory__limit_delete"
     args: ["--bolt-port", *bolt_port]
-    <<: *disk_450_MiB_limit_cluster
+    <<: *disk_300_MiB_limit_cluster
 
   - name: "Memory control for accumulation"
     binary: "tests/e2e/memory/memgraph__e2e__memory__limit_accumulation"
@@ -170,17 +209,11 @@ workloads:
     args: ["--bolt-port", *bolt_port]
     <<: *disk_450_MiB_limit_cluster
 
-  - name: "Memory control for create from multi thread proc create"
-    binary: "tests/e2e/memory/memgraph__e2e__memory_limit_global_multi_thread_proc_create"
-    proc: "tests/e2e/memory/procedures/"
-    args: ["--bolt-port", *bolt_port]
-    <<: *in_memory_cluster
-
   - name: "Memory control for memory limit global thread alloc"
     binary: "tests/e2e/memory/memgraph__e2e__memory_limit_global_thread_alloc_proc"
     proc: "tests/e2e/memory/procedures/"
     args: ["--bolt-port", *bolt_port]
-    <<: *in_memory_cluster
+    <<: *in_memory_1024_MiB_limit_cluster
 
   - name: "Procedure memory control for single procedure"
     binary: "tests/e2e/memory/memgraph__e2e__procedure_memory_limit"

tests/unit/storage_v2_replication.cpp

@@ -142,7 +142,7 @@ TEST_F(ReplicationTest, BasicSynchronousReplicationTest) {
   MinMemgraph replica(repl_conf);
 
   auto replica_store_handler = replica.repl_handler;
-  replica_store_handler.SetReplicationRoleReplica(
+  replica_store_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = ports[0],
@@ -439,13 +439,13 @@ TEST_F(ReplicationTest, MultipleSynchronousReplicationTest) {
   MinMemgraph replica1(repl_conf);
   MinMemgraph replica2(repl2_conf);
 
-  replica1.repl_handler.SetReplicationRoleReplica(
+  replica1.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = ports[0],
       },
       std::nullopt);
-  replica2.repl_handler.SetReplicationRoleReplica(
+  replica2.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = ports[1],
@@ -597,7 +597,7 @@ TEST_F(ReplicationTest, RecoveryProcess) {
     MinMemgraph replica(repl_conf);
     auto replica_store_handler = replica.repl_handler;
 
-    replica_store_handler.SetReplicationRoleReplica(
+    replica_store_handler.TrySetReplicationRoleReplica(
         ReplicationServerConfig{
             .ip_address = local_host,
             .port = ports[0],
@@ -676,7 +676,7 @@ TEST_F(ReplicationTest, BasicAsynchronousReplicationTest) {
   MinMemgraph replica_async(repl_conf);
 
   auto replica_store_handler = replica_async.repl_handler;
-  replica_store_handler.SetReplicationRoleReplica(
+  replica_store_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = ports[1],
@@ -726,7 +726,7 @@ TEST_F(ReplicationTest, EpochTest) {
   MinMemgraph main(main_conf);
   MinMemgraph replica1(repl_conf);
 
-  replica1.repl_handler.SetReplicationRoleReplica(
+  replica1.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = ports[0],
@@ -734,7 +734,7 @@ TEST_F(ReplicationTest, EpochTest) {
       std::nullopt);
 
   MinMemgraph replica2(repl2_conf);
-  replica2.repl_handler.SetReplicationRoleReplica(
+  replica2.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = 10001,
@@ -819,7 +819,7 @@ TEST_F(ReplicationTest, EpochTest) {
     ASSERT_FALSE(acc->Commit().HasError());
   }
 
-  replica1.repl_handler.SetReplicationRoleReplica(
+  replica1.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = ports[0],
@@ -858,7 +858,7 @@ TEST_F(ReplicationTest, ReplicationInformation) {
   MinMemgraph replica1(repl_conf);
 
   uint16_t replica1_port = 10001;
-  replica1.repl_handler.SetReplicationRoleReplica(
+  replica1.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = replica1_port,
@@ -867,7 +867,7 @@ TEST_F(ReplicationTest, ReplicationInformation) {
 
   uint16_t replica2_port = 10002;
   MinMemgraph replica2(repl2_conf);
-  replica2.repl_handler.SetReplicationRoleReplica(
+  replica2.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = replica2_port,
@@ -923,7 +923,7 @@ TEST_F(ReplicationTest, ReplicationReplicaWithExistingName) {
   MinMemgraph replica1(repl_conf);
 
   uint16_t replica1_port = 10001;
-  replica1.repl_handler.SetReplicationRoleReplica(
+  replica1.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = replica1_port,
@@ -932,7 +932,7 @@ TEST_F(ReplicationTest, ReplicationReplicaWithExistingName) {
 
   uint16_t replica2_port = 10002;
   MinMemgraph replica2(repl2_conf);
-  replica2.repl_handler.SetReplicationRoleReplica(
+  replica2.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = replica2_port,
@@ -966,7 +966,7 @@ TEST_F(ReplicationTest, ReplicationReplicaWithExistingEndPoint) {
 
   MinMemgraph main(main_conf);
   MinMemgraph replica1(repl_conf);
-  replica1.repl_handler.SetReplicationRoleReplica(
+  replica1.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = common_port,
@@ -974,7 +974,7 @@ TEST_F(ReplicationTest, ReplicationReplicaWithExistingEndPoint) {
       std::nullopt);
 
   MinMemgraph replica2(repl2_conf);
-  replica2.repl_handler.SetReplicationRoleReplica(
+  replica2.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = common_port,
@@ -1023,7 +1023,7 @@ TEST_F(ReplicationTest, RestoringReplicationAtStartupAfterDroppingReplica) {
   std::optional<MinMemgraph> main(main_config);
   MinMemgraph replica1(replica1_config);
 
-  replica1.repl_handler.SetReplicationRoleReplica(
+  replica1.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = ports[0],
@@ -1031,7 +1031,7 @@ TEST_F(ReplicationTest, RestoringReplicationAtStartupAfterDroppingReplica) {
       std::nullopt);
 
   MinMemgraph replica2(replica2_config);
-  replica2.repl_handler.SetReplicationRoleReplica(
+  replica2.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = ports[1],
@@ -1088,7 +1088,7 @@ TEST_F(ReplicationTest, RestoringReplicationAtStartup) {
   std::optional<MinMemgraph> main(main_config);
   MinMemgraph replica1(repl_conf);
 
-  replica1.repl_handler.SetReplicationRoleReplica(
+  replica1.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = ports[0],
@@ -1097,7 +1097,7 @@ TEST_F(ReplicationTest, RestoringReplicationAtStartup) {
 
   MinMemgraph replica2(repl2_conf);
 
-  replica2.repl_handler.SetReplicationRoleReplica(
+  replica2.repl_handler.TrySetReplicationRoleReplica(
       ReplicationServerConfig{
           .ip_address = local_host,
           .port = ports[1],

tests/unit/utils_memory_tracker.cpp

@@ -1,4 +1,4 @@
-// Copyright 2022 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
@@ -36,13 +36,13 @@ TEST(MemoryTrackerTest, ExceptionEnabler) {
       can_continue = true;
     }};
 
-    ASSERT_NO_THROW(memory_tracker.Alloc(hard_limit + 1));
+    ASSERT_TRUE(memory_tracker.Alloc(hard_limit + 1));
   }};
 
   std::thread t2{[&] {
     memgraph::utils::MemoryTracker::OutOfMemoryExceptionEnabler exception_enabler;
     enabler_created = true;
-    ASSERT_THROW(memory_tracker.Alloc(hard_limit + 1), memgraph::utils::OutOfMemoryException);
+    ASSERT_FALSE(memory_tracker.Alloc(hard_limit + 1));
 
     // hold the enabler until the first thread finishes
     while (!can_continue)
@@ -63,8 +63,8 @@ TEST(MemoryTrackerTest, ExceptionBlocker) {
   {
     memgraph::utils::MemoryTracker::OutOfMemoryExceptionBlocker exception_blocker;
 
-    ASSERT_NO_THROW(memory_tracker.Alloc(hard_limit + 1));
+    ASSERT_TRUE(memory_tracker.Alloc(hard_limit + 1));
     ASSERT_EQ(memory_tracker.Amount(), hard_limit + 1);
   }
-  ASSERT_THROW(memory_tracker.Alloc(hard_limit + 1), memgraph::utils::OutOfMemoryException);
+  ASSERT_FALSE(memory_tracker.Alloc(hard_limit + 1));
 }