fix: Increase code coverage

coveralls.json

@@ -16,6 +16,7 @@
         "lib/realtime/adapters/postgres/oid_database.ex",
         "lib/realtime/adapters/postgres/protocol/",
         "lib/realtime/application.ex",
+        "lib/realtime/monitoring/prom_ex/plugins/phoenix.ex",
         "lib/realtime/operations.ex",
         "lib/realtime/release.ex",
         "lib/realtime/tenants/authorization/policies/broadcast_policies.ex",

lib/realtime/application.ex

@@ -36,7 +36,7 @@ defmodule Realtime.Application do
       :gen_event.swap_sup_handler(
         :erl_signal_server,
         {:erl_signal_handler, []},
-        {Realtime.SignalHandler, []}
+        {Realtime.SignalHandler, %{handler_mod: :erl_signal_handler}}
       )
 
     Realtime.PromEx.set_metrics_tags()
@@ -53,7 +53,7 @@ defmodule Realtime.Application do
       [
         Realtime.ErlSysMon,
         Realtime.PromEx,
-        Realtime.Telemetry.Logger,
+        {Realtime.Telemetry.Logger, handler_id: "telemetry-logger"},
         Realtime.Repo,
         RealtimeWeb.Telemetry,
         {Cluster.Supervisor, [topologies, [name: Realtime.ClusterSupervisor]]},
@@ -84,8 +84,7 @@ defmodule Realtime.Application do
          strategy: :one_for_one,
          name: Realtime.Tenants.Listen.DynamicSupervisor},
         RealtimeWeb.Endpoint,
-        RealtimeWeb.Presence,
-        Realtime.MetricsCleaner
+        RealtimeWeb.Presence
       ] ++ extensions_supervisors() ++ janitor_tasks()
 
     children =
@@ -130,7 +129,8 @@ defmodule Realtime.Application do
          max_children: janitor_max_children,
          max_seconds: janitor_children_timeout,
          max_restarts: 1},
-        Realtime.Tenants.Janitor
+        Realtime.Tenants.Janitor,
+        Realtime.MetricsCleaner
       ]
     else
       []

lib/realtime/metrics_cleaner.ex

@@ -6,8 +6,7 @@ defmodule Realtime.MetricsCleaner do
 
   defstruct [:check_ref, :interval]
 
-  def start_link(args),
-    do: GenServer.start_link(__MODULE__, args, name: __MODULE__)
+  def start_link(args), do: GenServer.start_link(__MODULE__, args)
 
   def init(_args) do
     interval = Application.get_env(:realtime, :metrics_cleaner_schedule_timer_in_ms)

lib/realtime/monitoring/erl_sys_mon.ex

@@ -7,24 +7,21 @@ defmodule Realtime.ErlSysMon do
 
   require Logger
 
-  def start_link(args) do
-    GenServer.start_link(__MODULE__, args, name: __MODULE__)
-  end
-
-  def init(_args) do
-    :erlang.system_monitor(self(), [
-      :busy_dist_port,
-      :busy_port,
-      {:long_gc, 250},
-      {:long_schedule, 100}
-    ])
+  @defults [
+    :busy_dist_port,
+    :busy_port,
+    {:long_gc, 250},
+    {:long_schedule, 100}
+  ]
+  def start_link(args \\ @defults), do: GenServer.start_link(__MODULE__, args)
 
+  def init(args) do
+    :erlang.system_monitor(self(), args)
     {:ok, []}
   end
 
   def handle_info(msg, state) do
     Logger.warning("#{__MODULE__} message: " <> inspect(msg))
-
     {:noreply, state}
   end
 end

lib/realtime/signal_handler.ex

@@ -12,20 +12,19 @@ defmodule Realtime.SignalHandler do
   end
 
   @impl true
-  def init(_) do
-    Logger.info("#{__MODULE__} is being initialized...")
-    {:ok, %{}}
+  def init({%{handler_mod: _} = args, :ok}) do
+    {:ok, args}
   end
 
   @impl true
-  def handle_event(signal, state) do
+  def handle_event(signal, %{handler_mod: handler_mod} = state) do
     Logger.warning("#{__MODULE__}: #{inspect(signal)} received")
 
     if signal == :sigterm do
       Application.put_env(:realtime, :shutdown_in_progress, true)
     end
 
-    :erl_signal_handler.handle_event(signal, state)
+    handler_mod.handle_event(signal, state)
   end
 
   @impl true

