chore: Ensure test processes are supervised by the test supervisor (#2841)

Ensures that all processes started are started under the ExUnit test
supervisor, in the hope that this gives us better guarantees about
cleanups occurring at the right time and we reduce flakes that seem to
potentially be related to timing issues.

Author: msfstef

packages/sync-service/lib/electric/postgres/replication_client.ex

@@ -215,6 +215,12 @@ defmodule Electric.Postgres.ReplicationClient do
     {:noreply, state}
   end
 
+  # This callback is invoked when the connection process receives a shutdown signal.
+  def handle_info({:EXIT, _pid, :shutdown}, _state) do
+    Logger.debug("Replication client #{inspect(self())} received shutdown signal, stopping")
+    {:disconnect, :shutdown}
+  end
+
   # The implementation of Postgrex.ReplicationConnection doesn't give us a convenient way to
   # check whether the START_REPLICATION_SLOT statement succeeded before switching the
   # connection into streaming mode. Returning {:query, "START_REPLICATION_SLOT ...", state}

packages/sync-service/test/electric/concurrent_stream_test.exs

@@ -7,7 +7,7 @@ defmodule Electric.ConcurrentStreamTest do
   describe "stream_to_end/2" do
     setup %{tmp_dir: tmp_dir, test: test} do
       db = :"cubdb_#{test}"
-      CubDB.start_link(data_dir: tmp_dir, name: db)
+      start_link_supervised!({CubDB, data_dir: tmp_dir, name: db})
       {:ok, %{db: db}}
     end
 

packages/sync-service/test/electric/connection/manager_test.exs

@@ -102,14 +102,20 @@ defmodule Electric.Connection.ConnectionManagerTest do
       wait_until_active(stack_id)
 
       # Start another lock process so that when ConnectionManager exits it is not able to restore its readiness immediately.
+      new_stack_id = stack_id <> "_new"
+      _registry = start_link_supervised!({Electric.ProcessRegistry, stack_id: new_stack_id})
+
       lock_opts = [
         connection_opts: ctx.connection_opts,
         connection_manager: self(),
         lock_name: Keyword.fetch!(ctx.replication_opts, :slot_name),
-        stack_id: stack_id
+        stack_id: new_stack_id
       ]
 
-      Electric.Postgres.LockConnection.start_link(lock_opts)
+      start_supervised!(%{
+        id: :alt_lock,
+        start: {Electric.Postgres.LockConnection, :start_link, [lock_opts]}
+      })
 
       monitor = monitor_replication_client(stack_id)
 

packages/sync-service/test/electric/postgres/lock_connection_test.exs

@@ -18,13 +18,19 @@ defmodule Electric.Postgres.LockConnectionTest do
     } do
       log =
         capture_log(fn ->
-          assert {:ok, _pid} =
-                   LockConnection.start_link(
-                     connection_opts: config,
-                     connection_manager: self(),
-                     lock_name: @lock_name,
-                     stack_id: stack_id
-                   )
+          start_supervised!(%{
+            id: :lock1,
+            start:
+              {LockConnection, :start_link,
+               [
+                 [
+                   connection_opts: config,
+                   connection_manager: self(),
+                   lock_name: @lock_name,
+                   stack_id: stack_id
+                 ]
+               ]}
+          })
 
           assert_lock_acquired()
         end)
@@ -44,33 +50,45 @@ defmodule Electric.Postgres.LockConnectionTest do
 
     test "should wait if lock is already acquired", %{db_config: config, stack_id: stack_id} do
       # grab lock with one connection
-      assert {:ok, pid1} =
-               LockConnection.start_link(
-                 connection_opts: config,
-                 connection_manager: self(),
-                 lock_name: @lock_name,
-                 stack_id: stack_id
-               )
+      start_supervised!(%{
+        id: :lock1,
+        start:
+          {LockConnection, :start_link,
+           [
+             [
+               connection_opts: config,
+               connection_manager: self(),
+               lock_name: @lock_name,
+               stack_id: stack_id
+             ]
+           ]}
+      })
 
       assert_lock_acquired()
 
       # try to grab the same lock using a different connection
       new_stack_id = stack_id <> "_new"
-      _registry = start_link_supervised!({Electric.ProcessRegistry, stack_id: new_stack_id})
+      start_link_supervised!({Electric.ProcessRegistry, stack_id: new_stack_id})
 
-      assert {:ok, _pid} =
-               LockConnection.start_link(
-                 connection_opts: config,
-                 connection_manager: self(),
-                 lock_name: @lock_name,
-                 stack_id: new_stack_id
-               )
+      start_supervised!(%{
+        id: :lock2,
+        start:
+          {LockConnection, :start_link,
+           [
+             [
+               connection_opts: config,
+               connection_manager: self(),
+               lock_name: @lock_name,
+               stack_id: new_stack_id
+             ]
+           ]}
+      })
 
       # should fail to grab it
       refute_lock_acquired()
 
       # should immediately grab it once previous lock is released
-      GenServer.stop(pid1)
+      stop_supervised!(:lock1)
       assert_lock_acquired()
     end
   end

packages/sync-service/test/electric/postgres/replication_client_test.exs

