more test fixing

Author: filipecabaco

config/runtime.exs

@@ -18,7 +18,8 @@ migration_partition_slots =
 connect_partition_slots =
   System.get_env("CONNECT_PARTITION_SLOTS", "#{System.schedulers_online() * 2}") |> String.to_integer()
 
-connect_throttle_limit = System.get_env("CONNECT_THROTTLE_LIMIT", "10") |> String.to_integer()
+connect_throttle_limit_per_second =
+  System.get_env("CONNECT_THROTTLE_LIMIT_per_second_PER_SECOND", "1") |> String.to_integer()
 
 if !(db_version in [nil, "ipv6", "ipv4"]),
   do: raise("Invalid IP version, please set either ipv6 or ipv4")
@@ -41,7 +42,7 @@ socket_options =
 config :realtime,
   migration_partition_slots: migration_partition_slots,
   connect_partition_slots: connect_partition_slots,
-  connect_throttle_limit: connect_throttle_limit,
+  connect_throttle_limit_per_second: connect_throttle_limit_per_second,
   tenant_max_bytes_per_second: System.get_env("TENANT_MAX_BYTES_PER_SECOND", "100000") |> String.to_integer(),
   tenant_max_channels_per_client: System.get_env("TENANT_MAX_CHANNELS_PER_CLIENT", "100") |> String.to_integer(),
   tenant_max_concurrent_users: System.get_env("TENANT_MAX_CONCURRENT_USERS", "200") |> String.to_integer(),

lib/realtime/tenants/connect.ex

@@ -307,7 +307,7 @@ defmodule Realtime.Tenants.Connect do
         %{db_conn_reference: db_conn_reference} = state
       ) do
     Logger.warning("Database connection has been terminated")
-    {:stop, :normal, state}
+    {:stop, :shutdown, state}
   end
 
   # Handle replication connection termination
@@ -316,7 +316,7 @@ defmodule Realtime.Tenants.Connect do
         %{replication_connection_reference: replication_connection_reference} = state
       ) do
     Logger.warning("Replication connection has died")
-    {:stop, :normal, state}
+    {:stop, :shutdown, state}
   end
 
   #  Handle listen connection termination
@@ -325,7 +325,7 @@ defmodule Realtime.Tenants.Connect do
         %{listen_reference: listen_reference} = state
       ) do
     Logger.warning("Listen has been terminated")
-    {:stop, :normal, state}
+    {:stop, :shutdown, state}
   end
 
   # Ignore messages to avoid handle_info unmatched functions

lib/realtime/tenants/connect/backoff.ex

@@ -10,11 +10,10 @@ defmodule Realtime.Tenants.Connect.Backoff do
   @impl Realtime.Tenants.Connect.Piper
   def run(acc) do
     %{tenant_id: tenant_id} = acc
-    connect_throttle_limit = Application.fetch_env!(:realtime, :connect_throttle_limit)
+    connect_throttle_limit_per_second = Application.fetch_env!(:realtime, :connect_throttle_limit_per_second)
 
     with {:ok, counter} <- start_connects_per_second_counter(tenant_id),
-         {:ok, %{avg: avg}} when avg < connect_throttle_limit <- RateCounter.get(counter) do
-      IO.inspect(avg)
+         {:ok, %{avg: avg}} when avg <= connect_throttle_limit_per_second <- RateCounter.get(counter) do
       GenCounter.add(counter)
       {:ok, acc}
     else
@@ -24,14 +23,16 @@ defmodule Realtime.Tenants.Connect.Backoff do
 
   defp start_connects_per_second_counter(tenant_id) do
     id = Tenants.connection_attempts_per_second_key(tenant_id)
-    GenCounter.new(id)
 
-    res = RateCounter.new(id, idle_shutdown: :infinity, tick: 200)
+    case RateCounter.get(id) do
+      {:ok, _} ->
+        :ok
 
