19370957114

Committed 14 Nov 2025 04:30PM UTC coverage: 62.682% (+1.4%) from 61.246%

Build # 19370957114

Build Type

Pull #744

github

Committed by

web-flow

Commit Message

Merge fd252a012 into 0224a24c8

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

Run Details

592 of 785 new or added lines in 22 files covered. (75.41%)

18 existing lines in 5 files now uncovered.

1809 of 2886 relevant lines covered (62.68%)

4508.83 hits per line

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

81.82

/lib/supavisor/client_handler/protocol_helpers.ex

defmodule Supavisor.ClientHandler.ProtocolHelpers do
  @moduledoc """
  Protocol parsing and analysis helpers for client connections.

  This module contains pure business logic for:
  - Startup packet parsing and validation
  - Protocol message analysis and routing
  - Client packet processing
  - Protocol data transformation utilities

  All functions are pure (other than potential logs).
  """

  require Logger

  alias Supavisor.{
    FeatureFlag,
    HandlerHelpers,
    Helpers,
    Protocol.MessageStreamer,
    Protocol.Server
  }

  require Supavisor.Protocol.PreparedStatements, as: PreparedStatements

  @type startup_result ::
          {:ok, {atom(), {String.t(), String.t(), String.t() | nil, String.t() | nil}},
           String.t(), atom() | nil}
          | {:error, term()}

  @type packet_processing_result ::
          {:ok, MessageStreamer.stream_state(), [PreparedStatements.handled_pkt()] | binary()}
          | {:error, :max_prepared_statements}
          | {:error, :max_prepared_statements_memory}
          | {:error, :prepared_statement_on_simple_query}
          | {:error, :duplicate_prepared_statement, PreparedStatements.statement_name()}
          | {:error, :prepared_statement_not_found, PreparedStatements.statement_name()}

  ## Startup Packet Processing

  @doc """
  Parses and validates startup packet data.

  Returns parsed user info, application name, and log level if successful.
  """
  @spec parse_startup_packet(binary()) :: startup_result()
  def parse_startup_packet(bin) do
    case Server.decode_startup_packet(bin) do
      {:ok, hello} ->
        Logger.debug("ClientHandler: Client startup message: #{inspect(hello)}")

        case extract_and_validate_user_info(hello.payload) do
          {:ok, {type, {user, tenant_or_alias, db_name, search_path}}} ->
            app_name = normalize_app_name(hello.payload["application_name"])
            log_level = extract_log_level(hello)
            {:ok, {type, {user, tenant_or_alias, db_name, search_path}}, app_name, log_level}

          {:error, reason} ->
            {:error, {:invalid_user_info, reason}}
        end

      {:error, error} ->
        {:error, {:decode_error, error}}
    end
  end

  @doc """
  Extracts and validates user information from startup payload.
  """
  @spec extract_and_validate_user_info(map()) ::
          {:ok, {atom(), {String.t(), String.t(), String.t() | nil, String.t() | nil}}}
          | {:error, term()}
  def extract_and_validate_user_info(payload) do
    {type, {user, tenant_or_alias, db_name}} = HandlerHelpers.parse_user_info(payload)

    if Helpers.validate_name(user) and Helpers.validate_name(db_name) do
      search_path = payload["options"]["--search_path"]
      {:ok, {type, {user, tenant_or_alias, db_name, search_path}}}
    else
      {:error, {:invalid_format, {user, db_name}}}
    end
  end

  ## Client Packet Processing

  @doc """
  Processes client packets for prepared statements based on mode and feature flags.

  Returns processed packets or passes through unchanged based on configuration.
  """
  @spec process_client_packets(binary(), atom(), map()) :: packet_processing_result()
  def process_client_packets(
        bin,
        :transaction,
        %{tenant_feature_flags: tenant_feature_flags} = data
      ) do
    if FeatureFlag.enabled?(tenant_feature_flags, "named_prepared_statements") do
      MessageStreamer.handle_packets(data.stream_state, bin)
    else
      {:ok, data.stream_state, bin}
    end
  end

  def process_client_packets(bin, _mode, data) do
    {:ok, data.stream_state, bin}
  end

  ## Protocol Utilities

  @doc """
  Normalizes application name from client connection.

  Returns sanitized string or default "Supavisor" for invalid names.
  """
  @spec normalize_app_name(any()) :: String.t()
  def normalize_app_name(name) when is_binary(name), do: name
  def normalize_app_name(nil), do: "Supavisor"

  def normalize_app_name(name) do
    Logger.debug("ClientHandler: Invalid application name #{inspect(name)}")
    "Supavisor"
  end

  @doc """
  Extracts log level from startup message options.

  Returns atom log level or nil if not specified or invalid.
  """
  @spec extract_log_level(map()) :: atom() | nil
  def extract_log_level(%{"payload" => %{"options" => options}}) do
    level = options["log_level"] && String.to_existing_atom(options["log_level"])

    if level in [:debug, :info, :notice, :warning, :error] do
      level
    else
      nil
    end
  end

  def extract_log_level(_), do: nil
end

