diff --git a/service/vector_store_client.cc b/service/vector_store_client.cc index 1ad577d2df..14e5a5e35e 100644 --- a/service/vector_store_client.cc +++ b/service/vector_store_client.cc @@ -31,6 +31,8 @@ namespace { +using namespace std::chrono_literals; + using ann_error = service::vector_store_client::ann_error; using configuration_exception = exceptions::configuration_exception; using duration = lowres_clock::duration; @@ -46,6 +48,7 @@ using port_number = service::vector_store_client::port_number; using primary_key = service::vector_store_client::primary_key; using primary_keys = service::vector_store_client::primary_keys; using service_reply_format_error = service::vector_store_client::service_reply_format_error; +using tcp_keepalive_params = net::tcp_keepalive_params; using time_point = lowres_clock::time_point; // Wait time before retrying after an exception occurred @@ -211,6 +214,27 @@ auto read_ann_json(rjson::value const& json, schema_ptr const& schema) -> std::e return std::move(keys); } +class client_connection_factory : public http::experimental::connection_factory { + socket_address _addr; + +public: + explicit client_connection_factory(socket_address addr) + : _addr(addr) { + } + + future make([[maybe_unused]] abort_source* as) override { + auto socket = co_await seastar::connect(_addr, {}, transport::TCP); + socket.set_nodelay(true); + socket.set_keepalive_parameters(tcp_keepalive_params{ + .idle = 60s, + .interval = 60s, + .count = 10, + }); + socket.set_keepalive(true); + co_return socket; + } +}; + class http_client { host_port _host_port; @@ -222,7 +246,7 @@ public: http_client(host_port host_port_, inet_address addr) : _host_port(std::move(host_port_)) , _addr(std::move(addr)) - , impl(socket_address(addr, _host_port.port)) { + , impl(std::make_unique(socket_address(addr, _host_port.port))) { } bool connects_to(inet_address const& a, port_number p) const {