fix: Add specific error for increased message rates

filipecabaco · filipecabaco · commit 9d0fb4df5d73 · 2025-07-09T12:58:59.000+01:00
Add specific error for increased message rate errors but also refactored some logging approach to start simplifying that code
diff --git a/README.md b/README.md
@@ -223,6 +223,7 @@ This is the list of operational codes that can help you understand your deployme
 | ChannelRateLimitReached            | The number of channels you can create has reached its limit                                                                                                                                           |
 | ConnectionRateLimitReached         | The number of connected clients as reached its limit                                                                                                                                                  |
 | ClientJoinRateLimitReached         | The rate of joins per second from your clients as reached the channel limits                                                                                                                          |
+| MessagePerSecondRateLimitReached   | The rate of messages per second from your clients as reached the channel limits                                                                                                                       |
 | RealtimeDisabledForTenant          | Realtime has been disabled for the tenant                                                                                                                                                             |
 | UnableToConnectToTenantDatabase    | Realtime was not able to connect to the tenant's database                                                                                                                                             |
 | DatabaseLackOfConnections          | Realtime was not able to connect to the tenant's database due to not having enough available connections                                                                                              |
diff --git a/config/config.exs b/config/config.exs
@@ -51,7 +51,7 @@ config :tailwind,
 # Configures Elixir's Logger
 config :logger, :console,
   format: "$time $metadata[$level] $message\n",
-  metadata: [:request_id, :project, :external_id, :application_name, :error_code]
+  metadata: [:request_id, :project, :external_id, :application_name, :error_code, :sub, :exp, :iss]
 
 # Use Jason for JSON parsing in Phoenix
 config :phoenix, :json_library, Jason
diff --git a/config/test.exs b/config/test.exs
@@ -48,7 +48,7 @@ config :logger,
 # Configures Elixir's Logger
 config :logger, :console,
   format: "$time $metadata[$level] $message\n",
-  metadata: [:request_id, :project, :external_id, :application_name, :sub, :exp, :iss]
+  metadata: [:request_id, :project, :external_id, :application_name, :error_code, :sub, :exp, :iss]
 
 config :opentelemetry,
   span_processor: :simple,
diff --git a/lib/realtime_web/channels/realtime_channel.ex b/lib/realtime_web/channels/realtime_channel.ex
@@ -29,8 +29,8 @@ defmodule RealtimeWeb.RealtimeChannel do
   @confirm_token_ms_interval :timer.minutes(5)
 
   @impl true
-  def join("realtime:", _params, _socket) do
-    Logging.log_error_message(:error, "TopicNameRequired", "You must provide a topic name")
+  def join("realtime:", _params, socket) do
+    Logging.log_error_message(:error, "TopicNameRequired", "You must provide a topic name", socket)
   end
 
   def join("realtime:" <> sub_topic = topic, params, socket) do
@@ -54,9 +54,9 @@ defmodule RealtimeWeb.RealtimeChannel do
 
     with :ok <- SignalHandler.shutdown_in_progress?(),
          :ok <- only_private?(tenant_id, socket),
-         :ok <- limit_joins(socket.assigns),
+         :ok <- limit_joins(socket),
          :ok <- limit_channels(socket),
-         :ok <- limit_max_users(socket.assigns),
+         :ok <- limit_max_users(socket),
          :ok <- start_db_rate_counter(tenant_id),
          {:ok, claims, confirm_token_ref, access_token, _} <- confirm_token(socket),
          {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant_id),
@@ -113,109 +113,84 @@ defmodule RealtimeWeb.RealtimeChannel do
       {:ok, state, assign(socket, assigns)}
     else
       {:error, :expired_token, msg} ->
-        Logging.log_error_with_token_metadata("InvalidJWTToken", msg, socket.assigns.access_token)
+        Logging.log_error_with_token_metadata("InvalidJWTToken", msg, socket)
 
       {:error, :missing_claims} ->
         msg = "Fields `role` and `exp` are required in JWT"
-        Logging.log_error_with_token_metadata("InvalidJWTToken", msg, socket.assigns.access_token)
+        Logging.log_error_with_token_metadata("InvalidJWTToken", msg, socket)
 
       {:error, :unauthorized, msg} ->
-        Logging.log_error_message(:error, "Unauthorized", msg)
+        Logging.log_error_message(:error, "Unauthorized", msg, socket)
 
       {:error, :too_many_channels} ->
         msg = "Too many channels"
-        Logging.log_error_message(:error, "ChannelRateLimitReached", msg)
+        Logging.log_error_message(:error, "ChannelRateLimitReached", msg, socket)
 
       {:error, :too_many_connections} ->
         msg = "Too many connected users"
-        Logging.log_error_message(:error, "ConnectionRateLimitReached", msg)
+        Logging.log_error_message(:error, "ConnectionRateLimitReached", msg, socket)
 
       {:error, :too_many_joins} ->
         msg = "Too many joins per second"
