Drop redundant materializer replay-skip guard

The materializer had a `new_changes` clause that skipped ranges already applied
(<= applied_offset), guarding against the source consumer re-notifying it with
transactions the persistent slot replays after a restart. That dedup now happens
at the source: the consumer skips already-applied transactions (at/below its
restored latest_offset) before `apply_event`, so it never re-notifies the
materializer for them — the guard is unreachable.

Remove it and rely on the single source-level dedup. Confirmed by the
optimized_refetch restart repro (which previously triggered the materializer's
"Key already exists" crash) staying green across repeated runs, plus the
materializer/consumer/subquery suites and the other restore tests.

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

Author: robacourt

packages/sync-service/lib/electric/shapes/consumer/materializer.ex

@@ -493,20 +493,6 @@ defmodule Electric.Shapes.Consumer.Materializer do
     {:reply, :ok, state}
   end
 
-  def handle_call(
-        {:new_changes, {_range_start, range_end}, _xid, _commit?},
-        _from,
-        %{applied_offset: applied_offset} = state
-      )
-      when is_log_offset_lte(range_end, applied_offset) do
-    # This range has already been applied — either during the startup history
-    # replay (`read_history_up_to_subscribed`) or a previous `new_changes` call.
-    # This happens on restart when the persistent replication slot re-delivers
-    # already-persisted transactions. Re-applying them would raise
-    # "Key already exists" in `apply_changes/2`, so skip the range entirely.
-    {:reply, :ok, state}
-  end
-
   def handle_call({:new_changes, {range_start, range_end}, xid, commit?}, _from, state) do
     stack_storage = Storage.for_stack(state.stack_id)
     storage = Storage.for_shape(state.shape_handle, stack_storage)

packages/sync-service/test/integration/oracle_restore_test.exs

@@ -138,10 +138,10 @@ defmodule Electric.Integration.OracleRestoreTest do
   test "optimized subquery shapes stay consistent when the slot replays a backlog after restart",
        ctx do
     # Two `optimized: true` subquery shapes over the same `projects` source. After
-    # the restart the persistent slot re-delivers batch_1's already-applied changes
-    # to the subquery materializer; the materializer must ignore ranges it already
-    # applied during its startup history replay so the shapes stay consistent and
-    # the polling client is not sent a 409 must-refetch.
+    # the restart the persistent slot replays batch_1's already-applied
+    # transactions; the source consumer skips those (at/below its restored
+    # `latest_offset`) before re-notifying the subquery materializer, so the shapes
+    # stay consistent and the polling client is not sent a 409 must-refetch.
     shapes = [
       %{
         name: "issues_of_active_projects",