fix: invalid PG changes subscriptions should not retry

* Table that does not exist
* Realtime publication does not exist
* Column does not exist

Author: edgurgel

lib/extensions/postgres_cdc_rls/cdc_rls.ex

@@ -49,7 +49,7 @@ defmodule Extensions.PostgresCdcRls do
           {:cont, {:ok, [%{id: params.id, claims: params.claims, subscription_params: subscription_params} | acc]}}
 
         {:error, reason} ->
-          {:halt, {:error, :malformed_subscription_params, reason}}
+          {:halt, {:error, {:malformed_subscription_params, reason}}}
       end
     end)
   end

lib/extensions/postgres_cdc_rls/subscriptions.ex

@@ -15,7 +15,8 @@ defmodule Extensions.PostgresCdcRls.Subscriptions do
 
   @spec create(conn(), String.t(), subscription_list, pid(), pid()) ::
           {:ok, Postgrex.Result.t()}
-          | {:error, Exception.t() | :malformed_subscription_params | {:subscription_insert_failed, map()}}
+          | {:error, Exception.t() | {:exit, term} | {:subscription_insert_failed, String.t()}}
+
   def create(conn, publication, subscription_list, manager, caller) do
     sql = "with sub_tables as (
         select
@@ -63,16 +64,20 @@ defmodule Extensions.PostgresCdcRls.Subscriptions do
             msg =
               "Unable to subscribe to changes with given parameters. Please check Realtime is enabled for the given connect parameters: [#{params_to_log(params)}]"
 
-            rollback(conn, msg)
+            rollback(conn, {:subscription_insert_failed, msg})
 
           {:error, exception} ->
             msg =
               "Unable to subscribe to changes with given parameters. An exception happened so please check your connect parameters: [#{params_to_log(params)}]. Exception: #{Exception.message(exception)}"
 
-            rollback(conn, msg)
+            rollback(conn, {:subscription_insert_failed, msg})
         end
       end)
     end)
+  rescue
+    e -> {:error, e}
+  catch
+    :exit, reason -> {:error, {:exit, reason}}
   end
 
   defp params_to_log({schema, table, filters}) do

lib/realtime_web/channels/realtime_channel.ex

@@ -297,7 +297,7 @@ defmodule RealtimeWeb.RealtimeChannel do
             push_system_message("postgres_changes", socket, "ok", message, channel_name)
             {:noreply, assign(socket, :pg_sub_ref, nil)}
 
-          {:error, :malformed_subscription_params, error} ->
+          {:error, {reason, error}} when reason in [:malformed_subscription_params, :subscription_insert_failed] ->
             maybe_log_warning(socket, "RealtimeDisabledForConfiguration", error)
             push_system_message("postgres_changes", socket, "error", error, channel_name)
             # No point in retrying if the params are invalid

mix.exs

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

priv/repo/dev_seeds.exs

@@ -1,5 +1,3 @@
-import Ecto.Adapters.SQL, only: [query!: 3]
-
 alias Realtime.Api.Tenant
 alias Realtime.Database
 alias Realtime.Repo

test/realtime/extensions/cdc_rls/subscriptions_test.exs

@@ -44,6 +44,35 @@ defmodule Realtime.Extensions.PostgresCdcRls.Subscriptions do
       %Postgrex.Result{rows: [[1]]} = Postgrex.query!(conn, "select count(*) from realtime.subscription", [])
     end
 
