22738398145

Committed 05 Mar 2026 09:50PM UTC coverage: 43.243%. First build

Build # 22738398145

Build Type

Pull #865

github

Committed by

web-flow

Commit Message

Merge 910c5a91f into 609d94708

Pull Request Pull Request #865: fix: use a Dynamic Supervisor for starting proxy connections

Coverage Stats

5 of 85 new or added lines in 6 files covered. (5.88%)

1328 of 3071 relevant lines covered (43.24%)

3949.25 hits per line

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

24.05

/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, _user}, context) do
    %{
      error: Server.error_message("08006", "connection failure during authentication"),
      log_message:
        "Timeout while waiting for message in state #{auth_context_description(context)}"
    }
  end

  defp process({:error, :auth_error, {:unexpected_message, details}, _user}, context) do
    %{
      error: Server.error_message("08P01", "protocol violation during authentication"),
      log_message:
        "#{auth_context_description(context)} unexpected message during authentication: #{inspect(details)}"
    }
  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
    %{
      error: Server.error_message("08006", "connection failure during authentication"),
      log_message:
        "Timeout while waiting for message in state #{auth_context_description(context)}"
    }
  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
    message =
      "Authentication error, reason: #{inspect(reason)}, context: #{auth_context_description(context)}"

    %{
      error: Server.error_message("XX000", message),
      log_message: message
    }
  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_proxy_connections_reached}, _context) do
    %{
      error: Server.error_message("XX000", "Max client connections reached"),
      log_message: "Max proxy connections reached"
    }
  end

  defp process({:error, :failed_to_start_proxy_connection}, _context) do
    %{
      error:
        Server.error_message(
          "XX000",
          "Failed to start proxy connection. Check logs for more information"
        ),
      log_message: "Failed to start proxy connection"
    }
  end

  defp process({:error, :proxy_supervisor_unavailable}, _context) do
    %{
      error: Server.error_message("XX000", "Proxy supervisor unavailable"),
      log_message: "Proxy supervisor unavailable after retries"
    }
  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(:handshake), do: "Handshake"
  defp auth_context_description(:auth_md5_wait), do: "MD5"
  defp auth_context_description(:auth_password_wait), do: "PASSWORD"
  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)	6✔
