gossiper: Enable features and register enabler from outside

It's a bit hairy. The maybe_enable_features() is called from two places
-- the feature_enabler upon notifications from gossiper and directory by
gossiper from wait_for_gossip_to_settle().

The _latter_ is called only when the wait_for_gossip_to_settle() is
called for the first time because of the _gossip_settled checks in it.
For the first time this method is called by storage_service when it
tries to join the ring (next it's called from main, but that's not of
interest here).

Next, despite feature_enabler is registered early -- when gossiper
instance is constructed by sharded<gossiper>::start() -- it checks for
the _gossip_settled to be true to take any actions.

Considering both -- calling maybe_enable_features() _and_ registering
enabler after storage_service's call to wait_for_gossip_to_settle()
doesn't break the code logic, but make further patching possible. In
particular, the feature_enabler will move to feature_service not to
pollute gossiper code with anything that's not gossiping.

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>

gms/gossiper.cc

@@ -84,11 +84,11 @@ public:
         (void)_sys_ks;
     }
     future<> on_join(inet_address ep, endpoint_state state) override {
-        return _g.maybe_enable_features();
+        return _g.do_enable_features();
     }
     future<> on_change(inet_address ep, application_state state, const versioned_value&) override {
         if (state == application_state::SUPPORTED_FEATURES) {
-            return _g.maybe_enable_features();
+            return _g.do_enable_features();
         }
         return make_ready_future();
     }
@@ -116,7 +116,6 @@ gossiper::gossiper(abort_source& as, feature_service& features, const locator::s
     _scheduled_gossip_task.set_callback(_gcfg.gossip_scheduling_group, [this] { run(); });
     // half of QUARATINE_DELAY, to ensure _just_removed_endpoints has enough leeway to prevent re-gossip
     fat_client_timeout = quarantine_delay() / 2;
-    register_(make_shared<persistent_feature_enabler>(*this, _feature_service, _sys_ks.local()));
     // Register this instance with JMX
     namespace sm = seastar::metrics;
     auto ep = get_broadcast_address();
@@ -2381,9 +2380,6 @@ future<> gossiper::wait_for_gossip_to_settle() {
     if (force_after != 0) {
         co_await wait_for_gossip(GOSSIP_SETTLE_MIN_WAIT_MS, force_after);
     }
-    if (!std::exchange(_gossip_settled, true)) {
-        co_await maybe_enable_features();
-    }
 }
 
 future<> gossiper::wait_for_range_setup() {
@@ -2555,10 +2551,13 @@ void gossiper::append_endpoint_state(std::stringstream& ss, const endpoint_state
     }
 }
 
-future<> gossiper::maybe_enable_features() {
-    if (!_gossip_settled) {
-        co_return;
-    }
+future<> gossiper::enable_features() {
+    auto enabler = make_shared<persistent_feature_enabler>(*this, _feature_service, _sys_ks.local());
+    register_(enabler);
+    return do_enable_features();
+}
+
+future<> gossiper::do_enable_features() {
     auto loaded_peer_features = co_await db::system_keyspace::load_peer_features();
     auto&& features = get_supported_features(loaded_peer_features, ignore_features_of_local_node::no);
     co_await container().invoke_on_all([&features] (gossiper& g) -> future<> {

gms/gossiper.hh

@@ -587,7 +587,6 @@ private:
 
     uint64_t _nr_run = 0;
     uint64_t _msg_processing = 0;
-    bool _gossip_settled = false;
 
     class msg_proc_guard;
 private:
@@ -606,7 +605,8 @@ private:
     locator::token_metadata_ptr get_token_metadata_ptr() const noexcept;
 public:
     void check_knows_remote_features(std::set<std::string_view>& local_features, const std::unordered_map<inet_address, sstring>& loaded_peer_features) const;
-    future<> maybe_enable_features();
+    future<> do_enable_features();
+    future<> enable_features();
 private:
     seastar::metrics::metric_groups _metrics;
 public:

service/storage_service.cc

@@ -1435,6 +1435,7 @@ future<> storage_service::join_token_ring(cdc::generation_service& cdc_gen_servi
     });
     _listeners.emplace_back(make_lw_shared(std::move(schema_change_announce)));
     co_await _gossiper.wait_for_gossip_to_settle();
+    co_await _gossiper.enable_features();
 
     set_mode(mode::JOINING);