fix(sync-service): reduce live request wait memory (#4680)

## Motivation

Live shape requests spend most of their long-poll lifetime parked in a
receive while waiting for either a shape change or timeout. Under high
concurrency, parking inside the larger request-handling function kept
more heap live than the small wait state actually needed, which made the
per-request memory floor significant.

## Changes

- Move the blocking live-request receive into a narrow
`wait_for_live_change/2` helper that only holds the change reference and
timeout while parked.
- Convert receive outcomes into small event values and handle response
construction after the wait returns.
- Keep garbage collection immediately before the parked receive, but run
it inside the narrow helper so collection happens with a smaller root
set.
- Remove a dead `SystemMonitor` timer fallback branch that Elixir 1.20
reports as unreachable under warnings-as-errors.
- Add changesets for sync-service and electric-telemetry.

## Decisions

This keeps the waiter in the existing request process instead of
introducing a central waiter registry. That avoids adding a new
register/deregister bottleneck while still reducing the heap retained by
parked live requests. The existing behavior is preserved for new
changes, shape rotation, out-of-bounds checks, read-only timeout
handling, and ordinary no-change responses.

Author: icehaunter

.changeset/reduce-live-request-wait-memory.md

@@ -0,0 +1,8 @@
+---
+'@core/sync-service': patch
+'@core/electric-telemetry': patch
+---
+
+Reduce memory retained by live shape requests while they wait for changes.
+
+Remove a dead SystemMonitor timer branch that fails warnings-as-errors on Elixir 1.20.

packages/electric-telemetry/lib/electric/telemetry/system_monitor.ex

@@ -170,6 +170,4 @@ defmodule ElectricTelemetry.SystemMonitor do
           Process.send_after(self(), @garbage_collect_message, @garbage_collect_interval)
     }
   end
-
-  defp schedule_garbage_collect(state), do: state
 end

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

@@ -859,45 +859,97 @@ defmodule Electric.Shapes.Api do
     %{
       new_changes_ref: ref,
       handle: shape_handle,
-      api: %{long_poll_timeout: long_poll_timeout} = api
+      api: %{long_poll_timeout: long_poll_timeout}
     } = request
 
     Logger.debug("Client #{inspect(self())} is waiting for changes to #{shape_handle}")
 
-    # Bandit reuses handler processes across requests. This process may have accumulated
-    # garbage from previous requests or from building the response for this request.
-    # Before blocking in receive (potentially for up to long_poll_timeout seconds),
-    # run garbage collection so we don't hold onto memory while idle.
-    # Combined with handler_fullsweep_after config, this helps keep handler memory usage low.
+    ref
+    |> wait_for_live_change(long_poll_timeout)
+    |> handle_live_change_result(request)
+  end
+
+  defp wait_for_live_change(ref, long_poll_timeout) do
+    # Bandit reuses handler processes across requests. Entering this narrow helper before
+    # collecting keeps the parked receive from retaining the larger request-handling frame.
     :erlang.garbage_collect()
 
     receive do
       {^ref, :new_changes, latest_log_offset} ->
-        # Stream new log since currently "held" offset
-        %{request | last_offset: latest_log_offset}
+        {:new_changes, latest_log_offset}
+
+      {^ref, :shape_rotation, new_handle} ->
+        {:shape_rotation, new_handle}
+
+      {^ref, :shape_rotation} ->
+        :shape_rotation
+
+      {^ref, :out_of_bounds_timeout} ->
+        :out_of_bounds_timeout
+    after
+      long_poll_timeout ->
+        :long_poll_timeout
+    end
+  end
+
+  defp handle_live_change_result({:new_changes, latest_log_offset}, request) do
+    # Stream new log since currently "held" offset
+    %{request | last_offset: latest_log_offset}
+    |> determine_global_last_seen_lsn()
+    |> determine_log_chunk_offset()
+    |> determine_up_to_date()
+    |> do_serve_shape_log()
+  end
+
+  defp handle_live_change_result({:shape_rotation, new_handle}, request) do
+    error = Api.Error.must_refetch()
+
+    Response.error(request, error.message,
+      handle: new_handle,
+      status: error.status
+    )
+  end
+
+  defp handle_live_change_result(:shape_rotation, request) do
+    error = Api.Error.must_refetch()
+    Response.error(request, error.message, status: error.status)
+  end
+
+  defp handle_live_change_result(:out_of_bounds_timeout, request) do
+    %{api: api, handle: shape_handle} = request
+
+    Logger.debug(fn ->
+      "Client #{inspect(self())} timed out waiting for " <>
+        "changes to #{shape_handle} (out-of-bounds check)"
+    end)
+
+    case check_for_disk_updates(request) do
+      {:updated, new_offset} ->
+        %{request | last_offset: new_offset}
         |> determine_global_last_seen_lsn()
         |> determine_log_chunk_offset()
         |> determine_up_to_date()
         |> do_serve_shape_log()
 
