Set coordinator role

src/dbms/replication_handler.cpp

@@ -75,6 +75,7 @@ bool ReplicationHandler::SetReplicationRoleMain() {
     return true;
   };
 
+  // COORDINATOR cannot become main
   auto const coordinator_handler = [](replication::RoleCoordinatorData const &) { return false; };
 
   // TODO: under lock
@@ -83,13 +84,14 @@ bool ReplicationHandler::SetReplicationRoleMain() {
 }
 
 bool ReplicationHandler::SetReplicationRoleCoordinator() {
-  auto const main_handler = [](RoleMainData const &) { return false; };
+  // Upgrading REPLICA to COORDINATOR is not supported
   auto const replica_handler = [](RoleReplicaData const &) { return false; };
-
-  auto const coordinator_handler = [](replication::RoleCoordinatorData const &) {
-    // TODO
-    //  set RPCs
-    return true;
+  // If we are already COORDINATOR, we don't want to change anything
+  auto const coordinator_handler = [](replication::RoleCoordinatorData const &) { return false; };
+  // Upgrade MAIN to COORDINATOR
+  // TODO: (andi) Probably more complex steps will be necessary here.
+  auto const main_handler = [this](RoleMainData const &) {
+    return dbms_handler_.ReplicationState().SetReplicationRoleCoordinator();
   };
 
   // TODO: under lock
@@ -99,7 +101,7 @@ bool ReplicationHandler::SetReplicationRoleCoordinator() {
 
 bool ReplicationHandler::SetReplicationRoleReplica(const memgraph::replication::ReplicationServerConfig &config) {
   // We don't want to restart the server if we're already a REPLICA
-  if (dbms_handler_.ReplicationState().IsReplica()) {
+  if (dbms_handler_.ReplicationState().IsReplica() || dbms_handler_.ReplicationState().IsCoordinator()) {
     return false;
   }
 
@@ -131,6 +133,7 @@ bool ReplicationHandler::SetReplicationRoleReplica(const memgraph::replication::
                                      }
                                      return true;
                                    },
+                                   /// TODO: (andi) Assert that this code cannot execute
                                    [](replication::RoleCoordinatorData const &) { return false; }},
                  dbms_handler_.ReplicationState().ReplicationData());
   // TODO Handle error (restore to main?)

src/query/interpreter.cpp

@@ -286,7 +286,7 @@ class ReplQueryHandler final : public query::ReplicationQueryHandler {
       }
     } else if (replication_role == ReplicationQuery::ReplicationRole::COORDINATOR) {
       if (!handler_.SetReplicationRoleCoordinator()) {
-        throw QueryRuntimeException("Couldn't set role to main!");
+        throw QueryRuntimeException("Couldn't set role to coordinator!");
       }
     } else {
       if (!port || *port < 0 || *port > std::numeric_limits<uint16_t>::max()) {

src/replication/include/replication/state.hpp

@@ -54,9 +54,18 @@ struct RoleReplicaData {
 };
 
 struct RoleCoordinatorData {
+  RoleCoordinatorData() = default;
+  ~RoleCoordinatorData() = default;
+
+  RoleCoordinatorData(RoleCoordinatorData const &) = delete;
+  RoleCoordinatorData &operator=(RoleCoordinatorData const &) = delete;
+  RoleCoordinatorData(RoleCoordinatorData &&) = default;
+  RoleCoordinatorData &operator=(RoleCoordinatorData &&) = default;
+
   std::list<ReplicationClient> registered_replicas_{};
   std::unique_ptr<ReplicationClient> main;
-  // TODO does it need epoch
+
+  // TODO: (andi) Does it need epoch or some other way or tracking what is going on?
 };
 
 // Global (instance) level object
@@ -94,6 +103,9 @@ struct ReplicationState {
   bool ShouldPersist() const { return nullptr != durability_; }
   bool TryPersistRoleMain(std::string new_epoch);
   bool TryPersistRoleReplica(const ReplicationServerConfig &config);
+  /// TODO: (andi) If we will need epoch or something to track, we will need to pass it here as argument
+  bool TryPersistRoleCoordinator();
+
   bool TryPersistUnregisterReplica(std::string_view name);
   bool TryPersistRegisteredReplica(const ReplicationClientConfig &config);
 
@@ -103,8 +115,8 @@ struct ReplicationState {
   utils::BasicResult<RegisterReplicaError, ReplicationClient *> RegisterReplica(const ReplicationClientConfig &config);
 
   bool SetReplicationRoleMain();
-
   bool SetReplicationRoleReplica(const ReplicationServerConfig &config);
+  bool SetReplicationRoleCoordinator();
 
  private:
   bool HandleVersionMigration(durability::ReplicationRoleEntry &data) const;

src/replication/include/replication/status.hpp

@@ -46,11 +46,17 @@ struct ReplicaRole {
   friend bool operator==(ReplicaRole const &, ReplicaRole const &) = default;
 };
 
+// fragment of key: "__replication_role"
+// TODO: (andi) Check if you will need to add some epoch or something for managing coordinator role
+struct CoordinatorRole {
+  friend bool operator==(CoordinatorRole const &, CoordinatorRole const &) = default;
+};
+
 // from key: "__replication_role"
 struct ReplicationRoleEntry {
   DurabilityVersion version =
       DurabilityVersion::V2;  // if not latest then migration required for kReplicationReplicaPrefix
-  std::variant<MainRole, ReplicaRole> role;
+  std::variant<MainRole, ReplicaRole, CoordinatorRole> role;
 
   friend bool operator==(ReplicationRoleEntry const &, ReplicationRoleEntry const &) = default;
 };

src/replication/state.cpp

@@ -91,6 +91,20 @@ bool ReplicationState::TryPersistRoleMain(std::string new_epoch) {
   return false;
 }
 
+bool ReplicationState::TryPersistRoleCoordinator() {
+  if (!ShouldPersist()) return true;
+
+  /// TODO: (andi) Epoch/identifier possible change
+  auto data = durability::ReplicationRoleEntry{.role = durability::CoordinatorRole{}};
+
+  if (durability_->Put(durability::kReplicationRoleName, nlohmann::json(data).dump())) {
+    role_persisted = RolePersisted::YES;
+    return true;
+  }
+  spdlog::error("Error when saving COORDINATOR replication role in settings.");
+  return false;
+}
+
 bool ReplicationState::TryPersistUnregisterReplica(std::string_view name) {
   if (!ShouldPersist()) return true;
 
@@ -151,6 +165,9 @@ auto ReplicationState::FetchReplicationData() -> FetchReplicationResult_t {
             [&](durability::ReplicaRole &&r) -> FetchReplicationResult_t {
               return {RoleReplicaData{r.config, std::make_unique<ReplicationServer>(r.config)}};
             },
+            /// TODO: (andi) This must change for sure because this is the step in which we should create
+            /// ReplicationClient for MAIN and for REPLICAs
+            [&](durability::CoordinatorRole &&) -> FetchReplicationResult_t { return {RoleCoordinatorData{}}; },
         },
         std::move(data.role));
   } catch (...) {
@@ -233,6 +250,16 @@ bool ReplicationState::SetReplicationRoleReplica(const ReplicationServerConfig &
   return true;
 }
 
+bool ReplicationState::SetReplicationRoleCoordinator() {
+  /// TODO: (andi) Think if need epoch and how is this going to be tracked.
+  if (!TryPersistRoleCoordinator()) {
+    return false;
+  }
+
+  replication_data_ = RoleCoordinatorData{};
+  return true;
+}
+
 utils::BasicResult<RegisterReplicaError, ReplicationClient *> ReplicationState::RegisterReplica(
     const ReplicationClientConfig &config) {
   auto const replica_handler = [](RoleReplicaData const &) { return RegisterReplicaError::NOT_MAIN; };

src/replication/status.cpp

@@ -28,6 +28,8 @@ constexpr auto *kReplicationRole = "replication_role";
 constexpr auto *kEpoch = "epoch";
 constexpr auto *kVersion = "durability_version";
 
+/// TODO: (andi) This will have to change for MAIN because now also MAIN will have its own server. Do we then have to
+/// introduce new durability version?
 void to_json(nlohmann::json &j, const ReplicationRoleEntry &p) {
   auto processMAIN = [&](MainRole const &main) {
     j = nlohmann::json{{kVersion, p.version}, {kReplicationRole, ReplicationRole::MAIN}, {kEpoch, main.epoch.id()}};
@@ -41,7 +43,13 @@ void to_json(nlohmann::json &j, const ReplicationRoleEntry &p) {
         // TODO: SSL
     };
   };
-  std::visit(utils::Overloaded{processMAIN, processREPLICA}, p.role);
+
+  /// TODO: (andi) Extend this if you need to introduce some kind of timestamp to coordinator.
+  auto processCOORDINATOR = [&](CoordinatorRole const &) {
+    j = nlohmann::json{{kVersion, p.version}, {kReplicationRole, ReplicationRole::COORDINATOR}};
+  };
+
+  std::visit(utils::Overloaded{processMAIN, processREPLICA, processCOORDINATOR}, p.role);
 }
 
 void from_json(const nlohmann::json &j, ReplicationRoleEntry &p) {

tests/e2e/replication/CMakeLists.txt

@@ -14,7 +14,9 @@ copy_e2e_python_files(replication_show conftest.py)
 copy_e2e_python_files(replication_show show.py)
 copy_e2e_python_files(replication_show show_while_creating_invalid_state.py)
 copy_e2e_python_files(replication_show edge_delete.py)
-copy_e2e_python_files(replication_show replication_general.py)
+copy_e2e_python_files(replication_show set_coordinator_role.py)
+copy_e2e_python_files(replication_show coordinator_test.py)
+copy_e2e_python_files(replication_show replica_test.py)
 copy_e2e_python_files(replication_show manual_failover.py)
 copy_e2e_python_files_from_parent_folder(replication_show ".." memgraph.py)
 copy_e2e_python_files_from_parent_folder(replication_show ".." interactive_mg_runner.py)

tests/e2e/replication/coordinator_test.py

@@ -0,0 +1,38 @@
+# Copyright 2023 Memgraph Ltd.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+# License, and you may not use this file except in compliance with the Business Source License.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0, included in the file
+# licenses/APL.txt.
+
+import sys
+
+import pytest
+from common import connect_default, execute_and_fetch_all
+
+
+def test_registering_only_replicas_on_coordinator():
+    # cursor = connect_default().cursor()
+    # execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO COORDINATOR;")
+    # execute_and_fetch_all(cursor, "REGISTER REPLICA replica_1 SYNC TO '127.0.0.1:10001';")
+    # execute_and_fetch_all(cursor, "REGISTER REPLICA replica_2 SYNC TO '127.0.0.1:10002';")
+    # execute_and_fetch_all(cursor, "SHOW REPLICATION CLUSTER;")
+    # TODO: (andi) Test that the replicas are registered
+    pass
+
+
+def test_registering_only_main_on_coordinator():
+    pass
+    # cursor = connect_default().cursor()
+    # execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO COORDINATOR;")
+    # execute_and_fetch_all(cursor, "REGISTER MAIN TO '127.0.0.1:10005';")
+    # execute_and_fetch_all(cursor, "SHOW REPLICATION CLUSTER;")
+    # TODO: (andi) Test that the main is registered
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/replication/main_test.py

@@ -0,0 +1,28 @@
+# Copyright 2023 Memgraph Ltd.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+# License, and you may not use this file except in compliance with the Business Source License.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0, included in the file
+# licenses/APL.txt.
+
+import sys
+
+import pytest
+from common import connect_default, execute_and_fetch_all
+
+
+def test_setting_port_on_main():
+    pass
+    # cursor = connect_default().cursor()
+    # execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO MAIN WITH PORT 10011;")
+    # res = execute_and_fetch_all(cursor, "SHOW REPLICATION ROLE;")
+    # assert cursor.description[0].name == "replication role"
+    # assert res[0][0] == "main"
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/replication/replica_test.py

@@ -0,0 +1,28 @@
+# Copyright 2023 Memgraph Ltd.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+# License, and you may not use this file except in compliance with the Business Source License.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0, included in the file
+# licenses/APL.txt.
+
+import sys
+
+import pytest
+from common import execute_and_fetch_all
+
+
+def test_replica_cannot_register_replica(connection):
+    cursor = connection(7688, "role").cursor()
+    try:
+        execute_and_fetch_all(cursor, "REGISTER REPLICA replica_1 SYNC TO '127.0.0.1:10001'")
+        assert False
+    except Exception as e:
+        assert str(e) == "Replica can't register another replica!"
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/replication/set_coordinator_role.py

@@ -33,44 +33,24 @@ def test_coordinator_role_no_port():
     assert res[0][0] == "coordinator"
 
 
-def test_setting_port_on_main():
-    cursor = connect_default().cursor()
-    execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO MAIN WITH PORT 10011;")
-    res = execute_and_fetch_all(cursor, "SHOW REPLICATION ROLE;")
-    assert cursor.description[0].name == "replication role"
-    assert res[0][0] == "main"
-
-
-def test_registering_replicas_no_main_on_coordinator():
-    cursor = connect_default().cursor()
-    execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO COORDINATOR;")
-    execute_and_fetch_all(cursor, "REGISTER REPLICA replica_1 SYNC TO '127.0.0.1:10001';")
-    execute_and_fetch_all(cursor, "REGISTER REPLICA replica_2 SYNC TO '127.0.0.1:10002';")
+def test_registering_only_replicas_on_coordinator():
+    # cursor = connect_default().cursor()
+    # execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO COORDINATOR;")
+    # execute_and_fetch_all(cursor, "REGISTER REPLICA replica_1 SYNC TO '127.0.0.1:10001';")
+    # execute_and_fetch_all(cursor, "REGISTER REPLICA replica_2 SYNC TO '127.0.0.1:10002';")
     # execute_and_fetch_all(cursor, "SHOW REPLICATION CLUSTER;")
     # TODO: (andi) Test that the replicas are registered
+    pass
 
 
-def test_registering_main_no_replicas_on_coordinator():
-    cursor = connect_default().cursor()
-    execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO COORDINATOR;")
-    execute_and_fetch_all(cursor, "REGISTER MAIN TO '127.0.0.1:10005';")
+def test_registering_only_main_on_coordinator():
+    pass
+    # cursor = connect_default().cursor()
+    # execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO COORDINATOR;")
+    # execute_and_fetch_all(cursor, "REGISTER MAIN TO '127.0.0.1:10005';")
     # execute_and_fetch_all(cursor, "SHOW REPLICATION CLUSTER;")
     # TODO: (andi) Test that the main is registered
 
 
-# TODO: Test that SHOW REPLICATION CLUSTER can be only called on coordinator
-# TODO: Test that REGISTER MAIN passes on coordinator
-
-
-def test_replica_cannot_register_replica():
-    cursor = connect_default().cursor()
-    execute_and_fetch_all(cursor, "SET REPLICATION ROLE TO REPLICA WITH PORT 10011;")
-    try:
-        execute_and_fetch_all(cursor, "REGISTER REPLICA replica_1 SYNC TO '127.0.0.1:10001'")
-        assert False
-    except Exception as e:
-        assert str(e) == "Replica can't register another replica!"
-
-
 if __name__ == "__main__":
     sys.exit(pytest.main([__file__, "-rA"]))

tests/e2e/replication/workloads.yaml

@@ -52,19 +52,38 @@ template_cluster: &template_cluster
       ]
       <<: *template_validation_queries
 
-in_memory_cluster: &in_memory_cluster
+set_coordinator_role_cluster: &set_coordinator_role_cluster
   cluster:
     main:
       args: ["--bolt-port", "7687", "--log-level=TRACE"]
-      log_file: "replication_general_e2e.log"
+      log_file: "set_coordinator_role_e2e.log"
       setup_queries: []
       validation_queries: []
 
+coordinator_role_cluster: &coordinator_role_cluster
+  cluster:
+    main:
+      args: ["--bolt-port", "7687", "--log-level=TRACE"]
+      log_file: "coordinator_role_e2e.log"
+      setup_queries: ["SET REPLICATION ROLE TO COORDINATOR;"]
+      validation_queries: []
+
+
 workloads:
-  - name: "Replication general"
+  - name: "Set coordinator role"
     binary: "tests/e2e/pytest_runner.sh"
-    args: ["replication/replication_general.py"]
-    <<: *in_memory_cluster
+    args: ["replication/set_coordinator_role.py"]
+    <<: *set_coordinator_role_cluster
+
+  - name: "Coordinator test"
+    binary: "tests/e2e/pytest_runner.sh"
+    args: ["replication/coordinator_test.py"]
+    <<: *coordinator_role_cluster
+
+  - name: "Replica test"
+    binary: "tests/e2e/pytest_runner.sh"
+    args: ["replication/replica_test.py"]
+    <<: *template_cluster
 
   - name: "Constraints"
     binary: "tests/e2e/replication/memgraph__e2e__replication__constraints"
@@ -81,40 +100,14 @@ workloads:
     args: []
     <<: *template_cluster
 
-  - name: "Manual failover cluster setup"
+  - name: "Show while creating invalid state"
     binary: "tests/e2e/pytest_runner.sh"
-    args: ["replication/manual_failover.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: []
-      main:
-        args: ["--bolt-port", "7687", "--log-level=trace"]
-        log_file: "replication-e2e-main.log"
-        setup_queries: [
-          "SET REPLICATION ROLE TO MAIN WITH PORT 10003;",
-          "REGISTER REPLICA REPLICA_1 SYNC TO '127.0.0.1:10001'",
-          "REGISTER REPLICA REPLICA_2 SYNC TO '127.0.0.1:10002'",
-        ]
-        validation_queries: []
-      coordinator:
-        args: ["--bolt-port", "7690", "--log-level=trace"]
-        log_file: "replication-e2e-coordinator.log"
-        setup_queries: [
-          "SET REPLICATION ROLE TO COORDINATOR;",
-          "REGISTER MAIN TO '127.0.0.1:10003'",
-          "REGISTER REPLICA REPLICA_1 SYNC TO '127.0.0.1:10001'",
-          "REGISTER REPLICA REPLICA_2 SYNC TO '127.0.0.1:10002'",
-        ]
-        validation_queries: []
+    args: ["replication/show_while_creating_invalid_state.py"]
 
+  - name: "Delete edge replication"
+    binary: "tests/e2e/pytest_runner.sh"
+    args: ["replication/edge_delete.py"]
+    <<: *template_simple_cluster
 
   - name: "show"
     binary: "tests/e2e/pytest_runner.sh"
@@ -144,12 +137,3 @@ workloads:
           "register replica replica_3 async to '127.0.0.1:10003'"
         ]
         validation_queries: []
-
-  - name: "Show while creating invalid state"
-    binary: "tests/e2e/pytest_runner.sh"
-    args: ["replication/show_while_creating_invalid_state.py"]
-
-  - name: "Delete edge replication"
-    binary: "tests/e2e/pytest_runner.sh"
-    args: ["replication/edge_delete.py"]
-    <<: *template_simple_cluster