move to erl sys mon

Author: filipecabaco

lib/realtime/monitoring/erl_sys_mon.ex

@@ -24,8 +24,8 @@ defmodule Realtime.ErlSysMon do
     {:ok, []}
   end
 
-  def handle_info({:monitor, pid, _type, _meta} = msg, state) when is_pid(pid) do
-    log_process_info(msg, pid)
+  def handle_info({:monitor, pid, type, _meta} = msg, state) when is_pid(pid) do
+    log_process_info(msg, pid, type)
     {:noreply, state}
   end
 
@@ -34,7 +34,7 @@ defmodule Realtime.ErlSysMon do
     {:noreply, state}
   end
 
-  defp log_process_info(msg, pid) do
+  defp log_process_info(msg, pid, type) do
     pid_info =
       pid
       |> Process.info(:dictionary)
@@ -47,6 +47,7 @@ defmodule Realtime.ErlSysMon do
       end
 
     extra_info = Process.info(pid, [:registered_name, :message_queue_len, :total_heap_size])
+    defensive_exit?(pid, pid_info, type)
 
     Logger.warning(
       "#{__MODULE__} message: " <>
@@ -56,4 +57,14 @@ defmodule Realtime.ErlSysMon do
     _ ->
       Logger.warning("#{__MODULE__} message: " <> inspect(msg))
   end
+
+  defp defensive_exit?(pid, pid_info, type) do
+    {{:"$initial_call", {mod, _, _}}, _} = pid_info
+
+    # credo:disable-for-next-line
+    cond do
+      mod == Realtime.Integration.WebsocketClient && type == :long_message_queue -> Process.exit(pid, :kill)
+      true -> :ok
+    end
+  end
 end

lib/realtime_web/channels/realtime_channel.ex

@@ -29,6 +29,7 @@ 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")
@@ -237,24 +238,16 @@ 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)
-
-    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}
+    if rate_counter.limit.triggered do
+      message = "Too many messages per second"
+      shutdown_response(socket, message)
+    else
+      {:noreply, socket}
     end
   end
 
@@ -380,7 +373,6 @@ 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.15",
+      version: "2.51.16",
       elixir: "~> 1.17.3",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,

test/integration/rt_channel_test.exs

@@ -2485,13 +2485,14 @@ defmodule Realtime.Integration.RtChannelTest do
   end
 
   test "websocket message queue length limit", %{tenant: tenant} do
-    change_tenant_configuration(tenant, :max_events_per_second, 100_000) |> dbg()
+    change_tenant_configuration(tenant, :max_events_per_second, 100_000)
 
     on_exit(fn ->
-      change_tenant_configuration(tenant, :max_events_per_second, 100) |> dbg()
+      change_tenant_configuration(tenant, :max_events_per_second, 100)
     end)
 
     {socket, _} = get_connection(tenant)
+    Process.unlink(socket)
 
     config = %{broadcast: %{self: true}, private: false}
     channel = random_string()
@@ -2504,23 +2505,9 @@ defmodule Realtime.Integration.RtChannelTest 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 %Message{event: "phx_close", topic: ^full_topic}, 500
 
-    assert_receive msg = %Message{
-                     event: "phx_close",
-                     topic: ^full_topic
-                   },
-                   500
+    refute Process.alive?(socket)
   end
 
   defp mode(%{mode: :distributed}) do