fix: run migrations on tenants nearest region

Author: filipecabaco

lib/realtime/tenants.ex

@@ -98,13 +98,11 @@ defmodule Realtime.Tenants do
 
       connected_cluster when is_integer(connected_cluster) ->
         tenant = Cache.get_tenant_by_external_id(external_id)
-        {:ok, db_conn} = Database.connect(tenant, "realtime_health_check")
-        Process.alive?(db_conn) && GenServer.stop(db_conn)
-        Migrations.run_migrations(tenant)
+        result? = Migrations.run_migrations(tenant)
 
         {:ok,
          %{
-           healthy: true,
+           healthy: result? == :ok || result? == :noop,
            db_connected: false,
            connected_cluster: connected_cluster,
            region: region,
@@ -475,10 +473,11 @@ defmodule Realtime.Tenants do
   @doc """
   Checks if migrations for a given tenant need to run.
   """
-  @spec run_migrations?(Tenant.t()) :: boolean()
-  def run_migrations?(%Tenant{} = tenant) do
-    tenant.migrations_ran < Enum.count(Migrations.migrations())
-  end
+  @spec run_migrations?(Tenant.t() | integer()) :: boolean()
+  def run_migrations?(%Tenant{} = tenant), do: run_migrations?(tenant.migrations_ran)
+
+  def run_migrations?(migrations_ran) when is_integer(migrations_ran),
+    do: migrations_ran < Enum.count(Migrations.migrations())
 
   @doc """
   Broadcasts an operation event to the tenant's operations channel.

lib/realtime/tenants/migrations.ex

@@ -11,6 +11,8 @@ defmodule Realtime.Tenants.Migrations do
   alias Realtime.Repo
   alias Realtime.Api.Tenant
   alias Realtime.Api
+  alias Realtime.Nodes
+  alias Realtime.GenRpc
 
   alias Realtime.Tenants.Migrations.{
     CreateRealtimeSubscriptionTable,
@@ -148,7 +150,7 @@ defmodule Realtime.Tenants.Migrations do
     {20_251_103_001_201, BroadcastSendIncludePayloadId}
   ]
 
-  defstruct [:tenant_external_id, :settings]
+  defstruct [:tenant_external_id, :settings, migrations_ran: 0]
 
   @type t :: %__MODULE__{
           tenant_external_id: binary(),
@@ -161,14 +163,31 @@ defmodule Realtime.Tenants.Migrations do
   @spec run_migrations(Tenant.t()) :: :ok | :noop | {:error, any()}
   def run_migrations(%Tenant{} = tenant) do
     %{extensions: [%{settings: settings} | _]} = tenant
-    attrs = %__MODULE__{tenant_external_id: tenant.external_id, settings: settings}
 
+    attrs = %__MODULE__{
+      tenant_external_id: tenant.external_id,
+      settings: settings,
+      migrations_ran: tenant.migrations_ran
+    }
+
+    region = Tenants.region(tenant)
+
+    node =
+      case Nodes.node_from_region(region, tenant.external_id) do
+        {:ok, node} -> node
+        {:error, :not_available} -> node()
+      end
+
+    GenRpc.call(node, __MODULE__, :start_migration, [attrs], tenant_id: tenant.external_id)
+  end
+
+  def start_migration(attrs) do
     supervisor =
-      {:via, PartitionSupervisor, {Realtime.Tenants.Migrations.DynamicSupervisor, tenant.external_id}}
+      {:via, PartitionSupervisor, {Realtime.Tenants.Migrations.DynamicSupervisor, attrs.tenant_external_id}}
 
     spec = {__MODULE__, attrs}
 
-    if Tenants.run_migrations?(tenant) do
+    if Tenants.run_migrations?(attrs.migrations_ran) do
       case DynamicSupervisor.start_child(supervisor, spec) do
         :ignore -> :ok
         error -> error

mix.exs

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

test/integration/region_aware_migrations_test.exs

@@ -0,0 +1,61 @@
+defmodule Realtime.Integration.RegionAwareMigrationsTest do
+  use Realtime.DataCase, async: false
+  use Mimic
+
+  alias Containers
+  alias Realtime.Tenants
+  alias Realtime.Tenants.Migrations
+
+  setup do
+    {:ok, port} = Containers.checkout()
+
+    settings = [
+      %{
+        "type" => "postgres_cdc_rls",
+        "settings" => %{
+          "db_host" => "127.0.0.1",
+          "db_name" => "postgres",
+          "db_user" => "supabase_admin",
+          "db_password" => "postgres",
+          "db_port" => "#{port}",
+          "poll_interval" => 100,
+          "poll_max_changes" => 100,
+          "poll_max_record_bytes" => 1_048_576,
+          "region" => "eu-west-2",
+          "publication" => "supabase_realtime_test",
+          "ssl_enforced" => false
+        }
+      }
+    ]
+
+    tenant = tenant_fixture(%{extensions: settings})
+
+    {:ok, node} = Clustered.start(nil, extra_config: [{:realtime, :region, Tenants.region(tenant)}])
+
+    on_exit(fn -> Clustered.stop() end)
+
+    Process.sleep(1000)
+    %{tenant: tenant, node: node}
+  end
+
+  test "run_migrations routes to node in tenant's region with expected arguments", %{tenant: tenant, node: node} do
+    assert tenant.migrations_ran == 0
+
+    Realtime.GenRpc
+    |> Mimic.expect(:call, fn called_node, mod, func, args, opts ->
+      assert called_node == node
+      assert mod == Migrations
+      assert func == :start_migration
+      assert opts[:tenant_id] == tenant.external_id
+
+      arg = hd(args)
+      assert arg.tenant_external_id == tenant.external_id
+      assert arg.migrations_ran == tenant.migrations_ran
+      assert arg.settings == hd(tenant.extensions).settings
+
+      call_original(Realtime.GenRpc, :call, [node, mod, func, args, opts])
+    end)
+
+    assert :ok = Migrations.run_migrations(tenant)
+  end
+end

test/realtime_web/controllers/tenant_controller_test.exs

@@ -419,7 +419,7 @@ defmodule RealtimeWeb.TenantControllerTest do
 
       conn = get(conn, ~p"/api/tenants/#{tenant.external_id}/health")
       data = json_response(conn, 200)["data"]
-      Process.sleep(2000)
+      Process.sleep(4000)
 
       assert {:ok, %{rows: []}} = Postgrex.query(db_conn, "SELECT * FROM realtime.messages", [])