From 59ef100c21118c8cec98ad4c53aa607d041499f4 Mon Sep 17 00:00:00 2001
From: rob <rob.acourt@gmail.com>
Date: Tue, 30 Jun 2026 14:11:51 +0100
Subject: [PATCH 1/8] Add failing subquery-restore oracle test

A deterministic `test_against_oracle` restart reproduction for optimized subquery shapes diverging from Postgres after a graceful server restart.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 .../test/integration/oracle_restore_test.exs  | 66 +++++++++++++++++++
 1 file changed, 66 insertions(+)

diff --git a/packages/sync-service/test/integration/oracle_restore_test.exs b/packages/sync-service/test/integration/oracle_restore_test.exs
index 6de429df03..a02bf09509 100644
--- a/packages/sync-service/test/integration/oracle_restore_test.exs
+++ b/packages/sync-service/test/integration/oracle_restore_test.exs
@@ -129,4 +129,70 @@ defmodule Electric.Integration.OracleRestoreTest do
 
     OracleHarness.test_against_oracle(ctx, shapes, batches, restart_server_every: 1)
   end
+
+  @tag :oracle_restore_optimized_refetch
+  # Build a large persisted backlog on the `projects` source shape: 200 toggles
+  # under a small `chunk_bytes_threshold` so its log spans many chunks. After the
+  # restart the persistent replication slot has to replay that backlog.
+  @tag chunk_bytes_threshold: 200
+  test "optimized subquery shape must-refetches after restart during slot catch-up replay",
+       ctx do
+    # Two `optimized: true` subquery shapes over the same `projects` source.
+    #
+    # Regression test. Before the fix: after the restart, the `projects` source
+    # consumer replays batch_1 from the persistent slot and re-delivers those
+    # already-applied changes to the subquery materializer via `new_changes`. The
+    # materializer re-applied them and crashed ("Key ... already exists"), which
+    # cascaded — handle_materializer_down -> stop_and_clean ->
+    # handle_writer_termination({:shutdown, :cleanup}) -> remove_shape_async ->
+    # notify_shape_rotation — removing the (healthy) shapes and sending the
+    # polling client a 409 must-refetch.
+    #
+    # The fix makes the materializer ignore `new_changes` ranges it already
+    # applied during its startup history replay, so the crash (and the whole
+    # removal cascade) no longer happens. NB the underlying cascade — a
+    # materializer crash tearing down and *removing* healthy dependent shapes —
+    # is a separate hardening concern (the "bug 6" cascade) this fix leaves open.
+    shapes = [
+      %{
+        name: "issues_of_active_projects",
+        table: "issues",
+        where: "project_id IN (SELECT id FROM projects WHERE active = true)",
+        columns: ["id", "project_id", "title"],
+        pk: ["id"],
+        optimized: true
+      },
+      %{
+        name: "issues_of_inactive_projects",
+        table: "issues",
+        where: "project_id IN (SELECT id FROM projects WHERE active = false)",
+        columns: ["id", "project_id", "title"],
+        pk: ["id"],
+        optimized: true
+      }
+    ]
+
+    # batch_1: 200 toggles of p5's `active` flag — the backlog. Under the small
+    # `chunk_bytes_threshold` above this makes the `projects` source log span many
+    # chunks. p5 ends active, so pre-restart both shapes match the oracle; the
+    # restart then replays this backlog from the slot.
+    toggles =
+      Enum.flat_map(1..100, fn _ ->
+        [
+          [%{name: "deactivate_p5", sql: "UPDATE projects SET active = false WHERE id = 'p5'"}],
+          [%{name: "reactivate_p5", sql: "UPDATE projects SET active = true WHERE id = 'p5'"}]
+        ]
+      end)
+
+    # batch_2: a single dependency move applied after the restart. In practice the
+    # test fails during the batch_1 replay before this is reached; it's kept so the
+    # harness runs a post-restart batch/check.
+    batch_2 = [
+      [%{name: "deactivate_p3", sql: "UPDATE projects SET active = false WHERE id = 'p3'"}]
+    ]
+
+    batches = [toggles, batch_2]
+
+    OracleHarness.test_against_oracle(ctx, shapes, batches, restart_server_every: 1)
+  end
 end

From 21f370c0c2f4eea5592f45d9e747c7961252169c Mon Sep 17 00:00:00 2001
From: rob <rob.acourt@gmail.com>
Date: Thu, 2 Jul 2026 12:58:16 +0100
Subject: [PATCH 2/8] Replay missed moves after restart

---
 .../electric/shape_cache/in_memory_storage.ex |  15 +
 .../electric/shape_cache/pure_file_storage.ex |  13 +
 .../lib/electric/shape_cache/storage.ex       |  27 ++
 .../lib/electric/shapes/consumer.ex           | 159 +++++++---
 .../shapes/consumer/event_handler_builder.ex  |  13 +-
 .../electric/shapes/consumer/materializer.ex  | 274 +++++++++++++++++-
 .../lib/electric/shapes/consumer/state.ex     |  15 +
 .../storage_implementations_test.exs          |  27 ++
 .../shapes/consumer/materializer_test.exs     | 120 ++++++++
 .../test/electric/shapes/consumer_test.exs    |  72 +++++
 .../sync-service/test/support/test_storage.ex |  12 +
 11 files changed, 695 insertions(+), 52 deletions(-)

diff --git a/packages/sync-service/lib/electric/shape_cache/in_memory_storage.ex b/packages/sync-service/lib/electric/shape_cache/in_memory_storage.ex
index a2223f8c60..4d4f072d9d 100644
--- a/packages/sync-service/lib/electric/shape_cache/in_memory_storage.ex
+++ b/packages/sync-service/lib/electric/shape_cache/in_memory_storage.ex
@@ -15,6 +15,7 @@ defmodule Electric.ShapeCache.InMemoryStorage do
   @snapshot_start_index 0
   @snapshot_end_index :end
   @pg_snapshot_key :pg_snapshot
