feat: configure internal proxy ports via explicit lists (#709)

Author: abc3

Makefile

@@ -29,6 +29,8 @@ dev.node2:
 	PROXY_PORT_SESSION="5442" \
 	PROXY_PORT_TRANSACTION="6553" \
 	PROXY_PORT="5402" \
+	SESSION_PROXY_PORTS="12200,12201,12202,12203" \
+	TRANSACTION_PROXY_PORTS="12204,12205,12206,12207" \
 	NODE_IP=localhost \
 	AVAILABILITY_ZONE="ap-southeast-1c" \
 	ERL_AFLAGS="-kernel shell_history enabled" \
@@ -45,6 +47,8 @@ dev.node3:
 	CLUSTER_POSTGRES="true" \
 	PROXY_PORT_SESSION="5443" \
 	PROXY_PORT_TRANSACTION="6554" \
+	SESSION_PROXY_PORTS="12300,12301,12302,12303" \
+	TRANSACTION_PROXY_PORTS="12304,12305,12306,12307" \
 	ERL_AFLAGS="-kernel shell_history enabled" \
 	iex --name [email protected] --cookie cookie -S mix phx.server
 

config/runtime.exs

@@ -1,7 +1,12 @@
 import Config
 
 require Logger
-alias Supavisor.Helpers
+
+parse_integer_list = fn numbers when is_binary(numbers) ->
+  numbers
+  |> String.split(",", trim: true)
+  |> Enum.map(&String.to_integer/1)
+end
 
 secret_key_base =
   if config_env() in [:dev, :test] do
@@ -158,7 +163,12 @@ reconnect_retries =
 
 if config_env() != :test do
   config :supavisor,
-    local_proxy_shards: System.get_env("LOCAL_PROXY_SHARDS", "4") |> String.to_integer(),
+    session_proxy_ports:
+      System.get_env("SESSION_PROXY_PORTS", "12100,12101,12102,12103")
+      |> parse_integer_list.(),
+    transaction_proxy_ports:
+      System.get_env("TRANSACTION_PROXY_PORTS", "12104,12105,12106,12107")
+      |> parse_integer_list.(),
     availability_zone: System.get_env("AVAILABILITY_ZONE"),
     region: System.get_env("REGION") || System.get_env("FLY_REGION"),
     fly_alloc_id: System.get_env("FLY_ALLOC_ID"),

config/test.exs

@@ -1,5 +1,11 @@
 import Config
 
+parse_integer_list = fn numbers when is_binary(numbers) ->
+  numbers
+  |> String.split(",", trim: true)
+  |> Enum.map(&String.to_integer/1)
+end
+
 config :supavisor,
   region: "eu",
   fly_alloc_id: "123e4567-e89b-12d3-a456-426614174000",
@@ -18,7 +24,10 @@ config :supavisor,
   metrics_blocklist: [],
   node_host: System.get_env("NODE_IP", "127.0.0.1"),
   availability_zone: System.get_env("AVAILABILITY_ZONE"),
-  local_proxy_shards: System.get_env("LOCAL_PROXY_SHARDS", "4") |> String.to_integer(),
+  session_proxy_ports:
+    System.get_env("SESSION_PROXY_PORTS", "12100,12101,12102,12103") |> parse_integer_list.(),
+  transaction_proxy_ports:
+    System.get_env("TRANSACTION_PROXY_PORTS", "12104,12105,12106,12107") |> parse_integer_list.(),
   max_pools: 5,
   reconnect_retries: System.get_env("RECONNECT_RETRIES", "5") |> String.to_integer(),
   subscribe_retries: System.get_env("SUBSCRIBE_RETRIES", "5") |> String.to_integer()

docs/configuration/env.md

@@ -32,7 +32,8 @@ This document provides a categorized list of all environment variables used in S
 | `GLOBAL_UPSTREAM_CA_PATH`     | Upstream CA certificate path                                | -             | No       |
 | `GLOBAL_DOWNSTREAM_CERT_PATH` | Downstream certificate path                                 | -             | No       |
 | `GLOBAL_DOWNSTREAM_KEY_PATH`  | Downstream private key path                                 | -             | No       |
-| `LOCAL_PROXY_SHARDS`          | Number of local proxy shards per mode (session/transaction) | `4`           | No       |
+| `SESSION_PROXY_PORTS`         | Comma-separated list of ports for session proxy shards    | `12100,12101,12102,12103` | No |
+| `TRANSACTION_PROXY_PORTS`     | Comma-separated list of ports for transaction proxy shards | `12104,12105,12106,12107` | No |
 
 ---
 

lib/supavisor.ex

@@ -413,8 +413,14 @@ defmodule Supavisor do
   @spec get_local_server(id, atom) :: map()
   def get_local_server(id, mode) do
     host = Application.get_env(:supavisor, :node_host)
-    local_proxy_shards = Application.fetch_env!(:supavisor, :local_proxy_shards)
-    shard = :erlang.phash2(id, local_proxy_shards)
+
+    ports =
+      case mode do
+        :session -> Application.fetch_env!(:supavisor, :session_proxy_ports)
+        :transaction -> Application.fetch_env!(:supavisor, :transaction_proxy_ports)
+      end
+
+    shard = :erlang.phash2(id, length(ports))
     %{host: host, port: :ranch.get_port({:pg_proxy_internal, mode, shard})}
   end
 

lib/supavisor/application.ex

@@ -48,13 +48,15 @@ defmodule Supavisor.Application do
         {Supavisor.SignalHandler, []}
       )
 
