topology_state_machine: introduce global_topology_request

`topology` currently contains the `requests` map, which is suitable for
node-specific requests such as "this node wants to join" or "this node
must be removed". But for requests for operations that affect the
cluster as a whole, a separate request type and field is more
appropriate. Introduce one.

The enum currently contains the option `new_cdc_generation` for requests
to create a new CDC generation in the cluster. We will implement the
whole procedure in later commits.

db/system_keyspace.cc

@@ -260,6 +260,7 @@ schema_ptr system_keyspace::topology() {
             .with_column("transition_state", utf8_type, column_kind::static_column)
             .with_column("current_cdc_generation_uuid", uuid_type, column_kind::static_column)
             .with_column("current_cdc_generation_timestamp", timestamp_type, column_kind::static_column)
+            .with_column("global_topology_request", utf8_type, column_kind::static_column)
             .set_comment("Current state of topology change machine")
             .with_version(generate_schema_version(id))
             .build();
@@ -3693,6 +3694,12 @@ future<service::topology> system_keyspace::load_topology_state() {
                     "load_topology_state: normal nodes present but no current CDC generation ID");
             }
         }
+
+        if (some_row.has("global_topology_request")) {
+            auto req = service::global_topology_request_from_string(
+                    some_row.get_as<sstring>("global_topology_request"));
+            ret.global_request.emplace(req);
+        }
     }
 
     co_return ret;

service/topology_state_machine.cc

@@ -109,6 +109,28 @@ topology_request topology_request_from_string(const sstring& s) {
     throw std::runtime_error(fmt::format("cannot map name {} to topology_request", s));
 }
 
+static std::unordered_map<global_topology_request, sstring> global_topology_request_to_name_map = {
+    {global_topology_request::new_cdc_generation, "new_cdc_generation"},
+};
+
+std::ostream& operator<<(std::ostream& os, const global_topology_request& req) {
+    auto it = global_topology_request_to_name_map.find(req);
+    if (it == global_topology_request_to_name_map.end()) {
+        on_internal_error(tsmlogger, format("cannot print global topology request {}", static_cast<uint8_t>(req)));
+    }
+    return os << it->second;
+}
+
+global_topology_request global_topology_request_from_string(const sstring& s) {
+    for (auto&& e : global_topology_request_to_name_map) {
+        if (e.second == s) {
+            return e.first;
+        }
+    }
+
+    on_internal_error(tsmlogger, format("cannot map name {} to global_topology_request", s));
+}
+
 std::ostream& operator<<(std::ostream& os, const raft_topology_cmd::command& cmd) {
     switch (cmd) {
         case raft_topology_cmd::command::barrier:

service/topology_state_machine.hh

@@ -47,6 +47,10 @@ enum class topology_request: uint16_t {
 
 using request_param = std::variant<raft::server_id, sstring, uint32_t>;
 
+enum class global_topology_request: uint16_t {
+    new_cdc_generation,
+};
+
 struct ring_slice {
     std::unordered_set<dht::token> tokens;
 };
@@ -87,6 +91,9 @@ struct topology {
     // operation untill the node becomes normal
     std::unordered_map<raft::server_id, request_param> req_param;
 
+    // Pending global topology request (i.e. not related to any specific node).
+    std::optional<global_topology_request> global_request;
+
     // The ID of the last introduced CDC generation.
     std::optional<cdc::generation_id_v2> current_cdc_generation_id;
 
@@ -145,5 +152,7 @@ std::ostream& operator<<(std::ostream& os, node_state s);
 node_state node_state_from_string(const sstring& s);
 std::ostream& operator<<(std::ostream& os, const topology_request& req);
 topology_request topology_request_from_string(const sstring& s);
+std::ostream& operator<<(std::ostream&, const global_topology_request&);
+global_topology_request global_topology_request_from_string(const sstring&);
 std::ostream& operator<<(std::ostream& os, const raft_topology_cmd::command& cmd);
 }