fix: tick most RateCounters every 5 seconds instead of every second (#1787)

* fix: ensure RateCounter avg is always per second
* fix: change tenant rate counters to tick every 5 seconds

Author: edgurgel

lib/realtime/rate_counter/rate_counter.ex

@@ -4,6 +4,8 @@ defmodule Realtime.RateCounter do
 
   These rate counters use the GenCounter module.
   Start your RateCounter here and increment it with a `GenCounter.add/1` call, for example.
+
+  Average is calculated as the average number of events per second
   """
 
   use GenServer
@@ -208,7 +210,8 @@ defmodule Realtime.RateCounter do
     bucket_len = Enum.count(bucket)
 
     sum = Enum.sum(bucket)
-    avg = sum / bucket_len
+
+    avg = sum / bucket_len / (state.tick / 1_000)
 
     state = %{state | bucket: bucket, sum: sum, avg: avg}
 

lib/realtime/tenants.ex

@@ -151,6 +151,8 @@ defmodule Realtime.Tenants do
   @spec joins_per_second_rate(String.t(), non_neg_integer) :: RateCounter.Args.t()
   def joins_per_second_rate(tenant_id, max_joins_per_second) when is_binary(tenant_id) do
     opts = [
+      tick: :timer.seconds(5),
+      max_bucket_len: 12,
       telemetry: %{
         event_name: [:channel, :joins],
         measurements: %{limit: max_joins_per_second},
@@ -197,6 +199,8 @@ defmodule Realtime.Tenants do
 
   def events_per_second_rate(tenant_id, max_events_per_second) do
     opts = [
+      tick: :timer.seconds(5),
+      max_bucket_len: 12,
       telemetry: %{
         event_name: [:channel, :events],
         measurements: %{limit: max_events_per_second},
@@ -244,6 +248,8 @@ defmodule Realtime.Tenants do
   @spec db_events_per_second_rate(String.t(), non_neg_integer) :: RateCounter.Args.t()
   def db_events_per_second_rate(tenant_id, max_events_per_second) when is_binary(tenant_id) do
     opts = [
+      tick: :timer.seconds(5),
+      max_bucket_len: 12,
       telemetry: %{
         event_name: [:channel, :db_events],
         measurements: %{},
@@ -290,6 +296,8 @@ defmodule Realtime.Tenants do
   @spec presence_events_per_second_rate(String.t(), non_neg_integer) :: RateCounter.Args.t()
   def presence_events_per_second_rate(tenant_id, max_presence_events_per_second) do
     opts = [
+      tick: :timer.seconds(5),
+      max_bucket_len: 12,
       telemetry: %{
         event_name: [:channel, :presence_events],
         measurements: %{limit: max_presence_events_per_second},

test/integration/rt_channel/connection_lifecycle_test.exs

@@ -305,7 +305,7 @@ defmodule Realtime.Integration.RtChannel.ConnectionLifecycleTest do
 
       log =
         capture_log(fn ->
-          for _ <- 1..300 do
+          for _ <- 1..1500 do
             WebsocketClient.join(socket, realtime_topic, %{config: config})
             assert_receive %Message{event: "phx_reply", payload: %{"status" => "ok"}, topic: ^realtime_topic}, 500
           end

test/realtime/rate_counter/rate_counter_test.exs

@@ -174,7 +174,7 @@ defmodule Realtime.RateCounterTest do
 
       log =
         capture_log(fn ->
-          GenCounter.add(args.id, 50)
+          GenCounter.add(args.id, 6)
           Process.sleep(300)
         end)
 
@@ -185,7 +185,7 @@ defmodule Realtime.RateCounterTest do
       # Splitting by the error message returns the error message and the rest of the log only
       assert length(String.split(log, "ErrorMessage: Reason")) == 2
 
-      Process.sleep(300)
+      Process.sleep(400)
 
       assert {:ok, %RateCounter{limit: %{triggered: false}}} = RateCounter.get(args)
     end
@@ -301,6 +301,72 @@ defmodule Realtime.RateCounterTest do
     end
   end
 
+  describe "avg normalization" do
+    test "avg represents events per second regardless of tick interval" do
+      # 1-second tick: add 10 events → avg should be ~10 events/second
+      id_1s = {:domain, :metric, Ecto.UUID.generate()}
+      args_1s = %Args{id: id_1s, opts: [tick: 1_000, max_bucket_len: 1]}
+      {:ok, pid} = RateCounter.new(args_1s)
+      # wait for init to complete
+      :sys.get_state(pid)
+
+      GenCounter.add(id_1s, 10)
+      {:ok, state_1s} = RateCounterHelper.tick!(args_1s)
+      assert_in_delta state_1s.avg, 10.0, 0.01
+
+      # 5-second tick: add 50 events (= 10 per second) → avg should also be ~10 events/second
+      id_5s = {:domain, :metric, Ecto.UUID.generate()}
+      args_5s = %Args{id: id_5s, opts: [tick: 5_000, max_bucket_len: 1]}
+      {:ok, pid} = RateCounter.new(args_5s)
+      # wait for init to complete
+      :sys.get_state(pid)
+
+      GenCounter.add(id_5s, 50)
+      {:ok, state_5s} = RateCounterHelper.tick!(args_5s)
+      assert_in_delta state_5s.avg, 10.0, 0.01
+    end
+
+    test "avg limit triggers and unsets correctly with a non-1-second tick" do
+      id = {:domain, :metric, Ecto.UUID.generate()}
+
+      args = %Args{
+        id: id,
+        opts: [
+          tick: 5_000,
+          max_bucket_len: 1,
+          limit: [
+            value: 10,
+            measurement: :avg,
+            log_fn: fn ->
+              Logger.warning("RateLimitReached", external_id: "tenant123", project: "tenant123")
+            end
+          ]
+        ]
+      }
+
+      {:ok, pid} = RateCounter.new(args)
+      # wait for init to complete
+      :sys.get_state(pid)
+
+      # 60 events over a 5-second tick = 12 events/second, above the 10/s limit
+      log =
+        capture_log(fn ->
+          GenCounter.add(id, 60)
+          RateCounterHelper.tick!(args)
+        end)
+
+      assert {:ok, %RateCounter{avg: avg, limit: %{triggered: true}}} = RateCounter.get(args)
+      assert_in_delta avg, 12.0, 0.01
+      assert log =~ "RateLimitReached"
+
+      # 40 events over a 5-second tick = 8 events/second, below the 10/s limit
+      GenCounter.add(id, 40)
+      RateCounterHelper.tick!(args)
+      assert {:ok, %RateCounter{avg: avg, limit: %{triggered: false}}} = RateCounter.get(args)
+      assert_in_delta avg, 8.0, 0.01
+    end
+  end
+
   describe "publish_update/1" do
     test "cause shutdown with update message from update topic" do
       args = %Args{id: {:domain, :metric, Ecto.UUID.generate()}}

test/realtime_web/channels/realtime_channel/presence_handler_test.exs

@@ -450,7 +450,7 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
 
       log =
         capture_log(fn ->
-          for _ <- 1..300, do: PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
+          for _ <- 1..1500, do: PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
 
           {:ok, _} = RateCounterHelper.tick!(Tenants.presence_events_per_second_rate(tenant))
 
@@ -466,7 +466,7 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
 
       log =
         capture_log(fn ->
-          for _ <- 1..300, do: PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
+          for _ <- 1..1500, do: PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
 
           {:ok, _} = RateCounterHelper.tick!(Tenants.presence_events_per_second_rate(tenant))
 
@@ -535,7 +535,7 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
 
       log =
         capture_log(fn ->
-          for _ <- 1..300, do: PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
+          for _ <- 1..1500, do: PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
 
           {:ok, _} = RateCounterHelper.tick!(Tenants.presence_events_per_second_rate(tenant))
 
@@ -552,7 +552,7 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
 
       log =
         capture_log(fn ->
-          for _ <- 1..300, do: PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
+          for _ <- 1..1500, do: PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
 
           {:ok, _} = RateCounterHelper.tick!(Tenants.presence_events_per_second_rate(tenant))