diff --git a/.changeset/flush-tracker-writer-monitors.md b/.changeset/flush-tracker-writer-monitors.md
new file mode 100644
index 0000000000..b1c5273ee6
--- /dev/null
+++ b/.changeset/flush-tracker-writer-monitors.md
@@ -0,0 +1,7 @@
+---
+"@core/sync-service": patch
+"@core/electric-telemetry": patch
+---
+
+Prevent a dead or stalled shape consumer from pinning the replication slot's `confirmed_flush_lsn` indefinitely, which caused unbounded WAL retention. The collector now monitors the writer behind every pending flush entry — a crashed writer unpins its entry immediately, and a shape making no flush progress past a grace period is challenged and invalidated if it doesn't respond.
+
diff --git a/packages/electric-telemetry/lib/electric/telemetry/stack_telemetry.ex b/packages/electric-telemetry/lib/electric/telemetry/stack_telemetry.ex
index bdd08a6465..b47daefbc1 100644
--- a/packages/electric-telemetry/lib/electric/telemetry/stack_telemetry.ex
+++ b/packages/electric-telemetry/lib/electric/telemetry/stack_telemetry.ex
@@ -135,7 +135,9 @@ defmodule ElectricTelemetry.StackTelemetry do
       last_value("electric.admission_control.acquire.limit", tags: [:kind]),
       sum("electric.admission_control.reject.count", tags: [:kind]),
       last_value("electric.admission_control.reject.limit", tags: [:kind]),
-      distribution("electric.shape_log_collector.transaction.affected_shape_count")
+      distribution("electric.shape_log_collector.transaction.affected_shape_count"),
+      counter("electric.flush_tracker.writer_down.count", tags: [:reason_class]),
+      sum("electric.flush_tracker.stall_detected.count")
       | additional_metrics(telemetry_opts)
     ]
     |> ElectricTelemetry.keep_for_stack(telemetry_opts.stack_id)
diff --git a/packages/sync-service/config/runtime.exs b/packages/sync-service/config/runtime.exs
index cd0e4c3ecd..9291ec07d8 100644
--- a/packages/sync-service/config/runtime.exs
+++ b/packages/sync-service/config/runtime.exs
@@ -173,6 +173,9 @@ shape_enable_suspend? = env!("ELECTRIC_SHAPE_SUSPEND_CONSUMER", :boolean, nil)
 shape_suspend_after =
   env!("ELECTRIC_SHAPE_SUSPEND_AFTER", &Electric.Config.parse_human_readable_time!/1, nil)
 
+flush_stall_grace_period =
+  env!("ELECTRIC_FLUSH_STALL_GRACE_PERIOD", &Electric.Config.parse_human_readable_time!/1, nil)
+
 system_metrics_poll_interval =
   env!(
     "ELECTRIC_SYSTEM_METRICS_POLL_INTERVAL",
@@ -276,6 +279,7 @@ config :electric,
   shape_hibernate_after: shape_hibernate_after,
   shape_enable_suspend?: shape_enable_suspend?,
   shape_suspend_after: shape_suspend_after,
+  flush_stall_grace_period: flush_stall_grace_period,
   storage_dir: storage_dir,
   storage: storage_spec,
   cleanup_interval_ms:
diff --git a/packages/sync-service/lib/electric/application.ex b/packages/sync-service/lib/electric/application.ex
index 63a4ed7ffe..bd4670b7d1 100644
--- a/packages/sync-service/lib/electric/application.ex
+++ b/packages/sync-service/lib/electric/application.ex
@@ -154,6 +154,7 @@ defmodule Electric.Application do
         shape_hibernate_after: get_env(opts, :shape_hibernate_after),
         shape_enable_suspend?: get_env(opts, :shape_enable_suspend?),
         shape_suspend_after: get_env(opts, :shape_suspend_after),
+        flush_stall_grace_period: get_env(opts, :flush_stall_grace_period),
         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),
diff --git a/packages/sync-service/lib/electric/config.ex b/packages/sync-service/lib/electric/config.ex
index 2966bc351f..24be45ee4b 100644
--- a/packages/sync-service/lib/electric/config.ex
+++ b/packages/sync-service/lib/electric/config.ex
@@ -97,6 +97,10 @@ defmodule Electric.Config do
     # After hibernating, wait this duration before suspending (terminating).
     # Only applies when shape_enable_suspend? is true.
     shape_suspend_after: :timer.minutes(10),
+    # How long a pending flush entry may sit without flush progress before its
+    # shape is invalidated to unpin the stack-wide flush boundary. The storage
+    # contract already says writes slower than this should raise.
+    flush_stall_grace_period: :timer.minutes(1),
     # Sets max_requests for Bandit handler processes:
     # https://hexdocs.pm/bandit/Bandit.html#t:http_1_options/0
     # "The maximum number of requests to serve in a single HTTP/{1,2}
diff --git a/packages/sync-service/lib/electric/replication/shape_log_collector.ex b/packages/sync-service/lib/electric/replication/shape_log_collector.ex
index 7c62f3fd37..ff1cf9bfa3 100644
--- a/packages/sync-service/lib/electric/replication/shape_log_collector.ex
+++ b/packages/sync-service/lib/electric/replication/shape_log_collector.ex
@@ -46,6 +46,13 @@ defmodule Electric.Replication.ShapeLogCollector do
             consumer_registry_opts: [type: :any]
           )
 
+  @consumer_cleanup_reason Electric.ShapeCache.ShapeCleaner.consumer_cleanup_reason()
+
+  # How often to scan the FlushTracker for entries that have made no flush progress
+  # past the grace period (see the :flush_stall_grace_period stack config value).
+  @stall_check_interval 10_000
+  @stall_check_interval_floor 1_000
+
   defguardp is_ready_to_process(state)
             when is_map_key(state, :last_processed_offset) and
                    not is_nil(state.last_processed_offset)
@@ -162,6 +169,20 @@ defmodule Electric.Replication.ShapeLogCollector do
     GenServer.cast(name(stack_id), {:writer_flushed, shape_handle, offset})
   end
 
