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

85.23

/lib/supavisor/secret_checker.ex

defmodule Supavisor.SecretChecker do
  @moduledoc false

  use GenServer
  require Logger

  alias Supavisor.Helpers

  @interval :timer.seconds(15)

  def start_link(args) do
    name = {:via, Registry, {Supavisor.Registry.Tenants, {:secret_checker, args.id}}}
    GenServer.start_link(__MODULE__, args, name: name)
  end

  @spec get_secrets(Supavisor.id()) ::
          {:ok, {method :: atom(), Supavisor.secrets()}} | {:error, :not_started}
  def get_secrets(id) do
    erpc_call_node(id, fn ->
      case Registry.lookup(Supavisor.Registry.Tenants, {:secret_checker, id}) do
        [] ->
          {:error, :not_started}

        [{pid, _}] ->
          GenServer.call(pid, :get_secrets)
      end
    end)
  end

  @spec update_credentials(Supavisor.id(), String.t(), (-> String.t())) ::
          :ok | {:error, :not_started}
  def update_credentials(id, new_user, password_fn) do
    erpc_call_node(id, fn ->
      case Registry.lookup(Supavisor.Registry.Tenants, {:secret_checker, id}) do
        [] ->
          {:error, :not_started}

        [{pid, _}] ->
          GenServer.call(pid, {:update_credentials, new_user, password_fn})
      end
    end)
  end

  def init(args) do
    Logger.debug("SecretChecker: Starting secret checker")
    {{_type, tenant_external_id}, pool_user, _mode, db_name, _search_path} = args.id

    # Get tenant and manager user to build auth config
    tenant = Supavisor.Tenants.get_tenant_by_external_id(tenant_external_id)
    manager_secrets = Supavisor.Tenants.get_manager_user_cache(tenant_external_id)

    auth =
      if manager_secrets do
        %{
          host: String.to_charlist(tenant.db_host),
          sni_hostname: if(tenant.sni_hostname != nil, do: to_charlist(tenant.sni_hostname)),
          port: tenant.db_port,
          user: manager_secrets.db_user,
          manager_secrets: manager_secrets,
          auth_query: tenant.auth_query,
          database: if(db_name != nil, do: db_name, else: tenant.db_database),
          password: fn -> manager_secrets.db_password end,
          application_name: "Supavisor",
          ip_version: Supavisor.Helpers.ip_version(tenant.ip_version, tenant.db_host),
          upstream_ssl: tenant.upstream_ssl,
          upstream_verify: tenant.upstream_verify,
          upstream_tls_ca: Supavisor.Helpers.upstream_cert(tenant.upstream_tls_ca)
        }
      else
        nil
      end

    state = %{
      tenant: tenant_external_id,
      auth: auth,
      user: pool_user,
      ttl: :timer.hours(24),
      conn: nil,
      check_ref: check()
    }

    Logger.metadata(project: tenant, user: pool_user)
    {:ok, state, {:continue, :init_conn}}
  end

  def handle_continue(:init_conn, %{auth: nil} = state) do
    # No auth config (require_user: true tenant), skip connection setup
    {:noreply, state}
  end

  def handle_continue(:init_conn, %{auth: auth} = state) do
    {:ok, conn} = start_postgrex_connection(auth)
    {:noreply, %{state | conn: conn}}
  end

  def handle_info(:check, %{auth: nil} = state) do
    {:noreply, %{state | check_ref: check()}}
  end

  def handle_info(:check, state) do
    check_secrets(state.user, state)
    {:noreply, %{state | check_ref: check()}}
  end

  def handle_info(msg, state) do
    Logger.error("Unexpected message: #{inspect(msg)}")
    {:noreply, state}
  end

  def terminate(_, state) do
    :gen_statem.stop(state.conn)
    :ok
  end

  def check(interval \\ @interval),
    do: Process.send_after(self(), :check, interval + jitter())

  def check_secrets(user, %{auth: auth, conn: conn} = state) do
    alias Supavisor.ClientHandler.Auth

    case Helpers.get_user_secret(conn, auth.auth_query, user) do
      {:ok, secret} ->
        method =
          case secret do
            %Auth.MD5Secrets{} -> :auth_query_md5
            %Auth.SASLSecrets{} -> :auth_query
          end

        update_cache =
          case Supavisor.SecretCache.get_validation_secrets(state.tenant, state.user) do
            {:ok, {old_method, old_secrets_fn}} ->
              method != old_method or
                Map.delete(secret, :client_key) != Map.delete(old_secrets_fn.(), :client_key)

            _other ->
              true
          end

        if update_cache do
          Logger.info("Secrets changed or not present, updating cache")

          Supavisor.SecretCache.put_validation_secrets(state.tenant, state.user, method, fn ->
            secret
          end)
        end

        {:ok, {method, fn -> secret end}}

      other ->
        Logger.error("Failed to get secret: #{inspect(other)}")
    end
  end

  def handle_call(:get_secrets, _from, %{auth: nil} = state) do
    {:reply, {:error, :no_auth_config}, state}
  end

  def handle_call(:get_secrets, _from, state) do
    {:reply, check_secrets(state.user, state), state}
  end

  def handle_call({:update_credentials, new_user, password_fn}, _from, state) do
    Logger.info("SecretChecker: changing auth_query user to #{new_user}")

    new_auth = %{
      state.auth
      | user: new_user,
        password: password_fn
    }

    {:ok, new_conn} = start_postgrex_connection(new_auth)

    old_conn = state.conn
    Process.unlink(old_conn)

    Task.start(fn ->
      try do
        GenServer.stop(old_conn, :normal, 5_000)
      catch
        :exit, _ -> Process.exit(old_conn, :kill)
      end
    end)

    # Clear the secrets cache for this tenant/user
    tenant = state.tenant
    user = state.user
    Cachex.del(Supavisor.Cache, {:secrets, tenant, user})

    Logger.info("SecretChecker: Successfully changed auth_query user")
    {:reply, :ok, %{state | auth: new_auth, conn: new_conn}}
  end

  defp start_postgrex_connection(auth) do
    ssl_opts =
      if auth.upstream_ssl and auth.upstream_verify == :peer do
        [
          verify: :verify_peer,
          cacerts: [Helpers.upstream_cert(auth.upstream_tls_ca)],
          server_name_indication: auth.host,
          customize_hostname_check: [{:match_fun, fn _, _ -> true end}]
        ]
      else
        [
          verify: :verify_none
        ]
      end

    Postgrex.start_link(
      hostname: auth.host,
      port: auth.port,
      database: auth.database,
      password: auth.password.(),
      username: auth.user,
      parameters: [application_name: "Supavisor (auth_query)"],
      ssl: auth.upstream_ssl,
      socket_options: [
        auth.ip_version
      ],
      queue_target: 1_000,
      queue_interval: 5_000,
      ssl_opts: ssl_opts
    )
  end

  defp jitter, do: :rand.uniform(div(@interval, 10))

  defp erpc_call_node(id, fun) do
    case Supavisor.get_global_sup(id) do
      nil ->
        {:error, :not_started}

      pid ->
        :erpc.call(node(pid), fun)
    end
  end
