Edges in tx

* Add --storage-mode flag

config/flags.yaml

@@ -99,6 +99,10 @@ 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,8 +10,10 @@
 // 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>;
 
@@ -20,10 +22,11 @@ namespace memgraph::dbms {
 Database::Database(const storage::Config &config)
     : trigger_store_(config.durability.storage_directory / "triggers"),
       streams_{config.durability.storage_directory / "streams"} {
-  if (config.force_on_disk || utils::DirExists(config.disk.main_storage_directory)) {
+  if (config.storage_mode == memgraph::storage::StorageMode::ON_DISK_TRANSACTIONAL || 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);
+    storage_ = std::make_unique<storage::InMemoryStorage>(config, config.storage_mode);
   }
 }
 

src/flags/CMakeLists.txt

@@ -4,6 +4,7 @@ add_library(mg-flags STATIC audit.cpp
                             isolation_level.cpp
                             log_level.cpp
                             memory_limit.cpp
-                            run_time_configurable.cpp)
+                            run_time_configurable.cpp
+                            storage_mode.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,3 +17,4 @@
 #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

@@ -0,0 +1,55 @@
+// 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

@@ -0,0 +1,19 @@
+// 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,7 +291,8 @@ 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"}};
+               .wal_directory = FLAGS_data_directory + "/rocksdb_wal"},
+      .storage_mode = memgraph::flags::ParseStorageMode()};
   if (FLAGS_storage_snapshot_interval_sec == 0) {
     if (FLAGS_storage_wal_enabled) {
       LOG_FATAL(

src/query/interpreter.cpp

@@ -238,9 +238,10 @@ 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(
-              io::network::Endpoint(storage::replication::kDefaultReplicationServerIp, static_cast<uint16_t>(*port)),
-              storage::replication::ReplicationServerConfig{})) {
+      if (!db_->SetReplicaRole(storage::replication::ReplicationServerConfig{
+              .ip_address = storage::replication::kDefaultReplicationServerIp,
+              .port = static_cast<uint16_t>(*port),
+          })) {
         throw QueryRuntimeException("Couldn't set role to replica!");
       }
     }
@@ -286,9 +287,14 @@ 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(name, {std::move(ip), port}, repl_mode,
-                                      storage::replication::RegistrationMode::MUST_BE_INSTANTLY_VALID,
-                                      {.replica_check_frequency = replica_check_frequency, .ssl = std::nullopt});
+      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});
       if (ret.HasError()) {
         throw QueryRuntimeException(fmt::format("Couldn't register replica '{}'!", name));
       }

src/storage/v2/config.hpp

@@ -15,6 +15,7 @@
 #include <cstdint>
 #include <filesystem>
 #include "storage/v2/isolation_level.hpp"
