Merge branch 'master' into temporal-min-max

src/coordination/CMakeLists.txt

@@ -23,6 +23,7 @@ target_sources(mg-coordination
         include/nuraft/coordinator_state_manager.hpp
 
         PRIVATE
+        coordinator_config.cpp
         coordinator_client.cpp
         coordinator_state.cpp
         coordinator_rpc.cpp

src/coordination/coordinator_config.cpp

@@ -0,0 +1,54 @@
+// 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.
+
+#ifdef MG_ENTERPRISE
+
+#include "coordination/coordinator_config.hpp"
+
+namespace memgraph::coordination {
+
+void to_json(nlohmann::json &j, ReplClientInfo const &config) {
+  j = nlohmann::json{{"instance_name", config.instance_name},
+                     {"replication_mode", config.replication_mode},
+                     {"replication_ip_address", config.replication_ip_address},
+                     {"replication_port", config.replication_port}};
+}
+
+void from_json(nlohmann::json const &j, ReplClientInfo &config) {
+  config.instance_name = j.at("instance_name").get<std::string>();
+  config.replication_mode = j.at("replication_mode").get<replication_coordination_glue::ReplicationMode>();
+  config.replication_ip_address = j.at("replication_ip_address").get<std::string>();
+  config.replication_port = j.at("replication_port").get<uint16_t>();
+}
+
+void to_json(nlohmann::json &j, CoordinatorClientConfig const &config) {
+  j = nlohmann::json{{"instance_name", config.instance_name},
+                     {"ip_address", config.ip_address},
+                     {"port", config.port},
+                     {"instance_health_check_frequency_sec", config.instance_health_check_frequency_sec.count()},
+                     {"instance_down_timeout_sec", config.instance_down_timeout_sec.count()},
+                     {"instance_get_uuid_frequency_sec", config.instance_get_uuid_frequency_sec.count()},
+                     {"replication_client_info", config.replication_client_info}};
+}
+
+void from_json(nlohmann::json const &j, CoordinatorClientConfig &config) {
+  config.instance_name = j.at("instance_name").get<std::string>();
+  config.ip_address = j.at("ip_address").get<std::string>();
+  config.port = j.at("port").get<uint16_t>();
+  config.instance_health_check_frequency_sec =
+      std::chrono::seconds{j.at("instance_health_check_frequency_sec").get<int>()};
+  config.instance_down_timeout_sec = std::chrono::seconds{j.at("instance_down_timeout_sec").get<int>()};
+  config.instance_get_uuid_frequency_sec = std::chrono::seconds{j.at("instance_get_uuid_frequency_sec").get<int>()};
+  config.replication_client_info = j.at("replication_client_info").get<ReplClientInfo>();
+}
+
+}  // namespace memgraph::coordination
+#endif

src/coordination/coordinator_instance.cpp

@@ -100,8 +100,7 @@ auto CoordinatorInstance::ShowInstances() const -> std::vector<InstanceStatus> {
             .health = "unknown"};  // TODO: (andi) Get this info from RAFT and test it or when we will move
                                    // CoordinatorState to every instance, we can be smarter about this using our RPC.
   };
