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/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/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

@@ -434,39 +434,174 @@ 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 = start_write_watchdog(stack_id, response)
 
     {conn, bytes_sent} =
-      response.body
-      |> Enum.reduce_while({conn, 0}, fn chunk, {conn, bytes_sent} ->
-        chunk_size = IO.iodata_length(chunk)
-
-        OpenTelemetry.with_span(
-          "shape_get.plug.stream_chunk",
-          Map.put(sample_rate_attrs, "chunk_size", chunk_size),
-          stack_id,
-          fn ->
-            case Plug.Conn.chunk(conn, chunk) do
-              {:ok, conn} ->
-                {:cont, {conn, bytes_sent + chunk_size}}
-
-              {:error, reason} when reason in ["closed", :closed] ->
-                error_str = "Connection closed unexpectedly while streaming response"
-                conn = Plug.Conn.assign(conn, :error_str, error_str)
-                {:halt, {conn, bytes_sent}}
-
-              {:error, reason} ->
-                error_str = "Error while streaming response: #{inspect(reason)}"
-                Logger.error(error_str)
-                conn = Plug.Conn.assign(conn, :error_str, error_str)
-                {:halt, {conn, bytes_sent}}
+      try do
+        response.body
+        |> Enum.reduce_while({conn, 0}, fn chunk, {conn, bytes_sent} ->
+          chunk_size = IO.iodata_length(chunk)
+
+          OpenTelemetry.with_span(
+            "shape_get.plug.stream_chunk",
+            Map.put(sample_rate_attrs, "chunk_size", chunk_size),
+            stack_id,
+            fn ->
+              case write_in_bounded_pieces(conn, chunk, chunk_size, watchdog) do
+                {:ok, conn} ->
+                  {:cont, {conn, bytes_sent + chunk_size}}
+
+                {:error, reason} when reason in ["closed", :closed] ->
+                  error_str = "Connection closed unexpectedly while streaming response"
+                  conn = Plug.Conn.assign(conn, :error_str, error_str)
+                  {:halt, {conn, bytes_sent}}
+
+                {:error, reason} ->
+                  error_str = "Error while streaming response: #{inspect(reason)}"
+                  Logger.error(error_str)
+                  conn = Plug.Conn.assign(conn, :error_str, error_str)
+                  {:halt, {conn, bytes_sent}}
+              end
             end
-          end
-        )
-      end)
+          )
+        end)
+      after
+        watchdog && Kernel.send(watchdog, :stop)
+      end
 
     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()