-        Logging.log_error_message(:error, "ClientJoinRateLimitReached", msg)
+        Logging.log_error_message(:error, "ClientJoinRateLimitReached", msg, socket)
 
       {:error, :increase_connection_pool} ->
         msg = "Please increase your connection pool size"
-        Logging.log_error_message(:error, "IncreaseConnectionPool", msg)
+        Logging.log_error_message(:error, "IncreaseConnectionPool", msg, socket)
 
       {:error, :tenant_db_too_many_connections} ->
         msg = "Database can't accept more connections, Realtime won't connect"
-        Logging.log_error_message(:error, "DatabaseLackOfConnections", msg)
+        Logging.log_error_message(:error, "DatabaseLackOfConnections", msg, socket)
 
       {:error, :unable_to_set_policies, error} ->
-        Logging.log_error_message(:error, "UnableToSetPolicies", error)
+        Logging.log_error_message(:error, "UnableToSetPolicies", error, socket)
         {:error, %{reason: "Realtime was unable to connect to the project database"}}
 
       {:error, :tenant_database_unavailable} ->
         Logging.log_error_message(
           :error,
           "UnableToConnectToProject",
-          "Realtime was unable to connect to the project database"
+          "Realtime was unable to connect to the project database",
+          socket
         )
 
       {:error, :rpc_error, :timeout} ->
-        Logging.log_error_message(:error, "TimeoutOnRpcCall", "Node request timeout")
+        Logging.log_error_message(:error, "TimeoutOnRpcCall", "Node request timeout", socket)
 
       {:error, :rpc_error, reason} ->
-        Logging.log_error_message(:error, "ErrorOnRpcCall", "RPC call error: " <> inspect(reason))
+        Logging.log_error_message(:error, "ErrorOnRpcCall", "RPC call error: " <> inspect(reason), socket)
 
       {:error, :initializing} ->
-        Logging.log_error_message(
-          :error,
-          "InitializingProjectConnection",
-          "Realtime is initializing the project connection"
-        )
+        msg = "Realtime is initializing the project connection"
+        Logging.log_error_message(:error, "InitializingProjectConnection", msg, socket)
 
       {:error, :tenant_database_connection_initializing} ->
-        Logging.log_error_message(
-          :error,
-          "InitializingProjectConnection",
-          "Connecting to the project database"
-        )
+        msg = "Connecting to the project database"
+        Logging.log_error_message(:error, "InitializingProjectConnection", msg, socket)
 
       {:error, :token_malformed, msg} ->
-        Logging.log_error_message(:error, "MalformedJWT", msg)
+        Logging.log_error_message(:error, "MalformedJWT", msg, socket)
 
       {:error, invalid_exp} when is_integer(invalid_exp) and invalid_exp <= 0 ->
-        Logging.log_error_with_token_metadata(
-          "InvalidJWTToken",
-          "Token expiration time is invalid",
-          socket.assigns.access_token
-        )
+        Logging.log_error_with_token_metadata("InvalidJWTToken", "Token expiration time is invalid", socket)
 
       {:error, :private_only} ->
-        Logging.log_error_message(
-          :error,
-          "PrivateOnly",
-          "This project only allows private channels"
-        )
+        Logging.log_error_message(:error, "PrivateOnly", "This project only allows private channels", socket)
 
       {:error, :tenant_not_found} ->
-        Logging.log_error_message(
-          :error,
-          "TenantNotFound",
-          "Tenant with the given ID does not exist"
-        )
+        Logging.log_error_message(:error, "TenantNotFound", "Tenant with the given ID does not exist", socket)
 
       {:error, :tenant_suspended} ->
-        Logging.log_error_message(
-          :error,
-          "RealtimeDisabledForTenant",
-          "Realtime disabled for this tenant"
-        )
+        Logging.log_error_message(:error, "RealtimeDisabledForTenant", "Realtime disabled for this tenant", socket)
 
       {:error, :signature_error} ->
-        Logging.log_error_message(:error, "JwtSignatureError", "Failed to validate JWT signature")
+        Logging.log_error_message(:error, "JwtSignatureError", "Failed to validate JWT signature", socket)
 
       {:error, :shutdown_in_progress} ->
-        Logging.log_error_message(
-          :error,
-          "RealtimeRestarting",
-          "Realtime is restarting, please standby"
-        )
+        Logging.log_error_message(:error, "RealtimeRestarting", "Realtime is restarting, please standby", socket)
 
       {:error, error} ->
-        Logging.log_error_message(:error, "UnknownErrorOnChannel", error)
+        Logging.log_error_message(:error, "UnknownErrorOnChannel", error, socket)
         {:error, %{reason: "Unknown Error on Channel"}}
     end
   end
