From 28d960ac229c6b5e617b2617b834409fe8162d9c Mon Sep 17 00:00:00 2001 From: ruslandoga Date: Mon, 3 Aug 2026 14:32:31 +0300 Subject: [PATCH 1/8] Add regression test for late HTTP exceptions --- test/ch/query_test.exs | 33 +++++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/test/ch/query_test.exs b/test/ch/query_test.exs index e5e9a20a..31a114ee 100644 --- a/test/ch/query_test.exs +++ b/test/ch/query_test.exs @@ -492,6 +492,39 @@ defmodule Ch.QueryTest do assert {:error, %Ch.Error{code: 62}} = Ch.query(conn, "wat", [], query_options) end + test "detects __exception__ block in HTTP 200 response", %{ + conn: conn, + query_options: query_options + } do + probe = Ch.query!(conn, "SELECT 1", %{}, query_options) + + # Older ClickHouse versions in the CI matrix do not emit tagged exception blocks. + if :proplists.get_value("x-clickhouse-exception-tag", probe.headers) do + assert {:error, %Ch.Error{message: message}} = + Ch.query( + conn, + """ + SELECT number, throwIf(number = 2, 'late exception') + FROM numbers(5) + """, + %{}, + Keyword.merge(query_options, + headers: [{"x-clickhouse-format", "CSV"}], + settings: [ + wait_end_of_query: 0, + http_response_buffer_size: 1, + output_format_parallel_formatting: 0, + max_threads: 1, + max_block_size: 1 + ] + ) + ) + + assert message =~ "Code: 395" + assert message =~ "late exception" + end + end + test "connection works after failure in execute", %{conn: conn, query_options: query_options} do assert {:error, %Ch.Error{}} = Ch.query(conn, "wat", [], query_options) assert [[42]] = Ch.query!(conn, "SELECT 42", [], query_options).rows From 7ce000bd654dcfb9e8777fa2ec192a03ba297088 Mon Sep 17 00:00:00 2001 From: ruslandoga Date: Mon, 3 Aug 2026 15:11:05 +0300 Subject: [PATCH 2/8] Use List.keyfind for exception header probe --- test/ch/query_test.exs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/ch/query_test.exs b/test/ch/query_test.exs index 31a114ee..06bf706d 100644 --- a/test/ch/query_test.exs +++ b/test/ch/query_test.exs @@ -499,7 +499,7 @@ defmodule Ch.QueryTest do probe = Ch.query!(conn, "SELECT 1", %{}, query_options) # Older ClickHouse versions in the CI matrix do not emit tagged exception blocks. - if :proplists.get_value("x-clickhouse-exception-tag", probe.headers) do + if List.keyfind(probe.headers, "x-clickhouse-exception-tag", 0) do assert {:error, %Ch.Error{message: message}} = Ch.query( conn, From 22c3f722dc34ee862ade3ce6de4052c79a1942b3 Mon Sep 17 00:00:00 2001 From: ruslandoga Date: Mon, 3 Aug 2026 15:19:58 +0300 Subject: [PATCH 3/8] Exercise late exceptions with the default format --- test/ch/query_test.exs | 1 - 1 file changed, 1 deletion(-) diff --git a/test/ch/query_test.exs b/test/ch/query_test.exs index 06bf706d..8510e51c 100644 --- a/test/ch/query_test.exs +++ b/test/ch/query_test.exs @@ -509,7 +509,6 @@ defmodule Ch.QueryTest do """, %{}, Keyword.merge(query_options, - headers: [{"x-clickhouse-format", "CSV"}], settings: [ wait_end_of_query: 0, http_response_buffer_size: 1, From 7aece5e1bdbeb70570889b2f997d1052283a1e08 Mon Sep 17 00:00:00 2001 From: ruslandoga Date: Mon, 3 Aug 2026 15:34:20 +0300 Subject: [PATCH 4/8] Handle late ClickHouse HTTP exceptions --- test/ch/query_test.exs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/ch/query_test.exs b/test/ch/query_test.exs index 8510e51c..18e06e62 100644 --- a/test/ch/query_test.exs +++ b/test/ch/query_test.exs @@ -500,7 +500,7 @@ defmodule Ch.QueryTest do # Older ClickHouse versions in the CI matrix do not emit tagged exception blocks. if List.keyfind(probe.headers, "x-clickhouse-exception-tag", 0) do - assert {:error, %Ch.Error{message: message}} = + assert {:error, %Ch.Error{code: 395, message: message}} = Ch.query( conn, """ From 8e46f60d91e3592c4664037c313fa4b952041673 Mon Sep 17 00:00:00 2001 From: ruslandoga Date: Mon, 3 Aug 2026 15:43:01 +0300 Subject: [PATCH 5/8] Restrict late exceptions to RowBinary responses --- lib/ch/row_binary.ex | 62 ++++++++++++++++++++++++++++++++++++++++++ test/ch/query_test.exs | 44 ++++++++++++++---------------- test/test_helper.exs | 2 +- 3 files changed, 83 insertions(+), 25 deletions(-) diff --git a/lib/ch/row_binary.ex b/lib/ch/row_binary.ex index 1ff0d7f2..b0d0ddab 100644 --- a/lib/ch/row_binary.ex +++ b/lib/ch/row_binary.ex @@ -8,6 +8,10 @@ defmodule Ch.RowBinary do @epoch_gregorian_seconds 62_167_219_200 @epoch_gregorian_days 719_528 + @exception_marker "__exception__" + @exception_tag_length 16 + @max_exception_size 16 * 1024 + @max_exception_length_digits 8 @doc false def encode_names_and_types(names, types) do @@ -660,6 +664,64 @@ defmodule Ch.RowBinary do end end + @doc false + def decode_names_and_rows(row_binary_with_names_and_types, exception_tag) do + case decode_exception(row_binary_with_names_and_types, exception_tag) do + {:ok, message} -> {:error, message} + :error -> {:ok, decode_names_and_rows(row_binary_with_names_and_types)} + end + end + + @doc false + def decode_exception(body, tag) + when is_binary(tag) and byte_size(tag) == @exception_tag_length do + opening = "\r\n#{@exception_marker}\r\n#{tag}\r\n" + closing = " #{tag}\r\n#{@exception_marker}\r\n" + body_size = byte_size(body) + closing_size = byte_size(closing) + closing_start = body_size - closing_size + + with true <- closing_start >= 0, + ^closing <- binary_part(body, closing_start, closing_size), + {:ok, length_start} <- trailing_decimal_start(body, closing_start), + {message_length, ""} <- + body |> binary_part(length_start, closing_start - length_start) |> Integer.parse(), + message_start = length_start - message_length, + opening_start = message_start - byte_size(opening), + true <- opening_start >= 0, + true <- body_size - opening_start <= @max_exception_size, + ^opening <- binary_part(body, opening_start, byte_size(opening)), + message <- binary_part(body, message_start, message_length), + true <- String.ends_with?(message, "\n") do + {:ok, message} + else + _ -> :error + end + end + + def decode_exception(_body, _tag), do: :error + + defp trailing_decimal_start(body, index), do: trailing_decimal_start(body, index, 0) + + defp trailing_decimal_start(body, index, digits) when index > 0 do + case :binary.at(body, index - 1) do + digit when digit in ?0..?9 and digits < @max_exception_length_digits -> + trailing_decimal_start(body, index - 1, digits + 1) + + digit when digit in ?0..?9 -> + :error + + _other when digits > 0 -> + {:ok, index} + + _other -> + :error + end + end + + defp trailing_decimal_start(_body, 0, digits) when digits > 0, do: {:ok, 0} + defp trailing_decimal_start(_body, 0, 0), do: :error + @doc """ Decodes [RowBinary](https://clickhouse.com/docs/en/interfaces/formats/RowBinary) into rows. diff --git a/test/ch/query_test.exs b/test/ch/query_test.exs index 18e06e62..26084b7e 100644 --- a/test/ch/query_test.exs +++ b/test/ch/query_test.exs @@ -492,36 +492,32 @@ defmodule Ch.QueryTest do assert {:error, %Ch.Error{code: 62}} = Ch.query(conn, "wat", [], query_options) end + @tag :exception test "detects __exception__ block in HTTP 200 response", %{ conn: conn, query_options: query_options } do - probe = Ch.query!(conn, "SELECT 1", %{}, query_options) - - # Older ClickHouse versions in the CI matrix do not emit tagged exception blocks. - if List.keyfind(probe.headers, "x-clickhouse-exception-tag", 0) do - assert {:error, %Ch.Error{code: 395, message: message}} = - Ch.query( - conn, - """ - SELECT number, throwIf(number = 2, 'late exception') - FROM numbers(5) - """, - %{}, - Keyword.merge(query_options, - settings: [ - wait_end_of_query: 0, - http_response_buffer_size: 1, - output_format_parallel_formatting: 0, - max_threads: 1, - max_block_size: 1 - ] - ) + assert {:error, %Ch.Error{code: 395, message: message}} = + Ch.query( + conn, + """ + SELECT number, throwIf(number = 2, 'late exception') + FROM numbers(5) + """, + %{}, + Keyword.merge(query_options, + settings: [ + wait_end_of_query: 0, + http_response_buffer_size: 1, + output_format_parallel_formatting: 0, + max_threads: 1, + max_block_size: 1 + ] ) + ) - assert message =~ "Code: 395" - assert message =~ "late exception" - end + assert message =~ "Code: 395" + assert message =~ "late exception" end test "connection works after failure in execute", %{conn: conn, query_options: query_options} do diff --git a/test/test_helper.exs b/test/test_helper.exs index 8e783ee5..305a464e 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -45,7 +45,7 @@ extra_exclude = true -> # Time, Variant, JSON, and Dynamic types are not supported in older ClickHouse versions we have in the CI - [:time, :variant, :json, :dynamic] + [:time, :variant, :json, :dynamic, :exception] end ExUnit.start(exclude: [:slow | extra_exclude]) From 87d7cd028732d05b13a70bcb57442166dfe0fd84 Mon Sep 17 00:00:00 2001 From: ruslandoga Date: Sun, 9 Aug 2026 18:11:36 +0300 Subject: [PATCH 6/8] Adapt late exceptions to DBConnection --- lib/ch/connection.ex | 75 ++++++++++++++++++++++++++++++++++---------- 1 file changed, 59 insertions(+), 16 deletions(-) diff --git a/lib/ch/connection.ex b/lib/ch/connection.ex index b53394ab..1a2195b6 100644 --- a/lib/ch/connection.ex +++ b/lib/ch/connection.ex @@ -343,6 +343,7 @@ defmodule Ch.Connection do case request(conn, "POST", path, headers, body, opts) do {:ok, conn, responses} -> {:ok, query, responses, conn} {:error, _reason, _conn} = client_error -> client_error + {:disconnect, %Error{} = reason, conn} -> {:disconnect, reason, conn} {:disconnect, reason, conn} -> {:disconnect_and_retry, reason, conn} end end @@ -408,29 +409,68 @@ defmodule Ch.Connection do @spec receive_full_response(conn, timeout) :: {:ok, conn, [response]} | {:error, Error.t(), conn} - | {:disconnect, Mint.Types.error(), conn} + | {:disconnect, Error.t() | Mint.Types.error(), conn} defp receive_full_response(conn, timeout) do - with {:ok, conn, responses} <- recv_all(conn, [], timeout) do - case responses do - [200, headers | _rest] -> - conn = ensure_same_server(conn, headers) - {:ok, conn, responses} - - [_status, headers | data] -> - message = IO.iodata_to_binary(data) - - code = - if code = get_header(headers, "x-clickhouse-exception-code") do - String.to_integer(code) - end + case recv_all(conn, [], timeout) do + {:ok, conn, responses} -> + case responses do + [200, headers | _rest] -> + conn = ensure_same_server(conn, headers) + {:ok, conn, responses} + + [_status, headers | data] -> + message = IO.iodata_to_binary(data) + + code = + if code = get_header(headers, "x-clickhouse-exception-code") do + String.to_integer(code) + end + + {:error, Error.exception(code: code, message: message), conn} + end - {:error, Error.exception(code: code, message: message), conn} + {:closed, conn, responses, reason} -> + case decode_row_binary_exception(responses) do + {:ok, error} -> {:disconnect, error, conn} + :error -> {:disconnect, reason, conn} + end + + {:disconnect, _reason, _conn} = disconnect -> + disconnect + end + end + + defp decode_row_binary_exception([200, headers | data]) do + try do + with "RowBinaryWithNamesAndTypes" <- get_header(headers, "x-clickhouse-format"), + nil <- get_header(headers, "content-encoding"), + tag when is_binary(tag) <- get_header(headers, "x-clickhouse-exception-tag"), + body = IO.iodata_to_binary(data), + {:error, message} <- RowBinary.decode_names_and_rows(body, tag) do + {:ok, Error.exception(code: exception_code(message), message: message)} + else + _ -> :error end + rescue + _ -> :error end end + defp decode_row_binary_exception(_responses), do: :error + + defp exception_code(<<"Code: ", rest::binary>>) do + case Integer.parse(rest) do + {code, <<".", _rest::binary>>} -> code + _ -> nil + end + end + + defp exception_code(_message), do: nil + @spec recv_all(conn, [response], timeout()) :: - {:ok, conn, [response]} | {:disconnect, Mint.Types.error(), conn} + {:ok, conn, [response]} + | {:closed, conn, [response], Mint.Types.error()} + | {:disconnect, Mint.Types.error(), conn} defp recv_all(conn, acc, timeout) do case HTTP.recv(conn, 0, timeout) do {:ok, conn, responses} -> @@ -439,6 +479,9 @@ defmodule Ch.Connection do {:more, acc} -> recv_all(conn, acc, timeout) end + {:error, conn, %Mint.TransportError{reason: :closed} = reason, _responses} -> + {:closed, conn, :lists.reverse(acc), reason} + {:error, conn, reason, _responses} -> {:disconnect, reason, conn} end From ed4e8317023e198ccb55aa23d4a0d66d2c49ec66 Mon Sep 17 00:00:00 2001 From: ruslandoga Date: Sun, 9 Aug 2026 18:20:49 +0300 Subject: [PATCH 7/8] Detect exceptions in completed responses --- lib/ch/connection.ex | 11 +++++++---- lib/ch/row_binary.ex | 36 +++++++++++++++++++++++++++++++----- test/ch/row_binary_test.exs | 30 ++++++++++++++++++++++++++++++ 3 files changed, 68 insertions(+), 9 deletions(-) diff --git a/lib/ch/connection.ex b/lib/ch/connection.ex index 1a2195b6..3169dd60 100644 --- a/lib/ch/connection.ex +++ b/lib/ch/connection.ex @@ -414,9 +414,13 @@ defmodule Ch.Connection do case recv_all(conn, [], timeout) do {:ok, conn, responses} -> case responses do - [200, headers | _rest] -> + [200, headers | _data] -> conn = ensure_same_server(conn, headers) - {:ok, conn, responses} + + case decode_row_binary_exception(responses) do + {:ok, error} -> {:error, error, conn} + :error -> {:ok, conn, responses} + end [_status, headers | data] -> message = IO.iodata_to_binary(data) @@ -445,8 +449,7 @@ defmodule Ch.Connection do with "RowBinaryWithNamesAndTypes" <- get_header(headers, "x-clickhouse-format"), nil <- get_header(headers, "content-encoding"), tag when is_binary(tag) <- get_header(headers, "x-clickhouse-exception-tag"), - body = IO.iodata_to_binary(data), - {:error, message} <- RowBinary.decode_names_and_rows(body, tag) do + {:ok, message} <- RowBinary.decode_exception(data, tag) do {:ok, Error.exception(code: exception_code(message), message: message)} else _ -> :error diff --git a/lib/ch/row_binary.ex b/lib/ch/row_binary.ex index b0d0ddab..e60d92f0 100644 --- a/lib/ch/row_binary.ex +++ b/lib/ch/row_binary.ex @@ -665,14 +665,19 @@ defmodule Ch.RowBinary do end @doc false - def decode_names_and_rows(row_binary_with_names_and_types, exception_tag) do - case decode_exception(row_binary_with_names_and_types, exception_tag) do - {:ok, message} -> {:error, message} - :error -> {:ok, decode_names_and_rows(row_binary_with_names_and_types)} + def decode_exception(body, tag) + when is_list(body) and is_binary(tag) and byte_size(tag) == @exception_tag_length do + closing = " #{tag}\r\n#{@exception_marker}\r\n" + + if exception_tail(body, byte_size(closing)) == closing do + body + |> exception_tail(@max_exception_size) + |> decode_exception(tag) + else + :error end end - @doc false def decode_exception(body, tag) when is_binary(tag) and byte_size(tag) == @exception_tag_length do opening = "\r\n#{@exception_marker}\r\n#{tag}\r\n" @@ -701,6 +706,27 @@ defmodule Ch.RowBinary do def decode_exception(_body, _tag), do: :error + defp exception_tail(chunks, limit) do + chunks + |> :lists.reverse() + |> exception_tail([], 0, limit) + |> IO.iodata_to_binary() + end + + defp exception_tail(_chunks, acc, size, limit) when size == limit, do: acc + + defp exception_tail([chunk | chunks], acc, size, limit) do + remaining = limit - size + + if byte_size(chunk) <= remaining do + exception_tail(chunks, [chunk | acc], size + byte_size(chunk), limit) + else + [binary_part(chunk, byte_size(chunk) - remaining, remaining) | acc] + end + end + + defp exception_tail([], acc, _size, _limit), do: acc + defp trailing_decimal_start(body, index), do: trailing_decimal_start(body, index, 0) defp trailing_decimal_start(body, index, digits) when index > 0 do diff --git a/test/ch/row_binary_test.exs b/test/ch/row_binary_test.exs index 1fcd0a86..9da74cdd 100644 --- a/test/ch/row_binary_test.exs +++ b/test/ch/row_binary_test.exs @@ -312,6 +312,36 @@ defmodule Ch.RowBinaryTest do end end + describe "decode_exception/2" do + test "decodes a tagged exception split across response chunks" do + tag = "0123456789abcdef" + message = "Code: 395. DB::Exception: late exception\n" + + trailer = + "\r\n__exception__\r\n#{tag}\r\n#{message}#{byte_size(message)} #{tag}\r\n__exception__\r\n" + + assert {:ok, ^message} = + decode_exception( + [ + <<1, 1, "x", 5, "UInt8", 42>>, + binary_part(trailer, 0, 13), + binary_part(trailer, 13, byte_size(trailer) - 13) + ], + tag + ) + end + + test "rejects a trailer with a different tag" do + tag = "0123456789abcdef" + message = "Code: 395. DB::Exception: late exception\n" + + trailer = + "\r\n__exception__\r\n#{tag}\r\n#{message}#{byte_size(message)} fedcba9876543210\r\n__exception__\r\n" + + assert :error = decode_exception(trailer, tag) + end + end + describe "decode_rows/2" do test "empty" do assert decode_rows(<<>>, [:u8, :string]) == [] From d51b851cca8de85d81ed379b9efedf3d5731d7ec Mon Sep 17 00:00:00 2001 From: ruslandoga Date: Sun, 9 Aug 2026 18:26:58 +0300 Subject: [PATCH 8/8] Include ClickHouse errors in request specs --- lib/ch/connection.ex | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/lib/ch/connection.ex b/lib/ch/connection.ex index 3169dd60..c9c65080 100644 --- a/lib/ch/connection.ex +++ b/lib/ch/connection.ex @@ -363,7 +363,7 @@ defmodule Ch.Connection do @spec request(conn, binary, binary, Mint.Types.headers(), iodata, [Ch.query_option()]) :: {:ok, conn, [response]} | {:error, Error.t(), conn} - | {:disconnect, Mint.Types.error(), conn} + | {:disconnect, Error.t() | Mint.Types.error(), conn} defp request(conn, method, path, headers, body, opts) do with {:ok, conn, _ref} <- send_request(conn, method, path, headers, body) do receive_full_response(conn, timeout(conn, opts)) @@ -373,7 +373,7 @@ defmodule Ch.Connection do @spec request_chunked(conn, binary, binary, Mint.Types.headers(), Enumerable.t(), Keyword.t()) :: {:ok, conn, [response]} | {:error, Error.t(), conn} - | {:disconnect, Mint.Types.error(), conn} + | {:disconnect, Error.t() | Mint.Types.error(), conn} def request_chunked(conn, method, path, headers, stream, opts) do with {:ok, conn, ref} <- send_request(conn, method, path, headers, :stream), {:ok, conn} <- stream_body(conn, ref, stream),