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/mix.exs b/mix.exs
index d0e42bf11..e98ac608f 100644
--- a/mix.exs
+++ b/mix.exs
@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
def project do
[
app: :realtime,
- version: "2.53.0",
+ version: "2.53.4",
elixir: "~> 1.17.3",
elixirc_paths: elixirc_paths(Mix.env()),
start_permanent: Mix.env() == :prod,
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 d12c0ba73..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,16 @@ 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)
@@ -324,8 +334,11 @@ 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,
@@ -335,6 +348,79 @@ defmodule Realtime.Extensions.CdcRlsTest do
}
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
defmodule Subscriber do
# Start CDC remotely
diff --git a/test/realtime/tenants/connect_test.exs b/test/realtime/tenants/connect_test.exs
index 741f6ecf7..804b3018f 100644
--- a/test/realtime/tenants/connect_test.exs
+++ b/test/realtime/tenants/connect_test.exs
@@ -328,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/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