From 47a6dd2486b24b7a4280e887528336717baa39fb Mon Sep 17 00:00:00 2001 From: Valter Balegas Date: Tue, 14 Jul 2026 19:00:31 +0100 Subject: [PATCH] test(sync-service): reproduce FlushTracker stall when a tracked consumer dies without cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reproduces the long-standing "system stops flushing the WAL" stall: a consumer that dies without running terminate (untrappable :kill) after its shape is tracked in the FlushTracker leaves a permanently incomplete entry. Publish-time DOWN detection only fires for events routed to that same shape, and the ShapeCleaner path requires terminate to run — so if the shape's table goes quiet, the global flush boundary is pinned forever and confirmed_flush_lsn never advances for the whole stack, causing unbounded WAL growth on Postgres. The test proves the boundary stays pinned while every live consumer is fully flushed, and that explicitly removing the dead shape from the ShapeLogCollector immediately unblocks it. Co-Authored-By: Claude Fable 5 --- .../replication/shape_log_collector_test.exs | 132 ++++++++++++++++++ 1 file changed, 132 insertions(+) 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..959c86aa5d 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,138 @@ defmodule Electric.Replication.ShapeLogCollectorTest do end end + # Reproduction of the "system stops flushing the WAL" stall: + # + # A consumer that dies WITHOUT running its terminate callback (e.g. an + # untrappable `:kill` exit) after its shape has been tracked in the + # FlushTracker leaves a permanently incomplete entry. The only two cleanup + # paths — DOWN detection during `ConsumerRegistry.publish/2` for that same + # shape, and `ShapeCleaner`-driven `ShapeLogCollector.remove_shape/2` from the + # consumer's `terminate/2` — both require either new traffic routed to the + # dead shape or a terminate that never ran. If the shape's table stays quiet, + # the FlushTracker's global minimum is pinned forever and + # `confirmed_flush_lsn` never advances for the whole stack, even though every + # live consumer is fully flushed => unbounded WAL growth on Postgres. + describe "FlushTracker stall when tracked consumer dies without cleanup on a quiet shape" 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} + ) + + %{consumer_alive: consumer_alive, consumer_doomed: consumer_doomed} + end + + test "flush boundary is pinned forever while the dead shape receives no traffic", ctx do + register_as_replication_client(ctx.stack_id) + + # Txn 1 (lsn 42) touches both tables: both consumers process it and both + # shapes are tracked in the FlushTracker at commit. + lsn1 = Lsn.from_integer(42) + + txn1 = + complete_txn_fragment(100, lsn1, [ + %Changes.NewRecord{ + relation: {"public", "test_table"}, + record: %{"id" => "1"}, + log_offset: LogOffset.new(lsn1, 0) + }, + %Changes.NewRecord{ + relation: {"public", "other_table"}, + record: %{"id" => "1"}, + log_offset: LogOffset.new(lsn1, 2) + } + ]) + + assert :ok = ShapeLogCollector.handle_event(txn1, ctx.stack_id) + assert_receive {Support.TransactionConsumer, {:alive, _}, [_]} + assert_receive {Support.TransactionConsumer, {:doomed, _}, [_]} + + # The doomed consumer dies before flushing, without terminate running + # (`:kill` is untrappable), so it neither deregisters from the SLC nor + # goes through ShapeCleaner. Its FlushTracker entry stays incomplete. + kill_consumer(ctx.consumer_doomed, :kill) + + # The alive consumer flushes txn 1 completely. The flush boundary can + # only advance up to just before the dead shape's tracked position. + ShapeLogCollector.notify_flushed(ctx.stack_id, "shape-alive", LogOffset.new(lsn1, 2)) + assert_receive {:flush_boundary_updated, 40} + + # Txn 2 (lsn 50) touches only test_table, so it is never routed to the + # dead consumer and publish-time DOWN detection never triggers. + lsn2 = Lsn.from_integer(50) + + txn2 = + complete_txn_fragment(101, lsn2, [ + %Changes.NewRecord{ + relation: {"public", "test_table"}, + record: %{"id" => "2"}, + log_offset: LogOffset.new(lsn2, 0) + } + ]) + + assert :ok = ShapeLogCollector.handle_event(txn2, ctx.stack_id) + assert_receive {Support.TransactionConsumer, {:alive, _}, [_]} + refute_receive {Support.TransactionConsumer, {:doomed, _}, _} + + # Even with every live consumer fully flushed to lsn 50, the flush + # boundary does not move: the dead shape's entry pins the global minimum. + # This is the WAL-growth stall as seen from Postgres. + ShapeLogCollector.notify_flushed(ctx.stack_id, "shape-alive", LogOffset.new(lsn2, 0)) + refute_receive {:flush_boundary_updated, _} + + # Root cause proof: explicitly removing the dead shape from the SLC (the + # step that out-of-band death skips) immediately unblocks the boundary. + ShapeLogCollector.remove_shape(ctx.stack_id, "shape-doomed") + assert_receive {:flush_boundary_updated, 50} + end + end + defp kill_consumer(pid, reason) do ref = Process.monitor(pid) Process.exit(pid, reason)