🔄 Sync with upstream changes

.github/workflows/tests.yml

@@ -16,6 +16,10 @@ on:
     branches:
       - main
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   tests:
     name: Tests

Makefile

@@ -9,10 +9,10 @@ PORT ?= 4000
 # Common commands
 
 dev: ## Start a dev server
-	ELIXIR_ERL_OPTIONS="+hmax 1000000000" SLOT_NAME_SUFFIX=some_sha PORT=$(PORT) MIX_ENV=dev SECURE_CHANNELS=true API_JWT_SECRET=dev METRICS_JWT_SECRET=dev REGION=fra DB_ENC_KEY="1234567890123456" CLUSTER_STRATEGIES=$(CLUSTER_STRATEGIES) ERL_AFLAGS="-kernel shell_history enabled" GEN_RPC_TCP_SERVER_PORT=5369 GEN_RPC_TCP_CLIENT_PORT=5469 iex --name $(NODE_NAME)@127.0.0.1 --cookie cookie  -S mix phx.server
+	ELIXIR_ERL_OPTIONS="+hmax 1000000000" SLOT_NAME_SUFFIX=some_sha PORT=$(PORT) MIX_ENV=dev SECURE_CHANNELS=true API_JWT_SECRET=dev METRICS_JWT_SECRET=dev REGION=us-east-1 DB_ENC_KEY="1234567890123456" CLUSTER_STRATEGIES=$(CLUSTER_STRATEGIES) ERL_AFLAGS="-kernel shell_history enabled" GEN_RPC_TCP_SERVER_PORT=5369 GEN_RPC_TCP_CLIENT_PORT=5469 iex --name $(NODE_NAME)@127.0.0.1 --cookie cookie  -S mix phx.server
 
 dev.orange: ## Start another dev server (orange) on port 4001
-	ELIXIR_ERL_OPTIONS="+hmax 1000000000" SLOT_NAME_SUFFIX=some_sha PORT=4001 MIX_ENV=dev SECURE_CHANNELS=true API_JWT_SECRET=dev METRICS_JWT_SECRET=dev DB_ENC_KEY="1234567890123456" CLUSTER_STRATEGIES=$(CLUSTER_STRATEGIES) ERL_AFLAGS="-kernel shell_history enabled" GEN_RPC_TCP_SERVER_PORT=5469 GEN_RPC_TCP_CLIENT_PORT=5369 iex --name orange@127.0.0.1 --cookie cookie  -S mix phx.server
+	ELIXIR_ERL_OPTIONS="+hmax 1000000000" SLOT_NAME_SUFFIX=some_sha PORT=4001 MIX_ENV=dev SECURE_CHANNELS=true API_JWT_SECRET=dev METRICS_JWT_SECRET=dev REGION=eu-west-1 DB_ENC_KEY="1234567890123456" CLUSTER_STRATEGIES=$(CLUSTER_STRATEGIES) ERL_AFLAGS="-kernel shell_history enabled" GEN_RPC_TCP_SERVER_PORT=5469 GEN_RPC_TCP_CLIENT_PORT=5369 iex --name orange@127.0.0.1 --cookie cookie  -S mix phx.server
 
 seed: ## Seed the database
 	DB_ENC_KEY="1234567890123456" FLY_ALLOC_ID=123e4567-e89b-12d3-a456-426614174000 mix run priv/repo/dev_seeds.exs

config/dev.exs

@@ -97,6 +97,8 @@ config :phoenix, :plug_init_mode, :runtime
 # Disable caching to ensure the rendered spec is refreshed
 config :open_api_spex, :cache_adapter, OpenApiSpex.Plug.NoneCache
 
-config :opentelemetry, traces_exporter: {:otel_exporter_stdout, []}
+# Disabled but can print to stdout with:
+# config :opentelemetry, traces_exporter: {:otel_exporter_stdout, []}
+config :opentelemetry, traces_exporter: :none
 
 config :mix_test_watch, clear: true

config/runtime.exs

