add consumer suspend to feature flags

Author: magnetised

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

@@ -38,7 +38,7 @@ defmodule Electric.Config do
 
   @build_env Mix.env()
 
-  @known_feature_flags ~w[allow_subqueries]
+  @known_feature_flags ~w[allow_subqueries suspend_consumers]
 
   @defaults [
     ## Database

packages/sync-service/lib/electric/shapes/consumer.ex

@@ -370,12 +370,7 @@ defmodule Electric.Shapes.Consumer do
     #   a. we have no dependent shapes
     #   b. we don't have a materializer subscribed
 
-    can_suspend? =
-      Electric.StackConfig.lookup(state.stack_id, :shape_enable_suspend?, true) and
-        state.snapshot_started and Enum.empty?(state.shape.shape_dependencies_handles) and
-        not state.materializer_subscribed?
-
-    if can_suspend? do
+    if consumer_suspend_enabled?(state) and consumer_can_suspend?(state) do
       Logger.debug(fn -> ["Suspending consumer ", to_string(state.shape_handle)] end)
       {:stop, ShapeCleaner.consumer_suspend_reason(), state}
     else
@@ -385,6 +380,17 @@ defmodule Electric.Shapes.Consumer do
     end
   end
 
+  defp consumer_suspend_enabled?(%{stack_id: stack_id}) do
+    # TODO: remove additional feature flag check once feature is stabilised
+    Electric.StackConfig.lookup(stack_id, :shape_enable_suspend?, true) and
+      "suspend_consumers" in Electric.StackConfig.lookup(stack_id, :feature_flags, [])
+  end
+
+  defp consumer_can_suspend?(state) do
+    state.snapshot_started and Enum.empty?(state.shape.shape_dependencies_handles) and
+      not state.materializer_subscribed?
+  end
+
   @impl GenServer
   def terminate(reason, state) do
     Logger.debug(fn ->

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

@@ -23,7 +23,9 @@ defmodule Electric.StackConfig do
     [
       snapshot_timeout_to_first_data: :timer.seconds(30),
       shape_hibernate_after: Electric.Config.default(:shape_hibernate_after),
-      chunk_bytes_threshold: Electric.ShapeCache.LogChunker.default_chunk_size_threshold()
+      shape_enable_suspend?: Electric.Config.default(:shape_enable_suspend?),
+      chunk_bytes_threshold: Electric.ShapeCache.LogChunker.default_chunk_size_threshold(),
+      feature_flags: []
     ]
   end
 

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

@@ -367,7 +367,8 @@ defmodule Electric.StackSupervisor do
              snapshot_timeout_to_first_data: config.tweaks[:snapshot_timeout_to_first_data],
              inspector: inspector,
              shape_hibernate_after: shape_hibernate_after,
-             shape_enable_suspend?: shape_enable_suspend?
+             shape_enable_suspend?: shape_enable_suspend?,
+             feature_flags: Map.get(config, :feature_flags, [])
            ]},
           {Electric.AsyncDeleter,
            stack_id: stack_id,

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

@@ -112,7 +112,8 @@ defmodule Support.ComponentSetup do
              inspector: Map.get(ctx, :inspector, nil),
              shape_changes_registry:
                Map.get(ctx, :registry, Electric.StackSupervisor.registry_name(stack_id)),
-             shape_hibernate_after: Map.get(ctx, :shape_hibernate_after, 1_000)
+             shape_hibernate_after: Map.get(ctx, :shape_hibernate_after, 1_000),
+             feature_flags: Electric.Config.get_env(:feature_flags)
            ],
            seed_config
          )}
@@ -456,7 +457,8 @@ defmodule Support.ComponentSetup do
            registry_partitions: 1,
            shape_cleaner_opts: shape_cleaner_opts(ctx)
          ],
-         manual_table_publishing?: Map.get(ctx, :manual_table_publishing?, false)},
+         manual_table_publishing?: Map.get(ctx, :manual_table_publishing?, false),
+         feature_flags: Electric.Config.get_env(:feature_flags)},
         restart: :temporary,
         significant: false
       )