-    case res do
-      {:ok, _} -> {:ok, id}
-      {:error, {:already_started, _}} -> {:ok, id}
-      {:error, reason} -> {:error, reason}
+      {:error, _} ->
+        GenCounter.new(id)
+        RateCounter.new(id, idle_shutdown: :infinity, tick: 100, idle_shutdown_ms: :timer.minutes(5))
     end
+
+    {:ok, id}
   end
 end

test/realtime/monitoring/prom_ex/plugins/tenants_test.exs

@@ -2,7 +2,6 @@ defmodule Realtime.PromEx.Plugins.TenantsTest do
   use Realtime.DataCase, async: false
 
   alias Realtime.PromEx.Plugins.Tenants
-  alias Realtime.Rpc
   alias Realtime.Tenants.Connect
 
   defmodule MetricsTest do
@@ -16,23 +15,14 @@ defmodule Realtime.PromEx.Plugins.TenantsTest do
   describe "pooling metrics" do
     setup do
       start_supervised!(MetricsTest)
-      local_tenant = Containers.checkout_tenant()
-      remote_tenant = Containers.checkout_tenant()
-
-      on_exit(fn ->
-        Containers.checkin_tenant(local_tenant)
-        Containers.checkin_tenant(remote_tenant)
-      end)
-
-      {:ok, node} = Clustered.start()
-      {:ok, _} = Rpc.enhanced_call(node, Connect, :lookup_or_start_connection, [remote_tenant.external_id])
+      local_tenant = Containers.checkout_tenant(true)
+      on_exit(fn -> Containers.checkin_tenant(local_tenant) end)
       {:ok, _} = Connect.lookup_or_start_connection(local_tenant.external_id)
-
-      %{local_tenant: local_tenant, remote_tenant: remote_tenant}
+      :ok
     end
 
     test "conneted based on Connect module information for local node only" do
-      Process.sleep(100 * 2)
+      Process.sleep(2000)
       assert PromEx.get_metrics(MetricsTest) =~ "realtime_tenants_connected 1"
     end
   end

test/realtime/tenants/connect/backoff_test.exs

@@ -4,8 +4,7 @@ defmodule Realtime.Tenants.Connect.BackoffTest do
 
   setup do
     tenant_id = random_string()
-    acc = %{tenant_id: tenant_id}
-    {:ok, acc: acc}
+    {:ok, acc: %{tenant_id: tenant_id}}
   end
 
   test "does not apply backoff for a given tenant if never called", %{acc: acc} do
@@ -14,19 +13,19 @@ defmodule Realtime.Tenants.Connect.BackoffTest do
 
   test "applies backoff if the user as called more than once during the configured space", %{acc: acc} do
     # emulate calls
-    for _ <- 1..100 do
+    for _ <- 1..10 do
       Backoff.run(acc)
-      Process.sleep(5)
+      Process.sleep(10)
     end
 
     assert {:error, :tenant_create_backoff} = Backoff.run(acc)
   end
 
   test "resets backoff after the configured space", %{acc: acc} do
     # emulate calls
-    for _ <- 1..100 do
+    for _ <- 1..10 do
       Backoff.run(acc)
-      Process.sleep(5)
+      Process.sleep(10)
     end
 
     # emulate block

test/realtime/tenants/connect_test.exs

@@ -189,7 +189,7 @@ defmodule Realtime.Tenants.ConnectTest do
       log =
         capture_log(fn ->
           Realtime.Tenants.suspend_tenant_by_external_id(tenant2.external_id)
-          Process.sleep(100)
+          Process.sleep(50)
         end)
 
       refute log =~ "Tenant was suspended"