+
+  # A serve whose client stops accepting data blocks inside the socket write
+  # and cannot recover on its own: the TCP send timeout only catches a write
+  # that is *fully* blocked for its whole window, so a client draining at a
+  # trickle — or a proxy buffering for a vanished client — can hold the
+  # serve, everything it pins, and its file descriptor forever. Such serves
+  # never complete, so they emit no telemetry, and they accumulate with
+  # connection count (reproduced in
+  # test/integration/stalled_serve_memory_test.exs).
+  #
+  # The watchdog is a small companion process that terminates the serve when
+  # a single bounded socket write (≤ @socket_write_bytes) fails to complete
+  # within `:stalled_serve_timeout`. Write completion is quantized by the OS:
+  # a blocked write may only complete after the kernel send buffer frees a
+  # substantial fraction of its capacity, so the effective contract is that a
+  # healthy client must drain roughly one OS send buffer per timeout window
+  # (e.g. ~10 KB/s for a megabyte-class buffer and a 60-second window, and
+  # far less on stacks that signal writability at finer granularity). The
+  # timer is armed only while a write is in flight, so a serve idling between
+  # body elements (a live long-poll hold, an SSE stream waiting for changes)
+  # is never at risk. The terminated client can reconnect and resume from its
+  # last offset.
+  defp start_write_watchdog(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.
+    case Electric.StackConfig.lookup(
+           stack_id,
+           :stalled_serve_timeout,
+           Electric.Config.default(:stalled_serve_timeout)
+         ) do
+      timeout when is_integer(timeout) and timeout > 0 ->
+        handler = self()
+        shape_handle = response.handle
+
+        {:ok, watchdog} =
+          Task.start(fn ->
+            # The label groups these processes under a low-cardinality
+            # process_type in the per-process telemetry, rather than leaving
+            # them anonymous — serves and their guards should be visible.
+            # Logger metadata is set explicitly: this process does not inherit
+            # the handler's, and the reap warning below is the primary signal
+            # an otherwise-invisible stalled serve emits, so it must carry
+            # enough to correlate (which stack, which shape).
+            Process.set_label({:serve_watchdog, shape_handle})
+            Logger.metadata(stack_id: stack_id, shape_handle: shape_handle)
+            ref = Process.monitor(handler)
+            write_watchdog_loop(handler, ref, timeout, :idle)
+          end)
+
+        watchdog
+
+      # 0 (or any non-positive value) disables reaping.
+      _disabled ->
+        nil
+    end
+  end
+
+  defp write_watchdog_loop(handler, ref, timeout, writing?) do
+    receive do
+      :write_start ->
+        write_watchdog_loop(handler, ref, timeout, :writing)
+
+      :write_done ->
+        write_watchdog_loop(handler, ref, timeout, :idle)
+
+      :stop ->
+        :ok
+
+      {:DOWN, ^ref, :process, ^handler, _reason} ->
+        :ok
+    after
+      watchdog_wait(writing?, timeout) ->
+        Logger.warning(
+          "Terminating stalled shape response serve: client accepted no data " <>
+            "for #{timeout}ms"
+        )
+
+        Process.exit(handler, :kill)
+    end
+  end
+
+  defp watchdog_wait(:writing, timeout), do: timeout
+  defp watchdog_wait(:idle, _timeout), do: :infinity
+
+  # 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, watchdog) do
+    watchdog && Kernel.send(watchdog, :write_start)
+    result = Plug.Conn.chunk(conn, data)
+    watchdog && Kernel.send(watchdog, :write_done)
+    result
+  end
+
   def etag(response, opts \\ [])
 
   # When response contains no changes, in order to uniquely identify it and avoid

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
 

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

@@ -158,6 +158,10 @@ defmodule Electric.StackSupervisor do
                        type: {:or, [:non_neg_integer, nil]},
                        default: Electric.Config.default(:consumer_gc_heap_threshold)
                      ],
+                     stalled_serve_timeout: [
+                       type: :non_neg_integer,
+                       default: Electric.Config.default(:stalled_serve_timeout)
+                     ],
                      consumer_partitions: [type: {:or, [:pos_integer, nil]}, default: nil]
                    ]
                  ],
@@ -362,6 +366,7 @@ defmodule Electric.StackSupervisor do
     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)
+    stalled_serve_timeout = Keyword.fetch!(config.tweaks, :stalled_serve_timeout)
 
     shape_cache_opts = [
       stack_id: stack_id
@@ -413,6 +418,7 @@ defmodule Electric.StackSupervisor do
            shape_suspend_after: shape_suspend_after,
            process_spawn_opts: process_spawn_opts,
            consumer_gc_heap_threshold: consumer_gc_heap_threshold,
+           stalled_serve_timeout: stalled_serve_timeout,
            feature_flags: Map.get(config, :feature_flags, [])
          ]},
         {Electric.AsyncDeleter,

packages/sync-service/test/integration/stalled_serve_memory_test.exs

@@ -4,7 +4,7 @@ defmodule Electric.Integration.StalledServeMemoryTest do
   log chunk (~`chunk_bytes_threshold`, 10 MB by default) per connection for an
   unbounded time.
 
-  Reproduces a production OOM (2026-07-01): a bulk write woke a fleet of live
+  Reproduces a production OOM (see #4708): a bulk write woke a fleet of live
   long-pollers into multi-chunk catch-up serves; connections whose clients
   stalled kept their entire in-flight chunk pinned — once in the handler
   process and once in the socket's driver queue — with nothing to reap them.