+    test "publication does not exist", %{conn: conn} do
+      {:ok, subscription_params} = Subscriptions.parse_subscription_params(%{"schema" => "public", "table" => "test"})
+
+      subscription_list = [%{claims: %{"role" => "anon"}, id: UUID.uuid1(), subscription_params: subscription_params}]
+
+      Postgrex.query!(conn, "drop publication if exists supabase_realtime_test", [])
+
+      assert {:error,
+              {:subscription_insert_failed,
+               "Unable to subscribe to changes with given parameters. Please check Realtime is enabled for the given connect parameters: [schema: public, table: test, filters: []]"}} =
+               Subscriptions.create(conn, "supabase_realtime_test", subscription_list, self(), self())
+
+      %Postgrex.Result{rows: [[0]]} = Postgrex.query!(conn, "select count(*) from realtime.subscription", [])
+    end
+
+    test "table does not exist", %{conn: conn} do
+      {:ok, subscription_params} =
+        Subscriptions.parse_subscription_params(%{"schema" => "public", "table" => "doesnotexist"})
+
+      subscription_list = [%{claims: %{"role" => "anon"}, id: UUID.uuid1(), subscription_params: subscription_params}]
+
+      assert {:error,
+              {:subscription_insert_failed,
+               "Unable to subscribe to changes with given parameters. Please check Realtime is enabled for the given connect parameters: [schema: public, table: doesnotexist, filters: []]"}} =
+               Subscriptions.create(conn, "supabase_realtime_test", subscription_list, self(), self())
+
+      %Postgrex.Result{rows: [[0]]} = Postgrex.query!(conn, "select count(*) from realtime.subscription", [])
+    end
+
     test "column does not exist", %{conn: conn} do
       {:ok, subscription_params} =
         Subscriptions.parse_subscription_params(%{
@@ -55,7 +84,8 @@ defmodule Realtime.Extensions.PostgresCdcRls.Subscriptions do
       subscription_list = [%{claims: %{"role" => "anon"}, id: UUID.uuid1(), subscription_params: subscription_params}]
 
       assert {:error,
-              "Unable to subscribe to changes with given parameters. An exception happened so please check your connect parameters: [schema: public, table: test, filters: [{\"subject\", \"eq\", \"hey\"}]]. Exception: ERROR P0001 (raise_exception) invalid column for filter subject"} =
+              {:subscription_insert_failed,
+               "Unable to subscribe to changes with given parameters. An exception happened so please check your connect parameters: [schema: public, table: test, filters: [{\"subject\", \"eq\", \"hey\"}]]. Exception: ERROR P0001 (raise_exception) invalid column for filter subject"}} =
                Subscriptions.create(conn, "supabase_realtime_test", subscription_list, self(), self())
 
       %Postgrex.Result{rows: [[0]]} = Postgrex.query!(conn, "select count(*) from realtime.subscription", [])
@@ -72,11 +102,34 @@ defmodule Realtime.Extensions.PostgresCdcRls.Subscriptions do
       subscription_list = [%{claims: %{"role" => "anon"}, id: UUID.uuid1(), subscription_params: subscription_params}]
 
       assert {:error,
-              "Unable to subscribe to changes with given parameters. An exception happened so please check your connect parameters: [schema: public, table: test, filters: [{\"id\", \"eq\", \"hey\"}]]. Exception: ERROR 22P02 (invalid_text_representation) invalid input syntax for type integer: \"hey\""} =
+              {:subscription_insert_failed,
+               "Unable to subscribe to changes with given parameters. An exception happened so please check your connect parameters: [schema: public, table: test, filters: [{\"id\", \"eq\", \"hey\"}]]. Exception: ERROR 22P02 (invalid_text_representation) invalid input syntax for type integer: \"hey\""}} =
                Subscriptions.create(conn, "supabase_realtime_test", subscription_list, self(), self())
 
       %Postgrex.Result{rows: [[0]]} = Postgrex.query!(conn, "select count(*) from realtime.subscription", [])
     end
