From bb20ee90697262584b593cb180d617cf729f9cca Mon Sep 17 00:00:00 2001 From: Andrew Slabko Date: Wed, 2 Sep 2026 12:19:57 +0200 Subject: [PATCH 1/3] Throw ProtocolError on packet decoding failures Previously, when a packet payload could not be decoded (e.g. the connection dropped mid-response), ReceivePacket() returned an empty std::monostate variant that callers treated as a clean end of stream: NextBlock() returned std::nullopt as if the query had completed successfully, silently losing data. Now every decoding failure throws ProtocolError with a message identifying the packet that failed to decode, and std::monostate is removed from the DecodedPacket variant. Behavior changes: - A connection broken mid-query now surfaces as ProtocolError instead of a seemingly successful end of stream. This applies regardless of ClientOptions::SetRethrowException(false), which only concerns server-reported exceptions. - QueryEvents::OnServerException is no longer invoked with a partially populated exception when decoding the exception packet fails. - If the server rejects the connection but its exception packet cannot be decoded, the handshake now fails with a descriptive ProtocolError. --- clickhouse/client.cpp | 120 +++++++++++++++++++----------------------- 1 file changed, 54 insertions(+), 66 deletions(-) diff --git a/clickhouse/client.cpp b/clickhouse/client.cpp index 05f03bcd..07e90a7f 100644 --- a/clickhouse/client.cpp +++ b/clickhouse/client.cpp @@ -9,6 +9,7 @@ #include "columns/factory.h" #include +#include #include #include #include @@ -157,7 +158,6 @@ struct ProfileEvents { struct EndOfStream { }; using DecodedPacket = std::variant< - std::monostate, Block, ServerException, Profile, @@ -386,7 +386,7 @@ std::optional Client::Impl::NextBlock() { return {std::move(block)}; } case VariantIndex(): - case VariantIndex(): + // See note in ReceivePacket() for the ServerCodes::Exception case case VariantIndex(): ResetState(); return std::nullopt; @@ -674,7 +674,7 @@ DecodedPacket Client::Impl::ReceivePacket(uint64_t* server_packet) { uint64_t packet_type = 0; if (!WireFormat::ReadVarint64(*input_, &packet_type)) { - return {}; + throw ProtocolError{"can't read packet type from input stream"}; } if (server_packet) { *server_packet = packet_type; @@ -684,15 +684,19 @@ DecodedPacket Client::Impl::ReceivePacket(uint64_t* server_packet) { case ServerCodes::Data: { Block ret{}; if (!ReceiveData(ret)) { - throw ProtocolError("can't read data packet from input stream"); + throw ProtocolError{"can't read data packet from input stream"}; } return ret; } case ServerCodes::Exception: { + // ReceiveException throws ServerExceptions when it succeeds reading the exception + // information from the server. So the execution does not usually reach this state. + // However, the user can suppress exceptions with + // `ClientOptions::SetRethrowException(false)` (the default is true). ServerError ret{std::make_shared()}; if (!ReceiveException(false, &ret)) { - throw ProtocolError("can't read exception packet from input stream"); + throw ProtocolError{"server reported an error, but the exception packet could not be decoded (error details lost)"}; } return ret; } @@ -700,23 +704,13 @@ DecodedPacket Client::Impl::ReceivePacket(uint64_t* server_packet) { case ServerCodes::ProfileInfo: { Profile ret{}; - if (!WireFormat::ReadUInt64(*input_, &ret.rows)) { - return {}; - } - if (!WireFormat::ReadUInt64(*input_, &ret.blocks)) { - return {}; - } - if (!WireFormat::ReadUInt64(*input_, &ret.bytes)) { - return {}; - } - if (!WireFormat::ReadFixed(*input_, &ret.applied_limit)) { - return {}; - } - if (!WireFormat::ReadUInt64(*input_, &ret.rows_before_limit)) { - return {}; - } - if (!WireFormat::ReadFixed(*input_, &ret.calculated_rows_before_limit)) { - return {}; + if (!WireFormat::ReadUInt64(*input_, &ret.rows) || + !WireFormat::ReadUInt64(*input_, &ret.blocks) || + !WireFormat::ReadUInt64(*input_, &ret.bytes) || + !WireFormat::ReadFixed(*input_, &ret.applied_limit) || + !WireFormat::ReadUInt64(*input_, &ret.rows_before_limit) || + !WireFormat::ReadFixed(*input_, &ret.calculated_rows_before_limit)) { + throw ProtocolError{"can't read profile info packet from input stream"}; } if (events_) { @@ -729,24 +723,21 @@ DecodedPacket Client::Impl::ReceivePacket(uint64_t* server_packet) { case ServerCodes::Progress: { Progress ret{}; - if (!WireFormat::ReadUInt64(*input_, &ret.rows)) { - return {}; - } - if (!WireFormat::ReadUInt64(*input_, &ret.bytes)) { - return {}; + if (!WireFormat::ReadUInt64(*input_, &ret.rows) || + !WireFormat::ReadUInt64(*input_, &ret.bytes)) { + throw ProtocolError{"can't read progress packet from input stream"}; } + if constexpr(DMBS_PROTOCOL_REVISION >= DBMS_MIN_REVISION_WITH_TOTAL_ROWS_IN_PROGRESS) { if (!WireFormat::ReadUInt64(*input_, &ret.total_rows)) { - return {}; + throw ProtocolError{"can't read progress packet from input stream"}; } } if (server_info_.revision >= DBMS_MIN_REVISION_WITH_CLIENT_WRITE_INFO) { - if (!WireFormat::ReadUInt64(*input_, &ret.written_rows)) { - return {}; - } - if (!WireFormat::ReadUInt64(*input_, &ret.written_bytes)) { - return {}; + if (!WireFormat::ReadUInt64(*input_, &ret.written_rows) || + !WireFormat::ReadUInt64(*input_, &ret.written_bytes)) { + throw ProtocolError{"can't read progress packet from input stream"}; } } @@ -774,14 +765,11 @@ DecodedPacket Client::Impl::ReceivePacket(uint64_t* server_packet) { case ServerCodes::Log: { // log tag - if (!WireFormat::SkipString(*input_)) { - return {}; - } Log ret; - - // Use uncompressed stream since log blocks usually contain only one row - if (!ReadBlock(*input_, &ret.block)) { - return {}; + if (!WireFormat::SkipString(*input_) || + // Use uncompressed stream since log blocks usually contain only one row + !ReadBlock(*input_, &ret.block)) { + throw ProtocolError{"can't read log packet from input stream"}; } if (events_) { @@ -792,25 +780,19 @@ DecodedPacket Client::Impl::ReceivePacket(uint64_t* server_packet) { case ServerCodes::TableColumns: { // external table name - if (!WireFormat::SkipString(*input_)) { - return {}; - } - - // columns metadata - if (!WireFormat::SkipString(*input_)) { - return {}; + if (!WireFormat::SkipString(*input_) || + // columns metadata + !WireFormat::SkipString(*input_)) { + throw ProtocolError{"can't read table columns packet from input stream"}; } return TableColumns{}; } case ServerCodes::ProfileEvents: { - if (!WireFormat::SkipString(*input_)) { - return {}; - } - ProfileEvents ret; - if (!ReadBlock(*input_, &ret.block)) { - return {}; + if (!WireFormat::SkipString(*input_) || + !ReadBlock(*input_, &ret.block)) { + throw ProtocolError{"can't read profile events packet from input stream"}; } if (events_) { @@ -829,7 +811,7 @@ bool Client::Impl::ProcessPacket(uint64_t* server_packet) { auto packet = ReceivePacket(server_packet); switch (packet.index()) { case VariantIndex(): - case VariantIndex(): + // See note in ReceivePacket() for the ServerCodes::Exception case case VariantIndex(): return false; default: @@ -949,25 +931,29 @@ bool Client::Impl::ReceiveData(Block & block) { bool Client::Impl::ReceiveException(bool rethrow, ServerError * error) { std::shared_ptr e(new Exception); bool has_nested = false; // obsolete: https://github.com/ClickHouse/ClickHouse/blob/ef11941cf5a/src/IO/ReadHelpers.cpp#L2017 + bool exception_received = false; - bool exception_received = - WireFormat::ReadFixed(*input_, &e->code) + if (WireFormat::ReadFixed(*input_, &e->code) && WireFormat::ReadString(*input_, &e->name) && WireFormat::ReadString(*input_, &e->display_text) && WireFormat::ReadString(*input_, &e->stack_trace) - && WireFormat::ReadFixed(*input_, &has_nested); + && WireFormat::ReadFixed(*input_, &has_nested)) { - if (events_) { - events_->OnServerException(*e); - } + if (events_) { + events_->OnServerException(*e); + } - if (rethrow || options_.rethrow_exceptions) { - throw ServerError(e); - } + if (rethrow || options_.rethrow_exceptions) { + throw ServerError(e); + } + + exception_received = true; - if (exception_received && error != nullptr) { - *error = ServerError(e); + if (error != nullptr) { + *error = ServerError(e); + } } + return exception_received; } @@ -1213,7 +1199,9 @@ bool Client::Impl::ReceiveHello() { return true; } else if (packet_type == ServerCodes::Exception) { - ReceiveException(true); + if (!ReceiveException(true)) { + throw ProtocolError{"server rejected the connection, but its exception packet could not be decoded"}; + } return false; } From d41de8b2985c920f6a04a36db2a30a77b792171d Mon Sep 17 00:00:00 2001 From: Andrew Slabko Date: Wed, 2 Sep 2026 11:25:26 +0200 Subject: [PATCH 2/3] Extract per-packet decoding into Receive* helpers Move packet payload decoding out of ReceivePacket() into dedicated ReceiveProfileInfo/ReceiveProgress/ReceiveLog/ReceiveTableColumns/ ReceiveProfileEvents helpers. ReceiveException no longer throws; the throw decision (gated by ClientOptions::SetRethrowException) now lives at the call sites in ReceivePacket() and ReceiveHello(). --- clickhouse/client.cpp | 191 +++++++++++++++++++++++++----------------- 1 file changed, 112 insertions(+), 79 deletions(-) diff --git a/clickhouse/client.cpp b/clickhouse/client.cpp index 07e90a7f..1bba7a9f 100644 --- a/clickhouse/client.cpp +++ b/clickhouse/client.cpp @@ -247,7 +247,17 @@ class Client::Impl { bool ReceiveData(Block & block); /// Reads exception packet form input stream. - bool ReceiveException(bool rethrow = false, ServerError * error = nullptr); + bool ReceiveException(ServerError & error); + + bool ReceiveProfileInfo(Profile & profile); + + bool ReceiveProgress(Progress & progress); + + bool ReceiveLog(Block & block); + + bool ReceiveTableColumns(); + + bool ReceiveProfileEvents(Block & block); void WriteBlock(const Block& block, OutputStream& output); @@ -386,7 +396,9 @@ std::optional Client::Impl::NextBlock() { return {std::move(block)}; } case VariantIndex(): - // See note in ReceivePacket() for the ServerCodes::Exception case + // Normally `ReceivePacket()` throws on server exceptions, but it can be + // suppressed by `ClientOptions::SetRethrowException`. In that case the + // error marks the end of the response. case VariantIndex(): ResetState(); return std::nullopt; @@ -690,61 +702,31 @@ DecodedPacket Client::Impl::ReceivePacket(uint64_t* server_packet) { } case ServerCodes::Exception: { - // ReceiveException throws ServerExceptions when it succeeds reading the exception - // information from the server. So the execution does not usually reach this state. - // However, the user can suppress exceptions with - // `ClientOptions::SetRethrowException(false)` (the default is true). ServerError ret{std::make_shared()}; - if (!ReceiveException(false, &ret)) { + if (!ReceiveException(ret)) { throw ProtocolError{"server reported an error, but the exception packet could not be decoded (error details lost)"}; } + + if (options_.rethrow_exceptions) { + throw ret; + } + return ret; } case ServerCodes::ProfileInfo: { Profile ret{}; - - if (!WireFormat::ReadUInt64(*input_, &ret.rows) || - !WireFormat::ReadUInt64(*input_, &ret.blocks) || - !WireFormat::ReadUInt64(*input_, &ret.bytes) || - !WireFormat::ReadFixed(*input_, &ret.applied_limit) || - !WireFormat::ReadUInt64(*input_, &ret.rows_before_limit) || - !WireFormat::ReadFixed(*input_, &ret.calculated_rows_before_limit)) { + if (!ReceiveProfileInfo(ret)) { throw ProtocolError{"can't read profile info packet from input stream"}; } - - if (events_) { - events_->OnProfile(ret); - } - return ret; } case ServerCodes::Progress: { Progress ret{}; - - if (!WireFormat::ReadUInt64(*input_, &ret.rows) || - !WireFormat::ReadUInt64(*input_, &ret.bytes)) { + if (!ReceiveProgress(ret)) { throw ProtocolError{"can't read progress packet from input stream"}; } - - if constexpr(DMBS_PROTOCOL_REVISION >= DBMS_MIN_REVISION_WITH_TOTAL_ROWS_IN_PROGRESS) { - if (!WireFormat::ReadUInt64(*input_, &ret.total_rows)) { - throw ProtocolError{"can't read progress packet from input stream"}; - } - } - if (server_info_.revision >= DBMS_MIN_REVISION_WITH_CLIENT_WRITE_INFO) - { - if (!WireFormat::ReadUInt64(*input_, &ret.written_rows) || - !WireFormat::ReadUInt64(*input_, &ret.written_bytes)) { - throw ProtocolError{"can't read progress packet from input stream"}; - } - } - - if (events_) { - events_->OnProgress(ret); - } - return ret; } @@ -764,25 +746,15 @@ DecodedPacket Client::Impl::ReceivePacket(uint64_t* server_packet) { } case ServerCodes::Log: { - // log tag Log ret; - if (!WireFormat::SkipString(*input_) || - // Use uncompressed stream since log blocks usually contain only one row - !ReadBlock(*input_, &ret.block)) { + if (!ReceiveLog(ret.block)) { throw ProtocolError{"can't read log packet from input stream"}; } - - if (events_) { - events_->OnServerLog(ret.block); - } return ret; } case ServerCodes::TableColumns: { - // external table name - if (!WireFormat::SkipString(*input_) || - // columns metadata - !WireFormat::SkipString(*input_)) { + if (!ReceiveTableColumns()) { throw ProtocolError{"can't read table columns packet from input stream"}; } return TableColumns{}; @@ -790,20 +762,14 @@ DecodedPacket Client::Impl::ReceivePacket(uint64_t* server_packet) { case ServerCodes::ProfileEvents: { ProfileEvents ret; - if (!WireFormat::SkipString(*input_) || - !ReadBlock(*input_, &ret.block)) { + if (!ReceiveProfileEvents(ret.block)) { throw ProtocolError{"can't read profile events packet from input stream"}; } - - if (events_) { - events_->OnProfileEvents(ret.block); - } return ret; } default: throw UnimplementedError("unimplemented " + std::to_string((int)packet_type)); - break; } } @@ -811,7 +777,9 @@ bool Client::Impl::ProcessPacket(uint64_t* server_packet) { auto packet = ReceivePacket(server_packet); switch (packet.index()) { case VariantIndex(): - // See note in ReceivePacket() for the ServerCodes::Exception case + // Normally `ReceivePacket()` throws on server exceptions, but it can be + // suppressed by `ClientOptions::SetRethrowException`. In that case the + // error marks the end of the response. case VariantIndex(): return false; default: @@ -928,33 +896,97 @@ bool Client::Impl::ReceiveData(Block & block) { return true; } -bool Client::Impl::ReceiveException(bool rethrow, ServerError * error) { +bool Client::Impl::ReceiveException(ServerError & error) { std::shared_ptr e(new Exception); bool has_nested = false; // obsolete: https://github.com/ClickHouse/ClickHouse/blob/ef11941cf5a/src/IO/ReadHelpers.cpp#L2017 - bool exception_received = false; - if (WireFormat::ReadFixed(*input_, &e->code) - && WireFormat::ReadString(*input_, &e->name) - && WireFormat::ReadString(*input_, &e->display_text) - && WireFormat::ReadString(*input_, &e->stack_trace) - && WireFormat::ReadFixed(*input_, &has_nested)) { + if (!WireFormat::ReadFixed(*input_, &e->code) || + !WireFormat::ReadString(*input_, &e->name) || + !WireFormat::ReadString(*input_, &e->display_text) || + !WireFormat::ReadString(*input_, &e->stack_trace) || + !WireFormat::ReadFixed(*input_, &has_nested)) { + return false; + } - if (events_) { - events_->OnServerException(*e); - } + if (events_) { + events_->OnServerException(*e); + } - if (rethrow || options_.rethrow_exceptions) { - throw ServerError(e); - } + error = ServerError(e); - exception_received = true; + return true; +} - if (error != nullptr) { - *error = ServerError(e); +bool Client::Impl::ReceiveProfileInfo(Profile & profile) { + if (!WireFormat::ReadUInt64(*input_, &profile.rows) || + !WireFormat::ReadUInt64(*input_, &profile.blocks) || + !WireFormat::ReadUInt64(*input_, &profile.bytes) || + !WireFormat::ReadFixed(*input_, &profile.applied_limit) || + !WireFormat::ReadUInt64(*input_, &profile.rows_before_limit) || + !WireFormat::ReadFixed(*input_, &profile.calculated_rows_before_limit)) { + return false; + } + + if (events_) { + events_->OnProfile(profile); + } + + return true; +} + +bool Client::Impl::ReceiveProgress(Progress & progress) { + if (!WireFormat::ReadUInt64(*input_, &progress.rows) || + !WireFormat::ReadUInt64(*input_, &progress.bytes)) { + return false; + } + + if constexpr(DMBS_PROTOCOL_REVISION >= DBMS_MIN_REVISION_WITH_TOTAL_ROWS_IN_PROGRESS) { + if (!WireFormat::ReadUInt64(*input_, &progress.total_rows)) { + return false; + } + } + if (server_info_.revision >= DBMS_MIN_REVISION_WITH_CLIENT_WRITE_INFO) + { + if (!WireFormat::ReadUInt64(*input_, &progress.written_rows) || + !WireFormat::ReadUInt64(*input_, &progress.written_bytes)) { + return false; } } - return exception_received; + if (events_) { + events_->OnProgress(progress); + } + + return true; +} + +bool Client::Impl::ReceiveLog(Block & block) { + if (!WireFormat::SkipString(*input_) || // log tag + !ReadBlock(*input_, &block)) { // Use uncompressed stream since log blocks usually contain only one row + return false; + } + if (events_) { + events_->OnServerLog(block); + } + return true; +} + +bool Client::Impl::ReceiveTableColumns() { + if (!WireFormat::SkipString(*input_) || // external table name + !WireFormat::SkipString(*input_)) { // columns metadata + return false; + } + return true; +} + +bool Client::Impl::ReceiveProfileEvents(Block & block) { + if (!WireFormat::SkipString(*input_) || !ReadBlock(*input_, &block)) { + return false; + } + if (events_) { + events_->OnProfileEvents(block); + } + return true; } void Client::Impl::SendCancel() { @@ -1199,10 +1231,11 @@ bool Client::Impl::ReceiveHello() { return true; } else if (packet_type == ServerCodes::Exception) { - if (!ReceiveException(true)) { + ServerError ret{std::make_shared()}; + if (!ReceiveException(ret)) { throw ProtocolError{"server rejected the connection, but its exception packet could not be decoded"}; } - return false; + throw ret; } return false; From 5671a8a8a636ac5ccd063bcf339b3232f4d2c3c6 Mon Sep 17 00:00:00 2001 From: Andrew Slabko Date: Wed, 2 Sep 2026 21:04:33 +0200 Subject: [PATCH 3/3] Add offline wire-protocol tests for packet decoding errors --- ut/CMakeLists.txt | 1 + ut/client_protocol_ut.cpp | 240 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 241 insertions(+) create mode 100644 ut/client_protocol_ut.cpp diff --git a/ut/CMakeLists.txt b/ut/CMakeLists.txt index c827e5e3..b830ae17 100644 --- a/ut/CMakeLists.txt +++ b/ut/CMakeLists.txt @@ -5,6 +5,7 @@ SET ( clickhouse-cpp-ut-src bignum_ut.cpp bignum_round_trip.cpp block_ut.cpp + client_protocol_ut.cpp client_ut.cpp columns_ut.cpp column_as_ut.cpp diff --git a/ut/client_protocol_ut.cpp b/ut/client_protocol_ut.cpp new file mode 100644 index 00000000..72cfdd5d --- /dev/null +++ b/ut/client_protocol_ut.cpp @@ -0,0 +1,240 @@ +#include +#include +#include +#include +#include + +#include + +#include +#include +#include +#include + +namespace { + +using namespace clickhouse; + +/// A socket that serves a pre-recorded byte script as the server response +/// and discards (but keeps) everything written by the client. +class ScriptedSocket : public SocketBase { +public: + explicit ScriptedSocket(std::vector script) + : script_(std::move(script)) + {} + + std::unique_ptr makeInputStream() const override { + return std::make_unique(script_.data(), script_.size()); + } + + std::unique_ptr makeOutputStream() const override { + return std::make_unique(&written_); + } + +private: + const std::vector script_; + mutable std::vector written_; +}; + +class ScriptedSocketFactory : public SocketFactory { +public: + explicit ScriptedSocketFactory(std::vector script) + : script_(std::move(script)) + {} + + std::unique_ptr connect(const ClientOptions&, const Endpoint&) override { + return std::make_unique(script_); + } + +private: + std::vector script_; +}; + +ClientOptions ScriptedClientOptions() { + return ClientOptions() + .SetHost("scripted.test") + .SetPingBeforeQuery(false) + .SetSendRetries(0); +} + +/// A minimal well-formed ServerHello. Revision 50000 is below +/// DBMS_MIN_REVISION_WITH_SERVER_TIMEZONE (54058), so the server sends only +/// name, version_major, version_minor and revision. +const std::vector kServerHello = { + 0x00, // packet type: ServerCodes::Hello (varint) + 0x02, 0x43, 0x48, // server name: "CH" (string, len 2) + 0x01, // version_major = 1 (varint) + 0x01, // version_minor = 1 (varint) + 0xD0, 0x86, 0x03, // revision = 50000 (varint) +}; + +/// An exception packet with a valid code but a corrupt `name` string. +const std::vector kCorruptExceptionPacket = { + 0x02, // packet type: ServerCodes::Exception (varint) + 0x3C, 0x00, 0x00, 0x00, // code = 60 (int32, little-endian) + 0xFF, 0xFF, 0xFF, 0xFF, 0x0F, // name string length = 0xFFFFFFFF (varint), + // exceeds the 0x00FFFFFF limit in WireFormat::ReadString +}; + +/// A complete well-formed exception packet. +const std::vector kWellFormedExceptionPacket = { + 0x02, // packet type: ServerCodes::Exception (varint) + 0x3C, 0x00, 0x00, 0x00, // code = 60 (int32, little-endian) + 0x02, 0x44, 0x42, // name: "DB" (string, len 2) + 0x04, 0x6F, 0x6F, 0x70, 0x73, // display_text: "oops" (string, len 4) + 0x00, // stack_trace: "" (string, len 0) + 0x00, // has_nested = false (fixed, 1 byte) +}; + +/// Performs the handshake against kServerHello, then runs a Select that +/// receives `response` and expects a ProtocolError containing +/// `expected_message`. +void ExpectSelectThrowsProtocolError(const std::vector& response, + const std::string& expected_message) { + std::vector script = kServerHello; + script.insert(script.end(), response.begin(), response.end()); + // Trailing garbage: decoding must fail on validation, not on end-of-stream. + script.insert(script.end(), 64, 0xAA); + + // The handshake must succeed; only Select is expected to throw. + Client client(ScriptedClientOptions(), + std::make_unique(script)); + + try { + client.Select("SELECT 1", [](const Block&) {}); + FAIL() << "expected ProtocolError"; + } catch (const ProtocolError& e) { + EXPECT_TRUE(std::string(e.what()).find(expected_message) != std::string::npos) + << "unexpected message: " << e.what(); + } +} + +} + +TEST(ClientProtocol, MalformedServerHelloThrowsProtocolError) { + std::vector script = { + 0x00, // packet type: ServerCodes::Hello (varint) + 0xFF, 0xFF, 0xFF, 0xFF, 0x0F, // server-name string length = 0xFFFFFFFF (varint), + // exceeds the 0x00FFFFFF limit in WireFormat::ReadString + }; + + // Trailing garbage: decoding must fail on validation, not on end-of-stream. + script.insert(script.end(), 64, 0xAA); + + try { + Client client(ScriptedClientOptions(), + std::make_unique(script)); + FAIL() << "expected ProtocolError"; + } catch (const ProtocolError& e) { + EXPECT_TRUE(std::string(e.what()).find("fail to connect") != std::string::npos) + << "unexpected message: " << e.what(); + } +} + +TEST(ClientProtocol, MalformedSelectResponseThrowsProtocolError) { + // Malformed packet type: a varint whose continuation bit never clears, + // so ReadVarint64 gives up after MAX_VARINT_BYTES. + const std::vector response(10, 0x80); + ExpectSelectThrowsProtocolError(response, "can't read packet type"); +} + +TEST(ClientProtocol, CorruptProgressPacketThrowsProtocolError) { + std::vector response = { + 0x03, // packet type: ServerCodes::Progress (varint) + }; + // Malformed `rows` varint: the continuation bit never clears. + response.insert(response.end(), 10, 0x80); + ExpectSelectThrowsProtocolError(response, "can't read progress packet"); +} + +TEST(ClientProtocol, CorruptProfileInfoPacketThrowsProtocolError) { + std::vector response = { + 0x06, // packet type: ServerCodes::ProfileInfo (varint) + }; + // Malformed `rows` varint: the continuation bit never clears. + response.insert(response.end(), 10, 0x80); + ExpectSelectThrowsProtocolError(response, "can't read profile info packet"); +} + +TEST(ClientProtocol, CorruptLogPacketThrowsProtocolError) { + const std::vector response = { + 0x0A, // packet type: ServerCodes::Log (varint) + 0xFF, 0xFF, 0xFF, 0xFF, 0x0F, // log-tag string length = 0xFFFFFFFF (varint), + // exceeds the 0x00FFFFFF limit in WireFormat::SkipString + }; + ExpectSelectThrowsProtocolError(response, "can't read log packet"); +} + +TEST(ClientProtocol, CorruptExceptionPacketInSelectThrowsProtocolError) { + ExpectSelectThrowsProtocolError(kCorruptExceptionPacket, "error details lost"); +} + +TEST(ClientProtocol, CorruptExceptionPacketInHandshakeThrowsProtocolError) { + // The exception packet arrives instead of ServerHello. + std::vector script = kCorruptExceptionPacket; + // Trailing garbage: decoding must fail on validation, not on end-of-stream. + script.insert(script.end(), 64, 0xAA); + + try { + Client client(ScriptedClientOptions(), + std::make_unique(script)); + FAIL() << "expected ProtocolError"; + } catch (const ProtocolError& e) { + EXPECT_TRUE(std::string(e.what()).find("exception packet could not be decoded") != std::string::npos) + << "unexpected message: " << e.what(); + } +} + +TEST(ClientProtocol, ExceptionPacketInSelectThrowsServerException) { + std::vector script = kServerHello; + script.insert(script.end(), + kWellFormedExceptionPacket.begin(), kWellFormedExceptionPacket.end()); + + // The handshake must succeed; only Select is expected to throw. + Client client(ScriptedClientOptions(), + std::make_unique(script)); + + try { + client.Select("SELECT 1", [](const Block&) {}); + FAIL() << "expected ServerException"; + } catch (const ServerException& e) { + EXPECT_EQ(e.GetCode(), 60); + EXPECT_EQ(e.GetException().name, "DB"); + EXPECT_STREQ(e.what(), "oops"); + } +} + +TEST(ClientProtocol, ExceptionPacketInHandshakeThrowsServerException) { + // The exception packet arrives instead of ServerHello. + try { + Client client(ScriptedClientOptions(), + std::make_unique(kWellFormedExceptionPacket)); + FAIL() << "expected ServerException"; + } catch (const ServerException& e) { + EXPECT_EQ(e.GetCode(), 60); + EXPECT_EQ(e.GetException().name, "DB"); + EXPECT_STREQ(e.what(), "oops"); + } +} + +TEST(ClientProtocol, WellFormedSelectResponseSucceeds) { + std::vector script = kServerHello; + const std::vector response = { + 0x01, // packet type: ServerCodes::Data (varint) + // (revision 50000 has neither temp-table name nor BlockInfo) + 0x00, // num_columns = 0 (varint) + 0x00, // num_rows = 0 (varint) + 0x05, // packet type: ServerCodes::EndOfStream (varint) + }; + script.insert(script.end(), response.begin(), response.end()); + + Client client(ScriptedClientOptions(), + std::make_unique(script)); + + size_t blocks = 0; + EXPECT_NO_THROW(client.Select("SELECT 1", [&blocks](const Block& block) { + ++blocks; + EXPECT_EQ(block.GetRowCount(), 0u); + })); + EXPECT_EQ(blocks, 1u); +}