Merge 'Do not yield while traversing the gossiper endpoint state map'…

… from Benny Halevy

This series introduces a new gossiper method: get_endpoints that returns a vector of endpoints (by value) based on the endpoint state map.

get_endpoints is used here by gossiper and storage_service for iterations that may preempt
instead of iterating direction over the endpoint state map (`_endpoint_state_map` in gossiper or via `get_endpoint_states()`) so to prevent use-after-free that may potentially happen if the map is rehashed while the function yields causing invalidation of the loop iterators.

Fixes #13899

Closes #13900

* github.com:scylladb/scylladb:
  storage_service: do not preempt while traversing endpoint_state_map
  gossiper: do not preempt while traversing endpoint_state_map

gms/gossiper.cc

@@ -702,21 +702,24 @@ future<> gossiper::do_status_check() {
 
     auto now = this->now();
 
-    for (auto it = _endpoint_state_map.begin(); it != _endpoint_state_map.end();) {
-        auto endpoint = it->first;
-        auto& ep_state = it->second;
-        it++;
-
-        bool is_alive = ep_state.is_alive();
+    for (const auto& endpoint : get_endpoints()) {
         if (endpoint == get_broadcast_address()) {
             continue;
         }
+        auto eps = get_endpoint_state_for_endpoint_ptr(endpoint);
+        if (!eps) {
+            continue;
+        }
+        auto& ep_state = *eps;
+        bool is_alive = ep_state.is_alive();
+        auto update_timestamp = ep_state.get_update_timestamp();
+        // ep_state cannot be used after yielding
 
         // check if this is a fat client. fat clients are removed automatically from
         // gossip after FatClientTimeout.  Do not remove dead states here.
         if (is_gossip_only_member(endpoint)
             && !_just_removed_endpoints.contains(endpoint)
-            && ((now - ep_state.get_update_timestamp()) > fat_client_timeout)) {
+            && ((now - update_timestamp) > fat_client_timeout)) {
             logger.info("FatClient {} has been silent for {}ms, removing from gossip", endpoint, fat_client_timeout.count());
             co_await remove_endpoint(endpoint); // will put it in _just_removed_endpoints to respect quarantine delay
             co_await evict_from_membership(endpoint); // can get rid of the state immediately
@@ -1372,6 +1375,10 @@ const std::unordered_map<inet_address, endpoint_state>& gms::gossiper::get_endpo
     return _endpoint_state_map;
 }
 
+std::vector<inet_address> gossiper::get_endpoints() const {
+    return boost::copy_range<std::vector<inet_address>>(_endpoint_state_map | boost::adaptors::map_keys);
+}
+
 bool gossiper::uses_host_id(inet_address endpoint) const {
     return _messaging.knows_version(endpoint) ||
             get_application_state_ptr(endpoint, application_state::NET_VERSION);

gms/gossiper.hh

@@ -399,6 +399,8 @@ public:
 
     const std::unordered_map<inet_address, endpoint_state>& get_endpoint_states() const noexcept;
 
+    std::vector<inet_address> get_endpoints() const;
+
     bool uses_host_id(inet_address endpoint) const;
 
     locator::host_id get_host_id(inet_address endpoint) const;

service/storage_service.cc

@@ -1817,7 +1817,7 @@ std::unordered_set<gms::inet_address> storage_service::parse_node_list(sstring c
 future<std::unordered_set<gms::inet_address>> storage_service::get_nodes_to_sync_with(
         const std::unordered_set<gms::inet_address>& ignore_nodes) {
     std::unordered_set<gms::inet_address> result;
-    for (const auto& [node, _] :_gossiper.get_endpoint_states()) {
+    for (const auto& node :_gossiper.get_endpoints()) {
         co_await coroutine::maybe_yield();
         slogger.info("Check node={}, status={}", node, _gossiper.get_gossip_status(node));
         if (node != get_broadcast_address() &&
@@ -2812,12 +2812,11 @@ future<> storage_service::check_for_endpoint_collision(std::unordered_set<gms::i
                 // Raft is responsible for consistency, so in case it is enable no need to check here
                 !_raft_topology_change_enabled) {
                 found_bootstrapping_node = false;
-                for (auto& x : _gossiper.get_endpoint_states()) {
-                    auto state = _gossiper.get_gossip_status(x.second);
+                for (const auto& addr : _gossiper.get_endpoints()) {
+                    auto state = _gossiper.get_gossip_status(addr);
                     if (state == sstring(versioned_value::STATUS_UNKNOWN)) {
-                        throw std::runtime_error(::format("Node {} has gossip status=UNKNOWN. Try fixing it before adding new node to the cluster.", x.first));
+                        throw std::runtime_error(::format("Node {} has gossip status=UNKNOWN. Try fixing it before adding new node to the cluster.", addr));
                     }
-                    auto addr = x.first;
                     slogger.debug("Checking bootstrapping/leaving/moving nodes: node={}, status={} (check_for_endpoint_collision)", addr, state);
                     if (state == sstring(versioned_value::STATUS_BOOTSTRAPPING) ||
                         state == sstring(versioned_value::STATUS_LEAVING) ||