WIP

.clang-tidy

@@ -88,4 +88,3 @@ CheckOptions:
   - key:             modernize-use-nullptr.NullMacros
     value:           'NULL'
 ...
-

src/memgraph.cpp

@@ -252,6 +252,11 @@ DEFINE_double(query_execution_timeout_sec, 600,
               "Maximum allowed query execution time. Queries exceeding this "
               "limit will be aborted. Value of 0 means no limit.");
 
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+DEFINE_uint64(replication_replica_check_frequency_sec, 1,
+              "The time duration between two replica checks/pings. If < 1, replicas will NOT be checked at all. NOTE: "
+              "The MAIN instance allocates a new thread for each REPLICA.");
+
 // NOLINTNEXTLINE (cppcoreguidelines-avoid-non-const-global-variables)
 DEFINE_uint64(
     memory_limit, 0,
@@ -1216,6 +1221,7 @@ int main(int argc, char **argv) {
       &db,
       {.query = {.allow_load_csv = FLAGS_allow_load_csv},
        .execution_timeout_sec = FLAGS_query_execution_timeout_sec,
+       .replication_replica_check_frequency = std::chrono::seconds(FLAGS_replication_replica_check_frequency_sec),
        .default_kafka_bootstrap_servers = FLAGS_kafka_bootstrap_servers,
        .default_pulsar_service_url = FLAGS_pulsar_service_url,
        .stream_transaction_conflict_retries = FLAGS_stream_transaction_conflict_retries,

src/query/config.hpp

@@ -21,6 +21,8 @@ struct InterpreterConfig {
 
   // The default execution timeout is 10 minutes.
   double execution_timeout_sec{600.0};
+  // The same as \ref memgraph::storage::replication::ReplicationClientConfig
+  std::chrono::seconds replication_replica_check_frequency{1};
 
   std::string default_kafka_bootstrap_servers;
   std::string default_pulsar_service_url;

src/query/db_accessor.hpp

@@ -324,7 +324,7 @@ class DbAccessor final {
 
   void AdvanceCommand() { accessor_->AdvanceCommand(); }
 
-  utils::BasicResult<storage::ConstraintViolation, void> Commit() { return accessor_->Commit(); }
+  utils::BasicResult<storage::CommitError, void> Commit() { return accessor_->Commit(); }
 
   void Abort() { accessor_->Abort(); }
 

src/query/interpreter.cpp

@@ -160,7 +160,8 @@ class ReplQueryHandler final : public query::ReplicationQueryHandler {
 
   /// @throw QueryRuntimeException if an error ocurred.
   void RegisterReplica(const std::string &name, const std::string &socket_address,
-                       const ReplicationQuery::SyncMode sync_mode, const std::optional<double> timeout) override {
+                       const ReplicationQuery::SyncMode sync_mode, const std::optional<double> timeout,
+                       const std::chrono::seconds replica_check_frequency) override {
     if (db_->GetReplicationRole() == storage::ReplicationRole::REPLICA) {
       // replica can't register another replica
       throw QueryRuntimeException("Replica can't register another replica!");
@@ -182,8 +183,9 @@ class ReplQueryHandler final : public query::ReplicationQueryHandler {
         io::network::Endpoint::ParseSocketOrIpAddress(socket_address, query::kDefaultReplicationPort);
     if (maybe_ip_and_port) {
       auto [ip, port] = *maybe_ip_and_port;
-      auto ret =
-          db_->RegisterReplica(name, {std::move(ip), port}, repl_mode, {.timeout = timeout, .ssl = std::nullopt});
+      auto ret = db_->RegisterReplica(
+          name, {std::move(ip), port}, repl_mode,
+          {.timeout = timeout, .replica_check_frequency = replica_check_frequency, .ssl = std::nullopt});
       if (ret.HasError()) {
         throw QueryRuntimeException(fmt::format("Couldn't register replica '{}'!", name));
       }
@@ -448,7 +450,7 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
       return callback;
     }
     case ReplicationQuery::Action::SHOW_REPLICATION_ROLE: {
-      callback.header = {"replication mode"};
+      callback.header = {"replication role"};
       callback.fn = [handler = ReplQueryHandler{interpreter_context->db}] {
         auto mode = handler.ShowReplicationRole();
         switch (mode) {
@@ -467,6 +469,7 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
       const auto &sync_mode = repl_query->sync_mode_;
       auto socket_address = repl_query->socket_address_->Accept(evaluator);
       auto timeout = EvaluateOptionalExpression(repl_query->timeout_, &evaluator);
+      const auto replica_check_frequency = interpreter_context->config.replication_replica_check_frequency;
       std::optional<double> maybe_timeout;
       if (timeout.IsDouble()) {
         maybe_timeout = timeout.ValueDouble();
@@ -474,8 +477,9 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
         maybe_timeout = static_cast<double>(timeout.ValueInt());
       }
       callback.fn = [handler = ReplQueryHandler{interpreter_context->db}, name, socket_address, sync_mode,
-                     maybe_timeout]() mutable {
-        handler.RegisterReplica(name, std::string(socket_address.ValueString()), sync_mode, maybe_timeout);
+                     maybe_timeout, replica_check_frequency]() mutable {
+        handler.RegisterReplica(name, std::string(socket_address.ValueString()), sync_mode, maybe_timeout,
+                                replica_check_frequency);
         return std::vector<std::vector<TypedValue>>();
       };
       notifications->emplace_back(SeverityLevel::INFO, NotificationCode::REGISTER_REPLICA,
@@ -512,7 +516,6 @@ Callback HandleReplicationQuery(ReplicationQuery *repl_query, const Parameters &
               typed_replica.emplace_back(TypedValue("async"));
               break;
           }
-          typed_replica.emplace_back(TypedValue(static_cast<int64_t>(replica.sync_mode)));
           if (replica.timeout) {
             typed_replica.emplace_back(TypedValue(*replica.timeout));
           } else {
@@ -2206,25 +2209,39 @@ void RunTriggersIndividually(const utils::SkipList<Trigger> &triggers, Interpret
       continue;
     }
 
-    auto maybe_constraint_violation = db_accessor.Commit();
-    if (maybe_constraint_violation.HasError()) {
-      const auto &constraint_violation = maybe_constraint_violation.GetError();
-      switch (constraint_violation.type) {
-        case storage::ConstraintViolation::Type::EXISTENCE: {
-          const auto &label_name = db_accessor.LabelToName(constraint_violation.label);
-          MG_ASSERT(constraint_violation.properties.size() == 1U);
-          const auto &property_name = db_accessor.PropertyToName(*constraint_violation.properties.begin());
-          spdlog::warn("Trigger '{}' failed to commit due to existence constraint violation on :{}({})", trigger.Name(),
-                       label_name, property_name);
+    auto maybe_commit_error = db_accessor.Commit();
+    if (maybe_commit_error.HasError()) {
+      const auto &commit_error = maybe_commit_error.GetError();
+      switch (commit_error.type) {
+        case storage::CommitError::Type::UNABLE_TO_SYNC_REPLICATE: {
+          // TODO(gitbuda): This is tricky because this is an internal
+          // operation. Consider stopping main Memgraph instance here.
+          spdlog::warn("Trigger '{}' failed to commit due to inability to replicate data to SYNC replica",
+                       trigger.Name());
           break;
         }
-        case storage::ConstraintViolation::Type::UNIQUE: {
-          const auto &label_name = db_accessor.LabelToName(constraint_violation.label);
-          std::stringstream property_names_stream;
-          utils::PrintIterable(property_names_stream, constraint_violation.properties, ", ",
-                               [&](auto &stream, const auto &prop) { stream << db_accessor.PropertyToName(prop); });
-          spdlog::warn("Trigger '{}' failed to commit due to unique constraint violation on :{}({})", trigger.Name(),
-                       label_name, property_names_stream.str());
+        case storage::CommitError::Type::CONSTRAINT_VIOLATION: {
+          MG_ASSERT(commit_error.maybe_constraint_violation.has_value());
+          const auto &constraint_violation = *commit_error.maybe_constraint_violation;
+          switch (constraint_violation.type) {
+            case storage::ConstraintViolation::Type::EXISTENCE: {
+              const auto &label_name = db_accessor.LabelToName(constraint_violation.label);
+              MG_ASSERT(constraint_violation.properties.size() == 1U);
+              const auto &property_name = db_accessor.PropertyToName(*constraint_violation.properties.begin());
+              spdlog::warn("Trigger '{}' failed to commit due to existence constraint violation on :{}({})",
+                           trigger.Name(), label_name, property_name);
+              break;
+            }
+            case storage::ConstraintViolation::Type::UNIQUE: {
+              const auto &label_name = db_accessor.LabelToName(constraint_violation.label);
+              std::stringstream property_names_stream;
+              utils::PrintIterable(property_names_stream, constraint_violation.properties, ", ",
+                                   [&](auto &stream, const auto &prop) { stream << db_accessor.PropertyToName(prop); });
+              spdlog::warn("Trigger '{}' failed to commit due to unique constraint violation on :{}({})",
+                           trigger.Name(), label_name, property_names_stream.str());
+              break;
+            }
+          }
           break;
         }
       }
@@ -2269,28 +2286,40 @@ void Interpreter::Commit() {
     trigger_context_collector_.reset();
   };
 
-  auto maybe_constraint_violation = db_accessor_->Commit();
-  if (maybe_constraint_violation.HasError()) {
-    const auto &constraint_violation = maybe_constraint_violation.GetError();
-    switch (constraint_violation.type) {
-      case storage::ConstraintViolation::Type::EXISTENCE: {
-        auto label_name = execution_db_accessor_->LabelToName(constraint_violation.label);
-        MG_ASSERT(constraint_violation.properties.size() == 1U);
-        auto property_name = execution_db_accessor_->PropertyToName(*constraint_violation.properties.begin());
+  auto maybe_commit_error = db_accessor_->Commit();
+  if (maybe_commit_error.HasError()) {
+    const auto &commit_error = maybe_commit_error.GetError();
+    switch (commit_error.type) {
+      case storage::CommitError::Type::UNABLE_TO_SYNC_REPLICATE: {
         reset_necessary_members();
-        throw QueryException("Unable to commit due to existence constraint violation on :{}({})", label_name,
-                             property_name);
+        throw QueryException("Unable to commit due to inability to replicate to SYNC replica");
         break;
       }
-      case storage::ConstraintViolation::Type::UNIQUE: {
-        auto label_name = execution_db_accessor_->LabelToName(constraint_violation.label);
-        std::stringstream property_names_stream;
-        utils::PrintIterable(
-            property_names_stream, constraint_violation.properties, ", ",
-            [this](auto &stream, const auto &prop) { stream << execution_db_accessor_->PropertyToName(prop); });
-        reset_necessary_members();
-        throw QueryException("Unable to commit due to unique constraint violation on :{}({})", label_name,
-                             property_names_stream.str());
+      case storage::CommitError::Type::CONSTRAINT_VIOLATION: {
+        MG_ASSERT(commit_error.maybe_constraint_violation.has_value());
+        const auto &constraint_violation = *commit_error.maybe_constraint_violation;
+        switch (constraint_violation.type) {
+          case storage::ConstraintViolation::Type::EXISTENCE: {
+            auto label_name = execution_db_accessor_->LabelToName(constraint_violation.label);
+            MG_ASSERT(constraint_violation.properties.size() == 1U);
+            auto property_name = execution_db_accessor_->PropertyToName(*constraint_violation.properties.begin());
+            reset_necessary_members();
+            throw QueryException("Unable to commit due to existence constraint violation on :{}({})", label_name,
+                                 property_name);
+            break;
+          }
+          case storage::ConstraintViolation::Type::UNIQUE: {
+            auto label_name = execution_db_accessor_->LabelToName(constraint_violation.label);
+            std::stringstream property_names_stream;
+            utils::PrintIterable(
+                property_names_stream, constraint_violation.properties, ", ",
+                [this](auto &stream, const auto &prop) { stream << execution_db_accessor_->PropertyToName(prop); });
+            reset_necessary_members();
+            throw QueryException("Unable to commit due to unique constraint violation on :{}({})", label_name,
+                                 property_names_stream.str());
+            break;
+          }
+        }
         break;
       }
     }

src/query/interpreter.hpp

@@ -137,7 +137,8 @@ class ReplicationQueryHandler {
 
   /// @throw QueryRuntimeException if an error ocurred.
   virtual void RegisterReplica(const std::string &name, const std::string &socket_address,
-                               const ReplicationQuery::SyncMode sync_mode, const std::optional<double> timeout) = 0;
+                               const ReplicationQuery::SyncMode sync_mode, const std::optional<double> timeout,
+                               const std::chrono::seconds replica_check_frequency) = 0;
 
   /// @throw QueryRuntimeException if an error ocurred.
   virtual void DropReplica(const std::string &replica_name) = 0;

src/storage/v2/CMakeLists.txt

@@ -1,5 +1,6 @@
 set(storage_v2_src_files
     commit_log.cpp
+    commit_error.cpp
     constraints.cpp
     temporal.cpp
     durability/durability.cpp

src/storage/v2/commit_error.cpp

@@ -0,0 +1,20 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#include "storage/v2/commit_error.hpp"
+
+namespace memgraph::storage {
+
+bool operator==(const CommitError &lhs, const CommitError &rhs) {
+  return lhs.type == rhs.type && lhs.maybe_constraint_violation == rhs.maybe_constraint_violation;
+}
+
+}  // namespace memgraph::storage

src/storage/v2/commit_error.hpp

@@ -0,0 +1,32 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#pragma once
+
+#include <optional>
+
+#include "storage/v2/constraints.hpp"
+
+namespace memgraph::storage {
+
+struct CommitError {
+  enum class Type {
+    CONSTRAINT_VIOLATION,
+    UNABLE_TO_SYNC_REPLICATE,
+  };
+  Type type;
+
+  std::optional<ConstraintViolation> maybe_constraint_violation;
+};
+
+bool operator==(const CommitError &lhs, const CommitError &rhs);
+
+}  // namespace memgraph::storage

src/storage/v2/replication/config.hpp

@@ -16,6 +16,10 @@
 namespace memgraph::storage::replication {
 struct ReplicationClientConfig {
   std::optional<double> timeout;
+  // 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.
+  std::chrono::seconds replica_check_frequency{1};
 
   struct SSL {
     std::string key_file = "";

src/storage/v2/replication/replication_client.cpp

@@ -41,12 +41,49 @@ Storage::ReplicationClient::ReplicationClient(std::string name, Storage *storage
   }
 
   rpc_client_.emplace(endpoint, &*rpc_context_);
-  TryInitializeClient();
+  TryInitializeClientSync();
 
   if (config.timeout && replica_state_ != replication::ReplicaState::INVALID) {
     timeout_.emplace(*config.timeout);
     timeout_dispatcher_.emplace();
   }
+
+  // Help the user to get the most accurate replica state possible.
+  if (config.replica_check_frequency > std::chrono::seconds(0)) {
+    replica_checker_.Run("Replica Checker", config.replica_check_frequency, [&] { FrequentCheck(); });
+  }
+}
+
+void Storage::ReplicationClient::TryInitializeClientAsync() {
+  thread_pool_.AddTask([this] {
+    rpc_client_->Abort();
+    this->TryInitializeClientSync();
+  });
+}
+
+void Storage::ReplicationClient::FrequentCheck() {
+  const auto is_success = std::invoke([this]() {
+    try {
+      auto stream{rpc_client_->Stream<replication::FrequentHeartbeatRpc>()};
+      const auto response = stream.AwaitResponse();
+      return response.success;
+    } catch (const rpc::RpcFailedException &) {
+      return false;
+    }
+  });
+  // States: READY, REPLICATING, RECOVERY, INVALID
+  // If success && ready, replicating, recovery -> stay the same because something good is going on.
+  // If success && INVALID -> [it's possible that replica came back to life] -> TryInitializeClient.
+  // If fail -> [replica is not reachable at all] -> INVALID state.
+  // NOTE: TryInitializeClient might return nothing if there is a branching point.
+  // NOTE: The early return pattern simplified the code, but the behavior should be as explained.
+  if (!is_success) {
+    replica_state_.store(replication::ReplicaState::INVALID);
+    return;
+  }
+  if (replica_state_.load() == replication::ReplicaState::INVALID) {
+    TryInitializeClientAsync();
+  }
 }
 
 /// @throws rpc::RpcFailedException
@@ -100,7 +137,7 @@ void Storage::ReplicationClient::InitializeClient() {
   }
 }
 
-void Storage::ReplicationClient::TryInitializeClient() {
+void Storage::ReplicationClient::TryInitializeClientSync() {
   try {
     InitializeClient();
   } catch (const rpc::RpcFailedException &) {
@@ -113,10 +150,7 @@ void Storage::ReplicationClient::TryInitializeClient() {
 
 void Storage::ReplicationClient::HandleRpcFailure() {
   spdlog::error(utils::MessageWithLink("Couldn't replicate data to {}.", name_, "https://memgr.ph/replication"));
-  thread_pool_.AddTask([this] {
-    rpc_client_->Abort();
-    this->TryInitializeClient();
-  });
+  TryInitializeClientAsync();
 }
 
 replication::SnapshotRes Storage::ReplicationClient::TransferSnapshot(const std::filesystem::path &path) {
@@ -193,17 +227,18 @@ void Storage::ReplicationClient::IfStreamingTransaction(const std::function<void
   }
 }
 
-void Storage::ReplicationClient::FinalizeTransactionReplication() {
+std::optional<bool> Storage::ReplicationClient::FinalizeTransactionReplication() {
   // We can only check the state because it guarantees to be only
   // valid during a single transaction replication (if the assumption
   // that this and other transaction replication functions can only be
   // called from a one thread stands)
   if (replica_state_ != replication::ReplicaState::REPLICATING) {
-    return;
+    return std::nullopt;
   }
 
   if (mode_ == replication::ReplicationMode::ASYNC) {
-    thread_pool_.AddTask([this] { this->FinalizeTransactionReplicationInternal(); });
+    thread_pool_.AddTask([this] { [[maybe_unused]] auto finalized = this->FinalizeTransactionReplicationInternal(); });
+    return true;
   } else if (timeout_) {
     MG_ASSERT(mode_ == replication::ReplicationMode::SYNC, "Only SYNC replica can have a timeout.");
     MG_ASSERT(timeout_dispatcher_, "Timeout thread is missing");
@@ -211,7 +246,7 @@ void Storage::ReplicationClient::FinalizeTransactionReplication() {
 
     timeout_dispatcher_->active = true;
     thread_pool_.AddTask([&, this] {
-      this->FinalizeTransactionReplicationInternal();
+      [[maybe_unused]] auto finalized = this->FinalizeTransactionReplicationInternal();
       std::unique_lock main_guard(timeout_dispatcher_->main_lock);
       // TimerThread can finish waiting for timeout
       timeout_dispatcher_->active = false;
@@ -239,12 +274,13 @@ void Storage::ReplicationClient::FinalizeTransactionReplication() {
       // and acces the `active` variable`
       thread_pool_.AddTask([this] { timeout_dispatcher_.reset(); });
     }
+    return true;
   } else {
-    FinalizeTransactionReplicationInternal();
+    return FinalizeTransactionReplicationInternal();
   }
 }
 
-void Storage::ReplicationClient::FinalizeTransactionReplicationInternal() {
+bool Storage::ReplicationClient::FinalizeTransactionReplicationInternal() {
   MG_ASSERT(replica_stream_, "Missing stream for transaction deltas");
   try {
     auto response = replica_stream_->Finalize();
@@ -255,6 +291,7 @@ void Storage::ReplicationClient::FinalizeTransactionReplicationInternal() {
       thread_pool_.AddTask([&, this] { this->RecoverReplica(response.current_commit_timestamp); });
     } else {
       replica_state_.store(replication::ReplicaState::READY);
+      return true;
     }
   } catch (const rpc::RpcFailedException &) {
     replica_stream_.reset();
@@ -264,6 +301,7 @@ void Storage::ReplicationClient::FinalizeTransactionReplicationInternal() {
     }
     HandleRpcFailure();
   }
+  return false;
 }
 
 void Storage::ReplicationClient::RecoverReplica(uint64_t replica_commit) {

src/storage/v2/replication/replication_client.hpp

@@ -103,7 +103,10 @@ class Storage::ReplicationClient {
   // StartTransactionReplication, stream is created.
   void IfStreamingTransaction(const std::function<void(ReplicaStream &handler)> &callback);
 
-  void FinalizeTransactionReplication();
+  // Return none  -> OK
+  // Return true  -> OK
+  // Return false -> FAIL
+  std::optional<bool> FinalizeTransactionReplication();
 
   // Transfer the snapshot file.
   // @param path Path of the snapshot file.
@@ -125,7 +128,7 @@ class Storage::ReplicationClient {
   const auto &Endpoint() const { return rpc_client_->Endpoint(); }
 
  private:
-  void FinalizeTransactionReplicationInternal();
+  [[nodiscard]] bool FinalizeTransactionReplicationInternal();
 
   void RecoverReplica(uint64_t replica_commit);
 
@@ -142,16 +145,14 @@ class Storage::ReplicationClient {
 
   std::vector<RecoveryStep> GetRecoverySteps(uint64_t replica_commit, utils::FileRetainer::FileLocker *file_locker);
 
+  void FrequentCheck();
   void InitializeClient();
-
-  void TryInitializeClient();
-
+  void TryInitializeClientSync();
+  void TryInitializeClientAsync();
   void HandleRpcFailure();
 
   std::string name_;
-
   Storage *storage_;
-
   std::optional<communication::ClientContext> rpc_context_;
   std::optional<rpc::Client> rpc_client_;
 
@@ -198,6 +199,8 @@ class Storage::ReplicationClient {
   //    to ignore concurrency problems inside the client.
   utils::ThreadPool thread_pool_{1};
   std::atomic<replication::ReplicaState> replica_state_{replication::ReplicaState::INVALID};
+
+  utils::Scheduler replica_checker_;
 };
 
 }  // namespace memgraph::storage

src/storage/v2/replication/replication_server.cpp

@@ -60,6 +60,10 @@ Storage::ReplicationServer::ReplicationServer(Storage *storage, io::network::End
     spdlog::debug("Received HeartbeatRpc");
     this->HeartbeatHandler(req_reader, res_builder);
   });
+  rpc_server_->Register<replication::FrequentHeartbeatRpc>([](auto *req_reader, auto *res_builder) {
+    spdlog::debug("Received FrequentHeartbeatRpc");
+    FrequentHeartbeatHandler(req_reader, res_builder);
+  });
   rpc_server_->Register<replication::AppendDeltasRpc>([this](auto *req_reader, auto *res_builder) {
     spdlog::debug("Received AppendDeltasRpc");
     this->AppendDeltasHandler(req_reader, res_builder);
@@ -86,6 +90,13 @@ void Storage::ReplicationServer::HeartbeatHandler(slk::Reader *req_reader, slk::
   slk::Save(res, res_builder);
 }
 
+void Storage::ReplicationServer::FrequentHeartbeatHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
+  replication::FrequentHeartbeatReq req;
+  slk::Load(&req, req_reader);
+  replication::FrequentHeartbeatRes res{true};
+  slk::Save(res, res_builder);
+}
+
 void Storage::ReplicationServer::AppendDeltasHandler(slk::Reader *req_reader, slk::Builder *res_builder) {
   replication::AppendDeltasReq req;
   slk::Load(&req, req_reader);

src/storage/v2/replication/replication_server.hpp

@@ -29,6 +29,7 @@ class Storage::ReplicationServer {
  private:
   // RPC handlers
   void HeartbeatHandler(slk::Reader *req_reader, slk::Builder *res_builder);
+  static void FrequentHeartbeatHandler(slk::Reader *req_reader, slk::Builder *res_builder);
   void AppendDeltasHandler(slk::Reader *req_reader, slk::Builder *res_builder);
   void SnapshotHandler(slk::Reader *req_reader, slk::Builder *res_builder);
   void WalFilesHandler(slk::Reader *req_reader, slk::Builder *res_builder);

src/storage/v2/replication/rpc.lcp

@@ -43,6 +43,12 @@ cpp<#
      (current-commit-timestamp :uint64_t)
      (epoch-id "std::string"))))
 
+;; FrequentHearthbeat is required because calling Heartbeat takes the storage lock.
+;; Configured by `replication_replica_check_delay`.
+(lcp:define-rpc frequent-heartbeat
+  (:request ())
+  (:response ((success :bool))))
+
 (lcp:define-rpc snapshot
   (:request ())
   (:response

src/storage/v2/storage.cpp

@@ -40,6 +40,7 @@
 #include "utils/uuid.hpp"
 
 /// REPLICATION ///
+#include "storage/v2/commit_error.hpp"
 #include "storage/v2/replication/replication_client.hpp"
 #include "storage/v2/replication/replication_server.hpp"
 #include "storage/v2/replication/rpc.hpp"
@@ -813,7 +814,7 @@ EdgeTypeId Storage::Accessor::NameToEdgeType(const std::string_view &name) { ret
 
 void Storage::Accessor::AdvanceCommand() { ++transaction_.command_id; }
 
-utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
+utils::BasicResult<CommitError, void> Storage::Accessor::Commit(
     const std::optional<uint64_t> desired_commit_timestamp) {
   MG_ASSERT(is_transaction_active_, "The transaction is already terminated!");
   MG_ASSERT(!transaction_.must_abort, "The transaction can't be committed!");
@@ -836,7 +837,8 @@ utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
       auto validation_result = ValidateExistenceConstraints(*prev.vertex, storage_->constraints_);
       if (validation_result) {
         Abort();
-        return *validation_result;
+        return CommitError{.type = CommitError::Type::CONSTRAINT_VIOLATION,
+                           .maybe_constraint_violation = *validation_result};
       }
     }
 
@@ -848,6 +850,29 @@ utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
     // Save these so we can mark them used in the commit log.
     uint64_t start_timestamp = transaction_.start_timestamp;
 
+    // Aboring here and after the locked block obviously has an issue if
+    // replica goes down and back up during the execution of the locked block of
+    // code.
+    // Not enough, before the actual commit check all SYNC replicas for availability.
+    bool unable_to_sync_replicate = false;
+    const auto check_replicas = [&]() {
+      storage_->replication_clients_.WithLock([&](auto &clients) {
+        for (auto &client : clients) {
+          // Exclusively SYNC replicas have to be available to commit the transaction.
+          if (client->Mode() == replication::ReplicationMode::SYNC && !client->Timeout().has_value() &&
+              client->State() == replication::ReplicaState::INVALID) {
+            unable_to_sync_replicate = true;
+          }
+        }
+      });
+    };
+
+    check_replicas();
+    if (unable_to_sync_replicate) {
+      Abort();
+      return CommitError{.type = CommitError::Type::UNABLE_TO_SYNC_REPLICATE};
+    }
+
     {
       std::unique_lock<utils::SpinLock> engine_guard(storage_->engine_lock_);
       commit_timestamp_.emplace(storage_->CommitTimestamp(desired_commit_timestamp));
@@ -893,6 +918,8 @@ utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
         // Replica can log only the write transaction received from Main
         // so the Wal files are consistent
         if (storage_->replication_role_ == ReplicationRole::MAIN || desired_commit_timestamp.has_value()) {
+          // TODO(gitbuda): Possible to abort data operation because in this context there is an abort operation.
+          // TODO(gitbuda): If AppendToWal returns false, we can exit this block and Abort.
           storage_->AppendToWal(transaction_, *commit_timestamp_);
         }
 
@@ -915,13 +942,27 @@ utils::BasicResult<ConstraintViolation, void> Storage::Accessor::Commit(
           engine_guard.unlock();
         });
 
+        // NOTE: This will finish/commit the transaction.
         storage_->commit_log_->MarkFinished(start_timestamp);
+        // TODO(gitbuda): Maybe the solution here is to 1. mark 2. check all
+        // the replicas 3. unmark if required... it's not possible to unmark
+        // easily because mark contains mark + update_latest_active which is
+        // not reversable operation
       }
     }
 
+    // TODO(gitbuda): This doesn't have any effect because (it will only if the
+    // constraints are also violated).
+    check_replicas();
+    if (unable_to_sync_replicate) {
+      Abort();
+      return CommitError{.type = CommitError::Type::UNABLE_TO_SYNC_REPLICATE};
+    }
+
     if (unique_constraint_violation) {
       Abort();
-      return *unique_constraint_violation;
+      return CommitError{.type = CommitError::Type::CONSTRAINT_VIOLATION,
+                         .maybe_constraint_violation = *unique_constraint_violation};
     }
   }
   is_transaction_active_ = false;
@@ -1124,6 +1165,18 @@ EdgeTypeId Storage::NameToEdgeType(const std::string_view &name) {
   return EdgeTypeId::FromUint(name_id_mapper_.NameToId(name));
 }
 
+// TODO(gitbuda): Hard to abort global operations in SYNC replication mode
+// because there is no an abort op for that yet, one idea is to just apply
+// reverse operation, e.g., CreateIndex <-> DropIndex.
+//
+// Another idea is to double check that all replicas have relevant data prior
+// to calling MarkFinished. That approach would work in both data replication
+// and global operation cases.
+//
+// EDGE CASE 1: What if the first SYNC replica is alive, receives the delta
+// object, while the second SYNC replica is dead? (replication clients are
+// stored in a vector and accessed one by one)
+
 bool Storage::CreateIndex(LabelId label, const std::optional<uint64_t> desired_commit_timestamp) {
   std::unique_lock<utils::RWLock> storage_guard(main_lock_);
   if (!indices_.label_index.CreateIndex(label, vertices_.access())) return false;
@@ -1572,6 +1625,9 @@ void Storage::FinalizeWalFile() {
   }
 }
 
+// TODO(gitbuda): Hard to abort data operation in SYNC replication mode because:
+//   * Just calling Abort inside AppendToWal for some reason causes infinite loop.
+
 void Storage::AppendToWal(const Transaction &transaction, uint64_t final_commit_timestamp) {
   if (!InitializeWalFile()) return;
   // Traverse deltas and append them to the WAL file.
@@ -1743,10 +1799,15 @@ void Storage::AppendToWal(const Transaction &transaction, uint64_t final_commit_
   FinalizeWalFile();
 
   replication_clients_.WithLock([&](auto &clients) {
+    bool all_sync_replicas_ok = true;
     for (auto &client : clients) {
+      // TODO(gitbuda): SEMI-SYNC should be exculded from here.
+      if (client->Mode() == replication::ReplicationMode::SYNC)
       client->IfStreamingTransaction([&](auto &stream) { stream.AppendTransactionEnd(final_commit_timestamp); });
+      // TODO(gitbuda): FinalizeTransactionReplication should also indicate that eveything went well for SYNC replicas.
       client->FinalizeTransactionReplication();
     }
+    return all_sync_replicas_ok;
   });
 }
 

src/storage/v2/storage.hpp

@@ -43,6 +43,7 @@
 
 /// REPLICATION ///
 #include "rpc/server.hpp"
+#include "storage/v2/commit_error.hpp"
 #include "storage/v2/replication/config.hpp"
 #include "storage/v2/replication/enums.hpp"
 #include "storage/v2/replication/rpc.hpp"
@@ -308,11 +309,12 @@ class Storage final {
 
     void AdvanceCommand();
 
-    /// Commit returns `ConstraintViolation` if the changes made by this
-    /// transaction violate an existence or unique constraint. In that case the
-    /// transaction is automatically aborted. Otherwise, void is returned.
+    /// Commit returns `CommitError` if the changes made by this transaction
+    /// violate an existence, unique constraint or data could NOT be replicated
+    /// to SYNC replica. In that case the transaction is automatically aborted.
+    /// Otherwise, void is returned.
     /// @throw std::bad_alloc
-    utils::BasicResult<ConstraintViolation, void> Commit(std::optional<uint64_t> desired_commit_timestamp = {});
+    utils::BasicResult<CommitError, void> Commit(std::optional<uint64_t> desired_commit_timestamp = {});
 
     /// @throw std::bad_alloc
     void Abort();

tests/e2e/replication/CMakeLists.txt

@@ -5,3 +5,7 @@ target_link_libraries(memgraph__e2e__replication__constraints gflags mgclient mg
 
 add_executable(memgraph__e2e__replication__read_write_benchmark read_write_benchmark.cpp)
 target_link_libraries(memgraph__e2e__replication__read_write_benchmark gflags json mgclient mg-utils mg-io Threads::Threads)
+
+copy_e2e_python_files(replication_show common.py)
+copy_e2e_python_files(replication_show conftest.py)
+copy_e2e_python_files(replication_show show.py)

tests/e2e/replication/common.py

@@ -0,0 +1,26 @@
+# Copyright 2022 Memgraph Ltd.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+# License, and you may not use this file except in compliance with the Business Source License.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0, included in the file
+# licenses/APL.txt.
+
+import mgclient
+import typing
+
+
+def execute_and_fetch_all(
+    cursor: mgclient.Cursor, query: str, params: dict = {}
+) -> typing.List[tuple]:
+    cursor.execute(query, params)
+    return cursor.fetchall()
+
+
+def connect(**kwargs) -> mgclient.Connection:
+    connection = mgclient.connect(**kwargs)
+    connection.autocommit = True
+    return connection

tests/e2e/replication/conftest.py

@@ -0,0 +1,44 @@
+# Copyright 2022 Memgraph Ltd.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+# License, and you may not use this file except in compliance with the Business Source License.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0, included in the file
+# licenses/APL.txt.
+
+import pytest
+
+from common import execute_and_fetch_all, connect
+
+
+# The fixture here is more complex because the connection has to be
+# parameterized based on the test parameters (info has to be available on both
+# sides).
+#
+# https://docs.pytest.org/en/latest/example/parametrize.html#indirect-parametrization
+# is not an elegant/feasible solution here.
+#
+# The solution was independently developed and then I stumbled upon the same
+# approach here https://stackoverflow.com/a/68286553/4888809 which I think is
+# optimal.
+@pytest.fixture(scope="function")
+def connection():
+    connection_holder = None
+    role_holder = None
+
+    def inner_connection(port, role):
+        nonlocal connection_holder, role_holder
+        connection_holder = connect(host="localhost", port=port)
+        role_holder = role
+        return connection_holder
+
+    yield inner_connection
+
+    # Only main instance can be cleaned up because replicas do NOT accept
+    # writes.
+    if role_holder == "main":
+        cursor = connection_holder.cursor()
+        execute_and_fetch_all(cursor, "MATCH (n) DETACH DELETE n;")

tests/e2e/replication/show.py

@@ -0,0 +1,46 @@
+# Copyright 2022 Memgraph Ltd.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+# License, and you may not use this file except in compliance with the Business Source License.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0, included in the file
+# licenses/APL.txt.
+
+import sys
+
+import pytest
+from common import execute_and_fetch_all
+
+
+@pytest.mark.parametrize(
+    "port, role",
+    [(7687, "main"), (7688, "replica"), (7689, "replica"), (7690, "replica")],
+)
+def test_show_replication_role(port, role, connection):
+    cursor = connection(port, role).cursor()
+    data = execute_and_fetch_all(cursor, "SHOW REPLICATION ROLE;")
+    assert cursor.description[0].name == "replication role"
+    assert data[0][0] == role
+
+
+def test_show_replicas(connection):
+    cursor = connection(7687, "main").cursor()
+    actual_data = set(execute_and_fetch_all(cursor, "SHOW REPLICAS;"))
+
+    expected_column_names = {"name", "socket_address", "sync_mode", "timeout"}
+    actual_column_names = {x.name for x in cursor.description}
+    assert expected_column_names == actual_column_names
+
+    expected_data = {
+        ("replica_1", "127.0.0.1:10001", "sync", 0),
+        ("replica_2", "127.0.0.1:10002", "sync", 1.0),
+        ("replica_3", "127.0.0.1:10003", "async", None),
+    }
+    assert expected_data == actual_data
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/replication/workloads.yaml

@@ -46,4 +46,31 @@ workloads:
     args: []
     <<: *template_cluster
 
-
+  - name: "Show"
+    binary: "tests/e2e/pytest_runner.sh"
+    args: ["replication/show.py"]
+    cluster:
+      replica_1:
+        args: ["--bolt-port", "7688", "--log-level=TRACE"]
+        log_file: "replication-e2e-replica1.log"
+        setup_queries: ["SET REPLICATION ROLE TO REPLICA WITH PORT 10001;"]
+        validation_queries: []
+      replica_2:
+        args: ["--bolt-port", "7689", "--log-level=TRACE"]
+        log_file: "replication-e2e-replica2.log"
+        setup_queries: ["SET REPLICATION ROLE TO REPLICA WITH PORT 10002;"]
+        validation_queries: []
+      replica_3:
+        args: ["--bolt-port", "7690", "--log-level=TRACE"]
+        log_file: "replication-e2e-replica3.log"
+        setup_queries: ["SET REPLICATION ROLE TO REPLICA WITH PORT 10003;"]
+        validation_queries: []
+      main:
+        args: ["--bolt-port", "7687", "--log-level=TRACE"]
+        log_file: "replication-e2e-main.log"
+        setup_queries: [
+          "REGISTER REPLICA replica_1 SYNC WITH TIMEOUT 0 TO '127.0.0.1:10001'",
+          "REGISTER REPLICA replica_2 SYNC WITH TIMEOUT 1 TO '127.0.0.1:10002'",
+          "REGISTER REPLICA replica_3 ASYNC TO '127.0.0.1:10003'"
+        ]
+        validation_queries: []

tests/unit/storage_v2_constraints.cpp

@@ -137,8 +137,10 @@ TEST_F(ConstraintsTest, ExistenceConstraintsViolationOnCommit) {
 
     auto res = acc.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(),
-              (ConstraintViolation{ConstraintViolation::Type::EXISTENCE, label1, std::set<PropertyId>{prop1}}));
+    EXPECT_EQ(
+        res.GetError(),
+        (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
+                     ConstraintViolation{ConstraintViolation::Type::EXISTENCE, label1, std::set<PropertyId>{prop1}}}));
   }
 
   {
@@ -157,8 +159,10 @@ TEST_F(ConstraintsTest, ExistenceConstraintsViolationOnCommit) {
 
     auto res = acc.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(),
-              (ConstraintViolation{ConstraintViolation::Type::EXISTENCE, label1, std::set<PropertyId>{prop1}}));
+    EXPECT_EQ(
+        res.GetError(),
+        (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
+                     ConstraintViolation{ConstraintViolation::Type::EXISTENCE, label1, std::set<PropertyId>{prop1}}}));
   }
 
   {
@@ -458,8 +462,10 @@ TEST_F(ConstraintsTest, UniqueConstraintsViolationOnCommit1) {
     ASSERT_NO_ERROR(vertex2.SetProperty(prop1, PropertyValue(1)));
     auto res = acc.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(),
-              (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}));
+    EXPECT_EQ(
+        res.GetError(),
+        (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
+                     ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}}));
   }
 }
 
@@ -500,8 +506,10 @@ TEST_F(ConstraintsTest, UniqueConstraintsViolationOnCommit2) {
     ASSERT_NO_ERROR(acc2.Commit());
     auto res = acc3.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(),
-              (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}));
+    EXPECT_EQ(
+        res.GetError(),
+        (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
+                     ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}}));
   }
 }
 
@@ -545,12 +553,16 @@ TEST_F(ConstraintsTest, UniqueConstraintsViolationOnCommit3) {
 
     auto res = acc2.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(),
-              (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}));
+    EXPECT_EQ(
+        res.GetError(),
+        (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
+                     ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}}));
     res = acc3.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(),
-              (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}));
+    EXPECT_EQ(
+        res.GetError(),
+        (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
+                     ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1}}}));
   }
 }
 
@@ -620,7 +632,9 @@ TEST_F(ConstraintsTest, UniqueConstraintsLabelAlteration) {
 
     auto res = acc.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(), (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1}}));
+    EXPECT_EQ(res.GetError(),
+              (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
+                           ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1}}}));
   }
 
   {
@@ -654,7 +668,9 @@ TEST_F(ConstraintsTest, UniqueConstraintsLabelAlteration) {
 
     auto res = acc1.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(), (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1}}));
+    EXPECT_EQ(res.GetError(),
+              (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
+                           ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1}}}));
   }
 }
 
@@ -749,8 +765,9 @@ TEST_F(ConstraintsTest, UniqueConstraintsMultipleProperties) {
     ASSERT_NO_ERROR(vertex2->SetProperty(prop2, PropertyValue(2)));
     auto res = acc.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(),
-              (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set<PropertyId>{prop1, prop2}}));
+    EXPECT_EQ(res.GetError(), (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
+                                           ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1,
+                                                               std::set<PropertyId>{prop1, prop2}}}));
   }
 
   // Then change the second property of both vertex to null. Property values of
@@ -861,7 +878,9 @@ TEST_F(ConstraintsTest, UniqueConstraintsInsertRemoveAbortInsert) {
 
     auto res = acc.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(), (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1, prop2}}));
+    EXPECT_EQ(res.GetError(),
+              (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
+                           ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1, prop2}}}));
   }
 }
 
@@ -900,7 +919,9 @@ TEST_F(ConstraintsTest, UniqueConstraintsDeleteVertexSetProperty) {
 
     auto res = acc1.Commit();
     ASSERT_TRUE(res.HasError());
-    EXPECT_EQ(res.GetError(), (ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1}}));
+    EXPECT_EQ(res.GetError(),
+              (CommitError{CommitError::Type::CONSTRAINT_VIOLATION,
+                           ConstraintViolation{ConstraintViolation::Type::UNIQUE, label1, std::set{prop1}}}));
 
     ASSERT_NO_ERROR(acc2.Commit());
   }