+  @doc """
+  Notifies the ShapeLogCollector that a shape's writer is alive and
+  deliberately deferring its flush notifications (e.g. buffering transactions
+  ahead of PG snapshot info or during a subquery move-in awaiting splice).
+
+  Sent by a writer in answer to a `:verify_flush_progress` challenge from the
+  stall check. Grants the shape's flush entry a fresh stall grace period so a
+  healthy deferral is not mistaken for a wedged writer.
+  """
+  @spec notify_flush_deferred(Electric.stack_id(), Electric.shape_handle()) :: :ok
+  def notify_flush_deferred(stack_id, shape_handle) do
+    GenServer.cast(name(stack_id), {:writer_flush_deferred, shape_handle})
+  end
+
   @doc """
   Returns the list of currently active shapes being tracked
   in the shape matching filters.
@@ -240,7 +261,12 @@ defmodule Electric.Replication.ShapeLogCollector do
         tracked_relations: tracker_state,
         partitions: Partitions.new(Keyword.new(opts)),
         dependency_layers: DependencyLayers.new(),
-        pids_by_shape_handle: %{},
+        # A pending FlushTracker entry always has a live monitor watching the
+        # writer pid responsible for completing it.
+        writer_monitors: %{},
+        # Shapes whose writer was challenged by the last stall check and has not
+        # shown flush progress since (see :check_stalled_flushes).
+        stall_suspects: MapSet.new(),
         event_router:
           opts
           |> Keyword.new()
@@ -257,6 +283,8 @@ defmodule Electric.Replication.ShapeLogCollector do
         registry_state: registry_state
       })
 
+    schedule_stall_check(stall_grace_period(stack_id))
+
     {:ok, state, {:continue, :restore_shapes}}
   end
 
@@ -413,9 +441,26 @@ defmodule Electric.Replication.ShapeLogCollector do
   end
 
   def handle_cast({:writer_flushed, shape_id, offset}, state) do
-    {:noreply,
-     state
-     |> Map.update!(:flush_tracker, &FlushTracker.handle_flush_notification(&1, shape_id, offset))}
+    state =
+      Map.update!(
+        state,
+        :flush_tracker,
+        &FlushTracker.handle_flush_notification(&1, shape_id, offset, now_ms())
+      )
+
+    # A flush that completes the entry removes it from the tracker; its writer no
+    # longer needs watching.
+    state =
+      if FlushTracker.tracked?(state.flush_tracker, shape_id),
+        do: state,
+        else: demonitor_writer(state, shape_id)
+
+    {:noreply, clear_stall_suspect(state, shape_id)}
+  end
+
+  def handle_cast({:writer_flush_deferred, shape_handle}, state) do
+    state = Map.update!(state, :flush_tracker, &FlushTracker.touch(&1, shape_handle, now_ms()))
+    {:noreply, clear_stall_suspect(state, shape_handle)}
   end
 
   def handle_cast(
@@ -488,6 +533,188 @@ defmodule Electric.Replication.ShapeLogCollector do
     )
   end
 
+  def handle_info({{:down, shape_handle}, ref, :process, _pid, reason}, state) do
+    state =
+      case Map.pop(state.writer_monitors, shape_handle) do
+        {{_pid, ^ref}, writer_monitors} ->
+          %{state | writer_monitors: writer_monitors}
+          |> handle_writer_down(shape_handle, reason)
+
+        {nil, _} ->
+          state
+      end
+
+    {:noreply, state}
+  end
+
+  def handle_info(:check_stalled_flushes, state) do
+    now = now_ms()
+
+    grace_period = stall_grace_period(state.stack_id)
+    schedule_stall_check(grace_period)
+
+    stalled = FlushTracker.stalled_shapes(state.flush_tracker, now, grace_period)
+
+    # Challenge-response: a stalled entry whose writer is still alive gets one
+    # chance to prove it is deliberately deferring its flushes. First time a
+    # shape shows up stalled, its monitored writer pid is sent a challenge; a
+    # healthy deferring consumer answers with a notify_flush_deferred cast,
+    # which touches the entry and clears the suspicion. Invalidated are only
+    # the suspects still stalled with no progress since the previous check's
+    # challenge, and shapes with no monitored writer left to challenge (their
+    # writer is already dead — e.g. a :killed DOWN left the entry pinned).
+    {challengeable, orphaned} =
+      Enum.split_with(stalled, &is_map_key(state.writer_monitors, &1))
+
+    {repeat_suspects, fresh_suspects} =
+      Enum.split_with(challengeable, &MapSet.member?(state.stall_suspects, &1))
+
+    Enum.each(fresh_suspects, fn shape_handle ->
+      {pid, _ref} = Map.fetch!(state.writer_monitors, shape_handle)
+      send(pid, :verify_flush_progress)
+    end)
+
+    state = %{state | stall_suspects: MapSet.new(fresh_suspects)}
+
+    case orphaned ++ repeat_suspects do
+      [] ->
+        {:noreply, state}
+
+      stalled ->
+        Logger.warning(
+          "Writers for shapes #{inspect(stalled)} have made no flush progress in over " <>
+            "#{grace_period}ms, removing the shapes to unpin the flush boundary"
+        )
+
+        OpenTelemetry.execute(
+          [:electric, :flush_tracker, :stall_detected],
+          %{count: length(stalled)},
+          %{stack_id: state.stack_id}
+        )
+
+        Electric.ShapeCache.ShapeCleaner.remove_shapes_async(state.stack_id, stalled)
+
+        # Touching re-arms the grace period: the unpinning happens via the removal
+        # chain, and if that chain is lost the stall simply re-fires one grace
+        # period later (shape removal is idempotent).
+        flush_tracker =
+          Enum.reduce(stalled, state.flush_tracker, &FlushTracker.touch(&2, &1, now))
+
+        {:noreply, %{state | flush_tracker: flush_tracker}}
+    end
+  end
+
+  # Preserve the default GenServer behaviour for unexpected messages.
+  def handle_info(msg, state) do
+    Logger.warning("#{inspect(__MODULE__)} received unexpected message: #{inspect(msg)}")
+    {:noreply, state}
+  end
+
+  # ShapeCleaner is driving this removal: shape invalidation is already in flight,
+  # so just unpin the flush entry.
+  defp handle_writer_down(state, shape_handle, @consumer_cleanup_reason) do
+    emit_writer_down_telemetry(state, :cleanup)
+    Map.update!(state, :flush_tracker, &FlushTracker.handle_shape_removed(&1, shape_handle))
+  end
+
+  # Assume supervisor teardown (deploy/stack shutdown): leave the entry pinned on
+  # purpose so a mass shutdown never mass-invalidates shapes. :noproc belongs here
+  # because it masks the real exit reason of a writer that died before we could
+  # monitor it — a deploy-time :shutdown as easily as a crash. If the assumption
+  # is wrong, the stall check self-heals one grace period later.
+  defp handle_writer_down(state, _shape_handle, reason)
+       when reason in [:shutdown, :killed, :noproc] do
+    emit_writer_down_telemetry(state, :shutdown)
+    state
+  end
+
+  # Anything else is a crash ({:shutdown, :suspend} included: a suspending consumer
+  # must have no pending flush entries, so a monitored suspend is a contract
+  # violation). Unpin immediately and make sure the shape is invalidated — it must
+  # not resume from storage that is behind the acked WAL.
+  defp handle_writer_down(state, shape_handle, reason) do
+    Logger.warning(
+      "Writer for shape #{shape_handle} exited with #{inspect(reason)} before completing " <>
+        "its flush, removing the shape"
+    )
+
+    emit_writer_down_telemetry(state, :crash)
+    Electric.ShapeCache.ShapeCleaner.remove_shapes_async(state.stack_id, [shape_handle])
+    Map.update!(state, :flush_tracker, &FlushTracker.handle_shape_removed(&1, shape_handle))
+  end
+
+  # An undeliverable reason from the registry is either a publish-level failure —
+  # the shape is gone from ShapeStatus or was removed after failing to resume, so
+  # removal is already in flight, same as a cleanup DOWN — or the exit reason the
+  # broadcast's own monitor observed for the writer.
+  defp undeliverable_down_reason({:publish, _}), do: @consumer_cleanup_reason
+  defp undeliverable_down_reason(reason), do: reason
+
+  defp emit_writer_down_telemetry(state, reason_class) do
+    OpenTelemetry.execute(
+      [:electric, :flush_tracker, :writer_down],
+      %{count: 1},
+      %{stack_id: state.stack_id, reason_class: reason_class}
+    )
+  end
+
+  defp monitor_writer(state, shape_handle, pid) do
+    case Map.fetch(state.writer_monitors, shape_handle) do
+      {:ok, {^pid, _ref}} ->
+        state
+
+      {:ok, {_old_pid, _old_ref}} ->
+        # A different pid now owns this shape's entry (resumed consumer): swap the
+        # monitor over to it.
+        state |> demonitor_writer(shape_handle) |> monitor_writer(shape_handle, pid)
+
+      :error ->
+        # Monitoring an already-dead pid yields an immediate :noproc DOWN, so a
+        # writer that dies before this call is never lost: the DOWN is classified
+        # as teardown and the stall check picks the entry up if that was wrong.
+        ref = Process.monitor(pid, tag: {:down, shape_handle})
+        %{state | writer_monitors: Map.put(state.writer_monitors, shape_handle, {pid, ref})}
+    end
+  end
+
+  defp demonitor_writer(state, shape_handle) do
+    case Map.pop(state.writer_monitors, shape_handle) do
+      {{_pid, ref}, writer_monitors} ->
+        Process.demonitor(ref, [:flush])
+        %{state | writer_monitors: writer_monitors}
+
+      {nil, _} ->
+        state
+    end
+  end
+
+  # Any flush progress answers an outstanding stall challenge: if the shape's
+  # entry stalls again later, its writer must be challenged afresh rather than
+  # invalidated as an unresponsive suspect. (The check interval is clamped to at
+  # least 1s, so with a sub-second grace period a suspect could otherwise stall
+  # again before the next check despite having answered in between.)
+  defp clear_stall_suspect(state, shape_handle) do
+    %{state | stall_suspects: MapSet.delete(state.stall_suspects, shape_handle)}
+  end
+
+  defp stall_grace_period(stack_id) do
+    Electric.StackConfig.lookup(
+      stack_id,
+      :flush_stall_grace_period,
+      Electric.Config.default(:flush_stall_grace_period)
+    )
+  end
+
+  # Check at the configured grace period when it is shorter than the default
+  # interval, so a small grace period is enforced at matching granularity
+  # (clamped below so a tiny value cannot turn the check into a busy loop).
+  defp schedule_stall_check(grace_period) do
+    grace_period
+    |> min(@stall_check_interval)
+    |> max(@stall_check_interval_floor)
+    |> then(&Process.send_after(self(), :check_stalled_flushes, &1))
+  end
+
   defp do_handle_event(%Relation{} = rel, state) do
     OpenTelemetry.with_span(
       "pg_txn.replication_client.relation_received",
@@ -565,7 +792,7 @@ defmodule Electric.Replication.ShapeLogCollector do
 
     flush_tracker =
       if txn_fragment.commit do
-        FlushTracker.handle_txn_fragment(state.flush_tracker, txn_fragment, [])
+        FlushTracker.handle_txn_fragment(state.flush_tracker, txn_fragment, [], now_ms())
       else
         state.flush_tracker
       end
@@ -647,45 +874,73 @@ defmodule Electric.Replication.ShapeLogCollector do
     OpenTelemetry.start_interval(:"shape_log_collector.publish.duration_µs")
     context = OpenTelemetry.get_current_context()
 
-    undeliverable_set =
+    {undeliverable, delivered_pids} =
       for layer <- DependencyLayers.get_for_handles(state.dependency_layers, affected_shapes),
-          reduce: MapSet.new() do
-        acc ->
+          reduce: {%{}, %{}} do
+        {undeliverable_acc, delivered_acc} ->
           # Each publish is synchronous, so layers will be processed in order
           layer_events =
             Map.new(layer, fn handle ->
               {handle, {:handle_event, Map.fetch!(events_by_handle, handle), context}}
             end)
 
-          layer_undeliverable = ConsumerRegistry.publish(layer_events, state.registry_state)
-          layer_undeliverable |> Map.keys() |> Enum.into(acc)
+          {layer_undeliverable, layer_delivered} =
+            ConsumerRegistry.publish(layer_events, state.registry_state)
+
+          {Map.merge(undeliverable_acc, layer_undeliverable),
+           Map.merge(delivered_acc, layer_delivered)}
       end
 
     OpenTelemetry.start_interval(:"shape_log_collector.set_last_processed_lsn.duration_µs")
 
     lsn = Lsn.from_integer(state.last_processed_offset.tx_offset)
     LsnTracker.set_last_processed_lsn(state.stack_id, lsn)
-    delivered_shapes = MapSet.difference(affected_shapes, undeliverable_set)
 
-    # Remove shapes from FlushTracker that were already tracked in earlier
-    # fragments but are now undeliverable. This prevents stuck flush when
-    # a consumer processes fragment 1 but crashes on fragment 2.
-    flush_tracker =
-      Enum.reduce(undeliverable_set, state.flush_tracker, fn shape_handle, tracker ->
-        FlushTracker.handle_shape_removed(tracker, shape_handle)
-      end)
+    delivered_shapes =
+      MapSet.difference(affected_shapes, undeliverable |> Map.keys() |> MapSet.new())
 
-    flush_tracker =
-      case event do
-        %TransactionFragment{commit: commit} when not is_nil(commit) ->
-          LsnTracker.broadcast_last_seen_lsn(state.stack_id, lsn)
-          FlushTracker.handle_txn_fragment(flush_tracker, event, delivered_shapes)
+    # An undeliverable shape may still hold a pending flush entry from an
+    # earlier commit. Run the failure through the same classification as a
+    # writer DOWN instead of blindly unpinning: a crash still invalidates the
+    # shape, and an ambiguous reason leaves the entry pinned for the stall
+    # check rather than disarming that backstop.
+    state =
+      Enum.reduce(undeliverable, state, fn {shape_handle, reason}, state ->
+        state = demonitor_writer(state, shape_handle)
 
-        _ ->
-          flush_tracker
-      end
+        if FlushTracker.tracked?(state.flush_tracker, shape_handle) do
+          handle_writer_down(state, shape_handle, undeliverable_down_reason(reason))
+        else
+          state
+        end
+      end)
+
+    case event do
+      %TransactionFragment{commit: commit} when not is_nil(commit) ->
+        LsnTracker.broadcast_last_seen_lsn(state.stack_id, lsn)
+
+        flush_tracker =
+          FlushTracker.handle_txn_fragment(state.flush_tracker, event, delivered_shapes, now_ms())
+
+        # Every delivered shape still tracked after this commit gets a monitor on
+        # the pid that actually received it — not just newly tracked shapes. The
+        # suspend-retry path in ConsumerRegistry.publish/2 can deliver a commit
+        # for an already-tracked shape to a fresh consumer pid while the previous
+        # pid's completing flush cast is still in our mailbox; monitor_writer
+        # swaps the monitor over and flushes the old pid's queued DOWN, so the
+        # suspended predecessor's exit is never misread as a crash.
+        Enum.reduce(delivered_shapes, %{state | flush_tracker: flush_tracker}, fn shape_handle,
+                                                                                  state ->
+          if FlushTracker.tracked?(state.flush_tracker, shape_handle) do
+            monitor_writer(state, shape_handle, Map.fetch!(delivered_pids, shape_handle))
+          else
+            state
+          end
+        end)
 
-    %{state | flush_tracker: flush_tracker}
+      _ ->
+        state
+    end
   end
 
   defp handle_relation(state, rel) do
@@ -779,11 +1034,8 @@ defmodule Electric.Replication.ShapeLogCollector do
           OpenTelemetry.start_interval(:"unsubscribe_shape.remove_from_partitions.duration_µs")
           partitions = Partitions.remove_shape(state.partitions, shape_handle)
 
-          OpenTelemetry.start_interval(
-            :"unsubscribe_shape.remove_pids_by_shape_handle.duration_µs"
-          )
-
-          pids_by_shape_handle = Map.delete(state.pids_by_shape_handle, shape_handle)
+          OpenTelemetry.start_interval(:"unsubscribe_shape.demonitor_writer.duration_µs")
+          state = demonitor_writer(state, shape_handle)
 
           OpenTelemetry.start_interval(:"unsubscribe_shape.remove_from_flush_tracker.duration_µs")
           flush_tracker = FlushTracker.handle_shape_removed(state.flush_tracker, shape_handle)
@@ -807,7 +1059,6 @@ defmodule Electric.Replication.ShapeLogCollector do
              | subscriptions: count - 1,
                event_router: event_router,
                partitions: partitions,
-               pids_by_shape_handle: pids_by_shape_handle,
                dependency_layers: dependency_layers,
                flush_tracker: flush_tracker
            }
@@ -882,4 +1133,6 @@ defmodule Electric.Replication.ShapeLogCollector do
         {:error, reason}
     end
   end
+
+  defp now_ms, do: System.monotonic_time(:millisecond)
 end
diff --git a/packages/sync-service/lib/electric/replication/shape_log_collector/flush_tracker.ex b/packages/sync-service/lib/electric/replication/shape_log_collector/flush_tracker.ex
index f377a3f162..49a858b0b6 100644
--- a/packages/sync-service/lib/electric/replication/shape_log_collector/flush_tracker.ex
+++ b/packages/sync-service/lib/electric/replication/shape_log_collector/flush_tracker.ex
@@ -16,7 +16,9 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTracker do
           last_global_flushed_offset: LogOffset.t(),
           last_seen_offset: LogOffset.t(),
           last_flushed: %{
-            optional(shape_id()) => {last_sent :: LogOffset.t(), last_flushed :: LogOffset.t()}
+            optional(shape_id()) =>
+              {last_sent :: LogOffset.t(), last_flushed :: LogOffset.t(),
+               last_progress_at :: integer()}
           },
           min_incomplete_flush_tree: :gb_trees.tree(LogOffset.t_tuple(), MapSet.t(shape_id())),
           notify_fn: (non_neg_integer() -> any())
@@ -41,18 +43,24 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTracker do
   - `last_seen_offset`
   - Pending writes Mapping:
     ```
-    Shape => {last_sent, last_flushed}
+    Shape => {last_sent, last_flushed, last_progress_at}
     ```
     - Shapes where `last_sent == last_flushed` can be considered caught-up, and can be discarded from the mapping
+    - `last_progress_at` is a monotonic timestamp (in milliseconds) always injected by the caller — this module
+      never reads the clock itself. It is set when the shape is first tracked and refreshed only by flush
+      notifications (including partial ones): progress means flush progress, so new transactions routed to an
+      already-tracked shape do not refresh it. Entries whose timestamp is too old are reported by
+      `stalled_shapes/3` so the caller can act on writers that stopped flushing.
 
   ### Algorithm:
 
-  On incoming transaction: expressed via `handle_transaction/3`
+  On incoming transaction: expressed via `handle_transaction/4`
   1. Update `last_seen_offset` to the max offset of the transaction/block we received
   2. Determine affected shapes
   3. For each shape,
-    1. If Mapping already has the shape, update `last_sent` to the max offset of the transaction
-    2. If Mapping doesn't have the shape, add it with `{last_sent, prev_log_offset}` where `prev_log_offset` is an
+    1. If Mapping already has the shape, update `last_sent` to the max offset of the transaction, keeping
+       `last_progress_at` unchanged
+    2. If Mapping doesn't have the shape, add it with `{last_sent, prev_log_offset, now}` where `prev_log_offset` is an
        artificial offset with its `tx_offset` set to one less than the incoming transaction. This is a safe upper bound
        to use, as the shape must have flushed all relevant data before this transaction, and thus even if the previous
        transaction did not affect this shape we can consider it "flushed" by the shape.
@@ -60,10 +68,10 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTracker do
      Set `last_global_flushed_offset` to equal `last_seen_offset` and notify appropriately. See step 2 of writer flush process.
   5. Wait for the writers to send the flushed offset
 
-  On writer flush (i.e. when writer notifies the central process of a flushed write) notifying with `newlast_flushed` expressed via `handle_flush_notification/3`
+  On writer flush (i.e. when writer notifies the central process of a flushed write) notifying with `newlast_flushed` expressed via `handle_flush_notification/4`
   1. Update the mapping for the shape:
     1. If `last_sent` equals to the new flush position, then we're caught up. Delete this shape from the mapping
-    2. Otherwise, replace `last_flushed` with this new value
+    2. Otherwise, replace `last_flushed` with this new value and refresh `last_progress_at`
   2. If Mapping is empty after the update, we're globally caught up - set `last_global_flushed_offset` to equal `last_seen_offset`
   3. Otherwise:
     1. Determine the new global flushed offset:
@@ -95,15 +103,25 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTracker do
     last_flushed == %{} and :gb_trees.is_empty(tree)
   end
 
-  @spec handle_txn_fragment(t(), TransactionFragment.t(), Enumerable.t(shape_id())) :: t()
+  @doc """
+  Returns true if the shape is awaiting a flush notification from storage.
+  """
+  @spec tracked?(t(), shape_id()) :: boolean()
+  def tracked?(%__MODULE__{last_flushed: last_flushed}, shape_id) do
+    is_map_key(last_flushed, shape_id)
+  end
+
+  @spec handle_txn_fragment(t(), TransactionFragment.t(), Enumerable.t(shape_id()), integer()) ::
+          t()
 
   # Commit fragment: track all shapes affected by all fragments of the transaction and update last_seen_offset.
   def handle_txn_fragment(
         %__MODULE__{} = state,
         %TransactionFragment{commit: %Commit{}, last_log_offset: last_log_offset},
-        affected_shapes
+        affected_shapes,
+        now
       ) do
-    state = track_shapes(state, last_log_offset, affected_shapes)
+    state = track_shapes(state, last_log_offset, affected_shapes, now)
 
     state = %{state | last_seen_offset: last_log_offset}
 
@@ -114,7 +132,7 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTracker do
     end
   end
 
-  defp track_shapes(state, _last_log_offset, []), do: state
+  defp track_shapes(state, _last_log_offset, [], _now), do: state
 
   defp track_shapes(
          %__MODULE__{
@@ -122,7 +140,8 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTracker do
            last_flushed: last_flushed
          } = state,
          last_log_offset,
-         affected_shapes
+         affected_shapes,
+         now
        ) do
     prev_log_offset = %LogOffset{tx_offset: last_log_offset.tx_offset - 1}
 
@@ -132,12 +151,15 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTracker do
         {last_flushed, MapSet.new()},
         fn shape, {new_last_flushed, new_shape_ids} ->
           case Map.fetch(new_last_flushed, shape) do
-            {:ok, {_, last_flushed_offset}} ->
-              {Map.put(new_last_flushed, shape, {last_log_offset, last_flushed_offset}),
-               new_shape_ids}
+            {:ok, {_, last_flushed_offset, last_progress_at}} ->
+              {Map.put(
+                 new_last_flushed,
+                 shape,
+                 {last_log_offset, last_flushed_offset, last_progress_at}
+               ), new_shape_ids}
 
             :error ->
-              {Map.put(new_last_flushed, shape, {last_log_offset, prev_log_offset}),
+              {Map.put(new_last_flushed, shape, {last_log_offset, prev_log_offset, now}),
                MapSet.put(new_shape_ids, shape)}
           end
         end
@@ -155,25 +177,26 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTracker do
     }
   end
 
-  @spec handle_flush_notification(t(), shape_id(), LogOffset.t()) :: t()
+  @spec handle_flush_notification(t(), shape_id(), LogOffset.t(), integer()) :: t()
   def handle_flush_notification(
         %__MODULE__{
           last_flushed: last_flushed,
           min_incomplete_flush_tree: min_incomplete_flush_tree
         } = state,
         shape_id,
-        last_flushed_offset
+        last_flushed_offset,
+        now
       )
       when is_map_key(last_flushed, shape_id) do
     {last_flushed, min_incomplete_flush_tree} =
       case Map.fetch!(last_flushed, shape_id) do
-        {^last_flushed_offset, prev_flushed_offset} ->
+        {^last_flushed_offset, prev_flushed_offset, _last_progress_at} ->
           {Map.delete(last_flushed, shape_id),
            min_incomplete_flush_tree
            |> delete_from_tree(prev_flushed_offset, shape_id)}
 
-        {last_sent, prev_flushed_offset} ->
-          {Map.put(last_flushed, shape_id, {last_sent, last_flushed_offset}),
+        {last_sent, prev_flushed_offset, _last_progress_at} ->
+          {Map.put(last_flushed, shape_id, {last_sent, last_flushed_offset, now}),
            min_incomplete_flush_tree
            |> delete_from_tree(prev_flushed_offset, shape_id)
            |> add_to_tree(last_flushed_offset, shape_id)}
@@ -189,13 +212,45 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTracker do
   end
 
   # If the shape is not in the mapping, then we're processing a flush notification for a shape that was removed
-  def handle_flush_notification(state, _, _last_flushed_offset) do
+  def handle_flush_notification(state, _, _last_flushed_offset, _now) do
     state
   end
 
+  @doc """
+  Return the ids of tracked shapes whose entries have made no flush progress for longer than `grace_ms`.
+
+  Every entry in the mapping is incomplete by construction (completed entries are deleted), so this
+  inspects them all.
+  """
+  @spec stalled_shapes(t(), integer(), non_neg_integer()) :: [shape_id()]
+  def stalled_shapes(%__MODULE__{last_flushed: last_flushed}, now, grace_ms) do
+    for {shape_id, {_last_sent, _last_flushed, last_progress_at}} <- last_flushed,
+        now - last_progress_at > grace_ms,
+        do: shape_id
+  end
+
+  @doc """
+  Reset `last_progress_at` for a tracked shape, granting it a fresh grace period.
+
+  No-op if the shape is not tracked.
+  """
+  @spec touch(t(), shape_id(), integer()) :: t()
+  def touch(%__MODULE__{last_flushed: last_flushed} = state, shape_id, now) do
+    case Map.fetch(last_flushed, shape_id) do
+      {:ok, {last_sent, last_flushed_offset, _last_progress_at}} ->
+        %{
+          state
+          | last_flushed: Map.put(last_flushed, shape_id, {last_sent, last_flushed_offset, now})
+        }
+
+      :error ->
+        state
+    end
+  end
+
   def handle_shape_removed(%__MODULE__{last_flushed: last_flushed} = state, shape_id) do
     case Map.fetch(last_flushed, shape_id) do
-      {:ok, {_, last_flushed_offset}} ->
+      {:ok, {_, last_flushed_offset, _}} ->
         %{
           state
           | last_flushed: Map.delete(last_flushed, shape_id),
diff --git a/packages/sync-service/lib/electric/shape_cache/shape_cleaner.ex b/packages/sync-service/lib/electric/shape_cache/shape_cleaner.ex
index 16d2cb5fef..717daeb925 100644
--- a/packages/sync-service/lib/electric/shape_cache/shape_cleaner.ex
+++ b/packages/sync-service/lib/electric/shape_cache/shape_cleaner.ex
@@ -112,9 +112,10 @@ defmodule Electric.ShapeCache.ShapeCleaner do
     Electric.Shapes.ConsumerRegistry.remove_consumer(shape_handle, stack_id)
   end
 
+  # Any `{:shutdown, x}` other than the two Electric-tagged reasons above is
+  # not a deliberate stop and falls through to the abnormal-shutdown clause.
   def handle_writer_termination(_stack_id, _shape_handle, reason)
-      when reason in [:normal, :killed, :shutdown] or
-             (is_tuple(reason) and elem(reason, 0) == :shutdown) do
+      when reason in [:normal, :killed, :shutdown] do
     :ok
   end
 
@@ -172,6 +173,11 @@ defmodule Electric.ShapeCache.ShapeCleaner do
         end
 
       {:error, _reason} ->
+        # The shape is already gone from ShapeStatus, but an earlier removal chain may
+        # have died between the ShapeStatus removal and the ShapeLogCollector removal,
+        # leaving the shape's flush entry pinned. The SLC removal is idempotent, so
+        # re-issue it instead of assuming it ever completed.
+        :ok = Electric.Replication.ShapeLogCollector.remove_shape(stack_id, shape_handle)
         {:error, :data_removed}
     end
   end
diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex
index 4991b8cdb9..c9b7e63621 100644
--- a/packages/sync-service/lib/electric/shapes/consumer.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer.ex
@@ -102,7 +102,12 @@ defmodule Electric.Shapes.Consumer do
       :ok
     end
   catch
-    :exit, _reason -> :ok
+    :exit, _reason ->
+      # The stop call timed out or the consumer exited mid-call. A consumer that is
+      # still alive at this point is wedged and would keep pinning the stack's flush
+      # boundary, so escalate to a kill.
+      Process.exit(pid, :kill)
+      :ok
   end
 
   def stop(stack_id, shape_handle, reason) do
@@ -440,13 +445,30 @@ defmodule Electric.Shapes.Consumer do
     {:noreply, state, :hibernate}
   end
 
+  # Stall challenge from the ShapeLogCollector: this shape's flush entry has made
+  # no progress past the grace period. Answer only while in a deliberate deferral
+  # phase — the touch re-arms the entry's grace period. Any other state stays
+  # silent: either flush progress is imminent anyway, or the stall check is right
+  # to invalidate this shape at its next pass.
+  def handle_info(:verify_flush_progress, state) do
+    if deferring_flush_notifications?(state) do
+      ShapeLogCollector.notify_flush_deferred(state.stack_id, state.shape_handle)
+    end
+
+    {:noreply, state, state.hibernate_after}
+  end
+
   defp consumer_suspend_enabled?(%{stack_id: stack_id}) do
     Electric.StackConfig.lookup(stack_id, :shape_enable_suspend?, true)
   end
 
+  # A suspending consumer must have flushed and notified everything it has written:
+  # an empty txn_offset_mapping and no deferred flush notification guarantee that the
+  # ShapeLogCollector holds no pending flush entry for this shape.
   defp consumer_can_suspend?(state) do
     is_snapshot_started(state) and not Shape.has_dependencies(state.shape) and
-      not state.materializer_subscribed? and is_nil(state.pending_txn)
+      not state.materializer_subscribed? and is_nil(state.pending_txn) and
+      state.txn_offset_mapping == [] and is_nil(state.pending_flush_offset)
   end
 
   defp schedule_suspend_timer(%{suspend_after: nil} = state), do: state
@@ -461,6 +483,16 @@ defmodule Electric.Shapes.Consumer do
     %{state | suspend_timer: nil}
   end
 
+  # The two phases in which the consumer deliberately sits on delivered
+  # transactions without producing flush notifications: buffering ahead of PG
+  # snapshot info, and a subquery move-in waiting to be spliced. A deferral
+  # outlasting the stall grace period would read as a wedged writer, so these are
+  # the states in which a :verify_flush_progress challenge is answered.
+  defp deferring_flush_notifications?(%State{buffering?: true}), do: true
+
+  defp deferring_flush_notifications?(%State{event_handler: event_handler}),
+    do: is_struct(event_handler, EventHandler.Subqueries.Buffering)
+
   @impl GenServer
   def terminate(reason, state) do
     Logger.debug(fn ->
@@ -1251,12 +1283,22 @@ defmodule Electric.Shapes.Consumer do
     Inspector.clean(table_oid, inspector)
   end
 
+  # Consumers must only ever stop with :normal/:shutdown or one of the two
+  # Electric-tagged shutdown reasons ({:shutdown, :cleanup} | {:shutdown, :suspend}) —
+  # anything else is classified as a crash by the ShapeLogCollector.
   defp handle_materializer_down(reason, state) do
     case {reason, state.terminating?} do
-      {_, true} -> {:noreply, state}
-      {{:shutdown, _}, false} -> {:stop, reason, state}
-      {:shutdown, false} -> {:stop, reason, state}
-      _ -> stop_and_clean(state)
+      {_, true} ->
+        {:noreply, state}
+
+      # Bare :shutdown means supervisor teardown (e.g. a deploy) — propagate quietly.
+      {:shutdown, false} ->
+        {:stop, :shutdown, state}
+
+      # A tagged shutdown means the dependency is being deliberately removed; this
+      # dependent shape cannot function without it, so clean it up like any crash.
+      _ ->
+        stop_and_clean(state)
     end
   end
 
diff --git a/packages/sync-service/lib/electric/shapes/consumer_registry.ex b/packages/sync-service/lib/electric/shapes/consumer_registry.ex
index b98ba330ae..a5a46f5165 100644
--- a/packages/sync-service/lib/electric/shapes/consumer_registry.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer_registry.ex
@@ -87,17 +87,21 @@ defmodule Electric.Shapes.ConsumerRegistry do
     :ets.insert_new(table, [{shape_handle, pid}])
   end
 
-  @spec publish(%{shape_handle() => term()}, t()) :: %{shape_handle() => term()}
-  def publish(events_by_handle, _registry_state) when events_by_handle == %{}, do: %{}
+  @spec publish(%{shape_handle() => term()}, t()) ::
+          {undeliverable :: %{shape_handle() => term()}, delivered :: %{shape_handle() => pid()}}
+  def publish(events_by_handle, _registry_state) when events_by_handle == %{}, do: {%{}, %{}}
 
   def publish(events_by_handle, registry_state) do
-    {suspended, undeliverable} = resolve_and_broadcast(events_by_handle, registry_state)
+    {suspended, undeliverable, delivered} =
+      resolve_and_broadcast(events_by_handle, registry_state)
 
     # Retry suspended consumers once with fresh consumer processes.
     # We don't expect new suspensions here since we're targeting previously
     # suspended consumers explicitly.
     Enum.each(suspended, fn {handle, _event} -> remove_consumer(handle, registry_state) end)
-    {still_suspended, retry_undeliverable} = resolve_and_broadcast(suspended, registry_state)
+
+    {still_suspended, retry_undeliverable, retry_delivered} =
+      resolve_and_broadcast(suspended, registry_state)
 
     removed_shapes =
       if still_suspended != %{} do
@@ -109,13 +113,16 @@ defmodule Electric.Shapes.ConsumerRegistry do
         %{}
       end
 
-    undeliverable
-    |> Map.merge(retry_undeliverable)
-    |> Map.merge(removed_shapes)
+    undeliverable =
+      undeliverable
+      |> Map.merge(retry_undeliverable)
+      |> Map.merge(removed_shapes)
+
+    {undeliverable, Map.merge(delivered, retry_delivered)}
   end
 
   defp resolve_and_broadcast(events_by_handle, _registry_state)
-       when events_by_handle == %{}, do: {%{}, %{}}
+       when events_by_handle == %{}, do: {%{}, %{}, %{}}
 
   defp resolve_and_broadcast(events_by_handle, %{table: table} = registry_state) do
     {to_broadcast, undeliverable} =
@@ -126,8 +133,8 @@ defmodule Electric.Shapes.ConsumerRegistry do
         end
       end)
 
-    {suspended, crashed_or_missing} = broadcast(to_broadcast)
-    {suspended, Map.merge(undeliverable, crashed_or_missing)}
+    {suspended, crashed_or_missing, delivered} = broadcast(to_broadcast)
+    {suspended, Map.merge(undeliverable, crashed_or_missing), delivered}
   end
 
   @spec remove_consumer(shape_handle(), t()) :: :ok
@@ -150,17 +157,19 @@ defmodule Electric.Shapes.ConsumerRegistry do
   Calls many GenServers asynchronously with per-handle messages and waits
   for their responses before returning.
 
-  Returns a tuple `{suspended, crashed}` where:
+  Returns a tuple `{suspended, crashed, delivered}` where:
   - `suspended` is a map of `shape_handle => event` for handles whose consumers
     suspended (these should be retried by the caller)
   - `crashed` is a map of `shape_handle => exit_reason` for handles whose consumers
     crashed (these should NOT be retried)
+  - `delivered` is a map of `shape_handle => pid` for handles whose consumers
+    processed the event
 
   There is no timeout so if the GenServers do not respond or die, this
   function will block indefinitely.
   """
   @spec broadcast([{shape_handle(), term(), pid() | nil}]) ::