+
+    test "connection error" do
+      {:ok, subscription_params} =
+        Subscriptions.parse_subscription_params(%{"schema" => "public", "table" => "test"})
+
+      subscription_list = [%{claims: %{"role" => "anon"}, id: UUID.uuid1(), subscription_params: subscription_params}]
+      conn = spawn(fn -> :ok end)
+
+      assert {:error, {:exit, _}} =
+               Subscriptions.create(conn, "supabase_realtime_test", subscription_list, self(), self())
+    end
+
+    test "timeout", %{conn: conn} do
+      {:ok, subscription_params} = Subscriptions.parse_subscription_params(%{"schema" => "public", "table" => "test"})
+
+      Task.start(fn -> Postgrex.query!(conn, "SELECT pg_sleep(20)", []) end)
+
+      subscription_list = [%{claims: %{"role" => "anon"}, id: UUID.uuid1(), subscription_params: subscription_params}]
+
+      assert {:error, %DBConnection.ConnectionError{reason: :queue_timeout}} =
+               Subscriptions.create(conn, "supabase_realtime_test", subscription_list, self(), self())
+    end
   end
 
   describe "delete_all/1" do

test/realtime_web/channels/realtime_channel_test.exs

@@ -69,12 +69,7 @@ defmodule RealtimeWeb.RealtimeChannelTest do
                reply
 
       assert_push "system",
-                  %{
-                    message: "Subscribed to PostgreSQL",
-                    status: "ok",
-                    extension: "postgres_changes",
-                    channel: "test"
-                  },
+                  %{message: "Subscribed to PostgreSQL", status: "ok", extension: "postgres_changes", channel: "test"},
                   3000
 
       {:ok, conn} = Connect.lookup_or_start_connection(tenant.external_id)
@@ -125,6 +120,66 @@ defmodule RealtimeWeb.RealtimeChannelTest do
       # It won't re-subscribe
       assert socket.assigns.pg_sub_ref == nil
     end
+
+    test "invalid subscription table does not exist", %{tenant: tenant} do
+      jwt = Generators.generate_jwt_token(tenant)
+      {:ok, %Socket{} = socket} = connect(UserSocket, %{}, conn_opts(tenant, jwt))
+
+      config = %{
+        "presence" => %{"enabled" => false},
+        "postgres_changes" => [%{"event" => "*", "schema" => "public", "table" => "doesnotexist"}]
+      }
+
+      assert {:ok, reply, socket} = subscribe_and_join(socket, "realtime:test", %{"config" => config})
+
+      assert %{postgres_changes: [%{"event" => "*", "schema" => "public", "table" => "doesnotexist"}]} = reply
+
+      assert_push "system",
+                  %{
+                    message:
+                      "Unable to subscribe to changes with given parameters. Please check Realtime is enabled for the given connect parameters: [schema: public, table: doesnotexist, filters: []]",
+                    status: "error",
+                    extension: "postgres_changes",
+                    channel: "test"
+                  },
+                  3000
+
+      socket = Server.socket(socket.channel_pid)
+
+      # It won't re-subscribe
+      assert socket.assigns.pg_sub_ref == nil
+    end
+
+    test "invalid subscription column does not exist", %{tenant: tenant} do
+      jwt = Generators.generate_jwt_token(tenant)
+      {:ok, %Socket{} = socket} = connect(UserSocket, %{}, conn_opts(tenant, jwt))
+
+      config = %{
+        "presence" => %{"enabled" => false},
+        "postgres_changes" => [
+          %{"event" => "*", "schema" => "public", "table" => "test", "filter" => "notacolumn=eq.123"}
+        ]
+      }
+
+      assert {:ok, reply, socket} = subscribe_and_join(socket, "realtime:test", %{"config" => config})
+
+      assert %{postgres_changes: [%{"event" => "*", "schema" => "public", "table" => "test"}]} = reply
+
+      assert_push "system",
+                  %{
+                    message:
+                      "Unable to subscribe to changes with given parameters. An exception happened so please check your connect parameters: [schema: public, table: test, filters: [{\"notacolumn\", \"eq\", \"123\"}]]. Exception: ERROR P0001 (raise_exception) invalid column for filter notacolumn",
+                    status: "error",
+                    extension: "postgres_changes",
+                    channel: "test"
+                  },
+                  3000
+
+      socket = Server.socket(socket.channel_pid)
+
+      # It won't re-subscribe
+      assert socket.assigns.pg_sub_ref == nil
+    end
   end
 
   describe "broadcast" do