fix: realtime channel error cases (#1472)

* fix: realtime channel error cases

Also:

* Don't reset Authorization policies on :confirm_token
* Avoid resetting Authorization policies twice on new "access_token"
* New access token was not being passed to authorization_context
* Capture Connect RPC errors

Author: edgurgel

lib/realtime_web/channels/realtime_channel.ex

@@ -64,8 +64,8 @@ defmodule RealtimeWeb.RealtimeChannel do
          :ok <- limit_channels(socket),
          :ok <- limit_max_users(socket.assigns),
          {:ok, claims, confirm_token_ref, access_token, _} <- confirm_token(socket),
-         {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant_id),
          socket = assign_authorization_context(socket, sub_topic, access_token, claims),
+         {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant_id),
          {:ok, socket} <- maybe_assign_policies(sub_topic, db_conn, socket) do
       tenant_topic = Tenants.tenant_topic(tenant_id, sub_topic, !socket.assigns.private?)
 
@@ -424,19 +424,19 @@ defmodule RealtimeWeb.RealtimeChannel do
   def handle_in("access_token", %{"access_token" => refresh_token}, socket) when is_binary(refresh_token) do
     %{
       assigns: %{
-        access_token: access_token,
+        tenant: tenant_id,
         pg_sub_ref: pg_sub_ref,
         channel_name: channel_name,
-        pg_change_params: pg_change_params,
-        tenant: tenant
+        pg_change_params: pg_change_params
       }
     } = socket
 
-    socket = assign(socket, :access_token, refresh_token)
+    # Update token and reset policies
+    socket = assign(socket, %{access_token: refresh_token, policies: nil})
 
     with {:ok, claims, confirm_token_ref, _, socket} <- confirm_token(socket),
-         socket = assign_authorization_context(socket, channel_name, access_token, claims),
-         {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant),
+         socket = assign_authorization_context(socket, channel_name, refresh_token, claims),
+         {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant_id),
          {:ok, socket} <- maybe_assign_policies(channel_name, db_conn, socket) do
       Helpers.cancel_timer(pg_sub_ref)
       pg_change_params = Enum.map(pg_change_params, &Map.put(&1, :claims, claims))
@@ -455,23 +455,23 @@ defmodule RealtimeWeb.RealtimeChannel do
 
       {:noreply, assign(socket, assigns)}
     else
-      {:error, :unauthorized, msg} ->
-        shutdown_response(socket, msg)
-
-      {:error, :expired_token, msg} ->
+      {:error, reason, msg} when reason in ~w(unauthorized expired_token token_malformed)a ->
         shutdown_response(socket, msg)
 
       {:error, :missing_claims} ->
         shutdown_response(socket, "Fields `role` and `exp` are required in JWT")
 
-      {:error, :token_malformed, msg} ->
-        shutdown_response(socket, msg)
-
       {:error, :unable_to_set_policies, _msg} ->
         shutdown_response(socket, "Realtime was unable to connect to the project database")
 
       {:error, error} ->
         shutdown_response(socket, inspect(error))
+
+      {:error, :rpc_error, :timeout} ->
+        shutdown_response(socket, "Node request timeout")
+
+      {:error, :rpc_error, reason} ->
+        shutdown_response(socket, "RPC call error: " <> inspect(reason))
     end
   end
 
@@ -595,22 +595,14 @@ defmodule RealtimeWeb.RealtimeChannel do
   end
 
   defp confirm_token(%{assigns: assigns} = socket) do
-    %{
-      jwt_secret: jwt_secret,
-      access_token: access_token,
-      tenant: tenant_id
-    } = assigns
+    %{jwt_secret: jwt_secret, access_token: access_token} = assigns
 
-    topic = Map.get(assigns, :topic)
-    socket = Map.put(socket, :policies, nil)
     jwt_jwks = Map.get(assigns, :jwt_jwks)
 
     with jwt_secret_dec <- Crypto.decrypt!(jwt_secret),
          {:ok, %{"exp" => exp} = claims} when is_integer(exp) <-
            ChannelsAuthorization.authorize_conn(access_token, jwt_secret_dec, jwt_jwks),
-         exp_diff when exp_diff > 0 <- exp - Joken.current_time(),
-         {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant_id),
-         {:ok, socket} <- maybe_assign_policies(topic, db_conn, socket) do
+         exp_diff when exp_diff > 0 <- exp - Joken.current_time() do
       if ref = assigns[:confirm_token_ref], do: Helpers.cancel_timer(ref)
 
       interval = min(@confirm_token_ms_interval, exp_diff * 1000)
@@ -761,7 +753,7 @@ defmodule RealtimeWeb.RealtimeChannel do
   end
 
   defp maybe_assign_policies(topic, db_conn, %{assigns: %{private?: true}} = socket)
-       when not is_nil(topic) and not is_nil(db_conn) do
+       when not is_nil(topic) do
     authorization_context = socket.assigns.authorization_context
     policies = socket.assigns.policies || %Policies{}
 
@@ -785,9 +777,7 @@ defmodule RealtimeWeb.RealtimeChannel do
     end
   end
 
-  defp maybe_assign_policies(_, _, socket) do
-    {:ok, assign(socket, policies: nil)}
-  end
+  defp maybe_assign_policies(_, _, socket), do: {:ok, assign(socket, policies: nil)}
 
   defp only_private?(tenant_id, %{assigns: %{private?: private?}}) do
     tenant = Tenants.Cache.get_tenant_by_external_id(tenant_id)

mix.exs

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

test/integration/rt_channel_test.exs

@@ -1621,8 +1621,7 @@ defmodule Realtime.Integration.RtChannelTest do
       tenant = Tenants.get_tenant_by_external_id(tenant.external_id)
       Realtime.Api.update_tenant(tenant, %{jwt_jwks: %{keys: ["potato"]}})
 
-      Process.sleep(500)
-      assert {:error, %Mint.TransportError{reason: :closed}} = WebsocketClient.send_heartbeat(socket)
+      assert_process_down(socket)
     end
 
     test "on jwt_secret the socket closes and sends a system message", %{tenant: tenant, topic: topic} do
@@ -1638,8 +1637,7 @@ defmodule Realtime.Integration.RtChannelTest do
       tenant = Tenants.get_tenant_by_external_id(tenant.external_id)
       Realtime.Api.update_tenant(tenant, %{jwt_secret: "potato"})
 
-      Process.sleep(500)
-      assert {:error, %Mint.TransportError{reason: :closed}} = WebsocketClient.send_heartbeat(socket)
+      assert_process_down(socket)
     end
 
     test "on private_only the socket closes and sends a system message", %{tenant: tenant, topic: topic} do
@@ -1655,8 +1653,7 @@ defmodule Realtime.Integration.RtChannelTest do
       tenant = Tenants.get_tenant_by_external_id(tenant.external_id)
       Realtime.Api.update_tenant(tenant, %{private_only: true})
 
-      Process.sleep(500)
-      assert {:error, %Mint.TransportError{reason: :closed}} = WebsocketClient.send_heartbeat(socket)
+      assert_process_down(socket)
     end
 
     test "on other param changes the socket won't close and no message is sent", %{tenant: tenant, topic: topic} do
@@ -1712,8 +1709,7 @@ defmodule Realtime.Integration.RtChannelTest do
 
       SocketDisconnect.distributed_disconnect(tenant)
 
-      Process.sleep(500)
-      assert {:error, %Mint.TransportError{reason: :closed}} = WebsocketClient.send_heartbeat(socket)
+      assert_process_down(socket)
     end
   end
 
@@ -1746,6 +1742,7 @@ defmodule Realtime.Integration.RtChannelTest do
 
     test "max_events_per_second limit respected", %{tenant: tenant} do
       %{max_events_per_second: max_concurrent_users} = Tenants.get_tenant_by_external_id(tenant.external_id)
+      on_exit(fn -> change_tenant_configuration(tenant, :max_events_per_second, max_concurrent_users) end)
       change_tenant_configuration(tenant, :max_events_per_second, 1)
 
       {socket, _} = get_connection(tenant, "authenticated")
@@ -1755,7 +1752,12 @@ defmodule Realtime.Integration.RtChannelTest do
       WebsocketClient.join(socket, realtime_topic, %{config: config})
 
       for _ <- 1..1000 do
-        WebsocketClient.send_event(socket, realtime_topic, "broadcast", %{})
+        try do
+          WebsocketClient.send_event(socket, realtime_topic, "broadcast", %{})
+        catch
+          _, _ -> :ok
+        end
+
         1..5 |> Enum.random() |> Process.sleep()
       end
 
@@ -1770,8 +1772,7 @@ defmodule Realtime.Integration.RtChannelTest do
                      2000
 
       assert_receive %Message{event: "phx_close"}
-
-      change_tenant_configuration(tenant, :max_events_per_second, max_concurrent_users)
+      assert_process_down(socket)
     end
 
     test "max_channels_per_client limit respected", %{tenant: tenant} do
@@ -2181,9 +2182,8 @@ defmodule Realtime.Integration.RtChannelTest do
     end
 
     # wait to trigger tracker
-    Process.sleep(5000)
+    assert_process_down(socket, 5000)
     assert [] = Tracker.list_pids()
-    assert {:error, %Mint.TransportError{reason: :closed}} = WebsocketClient.send_heartbeat(socket)
   end
 
   test "failed connections are present in tracker with counter counter lower than 0 so they are actioned on by tracker",
@@ -2341,4 +2341,9 @@ defmodule Realtime.Integration.RtChannelTest do
 
     Realtime.Tenants.Cache.invalidate_tenant_cache(external_id)
   end
+
+  defp assert_process_down(pid, timeout \\ 100) do
+    ref = Process.monitor(pid)
+    assert_receive {:DOWN, ^ref, :process, ^pid, _reason}, timeout
+  end
 end

test/realtime/gen_rpc_test.exs

@@ -121,6 +121,36 @@ defmodule Realtime.GenRpcTest do
                       }}
     end
 