-          {%{shape_handle() => term()}, %{shape_handle() => term()}}
+          {%{shape_handle() => term()}, %{shape_handle() => term()}, %{shape_handle() => pid()}}
   def broadcast(handle_event_pids) do
     # Based on OTP GenServer.call, see:
     # https://github.com/erlang/otp/blob/090c308d7c925e154240685174addaa516ea2f69/lib/stdlib/src/gen.erl#L243
@@ -173,29 +182,30 @@ defmodule Electric.Shapes.ConsumerRegistry do
     |> Enum.map(fn {handle, event, pid} ->
       ref = Process.monitor(pid)
       send(pid, {:"$gen_call", {self(), ref}, event})
-      {handle, event, ref}
+      {handle, event, ref, pid}
     end)
-    |> Enum.reduce({%{}, %{}}, fn {handle, event, ref}, {suspended, crashed} ->
+    |> Enum.reduce({%{}, %{}, %{}}, fn {handle, event, ref, pid},
+                                       {suspended, crashed, delivered} ->
       receive do
         {^ref, _reply} ->
           Process.demonitor(ref, [:flush])
-          {suspended, crashed}
+          {suspended, crashed, Map.put(delivered, handle, pid)}
 
         {:DOWN, ^ref, _, _, @consumer_suspend_reason} ->
           # Consumer is in the act of suspending as the txn arrives.
           # Return for retry (publish/2 will start a new consumer instance).
-          {Map.put(suspended, handle, event), crashed}
+          {Map.put(suspended, handle, event), crashed, delivered}
 
         {:DOWN, ^ref, _, _, reason} ->
           # Consumer crashed — do not retry, return the crash reason.
-          {suspended, Map.put(crashed, handle, reason)}
+          {suspended, Map.put(crashed, handle, reason), delivered}
       end
     end)
     |> tap(fn
-      {suspended, crashed} when suspended == %{} and crashed == %{} ->
+      {suspended, crashed, _delivered} when suspended == %{} and crashed == %{} ->
         :ok
 
-      {suspended, crashed} ->
+      {suspended, crashed, _delivered} ->
         if suspended != %{} do
           Logger.debug(fn ->
             ["Re-trying suspended shape handles ", inspect(Map.keys(suspended))]
diff --git a/packages/sync-service/lib/electric/stack_config.ex b/packages/sync-service/lib/electric/stack_config.ex
index cf51621418..0d73d136a8 100644
--- a/packages/sync-service/lib/electric/stack_config.ex
+++ b/packages/sync-service/lib/electric/stack_config.ex
@@ -31,6 +31,7 @@ defmodule Electric.StackConfig do
       shape_hibernate_after: Electric.Config.default(:shape_hibernate_after),
       shape_enable_suspend?: Electric.Config.default(:shape_enable_suspend?),
       shape_suspend_after: Electric.Config.default(:shape_suspend_after),
+      flush_stall_grace_period: Electric.Config.default(:flush_stall_grace_period),
       chunk_bytes_threshold: Electric.ShapeCache.LogChunker.default_chunk_size_threshold(),
       feature_flags: [],
       process_spawn_opts: %{},
diff --git a/packages/sync-service/lib/electric/stack_supervisor.ex b/packages/sync-service/lib/electric/stack_supervisor.ex
index 3d8a35b078..6ea0367e80 100644
--- a/packages/sync-service/lib/electric/stack_supervisor.ex
+++ b/packages/sync-service/lib/electric/stack_supervisor.ex
@@ -140,6 +140,10 @@ defmodule Electric.StackSupervisor do
                        type: :non_neg_integer,
                        default: Electric.Config.default(:shape_suspend_after)
                      ],
+                     flush_stall_grace_period: [
+                       type: :non_neg_integer,
+                       default: Electric.Config.default(:flush_stall_grace_period)
+                     ],
                      snapshot_timeout_to_first_data: [
                        type: :pos_integer,
                        default: Electric.Config.default(:snapshot_timeout_to_first_data)
@@ -360,6 +364,7 @@ defmodule Electric.StackSupervisor do
     shape_hibernate_after = Keyword.fetch!(config.tweaks, :shape_hibernate_after)
     shape_enable_suspend? = Keyword.fetch!(config.tweaks, :shape_enable_suspend?)
     shape_suspend_after = Keyword.fetch!(config.tweaks, :shape_suspend_after)
+    flush_stall_grace_period = Keyword.fetch!(config.tweaks, :flush_stall_grace_period)
     process_spawn_opts = Keyword.fetch!(config.tweaks, :process_spawn_opts)
     consumer_gc_heap_threshold = Keyword.fetch!(config.tweaks, :consumer_gc_heap_threshold)
 
@@ -411,6 +416,7 @@ defmodule Electric.StackSupervisor do
            shape_hibernate_after: shape_hibernate_after,
            shape_enable_suspend?: shape_enable_suspend?,
            shape_suspend_after: shape_suspend_after,
+           flush_stall_grace_period: flush_stall_grace_period,
            process_spawn_opts: process_spawn_opts,
            consumer_gc_heap_threshold: consumer_gc_heap_threshold,
            feature_flags: Map.get(config, :feature_flags, [])
diff --git a/packages/sync-service/test/electric/replication/shape_log_collector/flush_tracker_test.exs b/packages/sync-service/test/electric/replication/shape_log_collector/flush_tracker_test.exs
index f0fbbec103..ae61a4480c 100644
--- a/packages/sync-service/test/electric/replication/shape_log_collector/flush_tracker_test.exs
+++ b/packages/sync-service/test/electric/replication/shape_log_collector/flush_tracker_test.exs
@@ -37,7 +37,7 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTrackerTest do
 
       # apply/3 hides the intentionally invalid (non-commit) fragment from the type checker.
       assert_raise FunctionClauseError, fn ->
-        apply(FlushTracker, :handle_txn_fragment, [tracker, fragment, ["shape1"]])
+        apply(FlushTracker, :handle_txn_fragment, [tracker, fragment, ["shape1"], 0])
       end
     end
 
@@ -53,16 +53,35 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTrackerTest do
 
       # apply/3 hides the intentionally invalid (non-commit) fragment from the type checker.
       assert_raise FunctionClauseError, fn ->
-        apply(FlushTracker, :handle_txn_fragment, [tracker, fragment, []])
+        apply(FlushTracker, :handle_txn_fragment, [tracker, fragment, [], 0])
       end
     end
+
+    test "tracks affected shapes until their flushes catch up", %{tracker: tracker} do
+      tracker =
+        FlushTracker.handle_txn_fragment(tracker, batch(lsn: 1, last_offset: 10), ["shape1"], 0)
+
+      assert FlushTracker.tracked?(tracker, "shape1")
+      refute FlushTracker.tracked?(tracker, "shape2")
+
+      tracker =
+        FlushTracker.handle_txn_fragment(
+          tracker,
+          batch(lsn: 2, last_offset: 10),
+          ["shape1", "shape2"],
+          0
+        )
+
+      assert FlushTracker.tracked?(tracker, "shape1")
+      assert FlushTracker.tracked?(tracker, "shape2")
+    end
   end
 
-  describe "handle_flush_notification/3" do
+  describe "handle_flush_notification/4" do
     test "should notify immediately when last shape catches up", %{tracker: tracker} do
       tracker = handle_txn(tracker, batch(lsn: 1, last_offset: 10), ["shape1"])
 
-      _ = FlushTracker.handle_flush_notification(tracker, "shape1", LogOffset.new(1, 10))
+      _ = FlushTracker.handle_flush_notification(tracker, "shape1", LogOffset.new(1, 10), 0)
       assert_receive {:flush_confirmed, 1}
     end
 
@@ -70,7 +89,7 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTrackerTest do
       tracker
       |> handle_txn(batch(lsn: 1, last_offset: 10), ["shape1"])
       |> handle_txn(batch(lsn: 3, last_offset: 10), [])
-      |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(1, 10))
+      |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(1, 10), 0)
 
       assert_receive {:flush_confirmed, 3}
     end
@@ -79,7 +98,7 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTrackerTest do
       tracker
       |> handle_txn(batch(lsn: 5, last_offset: 10), ["shape1"])
       # Pretend we've flushed only half of this batch
-      |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(5, 5))
+      |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(5, 5), 0)
 
       assert_receive {:flush_confirmed, 4}
     end
