19715527181

Committed 26 Nov 2025 07:41PM UTC coverage: 74.358% (+13.1%) from 61.246%

Build # 19715527181

Build Type

Pull #744

github

Committed by

web-flow

Commit Message

Merge 9e08c361d into 0224a24c8

Pull Request Pull Request #744: fix(defrag): improve statems, caching, logs, circuit breaking

Run Details

771 of 916 new or added lines in 23 files covered. (84.17%)

3 existing lines in 2 files now uncovered.

2404 of 3233 relevant lines covered (74.36%)

4254.35 hits per line

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

27.03

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

supabase / supavisor / 19715527181

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