Continue to debug

config/flags.yaml

@@ -99,10 +99,6 @@ modifications:
     value: "SNAPSHOT_ISOLATION"
     override: true
 
-  - name: "storage_mode"
-    value: "IN_MEMORY_TRANSACTIONAL"
-    override: true
-
   - name: "allow_load_csv"
     value: "true"
     override: false

src/dbms/database.cpp

@@ -10,10 +10,8 @@
 // licenses/APL.txt.
 
 #include "dbms/database.hpp"
-#include "flags/storage_mode.hpp"
 #include "storage/v2/disk/storage.hpp"
 #include "storage/v2/inmemory/storage.hpp"
-#include "storage/v2/storage_mode.hpp"
 
 template struct memgraph::utils::Gatekeeper<memgraph::dbms::Database>;
 
@@ -22,11 +20,10 @@ namespace memgraph::dbms {
 Database::Database(const storage::Config &config)
     : trigger_store_(config.durability.storage_directory / "triggers"),
       streams_{config.durability.storage_directory / "streams"} {
-  if (config.storage_mode == memgraph::storage::StorageMode::ON_DISK_TRANSACTIONAL || config.force_on_disk ||
-      utils::DirExists(config.disk.main_storage_directory)) {
+  if (config.force_on_disk || utils::DirExists(config.disk.main_storage_directory)) {
     storage_ = std::make_unique<storage::DiskStorage>(config);
   } else {
-    storage_ = std::make_unique<storage::InMemoryStorage>(config, config.storage_mode);
+    storage_ = std::make_unique<storage::InMemoryStorage>(config);
   }
 }
 

src/flags/CMakeLists.txt

@@ -4,7 +4,6 @@ add_library(mg-flags STATIC audit.cpp
                             isolation_level.cpp
                             log_level.cpp
                             memory_limit.cpp
-                            run_time_configurable.cpp
-                            storage_mode.cpp)
+                            run_time_configurable.cpp)
 target_include_directories(mg-flags PUBLIC ${CMAKE_SOURCE_DIR}/include)
 target_link_libraries(mg-flags PUBLIC spdlog::spdlog mg-settings mg-utils)

src/flags/all.hpp

@@ -17,4 +17,3 @@
 #include "flags/log_level.hpp"
 #include "flags/memory_limit.hpp"
 #include "flags/run_time_configurable.hpp"
-#include "flags/storage_mode.hpp"

src/flags/storage_mode.cpp

@@ -1,55 +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 "flags/storage_mode.hpp"
-
-#include "storage/v2/storage_mode.hpp"
-#include "utils/enum.hpp"
-#include "utils/flag_validation.hpp"
-
-#include "gflags/gflags.h"
-
-#include <array>
-
-inline constexpr std::array storage_mode_mappings{
-    std::pair{std::string_view{"IN_MEMORY_TRANSACTIONAL"}, memgraph::storage::StorageMode::IN_MEMORY_TRANSACTIONAL},
-    std::pair{std::string_view{"IN_MEMORY_ANALYTICAL"}, memgraph::storage::StorageMode::IN_MEMORY_ANALYTICAL},
-    std::pair{std::string_view{"ON_DISK_TRANSACTIONAL"}, memgraph::storage::StorageMode::ON_DISK_TRANSACTIONAL}};
-
-const std::string storage_mode_help_string =
-    fmt::format("Default storage mode Memgraph uses. Allowed values: {}",
-                memgraph::utils::GetAllowedEnumValuesString(storage_mode_mappings));
-
-// NOLINTNEXTLINE (cppcoreguidelines-avoid-non-const-global-variables)
-DEFINE_VALIDATED_string(storage_mode, "IN_MEMORY_TRANSACTIONAL", storage_mode_help_string.c_str(), {
-  if (const auto result = memgraph::utils::IsValidEnumValueString(value, storage_mode_mappings); result.HasError()) {
-    switch (result.GetError()) {
-      case memgraph::utils::ValidationError::EmptyValue: {
-        std::cout << "Storage mode cannot be empty." << std::endl;
-        break;
-      }
-      case memgraph::utils::ValidationError::InvalidValue: {
-        std::cout << "Invalid value for storage mode. Allowed values: "
-                  << memgraph::utils::GetAllowedEnumValuesString(storage_mode_mappings) << std::endl;
-        break;
-      }
-    }
-    return false;
-  }
-  return true;
-});
-
-memgraph::storage::StorageMode memgraph::flags::ParseStorageMode() {
-  const auto storage_mode =
-      memgraph::utils::StringToEnum<memgraph::storage::StorageMode>(FLAGS_storage_mode, storage_mode_mappings);
-  MG_ASSERT(storage_mode, "Invalid storage mode");
-  return *storage_mode;
-}

src/flags/storage_mode.hpp

@@ -1,19 +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.
-#pragma once
-
-#include "storage/v2/storage_mode.hpp"
-
-namespace memgraph::flags {
-
-memgraph::storage::StorageMode ParseStorageMode();
-
-}  // namespace memgraph::flags

src/memgraph.cpp

@@ -291,8 +291,7 @@ int main(int argc, char **argv) {
                .name_id_mapper_directory = FLAGS_data_directory + "/rocksdb_name_id_mapper",
                .id_name_mapper_directory = FLAGS_data_directory + "/rocksdb_id_name_mapper",
                .durability_directory = FLAGS_data_directory + "/rocksdb_durability",
-               .wal_directory = FLAGS_data_directory + "/rocksdb_wal"},
-      .storage_mode = memgraph::flags::ParseStorageMode()};
+               .wal_directory = FLAGS_data_directory + "/rocksdb_wal"}};
   if (FLAGS_storage_snapshot_interval_sec == 0) {
     if (FLAGS_storage_wal_enabled) {
       LOG_FATAL(

src/query/interpreter.cpp

@@ -238,10 +238,9 @@ class ReplQueryHandler final : public query::ReplicationQueryHandler {
       if (!port || *port < 0 || *port > std::numeric_limits<uint16_t>::max()) {
         throw QueryRuntimeException("Port number invalid!");
       }
-      if (!db_->SetReplicaRole(storage::replication::ReplicationServerConfig{
-              .ip_address = storage::replication::kDefaultReplicationServerIp,
-              .port = static_cast<uint16_t>(*port),
-          })) {
+      if (!db_->SetReplicaRole(
+              io::network::Endpoint(storage::replication::kDefaultReplicationServerIp, static_cast<uint16_t>(*port)),
+              storage::replication::ReplicationServerConfig{})) {
         throw QueryRuntimeException("Couldn't set role to replica!");
       }
     }
@@ -287,14 +286,9 @@ class ReplQueryHandler final : public query::ReplicationQueryHandler {
         io::network::Endpoint::ParseSocketOrIpAddress(socket_address, storage::replication::kDefaultReplicationPort);
     if (maybe_ip_and_port) {
       auto [ip, port] = *maybe_ip_and_port;
-      auto ret = db_->RegisterReplica(
-          storage::replication::RegistrationMode::MUST_BE_INSTANTLY_VALID,
-          storage::replication::ReplicationClientConfig{.name = name,
-                                                        .mode = repl_mode,
-                                                        .ip_address = ip,
-                                                        .port = port,
-                                                        .replica_check_frequency = replica_check_frequency,
-                                                        .ssl = std::nullopt});
+      auto ret = db_->RegisterReplica(name, {std::move(ip), port}, repl_mode,
+                                      storage::replication::RegistrationMode::MUST_BE_INSTANTLY_VALID,
+                                      {.replica_check_frequency = replica_check_frequency, .ssl = std::nullopt});
       if (ret.HasError()) {
         throw QueryRuntimeException(fmt::format("Couldn't register replica '{}'!", name));
       }
@@ -3628,6 +3622,8 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
          utils::Downcast<TransactionQueueQuery>(parsed_query.query))) {
       memgraph::metrics::IncrementCounter(memgraph::metrics::ActiveTransactions);
       auto &db_acc = *db_acc_;
+      spdlog::error("{} Creating Accessor in interpreter. db_accessor_ {}", std::this_thread::get_id(),
+                    (db_accessor_ ? "has value" : "nullptr"));
       db_accessor_ = db_acc->Access(GetIsolationLevelOverride());
       execution_db_accessor_.emplace(db_accessor_.get());
       transaction_status_.store(TransactionStatus::ACTIVE, std::memory_order_release);
@@ -3668,6 +3664,7 @@ Interpreter::PrepareResult Interpreter::Prepare(const std::string &query_string,
       prepared_query = PrepareDumpQuery(std::move(parsed_query), &query_execution->summary, &*execution_db_accessor_,
                                         memory_resource);
     } else if (utils::Downcast<IndexQuery>(parsed_query.query)) {
+      spdlog::error("{} PrepareIndexQuery in interpreter.", std::this_thread::get_id());
       prepared_query = PrepareIndexQuery(std::move(parsed_query), in_explicit_transaction_,
                                          &query_execution->notifications, db->storage(), get_plan_cache());
     } else if (utils::Downcast<AnalyzeGraphQuery>(parsed_query.query)) {
@@ -4005,6 +4002,7 @@ void Interpreter::Commit() {
   if (!commit_confirmed_by_all_sync_repplicas) {
     throw ReplicationException("At least one SYNC replica has not confirmed committing last transaction.");
   }
+  db_accessor_.reset();
 }
 
 void Interpreter::AdvanceCommand() {

src/storage/v2/config.hpp

@@ -15,7 +15,6 @@
 #include <cstdint>
 #include <filesystem>
 #include "storage/v2/isolation_level.hpp"
-#include "storage/v2/storage_mode.hpp"
 #include "utils/exceptions.hpp"
 
 namespace memgraph::storage {
@@ -80,7 +79,6 @@ struct Config {
 
   std::string name;
   bool force_on_disk{false};
-  StorageMode storage_mode{StorageMode::IN_MEMORY_TRANSACTIONAL};
 };
 
 static inline void UpdatePaths(Config &config, const std::filesystem::path &storage_dir) {

src/storage/v2/disk/rocksdb_storage.hpp

@@ -78,7 +78,7 @@ class ComparatorWithU64TsImpl : public rocksdb::Comparator {
 };
 
 struct DiskEdgeKey {
-  explicit DiskEdgeKey(const std::string_view keyView) : key(keyView) {}
+  DiskEdgeKey(const std::string_view keyView) : key(keyView) {}
 
   /// @tparam src_vertex_gid, dest_vertex_gid: Gid of the source and destination vertices
   /// @tparam edge_type_id: EdgeTypeId of the edge

src/storage/v2/disk/storage.cpp

@@ -391,7 +391,7 @@ std::optional<EdgeAccessor> DiskStorage::DiskAccessor::DeserializeEdge(const roc
   const auto edge_parts = utils::Split(key.ToStringView(), "|");
   const Gid edge_gid = Gid::FromString(edge_parts[4]);
 
-  auto edge_acc = transaction_.edges_.access();
+  auto edge_acc = edges_.access();
   auto res = edge_acc.find(edge_gid);
   if (res != edge_acc.end()) {
     return std::nullopt;
@@ -1203,8 +1203,7 @@ Result<EdgeAccessor> DiskStorage::DiskAccessor::CreateEdgeFromDisk(const VertexA
 
   EdgeRef edge(gid);
   if (config_.properties_on_edges) {
-    auto acc =
-        edge_import_mode_active ? disk_storage->edge_import_mode_cache_->AccessToEdges() : transaction_.edges_.access();
+    auto acc = edge_import_mode_active ? disk_storage->edge_import_mode_cache_->AccessToEdges() : edges_.access();
     auto *delta = CreateDeleteDeserializedObjectDelta(&transaction_, std::move(old_disk_key), std::move(read_ts));
     auto [it, inserted] = acc.insert(Edge(gid, delta));
     MG_ASSERT(it != acc.end(), "Invalid Edge accessor!");
@@ -1241,8 +1240,7 @@ Result<EdgeAccessor> DiskStorage::DiskAccessor::CreateEdge(VertexAccessor *from,
   bool edge_import_mode_active = disk_storage->edge_import_status_ == EdgeImportMode::ACTIVE;
 
   if (config_.properties_on_edges) {
-    auto acc =
-        edge_import_mode_active ? disk_storage->edge_import_mode_cache_->AccessToEdges() : transaction_.edges_.access();
+    auto acc = edge_import_mode_active ? disk_storage->edge_import_mode_cache_->AccessToEdges() : edges_.access();
     auto *delta = CreateDeleteObjectDelta(&transaction_);
     auto [it, inserted] = acc.insert(Edge(gid, delta));
     MG_ASSERT(inserted, "The edge must be inserted here!");
@@ -1576,7 +1574,7 @@ utils::BasicResult<StorageDataManipulationError, void> DiskStorage::DiskAccessor
         return vertices_flush_res.GetError();
       }
 
-      if (auto modified_edges_res = FlushModifiedEdges(transaction_.edges_.access()); modified_edges_res.HasError()) {
+      if (auto modified_edges_res = FlushModifiedEdges(edges_.access()); modified_edges_res.HasError()) {
         Abort();
         return modified_edges_res.GetError();
       }

src/storage/v2/disk/storage.hpp

@@ -271,6 +271,7 @@ class DiskStorage final : public Storage {
     /// We need them because query context for indexed reading is cleared after the query is done not after the
     /// transaction is done
     std::vector<std::list<Delta>> index_deltas_storage_;
+    utils::SkipList<Edge> edges_;
     Config::Items config_;
     std::unordered_set<std::string> edges_to_delete_;
     std::vector<std::pair<std::string, std::string>> vertices_to_delete_;
@@ -374,12 +375,13 @@ class DiskStorage final : public Storage {
   EdgeImportMode edge_import_status_{EdgeImportMode::INACTIVE};
   std::unique_ptr<EdgeImportModeCache> edge_import_mode_cache_{nullptr};
 
-  auto CreateReplicationClient(replication::ReplicationClientConfig const &config)
+  auto CreateReplicationClient(std::string name, io::network::Endpoint endpoint, replication::ReplicationMode mode,
+                               const replication::ReplicationClientConfig &config)
       -> std::unique_ptr<ReplicationClient> override {
     throw utils::BasicException("Disk storage mode does not support replication.");
   }
 
-  auto CreateReplicationServer(const replication::ReplicationServerConfig &config)
+  auto CreateReplicationServer(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config)
       -> std::unique_ptr<ReplicationServer> override {
     throw utils::BasicException("Disk storage mode does not support replication.");
   }

src/storage/v2/inmemory/replication/replication_client.cpp

@@ -102,9 +102,10 @@ uint64_t ReplicateCurrentWal(CurrentWalHandler &stream, durability::WalFile cons
 
 ////// ReplicationClient //////
 
-InMemoryReplicationClient::InMemoryReplicationClient(InMemoryStorage *storage,
+InMemoryReplicationClient::InMemoryReplicationClient(InMemoryStorage *storage, std::string name,
+                                                     io::network::Endpoint endpoint, replication::ReplicationMode mode,
                                                      const replication::ReplicationClientConfig &config)
-    : ReplicationClient{storage, config} {}
+    : ReplicationClient{storage, std::move(name), std::move(endpoint), mode, config} {}
 
 void InMemoryReplicationClient::RecoverReplica(uint64_t replica_commit) {
   spdlog::debug("Starting replica recover");

src/storage/v2/inmemory/replication/replication_client.hpp

@@ -18,7 +18,8 @@ class InMemoryStorage;
 
 class InMemoryReplicationClient : public ReplicationClient {
  public:
-  InMemoryReplicationClient(InMemoryStorage *storage, const replication::ReplicationClientConfig &config);
+  InMemoryReplicationClient(InMemoryStorage *storage, std::string name, io::network::Endpoint endpoint,
+                            replication::ReplicationMode mode, const replication::ReplicationClientConfig &config = {});
 
  protected:
   void RecoverReplica(uint64_t replica_commit) override;

src/storage/v2/inmemory/replication/replication_server.cpp

@@ -32,9 +32,9 @@ std::pair<uint64_t, durability::WalDeltaData> ReadDelta(durability::BaseDecoder
 };
 }  // namespace
 
-InMemoryReplicationServer::InMemoryReplicationServer(InMemoryStorage *storage,
+InMemoryReplicationServer::InMemoryReplicationServer(InMemoryStorage *storage, memgraph::io::network::Endpoint endpoint,
                                                      const replication::ReplicationServerConfig &config)
-    : ReplicationServer{config}, storage_(storage) {
+    : ReplicationServer{std::move(endpoint), config}, storage_(storage) {
   rpc_server_.Register<replication::HeartbeatRpc>([this](auto *req_reader, auto *res_builder) {
     spdlog::debug("Received HeartbeatRpc");
     this->HeartbeatHandler(req_reader, res_builder);

src/storage/v2/inmemory/replication/replication_server.hpp

@@ -20,7 +20,8 @@ class InMemoryStorage;
 
 class InMemoryReplicationServer : public ReplicationServer {
  public:
-  explicit InMemoryReplicationServer(InMemoryStorage *storage, const replication::ReplicationServerConfig &config);
+  explicit InMemoryReplicationServer(InMemoryStorage *storage, io::network::Endpoint endpoint,
+                                     const replication::ReplicationServerConfig &config);
 
  private:
   // RPC handlers

src/storage/v2/inmemory/storage.cpp

@@ -22,15 +22,13 @@ namespace memgraph::storage {
 
 using OOMExceptionEnabler = utils::MemoryTracker::OutOfMemoryExceptionEnabler;
 
-InMemoryStorage::InMemoryStorage(Config config, StorageMode storage_mode)
-    : Storage(config, storage_mode),
+InMemoryStorage::InMemoryStorage(Config config)
+    : Storage(config, StorageMode::IN_MEMORY_TRANSACTIONAL),
       snapshot_directory_(config.durability.storage_directory / durability::kSnapshotDirectory),
       lock_file_path_(config.durability.storage_directory / durability::kLockFile),
       wal_directory_(config.durability.storage_directory / durability::kWalDirectory),
       uuid_(utils::GenerateUUID()),
       global_locker_(file_retainer_.AddLocker()) {
-  MG_ASSERT(storage_mode != StorageMode::ON_DISK_TRANSACTIONAL,
-            "Invalid storage mode sent to InMemoryStorage constructor!");
   if (config_.durability.snapshot_wal_mode != Config::Durability::SnapshotWalMode::DISABLED ||
       config_.durability.snapshot_on_exit || config_.durability.recover_on_startup) {
     // Create the directory initially to crash the database in case of
@@ -149,8 +147,6 @@ InMemoryStorage::InMemoryStorage(Config config, StorageMode storage_mode)
   }
 }
 
-InMemoryStorage::InMemoryStorage(Config config) : InMemoryStorage(config, StorageMode::IN_MEMORY_TRANSACTIONAL) {}
-
 InMemoryStorage::~InMemoryStorage() {
   if (config_.gc.type == Config::Gc::Type::PERIODIC) {
     gc_runner_.Stop();
@@ -199,6 +195,8 @@ InMemoryStorage::InMemoryAccessor::~InMemoryAccessor() {
   }
 
   FinalizeTransaction();
+  spdlog::error("{} InMemoryAccessor::~InMemoryAccessor deleting accessor.\n Stack: {}", std::this_thread::get_id(),
+                "" /*utils::Stacktrace().dump()*/);
 }
 
 VertexAccessor InMemoryStorage::InMemoryAccessor::CreateVertex() {
@@ -953,7 +951,9 @@ void InMemoryStorage::InMemoryAccessor::FinalizeTransaction() {
 
 utils::BasicResult<StorageIndexDefinitionError, void> InMemoryStorage::CreateIndex(
     LabelId label, const std::optional<uint64_t> desired_commit_timestamp) {
+  spdlog::error("{} CreateIndex trying to acquire lock", std::this_thread::get_id());
   std::unique_lock<utils::RWLock> storage_guard(main_lock_);
+  spdlog::error("{} CreateIndex acquired lock", std::this_thread::get_id());
   auto *mem_label_index = static_cast<InMemoryLabelIndex *>(indices_.label_index_.get());
   if (!mem_label_index->CreateIndex(label, vertices_.access(), std::nullopt)) {
     return StorageIndexDefinitionError{IndexDefinitionError{}};
@@ -976,7 +976,9 @@ utils::BasicResult<StorageIndexDefinitionError, void> InMemoryStorage::CreateInd
 
 utils::BasicResult<StorageIndexDefinitionError, void> InMemoryStorage::CreateIndex(
     LabelId label, PropertyId property, const std::optional<uint64_t> desired_commit_timestamp) {
+  spdlog::error("{} CreatedIndex2 trying to acquire lock", std::this_thread::get_id());
   std::unique_lock<utils::RWLock> storage_guard(main_lock_);
+  spdlog::error("{} CreatedIndex2 acquired lock", std::this_thread::get_id());
   auto *mem_label_property_index = static_cast<InMemoryLabelPropertyIndex *>(indices_.label_property_index_.get());
   if (!mem_label_property_index->CreateIndex(label, property, vertices_.access(), std::nullopt)) {
     return StorageIndexDefinitionError{IndexDefinitionError{}};
@@ -1045,7 +1047,9 @@ utils::BasicResult<StorageIndexDefinitionError, void> InMemoryStorage::DropIndex
 
 utils::BasicResult<StorageExistenceConstraintDefinitionError, void> InMemoryStorage::CreateExistenceConstraint(
     LabelId label, PropertyId property, const std::optional<uint64_t> desired_commit_timestamp) {
+  spdlog::error("{} CreateExistenceConstraint trying to acquire lock", std::this_thread::get_id());
   std::unique_lock<utils::RWLock> storage_guard(main_lock_);
+  spdlog::error("{} CreateExistenceConstraint acquired lock", std::this_thread::get_id());
 
   if (constraints_.existence_constraints_->ConstraintExists(label, property)) {
     return StorageExistenceConstraintDefinitionError{ConstraintDefinitionError{}};
@@ -1093,7 +1097,9 @@ utils::BasicResult<StorageExistenceConstraintDroppingError, void> InMemoryStorag
 utils::BasicResult<StorageUniqueConstraintDefinitionError, UniqueConstraints::CreationStatus>
 InMemoryStorage::CreateUniqueConstraint(LabelId label, const std::set<PropertyId> &properties,
                                         const std::optional<uint64_t> desired_commit_timestamp) {
+  spdlog::error("{} CreateUniqueConstraint trying to acquire lock", std::this_thread::get_id());
   std::unique_lock<utils::RWLock> storage_guard(main_lock_);
+  spdlog::error("{} CreateUniqueConstraint acquired lock", std::this_thread::get_id());
   auto *mem_unique_constraints = static_cast<InMemoryUniqueConstraints *>(constraints_.unique_constraints_.get());
   auto ret = mem_unique_constraints->CreateConstraint(label, properties, vertices_.access());
   if (ret.HasError()) {
@@ -1204,15 +1210,19 @@ void InMemoryStorage::CollectGarbage(std::unique_lock<utils::RWLock> main_guard)
     if constexpr (force) {
       // We take the unique lock on the main storage lock, so we can forcefully clean
       // everything we can
+      spdlog::error("{} CollectGarbage trying to acquire unique lock", std::this_thread::get_id());
       if (!main_lock_.try_lock()) {
         CollectGarbage<false>();
         return;
       }
+      spdlog::error("{} CollectGarbage acquired unique lock", std::this_thread::get_id());
     } else {
       // Because the garbage collector iterates through the indices and constraints
       // to clean them up, it must take the main lock for reading to make sure that
       // the indices and constraints aren't concurrently being modified.
+      spdlog::error("{} CollectGarbage trying to acquire shared lock", std::this_thread::get_id());
       main_lock_.lock_shared();
+      spdlog::error("{} CollectGarbage acquired shared lock", std::this_thread::get_id());
     }
   } else {
     MG_ASSERT(main_guard.mutex() == std::addressof(main_lock_), "main_guard should be only for the main_lock_");
@@ -1756,13 +1766,17 @@ utils::BasicResult<InMemoryStorage::CreateSnapshotError> InMemoryStorage::Create
   auto max_num_tries{10};
   while (max_num_tries) {
     if (should_try_shared) {
+      spdlog::error("{} CreateSnapshot trying to acquire shared lock", std::this_thread::get_id());
       std::shared_lock<utils::RWLock> storage_guard(main_lock_);
+      spdlog::error("{} CreateSnapshot acquired shared lock", std::this_thread::get_id());
       if (storage_mode_ == memgraph::storage::StorageMode::IN_MEMORY_TRANSACTIONAL) {
         snapshot_creator();
         return {};
       }
     } else {
+      spdlog::error("{} CreateSnapshot trying to acquire unique lock", std::this_thread::get_id());
       std::unique_lock main_guard{main_lock_};
+      spdlog::error("{} CreateSnapshot acquired unique lock", std::this_thread::get_id());
       if (storage_mode_ == memgraph::storage::StorageMode::IN_MEMORY_ANALYTICAL) {
         if (is_periodic && *is_periodic) {
           return CreateSnapshotError::DisabledForAnalyticsPeriodicCommit;
@@ -1832,14 +1846,16 @@ utils::FileRetainer::FileLockerAccessor::ret_type InMemoryStorage::UnlockPath()
   return true;
 }
 
-auto InMemoryStorage::CreateReplicationClient(replication::ReplicationClientConfig const &config)
+auto InMemoryStorage::CreateReplicationClient(std::string name, io::network::Endpoint endpoint,
+                                              replication::ReplicationMode mode,
+                                              replication::ReplicationClientConfig const &config)
     -> std::unique_ptr<ReplicationClient> {
-  return std::make_unique<InMemoryReplicationClient>(this, config);
+  return std::make_unique<InMemoryReplicationClient>(this, std::move(name), std::move(endpoint), mode, config);
 }
 
 std::unique_ptr<ReplicationServer> InMemoryStorage::CreateReplicationServer(
-    const replication::ReplicationServerConfig &config) {
-  return std::make_unique<InMemoryReplicationServer>(this, config);
+    io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config) {
+  return std::make_unique<InMemoryReplicationServer>(this, std::move(endpoint), config);
 }
 
 }  // namespace memgraph::storage

src/storage/v2/inmemory/storage.hpp

@@ -51,7 +51,6 @@ class InMemoryStorage final : public Storage {
   /// @throw std::system_error
   /// @throw std::bad_alloc
   explicit InMemoryStorage(Config config = Config());
-  InMemoryStorage(Config config, StorageMode storage_mode);
 
   InMemoryStorage(const InMemoryStorage &) = delete;
   InMemoryStorage(InMemoryStorage &&) = delete;
@@ -371,10 +370,11 @@ class InMemoryStorage final : public Storage {
 
   Transaction CreateTransaction(IsolationLevel isolation_level, StorageMode storage_mode) override;
 
-  auto CreateReplicationClient(replication::ReplicationClientConfig const &config)
+  auto CreateReplicationClient(std::string name, io::network::Endpoint endpoint, replication::ReplicationMode mode,
+                               replication::ReplicationClientConfig const &config)
       -> std::unique_ptr<ReplicationClient> override;
 
-  auto CreateReplicationServer(const replication::ReplicationServerConfig &config)
+  auto CreateReplicationServer(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config)
       -> std::unique_ptr<ReplicationServer> override;
 
  private:

src/storage/v2/replication/config.hpp

@@ -1,4 +1,4 @@
-// Copyright 2023 Memgraph Ltd.
+// 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
@@ -15,15 +15,8 @@
 #include <optional>
 #include <string>
 
-#include "storage/v2/replication/enums.hpp"
-
 namespace memgraph::storage::replication {
 struct ReplicationClientConfig {
-  std::string name;
-  ReplicationMode mode;
-  std::string ip_address;
-  uint16_t port;
-
   // The default delay between main checking/pinging replicas is 1s because
   // that seems like a reasonable timeframe in which main should notice a
   // replica is down.
@@ -40,8 +33,6 @@ struct ReplicationClientConfig {
 };
 
 struct ReplicationServerConfig {
-  std::string ip_address;
-  uint16_t port;
   struct SSL {
     std::string key_file;
     std::string cert_file;

src/storage/v2/replication/replication.cpp

@@ -23,8 +23,8 @@ using OOMExceptionEnabler = utils::MemoryTracker::OutOfMemoryExceptionEnabler;
 
 namespace {
 
-std::string RegisterReplicaErrorToString(RegisterReplicaError error) {
-  using enum RegisterReplicaError;
+std::string RegisterReplicaErrorToString(ReplicationState::RegisterReplicaError error) {
+  using enum ReplicationState::RegisterReplicaError;
   switch (error) {
     case NAME_EXISTS:
       return "NAME_EXISTS";
@@ -147,75 +147,84 @@ bool storage::ReplicationState::FinalizeTransaction(uint64_t timestamp) {
   return finalized_on_all_replicas;
 }
 
-utils::BasicResult<RegisterReplicaError> ReplicationState::RegisterReplica(
+utils::BasicResult<ReplicationState::RegisterReplicaError> ReplicationState::RegisterReplica(
+    std::string name, io::network::Endpoint endpoint, const replication::ReplicationMode replication_mode,
     const replication::RegistrationMode registration_mode, const replication::ReplicationClientConfig &config,
     Storage *storage) {
   MG_ASSERT(GetRole() == replication::ReplicationRole::MAIN, "Only main instance can register a replica!");
 
-  auto name_check = [&config](auto &clients) {
-    auto name_matches = [&name = config.name](const auto &client) { return client->Name() == name; };
-    return std::any_of(clients.begin(), clients.end(), name_matches);
-  };
+  const bool name_exists = replication_clients_.WithLock([&](auto &clients) {
+    return std::any_of(clients.begin(), clients.end(), [&name](const auto &client) { return client->Name() == name; });
+  });
 
-  auto desired_endpoint = io::network::Endpoint{config.ip_address, config.port};
-  auto endpoint_check = [&](auto &clients) {
-    auto endpoint_matches = [&](const auto &client) { return client->Endpoint() == desired_endpoint; };
-    return std::any_of(clients.begin(), clients.end(), endpoint_matches);
-  };
+  if (name_exists) {
+    return RegisterReplicaError::NAME_EXISTS;
+  }
 
-  auto task = [&](auto &clients) -> utils::BasicResult<RegisterReplicaError> {
-    if (name_check(clients)) {
-      return RegisterReplicaError::NAME_EXISTS;
-    }
+  const auto end_point_exists = replication_clients_.WithLock([&endpoint](auto &clients) {
+    return std::any_of(clients.begin(), clients.end(),
+                       [&endpoint](const auto &client) { return client->Endpoint() == endpoint; });
+  });
 
-    if (endpoint_check(clients)) {
-      return RegisterReplicaError::END_POINT_EXISTS;
-    }
+  if (end_point_exists) {
+    return RegisterReplicaError::END_POINT_EXISTS;
+  }
 
-    if (!TryPersistReplicaClient(config)) {
+  if (ShouldStoreAndRestoreReplicationState()) {
+    auto data = replication::ReplicationStatusToJSON(
+        replication::ReplicationStatus{.name = name,
+                                       .ip_address = endpoint.address,
+                                       .port = endpoint.port,
+                                       .sync_mode = replication_mode,
+                                       .replica_check_frequency = config.replica_check_frequency,
+                                       .ssl = config.ssl,
+                                       .role = replication::ReplicationRole::REPLICA});
+    if (!durability_->Put(name, data.dump())) {
+      spdlog::error("Error when saving replica {} in settings.", name);
       return RegisterReplicaError::COULD_NOT_BE_PERSISTED;
     }
+  }
 
-    auto client = storage->CreateReplicationClient(config);
-    client->Start();
+  auto client = storage->CreateReplicationClient(std::move(name), std::move(endpoint), replication_mode, config);
+  client->Start();
 
-    if (client->State() == replication::ReplicaState::INVALID) {
-      if (replication::RegistrationMode::CAN_BE_INVALID != registration_mode) {
-        return RegisterReplicaError::CONNECTION_FAILED;
-      }
-
-      spdlog::warn("Connection failed when registering replica {}. Replica will still be registered.", client->Name());
+  if (client->State() == replication::ReplicaState::INVALID) {
+    if (replication::RegistrationMode::CAN_BE_INVALID != registration_mode) {
+      return RegisterReplicaError::CONNECTION_FAILED;
     }
 
-    clients.push_back(std::move(client));
-    return {};
-  };
+    spdlog::warn("Connection failed when registering replica {}. Replica will still be registered.", client->Name());
+  }
 
-  return replication_clients_.WithLock(task);
+  return replication_clients_.WithLock(
+      [&](auto &clients) -> utils::BasicResult<ReplicationState::RegisterReplicaError> {
+        // Another thread could have added a client with same name while
+        // we were connecting to this client.
+        if (std::any_of(clients.begin(), clients.end(),
+                        [&](const auto &other_client) { return client->Name() == other_client->Name(); })) {
+          return RegisterReplicaError::NAME_EXISTS;
+        }
+
+        if (std::any_of(clients.begin(), clients.end(), [&client](const auto &other_client) {
+              return client->Endpoint() == other_client->Endpoint();
+            })) {
+          return RegisterReplicaError::END_POINT_EXISTS;
+        }
+
+        clients.push_back(std::move(client));
+        return {};
+      });
 }
 
-bool ReplicationState::TryPersistReplicaClient(const replication::ReplicationClientConfig &config) {
-  if (!ShouldStoreAndRestoreReplicationState()) return true;
-  auto data = replication::ReplicationStatusToJSON(
-      replication::ReplicationStatus{.name = config.name,
-                                     .ip_address = config.ip_address,
-                                     .port = config.port,
-                                     .sync_mode = config.mode,
-                                     .replica_check_frequency = config.replica_check_frequency,
-                                     .ssl = config.ssl,
-                                     .role = replication::ReplicationRole::REPLICA});
-  if (durability_->Put(config.name, data.dump())) return true;
-  spdlog::error("Error when saving replica {} in settings.", config.name);
-  return false;
-}
-
-bool ReplicationState::SetReplicaRole(const replication::ReplicationServerConfig &config, Storage *storage) {
+bool ReplicationState::SetReplicaRole(io::network::Endpoint endpoint,
+                                      const replication::ReplicationServerConfig &config, Storage *storage) {
   // We don't want to restart the server if we're already a REPLICA
   if (GetRole() == replication::ReplicationRole::REPLICA) {
     return false;
   }
 
-  replication_server_ = storage->CreateReplicationServer(config);
+  auto port = endpoint.port;  // assigning because we will move the endpoint
+  replication_server_ = storage->CreateReplicationServer(std::move(endpoint), config);
   bool res = replication_server_->Start();
   if (!res) {
     spdlog::error("Unable to start the replication server.");
@@ -226,8 +235,8 @@ bool ReplicationState::SetReplicaRole(const replication::ReplicationServerConfig
     // Only thing that matters here is the role saved as REPLICA and the listening port
     auto data = replication::ReplicationStatusToJSON(
         replication::ReplicationStatus{.name = replication::kReservedReplicationRoleName,
-                                       .ip_address = config.ip_address,
-                                       .port = config.port,
+                                       .ip_address = "",
+                                       .port = port,
                                        .sync_mode = replication::ReplicationMode::SYNC,
                                        .replica_check_frequency = std::chrono::seconds(0),
                                        .ssl = std::nullopt,
@@ -309,10 +318,8 @@ void ReplicationState::RestoreReplicationRole(Storage *storage) {
   }
 
   if (GetRole() == replication::ReplicationRole::REPLICA) {
-    replication_server_ = storage->CreateReplicationServer(replication::ReplicationServerConfig{
-        .ip_address = replication::kDefaultReplicationServerIp,
-        .port = port,
-    });
+    io::network::Endpoint endpoint(replication::kDefaultReplicationServerIp, port);
+    replication_server_ = storage->CreateReplicationServer(std::move(endpoint), {});
     bool res = replication_server_->Start();
     if (!res) {
       LOG_FATAL("Unable to start the replication server.");
@@ -345,16 +352,14 @@ void ReplicationState::RestoreReplicas(Storage *storage) {
       continue;
     }
 
-    auto ret = RegisterReplica(replication::RegistrationMode::CAN_BE_INVALID,
-                               replication::ReplicationClientConfig{
-                                   .name = replica_status.name,
-                                   .mode = replica_status.sync_mode,
-                                   .ip_address = replica_status.ip_address,
-                                   .port = replica_status.port,
-                                   .replica_check_frequency = replica_status.replica_check_frequency,
-                                   .ssl = replica_status.ssl,
-                               },
-                               storage);
+    auto ret =
+        RegisterReplica(std::move(replica_status.name), {std::move(replica_status.ip_address), replica_status.port},
+                        replica_status.sync_mode, replication::RegistrationMode::CAN_BE_INVALID,
+                        {
+                            .replica_check_frequency = replica_status.replica_check_frequency,
+                            .ssl = replica_status.ssl,
+                        },
+                        storage);
 
     if (ret.HasError()) {
       MG_ASSERT(RegisterReplicaError::CONNECTION_FAILED != ret.GetError());

src/storage/v2/replication/replication.hpp

@@ -32,9 +32,14 @@ class Storage;
 class ReplicationServer;
 class ReplicationClient;
 
-enum class RegisterReplicaError : uint8_t { NAME_EXISTS, END_POINT_EXISTS, CONNECTION_FAILED, COULD_NOT_BE_PERSISTED };
-
 struct ReplicationState {
+  enum class RegisterReplicaError : uint8_t {
+    NAME_EXISTS,
+    END_POINT_EXISTS,
+    CONNECTION_FAILED,
+    COULD_NOT_BE_PERSISTED
+  };
+
   // TODO: This mirrors the logic in InMemoryConstructor; make it independent
   ReplicationState(bool restore, std::filesystem::path durability_dir);
 
@@ -45,7 +50,7 @@ struct ReplicationState {
 
   bool SetMainReplicationRole(Storage *storage);  // Set the instance to MAIN
   // TODO: ReplicationServer/Client uses Storage* for RPC callbacks
-  bool SetReplicaRole(const replication::ReplicationServerConfig &config,
+  bool SetReplicaRole(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config,
                       Storage *storage);  // Sets the instance to REPLICA
   // Generic restoration
   void RestoreReplicationRole(Storage *storage);
@@ -59,7 +64,9 @@ struct ReplicationState {
   bool FinalizeTransaction(uint64_t timestamp);
 
   // MAIN connecting to replicas
-  utils::BasicResult<RegisterReplicaError> RegisterReplica(const replication::RegistrationMode registration_mode,
+  utils::BasicResult<RegisterReplicaError> RegisterReplica(std::string name, io::network::Endpoint endpoint,
+                                                           const replication::ReplicationMode replication_mode,
+                                                           const replication::RegistrationMode registration_mode,
                                                            const replication::ReplicationClientConfig &config,
                                                            Storage *storage);
   bool UnregisterReplica(std::string_view name);
@@ -90,8 +97,8 @@ struct ReplicationState {
   void AppendEpoch(std::string new_epoch);
 
  private:
-  bool TryPersistReplicaClient(const replication::ReplicationClientConfig &config);
   bool ShouldStoreAndRestoreReplicationState() const { return nullptr != durability_; }
+
   void SetRole(replication::ReplicationRole role) { return replication_role_.store(role); }
 
   // NOTE: Server is not in MAIN it is in REPLICA

src/storage/v2/replication/replication_client.cpp

@@ -30,12 +30,14 @@ static auto CreateClientContext(const replication::ReplicationClientConfig &conf
                       : communication::ClientContext{};
 }
 
-ReplicationClient::ReplicationClient(Storage *storage, replication::ReplicationClientConfig const &config)
-    : name_{config.name},
+ReplicationClient::ReplicationClient(Storage *storage, std::string name, memgraph::io::network::Endpoint endpoint,
+                                     replication::ReplicationMode mode,
+                                     replication::ReplicationClientConfig const &config)
+    : name_{std::move(name)},
       rpc_context_{CreateClientContext(config)},
-      rpc_client_{io::network::Endpoint(config.ip_address, config.port), &rpc_context_},
+      rpc_client_{std::move(endpoint), &rpc_context_},
       replica_check_frequency_{config.replica_check_frequency},
-      mode_{config.mode},
+      mode_{mode},
       storage_{storage} {}
 
 ReplicationClient::~ReplicationClient() {

src/storage/v2/replication/replication_client.hpp

@@ -66,7 +66,8 @@ class ReplicationClient {
   friend class ReplicaStream;
 
  public:
-  ReplicationClient(Storage *storage, replication::ReplicationClientConfig const &config);
+  ReplicationClient(Storage *storage, std::string name, memgraph::io::network::Endpoint endpoint,
+                    replication::ReplicationMode mode, const replication::ReplicationClientConfig &config);
 
   ReplicationClient(ReplicationClient const &) = delete;
   ReplicationClient &operator=(ReplicationClient const &) = delete;

src/storage/v2/replication/replication_server.cpp

@@ -30,10 +30,9 @@ auto CreateServerContext(const replication::ReplicationServerConfig &config) ->
 constexpr auto kReplictionServerThreads = 1;
 }  // namespace
 
-ReplicationServer::ReplicationServer(const replication::ReplicationServerConfig &config)
+ReplicationServer::ReplicationServer(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config)
     : rpc_server_context_{CreateServerContext(config)},
-      rpc_server_{io::network::Endpoint{config.ip_address, config.port}, &rpc_server_context_,
-                  kReplictionServerThreads} {
+      rpc_server_{std::move(endpoint), &rpc_server_context_, kReplictionServerThreads} {
   rpc_server_.Register<replication::FrequentHeartbeatRpc>([](auto *req_reader, auto *res_builder) {
     spdlog::debug("Received FrequentHeartbeatRpc");
     FrequentHeartbeatHandler(req_reader, res_builder);

src/storage/v2/replication/replication_server.hpp

@@ -20,7 +20,7 @@ namespace memgraph::storage {
 
 class ReplicationServer {
  public:
-  explicit ReplicationServer(const replication::ReplicationServerConfig &config);
+  explicit ReplicationServer(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config);
   ReplicationServer(const ReplicationServer &) = delete;
   ReplicationServer(ReplicationServer &&) = delete;
   ReplicationServer &operator=(const ReplicationServer &) = delete;

src/storage/v2/storage.cpp

@@ -9,6 +9,7 @@
 // by the Apache License, Version 2.0, included in the file
 // licenses/APL.txt.
 
+#include <thread>
 #include "absl/container/flat_hash_set.h"
 #include "spdlog/spdlog.h"
 
@@ -21,6 +22,7 @@
 #include "utils/exceptions.hpp"
 #include "utils/file.hpp"
 #include "utils/logging.hpp"
+#include "utils/stacktrace.hpp"
 #include "utils/stat.hpp"
 #include "utils/timer.hpp"
 #include "utils/typeinfo.hpp"
@@ -64,7 +66,10 @@ Storage::Accessor::Accessor(Storage *storage, IsolationLevel isolation_level, St
       storage_guard_(storage_->main_lock_),
       transaction_(storage->CreateTransaction(isolation_level, storage_mode)),
       is_transaction_active_(true),
-      creation_storage_mode_(storage_mode) {}
+      creation_storage_mode_(storage_mode) {
+  spdlog::error("{} Accessor::Accessor acquired shared lock.\nStack: {}", std::this_thread::get_id(),
+                "" /*utils::Stacktrace().dump()*/);
+}
 
 Storage::Accessor::Accessor(Accessor &&other) noexcept
     : storage_(other.storage_),
@@ -74,23 +79,30 @@ Storage::Accessor::Accessor(Accessor &&other) noexcept
       is_transaction_active_(other.is_transaction_active_),
       creation_storage_mode_(other.creation_storage_mode_) {
   // Don't allow the other accessor to abort our transaction in destructor.
+  spdlog::error("{} move Accessor::Accessor acquired shared lock", std::this_thread::get_id());
   other.is_transaction_active_ = false;
   other.commit_timestamp_.reset();
 }
 
 IndicesInfo Storage::ListAllIndices() const {
+  spdlog::error("{} ListAllIndices trying to acquire shared lock", std::this_thread::get_id());
   std::shared_lock<utils::RWLock> storage_guard_(main_lock_);
+  spdlog::error("{} ListAllIndices acquired shared lock", std::this_thread::get_id());
   return {indices_.label_index_->ListIndices(), indices_.label_property_index_->ListIndices()};
 }
 
 ConstraintsInfo Storage::ListAllConstraints() const {
+  spdlog::error("{} ListAllConstraints trying to acquire shared lock", std::this_thread::get_id());
   std::shared_lock<utils::RWLock> storage_guard_(main_lock_);
+  spdlog::error("{} ListAllConstraints acquired shared lock", std::this_thread::get_id());
   return {constraints_.existence_constraints_->ListConstraints(), constraints_.unique_constraints_->ListConstraints()};
 }
 
 /// Main lock is taken by the caller.
 void Storage::SetStorageMode(StorageMode storage_mode) {
+  spdlog::error("{} SetStorageMode trying to acquire lock", std::this_thread::get_id());
   std::unique_lock main_guard{main_lock_};
+  spdlog::error("{} SetStorageMode acquired lock", std::this_thread::get_id());
   MG_ASSERT(
       (storage_mode_ == StorageMode::IN_MEMORY_ANALYTICAL || storage_mode_ == StorageMode::IN_MEMORY_TRANSACTIONAL) &&
       (storage_mode == StorageMode::IN_MEMORY_ANALYTICAL || storage_mode == StorageMode::IN_MEMORY_TRANSACTIONAL));
@@ -105,7 +117,9 @@ StorageMode Storage::GetStorageMode() const { return storage_mode_; }
 IsolationLevel Storage::GetIsolationLevel() const noexcept { return isolation_level_; }
 
 utils::BasicResult<Storage::SetIsolationLevelError> Storage::SetIsolationLevel(IsolationLevel isolation_level) {
+  spdlog::error("{} SetIsolationLevel trying to acquire lock", std::this_thread::get_id());
   std::unique_lock main_guard{main_lock_};
+  spdlog::error("{} SetIsolationLevel acquired lock", std::this_thread::get_id());
   if (storage_mode_ == storage::StorageMode::IN_MEMORY_ANALYTICAL) {
     return Storage::SetIsolationLevelError::DisabledForAnalyticalMode;
   }

src/storage/v2/storage.hpp

@@ -101,7 +101,7 @@ class Storage {
 
     Accessor(Accessor &&other) noexcept;
 
-    virtual ~Accessor() {}
+    virtual ~Accessor() { spdlog::error("{} Accessor::Accessor unlock shared lock", std::this_thread::get_id()); }
 
     virtual VertexAccessor CreateVertex() = 0;
 
@@ -315,23 +315,29 @@ class Storage {
 
   virtual void EstablishNewEpoch() = 0;
 
-  virtual auto CreateReplicationClient(replication::ReplicationClientConfig const &config)
+  virtual auto CreateReplicationClient(std::string name, io::network::Endpoint endpoint,
+                                       replication::ReplicationMode mode,
+                                       replication::ReplicationClientConfig const &config)
       -> std::unique_ptr<ReplicationClient> = 0;
 
-  virtual auto CreateReplicationServer(const replication::ReplicationServerConfig &config)
+  virtual auto CreateReplicationServer(io::network::Endpoint endpoint,
+                                       replication::ReplicationServerConfig const &config)
       -> std::unique_ptr<ReplicationServer> = 0;
 
   /// REPLICATION
-  bool SetReplicaRole(const replication::ReplicationServerConfig &config) {
-    return replication_state_.SetReplicaRole(config, this);
+  bool SetReplicaRole(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config) {
+    return replication_state_.SetReplicaRole(std::move(endpoint), config, this);
   }
   bool SetMainReplicationRole() { return replication_state_.SetMainReplicationRole(this); }
 
   /// @pre The instance should have a MAIN role
   /// @pre Timeout can only be set for SYNC replication
-  auto RegisterReplica(const replication::RegistrationMode registration_mode,
+  auto RegisterReplica(std::string name, io::network::Endpoint endpoint,
+                       const replication::ReplicationMode replication_mode,
+                       const replication::RegistrationMode registration_mode,
                        const replication::ReplicationClientConfig &config) {
-    return replication_state_.RegisterReplica(registration_mode, config, this);
+    return replication_state_.RegisterReplica(std::move(name), std::move(endpoint), replication_mode, registration_mode,
+                                              config, this);
   }
   /// @pre The instance should have a MAIN role
   bool UnregisterReplica(const std::string &name) { return replication_state_.UnregisterReplica(name); }

src/storage/v2/transaction.hpp

@@ -105,7 +105,6 @@ struct Transaction {
 
   // Store modified edges GID mapped to changed Delta and serialized edge key
   ModifiedEdgesMap modified_edges_;
-  utils::SkipList<Edge> edges_;
 };
 
 inline bool operator==(const Transaction &first, const Transaction &second) {

src/utils/rw_lock.hpp

@@ -1,4 +1,4 @@
-// Copyright 2022 Memgraph Ltd.
+// 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
@@ -14,6 +14,7 @@
 
 #include <pthread.h>
 #include <unistd.h>
+#include <thread>
 
 #include <cerrno>
 
@@ -74,7 +75,11 @@ class RWLock {
 
   ~RWLock() { pthread_rwlock_destroy(&lock_); }
 
-  void lock() { MG_ASSERT(pthread_rwlock_wrlock(&lock_) == 0, "Couldn't lock utils::RWLock!"); }
+  void lock() {
+    spdlog::error("{} Trying to lock unique lock", std::this_thread::get_id());
+    MG_ASSERT(pthread_rwlock_wrlock(&lock_) == 0, "Couldn't lock utils::RWLock!");
+    spdlog::error("{} Lock unique lock", std::this_thread::get_id());
+  }
 
   bool try_lock() {
     int err = pthread_rwlock_trywrlock(&lock_);
@@ -83,13 +88,19 @@ class RWLock {
     return false;
   }
 
-  void unlock() { MG_ASSERT(pthread_rwlock_unlock(&lock_) == 0, "Couldn't unlock utils::RWLock!"); }
+  void unlock() {
+    spdlog::error("{} Trying to unlock unique lock", std::this_thread::get_id());
+    MG_ASSERT(pthread_rwlock_unlock(&lock_) == 0, "Couldn't unlock utils::RWLock!");
+    spdlog::error("{} Unlock unique lock", std::this_thread::get_id());
+  }
 
   void lock_shared() {
     int err;
     while (true) {
+      spdlog::error("{} Trying to acquire lock_shared lock", std::this_thread::get_id());
       err = pthread_rwlock_rdlock(&lock_);
       if (err == 0) {
+        spdlog::error("{} Acquired lock_shared lock", std::this_thread::get_id());
         return;
       } else if (err == EAGAIN) {
         continue;
@@ -115,7 +126,11 @@ class RWLock {
     }
   }
 
-  void unlock_shared() { MG_ASSERT(pthread_rwlock_unlock(&lock_) == 0, "Couldn't unlock shared utils::RWLock!"); }
+  void unlock_shared() {
+    spdlog::error("{} Trying to  unlock shared lock", std::this_thread::get_id());
+    MG_ASSERT(pthread_rwlock_unlock(&lock_) == 0, "Couldn't unlock shared utils::RWLock!");
+    spdlog::error("{} unlock shared lock", std::this_thread::get_id());
+  }
 
  private:
   pthread_rwlock_t lock_ = PTHREAD_RWLOCK_INITIALIZER;

tests/e2e/configuration/default_config.py

@@ -156,11 +156,6 @@ startup_config_dict = {
         "100000",
         "Issue a 'fsync' call after this amount of transactions are written to the WAL file. Set to 1 for fully synchronous operation.",
     ),
-    "storage_mode": (
-        "IN_MEMORY_TRANSACTIONAL",
-        "IN_MEMORY_TRANSACTIONAL",
-        "Default storage mode Memgraph uses. Allowed values: IN_MEMORY_TRANSACTIONAL, IN_MEMORY_ANALYTICAL, ON_DISK_TRANSACTIONAL",
-    ),
     "storage_wal_file_size_kib": ("20480", "20480", "Minimum file size of each WAL file."),
     "storage_delete_on_drop": (
         "true",