fix: Database.from_settings/from_tenant can error out (#1841)

Author: edgurgel

lib/extensions/postgres_cdc_rls/replication_poller.ex

@@ -60,9 +60,8 @@ defmodule Extensions.PostgresCdcRls.ReplicationPoller do
 
   @impl true
   def handle_continue({:connect, tenant}, state) do
-    realtime_rls_settings = Database.from_tenant(tenant, "realtime_rls")
-
-    with {:ok, conn} <- Database.connect_db(realtime_rls_settings) do
+    with {:ok, realtime_rls_settings} <- Database.from_tenant(tenant, "realtime_rls"),
+         {:ok, conn} <- Database.connect_db(realtime_rls_settings) do
       {:noreply, Map.put(state, :conn, conn), {:continue, :prepare}}
     else
       {:error, reason} ->

lib/extensions/postgres_cdc_rls/subscription_manager.ex

@@ -79,10 +79,10 @@ defmodule Extensions.PostgresCdcRls.SubscriptionManager do
     extension = Realtime.PostgresCdc.filter_settings("postgres_cdc_rls", tenant.extensions)
     extension = Map.merge(extension, %{"subs_pool_size" => Map.get(extension, "subcriber_pool_size", 4)})
 
-    subscription_manager_settings = Database.from_settings(extension, "realtime_subscription_manager")
-    subscription_manager_pub_settings = Database.from_settings(extension, "realtime_subscription_manager_pub")
-
-    with {:ok, conn} <- Database.connect_db(subscription_manager_settings),
+    with {:ok, subscription_manager_settings} <- Database.from_settings(extension, "realtime_subscription_manager"),
+         {:ok, subscription_manager_pub_settings} <-
+           Database.from_settings(extension, "realtime_subscription_manager_pub"),
+         {:ok, conn} <- Database.connect_db(subscription_manager_settings),
          {:ok, conn_pub} <- Database.connect_db(subscription_manager_pub_settings) do
       Subscriptions.delete_all_if_table_exists(conn)
 

lib/extensions/postgres_cdc_rls/subscriptions_checker.ex

@@ -56,9 +56,8 @@ defmodule Extensions.PostgresCdcRls.SubscriptionsChecker do
 
     %Realtime.Api.Tenant{} = tenant = Realtime.Tenants.Cache.get_tenant_by_external_id(id)
 
-    realtime_subscription_checker_settings = Database.from_tenant(tenant, "realtime_subscription_checker")
-
-    with {:ok, conn} <- Database.connect_db(realtime_subscription_checker_settings) do
+    with {:ok, realtime_subscription_checker_settings} <- Database.from_tenant(tenant, "realtime_subscription_checker"),
+         {:ok, conn} <- Database.connect_db(realtime_subscription_checker_settings) do
       state = %State{
         id: id,
         conn: conn,

lib/realtime/database.ex

@@ -44,7 +44,7 @@ defmodule Realtime.Database do
   @doc """
   Creates a database connection struct from the given tenant.
   """
-  @spec from_tenant(Tenant.t(), binary(), :stop | :exp | :rand | :rand_exp) :: t()
+  @spec from_tenant(Tenant.t(), binary(), :stop | :exp | :rand | :rand_exp) :: {:ok, t()} | {:error, :nxdomain}
   def from_tenant(%Tenant{} = tenant, application_name, backoff \\ :rand_exp) do
     tenant
     |> then(&Realtime.PostgresCdc.filter_settings(@cdc, &1.extensions))
@@ -54,7 +54,7 @@ defmodule Realtime.Database do
   @doc """
   Creates a database connection struct from the given settings.
   """
-  @spec from_settings(map(), binary(), :stop | :exp | :rand | :rand_exp) :: t()
+  @spec from_settings(map(), binary(), :stop | :exp | :rand | :rand_exp) :: {:ok, t()} | {:error, :nxdomain}
   def from_settings(settings, application_name, backoff \\ :rand_exp) do
     pool = pool_size_by_application_name(application_name, settings)
 
@@ -65,22 +65,24 @@ defmodule Realtime.Database do
       |> Map.new()
       |> then(&Map.merge(settings, &1))
 
-    {:ok, addrtype} = detect_ip_version(settings["db_host"])
-    ssl = if default_ssl_param(settings), do: [verify: :verify_none], else: false
-
-    %__MODULE__{
-      hostname: settings["db_host"],
-      port: String.to_integer(settings["db_port"]),
-      database: settings["db_name"],
-      username: settings["db_user"],
-      password: settings["db_password"],
-      pool_size: pool,
-      queue_target: settings["db_queue_target"] || 5_000,
-      application_name: application_name,
-      backoff_type: backoff,
-      socket_options: [addrtype],
-      ssl: ssl
-    }
+    with {:ok, addrtype} <- detect_ip_version(settings["db_host"]) do
+      ssl = if default_ssl_param(settings), do: [verify: :verify_none], else: false
+
+      {:ok,
+       %__MODULE__{
+         hostname: settings["db_host"],
+         port: String.to_integer(settings["db_port"]),
+         database: settings["db_name"],
+         username: settings["db_user"],
+         password: settings["db_password"],
+         pool_size: pool,
+         queue_target: settings["db_queue_target"] || 5_000,
+         application_name: application_name,
+         backoff_type: backoff,
+         socket_options: [addrtype],
+         ssl: ssl
+       }}
+    end
   end
 
   @available_connection_factor 0.95
@@ -97,10 +99,10 @@ defmodule Realtime.Database do
     |> then(&PostgresCdc.filter_settings(@cdc, &1.extensions))
     |> then(fn settings ->
       required_pool = tenant_pool_requirements(settings)
-      check_settings = from_settings(settings, "realtime_connect", :stop)
-      check_settings = Map.put(check_settings, :max_restarts, 0)
 
-      with {:ok, conn} <- connect_db(check_settings),
+      with {:ok, base_settings} <- from_settings(settings, "realtime_connect", :stop),
+           check_settings = %{base_settings | max_restarts: 0},
+           {:ok, conn} <- connect_db(check_settings),
            {:ok, [available_connections, migrations_ran]} <- query_connection_info(conn) do
         requirement = ceil(required_pool * @available_connection_factor)
 
@@ -154,9 +156,9 @@ defmodule Realtime.Database do
   @spec connect(Tenant.t(), binary(), :stop | :exp | :rand | :rand_exp) ::
           {:ok, pid()} | {:error, any()}
   def connect(tenant, application_name, backoff \\ :stop) do
-    tenant
-    |> from_tenant(application_name, backoff)
-    |> connect_db()
+    with {:ok, settings} <- from_tenant(tenant, application_name, backoff) do
+      connect_db(settings)
+    end
   end
 
   @doc """

lib/realtime/tenants/migrations.ex

@@ -226,35 +226,35 @@ defmodule Realtime.Tenants.Migrations do
   end
 
   defp migrate(settings) do
-    settings = Database.from_settings(settings, "realtime_migrations", :stop)
+    with {:ok, settings} <- Database.from_settings(settings, "realtime_migrations", :stop) do
+      [
+        hostname: settings.hostname,
+        port: settings.port,
+        database: settings.database,
+        password: settings.password,
+        username: settings.username,
+        pool_size: settings.pool_size,
+        backoff_type: settings.backoff_type,
+        socket_options: settings.socket_options,
+        parameters: [application_name: settings.application_name],
+        ssl: settings.ssl
+      ]
+      |> Repo.with_dynamic_repo(fn repo ->
+        Logger.info("Applying migrations to #{settings.hostname}")
 
-    [
-      hostname: settings.hostname,
-      port: settings.port,
-      database: settings.database,
-      password: settings.password,
-      username: settings.username,
-      pool_size: settings.pool_size,
-      backoff_type: settings.backoff_type,
-      socket_options: settings.socket_options,
-      parameters: [application_name: settings.application_name],
-      ssl: settings.ssl
-    ]
-    |> Repo.with_dynamic_repo(fn repo ->
-      Logger.info("Applying migrations to #{settings.hostname}")
+        try do
+          opts = [all: true, prefix: "realtime", dynamic_repo: repo]
+          {time, _} = :timer.tc(fn -> Ecto.Migrator.run(Repo, @migrations, :up, opts) end)
+          Logger.info("Finished applying tenant migrations in #{div(time, 1000)}ms")
 
-      try do
-        opts = [all: true, prefix: "realtime", dynamic_repo: repo]
-        {time, _} = :timer.tc(fn -> Ecto.Migrator.run(Repo, @migrations, :up, opts) end)
-        Logger.info("Finished applying tenant migrations in #{div(time, 1000)}ms")
-
-        :ok
-      rescue
-        error ->
-          log_error("MigrationsFailedToRun", error, migration_error_metadata(error))
-          {:error, error}
-      end
-    end)
+          :ok
+        rescue
+          error ->
+            log_error("MigrationsFailedToRun", error, migration_error_metadata(error))
+            {:error, error}
+        end
+      end)
+    end
   end
 
   defp migration_error_metadata(%Postgrex.Error{postgres: postgres}) when is_map(postgres) do

lib/realtime/tenants/replication_connection.ex

@@ -126,28 +126,29 @@ defmodule Realtime.Tenants.ReplicationConnection do
 
   def start_link(%__MODULE__{tenant_id: tenant_id} = attrs) do
     tenant = Cache.get_tenant_by_external_id(tenant_id)
-    connection_opts = Database.from_tenant(tenant, "realtime_broadcast_changes", :stop)
-
-    connection_opts =
-      [
-        name: {:via, Registry, {Realtime.Registry.Unique, {__MODULE__, tenant_id}}},
-        hostname: connection_opts.hostname,
-        username: connection_opts.username,
-        password: connection_opts.password,
-        database: connection_opts.database,
-        port: connection_opts.port,
-        socket_options: connection_opts.socket_options,
-        ssl: connection_opts.ssl,
-        sync_connect: true,
-        auto_reconnect: false,
-        parameters: [application_name: "realtime_replication_connection"]
-      ]
-
-    case Postgrex.ReplicationConnection.start_link(__MODULE__, attrs, connection_opts) do
-      {:ok, pid} -> {:ok, pid}
-      {:error, {:already_started, pid}} -> {:ok, pid}
-      {:error, {:bad_return_from_init, {:stop, error}}} -> {:error, error}
-      {:error, error} -> {:error, error}
+
+    with {:ok, db_settings} <- Database.from_tenant(tenant, "realtime_broadcast_changes", :stop) do
+      connection_opts =
+        [
+          name: {:via, Registry, {Realtime.Registry.Unique, {__MODULE__, tenant_id}}},
+          hostname: db_settings.hostname,
+          username: db_settings.username,
+          password: db_settings.password,
+          database: db_settings.database,
+          port: db_settings.port,
+          socket_options: db_settings.socket_options,
+          ssl: db_settings.ssl,
+          sync_connect: true,
+          auto_reconnect: false,
+          parameters: [application_name: "realtime_replication_connection"]
+        ]
+
+      case Postgrex.ReplicationConnection.start_link(__MODULE__, attrs, connection_opts) do
+        {:ok, pid} -> {:ok, pid}
+        {:error, {:already_started, pid}} -> {:ok, pid}
+        {:error, {:bad_return_from_init, {:stop, error}}} -> {:error, error}
+        {:error, error} -> {:error, error}
+      end
     end
   end
 

test/realtime/database_test.exs

@@ -265,15 +265,15 @@ defmodule Realtime.DatabaseTest do
 
   describe "from_tenant/3" do
     test "uses default backoff when not provided", %{tenant: tenant} do
-      settings = Database.from_tenant(tenant, "realtime_test")
+      {:ok, settings} = Database.from_tenant(tenant, "realtime_test")
       assert settings.backoff_type == :rand_exp
     end
   end
 
   describe "from_settings/3" do
     test "uses default backoff when not provided", %{tenant: tenant} do
       settings = Realtime.PostgresCdc.filter_settings("postgres_cdc_rls", tenant.extensions)
-      result = Database.from_settings(settings, "realtime_connect")
+      {:ok, result} = Database.from_settings(settings, "realtime_connect")
       assert result.backoff_type == :rand_exp
     end
 
@@ -283,7 +283,7 @@ defmodule Realtime.DatabaseTest do
       {:ok, ip_version} = Database.detect_ip_version("127.0.0.1")
       socket_options = [ip_version]
       settings = Realtime.PostgresCdc.filter_settings("postgres_cdc_rls", tenant.extensions)
-      settings = Database.from_settings(settings, application_name, backoff)
+      {:ok, settings} = Database.from_settings(settings, application_name, backoff)
       port = settings.port
 
       assert %Realtime.Database{
@@ -313,12 +313,12 @@ defmodule Realtime.DatabaseTest do
 
       settings = Realtime.PostgresCdc.filter_settings("postgres_cdc_rls", tenant.extensions)
       settings = Map.put(settings, "ssl_enforced", true)
-      settings = Database.from_settings(settings, application_name, backoff)
+      {:ok, settings} = Database.from_settings(settings, application_name, backoff)
       assert settings.ssl == [verify: :verify_none]
 
       settings = Realtime.PostgresCdc.filter_settings("postgres_cdc_rls", tenant.extensions)
       settings = Map.put(settings, "ssl_enforced", false)
-      settings = Database.from_settings(settings, application_name, backoff)
+      {:ok, settings} = Database.from_settings(settings, application_name, backoff)
       assert settings.ssl == false
     end
   end

test/realtime/extensions/cdc_rls/subscriptions_test.exs

@@ -11,9 +11,10 @@ defmodule Realtime.Extensions.PostgresCdcRls.SubscriptionsTest do
   setup do
     tenant = Containers.checkout_tenant(run_migrations: true)
 
+    {:ok, db_settings} = Database.from_tenant(tenant, "realtime_rls")
+
     {:ok, conn} =
-      tenant
-      |> Database.from_tenant("realtime_rls")
+      db_settings
       |> Map.from_struct()
       |> Keyword.new()
       |> Postgrex.start_link()
@@ -279,9 +280,10 @@ defmodule Realtime.Extensions.PostgresCdcRls.SubscriptionsTest do
       )
 
       on_exit(fn ->
+        {:ok, db_settings} = Database.from_tenant(tenant, "realtime_rls")
+
         {:ok, cleanup_conn} =
-          tenant
-          |> Database.from_tenant("realtime_rls")
+          db_settings
           |> Map.from_struct()
           |> Keyword.new()
           |> Postgrex.start_link()

test/realtime/tenants/connect_test.exs

@@ -569,7 +569,8 @@ defmodule Realtime.Tenants.ConnectTest do
     end
 
     test "handles max_wal_senders by logging the correct operational code", %{tenant: tenant} do
-      opts = tenant |> Database.from_tenant("realtime_test", :stop) |> Database.opts()
+      {:ok, settings} = Database.from_tenant(tenant, "realtime_test", :stop)
+      opts = Database.opts(settings)
       parent = self()
 
       pids =

test/realtime/tenants/replication_connection_test.exs

@@ -544,7 +544,8 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
     end
 
     test "handle standby connections exceeds max_wal_senders", %{tenant: tenant} do
-      opts = Database.from_tenant(tenant, "realtime_test", :stop) |> Database.opts()
+      {:ok, settings} = Database.from_tenant(tenant, "realtime_test", :stop)
+      opts = Database.opts(settings)
       parent = self()
 
       # This creates a loop of errors that occupies all WAL senders and lets us test the error handling