@@ -377,10 +377,10 @@ defmodule Realtime.Tenants.ConnectTest do
     test "respects backoff pipe", %{tenant: tenant} do
       log =
         capture_log(fn ->
-          for _ <- 1..100 do
+          for _ <- 1..10 do
             Connect.connect(tenant.external_id)
+            Process.sleep(10)
             Connect.shutdown(tenant.external_id)
-            Process.sleep(5)
           end
 
           assert {:error, :tenant_create_backoff} = Connect.connect(tenant.external_id)
@@ -390,10 +390,10 @@ defmodule Realtime.Tenants.ConnectTest do
     end
 
     test "after timer, is able to connect", %{tenant: tenant} do
-      for _ <- 1..100 do
+      for _ <- 1..10 do
         Connect.connect(tenant.external_id)
+        Process.sleep(10)
         Connect.shutdown(tenant.external_id)
-        Process.sleep(5)
       end
 
       assert {:error, :tenant_create_backoff} = Connect.connect(tenant.external_id)

test/realtime/tenants/replication_connection_test.exs

@@ -1,4 +1,5 @@
 defmodule Realtime.Tenants.ReplicationConnectionTest do
+  alias Realtime.RateCounter
   use Realtime.DataCase, async: false
 
   alias Realtime.Api.Message
@@ -13,6 +14,12 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
 
     tenant = Containers.checkout_tenant(true)
 
+    {:ok, db_conn} = Database.connect(tenant, "realtime_test", :stop)
+    name = "supabase_realtime_messages_replication_slot_test"
+    Postgrex.query(db_conn, "SELECT pg_drop_replication_slot($1)", [name])
+    Process.exit(db_conn, :normal)
+    tenant |> Tenants.limiter_keys() |> Enum.each(&RateCounter.new(&1))
+
     on_exit(fn ->
       Application.put_env(:realtime, :slot_name_suffix, slot)
       Containers.checkin_tenant(tenant)
@@ -117,7 +124,8 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
 
   test "fails on existing replication slot", %{tenant: tenant} do
     {:ok, db_conn} = Database.connect(tenant, "realtime_test", :stop)
-    name = "supabase_realtime_messages_replication_slot_"
+    name = "supabase_realtime_messages_replication_slot_test"
+
     Postgrex.query!(db_conn, "SELECT pg_create_logical_replication_slot($1, 'test_decoding')", [name])
 
     assert {:error, "Temporary Replication slot already exists and in use"} =

test/realtime_web/controllers/tenant_controller_test.exs

@@ -99,22 +99,15 @@ defmodule RealtimeWeb.TenantControllerTest do
   end
 
   describe "upsert with post" do
-    test "renders tenant when data is valid", %{conn: conn} do
-      external_id = random_string()
-
-      port =
-        5500..9000
-        |> Enum.reject(&(&1 in Enum.map(:ets.tab2list(:test_ports), fn {port} -> port end)))
-        |> Enum.random()
+    setup [:with_tenant]
 
+    test "renders tenant when data is valid", %{conn: conn, tenant: tenant} do
+      external_id = tenant.external_id
+      port = Database.from_tenant(tenant, "realtime_test", :stop).port
       attrs = default_tenant_attrs(port)
       attrs = Map.put(attrs, "external_id", external_id)
-
-      Containers.initialize_no_tenant(external_id, port)
-      on_exit(fn -> Containers.stop_container(external_id) end)
-
       conn = post(conn, ~p"/api/tenants", tenant: attrs)
-      assert %{"id" => _id, "external_id" => ^external_id} = json_response(conn, 201)["data"]
+      assert %{"id" => _id, "external_id" => ^external_id} = json_response(conn, 200)["data"]
 
       conn = get(conn, Routes.tenant_path(conn, :show, external_id))
       assert ^external_id = json_response(conn, 200)["data"]["external_id"]
@@ -166,23 +159,13 @@ defmodule RealtimeWeb.TenantControllerTest do
   describe "upsert with put" do
     setup [:with_tenant]
 
-    test "renders tenant when data is valid", %{conn: conn} do
-      external_id = random_string()
-
-      port =
-        5500..9000
-        |> Enum.reject(&(&1 in Enum.map(:ets.tab2list(:test_ports), fn {port} -> port end)))
-        |> Enum.random()
-
-      :ets.insert(:test_ports, {port})
-
+    test "renders tenant when data is valid", %{tenant: tenant, conn: conn} do
+      external_id = tenant.external_id
+      port = Database.from_tenant(tenant, "realtime_test", :stop).port
       attrs = default_tenant_attrs(port)
 
-      Containers.initialize_no_tenant(external_id, port)
-      on_exit(fn -> Containers.stop_container(external_id) end)
-
       conn = put(conn, ~p"/api/tenants/#{external_id}", tenant: attrs)
-      assert %{"id" => _id, "external_id" => ^external_id} = json_response(conn, 201)["data"]
+      assert %{"id" => _id, "external_id" => ^external_id} = json_response(conn, 200)["data"]
 
       conn = get(conn, Routes.tenant_path(conn, :show, external_id))
       assert ^external_id = json_response(conn, 200)["data"]["external_id"]

test/support/containers.ex

@@ -90,55 +90,48 @@ defmodule Containers do
     tenant = Enum.random(tenants)
     :ets.insert(:containers, {tenant.external_id, %{tenant: tenant, using?: true}})
 
-    settings = Database.from_tenant(tenant, "realtime_test", :stop)
-    settings = %{settings | max_restarts: 0, ssl: false}
-    {:ok, conn} = Database.connect_db(settings)
+    capture_log(fn ->
+      settings = Database.from_tenant(tenant, "realtime_test", :stop)
+      settings = %{settings | max_restarts: 0, ssl: false}
+      {:ok, conn} = Database.connect_db(settings)
 
-    Postgrex.transaction(conn, fn db_conn ->
-      pid = Connect.whereis(tenant.external_id)
-      if pid && Process.alive?(pid), do: Connect.shutdown(tenant.external_id)
+      Postgrex.transaction(conn, fn db_conn ->
+        Postgrex.query!(
+          db_conn,
+          "SELECT pg_terminate_backend(pid) from pg_stat_activity where application_name like 'realtime_%' and application_name != 'realtime_test'",
+          []
+        )
 
-      tenant
-      |> Tenants.limiter_keys()
-      |> Enum.each(fn key ->
         RateCounter.stop(tenant.external_id)
         GenCounter.stop(tenant.external_id)
-        RateCounter.new(key)
-        GenCounter.new(key)
-      end)
 
-      Postgrex.query!(
-        db_conn,
-        "SELECT pg_terminate_backend(pid) from pg_stat_activity where application_name like 'realtime_%' and application_name != 'realtime_test'",
-        []
-      )
+        Postgrex.query!(db_conn, "DROP SCHEMA realtime CASCADE", [])
+        Postgrex.query!(db_conn, "CREATE SCHEMA realtime", [])
 
-      Postgrex.query!(db_conn, "DROP SCHEMA realtime CASCADE", [])
-      Postgrex.query!(db_conn, "CREATE SCHEMA realtime", [])
+        if Tenants.get_tenant_by_external_id(tenant.external_id) do
+          Tenants.update_migrations_ran(tenant.external_id, 0)
+        end
 
-      if Tenants.get_tenant_by_external_id(tenant.external_id) do
-        Tenants.update_migrations_ran(tenant.external_id, 0)
+        :ok
+      end)
+
+      if run_migrations? do
+        Migrations.run_migrations(tenant)
+        {:ok, pid} = Database.connect(tenant, "realtime_test", :stop)
+        :ok = Migrations.create_partitions(pid)
       end
 
-      :ok
+      Process.sleep(1000)
     end)
 
-    if run_migrations? do
-      Migrations.run_migrations(tenant)
-      {:ok, pid} = Database.connect(tenant, "realtime_test", :stop)
-      :ok = Migrations.create_partitions(pid)
-    end
-
-    Process.sleep(1000)
-
     tenant
   end
 
   def checkin_tenant(tenant) do
     :ets.insert(:containers, {tenant.external_id, %{tenant: tenant, using?: false}})
   end
 
-  @spec stop_container(any()) :: {any(), non_neg_integer()}
+  @spec stop_container(Tenant.t() | binary()) :: {any(), non_neg_integer()}
   def stop_container(%Tenant{} = tenant) do
     :ets.delete(:containers, tenant.external_id)
     pid = Connect.whereis(tenant.external_id)