diff --git a/.changeset/reap-stalled-serves.md b/.changeset/reap-stalled-serves.md
new file mode 100644
index 0000000000..b4160396d5
--- /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 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/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/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/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/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
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 6a99d06335..2953c9b670 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,6 +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 = watchdog_context(conn, stack_id, response)
 
     {conn, bytes_sent} =
       response.body
@@ -445,7 +447,7 @@ defmodule Electric.Shapes.Api.Response do
           Map.put(sample_rate_attrs, "chunk_size", chunk_size),
           stack_id,
           fn ->
-            case Plug.Conn.chunk(conn, chunk) do
+            case write_in_bounded_pieces(conn, chunk, chunk_size, watchdog) do
               {:ok, conn} ->
                 {:cont, {conn, bytes_sent + chunk_size}}
 
@@ -467,6 +469,91 @@ defmodule Electric.Shapes.Api.Response do
     Plug.Conn.assign(conn, :streaming_bytes_sent, bytes_sent)
   end
 
+  # Socket writes are split into pieces of at most this size. Bounding the
+  # write unit bounds the response data queued in the socket's driver queue
+  # at any moment and — because a bounded write to a live client completes
+  # once the transport buffers drain — gives the stall watchdog below a
+  # progress signal: each completed piece is proof the client accepted data.
+  #
+  # Derived from the encoder's batch cap so that ordinary body elements are
+  # written as-is (no flattening or splitting) and only oversized single
+  # items (e.g. one very large row) are split. Making pieces smaller would
+  # not sharpen the watchdog: write completion is quantized by the OS at up
+  # to a kernel send buffer of drain, which dwarfs the piece size.
+  @socket_write_bytes Electric.Shapes.Api.Encoder.JSON.max_batch_bytes()
+
+  # See Electric.Shapes.Api.ServeWatchdog for the design rationale. The
+  # 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 =
+      Electric.StackConfig.lookup(
+        stack_id,
+        :stalled_serve_timeout,
+        Electric.Config.default(:stalled_serve_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
+  # written directly without flattening.
+  defp write_in_bounded_pieces(conn, chunk, chunk_size, watchdog)
+       when chunk_size <= @socket_write_bytes do
+    guarded_chunk(conn, chunk, watchdog)
+  end
+
+  defp write_in_bounded_pieces(conn, chunk, _chunk_size, watchdog) do
+    chunk
+    |> IO.iodata_to_binary()
+    |> write_pieces(conn, watchdog)
+  end
+
+  defp write_pieces(<<piece::binary-size(@socket_write_bytes), rest::binary>>, conn, watchdog)
+       when byte_size(rest) > 0 do
+    case guarded_chunk(conn, piece, watchdog) do
+      {:ok, conn} -> write_pieces(rest, conn, watchdog)
+      error -> error
+    end
+  end
+
+  defp write_pieces(piece, conn, watchdog) do
+    guarded_chunk(conn, piece, watchdog)
+  end
+
+  defp guarded_chunk(conn, data, nil) do
+    Plug.Conn.chunk(conn, data)
+  end
+
+  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 \\ [])
 
   # When response contains no changes, in order to uniquely identify it and avoid
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..6f0186280f
--- /dev/null
+++ b/packages/sync-service/lib/electric/shapes/api/serve_watchdog.ex
@@ -0,0 +1,183 @@
+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).
+
+  ## 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
+
+  @write_ref_key {__MODULE__, :write_ref}
+
+  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 """
+  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 resolve(stack_id) do
+    GenServer.whereis(name(stack_id))
+  rescue
+    ArgumentError -> nil
+  end
+
+  @doc """
+  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 disarm(ref) do
+    Process.delete(@write_ref_key)
+    :erlang.cancel_timer(ref, async: true, info: false)
+    :ok
+  end
+
+  @impl GenServer
+  def init(stack_id) do
+    Process.set_label({:serve_watchdog, stack_id})
+    Logger.metadata(stack_id: stack_id)
+    {:ok, %{stack_id: stack_id, pending_kills: %{}}}
+  end
+
+  @impl GenServer
+  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
+
+    {:noreply, state}
+  end
+
+  def handle_info({:DOWN, mref, :process, _pid, reason}, state) do
+    {entry, pending_kills} = Map.pop(state.pending_kills, mref)
+
+    if entry && reason == :killed do
+      release_permit(entry.permit)
+
+      :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 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/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..40eadbb28d 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,8 +418,10 @@ 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.Shapes.Api.ServeWatchdog, stack_id: stack_id},
         {Electric.AsyncDeleter,
          stack_id: stack_id,
          storage_dir: config.storage_dir,
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..34d913a5f5
--- /dev/null
+++ b/packages/sync-service/test/electric/shapes/api/serve_watchdog_test.exs
@@ -0,0 +1,191 @@
+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.AdmissionControl
+  alias Electric.Shapes.Api.ServeWatchdog
+
+  @shape_handle "the-shape-handle"
+
+  setup :with_stack_id_from_test
+
+  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 `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 ->
+        result = setup_fun.()
+        send(parent, {:registered, result})
+        Process.sleep(:infinity)
+      end)
+
+    ref = Process.monitor(pid)
+    assert_receive {:registered, result}
+    {pid, ref, result}
+  end
+
+  test "kills a process whose write outlives its deadline", ctx do
+    watchdog = ctx.watchdog
+
+    log =
+      capture_log(fn ->
+        {pid, ref, _} = victim(fn -> ServeWatchdog.arm(watchdog, @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
+    watchdog = ctx.watchdog
+    {pid, ref, _} = victim(fn -> ServeWatchdog.arm(watchdog, @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}, 200
+    assert Process.alive?(pid)
+  end
+
+  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, timer_ref} =
+      victim(fn ->
+        timer = ServeWatchdog.arm(watchdog, @shape_handle, 60_000)
+        ServeWatchdog.disarm(timer)
+        timer
+      end)
+
+    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 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, old_timer} =
+      victim(fn ->
+        timer = ServeWatchdog.arm(watchdog, @shape_handle, 30)
+        ServeWatchdog.disarm(timer)
+        _new_timer = ServeWatchdog.arm(watchdog, @shape_handle, 60_000)
+        timer
+      end)
+
+    send(watchdog, {:timeout, old_timer, {:stalled_write, pid, @shape_handle}})
+
+    refute_receive {:DOWN, ^ref, :process, ^pid, _reason}, 200
+    assert Process.alive?(pid)
+  end
+
+  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.arm(watchdog, @shape_handle, 60_000)
+        send(parent, :registered)
+      end)
+
+    ref = Process.monitor(pid)
+    assert_receive :registered
+    assert_receive {:DOWN, ^ref, :process, ^pid, :normal}
+
+    # 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 "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
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
new file mode 100644
index 0000000000..f793f9895a
--- /dev/null
+++ b/packages/sync-service/test/integration/stalled_serve_reaping_test.exs
@@ -0,0 +1,200 @@
+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 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 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
+
+  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 2_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
+
+      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, :killed},
+                       @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
+
+  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
+        ])
+
+      :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))
+
+    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