feat(sync-service): add txids to move-in/move-out control messages (#4304)

## Summary

Move-in/move-out control messages now carry `headers.txids`, mirroring
the per-row `headers.txids` already emitted on insert/update/delete log
entries. The list contains the upstream Postgres xid(s) whose commit
caused the dependency boundary to flip.

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: robacourt

.changeset/move-message-txids.md

@@ -0,0 +1,5 @@
+---
+'@core/sync-service': patch
+---
+
+Add `txids` to `move-in`/`move-out` control messages, mirroring the per-row `headers.txids` already emitted on insert/update/delete log entries. The list contains the upstream Postgres xid(s) whose commit caused the dependency boundary to flip.

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

@@ -308,7 +308,7 @@ defmodule Electric.Shapes.Consumer do
   end
 
   def handle_info(
-        {:materializer_changes, dep_handle, %{move_in: move_in, move_out: move_out}},
+        {:materializer_changes, dep_handle, %{move_in: move_in, move_out: move_out} = payload},
         state
       ) do
     Logger.debug(fn ->
@@ -317,10 +317,7 @@ defmodule Electric.Shapes.Consumer do
 
     handle_apply_event_result(
       state,
-      apply_event(
-        state,
-        {:materializer_changes, dep_handle, %{move_in: move_in, move_out: move_out}}
-      )
+      apply_event(state, {:materializer_changes, dep_handle, payload})
     )
   end
 
@@ -649,7 +646,8 @@ defmodule Electric.Shapes.Consumer do
         # The Materializer must see all txn changes for correct tracking of move-ins and
         # move-outs for the outer shape. The commit=false flag ensure it doesn't yet notify
         # outer consumers about those changes.
-        :ok = notify_materializer_of_new_changes(state, converted_changes, commit: false)
+        :ok =
+          notify_materializer_of_new_changes(state, converted_changes, commit: false, xid: xid)
 
         txn =
           PendingTxn.update_with_changes(
@@ -720,7 +718,7 @@ defmodule Electric.Shapes.Consumer do
       # Signal commit to storage to allow it to advance its internal txn offset
       writer = ShapeCache.Storage.signal_txn_commit!(txn.xid, writer)
 
-      :ok = notify_new_changes(state, [], state.latest_offset)
+      :ok = notify_new_changes_with_offset(state, [], state.latest_offset, xid: txn.xid)
 
       lag = calculate_replication_lag(txn_fragment.commit.commit_timestamp)
 
@@ -805,7 +803,7 @@ defmodule Electric.Shapes.Consumer do
 
       {state, notification, num_changes, total_size} ->
         if notification do
-          :ok = notify_new_changes(state, notification)
+          :ok = notify_new_changes(state, notification, xid: txn.xid)
 
           OpenTelemetry.add_span_attributes(%{
             num_bytes: total_size,
@@ -897,19 +895,21 @@ defmodule Electric.Shapes.Consumer do
     mark_for_removal(state)
   end
 
-  defp notify_new_changes(_state, nil), do: :ok
+  defp notify_new_changes(state, notification, opts \\ [])
+  defp notify_new_changes(_state, nil, _opts), do: :ok
 
-  defp notify_new_changes(state, {changes, upper_bound}) do
-    notify_new_changes(state, changes, upper_bound)
+  defp notify_new_changes(state, {changes, upper_bound}, opts) do
+    notify_new_changes_with_offset(state, changes, upper_bound, opts)
   end
 
-  @spec notify_new_changes(
+  @spec notify_new_changes_with_offset(
           state :: map(),
           changes_or_bounds :: list(Changes.change()) | {LogOffset.t(), LogOffset.t()},
-          latest_log_offset :: LogOffset.t()
+          latest_log_offset :: LogOffset.t(),
+          opts :: keyword()
         ) :: :ok
-  defp notify_new_changes(state, changes_or_bounds, latest_log_offset) do
-    :ok = notify_materializer_of_new_changes(state, changes_or_bounds)
+  defp notify_new_changes_with_offset(state, changes_or_bounds, latest_log_offset, opts) do
+    :ok = notify_materializer_of_new_changes(state, changes_or_bounds, opts)
     :ok = notify_clients_of_new_changes(state, latest_log_offset)
   end
 
@@ -937,8 +937,6 @@ defmodule Electric.Shapes.Consumer do
           changes_or_bounds :: list(Changes.change()) | {LogOffset.t(), LogOffset.t()},
           opts :: keyword()
         ) :: :ok
-  defp notify_materializer_of_new_changes(state, changes_or_bounds, opts \\ [])
-
   defp notify_materializer_of_new_changes(
          %{materializer_subscribed?: true} = state,
          changes_or_bounds,

packages/sync-service/lib/electric/shapes/consumer/event_handler/subqueries/buffering.ex

@@ -37,7 +37,7 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Buffering do
         %ShapeInfo{} = shape_info,
         views,
         %MoveQueue{} = queue,
-        {dep_move_kind, dep_index, values} = move,
+        {dep_move_kind, dep_index, values, txids} = move,
         subquery_ref,
         opts \\ []
       )
@@ -47,7 +47,7 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Buffering do
       queue: queue,
       active_move:
         views
-        |> ActiveMove.start(dep_index, dep_move_kind, subquery_ref, values)
+        |> ActiveMove.start(dep_index, dep_move_kind, subquery_ref, values, txids)
         |> ActiveMove.carry_latest_seen_lsn(Keyword.get(opts, :latest_seen_lsn))
     }
 
@@ -129,7 +129,8 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Buffering do
       index_effects =
         IndexChanges.effects_for_complete(
           state.shape_info.dnf_plan,
-          {active_move.dep_move_kind, active_move.dep_index, active_move.values},
+          {active_move.dep_move_kind, active_move.dep_index, active_move.values,
+           active_move.txids},
           active_move.subquery_ref
         )
 

packages/sync-service/lib/electric/shapes/consumer/event_handler/subqueries/steady.ex

@@ -75,7 +75,7 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Steady do
       nil ->
         {:ok, state, effects}
 
-      {{dep_move_kind, dep_index, values} = move, queue} ->
+      {{dep_move_kind, dep_index, values, txids} = move, queue} ->
         subquery_ref = RefResolver.ref_from_dep_index!(state.shape_info.ref_resolver, dep_index)
         subscription_active? = Keyword.get(opts, :subscription_active?, false)
         latest_seen_lsn = Keyword.get(opts, :latest_seen_lsn)
@@ -108,7 +108,7 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Steady do
             effects =
               effects
               |> EffectList.append(
-                MoveBroadcast.effect_for_move_out(dep_index, values, state.shape_info)
+                MoveBroadcast.effect_for_move_out(dep_index, values, txids, state.shape_info)
               )
               |> EffectList.append_all(index_effects)
 

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

@@ -41,7 +41,8 @@ defmodule Electric.Shapes.Consumer.Materializer do
           :ok
   def new_changes(state, changes, opts \\ []) do
     commit? = Keyword.get(opts, :commit, true)
-    GenServer.call(name(state), {:new_changes, changes, commit?}, :infinity)
+    xid = Keyword.get(opts, :xid)
+    GenServer.call(name(state), {:new_changes, changes, xid, commit?}, :infinity)
   end
 
   def wait_until_ready(state) do
@@ -205,23 +206,23 @@ defmodule Electric.Shapes.Consumer.Materializer do
     {:reply, :ok, state}
   end
 
-  def handle_call({:new_changes, {range_start, range_end}, commit?}, _from, state) do
+  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)
 
     state =
       Storage.get_log_stream(range_start, range_end, storage)
       |> decode_json_stream()
-      |> apply_and_accumulate_events(state)
+      |> apply_and_accumulate_events(xid, state)
       |> maybe_flush_pending_events(commit?)
 
     {:reply, :ok, state}
   end
 
-  def handle_call({:new_changes, changes, commit?}, _from, state) when is_list(changes) do
+  def handle_call({:new_changes, changes, xid, commit?}, _from, state) when is_list(changes) do
     state =
       changes
-      |> apply_and_accumulate_events(state)
+      |> apply_and_accumulate_events(xid, state)
       |> maybe_flush_pending_events(commit?)
 
     {:reply, :ok, state}
@@ -343,21 +344,21 @@ defmodule Electric.Shapes.Consumer.Materializer do
             }
         end
 
-      %{"headers" => %{"event" => "move-out", "patterns" => patterns}} ->
+      %{"headers" => %{"event" => "move-out", "patterns" => patterns} = headers} ->
         patterns =
           Enum.map(patterns, fn %{"pos" => pos, "value" => value} ->
             %{pos: pos, value: value}
           end)
 
-        %{headers: %{event: "move-out", patterns: patterns}}
+        %{headers: %{event: "move-out", patterns: patterns, txids: Map.get(headers, "txids", [])}}
 
-      %{"headers" => %{"event" => "move-in", "patterns" => patterns}} ->
+      %{"headers" => %{"event" => "move-in", "patterns" => patterns} = headers} ->
         patterns =
           Enum.map(patterns, fn %{"pos" => pos, "value" => value} ->
             %{pos: pos, value: value}
           end)
 
-        %{headers: %{event: "move-in", patterns: patterns}}
+        %{headers: %{event: "move-in", patterns: patterns, txids: Map.get(headers, "txids", [])}}
     end)
   end
 
@@ -390,16 +391,28 @@ defmodule Electric.Shapes.Consumer.Materializer do
     Eval.Env.const_to_pg_string(Eval.Env.new(), value, type)
   end
 
-  defp apply_and_accumulate_events(changes, state) do
+  defp apply_and_accumulate_events(changes, xid, state) do
     {state, events} = apply_changes(changes, state)
+    events = with_txids(events, xid)
     %{state | pending_events: merge_events(state.pending_events, events)}
   end
 
+  defp with_txids(events, _xid) when events == %{}, do: events
+
+  defp with_txids(events, xid) do
+    Map.put(events, :txids, xid_set(xid))
+  end
+
+  defp xid_set(nil), do: MapSet.new()
+  defp xid_set(xid) when is_integer(xid) and xid > 0, do: MapSet.new([xid])
+
   defp maybe_flush_pending_events(state, true) do
     events =
       cancel_matching_move_events(state.pending_events)
 
     if events != %{} do
+      events = finalize_txids(events)
+
       for pid <- state.subscribers do
         send(pid, {:materializer_changes, state.shape_handle, events})
       end
@@ -412,13 +425,19 @@ defmodule Electric.Shapes.Consumer.Materializer do
 
   defp maybe_flush_pending_events(state, _commit?), do: state
 
+  defp finalize_txids(events) do
+    Map.update(events, :txids, [], &Enum.sort(&1))
+  end
+
   defp merge_events(pending, new) when pending == %{}, do: new
   defp merge_events(pending, new) when new == %{}, do: pending
 
   defp merge_events(pending, new) do
     %{
       move_in: Map.get(new, :move_in, []) ++ Map.get(pending, :move_in, []),
-      move_out: Map.get(new, :move_out, []) ++ Map.get(pending, :move_out, [])
+      move_out: Map.get(new, :move_out, []) ++ Map.get(pending, :move_out, []),
+      txids:
+        MapSet.union(Map.get(pending, :txids, MapSet.new()), Map.get(new, :txids, MapSet.new()))
     }
   end
 
@@ -433,7 +452,11 @@ defmodule Electric.Shapes.Consumer.Materializer do
   defp cancel_matching_move_events(events) do
     ins = events |> Map.get(:move_in, []) |> Enum.sort_by(fn {v, _} -> v end)
     outs = events |> Map.get(:move_out, []) |> Enum.sort_by(fn {v, _} -> v end)
-    cancel_sorted_pairs(ins, outs, %{move_in: [], move_out: []})
+
+    case cancel_sorted_pairs(ins, outs, %{move_in: [], move_out: []}) do
+      empty when empty == %{} -> empty
+      result -> Map.put(result, :txids, Map.get(events, :txids, MapSet.new()))
+    end
   end
 
   defp cancel_sorted_pairs([{v, _} | ins], [{v, _} | outs], acc),

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

@@ -22,6 +22,7 @@ defmodule Electric.Shapes.Consumer.Subqueries.ActiveMove do
     :values,
     :views_before_move,
     :views_after_move,
+    txids: [],
     snapshot: nil,
     move_in_snapshot_name: nil,
     move_in_row_count: nil,
@@ -40,6 +41,7 @@ defmodule Electric.Shapes.Consumer.Subqueries.ActiveMove do
           values: [move_value()],
           views_before_move: Views.t(),
           views_after_move: Views.t(),
+          txids: [non_neg_integer()],
           snapshot: {term(), term(), [term()]} | nil,
           move_in_snapshot_name: String.t() | nil,
           move_in_row_count: non_neg_integer() | nil,
@@ -51,14 +53,22 @@ defmodule Electric.Shapes.Consumer.Subqueries.ActiveMove do
           buffered_txns: [Transaction.t()]
         }
 
