19911268462

Committed 03 Dec 2025 10:43PM UTC coverage: 63.801% (-0.2%) from 63.971%

Build # 19911268462

Build Type

push

github

Committed by

web-flow

Commit Message

chore: more buckets for higher resolution on histograms (#784)

Run Details

1870 of 2931 relevant lines covered (63.8%)

4193.69 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

40.54

/lib/supavisor/client_handler/error.ex

defmodule Supavisor.ClientHandler.Error do
  @moduledoc """
  Error handling and message formatting for ClientHandler.
  """

  alias Supavisor.{HandlerHelpers, Protocol.Server}

  require Supavisor.Protocol.PreparedStatements, as: PreparedStatements
  require Logger

  @doc """
  Handles error by logging and sending appropriate error message to the client socket.

  Optional context parameter is used for generic errors to indicate where they occurred.
  """
  @spec maybe_log_and_send_error(term(), term(), term()) :: :ok
  def maybe_log_and_send_error(sock, error, context \\ nil) do
    error_actions = process(error, context)
    message = Map.fetch!(error_actions, :error)
    log_message = Map.get(error_actions, :log_message)
    send_ready_for_query = Map.get(error_actions, :send_ready_for_query, false)
    auth_error = Map.get(error_actions, :auth_error, false)

    if log_message do
      Logger.error("ClientHandler: #{log_message}", auth_error: auth_error)
    end

    if send_ready_for_query do
      HandlerHelpers.sock_send(sock, [message, Server.ready_for_query()])
    else
      HandlerHelpers.sock_send(sock, message)
    end
  end

  @spec process(term(), term()) :: map()
  defp process({:error, :max_prepared_statements}, _context) do
    message_text =
      "max prepared statements limit reached. Limit: #{PreparedStatements.client_limit()} per connection"

    %{
      error: Server.error_message("XX000", message_text),
      log_message: message_text
    }
  end

  defp process({:error, :prepared_statement_on_simple_query}, _context) do
    message_text =
      "Supavisor transaction mode only supports prepared statements using the Extended Query Protocol"

    %{
      error: Server.error_message("XX000", message_text),
      log_message: message_text,
      send_ready_for_query: true
    }
  end

  defp process({:error, :max_prepared_statements_memory}, _context) do
    limit_mb = PreparedStatements.client_memory_limit_bytes() / 1_000_000

    message_text =
      "max prepared statements memory limit reached. Limit: #{limit_mb}MB per connection"

    %{
      error: Server.error_message("XX000", message_text),
      log_message: message_text
    }
  end

  defp process({:error, :prepared_statement_not_found, name}, _context) do
    message_text = "prepared statement #{inspect(name)} does not exist"

    %{
      error: Server.error_message("26000", message_text),
      log_message: message_text
    }
  end

  defp process({:error, :duplicate_prepared_statement, name}, _context) do
    message_text = "prepared statement #{inspect(name)} already exists"

    %{
      error: Server.error_message("42P05", message_text),
      log_message: message_text
    }
  end

  defp process({:error, :ssl_required, user}, _context) do
    %{
      error: Server.error_message("XX000", "SSL connection is required"),
      log_message: "Tenant is not allowed to connect without SSL, user #{user}"
    }
  end

  defp process({:error, :address_not_allowed, addr}, _context) do
    message = "Address not in tenant allow_list: " <> inspect(addr)

    %{
      error: Server.error_message("XX000", message),
      log_message: message
    }
  end

  defp process({:error, :tenant_not_found}, _context) do
    %{
      error: Server.error_message("XX000", "Tenant or user not found"),
      log_message: "Tenant not found"
    }
  end

  defp process({:error, :tenant_not_found, reason, type, user, tenant_or_alias}, _context) do
    %{
      error: Server.error_message("XX000", "Tenant or user not found"),
      log_message: "User not found: #{inspect(reason)} #{inspect({type, user, tenant_or_alias})}",
      auth_error: true
    }
  end

  defp process({:error, :auth_error, :wrong_password, user}, _context) do
    %{
      error: Server.error_message("28P01", "password authentication failed for user \"#{user}\""),
      log_message: "Exchange error: password authentication failed for user \"#{user}\"",
      auth_error: true
    }
  end

  defp process({:error, :auth_error, {:timeout, _message}, _user}, _context) do
    %{
      error: Server.error_message("08006", "connection failure during authentication")
    }
  end

  defp process({:error, :auth_error, {:unexpected_message, _details}, _user}, _context) do
    %{
      error: Server.error_message("08P01", "protocol violation during authentication")
    }
  end

  defp process({:error, :auth_error, {:decode_error, error}}, context) do
    auth_stage = auth_context_description(context)

    %{
      error: Server.error_message("08P01", "protocol violation during authentication"),
      log_message: "#{auth_stage} auth decode error: #{inspect(error)}"
    }
  end

  defp process({:error, :auth_error, {:unexpected_message, other}}, context) do
    auth_stage = auth_context_description(context)

    %{
      error: Server.error_message("08P01", "protocol violation during authentication"),
      log_message: "#{auth_stage} auth unexpected message: #{inspect(other)}"
    }
  end

  defp process({:error, :auth_error, {:timeout, _}}, context) do
    log_message =
      case context do
        :auth_md5_wait -> "Timeout while waiting for MD5 password"
        :auth_scram_first_wait -> "Timeout while waiting for first SCRAM message"
        :auth_scram_final_wait -> "Timeout while waiting for final SCRAM message"
        _ -> "Authentication timeout"
      end

    %{
      error: Server.error_message("08006", "connection failure during authentication"),
      log_message: log_message
    }
  end

  defp process({:error, {:invalid_user_info, {:invalid_format, {user, db_name}}}}, _context) do
    %{
      error:
        Server.error_message(
          "XX000",
          "Authentication error, reason: \"Invalid format for user or db_name\""
        ),
      log_message: "Invalid format for user or db_name: #{inspect({user, db_name})}"
    }
  end

  defp process({:error, :auth_error, reason}, _context) do
    %{
      error: Server.error_message("XX000", "Authentication error, reason: #{inspect(reason)}")
    }
  end

  defp process({:error, :max_clients_reached}, _context) do
    %{
      error: Server.error_message("XX000", "Max client connections reached"),
      log_message: "Max client connections reached"
    }
  end

  defp process({:error, :max_clients_reached_session}, _context) do
    message =
      "MaxClientsInSessionMode: max clients reached - in Session mode max clients are limited to pool_size"

    %{
      error: Server.error_message("XX000", message),
      log_message: message
    }
  end

  defp process({:error, :max_pools_reached}, _context) do
    %{
      error: Server.error_message("XX000", "Max pools count reached"),
      log_message: "Max pools count reached"
    }
  end

  defp process({:error, :db_handler_exited, pid, reason}, _context) do
    message =
      case reason do
        :db_termination -> "Connection to database closed. Check logs for more information"
        _ -> "DbHandler exited. Check logs for more information."
      end

    %{
      error: Server.error_message("XX000", message),
      log_message: "DbHandler #{inspect(pid)} exited #{inspect(reason)}"
    }
  end

  defp process({:error, :session_timeout}, _context) do
    message =
      "MaxClientsInSessionMode: max clients reached - in Session mode max clients are limited to pool_size"

    %{
      error: Server.error_message("XX000", message),
      log_message: message
    }
  end

  defp process({:error, :transaction_timeout}, _context) do
    message = "Unable to check out connection from the pool due to timeout"

    %{
      error: Server.error_message("XX000", message),
      log_message: message
    }
  end

  defp process({:error, :subscribe_retries_exhausted}, _context) do
    message = "Failed to subscribe to tenant after multiple retries. Terminating."

    %{
      error: Server.error_message("XX000", message),
      log_message: message
    }
  end

  defp process({:error, :circuit_breaker_open, operation, blocked_until}, _context) do
    explanation = Supavisor.CircuitBreaker.explanation(operation)
    message = "Circuit breaker open for operation: #{operation}, blocked until: #{blocked_until}"

    %{
      error: Server.error_message("XX000", "Circuit breaker open: #{explanation}"),
      log_message: message
    }
  end

  defp process(error, context) do
    message =
      case context do
        nil -> "Internal error: #{inspect(error)}"
        context -> "Internal error (#{context}): #{inspect(error)}"
      end

    %{
      error: Server.error_message("XX000", message),
      log_message: message
    }
  end

  defp auth_context_description(:auth_md5_wait), do: "MD5"
  defp auth_context_description(:auth_scram_first_wait), do: "SCRAM first"
  defp auth_context_description(:auth_scram_final_wait), do: "SCRAM final"
  defp auth_context_description(_), do: "Unknown"
end

1	defmodule Supavisor.ClientHandler.Error do
2	@moduledoc """
3	Error handling and message formatting for ClientHandler.
4	"""
5
6	alias Supavisor.{HandlerHelpers, Protocol.Server}
7
8	require Supavisor.Protocol.PreparedStatements, as: PreparedStatements
9	require Logger
10
11	@doc """
12	Handles error by logging and sending appropriate error message to the client socket.
13
14	Optional context parameter is used for generic errors to indicate where they occurred.
15	"""
16	@spec maybe_log_and_send_error(term(), term(), term()) :: :ok
17	def maybe_log_and_send_error(sock, error, context \\ nil) do
18	error_actions = process(error, context)	25✔
19	message = Map.fetch!(error_actions, :error)	25✔
20	log_message = Map.get(error_actions, :log_message)	25✔
21	send_ready_for_query = Map.get(error_actions, :send_ready_for_query, false)	25✔
22	auth_error = Map.get(error_actions, :auth_error, false)	25✔
23
24	if log_message do	25✔
25	Logger.error("ClientHandler: #{log_message}", auth_error: auth_error)	25✔
26	end
27
28	if send_ready_for_query do	25✔
29	HandlerHelpers.sock_send(sock, [message, Server.ready_for_query()])	4✔
30	else
31	HandlerHelpers.sock_send(sock, message)	21✔
32	end
33	end
34
35	@spec process(term(), term()) :: map()
36	defp process({:error, :max_prepared_statements}, _context) do
37	message_text =	1✔
38	"max prepared statements limit reached. Limit: #{PreparedStatements.client_limit()} per connection"	1✔
39
40	%{	1✔
41	error: Server.error_message("XX000", message_text),
42	log_message: message_text
43	}
44	end
45
46	defp process({:error, :prepared_statement_on_simple_query}, _context) do
47	message_text =	4✔
48	"Supavisor transaction mode only supports prepared statements using the Extended Query Protocol"
49
50	%{	4✔
51	error: Server.error_message("XX000", message_text),
52	log_message: message_text,
53	send_ready_for_query: true
54	}
55	end
56
57	defp process({:error, :max_prepared_statements_memory}, _context) do
58	limit_mb = PreparedStatements.client_memory_limit_bytes() / 1_000_000	1✔
59
60	message_text =	1✔
61	"max prepared statements memory limit reached. Limit: #{limit_mb}MB per connection"	1✔
62
63	%{	1✔
64	error: Server.error_message("XX000", message_text),
65	log_message: message_text
66	}
67	end
68
69	defp process({:error, :prepared_statement_not_found, name}, _context) do
70	message_text = "prepared statement #{inspect(name)} does not exist"	×
71
72	%{	×
73	error: Server.error_message("26000", message_text),
74	log_message: message_text
75	}
76	end
77
78	defp process({:error, :duplicate_prepared_statement, name}, _context) do
79	message_text = "prepared statement #{inspect(name)} already exists"	×
80
81	%{	×
82	error: Server.error_message("42P05", message_text),
83	log_message: message_text
84	}
85	end
86
87	defp process({:error, :ssl_required, user}, _context) do
88	%{	×
89	error: Server.error_message("XX000", "SSL connection is required"),
90	log_message: "Tenant is not allowed to connect without SSL, user #{user}"	×
91	}
92	end
93
94	defp process({:error, :address_not_allowed, addr}, _context) do
95	message = "Address not in tenant allow_list: " <> inspect(addr)	×
96
97	%{	×
98	error: Server.error_message("XX000", message),
99	log_message: message
100	}
101	end
102
103	defp process({:error, :tenant_not_found}, _context) do
104	%{	×
105	error: Server.error_message("XX000", "Tenant or user not found"),
106	log_message: "Tenant not found"
107	}
108	end
109
110	defp process({:error, :tenant_not_found, reason, type, user, tenant_or_alias}, _context) do
111	%{	×
112	error: Server.error_message("XX000", "Tenant or user not found"),
113	log_message: "User not found: #{inspect(reason)} #{inspect({type, user, tenant_or_alias})}",
114	auth_error: true
115	}
116	end
117
118	defp process({:error, :auth_error, :wrong_password, user}, _context) do
119	%{	13✔
120	error: Server.error_message("28P01", "password authentication failed for user \"#{user}\""),	13✔
121	log_message: "Exchange error: password authentication failed for user \"#{user}\"",	13✔
122	auth_error: true
123	}
124	end
125
126	defp process({:error, :auth_error, {:timeout, _message}, _user}, _context) do
127	%{	×
128	error: Server.error_message("08006", "connection failure during authentication")
129	}
130	end
131
132	defp process({:error, :auth_error, {:unexpected_message, _details}, _user}, _context) do
133	%{	×
134	error: Server.error_message("08P01", "protocol violation during authentication")
135	}
136	end
137
138	defp process({:error, :auth_error, {:decode_error, error}}, context) do
139	auth_stage = auth_context_description(context)	×
140
141	%{	×
142	error: Server.error_message("08P01", "protocol violation during authentication"),
143	log_message: "#{auth_stage} auth decode error: #{inspect(error)}"	×
144	}
145	end
146
147	defp process({:error, :auth_error, {:unexpected_message, other}}, context) do
148	auth_stage = auth_context_description(context)	×
149
150	%{	×
151	error: Server.error_message("08P01", "protocol violation during authentication"),
152	log_message: "#{auth_stage} auth unexpected message: #{inspect(other)}"	×
153	}
154	end
155
156	defp process({:error, :auth_error, {:timeout, _}}, context) do
157	log_message =	×
158	case context do
159	:auth_md5_wait -> "Timeout while waiting for MD5 password"	×
160	:auth_scram_first_wait -> "Timeout while waiting for first SCRAM message"	×
161	:auth_scram_final_wait -> "Timeout while waiting for final SCRAM message"	×
162	_ -> "Authentication timeout"	×
163	end
164
165	%{	×
166	error: Server.error_message("08006", "connection failure during authentication"),
167	log_message: log_message
168	}
169	end
170
171	defp process({:error, {:invalid_user_info, {:invalid_format, {user, db_name}}}}, _context) do
172	%{	1✔
173	error:
174	Server.error_message(
175	"XX000",
176	"Authentication error, reason: \"Invalid format for user or db_name\""
177	),
178	log_message: "Invalid format for user or db_name: #{inspect({user, db_name})}"
179	}
180	end
181
182	defp process({:error, :auth_error, reason}, _context) do
183	%{	×
184	error: Server.error_message("XX000", "Authentication error, reason: #{inspect(reason)}")
185	}
186	end
187
188	defp process({:error, :max_clients_reached}, _context) do
189	%{	1✔
190	error: Server.error_message("XX000", "Max client connections reached"),
191	log_message: "Max client connections reached"
192	}
193	end
194
195	defp process({:error, :max_clients_reached_session}, _context) do
196	message =	1✔
197	"MaxClientsInSessionMode: max clients reached - in Session mode max clients are limited to pool_size"
198
199	%{	1✔
200	error: Server.error_message("XX000", message),
201	log_message: message
202	}
203	end
204
205	defp process({:error, :max_pools_reached}, _context) do
206	%{	×
207	error: Server.error_message("XX000", "Max pools count reached"),
208	log_message: "Max pools count reached"
209	}
210	end
211
212	defp process({:error, :db_handler_exited, pid, reason}, _context) do
213	message =	×
214	case reason do
215	:db_termination -> "Connection to database closed. Check logs for more information"	×
216	_ -> "DbHandler exited. Check logs for more information."	×
217	end
218
219	%{	×
220	error: Server.error_message("XX000", message),
221	log_message: "DbHandler #{inspect(pid)} exited #{inspect(reason)}"
222	}
223	end
224
225	defp process({:error, :session_timeout}, _context) do
226	message =	×
227	"MaxClientsInSessionMode: max clients reached - in Session mode max clients are limited to pool_size"
228
229	%{	×
230	error: Server.error_message("XX000", message),
231	log_message: message
232	}
233	end
234
235	defp process({:error, :transaction_timeout}, _context) do
236	message = "Unable to check out connection from the pool due to timeout"	×
237
238	%{	×
239	error: Server.error_message("XX000", message),
240	log_message: message
241	}
242	end
243
244	defp process({:error, :subscribe_retries_exhausted}, _context) do
245	message = "Failed to subscribe to tenant after multiple retries. Terminating."	×
246
247	%{	×
248	error: Server.error_message("XX000", message),
249	log_message: message
250	}
251	end
252
253	defp process({:error, :circuit_breaker_open, operation, blocked_until}, _context) do
254	explanation = Supavisor.CircuitBreaker.explanation(operation)	3✔
255	message = "Circuit breaker open for operation: #{operation}, blocked until: #{blocked_until}"	3✔
256
257	%{	3✔
258	error: Server.error_message("XX000", "Circuit breaker open: #{explanation}"),	3✔
259	log_message: message
260	}
261	end
262
263	defp process(error, context) do
264	message =	×
265	case context do
266	nil -> "Internal error: #{inspect(error)}"	×
267	context -> "Internal error (#{context}): #{inspect(error)}"	×
268	end
269
270	%{	×
271	error: Server.error_message("XX000", message),
272	log_message: message
273	}
274	end
275
276	defp auth_context_description(:auth_md5_wait), do: "MD5"	×
277	defp auth_context_description(:auth_scram_first_wait), do: "SCRAM first"	×
278	defp auth_context_description(:auth_scram_final_wait), do: "SCRAM final"	×
279	defp auth_context_description(_), do: "Unknown"	×
280	end

supabase / supavisor / 19911268462

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous