feat(sync-service): All configuring process GC settings for ShapeLogCollector (#3545)

Add an env var for permanent configuration of per-process flags to
configure gc settings and add to ShapeLogCollector

Also add a simple call to manually set values on the running SLC

Author: magnetised

packages/sync-service/config/runtime.exs

@@ -247,6 +247,8 @@ config :electric,
       nil
     ),
   process_registry_partitions: env!("ELECTRIC_TWEAKS_PROCESS_REGISTRY_PARTITIONS", :integer, nil),
+  process_spawn_opts:
+    env!("ELECTRIC_PROCESS_SPAWN_OPTS", &Electric.Config.parse_spawn_opts!/1, %{}),
   http_api_num_acceptors: env!("ELECTRIC_TWEAKS_HTTP_API_NUM_ACCEPTORS", :integer, 100),
   conn_max_requests: env!("ELECTRIC_TWEAKS_CONN_MAX_REQUESTS", :integer, nil),
   tcp_send_timeout:

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

@@ -142,7 +142,8 @@ defmodule Electric.Application do
         cleanup_interval_ms: get_env(opts, :cleanup_interval_ms),
         shape_hibernate_after: get_env(opts, :shape_hibernate_after),
         shape_enable_suspend?: get_env(opts, :shape_enable_suspend?),
-        conn_max_requests: get_env(opts, :conn_max_requests)
+        conn_max_requests: get_env(opts, :conn_max_requests),
+        process_spawn_opts: get_env(opts, :process_spawn_opts)
       ],
       manual_table_publishing?: get_env(opts, :manual_table_publishing?)
     )

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

@@ -93,6 +93,12 @@ defmodule Electric.Config do
     conn_max_requests: 50,
     ## Performance tweaks
     publication_alter_debounce_ms: 0,
+    # allow for configuring per-process `Process.spawn_opt()`. In the form
+    # %{process_id :: atom() => [Process.spawn_opt()]}
+    # See `Process.flag/2`
+    #
+    # e.g. %{shape_log_collector: [min_heap_size: 1024 * 1024, min_bin_vheap_size: 1024 * 1024]}
+    process_spawn_opts: %{},
     ## Misc
     process_registry_partitions: &Electric.Config.Defaults.process_registry_partitions/0,
     feature_flags: if(Mix.env() == :test, do: @known_feature_flags, else: []),
@@ -466,6 +472,69 @@ defmodule Electric.Config do
     end
   end
 
+  @valid_spawn_opts ~w[min_bin_vheap_size min_heap_size priority fullsweep_after message_queue_data]
+
+  @doc """
+  Parse `spawn_opts` from environment variable to keyword list suitable for passing to `GenServer.start_link/2`
+
+  ## Examples
+
+      iex> parse_spawn_opts!(~S({"shape_log_collector":{"min_heap_size":234,"min_bin_vheap_size":123,"message_queue_data":"on_heap","priority":"high","fullsweep_after":104}}))
+      %{shape_log_collector: [fullsweep_after: 104, message_queue_data: :on_heap, min_bin_vheap_size: 123, min_heap_size: 234, priority: :high]}
+
+      iex> parse_spawn_opts!(~S({"shape_log_collector":{"monkey":123,"message_queue_data":"on_fire","min_bin_vheap_size":-1,"priority":"high"}}))
+      %{shape_log_collector: [priority: :high]}
+
+      iex> parse_spawn_opts!("")
+      %{}
+
+      iex> parse_spawn_opts!("{}")
+      %{}
+  """
+  def parse_spawn_opts!("") do
+    %{}
+  end
+
+  def parse_spawn_opts!(str) do
+    str
+    |> Jason.decode!()
+    |> then(fn
+      opts when is_map(opts) ->
+        for {process_name, process_opts} <- opts, is_map(process_opts), into: %{} do
+          opts =
+            for {opt_key, opt_val} <- process_opts,
+                opt_key in @valid_spawn_opts,
+                key = String.to_atom(opt_key),
+                val = validate_spawn_opt(key, opt_val) do
+              {key, val}
+            end
+
+          {String.to_atom(process_name), opts}
+        end
+
+      _invalid ->
+        raise ArgumentError, message: "Invalid spawn opts: #{inspect(str)}"
+    end)
+    |> tap(fn process_spawn_opts ->
+      if map_size(process_spawn_opts) > 0 do
+        Logger.info("Process spawn opts: #{inspect(process_spawn_opts)}")
+      end
+    end)
+  end
+
+  defp validate_spawn_opt(:min_bin_vheap_size, val) when is_integer(val) and val >= 0, do: val
+  defp validate_spawn_opt(:min_heap_size, val) when is_integer(val) and val >= 0, do: val
+
+  defp validate_spawn_opt(:priority, val) when val in ["low", "normal", "high"],
+    do: String.to_atom(val)
+
+  defp validate_spawn_opt(:fullsweep_after, val) when is_integer(val) and val >= 0, do: val
+
+  defp validate_spawn_opt(:message_queue_data, val) when val in ["off_heap", "on_heap"],
+    do: String.to_atom(val)
+
+  defp validate_spawn_opt(_, _), do: nil
+
   @doc false
   # helper function for use in doc tests
   def deobfuscate({:ok, connection_opts}),

