fix: rate limit presence events

Author: filipecabaco

lib/realtime/api/tenant.ex

@@ -19,6 +19,7 @@ defmodule Realtime.Api.Tenant do
     field(:postgres_cdc_default, :string)
     field(:max_concurrent_users, :integer)
     field(:max_events_per_second, :integer)
+    field(:max_presence_events_per_second, :integer, default: 100)
     field(:max_bytes_per_second, :integer)
     field(:max_channels_per_client, :integer)
     field(:max_joins_per_second, :integer)

lib/realtime_web/channels/realtime_channel/presence_handler.ex

@@ -10,6 +10,8 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandler do
   alias Phoenix.Socket
   alias Phoenix.Tracker.Shard
   alias Realtime.GenCounter
+  alias Realtime.RateCounter
+  alias Realtime.Tenants
   alias Realtime.Tenants.Authorization
   alias Realtime.Tenants.Authorization.Policies
   alias Realtime.Tenants.Authorization.Policies.PresencePolicies
@@ -24,29 +26,33 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandler do
 
   def sync(%{assigns: %{private?: false}} = socket) do
     %{assigns: %{tenant_topic: topic}} = socket
+
     socket = Logging.maybe_log_handle_info(socket, :sync_presence)
-    count(socket)
-    push(socket, "presence_state", presence_dirty_list(topic))
-    {:noreply, socket}
+
+    with :ok <- limit_presence_event(socket) do
+      push(socket, "presence_state", presence_dirty_list(topic))
+      {:noreply, socket}
+    else
+      {:error, :rate_limit_exceeded} -> {:stop, :normal, socket}
+    end
   end
 
   def sync(%{assigns: assigns} = socket) do
     %{tenant_topic: topic, policies: policies} = assigns
 
-    socket =
-      case policies do
-        %Policies{presence: %PresencePolicies{read: false}} ->
-          Logger.info("Presence track message ignored on #{topic}")
-          socket
-
-        _ ->
-          socket = Logging.maybe_log_handle_info(socket, :sync_presence)
-          count(socket)
-          push(socket, "presence_state", presence_dirty_list(topic))
-          socket
-      end
-
-    {:noreply, socket}
+    with :ok <- limit_presence_event(socket),
+         %Policies{presence: %PresencePolicies{read: true}} <- policies do
+      socket = Logging.maybe_log_handle_info(socket, :sync_presence)
+      push(socket, "presence_state", presence_dirty_list(topic))
+      {:noreply, socket}
+    else
+      %Policies{presence: %PresencePolicies{read: false}} ->
+        Logger.info("Presence track message ignored on #{topic}")
+        {:noreply, socket}
+
+      {:error, :rate_limit_exceeded} ->
+        {:stop, :normal, socket}
+    end
   end
 
   @spec handle(map(), Socket.t()) :: {:reply, :error | :ok, Socket.t()}
@@ -58,15 +64,7 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandler do
 
   def handle(%{"event" => event} = payload, db_conn, socket) do
     event = String.downcase(event, :ascii)
-
-    case handle_presence_event(event, payload, db_conn, socket) do
-      {:ok, socket} ->
-        count(socket)
-        {:reply, :ok, socket}
-
-      {:error, socket} ->
-        {:reply, :error, socket}
-    end
+    handle_presence_event(event, payload, db_conn, socket)
   end
 
   def handle(_payload, _db_conn, socket), do: {:noreply, socket}
@@ -90,11 +88,11 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandler do
 
       {:error, :rls_policy_error, error} ->
         log_error("RlsPolicyError", error)
-        {:error, socket}
+        {:reply, :error, socket}
 
       {:error, error} ->
         log_error("UnableToSetPolicies", error)
-        {:error, socket}
+        {:reply, :error, socket}
     end
   end
 
@@ -113,41 +111,43 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandler do
          _db_conn,
          %{assigns: %{private?: true, policies: %Policies{presence: %PresencePolicies{write: false}}}} = socket
        ) do
-    {:error, socket}
+    {:reply, :error, socket}
   end
 
   defp handle_presence_event("untrack", _, _, socket) do
     %{assigns: %{presence_key: presence_key, tenant_topic: tenant_topic}} = socket
-    {Presence.untrack(self(), tenant_topic, presence_key), socket}
+    :ok = Presence.untrack(self(), tenant_topic, presence_key)
+    {:reply, :ok, socket}
   end
 
   defp handle_presence_event(event, _, _, socket) do
     log_error("UnknownPresenceEvent", event)
-    {:error, socket}
+    {:reply, :error, socket}
   end
 
   defp track(socket, payload) do
     %{assigns: %{presence_key: presence_key, tenant_topic: tenant_topic}} = socket
     payload = Map.get(payload, "payload", %{})
 
-    case Presence.track(self(), tenant_topic, presence_key, payload) do
-      {:ok, _} ->
-        {:ok, socket}
-
+    with :ok <- limit_presence_event(socket),
+         {:ok, _} <- Presence.track(self(), tenant_topic, presence_key, payload) do
+      {:reply, :ok, socket}
+    else
       {:error, {:already_tracked, pid, _, _}} ->
         case Presence.update(pid, tenant_topic, presence_key, payload) do
