fix: add tenant_id tag to GenRpc calls from CdcRls (#1598)

Author: edgurgel

lib/extensions/postgres_cdc_rls/cdc_rls.ex

@@ -28,14 +28,14 @@ defmodule Extensions.PostgresCdcRls do
   end
 
   @impl true
-  def handle_after_connect({manager_pid, conn}, settings, params_list) do
+  def handle_after_connect({manager_pid, conn}, settings, params_list, tenant) 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)
+        GenRpc.call(conn_node, Subscriptions, :create, opts, timeout: 15_000, tenant_id: tenant)
       else
         apply(Subscriptions, :create, opts)
       end
@@ -90,7 +90,7 @@ defmodule Extensions.PostgresCdcRls do
       "Starting distributed postgres extension #{inspect(lauch_node: launch_node, region: region, platform_region: platform_region)}"
     )
 
-    case GenRpc.call(launch_node, __MODULE__, :start, [args], timeout: 30_000, tenant: tenant) do
+    case GenRpc.call(launch_node, __MODULE__, :start, [args], timeout: 30_000, tenant_id: tenant) do
       {:ok, _pid} = ok ->
         ok
 

lib/realtime/postgres_cdc.ex

@@ -16,8 +16,8 @@ defmodule Realtime.PostgresCdc do
     apply(module, :handle_connect, [opts])
   end
 
-  def after_connect(module, connect_response, extension, params) do
-    apply(module, :handle_after_connect, [connect_response, extension, params])
+  def after_connect(module, connect_response, extension, params, tenant) do
+    apply(module, :handle_after_connect, [connect_response, extension, params, tenant])
   end
 
   def subscribe(module, pg_change_params, tenant, metadata) do
@@ -80,7 +80,8 @@ defmodule Realtime.PostgresCdc do
   end
 
   @callback handle_connect(any()) :: {:ok, any()} | nil
-  @callback handle_after_connect(any(), any(), any()) :: {:ok, any()} | {:error, any()} | {:error, any(), any()}
+  @callback handle_after_connect(any(), any(), any(), tenant_id :: String.t()) ::
+              {: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

@@ -290,7 +290,7 @@ defmodule RealtimeWeb.RealtimeChannel do
 
     case PostgresCdc.connect(module, args) do
       {:ok, response} ->
-        case PostgresCdc.after_connect(module, response, postgres_extension, pg_change_params) do
+        case PostgresCdc.after_connect(module, response, postgres_extension, pg_change_params, tenant) do
           {:ok, _response} ->
             message = "Subscribed to PostgreSQL"
             maybe_log_info(socket, message)

mix.exs

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

test/realtime/extensions/cdc_rls/cdc_rls_test.exs

@@ -86,7 +86,7 @@ defmodule Realtime.Extensions.CdcRlsTest do
       {:ok, response} = PostgresCdcRls.handle_connect(args)
 
       # Now subscribe to the Postgres Changes
-      {:ok, _} = PostgresCdcRls.handle_after_connect(response, postgres_extension, pg_change_params)
+      {:ok, _} = PostgresCdcRls.handle_after_connect(response, postgres_extension, pg_change_params, external_id)
 
       RealtimeWeb.Endpoint.unsubscribe(Realtime.Syn.PostgresCdc.syn_topic(tenant.external_id))
       %{tenant: tenant}
@@ -242,9 +242,9 @@ defmodule Realtime.Extensions.CdcRlsTest do
 
       on_exit(fn -> :telemetry.detach(__MODULE__) end)
 
-      :telemetry.attach(
+      :telemetry.attach_many(
         __MODULE__,
-        [:realtime, :tenants, :payload, :size],
+        [[:realtime, :tenants, :payload, :size], [:realtime, :rpc]],
         &__MODULE__.handle_telemetry/4,
         pid: self()
       )
@@ -291,8 +291,19 @@ defmodule Realtime.Extensions.CdcRlsTest do
       Process.sleep(3000)
       {:ok, response} = PostgresCdcRls.handle_connect(args)
 
+      assert_receive {
+        :telemetry,
+        [:realtime, :rpc],
+        %{latency: _},
+        %{
+          tenant: "dev_tenant",
+          mechanism: :gen_rpc,
+          success: true
+        }
+      }
+
       # Now subscribe to the Postgres Changes
-      {:ok, _} = PostgresCdcRls.handle_after_connect(response, postgres_extension, pg_change_params)
+      {:ok, _} = PostgresCdcRls.handle_after_connect(response, postgres_extension, pg_change_params, external_id)
       assert %Postgrex.Result{rows: [[1]]} = Postgrex.query!(conn, "select count(*) from realtime.subscription", [])
 
       # Insert a record
@@ -382,7 +393,7 @@ defmodule Realtime.Extensions.CdcRlsTest do
       {:ok, response} = PostgresCdcRls.handle_connect(args)
 
       # Now subscribe to the Postgres Changes
-      {:ok, _} = PostgresCdcRls.handle_after_connect(response, postgres_extension, pg_change_params)
+      {:ok, _} = PostgresCdcRls.handle_after_connect(response, postgres_extension, pg_change_params, external_id)
       assert %Postgrex.Result{rows: [[1]]} = Postgrex.query!(conn, "select count(*) from realtime.subscription", [])
 
       log =
@@ -468,7 +479,7 @@ defmodule Realtime.Extensions.CdcRlsTest do
       :ok = PostgresCdc.subscribe(PostgresCdcRls, pg_change_params, external_id, metadata)
 
       # Now subscribe to the Postgres Changes
-      {:ok, _} = PostgresCdcRls.handle_after_connect(response, postgres_extension, pg_change_params)
+      {:ok, _} = PostgresCdcRls.handle_after_connect(response, postgres_extension, pg_change_params, external_id)
       assert %Postgrex.Result{rows: [[1]]} = Postgrex.query!(conn, "select count(*) from realtime.subscription", [])
 
       # Insert a record
@@ -507,6 +518,19 @@ defmodule Realtime.Extensions.CdcRlsTest do
       assert {:ok, %RateCounter{id: {:channel, :db_events, "dev_tenant"}, bucket: bucket}} = RateCounter.get(rate)
       assert 1 in bucket
 
+      assert_receive {
+        :telemetry,
+        [:realtime, :rpc],
+        %{latency: _},
+        %{
+          tenant: "dev_tenant",
+          mechanism: :gen_rpc,
+          origin_node: _,
+          success: true,
+          target_node: ^node
+        }
+      }
+
       :erpc.call(node, PostgresCdcRls, :handle_stop, [tenant.external_id, 10_000])
     end
   end