From 92ba4f552c5f46e816f39656a3b4c82c6b992ddb Mon Sep 17 00:00:00 2001 From: rob Date: Wed, 1 Jul 2026 17:28:14 +0100 Subject: [PATCH 1/2] fix(sync-service): shed live shape subscribers that stop draining their mailbox MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A live `GET /v1/shape` request whose client can't keep up — a stalled or dead socket while changes keep streaming — registered as a change subscriber and was notified via a fire-and-forget `send/2` per transaction, with no backpressure to the producer and no cap on the subscriber's mailbox. The request process, blocked writing to the socket, stopped draining its mailbox while the consumer kept enqueuing `:new_changes`, so the message queue (and the reference-counted binary each message pins) grew without bound until the node OOMed. Observed in production as a single `GET /v1/shape` process with a ~200k message queue and ~3.9 GB of `vm.memory.binary`. Fix (producer-side mailbox watermark): before notifying each subscriber, `Consumer.notify_clients_of_new_changes/2` checks the subscriber's `message_queue_len`; if it exceeds `:slow_subscriber_max_queue_len` the subscriber is terminated instead of notified. The client reconnects and resumes from its last offset (the live protocol is resumable), so shedding is safe. This is the only mechanism that catches a handler blocked inside `Plug.Conn.chunk` — it can't wake to check itself, but the producer inspects it externally. The watermark is always on (a positive integer, default 10,000; non-positive values are rejected at stack startup) and tunable via the `ELECTRIC_SLOW_SUBSCRIBER_MAX_QUEUE_LEN` env var. Co-Authored-By: Claude Opus 4.8 (1M context) --- .changeset/shed-slow-live-subscribers.md | 5 ++ packages/sync-service/config/runtime.exs | 1 + .../sync-service/lib/electric/application.ex | 3 +- packages/sync-service/lib/electric/config.ex | 5 ++ .../lib/electric/shapes/consumer.ex | 50 ++++++++++- .../sync-service/lib/electric/stack_config.ex | 3 +- .../lib/electric/stack_supervisor.ex | 8 ++ .../test/electric/shapes/consumer_test.exs | 86 +++++++++++++++++++ 8 files changed, 157 insertions(+), 4 deletions(-) create mode 100644 .changeset/shed-slow-live-subscribers.md diff --git a/.changeset/shed-slow-live-subscribers.md b/.changeset/shed-slow-live-subscribers.md new file mode 100644 index 0000000000..3cf34c96e5 --- /dev/null +++ b/.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. diff --git a/packages/sync-service/config/runtime.exs b/packages/sync-service/config/runtime.exs index 6c73b12631..d2b43ca744 100644 --- a/packages/sync-service/config/runtime.exs +++ b/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), diff --git a/packages/sync-service/lib/electric/application.ex b/packages/sync-service/lib/electric/application.ex index 63a4ed7ffe..1b7335ee20 100644 --- a/packages/sync-service/lib/electric/application.ex +++ b/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: [ diff --git a/packages/sync-service/lib/electric/config.ex b/packages/sync-service/lib/electric/config.ex index c6d77a830c..0230c53eb4 100644 --- a/packages/sync-service/lib/electric/config.ex +++ b/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: []), diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex index f60a57dcdd..d94de96b70 100644 --- a/packages/sync-service/lib/electric/shapes/consumer.ex +++ b/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( diff --git a/packages/sync-service/lib/electric/stack_config.ex b/packages/sync-service/lib/electric/stack_config.ex index cf51621418..d7b5ccd8bb 100644 --- a/packages/sync-service/lib/electric/stack_config.ex +++ b/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 diff --git a/packages/sync-service/lib/electric/stack_supervisor.ex b/packages/sync-service/lib/electric/stack_supervisor.ex index 3d8a35b078..cfc4d19f68 100644 --- a/packages/sync-service/lib/electric/stack_supervisor.ex +++ b/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, diff --git a/packages/sync-service/test/electric/shapes/consumer_test.exs b/packages/sync-service/test/electric/shapes/consumer_test.exs index a2554f54d9..1e15dff2f9 100644 --- a/packages/sync-service/test/electric/shapes/consumer_test.exs +++ b/packages/sync-service/test/electric/shapes/consumer_test.exs @@ -223,6 +223,92 @@ defmodule Electric.Shapes.ConsumerTest do [consumers: consumers] end + test "sheds a live subscriber whose mailbox grows without bound", ctx do + # Regression: a live `GET /v1/shape` subscriber that can't drain its + # mailbox (stalled/dead socket) while changes keep streaming accumulates + # one `{:new_changes, _}` message per transaction with no upper bound. + # Each message pins reference-counted binary, so memory grows unbounded + # until the node OOMs. The consumer must shed such a subscriber once its + # mailbox crosses a watermark instead of notifying it forever. + 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 subscriber that registers for changes but never drains its mailbox. + subscriber = + spawn(fn -> + {:ok, _} = Registry.register(ctx.registry, @shape_handle1, make_ref()) + send(test_pid, :subscribed) + Process.sleep(:infinity) + end) + + monitor = Process.monitor(subscriber) + assert_receive :subscribed, @receive_timeout + + # 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 non-draining subscriber must be shed once its mailbox crosses the + # watermark, rather than being allowed to grow without bound. + assert_receive {:DOWN, ^monitor, :process, ^subscriber, _reason}, @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) From a3f8e0ba9c8434c6e48cfc480b1484c4afe60900 Mon Sep 17 00:00:00 2001 From: rob Date: Thu, 2 Jul 2026 09:20:27 +0100 Subject: [PATCH 2/2] test(sync-service): block the shed reproduction in a real stalled socket MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review: the reproduction simulated the stuck subscriber with Process.sleep(:infinity) (idle in receive), which is a different scheduler state from the production failure — a handler blocked inside Plug.Conn.chunk. It proved the watermark comparison, not that the mechanism fires against a genuinely socket-blocked handler, and didn't cover the risk that Process.info(_, :message_queue_len) could itself stall on the blocked target and freeze the per-shape consumer. The subscriber is now blocked in a real :gen_tcp.send to a peer that never reads — the actual production state (`:waiting` in the inet_reply receive, verified on OTP 28). The shed still firing proves process_info did not block the consumer; an added assertion registers a healthy subscriber afterwards and confirms it is still notified, explicitly guarding consumer liveness. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../test/electric/shapes/consumer_test.exs | 89 ++++++++++++++++--- 1 file changed, 79 insertions(+), 10 deletions(-) diff --git a/packages/sync-service/test/electric/shapes/consumer_test.exs b/packages/sync-service/test/electric/shapes/consumer_test.exs index 1e15dff2f9..7cba826a82 100644 --- a/packages/sync-service/test/electric/shapes/consumer_test.exs +++ b/packages/sync-service/test/electric/shapes/consumer_test.exs @@ -223,29 +223,68 @@ defmodule Electric.Shapes.ConsumerTest do [consumers: consumers] end - test "sheds a live subscriber whose mailbox grows without bound", ctx do - # Regression: a live `GET /v1/shape` subscriber that can't drain its - # mailbox (stalled/dead socket) while changes keep streaming accumulates - # one `{:new_changes, _}` message per transaction with no upper bound. - # Each message pins reference-counted binary, so memory grows unbounded + 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 instead of notifying it forever. + # 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 subscriber that registers for changes but never drains its mailbox. + # 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) - Process.sleep(:infinity) + + # 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 @@ -263,9 +302,39 @@ defmodule Electric.Shapes.ConsumerTest do assert :ok = ShapeLogCollector.handle_event(txn, ctx.stack_id) end - # The non-draining subscriber must be shed once its mailbox crosses the - # watermark, rather than being allowed to grow without bound. + # 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