-          {:ok, _} -> {:ok, socket}
-          {:error, _} -> {:error, socket}
+          {:ok, _} -> {:reply, :ok, socket}
+          {:error, _} -> {:reply, :error, socket}
         end
 
+      {:error, :rate_limit_exceeded} ->
+        {:reply, :error, socket}
+
       {:error, error} ->
         log_error("UnableToTrackPresence", error)
-        {:error, socket}
+        {:reply, :error, socket}
     end
   end
 
-  defp count(%{assigns: %{presence_rate_counter: presence_counter}}), do: GenCounter.add(presence_counter.id)
-
   defp presence_dirty_list(topic) do
     [{:pool_size, size}] = :ets.lookup(Presence, :pool_size)
 
@@ -156,4 +156,20 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandler do
     |> Shard.dirty_list(topic)
     |> Phoenix.Presence.group()
   end
+
+  defp limit_presence_event(socket) do
+    %{assigns: %{presence_rate_counter: presence_counter, tenant: tenant_id}} = socket
+    GenCounter.add(presence_counter.id)
+    {:ok, rate_counter} = RateCounter.get(presence_counter)
+
+    tenant = Tenants.Cache.get_tenant_by_external_id(tenant_id)
+
+    if rate_counter.avg > tenant.max_presence_events_per_second do
+      message = "Too many presence messages per second"
+      log_warning("TooManyPresenceMessages", message)
+      {:error, :rate_limit_exceeded}
+    else
+      :ok
+    end
+  end
 end

mix.exs

@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
   def project do
     [
       app: :realtime,
-      version: "2.41.23",
+      version: "2.41.24",
       elixir: "~> 1.17.3",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,

priv/repo/migrations/20250811121559_add_max_presence_events_per_second.exs

@@ -0,0 +1,9 @@
+defmodule Realtime.Repo.Migrations.AddMaxPresenceEventsPerSecond do
+  use Ecto.Migration
+
+  def change do
+    alter table(:tenants) do
+      add :max_presence_events_per_second, :integer, default: 100
+    end
+  end
+end

test/realtime_web/channels/realtime_channel/presence_handler_test.exs

@@ -25,9 +25,10 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
       db_conn: db_conn
     } do
       key = random_string()
+      policies = %Policies{presence: %PresencePolicies{read: true, write: true}}
 
       socket =
-        socket_fixture(tenant, topic, key, %Policies{presence: %PresencePolicies{read: true, write: true}})
+        socket_fixture(tenant, topic, key, policies: policies)
 
       PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
       topic = socket.assigns.tenant_topic
@@ -36,15 +37,10 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
       assert Map.has_key?(joins, key)
     end
 
-    test "when tracking already existing user, metadata updated", %{
-      tenant: tenant,
-      topic: topic,
-      db_conn: db_conn
-    } do
+    test "when tracking already existing user, metadata updated", %{tenant: tenant, topic: topic, db_conn: db_conn} do
       key = random_string()
-
-      socket =
-        socket_fixture(tenant, topic, key, %Policies{presence: %PresencePolicies{read: true, write: true}})
+      policies = %Policies{presence: %PresencePolicies{read: true, write: true}}
+      socket = socket_fixture(tenant, topic, key, policies: policies)
 
       assert {:reply, :ok, socket} = PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
 
@@ -62,15 +58,8 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
 
     test "with false policy and is public, user can track their presence and changes", %{tenant: tenant, topic: topic} do
       key = random_string()
-
-      socket =
-        socket_fixture(
-          tenant,
-          topic,
-          key,
-          %Policies{presence: %PresencePolicies{read: false, write: false}},
-          false
-        )
+      policies = %Policies{presence: %PresencePolicies{read: false, write: false}}
+      socket = socket_fixture(tenant, topic, key, policies: policies, private?: false)
 
       assert {:reply, :ok, _socket} = PresenceHandler.handle(%{"event" => "track"}, socket)
 
@@ -81,9 +70,8 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
 
     test "user can untrack when they want", %{tenant: tenant, topic: topic, db_conn: db_conn} do
       key = random_string()
-
-      socket =
-        socket_fixture(tenant, topic, key, %Policies{presence: %PresencePolicies{read: true, write: true}})
+      policies = %Policies{presence: %PresencePolicies{read: true, write: true}}
+      socket = socket_fixture(tenant, topic, key, policies: policies)
 
       assert {:reply, :ok, socket} = PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
 
@@ -151,10 +139,8 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
       reject(&Authorization.get_write_authorizations/3)
 
       key = random_string()
-
-      socket =
-        socket_fixture(tenant, topic, key, %Policies{broadcast: %BroadcastPolicies{read: false}}, false)
-
+      policies = %Policies{broadcast: %BroadcastPolicies{read: false}}
+      socket = socket_fixture(tenant, topic, key, policies: policies, private?: false)
       topic = socket.assigns.tenant_topic
 
       for _ <- 1..100, reduce: socket do
