From 2287d9699da39487271dbed14bbd5c4e4bca4f9d 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/6] Add failing test

---
 .../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 d3bc8cf50ff1d0bdd0e4981c1166cfb7325e7081 Mon Sep 17 00:00:00 2001
From: rob <rob.acourt@gmail.com>
Date: Thu, 25 Jun 2026 15:01:36 +0100
Subject: [PATCH 2/6] Prevent shape removal and request crashes during restart
 (bug 6)

---
 .changeset/restore-shutdown-shape-removal.md  |  9 +++
 .../lib/electric/shapes/consumer.ex           | 12 ++++
 .../test/electric/shapes/consumer_test.exs    | 58 +++++++++++++++++++
 3 files changed, 79 insertions(+)
 create mode 100644 .changeset/restore-shutdown-shape-removal.md

diff --git a/.changeset/restore-shutdown-shape-removal.md b/.changeset/restore-shutdown-shape-removal.md
new file mode 100644
index 0000000000..1b2021a2f2
--- /dev/null
+++ b/.changeset/restore-shutdown-shape-removal.md
@@ -0,0 +1,9 @@
+---
+'@core/sync-service': patch
+---
+
+Stop subquery shapes from being spuriously removed during a server restart. When
+a dependency consumer's inline call to its materializer raced the materializer's
+shutdown, the resulting `:noproc` exit crashed the consumer and removed the shape
+from disk, causing a `409 must-refetch` after the restart. The consumer now
+absorbs that exit and lets the monitored `:DOWN` drive a clean stop.
diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex
index 4991b8cdb9..0935f7c9ea 100644
--- a/packages/sync-service/lib/electric/shapes/consumer.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer.ex
@@ -1063,6 +1063,18 @@ defmodule Electric.Shapes.Consumer do
          opts
        ) do
     Materializer.new_changes(Map.take(state, [:stack_id, :shape_handle]), changes_or_bounds, opts)
+  catch
+    # The consumer monitors the materializer; if the materializer died the
+    # :DOWN message is already in our mailbox and handle_materializer_down/2
+    # will run after the current handle_event/handle_call completes.
+    # Treat a `:noproc` (or transient `:normal`/`:shutdown` exit) here as
+    # the same condition: don't crash the consumer (which would route into
+    # the abnormal-shutdown path of handle_writer_termination and remove
+    # the shape from disk).
+    :exit, {:noproc, _} -> :ok
+    :exit, :noproc -> :ok
+    :exit, {:normal, _} -> :ok
+    :exit, {:shutdown, _} -> :ok
   end
 
   defp notify_materializer_of_new_changes(_state, _changes_or_bounds, _opts), do: :ok
diff --git a/packages/sync-service/test/electric/shapes/consumer_test.exs b/packages/sync-service/test/electric/shapes/consumer_test.exs
index 4fe0bb44d9..afeddb443a 100644
--- a/packages/sync-service/test/electric/shapes/consumer_test.exs
+++ b/packages/sync-service/test/electric/shapes/consumer_test.exs
@@ -2760,6 +2760,64 @@ defmodule Electric.Shapes.ConsumerTest do
       # After cleanup, the shape's rows should be removed from the index
       refute SubqueryIndex.has_positions?(index, shape_handle)
     end
+
+    test "dependency consumer survives a :noproc from its materializer without removing the shape",
+         ctx do
+      # Bug 6 cascade route: during a stack restart's shutdown, the dependency
+      # consumer's inline call into its materializer can race the
+      # materializer's death and exit with :noproc. Without the catch in
+      # notify_materializer_of_new_changes/3, that crashes the consumer with a
+      # non-shutdown reason, which routes through handle_writer_termination and
+      # removes the shape from disk — mid stack-shutdown that leaves the shape
+      # half-removed and 409s on the next poll after restart. The catch must
+      # absorb the exit so the pending :DOWN can drive a clean stop instead.
+
+      # Make the dependency consumer's notification call into the materializer
+      # exit exactly as a GenServer.call to an already-dead process would.
+      Repatch.patch(Consumer.Materializer, :new_changes, [mode: :shared], fn _, _, _ ->
+        exit({:noproc, {GenServer, :call, [:materializer, :new_changes, 5000]}})
+      end)
+
+      Support.TestUtils.activate_mocks_for_descendant_procs(Consumer)
+
+      # If the bug were present the consumer would crash and remove the shape;
+      # assert remove_shape is never called.
+      patch_shape_status(
+        remove_shape: fn _, handle ->
+          raise "Unexpected remove_shape for #{handle}"
+        end
+      )
+
+      {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
+
+      dep_consumer = Consumer.whereis(ctx.stack_id, dep_handle)
+      assert is_pid(dep_consumer)
+      ref = Process.monitor(dep_consumer)
+
+      # A change to the dependency table makes the dependency consumer notify
+      # its materializer — hitting the patched, exiting call.
+      ShapeLogCollector.handle_event(
+        complete_txn_fragment(100, Lsn.from_integer(50), [
+          %Changes.NewRecord{
+            relation: {"public", "other_table"},
+            record: %{"id" => "1"},
+            log_offset: LogOffset.new(Lsn.from_integer(50), 0)
+          }
+        ]),
+        ctx.stack_id
+      )
+
+      # With the catch, the dependency consumer absorbs the :noproc and stays
+      # alive; the shape is not removed.
+      refute_receive {:DOWN, ^ref, :process, _, _}, 500
+      assert Consumer.whereis(ctx.stack_id, dep_handle) == dep_consumer
+    end
   end
 
   defp refute_storage_calls_for_txn_fragment(shape_handle) do

From 060a23805c3f4b4cae756396b1b8c05534392210 Mon Sep 17 00:00:00 2001
From: rob <rob.acourt@gmail.com>
Date: Thu, 2 Jul 2026 12:58:16 +0100
Subject: [PATCH 3/6] 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  | 260 +++++++++++++++++-
 .../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, 687 insertions(+), 46 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 30405459c0..29cacef15d 100644
--- a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
@@ -103,11 +103,29 @@ defmodule Electric.Shapes.Consumer.Materializer do
     end)
   end
 