-  @spec start(Views.t(), non_neg_integer(), :move_in | :move_out, [String.t()], [move_value()]) ::
-          t()
-  def start(views, dep_index, dep_move_kind, subquery_ref, values) when is_map(views) do
+  @spec start(
+          Views.t(),
+          non_neg_integer(),
+          :move_in | :move_out,
+          [String.t()],
+          [move_value()],
+          [non_neg_integer()]
+        ) :: t()
+  def start(views, dep_index, dep_move_kind, subquery_ref, values, txids \\ [])
+      when is_map(views) do
     %__MODULE__{
       dep_index: dep_index,
       dep_move_kind: dep_move_kind,
       subquery_ref: subquery_ref,
       values: values,
+      txids: txids,
       views_before_move: views,
       views_after_move: Views.apply_move(views, subquery_ref, values, dep_move_kind)
     }

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

@@ -57,7 +57,7 @@ defmodule Electric.Shapes.Consumer.Subqueries.IndexChanges do
   alias Electric.Shapes.Consumer.Effects
   alias Electric.Shapes.DnfPlan
 
-  @type move :: {:move_in | :move_out, non_neg_integer(), list()}
+  @type move :: {:move_in | :move_out, non_neg_integer(), list(), [non_neg_integer()]}
 
   @doc """
   Returns index effects to apply when a dependency move event starts buffering.
@@ -68,7 +68,7 @@ defmodule Electric.Shapes.Consumer.Subqueries.IndexChanges do
   """
   @spec effects_for_buffering(DnfPlan.t(), move(), [String.t()]) ::
           [Effects.AddToSubqueryIndex.t() | Effects.RemoveFromSubqueryIndex.t()]
