fix(sync-service): prevent permanent flush-boundary stalls from unexpected consumer lifecycle transitions (#4730)

## Problem

The FlushTracker pins `confirmed_flush_lsn` for the whole stack on the
minimum incomplete flush entry, and cleanup of entries is purely
event-driven: either new traffic routed to the same shape (publish-time
DOWN detection) or an explicit `ShapeLogCollector.remove_shape/2`
arriving via the ShapeCleaner chain. Several real-world paths break both
— a storage error that crashes the consumer *and* aborts its
`terminate/2` before `handle_writer_termination` runs, a removal chain
that dies between the `ShapeStatus` removal and the SLC removal
(unretryable: the retry short-circuits at `{:error, :data_removed}`),
untagged `{:shutdown, x}` exit reasons classified as benign, and a
consumer that stays alive but stops flushing. On a subsequently-quiet
table any of these pins the WAL ack forever, causing unbounded WAL
growth on Postgres. PR #4713 has a minimal reproduction of the stall.

## Solution

The SLC becomes the authority on consumer-death handling for flush
accounting. Invariant: an incomplete FlushTracker entry always has a
live monitor watching the pid responsible for completing it, and a
timestamp bounding how long it may sit without flush progress.

- **Writer monitors**: the SLC monitors every consumer with an
incomplete FlushTracker entry (created when a commit-fragment publish
tracks the shape, released when the entry completes or the shape is
removed). DOWNs are classified by exit reason: crashes unpin the flush
boundary immediately and trigger shape invalidation; deliberate-cleanup
DOWNs (`{:shutdown, :cleanup}`) unpin only; bare `:shutdown`/`:killed`
are assumed to be supervisor teardown and left alone.
- **Stall detection**: incomplete entries carry a `last_progress_at`
timestamp refreshed only by flush notifications. A periodic tick finds
shapes whose entries make no flush progress for
`ELECTRIC_FLUSH_STALL_GRACE_PERIOD` (default 60s) and invalidates them
via the normal ShapeCleaner path, re-arming on each fire so a lost
cleanup chain retries instead of pinning forever. Misclassified DOWNs
self-heal through the same mechanism.
- **Stall challenge-response**: a stalled entry whose writer is still
alive is challenged (`:verify_flush_progress`) before being acted on. A
consumer answers only while legitimately deferring flush notifications
(buffering ahead of PG snapshot info, subquery move-in awaiting splice);
the answer re-arms the entry's grace period so healthy deferral phases
are not misread as stalls. Only a suspect still stalled and silent at
the next check — or a shape whose writer is already dead and
demonitored, which skips the challenge — is invalidated. Consumers stay
silent in steady state: no periodic messaging, and a wedged consumer
can't answer, so the backstop stays armed.
- **Reason normalization**: the benign clause of
`handle_writer_termination` shrinks to exactly
`:normal`/`:killed`/`:shutdown`, so untagged `{:shutdown, x}` reasons
now trigger removal; tagged dependency-materializer shutdowns map to
`stop_and_clean`; suspend additionally requires all txn state flushed
and notified (guaranteeing a suspended consumer never leaves an
incomplete entry behind); `Consumer.stop` escalates a wedged consumer to
a kill after the stop call times out.
- ShapeCleaner re-issues the idempotent SLC removal even when the shape
is already gone from `ShapeStatus`, healing pins left by removal chains
that died partway through.

Telemetry: new `[:electric, :flush_tracker, :writer_down]` and
`[:electric, :flush_tracker, :stall_detected]` events.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Garry Hill <garry@electric-sql.com>

Author: 3 people

.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.
+

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)

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:

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),

packages/sync-service/lib/electric/config.ex

@@ -104,6 +104,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}