@@ -92,16 +111,16 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTrackerTest do
         |> handle_txn(batch(lsn: 5, last_offset: 10), ["shape1", "shape2"])
         |> handle_txn(batch(lsn: 6, last_offset: 10), ["shape1"])
         |> handle_txn(batch(lsn: 7, last_offset: 10), ["shape2"])
-        |> FlushTracker.handle_flush_notification("shape2", LogOffset.new(7, 4))
-        |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(6, 3))
+        |> FlushTracker.handle_flush_notification("shape2", LogOffset.new(7, 4), 0)
+        |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(6, 3), 0)
 
       assert_receive {:flush_confirmed, 5}
 
-      tracker = FlushTracker.handle_flush_notification(tracker, "shape1", LogOffset.new(6, 10))
+      tracker = FlushTracker.handle_flush_notification(tracker, "shape1", LogOffset.new(6, 10), 0)
 
       assert_receive {:flush_confirmed, 6}
 
-      FlushTracker.handle_flush_notification(tracker, "shape2", LogOffset.new(7, 10))
+      FlushTracker.handle_flush_notification(tracker, "shape2", LogOffset.new(7, 10), 0)
 
       assert_receive {:flush_confirmed, 7}
     end
@@ -113,24 +132,32 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTrackerTest do
         tracker
         |> handle_txn(batch(lsn: 5, last_offset: 10), ["shape1", "shape2"])
         |> handle_txn(batch(lsn: 6, last_offset: 10), ["shape1", "shape2"])
