service/raft: introduce group0_upgrade_state

Define an enum class, `group0_upgrade_state`, describing the state of
the upgrade procedure (implemented in later commits).

Provide IDL definitions for (de)serialization.

The node will have its current upgrade state stored on disk in
`system.scylla_local` under the `group0_upgrade_state` key. If the key
is not present we assume `use_pre_raft_procedures` (meaning we haven't
started upgrading yet or we're at the beginning of upgrade).

Introduce `system_keyspace` accessor methods for storing and retrieving
the on-disk state.

db/system_keyspace.cc

@@ -3262,6 +3262,56 @@ future<mutation> system_keyspace::get_group0_history(distributed<service::storag
     co_return mutation(s, partition_key::from_singular(*s, GROUP0_HISTORY_KEY));
 }
 
+static constexpr auto GROUP0_UPGRADE_STATE_KEY = "group0_upgrade_state";
+
+future<service::group0_upgrade_state> system_keyspace::load_group0_upgrade_state() {
+    auto s = co_await get_scylla_local_param_as<sstring>(GROUP0_UPGRADE_STATE_KEY);
+
+    if (!s || *s == "use_pre_raft_procedures") {
+        co_return service::group0_upgrade_state::use_pre_raft_procedures;
+    } else if (*s == "synchronize") {
+        co_return service::group0_upgrade_state::synchronize;
+    } else if (*s == "use_post_raft_procedures") {
+        co_return service::group0_upgrade_state::use_post_raft_procedures;
+    } else if (*s == "recovery") {
+        co_return service::group0_upgrade_state::recovery;
+    }
+
+    slogger.error(
+            "load_group0_upgrade_state(): unknown value '{}' for key 'group0_upgrade_state' in Scylla local table."
+            " Did you change the value manually?"
+            " Correct values are: 'use_pre_raft_procedures', 'synchronize', 'use_post_raft_procedures', 'recovery'."
+            " Assuming 'recovery'.", *s);
+    // We don't call `on_internal_error` which would probably prevent the node from starting, but enter `recovery`
+    // allowing the user to fix their cluster.
+    co_return service::group0_upgrade_state::recovery;
+}
+
+future<> system_keyspace::save_group0_upgrade_state(service::group0_upgrade_state s) {
+    auto value = [s] () constexpr {
+        switch (s) {
+            case service::group0_upgrade_state::use_post_raft_procedures:
+                return "use_post_raft_procedures";
+            case service::group0_upgrade_state::synchronize:
+                return "synchronize";
+            case service::group0_upgrade_state::recovery:
+                // It should not be necessary to ever save this state internally - the user sets it manually
+                // (e.g. from cqlsh) if recovery is needed - but handle the case anyway.
+                return "recovery";
+            case service::group0_upgrade_state::use_pre_raft_procedures:
+                // It should not be necessary to ever save this state, but handle the case anyway.
+                return "use_pre_raft_procedures";
+        }
+
+        on_internal_error(slogger, format(
+                "save_group0_upgrade_state: given value is outside the set of possible values (integer value: {})."
+                " This may have been caused by undefined behavior; best restart your system.",
+                static_cast<uint8_t>(s)));
+    }();
+
+    return set_scylla_local_param(GROUP0_UPGRADE_STATE_KEY, value);
+}
+
 sstring system_keyspace_name() {
     return system_keyspace::NAME;
 }

db/system_keyspace.hh

@@ -25,6 +25,7 @@
 #include <seastar/core/distributed.hh>
 #include "cdc/generation_id.hh"
 #include "locator/host_id.hh"
+#include "service/raft/group0_upgrade.hh"
 
 namespace service {
 
@@ -457,6 +458,9 @@ public:
     // Assumes that the history table exists, i.e. Raft experimental feature is enabled.
     static future<mutation> get_group0_history(distributed<service::storage_proxy>&);
 
+    future<service::group0_upgrade_state> load_group0_upgrade_state();
+    future<> save_group0_upgrade_state(service::group0_upgrade_state);
+
     system_keyspace(sharded<cql3::query_processor>& qp, sharded<replica::database>& db) noexcept;
     ~system_keyspace();
     future<> start();

idl/group0.idl.hh

@@ -19,6 +19,13 @@ struct group0_peer_exchange {
     std::variant<std::monostate, service::group0_info, std::vector<raft::server_address>> info;
 };
 
+enum class group0_upgrade_state : uint8_t {
+    recovery = 0,
+    use_pre_raft_procedures = 1,
+    synchronize = 2,
+    use_post_raft_procedures = 3,
+};
+
 verb [[with_client_info, with_timeout]] group0_peer_exchange (std::vector<raft::server_address> peers) -> service::group0_peer_exchange;
 verb [[with_client_info, with_timeout]] group0_modify_config (raft::group_id gid, std::vector<raft::config_member> add, std::vector<raft::server_id> del);
 

message/messaging_service.cc

@@ -41,6 +41,7 @@
 #include "cache_temperature.hh"
 #include "raft/raft.hh"
 #include "service/raft/messaging.hh"
+#include "service/raft/group0_upgrade.hh"
 #include "replica/exceptions.hh"
 #include "serializer.hh"
 #include "full_position.hh"

service/raft/group0_upgrade.hh

@@ -0,0 +1,40 @@
+/*
+ * Copyright (C) 2022-present ScyllaDB
+ */
+
+/*
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+#pragma once
+
+#include <iostream>
+
+namespace service {
+
+enum class group0_upgrade_state : uint8_t {
+    // In recovery state group 0 is disabled.
+    // The node does not attempt any Raft/group 0 related operations, including upgrade.
+    recovery = 0,
+
+    // In `use_pre_raft_procedures` state the node performs schema changes without using group 0,
+    // using the 'old ways': by updating its local schema tables and attempting to push the change to others
+    // through direct RPC. If necessary, changes are later propagated with gossip.
+    use_pre_raft_procedures = 1,
+
+    // In `synchronize` state the node rejects any attempts for changing the schema.
+    // Schema changes may still arrive from other nodes for some time. However, if no failures occur
+    // during the upgrade procedure, eventually all nodes should enter `synchronize` state. Then
+    // the nodes ensure that schema is synchronized across the entire cluster before entering `use_post_raft_procedures`.
+    synchronize = 2,
+
+    // In `use_post_raft_procedures` state the upgrade is finished. The node performs schema changes
+    // using group 0, i.e. by constructing appropriate Raft commands and sending them to the Raft group 0 cluster.
+    use_post_raft_procedures = 3,
+};
+
+inline constexpr uint8_t group0_upgrade_state_last = 3;
+
+std::ostream& operator<<(std::ostream&, group0_upgrade_state);
+
+} // namespace service

service/raft/raft_group0.cc

@@ -570,5 +570,24 @@ persistent_discovery::persistent_discovery(raft::server_address self, const peer
     }
 }
 
+std::ostream& operator<<(std::ostream& os, group0_upgrade_state state) {
+    switch (state) {
+        case group0_upgrade_state::recovery:
+            os << "recovery";
+            break;
+        case group0_upgrade_state::use_post_raft_procedures:
+            os << "use_post_raft_procedures";
+            break;
+        case group0_upgrade_state::synchronize:
+            os << "synchronize";
+            break;
+        case group0_upgrade_state::use_pre_raft_procedures:
+            os << "use_pre_raft_procedures";
+            break;
+    }
+
+    return os;
+}
+
 } // end of namespace service