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 == RealtimeWeb.RealtimeChannel && 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

@@ -2484,45 +2484,6 @@ 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")
 

test/realtime/monitoring/erl_sys_mon_test.exs

@@ -12,7 +12,7 @@ defmodule Realtime.Monitoring.ErlSysMonTest do
           Task.async(fn ->
             Process.register(self(), TestProcess)
             Enum.map(1..1000, &send(self(), &1))
-            # Wait  for ErlSysMon to notice
+
             Process.sleep(4000)
           end)
           |> Task.await()
@@ -26,4 +26,81 @@ defmodule Realtime.Monitoring.ErlSysMonTest do
       assert log =~ "total_heap_size: "
     end
   end
+
+  test "ErlSysMon kills RealtimeChannel process with long message queue" do
+    start_supervised!({ErlSysMon, config: [{:long_message_queue, {0, 50}}]})
+    {:ok, channel_pid} = create_mock_realtime_channel()
+    ref = Process.monitor(channel_pid)
+    Process.unlink(channel_pid)
+    for i <- 1..10_000, do: send(channel_pid, {:test_message, "message_#{i}"})
+
+    assert_receive {:DOWN, ^ref, :process, ^channel_pid, :killed}, 5000
+    refute Process.alive?(channel_pid)
+  end
+
+  test "ErlSysMon does not kill non-RealtimeChannel processes with long message queue" do
+    start_supervised!({ErlSysMon, config: [{:long_message_queue, {0, 50}}]})
+    {:ok, regular_pid} = create_regular_process()
+    Process.unlink(regular_pid)
+    ref = Process.monitor(regular_pid)
+    for i <- 1..10_000, do: send(regular_pid, {:test_message, "message_#{i}"})
+    Process.sleep(2000)
+
+    assert Process.alive?(regular_pid)
+
+    Process.exit(regular_pid, :kill)
+    assert_receive {:DOWN, ^ref, :process, ^regular_pid, :killed}
+  end
+
+  test "ErlSysMon logs warning for RealtimeChannel long message queue" do
+    start_supervised!({ErlSysMon, config: [{:long_message_queue, {0, 50}}]})
+    {:ok, channel_pid} = create_mock_realtime_channel()
+    Process.unlink(channel_pid)
+
+    log =
+      capture_log(fn ->
+        for i <- 1..10_000, do: send(channel_pid, {:test_message, "message_#{i}"})
+        Process.sleep(1000)
+      end)
+
+    assert log =~ "Realtime.ErlSysMon message:"
+    assert log =~ "RealtimeWeb.RealtimeChannel"
+    assert log =~ "long_message_queue"
+  end
+
+  defp create_mock_realtime_channel do
+    pid =
+      spawn_link(fn ->
+        Process.put(:"$initial_call", {RealtimeWeb.RealtimeChannel, :init, 1})
+
+        slow_message_processor()
+      end)
+
+    {:ok, pid}
+  end
+
+  defp create_regular_process do
+    pid =
+      spawn_link(fn ->
+        Process.put(:"$initial_call", {SomeOtherModule, :init, 1})
+
+        slow_message_processor()
+      end)
+
+    {:ok, pid}
+  end
+
+  defp slow_message_processor do
+    receive do
+      {:test_message, _content} ->
+        Process.sleep(10)
+        slow_message_processor()
+
+      :stop ->
+        :ok
+
+      _other ->
+        slow_message_processor()
+    end
+  end
 end