-        |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(5, 10))
+        |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(5, 10), 0)
 
       assert_receive {:flush_confirmed, 3}
 
-      tracker = tracker |> FlushTracker.handle_flush_notification("shape2", LogOffset.new(5, 10))
+      tracker =
+        tracker |> FlushTracker.handle_flush_notification("shape2", LogOffset.new(5, 10), 0)
+
       assert_receive {:flush_confirmed, 4}
       refute_receive {:flush_confirmed, _}
 
-      tracker = tracker |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(6, 3))
+      tracker =
+        tracker |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(6, 3), 0)
+
       refute_receive {:flush_confirmed, _}
 
-      tracker = tracker |> FlushTracker.handle_flush_notification("shape2", LogOffset.new(6, 3))
+      tracker =
+        tracker |> FlushTracker.handle_flush_notification("shape2", LogOffset.new(6, 3), 0)
+
       assert_receive {:flush_confirmed, 5}
 
-      tracker = tracker |> FlushTracker.handle_flush_notification("shape2", LogOffset.new(6, 10))
+      tracker =
+        tracker |> FlushTracker.handle_flush_notification("shape2", LogOffset.new(6, 10), 0)
+
       refute_receive {:flush_confirmed, _}
 
-      FlushTracker.handle_flush_notification(tracker, "shape1", LogOffset.new(6, 10))
+      FlushTracker.handle_flush_notification(tracker, "shape1", LogOffset.new(6, 10), 0)
       assert_receive {:flush_confirmed, 6}
     end
 
@@ -148,13 +175,13 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTrackerTest do
         |> handle_txn(batch(lsn: 10, last_offset: 10), ["shape1"])
         |> handle_txn(batch(lsn: 11, last_offset: 10), [])
         |> handle_txn(batch(lsn: 12, last_offset: 10), ["shape2"])
-        |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(10, 10))
+        |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(10, 10), 0)
 
       assert_receive {:flush_confirmed, 10}
 
       tracker =
         tracker
-        |> FlushTracker.handle_flush_notification("shape2", LogOffset.new(12, 10))
+        |> FlushTracker.handle_flush_notification("shape2", LogOffset.new(12, 10), 0)
 
       # There's no notification for lsn=11 because by the time that txn fragment is
       # processed, flush tracker's last_flushed is non-empty: it was populated with
@@ -169,11 +196,11 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTrackerTest do
       tracker
       |> handle_txn(batch(lsn: 10, last_offset: 10), ["shape1"])
       |> handle_txn(batch(lsn: 11, last_offset: 10), ["shape2"])
-      |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(10, 10))
+      |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(10, 10), 0)
       |> handle_txn(batch(lsn: 12, last_offset: 10), ["shape1"])
-      |> FlushTracker.handle_flush_notification("shape2", LogOffset.new(11, 10))
+      |> FlushTracker.handle_flush_notification("shape2", LogOffset.new(11, 10), 0)
       |> handle_txn(batch(lsn: 13, last_offset: 10), ["shape2"])
-      |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(12, 10))
+      |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(12, 10), 0)
 
       assert_receive {:flush_confirmed, 9}
       assert_receive {:flush_confirmed, 10}
@@ -201,7 +228,7 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTrackerTest do
       # alive_shape flushes completely — global offset advances to one behind
       # the minimum incomplete (dead_shape is stuck at prev_log_offset with tx_offset=4)
       tracker =
-        FlushTracker.handle_flush_notification(tracker, "alive_shape", LogOffset.new(5, 10))
+        FlushTracker.handle_flush_notification(tracker, "alive_shape", LogOffset.new(5, 10), 0)
 
       assert_receive {:flush_confirmed, 3}
 
@@ -210,7 +237,7 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTrackerTest do
 
       # alive_shape flushes again
       tracker =
-        FlushTracker.handle_flush_notification(tracker, "alive_shape", LogOffset.new(6, 10))
+        FlushTracker.handle_flush_notification(tracker, "alive_shape", LogOffset.new(6, 10), 0)
 
       # Still stuck — dead_shape holds the global offset at 3 (tx_offset 4 minus 1)
       refute_receive {:flush_confirmed, _}
@@ -226,7 +253,7 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTrackerTest do
 
       # alive_shape flushes completely
       tracker =
-        FlushTracker.handle_flush_notification(tracker, "alive_shape", LogOffset.new(5, 10))
+        FlushTracker.handle_flush_notification(tracker, "alive_shape", LogOffset.new(5, 10), 0)
 
       assert_receive {:flush_confirmed, 3}
 
@@ -252,13 +279,13 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTrackerTest do
 
       # alive_shape flushes both txns
       tracker =
-        FlushTracker.handle_flush_notification(tracker, "alive_shape", LogOffset.new(5, 10))
+        FlushTracker.handle_flush_notification(tracker, "alive_shape", LogOffset.new(5, 10), 0)
 
       # First flush notification: dead_shape_1 is stuck at prev_log_offset tx_offset=4
       assert_receive {:flush_confirmed, 3}
 
       tracker =
-        FlushTracker.handle_flush_notification(tracker, "alive_shape", LogOffset.new(6, 10))
+        FlushTracker.handle_flush_notification(tracker, "alive_shape", LogOffset.new(6, 10), 0)
 
       # No additional notification — global offset is still stuck at dead_shape_1's position
       refute_receive {:flush_confirmed, _}
@@ -275,10 +302,79 @@ defmodule Electric.Replication.ShapeLogCollector.FlushTrackerTest do
     end
   end
 
+  describe "stall detection" do
+    test "last_progress_at is set when a shape is first tracked", %{tracker: tracker} do
+      tracker = handle_txn(tracker, batch(lsn: 5, last_offset: 10), ["shape1"], 100)
+
+      assert FlushTracker.stalled_shapes(tracker, 201, 100) == ["shape1"]
+    end
+
+    test "stalled_shapes uses a strict comparison against grace_ms", %{tracker: tracker} do
+      tracker = handle_txn(tracker, batch(lsn: 5, last_offset: 10), ["shape1"], 100)
+
+      # Exactly grace_ms since last progress is not yet stalled
+      assert FlushTracker.stalled_shapes(tracker, 200, 100) == []
+      assert FlushTracker.stalled_shapes(tracker, 201, 100) == ["shape1"]
+    end
+
+    test "last_progress_at is not refreshed by subsequent txns", %{tracker: tracker} do
+      tracker =
+        tracker
+        |> handle_txn(batch(lsn: 5, last_offset: 10), ["shape1"], 100)
+        |> handle_txn(batch(lsn: 6, last_offset: 10), ["shape1"], 500)
+
+      # Still measured from the first tracking at 100
+      assert FlushTracker.stalled_shapes(tracker, 600, 450) == ["shape1"]
+    end
+
+    test "last_progress_at is refreshed by a partial flush notification", %{tracker: tracker} do
+      tracker =
+        tracker
+        |> handle_txn(batch(lsn: 5, last_offset: 10), ["shape1"], 100)
+        # Partial flush: entry stays incomplete but counts as progress
+        |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(5, 5), 300)
+
+      assert FlushTracker.stalled_shapes(tracker, 400, 100) == []
+      assert FlushTracker.stalled_shapes(tracker, 401, 100) == ["shape1"]
+    end
+
+    test "completed entries are never reported as stalled", %{tracker: tracker} do
+      tracker =
+        tracker
+        |> handle_txn(batch(lsn: 5, last_offset: 10), ["shape1"], 100)
+        |> FlushTracker.handle_flush_notification("shape1", LogOffset.new(5, 10), 150)
+
+      assert FlushTracker.stalled_shapes(tracker, 1_000_000, 0) == []
+    end
+
+    test "touch re-arms the grace period", %{tracker: tracker} do
+      tracker =
+        tracker
+        |> handle_txn(batch(lsn: 5, last_offset: 10), ["shape1"], 100)
+        |> FlushTracker.touch("shape1", 500)
+
+      assert FlushTracker.stalled_shapes(tracker, 550, 100) == []
+      assert FlushTracker.stalled_shapes(tracker, 700, 100) == ["shape1"]
+    end
+
+    test "touch is a no-op for an untracked shape", %{tracker: tracker} do
+      assert FlushTracker.touch(tracker, "shape1", 500) == tracker
+    end
+
+    test "only shapes past the grace period are reported", %{tracker: tracker} do
+      tracker =
+        tracker
+        |> handle_txn(batch(lsn: 5, last_offset: 10), ["shape1"], 100)
+        |> handle_txn(batch(lsn: 6, last_offset: 10), ["shape2"], 400)
+
+      assert FlushTracker.stalled_shapes(tracker, 480, 100) == ["shape1"]
+    end
+  end
+
   # Helper: calls handle_txn_fragment with shapes_with_changes defaulting to
   # all affected shapes (the common case for single-fragment transactions).
-  defp handle_txn(tracker, fragment, affected_shapes) do
-    FlushTracker.handle_txn_fragment(tracker, fragment, affected_shapes)
+  defp handle_txn(tracker, fragment, affected_shapes, now \\ 0) do
+    FlushTracker.handle_txn_fragment(tracker, fragment, affected_shapes, now)
   end
 
   defp batch(opts) do
diff --git a/packages/sync-service/test/electric/replication/shape_log_collector_test.exs b/packages/sync-service/test/electric/replication/shape_log_collector_test.exs
index e4d620768b..ed449a8c48 100644
--- a/packages/sync-service/test/electric/replication/shape_log_collector_test.exs
+++ b/packages/sync-service/test/electric/replication/shape_log_collector_test.exs
@@ -1665,6 +1665,411 @@ defmodule Electric.Replication.ShapeLogCollectorTest do
     end
   end
 
