move to use DBConnection.Backoff

Author: filipecabaco

README.md

@@ -259,6 +259,7 @@ This is the list of operational codes that can help you understand your deployme
 | RealtimeNodeDisconnected           | Realtime is a distributed application and this means that one the system is unable to communicate with one of the distributed nodes                                                                   |
 | MigrationsFailedToRun              | Error when running the migrations against the Tenant database that are required by Realtime                                                                                                           |
 | StartReplicationFailed             | Error when starting the replication and listening of errors for database broadcasting                                                                                                                 |
+| ReplicationConnectionTimeout       | Replication connection timed out during initialization                                                                                                                                                |
 | ReplicationMaxWalSendersReached    | Maximum number of WAL senders reached in tenant database, check how to increase this value in this [link](https://supabase.com/docs/guides/database/custom-postgres-config#cli-configurable-settings) |
 | MigrationCheckFailed               | Check to see if we require to run migrations fails                                                                                                                                                    |
 | PartitionCreationFailed            | Error when creating partitions for realtime.messages                                                                                                                                                  |

lib/realtime/tenants/connect.ex

@@ -24,17 +24,29 @@ defmodule Realtime.Tenants.Connect do
   alias Realtime.Tenants.Rebalancer
   alias Realtime.Tenants.ReplicationConnection
   alias Realtime.UsersCounter
+  alias DBConnection.Backoff
 
   @rpc_timeout_default 30_000
   @check_connected_user_interval_default 50_000
   @connected_users_bucket_shutdown [0, 0, 0, 0, 0, 0]
+  @type t :: %__MODULE__{
+          tenant_id: binary(),
+          db_conn_reference: reference(),
+          db_conn_pid: pid(),
+          replication_connection_pid: pid(),
+          replication_connection_reference: reference(),
+          backoff: Backoff.t(),
+          check_connected_user_interval: non_neg_integer(),
+          connected_users_bucket: list(non_neg_integer()),
+          check_connect_region_interval: non_neg_integer()
+        }
 
   defstruct tenant_id: nil,
             db_conn_reference: nil,
             db_conn_pid: nil,
             replication_connection_pid: nil,
             replication_connection_reference: nil,
-            replication_recovery_attempts: 0,
+            backoff: nil,
             check_connected_user_interval: nil,
             connected_users_bucket: [1],
             check_connect_region_interval: nil
@@ -216,7 +228,8 @@ defmodule Realtime.Tenants.Connect do
     state = %__MODULE__{
       tenant_id: tenant_id,
       check_connected_user_interval: check_connected_user_interval,
-      check_connect_region_interval: check_connect_region_interval
+      check_connect_region_interval: check_connect_region_interval,
+      backoff: Backoff.new(min: :timer.seconds(1), max: :timer.seconds(15), type: :rand_exp)
     }
 
     opts = Keyword.put(opts, :name, {:via, :syn, name})
@@ -351,40 +364,40 @@ defmodule Realtime.Tenants.Connect do
     {:stop, :shutdown, state}
   end
 
-  @replication_recovery_backoff_min :timer.seconds(1)
-
   # Handle replication connection termination
   def handle_info(
         {:DOWN, replication_connection_reference, _, _, _},
         %{replication_connection_reference: replication_connection_reference} = state
       ) do
+    %{backoff: backoff} = state
     log_warning("ReplicationConnectionDown", "Replication connection has been terminated")
-    backoff = calculate_recovery_backoff(state.replication_recovery_attempts)
-    Process.send_after(self(), :recover_replication_connection, backoff)
-    state = %{state | replication_connection_pid: nil, replication_connection_reference: nil}
+    {timeout, backoff} = Backoff.backoff(backoff)
+    Process.send_after(self(), :recover_replication_connection, timeout)
+    state = %{state | replication_connection_pid: nil, replication_connection_reference: nil, backoff: backoff}
     {:noreply, state}
   end
 
   @replication_connection_query "SELECT 1 from pg_stat_activity where application_name='realtime_replication_connection'"
   def handle_info(:recover_replication_connection, state) do
-    with %{num_rows: 0} <- Postgrex.query!(state.db_conn_pid, @replication_connection_query, []),
+    %{backoff: backoff, db_conn_pid: db_conn_pid} = state
+
+    with %{num_rows: 0} <- Postgrex.query!(db_conn_pid, @replication_connection_query, []),
          {:ok, state} <- start_replication_connection(state) do
-      {:noreply, %{state | replication_recovery_attempts: 0}}
+      {:noreply, %{state | backoff: Backoff.reset(backoff)}}
     else
-      %{num_rows: 0} ->
-        {:noreply, %{state | replication_recovery_attempts: 0}}
+      %{num_rows: _} ->
+        {:noreply, %{state | backoff: Backoff.reset(backoff)}}
 
       {:error, error} ->
-        attempts = state.replication_recovery_attempts + 1
-        backoff = calculate_recovery_backoff(attempts)
+        {timeout, backoff} = Backoff.backoff(backoff)
 
         log_error(
           "ReplicationConnectionRecoveryFailed",
-          "Replication connection recovery failed, attempt #{attempts}, next retry in #{backoff}ms : #{inspect(error)}"
+          "Replication connection recovery failed, next retry in #{timeout}ms : #{inspect(error)}"
         )
 
-        Process.send_after(self(), :recover_replication_connection, backoff)
-        {:noreply, %{state | replication_recovery_attempts: attempts}}
+        Process.send_after(self(), :recover_replication_connection, timeout)
+        {:noreply, %{state | backoff: backoff}}
     end
   end
 
@@ -476,10 +489,4 @@ defmodule Realtime.Tenants.Connect do
       log_error("StartReplicationFailed", error)
       {:error, state}
   end
-
-  defp calculate_recovery_backoff(attempts) do
-    max_backoff = @replication_recovery_backoff_min * Integer.pow(2, attempts)
-    half = div(max_backoff, 2)
-    half + Enum.random(0..half)
-  end
 end

test/realtime/tenants/connect_test.exs

@@ -484,8 +484,7 @@ defmodule Realtime.Tenants.ConnectTest do
 
       assert_receive {:DOWN, _, :process, ^replication_connection_pid, _}
 
-      Process.sleep(1500)
-      new_replication_connection_pid = ReplicationConnection.whereis(tenant.external_id)
+      new_replication_connection_pid = assert_pid(fn -> ReplicationConnection.whereis(tenant.external_id) end)
 
       assert replication_connection_pid != new_replication_connection_pid
       assert Process.alive?(new_replication_connection_pid)
@@ -503,8 +502,7 @@ defmodule Realtime.Tenants.ConnectTest do
       Process.exit(replication_connection_pid, :kill)
       assert_receive {:DOWN, _, :process, ^replication_connection_pid, _}
 
-      Process.sleep(1500)
-      new_replication_connection_pid = ReplicationConnection.whereis(tenant.external_id)
+      new_replication_connection_pid = assert_pid(fn -> ReplicationConnection.whereis(tenant.external_id) end)
 
       assert replication_connection_pid != new_replication_connection_pid
       assert Process.alive?(new_replication_connection_pid)
@@ -687,4 +685,18 @@ defmodule Realtime.Tenants.ConnectTest do
 
     Realtime.Api.update_tenant_by_external_id(tenant.external_id, %{extensions: extensions})
   end
+  defp assert_pid(call, attempts \\ 10)
+
+  defp assert_pid(_call, 0) do
+    raise "Timeout waiting for pid"
+  end
+
+  defp assert_pid(call, attempts) do
+    case call.() do
+      pid when is_pid(pid) -> pid
+      _ ->
+        Process.sleep(500)
+        assert_pid(call, attempts - 1)
+    end
+  end
 end