chore(sync-service): Add MonitoredCoreSupervisor (#3307)

Part of a series of PRs to separate the shape and connection subsystems.

This PR moves the StatusMonitor further up the supervision tree so that
it can supervise both subsystems. Currently the StatusMonitor is
supervised by the Connection.Supervisor which means once the shape
subsystem has been moved out of the Connection.Supervisor's descendants
it will no longer be at the right level.

This PR introduces the idea of the electric core - the shape and
connection subsystems, supervised by the CoreSupervisor. There is also a
MonitoredCoreSupervisor which supervises the CoreSupervisor along with
the StatusMonitor with the the rest_for_one strategy which allows the
StatusMonitor to stay consistent (everything it monitors is restarted if
it dies, so there is no chance it will die and lose references to what
it is monitoring).

The supervision tree now looks like this:

StackSupervisor
- utility processes such as the EtsInspector that can be restarted
indecently
- MonitoredCoreSupervisor
  - StatusMonitor
  - CoreSupervisor
- (to be added in a later PR) Replication.Supervisor (to later be
renamed to ShapeSubsystemSupervisor)
- Connection.Manager.Supervisor (to later be renamed to
ConnectionSubsystemSupervisor)

Author: robacourt

packages/sync-service/lib/electric/connection/supervisor.ex

@@ -22,8 +22,6 @@ defmodule Electric.Connection.Supervisor do
   whole OTP application shutting down.
   """
 
-  # This supervisor is meant to be a child of Electric.StackSupervisor.
-  #
   # The `restart: :transient, significant: true` combo allows for shutting the supervisor down
   # and signalling the parent supervisor to shut itself down as well if that one has
   # `:auto_shutdown` set to `:any_significant` or `:all_significant`.
@@ -63,15 +61,10 @@ defmodule Electric.Connection.Supervisor do
     Electric.Telemetry.Sentry.set_tags_context(stack_id: stack_id)
 
     children = [
-      {Electric.StatusMonitor, stack_id: stack_id},
       {Electric.Connection.Restarter, stack_id: stack_id},
       {Electric.Connection.Manager.Supervisor, opts}
     ]
 
-    # The :rest_for_one strategy is used here to ensure that if the StatusMonitor unexpectedly dies,
-    # all subsequent child processes are also restarted. Since the StatusMonitor keeps track of the
-    # statuses of the other children, losing it means losing that state. Restarting the other children
-    # ensures they re-notify the StatusMonitor, allowing it to rebuild its internal state correctly.
     Supervisor.init(children, strategy: :rest_for_one)
   end
 end

packages/sync-service/lib/electric/core_supervisor.ex

@@ -0,0 +1,33 @@
+defmodule Electric.CoreSupervisor do
+  @moduledoc """
+  A supervisor that starts the core components of the Electric system.
+  This is divided into two subsystems:
+  1. The connection subsystem (processes that may exit on a connection failure), started with Connection.Manager.Supervisor
+  2. The shape subsystem (processes that are resilient to connection failures), started with Electric.Replication.Supervisor
+
+  NOTE: Currently the ShapeSubsystem is not directly supervised here, but this change with happen in an upcoming PR.
+  """
+
+  use Supervisor, restart: :transient, significant: true
+
+  def start_link(opts) do
+    Supervisor.start_link(__MODULE__, opts)
+  end
+
+  @impl true
+  def init(opts) do
+    stack_id = Keyword.fetch!(opts, :stack_id)
+
+    Process.set_label({:core_supervisor, stack_id})
+    Logger.metadata(stack_id: stack_id)
+    Electric.Telemetry.Sentry.set_tags_context(stack_id: stack_id)
+
+    connection_manager_opts = Keyword.fetch!(opts, :connection_manager_opts)
+
+    children = [
+      {Electric.Connection.Supervisor, connection_manager_opts}
+    ]
+
+    Supervisor.init(children, strategy: :one_for_one, auto_shutdown: :any_significant)
+  end
+end

packages/sync-service/lib/electric/monitored_core_supervisor.ex

@@ -0,0 +1,33 @@
+defmodule Electric.MonitoredCoreSupervisor do
+  @moduledoc """
+  A supervisor that starts and monitors the core components of the Electric system.
+  It needs to be a separate supervisor from the CoreSupervisor because of the way
+  the StatusMonitor works (see the rest_for_one comments below).
+  """
+
+  use Supervisor, restart: :transient, significant: true
+
+  def start_link(opts) do
+    Supervisor.start_link(__MODULE__, opts)
+  end
+
+  @impl true
+  def init(opts) do
+    stack_id = Keyword.fetch!(opts, :stack_id)
+
+    Process.set_label({:monitored_core_supervisor, stack_id})
+    Logger.metadata(stack_id: stack_id)
+    Electric.Telemetry.Sentry.set_tags_context(stack_id: stack_id)
+
+    children = [
+      {Electric.StatusMonitor, stack_id: stack_id},
+      {Electric.CoreSupervisor, opts}
+    ]
+
+    # The :rest_for_one strategy is used here to ensure that if the StatusMonitor unexpectedly dies,
+    # all the processes it is monitoring are also restarted. Since the StatusMonitor keeps track of the
+    # statuses of the other processes, losing it means losing that state. Restarting the other children
+    # ensures they re-notify the StatusMonitor, allowing it to rebuild its internal state correctly.
+    Supervisor.init(children, strategy: :rest_for_one, auto_shutdown: :any_significant)
+  end
+end

packages/sync-service/lib/electric/stack_supervisor.ex

@@ -310,7 +310,7 @@ defmodule Electric.StackSupervisor do
     shape_log_collector =
       Electric.Replication.ShapeLogCollector.name(stack_id)
 
-    new_connection_manager_opts = [
+    connection_manager_opts = [
       stack_id: stack_id,
       # Coming from the outside, need validation
       connection_opts: config.connection_opts,
@@ -374,7 +374,8 @@ defmodule Electric.StackSupervisor do
              Keyword.take(shape_cache_opts, [:publication_manager])
            ])},
           {Electric.ShapeCache.ShapeStatusOwner, [stack_id: stack_id, storage: storage]},
-          {Electric.Connection.Supervisor, new_connection_manager_opts}
+          {Electric.MonitoredCoreSupervisor,
+           stack_id: stack_id, connection_manager_opts: connection_manager_opts}
         ]
 
     # Store the telemetry span attributes in the persistent term for this stack

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

@@ -11,6 +11,7 @@ defmodule Electric.Connection.ConnectionManagerTest do
   setup [
     :with_unique_db,
     :with_stack_id_from_test,
+    :with_status_monitor,
     :with_persistent_kv,
     :with_inspector,
     :with_slot_name_and_stream_id,