packages/sync-service/lib/electric/replication/shape_log_collector.ex

@@ -38,8 +38,13 @@ defmodule Electric.Replication.ShapeLogCollector do
             when is_map_key(state, :last_processed_lsn) and not is_nil(state.last_processed_lsn)
 
   def start_link(opts) do
-    with {:ok, opts} <- NimbleOptions.validate(opts, @schema) do
-      GenServer.start_link(__MODULE__, Map.new(opts), name: name(opts[:stack_id]))
+    with {:ok, opts} <- NimbleOptions.validate(Map.new(opts), @schema) do
+      stack_id = opts[:stack_id]
+
+      GenServer.start_link(__MODULE__, opts,
+        name: name(stack_id),
+        spawn_opt: Electric.StackConfig.spawn_opts(stack_id, :shape_log_collector)
+      )
     end
   end
 
@@ -93,6 +98,34 @@ defmodule Electric.Replication.ShapeLogCollector do
     GenServer.call(server(server_ref), :active_shapes)
   end
 
+  @doc """
+  Set process flags on the given ShapeLogCollector process.
+
+  Accepts a list of flags to set, see `Process.flag/2` for valid settings.
+
+  Doesn't crash if given an invalid flag or value - instead returns the list of
+  invalid flags.
+
+      iex> ShapeLogCollector.set_process_flags("my-stack-id", min_heap_size: 1024 * 1024, min_bin_vheap_size: 1024 * 1024)
+      {:ok, settings: [min_heap_size: 1024 * 1024, min_bin_vheap_size: 1024 * 1024], invalid: []}
+  """
+  def set_process_flags(server_ref, flags) do
+    GenServer.call(server(server_ref), {:set_process_flags, flags}, :infinity)
+  end
+
+  def get_process_flags(server_ref) do
+    if pid = server(server_ref) |> GenServer.whereis() do
+      {:garbage_collection, gc_flags} = :erlang.process_info(pid, :garbage_collection)
+      {:priority, priority} = :erlang.process_info(pid, :priority)
+
+      {:ok,
+       [priority: priority] ++
+         Keyword.take(gc_flags, [:min_bin_vheap_size, :min_heap_size, :fullsweep_after])}
+    else
+      :error
+    end
+  end
+
   def init(opts) do
     activate_mocked_functions_from_test_process()
 
@@ -234,6 +267,20 @@ defmodule Electric.Replication.ShapeLogCollector do
     {:reply, Filter.active_shapes(state.filter), state}
   end
 