+  # Adapted from the stall reproduction in PR #4713, with the outcome inverted:
+  # a consumer that dies without running its terminate callback (or wedges alive
+  # without flushing) no longer pins the FlushTracker's global minimum forever.
+  # The SLC monitors the writer pid behind every pending flush entry and runs a
+  # periodic stall check, so the boundary is unpinned and the shape invalidated
+  # without requiring new traffic or an explicit `remove_shape` call.
+  describe "FlushTracker writer monitors and stall detection" do
+    @quiet_inspector Support.StubInspector.new(
+                       tables: [{5678, {"public", "other_table"}}],
+                       columns: [%{name: "id", type: "int8", pk_position: 0}]
+                     )
+    @quiet_shape Shape.new!("other_table", inspector: @quiet_inspector)
+
+    setup :setup_log_collector
+
+    setup ctx do
+      parent = self()
+
+      stub_inspector(
+        load_relation_oid: fn
+          {"public", "test_table"}, _ -> {:ok, {1234, {"public", "test_table"}}}
+          {"public", "other_table"}, _ -> {:ok, {5678, {"public", "other_table"}}}
+        end,
+        load_relation_info: fn
+          1234, _ ->
+            {:ok, %{id: 1234, schema: "public", name: "test_table", parent: nil, children: nil}}
+
+          5678, _ ->
+            {:ok, %{id: 5678, schema: "public", name: "other_table", parent: nil, children: nil}}
+        end,
+        load_column_info: fn
+          1234, _ -> {:ok, [%{pk_position: 0, name: "id", is_generated: false}]}
+          5678, _ -> {:ok, [%{pk_position: 0, name: "id", is_generated: false}]}
+        end
+      )
+
+      consumer_alive =
+        start_supervised!(
+          {Support.TransactionConsumer,
+           id: :alive,
+           stack_id: ctx.stack_id,
+           parent: parent,
+           shape: @shape,
+           shape_handle: "shape-alive"},
+          id: {:consumer, :alive}
+        )
+
+      consumer_doomed =
+        start_supervised!(
+          {Support.TransactionConsumer,
+           id: :doomed,
+           stack_id: ctx.stack_id,
+           parent: parent,
+           shape: @quiet_shape,
+           shape_handle: "shape-doomed"},
+          id: {:consumer, :doomed}
+        )
+
+      register_as_replication_client(ctx.stack_id)
+
+      %{consumer_alive: consumer_alive, consumer_doomed: consumer_doomed}
+    end
+
+    test "crashed writer DOWN unpins the boundary and schedules removal without traffic", ctx do
+      stub_shape_cleaner(ctx)
+      attach_writer_down_telemetry(ctx)
+
+      seed_pinned_flush_entry(ctx)
+
+      # The doomed consumer crashes out-of-band (terminate/2 never runs) while its
+      # shape's table stays quiet. The writer monitor's DOWN unpins the entry
+      # immediately and schedules the shape's removal — no traffic needed.
+      Support.TransactionConsumer.crash(ctx.consumer_doomed, {:error, :simulated_disk_failure})
+
+      assert_receive {:remove_shapes_async, ["shape-doomed"]}
+      assert_receive {:flush_boundary_updated, 42}
+
+      assert_receive {:writer_down_telemetry, %{count: 1}, %{reason_class: :crash}}
+    end
+
+    test "consumer killed without cleanup is self-healed by the stall check", ctx do
+      # Emulate the real removal chain: ShapeCleaner.remove_shapes eventually issues
+      # ShapeLogCollector.remove_shape for each handle from a cleanup task.
+      stack_id = ctx.stack_id
+      parent = self()
+
+      patch_calls(Electric.ShapeCache.ShapeCleaner, [],
+        remove_shapes_async: fn ^stack_id, handles ->
+          send(parent, {:remove_shapes_async, handles})
+          spawn(fn -> Enum.each(handles, &ShapeLogCollector.remove_shape(stack_id, &1)) end)
+          :ok
+        end
+      )
+
+      seed_pinned_flush_entry(ctx)
+
+      # `:kill` is untrappable: the consumer dies without terminate/2 and the DOWN
+      # reason `:killed` is classified as supervisor teardown, so the entry stays
+      # pinned for now...
+      kill_consumer(ctx.consumer_doomed, :kill)
+      refute_receive {:flush_boundary_updated, _}
+      refute_receive {:remove_shapes_async, _}
+
+      # ...until the stall check finds it past the grace period and removes the
+      # shape, which unpins the boundary — without new traffic and without an
+      # explicit remove_shape call.
+      Electric.StackConfig.put(ctx.stack_id, :flush_stall_grace_period, 20)
+      Process.sleep(50)
+      trigger_stall_check(ctx.stack_id)
+
+      assert_receive {:remove_shapes_async, ["shape-doomed"]}
+      assert_receive {:flush_boundary_updated, 42}
+    end
+
+    test "consumer stopped with bare :shutdown keeps the entry pinned", ctx do
+      stub_shape_cleaner(ctx)
+      attach_writer_down_telemetry(ctx)
+
+      seed_pinned_flush_entry(ctx)
+
+      # Bare :shutdown is assumed to be a supervisor teardown: the entry must stay
+      # pinned and the shape must not be invalidated.
+      Support.TransactionConsumer.crash(ctx.consumer_doomed, :shutdown)
+
+      assert_receive {:writer_down_telemetry, %{count: 1}, %{reason_class: :shutdown}}
+      refute_receive {:remove_shapes_async, _}
+      refute_receive {:flush_boundary_updated, _}
+    end
+
+    test "undeliverable :noproc keeps the entry pinned and the stall check armed", ctx do
+      stub_shape_cleaner(ctx)
+      attach_writer_down_telemetry(ctx)
+
+      seed_pinned_flush_entry(ctx)
+
+      # The doomed consumer is killed; the `:killed` DOWN is classified as
+      # supervisor teardown, so the entry stays pinned and the monitor is gone.
+      kill_consumer(ctx.consumer_doomed, :kill)
+      assert_receive {:writer_down_telemetry, %{count: 1}, %{reason_class: :shutdown}}
+
+      # New traffic for its table resolves the stale registry entry to the dead
+      # pid, so the publish observes a `:noproc` — a masked reason that must not
+      # be treated as a crash (no invalidation) nor blindly unpinned (the entry
+      # must stay for the stall check).
+      lsn = Lsn.from_integer(43)
+
+      txn =
+        complete_txn_fragment(101, lsn, [
+          %Changes.NewRecord{
+            relation: {"public", "other_table"},
+            record: %{"id" => "2"},
+            log_offset: LogOffset.new(lsn, 0)
+          }
+        ])
+
+      assert :ok = ShapeLogCollector.handle_event(txn, ctx.stack_id)
+
+      assert_receive {:writer_down_telemetry, %{count: 1}, %{reason_class: :shutdown}}
+      refute_receive {:remove_shapes_async, _}
+      refute_receive {:flush_boundary_updated, _}
+
+      # The entry is still pinned, so the stall check self-heals one grace
+      # period later.
+      Electric.StackConfig.put(ctx.stack_id, :flush_stall_grace_period, 20)
+      Process.sleep(50)
+      trigger_stall_check(ctx.stack_id)
+
+      assert_receive {:remove_shapes_async, ["shape-doomed"]}
+    end
+
+    test "stall check challenges a wedged-alive consumer, then invalidates it", ctx do
+      stub_shape_cleaner(ctx)
+
+      seed_pinned_flush_entry(ctx)
+
+      # The doomed consumer is alive but never flushes. Once its entry exceeds
+      # the grace period, the stall check first challenges the writer rather
+      # than removing the shape outright.
+      Electric.StackConfig.put(ctx.stack_id, :flush_stall_grace_period, 20)
+      Process.sleep(50)
+      trigger_stall_check(ctx.stack_id)
+      assert_receive {:flush_progress_challenged, _pid}
+      refute_receive {:remove_shapes_async, _}
+
+      # The wedged consumer never answers: the next check invalidates it.
+      trigger_stall_check(ctx.stack_id)
+      assert_receive {:remove_shapes_async, ["shape-doomed"]}
+
+      # The touch re-armed the grace period: an immediate re-check does not re-fire.
+      trigger_stall_check(ctx.stack_id)
+      refute_receive {:remove_shapes_async, _}
+
+      # If the removal chain is lost (our stub drops it), the stall re-fires one
+      # grace period later — again as a challenge first, then invalidation.
+      Process.sleep(50)
+      trigger_stall_check(ctx.stack_id)
+      assert_receive {:flush_progress_challenged, _pid}
+      refute_receive {:remove_shapes_async, _}
+      trigger_stall_check(ctx.stack_id)
+      assert_receive {:remove_shapes_async, ["shape-doomed"]}
+    end
+
+    test "a challenged deferring writer answers and re-arms the grace period", ctx do
+      stub_shape_cleaner(ctx)
+
+      seed_pinned_flush_entry(ctx)
+
+      Electric.StackConfig.put(ctx.stack_id, :flush_stall_grace_period, 20)
+      Process.sleep(50)
+
+      # The stalled entry's writer is challenged; answering with
+      # notify_flush_deferred (as a deliberately deferring consumer would)
+      # touches the entry, so an immediate re-check finds nothing stalled.
+      trigger_stall_check(ctx.stack_id)
+      assert_receive {:flush_progress_challenged, _pid}
+      ShapeLogCollector.notify_flush_deferred(ctx.stack_id, "shape-doomed")
+      trigger_stall_check(ctx.stack_id)
+      refute_receive {:flush_progress_challenged, _pid}
+      refute_receive {:remove_shapes_async, _}
+
+      # A challenge answered without any stall check in between: the answer must
+      # clear the suspicion, so when the entry stalls again a full grace period
+      # later the writer is challenged afresh instead of being invalidated as an
+      # unresponsive suspect...
+      Process.sleep(50)
+      trigger_stall_check(ctx.stack_id)
+      assert_receive {:flush_progress_challenged, _pid}
+      refute_receive {:remove_shapes_async, _}
+      ShapeLogCollector.notify_flush_deferred(ctx.stack_id, "shape-doomed")
+      Process.sleep(50)
+      trigger_stall_check(ctx.stack_id)
+      assert_receive {:flush_progress_challenged, _pid}
+      refute_receive {:remove_shapes_async, _}
+
+      # ...and only an unanswered challenge invalidates the shape.
+      trigger_stall_check(ctx.stack_id)
+      assert_receive {:remove_shapes_async, ["shape-doomed"]}
+    end
+
+    test "completed entry is demonitored so later writer death has no effect", ctx do
+      stub_shape_cleaner(ctx)
+      attach_writer_down_telemetry(ctx)
+
+      seed_pinned_flush_entry(ctx)
+
+      # The doomed consumer flushes everything: its entry completes and its monitor
+      # is dropped.
+      ShapeLogCollector.notify_flushed(ctx.stack_id, "shape-doomed", LogOffset.new(42, 2))
+      assert_receive {:flush_boundary_updated, 42}
+
+      # Its subsequent death is nobody's business: no DOWN side effects.
+      Support.TransactionConsumer.crash(ctx.consumer_doomed, {:error, :simulated_disk_failure})
+
+      refute_receive {:writer_down_telemetry, _, _}
+      refute_receive {:remove_shapes_async, _}
+      refute_receive {:flush_boundary_updated, _}
+    end
+
+    test "monitor follows the writer when a commit is redelivered to a fresh pid", ctx do
+      stub_shape_cleaner(ctx)
+      attach_writer_down_telemetry(ctx)
+
+      seed_pinned_flush_entry(ctx)
+
+      # Simulate the suspend-retry hand-over: the registry entry for the
+      # still-incomplete shape is replaced by a fresh consumer before the old
+      # one's exit is observed by the SLC.
+      Electric.Shapes.ConsumerRegistry.remove_consumer("shape-doomed", ctx.stack_id)
+
+      consumer_fresh =
+        start_supervised!(
+          {Support.TransactionConsumer,
+           id: :doomed_fresh,
+           stack_id: ctx.stack_id,
+           parent: self(),
+           shape: @quiet_shape,
+           shape_handle: "shape-doomed",
+           action: :restore},
+          id: {:consumer, :doomed_fresh}
+        )
+
+      lsn = Lsn.from_integer(50)
+
+      txn =
+        complete_txn_fragment(101, lsn, [
+          %Changes.NewRecord{
+            relation: {"public", "other_table"},
+            record: %{"id" => "2"},
+            log_offset: LogOffset.new(lsn, 0)
+          }
+        ])
+
+      assert :ok = ShapeLogCollector.handle_event(txn, ctx.stack_id)
+      assert_receive {Support.TransactionConsumer, {:doomed_fresh, _}, [_]}
+
+      # The predecessor's late exit — e.g. its deferred {:shutdown, :suspend} —
+      # is no longer this shape's business: no crash classification, no
+      # invalidation. Without the monitor swap this reads as a contract
+      # violation and spuriously invalidates a healthy shape.
+      Support.TransactionConsumer.crash(ctx.consumer_doomed, {:shutdown, :suspend})
+      refute_receive {:writer_down_telemetry, _, _}
+      refute_receive {:remove_shapes_async, _}
+
+      # The monitor followed the fresh pid: its crash unpins and invalidates.
+      Support.TransactionConsumer.crash(consumer_fresh, {:error, :simulated_disk_failure})
+      assert_receive {:writer_down_telemetry, %{count: 1}, %{reason_class: :crash}}
+      assert_receive {:remove_shapes_async, ["shape-doomed"]}
+      assert_receive {:flush_boundary_updated, 50}
+    end
+
+    test "writer crashing during publish is classified as a crash for its tracked entry", ctx do
+      stub_shape_cleaner(ctx)
+      attach_writer_down_telemetry(ctx)
+
+      seed_pinned_flush_entry(ctx)
+
+      # The doomed consumer exits with a real crash reason while handling the
+      # next commit: the publish-time exit observation must classify it exactly
+      # like a monitor DOWN — immediate unpin plus invalidation, with the
+      # writer monitor's own queued DOWN flushed rather than double-handled.
+      lsn = Lsn.from_integer(50)
+
+      txn =
+        complete_txn_fragment(101, lsn, [
+          %Changes.NewRecord{
+            relation: {"public", "other_table"},
+            record: %{
+              "id" => "stop-with-reason",
+              "handle" => "shape-doomed",
+              "reason" => {:error, :simulated_disk_failure}
+            },
+            log_offset: LogOffset.new(lsn, 0)
+          }
+        ])
+
+      assert :ok = ShapeLogCollector.handle_event(txn, ctx.stack_id)
+
+      assert_receive {:writer_down_telemetry, %{count: 1}, %{reason_class: :crash}}
+      assert_receive {:remove_shapes_async, ["shape-doomed"]}
+      assert_receive {:flush_boundary_updated, 42}
+      assert_receive {:flush_boundary_updated, 50}
+    end
+  end
+
+  # Publish a txn (lsn 42) touching both test_table (shape-alive) and other_table
+  # (shape-doomed), then flush shape-alive completely. shape-doomed's entry is left
+  # as the only incomplete one, holding the flush boundary just before the txn.
+  defp seed_pinned_flush_entry(ctx) do
+    lsn = Lsn.from_integer(42)
+
+    txn =
+      complete_txn_fragment(100, lsn, [
+        %Changes.NewRecord{
+          relation: {"public", "test_table"},
+          record: %{"id" => "1"},
+          log_offset: LogOffset.new(lsn, 0)
+        },
+        %Changes.NewRecord{
+          relation: {"public", "other_table"},
+          record: %{"id" => "1"},
+          log_offset: LogOffset.new(lsn, 2)
+        }
+      ])
+
+    assert :ok = ShapeLogCollector.handle_event(txn, ctx.stack_id)
+    assert_receive {Support.TransactionConsumer, {:alive, _}, [_]}
+    assert_receive {Support.TransactionConsumer, {:doomed, _}, [_]}
+
+    ShapeLogCollector.notify_flushed(ctx.stack_id, "shape-alive", LogOffset.new(lsn, 2))
+    assert_receive {:flush_boundary_updated, 40}
+  end
+
+  defp stub_shape_cleaner(%{stack_id: stack_id}) do
+    parent = self()
+
+    patch_calls(Electric.ShapeCache.ShapeCleaner, [],
+      remove_shapes_async: fn ^stack_id, handles ->
+        send(parent, {:remove_shapes_async, handles})
+        :ok
+      end
+    )
+  end
+
+  defp attach_writer_down_telemetry(%{stack_id: stack_id, test: test}) do
+    parent = self()
+    handler_id = "writer-down-#{inspect(test)}"
+
+    :telemetry.attach(
+      handler_id,
+      [:electric, :flush_tracker, :writer_down],
+      fn _event, measurements, metadata, _config ->
+        if metadata.stack_id == stack_id do
+          send(parent, {:writer_down_telemetry, measurements, metadata})
+        end
+      end,
+      nil
+    )
+
+    on_exit(fn -> :telemetry.detach(handler_id) end)
+  end
+
+  defp trigger_stall_check(stack_id) do
+    stack_id |> ShapeLogCollector.name() |> GenServer.whereis() |> send(:check_stalled_flushes)
+  end
+
   defp kill_consumer(pid, reason) do
     ref = Process.monitor(pid)
     Process.exit(pid, reason)
