Fix Materializer startup race condition via offset coordination (#3794)

## Summary

Fixes #3787 - addresses the snapshot/replication race condition in the
Materializer that caused "Key already exists" crashes.

### The race condition

The Materializer subscribed to the Consumer **before** reading from
storage:

```elixir
# BEFORE: In handle_continue(:start_materializer, ...)
Consumer.subscribe_materializer(stack_id, shape_handle, self())  # <- Subscribes first
{:noreply, state, {:continue, {:read_stream, shape_storage}}}    # <- Then reads ALL storage
```

During the window between subscribing and reading, any changes that
arrive via `Consumer.new_changes()` would be delivered to the
Materializer. If those changes included records that were also in the
snapshot being read, the Materializer received duplicates and crashed.

### Production example (maxwell instance, 27 Jan 2026)

```
18:10:10.437 [error] GenServer Materializer "97489818-..." terminating
** (RuntimeError) Key "public"."offers"/"d3c8d8a5-5060-4a36-a67d-240de0c95a88" already exists
```

The transaction that triggered it:
```elixir
%Electric.Replication.Changes.NewRecord{
  relation: {"public", "offers"},
  record: %{"id" => "d3c8d8a5-5060-4a36-a67d-240de0c95a88"},
  key: "\"public\".\"offers\"/\"d3c8d8a5-5060-4a36-a67d-240de0c95a88\"",
  move_tags: ["e12422d3af57a36d01a50b4645a517e4"]  # <- Move-in event
}
```

The record was already in the snapshot (matched via `is_template = true`
OR the subquery), AND was delivered via replication with `move_tags` as
a move-in event.

### The fix

Consumer now returns its current offset when a Materializer subscribes.
The Materializer reads storage **only up to that offset**, not beyond.
Changes after that offset will be delivered via `new_changes` messages,
ensuring each change is delivered exactly once.

```elixir
# AFTER: Consumer returns offset on subscription
def handle_call({:subscribe_materializer, pid}, _from, state) do
  {:reply, {:ok, state.latest_offset}, %{state | materializer_subscribed?: true}, ...}
end

# AFTER: Materializer uses offset to bound storage reads
{:ok, subscribed_offset} = Consumer.subscribe_materializer(stack_id, shape_handle, self())
{:noreply, %{state | subscribed_offset: subscribed_offset}, {:continue, {:read_stream, ...}}}

# In handle_continue({:read_stream, ...})
{:ok, offset, stream} = get_stream_up_to_offset(state.offset, state.subscribed_offset, storage)
```

## Test plan

- [x] Added test verifying offset coordination prevents duplicates
- [x] All existing Materializer tests pass (35 tests)
- [x] Full test suite passes (1334 tests)

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

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: robacourt

.changeset/fix-materializer-startup-race.md

@@ -0,0 +1,7 @@
+---
+'@core/sync-service': patch
+---
+
+Fix Materializer startup race condition that caused "Key already exists" crashes
+
+The Materializer subscribed to the Consumer before reading from storage, creating a window where the same record could be delivered twice (via storage AND via new_changes). Now the Consumer returns its current offset on subscription, and the Materializer reads storage only up to that offset.

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

@@ -54,7 +54,8 @@ defmodule Electric.Shapes.Consumer do
     |> GenServer.call(:await_snapshot_start, timeout)
   end
 
-  @spec subscribe_materializer(Electric.stack_id(), Electric.shape_handle(), pid()) :: :ok
+  @spec subscribe_materializer(Electric.stack_id(), Electric.shape_handle(), pid()) ::
+          {:ok, LogOffset.t()}
   def subscribe_materializer(stack_id, shape_handle, pid) do
     stack_id
     |> consumer_pid(shape_handle)
@@ -210,7 +211,9 @@ defmodule Electric.Shapes.Consumer do
   def handle_call({:subscribe_materializer, pid}, _from, state) do
     Logger.debug("Subscribing materializer for #{state.shape_handle}")
     Process.monitor(pid, tag: :materializer_down)
-    {:reply, :ok, %{state | materializer_subscribed?: true}, state.hibernate_after}
+
+    {:reply, {:ok, state.latest_offset}, %{state | materializer_subscribed?: true},
+     state.hibernate_after}
   end
 
   def handle_call({:stop, reason}, _from, state) do

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

@@ -97,6 +97,7 @@ defmodule Electric.Shapes.Consumer.Materializer do
         tag_indices: %{},
         value_counts: %{},
         offset: LogOffset.before_all(),
+        subscribed_offset: nil,
         ref: nil,
         subscribers: MapSet.new()
       })
