messaging_service: Do TLS init early

Fixes #14299

failure_detector can try sending messages to TLS endpoints before start_listen
has been called (why?). Need TLS initialized before this. So do on service creation.

Closes #14493

main.cc

@@ -1157,6 +1157,11 @@ To start the scylla server proper, simply invoke as: scylla server (or just scyl
                 messaging.invoke_on_all(&netw::messaging_service::stop).get();
             });
 
+            // #14299 - do early init of messaging_service (or rather its TLS structures)
+            // since other things (failure_detector) might try to send messages vie it
+            // before start_listen is called.
+            messaging.invoke_on_all(&netw::messaging_service::start).get();
+
             supervisor::notify("starting gossiper");
             gms::gossip_config gcfg;
             gcfg.gossip_scheduling_group = dbcfg.gossip_scheduling_group;

message/messaging_service.cc

@@ -249,8 +249,7 @@ rpc_resource_limits(size_t memory_limit) {
     return limits;
 }
 
-future<> messaging_service::start_listen(locator::shared_token_metadata& stm) {
-    _token_metadata = &stm;
+future<> messaging_service::start() {
     if (_credentials_builder && !_credentials) {
         return _credentials_builder->build_reloadable_server_credentials([](const std::unordered_set<sstring>& files, std::exception_ptr ep) {
             if (ep) {
@@ -260,9 +259,13 @@ future<> messaging_service::start_listen(locator::shared_token_metadata& stm) {
             }
         }).then([this](shared_ptr<seastar::tls::server_credentials> creds) {
             _credentials = std::move(creds);
-            do_start_listen();
         });
     }
+    return make_ready_future<>();
+}
+
+future<> messaging_service::start_listen(locator::shared_token_metadata& stm) {
+    _token_metadata = &stm;
     do_start_listen();
     return make_ready_future<>();
 }
@@ -866,6 +869,8 @@ shared_ptr<messaging_service::rpc_protocol_client_wrapper> messaging_service::ge
     opts.reuseaddr = true;
     opts.isolation_cookie = _scheduling_info_for_connection_index[idx].isolation_cookie;
 
+    assert(!must_encrypt || _credentials);
+
     auto client = must_encrypt ?
                     ::make_shared<rpc_protocol_client_wrapper>(_rpc->protocol(), std::move(opts),
                                     remote_addr, laddr, _credentials) :

message/messaging_service.hh

@@ -332,6 +332,7 @@ public:
     messaging_service(config cfg, scheduling_config scfg, std::shared_ptr<seastar::tls::credentials_builder>);
     ~messaging_service();
 
+    future<> start();
     future<> start_listen(locator::shared_token_metadata& stm);
     uint16_t port();
     gms::inet_address listen_address();