feat(consumer): bound Shape.Consumer heap growth via spawn opts + adaptive GC (#4539)

Closes #4476

## Summary

Bounds runaway old-heap growth of `Shape.Consumer` processes (issue
#4476). On a production node with 5,043 consumers, ~36 GB was
unreclaimable garbage (7–9 MB heaps each holding ~8 KB of live state)
because consumers inherit the BEAM default `fullsweep_after: 65535` and
rarely hibernate, so a full sweep is effectively never triggered.

Two complementary, independently-tunable levers:

1. **Per-process spawn opts (env-driven).** `Consumer`,
`Consumer.Snapshotter`, and `Consumer.Materializer` now pass `spawn_opt:
Electric.StackConfig.spawn_opts(stack_id, key)` (keys `:consumer`,
`:consumer_snapshotter`, `:consumer_materializer`), mirroring
`ShapeLogCollector`. This lets `ELECTRIC_PROCESS_SPAWN_OPTS` set
`fullsweep_after` per process type. No baked-in defaults — purely
env-configured.

2. **Adaptive GC, opt-in.** After processing a transaction fragment
(including during the startup buffer drain), the consumer forces
`:erlang.garbage_collect()` only if its heap exceeds
`consumer_gc_heap_threshold` (bytes; `nil` = off, the default). The
threshold is cached per-consumer at startup (`State.new/2`), so changing
it affects consumers started afterwards. The forced sweep runs **off the
reply path** via a `{:continue, :maybe_gc}` step, so it never blocks the
`ShapeLogCollector`'s synchronous publish call, and a hysteresis
interval (`@gc_min_interval_ms`) caps how much CPU a busy consumer
spends sweeping. See `Electric.Shapes.Consumer.set_gc_heap_threshold/2`.

Request-handler (`GET /v1/shape`) processes are intentionally untouched
— they already expose `fullsweep_after` via
`ELECTRIC_TWEAKS_HANDLER_FULLSWEEP_AFTER` and force GC before long-poll
blocking.

## Config

| Env var | Meaning | Default |
|---|---|---|
| `ELECTRIC_PROCESS_SPAWN_OPTS` | JSON map; now accepts
`consumer`/`consumer_snapshotter`/`consumer_materializer` keys (e.g.
`{"consumer":{"fullsweep_after":4}}`) | `%{}` |
| `ELECTRIC_CONSUMER_GC_HEAP_THRESHOLD` | adaptive-GC heap threshold in
bytes; `nil` disables | `nil` |

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Oleksii Sholik <oleksii@sholik.dev>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: 3 people

.changeset/consumer-heap-gc.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Bound Shape.Consumer heap growth: make the consumer family's process spawn options (incl. `fullsweep_after`) configurable per process via `ELECTRIC_PROCESS_SPAWN_OPTS`, and add an opt-in adaptive GC that runs after a transaction fragment when the consumer's heap exceeds the runtime-tunable `ELECTRIC_CONSUMER_GC_HEAP_THRESHOLD` (off by default).

packages/sync-service/config/runtime.exs

@@ -291,6 +291,7 @@ config :electric,
   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, %{}),
+  consumer_gc_heap_threshold: env!("ELECTRIC_CONSUMER_GC_HEAP_THRESHOLD", :integer, nil),
   http_api_num_acceptors: env!("ELECTRIC_TWEAKS_HTTP_API_NUM_ACCEPTORS", :integer, 100),
   conn_max_requests: env!("ELECTRIC_TWEAKS_CONN_MAX_REQUESTS", :integer, nil),
   handler_fullsweep_after: env!("ELECTRIC_TWEAKS_HANDLER_FULLSWEEP_AFTER", :integer, nil),

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

@@ -156,7 +156,8 @@ defmodule Electric.Application do
         shape_suspend_after: get_env(opts, :shape_suspend_after),
         conn_max_requests: get_env(opts, :conn_max_requests),
         handler_fullsweep_after: get_env(opts, :handler_fullsweep_after),
-        process_spawn_opts: get_env(opts, :process_spawn_opts)
+        process_spawn_opts: get_env(opts, :process_spawn_opts),
+        consumer_gc_heap_threshold: get_env(opts, :consumer_gc_heap_threshold)
       ],
       manual_table_publishing?: get_env(opts, :manual_table_publishing?),
       shape_db_opts: [

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

@@ -114,6 +114,9 @@ defmodule Electric.Config do
     #
     # e.g. %{shape_log_collector: [min_heap_size: 1024 * 1024, min_bin_vheap_size: 1024 * 1024]}
     process_spawn_opts: %{},
+    # Heap-size threshold (in BYTES) above which a consumer runs :erlang.garbage_collect()
+    # after processing a transaction fragment.
+    consumer_gc_heap_threshold: nil,
     ## Misc
     process_registry_partitions: &Electric.Config.Defaults.process_registry_partitions/0,
     feature_flags: if(Mix.env() == :test, do: @known_feature_flags, else: []),

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

@@ -38,6 +38,12 @@ defmodule Electric.Shapes.Consumer do
   @default_snapshot_timeout 45_000
   @stop_and_clean_timeout 30_000
   @stop_and_clean_reason ShapeCleaner.consumer_cleanup_reason()
+  @word_size :erlang.system_info(:wordsize)
+
+  # Minimum wall-clock interval (ms) between consumer-forced full GC sweeps. Caps how much CPU
+  # a busy consumer spends on full sweeps to at most one per @gc_min_interval_ms regardless of
+  # fragment rate.
+  @gc_min_interval_ms 1_000
 
   @type initialize_shape_opts() :: %{
           :action => :create | :restore,
@@ -110,9 +116,24 @@ defmodule Electric.Shapes.Consumer do
     ConsumerRegistry.whereis(stack_id, shape_handle)
   end
 
+  @doc """
+  Set the adaptive-GC heap threshold (bytes, or nil to disable) for a single stack.
+
+  Consumers cache this value at startup (see `State.new/2`), so the new threshold only
+  applies to consumers started after this call — already-running consumers keep the
+  threshold they read when they booted. Safe to call from IEx.
+  """
+  @spec set_gc_heap_threshold(Electric.stack_id(), non_neg_integer() | nil) :: :ok
+  def set_gc_heap_threshold(stack_id, threshold_bytes)
+      when is_nil(threshold_bytes) or (is_integer(threshold_bytes) and threshold_bytes >= 0) do
+    Electric.StackConfig.put(stack_id, :consumer_gc_heap_threshold, threshold_bytes)
+    :ok
+  end
+
   def start_link(%{stack_id: stack_id, shape_handle: shape_handle} = _config) do
     GenServer.start_link(__MODULE__, %{stack_id: stack_id, shape_handle: shape_handle},
-      name: name(stack_id, shape_handle)
+      name: name(stack_id, shape_handle),
+      spawn_opt: Electric.StackConfig.spawn_opts(stack_id, :consumer)
     )
   end
 
@@ -170,10 +191,19 @@ defmodule Electric.Shapes.Consumer do
     if state.terminating? do
       {:noreply, state, {:continue, :stop_and_clean}}
     else
-      {:noreply, state, state.hibernate_after}
+      {:noreply, state, {:continue, :maybe_gc}}
     end
   end
 
+  # Deferred adaptive GC. Reached via {:continue, :maybe_gc} after a fragment (or a
+  # full buffer drain) has been processed, so a forced full sweep runs off the
+  # reply/critical path rather than blocking the ShapeLogCollector. Re-establishes
+  # the hibernate_after timeout that the {:continue, …} return could not carry.
+  def handle_continue(:maybe_gc, state) do
+    state = maybe_garbage_collect(state)
+    {:noreply, state, state.hibernate_after}
+  end
+
   @impl GenServer
   # Any incoming message counts as activity: cancel the pending suspend timer (if
   # any) and recurse for actual handling of the call.
@@ -204,7 +234,7 @@ defmodule Electric.Shapes.Consumer do
         {:reply, :ok, state, {:continue, :stop_and_clean}}
 
       state ->
-        {:reply, :ok, state, state.hibernate_after}
+        {:reply, :ok, state, {:continue, :maybe_gc}}
     end
   end
 
@@ -520,6 +550,56 @@ defmodule Electric.Shapes.Consumer do
     handle_txn_fragment(txn_fragment, state)
   end
 
+  # Adaptive GC check. Always invoked from a {:continue, :maybe_gc} handler so the
+  # forced full sweep runs off the ShapeLogCollector's synchronous publish path: the
+  # SLC has already received :ok by the time this runs. Because the SLC publishes
+  # fragments to a given consumer sequentially, the continue typically completes
+  # before the next fragment arrives, so it does not block steady-state throughput.
+
+  # Fast path: adaptive GC is disabled — skip all process_info/time calls.
+  defp maybe_garbage_collect(%State{gc_heap_threshold: nil} = state), do: state
+
+  defp maybe_garbage_collect(%State{gc_heap_threshold: threshold_bytes} = state) do
+    {:total_heap_size, heap_words} = :erlang.process_info(self(), :total_heap_size)
+    heap_bytes = heap_words * @word_size
+    now = System.monotonic_time(:millisecond)
+
+    if should_force_gc?(heap_bytes, threshold_bytes, state.last_forced_gc_at, now) do
+      :erlang.garbage_collect()
+      %{state | last_forced_gc_at: now}
+    else
+      state
+    end
+  end
+
+  @doc false
+  # Decide whether to force a full GC sweep: heap (bytes) must be over the
+  # threshold (bytes) AND at least @gc_min_interval_ms must have elapsed since the
+  # last forced GC. last_gc_at / now_ms are monotonic milliseconds; last_gc_at is
+  # nil if this consumer has never forced a GC (always fire on first over-threshold
+  # event). Passing explicit min_interval_ms enables deterministic unit tests.
+  @spec should_force_gc?(
+          non_neg_integer(),
+          non_neg_integer() | nil,
+          integer() | nil,
+          integer(),
+          non_neg_integer()
+        ) :: boolean()
+  def should_force_gc?(
+        heap_bytes,
+        threshold_bytes,
+        last_gc_at,
+        now_ms,
+        min_interval_ms \\ @gc_min_interval_ms
+      )
+
+  def should_force_gc?(_heap_bytes, nil, _last_gc_at, _now_ms, _min_interval_ms), do: false
+
+  def should_force_gc?(heap_bytes, threshold_bytes, last_gc_at, now_ms, min_interval_ms) do
+    heap_bytes > threshold_bytes and
+      (is_nil(last_gc_at) or now_ms - last_gc_at >= min_interval_ms)
+  end
+
   # A consumer process starts with buffering?=true before it has PG snapshot info (xmin, xmax, xip_list).
   # In this phase we have to buffer incoming txn fragments because we can't yet decide what to
   # do with the transaction: skip it or write it to the shape log.

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

@@ -111,7 +111,10 @@ defmodule Electric.Shapes.Consumer.Materializer do
     do: subscribe(%{stack_id: stack_id, shape_handle: shape_handle})
 
   def start_link(opts) do
-    GenServer.start_link(__MODULE__, opts, name: name(opts))
+    GenServer.start_link(__MODULE__, opts,
+      name: name(opts),
+      spawn_opt: Electric.StackConfig.spawn_opts(opts.stack_id, :consumer_materializer)
+    )
   end
 
   def init(opts) do

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

@@ -20,7 +20,10 @@ defmodule Electric.Shapes.Consumer.Snapshotter do
   end
 
   def start_link(config) when is_map(config) do
-    GenServer.start_link(__MODULE__, config, name: name(config))
+    GenServer.start_link(__MODULE__, config,
+      name: name(config),
+      spawn_opt: Electric.StackConfig.spawn_opts(config.stack_id, :consumer_snapshotter)
+    )
   end
 
   def init(config) do

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

@@ -47,7 +47,13 @@ defmodule Electric.Shapes.Consumer.State do
     # as any message arrives (activity), so at most one is ever live at a time.
     suspend_timer: nil,
     # How long after hibernation to suspend (in ms)
-    suspend_after: nil
+    suspend_after: nil,
+    # Monotonic millisecond timestamp of the last consumer-forced GC (nil if never).
+    # Used by hysteresis logic in maybe_garbage_collect/1 to cap forced-GC frequency.
+    last_forced_gc_at: nil,
+    # Adaptive-GC heap threshold (bytes) cached at consumer startup, or nil when
+    # disabled.
+    gc_heap_threshold: nil
   ]
 
   @type pg_snapshot() :: SnapshotQuery.pg_snapshot()
@@ -107,6 +113,7 @@ defmodule Electric.Shapes.Consumer.State do
           :shape_suspend_after,
           Electric.Config.default(:shape_suspend_after)
         ),
+      gc_heap_threshold: Electric.StackConfig.lookup(stack_id, :consumer_gc_heap_threshold, nil),
       buffering?: true
     }
   end

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

@@ -33,7 +33,8 @@ defmodule Electric.StackConfig do
       shape_suspend_after: Electric.Config.default(:shape_suspend_after),
       chunk_bytes_threshold: Electric.ShapeCache.LogChunker.default_chunk_size_threshold(),
       feature_flags: [],
-      process_spawn_opts: %{}
+      process_spawn_opts: %{},
+      consumer_gc_heap_threshold: Electric.Config.default(:consumer_gc_heap_threshold)
     ]
   end
 

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

@@ -154,6 +154,10 @@ defmodule Electric.StackSupervisor do
                        default: nil
                      ],
                      process_spawn_opts: [type: :map, default: %{}],
+                     consumer_gc_heap_threshold: [
+                       type: {:or, [:non_neg_integer, nil]},
+                       default: Electric.Config.default(:consumer_gc_heap_threshold)
+                     ],
                      consumer_partitions: [type: {:or, [:pos_integer, nil]}, default: nil]
                    ]
                  ],
@@ -357,6 +361,7 @@ defmodule Electric.StackSupervisor do
     shape_enable_suspend? = Keyword.fetch!(config.tweaks, :shape_enable_suspend?)
     shape_suspend_after = Keyword.fetch!(config.tweaks, :shape_suspend_after)
     process_spawn_opts = Keyword.fetch!(config.tweaks, :process_spawn_opts)
+    consumer_gc_heap_threshold = Keyword.fetch!(config.tweaks, :consumer_gc_heap_threshold)
 
     shape_cache_opts = [
       stack_id: stack_id
@@ -407,6 +412,7 @@ defmodule Electric.StackSupervisor do
            shape_enable_suspend?: shape_enable_suspend?,
            shape_suspend_after: shape_suspend_after,
            process_spawn_opts: process_spawn_opts,
+           consumer_gc_heap_threshold: consumer_gc_heap_threshold,
            feature_flags: Map.get(config, :feature_flags, [])
          ]},
         {Electric.AsyncDeleter,