ebd445de3bb26ad0378512ac820c2735bcde2787-PR-1676

Committed 23 Jan 2026 11:08AM UTC coverage: 88.629% (+0.3%) from 88.359%

Build # ebd445de3bb26ad0378512ac820c2735bcde2787-PR-1676

Build Type

Pull #1676

github

Committed by

filipecabaco

Commit Message

restrict balancing for the first 5 minutes (configurable)

Pull Request Pull Request #1676: fix: Load aware node picker using avg 15 minutes

Run Details

10 of 11 new or added lines in 1 file covered. (90.91%)

2 existing lines in 1 file now uncovered.

2572 of 2902 relevant lines covered (88.63%)

9659.2 hits per line

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

62.96

/lib/realtime/nodes.ex

defmodule Realtime.Nodes do
  @moduledoc """
  Handles common needs for :syn module operations
  """
  require Logger
  alias Realtime.Api.Tenant
  alias Realtime.Tenants

  @doc """
  Gets the node to launch the Postgres connection on for a tenant.
  """
  @spec get_node_for_tenant(Tenant.t()) :: {:ok, node(), binary()} | {:error, term()}
  def get_node_for_tenant(nil), do: {:error, :tenant_not_found}

  def get_node_for_tenant(%Tenant{external_id: tenant_id} = tenant) do
    with region <- Tenants.region(tenant),
         tenant_region <- platform_region_translator(region),
         node <- launch_node(tenant_id, tenant_region, node()) do
      {:ok, node, tenant_region}
    end
  end

  @doc """
  Translates a region from a platform to the closest Supabase tenant region
  """
  @spec platform_region_translator(String.t() | nil) :: nil | binary()
  def platform_region_translator(nil), do: nil

  def platform_region_translator(tenant_region) when is_binary(tenant_region) do
    case tenant_region do
      "ap-east-1" -> "ap-southeast-1"
      "ap-northeast-1" -> "ap-southeast-1"
      "ap-northeast-2" -> "ap-southeast-1"
      "ap-south-1" -> "ap-southeast-1"
      "ap-southeast-1" -> "ap-southeast-1"
      "ap-southeast-2" -> "ap-southeast-2"
      "ca-central-1" -> "us-east-1"
      "eu-central-1" -> "eu-west-2"
      "eu-central-2" -> "eu-west-2"
      "eu-north-1" -> "eu-west-2"
      "eu-west-1" -> "eu-west-2"
      "eu-west-2" -> "eu-west-2"
      "eu-west-3" -> "eu-west-2"
      "sa-east-1" -> "us-east-1"
      "us-east-1" -> "us-east-1"
      "us-east-2" -> "us-east-1"
      "us-west-1" -> "us-west-1"
      "us-west-2" -> "us-west-1"
      _ -> nil
    end
  end

  @doc """
  Lists the nodes in a region. Sorts by node name in case the list order
  is unstable.
  """

  @spec region_nodes(String.t() | nil) :: [atom()]
  def region_nodes(region) when is_binary(region) do
    :syn.members(RegionNodes, region)
    |> Enum.map(fn {_pid, [node: node]} -> node end)
    |> Enum.sort()
  end

  def region_nodes(nil), do: []

  @doc """
  Picks a node from a region based on the provided key
  """
  @spec node_from_region(String.t(), term()) :: {:ok, node} | {:error, :not_available}
  def node_from_region(region, key) when is_binary(region) do
    nodes = region_nodes(region)

    case nodes do
      [] ->
        {:error, :not_available}

      _ ->
        member_count = Enum.count(nodes)
        index = :erlang.phash2(key, member_count)

        {:ok, Enum.fetch!(nodes, index)}
    end
  end

  def node_from_region(_, _), do: {:error, :not_available}

  @doc """
  Picks the node to launch the Postgres connection on.

  If there are not two nodes in a region the connection is established from
  the `default` node given.
  """
  @spec launch_node(String.t(), String.t() | nil, atom()) :: atom()
  def launch_node(_tenant_id, region, default) do
    cond do
      region_nodes(region) == [] ->
        Logger.warning("Zero region nodes for #{region} using #{inspect(default)}")
        default

      uptime_ms() < Application.fetch_env!(:realtime, :node_balance_uptime_threshold_in_ms) ->
        Enum.random(region_nodes(region))

      true ->
        load_aware_node_picker(region_nodes(region))
    end
  end

  defp load_aware_node_picker(regions_nodes) do
    regions_nodes
    |> Enum.take_random(2)
    |> Enum.min_by(&node_load(&1))
  end

  def node_load(node) when node() == node, do: :cpu_sup.avg5()
  def node_load(node) when node() != node, do: Realtime.GenRpc.call(node, __MODULE__, :node_load, [node], [])

  @doc """
  Gets a short node name from a node name when a node name looks like `realtime-prod@fdaa:0:cc:a7b:b385:83c3:cfe3:2`

  ## Examples

      iex> node = Node.self()
      iex> Realtime.Helpers.short_node_id_from_name(node)
      "nohost"

      iex> node = :"realtime-prod@fdaa:0:cc:a7b:b385:83c3:cfe3:2"
      iex> Realtime.Helpers.short_node_id_from_name(node)
      "83c3cfe3"

      iex> node = :"pink@127.0.0.1"
      iex> Realtime.Helpers.short_node_id_from_name(node)
      "pink@127.0.0.1"

      iex> node = :"pink@10.0.1.1"
      iex> Realtime.Helpers.short_node_id_from_name(node)
      "10.0.1.1"

      iex> node = :"realtime@host.name.internal"
      iex> Realtime.Helpers.short_node_id_from_name(node)
      "host.name.internal"
  """

  @spec short_node_id_from_name(atom()) :: String.t()
  def short_node_id_from_name(name) when is_atom(name) do
    [_, host] = name |> Atom.to_string() |> String.split("@", parts: 2)

    case String.split(host, ":", parts: 8) do
      [_, _, _, _, _, one, two, _] ->
        one <> two

      ["127.0.0.1"] ->
        Atom.to_string(name)

      _other ->
        host
    end
  end

  @spec all_node_regions() :: [String.t()]
  @doc "List all the regions where nodes can be launched"
  def all_node_regions(), do: :syn.group_names(RegionNodes)

  defp uptime_ms do
    start_time = :erlang.system_info(:start_time)
    now = :erlang.monotonic_time()
    :erlang.convert_time_unit(now - start_time, :native, :millisecond)
  end
