Skip already-processed transactions on restart (#4668)

## Summary

On restart, the persistent replication slot can replay transactions the
consumer has already applied and persisted. The consumer deduplicated
these on its multi-fragment path, but the complete-transaction **fast
paths short-circuited before reaching that check** — 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 with `Key ... already exists`.

This is **bug 2** from the restart-aware oracle property-test triage
(see #4648 / `packages/sync-service/bugs.md`).

## Problem

After a `StackSupervisor` restart, the source consumer reconnects to the
persistent slot and replays from its confirmed LSN, which can lag the
shape's on-disk offset. Complete single-fragment transactions (the
common case — one statement per txn) took the fast path and were
re-processed:

- **Normal shapes:** duplicate / orphan operations appended to the shape
log.
- **Subquery shapes:** the re-notified duplicate is re-applied by the
dependency materializer, which is a stateful reducer and crashes on the
duplicate insert (`"Key ... already exists"`). That crash then cascades
into tearing down and *removing* the healthy dependent shapes and
returning `409 must-refetch` to clients.

## Solution

Consolidate the offset dedup into a **single `handle_txn_fragment/2`
clause** that skips any fragment already at or below `latest_offset`
(routing to the existing `skip_txn_fragment`), placed before the
complete-transaction fast paths. The dedup now lives in one place and
covers single- and multi-fragment paths uniformly, replacing the
separate `fragment_already_processed?/2` check. Normal operation is
untouched: new transactions have `offset > latest_offset`, so the guard
is false and they fall through to the existing clauses.

This relies on a transaction's fragments being entirely at-or-below or
entirely above `latest_offset` (which only advances at commit
boundaries) — the same assumption the previous per-fragment check
already made.

## Test Plan

- [x] `test/electric/shapes/consumer_test.exs` — two regression tests
(single-fragment and multi-fragment) replay an already-applied
transaction straight to the consumer, bypassing the log collector's own
dedup (as happens on restart with a fresh collector), and assert it is
skipped: no `append_to_log!`, no `:new_changes` notification, log
unchanged. Fail without the fix.
- [x] `test/electric/shapes/` + `test/electric/replication/` — 937
tests, 100 doctests, 6 properties, 0 failures.

---
Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: robacourt

.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]