From 772baff2de94f09f25877372f4b7e9fa6cffa537 Mon Sep 17 00:00:00 2001 From: rob Date: Wed, 1 Jul 2026 11:07:56 +0100 Subject: [PATCH 1/8] Skip already processed changes --- .changeset/consumer-skip-replayed-txns.md | 12 ++++ .../lib/electric/shapes/consumer.ex | 13 +++++ .../test/electric/shapes/consumer_test.exs | 56 +++++++++++++++++++ 3 files changed, 81 insertions(+) create mode 100644 .changeset/consumer-skip-replayed-txns.md diff --git a/.changeset/consumer-skip-replayed-txns.md b/.changeset/consumer-skip-replayed-txns.md new file mode 100644 index 0000000000..ae4ea9ad33 --- /dev/null +++ b/.changeset/consumer-skip-replayed-txns.md @@ -0,0 +1,12 @@ +--- +'@core/sync-service': patch +--- + +Fix shapes processing duplicate operations after a server restart. On restart the +persistent replication slot can replay transactions the consumer has already +applied and persisted. The multi-fragment path skipped these via +`fragment_already_processed?/2`, but the single-fragment complete-transaction fast +path did not — so a replayed transaction was re-written to the shape log +(duplicating ops) and re-notified to dependent subquery materializers, which +re-applied it and crashed. The consumer now skips a complete transaction already +at or below its persisted `latest_offset` on every path. diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex index f60a57dcdd..76f96c3d80 100644 --- a/packages/sync-service/lib/electric/shapes/consumer.ex +++ b/packages/sync-service/lib/electric/shapes/consumer.ex @@ -587,6 +587,19 @@ defmodule Electric.Shapes.Consumer do State.add_to_buffer(state, txn_fragment) end + # On restart the persistent replication slot can replay transactions we have + # already applied and persisted. The multi-fragment path skips those via + # `fragment_already_processed?/2` (see `process_txn_fragment/2`), but the + # complete-transaction fast paths below did not — so a replayed single-fragment + # transaction was re-written to the shape log (duplicating ops) and re-notified + # to dependent materializers, which re-apply it and crash. Skip it here, so the + # dedup is applied on every path. + defp handle_txn_fragment(%TransactionFragment{last_log_offset: offset} = txn_fragment, state) + when TransactionFragment.complete_transaction?(txn_fragment) and + LogOffset.is_log_offset_lte(offset, state.latest_offset) do + consider_flushed(state, offset) + 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 diff --git a/packages/sync-service/test/electric/shapes/consumer_test.exs b/packages/sync-service/test/electric/shapes/consumer_test.exs index a2554f54d9..025f0a705e 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) From 8b309ccea4326db0fde4e9fb6d70b9660a44dcd2 Mon Sep 17 00:00:00 2001 From: rob Date: Wed, 1 Jul 2026 12:07:50 +0100 Subject: [PATCH 2/8] Consolidate replayed-transaction dedup into a single clause Replace the complete-transaction offset check and process_txn_fragment's fragment_already_processed?/2 with one offset-only handle_txn_fragment clause placed after the buffering? clause, routing to the existing skip_txn_fragment. The dedup now lives in one place and covers single- and multi-fragment paths uniformly. Behaviour-preserving. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../lib/electric/shapes/consumer.ex | 31 +++++++++---------- 1 file changed, 15 insertions(+), 16 deletions(-) diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex index 76f96c3d80..5397a98b9d 100644 --- a/packages/sync-service/lib/electric/shapes/consumer.ex +++ b/packages/sync-service/lib/electric/shapes/consumer.ex @@ -587,17 +587,18 @@ defmodule Electric.Shapes.Consumer do State.add_to_buffer(state, txn_fragment) end - # On restart the persistent replication slot can replay transactions we have - # already applied and persisted. The multi-fragment path skips those via - # `fragment_already_processed?/2` (see `process_txn_fragment/2`), but the - # complete-transaction fast paths below did not — so a replayed single-fragment - # transaction was re-written to the shape log (duplicating ops) and re-notified - # to dependent materializers, which re-apply it and crash. Skip it here, so the - # dedup is applied on every path. + # A transaction already applied and persisted (e.g. replayed from the persistent + # replication slot on restart) sits at or below `latest_offset`. Skip it here so + # the dedup covers every path — including the complete-transaction fast paths + # below, which short-circuit before reaching `process_txn_fragment/2`. Re-applying + # such a transaction would duplicate ops in the shape log and, for dependent + # subquery materializers, re-apply and crash them. + # + # This relies on a transaction's fragments being entirely at-or-below or entirely + # above `latest_offset` (it only advances at commit boundaries). defp handle_txn_fragment(%TransactionFragment{last_log_offset: offset} = txn_fragment, state) - when TransactionFragment.complete_transaction?(txn_fragment) and - LogOffset.is_log_offset_lte(offset, state.latest_offset) do - consider_flushed(state, offset) + 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 @@ -670,8 +671,10 @@ defmodule Electric.Shapes.Consumer do %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 @@ -1110,10 +1113,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 From e245d272a20b4432deb9b70b2475de2bf73459e2 Mon Sep 17 00:00:00 2001 From: rob Date: Wed, 1 Jul 2026 12:28:55 +0100 Subject: [PATCH 3/8] Add multi-fragment replay regression test; tidy changeset wording Cover the multi-fragment replay path (BEGIN/middle/COMMIT fragments skipped at the offset-dedup clause without setting up pending_txn), which the existing single-fragment replay test doesn't exercise. Reword the changeset to drop implementation details that no longer match after the dedup consolidation. Co-Authored-By: Claude Opus 4.8 (1M context) --- .changeset/consumer-skip-replayed-txns.md | 10 +-- .../test/electric/shapes/consumer_test.exs | 87 +++++++++++++++++++ 2 files changed, 91 insertions(+), 6 deletions(-) diff --git a/.changeset/consumer-skip-replayed-txns.md b/.changeset/consumer-skip-replayed-txns.md index ae4ea9ad33..4ffc0efe43 100644 --- a/.changeset/consumer-skip-replayed-txns.md +++ b/.changeset/consumer-skip-replayed-txns.md @@ -4,9 +4,7 @@ Fix shapes processing duplicate operations after a server restart. On restart the persistent replication slot can replay transactions the consumer has already -applied and persisted. The multi-fragment path skipped these via -`fragment_already_processed?/2`, but the single-fragment complete-transaction fast -path did not — so a replayed transaction was re-written to the shape log -(duplicating ops) and re-notified to dependent subquery materializers, which -re-applied it and crashed. The consumer now skips a complete transaction already -at or below its persisted `latest_offset` on every path. +applied and persisted. The consumer now skips any transaction already at or below +its persisted offset, so a replayed transaction is no longer re-written to the +shape log (duplicating ops) or re-notified to dependent subquery materializers +(which would re-apply it and crash). diff --git a/packages/sync-service/test/electric/shapes/consumer_test.exs b/packages/sync-service/test/electric/shapes/consumer_test.exs index 025f0a705e..4fe0bb44d9 100644 --- a/packages/sync-service/test/electric/shapes/consumer_test.exs +++ b/packages/sync-service/test/electric/shapes/consumer_test.exs @@ -890,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] From 44f957321d0a9215807b7d09f3d3793d6f36b7d5 Mon Sep 17 00:00:00 2001 From: Rob A'Court Date: Wed, 1 Jul 2026 12:40:19 +0100 Subject: [PATCH 4/8] Apply suggestion from @robacourt --- .changeset/consumer-skip-replayed-txns.md | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/.changeset/consumer-skip-replayed-txns.md b/.changeset/consumer-skip-replayed-txns.md index 4ffc0efe43..1328512ab4 100644 --- a/.changeset/consumer-skip-replayed-txns.md +++ b/.changeset/consumer-skip-replayed-txns.md @@ -2,9 +2,4 @@ '@core/sync-service': patch --- -Fix shapes processing duplicate operations after a server restart. On restart the -persistent replication slot can replay transactions the consumer has already -applied and persisted. The consumer now skips any transaction already at or below -its persisted offset, so a replayed transaction is no longer re-written to the -shape log (duplicating ops) or re-notified to dependent subquery materializers -(which would re-apply it and crash). +Fix shapes processing duplicate operations after a server restart. From 8cf97106803dd93572c406d47fbafd319ed21e87 Mon Sep 17 00:00:00 2001 From: Rob A'Court Date: Wed, 1 Jul 2026 12:40:37 +0100 Subject: [PATCH 5/8] Update packages/sync-service/lib/electric/shapes/consumer.ex --- packages/sync-service/lib/electric/shapes/consumer.ex | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex index 5397a98b9d..07c3730a7f 100644 --- a/packages/sync-service/lib/electric/shapes/consumer.ex +++ b/packages/sync-service/lib/electric/shapes/consumer.ex @@ -587,12 +587,8 @@ defmodule Electric.Shapes.Consumer do State.add_to_buffer(state, txn_fragment) end - # A transaction already applied and persisted (e.g. replayed from the persistent - # replication slot on restart) sits at or below `latest_offset`. Skip it here so - # the dedup covers every path — including the complete-transaction fast paths - # below, which short-circuit before reaching `process_txn_fragment/2`. Re-applying - # such a transaction would duplicate ops in the shape log and, for dependent - # subquery materializers, re-apply and crash them. + # Skip transactions already applied and persisted (e.g. replayed from the persistent + # replication slot on restart) - ones at or below `latest_offset`. # # This relies on a transaction's fragments being entirely at-or-below or entirely # above `latest_offset` (it only advances at commit boundaries). From 2ac92d9da12f1a827b045d02e620148b9e25d714 Mon Sep 17 00:00:00 2001 From: rob Date: Wed, 1 Jul 2026 13:51:01 +0100 Subject: [PATCH 6/8] Correct replay-skip invariant comment latest_offset advances per written fragment, not only at commit, so reword the justification: the whole-transaction skip is safe on the replay path because the restored latest_offset is commit-aligned (Storage.fetch_latest_offset/1) and the slot replays whole transactions from a commit boundary. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/sync-service/lib/electric/shapes/consumer.ex | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex index 07c3730a7f..ee712d028d 100644 --- a/packages/sync-service/lib/electric/shapes/consumer.ex +++ b/packages/sync-service/lib/electric/shapes/consumer.ex @@ -590,8 +590,14 @@ defmodule Electric.Shapes.Consumer do # Skip transactions already applied and persisted (e.g. replayed from the persistent # replication slot on restart) - ones at or below `latest_offset`. # - # This relies on a transaction's fragments being entirely at-or-below or entirely - # above `latest_offset` (it only advances at commit boundaries). + # 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) From e74a616bf09e1204d59a58b09afa085ba3a44675 Mon Sep 17 00:00:00 2001 From: rob Date: Wed, 1 Jul 2026 13:54:36 +0100 Subject: [PATCH 7/8] Remove trailing whitespace in comment Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/sync-service/lib/electric/shapes/consumer.ex | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex index ee712d028d..3ffb2d9205 100644 --- a/packages/sync-service/lib/electric/shapes/consumer.ex +++ b/packages/sync-service/lib/electric/shapes/consumer.ex @@ -588,7 +588,7 @@ defmodule Electric.Shapes.Consumer do end # Skip transactions already applied and persisted (e.g. replayed from the persistent - # replication slot on restart) - ones at or below `latest_offset`. + # 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 From fcb5ffed7bcbc0de914a0734fd7232af18f428f4 Mon Sep 17 00:00:00 2001 From: rob Date: Wed, 1 Jul 2026 13:59:18 +0100 Subject: [PATCH 8/8] Fail loudly if a straddling transaction reaches process_txn_fragment Skipping a BEGIN fragment via the already-applied clause now leaves pending_txn: nil. In normal operation process_txn_fragment is only reached with pending_txn set (the BEGIN fragment creates it), so this is safe. But a transaction straddling latest_offset would reach it with pending_txn: nil and crash on `nil.consider_flushed?`. Add an explicit pending_txn: nil clause that raises a clear, documented error instead of the implicit BadMapError. The straddle is ruled out by the commit-aligned restore + whole-transaction replay invariant. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../sync-service/lib/electric/shapes/consumer.ex | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex index 3ffb2d9205..1c9d36cb25 100644 --- a/packages/sync-service/lib/electric/shapes/consumer.ex +++ b/packages/sync-service/lib/electric/shapes/consumer.ex @@ -668,6 +668,21 @@ 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