1	defmodule Supavisor.ClientHandler.ProtocolHelpers do
2	@moduledoc """
3	Protocol parsing and analysis helpers for client connections.
4
5	This module contains pure business logic for:
6	- Startup packet parsing and validation
7	- Protocol message analysis and routing
8	- Client packet processing
9	- Protocol data transformation utilities
10
11	All functions are pure (other than potential logs).
12	"""
13
14	require Logger
15
16	alias Supavisor.{
17	FeatureFlag,
18	HandlerHelpers,
19	Helpers,
20	Protocol.MessageStreamer,
21	Protocol.Server
22	}
23
24	require Supavisor.Protocol.PreparedStatements, as: PreparedStatements
25
26	@type startup_result ::
27	{:ok, {atom(), {String.t(), String.t(), String.t() \| nil, String.t() \| nil}},
28	String.t(), atom() \| nil}
29	\| {:error, term()}
30
31	@type packet_processing_result ::
32	{:ok, MessageStreamer.stream_state(), [PreparedStatements.handled_pkt()] \| binary()}
33	\| {:error, :max_prepared_statements}
34	\| {:error, :max_prepared_statements_memory}
35	\| {:error, :prepared_statement_on_simple_query}
36	\| {:error, :duplicate_prepared_statement, PreparedStatements.statement_name()}
37	\| {:error, :prepared_statement_not_found, PreparedStatements.statement_name()}
38
39	## Startup Packet Processing
40
41	@doc """
42	Parses and validates startup packet data.
43
44	Returns parsed user info, application name, and log level if successful.
45	"""
46	@spec parse_startup_packet(binary()) :: startup_result()
47	def parse_startup_packet(bin) do
48	case Server.decode_startup_packet(bin) do	622✔
49	{:ok, hello} ->
50	Logger.debug("ClientHandler: Client startup message: #{inspect(hello)}")	616✔
51
52	case extract_and_validate_user_info(hello.payload) do	616✔
53	{:ok, {type, {user, tenant_or_alias, db_name, search_path}}} ->
54	app_name = normalize_app_name(hello.payload["application_name"])	615✔
55	log_level = extract_log_level(hello)	615✔
56	{:ok, {type, {user, tenant_or_alias, db_name, search_path}}, app_name, log_level}	615✔
57
58	{:error, reason} ->	1✔
59	{:error, {:invalid_user_info, reason}}
60	end
61
62	{:error, error} ->	6✔
63	{:error, {:decode_error, error}}
64	end
65	end
66
67	@doc """
68	Extracts and validates user information from startup payload.
69	"""
70	@spec extract_and_validate_user_info(map()) ::
71	{:ok, {atom(), {String.t(), String.t(), String.t() \| nil, String.t() \| nil}}}
72	\| {:error, term()}
73	def extract_and_validate_user_info(payload) do
74	{type, {user, tenant_or_alias, db_name}} = HandlerHelpers.parse_user_info(payload)	616✔
75
76	if Helpers.validate_name(user) and Helpers.validate_name(db_name) do	616✔
77	search_path = payload["options"]["--search_path"]	615✔
78	{:ok, {type, {user, tenant_or_alias, db_name, search_path}}}
79	else
80	{:error, {:invalid_format, {user, db_name}}}
81	end
82	end
83
84	## Client Packet Processing
85
86	@doc """
87	Processes client packets for prepared statements based on mode and feature flags.
88
89	Returns processed packets or passes through unchanged based on configuration.
90	"""
91	@spec process_client_packets(binary(), atom(), map()) :: packet_processing_result()
92	def process_client_packets(
93	bin,
94	:transaction,
95	%{tenant_feature_flags: tenant_feature_flags} = data
96	) do
97	if FeatureFlag.enabled?(tenant_feature_flags, "named_prepared_statements") do	9,249✔
98	MessageStreamer.handle_packets(data.stream_state, bin)	9,224✔
99	else
100	{:ok, data.stream_state, bin}	25✔
101	end
102	end
103
104	def process_client_packets(bin, _mode, data) do
105	{:ok, data.stream_state, bin}	5,974✔
106	end
107
108	## Protocol Utilities
109
110	@doc """
111	Normalizes application name from client connection.
112
113	Returns sanitized string or default "Supavisor" for invalid names.
114	"""
115	@spec normalize_app_name(any()) :: String.t()
116	def normalize_app_name(name) when is_binary(name), do: name	439✔
117	def normalize_app_name(nil), do: "Supavisor"	176✔
118
119	def normalize_app_name(name) do
NEW 120	Logger.debug("ClientHandler: Invalid application name #{inspect(name)}")	×
121	"Supavisor"
122	end
123
124	@doc """
125	Extracts log level from startup message options.
126
127	Returns atom log level or nil if not specified or invalid.
128	"""
129	@spec extract_log_level(map()) :: atom() \| nil
130	def extract_log_level(%{"payload" => %{"options" => options}}) do
NEW 131	level = options["log_level"] && String.to_existing_atom(options["log_level"])	×
132
NEW 133	if level in [:debug, :info, :notice, :warning, :error] do	×
NEW 134	level	×
135	else
136	nil
137	end
138	end
139
140	def extract_log_level(_), do: nil	615✔
141	end

supabase / supavisor / 19370957114

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