-  def effects_for_buffering(dnf_plan, {dep_move_kind, dep_index, values}, subquery_ref) do
+  def effects_for_buffering(dnf_plan, {dep_move_kind, dep_index, values, _txids}, subquery_ref) do
     polarity = Map.get(dnf_plan.dependency_polarities, dep_index, :positive)
 
     case {polarity, dep_move_kind} do
@@ -104,7 +104,7 @@ defmodule Electric.Shapes.Consumer.Subqueries.IndexChanges do
   """
   @spec effects_for_complete(DnfPlan.t(), move(), [String.t()]) ::
           [Effects.AddToSubqueryIndex.t() | Effects.RemoveFromSubqueryIndex.t()]
-  def effects_for_complete(dnf_plan, {dep_move_kind, dep_index, values}, subquery_ref) do
+  def effects_for_complete(dnf_plan, {dep_move_kind, dep_index, values, _txids}, subquery_ref) do
     polarity = Map.get(dnf_plan.dependency_polarities, dep_index, :positive)
 
     case {polarity, dep_move_kind} do

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

@@ -7,6 +7,7 @@ defmodule Electric.Shapes.Consumer.Subqueries.MoveBroadcast do
   alias Electric.Shapes.SubqueryTags
 
   @type move_value() :: {term(), term()}
+  @type txid() :: pos_integer()
   @type move() :: %{dep_index: non_neg_integer(), values: [move_value()]}
 
   @spec effect_for_move_in(move(), ShapeInfo.t()) :: %Effects.AppendControl{}
@@ -17,22 +18,28 @@ defmodule Electric.Shapes.Consumer.Subqueries.MoveBroadcast do
           shape_info.dnf_plan,
           active_move.dep_index,
           active_move.values,
+          active_move.txids,
           "move-in",
           shape_info.stack_id,
           shape_info.shape_handle
         )
     }
   end
 
-  @spec effect_for_move_out(non_neg_integer(), [move_value()], ShapeInfo.t()) ::
-          %Effects.AppendControl{}
-  def effect_for_move_out(dep_index, values, %ShapeInfo{} = shape_info) do
+  @spec effect_for_move_out(
+          non_neg_integer(),
+          [move_value()],
+          [txid()],
+          ShapeInfo.t()
+        ) :: %Effects.AppendControl{}
+  def effect_for_move_out(dep_index, values, txids, %ShapeInfo{} = shape_info) do
     %Effects.AppendControl{
       message:
         make(
           shape_info.dnf_plan,
           dep_index,
           values,
+          txids,
           "move-out",
           shape_info.stack_id,
           shape_info.shape_handle
@@ -44,11 +51,12 @@ defmodule Electric.Shapes.Consumer.Subqueries.MoveBroadcast do
           DnfPlan.t(),
           non_neg_integer(),
           [move_value()],
+          [txid()],
           String.t(),
           String.t(),
           String.t()
         ) :: map()
-  defp make(plan, dep_index, values, event, stack_id, shape_handle)
+  defp make(plan, dep_index, values, txids, event, stack_id, shape_handle)
        when event in ["move-in", "move-out"] do
     positions = Map.get(plan.dependency_positions, dep_index, [])
 
@@ -61,7 +69,7 @@ defmodule Electric.Shapes.Consumer.Subqueries.MoveBroadcast do
         end)
       end)
 
-    %{headers: %{event: event, patterns: patterns}}
+    %{headers: %{event: event, patterns: patterns, txids: txids}}
   end
 
   defp make_hash(%{tag_columns: [_col]}, stack_id, shape_handle, value) do