FlushTracker has no active mechanism to detect shapes that stop reporting flush progress

[alco]

Summary

The FlushTracker in ShapeLogCollector is a passive data structure that relies entirely on external notifications to advance last_global_flushed_offset. When a shape enters tracking via handle_txn_fragment, the tracker expects one of two things to eventually happen:

1. handle_flush_notification — the consumer reports it flushed the offset
2. handle_shape_removed — the shape is cleaned up

If neither happens, that shape's entry becomes the minimum in min_incomplete_flush_tree and permanently blocks WAL flush advancement for the entire stack.

ShapeLogCollector does not monitor consumer processes. There is no timeout, no periodic sweep, and no liveness check for shapes tracked by FlushTracker. The system is entirely dependent on every consumer eventually calling notify_flushed or being properly removed — a property that is not guaranteed across all failure modes.

Impact

When last_global_flushed_offset stops advancing, the replication client never receives {:flush_boundary_updated, lsn}, Postgres never gets WAL flush acknowledgement past that point, and WAL accumulates indefinitely until disk is exhausted.

What's already addressed

• fix(sync-service): recover from dead consumers blocking WAL flush advancement #3975 fixes the case where a consumer dies during ConsumerRegistry.broadcast — the dead consumer is detected via :DOWN, cleaned from ETS, and a replacement is started. Events for shapes that were removed between routing and delivery are excluded from FlushTracker entirely.

What remains unaddressed

The child issues below describe specific failure modes that can leave shapes permanently tracked in FlushTracker with no path to resolution.

Suggested fix

Add a defense-in-depth mechanism in ShapeLogCollector to actively detect when FlushTracker-tracked shapes have stopped progressing. Two approaches:

Option A: Monitor consumer PIDs. When ConsumerRegistry.publish delivers events, capture the consumer PIDs. ShapeLogCollector monitors them. On :DOWN, call FlushTracker.handle_shape_removed. This is fast (immediate detection) but requires ShapeLogCollector to handle :DOWN messages and maintain a PID→shape mapping.

Option B: Periodic liveness sweep. On each incoming transaction (or on a timer), ShapeLogCollector iterates FlushTracker.last_flushed and checks whether each shape still has a live consumer via ConsumerRegistry.whereis + Process.alive?. Dead shapes get handle_shape_removed. This is simpler but detection is delayed until the next sweep.

Option A catches all crash/kill scenarios immediately. Option B also catches the "alive but stuck" scenario (child issue #3) if extended with a staleness timeout.

[msfstef]

On Option A - we had moved to the request batcher to avoid flooding the shape log collector with shape registration/lifecycle events since its main job is to handle the transactions and push to the right consumers. I'd be wary about reintroducing that. Maybe the flush tracker should not live on the shape log collector, and should be its own process?

[alco]

@msfstef The issue description is not telling the full story. It focuses on ShapeLogCollector alone, but it calls ConsumerRegistry.publish(), which in turn calls broadcast() and that function monitors every consumer it sends the txn fragment to, as is the standard GenServer.call() protocol.

In any case, I think the actual approach for resolving the problem is somewhere between #3975 and #3979

[alco]

Proposed approach: Periodic FlushTracker liveness sweep (Option B)

Rather than monitoring consumer PIDs in ShapeLogCollector (Option A), add a periodic sweep that detects when last_global_flushed_offset has been stuck for too long and checks the liveness of the consumers responsible for blocking it.

Design

FlushTracker gains a single query function blocking_shape_ids/1 that returns the shape IDs at the smallest position in min_incomplete_flush_tree — these are the shapes currently preventing the global flush offset from advancing.

ShapeLogCollector adds a recurring timer (default every 30s) with two-phase staleness detection:

1. On each tick, compare last_global_flushed_offset with the previous sweep's cached value
2. If the offset hasn't advanced and FlushTracker is non-empty, record the timestamp of first detection
3. Only act when the offset has been stuck for longer than a configurable threshold (default 60s)

When the threshold is exceeded, the sweep checks each blocking shape:

• Orphaned entry (shape no longer in EventRouter but still in FlushTracker — partial cleanup) → call handle_shape_removed directly
• Dead consumer (shape in EventRouter but ConsumerRegistry.whereis returns nil or dead PID) → trigger ShapeCleaner.remove_shape_async, which invalidates the shape first (clients get 409), then cleans up FlushTracker through the standard remove_subscription path. This guarantees no data loss — the FlushTracker entry is only removed after the shape is fully invalidated.
• Alive but stuck (consumer exists and is alive but not making progress) → log a warning. This helps surface non-progressing consumers as described in FlushTracker stall: consumer alive but permanently stuck (no progress detection) #3983, even though the sweep doesn't take direct action on them — the warning provides observability into the issue while a separate mechanism can be designed to handle that case.

A shapes_pending_sweep_removal set prevents duplicate remove_shape_async calls across consecutive sweep ticks while waiting for the async cleanup pipeline to complete. Telemetry on [:electric, :flush_tracker, :sweep] reports counts of orphaned/dead/alive-stuck detections.

Safety

• No data loss: We never remove a FlushTracker entry directly for a shape that still exists. The cleanup goes through ShapeCleaner.remove_shape_async → ShapeStatus.remove_shape → Consumer.stop → Storage.cleanup! → ShapeLogCollector.remove_shape → FlushTracker.handle_shape_removed.
• Double-cleanup is safe: ShapeStatus.remove_shape is idempotent, EventRouter.has_shape? guards remove_subscription, and FlushTracker.handle_shape_removed on an unknown shape is a no-op.
• Suspended consumers: A suspended consumer has no PID in ConsumerRegistry (clause 2 of handle_writer_termination removes it). If the shape blocks FlushTracker for >60s with no new transactions arriving, the sweep will invalidate it. This is a deliberate trade-off — the cost of a client refetch is low compared to unbounded WAL accumulation.

Primary catch scenarios

This sweep is defense-in-depth on top of the primary fix in #4011. Its two unique catches are:

Scenario	Why primary fixes don't help
Consumer killed with kill -9	terminate/2 doesn't run, so no cleanup fires
Consumer alive but stuck	No existing mechanism detects non-progressing consumers

Scope

~3 files modified: flush_tracker.ex (add blocking_shape_ids/1), shape_log_collector.ex (sweep timer + handler), stack_config.ex (default config values). No changes to FlushTracker's data model — it remains a pure data structure.