-  def subscribe(pid) when is_pid(pid), do: GenServer.call(pid, :subscribe)
+  @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.
+  """
+  def subscribe(pid, from_lsn \\ nil)
+
+  def subscribe(pid, from_lsn) when is_pid(pid),
+    do: GenServer.call(pid, {:subscribe, from_lsn})
 
-  def subscribe(opts) when is_map(opts), do: GenServer.call(name(opts), :subscribe)
+  def subscribe(opts, from_lsn) when is_map(opts),
+    do: GenServer.call(name(opts), {:subscribe, from_lsn})
 
-  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
@@ -132,6 +150,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()
@@ -172,6 +194,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
@@ -195,7 +225,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
@@ -205,7 +235,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
@@ -246,12 +276,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
@@ -260,10 +473,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()
@@ -282,10 +513,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
@@ -465,7 +704,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 afeddb443a..2e1d51381f 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 372b3baaa12db23115216260774e830eecbbf403 Mon Sep 17 00:00:00 2001
From: rob <rob.acourt@gmail.com>
Date: Thu, 2 Jul 2026 13:40:05 +0100
Subject: [PATCH 4/6] 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           | 99 ++++++++++++++++---
 .../lib/electric/shapes/consumer/state.ex     | 14 ++-
 .../test/electric/shapes/consumer_test.exs    | 12 ++-
 .../test/integration/oracle_restore_test.exs  |  7 +-
 5 files changed, 125 insertions(+), 16 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..ef73cf3768 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 ->
@@ -1015,25 +1021,96 @@ defmodule Electric.Shapes.Consumer do
   # 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.
+  defp commit_all_move_positions(%State{staged_move_positions: staged} = state)
+       when staged == %{},
+       do: state
+
+  defp commit_all_move_positions(%State{storage: storage} = state) when not is_nil(storage) do
+    positions =
+      Enum.reduce(state.staged_move_positions, state.move_positions, fn {handle, entries}, acc ->
+        {_threshold, lsn} = List.last(entries)
+        Map.update(acc, handle, lsn, &LogOffset.max(&1, lsn))
+      end)
+
+    try do
+      ShapeCache.Storage.set_move_positions!(positions, storage)
+    rescue
+      _ -> :ok
+    end
+
+    %{state | move_positions: positions, staged_move_positions: %{}}
+  end
+
+  defp commit_all_move_positions(state), do: state
+
   defp move_pipeline_fully_drained?(%EventHandler.Subqueries.Steady{queue: queue}),
     do: MoveQueue.length(queue) == 0
 
@@ -1216,7 +1293,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 2e1d51381f..0bbe988123 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 aafa5c169de63e276c789cfd0df5f92cf3ad89d8 Mon Sep 17 00:00:00 2001
From: rob <rob.acourt@gmail.com>
Date: Thu, 2 Jul 2026 15:18:27 +0100
Subject: [PATCH 5/6] Fix never-matches warnings in commit_all_move_positions/1

It is called from terminate/2 after terminate_writer/1 has popped :writer off
the state, so the value is no longer typed as %State{} and the %State{} clauses
were flagged as never matching. Match a bare map and read the fields via
Map.get instead.

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

diff --git a/packages/sync-service/lib/electric/shapes/consumer.ex b/packages/sync-service/lib/electric/shapes/consumer.ex
index ef73cf3768..419be52ff6 100644
--- a/packages/sync-service/lib/electric/shapes/consumer.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer.ex
@@ -1088,29 +1088,32 @@ defmodule Electric.Shapes.Consumer do
 
   # 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.
-  defp commit_all_move_positions(%State{staged_move_positions: staged} = state)
-       when staged == %{},
-       do: state
+  # 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)
 
-  defp commit_all_move_positions(%State{storage: storage} = state) when not is_nil(storage) do
-    positions =
-      Enum.reduce(state.staged_move_positions, state.move_positions, fn {handle, entries}, acc ->
-        {_threshold, lsn} = List.last(entries)
-        Map.update(acc, handle, lsn, &LogOffset.max(&1, lsn))
-      end)
+      try do
+        ShapeCache.Storage.set_move_positions!(positions, storage)
+      rescue
+        _ -> :ok
+      end
 
-    try do
-      ShapeCache.Storage.set_move_positions!(positions, storage)
-    rescue
-      _ -> :ok
+      %{state | move_positions: positions, staged_move_positions: %{}}
     end
-
-    %{state | move_positions: positions, staged_move_positions: %{}}
   end
 
-  defp commit_all_move_positions(state), do: state
-
   defp move_pipeline_fully_drained?(%EventHandler.Subqueries.Steady{queue: queue}),
     do: MoveQueue.length(queue) == 0
 

From a5981eac04ddf0da0bc928f7281d55ee2fc93632 Mon Sep 17 00:00:00 2001
From: joshdchang <joshchang04@gmail.com>
Date: Sun, 12 Jul 2026 22:39:48 -0400
Subject: [PATCH 6/6] Fix nested subquery move replay offsets

---
 .changeset/subquery-move-replay-on-restart.md |   4 +-
 .../electric/shape_cache/in_memory_storage.ex |  39 ++-
 .../electric/shape_cache/pure_file_storage.ex | 230 +++++++++++++----
 .../shape_cache/pure_file_storage/log_file.ex |  74 ++++--
 .../lib/electric/shape_cache/storage.ex       |  41 +++
 .../electric/shapes/consumer/materializer.ex  | 126 +++++-----
 .../pure_file_storage/log_file_test.exs       |  50 ++++
 .../storage_implementations_test.exs          | 236 ++++++++++++++++++
 .../shapes/consumer/materializer_test.exs     | 134 ++++++++++
 .../test/integration/oracle_restore_test.exs  |  58 ++++-
 .../sync-service/test/support/test_storage.ex |  13 +
 11 files changed, 872 insertions(+), 133 deletions(-)
 create mode 100644 packages/sync-service/test/electric/shape_cache/pure_file_storage/log_file_test.exs

diff --git a/.changeset/subquery-move-replay-on-restart.md b/.changeset/subquery-move-replay-on-restart.md
index 8a9b26302f..2c4156c0a9 100644
--- a/.changeset/subquery-move-replay-on-restart.md
+++ b/.changeset/subquery-move-replay-on-restart.md
@@ -6,4 +6,6 @@ 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.
+diverging from Postgres. Replay uses the authoritative persisted shape-log
+offset for control messages and spliced move-in rows, including nested
+subqueries whose generated JSON records do not contain source LSNs.
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..dbfc023839 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
@@ -170,7 +170,7 @@ defmodule Electric.ShapeCache.InMemoryStorage do
   defp snapshot_end(),
     do: snapshot_chunk_end(storage_offset(LogOffset.last_before_real_offsets()))
 
-  defp get_offset_indexed_stream(offset, max_offset, offset_indexed_table) do
+  defp get_offset_indexed_stream(offset, max_offset, offset_indexed_table, project_item) do
     offset = storage_offset(offset)
     max_offset = storage_offset(max_offset)
 
@@ -183,11 +183,15 @@ defmodule Electric.ShapeCache.InMemoryStorage do
           nil
 
         {{:offset, position}, [{_, item}]} ->
-          {item, position}
+          {project_item.(LogOffset.new(position), item), position}
       end
     end)
   end
 
+  defp get_offset_indexed_stream(offset, max_offset, offset_indexed_table) do
+    get_offset_indexed_stream(offset, max_offset, offset_indexed_table, fn _, item -> item end)
+  end
+
   @snapshot_boundary_offset LogOffset.last_before_real_offsets()
   @impl Electric.ShapeCache.Storage
   def get_log_stream(offset, max_offset, %MS{} = opts)
@@ -203,6 +207,37 @@ 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
+
+  defp get_offset_indexed_stream_with_offsets(offset, max_offset, offset_indexed_table) do
+    get_offset_indexed_stream(offset, max_offset, offset_indexed_table, fn offset, item ->
+      {offset, item}
+    end)
+  end
+
+  defp stream_from_snapshot_with_offsets(offset, max_offset, opts) do
+    offset
+    |> stream_from_snapshot(max_offset, opts)
+    |> Stream.map(&{nil, &1})
+  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..5487202023 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,71 @@ defmodule Electric.ShapeCache.PureFileStorage do
     end
   end
 
+  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_with_offsets(min_offset, max_offset, opts)
+  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
+
+    {stream_kind, stream} =
+      case {last_snapshot_chunk, min_offset} do
+        {_, x} when is_min_offset(x) ->
+          {:snapshot, Snapshot.stream_chunk_lines(opts, 0)}
+
+        {%LogOffset{} = latest, min_offset} when is_log_offset_lt(min_offset, latest) ->
+          {:snapshot, Snapshot.stream_chunk_lines(opts, op_offset + 1)}
+
+        {nil, _offset} ->
+          {:snapshot, wait_for_chunk_file_or_snapshot_end(opts, op_offset + 1)}
+
+        {%LogOffset{}, offset} ->
+          {:main_log, stream_main_log_with_offsets(offset, max_offset, opts)}
+      end
+
+    case stream_kind do
+      :snapshot -> Stream.map(stream, &{nil, &1})
+      :main_log -> stream
+    end
+  end
+
+  defp stream_main_log_with_offsets(min_offset, max_offset, opts) do
+    stream_main_log(
+      min_offset,
+      max_offset,
+      opts,
+      fn offset, item -> {offset, item} end,
+      true
+    )
+  end
+
   defp stream_main_log(min_offset, max_offset, %__MODULE__{} = opts) do
+    stream_main_log(min_offset, max_offset, opts, fn _, item -> item end, false)
+  end
+
+  defp stream_main_log(
+         min_offset,
+         max_offset,
+         %__MODULE__{} = opts,
+         project_item,
+         exact_upper_bound?
+       ) do
     storage_meta(
       ets_table: ets,
       last_persisted_offset: last_persisted,
@@ -1058,43 +1122,66 @@ defmodule Electric.ShapeCache.PureFileStorage do
 
     upper_read_bound = LogOffset.min(max_offset, last_seen)
 
-    # Convert upper_read_bound to tuple for comparison with ETS offsets
-    upper_read_bound_tuple = LogOffset.to_tuple(upper_read_bound)
-
     cond do
       is_log_offset_lte(last_persisted, min_offset) and is_nil(ets) ->
         []
 
       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
-          {_data, last_offset}
-          when is_nil(last_offset) or last_offset < upper_read_bound_tuple ->
+        case read_range_from_ets_cache(ets, min_offset, upper_read_bound, project_item) do
+          {_data, :incomplete} ->
             # 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)
-
-          {data, _last_offset} ->
+            stream_from_disk(
+              opts,
+              min_offset,
+              upper_read_bound,
+              boundary_info,
+              project_item,
+              exact_upper_bound?
+            )
+
+          {data, :complete} ->
             data
         end
 
       is_log_offset_lte(upper_read_bound, last_persisted) ->
-        stream_from_disk(opts, min_offset, upper_read_bound, boundary_info)
+        stream_from_disk(
+          opts,
+          min_offset,
+          upper_read_bound,
+          boundary_info,
+          project_item,
+          exact_upper_bound?
+        )
 
       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
-          {_upper_range, last_offset}
-          when is_nil(last_offset) or last_offset < upper_read_bound_tuple ->
+        case read_range_from_ets_cache(ets, last_persisted, upper_read_bound, project_item) do
+          {_upper_range, :incomplete} ->
             # 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)
-
-          {upper_range, _last_offset} ->
-            stream_from_disk(opts, min_offset, last_persisted, boundary_info)
+            stream_from_disk(
+              opts,
+              min_offset,
+              upper_read_bound,
+              boundary_info,
+              project_item,
+              exact_upper_bound?
+            )
+
+          {upper_range, :complete} ->
+            stream_from_disk(
+              opts,
+              min_offset,
+              last_persisted,
+              boundary_info,
+              project_item,
+              exact_upper_bound?
+            )
             |> Stream.concat(upper_range)
         end
     end
@@ -1139,15 +1226,27 @@ defmodule Electric.ShapeCache.PureFileStorage do
     end
   end
 
-  # Returns {data, last_offset_read} where last_offset_read is the offset tuple of the
-  # last entry read, or nil if no entries were read. This allows callers to detect
-  # partial reads due to concurrent ETS clearing.
-  @spec read_range_from_ets_cache(:ets.tid() | nil, LogOffset.t(), LogOffset.t()) ::
-          {list(), LogOffset.t_tuple() | nil}
-  defp read_range_from_ets_cache(nil, _min, _max), do: {[], nil}
-
-  defp read_range_from_ets_cache(ets, %LogOffset{} = min, %LogOffset{} = max) do
-    read_range_from_ets_cache(ets, LogOffset.to_tuple(min), LogOffset.to_tuple(max), [], nil)
+  # Returns whether the requested range was read completely. A range can be
+  # complete even when no item exists at its upper bound because log offsets are
+  # sparse. `:incomplete` means ETS disappeared or ended before the known live
+  # tail, so callers must retry from disk after a concurrent flush.
+  @spec read_range_from_ets_cache(
+          :ets.tid() | nil,
+          LogOffset.t(),
+          LogOffset.t(),
+          (LogOffset.t(), binary() -> term())
+        ) ::
+          {list(), :complete | :incomplete}
+  defp read_range_from_ets_cache(nil, _min, _max, _project_item), do: {[], :incomplete}
+
+  defp read_range_from_ets_cache(ets, %LogOffset{} = min, %LogOffset{} = max, project_item) do
+    read_range_from_ets_cache(
+      ets,
+      LogOffset.to_tuple(min),
+      LogOffset.to_tuple(max),
+      [],
+      project_item
+    )
   end
 
   @spec read_range_from_ets_cache(
@@ -1155,21 +1254,33 @@ defmodule Electric.ShapeCache.PureFileStorage do
           LogOffset.t_tuple(),
           LogOffset.t_tuple(),
           list(),
-          LogOffset.t_tuple() | nil
-        ) :: {list(), LogOffset.t_tuple() | nil}
-  defp read_range_from_ets_cache(ets, min, {max_tx, max_op} = max, acc, last_offset) do
+          (LogOffset.t(), binary() -> term())
+        ) :: {list(), :complete | :incomplete}
+  defp read_range_from_ets_cache(
+         ets,
+         min,
+         {max_tx, max_op} = max,
+         acc,
+         project_item
+       ) do
     case safe_next_lookup(ets, min) do
       :ets_dead ->
-        {Enum.reverse(acc), last_offset}
+        {Enum.reverse(acc), :incomplete}
 
       :"$end_of_table" ->
-        {Enum.reverse(acc), last_offset}
+        {Enum.reverse(acc), :incomplete}
 
       {{min_tx, min_op}, _} when min_tx > max_tx or (min_tx == max_tx and min_op > max_op) ->
-        {Enum.reverse(acc), last_offset}
+        {Enum.reverse(acc), :complete}
 
       {new_min, [{_, item}]} ->
-        read_range_from_ets_cache(ets, new_min, max, [item | acc], new_min)
+        projected_item = project_item.(LogOffset.new(new_min), item)
+
+        if new_min == max do
+          {Enum.reverse([projected_item | acc]), :complete}
+        else
+          read_range_from_ets_cache(ets, new_min, max, [projected_item | acc], project_item)
+        end
     end
   end
 
@@ -1181,7 +1292,14 @@ defmodule Electric.ShapeCache.PureFileStorage do
     ArgumentError -> :ets_dead
   end
 
-  defp stream_from_disk(%__MODULE__{}, min_offset, max_offset, _)
+  defp stream_from_disk(
+         %__MODULE__{},
+         min_offset,
+         max_offset,
+         _,
+         _project_item,
+         _exact_upper_bound?
+       )
        when is_log_offset_lte(max_offset, min_offset),
        do: []
 
@@ -1189,20 +1307,43 @@ defmodule Electric.ShapeCache.PureFileStorage do
          %__MODULE__{} = opts,
          min_offset,
          max_offset,
-         boundary_info
+         boundary_info,
+         project_item,
+         exact_upper_bound?
        ) 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_jsons(
-          opts,
-          json_file(opts, suffix),
-          start_pos,
-          end_pos,
-          min_offset
-        )
-        |> Stream.concat(stream_from_disk(opts, chunk_end_offset, max_offset, boundary_info))
+        if exact_upper_bound? and is_log_offset_lt(max_offset, chunk_end_offset) do
+          LogFile.stream_jsons_until_offset(
+            opts,
+            json_file(opts, suffix),
+            start_pos,
+            min_offset,
+            max_offset,
+            project_item
+          )
+        else
+          LogFile.stream_jsons(
+            opts,
+            json_file(opts, suffix),
+            start_pos,
+            end_pos,
+            min_offset,
+            project_item
+          )
+          |> Stream.concat(
+            stream_from_disk(
+              opts,
+              chunk_end_offset,
+              max_offset,
+              boundary_info,
+              project_item,
+              exact_upper_bound?
+            )
+          )
+        end
 
       {:ok, nil, {start_pos, nil}} ->
         LogFile.stream_jsons_until_offset(
@@ -1210,7 +1351,8 @@ defmodule Electric.ShapeCache.PureFileStorage do
           json_file(opts, suffix),
           start_pos,
           min_offset,
-          max_offset
+          max_offset,
+          project_item
         )
 
       :error ->
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 06761c8043..f7c1c0b317 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
@@ -179,7 +179,8 @@ defmodule Electric.ShapeCache.PureFileStorage.LogFile do
         log_file_path,
         start_position,
         end_position,
-        exclusive_min_offset
+        exclusive_min_offset,
+        project_item \\ &json_only/2
       ) do
     # We can read ahead entire chunk into memory since chunk sizes are expected to be ~10MB by default,
     case safely_open_file!(opts, log_file_path, [:read, :raw]) do
@@ -189,7 +190,9 @@ defmodule Electric.ShapeCache.PureFileStorage.LogFile do
       {:ok, file} ->
         try do
           with {:ok, data} <- :file.pread(file, start_position, end_position - start_position) do
-            {jsons, _} = extract_jsons_from_binary(data, exclusive_min_offset, nil)
+            {jsons, _} =
+              extract_jsons_from_binary(data, exclusive_min_offset, nil, project_item)
+
             jsons
           else
             :eof ->
@@ -212,7 +215,9 @@ defmodule Electric.ShapeCache.PureFileStorage.LogFile do
         log_file_path,
         start_position,
         exclusive_min_offset,
-        inclusive_max_offset
+        inclusive_max_offset,
+        project_item \\ &json_only/2,
+        read_fun \\ &:file.read/2
       ) do
     Stream.resource(
       fn ->
@@ -229,14 +234,18 @@ defmodule Electric.ShapeCache.PureFileStorage.LogFile do
         :halt ->
           {:halt, []}
 
+        {file, :halt} ->
+          {:halt, {file, ""}}
+
         {file, binary_rest} ->
-          case :file.read(file, 4096) do
+          case read_fun.(file, 4096) do
             {:ok, data} ->
               {jsons, rest} =
                 extract_jsons_from_binary(
                   binary_rest <> data,
                   exclusive_min_offset,
-                  inclusive_max_offset
+                  inclusive_max_offset,
+                  project_item
                 )
 
               {jsons, {file, rest}}
@@ -252,10 +261,23 @@ defmodule Electric.ShapeCache.PureFileStorage.LogFile do
     )
   end
 
-  @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 \\ [])
-  defp extract_jsons_from_binary(<<>>, _, _, acc), do: {Enum.reverse(acc), ""}
+  defp json_only(_offset, json), do: json
+
+  @spec extract_jsons_from_binary(
+          binary(),
+          LogOffset.t(),
+          LogOffset.t() | nil,
+          (LogOffset.t(), String.t() -> term())
+        ) :: {list(), binary() | :halt}
+  defp extract_jsons_from_binary(
+         binary,
+         exclusive_min_offset,
+         inclusive_max_offset,
+         project_item,
+         acc \\ []
+       )
+
+  defp extract_jsons_from_binary(<<>>, _, _, _, acc), do: {Enum.reverse(acc), ""}
 
   defp extract_jsons_from_binary(
          <<tx_offset1::64, op_offset1::64, key_size::32, _::binary-size(key_size), _::8, _flag::8,
@@ -265,41 +287,53 @@ defmodule Electric.ShapeCache.PureFileStorage.LogFile do
            op_offset: op_offset2
          } = log_offset,
          inclusive_max_offset,
+         project_item,
          acc
        )
        when tx_offset1 < tx_offset2 or (tx_offset1 == tx_offset2 and op_offset1 <= op_offset2),
-       do: extract_jsons_from_binary(rest, log_offset, inclusive_max_offset, acc)
+       do: extract_jsons_from_binary(rest, log_offset, inclusive_max_offset, project_item, acc)
 
   defp extract_jsons_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,
+         _log_offset,
+         %LogOffset{tx_offset: tx_offset2, op_offset: op_offset2},
+         project_item,
          acc
        )
        when tx_offset1 == tx_offset2 and op_offset1 == op_offset2,
-       do: extract_jsons_from_binary("", log_offset, inclusive_max_offset, [json | acc])
+       do: {
+         Enum.reverse([
+           project_item.(LogOffset.new(tx_offset1, op_offset1), json) | acc
+         ]),
+         :halt
+       }
 
   defp extract_jsons_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,
+         _log_offset,
+         %LogOffset{tx_offset: tx_offset2, op_offset: op_offset2},
+         _project_item,
          acc
        )
        when tx_offset1 > tx_offset2 or (tx_offset1 == tx_offset2 and op_offset1 > op_offset2),
-       do: extract_jsons_from_binary("", log_offset, inclusive_max_offset, acc)
+       do: {Enum.reverse(acc), :halt}
 
   defp extract_jsons_from_binary(
-         <<_::128, key_size::32, _::binary-size(key_size), _::8, _flag::8, json_size::64,
-           json::binary-size(json_size), rest::binary>>,
+         <<tx_offset::64, op_offset::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,
+         project_item,
          acc
        ),
-       do: extract_jsons_from_binary(rest, log_offset, inclusive_max_offset, [json | acc])
+       do:
+         extract_jsons_from_binary(rest, log_offset, inclusive_max_offset, project_item, [
+           project_item.(LogOffset.new(tx_offset, op_offset), json) | acc
+         ])
 
-  defp extract_jsons_from_binary(rest, _, _, acc),
+  defp extract_jsons_from_binary(rest, _, _, _, acc),
     do: {Enum.reverse(acc), rest}
 
   defp get_op_type(:insert), do: ?i
diff --git a/packages/sync-service/lib/electric/shape_cache/storage.ex b/packages/sync-service/lib/electric/shape_cache/storage.ex
index 1998612294..f0433814db 100644
--- a/packages/sync-service/lib/electric/shape_cache/storage.ex
+++ b/packages/sync-service/lib/electric/shape_cache/storage.ex
@@ -36,6 +36,7 @@ defmodule Electric.ShapeCache.Storage do
           {LogOffset.t(), key :: String.t(), operation_type :: operation_type(),
            Querying.json_iodata()}
   @type log :: Enumerable.t(Querying.json_iodata())
+  @type offset_log :: Enumerable.t({LogOffset.t() | nil, Querying.json_iodata()})
 
   @typedoc """
   A move-in snapshot row, represented as a 3-element list `[key, tags, json]`:
@@ -188,6 +189,21 @@ defmodule Electric.ShapeCache.Storage do
   @callback get_log_stream(offset :: LogOffset.t(), max_offset :: LogOffset.t(), shape_opts()) ::
               log()
 
+  @doc """
+  Get a stream of the log together with the authoritative offsets assigned by
+  storage.
+
+  Main-log entries are returned as `{offset, item}`. Initial snapshot entries
+  have no main-log offset and are returned as `{nil, item}`.
+  """
+  @callback get_log_stream_with_offsets(
+              offset :: LogOffset.t(),
+              max_offset :: LogOffset.t(),
+              shape_opts()
+            ) :: offset_log()
+
+  @optional_callbacks get_log_stream_with_offsets: 3
+
   @doc """
   Get the last exclusive offset of the chunk starting from the given offset.
 
@@ -436,6 +452,31 @@ defmodule Electric.ShapeCache.Storage do
     []
   end
 
+  @doc """
+  Get a shape log stream with the authoritative storage offset for each item.
+
+  The lower bound is exclusive and the upper bound is inclusive, matching
+  `get_log_stream/3`. Snapshot entries are paired with `nil` because they do
+  not have main-log offsets.
+  """
+  @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
+    if Code.ensure_loaded?(mod) and function_exported?(mod, :get_log_stream_with_offsets, 3) do
+      mod.get_log_stream_with_offsets(offset, max_offset, shape_opts)
+    else
+      raise Error,
+        message: "Storage adapter #{inspect(mod)} does not support offset-preserving log streams"
+    end
+  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 29cacef15d..8c81cc362f 100644
--- a/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
+++ b/packages/sync-service/lib/electric/shapes/consumer/materializer.ex
@@ -310,75 +310,88 @@ defmodule Electric.Shapes.Consumer.Materializer 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.
+    seed_state = replay_seed_at(state, from_lsn, storage)
+    seed = link_values_from_counts(seed_state.value_counts)
+
+    # Replay only the missing tail using storage's authoritative offsets.
+    # Move controls, snapshot-end controls, and spliced move-in rows do not
+    # encode Postgres offsets in their JSON, so JSON headers are not a safe
+    # replay cursor.
     replay0 = %{
+      seed_state
+      | offset: from_lsn,
+        subscribed_offset: state.applied_offset,
+        pending_events: %{pid: pid}
+    }
+
+    Storage.get_log_stream_with_offsets(from_lsn, state.applied_offset, storage)
+    |> apply_replay_stream(replay0)
+
+    seed
+  end
+
+  # Reconstruct the exact dependency view at the persisted subscriber cursor.
+  # Initial snapshot entries have virtual chunk offsets rather than per-row
+  # offsets, so replay that section with the existing snapshot iterator. Read
+  # the main-log section separately through the strictly bounded offset-aware
+  # storage API; ordinary get_log_stream/3 reads whole storage chunks and may
+  # intentionally include entries past its requested max offset.
+  defp replay_seed_at(state, from_lsn, storage) do
+    snapshot_end = LogOffset.min(from_lsn, LogOffset.last_before_real_offsets())
+
+    seed0 = %{
       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}
+        subscribed_offset: snapshot_end,
+        pending_events: %{}
     }
 
-    replay =
-      read_history_up_to_subscribed(replay0, storage, fn stream, acc ->
-        apply_replay_stream(stream, acc)
-      end)
+    seed_state = read_history_up_to_subscribed(seed0, storage)
 
-    case replay.pending_events.seed do
-      nil -> link_values_from_counts(replay.value_counts)
-      seed -> seed
+    if is_log_offset_lt(snapshot_end, from_lsn) do
+      {seed_state, _events} =
+        Storage.get_log_stream_with_offsets(snapshot_end, from_lsn, storage)
+        |> Stream.map(fn {_offset, item} -> item end)
+        |> decode_json_stream()
+        |> apply_changes(seed_state)
+
+      %{seed_state | offset: from_lsn, subscribed_offset: from_lsn}
+    else
+      seed_state
     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).
+  # Apply function used during move replay. The supplied stream is the missing
+  # tail `(from_lsn, applied_offset]`, paired with authoritative storage
+  # offsets. Group each transaction, apply it to the throwaway seed state, and
+  # emit its net moves at the persisted offset.
   defp apply_replay_stream(stream, acc) do
-    %{from_lsn: from_lsn, pid: pid} = acc.pending_events
+    %{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()
+      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)
+      if events != %{} do
+        payload =
+          events
+          |> finalize_txids()
+          |> Map.put(:lsn, txn_offset)
 
-          send(pid, {:materializer_changes, handle, payload})
-        end
+        send(pid, {:materializer_changes, handle, payload})
       end
 
       acc
@@ -386,25 +399,20 @@ defmodule Electric.Shapes.Consumer.Materializer do
   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.
+  # preserving the authoritative per-item storage offset 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 ->
+    |> Stream.map(fn {offset, item} -> {offset, Jason.decode!(item)} end)
+    |> Stream.filter(fn {_offset, decoded} ->
       Map.has_key?(decoded, "key") || Map.has_key?(decoded["headers"], "event")
     end)
-    |> Stream.map(fn decoded ->
+    |> Stream.map(fn {offset, decoded} ->
       headers = decoded["headers"]
-      {decode_offset(headers), decode_txids(headers), decode_change(decoded)}
+      {offset, 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: []
 
@@ -456,8 +464,8 @@ defmodule Electric.Shapes.Consumer.Materializer do
   # 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 ->
+    |> Stream.chunk_by(fn {offset, _txids, _change} -> offset.tx_offset end)
+    |> Stream.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()
diff --git a/packages/sync-service/test/electric/shape_cache/pure_file_storage/log_file_test.exs b/packages/sync-service/test/electric/shape_cache/pure_file_storage/log_file_test.exs
new file mode 100644
index 0000000000..d017620957
--- /dev/null
+++ b/packages/sync-service/test/electric/shape_cache/pure_file_storage/log_file_test.exs
@@ -0,0 +1,50 @@
+defmodule Electric.ShapeCache.PureFileStorage.LogFileTest do
+  use ExUnit.Case, async: false
+
+  alias Electric.Replication.LogOffset
+  alias Electric.ShapeCache.PureFileStorage
+  alias Electric.ShapeCache.PureFileStorage.LogFile
+
+  @moduletag :tmp_dir
+
+  test "bounded streaming stops reading the file after the inclusive max", %{tmp_dir: tmp_dir} do
+    path = Path.join(tmp_dir, "bounded-log.bin")
+    max_offset = LogOffset.new(2, 0)
+
+    entries = [
+      {LogOffset.new(1, 0), "first", :insert, Jason.encode!(%{id: "first"})},
+      {max_offset, "max", :insert, Jason.encode!(%{id: "max"})}
+    ]
+
+    tail =
+      for tx <- 3..8 do
+        json = Jason.encode!(%{id: "tail-#{tx}", padding: String.duplicate("x", 4_096)})
+        {LogOffset.new(tx, 0), "tail-#{tx}", :insert, json}
+      end
+
+    LogFile.write_log_file(entries ++ tail, path, 1_000_000)
+
+    test_pid = self()
+
+    read_fun = fn file, size ->
+      send(test_pid, :file_read)
+      :file.read(file, size)
+    end
+
+    opts = %PureFileStorage{stack_id: "test", shape_handle: "test"}
+
+    assert LogFile.stream_jsons_until_offset(
+             opts,
+             path,
+             0,
+             LogOffset.first(),
+             max_offset,
+             fn _offset, json -> json end,
+             read_fun
+           )
+           |> Enum.map(&Jason.decode!/1) == [%{"id" => "first"}, %{"id" => "max"}]
+
+    assert_receive :file_read
+    refute_receive :file_read
+  end
+end
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 a65ff3b32b..fd3ccd8499 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
@@ -14,6 +14,9 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
   import Support.ComponentSetup
   import Support.TestUtils
 
+  defmodule StorageWithoutOffsetStream do
+  end
+
   @moduletag :tmp_dir
 
   @shape_handle "the-shape-handle"
@@ -49,6 +52,18 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
 
   setup [:with_stack_id_from_test, :with_async_deleter]
 
+  test "offset-preserving reads fail clearly for adapters without the optional capability" do
+    assert_raise Storage.Error,
+                 "Storage adapter #{inspect(StorageWithoutOffsetStream)} does not support offset-preserving log streams",
+                 fn ->
+                   Storage.get_log_stream_with_offsets(
+                     LogOffset.first(),
+                     LogOffset.last(),
+                     {StorageWithoutOffsetStream, :shape_opts}
+                   )
+                 end
+  end
+
   for module <- [InMemoryStorage, PureFileStorage] do
     module_name = module |> Module.split() |> List.last()
 
@@ -439,6 +454,158 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
       end
     end
 
+    describe "#{module_name}.get_log_stream_with_offsets/3" do
+      setup do
+        {:ok, %{module: unquote(module)}}
+      end
+
+      setup :start_storage
+
+      test "pairs snapshot entries with nil offsets", %{storage: storage} do
+        Storage.mark_snapshot_as_started(storage)
+        Storage.make_new_snapshot!(@data_stream, storage)
+
+        entries =
+          Storage.get_log_stream_with_offsets(
+            LogOffset.before_all(),
+            LogOffset.last_before_real_offsets(),
+            storage
+          )
+          |> Enum.to_list()
+
+        assert [{nil, _}, {nil, _}] = entries
+      end
+
+      @tag chunk_size: 1
+      test "preserves exact main-log offsets and bounds across gaps, generated entries, and hibernate",
+           %{
+             storage: storage,
+             writer: writer
+           } do
+        Storage.mark_snapshot_as_started(storage)
+        Storage.make_new_snapshot!([], storage)
+
+        base_offset = LogOffset.new(1_000, 0)
+
+        writer =
+          Storage.append_to_log!(
+            [
+              {base_offset, "regular", :insert,
+               Jason.encode!(%{
+                 key: "regular",
+                 value: %{id: "regular"},
+                 headers: %{operation: "insert"}
+               })}
+            ],
+            writer
+          )
+
+        {{^base_offset, control_offset}, writer} =
+          Storage.append_control_message!(
+            %{headers: %{event: "move-out", patterns: [%{pos: 0, value: "tag"}]}},
+            writer
+          )
+
+        Storage.write_move_in_snapshot!(
+          [
+            [
+              "move-in",
+              ["tag"],
+              Jason.encode!(%{
+                key: "move-in",
+                value: %{id: "move-in"},
+                headers: %{operation: "insert", is_move_in: true}
+              })
+            ]
+          ],
+          "move-in-snapshot",
+          storage
+        )
+
+        {{^control_offset, move_in_offset}, writer} =
+          Storage.append_move_in_snapshot_to_log!("move-in-snapshot", writer)
+
+        gapped_offset = LogOffset.new(2_000, 5)
+
+        writer =
+          Storage.append_to_log!(
+            [
+              {gapped_offset, "after-gap", :insert,
+               Jason.encode!(%{
+                 key: "after-gap",
+                 value: %{id: "after-gap"},
+                 headers: %{operation: "insert"}
+               })}
+            ],
+            writer
+          )
+
+        {{^gapped_offset, snapshot_end_offset}, writer} =
+          Storage.append_control_message!(
+            %{headers: %{control: "snapshot-end"}},
+            writer
+          )
+
+        expected_offsets = [
+          base_offset,
+          control_offset,
+          move_in_offset,
+          gapped_offset,
+          snapshot_end_offset
+        ]
+
+        read_entries = fn min_offset, max_offset ->
+          storage
+          |> then(
+            &Storage.get_log_stream_with_offsets(
+              min_offset,
+              max_offset,
+              &1
+            )
+          )
+          |> Enum.map(fn {offset, json} ->
+            label =
+              case Jason.decode!(json) do
+                %{"key" => key} -> key
+                %{"headers" => %{"event" => event}} -> event
+                %{"headers" => %{"control" => control}} -> control
+              end
+
+            {offset, label}
+          end)
+        end
+
+        expected_entries =
+          Enum.zip(expected_offsets, [
+            "regular",
+            "move-out",
+            "move-in",
+            "after-gap",
+            "snapshot-end"
+          ])
+
+        assert read_entries.(LogOffset.last_before_real_offsets(), LogOffset.last()) ==
+                 expected_entries
+
+        assert read_entries.(base_offset, LogOffset.new(1_500, 0)) ==
+                 Enum.slice(expected_entries, 1, 2)
+
+        assert read_entries.(control_offset, gapped_offset) ==
+                 Enum.slice(expected_entries, 2, 2)
+
+        _writer = Storage.hibernate(writer)
+
+        assert read_entries.(LogOffset.last_before_real_offsets(), LogOffset.last()) ==
+                 expected_entries
+
+        assert read_entries.(base_offset, LogOffset.new(1_500, 0)) ==
+                 Enum.slice(expected_entries, 1, 2)
+
+        assert read_entries.(control_offset, gapped_offset) ==
+                 Enum.slice(expected_entries, 2, 2)
+      end
+    end
+
     describe "#{module_name}.append_control_message!/2" do
       setup do
         {:ok, %{module: unquote(module)}}
@@ -958,6 +1125,75 @@ defmodule Electric.ShapeCache.StorageImplimentationsTest do
     # FS is occasionally slow so we set a high timeout for compaction
     @default_compaction_timeout 2000
 
+    describe "#{module_name}.get_log_stream_with_offsets/3 disk and live tail bounds" do
+      @tag chunk_size: 1
+      setup :start_storage
+      setup :start_empty_snapshot
+
+      test "keeps sparse lower-exclusive and upper-inclusive bounds across disk and ETS", %{
+        storage: storage,
+        writer: writer
+      } do
+        entry = fn offset, label ->
+          {offset, label, :insert,
+           Jason.encode!(%{
+             key: label,
+             value: %{id: label},
+             headers: %{operation: "insert"}
+           })}
+        end
+
+        disk_first = LogOffset.new(100, 0)
+        disk_second = LogOffset.new(100, 10)
+        disk_last = LogOffset.new(200, 0)
+
+        writer =
+          Storage.append_to_log!(
+            [entry.(disk_first, "disk-first"), entry.(disk_second, "disk-second")],
+            writer
+          )
+
+        writer = Storage.append_to_log!([entry.(disk_last, "disk-last")], writer)
+        writer = Storage.hibernate(writer)
+
+        live_first = LogOffset.new(1_000, 0)
+        live_last = LogOffset.new(2_000, 0)
+        writer = Storage.append_to_log!([entry.(live_first, "live-first")], writer)
+        writer = Storage.append_to_log!([entry.(live_last, "live-last")], writer)
+
+        read = fn min_offset, max_offset ->
+          Storage.get_log_stream_with_offsets(min_offset, max_offset, storage)
+          |> Enum.map(fn {offset, json} -> {offset, Jason.decode!(json)["key"]} end)
+        end
+
+        # The upper bound falls in a deliberate offset gap inside a complete
+        # on-disk chunk. The later item from that chunk must not leak through.
+        assert read.(LogOffset.last_before_real_offsets(), LogOffset.new(100, 5)) == [
+                 {disk_first, "disk-first"}
+               ]
+
+        # The requested range crosses the persisted/live boundary and ends in
+        # another deliberate gap. Disk and ETS entries are joined once each.
+        assert read.(disk_first, LogOffset.new(1_500, 0)) == [
+                 {disk_second, "disk-second"},
+                 {disk_last, "disk-last"},
+                 {live_first, "live-first"}
+               ]
+
+        _writer = Storage.hibernate(writer)
+
+        assert read.(LogOffset.last_before_real_offsets(), LogOffset.new(100, 5)) == [
+                 {disk_first, "disk-first"}
+               ]
+
+        assert read.(disk_first, LogOffset.new(1_500, 0)) == [
+                 {disk_second, "disk-second"},
+                 {disk_last, "disk-last"},
+                 {live_first, "live-first"}
+               ]
+      end
+    end
+
     describe "#{module_name}.compact/1" do
       setup :start_storage
 
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..c4626275b2 100644
--- a/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs
+++ b/packages/sync-service/test/electric/shapes/consumer/materializer_test.exs
@@ -1644,6 +1644,140 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
       assert seed == MapSet.new([10, 20, 30])
       refute_received {:materializer_changes, _handle, _payload}
     end
+
+    @tag chunk_size: 1
+    test "replays persisted move control messages at their authoritative log offsets", ctx do
+      shape_handle = "nested-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)
+
+      Storage.make_new_snapshot!(
+        make_snapshot_data([
+          %Changes.NewRecord{
+            record: %{"id" => "1", "value" => "10"},
+            move_tags: ["nested-tag"],
+            active_conditions: [true]
+          }
+        ]),
+        storage
+      )
+
+      cursor = LogOffset.new(100, 0)
+      writer = Storage.append_to_log!(main_log_insert(cursor, "2", "20"), writer)
+
+      {{^cursor, ignored_offset}, writer} =
+        Storage.append_control_message!(
+          Jason.encode!(%{headers: %{control: "up_to_date"}}),
+          writer
+        )
+
+      {{^ignored_offset, control_offset}, writer} =
+        Storage.append_control_message!(
+          Jason.encode!(%{
+            headers: %{
+              event: "move-out",
+              patterns: [%{pos: 0, value: "nested-tag"}],
+              txids: [42]
+            }
+          }),
+          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, control_offset})
+
+      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([20])
+
+      assert {:ok, seed, ^control_offset} = Materializer.subscribe(mat_ctx, cursor)
+      assert seed == MapSet.new([10, 20])
+
+      assert_receive {:materializer_changes, ^shape_handle,
+                      %{
+                        move_out: [{10, "10"}],
+                        lsn: ^control_offset,
+                        txids: [42]
+                      }}
+    end
+
+    @tag chunk_size: 1
+    test "replays persisted move-in rows at their authoritative log offsets", ctx do
+      shape_handle = "move-in-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)
+
+      Storage.make_new_snapshot!(
+        make_snapshot_data([%Changes.NewRecord{record: %{"id" => "1", "value" => "10"}}]),
+        storage
+      )
+
+      cursor = LogOffset.new(100, 0)
+      writer = Storage.append_to_log!(main_log_insert(cursor, "2", "20"), writer)
+
+      move_in_item =
+        Jason.encode!(%{
+          key: ~s|"public"."test_table"/"3"|,
+          value: %{"id" => "3", "value" => "30"},
+          headers: %{operation: "insert", tags: ["nested-tag"], active_conditions: [true]}
+        })
+
+      Storage.write_move_in_snapshot!(
+        [[~s|"public"."test_table"/"3"|, ["nested-tag"], move_in_item]],
+        "nested-move-in",
+        storage
+      )
+
+      {{^cursor, move_in_offset}, writer} =
+        Storage.append_move_in_snapshot_to_log!("nested-move-in", 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, move_in_offset})
+
+      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])
+
+      assert {:ok, seed, ^move_in_offset} = Materializer.subscribe(mat_ctx, cursor)
+      assert seed == MapSet.new([10, 20])
+
+      assert_receive {:materializer_changes, ^shape_handle,
+                      %{
+                        move_in: [{30, "30"}],
+                        lsn: ^move_in_offset,
+                        txids: []
+                      }}
+    end
   end
 
   describe "startup race condition handling" do
diff --git a/packages/sync-service/test/integration/oracle_restore_test.exs b/packages/sync-service/test/integration/oracle_restore_test.exs
index 93e677575e..9a0671746a 100644
--- a/packages/sync-service/test/integration/oracle_restore_test.exs
+++ b/packages/sync-service/test/integration/oracle_restore_test.exs
@@ -7,7 +7,8 @@ defmodule Electric.Integration.OracleRestoreTest do
   These tests use a small, readable "issue tracker" domain schema rather than
   the abstract `level_N` hierarchy from `Support.OracleHarness.StandardSchema`:
 
-      projects (id, active)
+      teams (id, active)
+          └── projects (id, team_id, active)
           └── issues (id, project_id, title)
 
   The shape under test is "issues belonging to an active project", expressed
@@ -55,17 +56,26 @@ defmodule Electric.Integration.OracleRestoreTest do
     %{replication_opts_overrides: [slot_temporary?: false]}
   end
 
-  # A two-table "issue tracker": projects own issues. `projects.active` drives
-  # the subquery shape under test. Seeded so that p1, p3, p5 start active and
-  # p2, p4 start inactive; issues are spread round-robin across the projects
-  # so each project owns four (e.g. p1 owns i1, i6, i11, i16).
+  # A three-table "issue tracker": teams own projects, which own issues.
+  # `projects.active` drives the existing subquery shapes, while `teams.active`
+  # drives the nested-subquery regression below. Issues are spread round-robin
+  # across the projects so each project owns four (e.g. p1 owns i1, i6, i11,
+  # i16).
   defp setup_issue_tracker_schema(ctx) do
     OracleHarness.apply_sql(ctx, [
       "DROP TABLE IF EXISTS issues CASCADE",
       "DROP TABLE IF EXISTS projects CASCADE",
+      "DROP TABLE IF EXISTS teams CASCADE",
+      """
+      CREATE TABLE teams (
+        id TEXT PRIMARY KEY,
+        active BOOLEAN NOT NULL DEFAULT true
+      )
+      """,
       """
       CREATE TABLE projects (
         id TEXT PRIMARY KEY,
+        team_id TEXT NOT NULL REFERENCES teams(id) ON DELETE CASCADE,
         active BOOLEAN NOT NULL DEFAULT true
       )
       """,
@@ -78,12 +88,16 @@ defmodule Electric.Integration.OracleRestoreTest do
       """
     ])
 
