🔄 Sync with upstream changes (#11)

* fix: runtime setup error (supabase#1520)

* fix: use primary instead of replica on rename_settings_field (supabase#1521)

* feat: upgrade cowboy & ranch (supabase#1523)

* fix: Fix GenRpc to not try to connect to nodes that are not alive (supabase#1525)

* fix: enable presence on track message (supabase#1527)

currently the user would need to have enabled from the beginning of the channel. this will enable users to enable presence later in the flow by sending a track message which will enable presence messages for them

* fix: set cowboy active_n=100 as cowboy 2.12.0 (supabase#1530)

cowboy 2.13.0 set the default active_n=1

* fix: provide error_code metadata on RealtimeChannel.Logging (supabase#1531)

* feat: disable UTF8 validation on websocket frames (supabase#1532)

Currently all text frames as handled only with JSON which already requires UTF-8

* fix: move DB setup to happen after Connect.init (supabase#1533)

This change reduces the impact of slow DB setup impacting other tenants
trying to connect at the same time that landed on the same partition

* fix: handle wal bloat (supabase#1528)

Verify that replication connection is able to reconnect when faced with WAL bloat issues

* feat: replay realtime.messages (supabase#1526)

A new index was created on inserted_at DESC, topic WHERE private IS TRUE AND extension = "broadast"

The hardcoded limit is 25 for now.

* feat: gen_rpc pub sub adapter (supabase#1529)

Add a PubSub adapter that uses gen_rpc to send messages to other nodes.

It uses :gen_rpc.abcast/3 instead of :erlang.send/2

The adapter works very similarly to the PG2 adapter. It consists of
multiple workers that forward to the local node using PubSub.local_broadcast.

The way to choose the worker to be used is based on the sending process
just like PG2 adapter does

The number of workers is controlled by `:pool_size` or `:broadcast_pool_size`.
This distinction exists because Phoenix.PubSub uses `:pool_size` to
define how many partitions the PubSub registry will use. It's possible
to control them separately by using `:broadcast_pool_size`

* fix: ensure message id doesn't raise on non-map payloads (supabase#1534)

* fix: match error on Connect (supabase#1536)



---------

Co-authored-by: Eduardo Gurgel Pinho <eduardo.gurgel@supabase.io>

* feat: websocket max heap size configuration (supabase#1538)

* fix: set max process heap size to 500MB instead of 8GB
* feat: set websocket transport max heap size

WEBSOCKET_MAX_HEAP_SIZE can be used to configure it

* fix: update gen_rpc to fix gen_rpc_dispatcher issues (supabase#1537)

Issues:

* Single gen_rpc_dispatcher that can be a bottleneck if the connecting takes some time
* Many calls can land on the dispatcher but the node might be gone already. If we don't validate the node it might keep trying to connect until it times out instead of quickly giving up due to not being an actively connected node.

* fix: improve ErlSysMon logging for processes (supabase#1540)

Include initial_call, ancestors, registered_name, message_queue_len and total_heap_size

Also bump long_schedule and long_gc

* fix: make pubsub adapter configurable (supabase#1539)

* fix: specify that only private channels are allowed when replaying (supabase#1543)

messages

* fix: rate limit connect module (supabase#1541)

On bad connection, we rate limit the Connect module so we prevent abuses and too much logging of errors

---------

Co-authored-by: Filipe Cabaço <filipe@supabase.io>
Co-authored-by: Eduardo Gurgel <eduardo.gurgel@supabase.io>
Co-authored-by: Bradley Haljendi <5642609+Fudster@users.noreply.github.com>

Author: 4 people

README.md

@@ -243,6 +243,7 @@ This is the list of operational codes that can help you understand your deployme
 | ChannelRateLimitReached            | The number of channels you can create has reached its limit                                                                                                                                           |
 | ConnectionRateLimitReached         | The number of connected clients as reached its limit                                                                                                                                                  |
 | ClientJoinRateLimitReached         | The rate of joins per second from your clients has reached the channel limits                                                                                                                         |
+| DatabaseConnectionRateLimitReached | The rate of attempts to connect to tenants database has reached the limit                                                                                                                             |
 | MessagePerSecondRateLimitReached   | The rate of messages per second from your clients has reached the channel limits                                                                                                                      |
 | RealtimeDisabledForTenant          | Realtime has been disabled for the tenant                                                                                                                                                             |
 | UnableToConnectToTenantDatabase    | Realtime was not able to connect to the tenant's database                                                                                                                                             |

lib/realtime/tenants.ex

@@ -328,6 +328,32 @@ defmodule Realtime.Tenants do
     %RateCounter.Args{id: {:channel, :authorization_errors, external_id}, opts: opts}
   end
 
+  @connect_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)
+  end
+
+  def connect_per_second_rate(tenant_id) do
+    opts = [
+      max_bucket_len: 10,
+      limit: [
+        value: @connect_per_second_default,
+        measurement: :sum,
+        log_fn: fn ->
+          Logger.critical(
+            "DatabaseConnectionRateLimitReached: Too many connection attempts against the tenant database",
+            external_id: tenant_id,
+            project: tenant_id
+          )
+        end
+      ]
+    ]
+
+    %RateCounter.Args{id: {:database, :connect, tenant_id}, opts: opts}
+  end
+
   defp pool_size(%{extensions: [%{settings: settings} | _]}) do
     Database.pool_size_by_application_name("realtime_connect", settings)
   end

lib/realtime/tenants/connect.ex

@@ -11,15 +11,17 @@ defmodule Realtime.Tenants.Connect do
 
   use Realtime.Logs
 
-  alias Realtime.Tenants.Rebalancer
   alias Realtime.Api.Tenant
+  alias Realtime.GenCounter
+  alias Realtime.RateCounter
   alias Realtime.Rpc
   alias Realtime.Tenants
   alias Realtime.Tenants.Connect.CheckConnection
   alias Realtime.Tenants.Connect.GetTenant
   alias Realtime.Tenants.Connect.Piper
   alias Realtime.Tenants.Connect.RegisterProcess
   alias Realtime.Tenants.Migrations
+  alias Realtime.Tenants.Rebalancer
   alias Realtime.Tenants.ReplicationConnection
   alias Realtime.UsersCounter
 
@@ -39,11 +41,8 @@ defmodule Realtime.Tenants.Connect do
   @doc "Check if Connect has finished setting up connections"
   def ready?(tenant_id) do
     case whereis(tenant_id) do
-      pid when is_pid(pid) ->
-        GenServer.call(pid, :ready?)
-
-      _ ->
-        false
+      pid when is_pid(pid) -> GenServer.call(pid, :ready?)
+      _ -> false
     end
   end
 
@@ -55,24 +54,29 @@ defmodule Realtime.Tenants.Connect do
           | {:error, :tenant_database_unavailable}
           | {:error, :initializing}
           | {:error, :tenant_database_connection_initializing}
-          | {:error, :tenant_db_too_many_connections}
+          | {:error, :connect_rate_limit_reached}
           | {:error, :rpc_error, term()}
   def lookup_or_start_connection(tenant_id, opts \\ []) when is_binary(tenant_id) do
-    case get_status(tenant_id) do
-      {:ok, conn} ->
-        {:ok, conn}
+    rate_args = Tenants.connect_per_second_rate(tenant_id)
+    RateCounter.new(rate_args)
 
-      {:error, :tenant_database_unavailable} ->
-        {:error, :tenant_database_unavailable}
+    with {:ok, %{limit: %{triggered: false}}} <- RateCounter.get(rate_args),
+         {:ok, conn} <- get_status(tenant_id) do
+      {:ok, conn}
+    else
+      {:ok, %{limit: %{triggered: true}}} ->
+        {:error, :connect_rate_limit_reached}
 
       {:error, :tenant_database_connection_initializing} ->
+        GenCounter.add(rate_args.id)
         call_external_node(tenant_id, opts)
 
       {:error, :initializing} ->
         {:error, :tenant_database_unavailable}
 
-      {:error, :tenant_db_too_many_connections} ->
-        {:error, :tenant_db_too_many_connections}
+      {:error, reason} ->
+        GenCounter.add(rate_args.id)
+        {:error, reason}
     end
   end
 

lib/realtime_web/channels/realtime_channel.ex

@@ -167,6 +167,10 @@ defmodule RealtimeWeb.RealtimeChannel do
         msg = "Database can't accept more connections, Realtime won't connect"
         log_error(socket, "DatabaseLackOfConnections", msg)
 
+      {:error, :connect_rate_limit_reached} ->
+        msg = "Too many database connections attempts per second"
+        log_error(socket, "DatabaseConnectionRateLimitReached", msg)
+
       {:error, :unable_to_set_policies, error} ->
         log_error(socket, "UnableToSetPolicies", error)
         {:error, %{reason: "Realtime was unable to connect to the project database"}}
@@ -213,6 +217,9 @@ defmodule RealtimeWeb.RealtimeChannel do
       {:error, :invalid_replay_params} ->
         log_error(socket, "UnableToReplayMessages", "Replay params are not valid")
 
+      {:error, :invalid_replay_channel} ->
+        log_error(socket, "UnableToReplayMessages", "Replay is not allowed for public channels")
+
       {:error, error} ->
         log_error(socket, "UnknownErrorOnChannel", error)
         {:error, %{reason: "Unknown Error on Channel"}}
@@ -790,7 +797,7 @@ defmodule RealtimeWeb.RealtimeChannel do
   end
 
   defp maybe_replay_messages(%{"broadcast" => %{"replay" => _}}, _sub_topic, _db_conn, false = _private?) do
-    {:error, :invalid_replay_params}
+    {:error, :invalid_replay_channel}
   end
 
   defp maybe_replay_messages(%{"broadcast" => %{"replay" => replay_params}}, sub_topic, db_conn, true = _private?)

mix.exs

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

test/realtime/tenants/connect_test.exs

@@ -515,6 +515,53 @@ defmodule Realtime.Tenants.ConnectTest do
       assert capture_log(fn -> assert {:error, :rpc_error, _} = Connect.lookup_or_start_connection("tenant") end) =~
                "project=tenant external_id=tenant [error] ErrorOnRpcCall"
     end
+
+    test "rate limit connect when too many connections against bad database", %{tenant: tenant} do
+      extension = %{
+        "type" => "postgres_cdc_rls",
+        "settings" => %{
+          "db_host" => "127.0.0.1",
+          "db_name" => "postgres",
+          "db_user" => "supabase_admin",
+          "db_password" => "postgres",
+          "poll_interval" => 100,
+          "poll_max_changes" => 100,
+          "poll_max_record_bytes" => 1_048_576,
+          "region" => "us-east-1",
+          "ssl_enforced" => true
+        }
+      }
+
+      {:ok, tenant} = update_extension(tenant, extension)
+
+      log =
+        capture_log(fn ->
+          res =
+            for _ <- 1..50 do
+              Process.sleep(200)
+              Connect.lookup_or_start_connection(tenant.external_id)
+            end
+
+          assert Enum.any?(res, fn {_, res} -> res == :connect_rate_limit_reached end)
+        end)
+
+      assert log =~ "DatabaseConnectionRateLimitReached: Too many connection attempts against the tenant database"
+    end
+
+    test "rate limit connect will not trigger if connection is successful", %{tenant: tenant} do
+      log =
+        capture_log(fn ->
+          res =
+            for _ <- 1..20 do
+              Process.sleep(500)
+              Connect.lookup_or_start_connection(tenant.external_id)
+            end
+
+          refute Enum.any?(res, fn {_, res} -> res == :tenant_db_too_many_connections end)
+        end)
+
+      refute log =~ "DatabaseConnectionRateLimitReached: Too many connection attempts against the tenant database"
+    end
   end
 
   describe "shutdown/1" do

test/realtime_web/channels/realtime_channel_test.exs

@@ -153,7 +153,7 @@ defmodule RealtimeWeb.RealtimeChannelTest do
 
       assert {
                :error,
-               %{reason: "UnableToReplayMessages: Replay params are not valid"}
+               %{reason: "UnableToReplayMessages: Replay is not allowed for public channels"}
              } = subscribe_and_join(socket, "realtime:test", %{"config" => config})
 
       refute_receive _any

test/realtime_web/controllers/broadcast_controller_test.exs

@@ -272,6 +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)
 
       messages_to_send =
@@ -290,7 +291,10 @@ defmodule RealtimeWeb.BroadcastControllerTest do
 
       GenCounter
       |> expect(:add, fn ^request_events_key -> :ok end)
-      |> expect(:add, length(messages), fn ^broadcast_events_key -> :ok end)
+      |> expect(:add, length(messages), fn
+        ^broadcast_events_key -> :ok
+        ^connect_events_key -> :ok
+      end)
 
       conn = post(conn, Routes.broadcast_path(conn, :broadcast), %{"messages" => messages})
 
@@ -326,6 +330,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)
 
       channels =
@@ -354,7 +359,10 @@ defmodule RealtimeWeb.BroadcastControllerTest do
 
       GenCounter
       |> expect(:add, fn ^request_events_key -> :ok end)
-      |> expect(:add, length(messages), fn ^broadcast_events_key -> :ok end)
+      |> expect(:add, length(messages), fn
+        ^broadcast_events_key -> :ok
+        ^connect_events_key -> :ok
+      end)
 
       conn = post(conn, Routes.broadcast_path(conn, :broadcast), %{"messages" => messages})
 
@@ -408,6 +416,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)
 
       messages_to_send =
@@ -428,7 +437,9 @@ defmodule RealtimeWeb.BroadcastControllerTest do
 
       GenCounter
       |> expect(:add, fn ^request_events_key -> :ok end)
-      |> expect(:add, length(messages_to_send), fn ^broadcast_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})
 
@@ -482,7 +493,6 @@ defmodule RealtimeWeb.BroadcastControllerTest do
 
       GenCounter
       |> expect(:add, fn ^request_events_key -> 1 end)
-      |> reject(:add, 1)
 
       conn = post(conn, Routes.broadcast_path(conn, :broadcast), %{"messages" => messages})