@@ -113,13 +114,15 @@ defmodule Electric.Shapes.Consumer.Materializer do
     try do
       case Consumer.await_snapshot_start(stack_id, shape_handle, :infinity) do
         :started ->
-          Consumer.subscribe_materializer(stack_id, shape_handle, self())
+          {:ok, subscribed_offset} =
+            Consumer.subscribe_materializer(stack_id, shape_handle, self())
 
           Process.monitor(Consumer.whereis(stack_id, shape_handle),
             tag: {:consumer_down, state.shape_handle}
           )
 
-          {:noreply, state, {:continue, {:read_stream, shape_storage}}}
+          {:noreply, %{state | subscribed_offset: subscribed_offset},
+           {:continue, {:read_stream, shape_storage}}}
 
         {:error, _reason} ->
           {:stop, :shutdown, state}
@@ -133,7 +136,8 @@ defmodule Electric.Shapes.Consumer.Materializer do
   end
 
   def handle_continue({:read_stream, storage}, state) do
-    {:ok, offset, stream} = get_stream_up_to_date(state.offset, storage)
+    {:ok, offset, stream} =
+      get_stream_up_to_offset(state.offset, state.subscribed_offset, storage)
 
     {state, _} =
       stream
@@ -143,22 +147,20 @@ defmodule Electric.Shapes.Consumer.Materializer do
     {:noreply, %{state | offset: offset}}
   end
 
-  def get_stream_up_to_date(min_offset, storage) do
-    case Storage.get_chunk_end_log_offset(min_offset, storage) do
-      nil ->
-        {:ok, max_offset} = Storage.fetch_latest_offset(storage)
-
-        if is_log_offset_lte(max_offset, min_offset) do
-          {:ok, min_offset, []}
-        else
-          stream = Storage.get_log_stream(min_offset, max_offset, storage)
-          {:ok, max_offset, stream}
-        end
-
-      max_offset ->
-        stream1 = Storage.get_log_stream(min_offset, max_offset, storage)
-        {:ok, offset, stream2} = get_stream_up_to_date(max_offset, storage)
-        {:ok, offset, Stream.concat(stream1, stream2)}
+  @doc """
+  Get a stream of log entries from storage, bounded by the subscribed offset.
+
+  The subscribed_offset is the Consumer's latest_offset at the time of subscription.
+  We only read up to this offset to avoid duplicates - any changes after this offset
+  will be delivered via new_changes messages from the Consumer.
+  """
+  def get_stream_up_to_offset(min_offset, subscribed_offset, storage) do
+    # If subscribed_offset is nil or at/before min_offset, nothing to read
+    if is_nil(subscribed_offset) or is_log_offset_lte(subscribed_offset, min_offset) do
+      {:ok, min_offset, []}
+    else
+      stream = Storage.get_log_stream(min_offset, subscribed_offset, storage)
+      {:ok, subscribed_offset, stream}
     end
   end
 

packages/sync-service/test/electric/shapes/consumer/materializer_test.exs

@@ -62,7 +62,11 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
       })
 
     respond_to_call(:await_snapshot_start, :started)
-    respond_to_call(:subscribe_materializer, :ok)
+
+    respond_to_call(
+      :subscribe_materializer,
+      {:ok, Electric.Replication.LogOffset.last_before_real_offsets()}
+    )
 
     assert Materializer.wait_until_ready(ctx) == :ok
   end
@@ -79,7 +83,11 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
       })
 
     respond_to_call(:await_snapshot_start, :started)
-    respond_to_call(:subscribe_materializer, :ok)
+
+    respond_to_call(
+      :subscribe_materializer,
+      {:ok, Electric.Replication.LogOffset.last_before_real_offsets()}
+    )
 
     assert Materializer.wait_until_ready(ctx) == :ok
 