@@ -68,7 +68,7 @@ janitor_children_timeout = Env.get_integer("JANITOR_CHILDREN_TIMEOUT", :timer.se
 janitor_schedule_timer = Env.get_integer("JANITOR_SCHEDULE_TIMER_IN_MS", :timer.hours(4))
 platform = if System.get_env("AWS_EXECUTION_ENV") == "AWS_ECS_FARGATE", do: :aws, else: :fly
 broadcast_pool_size = Env.get_integer("BROADCAST_POOL_SIZE", 10)
-pubsub_adapter = System.get_env("PUBSUB_ADAPTER", "pg2") |> String.to_atom()
+pubsub_adapter = System.get_env("PUBSUB_ADAPTER", "gen_rpc") |> String.to_atom()
 websocket_max_heap_size = div(Env.get_integer("WEBSOCKET_MAX_HEAP_SIZE", 50_000_000), :erlang.system_info(:wordsize))
 
 no_channel_timeout_in_ms =

lib/extensions/postgres_cdc_rls/replication_poller.ex

@@ -183,7 +183,7 @@ defmodule Extensions.PostgresCdcRls.ReplicationPoller do
         change <- columns |> Enum.zip(row) |> generate_record() |> List.wrap() do
       topic = "realtime:postgres:" <> tenant_id
 
-      RealtimeWeb.TenantBroadcaster.pubsub_broadcast(tenant_id, topic, change, MessageDispatcher)
+      RealtimeWeb.TenantBroadcaster.pubsub_broadcast(tenant_id, topic, change, MessageDispatcher, :postgres_changes)
     end
 
     {:ok, rows_count}

lib/realtime/gen_rpc/pub_sub.ex

@@ -65,7 +65,11 @@ defmodule Realtime.GenRpcPubSub.Worker do
   def start_link({pubsub, worker}), do: GenServer.start_link(__MODULE__, pubsub, name: worker)
 
   @impl true
-  def init(pubsub), do: {:ok, pubsub}
+  def init(pubsub) do
+    Process.flag(:message_queue_data, :off_heap)
+    Process.flag(:fullsweep_after, 100)
+    {:ok, pubsub}
+  end
 
   @impl true
   def handle_info({:ftl, topic, message, dispatcher}, pubsub) do

lib/realtime/monitoring/latency.ex

@@ -7,7 +7,7 @@ defmodule Realtime.Latency do
   use Realtime.Logs
 
   alias Realtime.Nodes
-  alias Realtime.Rpc
+  alias Realtime.GenRpc
 
   defmodule Payload do
     @moduledoc false
@@ -33,7 +33,7 @@ defmodule Realtime.Latency do
           }
   end
 
-  @every 5_000
+  @every 15_000
   def start_link(args) do
     GenServer.start_link(__MODULE__, args, name: __MODULE__)
   end
@@ -76,7 +76,7 @@ defmodule Realtime.Latency do
         Task.Supervisor.async(Realtime.TaskSupervisor, fn ->
           {latency, response} =
             :timer.tc(fn ->
-              Rpc.call(n, __MODULE__, :pong, [pong_timeout], timeout: timer_timeout)
+              GenRpc.call(n, __MODULE__, :pong, [pong_timeout], timeout: timer_timeout)
             end)
 
           latency_ms = latency / 1_000
@@ -85,7 +85,7 @@ defmodule Realtime.Latency do
           from_node = Nodes.short_node_id_from_name(Node.self())
 
           case response do
-            {:badrpc, reason} ->
+            {:error, :rpc_error, reason} ->
               log_error(
                 "RealtimeNodeDisconnected",
                 "Unable to connect to #{short_name} from #{region}: #{reason}"

lib/realtime/monitoring/prom_ex/plugins/tenant.ex

@@ -36,20 +36,21 @@ defmodule Realtime.PromEx.Plugins.Tenant do
           event_name: [:realtime, :tenants, :payload, :size],
           measurement: :size,
           description: "Tenant payload size",
-          tags: [:tenant],
+          tags: [:tenant, :message_type],
           unit: :byte,
           reporter_options: [
-            buckets: [100, 250, 500, 1000, 2000, 3000, 5000, 10_000, 25_000]
+            buckets: [250, 500, 1000, 3000, 5000, 10_000, 25_000, 100_000, 500_000, 1_000_000, 3_000_000]
           ]
         ),
         distribution(
           [:realtime, :payload, :size],
           event_name: [:realtime, :tenants, :payload, :size],
           measurement: :size,
           description: "Payload size",
+          tags: [:message_type],
           unit: :byte,
           reporter_options: [
-            buckets: [100, 250, 500, 1000, 2000, 3000, 5000, 10_000, 25_000]
+            buckets: [250, 500, 1000, 3000, 5000, 10_000, 25_000, 100_000, 500_000, 1_000_000, 3_000_000]
           ]
         )
       ]