-
-  auto instances_status = utils::fmap(coord_instance_to_status, raft_state_.GetAllCoordinators());
+  auto instances_status = utils::fmap(raft_state_.GetAllCoordinators(), coord_instance_to_status);
 
   if (raft_state_.IsLeader()) {
     auto const stringify_repl_role = [this](ReplicationInstance const &instance) -> std::string {

src/coordination/coordinator_state_machine.cpp

@@ -30,77 +30,67 @@ auto CoordinatorStateMachine::IsReplica(std::string_view instance_name) const ->
   return cluster_state_.IsReplica(instance_name);
 }
 
-auto CoordinatorStateMachine::CreateLog(std::string_view log) -> ptr<buffer> {
-  ptr<buffer> log_buf = buffer::alloc(sizeof(uint32_t) + log.size());
+auto CoordinatorStateMachine::CreateLog(nlohmann::json &&log) -> ptr<buffer> {
+  auto const log_dump = log.dump();
+  ptr<buffer> log_buf = buffer::alloc(sizeof(uint32_t) + log_dump.size());
   buffer_serializer bs(log_buf);
-  bs.put_str(log.data());
+  bs.put_str(log_dump);
   return log_buf;
 }
 
 auto CoordinatorStateMachine::SerializeRegisterInstance(CoordinatorClientConfig const &config) -> ptr<buffer> {
-  auto const str_log = fmt::format("{}*register", config.ToString());
-  return CreateLog(str_log);
+  return CreateLog({{"action", RaftLogAction::REGISTER_REPLICATION_INSTANCE}, {"info", config}});
 }
 
 auto CoordinatorStateMachine::SerializeUnregisterInstance(std::string_view instance_name) -> ptr<buffer> {
-  auto const str_log = fmt::format("{}*unregister", instance_name);
-  return CreateLog(str_log);
+  return CreateLog({{"action", RaftLogAction::UNREGISTER_REPLICATION_INSTANCE}, {"info", instance_name}});
 }
 
 auto CoordinatorStateMachine::SerializeSetInstanceAsMain(std::string_view instance_name) -> ptr<buffer> {
-  auto const str_log = fmt::format("{}*promote", instance_name);
-  return CreateLog(str_log);
+  return CreateLog({{"action", RaftLogAction::SET_INSTANCE_AS_MAIN}, {"info", instance_name}});
 }
 
 auto CoordinatorStateMachine::SerializeSetInstanceAsReplica(std::string_view instance_name) -> ptr<buffer> {
-  auto const str_log = fmt::format("{}*demote", instance_name);
-  return CreateLog(str_log);
+  return CreateLog({{"action", RaftLogAction::SET_INSTANCE_AS_REPLICA}, {"info", instance_name}});
 }
 
 auto CoordinatorStateMachine::SerializeUpdateUUID(utils::UUID const &uuid) -> ptr<buffer> {
-  auto const str_log = fmt::format("{}*update_uuid", nlohmann::json{{"uuid", uuid}}.dump());
-  return CreateLog(str_log);
+  return CreateLog({{"action", RaftLogAction::UPDATE_UUID}, {"info", uuid}});
 }
 
 auto CoordinatorStateMachine::DecodeLog(buffer &data) -> std::pair<TRaftLog, RaftLogAction> {
   buffer_serializer bs(data);
+  auto const json = nlohmann::json::parse(bs.get_str());
 
-  auto const log_str = bs.get_str();
-  auto const sep = log_str.find('*');
-  auto const action = log_str.substr(sep + 1);
-  auto const info = log_str.substr(0, sep);
+  auto const action = json["action"].get<RaftLogAction>();
+  auto const &info = json["info"];
 
-  if (action == "register") {
-    return {CoordinatorClientConfig::FromString(info), RaftLogAction::REGISTER_REPLICATION_INSTANCE};
+  switch (action) {
+    case RaftLogAction::REGISTER_REPLICATION_INSTANCE:
+      return {info.get<CoordinatorClientConfig>(), action};
+    case RaftLogAction::UPDATE_UUID:
+      return {info.get<utils::UUID>(), action};
+    case RaftLogAction::UNREGISTER_REPLICATION_INSTANCE:
+    case RaftLogAction::SET_INSTANCE_AS_MAIN:
+      [[fallthrough]];
+    case RaftLogAction::SET_INSTANCE_AS_REPLICA:
+      return {info.get<std::string>(), action};
   }
-  if (action == "unregister") {
-    return {info, RaftLogAction::UNREGISTER_REPLICATION_INSTANCE};
-  }
-  if (action == "promote") {
-    return {info, RaftLogAction::SET_INSTANCE_AS_MAIN};
-  }
-  if (action == "demote") {
-    return {info, RaftLogAction::SET_INSTANCE_AS_REPLICA};
-  }
-  if (action == "update_uuid") {
-    auto const json = nlohmann::json::parse(info);
-    return {json.at("uuid").get<utils::UUID>(), RaftLogAction::UPDATE_UUID};
-  }
-
   throw std::runtime_error("Unknown action");
 }
 
 auto CoordinatorStateMachine::pre_commit(ulong const /*log_idx*/, buffer & /*data*/) -> ptr<buffer> { return nullptr; }
 
 auto CoordinatorStateMachine::commit(ulong const log_idx, buffer &data) -> ptr<buffer> {
-  buffer_serializer bs(data);
-
   auto const [parsed_data, log_action] = DecodeLog(data);
   cluster_state_.DoAction(parsed_data, log_action);
-
   last_committed_idx_ = log_idx;
-  // TODO: (andi) Don't return nullptr
-  return nullptr;
+
+  // Return raft log number
+  ptr<buffer> ret = buffer::alloc(sizeof(log_idx));
+  buffer_serializer bs_ret(ret);
+  bs_ret.put_u64(log_idx);
+  return ret;
 }
 
 auto CoordinatorStateMachine::commit_config(ulong const log_idx, ptr<cluster_config> & /*new_conf*/) -> void {

src/coordination/include/coordination/coordinator_config.hpp

@@ -22,12 +22,12 @@
 #include <string>
 
 #include <fmt/format.h>
+#include "json/json.hpp"
 
 namespace memgraph::coordination {
 
 inline constexpr auto *kDefaultReplicationServerIp = "0.0.0.0";
 
-// TODO: (andi) JSON serialization for RAFT log.
 struct CoordinatorClientConfig {
   std::string instance_name;
   std::string ip_address;
@@ -43,28 +43,11 @@ struct CoordinatorClientConfig {
   }
 
   struct ReplicationClientInfo {
-    // TODO: (andi) Do we even need here instance_name for this struct?
     std::string instance_name;
     replication_coordination_glue::ReplicationMode replication_mode{};
     std::string replication_ip_address;
     uint16_t replication_port{};
 
-    auto ToString() const -> std::string {
-      return fmt::format("{}#{}#{}#{}", instance_name, replication_ip_address, replication_port,
-                         replication_coordination_glue::ReplicationModeToString(replication_mode));
-    }
-
-    // TODO: (andi) How can I make use of monadic parsers here?
-    static auto FromString(std::string_view log) -> ReplicationClientInfo {
-      ReplicationClientInfo replication_client_info;
-      auto splitted = utils::Split(log, "#");
-      replication_client_info.instance_name = splitted[0];
-      replication_client_info.replication_ip_address = splitted[1];
-      replication_client_info.replication_port = std::stoi(splitted[2]);
-      replication_client_info.replication_mode = replication_coordination_glue::ReplicationModeFromString(splitted[3]);
-      return replication_client_info;
-    }
-
     friend bool operator==(ReplicationClientInfo const &, ReplicationClientInfo const &) = default;
   };
 
@@ -79,25 +62,6 @@ struct CoordinatorClientConfig {
 
   std::optional<SSL> ssl;
 
-  auto ToString() const -> std::string {
-    return fmt::format("{}|{}|{}|{}|{}|{}|{}", instance_name, ip_address, port,
-                       instance_health_check_frequency_sec.count(), instance_down_timeout_sec.count(),
-                       instance_get_uuid_frequency_sec.count(), replication_client_info.ToString());
-  }
-
-  static auto FromString(std::string_view log) -> CoordinatorClientConfig {
-    CoordinatorClientConfig config;
-    auto splitted = utils::Split(log, "|");
-    config.instance_name = splitted[0];
-    config.ip_address = splitted[1];
-    config.port = std::stoi(splitted[2]);
-    config.instance_health_check_frequency_sec = std::chrono::seconds(std::stoi(splitted[3]));
-    config.instance_down_timeout_sec = std::chrono::seconds(std::stoi(splitted[4]));
-    config.instance_get_uuid_frequency_sec = std::chrono::seconds(std::stoi(splitted[5]));
-    config.replication_client_info = ReplicationClientInfo::FromString(splitted[6]);
-    return config;
-  }
-
   friend bool operator==(CoordinatorClientConfig const &, CoordinatorClientConfig const &) = default;
 };
 
@@ -119,5 +83,11 @@ struct CoordinatorServerConfig {
   friend bool operator==(CoordinatorServerConfig const &, CoordinatorServerConfig const &) = default;
 };
 
+void to_json(nlohmann::json &j, CoordinatorClientConfig const &config);
+void from_json(nlohmann::json const &j, CoordinatorClientConfig &config);
+
+void to_json(nlohmann::json &j, ReplClientInfo const &config);
+void from_json(nlohmann::json const &j, ReplClientInfo &config);
+
 }  // namespace memgraph::coordination
 #endif

src/coordination/include/nuraft/coordinator_state_machine.hpp

@@ -47,7 +47,7 @@ class CoordinatorStateMachine : public state_machine {
   auto IsMain(std::string_view instance_name) const -> bool;
   auto IsReplica(std::string_view instance_name) const -> bool;
 
-  static auto CreateLog(std::string_view log) -> ptr<buffer>;
+  static auto CreateLog(nlohmann::json &&log) -> ptr<buffer>;
   static auto SerializeRegisterInstance(CoordinatorClientConfig const &config) -> ptr<buffer>;
   static auto SerializeUnregisterInstance(std::string_view instance_name) -> ptr<buffer>;
   static auto SerializeSetInstanceAsMain(std::string_view instance_name) -> ptr<buffer>;

src/coordination/include/nuraft/raft_log_action.hpp

@@ -18,6 +18,8 @@
 #include <cstdint>
 #include <string>
 
+#include "json/json.hpp"
+
 namespace memgraph::coordination {
 
 enum class RaftLogAction : uint8_t {
@@ -28,6 +30,14 @@ enum class RaftLogAction : uint8_t {
   UPDATE_UUID
 };
 
+NLOHMANN_JSON_SERIALIZE_ENUM(RaftLogAction, {
+                                                {RaftLogAction::REGISTER_REPLICATION_INSTANCE, "register"},
+                                                {RaftLogAction::UNREGISTER_REPLICATION_INSTANCE, "unregister"},
+                                                {RaftLogAction::SET_INSTANCE_AS_MAIN, "promote"},
+                                                {RaftLogAction::SET_INSTANCE_AS_REPLICA, "demote"},
+                                                {RaftLogAction::UPDATE_UUID, "update_uuid"},
+                                            })
+
 inline auto ParseRaftLogAction(std::string_view action) -> RaftLogAction {
   if (action == "register") {
     return RaftLogAction::REGISTER_REPLICATION_INSTANCE;

src/dbms/inmemory/replication_handlers.cpp

@@ -118,9 +118,14 @@ void InMemoryReplicationHandlers::Register(dbms::DbmsHandler *dbms_handler, repl
       });
   server.rpc_server_.Register<replication_coordination_glue::SwapMainUUIDRpc>(
       [&data, dbms_handler](auto *req_reader, auto *res_builder) {
-        spdlog::debug("Received SwapMainUUIDHandler");
+        spdlog::debug("Received SwapMainUUIDRpc");
         InMemoryReplicationHandlers::SwapMainUUIDHandler(dbms_handler, data, req_reader, res_builder);
       });
+  server.rpc_server_.Register<storage::replication::ForceResetStorageRpc>(
+      [&data, dbms_handler](auto *req_reader, auto *res_builder) {
+        spdlog::debug("Received ForceResetStorageRpc");
+        InMemoryReplicationHandlers::ForceResetStorageHandler(dbms_handler, data.uuid_, req_reader, res_builder);
+      });
 }
 
 void InMemoryReplicationHandlers::SwapMainUUIDHandler(dbms::DbmsHandler *dbms_handler,
@@ -329,6 +334,78 @@ void InMemoryReplicationHandlers::SnapshotHandler(dbms::DbmsHandler *dbms_handle
   spdlog::debug("Replication recovery from snapshot finished!");
 }
 
+void InMemoryReplicationHandlers::ForceResetStorageHandler(dbms::DbmsHandler *dbms_handler,
+                                                           const std::optional<utils::UUID> &current_main_uuid,
+                                                           slk::Reader *req_reader, slk::Builder *res_builder) {
+  storage::replication::ForceResetStorageReq req;
+  slk::Load(&req, req_reader);
+  auto db_acc = GetDatabaseAccessor(dbms_handler, req.db_uuid);
+  if (!db_acc) {
+    storage::replication::ForceResetStorageRes res{false, 0};
+    slk::Save(res, res_builder);
+    return;
+  }
+  if (!current_main_uuid.has_value() || req.main_uuid != current_main_uuid) [[unlikely]] {
+    LogWrongMain(current_main_uuid, req.main_uuid, storage::replication::SnapshotReq::kType.name);
+    storage::replication::ForceResetStorageRes res{false, 0};
+    slk::Save(res, res_builder);
+    return;
+  }
+
+  storage::replication::Decoder decoder(req_reader);
+
+  auto *storage = static_cast<storage::InMemoryStorage *>(db_acc->get()->storage());
+
+  auto storage_guard = std::unique_lock{storage->main_lock_};
+
+  // Clear the database
+  storage->vertices_.clear();
+  storage->edges_.clear();
+  storage->commit_log_.reset();
+  storage->commit_log_.emplace();
+
+  storage->constraints_.existence_constraints_ = std::make_unique<storage::ExistenceConstraints>();
+  storage->constraints_.unique_constraints_ = std::make_unique<storage::InMemoryUniqueConstraints>();
+  storage->indices_.label_index_ = std::make_unique<storage::InMemoryLabelIndex>();
+  storage->indices_.label_property_index_ = std::make_unique<storage::InMemoryLabelPropertyIndex>();
+
+  // Fine since we will force push when reading from WAL just random epoch with 0 timestamp, as it should be if it
+  // acted as MAIN before
+  storage->repl_storage_state_.epoch_.SetEpoch(std::string(utils::UUID{}));
+  storage->repl_storage_state_.last_commit_timestamp_ = 0;
+
+  storage->repl_storage_state_.history.clear();
+  storage->vertex_id_ = 0;
+  storage->edge_id_ = 0;
+  storage->timestamp_ = storage::kTimestampInitialId;
+
+  storage->CollectGarbage<true>(std::move(storage_guard), false);
+  storage->vertices_.run_gc();
+  storage->edges_.run_gc();
+
+  storage::replication::ForceResetStorageRes res{true, storage->repl_storage_state_.last_commit_timestamp_.load()};
+  slk::Save(res, res_builder);
+
+  spdlog::trace("Deleting old snapshot files.");
+  // Delete other durability files
+  auto snapshot_files = storage::durability::GetSnapshotFiles(storage->recovery_.snapshot_directory_, storage->uuid_);
+  for (const auto &[path, uuid, _] : snapshot_files) {
+    spdlog::trace("Deleting snapshot file {}", path);
+    storage->file_retainer_.DeleteFile(path);
+  }
+
+  spdlog::trace("Deleting old WAL files.");
+  auto wal_files = storage::durability::GetWalFiles(storage->recovery_.wal_directory_, storage->uuid_);
+  if (wal_files) {
+    for (const auto &wal_file : *wal_files) {
+      spdlog::trace("Deleting WAL file {}", wal_file.path);
+      storage->file_retainer_.DeleteFile(wal_file.path);
+    }
+
+    storage->wal_file_.reset();
+  }
+}
+
 void InMemoryReplicationHandlers::WalFilesHandler(dbms::DbmsHandler *dbms_handler,
                                                   const std::optional<utils::UUID> &current_main_uuid,
                                                   slk::Reader *req_reader, slk::Builder *res_builder) {

src/dbms/inmemory/replication_handlers.hpp

@@ -48,6 +48,9 @@ class InMemoryReplicationHandlers {
 
   static void SwapMainUUIDHandler(dbms::DbmsHandler *dbms_handler, replication::RoleReplicaData &role_replica_data,
                                   slk::Reader *req_reader, slk::Builder *res_builder);
+  static void ForceResetStorageHandler(dbms::DbmsHandler *dbms_handler,
+                                       const std::optional<utils::UUID> &current_main_uuid, slk::Reader *req_reader,
+                                       slk::Builder *res_builder);
 
   static void LoadWal(storage::InMemoryStorage *storage, storage::replication::Decoder *decoder);
 

src/query/interpreter.cpp

@@ -1263,7 +1263,7 @@ Callback HandleCoordinatorQuery(CoordinatorQuery *coordinator_query, const Param
                   TypedValue{status.coord_socket_address}, TypedValue{status.health}, TypedValue{status.cluster_role}};
         };
 
-        return utils::fmap(converter, instances);
+        return utils::fmap(instances, converter);
       };
       return callback;
     }

src/query/plan/operator.cpp

@@ -3912,11 +3912,13 @@ class AggregateCursor : public Cursor {
       case TypedValue::Type::Int:
       case TypedValue::Type::Double:
       case TypedValue::Type::String:
+      case TypedValue::Type::Date:
+      case TypedValue::Type::LocalTime:
+      case TypedValue::Type::LocalDateTime:
         return;
       default:
         throw QueryRuntimeException(
-            "Only boolean, numeric and string values are allowed in "
-            "MIN and MAX aggregations.");
+            "Only boolean, numeric, string, and non-duration temporal values are allowed in MIN and MAX aggregations.");
     }
   }
 

src/replication_coordination_glue/mode.hpp

@@ -16,28 +16,15 @@
 #include <stdexcept>
 #include <string>
 
+#include "json/json.hpp"
+
 namespace memgraph::replication_coordination_glue {
 
 enum class ReplicationMode : std::uint8_t { SYNC, ASYNC };
 
-inline auto ReplicationModeToString(ReplicationMode mode) -> std::string {
-  switch (mode) {
-    case ReplicationMode::SYNC:
-      return "SYNC";
-    case ReplicationMode::ASYNC:
-      return "ASYNC";
-  }
-  throw std::invalid_argument("Invalid replication mode");
-}
-
-inline auto ReplicationModeFromString(std::string_view mode) -> ReplicationMode {
-  if (mode == "SYNC") {
-    return ReplicationMode::SYNC;
-  }
-  if (mode == "ASYNC") {
-    return ReplicationMode::ASYNC;
-  }
-  throw std::invalid_argument("Invalid replication mode");
-}
+NLOHMANN_JSON_SERIALIZE_ENUM(ReplicationMode, {
+                                                  {ReplicationMode::SYNC, "sync"},
+                                                  {ReplicationMode::ASYNC, "async"},
+                                              })
 
 }  // namespace memgraph::replication_coordination_glue

src/replication_handler/include/replication_handler/replication_handler.hpp

@@ -210,8 +210,13 @@ struct ReplicationHandler : public memgraph::query::ReplicationQueryHandler {
             auto client = std::make_unique<storage::ReplicationStorageClient>(*instance_client_ptr, main_uuid);
             client->Start(storage, std::move(db_acc));
             bool const success = std::invoke([state = client->State()]() {
+              // We force sync replicas in other situation
               if (state == storage::replication::ReplicaState::DIVERGED_FROM_MAIN) {
+#ifdef MG_ENTERPRISE
+                return FLAGS_coordinator_server_port != 0;
+#else
                 return false;
+#endif
               }
               return true;
             });

src/replication_handler/replication_handler.cpp

@@ -271,8 +271,7 @@ auto ReplicationHandler::GetDatabasesHistories() -> replication_coordination_glu
   dbms_handler_.ForEach([&results](memgraph::dbms::DatabaseAccess db_acc) {
     auto &repl_storage_state = db_acc->storage()->repl_storage_state_;
 
-    std::vector<std::pair<std::string, uint64_t>> history = utils::fmap(
-        [](const auto &elem) { return std::make_pair(elem.first, elem.second); }, repl_storage_state.history);
+    std::vector<std::pair<std::string, uint64_t>> history = utils::fmap(repl_storage_state.history);
 
     history.emplace_back(std::string(repl_storage_state.epoch_.id()), repl_storage_state.last_commit_timestamp_.load());
     replication_coordination_glue::DatabaseHistory repl{

src/storage/v2/durability/durability.cpp

@@ -358,7 +358,6 @@ std::optional<RecoveryInfo> Recovery::RecoverData(std::string *uuid, Replication
       spdlog::warn(utils::MessageWithLink("No snapshot or WAL file found.", "https://memgr.ph/durability"));
       return std::nullopt;
     }
-    // TODO(antoniofilipovic) What is the logic here?
     std::sort(wal_files.begin(), wal_files.end());
     // UUID used for durability is the UUID of the last WAL file.
     // Same for the epoch id.
@@ -437,17 +436,13 @@ std::optional<RecoveryInfo> Recovery::RecoverData(std::string *uuid, Replication
           last_loaded_timestamp.emplace(recovery_info.next_timestamp - 1);
         }
 
-        bool epoch_history_empty = epoch_history->empty();
-        bool epoch_not_recorded = !epoch_history_empty && epoch_history->back().first != wal_file.epoch_id;
         auto last_loaded_timestamp_value = last_loaded_timestamp.value_or(0);
-
-        if (epoch_history_empty || epoch_not_recorded) {
-          epoch_history->emplace_back(std::string(wal_file.epoch_id), last_loaded_timestamp_value);
-        }
-
-        auto last_epoch_updated = !epoch_history_empty && epoch_history->back().first == wal_file.epoch_id &&
-                                  epoch_history->back().second < last_loaded_timestamp_value;
-        if (last_epoch_updated) {
+        if (epoch_history->empty() || epoch_history->back().first != wal_file.epoch_id) {
+          // no history or new epoch, add it
+          epoch_history->emplace_back(wal_file.epoch_id, last_loaded_timestamp_value);
+          repl_storage_state.epoch_.SetEpoch(wal_file.epoch_id);
+        } else if (epoch_history->back().second < last_loaded_timestamp_value) {
+          // existing epoch, update with newer timestamp
           epoch_history->back().second = last_loaded_timestamp_value;
         }
 
@@ -469,11 +464,11 @@ std::optional<RecoveryInfo> Recovery::RecoverData(std::string *uuid, Replication
 
   memgraph::metrics::Measure(memgraph::metrics::SnapshotRecoveryLatency_us,
                              std::chrono::duration_cast<std::chrono::microseconds>(timer.Elapsed()).count());
-  spdlog::info("Set epoch id: {}  with commit timestamp {}", std::string(repl_storage_state.epoch_.id()),
-               repl_storage_state.last_commit_timestamp_);
+  spdlog::trace("Set epoch id: {}  with commit timestamp {}", std::string(repl_storage_state.epoch_.id()),
+                repl_storage_state.last_commit_timestamp_);
 
   std::for_each(repl_storage_state.history.begin(), repl_storage_state.history.end(), [](auto &history) {
-    spdlog::info("epoch id: {}  with commit timestamp {}", std::string(history.first), history.second);
+    spdlog::trace("epoch id: {}  with commit timestamp {}", std::string(history.first), history.second);
   });
   return recovery_info;
 }

src/storage/v2/inmemory/storage.cpp

@@ -109,7 +109,7 @@ InMemoryStorage::InMemoryStorage(Config config)
       timestamp_ = std::max(timestamp_, info->next_timestamp);
       if (info->last_commit_timestamp) {
         repl_storage_state_.last_commit_timestamp_ = *info->last_commit_timestamp;
-        spdlog::info("Recovering last commit timestamp {}", *info->last_commit_timestamp);
+        spdlog::trace("Recovering last commit timestamp {}", *info->last_commit_timestamp);
       }
     }
   } else if (config_.durability.snapshot_wal_mode != Config::Durability::SnapshotWalMode::DISABLED ||

src/storage/v2/replication/replication_client.cpp

@@ -76,13 +76,37 @@ void ReplicationStorageClient::UpdateReplicaState(Storage *storage, DatabaseAcce
     }
   }
   if (branching_point) {
-    spdlog::error(
-        "You cannot register Replica {} to this Main because at one point "
-        "Replica {} acted as the Main instance. Both the Main and Replica {} "
-        "now hold unique data. Please resolve data conflicts and start the "
-        "replication on a clean instance.",
-        client_.name_, client_.name_, client_.name_);
-    replica_state_.WithLock([](auto &val) { val = replication::ReplicaState::DIVERGED_FROM_MAIN; });
+    auto replica_state = replica_state_.Lock();
+    if (*replica_state == replication::ReplicaState::DIVERGED_FROM_MAIN) {
+      return;
+    }
+    *replica_state = replication::ReplicaState::DIVERGED_FROM_MAIN;
+
+    auto log_error = [client_name = client_.name_]() {
+      spdlog::error(
+          "You cannot register Replica {} to this Main because at one point "
+          "Replica {} acted as the Main instance. Both the Main and Replica {} "
+          "now hold unique data. Please resolve data conflicts and start the "
+          "replication on a clean instance.",
+          client_name, client_name, client_name);
+    };
+#ifdef MG_ENTERPRISE
+    if (!FLAGS_coordinator_server_port) {
+      log_error();
+      return;
+    }
+    client_.thread_pool_.AddTask([storage, gk = std::move(db_acc), this] {
+      const auto [success, timestamp] = this->ForceResetStorage(storage);
+      if (success) {
+        spdlog::info("Successfully reset storage of REPLICA {} to timestamp {}.", client_.name_, timestamp);
+        return;
+      }
+      spdlog::error("You cannot register REPLICA {} to this MAIN because MAIN couldn't reset REPLICA's storage.",
+                    client_.name_);
+    });
+#else
+    log_error();
+#endif
     return;
   }
 
@@ -265,8 +289,6 @@ void ReplicationStorageClient::RecoverReplica(uint64_t replica_commit, memgraph:
   spdlog::debug("Starting replica recovery");
   auto *mem_storage = static_cast<InMemoryStorage *>(storage);
 
-  // TODO(antoniofilipovic): Can we get stuck here in while loop if replica commit timestamp is not updated when using
-  // only snapshot
   while (true) {
     auto file_locker = mem_storage->file_retainer_.AddLocker();
 
@@ -335,6 +357,21 @@ void ReplicationStorageClient::RecoverReplica(uint64_t replica_commit, memgraph:
   }
 }
 
+std::pair<bool, uint64_t> ReplicationStorageClient::ForceResetStorage(memgraph::storage::Storage *storage) {
+  utils::OnScopeExit set_to_maybe_behind{
+      [this]() { replica_state_.WithLock([](auto &state) { state = replication::ReplicaState::MAYBE_BEHIND; }); }};
+  try {
+    auto stream{client_.rpc_client_.Stream<replication::ForceResetStorageRpc>(main_uuid_, storage->uuid())};
+    const auto res = stream.AwaitResponse();
+    return std::pair{res.success, res.current_commit_timestamp};
+  } catch (const rpc::RpcFailedException &) {
+    spdlog::error(
+        utils::MessageWithLink("Couldn't ForceReset data to {}.", client_.name_, "https://memgr.ph/replication"));
+  }
+
+  return {false, 0};
+}
+
 ////// ReplicaStream //////
 ReplicaStream::ReplicaStream(Storage *storage, rpc::Client &rpc_client, const uint64_t current_seq_num,
                              utils::UUID main_uuid)

src/storage/v2/replication/replication_client.hpp

@@ -188,6 +188,13 @@ class ReplicationStorageClient {
    */
   void UpdateReplicaState(Storage *storage, DatabaseAccessProtector db_acc);
 
+  /**
+   * @brief Forcefully reset storage to as it is when started from scratch.
+   *
+   * @param storage pointer to the storage associated with the client
+   */
+  std::pair<bool, uint64_t> ForceResetStorage(Storage *storage);
+
   void LogRpcFailure();
 
   /**

src/storage/v2/replication/rpc.cpp

@@ -59,6 +59,19 @@ void TimestampRes::Save(const TimestampRes &self, memgraph::slk::Builder *builde
   memgraph::slk::Save(self, builder);
 }
 void TimestampRes::Load(TimestampRes *self, memgraph::slk::Reader *reader) { memgraph::slk::Load(self, reader); }
+
+void ForceResetStorageReq::Save(const ForceResetStorageReq &self, memgraph::slk::Builder *builder) {
+  memgraph::slk::Save(self, builder);
+}
+void ForceResetStorageReq::Load(ForceResetStorageReq *self, memgraph::slk::Reader *reader) {
+  memgraph::slk::Load(self, reader);
+}
+void ForceResetStorageRes::Save(const ForceResetStorageRes &self, memgraph::slk::Builder *builder) {
+  memgraph::slk::Save(self, builder);
+}
+void ForceResetStorageRes::Load(ForceResetStorageRes *self, memgraph::slk::Reader *reader) {
+  memgraph::slk::Load(self, reader);
+}
 }  // namespace storage::replication
 
 constexpr utils::TypeInfo storage::replication::AppendDeltasReq::kType{utils::TypeId::REP_APPEND_DELTAS_REQ,
@@ -97,6 +110,12 @@ constexpr utils::TypeInfo storage::replication::TimestampReq::kType{utils::TypeI
 constexpr utils::TypeInfo storage::replication::TimestampRes::kType{utils::TypeId::REP_TIMESTAMP_RES, "TimestampRes",
                                                                     nullptr};
 
+constexpr utils::TypeInfo storage::replication::ForceResetStorageReq::kType{utils::TypeId::REP_FORCE_RESET_STORAGE_REQ,
+                                                                            "ForceResetStorageReq", nullptr};
+
+constexpr utils::TypeInfo storage::replication::ForceResetStorageRes::kType{utils::TypeId::REP_FORCE_RESET_STORAGE_RES,
+                                                                            "ForceResetStorageRes", nullptr};
+
 // Autogenerated SLK serialization code
 namespace slk {
 // Serialize code for TimestampRes
@@ -255,6 +274,30 @@ void Load(memgraph::storage::replication::AppendDeltasReq *self, memgraph::slk::
   memgraph::slk::Load(&self->seq_num, reader);
 }
 
+// Serialize code for ForceResetStorageReq
+
+void Save(const memgraph::storage::replication::ForceResetStorageReq &self, memgraph::slk::Builder *builder) {
+  memgraph::slk::Save(self.main_uuid, builder);
+  memgraph::slk::Save(self.db_uuid, builder);
+}
+
+void Load(memgraph::storage::replication::ForceResetStorageReq *self, memgraph::slk::Reader *reader) {
+  memgraph::slk::Load(&self->main_uuid, reader);
+  memgraph::slk::Load(&self->db_uuid, reader);
+}
+
+// Serialize code for ForceResetStorageRes
+
+void Save(const memgraph::storage::replication::ForceResetStorageRes &self, memgraph::slk::Builder *builder) {
+  memgraph::slk::Save(self.success, builder);
+  memgraph::slk::Save(self.current_commit_timestamp, builder);
+}
+
+void Load(memgraph::storage::replication::ForceResetStorageRes *self, memgraph::slk::Reader *reader) {
+  memgraph::slk::Load(&self->success, reader);
+  memgraph::slk::Load(&self->current_commit_timestamp, reader);
+}
+
 // Serialize SalientConfig
 
 void Save(const memgraph::storage::SalientConfig &self, memgraph::slk::Builder *builder) {

src/storage/v2/replication/rpc.hpp

@@ -210,6 +210,36 @@ struct TimestampRes {
 
 using TimestampRpc = rpc::RequestResponse<TimestampReq, TimestampRes>;
 
+struct ForceResetStorageReq {
+  static const utils::TypeInfo kType;
+  static const utils::TypeInfo &GetTypeInfo() { return kType; }
+
+  static void Load(ForceResetStorageReq *self, memgraph::slk::Reader *reader);
+  static void Save(const ForceResetStorageReq &self, memgraph::slk::Builder *builder);
+  ForceResetStorageReq() = default;
+  explicit ForceResetStorageReq(const utils::UUID &main_uuid, const utils::UUID &db_uuid)
+      : main_uuid{main_uuid}, db_uuid{db_uuid} {}
+
+  utils::UUID main_uuid;
+  utils::UUID db_uuid;
+};
+
+struct ForceResetStorageRes {
+  static const utils::TypeInfo kType;
+  static const utils::TypeInfo &GetTypeInfo() { return kType; }
+
+  static void Load(ForceResetStorageRes *self, memgraph::slk::Reader *reader);
+  static void Save(const ForceResetStorageRes &self, memgraph::slk::Builder *builder);
+  ForceResetStorageRes() = default;
+  ForceResetStorageRes(bool success, uint64_t current_commit_timestamp)
+      : success(success), current_commit_timestamp(current_commit_timestamp) {}
+
+  bool success;
+  uint64_t current_commit_timestamp;
+};
+
+using ForceResetStorageRpc = rpc::RequestResponse<ForceResetStorageReq, ForceResetStorageRes>;
+
 }  // namespace memgraph::storage::replication
 
 // SLK serialization declarations
@@ -267,4 +297,12 @@ void Save(const memgraph::storage::SalientConfig &self, memgraph::slk::Builder *
 
 void Load(memgraph::storage::SalientConfig *self, memgraph::slk::Reader *reader);
 
+void Save(const memgraph::storage::replication::ForceResetStorageReq &self, memgraph::slk::Builder *builder);
+
+void Load(memgraph::storage::replication::ForceResetStorageReq *self, memgraph::slk::Reader *reader);
+
+void Save(const memgraph::storage::replication::ForceResetStorageRes &self, memgraph::slk::Builder *builder);
+
+void Load(memgraph::storage::replication::ForceResetStorageRes *self, memgraph::slk::Reader *reader);
+
 }  // namespace memgraph::slk

src/utils/functional.hpp

@@ -19,9 +19,9 @@
 namespace memgraph::utils {
 
 template <template <typename, typename...> class Container, typename T, typename Allocator = std::allocator<T>,
-          typename F, typename R = std::invoke_result_t<F, T>>
+          typename F = std::identity, typename R = std::decay_t<std::invoke_result_t<F, T>>>
 requires ranges::range<Container<T, Allocator>> &&
-    (!std::same_as<Container<T, Allocator>, std::string>)auto fmap(F &&f, const Container<T, Allocator> &v)
+    (!std::same_as<Container<T, Allocator>, std::string>)auto fmap(const Container<T, Allocator> &v, F &&f = {})
         -> std::vector<R> {
   return v | ranges::views::transform(std::forward<F>(f)) | ranges::to<std::vector<R>>();
 }

src/utils/typeinfo.hpp

@@ -99,6 +99,8 @@ enum class TypeId : uint64_t {
   REP_DROP_AUTH_DATA_RES,
   REP_TRY_SET_MAIN_UUID_REQ,
   REP_TRY_SET_MAIN_UUID_RES,
+  REP_FORCE_RESET_STORAGE_REQ,
+  REP_FORCE_RESET_STORAGE_RES,
 
   // Coordinator
   COORD_FAILOVER_REQ,

tests/e2e/high_availability/single_coordinator.py

@@ -596,9 +596,8 @@ def test_replication_works_on_failover_replica_2_epochs_more_commits_away(data_r
     mg_sleep_and_assert(5, get_vertex_count)
 
 
-@pytest.mark.parametrize("data_recovery", ["false"])
+@pytest.mark.parametrize("data_recovery", ["true"])
 def test_replication_forcefully_works_on_failover_replica_misses_epoch(data_recovery):
-    # TODO(antoniofilipovic) Test should pass when logic is added
     # Goal of this test is to check the replication works forcefully if replica misses epoch
     # 1. We start all replicas, main and coordinator manually
     # 2. We check that main has correct state
@@ -606,9 +605,9 @@ def test_replication_forcefully_works_on_failover_replica_misses_epoch(data_reco
     # 4. Expect data to be copied on all replicas
     # 5. Kill instance_1 ( this one will miss complete epoch)
     # 6. Kill main (instance_3)
-    # 7. Instance_2 or instance_4 new main
-    # 8. New main commits
-    # 9. Instance_2 down (not main)
+    # 7. Instance_2
+    # 8. Instance_2 commits
+    # 9. Instance_2 down
     # 10. instance_4 down
     # 11. Instance 1 up (missed epoch)
     # 12 Instance 1 new main
@@ -617,7 +616,272 @@ def test_replication_forcefully_works_on_failover_replica_misses_epoch(data_reco
 
     temp_dir = tempfile.TemporaryDirectory().name
 
-    pass
+    MEMGRAPH_INNER_INSTANCES_DESCRIPTION = {
+        "instance_1": {
+            "args": [
+                "--experimental-enabled=high-availability",
+                "--bolt-port",
+                "7688",
+                "--log-level",
+                "TRACE",
+                "--coordinator-server-port",
+                "10011",
+                "--replication-restore-state-on-startup",
+                "true",
+                f"--data-recovery-on-startup={data_recovery}",
+                "--storage-recover-on-startup=false",
+            ],
+            "log_file": "instance_1.log",
+            "data_directory": f"{temp_dir}/instance_1",
+            "setup_queries": [],
+        },
+        "instance_2": {
+            "args": [
+                "--experimental-enabled=high-availability",
+                "--bolt-port",
+                "7689",
+                "--log-level",
+                "TRACE",
+                "--coordinator-server-port",
+                "10012",
+                "--replication-restore-state-on-startup",
+                "true",
+                f"--data-recovery-on-startup={data_recovery}",
+                "--storage-recover-on-startup=false",
+            ],
+            "log_file": "instance_2.log",
+            "data_directory": f"{temp_dir}/instance_2",
+            "setup_queries": [],
+        },
+        "instance_3": {
+            "args": [
+                "--experimental-enabled=high-availability",
+                "--bolt-port",
+                "7687",
+                "--log-level",
+                "TRACE",
+                "--coordinator-server-port",
+                "10013",
+                "--replication-restore-state-on-startup",
+                "true",
+                "--data-recovery-on-startup",
+                f"{data_recovery}",
+                "--storage-recover-on-startup=false",
+            ],
+            "log_file": "instance_3.log",
+            "data_directory": f"{temp_dir}/instance_3",
+            "setup_queries": [],
+        },
+        "instance_4": {
+            "args": [
+                "--experimental-enabled=high-availability",
+                "--bolt-port",
+                "7691",
+                "--log-level",
+                "TRACE",
+                "--coordinator-server-port",
+                "10014",
+                "--replication-restore-state-on-startup",
+                "true",
+                "--data-recovery-on-startup",
+                f"{data_recovery}",
+                "--storage-recover-on-startup=false",
+            ],
+            "log_file": "instance_4.log",
+            "data_directory": f"{temp_dir}/instance_4",
+            "setup_queries": [],
+        },
+        "coordinator": {
+            "args": [
+                "--experimental-enabled=high-availability",
+                "--bolt-port",
+                "7690",
+                "--log-level=TRACE",
+                "--raft-server-id=1",
+                "--raft-server-port=10111",
+            ],
+            "log_file": "coordinator.log",
+            "setup_queries": [
+                "REGISTER INSTANCE instance_1 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';",
+                "REGISTER INSTANCE instance_3 ON '127.0.0.1:10013' WITH '127.0.0.1:10003';",
+                "REGISTER INSTANCE instance_4 ON '127.0.0.1:10014' WITH '127.0.0.1:10004';",
+                "SET INSTANCE instance_3 TO MAIN",
+            ],
+        },
+    }
+
+    # 1
+
+    interactive_mg_runner.start_all(MEMGRAPH_INNER_INSTANCES_DESCRIPTION)
+
+    # 2
+
+    main_cursor = connect(host="localhost", port=7687).cursor()
+    expected_data_on_main = [
+        (
+            "instance_1",
+            "127.0.0.1:10001",
+            "sync",
+            {"behind": None, "status": "ready", "ts": 0},
+            {"memgraph": {"behind": 0, "status": "ready", "ts": 0}},
+        ),
+        (
+            "instance_2",
+            "127.0.0.1:10002",
+            "sync",
+            {"behind": None, "status": "ready", "ts": 0},
+            {"memgraph": {"behind": 0, "status": "ready", "ts": 0}},
+        ),
+        (
+            "instance_4",
+            "127.0.0.1:10004",
+            "sync",
+            {"behind": None, "status": "ready", "ts": 0},
+            {"memgraph": {"behind": 0, "status": "ready", "ts": 0}},
+        ),
+    ]
+
+    main_cursor = connect(host="localhost", port=7687).cursor()
+
+    def retrieve_data_show_replicas():
+        return sorted(list(execute_and_fetch_all(main_cursor, "SHOW REPLICAS;")))
+
+    mg_sleep_and_assert_collection(expected_data_on_main, retrieve_data_show_replicas)
+
+    coord_cursor = connect(host="localhost", port=7690).cursor()
+
+    def retrieve_data_show_instances():
+        return sorted(list(execute_and_fetch_all(coord_cursor, "SHOW INSTANCES;")))
+
+    expected_data_on_coord = [
+        ("coordinator_1", "127.0.0.1:10111", "", "unknown", "coordinator"),
+        ("instance_1", "", "127.0.0.1:10011", "up", "replica"),
+        ("instance_2", "", "127.0.0.1:10012", "up", "replica"),
+        ("instance_3", "", "127.0.0.1:10013", "up", "main"),
+        ("instance_4", "", "127.0.0.1:10014", "up", "replica"),
+    ]
+    mg_sleep_and_assert(expected_data_on_coord, retrieve_data_show_instances)
+
+    # 3
+
+    execute_and_fetch_all(main_cursor, "CREATE (:Epoch1Vertex {prop:1});")
+    execute_and_fetch_all(main_cursor, "CREATE (:Epoch1Vertex {prop:2});")
+
+    # 4
+    instance_1_cursor = connect(host="localhost", port=7688).cursor()
+    instance_2_cursor = connect(host="localhost", port=7689).cursor()
+    instance_4_cursor = connect(host="localhost", port=7691).cursor()
+
+    assert execute_and_fetch_all(instance_1_cursor, "MATCH (n) RETURN count(n);")[0][0] == 2
+    assert execute_and_fetch_all(instance_2_cursor, "MATCH (n) RETURN count(n);")[0][0] == 2
+    assert execute_and_fetch_all(instance_4_cursor, "MATCH (n) RETURN count(n);")[0][0] == 2
+
+    # 5
+
+    interactive_mg_runner.kill(MEMGRAPH_INNER_INSTANCES_DESCRIPTION, "instance_1")
+
+    # 6
+    interactive_mg_runner.kill(MEMGRAPH_INNER_INSTANCES_DESCRIPTION, "instance_3")
+
+    # 7
+
+    expected_data_on_coord = [
+        ("coordinator_1", "127.0.0.1:10111", "", "unknown", "coordinator"),
+        ("instance_1", "", "127.0.0.1:10011", "down", "unknown"),
+        ("instance_2", "", "127.0.0.1:10012", "up", "main"),
+        ("instance_3", "", "127.0.0.1:10013", "down", "unknown"),
+        ("instance_4", "", "127.0.0.1:10014", "up", "replica"),
+    ]
+    mg_sleep_and_assert(expected_data_on_coord, retrieve_data_show_instances)
+
+    # 8
+
+    with pytest.raises(Exception) as e:
+        execute_and_fetch_all(instance_2_cursor, "CREATE (:Epoch2Vertex {prop:1});")
+    assert "At least one SYNC replica has not confirmed committing last transaction." in str(e.value)
+
+    def get_vertex_count():
+        return execute_and_fetch_all(instance_4_cursor, "MATCH (n) RETURN count(n)")[0][0]
+
+    mg_sleep_and_assert(3, get_vertex_count)
+
+    assert execute_and_fetch_all(instance_4_cursor, "MATCH (n) RETURN count(n);")[0][0] == 3
+
+    # 9
+
+    interactive_mg_runner.kill(MEMGRAPH_INNER_INSTANCES_DESCRIPTION, "instance_2")
+
+    # 10
+
+    interactive_mg_runner.kill(MEMGRAPH_INNER_INSTANCES_DESCRIPTION, "instance_4")
+
+    # 11
+
+    interactive_mg_runner.start(MEMGRAPH_INNER_INSTANCES_DESCRIPTION, "instance_1")
+
+    expected_data_on_coord = [
+        ("coordinator_1", "127.0.0.1:10111", "", "unknown", "coordinator"),
+        ("instance_1", "", "127.0.0.1:10011", "up", "main"),
+        ("instance_2", "", "127.0.0.1:10012", "down", "unknown"),
+        ("instance_3", "", "127.0.0.1:10013", "down", "unknown"),
+        ("instance_4", "", "127.0.0.1:10014", "down", "unknown"),
+    ]
+    mg_sleep_and_assert(expected_data_on_coord, retrieve_data_show_instances)
+
+    # 12
+
+    interactive_mg_runner.start(MEMGRAPH_INNER_INSTANCES_DESCRIPTION, "instance_2")
+
+    # 13
+
+    expected_data_on_coord = [
+        ("coordinator_1", "127.0.0.1:10111", "", "unknown", "coordinator"),
+        ("instance_1", "", "127.0.0.1:10011", "up", "main"),
+        ("instance_2", "", "127.0.0.1:10012", "up", "replica"),
+        ("instance_3", "", "127.0.0.1:10013", "down", "unknown"),
+        ("instance_4", "", "127.0.0.1:10014", "down", "unknown"),
+    ]
+    mg_sleep_and_assert(expected_data_on_coord, retrieve_data_show_instances)
+
+    # 12
+    instance_1_cursor = connect(host="localhost", port=7688).cursor()
+    instance_2_cursor = connect(host="localhost", port=7689).cursor()
+
+    def get_vertex_count():
+        return execute_and_fetch_all(instance_1_cursor, "MATCH (n) RETURN count(n)")[0][0]
+
+    mg_sleep_and_assert(2, get_vertex_count)
+
+    def get_vertex_count():
+        return execute_and_fetch_all(instance_2_cursor, "MATCH (n) RETURN count(n)")[0][0]
+
+    mg_sleep_and_assert(2, get_vertex_count)
+
+    # 13
+    with pytest.raises(Exception) as e:
+        execute_and_fetch_all(instance_1_cursor, "CREATE (:Epoch3Vertex {prop:1});")
+    assert "At least one SYNC replica has not confirmed committing last transaction." in str(e.value)
+
+    # 14
+
+    def get_vertex_objects_func_creator(cursor):
+        def get_vertex_objects():
+            return list(
+                execute_and_fetch_all(
+                    cursor, "MATCH (n) " "WITH labels(n) as labels, properties(n) as props " "RETURN labels[0], props;"
+                )
+            )
+
+        return get_vertex_objects
+
+    vertex_objects = [("Epoch1Vertex", {"prop": 1}), ("Epoch1Vertex", {"prop": 2}), ("Epoch3Vertex", {"prop": 1})]
+
+    mg_sleep_and_assert_collection(vertex_objects, get_vertex_objects_func_creator(instance_1_cursor))
+
+    mg_sleep_and_assert_collection(vertex_objects, get_vertex_objects_func_creator(instance_2_cursor))
+
+    # 15
 
 
 @pytest.mark.parametrize("data_recovery", ["false", "true"])

tests/gql_behave/tests/memgraph_V1/features/aggregations.feature

@@ -236,6 +236,56 @@ Feature: Aggregations
             | min(null) |
             | null      |
 
+    Scenario: Min test 05: Date
+        Given any graph
+        When executing query:
+            """
+            UNWIND [date("2024-03-05"), date("2023-04-05"), date("2023-03-06")] AS i
+            RETURN min(i)
+            """
+        Then the result should be:
+            | min(i)     |
+            | 2023-03-06 |
+        And no side effects
+
+    Scenario: Min test 06: LocalTime
+        Given any graph
+        When executing query:
+            """
+            UNWIND [localTime("09:16:00"), localTime("09:15:59"), localTime("10:15:00")] AS i
+            RETURN min(i)
+            """
+        Then the result should be:
+            | min(i)             |
+            | 09:15:59.000000000 |
+        And no side effects
+
+    Scenario: Min test 07: LocalDateTime
+        Given any graph
+        When executing query:
+            """
+            UNWIND [localDateTime("2024-03-05T09:15:00"),
+            localDateTime("2023-04-05T09:15:00"),
+            localDateTime("2023-03-06T09:15:00"),
+            localDateTime("2023-03-05T10:15:00"),
+            localDateTime("2023-03-05T09:16:00"),
+            localDateTime("2023-03-05T09:15:59")] AS i
+            RETURN min(i)
+            """
+        Then the result should be:
+            | min(i)                        |
+            | 2023-03-05T09:15:59.000000000 |
+        And no side effects
+
+    Scenario: Min test 08: Duration
+        Given any graph
+        When executing query:
+            """
+            UNWIND [duration("PT2M2.33S"), duration("PT2M2.33S")] AS i
+            RETURN min(i)
+            """
+        Then an error should be raised
+
     Scenario: Max test 01:
         Given an empty graph
         And having executed
@@ -287,6 +337,56 @@ Feature: Aggregations
             | max(null) |
             | null      |
 
+    Scenario: Max test 05: Date
+        Given any graph
+        When executing query:
+            """
+            UNWIND [date("2024-03-05"), date("2023-04-05"), date("2023-03-06")] AS i
+            RETURN max(i)
+            """
+        Then the result should be:
+            | max(i)     |
+            | 2024-03-05 |
+        And no side effects
+
+    Scenario: Max test 06: LocalTime
+        Given any graph
+        When executing query:
+            """
+            UNWIND [localTime("10:15:00"), localTime("09:16:00"), localTime("09:15:59")] AS i
+            RETURN max(i)
+            """
+        Then the result should be:
+            | max(i)             |
+            | 10:15:00.000000000 |
+        And no side effects
+
+    Scenario: Max test 07: LocalDateTime
+        Given any graph
+        When executing query:
+            """
+            UNWIND [localDateTime("2024-03-05T09:15:00"),
+            localDateTime("2023-04-05T09:15:00"),
+            localDateTime("2023-03-06T09:15:00"),
+            localDateTime("2023-03-05T10:15:00"),
+            localDateTime("2023-03-05T09:16:00"),
+            localDateTime("2023-03-05T09:15:59")] AS i
+            RETURN max(i)
+            """
+        Then the result should be:
+            | max(i)                        |
+            | 2024-03-05T09:15:00.000000000 |
+        And no side effects
+
+    Scenario: Max test 08: Duration
+        Given any graph
+        When executing query:
+            """
+            UNWIND [duration("PT2M2.33S"), duration("PT2M2.33S")] AS i
+            RETURN max(i)
+            """
+        Then an error should be raised
+
     Scenario: Collect test 01:
         Given an empty graph
         And having executed
@@ -329,7 +429,7 @@ Feature: Aggregations
             | n                                 |
             | {a_key: 13, b_key: 11, c_key: 12} |
 
-        Scenario: Combined aggregations - some evauluates to null:
+        Scenario: Combined aggregations - some evaluate to null:
         Given an empty graph
         And having executed
             """

tests/gql_behave/tests/memgraph_V1_on_disk/features/aggregations.feature

@@ -236,6 +236,56 @@ Feature: Aggregations
             | min(null) |
             | null      |
 
+    Scenario: Min test 05: Date
+        Given any graph
+        When executing query:
+            """
+            UNWIND [date("2024-03-05"), date("2023-04-05"), date("2023-03-06")] AS i
+            RETURN min(i)
+            """
+        Then the result should be:
+            | min(i)     |
+            | 2023-03-06 |
+        And no side effects
+
+    Scenario: Min test 06: LocalTime
+        Given any graph
+        When executing query:
+            """
+            UNWIND [localTime("09:16:00"), localTime("09:15:59"), localTime("10:15:00")] AS i
+            RETURN min(i)
+            """
+        Then the result should be:
+            | min(i)             |
+            | 09:15:59.000000000 |
+        And no side effects
+
+    Scenario: Min test 07: LocalDateTime
+        Given any graph
+        When executing query:
+            """
+            UNWIND [localDateTime("2024-03-05T09:15:00"),
+            localDateTime("2023-04-05T09:15:00"),
+            localDateTime("2023-03-06T09:15:00"),
+            localDateTime("2023-03-05T10:15:00"),
+            localDateTime("2023-03-05T09:16:00"),
+            localDateTime("2023-03-05T09:15:59")] AS i
+            RETURN min(i)
+            """
+        Then the result should be:
+            | min(i)                        |
+            | 2023-03-05T09:15:59.000000000 |
+        And no side effects
+
+    Scenario: Min test 08: Duration
+        Given any graph
+        When executing query:
+            """
+            UNWIND [duration("PT2M2.33S"), duration("PT2M2.33S")] AS i
+            RETURN min(i)
+            """
+        Then an error should be raised
+
     Scenario: Max test 01:
         Given an empty graph
         And having executed
@@ -287,6 +337,56 @@ Feature: Aggregations
             | max(null) |
             | null      |
 
+    Scenario: Max test 05: Date
+        Given any graph
+        When executing query:
+            """
+            UNWIND [date("2024-03-05"), date("2023-04-05"), date("2023-03-06")] AS i
+            RETURN max(i)
+            """
+        Then the result should be:
+            | max(i)     |
+            | 2024-03-05 |
+        And no side effects
+
+    Scenario: Max test 06: LocalTime
+        Given any graph
+        When executing query:
+            """
+            UNWIND [localTime("10:15:00"), localTime("09:16:00"), localTime("09:15:59")] AS i
+            RETURN max(i)
+            """
+        Then the result should be:
+            | max(i)             |
+            | 10:15:00.000000000 |
+        And no side effects
+
+    Scenario: Max test 07: LocalDateTime
+        Given any graph
+        When executing query:
+            """
+            UNWIND [localDateTime("2024-03-05T09:15:00"),
+            localDateTime("2023-04-05T09:15:00"),
+            localDateTime("2023-03-06T09:15:00"),
+            localDateTime("2023-03-05T10:15:00"),
+            localDateTime("2023-03-05T09:16:00"),
+            localDateTime("2023-03-05T09:15:59")] AS i
+            RETURN max(i)
+            """
+        Then the result should be:
+            | max(i)                        |
+            | 2024-03-05T09:15:00.000000000 |
+        And no side effects
+
+    Scenario: Max test 08: Duration
+        Given any graph
+        When executing query:
+            """
+            UNWIND [duration("PT2M2.33S"), duration("PT2M2.33S")] AS i
+            RETURN max(i)
+            """
+        Then an error should be raised
+
     Scenario: Collect test 01:
         Given an empty graph
         And having executed
@@ -329,7 +429,7 @@ Feature: Aggregations
             | n                                 |
             | {a_key: 13, b_key: 11, c_key: 12} |
 
-        Scenario: Combined aggregations - some evauluates to null:
+        Scenario: Combined aggregations - some evaluate to null:
         Given an empty graph
         And having executed
             """

tests/unit/CMakeLists.txt

@@ -438,3 +438,10 @@ add_unit_test(coordination_utils.cpp)
 target_link_libraries(${test_prefix}coordination_utils gflags mg-coordination mg-repl_coord_glue)
 target_include_directories(${test_prefix}coordination_utils PRIVATE ${CMAKE_SOURCE_DIR}/include)
 endif()
+
+# Test Raft log serialization
+if(MG_ENTERPRISE)
+add_unit_test(raft_log_serialization.cpp)
+target_link_libraries(${test_prefix}raft_log_serialization gflags mg-coordination mg-repl_coord_glue)
+target_include_directories(${test_prefix}raft_log_serialization PRIVATE ${CMAKE_SOURCE_DIR}/include)
+endif()

tests/unit/coordination_utils.cpp

@@ -45,11 +45,9 @@ TEST_F(CoordinationUtils, MemgraphDbHistorySimple) {
   memgraph::utils::UUID db_uuid;
   std::string default_name = std::string(memgraph::dbms::kDefaultDB);
 
-  auto db_histories = memgraph::utils::fmap(
-      [](const std::pair<memgraph::utils::UUID, uint64_t> &pair) {
-        return std::make_pair(std::string(pair.first), pair.second);
-      },
-      histories);
+  auto db_histories = memgraph::utils::fmap(histories, [](const std::pair<memgraph::utils::UUID, uint64_t> &pair) {
+    return std::make_pair(std::string(pair.first), pair.second);
+  });
 
   memgraph::replication_coordination_glue::DatabaseHistory history{
       .db_uuid = db_uuid, .history = db_histories, .name = default_name};
@@ -91,11 +89,9 @@ TEST_F(CoordinationUtils, MemgraphDbHistoryLastEpochDifferent) {
   memgraph::utils::UUID db_uuid;
   std::string default_name = std::string(memgraph::dbms::kDefaultDB);
 
-  auto db_histories = memgraph::utils::fmap(
-      [](const std::pair<memgraph::utils::UUID, uint64_t> &pair) {
-        return std::make_pair(std::string(pair.first), pair.second);
-      },
-      histories);
+  auto db_histories = memgraph::utils::fmap(histories, [](const std::pair<memgraph::utils::UUID, uint64_t> &pair) {
+    return std::make_pair(std::string(pair.first), pair.second);
+  });
 
   db_histories.back().second = 51;
   memgraph::replication_coordination_glue::DatabaseHistory history1{
@@ -145,11 +141,9 @@ TEST_F(CoordinationUtils, MemgraphDbHistoryOneInstanceAheadFewEpochs) {
   memgraph::utils::UUID db_uuid;
   std::string default_name = std::string(memgraph::dbms::kDefaultDB);
 
-  auto db_histories = memgraph::utils::fmap(
-      [](const std::pair<memgraph::utils::UUID, uint64_t> &pair) {
-        return std::make_pair(std::string(pair.first), pair.second);
-      },
-      histories);
+  auto db_histories = memgraph::utils::fmap(histories, [](const std::pair<memgraph::utils::UUID, uint64_t> &pair) {
+    return std::make_pair(std::string(pair.first), pair.second);
+  });
 
   memgraph::replication_coordination_glue::DatabaseHistory history{
       .db_uuid = db_uuid, .history = db_histories, .name = default_name};
@@ -162,11 +156,10 @@ TEST_F(CoordinationUtils, MemgraphDbHistoryOneInstanceAheadFewEpochs) {
 
   histories.emplace_back(memgraph::utils::UUID{}, 60);
   histories.emplace_back(memgraph::utils::UUID{}, 65);
-  auto db_histories_longest = memgraph::utils::fmap(
-      [](const std::pair<memgraph::utils::UUID, uint64_t> &pair) {
+  auto db_histories_longest =
+      memgraph::utils::fmap(histories, [](const std::pair<memgraph::utils::UUID, uint64_t> &pair) {
         return std::make_pair(std::string(pair.first), pair.second);
-      },
-      histories);
+      });
 
   memgraph::replication_coordination_glue::DatabaseHistory history_longest{
       .db_uuid = db_uuid, .history = db_histories_longest, .name = default_name};
@@ -200,11 +193,9 @@ TEST_F(CoordinationUtils, MemgraphDbHistoryInstancesHistoryDiverged) {
   memgraph::utils::UUID db_uuid;
   std::string default_name = std::string(memgraph::dbms::kDefaultDB);
 
-  auto db_histories = memgraph::utils::fmap(
-      [](const std::pair<memgraph::utils::UUID, uint64_t> &pair) {
-        return std::make_pair(std::string(pair.first), pair.second);
-      },
-      histories);
+  auto db_histories = memgraph::utils::fmap(histories, [](const std::pair<memgraph::utils::UUID, uint64_t> &pair) {
+    return std::make_pair(std::string(pair.first), pair.second);
+  });
 
   memgraph::replication_coordination_glue::DatabaseHistory history{
       .db_uuid = db_uuid, .history = db_histories, .name = default_name};
@@ -217,11 +208,10 @@ TEST_F(CoordinationUtils, MemgraphDbHistoryInstancesHistoryDiverged) {
   auto oldest_commit_timestamp{5};
   auto newest_different_epoch = memgraph::utils::UUID{};
   histories.emplace_back(newest_different_epoch, oldest_commit_timestamp);
-  auto db_histories_different = memgraph::utils::fmap(
-      [](const std::pair<memgraph::utils::UUID, uint64_t> &pair) {
+  auto db_histories_different =
+      memgraph::utils::fmap(histories, [](const std::pair<memgraph::utils::UUID, uint64_t> &pair) {
         return std::make_pair(std::string(pair.first), pair.second);
-      },
-      histories);
+      });
 
   memgraph::replication_coordination_glue::DatabaseHistory history_3{
       .db_uuid = db_uuid, .history = db_histories_different, .name = default_name};

tests/unit/raft_log_serialization.cpp

@@ -0,0 +1,151 @@
+// Copyright 2024 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#include "coordination/coordinator_config.hpp"
+#include "nuraft/coordinator_state_machine.hpp"
+#include "nuraft/raft_log_action.hpp"
+#include "utils/file.hpp"
+#include "utils/uuid.hpp"
+
+#include <gflags/gflags.h>
+#include <gtest/gtest.h>
+#include "json/json.hpp"
+
+using memgraph::coordination::CoordinatorClientConfig;
+using memgraph::coordination::CoordinatorStateMachine;
+using memgraph::coordination::RaftLogAction;
+using memgraph::coordination::ReplClientInfo;
+using memgraph::replication_coordination_glue::ReplicationMode;
+using memgraph::utils::UUID;
+
+class RaftLogSerialization : public ::testing::Test {
+ protected:
+  void SetUp() override {}
+
+  void TearDown() override {}
+
+  std::filesystem::path test_folder_{std::filesystem::temp_directory_path() / "MG_tests_unit_raft_log_serialization"};
+};
+
+TEST_F(RaftLogSerialization, ReplClientInfo) {
+  ReplClientInfo info{"instance_name", ReplicationMode::SYNC, "127.0.0.1", 10111};
+
+  nlohmann::json j = info;
+  ReplClientInfo info2 = j.get<memgraph::coordination::ReplClientInfo>();
+
+  ASSERT_EQ(info, info2);
+}
+
+TEST_F(RaftLogSerialization, CoordinatorClientConfig) {
+  CoordinatorClientConfig config{"instance3",
+                                 "127.0.0.1",
+                                 10112,
+                                 std::chrono::seconds{1},
+                                 std::chrono::seconds{5},
+                                 std::chrono::seconds{10},
+                                 {"instance_name", ReplicationMode::ASYNC, "replication_ip_address", 10001},
+                                 .ssl = std::nullopt};
+
+  nlohmann::json j = config;
+  CoordinatorClientConfig config2 = j.get<memgraph::coordination::CoordinatorClientConfig>();
+
+  ASSERT_EQ(config, config2);
+}
+
+TEST_F(RaftLogSerialization, RaftLogActionRegister) {
+  auto action = RaftLogAction::REGISTER_REPLICATION_INSTANCE;
+
+  nlohmann::json j = action;
+  RaftLogAction action2 = j.get<memgraph::coordination::RaftLogAction>();
+
+  ASSERT_EQ(action, action2);
+}
+
+TEST_F(RaftLogSerialization, RaftLogActionUnregister) {
+  auto action = RaftLogAction::UNREGISTER_REPLICATION_INSTANCE;
+
+  nlohmann::json j = action;
+  RaftLogAction action2 = j.get<memgraph::coordination::RaftLogAction>();
+
+  ASSERT_EQ(action, action2);
+}
+
+TEST_F(RaftLogSerialization, RaftLogActionPromote) {
+  auto action = RaftLogAction::SET_INSTANCE_AS_MAIN;
+
+  nlohmann::json j = action;
+  RaftLogAction action2 = j.get<memgraph::coordination::RaftLogAction>();
+
+  ASSERT_EQ(action, action2);
+}
+
+TEST_F(RaftLogSerialization, RaftLogActionDemote) {
+  auto action = RaftLogAction::SET_INSTANCE_AS_REPLICA;
+
+  nlohmann::json j = action;
+  RaftLogAction action2 = j.get<memgraph::coordination::RaftLogAction>();
+
+  ASSERT_EQ(action, action2);
+}
+
+TEST_F(RaftLogSerialization, RaftLogActionUpdateUUID) {
+  auto action = RaftLogAction::UPDATE_UUID;
+
+  nlohmann::json j = action;
+  RaftLogAction action2 = j.get<memgraph::coordination::RaftLogAction>();
+
+  ASSERT_EQ(action, action2);
+}
+
+TEST_F(RaftLogSerialization, RegisterInstance) {
+  CoordinatorClientConfig config{"instance3",
+                                 "127.0.0.1",
+                                 10112,
+                                 std::chrono::seconds{1},
+                                 std::chrono::seconds{5},
+                                 std::chrono::seconds{10},
+                                 {"instance_name", ReplicationMode::ASYNC, "replication_ip_address", 10001},
+                                 .ssl = std::nullopt};
+
+  auto buffer = CoordinatorStateMachine::SerializeRegisterInstance(config);
+  auto [payload, action] = CoordinatorStateMachine::DecodeLog(*buffer);
+  ASSERT_EQ(action, RaftLogAction::REGISTER_REPLICATION_INSTANCE);
+  ASSERT_EQ(config, std::get<CoordinatorClientConfig>(payload));
+}
+
+TEST_F(RaftLogSerialization, UnregisterInstance) {
+  auto buffer = CoordinatorStateMachine::SerializeUnregisterInstance("instance3");
+  auto [payload, action] = CoordinatorStateMachine::DecodeLog(*buffer);
+  ASSERT_EQ(action, RaftLogAction::UNREGISTER_REPLICATION_INSTANCE);
+  ASSERT_EQ("instance3", std::get<std::string>(payload));
+}
+
+TEST_F(RaftLogSerialization, SetInstanceAsMain) {
+  auto buffer = CoordinatorStateMachine::SerializeSetInstanceAsMain("instance3");
+  auto [payload, action] = CoordinatorStateMachine::DecodeLog(*buffer);
+  ASSERT_EQ(action, RaftLogAction::SET_INSTANCE_AS_MAIN);
+  ASSERT_EQ("instance3", std::get<std::string>(payload));
+}
+
+TEST_F(RaftLogSerialization, SetInstanceAsReplica) {
+  auto buffer = CoordinatorStateMachine::SerializeSetInstanceAsReplica("instance3");
+  auto [payload, action] = CoordinatorStateMachine::DecodeLog(*buffer);
+  ASSERT_EQ(action, RaftLogAction::SET_INSTANCE_AS_REPLICA);
+  ASSERT_EQ("instance3", std::get<std::string>(payload));
+}
+
+TEST_F(RaftLogSerialization, UpdateUUID) {
+  UUID uuid;
+  auto buffer = CoordinatorStateMachine::SerializeUpdateUUID(uuid);
+  auto [payload, action] = CoordinatorStateMachine::DecodeLog(*buffer);
+  ASSERT_EQ(action, RaftLogAction::UPDATE_UUID);
+  ASSERT_EQ(uuid, std::get<UUID>(payload));
+}