diff --git a/.changeset/consumer-skip-replayed-txns.md b/.changeset/consumer-skip-replayed-txns.md new file mode 100644 index 0000000000..1328512ab4 --- /dev/null +++ b/.changeset/consumer-skip-replayed-txns.md @@ -0,0 +1,5 @@ +--- +'@core/sync-service': patch +--- + +Fix shapes processing duplicate operations after a server restart. diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex index f60a57dcdd..1c9d36cb25 100644 --- a/packages/sync-service/lib/electric/shapes/consumer.ex +++ b/packages/sync-service/lib/electric/shapes/consumer.ex @@ -587,6 +587,22 @@ defmodule Electric.Shapes.Consumer do State.add_to_buffer(state, txn_fragment) end + # Skip transactions already applied and persisted (e.g. replayed from the persistent + # replication slot on restart) - ones at or below `latest_offset`. + # + # This skips whole transactions, relying on a replayed transaction being entirely + # at-or-below or entirely above `latest_offset`, never straddling it. In-memory + # `latest_offset` advances per written fragment (see `write_txn_fragment_to_storage/2`), + # not only at commit — but on the replay path the guarantee holds: after a restart + # `latest_offset` is restored from storage at a commit boundary + # (`Storage.fetch_latest_offset/1`, from `last_{seen,persisted}_txn_offset`), and the + # replication slot replays whole transactions from a commit boundary + # (`confirmed_flush_lsn`). + defp handle_txn_fragment(%TransactionFragment{last_log_offset: offset} = txn_fragment, state) + when LogOffset.is_log_offset_lte(offset, state.latest_offset) do + skip_txn_fragment(state, txn_fragment) + end + # Short-circuit clauses for the most common case of a single-fragment transaction defp handle_txn_fragment(%TransactionFragment{} = txn_fragment, state) when TransactionFragment.complete_transaction?(txn_fragment) and @@ -652,13 +668,30 @@ defmodule Electric.Shapes.Consumer do defp handle_txn_fragment(txn_fragment, state), do: process_txn_fragment(txn_fragment, state) + # Defensive: this is only ever reached with a `pending_txn` set — a transaction's + # BEGIN fragment creates it (see the `has_begin?` clauses above) before any fragment + # gets here. The only way to arrive with `pending_txn: nil` is a transaction that + # straddled `latest_offset`: its BEGIN fragment skipped by the already-applied clause + # while a later fragment was not. The commit-aligned restore + whole-transaction replay + # invariant (documented on that clause) rules this out; fail loudly with a clear message + # rather than crash on `nil.consider_flushed?` if it is ever violated. + defp process_txn_fragment(%TransactionFragment{last_log_offset: offset}, %State{ + pending_txn: nil + }) do + raise "consumer received a transaction fragment at #{inspect(offset)} with no pending " <> + "transaction — a transaction straddling latest_offset was partially skipped, " <> + "which should be impossible" + end + defp process_txn_fragment( %TransactionFragment{} = txn_fragment, %State{pending_txn: txn} = state ) do cond do - # Fragments belonging to the same transaction can all be skipped either via xid-filtering or log offset filtering. - txn.consider_flushed? or fragment_already_processed?(txn_fragment, state) -> + # Fragments of a transaction whose xid is already in the initial snapshot are + # skipped here. (Offset-based dedup of replayed transactions is handled earlier, + # in `handle_txn_fragment/2`.) + txn.consider_flushed? -> skip_txn_fragment(state, txn_fragment) # With write_unit=txn all fragments are buffered until the Commit change is seen. At that @@ -1097,10 +1130,6 @@ defmodule Electric.Shapes.Consumer do Kernel.max(0, DateTime.diff(now, commit_timestamp, :millisecond)) end - defp fragment_already_processed?(%TransactionFragment{last_log_offset: offset}, state) do - LogOffset.is_log_offset_lte(offset, state.latest_offset) - end - defp consider_flushed(%State{} = state, log_offset) do if state.txn_offset_mapping == [] do # No relevant txns have been observed and unflushed, we can notify immediately diff --git a/packages/sync-service/test/electric/shapes/consumer_test.exs b/packages/sync-service/test/electric/shapes/consumer_test.exs index a2554f54d9..4fe0bb44d9 100644 --- a/packages/sync-service/test/electric/shapes/consumer_test.exs +++ b/packages/sync-service/test/electric/shapes/consumer_test.exs @@ -725,6 +725,62 @@ defmodule Electric.Shapes.ConsumerTest do get_log_items_from_storage(LogOffset.last_before_real_offsets(), shape_storage) end + test "skips an already-applied transaction replayed past a fresh log collector", ctx do + # Simulates a restart: the persistent replication slot replays a transaction + # the consumer has already applied and persisted. A freshly-started + # ShapeLogCollector hasn't seen the offset, so (unlike the test above) it + # won't drop it — the consumer itself must skip it, because its restored + # `latest_offset` is already at/past the transaction. Otherwise the fragment + # is re-written to the log (duplicating ops) and re-notified to dependent + # materializers, which re-apply it and crash. + {shape_handle, _} = ShapeCache.get_or_create_shape_handle(@shape1, ctx.stack_id) + :started = ShapeCache.await_snapshot_start(shape_handle, ctx.stack_id) + + ref = Shapes.Consumer.register_for_changes(ctx.stack_id, shape_handle) + + xid = 11 + lsn = Lsn.from_integer(10) + + txn = + complete_txn_fragment(xid, lsn, [ + %Changes.NewRecord{ + relation: {"public", "test_table"}, + record: %{"id" => "1"}, + log_offset: LogOffset.new(lsn, 0) + } + ]) + + consumer_pid = Shapes.Consumer.whereis(ctx.stack_id, shape_handle) + shape_storage = Storage.for_shape(shape_handle, ctx.storage) + + # First delivery: applied normally, advancing the consumer's latest_offset. + assert :ok = ShapeLogCollector.handle_event(txn, ctx.stack_id) + last_log_offset = LogOffset.new(lsn, 0) + assert_receive {^ref, :new_changes, ^last_log_offset} + + assert [op1] = + get_log_items_from_storage(LogOffset.last_before_real_offsets(), shape_storage) + + # Replay the same, already-applied transaction straight to the consumer, + # bypassing the collector's own offset de-dup (as happens on restart with a + # fresh collector). The consumer must skip it: no storage write, no + # notification, log unchanged. + enable_storage_tracer_for(consumer_pid) + + assert :ok = + GenServer.call( + Shapes.Consumer.name(ctx.stack_id, shape_handle), + {:handle_event, txn, Electric.Telemetry.OpenTelemetry.get_current_context()}, + :infinity + ) + + assert [] == Support.Trace.collect_traced_calls() + refute_receive {^ref, :new_changes, _} + + assert [op1] == + get_log_items_from_storage(LogOffset.last_before_real_offsets(), shape_storage) + end + @tag allow_subqueries: false test "duplicate txn fragment handling is idempotent", ctx do {shape_handle, _} = ShapeCache.get_or_create_shape_handle(@shape1, ctx.stack_id) @@ -834,6 +890,93 @@ defmodule Electric.Shapes.ConsumerTest do refute_receive {^ref, :new_changes, _} end + @tag allow_subqueries: false + test "skips an already-applied multi-fragment transaction replayed past a fresh log collector", + ctx do + # Multi-fragment variant of "skips an already-applied transaction replayed + # past a fresh log collector". On restart the persistent slot can replay a + # multi-statement transaction the consumer has already applied. This drives + # the offset-dedup on the multi-fragment path (BEGIN / middle / COMMIT + # fragments), not the single-fragment fast path — the consumer must skip + # every fragment without re-writing or re-notifying. + {shape_handle, _} = ShapeCache.get_or_create_shape_handle(@shape1, ctx.stack_id) + :started = ShapeCache.await_snapshot_start(shape_handle, ctx.stack_id) + + ref = Shapes.Consumer.register_for_changes(ctx.stack_id, shape_handle) + + xid = 11 + lsn = Lsn.from_integer(10) + + [f1, f2, f3] = + txn_fragments(xid, lsn, [ + %{ + has_begin?: true, + changes: [ + %Changes.NewRecord{ + relation: {"public", "test_table"}, + record: %{"id" => "1"}, + log_offset: LogOffset.new(lsn, 0) + } + ] + }, + %{ + changes: [ + %Changes.NewRecord{ + relation: {"public", "test_table"}, + record: %{"id" => "2"}, + log_offset: LogOffset.new(lsn, 2) + } + ] + }, + %{ + has_commit?: true, + changes: [ + %Changes.NewRecord{ + relation: {"public", "test_table"}, + record: %{"id" => "3"}, + log_offset: LogOffset.new(lsn, 4) + } + ] + } + ]) + + consumer_pid = Shapes.Consumer.whereis(ctx.stack_id, shape_handle) + shape_storage = Storage.for_shape(shape_handle, ctx.storage) + + # First delivery via the collector: applied normally, advancing latest_offset + # to the commit offset. + Enum.each([f1, f2, f3], &assert(:ok = ShapeLogCollector.handle_event(&1, ctx.stack_id))) + + commit_offset = LogOffset.new(lsn, 4) + assert_receive {^ref, :new_changes, ^commit_offset} + + assert [_, _, _] = + ops = + get_log_items_from_storage(LogOffset.last_before_real_offsets(), shape_storage) + + # Replay every fragment straight to the consumer, bypassing the collector's + # offset de-dup (as on restart with a fresh collector). The consumer's restored + # latest_offset is already at the commit, so all fragments — including the + # BEGIN fragment, which now skips without ever setting up `pending_txn` — must + # be skipped: no storage writes, no notification, log unchanged. + enable_storage_tracer_for(consumer_pid) + + Enum.each([f1, f2, f3], fn f -> + assert :ok = + GenServer.call( + Shapes.Consumer.name(ctx.stack_id, shape_handle), + {:handle_event, f, Electric.Telemetry.OpenTelemetry.get_current_context()}, + :infinity + ) + end) + + assert [] == Support.Trace.collect_traced_calls() + refute_receive {^ref, :new_changes, _} + + assert ops == + get_log_items_from_storage(LogOffset.last_before_real_offsets(), shape_storage) + end + @tag pg_snapshot: {10, 13, [10, 12]}, delay_snapshot_creation?: true, with_pure_file_storage_opts: [flush_period: 1]