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

.changeset/flush-tracker-writer-monitors.md

@@ -0,0 +1,6 @@
+---
+"@core/sync-service": patch
+"@core/electric-telemetry": patch
+---
+
+Stop a dead or wedged shape consumer from pinning the replication slot's `confirmed_flush_lsn` forever. The flush tracker advances the stack-wide flush boundary on the minimum incomplete entry, but entry cleanup was purely event-driven: a consumer that died without running its cleanup (e.g. a storage failure crashing it mid-terminate) or that stayed alive without flushing left its entry incomplete indefinitely, and on a subsequently quiet table nothing ever removed it — so the WAL ack froze and Postgres retained WAL without bound. The collector now monitors the writer process behind every incomplete flush entry: a crashed writer unpins the entry immediately and schedules the shape's removal, while a periodic stall check invalidates shapes whose entries make no flush progress past a grace period (configurable via `ELECTRIC_FLUSH_STALL_GRACE_PERIOD`, default 1 minute).

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

@@ -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 an incomplete 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}