@@ -157,20 +158,38 @@ defmodule Realtime.PromEx.Plugins.Tenant do
           description: "Sum of messages sent on a Realtime Channel.",
           tags: [:tenant]
         ),
+        sum(
+          [:realtime, :channel, :global, :events],
+          event_name: [:realtime, :rate_counter, :channel, :events],
+          measurement: :sum,
+          description: "Global sum of messages sent on a Realtime Channel."
+        ),
         sum(
           [:realtime, :channel, :presence_events],
           event_name: [:realtime, :rate_counter, :channel, :presence_events],
           measurement: :sum,
           description: "Sum of presence messages sent on a Realtime Channel.",
           tags: [:tenant]
         ),
+        sum(
+          [:realtime, :channel, :global, :presence_events],
+          event_name: [:realtime, :rate_counter, :channel, :presence_events],
+          measurement: :sum,
+          description: "Global sum of presence messages sent on a Realtime Channel."
+        ),
         sum(
           [:realtime, :channel, :db_events],
           event_name: [:realtime, :rate_counter, :channel, :db_events],
           measurement: :sum,
           description: "Sum of db messages sent on a Realtime Channel.",
           tags: [:tenant]
         ),
+        sum(
+          [:realtime, :channel, :global, :db_events],
+          event_name: [:realtime, :rate_counter, :channel, :db_events],
+          measurement: :sum,
+          description: "Global sum of db messages sent on a Realtime Channel."
+        ),
         sum(
           [:realtime, :channel, :joins],
           event_name: [:realtime, :rate_counter, :channel, :joins],

lib/realtime/monitoring/prom_ex/plugins/tenants.ex

@@ -21,6 +21,15 @@ defmodule Realtime.PromEx.Plugins.Tenants do
         unit: {:microsecond, :millisecond},
         tags: [:success, :tenant, :mechanism],
         reporter_options: [buckets: [10, 250, 5000, 15_000]]
+      ),
+      distribution(
+        [:realtime, :global, :rpc],
+        event_name: [:realtime, :rpc],
+        description: "Global Latency of rpc calls",
+        measurement: :latency,
+        unit: {:microsecond, :millisecond},
+        tags: [:success, :mechanism],
+        reporter_options: [buckets: [10, 250, 5000, 15_000]]
       )
     ])
   end

lib/realtime/tenants.ex

@@ -328,18 +328,18 @@ defmodule Realtime.Tenants do
     %RateCounter.Args{id: {:channel, :authorization_errors, external_id}, opts: opts}
   end
 
-  @connect_per_second_default 10
+  @connect_errors_per_second_default 10
   @doc "RateCounter arguments for counting connect per second."
-  @spec connect_per_second_rate(Tenant.t() | String.t()) :: RateCounter.Args.t()
-  def connect_per_second_rate(%Tenant{external_id: external_id}) do
-    connect_per_second_rate(external_id)
+  @spec connect_errors_per_second_rate(Tenant.t() | String.t()) :: RateCounter.Args.t()
+  def connect_errors_per_second_rate(%Tenant{external_id: external_id}) do
+    connect_errors_per_second_rate(external_id)
   end
 
