19506942358

Committed 19 Nov 2025 03:34PM UTC coverage: 74.52% (+13.3%) from 61.246%

Build # 19506942358

Build Type

Pull #744

github

Committed by

web-flow

Commit Message

Merge 9a95963fc into 0224a24c8

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

Coverage Stats

750 of 891 new or added lines in 23 files covered. (84.18%)

3 existing lines in 2 files now uncovered.

2407 of 3230 relevant lines covered (74.52%)

4256.17 hits per line

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

27.78

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

supabase / supavisor / 19506942358

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