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..7cba826a82 100644 --- a/packages/sync-service/test/electric/shapes/consumer_test.exs +++ b/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)