-      {^ref, :shape_rotation, new_handle} ->
-        error = Api.Error.must_refetch()
-
-        Response.error(request, error.message,
-          handle: new_handle,
-          status: error.status
-        )
+      _ ->
+        Response.invalid_request(api, errors: @offset_out_of_bounds)
+    end
+  end
 
-      {^ref, :shape_rotation} ->
-        error = Api.Error.must_refetch()
-        Response.error(request, error.message, status: error.status)
+  defp handle_live_change_result(:long_poll_timeout, request) do
+    %{api: api} = request
+    request = update_attrs(request, %{ot_is_long_poll_timeout: true})
+    status = Electric.StatusMonitor.status(api.stack_id)
 
-      {^ref, :out_of_bounds_timeout} ->
-        Logger.debug(fn ->
-          "Client #{inspect(self())} timed out waiting for " <>
-            "changes to #{shape_handle} (out-of-bounds check)"
-        end)
+    cond do
+      request.read_only? or status.shape == :read_only ->
+        # Align the request flag with the current runtime status so that
+        # downstream functions (determine_log_chunk_offset, get_merged_log_stream,
+        # etc.) use the correct read-only strategy.
+        request = %{request | read_only?: true}
 
+        # No consumer is running (or it stopped), so check if the
+        # active instance has flushed new data to disk.
         case check_for_disk_updates(request) do
           {:updated, new_offset} ->
             %{request | last_offset: new_offset}
@@ -907,45 +959,19 @@ defmodule Electric.Shapes.Api do
             |> do_serve_shape_log()
 
           _ ->
-            Response.invalid_request(api, errors: @offset_out_of_bounds)
-        end
-    after
-      long_poll_timeout ->
-        request = update_attrs(request, %{ot_is_long_poll_timeout: true})
-        status = Electric.StatusMonitor.status(api.stack_id)
-
-        cond do
-          request.read_only? or status.shape == :read_only ->
-            # Align the request flag with the current runtime status so that
-            # downstream functions (determine_log_chunk_offset, get_merged_log_stream,
-            # etc.) use the correct read-only strategy.
-            request = %{request | read_only?: true}
-
-            # No consumer is running (or it stopped), so check if the
-            # active instance has flushed new data to disk.
-            case check_for_disk_updates(request) do
-              {:updated, new_offset} ->
-                %{request | last_offset: new_offset}
-                |> determine_global_last_seen_lsn()
-                |> determine_log_chunk_offset()
-                |> determine_up_to_date()
-                |> do_serve_shape_log()
-
-              _ ->
-                request
-                |> determine_global_last_seen_lsn()
-                |> no_change_response()
-            end
-
-          status.shape == :up ->
             request
             |> determine_global_last_seen_lsn()
             |> no_change_response()
-
-          true ->
-            message = Electric.StatusMonitor.timeout_message(api.stack_id)
-            Response.error(request, message, status: 503, retry_after: 10)
         end
+
+      status.shape == :up ->
+        request
+        |> determine_global_last_seen_lsn()
+        |> no_change_response()
+
+      true ->
+        message = Electric.StatusMonitor.timeout_message(api.stack_id)
+        Response.error(request, message, status: 503, retry_after: 10)
     end
   end