+    team_values = "('t1', true), ('t2', false)"
     project_ids = for n <- 1..5, do: "p#{n}"
 
     project_values =
       project_ids
       |> Enum.with_index()
-      |> Enum.map_join(", ", fn {id, idx} -> "('#{id}', #{rem(idx, 2) == 0})" end)
+      |> Enum.map_join(", ", fn {id, idx} ->
+        team_id = if rem(idx, 2) == 0, do: "t1", else: "t2"
+        "('#{id}', '#{team_id}', #{rem(idx, 2) == 0})"
+      end)
 
     issue_values =
       for n <- 1..20 do
@@ -93,7 +107,8 @@ defmodule Electric.Integration.OracleRestoreTest do
       |> Enum.join(", ")
 
     OracleHarness.apply_sql(ctx, [
-      "INSERT INTO projects (id, active) VALUES #{project_values}",
+      "INSERT INTO teams (id, active) VALUES #{team_values}",
+      "INSERT INTO projects (id, team_id, active) VALUES #{project_values}",
       "INSERT INTO issues (id, project_id, title) VALUES #{issue_values}"
     ])
 
@@ -135,6 +150,35 @@ defmodule Electric.Integration.OracleRestoreTest do
     OracleHarness.test_against_oracle(ctx, shapes, batches, restart_server_every: 1)
   end
 
