From 407ee51bbcb1c9f30cdeb2d489f506dfcf17c168 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/5] 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 0ee4162f198a56fe3c968fd519dc11de44ba8fba 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/5] 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 2726402759dc26346aedc935cd7f8389e3cc8e95 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/5] 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 ed23cd9429..89846d7412 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"})
@@ -1439,6 +1460,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 c53fd34e4cf995ad9352ee0d6ec3b65914fc4984 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/5] 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 fdf08f4e0335f844973987c9914dabef1df6c34d 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/5] 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