Understand the latency between nodes across the Realtime cluster.
-
- <%= for {_pair, p} <- @pings do %>
-
-
From: <%= p.from_region %> - <%= p.from_node %>
-
To: <%= p.region %> - <%= p.node %>
-
<%= p.latency %> ms
-
<%= p.timestamp %>
-
- <% end %>
+
+
+
From: <%= p.payload.from_region %> - <%= p.payload.from_node %>
+
To: <%= p.payload.region %> - <%= p.payload.node %>
+
<%= p.payload.latency %> ms
+
<%= p.payload.timestamp %>
+
diff --git a/lib/realtime_web/tenant_broadcaster.ex b/lib/realtime_web/tenant_broadcaster.ex
index da02df79e..f8b739a0b 100644
--- a/lib/realtime_web/tenant_broadcaster.ex
+++ b/lib/realtime_web/tenant_broadcaster.ex
@@ -5,9 +5,12 @@ defmodule RealtimeWeb.TenantBroadcaster do
alias Phoenix.PubSub
- @spec pubsub_broadcast(tenant_id :: String.t(), PubSub.topic(), PubSub.message(), PubSub.dispatcher()) :: :ok
- def pubsub_broadcast(tenant_id, topic, message, dispatcher) do
- collect_payload_size(tenant_id, message)
+ @type message_type :: :broadcast | :presence | :postgres_changes
+
+ @spec pubsub_broadcast(tenant_id :: String.t(), PubSub.topic(), PubSub.message(), PubSub.dispatcher(), message_type) ::
+ :ok
+ def pubsub_broadcast(tenant_id, topic, message, dispatcher, message_type) do
+ collect_payload_size(tenant_id, message, message_type)
if pubsub_adapter() == :gen_rpc do
PubSub.broadcast(Realtime.PubSub, topic, message, dispatcher)
@@ -23,11 +26,12 @@ defmodule RealtimeWeb.TenantBroadcaster do
from :: pid,
PubSub.topic(),
PubSub.message(),
- PubSub.dispatcher()
+ PubSub.dispatcher(),
+ message_type
) ::
:ok
- def pubsub_broadcast_from(tenant_id, from, topic, message, dispatcher) do
- collect_payload_size(tenant_id, message)
+ def pubsub_broadcast_from(tenant_id, from, topic, message, dispatcher, message_type) do
+ collect_payload_size(tenant_id, message, message_type)
if pubsub_adapter() == :gen_rpc do
PubSub.broadcast_from(Realtime.PubSub, from, topic, message, dispatcher)
@@ -45,16 +49,18 @@ defmodule RealtimeWeb.TenantBroadcaster do
@payload_size_event [:realtime, :tenants, :payload, :size]
- defp collect_payload_size(tenant_id, payload) when is_struct(payload) do
+ @spec collect_payload_size(tenant_id :: String.t(), payload :: term, message_type :: message_type) :: :ok
+ def collect_payload_size(tenant_id, payload, message_type) when is_struct(payload) do
# Extracting from struct so the __struct__ bit is not calculated as part of the payload
- collect_payload_size(tenant_id, Map.from_struct(payload))
+ collect_payload_size(tenant_id, Map.from_struct(payload), message_type)
end
- defp collect_payload_size(tenant_id, payload) do
- :telemetry.execute(@payload_size_event, %{size: :erlang.external_size(payload)}, %{tenant: tenant_id})
+ def collect_payload_size(tenant_id, payload, message_type) do
+ :telemetry.execute(@payload_size_event, %{size: :erlang.external_size(payload)}, %{
+ tenant: tenant_id,
+ message_type: message_type
+ })
end
- defp pubsub_adapter do
- Application.fetch_env!(:realtime, :pubsub_adapter)
- end
+ defp pubsub_adapter, do: Application.fetch_env!(:realtime, :pubsub_adapter)
end
diff --git a/mix.exs b/mix.exs
index 4b0b1f40c..e98ac608f 100644
--- a/mix.exs
+++ b/mix.exs
@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
def project do
[
app: :realtime,
- version: "2.51.5",
+ version: "2.53.4",
elixir: "~> 1.17.3",
elixirc_paths: elixirc_paths(Mix.env()),
start_permanent: Mix.env() == :prod,
@@ -53,7 +53,7 @@ defmodule Realtime.MixProject do
# Type `mix help deps` for examples and options.
defp deps do
[
- {:phoenix, "~> 1.7.0"},
+ {:phoenix, override: true, github: "supabase/phoenix", branch: "feat/presence-custom-dispatcher-1.7.19"},
{:phoenix_ecto, "~> 4.4.0"},
{:ecto_sql, "~> 3.11"},
{:ecto_psql_extras, "~> 0.8"},
diff --git a/mix.lock b/mix.lock
index c5fce6022..ba6f47328 100644
--- a/mix.lock
+++ b/mix.lock
@@ -66,7 +66,7 @@
"opentelemetry_semantic_conventions": {:hex, :opentelemetry_semantic_conventions, "1.27.0", "acd0194a94a1e57d63da982ee9f4a9f88834ae0b31b0bd850815fe9be4bbb45f", [:mix, :rebar3], [], "hexpm", "9681ccaa24fd3d810b4461581717661fd85ff7019b082c2dff89c7d5b1fc2864"},
"opentelemetry_telemetry": {:hex, :opentelemetry_telemetry, "1.1.2", "410ab4d76b0921f42dbccbe5a7c831b8125282850be649ee1f70050d3961118a", [:mix, :rebar3], [{:opentelemetry_api, "~> 1.3", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.1", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "641ab469deb181957ac6d59bce6e1321d5fe2a56df444fc9c19afcad623ab253"},
"otel_http": {:hex, :otel_http, "0.2.0", "b17385986c7f1b862f5d577f72614ecaa29de40392b7618869999326b9a61d8a", [:rebar3], [], "hexpm", "f2beadf922c8cfeb0965488dd736c95cc6ea8b9efce89466b3904d317d7cc717"},
- "phoenix": {:hex, :phoenix, "1.7.19", "36617efe5afbd821099a8b994ff4618a340a5bfb25531a1802c4d4c634017a57", [:mix], [{:castore, ">= 0.0.0", [hex: :castore, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.1", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: true]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.7", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:plug_crypto, "~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:websock_adapter, "~> 0.5.3", [hex: :websock_adapter, repo: "hexpm", optional: false]}], "hexpm", "ba4dc14458278773f905f8ae6c2ec743d52c3a35b6b353733f64f02dfe096cd6"},
+ "phoenix": {:git, "https://github.com/supabase/phoenix.git", "7b884cc0cc1a49ad2bc272acda2e622b3e11c139", [branch: "feat/presence-custom-dispatcher-1.7.19"]},
"phoenix_ecto": {:hex, :phoenix_ecto, "4.4.3", "86e9878f833829c3f66da03d75254c155d91d72a201eb56ae83482328dc7ca93", [:mix], [{:ecto, "~> 3.5", [hex: :ecto, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 2.14.2 or ~> 3.0 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:plug, "~> 1.9", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "d36c401206f3011fefd63d04e8ef626ec8791975d9d107f9a0817d426f61ac07"},
"phoenix_html": {:hex, :phoenix_html, "3.3.4", "42a09fc443bbc1da37e372a5c8e6755d046f22b9b11343bf885067357da21cb3", [:mix], [{:plug, "~> 1.5", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "0249d3abec3714aff3415e7ee3d9786cb325be3151e6c4b3021502c585bf53fb"},
"phoenix_live_dashboard": {:hex, :phoenix_live_dashboard, "0.8.6", "7b1f0327f54c9eb69845fd09a77accf922f488c549a7e7b8618775eb603a62c7", [:mix], [{:ecto, "~> 3.6.2 or ~> 3.7", [hex: :ecto, repo: "hexpm", optional: true]}, {:ecto_mysql_extras, "~> 0.5", [hex: :ecto_mysql_extras, repo: "hexpm", optional: true]}, {:ecto_psql_extras, "~> 0.7", [hex: :ecto_psql_extras, repo: "hexpm", optional: true]}, {:ecto_sqlite3_extras, "~> 1.1.7 or ~> 1.2.0", [hex: :ecto_sqlite3_extras, repo: "hexpm", optional: true]}, {:mime, "~> 1.6 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:phoenix_live_view, "~> 0.19 or ~> 1.0", [hex: :phoenix_live_view, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 0.6 or ~> 1.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}], "hexpm", "1681ab813ec26ca6915beb3414aa138f298e17721dc6a2bde9e6eb8a62360ff6"},
diff --git a/test/integration/rt_channel_test.exs b/test/integration/rt_channel_test.exs
index 23b1a3a7f..8f4607b01 100644
--- a/test/integration/rt_channel_test.exs
+++ b/test/integration/rt_channel_test.exs
@@ -1159,7 +1159,7 @@ defmodule Realtime.Integration.RtChannelTest do
}
end)
- assert log =~ "ChannelShutdown: Token has expired 1000 seconds ago"
+ assert log =~ "ChannelShutdown: Token has expired"
end
test "ChannelShutdown include sub if available in jwt claims", %{tenant: tenant, topic: topic} do
@@ -2240,7 +2240,8 @@ defmodule Realtime.Integration.RtChannelTest do
# 0 events as no broadcast used
assert 2 = get_count([:realtime, :rate_counter, :channel, :joins], external_id)
assert 2 = get_count([:realtime, :rate_counter, :channel, :presence_events], external_id)
- assert 10 = get_count([:realtime, :rate_counter, :channel, :db_events], external_id)
+ # 5 + 5 + 5 (5 for each websocket and 5 while publishing)
+ assert 15 = get_count([:realtime, :rate_counter, :channel, :db_events], external_id)
assert 0 = get_count([:realtime, :rate_counter, :channel, :events], external_id)
end
@@ -2587,7 +2588,7 @@ defmodule Realtime.Integration.RtChannelTest do
Realtime.Tenants.Cache.invalidate_tenant_cache(external_id)
end
- defp assert_process_down(pid, timeout \\ 100) do
+ defp assert_process_down(pid, timeout \\ 300) do
ref = Process.monitor(pid)
assert_receive {:DOWN, ^ref, :process, ^pid, _reason}, timeout
end
diff --git a/test/realtime/extensions/cdc_rls/cdc_rls_test.exs b/test/realtime/extensions/cdc_rls/cdc_rls_test.exs
index 5f341c134..704346aab 100644
--- a/test/realtime/extensions/cdc_rls/cdc_rls_test.exs
+++ b/test/realtime/extensions/cdc_rls/cdc_rls_test.exs
@@ -235,6 +235,25 @@ defmodule Realtime.Extensions.CdcRlsTest do
end)
RateCounter.stop(tenant.external_id)
+ on_exit(fn -> RateCounter.stop(tenant.external_id) end)
+
+ on_exit(fn -> :telemetry.detach(__MODULE__) end)
+
+ :telemetry.attach(
+ __MODULE__,
+ [:realtime, :tenants, :payload, :size],
+ &__MODULE__.handle_telemetry/4,
+ pid: self()
+ )
+
+ on_exit(fn -> :telemetry.detach(__MODULE__) end)
+
+ :telemetry.attach(
+ __MODULE__,
+ [:realtime, :tenants, :payload, :size],
+ &__MODULE__.handle_telemetry/4,
+ pid: self()
+ )
%{tenant: tenant, conn: conn}
end
@@ -315,8 +334,91 @@ defmodule Realtime.Extensions.CdcRlsTest do
rate = Realtime.Tenants.db_events_per_second_rate(tenant)
- assert {:ok, %RateCounter{id: {:channel, :db_events, "dev_tenant"}, bucket: bucket}} = RateCounter.get(rate)
- assert 1 in bucket
+ assert {:ok, %RateCounter{id: {:channel, :db_events, "dev_tenant"}, bucket: bucket}} =
+ RateCounter.get(rate)
+
+ # 1 from ReplicationPoller and 1 from MessageDispatcher
+ assert Enum.sum(bucket) == 2
+
+ assert_receive {
+ :telemetry,
+ [:realtime, :tenants, :payload, :size],
+ %{size: 341},
+ %{tenant: "dev_tenant", message_type: :postgres_changes}
+ }
+ end
+
+ test "rate limit works", %{tenant: tenant, conn: conn} do
+ on_exit(fn -> PostgresCdcRls.handle_stop(tenant.external_id, 10_000) end)
+
+ %Tenant{extensions: extensions, external_id: external_id} = tenant
+ postgres_extension = PostgresCdc.filter_settings("postgres_cdc_rls", extensions)
+ args = Map.put(postgres_extension, "id", external_id)
+
+ pg_change_params = [
+ %{
+ id: UUID.uuid1(),
+ params: %{"event" => "*", "schema" => "public"},
+ channel_pid: self(),
+ claims: %{
+ "exp" => System.system_time(:second) + 100_000,
+ "iat" => 0,
+ "ref" => "127.0.0.1",
+ "role" => "anon"
+ }
+ }
+ ]
+
+ ids =
+ Enum.map(pg_change_params, fn %{id: id, params: params} ->
+ {UUID.string_to_binary!(id), :erlang.phash2(params)}
+ end)
+
+ topic = "realtime:test"
+ serializer = Phoenix.Socket.V1.JSONSerializer
+
+ subscription_metadata = {:subscriber_fastlane, self(), serializer, ids, topic, external_id, true}
+ metadata = [metadata: subscription_metadata]
+ :ok = PostgresCdc.subscribe(PostgresCdcRls, pg_change_params, external_id, metadata)
+
+ # First time it will return nil
+ PostgresCdcRls.handle_connect(args)
+ # Wait for it to start
+ Process.sleep(3000)
+ {:ok, response} = PostgresCdcRls.handle_connect(args)
+
+ # Now subscribe to the Postgres Changes
+ {:ok, _} = PostgresCdcRls.handle_after_connect(response, postgres_extension, pg_change_params)
+ assert %Postgrex.Result{rows: [[1]]} = Postgrex.query!(conn, "select count(*) from realtime.subscription", [])
+
+ log =
+ capture_log(fn ->
+ # increment artifically the counter to reach the limit
+ tenant.external_id
+ |> Realtime.Tenants.db_events_per_second_key()
+ |> Realtime.GenCounter.add(100_000_000)
+
+ # Wait for RateCounter to update
+ Process.sleep(1500)
+ end)
+
+ assert log =~ "MessagePerSecondRateLimitReached: Too many postgres changes messages per second"
+
+ # Insert a record
+ %{rows: [[_id]]} = Postgrex.query!(conn, "insert into test (details) values ('test') returning id", [])
+
+ refute_receive {:socket_push, :text, _}, 5000
+
+ # Wait for RateCounter to update
+ Process.sleep(2000)
+
+ rate = Realtime.Tenants.db_events_per_second_rate(tenant)
+
+ assert {:ok, %RateCounter{id: {:channel, :db_events, "dev_tenant"}, bucket: bucket, limit: %{triggered: true}}} =
+ RateCounter.get(rate)
+
+ # Nothing has changed
+ assert Enum.sum(bucket) == 100_000_000
end
@aux_mod (quote do
@@ -414,4 +516,6 @@ defmodule Realtime.Extensions.CdcRlsTest do
:erpc.call(node, PostgresCdcRls, :handle_stop, [tenant.external_id, 10_000])
end
end
+
+ def handle_telemetry(event, measures, metadata, pid: pid), do: send(pid, {:telemetry, event, measures, metadata})
end
diff --git a/test/realtime/gen_rpc_pub_sub_test.exs b/test/realtime/gen_rpc_pub_sub_test.exs
index 0013c2e7b..517c6c369 100644
--- a/test/realtime/gen_rpc_pub_sub_test.exs
+++ b/test/realtime/gen_rpc_pub_sub_test.exs
@@ -1,2 +1,18 @@
Application.put_env(:phoenix_pubsub, :test_adapter, {Realtime.GenRpcPubSub, []})
Code.require_file("../../deps/phoenix_pubsub/test/shared/pubsub_test.exs", __DIR__)
+
+defmodule Realtime.GenRpcPubSubTest do
+ use ExUnit.Case, async: true
+
+ test "it sets off_heap message_queue_data flag on the workers" do
+ assert Realtime.PubSubElixir.Realtime.PubSub.Adapter_1
+ |> Process.whereis()
+ |> Process.info(:message_queue_data) == {:message_queue_data, :off_heap}
+ end
+
+ test "it sets fullsweep_after flag on the workers" do
+ assert Realtime.PubSubElixir.Realtime.PubSub.Adapter_1
+ |> Process.whereis()
+ |> Process.info(:fullsweep_after) == {:fullsweep_after, 100}
+ end
+end
diff --git a/test/realtime/monitoring/prom_ex/plugins/tenant_test.exs b/test/realtime/monitoring/prom_ex/plugins/tenant_test.exs
index 164c8d2eb..77c1dc7cf 100644
--- a/test/realtime/monitoring/prom_ex/plugins/tenant_test.exs
+++ b/test/realtime/monitoring/prom_ex/plugins/tenant_test.exs
@@ -129,6 +129,17 @@ defmodule Realtime.PromEx.Plugins.TenantTest do
assert metric_value(pattern) == metric_value + 1
end
+ test "global event exists after counter added", %{tenant: %{external_id: external_id}} do
+ pattern =
+ ~r/realtime_channel_global_events\s(?
\d+)/
+
+ metric_value = metric_value(pattern)
+ FakeUserCounter.fake_event(external_id)
+
+ Process.sleep(200)
+ assert metric_value(pattern) == metric_value + 1
+ end
+
test "db_event exists after counter added", %{tenant: %{external_id: external_id}} do
pattern =
~r/realtime_channel_db_events{tenant="#{external_id}"}\s(?\d+)/
@@ -139,6 +150,16 @@ defmodule Realtime.PromEx.Plugins.TenantTest do
assert metric_value(pattern) == metric_value + 1
end
+ test "global db_event exists after counter added", %{tenant: %{external_id: external_id}} do
+ pattern =
+ ~r/realtime_channel_global_db_events\s(?\d+)/
+
+ metric_value = metric_value(pattern)
+ FakeUserCounter.fake_db_event(external_id)
+ Process.sleep(200)
+ assert metric_value(pattern) == metric_value + 1
+ end
+
test "presence_event exists after counter added", %{tenant: %{external_id: external_id}} do
pattern =
~r/realtime_channel_presence_events{tenant="#{external_id}"}\s(?\d+)/
@@ -149,6 +170,16 @@ defmodule Realtime.PromEx.Plugins.TenantTest do
assert metric_value(pattern) == metric_value + 1
end
+ test "global presence_event exists after counter added", %{tenant: %{external_id: external_id}} do
+ pattern =
+ ~r/realtime_channel_global_presence_events\s(?\d+)/
+
+ metric_value = metric_value(pattern)
+ FakeUserCounter.fake_presence_event(external_id)
+ Process.sleep(200)
+ assert metric_value(pattern) == metric_value + 1
+ end
+
test "metric read_authorization_check exists after check", context do
pattern =
~r/realtime_tenants_read_authorization_check_count{tenant="#{context.tenant.external_id}"}\s(?\d+)/
@@ -231,18 +262,18 @@ defmodule Realtime.PromEx.Plugins.TenantTest do
external_id = context.tenant.external_id
pattern =
- ~r/realtime_tenants_payload_size_count{tenant="#{external_id}"}\s(?\d+)/
+ ~r/realtime_tenants_payload_size_count{message_type=\"presence\",tenant="#{external_id}"}\s(?\d+)/
metric_value = metric_value(pattern)
message = %{topic: "a topic", event: "an event", payload: ["a", %{"b" => "c"}, 1, 23]}
- RealtimeWeb.TenantBroadcaster.pubsub_broadcast(external_id, "a topic", message, Phoenix.PubSub)
+ RealtimeWeb.TenantBroadcaster.pubsub_broadcast(external_id, "a topic", message, Phoenix.PubSub, :presence)
Process.sleep(200)
assert metric_value(pattern) == metric_value + 1
bucket_pattern =
- ~r/realtime_tenants_payload_size_bucket{tenant="#{external_id}",le="100"}\s(?\d+)/
+ ~r/realtime_tenants_payload_size_bucket{message_type=\"presence\",tenant="#{external_id}",le="250"}\s(?\d+)/
assert metric_value(bucket_pattern) > 0
end
@@ -250,17 +281,17 @@ defmodule Realtime.PromEx.Plugins.TenantTest do
test "global metric payload size", context do
external_id = context.tenant.external_id
- pattern = ~r/realtime_payload_size_count\s(?\d+)/
+ pattern = ~r/realtime_payload_size_count{message_type=\"broadcast\"}\s(?\d+)/
metric_value = metric_value(pattern)
message = %{topic: "a topic", event: "an event", payload: ["a", %{"b" => "c"}, 1, 23]}
- RealtimeWeb.TenantBroadcaster.pubsub_broadcast(external_id, "a topic", message, Phoenix.PubSub)
+ RealtimeWeb.TenantBroadcaster.pubsub_broadcast(external_id, "a topic", message, Phoenix.PubSub, :broadcast)
Process.sleep(200)
assert metric_value(pattern) == metric_value + 1
- bucket_pattern = ~r/realtime_payload_size_bucket{le="100"}\s(?\d+)/
+ bucket_pattern = ~r/realtime_payload_size_bucket{message_type=\"broadcast\",le="250"}\s(?\d+)/
assert metric_value(bucket_pattern) > 0
end
diff --git a/test/realtime/monitoring/prom_ex/plugins/tenants_test.exs b/test/realtime/monitoring/prom_ex/plugins/tenants_test.exs
index 080fd3cfb..ded087c74 100644
--- a/test/realtime/monitoring/prom_ex/plugins/tenants_test.exs
+++ b/test/realtime/monitoring/prom_ex/plugins/tenants_test.exs
@@ -37,6 +37,16 @@ defmodule Realtime.PromEx.Plugins.TenantsTest do
assert metric_value(pattern) == previous_value + 1
end
+ test "global success" do
+ pattern = ~r/realtime_global_rpc_count{mechanism=\"erpc\",success="true"}\s(?\d+)/
+ # Enough time for the poll rate to be triggered at least once
+ Process.sleep(200)
+ previous_value = metric_value(pattern)
+ assert {:ok, "success"} = Rpc.enhanced_call(node(), Test, :success, [], tenant_id: "123")
+ Process.sleep(200)
+ assert metric_value(pattern) == previous_value + 1
+ end
+
test "failure" do
pattern = ~r/realtime_rpc_count{mechanism=\"erpc\",success="false",tenant="123"}\s(?\d+)/
# Enough time for the poll rate to be triggered at least once
@@ -47,6 +57,16 @@ defmodule Realtime.PromEx.Plugins.TenantsTest do
assert metric_value(pattern) == previous_value + 1
end
+ test "global failure" do
+ pattern = ~r/realtime_global_rpc_count{mechanism=\"erpc\",success="false"}\s(?\d+)/
+ # Enough time for the poll rate to be triggered at least once
+ Process.sleep(200)
+ previous_value = metric_value(pattern)
+ assert {:error, "failure"} = Rpc.enhanced_call(node(), Test, :failure, [], tenant_id: "123")
+ Process.sleep(200)
+ assert metric_value(pattern) == previous_value + 1
+ end
+
test "exception" do
pattern = ~r/realtime_rpc_count{mechanism=\"erpc\",success="false",tenant="123"}\s(?\d+)/
# Enough time for the poll rate to be triggered at least once
@@ -59,6 +79,19 @@ defmodule Realtime.PromEx.Plugins.TenantsTest do
Process.sleep(200)
assert metric_value(pattern) == previous_value + 1
end
+
+ test "global exception" do
+ pattern = ~r/realtime_global_rpc_count{mechanism=\"erpc\",success="false"}\s(?\d+)/
+ # Enough time for the poll rate to be triggered at least once
+ Process.sleep(200)
+ previous_value = metric_value(pattern)
+
+ assert {:error, :rpc_error, %RuntimeError{message: "runtime error"}} =
+ Rpc.enhanced_call(node(), Test, :exception, [], tenant_id: "123")
+
+ Process.sleep(200)
+ assert metric_value(pattern) == previous_value + 1
+ end
end
test "event_metrics rpc" do
diff --git a/test/realtime/tenants/connect_test.exs b/test/realtime/tenants/connect_test.exs
index a52973d53..804b3018f 100644
--- a/test/realtime/tenants/connect_test.exs
+++ b/test/realtime/tenants/connect_test.exs
@@ -51,6 +51,36 @@ defmodule Realtime.Tenants.ConnectTest do
end
describe "handle cold start" do
+ test "multiple processes connecting calling Connect.connect", %{tenant: tenant} do
+ parent = self()
+
+ # Let's slow down Connect.connect so that multiple RPC calls are executed
+ stub(Connect, :connect, fn x, y, z ->
+ :timer.sleep(1000)
+ call_original(Connect, :connect, [x, y, z])
+ end)
+
+ connect = fn -> send(parent, Connect.lookup_or_start_connection(tenant.external_id)) end
+ # Let's call enough times to potentially trigger the Connect RateCounter
+
+ for _ <- 1..50, do: spawn(connect)
+
+ assert_receive({:ok, pid}, 1100)
+
+ for _ <- 1..49, do: assert_receive({:ok, ^pid})
+
+ # Does not trigger rate limit as connections eventually succeeded
+
+ {:ok, rate_counter} =
+ tenant.external_id
+ |> Tenants.connect_errors_per_second_rate()
+ |> Realtime.RateCounter.get()
+
+ assert rate_counter.sum == 0
+ assert rate_counter.avg == 0.0
+ assert rate_counter.limit.triggered == false
+ end
+
test "multiple proccesses succeed together", %{tenant: tenant} do
parent = self()
@@ -298,9 +328,9 @@ defmodule Realtime.Tenants.ConnectTest do
region = Tenants.region(tenant)
assert {_pid, %{conn: ^db_conn, region: ^region}} = :syn.lookup(Connect, external_id)
Process.sleep(1000)
- :syn.leave(:users, external_id, self())
+ external_id |> UsersCounter.scope() |> :syn.leave(external_id, self())
Process.sleep(1000)
- assert :undefined = :syn.lookup(Connect, external_id)
+ assert :undefined = external_id |> UsersCounter.scope() |> :syn.lookup(external_id)
refute Process.alive?(db_conn)
Connect.shutdown(external_id)
end
diff --git a/test/realtime_web/channels/payloads/join_test.exs b/test/realtime_web/channels/payloads/join_test.exs
index c1ea54a67..f02c2a73d 100644
--- a/test/realtime_web/channels/payloads/join_test.exs
+++ b/test/realtime_web/channels/payloads/join_test.exs
@@ -58,6 +58,14 @@ defmodule RealtimeWeb.Channels.Payloads.JoinTest do
assert is_binary(key)
end
+ test "presence key can be number" do
+ config = %{"config" => %{"presence" => %{"enabled" => true, "key" => 123}}}
+
+ assert {:ok, %Join{config: %Config{presence: %Presence{key: key}}}} = Join.validate(config)
+
+ assert key == 123
+ end
+
test "invalid replay" do
config = %{"config" => %{"broadcast" => %{"replay" => 123}}}
@@ -105,5 +113,11 @@ defmodule RealtimeWeb.Channels.Payloads.JoinTest do
user_token: ["unable to parse, expected string"]
}
end
+
+ test "handles postgres changes with nil value in array as empty array" do
+ config = %{"config" => %{"postgres_changes" => [nil]}}
+
+ assert {:ok, %Join{config: %Config{postgres_changes: []}}} = Join.validate(config)
+ end
end
end
diff --git a/test/realtime_web/channels/realtime_channel/message_dispatcher_test.exs b/test/realtime_web/channels/realtime_channel/message_dispatcher_test.exs
index 44ce83b99..53be2e51f 100644
--- a/test/realtime_web/channels/realtime_channel/message_dispatcher_test.exs
+++ b/test/realtime_web/channels/realtime_channel/message_dispatcher_test.exs
@@ -16,12 +16,24 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcherTest do
describe "fastlane_metadata/5" do
test "info level" do
assert MessageDispatcher.fastlane_metadata(self(), Serializer, "realtime:topic", :info, "tenant_id") ==
- {:rc_fastlane, self(), Serializer, "realtime:topic", {:log, "tenant_id"}, MapSet.new()}
+ {:rc_fastlane, self(), Serializer, "realtime:topic", :info, "tenant_id", MapSet.new()}
end
test "non-info level" do
assert MessageDispatcher.fastlane_metadata(self(), Serializer, "realtime:topic", :warning, "tenant_id") ==
- {:rc_fastlane, self(), Serializer, "realtime:topic", MapSet.new()}
+ {:rc_fastlane, self(), Serializer, "realtime:topic", :warning, "tenant_id", MapSet.new()}
+ end
+
+ test "replayed message ids" do
+ assert MessageDispatcher.fastlane_metadata(
+ self(),
+ Serializer,
+ "realtime:topic",
+ :warning,
+ "tenant_id",
+ MapSet.new([1])
+ ) ==
+ {:rc_fastlane, self(), Serializer, "realtime:topic", :warning, "tenant_id", MapSet.new([1])}
end
end
@@ -50,8 +62,8 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcherTest do
from_pid = :erlang.list_to_pid(~c'<0.2.1>')
subscribers = [
- {subscriber_pid, {:rc_fastlane, self(), TestSerializer, "realtime:topic", {:log, "tenant123"}, MapSet.new()}},
- {subscriber_pid, {:rc_fastlane, self(), TestSerializer, "realtime:topic", MapSet.new()}}
+ {subscriber_pid, {:rc_fastlane, self(), TestSerializer, "realtime:topic", :info, "tenant123", MapSet.new()}},
+ {subscriber_pid, {:rc_fastlane, self(), TestSerializer, "realtime:topic", :warning, "tenant123", MapSet.new()}}
]
msg = %Broadcast{topic: "some:other:topic", event: "event", payload: %{data: "test"}}
@@ -74,6 +86,48 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcherTest do
refute_receive _any
end
+ test "dispatches 'presence_diff' messages to fastlane subscribers" do
+ parent = self()
+
+ subscriber_pid =
+ spawn(fn ->
+ loop = fn loop ->
+ receive do
+ msg ->
+ send(parent, {:subscriber, msg})
+ loop.(loop)
+ end
+ end
+
+ loop.(loop)
+ end)
+
+ from_pid = :erlang.list_to_pid(~c'<0.2.1>')
+
+ subscribers = [
+ {subscriber_pid, {:rc_fastlane, self(), TestSerializer, "realtime:topic", :info, "tenant456", MapSet.new()}},
+ {subscriber_pid, {:rc_fastlane, self(), TestSerializer, "realtime:topic", :warning, "tenant456", MapSet.new()}}
+ ]
+
+ msg = %Broadcast{topic: "some:other:topic", event: "presence_diff", payload: %{data: "test"}}
+
+ log =
+ capture_log(fn ->
+ assert MessageDispatcher.dispatch(subscribers, from_pid, msg) == :ok
+ end)
+
+ assert log =~ "Received message on realtime:topic with payload: #{inspect(msg, pretty: true)}"
+
+ assert_receive {:encoded, %Broadcast{event: "presence_diff", payload: %{data: "test"}, topic: "realtime:topic"}}
+ assert_receive {:encoded, %Broadcast{event: "presence_diff", payload: %{data: "test"}, topic: "realtime:topic"}}
+
+ assert Agent.get(TestSerializer, & &1) == 1
+
+ assert Realtime.GenCounter.get(Realtime.Tenants.presence_events_per_second_key("tenant456")) == 2
+
+ refute_receive _any
+ end
+
test "does not dispatch messages to fastlane subscribers if they already replayed it" do
parent = self()
@@ -95,8 +149,9 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcherTest do
subscribers = [
{subscriber_pid,
- {:rc_fastlane, self(), TestSerializer, "realtime:topic", {:log, "tenant123"}, replaeyd_message_ids}},
- {subscriber_pid, {:rc_fastlane, self(), TestSerializer, "realtime:topic", replaeyd_message_ids}}
+ {:rc_fastlane, self(), TestSerializer, "realtime:topic", :info, "tenant123", replaeyd_message_ids}},
+ {subscriber_pid,
+ {:rc_fastlane, self(), TestSerializer, "realtime:topic", :warning, "tenant123", replaeyd_message_ids}}
]
msg = %Broadcast{
@@ -131,8 +186,8 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcherTest do
from_pid = :erlang.list_to_pid(~c'<0.2.1>')
subscribers = [
- {subscriber_pid, {:rc_fastlane, self(), TestSerializer, "realtime:topic", {:log, "tenant123"}, MapSet.new()}},
- {subscriber_pid, {:rc_fastlane, self(), TestSerializer, "realtime:topic", MapSet.new()}}
+ {subscriber_pid, {:rc_fastlane, self(), TestSerializer, "realtime:topic", :info, "tenant123", MapSet.new()}},
+ {subscriber_pid, {:rc_fastlane, self(), TestSerializer, "realtime:topic", :warning, "tenant123", MapSet.new()}}
]
msg = %Broadcast{topic: "some:other:topic", event: "event", payload: "not a map"}
diff --git a/test/realtime_web/channels/realtime_channel/presence_handler_test.exs b/test/realtime_web/channels/realtime_channel/presence_handler_test.exs
index 0cdf422e2..219f13e55 100644
--- a/test/realtime_web/channels/realtime_channel/presence_handler_test.exs
+++ b/test/realtime_web/channels/realtime_channel/presence_handler_test.exs
@@ -100,25 +100,41 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
end
describe "handle/3" do
+ setup do
+ on_exit(fn -> :telemetry.detach(__MODULE__) end)
+
+ :telemetry.attach(
+ __MODULE__,
+ [:realtime, :tenants, :payload, :size],
+ &__MODULE__.handle_telemetry/4,
+ pid: self()
+ )
+ end
+
test "with true policy and is private, user can track their presence and changes", %{
tenant: tenant,
topic: topic,
db_conn: db_conn
} do
+ external_id = tenant.external_id
key = random_string()
policies = %Policies{presence: %PresencePolicies{read: true, write: true}}
socket =
socket_fixture(tenant, topic, key, policies: policies)
- PresenceHandler.handle(%{"event" => "track"}, db_conn, socket)
+ PresenceHandler.handle(%{"event" => "track", "payload" => %{"A" => "b", "c" => "b"}}, db_conn, socket)
topic = socket.assigns.tenant_topic
assert_receive %Broadcast{topic: ^topic, event: "presence_diff", payload: %{joins: joins, leaves: %{}}}
assert Map.has_key?(joins, key)
+
+ assert_receive {:telemetry, [:realtime, :tenants, :payload, :size], %{size: 30},
+ %{tenant: ^external_id, message_type: :presence}}
end
test "when tracking already existing user, metadata updated", %{tenant: tenant, topic: topic, db_conn: db_conn} do
+ external_id = tenant.external_id
key = random_string()
policies = %Policies{presence: %PresencePolicies{read: true, write: true}}
socket = socket_fixture(tenant, topic, key, policies: policies)
@@ -134,10 +150,18 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
assert_receive %Broadcast{topic: ^topic, event: "presence_diff", payload: %{joins: joins, leaves: %{}}}
assert Map.has_key?(joins, key)
+
+ assert_receive {:telemetry, [:realtime, :tenants, :payload, :size], %{size: 6},
+ %{tenant: ^external_id, message_type: :presence}}
+
+ assert_receive {:telemetry, [:realtime, :tenants, :payload, :size], %{size: 55},
+ %{tenant: ^external_id, message_type: :presence}}
+
refute_receive :_
end
test "with false policy and is public, user can track their presence and changes", %{tenant: tenant, topic: topic} do
+ external_id = tenant.external_id
key = random_string()
policies = %Policies{presence: %PresencePolicies{read: false, write: false}}
socket = socket_fixture(tenant, topic, key, policies: policies, private?: false)
@@ -147,6 +171,9 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
topic = socket.assigns.tenant_topic
assert_receive %Broadcast{topic: ^topic, event: "presence_diff", payload: %{joins: joins, leaves: %{}}}
assert Map.has_key?(joins, key)
+
+ assert_receive {:telemetry, [:realtime, :tenants, :payload, :size], %{size: 6},
+ %{tenant: ^external_id, message_type: :presence}}
end
test "user can untrack when they want", %{tenant: tenant, topic: topic, db_conn: db_conn} do
@@ -434,6 +461,7 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
assert log =~ "PresenceRateLimitReached"
end
+ @tag :skip
@tag policies: [:authenticated_read_broadcast_and_presence, :authenticated_write_broadcast_and_presence]
test "respects rate limits on private channels", %{tenant: tenant, topic: topic, db_conn: db_conn} do
key = random_string()
@@ -517,4 +545,6 @@ defmodule RealtimeWeb.RealtimeChannel.PresenceHandlerTest do
}
}
end
+
+ def handle_telemetry(event, measures, metadata, pid: pid), do: send(pid, {:telemetry, event, measures, metadata})
end
diff --git a/test/realtime_web/channels/realtime_channel_test.exs b/test/realtime_web/channels/realtime_channel_test.exs
index ae6c1734a..8022d6ebd 100644
--- a/test/realtime_web/channels/realtime_channel_test.exs
+++ b/test/realtime_web/channels/realtime_channel_test.exs
@@ -239,23 +239,14 @@ defmodule RealtimeWeb.RealtimeChannelTest do
end
describe "presence" do
- test "events are counted", %{tenant: tenant} do
+ test "presence state event is counted", %{tenant: tenant} do
jwt = Generators.generate_jwt_token(tenant)
{:ok, %Socket{} = socket} = connect(UserSocket, %{"log_level" => "warning"}, conn_opts(tenant, jwt))
assert {:ok, _, %Socket{} = socket} = subscribe_and_join(socket, "realtime:test", %{})
- presence_diff = %Socket.Broadcast{event: "presence_diff", payload: %{joins: %{}, leaves: %{}}}
- send(socket.channel_pid, presence_diff)
-
assert_receive %Socket.Message{topic: "realtime:test", event: "presence_state", payload: %{}}
- assert_receive %Socket.Message{
- topic: "realtime:test",
- event: "presence_diff",
- payload: %{joins: %{}, leaves: %{}}
- }
-
tenant_id = tenant.external_id
# Wait for RateCounter to tick
@@ -264,8 +255,8 @@ defmodule RealtimeWeb.RealtimeChannelTest do
assert {:ok, %RateCounter{id: {:channel, :presence_events, ^tenant_id}, bucket: bucket}} =
RateCounter.get(socket.assigns.presence_rate_counter)
- # presence_state + presence_diff
- assert 2 in bucket
+ # presence_state
+ assert Enum.sum(bucket) == 1
end
end
diff --git a/test/realtime_web/channels/tenant_rate_limiters_test.exs b/test/realtime_web/channels/tenant_rate_limiters_test.exs
new file mode 100644
index 000000000..05d56ec82
--- /dev/null
+++ b/test/realtime_web/channels/tenant_rate_limiters_test.exs
@@ -0,0 +1,31 @@
+defmodule RealtimeWeb.TenantRateLimitersTest do
+ use Realtime.DataCase, async: true
+
+ use Mimic
+ alias RealtimeWeb.TenantRateLimiters
+ alias Realtime.Api.Tenant
+
+ setup do
+ tenant = %Tenant{external_id: random_string(), max_concurrent_users: 1, max_joins_per_second: 1}
+
+ %{tenant: tenant}
+ end
+
+ describe "check_tenant/1" do
+ test "rate is not exceeded", %{tenant: tenant} do
+ assert TenantRateLimiters.check_tenant(tenant) == :ok
+ end
+
+ test "max concurrent users is exceeded", %{tenant: tenant} do
+ Realtime.UsersCounter.add(self(), tenant.external_id)
+
+ assert TenantRateLimiters.check_tenant(tenant) == {:error, :too_many_connections}
+ end
+
+ test "max joins is exceeded", %{tenant: tenant} do
+ expect(Realtime.RateCounter, :get, fn _ -> {:ok, %{limit: %{triggered: true}}} end)
+
+ assert TenantRateLimiters.check_tenant(tenant) == {:error, :too_many_joins}
+ end
+ end
+end
diff --git a/test/realtime_web/controllers/broadcast_controller_test.exs b/test/realtime_web/controllers/broadcast_controller_test.exs
index 7bd426353..900eb7aa9 100644
--- a/test/realtime_web/controllers/broadcast_controller_test.exs
+++ b/test/realtime_web/controllers/broadcast_controller_test.exs
@@ -272,8 +272,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
} do
request_events_key = Tenants.requests_per_second_key(tenant)
broadcast_events_key = Tenants.events_per_second_key(tenant)
- connect_events_key = Tenants.connect_per_second_rate(tenant).id
- expect(TenantBroadcaster, :pubsub_broadcast, 5, fn _, _, _, _ -> :ok end)
+ expect(TenantBroadcaster, :pubsub_broadcast, 5, fn _, _, _, _, _ -> :ok end)
messages_to_send =
Stream.repeatedly(fn -> generate_message_with_policies(db_conn, tenant) end)
@@ -298,7 +297,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
conn = post(conn, Routes.broadcast_path(conn, :broadcast), %{"messages" => messages})
- broadcast_calls = calls(&TenantBroadcaster.pubsub_broadcast/4)
+ broadcast_calls = calls(&TenantBroadcaster.pubsub_broadcast/5)
Enum.each(messages_to_send, fn %{topic: topic} ->
broadcast_topic = Tenants.tenant_topic(tenant, topic, false)
@@ -314,7 +313,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
}
assert Enum.any?(broadcast_calls, fn
- [_, ^broadcast_topic, ^message, RealtimeChannel.MessageDispatcher] -> true
+ [_, ^broadcast_topic, ^message, RealtimeChannel.MessageDispatcher, :broadcast] -> true
_ -> false
end)
end)
@@ -330,8 +329,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
} do
request_events_key = Tenants.requests_per_second_key(tenant)
broadcast_events_key = Tenants.events_per_second_key(tenant)
- connect_events_key = Tenants.connect_per_second_rate(tenant).id
- expect(TenantBroadcaster, :pubsub_broadcast, 6, fn _, _, _, _ -> :ok end)
+ expect(TenantBroadcaster, :pubsub_broadcast, 6, fn _, _, _, _, _ -> :ok end)
channels =
Stream.repeatedly(fn -> generate_message_with_policies(db_conn, tenant) end)
@@ -366,7 +364,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
conn = post(conn, Routes.broadcast_path(conn, :broadcast), %{"messages" => messages})
- broadcast_calls = calls(&TenantBroadcaster.pubsub_broadcast/4)
+ broadcast_calls = calls(&TenantBroadcaster.pubsub_broadcast/5)
Enum.each(channels, fn %{topic: topic} ->
broadcast_topic = Tenants.tenant_topic(tenant, topic, false)
@@ -382,7 +380,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
}
assert Enum.count(broadcast_calls, fn
- [_, ^broadcast_topic, ^message, RealtimeChannel.MessageDispatcher] -> true
+ [_, ^broadcast_topic, ^message, RealtimeChannel.MessageDispatcher, :broadcast] -> true
_ -> false
end) == 1
end)
@@ -401,7 +399,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
open_channel_topic = Tenants.tenant_topic(tenant, "open_channel", true)
assert Enum.count(broadcast_calls, fn
- [_, ^open_channel_topic, ^message, RealtimeChannel.MessageDispatcher] -> true
+ [_, ^open_channel_topic, ^message, RealtimeChannel.MessageDispatcher, :broadcast] -> true
_ -> false
end) == 1
@@ -416,8 +414,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
} do
request_events_key = Tenants.requests_per_second_key(tenant)
broadcast_events_key = Tenants.events_per_second_key(tenant)
- connect_events_key = Tenants.connect_per_second_rate(tenant).id
- expect(TenantBroadcaster, :pubsub_broadcast, 5, fn _, _, _, _ -> :ok end)
+ expect(TenantBroadcaster, :pubsub_broadcast, 5, fn _, _, _, _, _ -> :ok end)
messages_to_send =
Stream.repeatedly(fn -> generate_message_with_policies(db_conn, tenant) end)
@@ -438,12 +435,11 @@ defmodule RealtimeWeb.BroadcastControllerTest do
GenCounter
|> expect(:add, fn ^request_events_key -> :ok end)
# remove the one message that won't be broadcasted for this user
- |> expect(:add, 1, fn ^connect_events_key -> :ok end)
|> expect(:add, length(messages) - 1, fn ^broadcast_events_key -> :ok end)
conn = post(conn, Routes.broadcast_path(conn, :broadcast), %{"messages" => messages})
- broadcast_calls = calls(&TenantBroadcaster.pubsub_broadcast/4)
+ broadcast_calls = calls(&TenantBroadcaster.pubsub_broadcast/5)
Enum.each(messages_to_send, fn %{topic: topic} ->
broadcast_topic = Tenants.tenant_topic(tenant, topic, false)
@@ -459,7 +455,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
}
assert Enum.count(broadcast_calls, fn
- [_, ^broadcast_topic, ^message, RealtimeChannel.MessageDispatcher] -> true
+ [_, ^broadcast_topic, ^message, RealtimeChannel.MessageDispatcher, :broadcast] -> true
_ -> false
end) == 1
end)
@@ -472,7 +468,7 @@ defmodule RealtimeWeb.BroadcastControllerTest do
@tag role: "anon"
test "user without permission won't broadcast", %{conn: conn, db_conn: db_conn, tenant: tenant} do
request_events_key = Tenants.requests_per_second_key(tenant)
- reject(&TenantBroadcaster.pubsub_broadcast/4)
+ reject(&TenantBroadcaster.pubsub_broadcast/5)
messages =
Stream.repeatedly(fn -> generate_message_with_policies(db_conn, tenant) end)
diff --git a/test/realtime_web/live/status_live/index_test.exs b/test/realtime_web/live/status_live/index_test.exs
new file mode 100644
index 000000000..ae3af0ad0
--- /dev/null
+++ b/test/realtime_web/live/status_live/index_test.exs
@@ -0,0 +1,33 @@
+defmodule RealtimeWeb.StatusLive.IndexTest do
+ use RealtimeWeb.ConnCase
+ import Phoenix.LiveViewTest
+
+ alias Realtime.Latency.Payload
+ alias Realtime.Nodes
+ alias RealtimeWeb.Endpoint
+
+ describe "Status LiveView" do
+ test "renders status page", %{conn: conn} do
+ {:ok, _view, html} = live(conn, ~p"/status")
+
+ assert html =~ "Realtime Status"
+ end
+
+ test "receives broadcast from PubSub", %{conn: conn} do
+ {:ok, view, _html} = live(conn, ~p"/status")
+
+ payload = %Payload{
+ from_node: Nodes.short_node_id_from_name(:"pink@127.0.0.1"),
+ node: Nodes.short_node_id_from_name(:"orange@127.0.0.1"),
+ latency: "42ms",
+ timestamp: DateTime.utc_now()
+ }
+
+ Endpoint.broadcast("admin:cluster", "ping", payload)
+
+ html = render(view)
+ assert html =~ "42ms"
+ assert html =~ "pink@127.0.0.1_orange@127.0.0.1"
+ end
+ end
+end
diff --git a/test/realtime_web/tenant_broadcaster_test.exs b/test/realtime_web/tenant_broadcaster_test.exs
index ddda381a1..bc3b4f90a 100644
--- a/test/realtime_web/tenant_broadcaster_test.exs
+++ b/test/realtime_web/tenant_broadcaster_test.exs
@@ -60,7 +60,7 @@ defmodule RealtimeWeb.TenantBroadcasterTest do
test "pubsub_broadcast", %{node: node} do
message = %Broadcast{topic: @topic, event: "an event", payload: %{"a" => "b"}}
- TenantBroadcaster.pubsub_broadcast("realtime-dev", @topic, message, Phoenix.PubSub)
+ TenantBroadcaster.pubsub_broadcast("realtime-dev", @topic, message, Phoenix.PubSub, :broadcast)
assert_receive ^message
@@ -71,13 +71,13 @@ defmodule RealtimeWeb.TenantBroadcasterTest do
:telemetry,
[:realtime, :tenants, :payload, :size],
%{size: 114},
- %{tenant: "realtime-dev"}
+ %{tenant: "realtime-dev", message_type: :broadcast}
}
end
test "pubsub_broadcast list payload", %{node: node} do
message = %Broadcast{topic: @topic, event: "an event", payload: ["a", %{"b" => "c"}, 1, 23]}
- TenantBroadcaster.pubsub_broadcast("realtime-dev", @topic, message, Phoenix.PubSub)
+ TenantBroadcaster.pubsub_broadcast("realtime-dev", @topic, message, Phoenix.PubSub, :broadcast)
assert_receive ^message
@@ -88,13 +88,13 @@ defmodule RealtimeWeb.TenantBroadcasterTest do
:telemetry,
[:realtime, :tenants, :payload, :size],
%{size: 130},
- %{tenant: "realtime-dev"}
+ %{tenant: "realtime-dev", message_type: :broadcast}
}
end
test "pubsub_broadcast string payload", %{node: node} do
message = %Broadcast{topic: @topic, event: "an event", payload: "some text payload"}
- TenantBroadcaster.pubsub_broadcast("realtime-dev", @topic, message, Phoenix.PubSub)
+ TenantBroadcaster.pubsub_broadcast("realtime-dev", @topic, message, Phoenix.PubSub, :broadcast)
assert_receive ^message
@@ -105,7 +105,7 @@ defmodule RealtimeWeb.TenantBroadcasterTest do
:telemetry,
[:realtime, :tenants, :payload, :size],
%{size: 119},
- %{tenant: "realtime-dev"}
+ %{tenant: "realtime-dev", message_type: :broadcast}
}
end
end
@@ -131,7 +131,7 @@ defmodule RealtimeWeb.TenantBroadcasterTest do
message = %Broadcast{topic: @topic, event: "an event", payload: %{"a" => "b"}}
- TenantBroadcaster.pubsub_broadcast_from("realtime-dev", self(), @topic, message, Phoenix.PubSub)
+ TenantBroadcaster.pubsub_broadcast_from("realtime-dev", self(), @topic, message, Phoenix.PubSub, :broadcast)
assert_receive {:other_process, ^message}
@@ -142,7 +142,7 @@ defmodule RealtimeWeb.TenantBroadcasterTest do
:telemetry,
[:realtime, :tenants, :payload, :size],
%{size: 114},
- %{tenant: "realtime-dev"}
+ %{tenant: "realtime-dev", message_type: :broadcast}
}
# This process does not receive the message
@@ -151,5 +151,38 @@ defmodule RealtimeWeb.TenantBroadcasterTest do
end
end
+ describe "collect_payload_size/3" do
+ @describetag pubsub_adapter: :gen_rpc
+
+ test "emit telemetry for struct" do
+ TenantBroadcaster.collect_payload_size(
+ "realtime-dev",
+ %Phoenix.Socket.Broadcast{event: "broadcast", payload: %{"a" => "b"}},
+ :broadcast
+ )
+
+ assert_receive {:telemetry, [:realtime, :tenants, :payload, :size], %{size: 65},
+ %{tenant: "realtime-dev", message_type: :broadcast}}
+ end
+
+ test "emit telemetry for map" do
+ TenantBroadcaster.collect_payload_size(
+ "realtime-dev",
+ %{event: "broadcast", payload: %{"a" => "b"}},
+ :postgres_changes
+ )
+
+ assert_receive {:telemetry, [:realtime, :tenants, :payload, :size], %{size: 53},
+ %{tenant: "realtime-dev", message_type: :postgres_changes}}
+ end
+
+ test "emit telemetry for non-map" do
+ TenantBroadcaster.collect_payload_size("realtime-dev", "some blob", :presence)
+
+ assert_receive {:telemetry, [:realtime, :tenants, :payload, :size], %{size: 15},
+ %{tenant: "realtime-dev", message_type: :presence}}
+ end
+ end
+
def handle_telemetry(event, measures, metadata, pid: pid), do: send(pid, {:telemetry, event, measures, metadata})
end