@@ -763,7 +771,11 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
       })
 
     respond_to_call(:await_snapshot_start, :started)
-    respond_to_call(:subscribe_materializer, :ok)
+
+    respond_to_call(
+      :subscribe_materializer,
+      {:ok, Electric.Replication.LogOffset.last_before_real_offsets()}
+    )
 
     assert Materializer.wait_until_ready(ctx) == :ok
     Materializer.subscribe(ctx)
@@ -790,6 +802,104 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
     |> Enum.map(&Changes.fill_key(&1, pk_cols))
   end
 
+  describe "startup offset coordination" do
+    test "no duplicate when offset coordination prevents overlap", ctx do
+      # This test verifies the fix: offset coordination prevents the duplicate.
+      # We use a mocked Consumer that returns an offset, and the Materializer
+      # should only read storage up to that offset, preventing duplicates.
+
+      Process.flag(:trap_exit, true)
+
+      shape_handle = "offset-test-#{System.unique_integer()}"
+
+      # Setup storage with a record at offset first()
+      storage = Storage.for_shape(shape_handle, ctx.storage)
+      Storage.start_link(storage)
+      writer = Storage.init_writer!(storage, @shape)
+      Storage.mark_snapshot_as_started(storage)
+
+      # Write a record to storage at LogOffset.first()
+      alias Electric.Replication.LogOffset
+      first_offset = LogOffset.first()
+
+      writer =
+        Storage.append_to_log!(
+          [
+            {first_offset, ~s|"public"."test_table"/"1"|, :insert,
+             ~s|{"key":"\\"public\\".\\"test_table\\"/\\"1\\"","value":{"id":"1","value":"10"},"headers":{"operation":"insert"}}|}
+          ],
+          writer
+        )
+
+      Storage.hibernate(writer)
+
+      # Mock Consumer that returns offset AND sends duplicate
+      test_pid = self()
+
+      consumer =
+        spawn(fn ->
+          receive do
+            {:"$gen_call", {from, ref}, :await_snapshot_start} ->
+              send(from, {ref, :started})
+          end
+
+          receive do
+            {:"$gen_call", {from, ref}, {:subscribe_materializer, mat_pid}} ->
+              # Return offset BEFORE the record - Materializer will read nothing from storage
+              send(from, {ref, {:ok, LogOffset.before_all()}})
+
+              # Simulate the race: send the same record via new_changes
+              # This should NOT crash because Materializer didn't read it from storage
+              GenServer.call(
+                mat_pid,
+                {:new_changes,
+                 [
+                   %Changes.NewRecord{
+                     relation: {"public", "test_table"},
+                     key: ~s|"public"."test_table"/"1"|,
+                     record: %{"id" => "1", "value" => "10"},
+                     move_tags: []
+                   }
+                 ]}
+              )
+
+              send(test_pid, :consumer_done)
+          end
+
+          Process.sleep(:infinity)
+        end)
+
+      ConsumerRegistry.register_consumer(consumer, shape_handle, ctx.stack_id)
+
+      # Start Materializer - should NOT crash
+      result =
+        Materializer.start_link(%{
+          stack_id: ctx.stack_id,
+          shape_handle: shape_handle,
+          storage: ctx.storage,
+          columns: ["value"],
+          materialized_type: {:array, :int8}
+        })
+
+      case result do
+        {:ok, pid} ->
+          # Wait for Consumer mock to finish
+          assert_receive :consumer_done, 5000
+          # Materializer should be alive
+          assert Process.alive?(pid)
+          # And should have the value
+          assert Materializer.get_link_values(%{
+                   stack_id: ctx.stack_id,
+                   shape_handle: shape_handle
+                 }) ==
+                   MapSet.new([10])
+
+        {:error, reason} ->
+          flunk("Materializer failed to start: #{inspect(reason)}")
+      end
+    end
+  end
+
   describe "startup race condition handling" do
     # Tests for the race condition where Consumer dies between await_snapshot_start
     # and subscribe_materializer. See concurrency_analysis/MATERIALIZER_RACE_ANALYSIS.md