supabase
diff --git a/‎README.md
Copy file name to clipboard
+56-55Lines changed: 56 additions & 55 deletions b/‎README.md
Copy file name to clipboard
+56-55Lines changed: 56 additions & 55 deletions
diff --git a/‎config/runtime.exs
Copy file name to clipboard
+11-1Lines changed: 11 additions & 1 deletion b/‎config/runtime.exs
Copy file name to clipboard
+11-1Lines changed: 11 additions & 1 deletion
diff --git a/‎lib/realtime/application.ex
Copy file name to clipboard
+4Lines changed: 4 additions & 0 deletions b/‎lib/realtime/application.ex
Copy file name to clipboard
+4Lines changed: 4 additions & 0 deletions
diff --git a/‎lib/realtime_web/channels/realtime_channel.ex
Copy file name to clipboard
+5-2Lines changed: 5 additions & 2 deletions b/‎lib/realtime_web/channels/realtime_channel.ex
Copy file name to clipboard
+5-2Lines changed: 5 additions & 2 deletions
diff --git a/‎lib/realtime_web/channels/realtime_channel/tracker.ex
Copy file name to clipboard
+88Lines changed: 88 additions & 0 deletions b/‎lib/realtime_web/channels/realtime_channel/tracker.ex
Copy file name to clipboard
+88Lines changed: 88 additions & 0 deletions
diff --git a/‎lib/realtime_web/channels/user_socket.ex
Copy file name to clipboard
+13-14Lines changed: 13 additions & 14 deletions b/‎lib/realtime_web/channels/user_socket.ex
Copy file name to clipboard
+13-14Lines changed: 13 additions & 14 deletions
diff --git a/‎mix.exs
Copy file name to clipboard
+1-1Lines changed: 1 addition & 1 deletion b/‎mix.exs
Copy file name to clipboard
+1-1Lines changed: 1 addition & 1 deletion
@@ -31,6 +31,9 @@ metrics_rpc_timeout_in_ms =
 rebalance_check_interval_in_ms =
   System.get_env("REBALANCE_CHECK_INTERVAL_IN_MS", to_string(:timer.minutes(10))) |> String.to_integer()
 
+disconnect_socket_on_no_channels_interval_in_ms =
+  System.get_env("DISCONNECT_SOCKET_ON_NO_CHANNELS_INTERVAL_IN_MS", "30000") |> String.to_integer()
+
 tenant_max_bytes_per_second = System.get_env("TENANT_MAX_BYTES_PER_SECOND", "100000") |> String.to_integer()
 tenant_max_channels_per_client = System.get_env("TENANT_MAX_CHANNELS_PER_CLIENT", "100") |> String.to_integer()
 tenant_max_concurrent_users = System.get_env("TENANT_MAX_CONCURRENT_USERS", "200") |> String.to_integer()
@@ -48,6 +51,11 @@ janitor_run_after_in_ms = System.get_env("JANITOR_RUN_AFTER_IN_MS", "600000") |>
 janitor_children_timeout = System.get_env("JANITOR_CHILDREN_TIMEOUT", "5000") |> String.to_integer()
 # 4 hours by default
 janitor_schedule_timer = System.get_env("JANITOR_SCHEDULE_TIMER_IN_MS", "14400000") |> String.to_integer()
+# defaults to 10 minutes
+no_channel_timeout_in_ms =
+  if config_env() == :test,
+    do: 3000,
+    else: System.get_env("NO_CHANNEL_TIMEOUT_IN_MS", "600000") |> String.to_integer()
 
 if !(db_version in [nil, "ipv6", "ipv4"]),
   do: raise("Invalid IP version, please set either ipv6 or ipv4")
@@ -79,8 +87,10 @@ config :realtime,
   metrics_cleaner_schedule_timer_in_ms: metrics_cleaner_schedule_timer_in_ms,
   metrics_rpc_timeout: metrics_rpc_timeout_in_ms,
   tenant_cache_expiration: tenant_cache_expiration,
+  disconnect_socket_on_no_channels_interval_in_ms: disconnect_socket_on_no_channels_interval_in_ms,
   rpc_timeout: rpc_timeout,
-  max_gen_rpc_clients: max_gen_rpc_clients
+  max_gen_rpc_clients: max_gen_rpc_clients,
+  no_channel_timeout_in_ms: no_channel_timeout_in_ms
 
 if config_env() == :test || !run_janitor? do
   config :realtime, run_janitor: false
 
@@ -10,6 +10,7 @@ defmodule Realtime.Application do
   alias Realtime.Tenants.ReplicationConnection
   alias Realtime.Tenants.Connect
   alias Realtime.Tenants.Migrations
