19741053250

Committed 27 Nov 2025 03:23PM UTC coverage: 63.26% (+2.0%) from 61.246%

Build # 19741053250

Build Type

Pull #744

github

Committed by

web-flow

Commit Message

Merge 15c0d1b2b into 0224a24c8

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

Coverage Stats

651 of 845 new or added lines in 25 files covered. (77.04%)

22 existing lines in 8 files now uncovered.

1851 of 2926 relevant lines covered (63.26%)

4207.28 hits per line

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

44.59

/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)	26✔
19	message = Map.fetch!(error_actions, :error)	26✔
20	log_message = Map.get(error_actions, :log_message)	26✔
21	send_ready_for_query = Map.get(error_actions, :send_ready_for_query, false)	26✔
22	auth_error = Map.get(error_actions, :auth_error, false)	26✔
23
24	if log_message do	26✔
25	Logger.error("ClientHandler: #{log_message}", auth_error: auth_error)	26✔
26	end
27
28	if send_ready_for_query do	26✔
29	HandlerHelpers.sock_send(sock, [message, Server.ready_for_query()])	4✔
30	else
31	HandlerHelpers.sock_send(sock, message)	22✔
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
NEW 70	message_text = "prepared statement #{inspect(name)} does not exist"	×
71
NEW 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
NEW 79	message_text = "prepared statement #{inspect(name)} already exists"	×
80
NEW 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
NEW 88	%{	×
89	error: Server.error_message("XX000", "SSL connection is required"),
NEW 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
NEW 95	message = "Address not in tenant allow_list: " <> inspect(addr)	×
96
NEW 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
NEW 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
NEW 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
NEW 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
NEW 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
NEW 139	auth_stage = auth_context_description(context)	×
140
NEW 141	%{	×
142	error: Server.error_message("08P01", "protocol violation during authentication"),
NEW 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
NEW 148	auth_stage = auth_context_description(context)	×
149
NEW 150	%{	×
151	error: Server.error_message("08P01", "protocol violation during authentication"),
NEW 152	log_message: "#{auth_stage} auth unexpected message: #{inspect(other)}"	×
153	}
154	end
155
156	defp process({:error, :auth_error, {:timeout, _}}, context) do
NEW 157	log_message =	×
158	case context do
NEW 159	:auth_md5_wait -> "Timeout while waiting for MD5 password"	×
NEW 160	:auth_scram_first_wait -> "Timeout while waiting for first SCRAM message"	×
NEW 161	:auth_scram_final_wait -> "Timeout while waiting for final SCRAM message"	×
NEW 162	_ -> "Authentication timeout"	×
163	end
164
NEW 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
NEW 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
NEW 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 =	1✔
214	case reason do
NEW 215	:db_termination -> "Connection to database closed. Check logs for more information"	×
216	_ -> "DbHandler exited. Check logs for more information."	1✔
217	end
218
219	%{	1✔
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
NEW 226	message =	×
227	"MaxClientsInSessionMode: max clients reached - in Session mode max clients are limited to pool_size"
228
NEW 229	%{	×
230	error: Server.error_message("XX000", message),
231	log_message: message
232	}
233	end
234
235	defp process({:error, :transaction_timeout}, _context) do
NEW 236	message = "Unable to check out connection from the pool due to timeout"	×
237
NEW 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
NEW 245	message = "Failed to subscribe to tenant after multiple retries. Terminating."	×
246
NEW 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
NEW 264	message =	×
265	case context do
NEW 266	nil -> "Internal error: #{inspect(error)}"	×
NEW 267	context -> "Internal error (#{context}): #{inspect(error)}"	×
268	end
269
NEW 270	%{	×
271	error: Server.error_message("XX000", message),
272	log_message: message
273	}
274	end
275
NEW 276	defp auth_context_description(:auth_md5_wait), do: "MD5"	×
NEW 277	defp auth_context_description(:auth_scram_first_wait), do: "SCRAM first"	×
NEW 278	defp auth_context_description(:auth_scram_final_wait), do: "SCRAM final"	×
NEW 279	defp auth_context_description(_), do: "Unknown"	×
280	end

supabase / supavisor / 19741053250

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