fixup! fix: ensure Watchdog kills ReplicationConnection on failing healthcheck

Author: edgurgel

lib/realtime/tenants/replication_connection.ex

@@ -73,7 +73,7 @@ defmodule Realtime.Tenants.ReplicationConnection do
   @doc """
   Starts the replication connection for a tenant and monitors a given pid to stop the ReplicationConnection.
   """
-  @spec start(Realtime.Api.Tenant.t(), pid()) :: {:ok, pid()} | {:error, any()}
+  @spec start(Realtime.Api.Tenant.t(), pid(), query_timeout :: timeout) :: {:ok, pid()} | {:error, any()}
   def start(tenant, monitored_pid, query_timeout \\ @default_query_timeout) do
     Logger.info("Starting replication for Broadcast Changes")
     opts = %__MODULE__{tenant_id: tenant.external_id, monitored_pid: monitored_pid, query_timeout: query_timeout}
@@ -242,7 +242,7 @@ defmodule Realtime.Tenants.ReplicationConnection do
       end)
 
     if valid_tables and rows != [] do
-      {:query, "SELECT 1", %{state | step: :start_replication_slot}}
+      {:query, "SELECT 1", [timeout: state.query_timeout], %{state | step: :start_replication_slot}}
     else
       query =
         "DROP PUBLICATION IF EXISTS #{publication_name}; CREATE PUBLICATION #{publication_name} FOR TABLE #{@schema}.#{@table}"

lib/realtime/tenants/replication_connection/watchdog.ex

@@ -1,7 +1,7 @@
 defmodule Realtime.Tenants.ReplicationConnection.Watchdog do
   @moduledoc """
   Monitors ReplicationConnection health by performing periodic call checks.
-  If the call times out, logs an error and shuts down, which cascades to ReplicationConnection.
+  If the call times out, logs an error and terminates the ReplicationConnection process to trigger a restart.
   """
   use GenServer
   use Realtime.Logs
@@ -60,7 +60,13 @@ defmodule Realtime.Tenants.ReplicationConnection.Watchdog do
       :exit, {:timeout, _} ->
         log_error("ReplicationConnectionWatchdogTimeout", "ReplicationConnection is not responding")
 
-        ReplicationConnection.stop(state.tenant_id, state.parent_pid)
+        case ReplicationConnection.stop(state.tenant_id, state.parent_pid) do
+          :ok ->
+            Logger.info("ReplicationConnection stopped successfully")
+
+          {:error, reason} ->
+            log_error(ReplicationConnectionWatchdogTimeout, "Failed to stop ReplicationConnection: #{inspect(reason)}")
+        end
 
         {:stop, :watchdog_timeout, state}
     end

test/realtime/tenants/replication_connection_test.exs

@@ -66,7 +66,7 @@ defmodule Realtime.Tenants.ReplicationConnectionTest do
           # Let's make it not reply to health checks
           :sys.suspend(pid)
 
-          reason = assert_process_down(pid, 200)
+          reason = assert_process_down(pid, 400)
           assert reason == :shutdown
         end)