From ebc231973f76a78c58dfae8548839667790cbed5 Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 14 Jul 2026 15:12:26 +0100 Subject: [PATCH 01/14] test(sync-service): stalled serves must be reaped within a stall timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Failing integration test for the defense-in-depth follow-up to #4708. Bounded write units cap what each stalled serve pins, but nothing reaps the serves themselves: a population of connections whose clients stop accepting data still accumulates memory and file descriptors in proportion to connection count, invisibly (stalled serves never complete, so they emit no spans and increment no request counters). The TCP send timeout only catches a fully blocked write; a trickling client — or a proxy buffering for a vanished one — evades it forever. The test parks 3 live long-pollers on never-read sockets, sets :stalled_serve_timeout to 1s, lands a transaction large enough that every serve wedges after the initial socket-buffer fill, and asserts each handler process is terminated within the timeout plus slack. Progress is defined as a completed socket write: with post-#4708 bounded write units, any client accepting even a trickle resets the deadline, so only serves accepting nothing for the full window die — and the client can reconnect and resume from its offset. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../stalled_serve_reaping_test.exs | 116 ++++++++++++++++++ 1 file changed, 116 insertions(+) create mode 100644 packages/sync-service/test/integration/stalled_serve_reaping_test.exs diff --git a/packages/sync-service/test/integration/stalled_serve_reaping_test.exs b/packages/sync-service/test/integration/stalled_serve_reaping_test.exs new file mode 100644 index 0000000000..423c926dc8 --- /dev/null +++ b/packages/sync-service/test/integration/stalled_serve_reaping_test.exs @@ -0,0 +1,116 @@ +defmodule Electric.Integration.StalledServeReapingTest do + @moduledoc """ + A shape response whose client stops accepting data must be terminated once + it has made no progress for `:stalled_serve_timeout`, releasing the + connection and everything the serve pins. + + Defense in depth for the 2026-07-01 production OOM: bounded write units + (#4708) cap what each stalled serve pins, but nothing reaps the serves + themselves — a population of stalled connections still accumulates memory + and file descriptors proportional to connection count, invisibly (stalled + serves never complete, so they emit no spans and increment no request + counters). The TCP send timeout only catches a *fully blocked* write; a + client draining at a trickle — or a proxy buffering for a vanished client — + evades it indefinitely. + + The reaping deadline is distinct from a slow-but-healthy client: progress + is measured per completed socket write, and the response body is streamed + in small units (post-#4708), so any client accepting even a trickle of + data resets the deadline. Only a serve accepting *nothing* for the full + window is terminated; the client can reconnect and resume from its offset. + """ + use ExUnit.Case, async: false + + import Support.ComponentSetup + import Support.DbSetup + import Support.DbStructureSetup + import Support.IntegrationSetup + + alias Electric.Plug.Router + + @moduletag :tmp_dir + + @stalled_clients 3 + @stalled_serve_timeout 1_000 + # The reaper must fire within the configured timeout plus scheduling slack. + @reap_deadline_ms 10_000 + + describe "stalled live clients" do + setup [:with_unique_db, :with_basic_tables, :with_sql_execute] + setup :with_complete_stack + setup :with_electric_client + + @tag timeout: 60_000 + test "a serve making no progress is terminated within the stall timeout", ctx do + %{stack_id: stack_id, db_conn: db_conn, port: port, server_pid: server_pid} = ctx + opts = Router.init(build_router_opts(ctx)) + registry = Electric.StackSupervisor.registry_name(stack_id) + + Electric.StackConfig.put(stack_id, :stalled_serve_timeout, @stalled_serve_timeout) + + # Create the shape and learn its handle. + snapshot = Plug.Test.conn(:get, "/v1/shape?table=items&offset=-1") |> Router.call(opts) + assert snapshot.status == 200 + [handle] = Plug.Conn.get_resp_header(snapshot, "electric-handle") + + # Park live long-pollers over raw sockets that will never be read from. + socks = + for _ <- 1..@stalled_clients do + {:ok, s} = + :gen_tcp.connect({127, 0, 0, 1}, port, [ + :binary, + active: false, + recbuf: 2048, + buffer: 2048 + ]) + + :ok = + :gen_tcp.send( + s, + "GET /v1/shape?table=items&handle=#{handle}&offset=0_0&live=true HTTP/1.1\r\n" <> + "Host: localhost\r\n\r\n" + ) + + s + end + + on_exit(fn -> Enum.each(socks, &:gen_tcp.close/1) end) + + wait_for_subscribers(registry, handle, @stalled_clients) + {:ok, handlers} = ThousandIsland.connection_pids(server_pid) + assert length(handlers) == @stalled_clients + + monitors = Map.new(handlers, fn pid -> {Process.monitor(pid), pid} end) + + # A large transaction wakes every waiter into a response far bigger than + # the client's socket buffers: every serve wedges after the initial + # buffer fill and never makes progress again. + Postgrex.query!( + db_conn, + "INSERT INTO items SELECT gen_random_uuid(), repeat('x', 50000) FROM generate_series(1, 400)", + [] + ) + + # Every stalled serve must be reaped within the configured stall timeout + # (plus slack). Without reaping they live — and pin memory and a file + # descriptor each — until the client goes away, which may be never. + for {ref, pid} <- monitors do + assert_receive {:DOWN, ^ref, :process, ^pid, _reason}, + @reap_deadline_ms, + "stalled serve #{inspect(pid)} was not terminated within " <> + "#{@stalled_serve_timeout}ms stall timeout (+ slack): nothing reaps " <> + "serves whose clients stop accepting data" + end + end + end + + defp wait_for_subscribers(registry, handle, n, tries \\ 400) do + found = length(Registry.lookup(registry, handle)) + + cond do + found >= n -> :ok + tries > 0 -> Process.sleep(25) && wait_for_subscribers(registry, handle, n, tries - 1) + true -> flunk("only #{found}/#{n} live requests subscribed") + end + end +end From fd747b945bd4fe9ef09670f0aa2c2b12261fed96 Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 14 Jul 2026 15:35:59 +0100 Subject: [PATCH 02/14] fix(sync-service): terminate serves whose clients stop accepting data MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Defense in depth for the 2026-07-01 OOM (follow-up to #4708). Serves to stalled clients never complete, so they emit no spans and increment no request counters — while each pins its in-flight response data and a file descriptor for as long as the client's connection survives, which may be forever: the TCP send timeout only catches a write that is fully blocked for its whole window, and empirically a blocked write against a full kernel send buffer can outlast it even while the client trickles. Response bodies are now written to the socket in pieces of at most 16 KiB, and a per-serve watchdog process terminates the handler when a single piece fails to complete within :stalled_serve_timeout (ELECTRIC_STALLED_SERVE_TIMEOUT, default 60s, 0 disables). Write completion is quantized by the OS at up to a send buffer's worth of drain, so the effective contract is that a healthy client drains roughly one OS send buffer per timeout window; a terminated client reconnects and resumes from its last offset. The watchdog is armed only while a write is in flight — serves idling between body elements (live long-poll holds, SSE waiting for changes) are never at risk. The bounded write unit also cuts the response data pinned in a stalled socket's driver queue from one body element (~256 KiB) to ~16 KiB, tightening #4708's per-connection bound. Two integration tests: stalled clients are reaped within the timeout (plus slack), and a slow-but-draining client whose serve spans many timeout windows is never reaped. Co-Authored-By: Claude Opus 4.8 (1M context) --- .changeset/reap-stalled-serves.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/api/response.ex | 164 +++++++++++++++--- .../sync-service/lib/electric/stack_config.ex | 3 +- .../lib/electric/stack_supervisor.ex | 6 + .../stalled_serve_reaping_test.exs | 92 +++++++++- 8 files changed, 246 insertions(+), 33 deletions(-) create mode 100644 .changeset/reap-stalled-serves.md diff --git a/.changeset/reap-stalled-serves.md b/.changeset/reap-stalled-serves.md new file mode 100644 index 0000000000..d2c87e17b5 --- /dev/null +++ b/.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 16 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. The smaller write unit also cuts the response data queued in a stalled socket's driver buffer from one body element (~256 KiB) to ~16 KiB. diff --git a/packages/sync-service/config/runtime.exs b/packages/sync-service/config/runtime.exs index cd0e4c3ecd..3345744e8b 100644 --- a/packages/sync-service/config/runtime.exs +++ b/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), diff --git a/packages/sync-service/lib/electric/application.ex b/packages/sync-service/lib/electric/application.ex index 63a4ed7ffe..550d9ffe5b 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), + stalled_serve_timeout: get_env(opts, :stalled_serve_timeout) ], 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 2966bc351f..193361f5ee 100644 --- a/packages/sync-service/lib/electric/config.ex +++ b/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: []), diff --git a/packages/sync-service/lib/electric/shapes/api/response.ex b/packages/sync-service/lib/electric/shapes/api/response.ex index 6a99d06335..bdffadbfc9 100644 --- a/packages/sync-service/lib/electric/shapes/api/response.ex +++ b/packages/sync-service/lib/electric/shapes/api/response.ex @@ -434,39 +434,151 @@ 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, 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 + @stalled_serve_timeout_default :timer.seconds(60) + + # Socket writes are split into pieces of at most this size. Bounding the + # write unit bounds both the response data queued in the socket's driver + # queue at any moment and — because a bounded write to a live client + # completes quickly once the transport buffers are full — gives the stall + # watchdog below a rate-independent progress signal: each completed piece + # is proof the client accepted data. + @socket_write_bytes 16 * 1024 + + # 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 (see the 2026-07-01 OOM 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 — + # on the order of 10 KB/s at the 60s default with megabyte-class buffers, + # 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 + case Electric.StackConfig.lookup( + stack_id, + :stalled_serve_timeout, + @stalled_serve_timeout_default + ) do + timeout when is_integer(timeout) and timeout > 0 -> + handler = self() + shape_handle = response.handle + + spawn(fn -> + ref = Process.monitor(handler) + write_watchdog_loop(handler, ref, shape_handle, timeout, :idle) + end) + + # 0 (or any non-positive value) disables reaping. + _disabled -> + nil + end + end + + defp write_watchdog_loop(handler, ref, shape_handle, timeout, writing?) do + receive do + :write_start -> + write_watchdog_loop(handler, ref, shape_handle, timeout, :writing) + + :write_done -> + write_watchdog_loop(handler, ref, shape_handle, 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", + shape_handle: shape_handle + ) + + 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. + defp write_in_bounded_pieces(conn, chunk, watchdog) do + chunk + |> IO.iodata_to_binary() + |> write_pieces(conn, watchdog) + end + + defp write_pieces(<>, 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 diff --git a/packages/sync-service/lib/electric/stack_config.ex b/packages/sync-service/lib/electric/stack_config.ex index cf51621418..58bced3dbc 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), + stalled_serve_timeout: Electric.Config.default(:stalled_serve_timeout) ] end diff --git a/packages/sync-service/lib/electric/stack_supervisor.ex b/packages/sync-service/lib/electric/stack_supervisor.ex index 3d8a35b078..4047cab785 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) ], + 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, diff --git a/packages/sync-service/test/integration/stalled_serve_reaping_test.exs b/packages/sync-service/test/integration/stalled_serve_reaping_test.exs index 423c926dc8..abbc2ce5ab 100644 --- a/packages/sync-service/test/integration/stalled_serve_reaping_test.exs +++ b/packages/sync-service/test/integration/stalled_serve_reaping_test.exs @@ -14,10 +14,11 @@ defmodule Electric.Integration.StalledServeReapingTest do evades it indefinitely. The reaping deadline is distinct from a slow-but-healthy client: progress - is measured per completed socket write, and the response body is streamed - in small units (post-#4708), so any client accepting even a trickle of - data resets the deadline. Only a serve accepting *nothing* for the full - window is terminated; the client can reconnect and resume from its offset. + is a completed bounded socket write, so a client sustaining a modest + throughput (roughly one OS send buffer per timeout window, worst case) + keeps resetting the deadline. A serve whose client accepts nothing for the + full window is terminated; the client can reconnect and resume from its + offset. """ use ExUnit.Case, async: false @@ -31,7 +32,7 @@ defmodule Electric.Integration.StalledServeReapingTest do @moduletag :tmp_dir @stalled_clients 3 - @stalled_serve_timeout 1_000 + @stalled_serve_timeout 2_000 # The reaper must fire within the configured timeout plus scheduling slack. @reap_deadline_ms 10_000 @@ -104,6 +105,87 @@ defmodule Electric.Integration.StalledServeReapingTest do end end + describe "slow but draining client" do + setup [:with_unique_db, :with_basic_tables, :with_sql_execute] + setup :with_complete_stack + setup :with_electric_client + + @tag timeout: 60_000 + test "a serve to a client draining at a trickle is never reaped", ctx do + %{stack_id: stack_id, db_conn: db_conn, port: port, server_pid: server_pid} = ctx + opts = Router.init(build_router_opts(ctx)) + registry = Electric.StackSupervisor.registry_name(stack_id) + + Electric.StackConfig.put(stack_id, :stalled_serve_timeout, @stalled_serve_timeout) + + snapshot = Plug.Test.conn(:get, "/v1/shape?table=items&offset=-1") |> Router.call(opts) + assert snapshot.status == 200 + [handle] = Plug.Conn.get_resp_header(snapshot, "electric-handle") + + {:ok, sock} = + :gen_tcp.connect({127, 0, 0, 1}, port, [ + :binary, + active: false, + recbuf: 65536, + buffer: 65536 + ]) + + on_exit(fn -> :gen_tcp.close(sock) end) + + :ok = + :gen_tcp.send( + sock, + "GET /v1/shape?table=items&handle=#{handle}&offset=0_0&live=true HTTP/1.1\r\n" <> + "Host: localhost\r\n\r\n" + ) + + wait_for_subscribers(registry, handle, 1) + {:ok, [handler]} = ThousandIsland.connection_pids(server_pid) + monitor = Process.monitor(handler) + + # A transaction whose serve, at our drain rate, spans many stall-timeout + # windows. The client drains more than an OS send buffer per window, so + # bounded writes keep completing and the serve must survive them all. + Postgrex.query!( + db_conn, + "INSERT INTO items SELECT gen_random_uuid(), repeat('x', 10000) FROM generate_series(1, 2000)", + [] + ) + + # Drain slowly for up to ~6x the stall timeout: read a piece, pause. A + # recv timeout means the (chunk-bounded) response completed under us. + deadline = System.monotonic_time(:millisecond) + 6 * @stalled_serve_timeout + drained = drain_slowly(sock, deadline, 0) + + # The serve must have spanned multiple stall-timeout windows. + assert drained > 5_000_000 + + refute_received {:DOWN, ^monitor, :process, ^handler, _reason} + + assert Process.alive?(handler), + "healthy slow-draining serve was reaped by the stall watchdog" + end + end + + defp drain_slowly(sock, deadline, acc) do + if System.monotonic_time(:millisecond) >= deadline do + acc + else + case :gen_tcp.recv(sock, 0, 2_000) do + {:ok, data} -> + Process.sleep(50) + drain_slowly(sock, deadline, acc + byte_size(data)) + + # No more data: the response completed while we were draining. + {:error, :timeout} -> + acc + + {:error, reason} -> + flunk("socket read failed after #{acc} bytes: #{inspect(reason)}") + end + end + end + defp wait_for_subscribers(registry, handle, n, tries \\ 400) do found = length(Registry.lookup(registry, handle)) From 3ca1bf2fb716b1f9c4a061635c20152c99a2b428 Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 14 Jul 2026 16:10:23 +0100 Subject: [PATCH 03/14] Start the serve watchdog via Task and give it a process label MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task.start provides $initial_call/$callers metadata and richer crash reports at no behavioral cost (unlinked and unsupervised like the bare spawn it replaces — restart semantics are meaningless for a watchdog whose protocol state lives with the serve). The explicit label groups the watchdogs under a low-cardinality `serve_watchdog` process_type in per-process telemetry instead of leaving them anonymous — the incident this guards against was prolonged by exactly that kind of process invisibility. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../lib/electric/shapes/api/response.ex | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/packages/sync-service/lib/electric/shapes/api/response.ex b/packages/sync-service/lib/electric/shapes/api/response.ex index bdffadbfc9..7668c99fd2 100644 --- a/packages/sync-service/lib/electric/shapes/api/response.ex +++ b/packages/sync-service/lib/electric/shapes/api/response.ex @@ -513,10 +513,17 @@ defmodule Electric.Shapes.Api.Response do handler = self() shape_handle = response.handle - spawn(fn -> - ref = Process.monitor(handler) - write_watchdog_loop(handler, ref, shape_handle, timeout, :idle) - end) + {: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. + Process.set_label({:serve_watchdog, shape_handle}) + ref = Process.monitor(handler) + write_watchdog_loop(handler, ref, shape_handle, timeout, :idle) + end) + + watchdog # 0 (or any non-positive value) disables reaping. _disabled -> From 877abf83e7f27df5e1ab1704d4587e9ea82421d8 Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 14 Jul 2026 16:40:44 +0100 Subject: [PATCH 04/14] Align the socket write piece size with the encoder batch cap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback: splitting every body element into 16 KiB pieces turned multi-megabyte single elements (e.g. one very large row, which the encoder byte cap cannot split) into a thousand-plus Plug.Conn.chunk calls. Under the Plug.Test adapter each call appends to an accumulated buffer, making body readback quadratic in the piece count — enough to time out RouterTest's large-binary and chunked-results tests in CI. Pieces are now 256 KiB, matching the encoder's batch byte cap, so ordinary elements are written as-is — via a fast path that skips flattening entirely, restoring the pre-watchdog zero-copy behavior for the common case — and only oversized single items are split (a 20 MB element becomes ~80 writes instead of ~1300). This does not weaken the watchdog: write completion is quantized by the OS at up to a kernel send buffer of drain, which dwarfs either piece size, so the effective progress contract is unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) --- .changeset/reap-stalled-serves.md | 2 +- .../lib/electric/shapes/api/response.ex | 30 +++++++++++++------ 2 files changed, 22 insertions(+), 10 deletions(-) diff --git a/.changeset/reap-stalled-serves.md b/.changeset/reap-stalled-serves.md index d2c87e17b5..b4160396d5 100644 --- a/.changeset/reap-stalled-serves.md +++ b/.changeset/reap-stalled-serves.md @@ -2,4 +2,4 @@ "@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 16 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. The smaller write unit also cuts the response data queued in a stalled socket's driver buffer from one body element (~256 KiB) to ~16 KiB. +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. diff --git a/packages/sync-service/lib/electric/shapes/api/response.ex b/packages/sync-service/lib/electric/shapes/api/response.ex index 7668c99fd2..0e77ce4e1c 100644 --- a/packages/sync-service/lib/electric/shapes/api/response.ex +++ b/packages/sync-service/lib/electric/shapes/api/response.ex @@ -447,7 +447,7 @@ defmodule Electric.Shapes.Api.Response do Map.put(sample_rate_attrs, "chunk_size", chunk_size), stack_id, fn -> - case write_in_bounded_pieces(conn, chunk, watchdog) do + case write_in_bounded_pieces(conn, chunk, chunk_size, watchdog) do {:ok, conn} -> {:cont, {conn, bytes_sent + chunk_size}} @@ -475,12 +475,17 @@ defmodule Electric.Shapes.Api.Response do @stalled_serve_timeout_default :timer.seconds(60) # Socket writes are split into pieces of at most this size. Bounding the - # write unit bounds both the response data queued in the socket's driver - # queue at any moment and — because a bounded write to a live client - # completes quickly once the transport buffers are full — gives the stall - # watchdog below a rate-independent progress signal: each completed piece - # is proof the client accepted data. - @socket_write_bytes 16 * 1024 + # 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. + # + # The size matches the encoder's batch byte cap, so ordinary body elements + # are written as-is (no flattening or splitting); 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 256 * 1024 # 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 @@ -560,8 +565,15 @@ defmodule Electric.Shapes.Api.Response do 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. - defp write_in_bounded_pieces(conn, chunk, watchdog) do + # notifying the watchdog around each one. Elements within the bound — the + # common case, since the encoder caps batches at the same size — 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) From 659e8ad2412c7df5044cacd59778847dc165020e Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 14 Jul 2026 17:09:32 +0100 Subject: [PATCH 05/14] Carry stack_id on the stalled-serve reap warning MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback: the watchdog runs in its own process and does not inherit the handler's Logger metadata, so the reap warning — the only signal an otherwise-invisible stalled serve emits — carried only the shape handle and could not be correlated to a stack. The watchdog now sets stack_id and shape_handle as its Logger metadata at start. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../lib/electric/shapes/api/response.ex | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/packages/sync-service/lib/electric/shapes/api/response.ex b/packages/sync-service/lib/electric/shapes/api/response.ex index 0e77ce4e1c..95dd111039 100644 --- a/packages/sync-service/lib/electric/shapes/api/response.ex +++ b/packages/sync-service/lib/electric/shapes/api/response.ex @@ -523,9 +523,14 @@ defmodule Electric.Shapes.Api.Response do # 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 only 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, shape_handle, timeout, :idle) + write_watchdog_loop(handler, ref, timeout, :idle) end) watchdog @@ -536,13 +541,13 @@ defmodule Electric.Shapes.Api.Response do end end - defp write_watchdog_loop(handler, ref, shape_handle, timeout, writing?) do + defp write_watchdog_loop(handler, ref, timeout, writing?) do receive do :write_start -> - write_watchdog_loop(handler, ref, shape_handle, timeout, :writing) + write_watchdog_loop(handler, ref, timeout, :writing) :write_done -> - write_watchdog_loop(handler, ref, shape_handle, timeout, :idle) + write_watchdog_loop(handler, ref, timeout, :idle) :stop -> :ok @@ -553,8 +558,7 @@ defmodule Electric.Shapes.Api.Response do watchdog_wait(writing?, timeout) -> Logger.warning( "Terminating stalled shape response serve: client accepted no data " <> - "for #{timeout}ms", - shape_handle: shape_handle + "for #{timeout}ms" ) Process.exit(handler, :kill) From 1d69f71e7dfc8b9f7b0229c75ac6713eee3076ef Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 14 Jul 2026 17:10:31 +0100 Subject: [PATCH 06/14] Use the Electric.Config default for the stalled-serve timeout fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback: the module attribute duplicated the config.ex default and was effectively dead — StackConfig's seed always provides the key — so the two values could silently drift. The lookup fallback now references Electric.Config.default/1, the single source of truth. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/sync-service/lib/electric/shapes/api/response.ex | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/packages/sync-service/lib/electric/shapes/api/response.ex b/packages/sync-service/lib/electric/shapes/api/response.ex index 95dd111039..08006bd1ca 100644 --- a/packages/sync-service/lib/electric/shapes/api/response.ex +++ b/packages/sync-service/lib/electric/shapes/api/response.ex @@ -472,8 +472,6 @@ defmodule Electric.Shapes.Api.Response do Plug.Conn.assign(conn, :streaming_bytes_sent, bytes_sent) end - @stalled_serve_timeout_default :timer.seconds(60) - # 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 @@ -509,10 +507,13 @@ defmodule Electric.Shapes.Api.Response do # is never at risk. The terminated client can reconnect and resume from its # last offset. defp start_write_watchdog(stack_id, response) do + # The fallback only applies to stacks whose seed config omits the key + # (e.g. minimal unit-test stacks); StackSupervisor-managed stacks always + # seed it. Electric.Config is the single source of truth for the default. case Electric.StackConfig.lookup( stack_id, :stalled_serve_timeout, - @stalled_serve_timeout_default + Electric.Config.default(:stalled_serve_timeout) ) do timeout when is_integer(timeout) and timeout > 0 -> handler = self() From e69bb158385774793e14f860a3cc3ebae25292d6 Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 14 Jul 2026 17:42:35 +0100 Subject: [PATCH 07/14] Derive the socket write bound from the encoder's batch cap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Comment audit follow-up: the write piece size matching the encoder's batch byte cap was a load-bearing coupling expressed only in prose — if either constant changed independently, the no-flatten fast path would silently degrade back to flattening and splitting every element. The encoder now exposes max_batch_bytes/0 and the response module derives its write bound from it, so the invariant is structural. Also de-drift a few comments: the throughput figure is phrased as an illustration rather than a claim about the current default, the reap warning is the "primary" (not "only") signal in anticipation of a telemetry counter, and the config fallback note no longer asserts another module's behavior. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../lib/electric/shapes/api/encoder.ex | 7 ++++++ .../lib/electric/shapes/api/response.ex | 25 +++++++++---------- 2 files changed, 19 insertions(+), 13 deletions(-) diff --git a/packages/sync-service/lib/electric/shapes/api/encoder.ex b/packages/sync-service/lib/electric/shapes/api/encoder.ex index 3f0acdc484..1e8aabe9e1 100644 --- a/packages/sync-service/lib/electric/shapes/api/encoder.ex +++ b/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], diff --git a/packages/sync-service/lib/electric/shapes/api/response.ex b/packages/sync-service/lib/electric/shapes/api/response.ex index 08006bd1ca..586029d7cc 100644 --- a/packages/sync-service/lib/electric/shapes/api/response.ex +++ b/packages/sync-service/lib/electric/shapes/api/response.ex @@ -478,12 +478,12 @@ defmodule Electric.Shapes.Api.Response do # once the transport buffers drain — gives the stall watchdog below a # progress signal: each completed piece is proof the client accepted data. # - # The size matches the encoder's batch byte cap, so ordinary body elements - # are written as-is (no flattening or splitting); only oversized single + # 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 256 * 1024 + @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 @@ -499,17 +499,16 @@ defmodule Electric.Shapes.Api.Response do # 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 — - # on the order of 10 KB/s at the 60s default with megabyte-class buffers, - # and far less on stacks that signal writability at finer granularity. The + # 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 - # The fallback only applies to stacks whose seed config omits the key - # (e.g. minimal unit-test stacks); StackSupervisor-managed stacks always - # seed it. Electric.Config is the single source of truth for the default. + # 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, @@ -525,9 +524,9 @@ defmodule Electric.Shapes.Api.Response do # 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 only signal an - # otherwise-invisible stalled serve emits, so it must carry enough - # to correlate (which stack, which shape). + # 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) @@ -571,7 +570,7 @@ defmodule Electric.Shapes.Api.Response do # 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, since the encoder caps batches at the same size — are + # 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 From f66065f37eac319b06b0ec0903c0638ee739b7ed Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 14 Jul 2026 17:46:23 +0100 Subject: [PATCH 08/14] Remove incident dates from code comments Incident details belong in commit messages and PR descriptions, where git blame already leads; a code comment should carry the rationale, which stands on its own. In a public repository a dated reference to a private incident is also a pointer no outside reader can resolve. The test moduledocs keep traceability through public PR references instead. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/sync-service/lib/electric/shapes/api/response.ex | 2 +- .../sync-service/test/integration/stalled_serve_memory_test.exs | 2 +- .../test/integration/stalled_serve_reaping_test.exs | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/packages/sync-service/lib/electric/shapes/api/response.ex b/packages/sync-service/lib/electric/shapes/api/response.ex index 586029d7cc..d226d78a97 100644 --- a/packages/sync-service/lib/electric/shapes/api/response.ex +++ b/packages/sync-service/lib/electric/shapes/api/response.ex @@ -491,7 +491,7 @@ defmodule Electric.Shapes.Api.Response do # 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 (see the 2026-07-01 OOM reproduced in + # connection count (reproduced in # test/integration/stalled_serve_memory_test.exs). # # The watchdog is a small companion process that terminates the serve when diff --git a/packages/sync-service/test/integration/stalled_serve_memory_test.exs b/packages/sync-service/test/integration/stalled_serve_memory_test.exs index 077cb38366..fd77681829 100644 --- a/packages/sync-service/test/integration/stalled_serve_memory_test.exs +++ b/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. diff --git a/packages/sync-service/test/integration/stalled_serve_reaping_test.exs b/packages/sync-service/test/integration/stalled_serve_reaping_test.exs index abbc2ce5ab..bfde83b404 100644 --- a/packages/sync-service/test/integration/stalled_serve_reaping_test.exs +++ b/packages/sync-service/test/integration/stalled_serve_reaping_test.exs @@ -4,7 +4,7 @@ defmodule Electric.Integration.StalledServeReapingTest do it has made no progress for `:stalled_serve_timeout`, releasing the connection and everything the serve pins. - Defense in depth for the 2026-07-01 production OOM: bounded write units + Defense in depth for the production OOM addressed by #4708: bounded write units (#4708) cap what each stalled serve pins, but nothing reaps the serves themselves — a population of stalled connections still accumulates memory and file descriptors proportional to connection count, invisibly (stalled From 7a54d42c5edf1612127d06102ee1c5d5a1f049f2 Mon Sep 17 00:00:00 2001 From: rob Date: Wed, 15 Jul 2026 12:45:27 +0100 Subject: [PATCH 09/14] Replace per-serve watchdog tasks with a per-stack ServeWatchdog GenServer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback: a watchdog process per in-flight serve doubles the processes involved in request handling, and the state machine had grown too complicated for private functions in Response. The watchdog is now a standalone, supervised GenServer, one per stack. Handlers register (cast) before each bounded socket write and deregister after it; a periodic sweep terminates handlers whose in-flight write has outlived its deadline and prunes entries for handlers that died on their own. Casts mean handlers never block on the server, casts to a stack without a running instance are silently dropped (serving degrades to unguarded rather than failing), and a restart self-heals: in-flight serves re-register on their next write, leaving at most one write unguarded — supervision semantics that a per-serve process could not have. Reap precision changes from exact to deadline-plus-sweep-interval (1s), which is noise against the 60s default. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../lib/electric/shapes/api/response.ex | 106 ++++----------- .../lib/electric/shapes/api/serve_watchdog.ex | 122 ++++++++++++++++++ .../lib/electric/stack_supervisor.ex | 1 + 3 files changed, 146 insertions(+), 83 deletions(-) create mode 100644 packages/sync-service/lib/electric/shapes/api/serve_watchdog.ex diff --git a/packages/sync-service/lib/electric/shapes/api/response.ex b/packages/sync-service/lib/electric/shapes/api/response.ex index d226d78a97..599f14c5b4 100644 --- a/packages/sync-service/lib/electric/shapes/api/response.ex +++ b/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,7 +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 = start_write_watchdog(stack_id, response) + watchdog = watchdog_context(stack_id, response) {conn, bytes_sent} = try do @@ -466,7 +467,7 @@ defmodule Electric.Shapes.Api.Response do ) end) after - watchdog && Kernel.send(watchdog, :stop) + watchdog && ServeWatchdog.write_finished(stack_id) end Plug.Conn.assign(conn, :streaming_bytes_sent, bytes_sent) @@ -485,89 +486,24 @@ defmodule Electric.Shapes.Api.Response do # 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 + # See Electric.Shapes.Api.ServeWatchdog for the design rationale. The + # context carries what each per-piece registration needs; `nil` (timeout + # not positive) disables reaping for this serve. + defp watchdog_context(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 + timeout = + Electric.StackConfig.lookup( + stack_id, + :stalled_serve_timeout, + Electric.Config.default(:stalled_serve_timeout) + ) + + if is_integer(timeout) and timeout > 0 do + {stack_id, response.handle, timeout} 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 @@ -595,10 +531,14 @@ defmodule Electric.Shapes.Api.Response do guarded_chunk(conn, piece, watchdog) end - defp guarded_chunk(conn, data, watchdog) do - watchdog && Kernel.send(watchdog, :write_start) + defp guarded_chunk(conn, data, nil) do + Plug.Conn.chunk(conn, data) + end + + defp guarded_chunk(conn, data, {stack_id, shape_handle, timeout}) do + ServeWatchdog.write_started(stack_id, shape_handle, timeout) result = Plug.Conn.chunk(conn, data) - watchdog && Kernel.send(watchdog, :write_done) + ServeWatchdog.write_finished(stack_id) result end diff --git a/packages/sync-service/lib/electric/shapes/api/serve_watchdog.ex b/packages/sync-service/lib/electric/shapes/api/serve_watchdog.ex new file mode 100644 index 0000000000..78009104af --- /dev/null +++ b/packages/sync-service/lib/electric/shapes/api/serve_watchdog.ex @@ -0,0 +1,122 @@ +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). + + This server instead times the completion of each bounded application-level + socket write: request handlers register before every write and deregister + after it, and a periodic sweep terminates any handler whose in-flight + write has outlived its deadline. The effective contract is a minimum + sustained throughput — one bounded write unit per deadline window — which + any live client trivially meets and a stalled one cannot. Write completion + is quantized by the OS at up to a kernel send buffer of drain, so + deadlines should comfortably exceed the time a healthy-but-slow client + needs to drain one (the 60s default is ample). A terminated client can + reconnect and resume from its last offset. + + Registration is cast-based: handlers never block on this server, and casts + to a stack without a running instance are dropped, so serving degrades to + unguarded rather than failing. If the server restarts, in-flight serves + re-register on their next write, leaving at most one write unguarded. + + One instance runs per stack. + """ + use GenServer + + require Logger + + @sweep_interval_ms 1_000 + + 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 """ + Register the calling process as being inside a socket write that must + complete within `timeout_ms`. + """ + def write_started(stack_id, shape_handle, timeout_ms) do + deadline = System.monotonic_time(:millisecond) + timeout_ms + GenServer.cast(name(stack_id), {:write_started, self(), deadline, timeout_ms, shape_handle}) + end + + @doc """ + Deregister the calling process's in-flight write. + """ + def write_finished(stack_id) do + GenServer.cast(name(stack_id), {:write_finished, self()}) + end + + @impl GenServer + def init(stack_id) do + Process.set_label({:serve_watchdog, stack_id}) + Logger.metadata(stack_id: stack_id) + schedule_sweep() + {:ok, %{stack_id: stack_id, writes: %{}}} + end + + @impl GenServer + def handle_cast({:write_started, pid, deadline, timeout_ms, shape_handle}, state) do + {:noreply, + %{state | writes: Map.put(state.writes, pid, {deadline, timeout_ms, shape_handle})}} + end + + def handle_cast({:write_finished, pid}, state) do + {:noreply, %{state | writes: Map.delete(state.writes, pid)}} + end + + @impl GenServer + def handle_info(:sweep, state) do + now = System.monotonic_time(:millisecond) + + writes = + for {pid, {deadline, timeout_ms, shape_handle}} = entry <- state.writes, + keep_entry?(pid, deadline, timeout_ms, shape_handle, now), + into: %{} do + entry + end + + schedule_sweep() + {:noreply, %{state | writes: writes}} + end + + defp keep_entry?(pid, deadline, timeout_ms, shape_handle, now) do + cond do + # The handler died on its own (client disconnect, error); nothing to do. + not Process.alive?(pid) -> + false + + now > deadline -> + Logger.warning( + "Terminating stalled shape response serve: client accepted no data " <> + "for #{timeout_ms}ms", + shape_handle: shape_handle + ) + + # The handler is blocked inside a socket write and cannot process + # messages, so only an untrappable exit can terminate it. + Process.exit(pid, :kill) + false + + true -> + true + end + end + + defp schedule_sweep do + Process.send_after(self(), :sweep, @sweep_interval_ms) + end +end diff --git a/packages/sync-service/lib/electric/stack_supervisor.ex b/packages/sync-service/lib/electric/stack_supervisor.ex index 4047cab785..40eadbb28d 100644 --- a/packages/sync-service/lib/electric/stack_supervisor.ex +++ b/packages/sync-service/lib/electric/stack_supervisor.ex @@ -421,6 +421,7 @@ defmodule Electric.StackSupervisor do stalled_serve_timeout: stalled_serve_timeout, feature_flags: Map.get(config, :feature_flags, []) ]}, + {Electric.Shapes.Api.ServeWatchdog, stack_id: stack_id}, {Electric.AsyncDeleter, stack_id: stack_id, storage_dir: config.storage_dir, From 859b16b15442e2762ae8f018ff2a2f4101e8006a Mon Sep 17 00:00:00 2001 From: rob Date: Wed, 15 Jul 2026 12:45:43 +0100 Subject: [PATCH 10/14] Assert the reap exit reason and drop redundant test socket cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback: matching :killed in the DOWN assertion sanity-checks that the termination came from the watchdog's untrappable kill rather than an incidental crash; and the explicit socket cleanup was dead weight — the sockets are owned by the test process and close with it. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../test/integration/stalled_serve_reaping_test.exs | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/packages/sync-service/test/integration/stalled_serve_reaping_test.exs b/packages/sync-service/test/integration/stalled_serve_reaping_test.exs index bfde83b404..07417158be 100644 --- a/packages/sync-service/test/integration/stalled_serve_reaping_test.exs +++ b/packages/sync-service/test/integration/stalled_serve_reaping_test.exs @@ -55,7 +55,7 @@ defmodule Electric.Integration.StalledServeReapingTest do [handle] = Plug.Conn.get_resp_header(snapshot, "electric-handle") # Park live long-pollers over raw sockets that will never be read from. - socks = + _socks = for _ <- 1..@stalled_clients do {:ok, s} = :gen_tcp.connect({127, 0, 0, 1}, port, [ @@ -75,8 +75,6 @@ defmodule Electric.Integration.StalledServeReapingTest do s end - on_exit(fn -> Enum.each(socks, &:gen_tcp.close/1) end) - wait_for_subscribers(registry, handle, @stalled_clients) {:ok, handlers} = ThousandIsland.connection_pids(server_pid) assert length(handlers) == @stalled_clients @@ -96,7 +94,7 @@ defmodule Electric.Integration.StalledServeReapingTest do # (plus slack). Without reaping they live — and pin memory and a file # descriptor each — until the client goes away, which may be never. for {ref, pid} <- monitors do - assert_receive {:DOWN, ^ref, :process, ^pid, _reason}, + assert_receive {:DOWN, ^ref, :process, ^pid, :killed}, @reap_deadline_ms, "stalled serve #{inspect(pid)} was not terminated within " <> "#{@stalled_serve_timeout}ms stall timeout (+ slack): nothing reaps " <> @@ -130,8 +128,6 @@ defmodule Electric.Integration.StalledServeReapingTest do buffer: 65536 ]) - on_exit(fn -> :gen_tcp.close(sock) end) - :ok = :gen_tcp.send( sock, From b20b420daf2927de4a9c9e3d07b6a5621e619672 Mon Sep 17 00:00:00 2001 From: rob Date: Wed, 15 Jul 2026 12:46:37 +0100 Subject: [PATCH 11/14] Sharpen the reaping test moduledoc on send-timeout and proxy semantics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback: the doc did not explain how the watchdog materially differs from the TCP send timeout (it times full completion of a bounded application write — a throughput floor — where the send timeout only detects a driver-level send with zero kernel acceptance for its whole window), and the buffering-proxy claim was compressed to the point of being misleading (while a proxy absorbs writes the serve completes and holds nothing; the deadline matters once the proxy stops accepting). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../stalled_serve_reaping_test.exs | 30 +++++++++++-------- 1 file changed, 18 insertions(+), 12 deletions(-) diff --git a/packages/sync-service/test/integration/stalled_serve_reaping_test.exs b/packages/sync-service/test/integration/stalled_serve_reaping_test.exs index 07417158be..f793f9895a 100644 --- a/packages/sync-service/test/integration/stalled_serve_reaping_test.exs +++ b/packages/sync-service/test/integration/stalled_serve_reaping_test.exs @@ -4,21 +4,27 @@ defmodule Electric.Integration.StalledServeReapingTest do it has made no progress for `:stalled_serve_timeout`, releasing the connection and everything the serve pins. - Defense in depth for the production OOM addressed by #4708: bounded write units - (#4708) cap what each stalled serve pins, but nothing reaps the serves + Defense in depth for the production OOM addressed by #4708: bounded write + units cap what each stalled serve pins, but nothing reaps the serves themselves — a population of stalled connections still accumulates memory and file descriptors proportional to connection count, invisibly (stalled serves never complete, so they emit no spans and increment no request - counters). The TCP send timeout only catches a *fully blocked* write; a - client draining at a trickle — or a proxy buffering for a vanished client — - evades it indefinitely. - - The reaping deadline is distinct from a slow-but-healthy client: progress - is a completed bounded socket write, so a client sustaining a modest - throughput (roughly one OS send buffer per timeout window, worst case) - keeps resetting the deadline. A serve whose client accepts nothing for the - full window is terminated; the client can reconnect and resume from its - offset. + counters). + + The kernel-level TCP send timeout does not reliably catch these: it fires + only when a single driver-level send sees no kernel acceptance for its + whole window, which a few stray bytes keep resetting — empirically, + stalled serves routinely outlive it. The watchdog times a different + quantity: the full completion of a bounded application-level write. That + turns the contract from "not perfectly frozen" into a minimum sustained + throughput, which a trickle cannot satisfy indefinitely. An intermediary + buffering on behalf of a vanished client only defers the same outcome: + while its buffer absorbs writes the serve completes and releases + everything anyway, and once it stops accepting, the deadline runs. + + The deadline never fires for a slow-but-healthy client: any client + sustaining roughly one write unit per timeout window keeps resetting it, + and a terminated client can reconnect and resume from its offset. """ use ExUnit.Case, async: false From d618f737b1fd9a4dfb1447ce832316075ca0e657 Mon Sep 17 00:00:00 2001 From: rob Date: Wed, 15 Jul 2026 13:05:05 +0100 Subject: [PATCH 12/14] Add unit tests for ServeWatchdog Direct coverage for the watchdog's contract, previously exercised only through the full-stack integration tests: a write outliving its deadline is killed (with the warning logged), writes within deadline and finished writes are left alone, a new write replaces the previous deadline, entries for processes that died on their own are pruned, and registration without a running server is silently dropped (serving degrades to unguarded). The sweep interval is now configurable via start_link opts so the tests run in under a second; the default is unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../lib/electric/shapes/api/serve_watchdog.ex | 15 +-- .../shapes/api/serve_watchdog_test.exs | 118 ++++++++++++++++++ 2 files changed, 126 insertions(+), 7 deletions(-) create mode 100644 packages/sync-service/test/electric/shapes/api/serve_watchdog_test.exs diff --git a/packages/sync-service/lib/electric/shapes/api/serve_watchdog.ex b/packages/sync-service/lib/electric/shapes/api/serve_watchdog.ex index 78009104af..d02269f6b8 100644 --- a/packages/sync-service/lib/electric/shapes/api/serve_watchdog.ex +++ b/packages/sync-service/lib/electric/shapes/api/serve_watchdog.ex @@ -37,7 +37,8 @@ defmodule Electric.Shapes.Api.ServeWatchdog do def start_link(opts) do stack_id = Keyword.fetch!(opts, :stack_id) - GenServer.start_link(__MODULE__, stack_id, name: name(stack_id)) + sweep_interval_ms = Keyword.get(opts, :sweep_interval_ms, @sweep_interval_ms) + GenServer.start_link(__MODULE__, {stack_id, sweep_interval_ms}, name: name(stack_id)) end def name(stack_id) do @@ -61,11 +62,11 @@ defmodule Electric.Shapes.Api.ServeWatchdog do end @impl GenServer - def init(stack_id) do + def init({stack_id, sweep_interval_ms}) do Process.set_label({:serve_watchdog, stack_id}) Logger.metadata(stack_id: stack_id) - schedule_sweep() - {:ok, %{stack_id: stack_id, writes: %{}}} + schedule_sweep(sweep_interval_ms) + {:ok, %{stack_id: stack_id, sweep_interval_ms: sweep_interval_ms, writes: %{}}} end @impl GenServer @@ -89,7 +90,7 @@ defmodule Electric.Shapes.Api.ServeWatchdog do entry end - schedule_sweep() + schedule_sweep(state.sweep_interval_ms) {:noreply, %{state | writes: writes}} end @@ -116,7 +117,7 @@ defmodule Electric.Shapes.Api.ServeWatchdog do end end - defp schedule_sweep do - Process.send_after(self(), :sweep, @sweep_interval_ms) + defp schedule_sweep(interval_ms) do + Process.send_after(self(), :sweep, interval_ms) end end diff --git a/packages/sync-service/test/electric/shapes/api/serve_watchdog_test.exs b/packages/sync-service/test/electric/shapes/api/serve_watchdog_test.exs new file mode 100644 index 0000000000..e801e92363 --- /dev/null +++ b/packages/sync-service/test/electric/shapes/api/serve_watchdog_test.exs @@ -0,0 +1,118 @@ +defmodule Electric.Shapes.Api.ServeWatchdogTest do + use ExUnit.Case, async: true + + import ExUnit.CaptureLog + import Support.ComponentSetup, only: [with_stack_id_from_test: 1] + + alias Electric.Shapes.Api.ServeWatchdog + + @sweep_interval_ms 25 + @shape_handle "the-shape-handle" + + setup :with_stack_id_from_test + + defp start_watchdog(ctx) do + start_supervised!( + {ServeWatchdog, stack_id: ctx.stack_id, sweep_interval_ms: @sweep_interval_ms} + ) + end + + # Spawns a process standing in for a request handler: it runs `register` + # (calls to write_started/write_finished must come from the process being + # watched), reports back, then hangs like a handler blocked in a socket + # write. + defp victim(register) do + parent = self() + + pid = + spawn(fn -> + register.() + send(parent, :registered) + Process.sleep(:infinity) + end) + + ref = Process.monitor(pid) + assert_receive :registered + {pid, ref} + end + + test "kills a process whose write outlives its deadline", ctx do + start_watchdog(ctx) + stack_id = ctx.stack_id + + log = + capture_log(fn -> + {pid, ref} = victim(fn -> ServeWatchdog.write_started(stack_id, @shape_handle, 30) end) + + assert_receive {:DOWN, ^ref, :process, ^pid, :killed}, 1_000 + end) + + assert log =~ "Terminating stalled shape response serve" + end + + test "does not kill a write that is still within its deadline", ctx do + start_watchdog(ctx) + stack_id = ctx.stack_id + + {pid, ref} = victim(fn -> ServeWatchdog.write_started(stack_id, @shape_handle, 60_000) end) + + refute_receive {:DOWN, ^ref, :process, ^pid, _reason}, 10 * @sweep_interval_ms + assert Process.alive?(pid) + end + + test "does not kill a process whose write finished", ctx do + start_watchdog(ctx) + stack_id = ctx.stack_id + + {pid, ref} = + victim(fn -> + ServeWatchdog.write_started(stack_id, @shape_handle, 30) + ServeWatchdog.write_finished(stack_id) + end) + + refute_receive {:DOWN, ^ref, :process, ^pid, _reason}, 10 * @sweep_interval_ms + assert Process.alive?(pid) + end + + test "a new write replaces the previous deadline", ctx do + start_watchdog(ctx) + stack_id = ctx.stack_id + + {pid, ref} = + victim(fn -> + ServeWatchdog.write_started(stack_id, @shape_handle, 30) + ServeWatchdog.write_finished(stack_id) + ServeWatchdog.write_started(stack_id, @shape_handle, 60_000) + end) + + refute_receive {:DOWN, ^ref, :process, ^pid, _reason}, 10 * @sweep_interval_ms + assert Process.alive?(pid) + end + + test "prunes entries for processes that died on their own", ctx do + start_watchdog(ctx) + stack_id = ctx.stack_id + parent = self() + + pid = + spawn(fn -> + ServeWatchdog.write_started(stack_id, @shape_handle, 60_000) + send(parent, :registered) + end) + + ref = Process.monitor(pid) + assert_receive :registered + assert_receive {:DOWN, ^ref, :process, ^pid, :normal} + + # After a sweep the entry is gone and the server is healthy. + Process.sleep(3 * @sweep_interval_ms) + assert %{writes: writes} = :sys.get_state(ServeWatchdog.name(stack_id)) + assert writes == %{} + end + + test "registration without a running server is silently dropped", ctx do + # No watchdog started for this stack: serving degrades to unguarded. + assert :ok = ServeWatchdog.write_started(ctx.stack_id, @shape_handle, 30) + assert :ok = ServeWatchdog.write_finished(ctx.stack_id) + end +end From b33f6d62c4f063336d549d7107c8f228812638ee Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 28 Jul 2026 09:41:48 +0100 Subject: [PATCH 13/14] Move the admission permit pd key into AdmissionControl Preparation for watchdog permit accounting: the key under which a handler stashes its held permit was private to ServeShapePlug, but the serve watchdog must read it from a killed handler's dictionary snapshot to release the permit the handler can no longer release itself. Owning the key in AdmissionControl makes that a shared definition rather than a duplicated literal. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/sync-service/lib/electric/admission_control.ex | 8 ++++++++ .../sync-service/lib/electric/plug/serve_shape_plug.ex | 2 +- 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/packages/sync-service/lib/electric/admission_control.ex b/packages/sync-service/lib/electric/admission_control.ex index eec0753fdd..31a6474798 100644 --- a/packages/sync-service/lib/electric/admission_control.ex +++ b/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. diff --git a/packages/sync-service/lib/electric/plug/serve_shape_plug.ex b/packages/sync-service/lib/electric/plug/serve_shape_plug.ex index 60eb91d8c8..c202a3e46b 100644 --- a/packages/sync-service/lib/electric/plug/serve_shape_plug.ex +++ b/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 From 9aa162a7cb16e9d6b5852591a08409f861c7b17e Mon Sep 17 00:00:00 2001 From: rob Date: Tue, 28 Jul 2026 09:49:58 +0100 Subject: [PATCH 14/14] Rework the watchdog on BIF timers; account for what a kill skips MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the top-level review (changes requested) and the transport critique. Transport: the cast-based design's message rate was proportional to write throughput with a single consumer — live-mode fanout (tens of thousands of woken long-pollers writing small elements) could queue hundreds of thousands of casts per transaction, and a backed-up mailbox made the sweep act on stale state, able to kill handlers whose writes had completed. Handlers now arm a per-scheduler BIF timer around each bounded write and cancel it on completion: the watchdog only hears about writes that overrun their deadline, so its load scales with stalls rather than throughput, and deadlines fire exactly instead of with sweep slack. A fired timer is honored only if the handler's process dictionary still carries that exact timer ref — stamped on arm, deleted on disarm *before* the asynchronous cancel, so a fired-but-being-cancelled timer cannot kill a handler whose write completed or that has moved on to another request on a keepalive connection. Kill accounting: Process.exit(:kill) destroys the handler without running its cleanup, which previously leaked the handler's admission permit (each reap permanently consuming an :existing slot until the limit returned 503s) and left reaps invisible to request metrics. From the same dictionary snapshot used for the ref check, the watchdog reads the handler's held permit, monitors, kills, and on observing the :killed exit releases the permit and emits [:electric, :plug, :serve_shape, :reaped] with the stack and shape. The theoretical window in which a handler completes its whole serve between snapshot and kill is accepted and documented, like the analogous arm/fire boundary race. Scope: serving is guarded only under Bandit, where the plug runs in the socket-owning process; under proxying adapters (Plug.Test, Cowboy in embedded mode) 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. resolve/1 fails open when no watchdog (or no stack registry) exists. Unit tests cover kill/no-kill/disarm, both stale-fire races, the died-on-its-own path, exactly-once permit release with the reap event, and fail-open resolution. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../lib/electric/shapes/api/response.ex | 94 ++++---- .../lib/electric/shapes/api/serve_watchdog.ex | 204 +++++++++++------- .../shapes/api/serve_watchdog_test.exs | 169 ++++++++++----- 3 files changed, 306 insertions(+), 161 deletions(-) diff --git a/packages/sync-service/lib/electric/shapes/api/response.ex b/packages/sync-service/lib/electric/shapes/api/response.ex index 599f14c5b4..2953c9b670 100644 --- a/packages/sync-service/lib/electric/shapes/api/response.ex +++ b/packages/sync-service/lib/electric/shapes/api/response.ex @@ -435,40 +435,36 @@ 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(stack_id, response) + watchdog = watchdog_context(conn, stack_id, response) {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 + 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) - after - watchdog && ServeWatchdog.write_finished(stack_id) - end + end + ) + end) Plug.Conn.assign(conn, :streaming_bytes_sent, bytes_sent) end @@ -487,9 +483,14 @@ defmodule Electric.Shapes.Api.Response do @socket_write_bytes Electric.Shapes.Api.Encoder.JSON.max_batch_bytes() # See Electric.Shapes.Api.ServeWatchdog for the design rationale. The - # context carries what each per-piece registration needs; `nil` (timeout - # not positive) disables reaping for this serve. - defp watchdog_context(stack_id, response) do + # 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 = @@ -499,11 +500,16 @@ defmodule Electric.Shapes.Api.Response do Electric.Config.default(:stalled_serve_timeout) ) - if is_integer(timeout) and timeout > 0 do - {stack_id, response.handle, 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 @@ -535,11 +541,17 @@ defmodule Electric.Shapes.Api.Response do Plug.Conn.chunk(conn, data) end - defp guarded_chunk(conn, data, {stack_id, shape_handle, timeout}) do - ServeWatchdog.write_started(stack_id, shape_handle, timeout) - result = Plug.Conn.chunk(conn, data) - ServeWatchdog.write_finished(stack_id) - result + 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 \\ []) diff --git a/packages/sync-service/lib/electric/shapes/api/serve_watchdog.ex b/packages/sync-service/lib/electric/shapes/api/serve_watchdog.ex index d02269f6b8..6f0186280f 100644 --- a/packages/sync-service/lib/electric/shapes/api/serve_watchdog.ex +++ b/packages/sync-service/lib/electric/shapes/api/serve_watchdog.ex @@ -11,34 +11,59 @@ defmodule Electric.Shapes.Api.ServeWatchdog do the connection survives — a population of them accumulates with connection count (reproduced in test/integration/stalled_serve_memory_test.exs). - This server instead times the completion of each bounded application-level - socket write: request handlers register before every write and deregister - after it, and a periodic sweep terminates any handler whose in-flight - write has outlived its deadline. The effective contract is a minimum - sustained throughput — one bounded write unit per deadline window — which - any live client trivially meets and a stalled one cannot. Write completion - is quantized by the OS at up to a kernel send buffer of drain, so - deadlines should comfortably exceed the time a healthy-but-slow client - needs to drain one (the 60s default is ample). A terminated client can - reconnect and resume from its last offset. - - Registration is cast-based: handlers never block on this server, and casts - to a stack without a running instance are dropped, so serving degrades to - unguarded rather than failing. If the server restarts, in-flight serves - re-register on their next write, leaving at most one write unguarded. - - One instance runs per stack. + ## 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 - @sweep_interval_ms 1_000 + @write_ref_key {__MODULE__, :write_ref} def start_link(opts) do stack_id = Keyword.fetch!(opts, :stack_id) - sweep_interval_ms = Keyword.get(opts, :sweep_interval_ms, @sweep_interval_ms) - GenServer.start_link(__MODULE__, {stack_id, sweep_interval_ms}, name: name(stack_id)) + GenServer.start_link(__MODULE__, stack_id, name: name(stack_id)) end def name(stack_id) do @@ -46,78 +71,113 @@ defmodule Electric.Shapes.Api.ServeWatchdog do end @doc """ - Register the calling process as being inside a socket write that must - complete within `timeout_ms`. + 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 write_started(stack_id, shape_handle, timeout_ms) do - deadline = System.monotonic_time(:millisecond) + timeout_ms - GenServer.cast(name(stack_id), {:write_started, self(), deadline, timeout_ms, shape_handle}) + def resolve(stack_id) do + GenServer.whereis(name(stack_id)) + rescue + ArgumentError -> nil end @doc """ - Deregister the calling process's in-flight write. + 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 write_finished(stack_id) do - GenServer.cast(name(stack_id), {:write_finished, self()}) + 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, sweep_interval_ms}) do + def init(stack_id) do Process.set_label({:serve_watchdog, stack_id}) Logger.metadata(stack_id: stack_id) - schedule_sweep(sweep_interval_ms) - {:ok, %{stack_id: stack_id, sweep_interval_ms: sweep_interval_ms, writes: %{}}} + {:ok, %{stack_id: stack_id, pending_kills: %{}}} end @impl GenServer - def handle_cast({:write_started, pid, deadline, timeout_ms, shape_handle}, state) do - {:noreply, - %{state | writes: Map.put(state.writes, pid, {deadline, timeout_ms, shape_handle})}} - end + 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 - def handle_cast({:write_finished, pid}, state) do - {:noreply, %{state | writes: Map.delete(state.writes, pid)}} + {:noreply, state} end - @impl GenServer - def handle_info(:sweep, state) do - now = System.monotonic_time(:millisecond) - - writes = - for {pid, {deadline, timeout_ms, shape_handle}} = entry <- state.writes, - keep_entry?(pid, deadline, timeout_ms, shape_handle, now), - into: %{} do - entry - end + def handle_info({:DOWN, mref, :process, _pid, reason}, state) do + {entry, pending_kills} = Map.pop(state.pending_kills, mref) - schedule_sweep(state.sweep_interval_ms) - {:noreply, %{state | writes: writes}} - end + if entry && reason == :killed do + release_permit(entry.permit) - defp keep_entry?(pid, deadline, timeout_ms, shape_handle, now) do - cond do - # The handler died on its own (client disconnect, error); nothing to do. - not Process.alive?(pid) -> - false - - now > deadline -> - Logger.warning( - "Terminating stalled shape response serve: client accepted no data " <> - "for #{timeout_ms}ms", - shape_handle: shape_handle - ) - - # The handler is blocked inside a socket write and cannot process - # messages, so only an untrappable exit can terminate it. - Process.exit(pid, :kill) - false - - true -> - true + :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 schedule_sweep(interval_ms) do - Process.send_after(self(), :sweep, interval_ms) + 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 diff --git a/packages/sync-service/test/electric/shapes/api/serve_watchdog_test.exs b/packages/sync-service/test/electric/shapes/api/serve_watchdog_test.exs index e801e92363..34d913a5f5 100644 --- a/packages/sync-service/test/electric/shapes/api/serve_watchdog_test.exs +++ b/packages/sync-service/test/electric/shapes/api/serve_watchdog_test.exs @@ -4,45 +4,42 @@ defmodule Electric.Shapes.Api.ServeWatchdogTest do import ExUnit.CaptureLog import Support.ComponentSetup, only: [with_stack_id_from_test: 1] + alias Electric.AdmissionControl alias Electric.Shapes.Api.ServeWatchdog - @sweep_interval_ms 25 @shape_handle "the-shape-handle" setup :with_stack_id_from_test - defp start_watchdog(ctx) do - start_supervised!( - {ServeWatchdog, stack_id: ctx.stack_id, sweep_interval_ms: @sweep_interval_ms} - ) + setup ctx do + watchdog = start_supervised!({ServeWatchdog, stack_id: ctx.stack_id}) + [watchdog: watchdog] end - # Spawns a process standing in for a request handler: it runs `register` - # (calls to write_started/write_finished must come from the process being - # watched), reports back, then hangs like a handler blocked in a socket - # write. - defp victim(register) do + # Spawns a process standing in for a request handler: it runs `setup_fun` + # (arm/disarm must be called from the process being watched), reports back, + # then hangs like a handler blocked in a socket write. + defp victim(setup_fun) do parent = self() pid = spawn(fn -> - register.() - send(parent, :registered) + result = setup_fun.() + send(parent, {:registered, result}) Process.sleep(:infinity) end) ref = Process.monitor(pid) - assert_receive :registered - {pid, ref} + assert_receive {:registered, result} + {pid, ref, result} end test "kills a process whose write outlives its deadline", ctx do - start_watchdog(ctx) - stack_id = ctx.stack_id + watchdog = ctx.watchdog log = capture_log(fn -> - {pid, ref} = victim(fn -> ServeWatchdog.write_started(stack_id, @shape_handle, 30) end) + {pid, ref, _} = victim(fn -> ServeWatchdog.arm(watchdog, @shape_handle, 30) end) assert_receive {:DOWN, ^ref, :process, ^pid, :killed}, 1_000 end) @@ -51,52 +48,72 @@ defmodule Electric.Shapes.Api.ServeWatchdogTest do end test "does not kill a write that is still within its deadline", ctx do - start_watchdog(ctx) - stack_id = ctx.stack_id + watchdog = ctx.watchdog + {pid, ref, _} = victim(fn -> ServeWatchdog.arm(watchdog, @shape_handle, 60_000) end) - {pid, ref} = victim(fn -> ServeWatchdog.write_started(stack_id, @shape_handle, 60_000) end) + refute_receive {:DOWN, ^ref, :process, ^pid, _reason}, 200 + assert Process.alive?(pid) + end + + test "does not kill a process whose write was disarmed", ctx do + watchdog = ctx.watchdog + + {pid, ref, _} = + victim(fn -> + timer = ServeWatchdog.arm(watchdog, @shape_handle, 30) + ServeWatchdog.disarm(timer) + end) - refute_receive {:DOWN, ^ref, :process, ^pid, _reason}, 10 * @sweep_interval_ms + refute_receive {:DOWN, ^ref, :process, ^pid, _reason}, 200 assert Process.alive?(pid) end - test "does not kill a process whose write finished", ctx do - start_watchdog(ctx) - stack_id = ctx.stack_id + test "a timeout that fires after the write completed is a no-op", ctx do + # The race an asynchronous cancel cannot prevent: the timer fires and its + # message is already in flight when the handler disarms. Delivering such + # a stale timeout must not kill the handler — disarm clears the stamped + # ref, which is what the watchdog checks. + watchdog = ctx.watchdog - {pid, ref} = + {pid, ref, timer_ref} = victim(fn -> - ServeWatchdog.write_started(stack_id, @shape_handle, 30) - ServeWatchdog.write_finished(stack_id) + timer = ServeWatchdog.arm(watchdog, @shape_handle, 60_000) + ServeWatchdog.disarm(timer) + timer end) - refute_receive {:DOWN, ^ref, :process, ^pid, _reason}, 10 * @sweep_interval_ms + send(watchdog, {:timeout, timer_ref, {:stalled_write, pid, @shape_handle}}) + + refute_receive {:DOWN, ^ref, :process, ^pid, _reason}, 200 assert Process.alive?(pid) end - test "a new write replaces the previous deadline", ctx do - start_watchdog(ctx) - stack_id = ctx.stack_id + test "a timeout for a re-armed write does not kill on the stale ref", ctx do + # Same race, one step later: the handler has moved on to its next write + # (new stamped ref) when a stale timeout for the previous one arrives. + watchdog = ctx.watchdog - {pid, ref} = + {pid, ref, old_timer} = victim(fn -> - ServeWatchdog.write_started(stack_id, @shape_handle, 30) - ServeWatchdog.write_finished(stack_id) - ServeWatchdog.write_started(stack_id, @shape_handle, 60_000) + timer = ServeWatchdog.arm(watchdog, @shape_handle, 30) + ServeWatchdog.disarm(timer) + _new_timer = ServeWatchdog.arm(watchdog, @shape_handle, 60_000) + timer end) - refute_receive {:DOWN, ^ref, :process, ^pid, _reason}, 10 * @sweep_interval_ms + send(watchdog, {:timeout, old_timer, {:stalled_write, pid, @shape_handle}}) + + refute_receive {:DOWN, ^ref, :process, ^pid, _reason}, 200 assert Process.alive?(pid) end - test "prunes entries for processes that died on their own", ctx do - start_watchdog(ctx) - stack_id = ctx.stack_id + test "a timeout for a handler that died on its own is a no-op", ctx do + watchdog = ctx.watchdog parent = self() pid = spawn(fn -> - ServeWatchdog.write_started(stack_id, @shape_handle, 60_000) + ServeWatchdog.arm(watchdog, @shape_handle, 60_000) send(parent, :registered) end) @@ -104,15 +121,71 @@ defmodule Electric.Shapes.Api.ServeWatchdogTest do assert_receive :registered assert_receive {:DOWN, ^ref, :process, ^pid, :normal} - # After a sweep the entry is gone and the server is healthy. - Process.sleep(3 * @sweep_interval_ms) - assert %{writes: writes} = :sys.get_state(ServeWatchdog.name(stack_id)) - assert writes == %{} + # Deliver a fire for the dead pid directly; the watchdog must survive. + send(watchdog, {:timeout, make_ref(), {:stalled_write, pid, @shape_handle}}) + assert Process.alive?(watchdog) end - test "registration without a running server is silently dropped", ctx do - # No watchdog started for this stack: serving degrades to unguarded. - assert :ok = ServeWatchdog.write_started(ctx.stack_id, @shape_handle, 30) - assert :ok = ServeWatchdog.write_finished(ctx.stack_id) + test "releases the killed handler's admission permit exactly once", ctx do + watchdog = ctx.watchdog + stack_id = ctx.stack_id + + events_ref = attach_reap_handler(ctx) + + {pid, ref, _} = + victim(fn -> + :ok = AdmissionControl.try_acquire(stack_id, :existing, max_concurrent: 10) + Process.put(AdmissionControl.permit_pd_key(), {stack_id, :existing}) + ServeWatchdog.arm(watchdog, @shape_handle, 30) + end) + + assert %{existing: 1} = AdmissionControl.get_current(stack_id) + + capture_log(fn -> + assert_receive {:DOWN, ^ref, :process, ^pid, :killed}, 1_000 + # The release happens when the watchdog observes the :killed exit. + assert_receive {:reaped, ^events_ref, %{stack_id: ^stack_id, shape_handle: @shape_handle}}, + 1_000 + end) + + assert %{existing: 0} = AdmissionControl.get_current(stack_id) + end + + test "a killed handler without a permit releases nothing", ctx do + watchdog = ctx.watchdog + stack_id = ctx.stack_id + + events_ref = attach_reap_handler(ctx) + + capture_log(fn -> + {pid, ref, _} = victim(fn -> ServeWatchdog.arm(watchdog, @shape_handle, 30) end) + + assert_receive {:DOWN, ^ref, :process, ^pid, :killed}, 1_000 + assert_receive {:reaped, ^events_ref, _meta}, 1_000 + end) + + assert %{existing: 0, initial: 0} = AdmissionControl.get_current(stack_id) + end + + test "resolve is nil without a running watchdog or without a stack at all", ctx do + stop_supervised!(ServeWatchdog) + assert ServeWatchdog.resolve(ctx.stack_id) == nil + assert ServeWatchdog.resolve(ctx.stack_id <> "-no-such-stack") == nil + end + + defp attach_reap_handler(ctx) do + events_ref = make_ref() + parent = self() + handler_id = {ctx.stack_id, :reap_test} + + :telemetry.attach( + handler_id, + [:electric, :plug, :serve_shape, :reaped], + fn _event, _measurements, meta, _config -> send(parent, {:reaped, events_ref, meta}) end, + nil + ) + + on_exit(fn -> :telemetry.detach(handler_id) end) + events_ref end end