+  @tag :oracle_restore_nested_subquery
+  test "nested subquery dependency moves replay across repeated server restarts", ctx do
+    shapes = [
+      %{
+        name: "issues_of_projects_in_active_teams",
+        table: "issues",
+        where:
+          "project_id IN (SELECT id FROM projects WHERE team_id IN " <>
+            "(SELECT id FROM teams WHERE active = true))",
+        columns: ["id", "project_id", "title"],
+        pk: ["id"],
+        optimized: true
+      }
+    ]
+
+    # Only dependency rows change: the projects and issues remain untouched.
+    # Restart after every batch so both move-out and move-in must be restored
+    # from the persisted nested dependency logs, twice, without duplication or
+    # shape rotation.
+    batches = [
+      [[%{name: "deactivate_t1", sql: "UPDATE teams SET active = false WHERE id = 't1'"}]],
+      [[%{name: "reactivate_t1", sql: "UPDATE teams SET active = true WHERE id = 't1'"}]],
+      [[%{name: "deactivate_t1_again", sql: "UPDATE teams SET active = false WHERE id = 't1'"}]],
+      [[%{name: "reactivate_t1_again", sql: "UPDATE teams SET active = true WHERE id = 't1'"}]]
+    ]
+
+    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
diff --git a/packages/sync-service/test/support/test_storage.ex b/packages/sync-service/test/support/test_storage.ex
index bdf493d49a..ec79a88149 100644
--- a/packages/sync-service/test/support/test_storage.ex
+++ b/packages/sync-service/test/support/test_storage.ex
@@ -125,6 +125,19 @@ 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})