lib/realtime/telemetry/logger.ex

@@ -14,17 +14,12 @@ defmodule Realtime.Telemetry.Logger do
     [:realtime, :rate_counter, :channel, :db_events]
   ]
 
-  def start_link(args \\ []) do
-    GenServer.start_link(__MODULE__, args, name: __MODULE__)
+  def start_link(args) do
+    GenServer.start_link(__MODULE__, args)
   end
 
-  def init(_args) do
-    :telemetry.attach_many(
-      "telemetry-logger",
-      @events,
-      &__MODULE__.handle_event/4,
-      []
-    )
+  def init(handler_id: handler_id) do
+    :telemetry.attach_many(handler_id, @events, &__MODULE__.handle_event/4, [])
 
     {:ok, []}
   end

lib/realtime/tenants.ex

@@ -18,12 +18,6 @@ defmodule Realtime.Tenants do
   @doc """
   Gets a list of connected tenant `external_id` strings in the cluster or a node.
   """
-
-  @spec list_connected_tenants :: [String.t()]
-  def list_connected_tenants() do
-    :syn.group_names(:users)
-  end
-
   @spec list_connected_tenants(atom()) :: [String.t()]
   def list_connected_tenants(node) do
     :syn.group_names(:users, node)

lib/realtime/tenants/janitor.ex

@@ -110,7 +110,11 @@ defmodule Realtime.Tenants.Janitor do
     {:noreply, state}
   end
 
+  # Ignore in coverage has the tests would require to await a random amount of minutes up to an hour
+  # coveralls-ignore-start
   defp timer(%{timer: timer, randomize: true}), do: timer + :timer.minutes(Enum.random(1..59))
+  # coveralls-ignore-stop
+
   defp timer(%{timer: timer}), do: timer
 
   defp perform_mantaince_tasks(tenants), do: Enum.map(tenants, &perform_mantaince_task/1)

mix.exs

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

test/realtime/database_test.exs

@@ -10,7 +10,7 @@ defmodule Realtime.DatabaseTest do
     tenant = tenant_fixture()
 
     # Ensure no replication slot is present before the test
-    ensure_no_replication_slot(tenant)
+    Cleanup.ensure_no_replication_slot()
 
     %{tenant: tenant}
   end
@@ -214,28 +214,4 @@ defmodule Realtime.DatabaseTest do
       assert settings.ssl == false
     end
   end
-
-  defp ensure_no_replication_slot(tenant, attempts \\ 5)
-  defp ensure_no_replication_slot(_tenant, 0), do: raise("Replication slot teardown failed")
-
-  defp ensure_no_replication_slot(tenant, attempts) do
-    {:ok, conn} = Database.connect(tenant, "realtime_test", :stop)
-
-    Database.replication_slot_teardown(tenant)
-
-    case Postgrex.query(conn, "SELECT slot_name FROM pg_replication_slots", []) do
-      {:ok, %{rows: []}} ->
-        :ok
-
-      {:ok, %{rows: rows}} ->
-        IO.inspect(rows)
-        Database.replication_slot_teardown(tenant)
-        Process.sleep(1000)
-        ensure_no_replication_slot(tenant, attempts - 1)
-
-      {:error, _} ->
-        Process.sleep(1000)
-        ensure_no_replication_slot(tenant, attempts - 1)
-    end
-  end
 end

test/realtime/logs_test.exs

@@ -0,0 +1,35 @@
+defmodule Realtime.LogsTest do
+  use ExUnit.Case
+
+  describe "Jason.Encoder implementation" do
+    test "encodes DBConnection.ConnectionError" do
+      error = %DBConnection.ConnectionError{
+        message: "connection lost",
+        reason: :timeout,
+        severity: :error
+      }
+
+      encoded = Jason.encode!(error)
+      assert encoded =~ "message: \"connection lost\""
+      assert encoded =~ "reason: :timeout"
+      assert encoded =~ "severity: :error"
+    end
+
+    test "encodes Postgrex.Error" do
+      error = %Postgrex.Error{
+        message: "relation not found",
+        postgres: %{
+          code: "42P01",
+          schema: "public",
+          table: "users"
+        }
+      }
+
+      encoded = Jason.encode!(error)
+      assert encoded =~ "message: \"relation not found\""
+      assert encoded =~ "schema: \"public\""
+      assert encoded =~ "table: \"users\""
+      assert encoded =~ "code: \"42P01\""
+    end
+  end
+end

