19397204070

Committed 15 Nov 2025 11:37PM UTC coverage: 62.812% (+1.6%) from 61.246%

Build # 19397204070

Build Type

Pull #744

github

Committed by

web-flow

Commit Message

Merge a0b83d534 into 0224a24c8

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

Run Details

595 of 787 new or added lines in 22 files covered. (75.6%)

18 existing lines in 5 files now uncovered.

1814 of 2888 relevant lines covered (62.81%)

4262.63 hits per line

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

43.48

/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_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
    %{
      error: Server.error_message("XX000", "DbHandler exited"),
      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 process 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)	14✔
19	message = Map.fetch!(error_actions, :error)	14✔
20	log_message = Map.get(error_actions, :log_message)	14✔
21	send_ready_for_query = Map.get(error_actions, :send_ready_for_query, false)	14✔
22	auth_error = Map.get(error_actions, :auth_error, false)	14✔
23
24	if log_message do	14✔
25	Logger.error("ClientHandler: #{log_message}", auth_error: auth_error)	14✔
26	end
27
28	if send_ready_for_query do	14✔
29	HandlerHelpers.sock_send(sock, [message, Server.ready_for_query()])	4✔
30	else
31	HandlerHelpers.sock_send(sock, message)	10✔
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	%{	3✔
120	error: Server.error_message("28P01", "password authentication failed for user \"#{user}\""),	3✔
121	log_message: "Exchange error: password authentication failed for user \"#{user}\"",	3✔
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_pools_reached}, _context) do
NEW 196	%{	×
197	error: Server.error_message("XX000", "Max pools count reached"),
198	log_message: "Max pools count reached"
199	}
200	end
201
202	defp process({:error, :db_handler_exited, pid, reason}, _context) do
NEW 203	%{	×
204	error: Server.error_message("XX000", "DbHandler exited"),
205	log_message: "DbHandler #{inspect(pid)} exited #{inspect(reason)}"
206	}
207	end
208
209	defp process({:error, :session_timeout}, _context) do
210	message =	1✔
211	"MaxClientsInSessionMode: max clients reached - in Session mode max clients are limited to pool_size"
212
213	%{	1✔
214	error: Server.error_message("XX000", message),
215	log_message: message
216	}
217	end
218
219	defp process({:error, :transaction_timeout}, _context) do
NEW 220	message = "Unable to check out process from the pool due to timeout"	×
221
NEW 222	%{	×
223	error: Server.error_message("XX000", message),
224	log_message: message
225	}
226	end
227
228	defp process({:error, :subscribe_retries_exhausted}, _context) do
NEW 229	message = "Failed to subscribe to tenant after multiple retries. Terminating."	×
230
NEW 231	%{	×
232	error: Server.error_message("XX000", message),
233	log_message: message
234	}
235	end
236
237	defp process({:error, :circuit_breaker_open, operation, blocked_until}, _context) do
238	explanation = Supavisor.CircuitBreaker.explanation(operation)	2✔
239	message = "Circuit breaker open for operation: #{operation}, blocked until: #{blocked_until}"	2✔
240
241	%{	2✔
242	error: Server.error_message("XX000", "Circuit breaker open: #{explanation}"),	2✔
243	log_message: message
244	}
245	end
246
247	defp process(error, context) do
NEW 248	message =	×
249	case context do
NEW 250	nil -> "Internal error: #{inspect(error)}"	×
NEW 251	context -> "Internal error (#{context}): #{inspect(error)}"	×
252	end
253
NEW 254	%{	×
255	error: Server.error_message("XX000", message),
256	log_message: message
257	}
258	end
259
NEW 260	defp auth_context_description(:auth_md5_wait), do: "MD5"	×
NEW 261	defp auth_context_description(:auth_scram_first_wait), do: "SCRAM first"	×
NEW 262	defp auth_context_description(:auth_scram_final_wait), do: "SCRAM final"	×
NEW 263	defp auth_context_description(_), do: "Unknown"	×
264	end

supabase / supavisor / 19397204070

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