fix: Performance improvements (#845)

* Remove the revert update on write checks in Channel, Broadcast and Presence * Separates the checks into a transaction PER feature to avoid extra locking and enable conn pool to be more heavily used
supabase · Apr 12, 2024 · 4a02133 · 4a02133
1 parent 6afbc8b
commit 4a02133
Show file tree

Hide file tree

Showing 5 changed files with 41 additions and 35 deletions.
diff --git a/lib/realtime/tenants/authorization.ex b/lib/realtime/tenants/authorization.ex
@@ -130,29 +130,40 @@ defmodule Realtime.Tenants.Authorization do
 
   @policies_mods [ChannelPolicies, BroadcastPolicies, PresencePolicies]
   defp get_policies_for_connection(conn, authorization_context) do
-    Helpers.transaction(conn, fn transaction_conn ->
-      set_conn_config(transaction_conn, authorization_context)
-
-      Enum.reduce_while(@policies_mods, %Policies{}, fn policies_mod, policies ->
-        with {:ok, policies} <-
-               policies_mod.check_write_policies(
-                 transaction_conn,
-                 policies,
-                 authorization_context
-               ),
-             {:ok, policies} <-
-               policies_mod.check_read_policies(
-                 transaction_conn,
-                 policies,
-                 authorization_context
-               ) do
-          {:cont, policies}
-        else
-          {:error, error} ->
-            Postgrex.rollback(transaction_conn, error)
-            {:halt, {:error, error}}
-        end
-      end)
+    Enum.reduce_while(@policies_mods, %Policies{}, fn policies_mod, policies ->
+      res =
+        Helpers.transaction(conn, fn transaction_conn ->
+          set_conn_config(transaction_conn, authorization_context)
+
+          get_policy_for_connection_and_feature(
+            policies_mod,
+            transaction_conn,
+            policies,
+            authorization_context
+          )
+        end)
+
+      case res do
+        {:error, error} -> {:halt, {:error, error}}
+        %DBConnection.TransactionError{} = err -> {:halt, err}
+        policy -> {:cont, policy}
+      end
     end)
   end
+
+  defp get_policy_for_connection_and_feature(
+         policy_mod,
+         transaction_conn,
+         policies,
+         authorization_context
+       ) do
+    with {:ok, policies} <-
+           policy_mod.check_write_policies(transaction_conn, policies, authorization_context),
+         {:ok, policies} <-
+           policy_mod.check_read_policies(transaction_conn, policies, authorization_context) do
+      policies
+    else
+      {:error, error} -> Postgrex.rollback(transaction_conn, error)
+    end
+  end
 end
diff --git a/lib/realtime/tenants/authorization/policies/broadcast_policies.ex b/lib/realtime/tenants/authorization/policies/broadcast_policies.ex
@@ -68,9 +68,8 @@ defmodule Realtime.Tenants.Authorization.Policies.BroadcastPolicies do
           changeset = Broadcast.check_changeset(broadcast, %{check: true})
 
           case Repo.update(conn, changeset, Broadcast, mode: :savepoint) do
-            {:ok, %Broadcast{check: true} = broadcast} ->
-              revert_changeset = Broadcast.check_changeset(broadcast, %{check: false})
-              {:ok, _} = Repo.update(conn, revert_changeset, Broadcast)
+            {:ok, %Broadcast{check: true}} ->
+              Postgrex.query!(transaction_conn, "ROLLBACK AND CHAIN", [])
               Policies.update_policies(policies, :broadcast, :write, true)
 
             {:error, %Postgrex.Error{postgres: %{code: :insufficient_privilege}}} ->
@@ -83,8 +82,6 @@ defmodule Realtime.Tenants.Authorization.Policies.BroadcastPolicies do
               Logger.error(
                 "Error getting broadcast write policies for connection: #{inspect(error)}"
               )
-
-              Postgrex.rollback(transaction_conn, error)
           end
 
         {:error, :not_found} ->

diff --git a/lib/realtime/tenants/authorization/policies/channel_policies.ex b/lib/realtime/tenants/authorization/policies/channel_policies.ex
@@ -80,9 +80,8 @@ defmodule Realtime.Tenants.Authorization.Policies.ChannelPolicies do
       changeset = Channel.check_changeset(channel, %{check: true})
 
       case Repo.update(transaction_conn, changeset, Channel, mode: :savepoint) do
-        {:ok, %Channel{check: true} = channel} ->
-          revert_changeset = Channel.check_changeset(channel, %{check: false})
-          {:ok, _} = Repo.update(transaction_conn, revert_changeset, Channel)
+        {:ok, %Channel{check: true}} ->
+          Postgrex.query!(transaction_conn, "ROLLBACK AND CHAIN", [])
           Policies.update_policies(policies, :channel, :write, true)
 
         {:ok, _} ->

diff --git a/lib/realtime/tenants/authorization/policies/presence_policies.ex b/lib/realtime/tenants/authorization/policies/presence_policies.ex
@@ -68,9 +68,8 @@ defmodule Realtime.Tenants.Authorization.Policies.PresencePolicies do
           changeset = Presence.check_changeset(broadcast, %{check: true})
 
           case Repo.update(conn, changeset, Presence, mode: :savepoint) do
-            {:ok, %Presence{check: true} = broadcast} ->
-              revert_changeset = Presence.check_changeset(broadcast, %{check: false})
-              {:ok, _} = Repo.update(conn, revert_changeset, Presence)
+            {:ok, %Presence{check: true}} ->
+              Postgrex.query!(transaction_conn, "ROLLBACK AND CHAIN", [])
               Policies.update_policies(policies, :presence, :write, true)
 
             {:error, %Postgrex.Error{postgres: %{code: :insufficient_privilege}}} ->

diff --git a/mix.exs b/mix.exs
@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
   def project do
     [
       app: :realtime,
-      version: "2.28.17",
+      version: "2.28.18",
       elixir: "~> 1.14.0",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,