end

1	defmodule Supavisor.SecretChecker do
2	@moduledoc false
3
4	use GenServer
5	require Logger
6
7	alias Supavisor.Helpers
8
9	@interval :timer.seconds(15)
10
11	def start_link(args) do
12	name = {:via, Registry, {Supavisor.Registry.Tenants, {:secret_checker, args.id}}}	34✔
13	GenServer.start_link(__MODULE__, args, name: name)	34✔
14	end
15
16	@spec get_secrets(Supavisor.id()) ::
17	{:ok, {method :: atom(), Supavisor.secrets()}} \| {:error, :not_started}
18	def get_secrets(id) do
19	erpc_call_node(id, fn ->	26✔
20	case Registry.lookup(Supavisor.Registry.Tenants, {:secret_checker, id}) do	1✔
21	[] ->	×
22	{:error, :not_started}
23
24	[{pid, _}] ->
25	GenServer.call(pid, :get_secrets)	1✔
26	end
27	end)
28	end
29
30	@spec update_credentials(Supavisor.id(), String.t(), (-> String.t())) ::
31	:ok \| {:error, :not_started}
32	def update_credentials(id, new_user, password_fn) do
33	erpc_call_node(id, fn ->	2✔
34	case Registry.lookup(Supavisor.Registry.Tenants, {:secret_checker, id}) do	2✔
35	[] ->	×
36	{:error, :not_started}
37
38	[{pid, _}] ->
39	GenServer.call(pid, {:update_credentials, new_user, password_fn})	2✔
40	end
41	end)
42	end
43
44	def init(args) do
45	Logger.debug("SecretChecker: Starting secret checker")	34✔
46	{{_type, tenant_external_id}, pool_user, _mode, db_name, _search_path} = args.id	34✔
47
48	# Get tenant and manager user to build auth config
49	tenant = Supavisor.Tenants.get_tenant_by_external_id(tenant_external_id)	34✔
50	manager_secrets = Supavisor.Tenants.get_manager_user_cache(tenant_external_id)	34✔
51
52	auth =	34✔
53	if manager_secrets do	9✔
54	%{	25✔
55	host: String.to_charlist(tenant.db_host),	25✔
56	sni_hostname: if(tenant.sni_hostname != nil, do: to_charlist(tenant.sni_hostname)),	25✔
57	port: tenant.db_port,	25✔
58	user: manager_secrets.db_user,	25✔
59	manager_secrets: manager_secrets,
60	auth_query: tenant.auth_query,	25✔
61	database: if(db_name != nil, do: db_name, else: tenant.db_database),	25✔
62	password: fn -> manager_secrets.db_password end,	25✔
63	application_name: "Supavisor",
64	ip_version: Supavisor.Helpers.ip_version(tenant.ip_version, tenant.db_host),	25✔
65	upstream_ssl: tenant.upstream_ssl,	25✔
66	upstream_verify: tenant.upstream_verify,	25✔
67	upstream_tls_ca: Supavisor.Helpers.upstream_cert(tenant.upstream_tls_ca)	25✔
68	}
69	else
70	nil
71	end
72
73	state = %{	34✔
74	tenant: tenant_external_id,
75	auth: auth,
76	user: pool_user,
77	ttl: :timer.hours(24),
78	conn: nil,
79	check_ref: check()
80	}
81
82	Logger.metadata(project: tenant, user: pool_user)	34✔
83	{:ok, state, {:continue, :init_conn}}	34✔
84	end
85
86	def handle_continue(:init_conn, %{auth: nil} = state) do	9✔
87	# No auth config (require_user: true tenant), skip connection setup
88	{:noreply, state}
89	end
90
91	def handle_continue(:init_conn, %{auth: auth} = state) do
92	{:ok, conn} = start_postgrex_connection(auth)	25✔
93	{:noreply, %{state \| conn: conn}}
94	end
95
NEW 96	def handle_info(:check, %{auth: nil} = state) do	×
97	{:noreply, %{state \| check_ref: check()}}
98	end
99
100	def handle_info(:check, state) do
101	check_secrets(state.user, state)	3✔
102	{:noreply, %{state \| check_ref: check()}}
103	end
104
105	def handle_info(msg, state) do
106	Logger.error("Unexpected message: #{inspect(msg)}")	×
107	{:noreply, state}
108	end
109
110	def terminate(_, state) do
NEW 111	:gen_statem.stop(state.conn)	×
112	:ok
113	end
114
115	def check(interval \\ @interval),	37✔
116	do: Process.send_after(self(), :check, interval + jitter())	37✔
117
118	def check_secrets(user, %{auth: auth, conn: conn} = state) do
119	alias Supavisor.ClientHandler.Auth
120
121	case Helpers.get_user_secret(conn, auth.auth_query, user) do	4✔
122	{:ok, secret} ->
123	method =	4✔
124	case secret do
NEW 125	%Auth.MD5Secrets{} -> :auth_query_md5	×
126	%Auth.SASLSecrets{} -> :auth_query	4✔
127	end
128
129	update_cache =	4✔
130	case Supavisor.SecretCache.get_validation_secrets(state.tenant, state.user) do	4✔
131	{:ok, {old_method, old_secrets_fn}} ->
132	method != old_method or	4✔
133	Map.delete(secret, :client_key) != Map.delete(old_secrets_fn.(), :client_key)	4✔
134
NEW 135	_other ->	×
136	true
137	end
138
139	if update_cache do	4✔
140	Logger.info("Secrets changed or not present, updating cache")	1✔
141
142	Supavisor.SecretCache.put_validation_secrets(state.tenant, state.user, method, fn ->	1✔
143	secret	1✔
144	end)
145	end
146
147	{:ok, {method, fn -> secret end}}	2✔
148
149	other ->
150	Logger.error("Failed to get secret: #{inspect(other)}")	×
151	end
152	end
153
154	def handle_call(:get_secrets, _from, %{auth: nil} = state) do
NEW 155	{:reply, {:error, :no_auth_config}, state}	×
156	end
157
158	def handle_call(:get_secrets, _from, state) do
159	{:reply, check_secrets(state.user, state), state}	1✔
160	end
161
162	def handle_call({:update_credentials, new_user, password_fn}, _from, state) do
163	Logger.info("SecretChecker: changing auth_query user to #{new_user}")	2✔
164
165	new_auth = %{	2✔
166	state.auth	2✔
167	\| user: new_user,
168	password: password_fn
169	}
170
171	{:ok, new_conn} = start_postgrex_connection(new_auth)	2✔
172
173	old_conn = state.conn	2✔
174	Process.unlink(old_conn)	2✔
175
176	Task.start(fn ->	2✔
177	try do	2✔
178	GenServer.stop(old_conn, :normal, 5_000)	2✔
179	catch
180	:exit, _ -> Process.exit(old_conn, :kill)	×
181	end
182	end)
183
184	# Clear the secrets cache for this tenant/user
185	tenant = state.tenant	2✔
186	user = state.user	2✔
187	Cachex.del(Supavisor.Cache, {:secrets, tenant, user})	2✔
188
189	Logger.info("SecretChecker: Successfully changed auth_query user")	2✔
190	{:reply, :ok, %{state \| auth: new_auth, conn: new_conn}}	2✔
191	end
192
193	defp start_postgrex_connection(auth) do
194	ssl_opts =	27✔
195	if auth.upstream_ssl and auth.upstream_verify == :peer do	27✔
196	[
197	verify: :verify_peer,
198	cacerts: [Helpers.upstream_cert(auth.upstream_tls_ca)],	×
199	server_name_indication: auth.host,	×
200	customize_hostname_check: [{:match_fun, fn _, _ -> true end}]	×
201	]
202	else
203	[
204	verify: :verify_none
205	]
206	end
207
208	Postgrex.start_link(	27✔
209	hostname: auth.host,	27✔
210	port: auth.port,	27✔
211	database: auth.database,	27✔
212	password: auth.password.(),	27✔
213	username: auth.user,	27✔
214	parameters: [application_name: "Supavisor (auth_query)"],
215	ssl: auth.upstream_ssl,	27✔
216	socket_options: [
217	auth.ip_version	27✔
218	],
219	queue_target: 1_000,
220	queue_interval: 5_000,
221	ssl_opts: ssl_opts
222	)
223	end
224
225	defp jitter, do: :rand.uniform(div(@interval, 10))	37✔
226
227	defp erpc_call_node(id, fun) do
228	case Supavisor.get_global_sup(id) do	28✔
229	nil ->	25✔
230	{:error, :not_started}
231
232	pid ->
233	:erpc.call(node(pid), fun)	3✔
234	end
235	end
236	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