fix: gracefully stop on connect error for cdc rls processes

.github/workflows/tests.yml

@@ -73,7 +73,7 @@ jobs:
       - name: Install dependencies
         run: mix deps.get
       - name: Set up Postgres
-        run: docker compose -f compose.dbs.yml up -d
+        run: docker compose -f compose.dbs.yml up -d --wait
       - name: Start epmd
         run: epmd -daemon
       - name: Run tests

lib/extensions/postgres_cdc_rls/replication_poller.ex

@@ -66,7 +66,7 @@ defmodule Extensions.PostgresCdcRls.ReplicationPoller do
     else
       {:error, reason} ->
         log_error("ReplicationPollerConnectionFailed", reason)
-        {:stop, reason, state}
+        {:stop, {:shutdown, reason}, state}
     end
   end
 

lib/extensions/postgres_cdc_rls/subscription_manager.ex

@@ -115,7 +115,7 @@ defmodule Extensions.PostgresCdcRls.SubscriptionManager do
     else
       {:error, reason} ->
         log_error("SubscriptionManagerConnectionFailed", reason)
-        {:stop, reason, nil}
+        {:stop, {:shutdown, reason}, nil}
     end
   end
 

lib/extensions/postgres_cdc_rls/subscriptions_checker.ex

@@ -71,7 +71,7 @@ defmodule Extensions.PostgresCdcRls.SubscriptionsChecker do
     else
       {:error, reason} ->
         log_error("SubscriptionsCheckerConnectionFailed", reason)
-        {:stop, reason, nil}
+        {:stop, {:shutdown, reason}, nil}
     end
   end
 

test/realtime/extensions/cdc_rls/replication_poller_test.exs

@@ -783,7 +783,7 @@ defmodule Realtime.Extensions.PostgresCdcRls.ReplicationPollerTest do
       pid = start_supervised!({Poller, args}, restart: :temporary)
       ref = Process.monitor(pid)
 
-      assert_receive {:DOWN, ^ref, :process, ^pid, :econnrefused}, 1000
+      assert_receive {:DOWN, ^ref, :process, ^pid, {:shutdown, :econnrefused}}, 1000
     end
   end
 

test/realtime/extensions/cdc_rls/subscription_manager_test.exs

@@ -246,7 +246,7 @@ defmodule Realtime.Extensions.CdcRls.SubscriptionManagerTest do
       pid = start_supervised!({SubscriptionManager, args}, restart: :temporary)
       ref = Process.monitor(pid)
 
-      assert_receive {:DOWN, ^ref, :process, ^pid, :econnrefused}, 1000
+      assert_receive {:DOWN, ^ref, :process, ^pid, {:shutdown, :econnrefused}}, 1000
     end
   end
 

test/realtime/extensions/cdc_rls/subscriptions_checker_test.exs

@@ -63,7 +63,7 @@ defmodule Realtime.Extensions.PostgresCdcRl.SubscriptionsCheckerTest do
       pid = start_supervised!({Checker, args}, restart: :temporary)
       ref = Process.monitor(pid)
 
-      assert_receive {:DOWN, ^ref, :process, ^pid, :econnrefused}, 2000
+      assert_receive {:DOWN, ^ref, :process, ^pid, {:shutdown, :econnrefused}}, 2000
     end
   end