-    local_proxy_shards_count = Application.fetch_env!(:supavisor, :local_proxy_shards)
+    session_shards =
+      :supavisor
+      |> Application.fetch_env!(:session_proxy_ports)
+      |> build_shards(:session)
 
-    local_proxy_shards =
-      for shard <- 0..(local_proxy_shards_count - 1), mode <- [:session, :transaction] do
-        {{:pg_proxy_internal, mode, shard}, 0, %{mode: mode, local: true, shard: shard},
-         Supavisor.ClientHandler}
-      end
+    transaction_shards =
+      :supavisor
+      |> Application.fetch_env!(:transaction_proxy_ports)
+      |> build_shards(:transaction)
 
     proxy_ports =
       [
@@ -64,7 +66,7 @@ defmodule Supavisor.Application do
          %{mode: :session, local: false}, Supavisor.ClientHandler},
         {:pg_proxy, Application.get_env(:supavisor, :proxy_port), %{mode: :proxy, local: false},
          Supavisor.ClientHandler}
-      ] ++ local_proxy_shards
+      ] ++ session_shards ++ transaction_shards
 
     for {key, port, opts, handler} <- proxy_ports do
       case :ranch.start_listener(
@@ -159,6 +161,14 @@ defmodule Supavisor.Application do
     :ok
   end
 
+  @spec build_shards([pos_integer()], atom()) :: term()
+  defp build_shards(ports, mode) do
+    for {port, shard} <- Enum.with_index(ports) do
+      {{:pg_proxy_internal, mode, shard}, port, %{mode: mode, local: true, shard: shard},
+       Supavisor.ClientHandler}
+    end
+  end
+
   @spec short_node_id() :: String.t() | nil
   defp short_node_id do
     with {:ok, fly_alloc_id} when is_binary(fly_alloc_id) <-

lib/supavisor/helpers.ex

@@ -337,6 +337,13 @@ defmodule Supavisor.Helpers do
     result -> result
   end
 
+  @spec parse_integer_list(String.t()) :: [integer()]
+  def parse_integer_list(numbers) when is_binary(numbers) do
+    numbers
+    |> String.split(",", trim: true)
+    |> Enum.map(&String.to_integer/1)
+  end
+
   @doc """
   Sets the maximum heap size for the current process. The `max_heap_size` parameter is in megabytes.
 

test/support/cluster.ex

@@ -52,6 +52,14 @@ defmodule Supavisor.Support.Cluster do
             {:supavisor, :availability_zone} ->
               "ap-southeast-1c"
 
+            {:supavisor, :session_proxy_ports} ->
+              Application.get_env(:supavisor, :session_proxy_ports)
+              |> Enum.map(&(&1 + 100))
+
+            {:supavisor, :transaction_proxy_ports} ->
+              Application.get_env(:supavisor, :transaction_proxy_ports)
+              |> Enum.map(&(&1 + 100))
+
             _ ->
               val
           end