22145415097

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

supabase / supavisor / 22145415097

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