19	message = Map.fetch!(error_actions, :error)	6✔
20	log_message = Map.get(error_actions, :log_message)	6✔
21	send_ready_for_query = Map.get(error_actions, :send_ready_for_query, false)	6✔
22	auth_error = Map.get(error_actions, :auth_error, false)	6✔
23
24	if log_message do	6✔
25	Logger.error("ClientHandler: #{log_message}", auth_error: auth_error)	6✔
26	end
27
28	if send_ready_for_query do	6✔
29	HandlerHelpers.sock_send(sock, [message, Server.ready_for_query()])	4✔
30	else
31	HandlerHelpers.sock_send(sock, message)	2✔
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	%{	×
120	error: Server.error_message("28P01", "password authentication failed for user \"#{user}\""),	×
121	log_message: "Exchange error: password authentication failed for user \"#{user}\"",	×
122	auth_error: true
123	}
124	end
125
126	defp process({:error, :auth_error, :timeout, _user}, context) do
127	%{	×
128	error: Server.error_message("08006", "connection failure during authentication"),
129	log_message:
130	"Timeout while waiting for message in state #{auth_context_description(context)}"	×
131	}
132	end
133
134	defp process({:error, :auth_error, {:unexpected_message, details}, _user}, context) do
135	%{	×
136	error: Server.error_message("08P01", "protocol violation during authentication"),
137	log_message:
138	"#{auth_context_description(context)} unexpected message during authentication: #{inspect(details)}"	×
139	}
140	end
141
142	defp process({:error, :auth_error, {:decode_error, error}}, context) do
143	auth_stage = auth_context_description(context)	×
144
145	%{	×
146	error: Server.error_message("08P01", "protocol violation during authentication"),
147	log_message: "#{auth_stage} auth decode error: #{inspect(error)}"	×
148	}
149	end
150
151	defp process({:error, :auth_error, {:unexpected_message, other}}, context) do
152	auth_stage = auth_context_description(context)	×
153
154	%{	×
155	error: Server.error_message("08P01", "protocol violation during authentication"),
156	log_message: "#{auth_stage} auth unexpected message: #{inspect(other)}"	×
157	}
158	end
159
160	defp process({:error, :auth_error, :timeout}, context) do
161	%{	×
162	error: Server.error_message("08006", "connection failure during authentication"),
163	log_message:
164	"Timeout while waiting for message in state #{auth_context_description(context)}"	×
165	}
166	end
167
168	defp process({:error, {:invalid_user_info, {:invalid_format, {user, db_name}}}}, _context) do
169	%{	×
170	error:
171	Server.error_message(
172	"XX000",
173	"Authentication error, reason: \"Invalid format for user or db_name\""
174	),
175	log_message: "Invalid format for user or db_name: #{inspect({user, db_name})}"
176	}
177	end
178
179	defp process({:error, :auth_error, reason}, context) do
180	message =	×
181	"Authentication error, reason: #{inspect(reason)}, context: #{auth_context_description(context)}"	×
182
183	%{	×
184	error: Server.error_message("XX000", message),
185	log_message: message
186	}
187	end
188
189	defp process({:error, :max_clients_reached}, _context) do
190	%{	×
191	error: Server.error_message("XX000", "Max client connections reached"),
192	log_message: "Max client connections reached"
193	}
194	end
195
196	defp process({:error, :max_proxy_connections_reached}, _context) do
NEW 197	%{	×
198	error: Server.error_message("XX000", "Max client connections reached"),
199	log_message: "Max proxy connections reached"
200	}
201	end
202
203	defp process({:error, :failed_to_start_proxy_connection}, _context) do
NEW 204	%{	×
205	error:
206	Server.error_message(
207	"XX000",
208	"Failed to start proxy connection. Check logs for more information"
209	),
210	log_message: "Failed to start proxy connection"
211	}
212	end
213
214	defp process({:error, :proxy_supervisor_unavailable}, _context) do
NEW 215	%{	×
216	error: Server.error_message("XX000", "Proxy supervisor unavailable"),
217	log_message: "Proxy supervisor unavailable after retries"
218	}
219	end
220
221	defp process({:error, :max_clients_reached_session}, _context) do
222	message =	×
223	"MaxClientsInSessionMode: max clients reached - in Session mode max clients are limited to pool_size"
224
225	%{	×
226	error: Server.error_message("XX000", message),
227	log_message: message
228	}
229	end
230
231	defp process({:error, :max_pools_reached}, _context) do
232	%{	×
233	error: Server.error_message("XX000", "Max pools count reached"),
234	log_message: "Max pools count reached"
235	}
236	end
237
238	defp process({:error, :db_handler_exited, pid, reason}, _context) do
239	message =	×
240	case reason do
241	:db_termination -> "Connection to database closed. Check logs for more information"	×
242	_ -> "DbHandler exited. Check logs for more information."	×
243	end
244
245	%{	×
246	error: Server.error_message("XX000", message),
247	log_message: "DbHandler #{inspect(pid)} exited #{inspect(reason)}"
248	}
249	end
250
251	defp process({:error, :session_timeout}, _context) do
252	message =	×
253	"MaxClientsInSessionMode: max clients reached - in Session mode max clients are limited to pool_size"
254
255	%{	×
256	error: Server.error_message("XX000", message),
257	log_message: message
258	}
259	end
260
261	defp process({:error, :transaction_timeout}, _context) do
262	message = "Unable to check out connection from the pool due to timeout"	×
263
264	%{	×
265	error: Server.error_message("XX000", message),
266	log_message: message
267	}
268	end
269
270	defp process({:error, :subscribe_retries_exhausted}, _context) do
271	message = "Failed to subscribe to tenant after multiple retries. Terminating."	×
272
273	%{	×
274	error: Server.error_message("XX000", message),
275	log_message: message
276	}
277	end
278
279	defp process({:error, :circuit_breaker_open, operation, blocked_until}, _context) do
280	explanation = Supavisor.CircuitBreaker.explanation(operation)	×
281	message = "Circuit breaker open for operation: #{operation}, blocked until: #{blocked_until}"	×
282
283	%{	×
284	error: Server.error_message("XX000", "Circuit breaker open: #{explanation}"),	×
285	log_message: message
286	}
287	end
288
289	defp process(error, context) do
290	message =	×
291	case context do
292	nil -> "Internal error: #{inspect(error)}"	×
293	context -> "Internal error (#{context}): #{inspect(error)}"	×
294	end
295
296	%{	×
297	error: Server.error_message("XX000", message),
298	log_message: message
299	}
300	end
301
302	defp auth_context_description(:handshake), do: "Handshake"	×
303	defp auth_context_description(:auth_md5_wait), do: "MD5"	×
304	defp auth_context_description(:auth_password_wait), do: "PASSWORD"	×
305	defp auth_context_description(:auth_scram_first_wait), do: "SCRAM first"	×
306	defp auth_context_description(:auth_scram_final_wait), do: "SCRAM final"	×
307	defp auth_context_description(_), do: "Unknown"	×
308	end

supabase / supavisor / 22738398145

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