+
   defmodule JwtSecretError, do: defexception([:message])
   defmodule JwtClaimValidatorsError, do: defexception([:message])
 
@@ -52,8 +53,10 @@ defmodule Realtime.Application do
 
     region = Application.get_env(:realtime, :region)
     :syn.join(RegionNodes, region, self(), node: node())
+
     migration_partition_slots = Application.get_env(:realtime, :migration_partition_slots)
     connect_partition_slots = Application.get_env(:realtime, :connect_partition_slots)
+    no_channel_timeout_in_ms = Application.get_env(:realtime, :no_channel_timeout_in_ms)
 
     children =
       [
@@ -92,6 +95,7 @@ defmodule Realtime.Application do
          strategy: :one_for_one,
          name: Connect.DynamicSupervisor,
          partitions: connect_partition_slots},
+        {RealtimeWeb.RealtimeChannel.Tracker, check_interval_in_ms: no_channel_timeout_in_ms},
         RealtimeWeb.Endpoint,
         RealtimeWeb.Presence
       ] ++ extensions_supervisors() ++ janitor_tasks()
 
@@ -26,6 +26,7 @@ defmodule RealtimeWeb.RealtimeChannel do
   alias RealtimeWeb.RealtimeChannel.Logging
   alias RealtimeWeb.RealtimeChannel.MessageDispatcher
   alias RealtimeWeb.RealtimeChannel.PresenceHandler
+  alias RealtimeWeb.RealtimeChannel.Tracker
 
   @confirm_token_ms_interval :timer.minutes(5)
 
@@ -42,6 +43,7 @@ defmodule RealtimeWeb.RealtimeChannel do
       transport_pid: transport_pid
     } = socket
 
+    Tracker.track(socket.transport_pid)
     Logger.metadata(external_id: tenant_id, project: tenant_id)
     Logger.put_process_level(self(), log_level)
 
@@ -351,7 +353,7 @@ defmodule RealtimeWeb.RealtimeChannel do
 
   def handle_info(:sync_presence, %{assigns: %{presence_enabled?: true}} = socket), do: PresenceHandler.sync(socket)
   def handle_info(:sync_presence, socket), do: {:noreply, socket}
-  def handle_info(_message, socket), do: {:noreply, socket}
+  def handle_info(_, socket), do: {:noreply, socket}
 
   @impl true
   def handle_in("broadcast", payload, %{assigns: %{private?: true}} = socket) do
@@ -473,9 +475,10 @@ defmodule RealtimeWeb.RealtimeChannel do
   end
 
   @impl true
-  def terminate(reason, _state) do
+  def terminate(reason, %{transport_pid: transport_pid}) do
     Logger.debug("Channel terminated with reason: #{reason}")
     :telemetry.execute([:prom_ex, :plugin, :realtime, :disconnected], %{})
+    Tracker.untrack(transport_pid)
     :ok
   end
 
 
@@ -0,0 +1,88 @@
+defmodule RealtimeWeb.RealtimeChannel.Tracker do
+  @moduledoc """
+  Tracks if the user has any channels open.
+
+  Stores in :ets table the data.
+
+  If the user has no channels open, we kill the transport pid.
+  """
+  use GenServer
+  require Logger
+
+  @table :channel_tracker
+  @zero_count_match [{{:"$1", :"$2"}, [{:"=<", :"$2", 0}], [:"$1"]}]
+  @zero_count_delete [{{:"$1", :"$2"}, [{:"=<", :"$2", 0}], [true]}]
+  @doc """
+  Tracks a transport pid.
+  """
+  @spec track(pid()) :: integer()
+  def track(pid), do: :ets.update_counter(@table, pid, 1, {pid, 0})
+
+  @doc """
+  Un-tracks a transport pid.
+  """
+  @spec untrack(pid()) :: integer()
+  def untrack(pid), do: :ets.update_counter(@table, pid, -1, {pid, 0})
+
+  @doc """
+  Returns the number of channels open for a transport pid.
+  """
+  @spec count(pid()) :: integer()
+  def count(pid) do
+    case :ets.lookup(@table, pid) do
+      [{^pid, count}] -> count
+      [] -> 0
+    end
+  end
+
+  @doc """
+  Returns a list of all pids in the table and their count.
+  """
+  @spec list_pids() :: [{pid(), integer()}]
+  def list_pids, do: :ets.tab2list(@table)
+
+  def start_link(opts) do
+    if :ets.whereis(@table) == :undefined do
+      :ets.new(@table, [
+        :set,
+        :public,
+        :named_table,
+        {:decentralized_counters, true},
+        {:write_concurrency, true}
+      ])
+    end
+
+    GenServer.start_link(__MODULE__, opts)
+  end
+
+  @impl true
+  def init(opts) do
+    check_interval_in_ms = Keyword.fetch!(opts, :check_interval_in_ms)
+    Process.send_after(self(), :check_channels, check_interval_in_ms)
+    {:ok, %{check_interval_in_ms: check_interval_in_ms}}
+  end
+
+  @impl true
+  def handle_info(:check_channels, state) do
+    chunked_killing()
+    :ets.select_delete(@table, @zero_count_delete)
+    Process.send_after(self(), :check_channels, state.check_interval_in_ms)
+    {:noreply, state}
+  end
+
+  defp chunked_killing(cont \\ nil) do
+    result = if cont, do: :ets.select(cont), else: :ets.select(@table, @zero_count_match, 1000)
+
+    case result do
+      :"$end_of_table" ->
+        :ok
+
+      {pids, cont} ->
+        Logger.info("Killing #{length(pids)} transport pids with no channels open")
+        Enum.each(pids, fn pid -> if Process.alive?(pid), do: Process.exit(pid, :kill) end)
+        chunked_killing(cont)
+    end
+  end
+
+  def table_name, do: @table
+end
@@ -7,10 +7,10 @@ defmodule RealtimeWeb.UserSocket do
   alias Realtime.Database
   alias Realtime.PostgresCdc
   alias Realtime.Tenants
