fix(sync-service): shed live shape subscribers that stop draining their mailbox

.changeset/shed-slow-live-subscribers.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Shed live shape subscribers that stop draining their mailbox. A live `GET /v1/shape` request whose client can't keep up (a stalled or dead socket) while changes keep streaming would accumulate one `:new_changes` notification per transaction with no upper bound, pinning reference-counted binary memory until the node ran out of memory. The consumer now terminates a subscriber once its mailbox exceeds the watermark set by `ELECTRIC_SLOW_SUBSCRIBER_MAX_QUEUE_LEN` (default 10,000; always on); the client reconnects and resumes from its last offset.

packages/sync-service/config/runtime.exs

@@ -292,6 +292,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),
+  slow_subscriber_max_queue_len: env!("ELECTRIC_SLOW_SUBSCRIBER_MAX_QUEUE_LEN", :integer, 10_000),
   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),
+        slow_subscriber_max_queue_len: get_env(opts, :slow_subscriber_max_queue_len)
       ],
       manual_table_publishing?: get_env(opts, :manual_table_publishing?),
       shape_db_opts: [

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

@@ -117,6 +117,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,
+    # Message-queue length above which a live shape subscriber that has stopped
+    # draining its mailbox (e.g. a stalled/dead client socket) is shed, to avoid
+    # unbounded memory growth. Always a positive integer; shedding cannot be
+    # disabled.
+    slow_subscriber_max_queue_len: 10_000,
     ## 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

@@ -45,6 +45,16 @@ defmodule Electric.Shapes.Consumer do
   # fragment rate.
   @gc_min_interval_ms 1_000
 
+  # Safety net for live subscribers that stop draining their mailbox (e.g. a
+  # stalled or dead client socket). A healthy subscriber drains :new_changes
+  # promptly and sits near zero; this default only sheds pathologically
+  # backed-up subscribers well before they can OOM the node. Always a positive
+  # integer — shedding cannot be disabled. Tunable via
+  # ELECTRIC_SLOW_SUBSCRIBER_MAX_QUEUE_LEN (stack config key
+  # :slow_subscriber_max_queue_len). Used only as a fallback default when the
+  # stack config has not been seeded (e.g. in unit tests).
+  @default_slow_subscriber_max_queue_len 10_000
+
   @type initialize_shape_opts() :: %{
           :action => :create | :restore,
           optional(:otel_ctx) => OpenTelemetry.otel_ctx() | nil,
@@ -1012,6 +1022,13 @@ defmodule Electric.Shapes.Consumer do
           latest_log_offset :: LogOffset.t()
         ) :: :ok
   defp notify_clients_of_new_changes(state, latest_log_offset) do
+    max_queue_len =
+      Electric.StackConfig.lookup(
+        state.stack_id,
+        :slow_subscriber_max_queue_len,
+        @default_slow_subscriber_max_queue_len
+      )
+
     Registry.dispatch(
       Electric.StackSupervisor.registry_name(state.stack_id),
       state.shape_handle,
@@ -1020,10 +1037,39 @@ defmodule Electric.Shapes.Consumer do
           "Notifying ~#{length(registered)} clients about new changes to #{state.shape_handle}"
         end)
 
-        for {pid, ref} <- registered,
-            do: send(pid, {ref, :new_changes, latest_log_offset})
+        for {pid, ref} <- registered do
+          if subscriber_overloaded?(pid, max_queue_len) do
+            shed_slow_subscriber(pid, state.shape_handle, max_queue_len)
+          else
+            send(pid, {ref, :new_changes, latest_log_offset})
+          end
+        end
       end
     )
+
+    :ok
+  end
+
+  # A live subscriber (e.g. a `GET /v1/shape` request) that can't drain its
+  # mailbox — a stalled or dead client socket while changes keep streaming —
+  # would otherwise accumulate one `:new_changes` message per transaction
+  # without bound, pinning reference-counted binary until the node runs out of
+  # memory. Once its mailbox crosses the watermark we stop notifying it and
+  # terminate it; the client reconnects and resumes from its last offset.
+  defp subscriber_overloaded?(pid, max_queue_len) do
+    case Process.info(pid, :message_queue_len) do
+      {:message_queue_len, len} -> len > max_queue_len
+      nil -> false
+    end
+  end
+
+  defp shed_slow_subscriber(pid, shape_handle, max_queue_len) do
+    Logger.warning(
+      "Shedding slow subscriber #{inspect(pid)} for shape #{shape_handle}: " <>
+        "mailbox exceeded #{max_queue_len} pending messages"
+    )
+
+    Process.exit(pid, :kill)
   end
 
   @spec notify_materializer_of_new_changes(

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),
+      slow_subscriber_max_queue_len: Electric.Config.default(:slow_subscriber_max_queue_len)
     ]
   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)
                      ],
