diff --git a/clickhouse/base/endpoints_iterator.cpp b/clickhouse/base/endpoints_iterator.cpp index 30d3593e..6680225e 100644 --- a/clickhouse/base/endpoints_iterator.cpp +++ b/clickhouse/base/endpoints_iterator.cpp @@ -3,8 +3,21 @@ namespace clickhouse { +namespace { + +const std::vector & ValidateEndpoints(const std::vector& endpoints) +{ + if (endpoints.empty()) { + throw ValidationError("The list of endpoints is empty"); + } + return endpoints; +} + +} // anonymous namespace + RoundRobinEndpointsIterator::RoundRobinEndpointsIterator(const std::vector& _endpoints) - : endpoints (_endpoints) + : endpoints (ValidateEndpoints(_endpoints)) + // set `current_index` to the value such that `Next` returns an element at index 0 , current_index (endpoints.size() - 1ull) { } diff --git a/clickhouse/client.cpp b/clickhouse/client.cpp index 0a8950c7..05f03bcd 100644 --- a/clickhouse/client.cpp +++ b/clickhouse/client.cpp @@ -179,15 +179,6 @@ std::unique_ptr GetSocketFactory(const ClientOptions& opts) { return std::make_unique(); } -std::unique_ptr GetEndpointsIterator(const ClientOptions& opts) { - if (opts.endpoints.empty()) - { - throw ValidationError("The list of endpoints is empty"); - } - - return std::make_unique(opts.endpoints); -} - } // anonymous namespace class Client::Impl { @@ -264,18 +255,11 @@ class Client::Impl { void InitializeStreams(std::unique_ptr&& socket); - inline size_t GetConnectionAttempts() const - { - return options_.endpoints.size() * options_.send_retries; - } - private: /// In case of network errors tries to reconnect to server and /// call fuc several times. void RetryGuard(std::function func); - void RetryConnectToTheEndpoint(std::function& func); - private: enum class State : uint8_t { Idle = 0, @@ -317,6 +301,8 @@ class Client::Impl { std::unique_ptr socket_; std::unique_ptr endpoints_iterator; + // current_endpoint_ points to the last successfully connected endpoint, and always + // holds a value. The variable remains wrapped as optional for backwards compatibility. std::optional current_endpoint_; ServerInfo server_info_; @@ -342,7 +328,8 @@ Client::Impl::Impl(const ClientOptions& opts, : options_(modifyClientOptions(opts)) , events_(nullptr) , socket_factory_(std::move(socket_factory)) - , endpoints_iterator(GetEndpointsIterator(options_)) + , endpoints_iterator(std::make_unique(options_.endpoints)) + , current_endpoint_(endpoints_iterator->Next()) { CreateConnection(); @@ -619,20 +606,23 @@ void Client::Impl::ResetConnection() { } void Client::Impl::ResetConnectionEndpoint() { - current_endpoint_.reset(); - for (size_t i = 0; i < options_.endpoints.size();) + std::optional last_endpoint = current_endpoint_; + for (size_t i = 1; ; ++i) { try { - current_endpoint_ = endpoints_iterator->Next(); ResetConnection(); return; } catch (const std::system_error&) { - if (++i == options_.endpoints.size()) + current_endpoint_ = endpoints_iterator->Next(); + if (i >= options_.endpoints.size()) { - current_endpoint_.reset(); + current_endpoint_ = last_endpoint; throw; } + } catch (...) { + current_endpoint_ = last_endpoint; + throw; } } } @@ -640,7 +630,7 @@ void Client::Impl::ResetConnectionEndpoint() { void Client::Impl::CreateConnection() { // make sure to try to connect to each endpoint at least once even if `options_.send_retries` is 0 const size_t max_attempts = (options_.send_retries ? options_.send_retries : 1); - for (size_t i = 0; i < max_attempts;) + for (size_t i = 1; ; ++i) { try { @@ -648,7 +638,7 @@ void Client::Impl::CreateConnection() { ResetConnectionEndpoint(); return; } catch (const std::system_error&) { - if (++i >= max_attempts) + if (i >= max_attempts) { throw; } @@ -1232,32 +1222,36 @@ bool Client::Impl::ReceiveHello() { void Client::Impl::RetryGuard(std::function func) { - if (current_endpoint_) - { - for (unsigned int i = 0; ; ++i) { - try { - func(); - return; - } catch (const std::system_error&) { - bool ok = true; + for (unsigned int i = 1; ; ++i) { + try { + func(); + return; + } catch (const std::system_error&) { + // if send_retries == 0 do not try anymore, throw right away + if (options_.send_retries == 0) { + throw; + } - try { - socket_factory_->sleepFor(options_.retry_timeout); - ResetConnection(); - } catch (...) { - ok = false; - } + // If `send_retries` attempts failed, try other endpoints + if (i >= options_.send_retries) { + break; + } - if (!ok && i == options_.send_retries) { - break; - } + // otherwise sleep and try again + try { + socket_factory_->sleepFor(options_.retry_timeout); + ResetConnection(); + } catch (const std::system_error&) { } + } } + // Connections with current_endpoint_ are broken. - // Trying to establish with the another one from the list. - size_t connection_attempts_count = GetConnectionAttempts(); - for (size_t i = 0; i < connection_attempts_count;) + // Trying to establish with another one from the list. + size_t connection_attempts_count = options_.endpoints.size() * options_.send_retries; + std::optional last_endpoint = current_endpoint_; + for (size_t i = 1; ; ++i) { try { @@ -1267,11 +1261,14 @@ void Client::Impl::RetryGuard(std::function func) { func(); return; } catch (const std::system_error&) { - if (++i == connection_attempts_count) + if (i >= connection_attempts_count) { - current_endpoint_.reset(); + current_endpoint_ = last_endpoint; throw; } + } catch (...) { + current_endpoint_ = last_endpoint; + throw; } } } diff --git a/clickhouse/client.h b/clickhouse/client.h index 78968497..5e13d269 100644 --- a/clickhouse/client.h +++ b/clickhouse/client.h @@ -338,11 +338,12 @@ class Client { const ServerInfo& GetServerInfo() const; - /// Get current connected endpoint. - /// In case when client is not connected to any endpoint, nullopt will returned. + /// Get current endpoint, i.e. the last successfully connected endpoint. + /// It remains optional for backward compatibility, but now always contains a value. const std::optional& GetCurrentEndpoint() const; - // Try to connect to different endpoints one by one only one time. If it doesn't work, throw an exception. + /// Try to reconnect to different endpoints one by one only one time. If it doesn't work, throw + /// an exception. The function starts with the last successfully connected endpoint. void ResetConnectionEndpoint(); struct Version diff --git a/ut/BUILD.bazel b/ut/BUILD.bazel index 6845fbb5..a41e3e54 100644 --- a/ut/BUILD.bazel +++ b/ut/BUILD.bazel @@ -110,6 +110,7 @@ cc_test( "readonly_client_test.cpp", "readonly_client_test.h", "roundtrip_tests.cpp", + "test_socket_factory_adapters.h", # Test entry point and shared support code. "main.cpp", "roundtrip_column.cpp", diff --git a/ut/client_ut.cpp b/ut/client_ut.cpp index 472087c0..0e5a020b 100644 --- a/ut/client_ut.cpp +++ b/ut/client_ut.cpp @@ -8,6 +8,7 @@ #include "readonly_client_test.h" #include "connection_failed_client_test.h" +#include "test_socket_factory_adapters.h" #include "ut/utils_comparison.h" #include "utils.h" #include "ut/roundtrip_column.h" @@ -1849,31 +1850,44 @@ INSTANTIATE_TEST_SUITE_P(MultipleEndpointsFailed, ConnectionFailedClientTest, class ResetConnectionTestCase : public testing::TestWithParam {}; -TEST_P(ResetConnectionTestCase, ResetConnectionEndpointTest) { - const auto & client_options = GetParam(); - std::unique_ptr client; +TEST(ResetConnectionEndpointTest, ReconnectsCurrentBeforeFailover) { + const Endpoint primary{"primary", 9000}; + const Endpoint secondary{"secondary", 9000}; + const Endpoint actual_endpoint{LocalHostEndpoint.host, LocalHostEndpoint.port}; - try { - client = std::make_unique(client_options); - auto endpoint = client->GetCurrentEndpoint().value(); - ASSERT_EQ("localhost", endpoint.host); - ASSERT_EQ(9000u, endpoint.port); + ClientOptions options(LocalHostEndpoint); + options.SetHost(""); + options.SetEndpoints({primary, secondary}); - client->ResetConnectionEndpoint(); - endpoint = client->GetCurrentEndpoint().value(); - ASSERT_EQ("127.0.0.1", endpoint.host); - ASSERT_EQ(9000u, endpoint.port); + // Redirect both logical endpoints to the same reachable test server. + auto base_socket_factory = std::make_unique(); + auto socket_factory = std::make_unique(*base_socket_factory, actual_endpoint); + auto * const adapter = socket_factory.get(); - client->ResetConnectionEndpoint(); + // The initial connection selects the first endpoint. + Client client(options, std::move(socket_factory)); + ASSERT_EQ(primary, client.GetCurrentEndpoint().value()); - endpoint = client->GetCurrentEndpoint().value(); - ASSERT_EQ("localhost", endpoint.host); - ASSERT_EQ(9000u, endpoint.port); + // A healthy current endpoint is retried without advancing. + adapter->SetFailEndpoint(std::nullopt); + adapter->ClearConnectRequests(); + client.ResetConnectionEndpoint(); + EXPECT_EQ(primary, client.GetCurrentEndpoint().value()); + EXPECT_EQ(std::vector{primary}, adapter->ConnectRequests()); - SUCCEED(); - } catch (const std::exception & e) { - FAIL() << "Got an unexpected exception : " << e.what(); - } + // Failure of the current endpoint advances to the next endpoint. + adapter->SetFailEndpoint(primary); + adapter->ClearConnectRequests(); + client.ResetConnectionEndpoint(); + EXPECT_EQ(secondary, client.GetCurrentEndpoint().value()); + EXPECT_EQ((std::vector{primary, secondary}), adapter->ConnectRequests()); + + // Failure of the last endpoint wraps around to the first endpoint. + adapter->SetFailEndpoint(secondary); + adapter->ClearConnectRequests(); + client.ResetConnectionEndpoint(); + EXPECT_EQ(primary, client.GetCurrentEndpoint().value()); + EXPECT_EQ((std::vector{secondary, primary}), adapter->ConnectRequests()); } TEST_P(ResetConnectionTestCase, ResetConnectionTest) { diff --git a/ut/test_socket_factory_adapters.h b/ut/test_socket_factory_adapters.h new file mode 100644 index 00000000..19624bbc --- /dev/null +++ b/ut/test_socket_factory_adapters.h @@ -0,0 +1,62 @@ +#pragma once + +#include "clickhouse/base/socket.h" + +#include +#include +#include +#include +#include +#include + +namespace clickhouse { + +/** Records requested endpoints and optionally fails one matching connection attempt. + * + * Successful attempts are redirected to actual_endpoint, allowing tests to exercise + * failover between distinct logical endpoints using a single reachable server. Setting + * fail_endpoint makes the next matching attempt throw connection_refused and then + * clears the value. The wrapped factory must outlive the adapter. + */ +struct FailOnceSocketFactoryAdapter : public SocketFactory { + SocketFactory & socket_factory; + Endpoint actual_endpoint; + std::vector connect_requests{}; + std::optional fail_endpoint{}; + + FailOnceSocketFactoryAdapter(SocketFactory & socket_factory, + Endpoint actual_endpoint) + : socket_factory(socket_factory) + , actual_endpoint(std::move(actual_endpoint)) + {} + + std::unique_ptr connect(const ClientOptions& opts, + const Endpoint& endpoint) override { + connect_requests.push_back(endpoint); + + if (fail_endpoint && fail_endpoint.value() == endpoint) { + fail_endpoint.reset(); + throw std::system_error(std::make_error_code(std::errc::connection_refused)); + } + + return socket_factory.connect(opts, actual_endpoint); + } + + void SetFailEndpoint(std::optional endpoint) { + fail_endpoint = std::move(endpoint); + } + + const std::vector & ConnectRequests() const { + return connect_requests; + } + + void ClearConnectRequests() { + connect_requests.clear(); + } + + void sleepFor(const std::chrono::milliseconds& duration) override { + socket_factory.sleepFor(duration); + } +}; + +} // namespace clickhouse