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

94.12

/lib/supavisor/client_handler/cancel.ex

defmodule Supavisor.ClientHandler.Cancel do
  @moduledoc """
  Helpers for dealing with query canceling.

  From the docs:

  > To issue a cancel request, the frontend opens a new connection to the server and sends a CancelRequest message,
  > rather than the StartupMessage message that would ordinarily be sent across a new connection. The server
  > will process this request and then close the connection. For security reasons, no direct reply is made to the
  > cancel request message.

  Cancel requests are sent through Phoenix PubSub. ClientHandlers listen to their key and other client
  handlers may send cancel requests on it.
  """

  require Logger
  alias Phoenix.PubSub
  alias Supavisor.HandlerHelpers
  require Supavisor.Protocol.Server, as: Server

  @doc """
  Called upon receiving cancel requests, broadcasts to relevant client handler
  """
  @spec send_cancel_query(non_neg_integer, non_neg_integer, term) :: :ok | {:errr, term}
  def send_cancel_query(pid, key, msg \\ :cancel_query) do
    PubSub.broadcast(
      Supavisor.PubSub,
      "cancel_req:#{pid}_#{key}",
      msg
    )
  end

  @doc """
  Called by ClientHandlers when starting a connection
  """
  @spec listen_cancel_query(non_neg_integer, non_neg_integer) :: :ok | {:errr, term}
  def listen_cancel_query(pid, key) do
    PubSub.subscribe(Supavisor.PubSub, "cancel_req:#{pid}_#{key}")
  end

  @doc """
  If there's an ongoing query, forward the message to cancel it to the checked out connection

  Called by the client handler when receiving a cancel requests
  """
  def maybe_forward_cancel_to_db(:busy, data) do
    key = {data.tenant, data.db_connection}
    Logger.debug("ClientHandler: Cancel query for #{inspect(key)}")
    {_pool, db_pid, _db_sock} = data.db_connection

    case db_pid_meta(key) do
      [{^db_pid, meta}] ->
        msg = Server.cancel_message(meta.pid, meta.key)
        opts = [:binary, {:packet, :raw}, {:active, true}, meta.ip_version]
        {:ok, sock} = :gen_tcp.connect(meta.host, meta.port, opts)
        sock = {:gen_tcp, sock}
        :ok = HandlerHelpers.sock_send(sock, msg)
        :ok = HandlerHelpers.sock_close(sock)

      error ->
        Logger.error(
          "ClientHandler: Received cancel but no proc was found #{inspect(key)} #{inspect(error)}"
        )
    end
  end

  def maybe_forward_cancel_to_db(_state, _data) do
    :ok
  end

  defp db_pid_meta({_, {_, pid, _}} = _key) do
    rkey = Supavisor.Registry.PoolPids

    pid
    |> node()
    |> :erpc.call(Registry, :lookup, [rkey, pid], 15_000)
  end
end

1	defmodule Supavisor.ClientHandler.Cancel do
2	@moduledoc """
3	Helpers for dealing with query canceling.
4
5	From the docs:
6
7	> To issue a cancel request, the frontend opens a new connection to the server and sends a CancelRequest message,
8	> rather than the StartupMessage message that would ordinarily be sent across a new connection. The server
9	> will process this request and then close the connection. For security reasons, no direct reply is made to the
10	> cancel request message.
11
12	Cancel requests are sent through Phoenix PubSub. ClientHandlers listen to their key and other client
13	handlers may send cancel requests on it.
14	"""
15
16	require Logger
17	alias Phoenix.PubSub
18	alias Supavisor.HandlerHelpers
19	require Supavisor.Protocol.Server, as: Server
20
21	@doc """
22	Called upon receiving cancel requests, broadcasts to relevant client handler
23	"""
24	@spec send_cancel_query(non_neg_integer, non_neg_integer, term) :: :ok \| {:errr, term}
25	def send_cancel_query(pid, key, msg \\ :cancel_query) do
26	PubSub.broadcast(	118✔
27	Supavisor.PubSub,
28	"cancel_req:#{pid}_#{key}",	118✔
29	msg
30	)
31	end
32
33	@doc """
34	Called by ClientHandlers when starting a connection
35	"""
36	@spec listen_cancel_query(non_neg_integer, non_neg_integer) :: :ok \| {:errr, term}
37	def listen_cancel_query(pid, key) do
38	PubSub.subscribe(Supavisor.PubSub, "cancel_req:#{pid}_#{key}")	563✔
39	end
40
41	@doc """
42	If there's an ongoing query, forward the message to cancel it to the checked out connection
43
44	Called by the client handler when receiving a cancel requests
45	"""
46	def maybe_forward_cancel_to_db(:busy, data) do
47	key = {data.tenant, data.db_connection}	9✔
48	Logger.debug("ClientHandler: Cancel query for #{inspect(key)}")	9✔
49	{_pool, db_pid, _db_sock} = data.db_connection	9✔
50
51	case db_pid_meta(key) do	9✔
52	[{^db_pid, meta}] ->
53	msg = Server.cancel_message(meta.pid, meta.key)	9✔
54	opts = [:binary, {:packet, :raw}, {:active, true}, meta.ip_version]	9✔
55	{:ok, sock} = :gen_tcp.connect(meta.host, meta.port, opts)	9✔
56	sock = {:gen_tcp, sock}	9✔
57	:ok = HandlerHelpers.sock_send(sock, msg)	9✔
58	:ok = HandlerHelpers.sock_close(sock)	9✔
59
60	error ->
NEW 61	Logger.error(	×
62	"ClientHandler: Received cancel but no proc was found #{inspect(key)} #{inspect(error)}"
63	)
64	end
65	end
66
67	def maybe_forward_cancel_to_db(_state, _data) do	107✔
68	:ok
69	end
70
71	defp db_pid_meta({_, {_, pid, _}} = _key) do
72	rkey = Supavisor.Registry.PoolPids	9✔
73
74	pid
75	\|> node()
76	\|> :erpc.call(Registry, :lookup, [rkey, pid], 15_000)	9✔
77	end
78	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