test/realtime/metrics_cleaner_test.exs

@@ -0,0 +1,44 @@
+defmodule Realtime.MetricsCleanerTest do
+  use Realtime.DataCase, async: false
+  alias Realtime.MetricsCleaner
+
+  setup do
+    interval = Application.get_env(:realtime, :metrics_cleaner_schedule_timer_in_ms)
+    Application.put_env(:realtime, :metrics_cleaner_schedule_timer_in_ms, 100)
+    tenant = tenant_fixture()
+
+    on_exit(fn ->
+      Application.put_env(:realtime, :metrics_cleaner_schedule_timer_in_ms, interval)
+    end)
+
+    %{tenant: tenant}
+  end
+
+  describe "metrics cleanup" do
+    test "cleans up metrics for users that have been disconnected", %{
+      tenant: %{external_id: external_id}
+    } do
+      start_supervised!(MetricsCleaner)
+      {:ok, _} = Realtime.Tenants.Connect.lookup_or_start_connection(external_id)
+      # Wait for promex to collect the metrics
+      Process.sleep(6000)
+
+      Realtime.Telemetry.execute(
+        [:realtime, :connections],
+        %{connected: 10, connected_cluster: 10, limit: 100},
+        %{tenant: external_id}
+      )
+
+      assert Realtime.PromEx.Metrics
+             |> :ets.select([{{{:_, %{tenant: :"$1"}}, :_}, [], [:"$1"]}])
+             |> Enum.any?(&(&1 == external_id))
+
+      Realtime.Tenants.Connect.shutdown(external_id)
+      Process.sleep(200)
+
+      refute Realtime.PromEx.Metrics
+             |> :ets.select([{{{:_, %{tenant: :"$1"}}, :_}, [], [:"$1"]}])
+             |> Enum.any?(&(&1 == external_id))
+    end
+  end
+end

test/realtime/monitoring/erl_sys_mon_test.exs

@@ -0,0 +1,21 @@
+defmodule Realtime.ErlSysMonTest do
+  use ExUnit.Case
+  import ExUnit.CaptureLog
+  alias Realtime.ErlSysMon
+
+  describe "system monitoring" do
+    setup do
+      Logger.configure(level: :warning)
+      on_exit(fn -> Logger.configure(level: :error) end)
+    end
+
+    test "logs system monitor events" do
+      start_supervised!({ErlSysMon, [{:long_message_queue, {1, 10}}]})
+
+      assert capture_log(fn ->
+               Task.async(fn -> Enum.map(1..10000, &send(self(), &1)) end)
+               |> Task.await()
+             end) =~ "Realtime.ErlSysMon message: "
+    end
+  end
+end

test/realtime/signal_handler_test.exs

@@ -0,0 +1,44 @@
+defmodule Realtime.SignalHandlerTest do
+  use ExUnit.Case
+  import ExUnit.CaptureLog
+  alias Realtime.SignalHandler
+  import Mock
+
+  defmodule FakeHandler do
+    def handle_event(:sigterm, _state), do: :ok
+  end
+
+  setup do
+    Logger.configure(level: :warning)
+
+    on_exit(fn ->
+      Logger.configure(level: :error)
+      Application.put_env(:realtime, :shutdown_in_progress, false)
+    end)
+  end
+
+  describe "signal handling" do
+    test "sends signal to handler_mod" do
+      with_mock FakeHandler, handle_event: fn :sigterm, _state -> :ok end do
+        {:ok, state} = SignalHandler.init({%{handler_mod: FakeHandler}, :ok})
+
+        assert capture_log(fn -> SignalHandler.handle_event(:sigterm, state) end) =~
+                 "SignalHandler: :sigterm received"
+
+        assert_called_exactly(FakeHandler.handle_event(:sigterm, :_), 1)
+      end
+    end
+  end
+
+  describe "shutdown_in_progress?/1" do
+    test "shutdown_in_progress? returns error when shutdown is in progress" do
+      Application.put_env(:realtime, :shutdown_in_progress, true)
+      assert SignalHandler.shutdown_in_progress?() == {:error, :shutdown_in_progress}
+    end
+
+    test "shutdown_in_progress? returns ok when no shutdown in progress" do
+      Application.put_env(:realtime, :shutdown_in_progress, false)
+      assert SignalHandler.shutdown_in_progress?() == :ok
+    end
+  end
+end