Preserve storage offsets when replaying nested subquery moves

.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.

.changeset/subquery-move-replay-on-restart.md

@@ -0,0 +1,11 @@
+---
+'@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. 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.

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()}
 
@@ -155,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)
 
@@ -168,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)
@@ -188,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)),