-  def connect_per_second_rate(tenant_id) do
+  def connect_errors_per_second_rate(tenant_id) do
     opts = [
-      max_bucket_len: 10,
+      max_bucket_len: 30,
       limit: [
-        value: @connect_per_second_default,
+        value: @connect_errors_per_second_default,
         measurement: :sum,
         log_fn: fn ->
           Logger.critical(

lib/realtime/tenants/batch_broadcast.ex

@@ -129,7 +129,14 @@ defmodule Realtime.Tenants.BatchBroadcast do
     broadcast = %Phoenix.Socket.Broadcast{topic: message.topic, event: @event_type, payload: payload}
 
     GenCounter.add(events_per_second_rate.id)
-    TenantBroadcaster.pubsub_broadcast(tenant.external_id, tenant_topic, broadcast, RealtimeChannel.MessageDispatcher)
+
+    TenantBroadcaster.pubsub_broadcast(
+      tenant.external_id,
+      tenant_topic,
+      broadcast,
+      RealtimeChannel.MessageDispatcher,
+      :broadcast
+    )
   end
 
   defp permissions_for_message(_, nil, _), do: nil

lib/realtime/tenants/connect.ex

@@ -57,7 +57,7 @@ defmodule Realtime.Tenants.Connect do
           | {:error, :connect_rate_limit_reached}
           | {:error, :rpc_error, term()}
   def lookup_or_start_connection(tenant_id, opts \\ []) when is_binary(tenant_id) do
-    rate_args = Tenants.connect_per_second_rate(tenant_id)
+    rate_args = Tenants.connect_errors_per_second_rate(tenant_id)
     RateCounter.new(rate_args)
 
     with {:ok, %{limit: %{triggered: false}}} <- RateCounter.get(rate_args),
@@ -68,8 +68,14 @@ defmodule Realtime.Tenants.Connect do
         {:error, :connect_rate_limit_reached}
 
       {:error, :tenant_database_connection_initializing} ->
-        GenCounter.add(rate_args.id)
-        call_external_node(tenant_id, opts)
+        case call_external_node(tenant_id, opts) do
+          {:ok, pid} ->
+            {:ok, pid}
+
+          error ->
+            GenCounter.add(rate_args.id)
+            error
+        end
 
       {:error, :initializing} ->
         {:error, :tenant_database_unavailable}

lib/realtime_web/channels/presence.ex

@@ -8,5 +8,6 @@ defmodule RealtimeWeb.Presence do
   use Phoenix.Presence,
     otp_app: :realtime,
     pubsub_server: Realtime.PubSub,
+    dispatcher: RealtimeWeb.RealtimeChannel.MessageDispatcher,
     pool_size: 10
 end

lib/realtime_web/channels/realtime_channel.ex

@@ -18,7 +18,6 @@ defmodule RealtimeWeb.RealtimeChannel do
   alias Realtime.Tenants.Authorization
   alias Realtime.Tenants.Authorization.Policies
   alias Realtime.Tenants.Authorization.Policies.BroadcastPolicies
-  alias Realtime.Tenants.Authorization.Policies.PresencePolicies
   alias Realtime.Tenants.Connect
 
   alias RealtimeWeb.Channels.Payloads.Join
@@ -259,27 +258,11 @@ defmodule RealtimeWeb.RealtimeChannel do
     {:noreply, assign(socket, %{pg_sub_ref: pg_sub_ref})}
   end
 
-  def handle_info(
-        %{event: "presence_diff"},
-        %{assigns: %{policies: %Policies{presence: %PresencePolicies{read: false}}}} = socket
-      ) do
-    Logger.warning("Presence message ignored")
-    {:noreply, socket}
-  end
-
   def handle_info(_msg, %{assigns: %{policies: %Policies{broadcast: %BroadcastPolicies{read: false}}}} = socket) do
     Logger.warning("Broadcast message ignored")
     {:noreply, socket}
   end
 
-  def handle_info(%{event: "presence_diff", payload: payload} = msg, socket) do
-    %{presence_rate_counter: presence_rate_counter} = socket.assigns
-    GenCounter.add(presence_rate_counter.id)
-    maybe_log_info(socket, msg)
-    push(socket, "presence_diff", payload)
-    {:noreply, socket}
-  end
-
   def handle_info(%{event: type, payload: payload} = msg, socket) do
     count(socket)
     maybe_log_info(socket, msg)

lib/realtime_web/channels/realtime_channel/broadcast_handler.ex

@@ -76,14 +76,21 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandler do
     broadcast = %Phoenix.Socket.Broadcast{topic: tenant_topic, event: @event_type, payload: payload}
 
     if self_broadcast do
-      TenantBroadcaster.pubsub_broadcast(tenant_id, tenant_topic, broadcast, RealtimeChannel.MessageDispatcher)
+      TenantBroadcaster.pubsub_broadcast(
+        tenant_id,
+        tenant_topic,
+        broadcast,
+        RealtimeChannel.MessageDispatcher,
+        :broadcast
+      )
     else
       TenantBroadcaster.pubsub_broadcast_from(
         tenant_id,
         self(),
         tenant_topic,
         broadcast,
-        RealtimeChannel.MessageDispatcher
+        RealtimeChannel.MessageDispatcher,
+        :broadcast
       )
     end
   end