fix(sync-service): terminate serves whose clients stop accepting data

.changeset/reap-stalled-serves.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Terminate shape response serves whose clients stop accepting data. Serves to stalled clients never complete and are invisible to request telemetry, while pinning response memory and a file descriptor each for as long as the client's connection survives — a population of them exhausted a production node. Response bodies are now written to the socket in pieces of at most 256 KiB, and a watchdog terminates the serve when a single piece fails to complete within `ELECTRIC_STALLED_SERVE_TIMEOUT` (default 60s; 0 disables). A healthy client only needs to drain roughly one OS send buffer per timeout window to stay clear of the deadline, and a terminated client can reconnect and resume from its last offset. Oversized single body elements (e.g. one very large row) no longer enter the socket driver queue whole.

packages/sync-service/config/runtime.exs

@@ -294,6 +294,7 @@ config :electric,
   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),
+  stalled_serve_timeout: env!("ELECTRIC_STALLED_SERVE_TIMEOUT", :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/admission_control.ex

@@ -54,6 +54,14 @@ defmodule Electric.AdmissionControl do
     defp tuple_pos(unquote(kind)), do: unquote(pos)
   end
 
+  @doc """
+  The process-dictionary key under which a request handler stashes its held
+  permit as `{stack_id, kind}`. Owned here so that other components that
+  must inspect or account for a handler's permit (e.g. the serve watchdog
+  releasing a killed handler's permit) share one definition.
+  """
+  def permit_pd_key, do: {__MODULE__, :permit}
+
   @doc """
   Try to acquire a permit for the given stack_id.
 

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

@@ -157,7 +157,8 @@ defmodule Electric.Application do
         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),
-        consumer_gc_heap_threshold: get_env(opts, :consumer_gc_heap_threshold)
+        consumer_gc_heap_threshold: get_env(opts, :consumer_gc_heap_threshold),
+        stalled_serve_timeout: get_env(opts, :stalled_serve_timeout)
       ],
       manual_table_publishing?: get_env(opts, :manual_table_publishing?),
       shape_db_opts: [

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

@@ -120,6 +120,11 @@ defmodule Electric.Config do
     # Heap-size threshold (in BYTES) above which a consumer runs :erlang.garbage_collect()
     # after processing a transaction fragment.
     consumer_gc_heap_threshold: nil,
+    # Terminate a shape response serve when a single socket write makes no
+    # progress for this long (ms). Catches serves to clients that stop
+    # accepting data, which otherwise pin memory and a file descriptor
+    # indefinitely. 0 disables reaping.
+    stalled_serve_timeout: :timer.seconds(60),
     ## 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/plug/serve_shape_plug.ex

@@ -43,7 +43,7 @@ defmodule Electric.Plug.ServeShapePlug do
 
   require Logger
 
-  @admission_permit_key {__MODULE__, :admission_permit}
+  @admission_permit_key Electric.AdmissionControl.permit_pd_key()
   @subquery_compaction_error "can't be enabled for shapes with subqueries"
 
   # These plugs are invoked inside the `call/2` function below, after `conn` has been preloaded with

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

@@ -74,6 +74,13 @@ defmodule Electric.Shapes.Api.Encoder.JSON do
   @max_batch_items 500
   @max_batch_bytes 256 * 1024
 
+  @doc """
+  Upper bound in bytes for a single encoded response body element (a batch of
+  log items). Consumers of encoded streams may size their write units to this:
+  elements at or under it can be written as-is.
+  """
+  def max_batch_bytes, do: @max_batch_bytes
+
   defp to_json_stream(items) do
     Stream.concat([
       [@json_list_start],

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

@@ -1,6 +1,7 @@
 defmodule Electric.Shapes.Api.Response do
   alias Electric.Plug.Utils
   alias Electric.Shapes.Api
+  alias Electric.Shapes.Api.ServeWatchdog
   alias Electric.Shapes.Shape
   alias Electric.Telemetry.OpenTelemetry
 
@@ -434,6 +435,7 @@ defmodule Electric.Shapes.Api.Response do
     sample_rate_attrs = Electric.Plug.TraceContextPlug.sample_rate_attrs(conn, status)
 
     conn = Plug.Conn.send_chunked(conn, status)
+    watchdog = watchdog_context(conn, stack_id, response)
 
     {conn, bytes_sent} =
       response.body
@@ -445,7 +447,7 @@ defmodule Electric.Shapes.Api.Response do
           Map.put(sample_rate_attrs, "chunk_size", chunk_size),
           stack_id,
           fn ->
-            case Plug.Conn.chunk(conn, chunk) do
+            case write_in_bounded_pieces(conn, chunk, chunk_size, watchdog) do
               {:ok, conn} ->
                 {:cont, {conn, bytes_sent + chunk_size}}
 
@@ -467,6 +469,91 @@ defmodule Electric.Shapes.Api.Response do
     Plug.Conn.assign(conn, :streaming_bytes_sent, bytes_sent)
   end
 
+  # Socket writes are split into pieces of at most this size. Bounding the
+  # write unit bounds the response data queued in the socket's driver queue
+  # at any moment and — because a bounded write to a live client completes
+  # once the transport buffers drain — gives the stall watchdog below a
+  # progress signal: each completed piece is proof the client accepted data.
+  #
+  # Derived from the encoder's batch cap so that ordinary body elements are
+  # written as-is (no flattening or splitting) and only oversized single
+  # items (e.g. one very large row) are split. Making pieces smaller would
+  # not sharpen the watchdog: write completion is quantized by the OS at up
+  # to a kernel send buffer of drain, which dwarfs the piece size.
+  @socket_write_bytes Electric.Shapes.Api.Encoder.JSON.max_batch_bytes()
+
+  # See Electric.Shapes.Api.ServeWatchdog for the design rationale. The
+  # context carries what arming each write's deadline needs; `nil` disables
+  # reaping for this serve: when the timeout is not positive, when no
+  # watchdog is running for the stack, or when the adapter does not run the
+  # plug in the socket-owning process (only Bandit does; under proxying
+  # adapters like Plug.Test or Cowboy, writes complete without touching a
+  # socket, so a deadline would measure nothing and a kill would hit a
+  # process whose cleanup semantics we do not own).
+  defp watchdog_context(%Plug.Conn{adapter: {Bandit.Adapter, _}}, stack_id, response) do
+    # Fallback for stacks whose seed config omits the key (e.g. minimal
+    # unit-test stacks); Electric.Config is the single source of truth.
+    timeout =
+      Electric.StackConfig.lookup(
+        stack_id,
+        :stalled_serve_timeout,
+        Electric.Config.default(:stalled_serve_timeout)
+      )
+
+    with true <- is_integer(timeout) and timeout > 0,
+         watchdog when is_pid(watchdog) <- ServeWatchdog.resolve(stack_id) do
+      {watchdog, response.handle, timeout}
+    else
+      _ -> nil
+    end
+  end
+
+  defp watchdog_context(%Plug.Conn{}, _stack_id, _response), do: nil
+
+  # Write one response body element as a sequence of bounded socket writes,
+  # notifying the watchdog around each one. Elements within the bound — the
+  # common case, as the bound is derived from the encoder's batch cap — are
+  # written directly without flattening.
+  defp write_in_bounded_pieces(conn, chunk, chunk_size, watchdog)
+       when chunk_size <= @socket_write_bytes do
+    guarded_chunk(conn, chunk, watchdog)
+  end
+
+  defp write_in_bounded_pieces(conn, chunk, _chunk_size, watchdog) do
+    chunk
+    |> IO.iodata_to_binary()
+    |> write_pieces(conn, watchdog)
+  end
+
+  defp write_pieces(<<piece::binary-size(@socket_write_bytes), rest::binary>>, conn, watchdog)
+       when byte_size(rest) > 0 do
+    case guarded_chunk(conn, piece, watchdog) do
+      {:ok, conn} -> write_pieces(rest, conn, watchdog)
+      error -> error
+    end
+  end
+
+  defp write_pieces(piece, conn, watchdog) do
+    guarded_chunk(conn, piece, watchdog)
+  end
+
+  defp guarded_chunk(conn, data, nil) do
+    Plug.Conn.chunk(conn, data)
+  end
+
+  defp guarded_chunk(conn, data, {watchdog, shape_handle, timeout}) do
+    ref = ServeWatchdog.arm(watchdog, shape_handle, timeout)
+
+    # Disarming in `after` keeps the stamped ref from outliving this write on
+    # any exit path; a stale stamp would let a later firing timer kill a
+    # handler that is no longer inside the write it was armed for.
+    try do
+      Plug.Conn.chunk(conn, data)
+    after
+      ServeWatchdog.disarm(ref)
+    end
+  end
+
   def etag(response, opts \\ [])
 
   # When response contains no changes, in order to uniquely identify it and avoid

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

@@ -0,0 +1,183 @@
+defmodule Electric.Shapes.Api.ServeWatchdog do
+  @moduledoc """
+  Terminates shape response serves whose clients stop accepting data.
+
+  A serve writing to a socket whose peer stops draining blocks inside the
+  socket write and cannot recover on its own: the kernel-level TCP send
+  timeout fires only when a driver-level send makes no progress for its
+  entire window, and in practice a connection draining even a trickle keeps
+  resetting it. Such serves never complete, so they emit no telemetry, and
+  each pins its in-flight response data and a file descriptor for as long as
+  the connection survives — a population of them accumulates with connection
+  count (reproduced in test/integration/stalled_serve_memory_test.exs).
+
+  ## Mechanism
+
+  Around each bounded socket write, the request handler arms a BIF timer
+  (`:erlang.start_timer/3`) addressed at this server and cancels it when the
+  write completes. Timers are per-scheduler O(1) and a cancelled timer never
+  sends anything, so this server's load is proportional to *stalls* — the
+  rare event it exists for — rather than to write throughput, and deadlines
+  fire exactly rather than with sweep slack.
+
+  A fired timer is only acted on if the handler's process dictionary still
+  carries that exact timer ref: the handler stamps the ref when arming and
+  deletes it when the write completes, so a timeout message racing a
+  completed write (or landing after the same process has moved on to another
+  request on a keepalive connection) is a no-op rather than a wrong kill.
+
+  ## What a kill must clean up
+
+  `Process.exit(pid, :kill)` destroys the handler without running another
+  instruction, so this server takes over the accounting the handler can no
+  longer do. From the same dictionary snapshot used for the ref check it
+  reads the handler's admission permit (see
+  `Electric.AdmissionControl.permit_pd_key/0`), then monitors, kills, and on
+  observing the `:killed` exit releases the permit and emits a
+  `[:electric, :plug, :serve_shape, :reaped]` telemetry event with the stack
+  and shape — otherwise reaped serves would be invisible to request metrics
+  (the handler's own telemetry emission dies with it) and would permanently
+  leak their admission slot. There is a theoretical window in which the
+  handler completes its entire serve and releases its own permit between the
+  dictionary snapshot and the kill landing — accepted, like the analogous
+  arm/fire boundary race, as it requires the whole remainder of a serve to
+  execute in the microseconds after its current write has already overrun a
+  deadline measured in tens of seconds.
+
+  The handler's root OTel span is not ended by the kill; orphaned spans are
+  dropped by the SDK's sweeper.
+
+  ## Scope
+
+  Serves are only guarded under adapters where the plug runs in the
+  socket-owning process (Bandit HTTP/1) — see `Response.send_stream/2` for
+  the gate. Arming toward a stack without a running server is a silent
+  no-op, so serving degrades to unguarded rather than failing; if this
+  server restarts, in-flight serves re-arm toward it on their next write.
+  """
+  use GenServer
+
+  require Logger
+
+  @write_ref_key {__MODULE__, :write_ref}
+
+  def start_link(opts) do
+    stack_id = Keyword.fetch!(opts, :stack_id)
+    GenServer.start_link(__MODULE__, stack_id, name: name(stack_id))
+  end
+
+  def name(stack_id) do
+    Electric.ProcessRegistry.name(stack_id, __MODULE__)
+  end
+
+  @doc """
+  Resolve the watchdog for a stack, or nil if none is running — including
+  when the stack's process registry itself is absent, so callers uniformly
+  degrade to unguarded serving.
+  """
+  def resolve(stack_id) do
+    GenServer.whereis(name(stack_id))
+  rescue
+    ArgumentError -> nil
+  end
+
+  @doc """
+  Arm a deadline for a socket write the calling process is about to make.
+
+  Returns the timer ref to pass to `disarm/1` once the write completes.
+  """
+  def arm(watchdog, shape_handle, timeout_ms) when is_pid(watchdog) do
+    ref = :erlang.start_timer(timeout_ms, watchdog, {:stalled_write, self(), shape_handle})
+    Process.put(@write_ref_key, ref)
+    ref
+  end
+
+  @doc """
+  Disarm a deadline after the write completed.
+
+  Clearing the stamped ref before cancelling matters: with an asynchronous
+  cancel, the timer may already have fired, and the stale stamp would
+  otherwise make the in-flight timeout message kill a handler whose write
+  completed.
+  """
+  def disarm(ref) do
+    Process.delete(@write_ref_key)
+    :erlang.cancel_timer(ref, async: true, info: false)
+    :ok
+  end
+
+  @impl GenServer
+  def init(stack_id) do
+    Process.set_label({:serve_watchdog, stack_id})
+    Logger.metadata(stack_id: stack_id)
+    {:ok, %{stack_id: stack_id, pending_kills: %{}}}
+  end
+
+  @impl GenServer
+  def handle_info({:timeout, ref, {:stalled_write, pid, shape_handle}}, state) do
+    state =
+      case Process.info(pid, :dictionary) do
+        {:dictionary, dict} ->
+          if List.keyfind(dict, @write_ref_key, 0) == {@write_ref_key, ref} do
+            kill(pid, shape_handle, dict, state)
+          else
+            # The write completed (or the process moved on to another
+            # request) before the cancel could beat the firing timer.
+            state
+          end
+
+        # The handler died on its own; nothing to do.
+        nil ->
+          state
+      end
+
+    {:noreply, state}
+  end
+
+  def handle_info({:DOWN, mref, :process, _pid, reason}, state) do
+    {entry, pending_kills} = Map.pop(state.pending_kills, mref)
+
+    if entry && reason == :killed do
+      release_permit(entry.permit)
+
+      :telemetry.execute([:electric, :plug, :serve_shape, :reaped], %{count: 1}, %{
+        stack_id: state.stack_id,
+        shape_handle: entry.shape_handle
+      })
+    end
+
+    {:noreply, %{state | pending_kills: pending_kills}}
+  end
+
+  defp kill(pid, shape_handle, dict, state) do
+    permit =
+      case List.keyfind(dict, Electric.AdmissionControl.permit_pd_key(), 0) do
+        {_key, permit} -> permit
+        nil -> nil
+      end
+
+    Logger.warning(
+      "Terminating stalled shape response serve: client did not accept an " <>
+        "in-flight write within its deadline",
+      shape_handle: shape_handle
+    )
+
+    mref = Process.monitor(pid)
+
+    # The handler is blocked inside a socket write and cannot process
+    # messages, so only an untrappable exit can terminate it. The permit
+    # release and reap telemetry happen when the resulting :killed exit is
+    # observed, standing in for the cleanup the killed handler can no
+    # longer run itself.
+    Process.exit(pid, :kill)
+
+    %{
+      state
+      | pending_kills:
+          Map.put(state.pending_kills, mref, %{permit: permit, shape_handle: shape_handle})
+    }
+  end
+
+  defp release_permit({stack_id, kind}), do: Electric.AdmissionControl.release(stack_id, kind)
+  defp release_permit(nil), do: :ok
+end

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

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