+                     slow_subscriber_max_queue_len: [
+                       type: :pos_integer,
+                       default: Electric.Config.default(:slow_subscriber_max_queue_len)
+                     ],
                      consumer_partitions: [type: {:or, [:pos_integer, nil]}, default: nil]
                    ]
                  ],
@@ -363,6 +367,9 @@ defmodule Electric.StackSupervisor do
     process_spawn_opts = Keyword.fetch!(config.tweaks, :process_spawn_opts)
     consumer_gc_heap_threshold = Keyword.fetch!(config.tweaks, :consumer_gc_heap_threshold)
 
+    slow_subscriber_max_queue_len =
+      Keyword.fetch!(config.tweaks, :slow_subscriber_max_queue_len)
+
     shape_cache_opts = [
       stack_id: stack_id
     ]
@@ -413,6 +420,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,
+           slow_subscriber_max_queue_len: slow_subscriber_max_queue_len,
            feature_flags: Map.get(config, :feature_flags, [])
          ]},
         {Electric.AsyncDeleter,

packages/sync-service/test/electric/shapes/consumer_test.exs

@@ -223,6 +223,161 @@ defmodule Electric.Shapes.ConsumerTest do
       [consumers: consumers]
     end
 
+    test "sheds a live subscriber blocked writing to a stalled socket", ctx do
+      # Regression: a live `GET /v1/shape` handler blocked in `Plug.Conn.chunk`
+      # writing to a stalled/dead client socket stops draining its mailbox while
+      # changes keep streaming. It accumulates one `{:new_changes, _}` message
+      # per transaction with no upper bound, pinning reference-counted binary
+      # until the node OOMs. The consumer must shed such a subscriber once its
+      # mailbox crosses a watermark.
+      #
+      # The stuck subscriber is blocked in a real `:gen_tcp.send` to a peer that
+      # never reads — the actual production scheduler state, `:waiting` in the
+      # `inet_reply` receive — so this also exercises that the consumer's
+      # `Process.info(_, :message_queue_len)` check does not itself stall on the
+      # blocked subscriber (which would freeze the per-shape consumer).
+      Electric.StackConfig.put(ctx.stack_id, :slow_subscriber_max_queue_len, 5)
+
+      xmin = snapshot_xmin(@shape_handle1, ctx)
+      base_lsn = lsn(@shape_handle1, ctx)
+      test_pid = self()
+
+      # A stalled "client": accepts the connection but never reads, so the
+      # subscriber's writes fill the buffers and block.
+      {:ok, listen} =
+        :gen_tcp.listen(0, [:binary, active: false, reuseaddr: true, recbuf: 1024])
+
+      {:ok, lport} = :inet.port(listen)
+      on_exit(fn -> :gen_tcp.close(listen) end)
+
+      acceptor =
+        spawn(fn ->
+          {:ok, _accepted} = :gen_tcp.accept(listen)
+          Process.sleep(:infinity)
+        end)
+
+      on_exit(fn -> Process.exit(acceptor, :kill) end)
+
+      subscriber =
+        spawn(fn ->
+          {:ok, sock} =
+            :gen_tcp.connect({127, 0, 0, 1}, lport, [
+              :binary,
+              active: false,
+              sndbuf: 1024,
+              buffer: 1024,
+              high_watermark: 2048,
+              low_watermark: 1024,
+              send_timeout: :infinity
+            ])
+
+          {:ok, _} = Registry.register(ctx.registry, @shape_handle1, make_ref())
+          send(test_pid, :subscribed)
+
+          # Block forever writing to the stalled socket, exactly like a live
+          # handler stuck in Plug.Conn.chunk. The process never drains its
+          # :new_changes mailbox while parked here.
+          chunk = :binary.copy(<<0>>, 1_000_000)
+          Enum.each(Stream.cycle([chunk]), &:gen_tcp.send(sock, &1))
+        end)
+
+      monitor = Process.monitor(subscriber)
+      assert_receive :subscribed, @receive_timeout
+      # Let it get stuck in the blocking send before changes start streaming.
+      Process.sleep(50)
+
+      # Stream a steady series of transactions that all match the shape.
+      for i <- 1..50 do
+        txn_lsn = Lsn.increment(base_lsn, i)
+
+        txn =
+          complete_txn_fragment(xmin + i, txn_lsn, [
+            %Changes.NewRecord{
+              relation: {"public", "test_table"},
+              record: %{"id" => "1"},
+              log_offset: LogOffset.new(txn_lsn, 0)
+            }
+          ])
+
+        assert :ok = ShapeLogCollector.handle_event(txn, ctx.stack_id)
+      end
+
+      # The stuck subscriber must be shed once its mailbox crosses the watermark.
+      # This arriving also proves the consumer's mailbox check did not block on
+      # the socket-stuck subscriber — otherwise the shed would never run.
+      assert_receive {:DOWN, ^monitor, :process, ^subscriber, _reason}, @receive_timeout
+
+      # And the consumer must still be alive and notifying afterwards: register a
+      # healthy subscriber and confirm it receives the next change.
+      healthy =
+        spawn(fn ->
+          {:ok, _} = Registry.register(ctx.registry, @shape_handle1, ref = make_ref())
+          send(test_pid, :healthy_subscribed)
+
+          receive do
+            {^ref, :new_changes, offset} -> send(test_pid, {:healthy_notified, offset})
+          end
+        end)
+
+      on_exit(fn -> Process.exit(healthy, :kill) end)
+      assert_receive :healthy_subscribed, @receive_timeout
+
+      healthy_lsn = Lsn.increment(base_lsn, 100)
+
+      healthy_txn =
+        complete_txn_fragment(xmin + 100, healthy_lsn, [
+          %Changes.NewRecord{
+            relation: {"public", "test_table"},
+            record: %{"id" => "1"},
+            log_offset: LogOffset.new(healthy_lsn, 0)
+          }
+        ])
+
+      assert :ok = ShapeLogCollector.handle_event(healthy_txn, ctx.stack_id)
+      assert_receive {:healthy_notified, _offset}, @receive_timeout
+    end
+
+    test "does not shed a healthy subscriber whose backlog stays under the watermark", ctx do
+      # The shedding safety net must not disturb normal live clients that drain
+      # their mailbox and never approach the watermark.
+      Electric.StackConfig.put(ctx.stack_id, :slow_subscriber_max_queue_len, 100)
+
+      xmin = snapshot_xmin(@shape_handle1, ctx)
+      base_lsn = lsn(@shape_handle1, ctx)
+      test_pid = self()
+
+      # A subscriber that drains every notification promptly.
+      subscriber =
+        spawn(fn ->
+          {:ok, _} = Registry.register(ctx.registry, @shape_handle1, ref = make_ref())
+          send(test_pid, :subscribed)
+          drain = fn drain -> receive(do: ({^ref, :new_changes, _} -> drain.(drain))) end
+          drain.(drain)
+        end)
+
+      monitor = Process.monitor(subscriber)
+      on_exit(fn -> Process.exit(subscriber, :kill) end)
+      assert_receive :subscribed, @receive_timeout
+
+      for i <- 1..50 do
+        txn_lsn = Lsn.increment(base_lsn, i)
+
+        txn =
+          complete_txn_fragment(xmin + i, txn_lsn, [
+            %Changes.NewRecord{
+              relation: {"public", "test_table"},
+              record: %{"id" => "1"},
+              log_offset: LogOffset.new(txn_lsn, 0)
+            }
+          ])
+
+        assert :ok = ShapeLogCollector.handle_event(txn, ctx.stack_id)
+      end
+
+      refute_receive {:DOWN, ^monitor, :process, ^subscriber, _reason}, 200
+      assert Process.alive?(subscriber)
+    end
+
     test "appends to log when xid >= xmin", ctx do
       xid = 150
       xmin = snapshot_xmin(@shape_handle1, ctx)