+
   alias RealtimeWeb.ChannelsAuthorization
   alias RealtimeWeb.RealtimeChannel
   alias RealtimeWeb.RealtimeChannel.Logging
-
   ## Channels
   channel "realtime:*", RealtimeChannel
 
@@ -34,21 +34,25 @@ defmodule RealtimeWeb.UserSocket do
       token = access_token(params, headers)
 
       with %Tenant{
-             extensions: extensions,
              jwt_secret: jwt_secret,
              jwt_jwks: jwt_jwks,
-             max_concurrent_users: max_conn_users,
-             max_events_per_second: max_events_per_second,
-             max_bytes_per_second: max_bytes_per_second,
-             max_joins_per_second: max_joins_per_second,
-             max_channels_per_client: max_channels_per_client,
              postgres_cdc_default: postgres_cdc_default,
              suspend: false
-           } <- Tenants.Cache.get_tenant_by_external_id(external_id),
+           } = tenant <- Tenants.Cache.get_tenant_by_external_id(external_id),
            token when is_binary(token) <- token,
            jwt_secret_dec <- Crypto.decrypt!(jwt_secret),
            {:ok, claims} <- ChannelsAuthorization.authorize_conn(token, jwt_secret_dec, jwt_jwks),
            {:ok, postgres_cdc_module} <- PostgresCdc.driver(postgres_cdc_default) do
+        %Tenant{
+          extensions: extensions,
+          max_concurrent_users: max_conn_users,
+          max_events_per_second: max_events_per_second,
+          max_bytes_per_second: max_bytes_per_second,
+          max_joins_per_second: max_joins_per_second,
+          max_channels_per_client: max_channels_per_client,
+          postgres_cdc_default: postgres_cdc_default
+        } = tenant
+
         assigns = %RealtimeChannel.Assigns{
           claims: claims,
           jwt_secret: jwt_secret,
@@ -77,12 +81,7 @@ defmodule RealtimeWeb.UserSocket do
           {:error, :tenant_not_found}
 
         %Tenant{suspend: true} ->
-          Logging.log_error_message(
-            :error,
-            "RealtimeDisabledForTenant",
-            "Realtime disabled for this tenant"
-          )
-
+          Logging.log_error_message(:error, "RealtimeDisabledForTenant", "Realtime disabled for this tenant")
           {:error, :tenant_suspended}
 
         {:error, :expired_token, msg} ->
 
@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
   def project do
     [
       app: :realtime,
-      version: "2.41.0",
+      version: "2.41.1",
       elixir: "~> 1.17.3",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,
Original file line number	Diff line number	Diff line change
`@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do`
`4`	`4`	`def project do`
`5`	`5`	`[`
`6`	`6`	`app: :realtime,`
`7`		`- version: "2.41.0",`
	`7`	`+ version: "2.41.1",`
`8`	`8`	`elixir: "~> 1.17.3",`
`9`	`9`	`elixirc_paths: elixirc_paths(Mix.env()),`
`10`	`10`	`start_permanent: Mix.env() == :prod,`