+    test "local node exception" do
+      current_node = node()
+
+      assert {:error, :rpc_error, _} = GenRpc.call(current_node, Map, :fetch!, [%{}, :a], tenant_id: "123")
+
+      assert_receive {[:realtime, :rpc], %{latency: _},
+                      %{
+                        origin_node: ^current_node,
+                        target_node: ^current_node,
+                        success: false,
+                        tenant: "123",
+                        mechanism: :gen_rpc
+                      }}
+    end
+
+    test "remote node exception", %{node: node} do
+      current_node = node()
+
+      assert {:error, :rpc_error, _} = GenRpc.call(node, Map, :fetch!, [%{}, :a], tenant_id: "123")
+
+      assert_receive {[:realtime, :rpc], %{latency: _},
+                      %{
+                        origin_node: ^current_node,
+                        target_node: ^node,
+                        success: false,
+                        tenant: "123",
+                        mechanism: :gen_rpc
+                      }}
+    end
+
     @tag extra_config: [{:gen_rpc, :tcp_server_port, 9999}]
     test "bad tcp error", %{node: node} do
       current_node = node()

test/realtime/monitoring/prom_ex/plugins/distributed_test.exs

@@ -57,7 +57,7 @@ defmodule Realtime.PromEx.Plugins.DistributedTest do
 
     test "queue_size", %{metrics: metrics, node: node} do
       pattern = ~r/dist_queue_size{origin_node=\"#{node()}\",target_node=\"#{node}\"}\s(?<number>\d+)/
-      assert metric_value(metrics, pattern) == 0
+      assert is_integer(metric_value(metrics, pattern))
     end
   end