Skip to content

Throw ProtocolError on packet decoding failures #564

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
slabko merged 3 commits into from
Sep 4, 2026
Merged

Throw ProtocolError on packet decoding failures #564

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
bb20ee9
Throw ProtocolError on packet decoding failures
slabko Sep 2, 2026
d41de8b
Extract per-packet decoding into Receive* helpers
slabko Sep 2, 2026
5671a8a
Add offline wire-protocol tests for packet decoding errors
slabko Sep 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
223 changes: 122 additions & 101 deletions clickhouse/client.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
#include "columns/factory.h"

#include <cassert>
#include <memory>
#include <optional>
#include <sstream>
#include <system_error>
Expand Down Expand Up @@ -157,7 +158,6 @@ struct ProfileEvents {
struct EndOfStream {
};
using DecodedPacket = std::variant<
std::monostate,
Block,
ServerException,
Profile,
Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -386,7 +396,9 @@ std::optional<Block> Client::Impl::NextBlock() {
return {std::move(block)};
}
case VariantIndex<ServerError, decltype(packet)>():
case VariantIndex<std::monostate, decltype(packet)>():
// 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<EndOfStream, decltype(packet)>():
ResetState();
return std::nullopt;
Expand Down Expand Up @@ -674,7 +686,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"};

@chernser chernser Sep 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

may be use different messages to mark point where it is failing?

@chernser chernser Sep 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

actually everything looks ok - never mind.
I usually like to see "read failed: packet type" - so I can grep on "read failed" and see what steps passed.

}
if (server_packet) {
*server_packet = packet_type;
Expand All @@ -684,76 +696,37 @@ 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: {
ServerError ret{std::make_shared<Exception>()};

@slabko slabko Sep 4, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is okay for this rare exception path.

if (!ReceiveException(false, &ret)) {
throw ProtocolError("can't read exception packet from input stream");
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)) {
return {};
}
if (!WireFormat::ReadUInt64(*input_, &ret.blocks)) {
return {};
}
if (!WireFormat::ReadUInt64(*input_, &ret.bytes)) {
return {};
}
if (!WireFormat::ReadFixed(*input_, &ret.applied_limit)) {
return {};
if (!ReceiveProfileInfo(ret)) {
throw ProtocolError{"can't read profile info packet from input stream"};
}
if (!WireFormat::ReadUInt64(*input_, &ret.rows_before_limit)) {
return {};
}
if (!WireFormat::ReadFixed(*input_, &ret.calculated_rows_before_limit)) {
return {};
}

if (events_) {
events_->OnProfile(ret);
}

return ret;
}

case ServerCodes::Progress: {
Progress ret{};

if (!WireFormat::ReadUInt64(*input_, &ret.rows)) {
return {};
}
if (!WireFormat::ReadUInt64(*input_, &ret.bytes)) {
return {};
}
if constexpr(DMBS_PROTOCOL_REVISION >= DBMS_MIN_REVISION_WITH_TOTAL_ROWS_IN_PROGRESS) {
if (!WireFormat::ReadUInt64(*input_, &ret.total_rows)) {
return {};
}
}
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 (events_) {
events_->OnProgress(ret);
if (!ReceiveProgress(ret)) {
throw ProtocolError{"can't read progress packet from input stream"};
}

return ret;
}

Expand All @@ -773,63 +746,40 @@ 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 (events_) {
events_->OnServerLog(ret.block);
if (!ReceiveLog(ret.block)) {
throw ProtocolError{"can't read log packet from input stream"};
}
return ret;
}

case ServerCodes::TableColumns: {
// external table name
if (!WireFormat::SkipString(*input_)) {
return {};
}

// columns metadata
if (!WireFormat::SkipString(*input_)) {
return {};
if (!ReceiveTableColumns()) {
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 (events_) {
events_->OnProfileEvents(ret.block);
if (!ReceiveProfileEvents(ret.block)) {
throw ProtocolError{"can't read profile events packet from input stream"};
}
return ret;
}

default:
throw UnimplementedError("unimplemented " + std::to_string((int)packet_type));
break;
}
}

bool Client::Impl::ProcessPacket(uint64_t* server_packet) {
auto packet = ReceivePacket(server_packet);
switch (packet.index()) {
case VariantIndex<ServerError, decltype(packet)>():
case VariantIndex<std::monostate, decltype(packet)>():
// 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<EndOfStream, decltype(packet)>():
return false;
default:
Expand Down Expand Up @@ -946,29 +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<Exception> e(new Exception);
bool has_nested = false; // obsolete: https://github.com/ClickHouse/ClickHouse/blob/ef11941cf5a/src/IO/ReadHelpers.cpp#L2017

bool exception_received =
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 (rethrow || options_.rethrow_exceptions) {
throw ServerError(e);
error = ServerError(e);

return true;
}

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 (exception_received && error != nullptr) {
*error = ServerError(e);
if (events_) {
events_->OnProfile(profile);
}
return exception_received;

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;
}
}

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() {
Expand Down Expand Up @@ -1213,8 +1231,11 @@ bool Client::Impl::ReceiveHello() {

return true;
} else if (packet_type == ServerCodes::Exception) {
ReceiveException(true);
return false;
ServerError ret{std::make_shared<Exception>()};
if (!ReceiveException(ret)) {
throw ProtocolError{"server rejected the connection, but its exception packet could not be decoded"};
}
throw ret;
}
Comment on lines 1233 to 1239

@slabko slabko Sep 4, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hello happens only during the handshake, and yes, in that case the client throws another exception.


return false;
Expand Down
1 change: 1 addition & 0 deletions ut/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading
Loading