diff --git a/packages/sync-service/test/electric/shape_cleaner_test.exs b/packages/sync-service/test/electric/shape_cleaner_test.exs
index 1130d40d65..2bc3655ec4 100644
--- a/packages/sync-service/test/electric/shape_cleaner_test.exs
+++ b/packages/sync-service/test/electric/shape_cleaner_test.exs
@@ -199,6 +199,24 @@ defmodule Electric.ShapeCleanerTest do
         {:ok, _} =
           with_log(fn -> ShapeCleaner.remove_shape(ctx.stack_id, shape_handle) end)
       end
+
+      # An earlier removal chain may have died between the ShapeStatus removal and
+      # the SLC removal, leaving the shape's flush entry pinned. A retry must not
+      # short-circuit past the SLC removal just because ShapeStatus already errors.
+      test "re-issues the SLC removal when the shape is already gone from ShapeStatus", ctx do
+        parent = self()
+
+        patch_calls(Electric.Replication.ShapeLogCollector,
+          remove_shape: fn _stack_id, handle ->
+            send(parent, {:slc_remove_shape, handle})
+            :ok
+          end
+        )
+
+        :ok = @cleanup_fn.(ctx.stack_id, "already-gone")
+
+        assert_receive {:slc_remove_shape, "already-gone"}
+      end
     end
   end
 
@@ -308,6 +326,49 @@ defmodule Electric.ShapeCleanerTest do
     end
   end
 
+  describe "handle_writer_termination/3" do
+    test "an untagged {:shutdown, term} reason triggers shape removal", ctx do
+      parent = self()
+
+      patch_calls(Electric.ShapeCache.ShapeCleaner.CleanupTaskSupervisor,
+        perform_async: fn _stack_id, _fun ->
+          send(parent, :removal_scheduled)
+          :ok
+        end
+      )
+
+      log =
+        capture_log(fn ->
+          assert :removed =
+                   ShapeCleaner.handle_writer_termination(
+                     ctx.stack_id,
+                     "some-handle",
+                     {:shutdown, :arbitrary}
+                   )
+        end)
+
+      assert log =~ "abnormal shutdown"
+      assert_receive :removal_scheduled
+    end
+
+    test "benign termination reasons do not trigger removal", ctx do
+      parent = self()
+
+      patch_calls(Electric.ShapeCache.ShapeCleaner.CleanupTaskSupervisor,
+        perform_async: fn _stack_id, _fun ->
+          send(parent, :removal_scheduled)
+          :ok
+        end
+      )
+
+      for reason <- [:normal, :killed, :shutdown] do
+        assert :ok = ShapeCleaner.handle_writer_termination(ctx.stack_id, "some-handle", reason)
+      end
+
+      refute_receive :removal_scheduled, 100
+    end
+  end
+
   defp assert_shape_log_collector_active_shapes(ctx, shape_handles_active, timeout \\ 500) do
     assert shape_handles_active ==
              Electric.Replication.ShapeLogCollector.active_shapes(ctx.stack_id)
diff --git a/packages/sync-service/test/electric/shapes/consumer_registry_test.exs b/packages/sync-service/test/electric/shapes/consumer_registry_test.exs
index 7419687747..71a1f7d144 100644
--- a/packages/sync-service/test/electric/shapes/consumer_registry_test.exs
+++ b/packages/sync-service/test/electric/shapes/consumer_registry_test.exs
@@ -60,9 +60,10 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
     test "starts consumer when receiving a message", ctx do
       assert ConsumerRegistry.active_consumer_count(ctx.stack_id) == 0
 
-      assert %{} ==
+      assert {%{}, %{"handle-1" => pid}} =
                ConsumerRegistry.publish(%{"handle-1" => {:txn, %{lsn: 1}}}, ctx.registry_state)
 
+      assert is_pid(pid)
       assert_receive {:start_consumer, "handle-1"}
       assert_receive {:broadcast, "handle-1", {:txn, %{lsn: 1}}}
       assert ConsumerRegistry.active_consumer_count(ctx.stack_id) == 1
@@ -71,14 +72,14 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
     test "uses existing consumer when already active", ctx do
       assert ConsumerRegistry.active_consumer_count(ctx.stack_id) == 0
 
-      assert %{} ==
+      assert {%{}, %{"handle-1" => pid}} =
                ConsumerRegistry.publish(%{"handle-1" => {:txn, %{lsn: 1}}}, ctx.registry_state)
 
       assert_receive {:start_consumer, "handle-1"}
       assert_receive {:broadcast, "handle-1", {:txn, %{lsn: 1}}}
       assert ConsumerRegistry.active_consumer_count(ctx.stack_id) == 1
 
-      assert %{} ==
+      assert {%{}, %{"handle-1" => ^pid}} =
                ConsumerRegistry.publish(%{"handle-1" => {:txn, %{lsn: 2}}}, ctx.registry_state)
 
       assert ConsumerRegistry.active_consumer_count(ctx.stack_id) == 1
@@ -89,14 +90,14 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
     test "starts any missing consumers", ctx do
       assert ConsumerRegistry.active_consumer_count(ctx.stack_id) == 0
 
-      assert %{} ==
+      assert {%{}, %{"handle-1" => _}} =
                ConsumerRegistry.publish(%{"handle-1" => {:txn, %{lsn: 1}}}, ctx.registry_state)
 
       assert_receive {:start_consumer, "handle-1"}
       assert_receive {:broadcast, "handle-1", {:txn, %{lsn: 1}}}
       assert ConsumerRegistry.active_consumer_count(ctx.stack_id) == 1
 
-      assert %{} ==
+      assert {%{}, %{"handle-1" => _, "handle-2" => _}} =
                ConsumerRegistry.publish(
                  %{"handle-1" => {:txn, %{lsn: 2}}, "handle-2" => {:txn, %{lsn: 2}}},
                  ctx.registry_state
@@ -155,7 +156,7 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
 
       assert ConsumerRegistry.active_consumer_count(ctx.stack_id) == 3
 
-      assert %{} ==
+      assert {%{}, delivered} =
                ConsumerRegistry.publish(
                  %{
                    "handle-1" => {:txn, %{lsn: 1}},
@@ -165,6 +166,8 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
                  ctx.registry_state
                )
 
+      assert Map.keys(delivered) |> Enum.sort() == ["handle-1", "handle-2", "handle-3"]
+
       assert_receive {:start_consumer, "handle-1"}
       assert_receive {:start_consumer, "handle-2"}, 10
 
@@ -187,13 +190,14 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
       assert ConsumerRegistry.active_consumer_count(ctx.stack_id) == 1
 
       # Crashed consumers are NOT retried — they are returned as undeliverable
-      result =
+      {undeliverable, delivered} =
         ConsumerRegistry.publish(
           %{"handle-crash" => {:txn, %{lsn: 1}}},
           ctx.registry_state
         )
 
-      assert :noproc == Map.fetch!(result, "handle-crash")
+      assert :noproc == Map.fetch!(undeliverable, "handle-crash")
+      assert delivered == %{}
 
       # No replacement consumer should have been started
       refute_receive {:start_consumer, "handle-crash"}
@@ -234,13 +238,14 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
       )
 
       # Dead PID is detected as crashed, returned as undeliverable
-      result =
+      {undeliverable, delivered} =
         ConsumerRegistry.publish(
           %{"handle-removed" => {:txn, %{lsn: 1}}},
           ctx.registry_state
         )
 
-      assert :noproc == Map.fetch!(result, "handle-removed")
+      assert :noproc == Map.fetch!(undeliverable, "handle-removed")
+      assert delivered == %{}
     end
 
     test "consumer that crashes during event processing is returned as undeliverable", ctx do
@@ -271,7 +276,7 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
           id: :healthy_subscriber
         )
 