+  def handle_call({:set_process_flags, flags}, _from, state) do
+    {settings, invalid} =
+      Enum.flat_map_reduce(flags, [], fn {flag, value}, invalid ->
+        try do
+          {[{flag, Process.flag(flag, value)}], invalid}
+        rescue
+          ArgumentError ->
+            {[], [flag | invalid]}
+        end
+      end)
+
+    {:reply, {:ok, [settings: settings, invalid: invalid]}, state}
+  end
+
   def handle_cast({:writer_flushed, shape_id, offset}, state) do
     {:noreply,
      state

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

@@ -9,6 +9,12 @@ defmodule Electric.StackConfig do
     :ets.lookup_element(table(stack_id), key, 2, default)
   end
 
+  def spawn_opts(stack_id, process_name) do
+    stack_id
+    |> lookup(:process_spawn_opts, %{})
+    |> Map.get(process_name, [])
+  end
+
   def lookup!(stack_id, key) do
     :ets.lookup_element(table(stack_id), key, 2)
   rescue
@@ -25,7 +31,8 @@ defmodule Electric.StackConfig do
       shape_hibernate_after: Electric.Config.default(:shape_hibernate_after),
       shape_enable_suspend?: Electric.Config.default(:shape_enable_suspend?),
       chunk_bytes_threshold: Electric.ShapeCache.LogChunker.default_chunk_size_threshold(),
-      feature_flags: []
+      feature_flags: [],
+      process_spawn_opts: %{}
     ]
   end
 

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

@@ -135,7 +135,8 @@ defmodule Electric.StackSupervisor do
                      conn_max_requests: [
                        type: :pos_integer,
                        default: Electric.Config.default(:conn_max_requests)
-                     ]
+                     ],
+                     process_spawn_opts: [type: :map, default: %{}]
                    ]
                  ],
                  manual_table_publishing?: [
@@ -317,6 +318,7 @@ defmodule Electric.StackSupervisor do
 
     shape_hibernate_after = Keyword.fetch!(config.tweaks, :shape_hibernate_after)
     shape_enable_suspend? = Keyword.fetch!(config.tweaks, :shape_enable_suspend?)
+    process_spawn_opts = Keyword.fetch!(config.tweaks, :process_spawn_opts)
 
     shape_cache_opts = [
       stack_id: stack_id
@@ -366,6 +368,7 @@ defmodule Electric.StackSupervisor do
            inspector: inspector,
            shape_hibernate_after: shape_hibernate_after,
            shape_enable_suspend?: shape_enable_suspend?,
+           process_spawn_opts: process_spawn_opts,
            feature_flags: Map.get(config, :feature_flags, [])
          ]},
         {Electric.AsyncDeleter,

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

@@ -82,6 +82,22 @@ defmodule Electric.Replication.ShapeLogCollectorTest do
     %{server: pid, registry: registry_name, shape_cache: shape_cache_pid}
   end
 
+  describe "process gc configuration" do
+    setup :setup_log_collector
+
+    @tag process_spawn_opts: %{
+           shape_log_collector: [priority: :high, min_bin_vheap_size: 1024 * 1024]
+         }
+    test "are correctly passed to process", ctx do
+      pid = ShapeLogCollector.name(ctx.stack_id) |> GenServer.whereis()
+
+      info = Process.info(pid)
+
+      assert :high == info[:priority]
+      assert info[:garbage_collection][:min_bin_vheap_size] >= 1024 * 1024
+    end
+  end
+
   describe "shape restoration" do
     setup :setup_log_collector
 

packages/sync-service/test/support/component_setup.ex

@@ -111,7 +111,8 @@ defmodule Support.ComponentSetup do
                Map.get(ctx, :registry, Electric.StackSupervisor.registry_name(stack_id)),
              shape_hibernate_after: Map.get(ctx, :shape_hibernate_after, 1_000),
              shape_enable_suspend?: Map.get(ctx, :suspend, false),
-             feature_flags: Electric.Config.get_env(:feature_flags)
+             feature_flags: Electric.Config.get_env(:feature_flags),
+             process_spawn_opts: Map.get(ctx, :process_spawn_opts, %{})
            ],
            seed_config
          )}