end

1	defmodule Realtime.Nodes do
2	@moduledoc """
3	Handles common needs for :syn module operations
4	"""
5	require Logger
6	alias Realtime.Api.Tenant
7	alias Realtime.Tenants
8
9	@doc """
10	Gets the node to launch the Postgres connection on for a tenant.
11	"""
12	@spec get_node_for_tenant(Tenant.t()) :: {:ok, node(), binary()} \| {:error, term()}
13	def get_node_for_tenant(nil), do: {:error, :tenant_not_found}	4✔
14
15	def get_node_for_tenant(%Tenant{external_id: tenant_id} = tenant) do
16	with region <- Tenants.region(tenant),	804✔
17	tenant_region <- platform_region_translator(region),	804✔
18	node <- launch_node(tenant_id, tenant_region, node()) do	804✔
19	{:ok, node, tenant_region}	802✔
20	end
21	end
22
23	@doc """
24	Translates a region from a platform to the closest Supabase tenant region
25	"""
26	@spec platform_region_translator(String.t() \| nil) :: nil \| binary()
27	def platform_region_translator(nil), do: nil	×
28
29	def platform_region_translator(tenant_region) when is_binary(tenant_region) do
30	case tenant_region do	900✔
31	"ap-east-1" -> "ap-southeast-1"	×
32	"ap-northeast-1" -> "ap-southeast-1"	×
33	"ap-northeast-2" -> "ap-southeast-1"	×
34	"ap-south-1" -> "ap-southeast-1"	×
35	"ap-southeast-1" -> "ap-southeast-1"	×
36	"ap-southeast-2" -> "ap-southeast-2"	2✔
37	"ca-central-1" -> "us-east-1"	×
38	"eu-central-1" -> "eu-west-2"	×
39	"eu-central-2" -> "eu-west-2"	×
40	"eu-north-1" -> "eu-west-2"	×
41	"eu-west-1" -> "eu-west-2"	×
42	"eu-west-2" -> "eu-west-2"	×
43	"eu-west-3" -> "eu-west-2"	×
44	"sa-east-1" -> "us-east-1"	×
45	"us-east-1" -> "us-east-1"	898✔
46	"us-east-2" -> "us-east-1"	×
47	"us-west-1" -> "us-west-1"	×
48	"us-west-2" -> "us-west-1"	×
49	_ -> nil	×
50	end
51	end
52
53	@doc """
54	Lists the nodes in a region. Sorts by node name in case the list order
55	is unstable.
56	"""
57
58	@spec region_nodes(String.t() \| nil) :: [atom()]
59	def region_nodes(region) when is_binary(region) do
60	:syn.members(RegionNodes, region)
61	\|> Enum.map(fn {_pid, [node: node]} -> node end)	2,016✔
62	\|> Enum.sort()	1,927✔
63	end
64
65	def region_nodes(nil), do: []	2✔
66
67	@doc """
68	Picks a node from a region based on the provided key
69	"""
70	@spec node_from_region(String.t(), term()) :: {:ok, node} \| {:error, :not_available}
71	def node_from_region(region, key) when is_binary(region) do
72	nodes = region_nodes(region)	29✔
73
74	case nodes do	29✔
75	[] ->	2✔
76	{:error, :not_available}
77
78	_ ->
79	member_count = Enum.count(nodes)	27✔
80	index = :erlang.phash2(key, member_count)	27✔
81
82	{:ok, Enum.fetch!(nodes, index)}
83	end
84	end
85
86	def node_from_region(_, _), do: {:error, :not_available}	2✔
87
88	@doc """
89	Picks the node to launch the Postgres connection on.
90
91	If there are not two nodes in a region the connection is established from
92	the `default` node given.
93	"""
94	@spec launch_node(String.t(), String.t() \| nil, atom()) :: atom()
95	def launch_node(_tenant_id, region, default) do
96	cond do	924✔
97	region_nodes(region) == [] ->
98	Logger.warning("Zero region nodes for #{region} using #{inspect(default)}")	4✔
99	default	4✔
100
101	uptime_ms() < Application.fetch_env!(:realtime, :node_balance_uptime_threshold_in_ms) ->	920✔
NEW UNCOV 102	Enum.random(region_nodes(region))	×
103
104	true ->	920✔
105	load_aware_node_picker(region_nodes(region))	920✔
106	end
107	end
108
109	defp load_aware_node_picker(regions_nodes) do
110	regions_nodes
111	\|> Enum.take_random(2)
112	\|> Enum.min_by(&node_load(&1))	920✔
113	end
114
115	def node_load(node) when node() == node, do: :cpu_sup.avg5()	920✔
116	def node_load(node) when node() != node, do: Realtime.GenRpc.call(node, __MODULE__, :node_load, [node], [])	28✔
117
118	@doc """
119	Gets a short node name from a node name when a node name looks like `realtime-prod@fdaa:0:cc:a7b:b385:83c3:cfe3:2`
120
121	## Examples
122
123	iex> node = Node.self()
124	iex> Realtime.Helpers.short_node_id_from_name(node)
125	"nohost"
126
127	iex> node = :"realtime-prod@fdaa:0:cc:a7b:b385:83c3:cfe3:2"
128	iex> Realtime.Helpers.short_node_id_from_name(node)
129	"83c3cfe3"
130
131	iex> node = :"pink@127.0.0.1"
132	iex> Realtime.Helpers.short_node_id_from_name(node)
133	"pink@127.0.0.1"
134
135	iex> node = :"pink@10.0.1.1"
136	iex> Realtime.Helpers.short_node_id_from_name(node)
137	"10.0.1.1"
138
139	iex> node = :"realtime@host.name.internal"
140	iex> Realtime.Helpers.short_node_id_from_name(node)
141	"host.name.internal"
142	"""
143
144	@spec short_node_id_from_name(atom()) :: String.t()
145	def short_node_id_from_name(name) when is_atom(name) do
146	[_, host] = name \|> Atom.to_string() \|> String.split("@", parts: 2)	170✔
147
148	case String.split(host, ":", parts: 8) do	170✔
149	[_, _, _, _, _, one, two, _] ->
UNCOV 150	one <> two	×
151
152	["127.0.0.1"] ->
153	Atom.to_string(name)	168✔
154
155	_other ->
156	host	2✔
157	end
158	end
159
160	@spec all_node_regions() :: [String.t()]
161	@doc "List all the regions where nodes can be launched"
162	def all_node_regions(), do: :syn.group_names(RegionNodes)	14✔
163
164	defp uptime_ms do
165	start_time = :erlang.system_info(:start_time)	920✔
166	now = :erlang.monotonic_time()	920✔
167	:erlang.convert_time_unit(now - start_time, :native, :millisecond)	920✔
168	end
169	end

supabase / realtime / ebd445de3bb26ad0378512ac820c2735bcde2787-PR-1676

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