@@ -227,7 +202,7 @@ defmodule RealtimeWeb.RealtimeChannel do
       )
       when avg > max do
     message = "Too many messages per second"
-
+    Logging.log_error_message(:warning, "MessagePerSecondRateLimitReached", message, socket)
     shutdown_response(socket, message)
   end
 
@@ -379,6 +354,7 @@ defmodule RealtimeWeb.RealtimeChannel do
   def handle_in(_, _, %{assigns: %{rate_counter: %{avg: avg}, limits: %{max_events_per_second: max}}} = socket)
       when avg > max do
     message = "Too many messages per second"
+    Logging.log_error_message(:error, "MessagePerSecondRateLimitReached", message, socket)
 
     shutdown_response(socket, message)
   end
@@ -478,7 +454,7 @@ defmodule RealtimeWeb.RealtimeChannel do
     wait
   end
 
-  def limit_joins(%{tenant: tenant, limits: limits}) do
+  def limit_joins(%{assigns: %{tenant: tenant, limits: limits}} = socket) do
     id = Tenants.joins_per_second_key(tenant)
     GenCounter.new(id)
 
@@ -501,7 +477,7 @@ defmodule RealtimeWeb.RealtimeChannel do
         {:error, :too_many_joins}
 
       error ->
-        Logging.log_error_message(:error, "UnknownErrorOnCounter", error)
+        Logging.log_error_message(:error, "UnknownErrorOnCounter", error, socket)
         {:error, error}
     end
   end
@@ -517,7 +493,7 @@ defmodule RealtimeWeb.RealtimeChannel do
     end
   end
 
-  defp limit_max_users(%{limits: %{max_concurrent_users: max_conn_users}, tenant: tenant}) do
+  defp limit_max_users(%{assigns: %{limits: %{max_concurrent_users: max_conn_users}, tenant: tenant}}) do
     conns = Realtime.UsersCounter.tenant_users(tenant)
 
     if conns < max_conn_users,
@@ -645,10 +621,9 @@ defmodule RealtimeWeb.RealtimeChannel do
   end
 
   defp shutdown_response(socket, message) when is_binary(message) do
-    %{assigns: %{channel_name: channel_name, access_token: access_token}} = socket
-    metadata = log_metadata(access_token)
+    %{assigns: %{channel_name: channel_name}} = socket
     push_system_message("system", socket, "error", message, channel_name)
-    log_warning("ChannelShutdown", message, metadata)
+    Logging.log_warning_with_token_metadata("ChannelShutdown", message, socket)
     {:stop, :normal, socket}
   end
 
@@ -820,17 +795,4 @@ defmodule RealtimeWeb.RealtimeChannel do
       do: {:error, :private_only},
       else: :ok
   end
-
-  defp log_metadata(access_token) do
-    access_token
-    |> Joken.peek_claims()
-    |> then(fn
-      {:ok, claims} -> Map.get(claims, "sub")
-      _ -> nil
-    end)
-    |> then(fn
-      nil -> []
-      sub -> [sub: sub]
-    end)
-  end
 end
diff --git a/lib/realtime_web/channels/realtime_channel/logging.ex b/lib/realtime_web/channels/realtime_channel/logging.ex
@@ -43,13 +43,12 @@ defmodule RealtimeWeb.RealtimeChannel.Logging do
   @spec log_error_with_token_metadata(
           code :: binary(),
           msg :: binary(),
-          token :: Joken.bearer_token(),
-          metadata :: keyword()
+          socket :: Phoenix.Socket.t()
         ) :: {:error, %{reason: binary()}}
