Merge pull request #20589 from WillemKauf/tls_client_shutdown_23_3

[CORE-4878] [v23.3.x] http: change `stop()` to `shutdown()` on TLS errors in `client` (manual backport)

src/v/http/client.cc

@@ -14,6 +14,7 @@
 #include "bytes/scattered_message.h"
 #include "config/base_property.h"
 #include "http/logger.h"
+#include "likely.h"
 #include "ssx/sformat.h"
 #include "vlog.h"
 
@@ -82,6 +83,11 @@ void client::check() const {
 
 ss::future<client::request_response_t> client::make_request(
   client::request_header&& header, ss::lowres_clock::duration timeout) {
+    if (unlikely(_stopped)) {
+        std::runtime_error err("client is stopped");
+        return ss::make_exception_future<client::request_response_t>(err);
+    }
+
     auto verb = header.method();
     auto target = header.target();
     ss::sstring target_str(target.data(), target.size());
@@ -122,19 +128,24 @@ ss::future<client::request_response_t> client::make_request(
           return ss::make_ready_future<request_response_t>(
             std::make_tuple(req, res));
       })
-      .handle_exception_type([this](ss::tls::verification_error err) {
-          return stop().then([err = std::move(err)] {
-              return ss::make_exception_future<client::request_response_t>(err);
-          });
+      .handle_exception_type([this, ctxlog](ss::tls::verification_error err) {
+          vlog(ctxlog.warn, "make_request tls verification error {}", err);
+          shutdown();
+          return ss::make_exception_future<client::request_response_t>(err);
       });
 }
 
 ss::future<reconnect_result_t> client::get_connected(
   ss::lowres_clock::duration timeout, prefix_logger ctxlog) {
+    if (unlikely(_stopped)) {
+        throw std::runtime_error("client is stopped");
+    }
     vlog(
       ctxlog.debug,
-      "about to start connecting, {}, is-closed {}",
+      "about to start connecting, is_valid: {}, connect gate closed: {}, "
+      "dispatch gate closed: {}",
       is_valid(),
+      _connect_gate.is_closed(),
       _dispatch_gate.is_closed());
     auto current = ss::lowres_clock::now();
     const auto deadline = current + timeout;
@@ -369,8 +380,9 @@ ss::future<iobuf> client::response_stream::recv_some() {
       })
       .handle_exception_type([this](const ss::tls::verification_error& err) {
           _client->_probe->register_transport_error();
-          return _client->stop().then(
-            [err] { return ss::make_exception_future<iobuf>(err); });
+          vlog(_ctxlog.warn, "receive tls verification error {}", err);
+          _client->shutdown();
+          return ss::make_exception_future<iobuf>(err);
       })
       .handle_exception_type([this](const boost::system::system_error& ec) {
           vlog(_ctxlog.warn, "receive error {}", ec);
@@ -466,8 +478,9 @@ ss::future<> client::request_stream::send_some(iobuf&& seq) {
             })
             .handle_exception_type(
               [this](const ss::tls::verification_error& err) {
-                  return _client->stop().then(
-                    [err] { return ss::make_exception_future<>(err); });
+                  vlog(_ctxlog.warn, "send tls verification error {}", err);
+                  _client->shutdown();
+                  return ss::make_exception_future<>(err);
               })
             .handle_exception_type([this](const std::system_error& ec) {
                 // Things like EPIPE, ERESET.  This happens routinely