@@ -187,7 +173,7 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
     } do
       key = random_string()
       policies = %Policies{presence: %PresencePolicies{read: true, write: true}}
-      socket = socket_fixture(tenant, topic, key, policies, false, false)
+      socket = socket_fixture(tenant, topic, key, policies: policies, private?: false, enabled?: false)
 
       assert {:reply, :ok, _socket} = PresenceHandler.handle(%{"event" => "track"}, socket)
       topic = socket.assigns.tenant_topic
@@ -201,12 +187,68 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
     } do
       key = random_string()
       policies = %Policies{presence: %PresencePolicies{read: true, write: true}}
-      socket = socket_fixture(tenant, topic, key, policies, false, false)
+      socket = socket_fixture(tenant, topic, key, policies: policies, private?: false, enabled?: false)
 
       assert {:reply, :ok, _socket} = PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
       topic = socket.assigns.tenant_topic
       refute_receive %Broadcast{topic: ^topic, event: "presence_diff"}
     end
+
+    @tag policies: [:authenticated_read_broadcast_and_presence, :authenticated_write_broadcast_and_presence]
+    test "rate limit is checked on private channel", %{tenant: tenant, topic: topic, db_conn: db_conn} do
+      key = random_string()
+      policies = %Policies{presence: %PresencePolicies{read: true, write: true}}
+      socket = socket_fixture(tenant, topic, key, policies: policies, private?: true)
+
+      for _ <- 1..300, do: PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
+      Process.sleep(1000)
+
+      assert {:reply, :error, _} = PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
+    end
+
+    test "rate limit is checked on public channel", %{tenant: tenant, topic: topic, db_conn: db_conn} do
+      key = random_string()
+      socket = socket_fixture(tenant, topic, key, private?: false)
+
+      for _ <- 1..300, do: PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
+      Process.sleep(1000)
+
+      assert {:reply, :error, _} = PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
+    end
+  end
+
+  describe "sync/1" do
+    test "syncs presence state for public channels", %{tenant: tenant, topic: topic} do
+      key = random_string()
+      policies = %Policies{presence: %PresencePolicies{read: false, write: false}}
+      socket = socket_fixture(tenant, topic, key, policies: policies, private?: false)
+
+      assert {:noreply, _socket} = PresenceHandler.sync(socket)
+    end
+
+    test "syncs presence state for private channels with read policy true", %{tenant: tenant, topic: topic} do
+      key = random_string()
+      policies = %Policies{presence: %PresencePolicies{read: true, write: true}}
+      socket = socket_fixture(tenant, topic, key, policies: policies, private?: true)
+
+      assert {:noreply, _socket} = PresenceHandler.sync(socket)
+    end
+
+    test "ignores sync for private channels with read policy false", %{tenant: tenant, topic: topic} do
+      key = random_string()
+      policies = %Policies{presence: %PresencePolicies{read: false, write: true}}
+      socket = socket_fixture(tenant, topic, key, policies: policies, private?: true)
+
+      assert {:noreply, _socket} = PresenceHandler.sync(socket)
+    end
+
+    test "ignores sync when presence is disabled", %{tenant: tenant, topic: topic} do
+      key = random_string()
+      policies = %Policies{presence: %PresencePolicies{read: true, write: true}}
+      socket = socket_fixture(tenant, topic, key, policies: policies, private?: true, enabled?: false)
+
+      assert {:noreply, _socket} = PresenceHandler.sync(socket)
+    end
   end
 
   defp initiate_tenant(context) do
@@ -223,20 +265,19 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
     {:ok, tenant: tenant, db_conn: db_conn, topic: topic}
   end
 
-  defp socket_fixture(
-         tenant,
-         topic,
-         presence_key,
-         policies \\ %Policies{
-           broadcast: %BroadcastPolicies{read: true},
-           presence: %PresencePolicies{read: true, write: nil}
-         },
-         private? \\ true,
-         enabled? \\ true
-       ) do
+  defp socket_fixture(tenant, topic, presence_key, opts \\ []) do
+    policies =
+      Keyword.get(opts, :policies, %Policies{
+        broadcast: %BroadcastPolicies{read: true},
+        presence: %PresencePolicies{read: true, write: nil}
+      })
+
+    private? = Keyword.get(opts, :private?, true)
+    enabled? = Keyword.get(opts, :enabled?, true)
+    log_level = Keyword.get(opts, :log_level, :error)
+
     claims = %{sub: random_string(), role: "authenticated", exp: Joken.current_time() + 1_000}
     signer = Joken.Signer.create("HS256", "secret")
-
     jwt = Joken.generate_and_sign!(%{}, claims, signer)
 
     authorization_context =
@@ -267,7 +308,10 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
         presence_rate_counter: rate,
         private?: private?,
         presence_key: presence_key,
-        presence_enabled?: enabled?
+        presence_enabled?: enabled?,
+        log_level: log_level,
+        channel_name: topic,
+        tenant: tenant.external_id
       }
     }
   end