-  def log_error_with_token_metadata(code, msg, token, metadata \\ []) do
-    if metadata == [], do: Logger.metadata()
-    metadata = update_metadata_with_token_claims(metadata, token)
-    log_error_message(:error, code, msg, metadata)
+  def log_error_with_token_metadata(code, msg, %{assigns: %{access_token: access_token, tenant: tenant_id}} = socket) do
+    Logger.metadata(external_id: tenant_id, project: tenant_id)
+    update_metadata_with_token_claims(access_token)
+    log_error_message(:error, code, msg, socket)
   end
 
   @doc """
@@ -58,25 +57,25 @@ defmodule RealtimeWeb.RealtimeChannel.Logging do
   @spec log_warning_with_token_metadata(
           code :: binary(),
           msg :: binary(),
-          token :: Joken.bearer_token(),
-          metadata :: keyword()
+          socket :: Phoenix.Socket.t()
         ) :: {:error, %{reason: binary()}}
-  def log_warning_with_token_metadata(code, msg, token, metadata \\ []) do
-    if metadata == [], do: Logger.metadata()
-    metadata = update_metadata_with_token_claims(metadata, token)
-    log_error_message(:warning, code, msg, metadata)
+  def log_warning_with_token_metadata(code, msg, %{assigns: %{access_token: access_token, tenant: tenant_id}} = socket) do
+    Logger.metadata(external_id: tenant_id, project: tenant_id)
+    update_metadata_with_token_claims(access_token)
+    log_error_message(:error, code, msg, socket)
   end
 
-  defp update_metadata_with_token_claims(metadata, token) do
+  defp update_metadata_with_token_claims(token) do
     case Joken.peek_claims(token) do
       {:ok, claims} ->
         sub = Map.get(claims, "sub")
         exp = Map.get(claims, "exp")
         iss = Map.get(claims, "iss")
-        metadata ++ [sub: sub, exp: exp, iss: iss]
+
+        Logger.metadata(sub: sub, exp: exp, iss: iss)
 
       _ ->
-        metadata
+        nil
     end
   end
 
@@ -122,19 +121,22 @@ defmodule RealtimeWeb.RealtimeChannel.Logging do
           level :: :error | :warning,
           code :: binary(),
           error :: term(),
-          keyword()
+          socket :: Phoenix.Socket.t()
         ) :: {:error, %{reason: binary()}}
-  def log_error_message(level, code, error, metadata \\ [])
+  def log_error_message(level, code, error, socket)
 
-  def log_error_message(:error, code, error, metadata) do
+  def log_error_message(:error, code, error, %{assigns: %{tenant: tenant_id}}) do
     if code in system_errors(), do: Telemetry.execute([:realtime, :channel, :error], %{code: code}, %{code: code})
+    Logger.metadata(external_id: tenant_id, project: tenant_id)
 
-    log_error(code, error, metadata)
+    log_error(code, error, Logger.metadata())
     {:error, %{reason: error}}
   end
 
-  def log_error_message(:warning, code, error, metadata) do
-    log_warning(code, error, metadata)
+  def log_error_message(:warning, code, error, %{assigns: %{tenant: tenant_id}}) do
+    Logger.metadata(external_id: tenant_id, project: tenant_id)
+
+    log_warning(code, error, Logger.metadata())
     {:error, %{reason: error}}
   end
 end
diff --git a/lib/realtime_web/channels/user_socket.ex b/lib/realtime_web/channels/user_socket.ex
@@ -28,11 +28,12 @@ defmodule RealtimeWeb.UserSocket do
       %{uri: %{host: host}, x_headers: headers} = opts
 
       {:ok, external_id} = Database.get_external_id(host)
+      token = access_token(params, headers)
+      socket = socket |> assign(:access_token, token) |> assign(:tenant, external_id)
+
       Logger.metadata(external_id: external_id, project: external_id)
       Logger.put_process_level(self(), :error)
 
-      token = access_token(params, headers)
-
       with %Tenant{
              extensions: extensions,
              jwt_secret: jwt_secret,
@@ -68,19 +69,20 @@ defmodule RealtimeWeb.UserSocket do
         }
 
         assigns = Map.from_struct(assigns)
-
-        {:ok, assign(socket, assigns)}
+        socket = assign(socket, assigns)
+        {:ok, socket}
       else
         nil ->
           log_error("TenantNotFound", "Tenant not found: #{external_id}")
           {:error, :tenant_not_found}
 
         {:error, :expired_token, msg} ->
-          Logging.log_error_with_token_metadata("InvalidJWTToken", msg, token)
+          Logging.log_error_with_token_metadata("InvalidJWTToken", msg, socket)
           {:error, :expired_token}
 
         {:error, :missing_claims} ->
-          Logging.log_error_with_token_metadata("InvalidJWTToken", "Fields `role` and `exp` are required in JWT", token)
+          msg = "Fields `role` and `exp` are required in JWT"
+          Logging.log_error_with_token_metadata("InvalidJWTToken", msg, socket)
           {:error, :missing_claims}
 
         {:error, :token_malformed} ->
diff --git a/mix.exs b/mix.exs
@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
   def project do
     [
       app: :realtime,
-      version: "2.40.2",
+      version: "2.40.3",
       elixir: "~> 1.17.3",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,
diff --git a/test/integration/rt_channel_test.exs b/test/integration/rt_channel_test.exs
diff --git a/test/realtime_web/channels/realtime_channel/logging_test.exs b/test/realtime_web/channels/realtime_channel/logging_test.exs
diff --git a/test/support/websocket_client.ex b/test/support/websocket_client.ex

Original file line number	Diff line number	Diff line change
`@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do`
`4`	`4`	`def project do`
`5`	`5`	`[`
`6`	`6`	`app: :realtime,`
`7`		`- version: "2.40.2",`
	`7`	`+ version: "2.40.3",`
`8`	`8`	`elixir: "~> 1.17.3",`
`9`	`9`	`elixirc_paths: elixirc_paths(Mix.env()),`
`10`	`10`	`start_permanent: Mix.env() == :prod,`