-      result =
+      {undeliverable, delivered} =
         ConsumerRegistry.publish(
           %{
             "handle-crash" => {:txn, %{lsn: 1}},
@@ -283,9 +288,10 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
       assert_receive {:broadcast, "handle-ok", {:txn, %{lsn: 1}}}
 
       # Crashed handle is undeliverable with the crash reason
-      assert :processing_error == Map.fetch!(result, "handle-crash")
+      assert :processing_error == Map.fetch!(undeliverable, "handle-crash")
       # Healthy handle delivered successfully
-      refute Map.has_key?(result, "handle-ok")
+      refute Map.has_key?(undeliverable, "handle-ok")
+      assert Map.keys(delivered) == ["handle-ok"]
     end
 
     test "suspended consumers are retried but crashed consumers are not", ctx do
@@ -330,7 +336,7 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
           id: :ok_subscriber
         )
 
-      result =
+      {undeliverable, delivered} =
         ConsumerRegistry.publish(
           %{
             "handle-suspend" => {:txn, %{lsn: 1}},
@@ -347,12 +353,13 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
       assert_receive {:broadcast, "handle-suspend", {:txn, %{lsn: 1}}}
 
       # Crashed handle is undeliverable — NOT retried
-      assert :boom == Map.fetch!(result, "handle-crash")
+      assert :boom == Map.fetch!(undeliverable, "handle-crash")
       refute_receive {:start_consumer, "handle-crash"}
 
       # Healthy and retried-suspended handles delivered successfully
-      refute Map.has_key?(result, "handle-ok")
-      refute Map.has_key?(result, "handle-suspend")
+      refute Map.has_key?(undeliverable, "handle-ok")
+      refute Map.has_key?(undeliverable, "handle-suspend")
+      assert Map.keys(delivered) |> Enum.sort() == ["handle-ok", "handle-suspend"]
     end
 
     test "persistently suspending consumer results in shape removal after retry",
@@ -392,12 +399,13 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
       end)
 
       # First broadcast: suspended → retry broadcast: also suspended → remove shape
-      result =
+      {undeliverable, delivered} =
         ConsumerRegistry.publish(%{"handle-stubborn" => {:txn, %{lsn: 1}}}, ctx.registry_state)
 
       assert_receive {ShapeCleaner, :remove_shapes, ["handle-stubborn"]}
 
-      assert %{"handle-stubborn" => {:publish, :shape_removed}} == result
+      assert %{"handle-stubborn" => {:publish, :shape_removed}} == undeliverable
+      assert delivered == %{}
 
       # A new consumer has been started and suspended twice during the test
       assert_receive {:consumer_pid, pid}
@@ -424,7 +432,9 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
 
       assert ConsumerRegistry.active_consumer_count(ctx.stack_id) == 1
 
-      assert %{} == ConsumerRegistry.publish(%{handle => {:txn, %{lsn: 1}}}, ctx.registry_state)
+      assert {%{}, %{^handle => ^pid}} =
+               ConsumerRegistry.publish(%{handle => {:txn, %{lsn: 1}}}, ctx.registry_state)
+
       assert_receive {:broadcast, ^handle, {:txn, %{lsn: 1}}}
       refute_receive {:start_consumer, ^handle}, 10
     end
@@ -460,7 +470,9 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
 
       assert ConsumerRegistry.active_consumer_count(ctx.stack_id) == 1
 
-      assert %{} == ConsumerRegistry.publish(%{handle => {:txn, %{lsn: 1}}}, ctx.registry_state)
+      assert {%{}, %{^handle => _}} =
+               ConsumerRegistry.publish(%{handle => {:txn, %{lsn: 1}}}, ctx.registry_state)
+
       assert_receive {:broadcast, ^handle, {:txn, %{lsn: 1}}}
       refute_receive {:start_consumer, ^handle}, 10
 
@@ -468,7 +480,7 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
 
       assert ConsumerRegistry.active_consumer_count(ctx.stack_id) == 0
 
-      assert %{} ==
+      assert {%{}, %{"handle-1" => _}} =
                ConsumerRegistry.publish(%{"handle-1" => {:txn, %{lsn: 1}}}, ctx.registry_state)
 
       assert_receive {:start_consumer, "handle-1"}
@@ -511,7 +523,7 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
           {:reply, :ok, state}
         end)
 
-      assert {%{}, %{}} =
+      assert {%{}, %{}, %{"handle-1" => ^sub1, "handle-2" => ^sub2}} =
                ConsumerRegistry.broadcast([
                  {"handle-1", :test_message_1, sub1},
                  {"handle-2", :test_message_2, sub2}
@@ -536,7 +548,7 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
       {:ok, sub2} = TestSubscriber.start_link(on_message)
 
       Task.async(fn ->
-        assert {%{}, %{}} =
+        assert {%{}, %{}, %{"h-1" => ^sub1, "h-2" => ^sub2}} =
                  ConsumerRegistry.broadcast([
                    {"h-1", :test_message, sub1},
                    {"h-2", :test_message, sub2}
@@ -572,7 +584,7 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
       pid = self()
 
       Task.async(fn ->
-        {_suspended, crashed} =
+        {_suspended, crashed, delivered} =
           ConsumerRegistry.broadcast([
             {"h-1", :test_message, sub1},
             {"h-2", :test_message, sub2}
@@ -580,6 +592,7 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
 
         # sub2 was killed, so it appears in crashed
         assert Map.has_key?(crashed, "h-2")
+        assert Map.keys(delivered) == ["h-1"]
 
         send(pid, :publish_finished)
       end)
@@ -614,7 +627,7 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
       {:ok, sub2} = start_supervised({TestSubscriber, on_message_suspend}, id: :subscriber2)
       {:ok, sub3} = start_supervised({TestSubscriber, on_message}, id: :subscriber3)
 
-      {suspended, crashed} =
+      {suspended, crashed, delivered} =
         ConsumerRegistry.broadcast([
           {"h-1", :test_message, sub1},
           {"h-2", :test_message, sub2},
@@ -623,6 +636,7 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
 
       assert Map.keys(suspended) |> Enum.sort() == ["h-1", "h-2"]
       assert crashed == %{}
+      assert Map.keys(delivered) == ["h-3"]
 
       assert_receive :message_received
       assert_receive :message_received
@@ -640,7 +654,7 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
           {:reply, :ok, state}
         end)
 
-      assert {%{}, %{}} =
+      assert {%{}, %{}, %{"valid-shape" => ^subscriber}} =
                ConsumerRegistry.broadcast([
                  {"valid-shape", :event, subscriber},
                  {"removed-shape", :event, nil}
@@ -673,7 +687,7 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
           id: :healthy_subscriber
         )
 
-      {suspended, crashed} =
+      {suspended, crashed, delivered} =
         ConsumerRegistry.broadcast([
           {"crash-handle", :test_event, crash_sub},
           {"healthy-handle", :test_event, healthy_sub}
@@ -687,6 +701,7 @@ defmodule Electric.Shapes.ConsumerRegistryTest do
       # The healthy handle should NOT appear in either map
       refute Map.has_key?(crashed, "healthy-handle")
       refute Map.has_key?(suspended, "healthy-handle")
+      assert Map.keys(delivered) == ["healthy-handle"]
     end
   end
 end
diff --git a/packages/sync-service/test/electric/shapes/consumer_test.exs b/packages/sync-service/test/electric/shapes/consumer_test.exs
index 4fe0bb44d9..9b53f96abb 100644
--- a/packages/sync-service/test/electric/shapes/consumer_test.exs
+++ b/packages/sync-service/test/electric/shapes/consumer_test.exs
@@ -1859,6 +1859,71 @@ defmodule Electric.Shapes.ConsumerTest do
       assert Process.alive?(consumer_pid)
     end
 
+    @tag hibernate_after: 10, shape_suspend_after: 300
+    @tag suspend: true
+    test "should not suspend while a written txn is not yet flush-notified", ctx do
+      {shape_handle, _} = ShapeCache.get_or_create_shape_handle(@shape1, ctx.stack_id)
+
+      :started = ShapeCache.await_snapshot_start(shape_handle, ctx.stack_id)
+
+      consumer_pid = Consumer.whereis(ctx.stack_id, shape_handle)
+      assert is_pid(consumer_pid)
+      ref = Process.monitor(consumer_pid)
+
+      # Once hibernated, a suspend timer is armed.
+      assert is_reference(await_hibernation(consumer_pid))
+
+      # Inject un-notified txn state, as if a write had not yet been confirmed flushed.
+      offset = LogOffset.new(Lsn.from_integer(300), 0)
+
+      :sys.replace_state(consumer_pid, fn state ->
+        %{state | txn_offset_mapping: [{offset, offset}]}
+      end)
+
+      # The armed suspend timer fires but the consumer must refuse to suspend while
+      # a flush notification is outstanding.
+      refute_receive {:DOWN, ^ref, :process, ^consumer_pid, {:shutdown, :suspend}}, 500
+      assert Process.alive?(consumer_pid)
+
+      # Once the flush is confirmed and notified, the next suspend cycle goes through.
+      :sys.replace_state(consumer_pid, fn state -> %{state | txn_offset_mapping: []} end)
+      send(consumer_pid, {:configure_suspend, 5, 5, 10})
+
+      assert_receive {:DOWN, ^ref, :process, ^consumer_pid, {:shutdown, :suspend}}, 500
+    end
+
+    @tag hibernate_after: 10, shape_suspend_after: 300
+    @tag suspend: true
+    test "should not suspend while a deferred flush notification is pending", ctx do
+      {shape_handle, _} = ShapeCache.get_or_create_shape_handle(@shape1, ctx.stack_id)
+
+      :started = ShapeCache.await_snapshot_start(shape_handle, ctx.stack_id)
+
+      consumer_pid = Consumer.whereis(ctx.stack_id, shape_handle)
+      assert is_pid(consumer_pid)
+      ref = Process.monitor(consumer_pid)
+
+      # Once hibernated, a suspend timer is armed.
+      assert is_reference(await_hibernation(consumer_pid))
+
+      # Inject a deferred flush notification, as if a flush had been signalled in the
+      # middle of a multi-fragment transaction.
+      offset = LogOffset.new(Lsn.from_integer(300), 0)
+
+      :sys.replace_state(consumer_pid, fn state -> %{state | pending_flush_offset: offset} end)
+
+      # The armed suspend timer fires but the consumer must refuse to suspend while
+      # the deferred notification has not been delivered.
+      refute_receive {:DOWN, ^ref, :process, ^consumer_pid, {:shutdown, :suspend}}, 500
+      assert Process.alive?(consumer_pid)
+
+      # Once the deferred notification is delivered, the next suspend cycle goes through.
+      :sys.replace_state(consumer_pid, fn state -> %{state | pending_flush_offset: nil} end)
+      send(consumer_pid, {:configure_suspend, 5, 5, 10})
+
+      assert_receive {:DOWN, ^ref, :process, ^consumer_pid, {:shutdown, :suspend}}, 500
+    end
+
     @tag with_pure_file_storage_opts: [compaction_period: 5, keep_complete_chunks: 133]
     test "compaction is scheduled and invoked for a shape that has compaction enabled", ctx do
       parent = self()
@@ -3103,6 +3168,54 @@ defmodule Electric.Shapes.ConsumerTest do
     end
   end
 
+  describe "stall challenge response" do
+    # with_stack_id_from_test (line 87) already starts ProcessRegistry + StackConfig
+    # for ctx.stack_id; the GenServer callbacks are invoked directly with a synthetic
+    # state, with the test process registered under the ShapeLogCollector's name to
+    # receive the consumer's casts.
+
+    setup ctx do
+      {:via, Registry, {registry_name, key}} = ShapeLogCollector.name(ctx.stack_id)
+      {:ok, _} = Registry.register(registry_name, key, nil)
+      :ok
+    end
+
+    test "challenge is answered while buffering ahead of PG snapshot info", ctx do
+      state = Consumer.State.new(ctx.stack_id, "deferring-shape")
+      assert state.buffering?
+
+      assert {:noreply, ^state, _} = Consumer.handle_info(:verify_flush_progress, state)
+
+      assert_receive {:"$gen_cast", {:writer_flush_deferred, "deferring-shape"}}
+    end
+
+    test "a subquery move-in buffering phase counts as deferring", ctx do
+      handler = %Consumer.EventHandler.Subqueries.Buffering{
+        shape_info: nil,
+        queue: nil,
+        active_move: nil
+      }
+
+      state = %{
+        Consumer.State.new(ctx.stack_id, "move-in-shape")
+        | buffering?: false,
+          event_handler: handler
+      }
+
+      assert {:noreply, ^state, _} = Consumer.handle_info(:verify_flush_progress, state)
+
+      assert_receive {:"$gen_cast", {:writer_flush_deferred, "move-in-shape"}}
+    end
+
+    test "challenge is left unanswered when the consumer is not deferring", ctx do
+      state = %{Consumer.State.new(ctx.stack_id, "steady-shape") | buffering?: false}
+
+      assert {:noreply, ^state, _} = Consumer.handle_info(:verify_flush_progress, state)
+
+      refute_receive {:"$gen_cast", _}, 100
+    end
+  end
+
   describe "set_gc_heap_threshold helpers" do
     # with_stack_id_from_test (line 87) already starts ProcessRegistry + StackConfig
     # for ctx.stack_id — no heavier setup is needed for these pure-config tests.
diff --git a/packages/sync-service/test/support/transaction_consumer.ex b/packages/sync-service/test/support/transaction_consumer.ex
index 811f49312a..6a435a4d0f 100644
--- a/packages/sync-service/test/support/transaction_consumer.ex
+++ b/packages/sync-service/test/support/transaction_consumer.ex
@@ -61,6 +61,14 @@ defmodule Support.TransactionConsumer do
     GenServer.cast(pid, {:stop, reason})
   end
 
+  @doc """
+  Make the consumer die with `reason` without running its `terminate/2` callback,
+  like a crash that aborts terminate would.
+  """
+  def crash(pid, reason) do
+    GenServer.cast(pid, {:crash, reason})
+  end
+
   def init(opts) do
     Process.flag(:trap_exit, true)
     {:ok, stack_id} = Keyword.fetch(opts, :stack_id)
@@ -106,6 +114,14 @@ defmodule Support.TransactionConsumer do
     {:stop, reason, state}
   end
 
+  def handle_cast({:crash, reason}, state) do
+    # Stop trapping exits and take an exit signal from a linked process: the
+    # process dies with `reason` and terminate/2 never runs.
+    Process.flag(:trap_exit, false)
+    spawn_link(fn -> exit(reason) end)
+    {:noreply, state}
+  end
+
   # we no longer monitor consumer processes in the ShapeLogCollector
   # so consumers must de-register themselves
   def terminate(reason, %{stack_id: stack_id, shape_handle: shape_handle} = state) do
@@ -113,5 +129,12 @@ defmodule Support.TransactionConsumer do
     Electric.Replication.ShapeLogCollector.remove_shape(stack_id, shape_handle)
   end
 
+  # Forward stall challenges from the ShapeLogCollector so tests can observe
+  # them; answering (or not) is up to the test.
+  def handle_info(:verify_flush_progress, state) do
+    send(state.parent, {:flush_progress_challenged, self()})
+    {:noreply, state}
+  end
+
   def handle_info(_msg, state), do: {:noreply, state}
 end