Skip already-processed transactions on restart

.changeset/consumer-skip-replayed-txns.md

@@ -0,0 +1,5 @@
+---
+'@core/sync-service': patch
+---
+
+Fix shapes processing duplicate operations after a server restart.

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

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]