fix: set max socket message queue length

We're using the event to update the counter to check if the websocket has too many messages. If so we will consider abusive behaviour and will kill the socket to prevent further damage.

Author: filipecabaco

lib/realtime/helpers.ex

@@ -4,6 +4,9 @@ defmodule Realtime.Helpers do
   """
   require Logger
 
+  @doc """
+  Cancels a timer.
+  """
   @spec cancel_timer(reference() | nil) :: non_neg_integer() | false | :ok | nil
   def cancel_timer(nil), do: nil
   def cancel_timer(ref), do: Process.cancel_timer(ref)

lib/realtime_web/channels/realtime_channel.ex

@@ -29,7 +29,6 @@ defmodule RealtimeWeb.RealtimeChannel do
   alias RealtimeWeb.RealtimeChannel.Tracker
 
   @confirm_token_ms_interval :timer.minutes(5)
-
   @impl true
   def join("realtime:", _params, socket) do
     log_error(socket, "TopicNameRequired", "You must provide a topic name")
@@ -238,16 +237,24 @@ defmodule RealtimeWeb.RealtimeChannel do
     {:noreply, socket}
   end
 
+  @websocket_message_queue_length_limit 1000
   def handle_info(:update_rate_counter, socket) do
     count(socket)
 
     {:ok, rate_counter} = RateCounter.get(socket.assigns.rate_counter)
+    {:message_queue_len, len} = Process.info(self(), :message_queue_len) |> dbg()
 
-    if rate_counter.limit.triggered do
-      message = "Too many messages per second"
-      shutdown_response(socket, message)
-    else
-      {:noreply, socket}
+    cond do
+      rate_counter.limit.triggered ->
+        message = "Too many messages per second"
+        shutdown_response(socket, message)
+
+      len > @websocket_message_queue_length_limit ->
+        message = "Websocket message queue length is too long"
+        shutdown_response(socket, message)
+
+      true ->
+        {:noreply, socket}
     end
   end
 
@@ -373,6 +380,7 @@ defmodule RealtimeWeb.RealtimeChannel do
   end
 
   def handle_info(:sync_presence, socket), do: {:noreply, socket}
+
   def handle_info(_, socket), do: {:noreply, socket}
 
   @impl true

mix.exs

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

test/integration/rt_channel_test.exs

@@ -2484,6 +2484,45 @@ defmodule Realtime.Integration.RtChannelTest do
     end
   end
 
+  test "websocket message queue length limit", %{tenant: tenant} do
+    change_tenant_configuration(tenant, :max_events_per_second, 100_000) |> dbg()
+
+    on_exit(fn ->
+      change_tenant_configuration(tenant, :max_events_per_second, 100) |> dbg()
+    end)
+
+    {socket, _} = get_connection(tenant)
+
+    config = %{broadcast: %{self: true}, private: false}
+    channel = random_string()
+    full_topic = "realtime:#{channel}"
+
+    WebsocketClient.join(socket, full_topic, %{config: config})
+    assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}, topic: ^full_topic}, 500
+
+    for _ <- 1..3000, Process.alive?(socket) do
+      spawn(fn -> WebsocketClient.send_event(socket, full_topic, "broadcast", %{"msg" => random_string()}) end)
+    end
+
+    assert_receive msg = %Message{
+                     event: "system",
+                     topic: ^full_topic,
+                     payload: %{
+                       "channel" => ^channel,
+                       "extension" => "system",
+                       "message" => "Websocket message queue length is too long",
+                       "status" => "error"
+                     }
+                   },
+                   500
+
+    assert_receive msg = %Message{
+                     event: "phx_close",
+                     topic: ^full_topic
+                   },
+                   500
+  end
+
   defp mode(%{mode: :distributed}) do
     tenant = Api.get_tenant_by_external_id("dev_tenant")