@@ -470,11 +470,14 @@ defmodule Electric.Postgres.ReplicationClientTest do
   defp start_client(ctx, overrides \\ []) do
     ctx = Enum.into(overrides, ctx)
 
-    {:ok, client_pid} =
-      ReplicationClient.start_link(
-        stack_id: ctx.stack_id,
-        replication_opts: ctx.replication_opts
-      )
+    client_pid =
+      start_link_supervised!(%{
+        id: ReplicationClient,
+        start:
+          {ReplicationClient, :start_link,
+           [[stack_id: ctx.stack_id, replication_opts: ctx.replication_opts]]},
+        restart: :temporary
+      })
 
     conn_mgr = ctx.connection_manager
     assert_receive {^conn_mgr, :streaming_started}, @assert_receive_db_timeout

packages/sync-service/test/electric/process_registry_test.exs

@@ -6,12 +6,10 @@ defmodule Electric.ProcessRegistryTest do
 
   describe "alive?/2" do
     test "should return false for inexistent process" do
-      {:ok, _} =
-        ProcessRegistry.start_link(
-          name: ProcessRegistry.registry_name(@stack_id),
-          keys: :duplicate,
-          stack_id: @stack_id
-        )
+      start_link_supervised!(
+        {ProcessRegistry,
+         name: ProcessRegistry.registry_name(@stack_id), keys: :duplicate, stack_id: @stack_id}
+      )
 
       assert false == ProcessRegistry.alive?(@stack_id, "bar")
     end

packages/sync-service/test/electric/replication/publication_manager_test.exs

@@ -240,7 +240,7 @@ defmodule Electric.Replication.PublicationManagerTest do
       ctx: ctx,
       opts: opts
     } do
-      GenServer.stop(opts[:server])
+      stop_supervised!(opts[:server])
 
       test_id = self()
 

packages/sync-service/test/electric/replication/shape_log_collector_test.exs

@@ -81,7 +81,7 @@ defmodule Electric.Replication.ShapeLogCollectorTest do
         registry: registry_name
       ]
 
-    {:ok, shape_cache_pid} = Electric.ShapeCache.start_link(shape_cache_opts)
+    shape_cache_pid = start_link_supervised!({Electric.ShapeCache, shape_cache_opts})
 
     %{server: pid, registry: registry_name, shape_cache: shape_cache_pid}
   end
@@ -103,13 +103,14 @@ defmodule Electric.Replication.ShapeLogCollectorTest do
 
       consumers =
         Enum.map(1..3, fn id ->
-          {:ok, consumer} =
-            Support.TransactionConsumer.start_link(
-              id: id,
-              parent: parent,
-              producer: ctx.server,
-              shape: @shape
-            )
+          consumer =
+            start_link_supervised!(%{
+              id: {:consumer, id},
+              start:
+                {Support.TransactionConsumer, :start_link,
+                 [[id: id, parent: parent, producer: ctx.server, shape: @shape]]},
+              restart: :temporary
+            })
 
           {id, consumer}
         end)
@@ -240,13 +241,14 @@ defmodule Electric.Replication.ShapeLogCollectorTest do
 
       consumers =
         Enum.map(1..3, fn id ->
-          {:ok, consumer} =
-            Support.TransactionConsumer.start_link(
-              id: id,
-              parent: parent,
-              producer: ctx.server,
-              shape: @shape
-            )
+          consumer =
+            start_link_supervised!(%{
+              id: {:consumer, id},
+              start:
+                {Support.TransactionConsumer, :start_link,
+                 [[id: id, parent: parent, producer: ctx.server, shape: @shape]]},
+              restart: :temporary
+            })
 
           {id, consumer}
         end)
@@ -338,12 +340,10 @@ defmodule Electric.Replication.ShapeLogCollectorTest do
 
     consumer_id = "test_consumer"
 
-    {:ok, consumer} =
-      Support.TransactionConsumer.start_link(
-        id: consumer_id,
-        parent: self(),
-        producer: pid,
-        shape: @shape
+    consumer =
+      start_link_supervised!(
+        {Support.TransactionConsumer,
+         id: consumer_id, parent: self(), producer: pid, shape: @shape}
       )
 
     consumers = [{consumer_id, consumer}]

packages/sync-service/test/electric/shape_cache/file_storage_test.exs

@@ -27,7 +27,7 @@ defmodule Electric.ShapeCache.FileStorageTest do
       )
 
     shape_opts = FileStorage.for_shape(@shape_handle, opts)
-    {:ok, pid} = FileStorage.start_link(shape_opts)
+    pid = start_link_supervised!({FileStorage, shape_opts})
     {:ok, %{opts: shape_opts, shared_opts: opts, pid: pid, storage: {FileStorage, shape_opts}}}
   end
 

packages/sync-service/test/electric/shape_cache/storage_implementations_test.exs

@@ -753,7 +753,7 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
   defp start_storage(%{module: module} = context) do
     opts = module |> opts(context) |> module.shared_opts()
     shape_opts = module.for_shape(@shape_handle, opts)
-    {:ok, pid} = module.start_link(shape_opts)
+    pid = start_link_supervised!({module, shape_opts})
     {:ok, %{opts: shape_opts, shared_opts: opts, pid: pid, storage: {module, shape_opts}}}
   end