fix: malformed PG changes subscriptions should not retry (#1593)

Author: edgurgel

lib/extensions/postgres_cdc_rls/cdc_rls.ex

@@ -11,6 +11,7 @@ defmodule Extensions.PostgresCdcRls do
   alias Rls.Subscriptions
   alias Realtime.GenRpc
 
+  @impl true
   @spec handle_connect(map()) :: {:ok, {pid(), pid()}} | nil
   def handle_connect(args) do
     case get_manager_conn(args["id"]) do
@@ -26,22 +27,39 @@ defmodule Extensions.PostgresCdcRls do
     end
   end
 
-  def handle_after_connect({manager_pid, conn}, settings, params) do
-    publication = settings["publication"]
-    opts = [conn, publication, params, manager_pid, self()]
-    conn_node = node(conn)
+  @impl true
+  def handle_after_connect({manager_pid, conn}, settings, params_list) do
+    with {:ok, subscription_list} <- subscription_list(params_list) do
+      publication = settings["publication"]
+      opts = [conn, publication, subscription_list, manager_pid, self()]
+      conn_node = node(conn)
 
-    if conn_node !== node() do
-      GenRpc.call(conn_node, Subscriptions, :create, opts, timeout: 15_000)
-    else
-      apply(Subscriptions, :create, opts)
+      if conn_node !== node() do
+        GenRpc.call(conn_node, Subscriptions, :create, opts, timeout: 15_000)
+      else
+        apply(Subscriptions, :create, opts)
+      end
     end
   end
 
+  defp subscription_list(params_list) do
+    Enum.reduce_while(params_list, [], fn params, acc ->
+      case Subscriptions.parse_subscription_params(params[:params]) do
+        {:ok, subscription_params} ->
+          {:cont, {:ok, [%{id: params.id, claims: params.claims, subscription_params: subscription_params} | acc]}}
+
+        {:error, reason} ->
+          {:halt, {:error, :malformed_subscription_params, reason}}
+      end
+    end)
+  end
+
+  @impl true
   def handle_subscribe(_, tenant, metadata) do
     Endpoint.subscribe("realtime:postgres:" <> tenant, metadata)
   end
 
+  @impl true
   @doc """
   Stops the Supervision tree for a tenant.
 

lib/extensions/postgres_cdc_rls/subscriptions.ex

@@ -7,13 +7,16 @@ defmodule Extensions.PostgresCdcRls.Subscriptions do
   import Postgrex, only: [transaction: 2, query: 3, rollback: 2]
 
   @type conn() :: Postgrex.conn()
+  @type filter :: {binary, binary, binary}
+  @type subscription_params :: {binary, binary, [filter]}
+  @type subscription_list :: [%{id: binary, claims: map, subscription_params: subscription_params}]
 
   @filter_types ["eq", "neq", "lt", "lte", "gt", "gte", "in"]
 
-  @spec create(conn(), String.t(), [map()], pid(), pid()) ::
+  @spec create(conn(), String.t(), subscription_list, pid(), pid()) ::
           {:ok, Postgrex.Result.t()}
           | {:error, Exception.t() | :malformed_subscription_params | {:subscription_insert_failed, map()}}
-  def create(conn, publication, params_list, manager, caller) do
+  def create(conn, publication, subscription_list, manager, caller) do
     sql = "with sub_tables as (
         select
         rr.entity
@@ -50,37 +53,30 @@ defmodule Extensions.PostgresCdcRls.Subscriptions do
          id"
 
     transaction(conn, fn conn ->
-      Enum.map(params_list, fn %{id: id, claims: claims, params: params} ->
-        case parse_subscription_params(params) do
-          {:ok, [schema, table, filters]} ->
-            case query(conn, sql, [publication, schema, table, id, claims, filters]) do
-              {:ok, %{num_rows: num} = result} when num > 0 ->
-                send(manager, {:subscribed, {caller, id}})
-                result
+      Enum.map(subscription_list, fn %{id: id, claims: claims, subscription_params: params = {schema, table, filters}} ->
+        case query(conn, sql, [publication, schema, table, id, claims, filters]) do
+          {:ok, %{num_rows: num} = result} when num > 0 ->
+            send(manager, {:subscribed, {caller, id}})
+            result
 
-              {:ok, _} ->
-                msg =
-                  "Unable to subscribe to changes with given parameters. Please check Realtime is enabled for the given connect parameters: [#{params_to_log(params)}]"
+          {:ok, _} ->
+            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, 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)}"
+          {: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)
-            end
-
-          {:error, reason} ->
-            rollback(conn, reason)
+            rollback(conn, msg)
         end
       end)
     end)
   end
 
-  defp params_to_log(map) do
-    map
-    |> Map.to_list()
+  defp params_to_log({schema, table, filters}) do
+    [schema: schema, table: table, filters: filters]
     |> Enum.map_join(", ", fn {k, v} -> "#{k}: #{to_log(v)}" end)
   end
 
@@ -166,39 +162,55 @@ defmodule Extensions.PostgresCdcRls.Subscriptions do
 
   ## Examples
 
-      iex> params = %{"schema" => "public", "table" => "messages", "filter" => "subject=eq.hey"}
-      iex> Extensions.PostgresCdcRls.Subscriptions.parse_subscription_params(params)
-      {:ok, ["public", "messages", [{"subject", "eq", "hey"}]]}
+      iex> parse_subscription_params(%{"schema" => "public", "table" => "messages", "filter" => "subject=eq.hey"})
+      {:ok, {"public", "messages", [{"subject", "eq", "hey"}]}}
 
   `in` filter:
 
-      iex> params = %{"schema" => "public", "table" => "messages", "filter" => "subject=in.(hidee,ho)"}
-      iex> Extensions.PostgresCdcRls.Subscriptions.parse_subscription_params(params)
-      {:ok, ["public", "messages", [{"subject", "in", "{hidee,ho}"}]]}
+      iex> parse_subscription_params(%{"schema" => "public", "table" => "messages", "filter" => "subject=in.(hidee,ho)"})
+      {:ok, {"public", "messages", [{"subject", "in", "{hidee,ho}"}]}}
+
+  no filter:
+
+      iex> parse_subscription_params(%{"schema" => "public", "table" => "messages"})
+      {:ok, {"public", "messages", []}}
+
+  only schema:
+
+      iex> parse_subscription_params(%{"schema" => "public"})
+      {:ok, {"public", "*", []}}
+
+  only table:
+
+      iex> parse_subscription_params(%{"table" => "messages"})
+      {:ok, {"public", "messages", []}}
 
   An unsupported filter will respond with an error tuple:
 
-      iex> params = %{"schema" => "public", "table" => "messages", "filter" => "subject=like.hey"}
-      iex> Extensions.PostgresCdcRls.Subscriptions.parse_subscription_params(params)
+      iex> parse_subscription_params(%{"schema" => "public", "table" => "messages", "filter" => "subject=like.hey"})
       {:error, ~s(Error parsing `filter` params: ["like", "hey"])}
 
   Catch `undefined` filters:
 
-      iex> params = %{"schema" => "public", "table" => "messages", "filter" => "undefined"}
-      iex> Extensions.PostgresCdcRls.Subscriptions.parse_subscription_params(params)
+      iex> parse_subscription_params(%{"schema" => "public", "table" => "messages", "filter" => "undefined"})
       {:error, ~s(Error parsing `filter` params: ["undefined"])}
 
+  Catch `missing params`:
+
+      iex> parse_subscription_params(%{})
+      {:error, ~s(No subscription params provided. Please provide at least a `schema` or `table` to subscribe to: %{})}
+
   """
 
-  @spec parse_subscription_params(map()) :: {:ok, list} | {:error, binary()}
+  @spec parse_subscription_params(map()) :: {:ok, subscription_params} | {:error, binary()}
   def parse_subscription_params(params) do
     case params do
       %{"schema" => schema, "table" => table, "filter" => filter} ->
         with [col, rest] <- String.split(filter, "=", parts: 2),
              [filter_type, value] when filter_type in @filter_types <-
                String.split(rest, ".", parts: 2),
              {:ok, formatted_value} <- format_filter_value(filter_type, value) do
-          {:ok, [schema, table, [{col, filter_type, formatted_value}]]}
+          {:ok, {schema, table, [{col, filter_type, formatted_value}]}}
         else
           {:error, msg} ->
             {:error, "Error parsing `filter` params: #{msg}"}
@@ -208,13 +220,13 @@ defmodule Extensions.PostgresCdcRls.Subscriptions do
         end
 
       %{"schema" => schema, "table" => table} ->
-        {:ok, [schema, table, []]}
+        {:ok, {schema, table, []}}
 
       %{"schema" => schema} ->
-        {:ok, [schema, "*", []]}
+        {:ok, {schema, "*", []}}
 
       %{"table" => table} ->
-        {:ok, ["public", table, []]}
+        {:ok, {"public", table, []}}
 
       map when is_map_key(map, "user_token") or is_map_key(map, "auth_token") ->
         {:error,

lib/extensions/postgres_cdc_rls/worker_supervisor.ex

@@ -6,7 +6,7 @@ defmodule Extensions.PostgresCdcRls.WorkerSupervisor do
   alias PostgresCdcRls.ReplicationPoller
   alias PostgresCdcRls.SubscriptionManager
   alias PostgresCdcRls.SubscriptionsChecker
-  alias Realtime.Api
+  alias Realtime.Tenants.Cache
   alias Realtime.PostgresCdc.Exception
 
   def start_link(args) do
@@ -17,7 +17,7 @@ defmodule Extensions.PostgresCdcRls.WorkerSupervisor do
   @impl true
   def init(%{"id" => tenant} = args) when is_binary(tenant) do
     Logger.metadata(external_id: tenant, project: tenant)
-    unless Api.get_tenant_by_external_id(tenant, :primary), do: raise(Exception)
+    unless Cache.get_tenant_by_external_id(tenant), do: raise(Exception)
 
     subscribers_pids_table = :ets.new(__MODULE__, [:public, :bag])
     subscribers_nodes_table = :ets.new(__MODULE__, [:public, :set])

lib/realtime/postgres_cdc.ex

@@ -80,7 +80,7 @@ defmodule Realtime.PostgresCdc do
   end
 
   @callback handle_connect(any()) :: {:ok, any()} | nil
-  @callback handle_after_connect(any(), any(), any()) :: {:ok, any()} | {:error, any()}
+  @callback handle_after_connect(any(), any(), any()) :: {:ok, any()} | {:error, any()} | {:error, any(), any()}
   @callback handle_subscribe(any(), any(), any()) :: :ok
   @callback handle_stop(any(), any()) :: any()
 end

lib/realtime_web/channels/realtime_channel.ex

@@ -297,6 +297,12 @@ 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} ->
+            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
+            {:noreply, assign(socket, :pg_sub_ref, nil)}
+
           error ->
             maybe_log_warning(socket, "RealtimeDisabledForConfiguration", error)
 

mix.exs

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

test/integration/rt_channel_test.exs

@@ -100,7 +100,7 @@ defmodule Realtime.Integration.RtChannelTest do
                              "channel" => "any",
                              "extension" => "postgres_changes",
                              "message" =>
-                               "{:error, \"Unable to subscribe to changes with given parameters. Please check Realtime is enabled for the given connect parameters: [event: INSERT, schema: public]\"}",
+                               "{:error, \"Unable to subscribe to changes with given parameters. Please check Realtime is enabled for the given connect parameters: [schema: public, table: *, filters: []]\"}",
                              "status" => "error"
                            },
                            ref: nil,

test/realtime/extensions/cdc_rls/subscription_manager_test.exs

@@ -145,7 +145,7 @@ defmodule Realtime.Extensions.CdcRls.SubscriptionManagerTest do
 
     pg_change_params = %{
       id: uuid,
-      params: %{"event" => "*", "schema" => "public"},
+      subscription_params: {"public", "*", []},
       claims: %{
         "exp" => System.system_time(:second) + 100_000,
         "iat" => 0,