+#include "storage/v2/storage_mode.hpp"
 #include "utils/exceptions.hpp"
 
 namespace memgraph::storage {
@@ -79,6 +80,7 @@ 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 {
-  DiskEdgeKey(const std::string_view keyView) : key(keyView) {}
+  explicit 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 = edges_.access();
+  auto edge_acc = transaction_.edges_.access();
   auto res = edge_acc.find(edge_gid);
   if (res != edge_acc.end()) {
     return std::nullopt;
@@ -1203,7 +1203,8 @@ 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() : edges_.access();
+    auto acc =
+        edge_import_mode_active ? disk_storage->edge_import_mode_cache_->AccessToEdges() : transaction_.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!");
@@ -1240,7 +1241,8 @@ 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() : edges_.access();
+    auto acc =
+        edge_import_mode_active ? disk_storage->edge_import_mode_cache_->AccessToEdges() : transaction_.edges_.access();
     auto *delta = CreateDeleteObjectDelta(&transaction_);
     auto [it, inserted] = acc.insert(Edge(gid, delta));
     MG_ASSERT(inserted, "The edge must be inserted here!");
@@ -1574,7 +1576,7 @@ utils::BasicResult<StorageDataManipulationError, void> DiskStorage::DiskAccessor
         return vertices_flush_res.GetError();
       }
 
-      if (auto modified_edges_res = FlushModifiedEdges(edges_.access()); modified_edges_res.HasError()) {
+      if (auto modified_edges_res = FlushModifiedEdges(transaction_.edges_.access()); modified_edges_res.HasError()) {
         Abort();
         return modified_edges_res.GetError();
       }

src/storage/v2/disk/storage.hpp

@@ -271,7 +271,6 @@ 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_;
@@ -375,13 +374,12 @@ class DiskStorage final : public Storage {
   EdgeImportMode edge_import_status_{EdgeImportMode::INACTIVE};
   std::unique_ptr<EdgeImportModeCache> edge_import_mode_cache_{nullptr};
 
-  auto CreateReplicationClient(std::string name, io::network::Endpoint endpoint, replication::ReplicationMode mode,
-                               const replication::ReplicationClientConfig &config)
+  auto CreateReplicationClient(replication::ReplicationClientConfig const &config)
       -> std::unique_ptr<ReplicationClient> override {
     throw utils::BasicException("Disk storage mode does not support replication.");
   }
 
-  auto CreateReplicationServer(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config)
+  auto CreateReplicationServer(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,10 +102,9 @@ uint64_t ReplicateCurrentWal(CurrentWalHandler &stream, durability::WalFile cons
 
 ////// ReplicationClient //////
 
-InMemoryReplicationClient::InMemoryReplicationClient(InMemoryStorage *storage, std::string name,
-                                                     io::network::Endpoint endpoint, replication::ReplicationMode mode,
+InMemoryReplicationClient::InMemoryReplicationClient(InMemoryStorage *storage,
                                                      const replication::ReplicationClientConfig &config)
-    : ReplicationClient{storage, std::move(name), std::move(endpoint), mode, config} {}
+    : ReplicationClient{storage, config} {}
 
 void InMemoryReplicationClient::RecoverReplica(uint64_t replica_commit) {
   spdlog::debug("Starting replica recover");

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

@@ -18,8 +18,7 @@ class InMemoryStorage;
 
 class InMemoryReplicationClient : public ReplicationClient {
  public:
-  InMemoryReplicationClient(InMemoryStorage *storage, std::string name, io::network::Endpoint endpoint,
-                            replication::ReplicationMode mode, const replication::ReplicationClientConfig &config = {});
+  InMemoryReplicationClient(InMemoryStorage *storage, 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, memgraph::io::network::Endpoint endpoint,
+InMemoryReplicationServer::InMemoryReplicationServer(InMemoryStorage *storage,
                                                      const replication::ReplicationServerConfig &config)
-    : ReplicationServer{std::move(endpoint), config}, storage_(storage) {
+    : ReplicationServer{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,8 +20,7 @@ class InMemoryStorage;
 
 class InMemoryReplicationServer : public ReplicationServer {
  public:
-  explicit InMemoryReplicationServer(InMemoryStorage *storage, io::network::Endpoint endpoint,
-                                     const replication::ReplicationServerConfig &config);
+  explicit InMemoryReplicationServer(InMemoryStorage *storage, const replication::ReplicationServerConfig &config);
 
  private:
   // RPC handlers

src/storage/v2/inmemory/storage.cpp

@@ -22,13 +22,15 @@ namespace memgraph::storage {
 
 using OOMExceptionEnabler = utils::MemoryTracker::OutOfMemoryExceptionEnabler;
 
-InMemoryStorage::InMemoryStorage(Config config)
-    : Storage(config, StorageMode::IN_MEMORY_TRANSACTIONAL),
+InMemoryStorage::InMemoryStorage(Config config, StorageMode storage_mode)
+    : Storage(config, storage_mode),
       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
@@ -147,6 +149,8 @@ InMemoryStorage::InMemoryStorage(Config config)
   }
 }
 
+InMemoryStorage::InMemoryStorage(Config config) : InMemoryStorage(config, StorageMode::IN_MEMORY_TRANSACTIONAL) {}
+
 InMemoryStorage::~InMemoryStorage() {
   if (config_.gc.type == Config::Gc::Type::PERIODIC) {
     gc_runner_.Stop();
@@ -1828,16 +1832,14 @@ utils::FileRetainer::FileLockerAccessor::ret_type InMemoryStorage::UnlockPath()
   return true;
 }
 
-auto InMemoryStorage::CreateReplicationClient(std::string name, io::network::Endpoint endpoint,
-                                              replication::ReplicationMode mode,
-                                              replication::ReplicationClientConfig const &config)
+auto InMemoryStorage::CreateReplicationClient(replication::ReplicationClientConfig const &config)
     -> std::unique_ptr<ReplicationClient> {
-  return std::make_unique<InMemoryReplicationClient>(this, std::move(name), std::move(endpoint), mode, config);
+  return std::make_unique<InMemoryReplicationClient>(this, config);
 }
 
 std::unique_ptr<ReplicationServer> InMemoryStorage::CreateReplicationServer(
-    io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config) {
-  return std::make_unique<InMemoryReplicationServer>(this, std::move(endpoint), config);
+    const replication::ReplicationServerConfig &config) {
+  return std::make_unique<InMemoryReplicationServer>(this, config);
 }
 
 }  // namespace memgraph::storage

src/storage/v2/inmemory/storage.hpp

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

src/storage/v2/replication/config.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
@@ -15,8 +15,15 @@
 #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.
@@ -33,6 +40,8 @@ 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(ReplicationState::RegisterReplicaError error) {
-  using enum ReplicationState::RegisterReplicaError;
+std::string RegisterReplicaErrorToString(RegisterReplicaError error) {
+  using enum RegisterReplicaError;
   switch (error) {
     case NAME_EXISTS:
       return "NAME_EXISTS";
@@ -147,84 +147,75 @@ bool storage::ReplicationState::FinalizeTransaction(uint64_t timestamp) {
   return finalized_on_all_replicas;
 }
 
-utils::BasicResult<ReplicationState::RegisterReplicaError> ReplicationState::RegisterReplica(
-    std::string name, io::network::Endpoint endpoint, const replication::ReplicationMode replication_mode,
+utils::BasicResult<RegisterReplicaError> ReplicationState::RegisterReplica(
     const replication::RegistrationMode registration_mode, const replication::ReplicationClientConfig &config,
     Storage *storage) {
   MG_ASSERT(GetRole() == replication::ReplicationRole::MAIN, "Only main instance can register a replica!");
 
-  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 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);
+  };
 
-  if (name_exists) {
-    return RegisterReplicaError::NAME_EXISTS;
-  }
+  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);
+  };
 
-  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; });
-  });
+  auto task = [&](auto &clients) -> utils::BasicResult<RegisterReplicaError> {
+    if (name_check(clients)) {
+      return RegisterReplicaError::NAME_EXISTS;
+    }
 
-  if (end_point_exists) {
-    return RegisterReplicaError::END_POINT_EXISTS;
-  }
+    if (endpoint_check(clients)) {
+      return RegisterReplicaError::END_POINT_EXISTS;
+    }
 
-  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);
+    if (!TryPersistReplicaClient(config)) {
       return RegisterReplicaError::COULD_NOT_BE_PERSISTED;
     }
-  }
 
-  auto client = storage->CreateReplicationClient(std::move(name), std::move(endpoint), replication_mode, config);
-  client->Start();
+    auto client = storage->CreateReplicationClient(config);
+    client->Start();
 
-  if (client->State() == replication::ReplicaState::INVALID) {
-    if (replication::RegistrationMode::CAN_BE_INVALID != registration_mode) {
-      return RegisterReplicaError::CONNECTION_FAILED;
+    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());
     }
 
-    spdlog::warn("Connection failed when registering replica {}. Replica will still be registered.", client->Name());
-  }
+    clients.push_back(std::move(client));
+    return {};
+  };
 
-  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 {};
-      });
+  return replication_clients_.WithLock(task);
 }
 
-bool ReplicationState::SetReplicaRole(io::network::Endpoint endpoint,
-                                      const replication::ReplicationServerConfig &config, Storage *storage) {
+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) {
   // We don't want to restart the server if we're already a REPLICA
   if (GetRole() == replication::ReplicationRole::REPLICA) {
     return false;
   }
 
-  auto port = endpoint.port;  // assigning because we will move the endpoint
-  replication_server_ = storage->CreateReplicationServer(std::move(endpoint), config);
+  replication_server_ = storage->CreateReplicationServer(config);
   bool res = replication_server_->Start();
   if (!res) {
     spdlog::error("Unable to start the replication server.");
@@ -235,8 +226,8 @@ bool ReplicationState::SetReplicaRole(io::network::Endpoint endpoint,
     // 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 = "",
-                                       .port = port,
+                                       .ip_address = config.ip_address,
+                                       .port = config.port,
                                        .sync_mode = replication::ReplicationMode::SYNC,
                                        .replica_check_frequency = std::chrono::seconds(0),
                                        .ssl = std::nullopt,
@@ -318,8 +309,10 @@ void ReplicationState::RestoreReplicationRole(Storage *storage) {
   }
 
   if (GetRole() == replication::ReplicationRole::REPLICA) {
-    io::network::Endpoint endpoint(replication::kDefaultReplicationServerIp, port);
-    replication_server_ = storage->CreateReplicationServer(std::move(endpoint), {});
+    replication_server_ = storage->CreateReplicationServer(replication::ReplicationServerConfig{
+        .ip_address = replication::kDefaultReplicationServerIp,
+        .port = port,
+    });
     bool res = replication_server_->Start();
     if (!res) {
       LOG_FATAL("Unable to start the replication server.");
@@ -352,14 +345,16 @@ void ReplicationState::RestoreReplicas(Storage *storage) {
       continue;
     }
 
-    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);
+    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);
 
     if (ret.HasError()) {
       MG_ASSERT(RegisterReplicaError::CONNECTION_FAILED != ret.GetError());

src/storage/v2/replication/replication.hpp

@@ -32,14 +32,9 @@ class Storage;
 class ReplicationServer;
 class ReplicationClient;
 
-struct ReplicationState {
-  enum class RegisterReplicaError : uint8_t {
-    NAME_EXISTS,
-    END_POINT_EXISTS,
-    CONNECTION_FAILED,
-    COULD_NOT_BE_PERSISTED
-  };
+enum class RegisterReplicaError : uint8_t { NAME_EXISTS, END_POINT_EXISTS, CONNECTION_FAILED, COULD_NOT_BE_PERSISTED };
 
+struct ReplicationState {
   // TODO: This mirrors the logic in InMemoryConstructor; make it independent
   ReplicationState(bool restore, std::filesystem::path durability_dir);
 
@@ -50,7 +45,7 @@ struct ReplicationState {
 
   bool SetMainReplicationRole(Storage *storage);  // Set the instance to MAIN
   // TODO: ReplicationServer/Client uses Storage* for RPC callbacks
-  bool SetReplicaRole(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config,
+  bool SetReplicaRole(const replication::ReplicationServerConfig &config,
                       Storage *storage);  // Sets the instance to REPLICA
   // Generic restoration
   void RestoreReplicationRole(Storage *storage);
@@ -64,9 +59,7 @@ struct ReplicationState {
   bool FinalizeTransaction(uint64_t timestamp);
 
   // MAIN connecting to replicas
-  utils::BasicResult<RegisterReplicaError> RegisterReplica(std::string name, io::network::Endpoint endpoint,
-                                                           const replication::ReplicationMode replication_mode,
-                                                           const replication::RegistrationMode registration_mode,
+  utils::BasicResult<RegisterReplicaError> RegisterReplica(const replication::RegistrationMode registration_mode,
                                                            const replication::ReplicationClientConfig &config,
                                                            Storage *storage);
   bool UnregisterReplica(std::string_view name);
@@ -97,8 +90,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,14 +30,12 @@ static auto CreateClientContext(const replication::ReplicationClientConfig &conf
                       : communication::ClientContext{};
 }
 
-ReplicationClient::ReplicationClient(Storage *storage, std::string name, memgraph::io::network::Endpoint endpoint,
-                                     replication::ReplicationMode mode,
-                                     replication::ReplicationClientConfig const &config)
-    : name_{std::move(name)},
+ReplicationClient::ReplicationClient(Storage *storage, replication::ReplicationClientConfig const &config)
+    : name_{config.name},
       rpc_context_{CreateClientContext(config)},
-      rpc_client_{std::move(endpoint), &rpc_context_},
+      rpc_client_{io::network::Endpoint(config.ip_address, config.port), &rpc_context_},
       replica_check_frequency_{config.replica_check_frequency},
-      mode_{mode},
+      mode_{config.mode},
       storage_{storage} {}
 
 ReplicationClient::~ReplicationClient() {

src/storage/v2/replication/replication_client.hpp

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

src/storage/v2/replication/replication_server.cpp

@@ -30,9 +30,10 @@ auto CreateServerContext(const replication::ReplicationServerConfig &config) ->
 constexpr auto kReplictionServerThreads = 1;
 }  // namespace
 
-ReplicationServer::ReplicationServer(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config)
+ReplicationServer::ReplicationServer(const replication::ReplicationServerConfig &config)
     : rpc_server_context_{CreateServerContext(config)},
-      rpc_server_{std::move(endpoint), &rpc_server_context_, kReplictionServerThreads} {
+      rpc_server_{io::network::Endpoint{config.ip_address, config.port}, &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(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config);
+  explicit ReplicationServer(const replication::ReplicationServerConfig &config);
   ReplicationServer(const ReplicationServer &) = delete;
   ReplicationServer(ReplicationServer &&) = delete;
   ReplicationServer &operator=(const ReplicationServer &) = delete;

src/storage/v2/storage.hpp

@@ -315,29 +315,23 @@ class Storage {
 
   virtual void EstablishNewEpoch() = 0;
 
-  virtual auto CreateReplicationClient(std::string name, io::network::Endpoint endpoint,
-                                       replication::ReplicationMode mode,
-                                       replication::ReplicationClientConfig const &config)
+  virtual auto CreateReplicationClient(replication::ReplicationClientConfig const &config)
       -> std::unique_ptr<ReplicationClient> = 0;
 
-  virtual auto CreateReplicationServer(io::network::Endpoint endpoint,
-                                       replication::ReplicationServerConfig const &config)
+  virtual auto CreateReplicationServer(const replication::ReplicationServerConfig &config)
       -> std::unique_ptr<ReplicationServer> = 0;
 
   /// REPLICATION
-  bool SetReplicaRole(io::network::Endpoint endpoint, const replication::ReplicationServerConfig &config) {
-    return replication_state_.SetReplicaRole(std::move(endpoint), config, this);
+  bool SetReplicaRole(const replication::ReplicationServerConfig &config) {
+    return replication_state_.SetReplicaRole(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(std::string name, io::network::Endpoint endpoint,
-                       const replication::ReplicationMode replication_mode,
-                       const replication::RegistrationMode registration_mode,
+  auto RegisterReplica(const replication::RegistrationMode registration_mode,
                        const replication::ReplicationClientConfig &config) {
-    return replication_state_.RegisterReplica(std::move(name), std::move(endpoint), replication_mode, registration_mode,
-                                              config, this);
+    return replication_state_.RegisterReplica(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,6 +105,7 @@ 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) {

tests/e2e/configuration/default_config.py

@@ -156,6 +156,11 @@ 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",