+  @move_positions_key :move_positions
   @latest_offset_key :latest_offset
 
   defstruct [
@@ -127,6 +128,20 @@ defmodule Electric.ShapeCache.InMemoryStorage do
     :ok
   end
 
+  @impl Electric.ShapeCache.Storage
+  def set_move_positions!(move_positions, %MS{} = opts) do
+    :ets.insert(opts.snapshot_table, {@move_positions_key, move_positions})
+    :ok
+  end
+
+  @impl Electric.ShapeCache.Storage
+  def fetch_move_positions(%MS{} = opts) do
+    case :ets.lookup(opts.snapshot_table, @move_positions_key) do
+      [{@move_positions_key, move_positions}] -> {:ok, move_positions}
+      [] -> {:ok, %{}}
+    end
+  end
+
   @impl Electric.ShapeCache.Storage
   def get_all_stored_shape_handles(_opts), do: {:ok, MapSet.new()}
 
diff --git a/packages/sync-service/lib/electric/shape_cache/pure_file_storage.ex b/packages/sync-service/lib/electric/shape_cache/pure_file_storage.ex
index 137146010c..88f05ee361 100644
--- a/packages/sync-service/lib/electric/shape_cache/pure_file_storage.ex
+++ b/packages/sync-service/lib/electric/shape_cache/pure_file_storage.ex
@@ -69,6 +69,7 @@ defmodule Electric.ShapeCache.PureFileStorage do
     :last_persisted_txn_offset,
     :snapshot_started?,
     :pg_snapshot,
+    :move_positions,
     :last_snapshot_chunk,
     :compaction_started?,
     :compaction_boundary
@@ -464,6 +465,18 @@ defmodule Electric.ShapeCache.PureFileStorage do
     {:ok, read_cached_metadata(opts, :pg_snapshot)}
   end
 
+  # move_positions is written only when an outer subquery consumer applies a
+  # dependency move and read only at consumer startup, so it is persisted
+  # directly to disk (term-encoded) rather than through the ETS metadata cache.
+  def set_move_positions!(move_positions, %__MODULE__{} = opts) do
+    write_metadata!(opts, :move_positions, move_positions)
+    :ok
+  end
+
+  def fetch_move_positions(%__MODULE__{} = opts) do
+    {:ok, read_metadata!(opts, :move_positions) || %{}}
+  end
+
   defp read_latest_offset(%__MODULE__{} = opts) do
     read_multiple_cached_metadata(opts, [
       :last_seen_txn_offset,
diff --git a/packages/sync-service/lib/electric/shape_cache/storage.ex b/packages/sync-service/lib/electric/shape_cache/storage.ex
index eededf50fb..1998612294 100644
--- a/packages/sync-service/lib/electric/shape_cache/storage.ex
+++ b/packages/sync-service/lib/electric/shape_cache/storage.ex
@@ -18,6 +18,11 @@ defmodule Electric.ShapeCache.Storage do
           filter_txns?: boolean()
         }
   @type offset :: LogOffset.t()
+  @typedoc """
+  Per-dependency "moves-applied-up-to" source LSN positions for an outer
+  subquery consumer, keyed by the dependency's shape handle.
+  """
+  @type move_positions :: %{shape_handle() => LogOffset.t()}
 
   @type compiled_opts :: term()
   @type shape_opts :: term()
@@ -77,6 +82,18 @@ defmodule Electric.ShapeCache.Storage do
 
   @callback set_pg_snapshot(pg_snapshot(), shape_opts()) :: :ok
 
+  @doc """
+  Persist the per-dependency moves-applied-up-to positions for an outer
+  subquery consumer.
+  """
+  @callback set_move_positions!(move_positions(), shape_opts()) :: :ok
+
+  @doc """
+  Fetch the per-dependency moves-applied-up-to positions for an outer subquery
+  consumer. Returns `{:ok, %{}}` when none have been persisted yet.
+  """
+  @callback fetch_move_positions(shape_opts()) :: {:ok, move_positions()} | {:error, term()}
+
   @doc "Check if snapshot for a given shape handle already exists"
   @callback snapshot_started?(shape_opts()) :: boolean()
 
@@ -320,6 +337,16 @@ defmodule Electric.ShapeCache.Storage do
     mod.set_pg_snapshot(pg_snapshot, shape_opts)
   end
 
+  @impl __MODULE__
+  def set_move_positions!(move_positions, {mod, shape_opts}) do
+    mod.set_move_positions!(move_positions, shape_opts)
+  end
+
+  @impl __MODULE__
+  def fetch_move_positions({mod, shape_opts}) do
+    mod.fetch_move_positions(shape_opts)
+  end
+
   @impl __MODULE__
   def snapshot_started?({mod, shape_opts}) do
     mod.snapshot_started?(shape_opts)
diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex
index 0935f7c9ea..61153aab3f 100644
--- a/packages/sync-service/lib/electric/shapes/consumer.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer.ex
@@ -8,6 +8,7 @@ defmodule Electric.Shapes.Consumer do
   alias Electric.Shapes.Consumer.PendingTxn
   alias Electric.Shapes.Consumer.SetupEffects
   alias Electric.Shapes.Consumer.State
+  alias Electric.Shapes.Consumer.Subqueries.MoveQueue
 
   import Electric.Shapes.Consumer.State, only: :macros
   require Electric.Replication.LogOffset
@@ -352,6 +353,10 @@ defmodule Electric.Shapes.Consumer do
       "Consumer reacting to #{length(move_in)} move ins and #{length(move_out)} move outs from its #{dep_handle} dependency"
     end)
 
+    # Remember the source LSN of this move so that the per-dependency
+    # moves-position can be advanced once the move pipeline is fully drained.
+    state = record_pending_move_lsn(state, dep_handle, payload)
+
     handle_apply_event_result(
       state,
       apply_event(state, {:materializer_changes, dep_handle, payload})
@@ -986,10 +991,55 @@ defmodule Electric.Shapes.Consumer do
             {{previous_offset, result.state.latest_offset}, result.state.latest_offset}
           end
 
-        {result.state, notification, result.num_changes, result.total_size}
+        final_state = maybe_commit_move_positions(result.state)
+
+        {final_state, notification, result.num_changes, result.total_size}
+    end
+  end
+
+  # Stash the source LSN carried by a materializer move payload, keyed by
+  # dependency handle. The position is only committed (advanced + persisted)
+  # once the move pipeline is fully drained (see `maybe_commit_move_positions/1`).
+  defp record_pending_move_lsn(state, dep_handle, payload) do
+    case Map.get(payload, :lsn) do
+      nil ->
+        state
+
+      lsn ->
+        %{state | pending_move_lsns: Map.put(state.pending_move_lsns, dep_handle, lsn)}
+    end
+  end
+
+  # Once the subquery move pipeline is fully drained (Steady with an empty move
+  # queue), every move received so far has been applied to storage, so the
+  # per-dependency moves-positions can be safely advanced to the latest received
+  # source LSNs and persisted. This is the dedup key used to replay only the
+  # missed tail after a restart.
+  defp maybe_commit_move_positions(%State{pending_move_lsns: pending} = state)
+       when pending == %{},
+       do: state
+
+  defp maybe_commit_move_positions(%State{} = state) do
+    if move_pipeline_fully_drained?(state.event_handler) do
+      move_positions =
+        Map.merge(state.move_positions, state.pending_move_lsns, fn _handle, old, new ->
+          LogOffset.max(old, new)
+        end)
+
+      ShapeCache.Storage.set_move_positions!(move_positions, state.storage)
+
+      %{state | move_positions: move_positions, pending_move_lsns: %{}}
+    else
+      state
     end
   end
 
+  defp move_pipeline_fully_drained?(%EventHandler.Subqueries.Steady{queue: queue}),
+    do: MoveQueue.length(queue) == 0
+
+  defp move_pipeline_fully_drained?(%EventHandler.Subqueries.Buffering{}), do: false
+  defp move_pipeline_fully_drained?(_handler), do: true
+
   defp handle_event_error(state, {:truncate, xid}) do
     handle_txn_with_truncate(xid, state)
   end
@@ -1199,41 +1249,68 @@ defmodule Electric.Shapes.Consumer do
   end
 
   defp finish_initialization(%State{} = state, action, otel_ctx) do
-    if all_materializers_alive?(state) do
-      case initialize_event_handler(state, action) do
-        {:ok, state} ->
-          Logger.debug("Writer for #{state.shape_handle} initialized")
-
-          # We start the snapshotter even if there's a snapshot because it also performs the call
-          # to PublicationManager.add_shape/3. We *could* do that call here and avoid spawning a
-          # process if the shape already has a snapshot but the current semantics rely on being able
-          # to wait for the snapshot asynchronously and if we called publication manager here it would
-          # block and prevent await_snapshot_start calls from adding snapshot subscribers.
-
-          {:ok, _pid} =
-            Shapes.DynamicConsumerSupervisor.start_snapshotter(
-              state.stack_id,
-              %{
-                stack_id: state.stack_id,
-                shape: state.shape,
-                shape_handle: state.shape_handle,
-                storage: state.storage,
-                otel_ctx: otel_ctx
-              }
-            )
-
-          {:noreply, state}
-
-        {:error, state} ->
-          stop_and_clean(state)
-      end
-    else
-      stop_and_clean(state)
+    case subscribe_to_materializers(state) do
+      {:ok, state} ->
+        case initialize_event_handler(state, action) do
+          {:ok, state} ->
+            Logger.debug("Writer for #{state.shape_handle} initialized")
+
+            # We start the snapshotter even if there's a snapshot because it also performs the call
+            # to PublicationManager.add_shape/3. We *could* do that call here and avoid spawning a
+            # process if the shape already has a snapshot but the current semantics rely on being able
+            # to wait for the snapshot asynchronously and if we called publication manager here it would
+            # block and prevent await_snapshot_start calls from adding snapshot subscribers.
+
+            {:ok, _pid} =
+              Shapes.DynamicConsumerSupervisor.start_snapshotter(
+                state.stack_id,
+                %{
+                  stack_id: state.stack_id,
+                  shape: state.shape,
+                  shape_handle: state.shape_handle,
+                  storage: state.storage,
+                  otel_ctx: otel_ctx
+                }
+              )
+
+            {:noreply, state}
+
+          {:error, state} ->
+            stop_and_clean(state)
+        end
+
+      :error ->
+        stop_and_clean(state)
+    end
+  end
+
+  # Subscribe to each dependency materializer, passing the persisted per-dep
+  # moves-position so the materializer replays any moves this consumer missed
+  # across a restart. Captures the returned seed views (as-of the position) for
+  # seeding the event handler's dependency views, and baselines a position for
+  # dependencies that don't have one yet so a first missed move can be replayed.
+  #
+  # Returns `{:ok, state}` with `dep_seed_views`/`move_positions` populated, or
+  # `:error` if any dependency materializer is not alive.
+  defp subscribe_to_materializers(state) do
+    case do_subscribe_to_materializers(state) do
+      {:ok, %State{move_positions: move_positions} = state} ->
+        # Persist baselined positions so a restart before the first move can
+        # still replay it (no-op writes if nothing changed are cheap at startup).
+        if state.shape.shape_dependencies_handles != [] do
+          ShapeCache.Storage.set_move_positions!(move_positions, state.storage)
+        end
+
+        {:ok, state}
+
+      :error ->
+        :error
     end
   end
 
-  defp all_materializers_alive?(state) do
-    Enum.all?(state.shape.shape_dependencies_handles, fn shape_handle ->
+  defp do_subscribe_to_materializers(state) do
+    Enum.reduce_while(state.shape.shape_dependencies_handles, {:ok, state}, fn shape_handle,
+                                                                               {:ok, state} ->
       name = Materializer.name(state.stack_id, shape_handle)
 
       with pid when is_pid(pid) <- GenServer.whereis(name),
@@ -1242,9 +1319,19 @@ defmodule Electric.Shapes.Consumer do
           tag: {:dependency_materializer_down, shape_handle}
         )
 
-        Materializer.subscribe(pid)
+        from_lsn = Map.get(state.move_positions, shape_handle)
+        {:ok, seed_view, applied_offset} = Materializer.subscribe(pid, from_lsn)
+
+        move_positions =
+          Map.put_new(state.move_positions, shape_handle, applied_offset)
+
+        state = %{
+          state
+          | dep_seed_views: Map.put(state.dep_seed_views, shape_handle, seed_view),
+            move_positions: move_positions
+        }
 
-        true
+        {:cont, {:ok, state}}
       else
         _ ->
           Logger.warning(
@@ -1253,7 +1340,7 @@ defmodule Electric.Shapes.Consumer do
             state_shape_handle: state.shape_handle
           )
 
-          false
+          {:halt, :error}
       end
     end)
   end
diff --git a/packages/sync-service/lib/electric/shapes/consumer/event_handler_builder.ex b/packages/sync-service/lib/electric/shapes/consumer/event_handler_builder.ex
index feb251e914..db2e7d3316 100644
--- a/packages/sync-service/lib/electric/shapes/consumer/event_handler_builder.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer/event_handler_builder.ex
@@ -22,7 +22,18 @@ defmodule Electric.Shapes.Consumer.EventHandlerBuilder do
                                          {views, handle_mapping, index_mapping} ->
         materializer_opts = %{stack_id: state.stack_id, shape_handle: handle}
         :ok = Materializer.wait_until_ready(materializer_opts)
-        view = Materializer.get_link_values(materializer_opts)
+
+        # Seed the dependency view from the value captured at subscribe time
+        # (as-of this consumer's persisted moves-position), so that any moves
+        # the materializer replays are not eliminated as redundant against a
+        # view that already reflects them. Falls back to the materializer's
+        # current link values if no seed was captured (non-restart paths).
+        view =
+          case Map.fetch(state.dep_seed_views, handle) do
+            {:ok, seed_view} -> seed_view
+            :error -> Materializer.get_link_values(materializer_opts)
+          end
+
         ref = ["$sublink", Integer.to_string(index)]
 
         {Map.put(views, ref, view), Map.put(handle_mapping, handle, {index, ref}),
diff --git a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
index 1315c277b0..1ca8bd1174 100644
--- a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
@@ -103,17 +103,37 @@ defmodule Electric.Shapes.Consumer.Materializer do
     end)
   end
 
-  # A materializer cannot answer any call until `handle_continue(:start_materializer)`
-  # returns, and that blocks on `await_snapshot_start(:infinity)` until its own snapshot
-  # starts. A cold dependency snapshot can take longer than the default 5s `GenServer.call`
-  # timeout, so we wait with `:infinity` (consistent with `wait_until_ready/1` and
-  # `new_changes/3`). Liveness is handled by the caller monitoring the materializer, so a
-  # dead materializer surfaces as a call exit rather than being masked by a short timeout.
-  def subscribe(pid) when is_pid(pid), do: GenServer.call(pid, :subscribe, :infinity)
+  @doc """
+  Subscribe `pid` to this materializer's move events.
+
+  `from_lsn` is the source LSN (LogOffset) up to which the subscribing outer
+  consumer has already applied moves from this dependency, or `nil` for a fresh
+  subscription. When `from_lsn` is behind the materializer's applied position,
+  the moves in `(from_lsn, applied_offset]` are replayed to `pid` so it can
+  catch up after a restart.
+
+  Returns `{:ok, seed_link_values, applied_offset}` where `seed_link_values` is
+  the set of link values as of `from_lsn` (used to seed the outer consumer's
+  dependency view so replayed moves are not redundancy-eliminated), and
+  `applied_offset` is the materializer's current applied source LSN.
+
+  Calls use an `:infinity` timeout: a materializer cannot answer until
+  `handle_continue(:start_materializer)` returns, which blocks on
+  `await_snapshot_start(:infinity)`, and a cold dependency snapshot can take
+  longer than the default 5s `GenServer.call` timeout (consistent with
+  `wait_until_ready/1` and `new_changes/3`). Liveness is handled by the caller
+  monitoring the materializer, so a dead materializer surfaces as a call exit
+  rather than being masked by a short timeout.
+  """
+  def subscribe(pid, from_lsn \\ nil)
+
+  def subscribe(pid, from_lsn) when is_pid(pid),
+    do: GenServer.call(pid, {:subscribe, from_lsn}, :infinity)
 
-  def subscribe(opts) when is_map(opts), do: GenServer.call(name(opts), :subscribe, :infinity)
+  def subscribe(opts, from_lsn) when is_map(opts),
+    do: GenServer.call(name(opts), {:subscribe, from_lsn}, :infinity)
 
-  def subscribe(stack_id, shape_handle),
+  def subscribe(stack_id, shape_handle) when is_stack_id(stack_id),
     do: subscribe(%{stack_id: stack_id, shape_handle: shape_handle})
 
   def start_link(opts) do
@@ -138,6 +158,10 @@ defmodule Electric.Shapes.Consumer.Materializer do
         value_counts: %{},
         pending_events: %{},
         offset: LogOffset.before_all(),
+        # The highest source LSN (LogOffset) up to which changes have been
+        # applied to `value_counts`. Used to tag emitted moves with their
+        # source LSN and to bound move replay on subscribe.
+        applied_offset: LogOffset.before_all(),
         subscribed_offset: nil,
         ref: nil,
         subscribers: MapSet.new()
@@ -178,6 +202,14 @@ defmodule Electric.Shapes.Consumer.Materializer do
 
   def handle_continue({:read_stream, storage}, state) do
     state = read_history_up_to_subscribed(state, storage)
+    # After the startup replay, everything up to `subscribed_offset` has been
+    # applied; seed `applied_offset` accordingly so live moves and replay are
+    # tagged/bounded from the right position.
+    state =
+      if is_nil(state.subscribed_offset),
+        do: state,
+        else: %{state | applied_offset: state.subscribed_offset}
+
     write_link_values(state)
     {:noreply, state}
   end
@@ -201,7 +233,7 @@ defmodule Electric.Shapes.Consumer.Materializer do
   changes after this offset will be delivered via new_changes messages
   from the Consumer.
   """
-  def read_history_up_to_subscribed(state, storage) do
+  def read_history_up_to_subscribed(state, storage, apply_fun \\ &default_history_apply/2) do
     cond do
       is_nil(state.subscribed_offset) ->
         state
@@ -211,7 +243,7 @@ defmodule Electric.Shapes.Consumer.Materializer do
 
       true ->
         stream = Storage.get_log_stream(state.offset, state.subscribed_offset, storage)
-        {state, _} = stream |> decode_json_stream() |> apply_changes(state)
+        state = apply_fun.(stream, state)
 
         # If the read just covered the main log (because either the
         # current offset is already past the snapshot or the next chunk
@@ -252,12 +284,195 @@ defmodule Electric.Shapes.Consumer.Materializer do
               %{state | offset: state.subscribed_offset}
 
             true ->
-              read_history_up_to_subscribed(%{state | offset: next_offset}, storage)
+              read_history_up_to_subscribed(%{state | offset: next_offset}, storage, apply_fun)
           end
         end
     end
   end
 
+  # Default apply function for `read_history_up_to_subscribed/3`: apply the
+  # decoded stream to `value_counts`, discarding the emitted move events (the
+  # startup replay only needs the resulting state).
+  defp default_history_apply(stream, state) do
+    {state, _events} = stream |> decode_json_stream() |> apply_changes(state)
+    state
+  end
+
+  # Replay the moves in `(from_lsn, applied_offset]` to `pid`, returning the set
+  # of link values as of `from_lsn` (the seed view for the outer consumer).
+  #
+  # When `from_lsn` is nil (fresh subscription) or is already at/after the
+  # materializer's applied position, there is nothing to replay and the current
+  # link values are returned as the seed.
+  defp maybe_replay_moves(state, _pid, nil), do: link_values_from_counts(state.value_counts)
+
+  defp maybe_replay_moves(state, pid, from_lsn) do
+    if is_log_offset_lte(state.applied_offset, from_lsn) do
+      link_values_from_counts(state.value_counts)
+    else
+      replay_moves(state, pid, from_lsn)
+    end
+  end
+
+  defp replay_moves(state, pid, from_lsn) do
+    stack_storage = Storage.for_stack(state.stack_id)
+    storage = Storage.for_shape(state.shape_handle, stack_storage)
+
+    # A throwaway accumulator shaped like the materializer state so we can reuse
+    # `apply_changes/2` and `cast!/2` without touching the live value_counts.
+    replay0 = %{
+      state
+      | index: %{},
+        tag_indices: %{},
+        value_counts: %{},
+        offset: LogOffset.before_all(),
+        subscribed_offset: state.applied_offset,
+        # replay bookkeeping (consumed only by the replay apply fun below)
+        pending_events: %{from_lsn: from_lsn, pid: pid, seed: nil}
+    }
+
+    replay =
+      read_history_up_to_subscribed(replay0, storage, fn stream, acc ->
+        apply_replay_stream(stream, acc)
+      end)
+
+    case replay.pending_events.seed do
+      nil -> link_values_from_counts(replay.value_counts)
+      seed -> seed
+    end
+  end
+
+  # Apply function used during move replay. Decodes the stream with per-item
+  # source offsets, groups items into per-transaction batches, and for each
+  # batch:
+  #   * captures the seed view (value counts as of `from_lsn`) on the first
+  #     batch strictly after `from_lsn`;
+  #   * applies the batch to the throwaway value_counts;
+  #   * emits the resulting moves to `pid`, tagged with the batch's source LSN,
+  #     for batches strictly after `from_lsn` (batches at or before `from_lsn`
+  #     only build state and are not emitted).
+  defp apply_replay_stream(stream, acc) do
+    %{from_lsn: from_lsn, pid: pid} = acc.pending_events
+    handle = acc.shape_handle
+
+    stream
+    |> decode_json_stream_with_offsets()
+    |> chunk_by_txn()
+    |> Enum.reduce(acc, fn {txn_offset, changes, txids}, acc ->
+      emit? = is_log_offset_lt(from_lsn, txn_offset)
+
+      acc =
+        if emit? and is_nil(acc.pending_events.seed) do
+          seed = link_values_from_counts(acc.value_counts)
+          put_in(acc.pending_events.seed, seed)
+        else
+          acc
+        end
+
+      {acc, events} = apply_changes(changes, acc)
+
+      if emit? do
+        events =
+          case events do
+            empty when empty == %{} -> %{}
+            events -> Map.put(events, :txids, MapSet.new(txids))
+          end
+          |> cancel_matching_move_events()
+
+        if events != %{} do
+          payload =
+            events
+            |> finalize_txids()
+            |> Map.put(:lsn, txn_offset)
+
+          send(pid, {:materializer_changes, handle, payload})
+        end
+      end
+
+      acc
+    end)
+  end
+
+  # Decode a raw JSON log stream into `{log_offset, txids, change}` tuples,
+  # preserving the per-item source offset (reconstructed from the item headers)
+  # so replay can delimit transactions and tag emitted moves.
+  defp decode_json_stream_with_offsets(stream) do
+    stream
+    |> Stream.map(&Jason.decode!/1)
+    |> Stream.filter(fn decoded ->
+      Map.has_key?(decoded, "key") || Map.has_key?(decoded["headers"], "event")
+    end)
+    |> Stream.map(fn decoded ->
+      headers = decoded["headers"]
+      {decode_offset(headers), decode_txids(headers), decode_change(decoded)}
+    end)
+  end
+
+  defp decode_offset(%{"lsn" => lsn, "op_position" => op}) when is_binary(lsn),
+    do: LogOffset.new(String.to_integer(lsn), op)
+
+  defp decode_offset(_headers), do: LogOffset.before_all()
+
+  defp decode_txids(%{"txids" => txids}) when is_list(txids), do: txids
+  defp decode_txids(_headers), do: []
+
+  defp decode_change(%{
+         "key" => key,
+         "value" => value,
+         "headers" => %{"operation" => operation} = headers
+       }) do
+    case operation do
+      "insert" ->
+        %Changes.NewRecord{
+          key: key,
+          record: value,
+          move_tags: Map.get(headers, "tags", []),
+          active_conditions: Map.get(headers, "active_conditions", [])
+        }
+
+      "update" ->
+        %Changes.UpdatedRecord{
+          key: key,
+          record: value,
+          move_tags: Map.get(headers, "tags", []),
+          removed_move_tags: Map.get(headers, "removed_tags", []),
+          active_conditions: Map.get(headers, "active_conditions", [])
+        }
+
+      "delete" ->
+        %Changes.DeletedRecord{
+          key: key,
+          old_record: value,
+          move_tags: Map.get(headers, "tags", []),
+          active_conditions: Map.get(headers, "active_conditions", [])
+        }
+    end
+  end
+
+  defp decode_change(%{"headers" => %{"event" => event, "patterns" => patterns} = headers})
+       when event in ["move-out", "move-in"] do
+    patterns =
+      Enum.map(patterns, fn %{"pos" => pos, "value" => value} ->
+        %{pos: pos, value: value}
+      end)
+
+    %{headers: %{event: event, patterns: patterns, txids: Map.get(headers, "txids", [])}}
+  end
+
+  # Group consecutive decoded items into per-transaction batches. Items sharing a
+  # `tx_offset` belong to the same source transaction; each batch is tagged with
+  # the batch's last (largest) offset as its source LSN.
+  defp chunk_by_txn(items) do
+    items
+    |> Enum.chunk_by(fn {offset, _txids, _change} -> offset.tx_offset end)
+    |> Enum.map(fn batch ->
+      {offset, _, _} = List.last(batch)
+      changes = Enum.map(batch, fn {_o, _t, change} -> change end)
+      txids = batch |> Enum.flat_map(fn {_o, t, _c} -> t end) |> Enum.uniq()
+      {offset, changes, txids}
+    end)
+  end
+
   def handle_call(:get_link_values, _from, %{value_counts: value_counts} = state) do
     {:reply, link_values_from_counts(value_counts), state}
   end
@@ -266,10 +481,28 @@ 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)
 
+    # Track the source LSN of this batch so emitted moves can be tagged with it
+    # (used by outer consumers to dedup/replay moves across a restart).
+    state = %{state | applied_offset: range_end}
+
     state =
       Storage.get_log_stream(range_start, range_end, storage)
       |> decode_json_stream()
@@ -288,10 +521,18 @@ defmodule Electric.Shapes.Consumer.Materializer do
     {:reply, :ok, state}
   end
 
-  def handle_call(:subscribe, {pid, _ref} = _from, state) do
+  def handle_call({:subscribe, from_lsn}, {pid, _ref} = _from, state) do
     Process.monitor(pid)
 
-    {:reply, :ok, %{state | subscribers: MapSet.put(state.subscribers, pid)}}
+    # Register the subscriber *before* replaying so that any live move flushed
+    # after this call is delivered to `pid` and interleaves correctly after the
+    # replayed tail (replay covers up to `applied_offset`; live moves are
+    # strictly beyond it).
+    state = %{state | subscribers: MapSet.put(state.subscribers, pid)}
+
+    seed_link_values = maybe_replay_moves(state, pid, from_lsn)
+
+    {:reply, {:ok, seed_link_values, state.applied_offset}, state}
   end
 
   # if the supervisor is going down then this process will also be taken down
@@ -471,7 +712,10 @@ defmodule Electric.Shapes.Consumer.Materializer do
       cancel_matching_move_events(state.pending_events)
 
     if events != %{} do
-      events = finalize_txids(events)
+      events =
+        events
+        |> finalize_txids()
+        |> Map.put(:lsn, state.applied_offset)
 
       for pid <- state.subscribers do
         send(pid, {:materializer_changes, state.shape_handle, events})
diff --git a/packages/sync-service/lib/electric/shapes/consumer/state.ex b/packages/sync-service/lib/electric/shapes/consumer/state.ex
index e23edee4dc..8968fa94bf 100644
--- a/packages/sync-service/lib/electric/shapes/consumer/state.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer/state.ex
@@ -26,6 +26,19 @@ defmodule Electric.Shapes.Consumer.State do
     buffer: [],
     txn_offset_mapping: [],
     materializer_subscribed?: false,
+    # Per-dependency "moves-applied-up-to" source LSNs (`%{dep_handle => LogOffset}`),
+    # persisted so that after a restart the outer subquery consumer can ask each
+    # dependency materializer to replay the moves it missed and dedup by position.
+    move_positions: %{},
+    # Per-dependency seed views (`%{dep_handle => MapSet}`) captured from the
+    # materializer at subscribe time (as-of `move_positions`), used to seed the
+    # event handler's dependency views so replayed moves are not
+    # redundancy-eliminated.
+    dep_seed_views: %{},
+    # Per-dependency source LSN of the most recently received (not yet committed)
+    # materializer move (`%{dep_handle => LogOffset}`); folded into
+    # `move_positions` once the move pipeline is fully drained.
+    pending_move_lsns: %{},
     terminating?: false,
     buffering?: false,
     # Based on the write unit value, consumer will either buffer txn fragments in memory until
@@ -146,6 +159,7 @@ defmodule Electric.Shapes.Consumer.State do
 
     {:ok, latest_offset} = Storage.fetch_latest_offset(storage)
     {:ok, pg_snapshot} = Storage.fetch_pg_snapshot(storage)
+    {:ok, move_positions} = Storage.fetch_move_positions(storage)
 
     initial_snapshot_state = InitialSnapshot.new(pg_snapshot)
 
@@ -154,6 +168,7 @@ defmodule Electric.Shapes.Consumer.State do
       | latest_offset: latest_offset,
         storage: storage,
         writer: writer,
+        move_positions: move_positions,
         initial_snapshot_state: initial_snapshot_state,
         buffering?: InitialSnapshot.needs_buffering?(initial_snapshot_state)
     }
diff --git a/packages/sync-service/test/electric/shape_cache/storage_implementations_test.exs b/packages/sync-service/test/electric/shape_cache/storage_implementations_test.exs
index d5cbeb674e..a65ff3b32b 100644
--- a/packages/sync-service/test/electric/shape_cache/storage_implementations_test.exs
+++ b/packages/sync-service/test/electric/shape_cache/storage_implementations_test.exs
@@ -124,6 +124,33 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
       end
     end
 
+    describe "#{module_name}.fetch_move_positions/1" do
+      setup :start_storage
+
+      test "returns an empty map on startup", %{storage: opts} do
+        assert Storage.fetch_move_positions(opts) == {:ok, %{}}
+      end
+
+      test "round-trips a per-dependency positions map", %{storage: opts} do
+        positions = %{
+          "dep-a" => LogOffset.new(10, 2),
+          "dep-b" => LogOffset.new(42, 0)
+        }
+
+        assert :ok = Storage.set_move_positions!(positions, opts)
+        assert Storage.fetch_move_positions(opts) == {:ok, positions}
+      end
+
+      test "overwrites previously persisted positions", %{storage: opts} do
+        assert :ok = Storage.set_move_positions!(%{"dep-a" => LogOffset.new(1, 0)}, opts)
+
+        updated = %{"dep-a" => LogOffset.new(5, 0), "dep-b" => LogOffset.new(7, 1)}
+        assert :ok = Storage.set_move_positions!(updated, opts)
+
+        assert Storage.fetch_move_positions(opts) == {:ok, updated}
+      end
+    end
+
     describe "#{module_name}.append_to_log!/3" do
       setup do
         {:ok, %{module: unquote(module)}}
diff --git a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs
index 5cee0c2428..bb3aeaf901 100644
--- a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs
+++ b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs
@@ -1308,6 +1308,27 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
     |> Enum.map(fn {_offset, item} -> Jason.encode!(item) end)
   end
 
+  # Build a single main-log insert log item at `offset` introducing `value`,
+  # encoded the same way the source consumer would write it (headers carry the
+  # `lsn`/`op_position` used to reconstruct the offset during replay).
+  defp main_log_insert(offset, id, value) do
+    change =
+      %Changes.NewRecord{
+        relation: {"public", "test_table"},
+        key: ~s|"public"."test_table"/"#{id}"|,
+        record: %{"id" => id, "value" => value},
+        log_offset: offset,
+        move_tags: []
+      }
+      |> Changes.fill_key(["id"])
+
+    change
+    |> then(&LogItems.from_change(&1, 1, ["id"], :default))
+    |> Enum.map(fn {item_offset, item} ->
+      {item_offset, change.key, :insert, Jason.encode!(item)}
+    end)
+  end
+
   defp prep_changes(changes, opts \\ []) do
     pk_cols = Keyword.get(opts, :pk_cols, ["id"])
     relation = Keyword.get(opts, :relation, {"public", "test_table"})
@@ -1526,6 +1547,105 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
     end
   end
 
+  describe "move replay on subscribe" do
+    # A subscriber that is behind (`from_lsn` < the materializer's applied
+    # position) is caught up by replaying only the moves it missed, each tagged
+    # with its source LSN, and is handed the link values as of `from_lsn` to seed
+    # its dependency view.
+    setup ctx do
+      shape_handle = "replay-test-#{System.unique_integer([:positive])}"
+
+      storage = Storage.for_shape(shape_handle, ctx.storage)
+      Storage.start_link(storage)
+      writer = Storage.init_writer!(storage, @shape)
+      Storage.mark_snapshot_as_started(storage)
+
+      # Snapshot establishes value 10.
+      Storage.make_new_snapshot!(
+        make_snapshot_data([%Changes.NewRecord{record: %{"id" => "1", "value" => "10"}}]),
+        storage
+      )
+
+      # Two main-log inserts at distinct offsets, each introducing a new value
+      # (a move-in): value 20 at (100,0), value 30 at (200,0).
+      writer =
+        Storage.append_to_log!(
+          main_log_insert(LogOffset.new(100, 0), "2", "20"),
+          writer
+        )
+
+      writer =
+        Storage.append_to_log!(
+          main_log_insert(LogOffset.new(200, 0), "3", "30"),
+          writer
+        )
+
+      Storage.hibernate(writer)
+
+      ConsumerRegistry.register_consumer(self(), shape_handle, ctx.stack_id)
+
+      {:ok, _pid} =
+        Materializer.start_link(%{
+          stack_id: ctx.stack_id,
+          shape_handle: shape_handle,
+          storage: ctx.storage,
+          columns: ["value"],
+          materialized_type: {:array, :int8}
+        })
+
+      respond_to_call(:await_snapshot_start, :started)
+      # Subscribed offset past both main-log entries so the materializer applies
+      # the full history at startup.
+      respond_to_call(:subscribe_materializer, {:ok, LogOffset.new(200, 0)})
+
+      mat_ctx = %{stack_id: ctx.stack_id, shape_handle: shape_handle}
+      assert Materializer.wait_until_ready(mat_ctx) == :ok
+      assert Materializer.get_link_values(mat_ctx) == MapSet.new([10, 20, 30])
+
+      Map.put(ctx, :mat_ctx, mat_ctx)
+    end
+
+    test "replays only moves after from_lsn, tagged with per-range source LSNs",
+         %{mat_ctx: mat_ctx} do
+      # Behind at (100,0): the move-in for value 20 (at (100,0)) is already
+      # applied, only value 30 (at (200,0)) must be replayed.
+      assert {:ok, seed, applied_offset} =
+               Materializer.subscribe(mat_ctx, LogOffset.new(100, 0))
+
+      # Seed view is the link values as of (100,0): value 30 not yet included.
+      assert seed == MapSet.new([10, 20])
+      assert applied_offset == LogOffset.new(200, 0)
+
+      assert_receive {:materializer_changes, _handle,
+                      %{move_in: [{30, "30"}], lsn: %LogOffset{tx_offset: 200, op_offset: 0}}}
+
+      # The already-applied move-in for value 20 is NOT replayed.
+      refute_received {:materializer_changes, _handle, %{move_in: [{20, "20"}]}}
+    end
+
+    test "replays every move when from_lsn is before all main-log moves",
+         %{mat_ctx: mat_ctx} do
+      assert {:ok, _seed, _applied} =
+               Materializer.subscribe(mat_ctx, LogOffset.new(50, 0))
+
+      assert_receive {:materializer_changes, _handle,
+                      %{move_in: [{20, "20"}], lsn: %LogOffset{tx_offset: 100, op_offset: 0}}}
+
+      assert_receive {:materializer_changes, _handle,
+                      %{move_in: [{30, "30"}], lsn: %LogOffset{tx_offset: 200, op_offset: 0}}}
+    end
+
+    test "does not replay when from_lsn is at or past the applied position",
+         %{mat_ctx: mat_ctx} do
+      assert {:ok, seed, _applied} =
+               Materializer.subscribe(mat_ctx, LogOffset.new(200, 0))
+
+      # Caught up: seed is the current link values and nothing is replayed.
+      assert seed == MapSet.new([10, 20, 30])
+      refute_received {:materializer_changes, _handle, _payload}
+    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
diff --git a/packages/sync-service/test/electric/shapes/consumer_test.exs b/packages/sync-service/test/electric/shapes/consumer_test.exs
index 2e180f657f..17aa25f262 100644
--- a/packages/sync-service/test/electric/shapes/consumer_test.exs
+++ b/packages/sync-service/test/electric/shapes/consumer_test.exs
@@ -2619,6 +2619,78 @@ defmodule Electric.Shapes.ConsumerTest do
              ] = get_log_items_from_storage(LogOffset.last_before_real_offsets(), shape_storage)
     end
 
+    test "consumer advances and persists the per-dependency moves-position on move application",
+         ctx do
+      parent = self()
+
+      Repatch.patch(
+        Electric.Shapes.Consumer.Effects,
+        :query_move_in_async,
+        [mode: :shared],
+        fn _task_sup, _consumer_state, _buffering_state, consumer_pid ->
+          send(parent, {:query_requested, consumer_pid})
+          :ok
+        end
+      )
+
+      Support.TestUtils.activate_mocks_for_descendant_procs(Consumer)
+
+      {shape_handle, _} =
+        ShapeCache.get_or_create_shape_handle(@shape_with_subquery, ctx.stack_id)
+
+      :started = ShapeCache.await_snapshot_start(shape_handle, ctx.stack_id)
+
+      {:ok, shape} = Electric.Shapes.fetch_shape_by_handle(ctx.stack_id, shape_handle)
+      [dep_handle] = shape.shape_dependencies_handles
+
+      consumer_pid = Consumer.whereis(ctx.stack_id, shape_handle)
+      ref = Shapes.Consumer.register_for_changes(ctx.stack_id, shape_handle)
+      shape_storage = Storage.for_shape(shape_handle, ctx.storage)
+
+      move_lsn = LogOffset.new(777, 0)
+
+      assert :ok = LsnTracker.broadcast_last_seen_lsn(ctx.stack_id, 100)
+
+      send(
+        consumer_pid,
+        {:materializer_changes, dep_handle, %{move_in: [{1, "1"}], move_out: [], lsn: move_lsn}}
+      )
+
+      assert_receive {:query_requested, ^consumer_pid}
+
+      # While the move-in is still buffering the position has NOT advanced to the
+      # move's LSN — it must only advance once the move is applied.
+      {:ok, buffering_positions} = Storage.fetch_move_positions(shape_storage)
+      refute Map.get(buffering_positions, dep_handle) == move_lsn
+
+      send(consumer_pid, {:pg_snapshot_known, {100, 300, []}})
+
+      send_stored_move_in_complete(
+        consumer_pid,
+        shape_storage,
+        [
+          [
+            ~s'"public"."test_table"/"1"',
+            [],
+            Jason.encode!(%{
+              "key" => ~s'"public"."test_table"/"1"',
+              "value" => %{"id" => "1", "value" => "val"},
+              "headers" => %{"operation" => "insert", "relation" => ["public", "test_table"]}
+            })
+          ]
+        ],
+        Lsn.from_integer(100)
+      )
+
+      assert_receive {^ref, :new_changes, _offset}, @receive_timeout
+
+      # Once the move-in splice completes and the pipeline is fully drained, the
+      # per-dependency moves-position is advanced to the move's source LSN and
+      # persisted to storage.
+      {:ok, applied_positions} = Storage.fetch_move_positions(shape_storage)
+      assert Map.get(applied_positions, dep_handle) == move_lsn
+    end
+
     test "consumer startup seeds the stack-scoped subquery index", ctx do
       alias Electric.Shapes.Filter.Indexes.SubqueryIndex
 
diff --git a/packages/sync-service/test/support/test_storage.ex b/packages/sync-service/test/support/test_storage.ex
index 32283ef824..bdf493d49a 100644
--- a/packages/sync-service/test/support/test_storage.ex
+++ b/packages/sync-service/test/support/test_storage.ex
@@ -101,6 +101,18 @@ defmodule Support.TestStorage do
     Storage.set_pg_snapshot(pg_snapshot, storage)
   end
 
+  @impl Electric.ShapeCache.Storage
+  def set_move_positions!(move_positions, {parent, shape_handle, _, storage}) do
+    send(parent, {__MODULE__, :set_move_positions!, shape_handle, move_positions})
+    Storage.set_move_positions!(move_positions, storage)
+  end
+
+  @impl Electric.ShapeCache.Storage
+  def fetch_move_positions({parent, shape_handle, _, storage}) do
+    send(parent, {__MODULE__, :fetch_move_positions, shape_handle})
+    Storage.fetch_move_positions(storage)
+  end
+
   @impl Electric.ShapeCache.Storage
   def snapshot_started?({parent, shape_handle, _, storage}) do
     send(parent, {__MODULE__, :snapshot_started?, shape_handle})

From 7fe71950ac510989a1b3de02f2ad06761544c0a7 Mon Sep 17 00:00:00 2001
From: rob <rob.acourt@gmail.com>
Date: Thu, 2 Jul 2026 13:40:05 +0100
Subject: [PATCH 3/8] Couple move-position persistence to durable flush;
 realistic long-poll timeout

The per-dependency moves-position is now staged when the move pipeline drains
and only committed+persisted once the writer confirms the flush has passed the
move's splice (and, at terminate, after the writer has been flushed). This
keeps the persisted position from running ahead of durable storage across a
restart, which could otherwise leave a subquery shape permanently missing rows.

Also raise the oracle-restore test's long_poll_timeout: a very short one trips
a separate post-restart long-poll readiness race (bug 3), unrelated to the
subquery-restore behaviour under test.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 .changeset/subquery-move-replay-on-restart.md |   9 ++
 .../lib/electric/shapes/consumer.ex           | 115 +++++++++++++++---
 .../lib/electric/shapes/consumer/state.ex     |  14 ++-
 .../test/electric/shapes/consumer_test.exs    |  12 +-
 .../test/integration/oracle_restore_test.exs  |   7 +-
 5 files changed, 136 insertions(+), 21 deletions(-)
 create mode 100644 .changeset/subquery-move-replay-on-restart.md

diff --git a/.changeset/subquery-move-replay-on-restart.md b/.changeset/subquery-move-replay-on-restart.md
new file mode 100644
index 0000000000..8a9b26302f
--- /dev/null
+++ b/.changeset/subquery-move-replay-on-restart.md
@@ -0,0 +1,9 @@
+---
+'@core/sync-service': patch
+---
+
+Fix optimized streaming subquery shapes losing dependency move-ins/move-outs
+across a graceful server restart. On restart the dependency materializer now
+replays the moves each outer consumer missed (deduplicated by a persisted
+per-dependency source-LSN position), so the outer shape catches up instead of
+diverging from Postgres.
diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex
index 61153aab3f..fb746a3501 100644
--- a/packages/sync-service/lib/electric/shapes/consumer.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer.ex
@@ -483,6 +483,11 @@ defmodule Electric.Shapes.Consumer do
     # shape but the alternative is leaking ets tables.
     state = terminate_writer(state)
 
+    # `terminate_writer/1` flushes the writer, so any staged move positions are
+    # now durable — commit them so the persisted position matches storage across
+    # a graceful restart.
+    state = commit_all_move_positions(state)
+
     ShapeCleaner.handle_writer_termination(state.stack_id, state.shape_handle, reason)
 
     State.reply_to_snapshot_waiters(state, {:error, "Shape terminated before snapshot was ready"})
@@ -991,15 +996,16 @@ defmodule Electric.Shapes.Consumer do
             {{previous_offset, result.state.latest_offset}, result.state.latest_offset}
           end
 
-        final_state = maybe_commit_move_positions(result.state)
+        final_state = maybe_stage_move_positions(result.state)
 
         {final_state, notification, result.num_changes, result.total_size}
     end
   end
 
   # Stash the source LSN carried by a materializer move payload, keyed by
-  # dependency handle. The position is only committed (advanced + persisted)
-  # once the move pipeline is fully drained (see `maybe_commit_move_positions/1`).
+  # dependency handle. It is staged once the move pipeline is fully drained (see
+  # `maybe_stage_move_positions/1`) and only persisted once the writer confirms
+  # the flush (see `commit_flushed_move_positions/2`).
   defp record_pending_move_lsn(state, dep_handle, payload) do
     case Map.get(payload, :lsn) do
       nil ->
@@ -1010,30 +1016,107 @@ defmodule Electric.Shapes.Consumer do
     end
   end
 
-  # Once the subquery move pipeline is fully drained (Steady with an empty move
-  # queue), every move received so far has been applied to storage, so the
-  # per-dependency moves-positions can be safely advanced to the latest received
-  # source LSNs and persisted. This is the dedup key used to replay only the
-  # missed tail after a restart.
-  defp maybe_commit_move_positions(%State{pending_move_lsns: pending} = state)
+  # Once the move pipeline is fully drained (Steady, empty queue) every received
+  # move has been applied to the writer buffer. Stage the received source LSNs,
+  # tagged with the current outer `latest_offset` as a flush threshold: the move's
+  # splice rows are at/below `latest_offset`, so they are durable once the writer
+  # has flushed to that offset. Staged entries are only advanced into (and
+  # persisted as) `move_positions` by `commit_flushed_move_positions/2` — so the
+  # persisted position never runs ahead of durable storage across a restart.
+  defp maybe_stage_move_positions(%State{pending_move_lsns: pending} = state)
        when pending == %{},
        do: state
 
-  defp maybe_commit_move_positions(%State{} = state) do
+  defp maybe_stage_move_positions(%State{} = state) do
     if move_pipeline_fully_drained?(state.event_handler) do
-      move_positions =
-        Map.merge(state.move_positions, state.pending_move_lsns, fn _handle, old, new ->
-          LogOffset.max(old, new)
+      threshold = state.latest_offset
+
+      staged =
+        Enum.reduce(state.pending_move_lsns, state.staged_move_positions, fn {handle, lsn}, acc ->
+          Map.update(acc, handle, [{threshold, lsn}], &(&1 ++ [{threshold, lsn}]))
         end)
 
-      ShapeCache.Storage.set_move_positions!(move_positions, state.storage)
+      %{state | staged_move_positions: staged, pending_move_lsns: %{}}
+    else
+      state
+    end
+  end
+
+  # Commit staged move positions whose splice rows are now durable (flush
+  # threshold at/below `flushed_offset`), advancing and persisting
+  # `move_positions`.
+  defp commit_flushed_move_positions(
+         %State{staged_move_positions: staged} = state,
+         _flushed_offset
+       )
+       when staged == %{},
+       do: state
+
+  defp commit_flushed_move_positions(%State{} = state, flushed_offset) do
+    {staged, positions, changed?} =
+      Enum.reduce(state.staged_move_positions, {%{}, state.move_positions, false}, fn
+        {handle, entries}, {staged_acc, positions_acc, changed} ->
+          {committed, remaining} =
+            Enum.split_while(entries, fn {threshold, _lsn} ->
+              LogOffset.is_log_offset_lte(threshold, flushed_offset)
+            end)
+
+          positions_acc =
+            case List.last(committed) do
+              nil -> positions_acc
+              {_threshold, lsn} -> Map.update(positions_acc, handle, lsn, &LogOffset.max(&1, lsn))
+            end
+
+          staged_acc =
+            if remaining == [], do: staged_acc, else: Map.put(staged_acc, handle, remaining)
+
+          {staged_acc, positions_acc, changed or committed != []}
+      end)
 
-      %{state | move_positions: move_positions, pending_move_lsns: %{}}
+    if changed? do
+      ShapeCache.Storage.set_move_positions!(positions, state.storage)
+      %{state | staged_move_positions: staged, move_positions: positions}
     else
       state
     end
   end
 
+  # Commit all staged move positions unconditionally. Called from `terminate/2`
+  # after `terminate_writer/1` has flushed the writer, so every staged move's
+  # splice rows are durable by then. Note this runs after `terminate_writer/1`
+  # has popped `:writer` off the state, so we match a bare map rather than
+  # `%State{}`.
+  defp commit_all_move_positions(state) do
+    staged = Map.get(state, :staged_move_positions, %{})
+    storage = Map.get(state, :storage)
+
+    if staged == %{} or is_nil(storage) do
+      state
+    else
+      positions =
+        Enum.reduce(staged, state.move_positions, fn {handle, entries}, acc ->
+          {_threshold, lsn} = List.last(entries)
+          Map.update(acc, handle, lsn, &LogOffset.max(&1, lsn))
+        end)
+
+      # A persistence failure here must not crash `terminate/2` (that would mask
+      # the shutdown reason, and storage may already be torn down), but it should
+      # not pass silently: a dropped position just means more moves are replayed
+      # on restart. Log it so a genuine failure is observable.
+      try do
+        ShapeCache.Storage.set_move_positions!(positions, storage)
+      rescue
+        error ->
+          Logger.warning(
+            "Failed to persist move positions for shape " <>
+              "#{Map.get(state, :shape_handle)} during terminate: #{Exception.message(error)}"
+          )
+      end
+
+      %{state | move_positions: positions, staged_move_positions: %{}}
+    end
+  end
+
   defp move_pipeline_fully_drained?(%EventHandler.Subqueries.Steady{queue: queue}),
     do: MoveQueue.length(queue) == 0
 
@@ -1216,7 +1299,7 @@ defmodule Electric.Shapes.Consumer do
   defp confirm_flushed_and_notify(state, flushed_offset) do
     {state, txn_offset} = State.align_offset_to_txn_boundary(state, flushed_offset)
     ShapeLogCollector.notify_flushed(state.stack_id, state.shape_handle, txn_offset)
-    state
+    commit_flushed_move_positions(state, flushed_offset)
   end
 
   # After a pending transaction completes and txn_offset_mapping is populated,
diff --git a/packages/sync-service/lib/electric/shapes/consumer/state.ex b/packages/sync-service/lib/electric/shapes/consumer/state.ex
index 8968fa94bf..26429aa8d2 100644
--- a/packages/sync-service/lib/electric/shapes/consumer/state.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer/state.ex
@@ -35,10 +35,18 @@ defmodule Electric.Shapes.Consumer.State do
     # event handler's dependency views so replayed moves are not
     # redundancy-eliminated.
     dep_seed_views: %{},
-    # Per-dependency source LSN of the most recently received (not yet committed)
-    # materializer move (`%{dep_handle => LogOffset}`); folded into
-    # `move_positions` once the move pipeline is fully drained.
+    # Per-dependency source LSN of the most recently received (not yet applied)
+    # materializer move (`%{dep_handle => LogOffset}`); moved into
+    # `staged_move_positions` once the move pipeline is fully drained (applied to
+    # the writer buffer).
     pending_move_lsns: %{},
+    # Per-dependency source LSNs that have been applied to the writer buffer but
+    # are not yet known to be durably flushed
+    # (`%{dep_handle => [{flush_threshold_offset, source_lsn}]}`, ascending).
+    # An entry is committed to `move_positions` (advanced + persisted) only once
+    # the writer confirms a flush at/after its threshold, so the persisted
+    # position never runs ahead of durable storage.
+    staged_move_positions: %{},
     terminating?: false,
     buffering?: false,
     # Based on the write unit value, consumer will either buffer txn fragments in memory until
diff --git a/packages/sync-service/test/electric/shapes/consumer_test.exs b/packages/sync-service/test/electric/shapes/consumer_test.exs
index 17aa25f262..62a38df45b 100644
--- a/packages/sync-service/test/electric/shapes/consumer_test.exs
+++ b/packages/sync-service/test/electric/shapes/consumer_test.exs
@@ -2684,9 +2684,19 @@ defmodule Electric.Shapes.ConsumerTest do
 
       assert_receive {^ref, :new_changes, _offset}, @receive_timeout
 
-      # Once the move-in splice completes and the pipeline is fully drained, the
+      # The splice has been applied to the writer buffer, but the moves-position
+      # is only *staged* — it must not be persisted ahead of a durable flush, or
+      # a restart could leave it pointing past storage. So until the writer
+      # confirms the flush, the persisted position has not advanced.
+      {:ok, staged_positions} = Storage.fetch_move_positions(shape_storage)
+      refute Map.get(staged_positions, dep_handle) == move_lsn
+
+      # Once the writer confirms a flush at/after the move's splice, the
       # per-dependency moves-position is advanced to the move's source LSN and
       # persisted to storage.
+      send(consumer_pid, {Storage, :flushed, LogOffset.new(1_000_000_000, 0)})
+      :sys.get_state(consumer_pid)
+
       {:ok, applied_positions} = Storage.fetch_move_positions(shape_storage)
       assert Map.get(applied_positions, dep_handle) == move_lsn
     end
diff --git a/packages/sync-service/test/integration/oracle_restore_test.exs b/packages/sync-service/test/integration/oracle_restore_test.exs
index a02bf09509..93e677575e 100644
--- a/packages/sync-service/test/integration/oracle_restore_test.exs
+++ b/packages/sync-service/test/integration/oracle_restore_test.exs
@@ -35,7 +35,12 @@ defmodule Electric.Integration.OracleRestoreTest do
   setup ctx do
     ctx =
       with_electric_client(ctx,
-        router_opts: [long_poll_timeout: 100],
+        # A realistic long-poll timeout. A very short one (e.g. 100ms) trips a
+        # separate post-restart long-poll readiness race (bug 3): the poll times
+        # out before replication has caught up after the restart, yielding a
+        # spurious 409. That is independent of the subquery-restore behaviour
+        # under test here.
+        router_opts: [long_poll_timeout: 5000],
         num_clients: 1
       )
 

From 3054a7115a9bb8086367689dfdbf58b2ded6db96 Mon Sep 17 00:00:00 2001
From: rob <rob.acourt@gmail.com>
Date: Tue, 14 Jul 2026 13:09:10 +0100
Subject: [PATCH 4/8] Add failing test for nested subquery replay dropping
 control messages

A nested optimized subquery (issues -> projects-subquery -> regions) produces a
dependency shape that is itself an optimized subquery, so its log contains
move-in/move-out control messages. On restart the dependency materializer's
replay drops those (they decode to before_all()), so the outer shape can be left
missing the issues of projects that moved via a control message.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 .../test/integration/oracle_restore_test.exs  | 150 ++++++++++++++----
 1 file changed, 117 insertions(+), 33 deletions(-)

diff --git a/packages/sync-service/test/integration/oracle_restore_test.exs b/packages/sync-service/test/integration/oracle_restore_test.exs
index 93e677575e..5af5a049e7 100644
--- a/packages/sync-service/test/integration/oracle_restore_test.exs
+++ b/packages/sync-service/test/integration/oracle_restore_test.exs
@@ -1,8 +1,8 @@
 defmodule Electric.Integration.OracleRestoreTest do
   @moduledoc """
-  Targeted regression tests for restore-from-file. Each test exercises a
-  scenario from `bugs.md` with a deterministic, minimal mutation sequence,
-  reusing `Support.OracleHarness.test_against_oracle/4`.
+  Targeted regression tests for subquery-shape restore across a server restart,
+  each with a deterministic, minimal mutation sequence, reusing
+  `Support.OracleHarness.test_against_oracle/4`.
 
   These tests use a small, readable "issue tracker" domain schema rather than
   the abstract `level_N` hierarchy from `Support.OracleHarness.StandardSchema`:
@@ -12,9 +12,6 @@ defmodule Electric.Integration.OracleRestoreTest do
 
   The shape under test is "issues belonging to an active project", expressed
   as a subquery over `projects.active`.
-
-  These tests are expected to fail until the underlying Electric bugs are
-  fixed.
   """
 
   use ExUnit.Case, async: false
@@ -35,10 +32,9 @@ defmodule Electric.Integration.OracleRestoreTest do
   setup ctx do
     ctx =
       with_electric_client(ctx,
-        # A realistic long-poll timeout. A very short one (e.g. 100ms) trips a
-        # separate post-restart long-poll readiness race (bug 3): the poll times
-        # out before replication has caught up after the restart, yielding a
-        # spurious 409. That is independent of the subquery-restore behaviour
+        # A realistic long-poll timeout. A very short one (e.g. 100ms) can time
+        # out before replication has caught up after a restart, yielding a
+        # spurious 409 that is independent of the subquery-restore behaviour
         # under test here.
         router_opts: [long_poll_timeout: 5000],
         num_clients: 1
@@ -101,12 +97,11 @@ defmodule Electric.Integration.OracleRestoreTest do
   end
 
   @tag :oracle_restore_bug_1
-  test "bug 1: subquery shape diverges from oracle after server restart", ctx do
-    # Shape on `issues` with a subquery predicate over `projects.active`.
-    # After the server is restarted, the subquery materializer state is not
-    # restored from disk, so toggling `projects.active` on either side of the
-    # restart produces a divergence between the client view and the oracle
-    # (or a 409 must-refetch on this `optimized: true` shape).
+  test "subquery shape stays consistent with oracle across a server restart", ctx do
+    # A single `optimized: true` shape on `issues` with a subquery predicate over
+    # `projects.active`. Toggling `projects.active` on either side of the restart
+    # moves issue rows in and out of the shape via the subquery materializer; the
+    # client view must stay consistent with the oracle across the restart.
     shapes = [
       %{
         name: "issues_of_active_projects",
@@ -140,24 +135,13 @@ defmodule Electric.Integration.OracleRestoreTest do
   # under a small `chunk_bytes_threshold` so its log spans many chunks. After the
   # restart the persistent replication slot has to replay that backlog.
   @tag chunk_bytes_threshold: 200
-  test "optimized subquery shape must-refetches after restart during slot catch-up replay",
+  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.
-    #
-    # Regression test. Before the fix: after the restart, the `projects` source
-    # consumer replays batch_1 from the persistent slot and re-delivers those
-    # already-applied changes to the subquery materializer via `new_changes`. The
-    # materializer re-applied them and crashed ("Key ... already exists"), which
-    # cascaded — handle_materializer_down -> stop_and_clean ->
-    # handle_writer_termination({:shutdown, :cleanup}) -> remove_shape_async ->
-    # notify_shape_rotation — removing the (healthy) shapes and sending the
-    # polling client a 409 must-refetch.
-    #
-    # The fix makes the materializer ignore `new_changes` ranges it already
-    # applied during its startup history replay, so the crash (and the whole
-    # removal cascade) no longer happens. NB the underlying cascade — a
-    # materializer crash tearing down and *removing* healthy dependent shapes —
-    # is a separate hardening concern (the "bug 6" cascade) this fix leaves open.
+    # 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.
     shapes = [
       %{
         name: "issues_of_active_projects",
@@ -200,4 +184,104 @@ defmodule Electric.Integration.OracleRestoreTest do
 
     OracleHarness.test_against_oracle(ctx, shapes, batches, restart_server_every: 1)
   end
+
+  # A three-level "issue tracker": regions own projects own issues.
+  #
+  #     regions (id, active)
+  #         └── projects (id, region_id)
+  #                 └── issues (id, project_id, title)
+  #
+  # r1, r3 start active; r2 starts inactive. Two projects per region, two issues
+  # per project.
+  defp setup_nested_schema(ctx) do
+    OracleHarness.apply_sql(ctx, [
+      "DROP TABLE IF EXISTS issues CASCADE",
+      "DROP TABLE IF EXISTS projects CASCADE",
+      "DROP TABLE IF EXISTS regions CASCADE",
+      """
+      CREATE TABLE regions (
+        id TEXT PRIMARY KEY,
+        active BOOLEAN NOT NULL DEFAULT true
+      )
+      """,
+      """
+      CREATE TABLE projects (
+        id TEXT PRIMARY KEY,
+        region_id TEXT NOT NULL REFERENCES regions(id) ON DELETE CASCADE
+      )
+      """,
+      """
+      CREATE TABLE issues (
+        id TEXT PRIMARY KEY,
+        project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
+        title TEXT NOT NULL
+      )
+      """
+    ])
+
+    OracleHarness.apply_sql(ctx, [
+      "INSERT INTO regions (id, active) VALUES ('r1', true), ('r2', false), ('r3', true)",
+      """
+      INSERT INTO projects (id, region_id) VALUES
+        ('p1', 'r1'), ('p2', 'r1'),
+        ('p3', 'r2'), ('p4', 'r2'),
+        ('p5', 'r3'), ('p6', 'r3')
+      """,
+      "INSERT INTO issues (id, project_id, title) VALUES " <>
+        (for(n <- 1..12, do: "('i#{n}', 'p#{div(n - 1, 2) + 1}', 'Issue #{n}')")
+         |> Enum.join(", "))
+    ])
+
+    :ok
+  end
+
+  @tag :oracle_restore_nested_subquery
+  # As with the single-level case, force the source shape's log to span multiple
+  # chunks so the post-restart replay path is exercised.
+  @tag chunk_bytes_threshold: 200
+  test "nested optimized subquery shape stays consistent with oracle across a restart", ctx do
+    setup_nested_schema(ctx)
+
+    # The shape under test is a *nested* subquery: issues whose project belongs to
+    # an active region. Its dependency shape — `projects WHERE region_id IN (SELECT
+    # id FROM regions WHERE active)` — is itself an optimized subquery shape, so
+    # its log contains move-in/move-out *control messages* (projects moving in and
+    # out as regions toggle). On restart the dependency materializer replays that
+    # log to catch the outer consumer up; the control-message moves must be
+    # re-emitted so the outer shape isn't left missing the issues of projects that
+    # moved via a control message.
+    shapes = [
+      %{
+        name: "issues_of_active_regions",
+        table: "issues",
+        where:
+          "project_id IN (SELECT id FROM projects WHERE region_id IN " <>
+            "(SELECT id FROM regions WHERE active = true))",
+        columns: ["id", "project_id", "title"],
+        pk: ["id"],
+        optimized: true
+      }
+    ]
+
+    # batch_1: 200 toggles of r3's `active` flag — each toggle moves p5/p6 in and
+    # out of the inner subquery shape, writing control messages to its log. r3 ends
+    # active, so pre-restart the shape matches the oracle.
+    toggles =
+      Enum.flat_map(1..100, fn _ ->
+        [
+          [%{name: "deactivate_r3", sql: "UPDATE regions SET active = false WHERE id = 'r3'"}],
+          [%{name: "reactivate_r3", sql: "UPDATE regions SET active = true WHERE id = 'r3'"}]
+        ]
+      end)
+
+    # batch_2 (after the restart): a single region deactivate that moves a project
+    # out of the inner subquery via a control message.
+    batch_2 = [
+      [%{name: "deactivate_r1", sql: "UPDATE regions SET active = false WHERE id = 'r1'"}]
+    ]
+
+    batches = [toggles, batch_2]
+
+    OracleHarness.test_against_oracle(ctx, shapes, batches, restart_server_every: 1)
+  end
 end

From ce1eef2aecf2270f5e682b27c0ac4be3ef31f228 Mon Sep 17 00:00:00 2001
From: rob <rob.acourt@gmail.com>
Date: Tue, 14 Jul 2026 16:15:09 +0100
Subject: [PATCH 5/8] Replay control-message moves with authoritative storage
 offsets
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

On restart a dependency materializer replays its log to catch up a behind
subscriber, re-emitting the moves it missed. Data-change log items carry their
offset in their headers, but control messages (subquery move-in/move-out events)
do not — so `decode_offset/1` collapsed them to `before_all()`, placing them at
or before every `from_lsn` and never re-emitting them. This only surfaces when a
materializer's own log contains control messages, i.e. a nested optimized
subquery whose dependency is itself an optimized subquery.

Add `Storage.get_log_stream_with_offsets/3`, which yields `{LogOffset, json}`
pairs from the authoritative storage framing (snapshot lines, which precede all
real offsets, are tagged `before_all()`). Materializer replay now reads with
offsets and positions every item — control messages included — by its real
transaction offset, so control-message moves after `from_lsn` are replayed.

Covered by a deterministic materializer replay unit test; the live `new_changes`
path is unchanged (it already tags batches by `range_end`).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 .../electric/shape_cache/in_memory_storage.ex |  44 ++++++
 .../electric/shape_cache/pure_file_storage.ex | 144 +++++++++++++++++-
 .../shape_cache/pure_file_storage/log_file.ex | 138 +++++++++++++++++
 .../lib/electric/shape_cache/storage.ex       |  29 ++++
 .../electric/shapes/consumer/materializer.ex  |  53 ++++---
 .../shapes/consumer/materializer_test.exs     | 107 ++++++++++++-
 .../sync-service/test/support/test_storage.ex |   6 +
 7 files changed, 493 insertions(+), 28 deletions(-)

diff --git a/packages/sync-service/lib/electric/shape_cache/in_memory_storage.ex b/packages/sync-service/lib/electric/shape_cache/in_memory_storage.ex
index 4d4f072d9d..04e371b050 100644
--- a/packages/sync-service/lib/electric/shape_cache/in_memory_storage.ex
+++ b/packages/sync-service/lib/electric/shape_cache/in_memory_storage.ex
@@ -203,6 +203,50 @@ defmodule Electric.ShapeCache.InMemoryStorage do
     get_offset_indexed_stream(offset, max_offset, opts.log_table)
   end
 
+  @impl Electric.ShapeCache.Storage
+  def get_log_stream_with_offsets(offset, max_offset, %MS{} = opts)
+      when is_log_offset_lt(offset, @snapshot_boundary_offset) do
+    case :ets.lookup_element(opts.snapshot_table, snapshot_end(), 2, nil) do
+      nil ->
+        stream_from_snapshot_with_offsets(offset, max_offset, opts)
+
+      max when is_log_offset_lt(offset, max) ->
+        stream_from_snapshot_with_offsets(offset, max_offset, opts)
+
+      _ ->
+        get_offset_indexed_stream_with_offsets(offset, max_offset, opts.log_table)
+    end
+  end
+
+  def get_log_stream_with_offsets(offset, max_offset, %MS{} = opts) do
+    get_offset_indexed_stream_with_offsets(offset, max_offset, opts.log_table)
+  end
+
+  # Snapshot lines precede all real offsets, so tag them with `before_all/0`.
+  defp stream_from_snapshot_with_offsets(offset, max_offset, %MS{} = opts) do
+    offset
+    |> stream_from_snapshot(max_offset, opts)
+    |> Stream.map(&{LogOffset.before_all(), &1})
+  end
+
+  defp get_offset_indexed_stream_with_offsets(offset, max_offset, offset_indexed_table) do
+    offset = storage_offset(offset)
+    max_offset = storage_offset(max_offset)
+
+    Stream.unfold(offset, fn offset ->
+      case :ets.next_lookup(offset_indexed_table, {:offset, offset}) do
+        :"$end_of_table" ->
+          nil
+
+        {{:offset, position}, _} when position > max_offset ->
+          nil
+
+        {{:offset, position}, [{_, item}]} ->
+          {{LogOffset.new(position), item}, position}
+      end
+    end)
+  end
+
   defp stream_from_snapshot(offset, max_offset, %MS{} = opts) do
     ConcurrentStream.stream_to_end(
       excluded_start_key: snapshot_chunk_end(storage_offset(offset)),
diff --git a/packages/sync-service/lib/electric/shape_cache/pure_file_storage.ex b/packages/sync-service/lib/electric/shape_cache/pure_file_storage.ex
index 88f05ee361..a2e2db9eaf 100644
--- a/packages/sync-service/lib/electric/shape_cache/pure_file_storage.ex
+++ b/packages/sync-service/lib/electric/shape_cache/pure_file_storage.ex
@@ -1035,7 +1035,59 @@ defmodule Electric.ShapeCache.PureFileStorage do
     end
   end
 
-  defp stream_main_log(min_offset, max_offset, %__MODULE__{} = opts) do
+  def get_log_stream_with_offsets(%LogOffset{} = min_offset, %LogOffset{} = max_offset, opts)
+      when is_last_virtual_offset(min_offset) or is_real_offset(min_offset) do
+    stream_main_log(min_offset, max_offset, opts, with_offsets?: true)
+  end
+
+  def get_log_stream_with_offsets(
+        %LogOffset{op_offset: op_offset} = min_offset,
+        %LogOffset{} = max_offset,
+        %__MODULE__{} = opts
+      ) do
+    metadata = read_multiple_cached_metadata(opts, [:snapshot_started?, :last_snapshot_chunk])
+
+    snapshot_started? = Keyword.get(metadata, :snapshot_started?) || false
+    last_snapshot_chunk = Keyword.get(metadata, :last_snapshot_chunk)
+
+    if not snapshot_started? and not shape_gone?(opts) do
+      raise(Storage.Error, message: "Snapshot not started")
+    end
+
+    case {last_snapshot_chunk, min_offset} do
+      {_, x} when is_min_offset(x) ->
+        snapshot_lines_with_offsets(Snapshot.stream_chunk_lines(opts, 0))
+
+      {%LogOffset{} = latest, min_offset} when is_log_offset_lt(min_offset, latest) ->
+        snapshot_lines_with_offsets(Snapshot.stream_chunk_lines(opts, op_offset + 1))
+
+      {nil, _offset} ->
+        snapshot_lines_with_offsets(wait_for_chunk_file_or_snapshot_end(opts, op_offset + 1))
+
+      {%LogOffset{}, offset} ->
+        stream_main_log(offset, max_offset, opts, with_offsets?: true)
+    end
+  end
+
+  # Snapshot lines precede all real offsets, so tag them with `before_all/0`.
+  defp snapshot_lines_with_offsets(stream),
+    do: Stream.map(stream, &{LogOffset.before_all(), &1})
+
+  defp stream_main_log(min_offset, max_offset, opts, gen_opts \\ [])
+
+  defp stream_main_log(min_offset, max_offset, %__MODULE__{} = opts, gen_opts) do
+    with_offsets? = Keyword.get(gen_opts, :with_offsets?, false)
+
+    ets_reader =
+      if with_offsets?,
+        do: &read_range_from_ets_cache_with_offsets/3,
+        else: &read_range_from_ets_cache/3
+
+    disk_streamer =
+      if with_offsets?,
+        do: &stream_from_disk_with_offsets/4,
+        else: &stream_from_disk/4
+
     storage_meta(
       ets_table: ets,
       last_persisted_offset: last_persisted,
@@ -1067,34 +1119,34 @@ defmodule Electric.ShapeCache.PureFileStorage do
 
       is_log_offset_lte(last_persisted, min_offset) ->
         # Pure ETS read case
-        case read_range_from_ets_cache(ets, min_offset, upper_read_bound) do
+        case ets_reader.(ets, min_offset, upper_read_bound) do
           {_data, last_offset}
           when is_nil(last_offset) or last_offset < upper_read_bound_tuple ->
             # Empty or partial read - ETS was cleared by a concurrent flush.
             # Data is now on disk (flush writes to disk before clearing ETS),
             # so read directly from there using existing boundary info.
-            stream_from_disk(opts, min_offset, upper_read_bound, boundary_info)
+            disk_streamer.(opts, min_offset, upper_read_bound, boundary_info)
 
           {data, _last_offset} ->
             data
         end
 
       is_log_offset_lte(upper_read_bound, last_persisted) ->
-        stream_from_disk(opts, min_offset, upper_read_bound, boundary_info)
+        disk_streamer.(opts, min_offset, upper_read_bound, boundary_info)
 
       true ->
         # Mixed disk + ETS case
         # Because ETS may be cleared by a flush in a parallel process, we're reading it out into memory.
         # It's expected to be fairly small in the worst case, up 64KB
-        case read_range_from_ets_cache(ets, last_persisted, upper_read_bound) do
+        case ets_reader.(ets, last_persisted, upper_read_bound) do
           {_upper_range, last_offset}
           when is_nil(last_offset) or last_offset < upper_read_bound_tuple ->
             # Empty or partial read - ETS was cleared by a concurrent flush.
             # Data is now on disk, so read the full range from there.
-            stream_from_disk(opts, min_offset, upper_read_bound, boundary_info)
+            disk_streamer.(opts, min_offset, upper_read_bound, boundary_info)
 
           {upper_range, _last_offset} ->
-            stream_from_disk(opts, min_offset, last_persisted, boundary_info)
+            disk_streamer.(opts, min_offset, last_persisted, boundary_info)
             |> Stream.concat(upper_range)
         end
     end
@@ -1173,6 +1225,43 @@ defmodule Electric.ShapeCache.PureFileStorage do
     end
   end
 
+  # Offset-preserving counterpart to `read_range_from_ets_cache/3`, yielding
+  # `{LogOffset.t(), item}` pairs. `last_offset` stays a tuple for comparison with
+  # `upper_read_bound_tuple` in `stream_main_log/4`.
+  defp read_range_from_ets_cache_with_offsets(nil, _min, _max), do: {[], nil}
+
+  defp read_range_from_ets_cache_with_offsets(ets, %LogOffset{} = min, %LogOffset{} = max) do
+    read_range_from_ets_cache_with_offsets(
+      ets,
+      LogOffset.to_tuple(min),
+      LogOffset.to_tuple(max),
+      [],
+      nil
+    )
+  end
+
+  defp read_range_from_ets_cache_with_offsets(ets, min, {max_tx, max_op} = max, acc, last_offset) do
+    case safe_next_lookup(ets, min) do
+      :ets_dead ->
+        {Enum.reverse(acc), last_offset}
+
+      :"$end_of_table" ->
+        {Enum.reverse(acc), last_offset}
+
+      {{min_tx, min_op}, _} when min_tx > max_tx or (min_tx == max_tx and min_op > max_op) ->
+        {Enum.reverse(acc), last_offset}
+
+      {new_min, [{_, item}]} ->
+        read_range_from_ets_cache_with_offsets(
+          ets,
+          new_min,
+          max,
+          [{LogOffset.new(new_min), item} | acc],
+          new_min
+        )
+    end
+  end
+
   # The owning Consumer's buffer ETS table may be destroyed during terminate.
   # Falling back is safe because terminate flushes to disk before deleting.
   defp safe_next_lookup(ets, min) do
@@ -1218,6 +1307,47 @@ defmodule Electric.ShapeCache.PureFileStorage do
     end
   end
 
+  # Offset-preserving counterpart to `stream_from_disk/4`, yielding
+  # `{LogOffset.t(), json}` pairs.
+  defp stream_from_disk_with_offsets(%__MODULE__{}, min_offset, max_offset, _)
+       when is_log_offset_lte(max_offset, min_offset),
+       do: []
+
+  defp stream_from_disk_with_offsets(
+         %__MODULE__{} = opts,
+         min_offset,
+         max_offset,
+         boundary_info
+       ) do
+    suffix = get_suffix(min_offset, boundary_info)
+
+    case fetch_chunk(min_offset, opts, boundary_info) do
+      {:ok, chunk_end_offset, {start_pos, end_pos}} when not is_nil(end_pos) ->
+        LogFile.stream_entries_with_offsets(
+          opts,
+          json_file(opts, suffix),
+          start_pos,
+          end_pos,
+          min_offset
+        )
+        |> Stream.concat(
+          stream_from_disk_with_offsets(opts, chunk_end_offset, max_offset, boundary_info)
+        )
+
+      {:ok, nil, {start_pos, nil}} ->
+        LogFile.stream_entries_until_offset_with_offsets(
+          opts,
+          json_file(opts, suffix),
+          start_pos,
+          min_offset,
+          max_offset
+        )
+
+      :error ->
+        []
+    end
+  end
+
   defp get_suffix(min_offset, {_, {compaction_boundary, compacted_name}, _})
        when is_log_offset_lt(min_offset, compaction_boundary),
        do: compacted_name
diff --git a/packages/sync-service/lib/electric/shape_cache/pure_file_storage/log_file.ex b/packages/sync-service/lib/electric/shape_cache/pure_file_storage/log_file.ex
index 65533706fb..bc7bb9a316 100644
--- a/packages/sync-service/lib/electric/shape_cache/pure_file_storage/log_file.ex
+++ b/packages/sync-service/lib/electric/shape_cache/pure_file_storage/log_file.ex
@@ -282,6 +282,144 @@ defmodule Electric.ShapeCache.PureFileStorage.LogFile do
     )
   end
 
+  @doc """
+  Like `stream_jsons/5` but yields `{LogOffset.t(), json}` pairs, preserving each
+  entry's on-disk offset (used to position control messages, which carry no
+  offset in their JSON headers).
+  """
+  def stream_entries_with_offsets(
+        %PFS{} = opts,
+        log_file_path,
+        start_position,
+        end_position,
+        exclusive_min_offset
+      ) do
+    case safely_open_file!(opts, log_file_path, [:read, :raw]) do
+      {:halt, :data_removed} ->
+        []
+
+      {:ok, file} ->
+        try do
+          with {:ok, data} <- :file.pread(file, start_position, end_position - start_position) do
+            {entries, _} = extract_entries_from_binary(data, exclusive_min_offset, nil)
+            entries
+          else
+            :eof ->
+              raise "unexpected end of file"
+
+            {:error, reason} ->
+              raise File.Error,
+                path: log_file_path,
+                reason: reason,
+                action: "pread(#{start_position}, #{end_position - start_position})"
+          end
+        after
+          File.close(file)
+        end
+    end
+  end
+
+  @doc "Like `stream_jsons_until_offset/5` but yields `{LogOffset.t(), json}` pairs."
+  def stream_entries_until_offset_with_offsets(
+        %PFS{} = opts,
+        log_file_path,
+        start_position,
+        exclusive_min_offset,
+        inclusive_max_offset
+      ) do
+    Stream.resource(
+      fn ->
+        case safely_open_file!(opts, log_file_path, [:read, :raw]) do
+          {:ok, file} ->
+            {:ok, ^start_position} = :file.position(file, start_position)
+            {file, ""}
+
+          {:halt, :data_removed} ->
+            :halt
+        end
+      end,
+      fn
+        :halt ->
+          {:halt, []}
+
+        {file, binary_rest} ->
+          case :file.read(file, 4096) do
+            {:ok, data} ->
+              {entries, rest} =
+                extract_entries_from_binary(
+                  binary_rest <> data,
+                  exclusive_min_offset,
+                  inclusive_max_offset
+                )
+
+              {entries, {file, rest}}
+
+            :eof ->
+              {:halt, {file, binary_rest}}
+          end
+      end,
+      fn
+        [] -> :ok
+        {file, _} -> File.close(file)
+      end
+    )
+  end
+
+  # Offset-preserving counterpart to `extract_jsons_from_binary/4`. Kept separate
+  # so the json-only hot path avoids the per-entry `LogOffset` allocation.
+  @spec extract_entries_from_binary(binary(), LogOffset.t(), LogOffset.t() | nil) ::
+          {[{LogOffset.t(), String.t()}], binary()}
+  defp extract_entries_from_binary(binary, exclusive_min_offset, inclusive_max_offset, acc \\ [])
+  defp extract_entries_from_binary(<<>>, _, _, acc), do: {Enum.reverse(acc), ""}
+
+  defp extract_entries_from_binary(
+         <<tx_offset1::64, op_offset1::64, key_size::32, _::binary-size(key_size), _::8, _flag::8,
+           json_size::64, _::binary-size(json_size), rest::binary>>,
+         %LogOffset{tx_offset: tx_offset2, op_offset: op_offset2} = log_offset,
+         inclusive_max_offset,
+         acc
+       )
+       when tx_offset1 < tx_offset2 or (tx_offset1 == tx_offset2 and op_offset1 <= op_offset2),
+       do: extract_entries_from_binary(rest, log_offset, inclusive_max_offset, acc)
+
+  defp extract_entries_from_binary(
+         <<tx_offset1::64, op_offset1::64, key_size::32, _::binary-size(key_size), _::8, _flag::8,
+           json_size::64, json::binary-size(json_size), _::binary>>,
+         log_offset,
+         %LogOffset{tx_offset: tx_offset2, op_offset: op_offset2} = inclusive_max_offset,
+         acc
+       )
+       when tx_offset1 == tx_offset2 and op_offset1 == op_offset2,
+       do:
+         extract_entries_from_binary("", log_offset, inclusive_max_offset, [
+           {LogOffset.new(tx_offset1, op_offset1), json} | acc
+         ])
+
+  defp extract_entries_from_binary(
+         <<tx_offset1::64, op_offset1::64, key_size::32, _::binary-size(key_size), _::8, _flag::8,
+           json_size::64, _json::binary-size(json_size), _::binary>>,
+         log_offset,
+         %LogOffset{tx_offset: tx_offset2, op_offset: op_offset2} = inclusive_max_offset,
+         acc
+       )
+       when tx_offset1 > tx_offset2 or (tx_offset1 == tx_offset2 and op_offset1 > op_offset2),
+       do: extract_entries_from_binary("", log_offset, inclusive_max_offset, acc)
+
+  defp extract_entries_from_binary(
+         <<tx_offset1::64, op_offset1::64, key_size::32, _::binary-size(key_size), _::8, _flag::8,
+           json_size::64, json::binary-size(json_size), rest::binary>>,
+         log_offset,
+         inclusive_max_offset,
+         acc
+       ),
+       do:
+         extract_entries_from_binary(rest, log_offset, inclusive_max_offset, [
+           {LogOffset.new(tx_offset1, op_offset1), json} | acc
+         ])
+
+  defp extract_entries_from_binary(rest, _, _, acc),
+    do: {Enum.reverse(acc), rest}
+
   @spec extract_jsons_from_binary(binary(), LogOffset.t(), LogOffset.t() | nil) ::
           Enumerable.t(String.t())
   defp extract_jsons_from_binary(binary, exclusive_min_offset, inclusive_max_offset, acc \\ [])
diff --git a/packages/sync-service/lib/electric/shape_cache/storage.ex b/packages/sync-service/lib/electric/shape_cache/storage.ex
index 1998612294..d62c8605d2 100644
--- a/packages/sync-service/lib/electric/shape_cache/storage.ex
+++ b/packages/sync-service/lib/electric/shape_cache/storage.ex
@@ -188,6 +188,22 @@ defmodule Electric.ShapeCache.Storage do
   @callback get_log_stream(offset :: LogOffset.t(), max_offset :: LogOffset.t(), shape_opts()) ::
               log()
 
+  @doc """
+  Like `get_log_stream/3` but yields `{LogOffset.t(), log_item_json}` pairs so the
+  caller gets each item's authoritative storage offset.
+
+  Data-change log items carry their offset in their headers, but control messages
+  (subquery move-in/move-out events) do not, so they cannot be positioned by
+  re-parsing the JSON alone. Materializer replay uses this to place control
+  messages in the correct transaction. Snapshot lines (which precede all real
+  offsets) are yielded with `LogOffset.before_all/0`.
+  """
+  @callback get_log_stream_with_offsets(
+              offset :: LogOffset.t(),
+              max_offset :: LogOffset.t(),
+              shape_opts()
+            ) :: Enumerable.t({LogOffset.t(), binary()})
+
   @doc """
   Get the last exclusive offset of the chunk starting from the given offset.
 
@@ -436,6 +452,19 @@ defmodule Electric.ShapeCache.Storage do
     []
   end
 
+  @impl __MODULE__
+  def get_log_stream_with_offsets(offset, max_offset \\ @last_log_offset, storage)
+
+  def get_log_stream_with_offsets(offset, max_offset, {mod, shape_opts})
+      when max_offset == @last_log_offset or not is_log_offset_lt(max_offset, offset) do
+    mod.get_log_stream_with_offsets(offset, max_offset, shape_opts)
+  end
+
+  def get_log_stream_with_offsets(offset, max_offset, _storage)
+      when is_log_offset_lt(max_offset, offset) do
+    []
+  end
+
   @impl __MODULE__
   def get_chunk_end_log_offset(offset, {mod, shape_opts}) do
     mod.get_chunk_end_log_offset(offset, shape_opts)
diff --git a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
index 1ca8bd1174..a7acbb9a7a 100644
--- a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
@@ -233,7 +233,12 @@ defmodule Electric.Shapes.Consumer.Materializer do
   changes after this offset will be delivered via new_changes messages
   from the Consumer.
   """
-  def read_history_up_to_subscribed(state, storage, apply_fun \\ &default_history_apply/2) do
+  def read_history_up_to_subscribed(
+        state,
+        storage,
+        apply_fun \\ &default_history_apply/2,
+        with_offsets? \\ false
+      ) do
     cond do
       is_nil(state.subscribed_offset) ->
         state
@@ -242,7 +247,13 @@ defmodule Electric.Shapes.Consumer.Materializer do
         state
 
       true ->
-        stream = Storage.get_log_stream(state.offset, state.subscribed_offset, storage)
+        stream =
+          if with_offsets? do
+            Storage.get_log_stream_with_offsets(state.offset, state.subscribed_offset, storage)
+          else
+            Storage.get_log_stream(state.offset, state.subscribed_offset, storage)
+          end
+
         state = apply_fun.(stream, state)
 
         # If the read just covered the main log (because either the
@@ -284,7 +295,12 @@ defmodule Electric.Shapes.Consumer.Materializer do
               %{state | offset: state.subscribed_offset}
 
             true ->
-              read_history_up_to_subscribed(%{state | offset: next_offset}, storage, apply_fun)
+              read_history_up_to_subscribed(
+                %{state | offset: next_offset},
+                storage,
+                apply_fun,
+                with_offsets?
+              )
           end
         end
     end
@@ -332,9 +348,12 @@ defmodule Electric.Shapes.Consumer.Materializer do
     }
 
     replay =
-      read_history_up_to_subscribed(replay0, storage, fn stream, acc ->
-        apply_replay_stream(stream, acc)
-      end)
+      read_history_up_to_subscribed(
+        replay0,
+        storage,
+        fn stream, acc -> apply_replay_stream(stream, acc) end,
+        true
+      )
 
     case replay.pending_events.seed do
       nil -> link_values_from_counts(replay.value_counts)
@@ -393,26 +412,22 @@ defmodule Electric.Shapes.Consumer.Materializer do
     end)
   end
 
-  # Decode a raw JSON log stream into `{log_offset, txids, change}` tuples,
-  # preserving the per-item source offset (reconstructed from the item headers)
-  # so replay can delimit transactions and tag emitted moves.
+  # Decode a `{log_offset, json}` log stream (from `get_log_stream_with_offsets/3`)
+  # into `{log_offset, txids, change}` tuples so replay can delimit transactions
+  # and tag emitted moves. Unlike the JSON headers, the storage offset is present
+  # for *every* item — including control messages, which carry no `lsn`/`op_position`
+  # in their headers and would otherwise collapse to `before_all/0`.
   defp decode_json_stream_with_offsets(stream) do
     stream
-    |> Stream.map(&Jason.decode!/1)
-    |> Stream.filter(fn decoded ->
+    |> Stream.map(fn {offset, json} -> {offset, Jason.decode!(json)} end)
+    |> Stream.filter(fn {_offset, decoded} ->
       Map.has_key?(decoded, "key") || Map.has_key?(decoded["headers"], "event")
     end)
-    |> Stream.map(fn decoded ->
-      headers = decoded["headers"]
-      {decode_offset(headers), decode_txids(headers), decode_change(decoded)}
+    |> Stream.map(fn {offset, decoded} ->
+      {offset, decode_txids(decoded["headers"]), decode_change(decoded)}
     end)
   end
 
-  defp decode_offset(%{"lsn" => lsn, "op_position" => op}) when is_binary(lsn),
-    do: LogOffset.new(String.to_integer(lsn), op)
-
-  defp decode_offset(_headers), do: LogOffset.before_all()
-
   defp decode_txids(%{"txids" => txids}) when is_list(txids), do: txids
   defp decode_txids(_headers), do: []
 
diff --git a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs
index bb3aeaf901..638694e3fc 100644
--- a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs
+++ b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs
@@ -1309,8 +1309,7 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
   end
 
   # Build a single main-log insert log item at `offset` introducing `value`,
-  # encoded the same way the source consumer would write it (headers carry the
-  # `lsn`/`op_position` used to reconstruct the offset during replay).
+  # encoded the same way the source consumer would write it.
   defp main_log_insert(offset, id, value) do
     change =
       %Changes.NewRecord{
@@ -1329,6 +1328,26 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
     end)
   end
 
+  # Like `main_log_insert/3` but attaches move tags so the row is indexed under
+  # those tags (and can later be moved out by a control message targeting them).
+  defp tagged_main_log_insert(offset, id, value, tags) do
+    change =
+      %Changes.NewRecord{
+        relation: {"public", "test_table"},
+        key: ~s|"public"."test_table"/"#{id}"|,
+        record: %{"id" => id, "value" => value},
+        log_offset: offset,
+        move_tags: tags
+      }
+      |> Changes.fill_key(["id"])
+
+    change
+    |> then(&LogItems.from_change(&1, 1, ["id"], :default))
+    |> Enum.map(fn {item_offset, item} ->
+      {item_offset, change.key, :insert, Jason.encode!(item)}
+    end)
+  end
+
   defp prep_changes(changes, opts \\ []) do
     pk_cols = Keyword.get(opts, :pk_cols, ["id"])
     relation = Keyword.get(opts, :relation, {"public", "test_table"})
@@ -1646,6 +1665,90 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
     end
   end
 
+  describe "move replay on subscribe with control messages" do
+    # A dependency shape that is itself an optimized subquery has move-in/move-out
+    # *control messages* in its own log (rows moving in/out as its nested
+    # dependency toggles). Control messages carry no `lsn`/`op_position` in their
+    # headers, so their position comes from the storage offset rather than the
+    # JSON. On restart the materializer replays that log to catch up a behind
+    # subscriber, and a control-message move after `from_lsn` must be re-emitted
+    # just like a data-change move.
+    setup ctx do
+      shape_handle = "control-replay-#{System.unique_integer([:positive])}"
+
+      storage = Storage.for_shape(shape_handle, ctx.storage)
+      Storage.start_link(storage)
+      writer = Storage.init_writer!(storage, @shape)
+      Storage.mark_snapshot_as_started(storage)
+      Storage.make_new_snapshot!([], storage)
+
+      # tx 100: value 10 enters the shape, tagged "tag1".
+      writer =
+        Storage.append_to_log!(
+          tagged_main_log_insert(LogOffset.new(100, 0), "1", "10", ["tag1"]),
+          writer
+        )
+
+      # tx 200: a second "tag1" row in a *later* transaction. Its only role is to
+      # open transaction 200 (so the control message below joins that transaction
+      # rather than tx 100); it shares value 10, so it emits no move of its own.
+      writer =
+        Storage.append_to_log!(
+          tagged_main_log_insert(LogOffset.new(200, 0), "2", "10", ["tag1"]),
+          writer
+        )
+
+      # (200,1): a move-out control message removing "tag1" — value 10 leaves the
+      # shape. Its source offset is only recoverable from storage, not the JSON.
+      {_range, writer} =
+        Storage.append_control_message!(
+          Jason.encode!(%{
+            headers: %{event: "move-out", patterns: [%{pos: 0, value: "tag1"}]}
+          }),
+          writer
+        )
+
+      Storage.hibernate(writer)
+
+      ConsumerRegistry.register_consumer(self(), shape_handle, ctx.stack_id)
+
+      {:ok, _pid} =
+        Materializer.start_link(%{
+          stack_id: ctx.stack_id,
+          shape_handle: shape_handle,
+          storage: ctx.storage,
+          columns: ["value"],
+          materialized_type: {:array, :int8}
+        })
+
+      respond_to_call(:await_snapshot_start, :started)
+      respond_to_call(:subscribe_materializer, {:ok, LogOffset.new(200, 1)})
+
+      mat_ctx = %{stack_id: ctx.stack_id, shape_handle: shape_handle}
+      assert Materializer.wait_until_ready(mat_ctx) == :ok
+      # Startup applied the whole history: value 10 entered, then left via the
+      # move-out control message, so nothing remains materialized.
+      assert Materializer.get_link_values(mat_ctx) == MapSet.new()
+
+      Map.put(ctx, :mat_ctx, mat_ctx)
+    end
+
+    test "replays a move-out carried by a control message after from_lsn",
+         %{mat_ctx: mat_ctx} do
+      # Behind at (150,0): value 10 was still present then, and the move-out at
+      # (200,1) is after from_lsn, so replay must re-emit it.
+      assert {:ok, seed, applied_offset} =
+               Materializer.subscribe(mat_ctx, LogOffset.new(150, 0))
+
+      # Seed view is the link values as of (150,0): value 10 still present.
+      assert seed == MapSet.new([10])
+      assert applied_offset == LogOffset.new(200, 1)
+
+      assert_receive {:materializer_changes, _handle,
+                      %{move_out: [{10, "10"}], lsn: %LogOffset{tx_offset: 200}}}
+    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
diff --git a/packages/sync-service/test/support/test_storage.ex b/packages/sync-service/test/support/test_storage.ex
index bdf493d49a..358cf13641 100644
--- a/packages/sync-service/test/support/test_storage.ex
+++ b/packages/sync-service/test/support/test_storage.ex
@@ -125,6 +125,12 @@ defmodule Support.TestStorage do
     Storage.get_log_stream(offset, max_offset, storage)
   end
 
+  @impl Electric.ShapeCache.Storage
+  def get_log_stream_with_offsets(offset, max_offset, {parent, shape_handle, _, storage}) do
+    send(parent, {__MODULE__, :get_log_stream_with_offsets, shape_handle, offset, max_offset})
+    Storage.get_log_stream_with_offsets(offset, max_offset, storage)
+  end
+
   @impl Electric.ShapeCache.Storage
   def get_chunk_end_log_offset(offset, {parent, shape_handle, _, storage}) do
     send(parent, {__MODULE__, :get_chunk_end_log_offset, shape_handle, offset})

From 062fce0eb3f6b6d08c65117cd3360894afa586f5 Mon Sep 17 00:00:00 2001
From: rob <rob.acourt@gmail.com>
Date: Tue, 14 Jul 2026 18:32:28 +0100
Subject: [PATCH 6/8] Chunk move replay by full offset, not by transaction
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A dependency materializer's replay grouped its log into batches by `tx_offset`
(a transaction/LSN boundary) before capturing the seed and emitting the moves
after `from_lsn`. That is wrong for nested optimized subqueries: dependency-driven
moves (subquery move-in/out) have no WAL LSN of their own, so they are appended as
successive `op_offset`s within a single `tx_offset`. Grouping by `tx_offset`
collapsed an entire nested subquery's move history into one batch, so the seed was
always the pre-log snapshot state and `from_lsn` was ignored — on restart the outer
consumer was seeded as though rows were still present and the corrective move was
redundancy-eliminated, leaving the outer shape diverged from Postgres.

Chunk by the full `(tx_offset, op_offset)` offset instead — the same space
`from_lsn`/`move_positions` live in — restoring the per-move granularity the seed
cut and emit boundary need. Snapshot lines all share `before_all/0`, so they still
group into a single pre-real-offsets batch. Only the replay path changes; the live
`new_changes` path (and its same-batch move cancellation) is untouched.

This makes the previously-failing nested optimized subquery restore test pass
deterministically (30/30, was ~4% divergence), with single-level restore and the
full materializer/consumer suites still green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 .../electric/shapes/consumer/materializer.ex  | 23 +++++++++++--------
 1 file changed, 14 insertions(+), 9 deletions(-)

diff --git a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
index a7acbb9a7a..c66d95e751 100644
--- a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
@@ -376,7 +376,7 @@ defmodule Electric.Shapes.Consumer.Materializer do
 
     stream
     |> decode_json_stream_with_offsets()
-    |> chunk_by_txn()
+    |> chunk_by_offset()
     |> Enum.reduce(acc, fn {txn_offset, changes, txids}, acc ->
       emit? = is_log_offset_lt(from_lsn, txn_offset)
 
@@ -414,9 +414,8 @@ defmodule Electric.Shapes.Consumer.Materializer do
 
   # Decode a `{log_offset, json}` log stream (from `get_log_stream_with_offsets/3`)
   # into `{log_offset, txids, change}` tuples so replay can delimit transactions
-  # and tag emitted moves. Unlike the JSON headers, the storage offset is present
-  # for *every* item — including control messages, which carry no `lsn`/`op_position`
-  # in their headers and would otherwise collapse to `before_all/0`.
+  # and tag emitted moves. The storage offset is used rather than the JSON
+  # headers because control messages carry no `lsn`/`op_position` there.
   defp decode_json_stream_with_offsets(stream) do
     stream
     |> Stream.map(fn {offset, json} -> {offset, Jason.decode!(json)} end)
@@ -474,12 +473,18 @@ defmodule Electric.Shapes.Consumer.Materializer do
     %{headers: %{event: event, patterns: patterns, txids: Map.get(headers, "txids", [])}}
   end
 
-  # Group consecutive decoded items into per-transaction batches. Items sharing a
-  # `tx_offset` belong to the same source transaction; each batch is tagged with
-  # the batch's last (largest) offset as its source LSN.
-  defp chunk_by_txn(items) do
+  # Group consecutive decoded items into replay batches keyed by their full
+  # `(tx_offset, op_offset)` offset — the same space `from_lsn`/`move_positions`
+  # live in, so the seed cut and emit boundary can fall between any two items.
+  # This matters because dependency-driven moves (subquery move-in/out) have no
+  # WAL LSN of their own and are appended as successive `op_offset`s within one
+  # `tx_offset`, so grouping by `tx_offset` alone cannot separate them. Each
+  # batch is tagged with its last (largest) offset as its source LSN. Snapshot
+  # lines all share `before_all/0`, so they group into a single pre-real-offsets
+  # batch.
+  defp chunk_by_offset(items) do
     items
-    |> Enum.chunk_by(fn {offset, _txids, _change} -> offset.tx_offset end)
+    |> Enum.chunk_by(fn {offset, _txids, _change} -> {offset.tx_offset, offset.op_offset} end)
     |> Enum.map(fn batch ->
       {offset, _, _} = List.last(batch)
       changes = Enum.map(batch, fn {_o, _t, change} -> change end)

From 0f2200c861ad3453453d25ba76e06ab44bbf60c1 Mon Sep 17 00:00:00 2001
From: rob <rob.acourt@gmail.com>
Date: Wed, 15 Jul 2026 09:35:39 +0100
Subject: [PATCH 7/8] Test replay of a value toggled multiple times within one
 transaction
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Replay keys batches by full offset, so a value that crosses the 0↔1 boundary
several times inside a single source transaction is re-emitted as separate
sequential payloads (one per op, in offset order) rather than a single netted
one. That is safe — the case `cancel_matching_move_events/1` guards against is a
single payload carrying both a move_in and a move_out for the same value; ordered
single-value payloads are just the normal per-transaction flow. Lock that in.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 .../shapes/consumer/materializer_test.exs     | 95 +++++++++++++++++++
 1 file changed, 95 insertions(+)

diff --git a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs
index 638694e3fc..59b98e81a2 100644
--- a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs
+++ b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs
@@ -1328,6 +1328,26 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
     end)
   end
 
+  # Build a single main-log delete log item at `offset` for row `id` (value only
+  # needs to match the corresponding insert so the materializer decrements it).
+  defp main_log_delete(offset, id, value) do
+    change =
+      %Changes.DeletedRecord{
+        relation: {"public", "test_table"},
+        key: ~s|"public"."test_table"/"#{id}"|,
+        old_record: %{"id" => id, "value" => value},
+        log_offset: offset,
+        move_tags: []
+      }
+      |> Changes.fill_key(["id"])
+
+    change
+    |> then(&LogItems.from_change(&1, 1, ["id"], :default))
+    |> Enum.map(fn {item_offset, item} ->
+      {item_offset, change.key, :delete, Jason.encode!(item)}
+    end)
+  end
+
   # Like `main_log_insert/3` but attaches move tags so the row is indexed under
   # those tags (and can later be moved out by a control message targeting them).
   defp tagged_main_log_insert(offset, id, value, tags) do
@@ -1749,6 +1769,81 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
     end
   end
 
+  describe "move replay of a value toggled multiple times within one transaction" do
+    # A value can cross the 0↔1 boundary several times inside a single source
+    # transaction (op_offsets sharing one tx_offset). Replay batches each op
+    # separately, so these are re-emitted as separate sequential payloads in
+    # offset order — never a single payload carrying both a move_in and a
+    # move_out for the same value (the case `cancel_matching_move_events/1`
+    # guards against, where the consumer's move-in query would race its own
+    # moved-out tag). Sequential single-value payloads are just the normal
+    # per-transaction flow applied in order.
+    setup ctx do
+      shape_handle = "toggle-replay-#{System.unique_integer([:positive])}"
+
+      storage = Storage.for_shape(shape_handle, ctx.storage)
+      Storage.start_link(storage)
+      writer = Storage.init_writer!(storage, @shape)
+      Storage.mark_snapshot_as_started(storage)
+      Storage.make_new_snapshot!([], storage)
+
+      # One transaction (tx 100) toggles value 10: in (row "a"), out (row "a"),
+      # in (row "b"). It ends present on "b", so startup materializes {10}.
+      writer = Storage.append_to_log!(main_log_insert(LogOffset.new(100, 0), "a", "10"), writer)
+      writer = Storage.append_to_log!(main_log_delete(LogOffset.new(100, 1), "a", "10"), writer)
+      writer = Storage.append_to_log!(main_log_insert(LogOffset.new(100, 2), "b", "10"), writer)
+      Storage.hibernate(writer)
+
+      ConsumerRegistry.register_consumer(self(), shape_handle, ctx.stack_id)
+
+      {:ok, _pid} =
+        Materializer.start_link(%{
+          stack_id: ctx.stack_id,
+          shape_handle: shape_handle,
+          storage: ctx.storage,
+          columns: ["value"],
+          materialized_type: {:array, :int8}
+        })
+
+      respond_to_call(:await_snapshot_start, :started)
+      respond_to_call(:subscribe_materializer, {:ok, LogOffset.new(100, 2)})
+
+      mat_ctx = %{stack_id: ctx.stack_id, shape_handle: shape_handle}
+      assert Materializer.wait_until_ready(mat_ctx) == :ok
+      assert Materializer.get_link_values(mat_ctx) == MapSet.new([10])
+
+      Map.put(ctx, :mat_ctx, mat_ctx)
+    end
+
+    test "re-emits each toggle as its own payload, in offset order", %{mat_ctx: mat_ctx} do
+      # Behind before the transaction: the whole toggle sequence is replayed.
+      assert {:ok, seed, applied_offset} =
+               Materializer.subscribe(mat_ctx, LogOffset.new(50, 0))
+
+      # As of (50,0) value 10 is not yet present.
+      assert seed == MapSet.new([])
+      assert applied_offset == LogOffset.new(100, 2)
+
+      # Collect the payloads in the order they arrive (replay sends are
+      # sequential from the subscribe call, so mailbox order is emit order) and
+      # assert the full sequence: three distinct payloads, one per op, in offset
+      # order — not a single netted payload.
+      payloads =
+        for _ <- 1..3 do
+          assert_receive {:materializer_changes, _h, payload}
+          payload
+        end
+
+      assert [
+               %{move_in: [{10, "10"}], lsn: %LogOffset{tx_offset: 100, op_offset: 0}},
+               %{move_out: [{10, "10"}], lsn: %LogOffset{tx_offset: 100, op_offset: 1}},
+               %{move_in: [{10, "10"}], lsn: %LogOffset{tx_offset: 100, op_offset: 2}}
+             ] = payloads
+
+      refute_received {:materializer_changes, _h, _payload}
+    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

From 28771b70b8d35aad51780b479745d6b5c6a30ab0 Mon Sep 17 00:00:00 2001
From: rob <rob.acourt@gmail.com>
Date: Wed, 15 Jul 2026 19:12:56 +0100
Subject: [PATCH 8/8] Drop redundant materializer replay-skip guard
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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>
---
 .../lib/electric/shapes/consumer/materializer.ex   | 14 --------------
 .../test/integration/oracle_restore_test.exs       |  8 ++++----
 2 files changed, 4 insertions(+), 18 deletions(-)

diff --git a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
index c66d95e751..d62f065cad 100644
--- a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
@@ -501,20 +501,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)
diff --git a/packages/sync-service/test/integration/oracle_restore_test.exs b/packages/sync-service/test/integration/oracle_restore_test.exs
index 5af5a049e7..8736e7dd22 100644
--- a/packages/sync-service/test/integration/oracle_restore_test.exs
+++ b/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",