addressing feedback

Author: icehaunter

packages/sync-service/lib/electric/postgres/snapshot_query.ex

@@ -1,4 +1,5 @@
 defmodule Electric.Postgres.SnapshotQuery do
+  alias Electric.SnapshotError
   alias Electric.Shapes.Shape
   alias Electric.Telemetry.OpenTelemetry
 
@@ -72,6 +73,9 @@ defmodule Electric.Postgres.SnapshotQuery do
       end,
       timeout: :infinity
     )
+  catch
+    :exit, {_, {DBConnection.Holder, :checkout, _}} ->
+      raise SnapshotError.connection_not_available()
   end
 
   defp shape_attrs(shape_handle, shape) do

packages/sync-service/lib/electric/shape_cache/pure_file_storage/file_info.ex

@@ -84,4 +84,8 @@ defmodule Electric.ShapeCache.PureFileStorage.FileInfo do
       :file.truncate(file)
     end)
   end
+
+  def delete(path) do
+    :prim_file.delete(path)
+  end
 end

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

@@ -1,8 +1,8 @@
 defmodule Electric.Shapes.Consumer do
   use GenServer, restart: :temporary
 
-  alias Electric.Shapes.Consumer.MoveHandlingState
-  alias Electric.Shapes.Consumer.InitialSnapshotState
+  alias Electric.Shapes.Consumer.MoveIns
+  alias Electric.Shapes.Consumer.InitialSnapshot
   alias Electric.Shapes.Consumer.MoveHandling
   alias Electric.Shapes.Consumer.State
 
@@ -437,7 +437,7 @@ defmodule Electric.Shapes.Consumer do
     do: {:cont, State.add_to_buffer(state, txn)}
 
   defp handle_txn(txn, state) when needs_initial_filtering(state) do
-    case InitialSnapshotState.filter(state.initial_snapshot_state, state.storage, txn) do
+    case InitialSnapshot.filter(state.initial_snapshot_state, state.storage, txn) do
       {:consider_flushed, initial_snapshot_state} ->
         {:cont, consider_flushed(%{state | initial_snapshot_state: initial_snapshot_state}, txn)}
 
@@ -455,7 +455,7 @@ defmodule Electric.Shapes.Consumer do
          txn,
          %State{move_handling_state: move_handling_state} = state
        ) do
-    case MoveHandlingState.check_txn(move_handling_state, txn) do
+    case MoveIns.check_txn(move_handling_state, txn) do
       {:start_buffering, new_move_handling_state} ->
         handle_txn(txn, %{state | move_handling_state: new_move_handling_state, buffering?: true})
 
@@ -666,7 +666,7 @@ defmodule Electric.Shapes.Consumer do
          change_acc,
          total_ops
        ) do
-    if not MoveHandlingState.change_already_visible?(filter_state, xid, change) do
+    if not MoveIns.change_already_visible?(filter_state, xid, change) do
       case Shape.convert_change(shape, change,
              extra_refs: extra_refs,
              stack_id: stack_id,

packages/sync-service/lib/electric/shapes/consumer/initial_snapshot_state.ex renamed to packages/sync-service/lib/electric/shapes/consumer/initial_snapshot.ex

@@ -1,4 +1,4 @@
-defmodule Electric.Shapes.Consumer.InitialSnapshotState do
+defmodule Electric.Shapes.Consumer.InitialSnapshot do
   @moduledoc false
   # Internal module, used as a part of the consumer state, dealing
   # with the initial snapshot state and the waiting for the snapshot to start.

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

@@ -149,6 +149,11 @@ defmodule Electric.Shapes.Consumer.Materializer do
           end
 
         %{"headers" => %{"event" => "move-out", "patterns" => patterns}} ->
+          patterns =
+            Enum.map(patterns, fn %{"pos" => pos, "value" => value} ->
+              %{pos: pos, value: value}
+            end)
+
           %{headers: %{event: "move-out", patterns: patterns}}
       end)
       |> apply_changes(state)
@@ -375,21 +380,21 @@ defmodule Electric.Shapes.Consumer.Materializer do
 
   defp add_row_to_tag_indices(tag_indices, key, move_tags) do
     # For now we only support one move tag per row (i.e. no `OR`s in the where clause if there's a subquery)
-    Enum.reduce(move_tags, tag_indices, fn [val1], acc ->
-      Map.update(acc, val1, MapSet.new([key]), &MapSet.put(&1, key))
+    Enum.reduce(move_tags, tag_indices, fn tag, acc when is_binary(tag) ->
+      Map.update(acc, tag, MapSet.new([key]), &MapSet.put(&1, key))
     end)
   end
 
   defp remove_row_from_tag_indices(tag_indices, key, move_tags) do
-    Enum.reduce(move_tags, tag_indices, fn [val1], acc ->
-      case Map.fetch(acc, val1) do
+    Enum.reduce(move_tags, tag_indices, fn tag, acc when is_binary(tag) ->
+      case Map.fetch(acc, tag) do
         {:ok, v} ->
           new_mapset = MapSet.delete(v, key)
 
           if MapSet.size(new_mapset) == 0 do
-            Map.delete(acc, val1)
+            Map.delete(acc, tag)
           else
-            Map.put(acc, val1, new_mapset)
+            Map.put(acc, tag, new_mapset)
           end
 
         :error ->
@@ -400,8 +405,9 @@ defmodule Electric.Shapes.Consumer.Materializer do
 
   defp pop_keys_from_tag_indices(tag_indices, patterns) do
     # This implementation is naive while we support only one tag per row and no composite tags.
-    Enum.reduce(patterns, {MapSet.new(), tag_indices}, fn [val1], {keys, acc} ->
-      case Map.pop(acc, val1) do
+    Enum.reduce(patterns, {MapSet.new(), tag_indices}, fn %{pos: _pos, value: value},
+                                                          {keys, acc} ->
+      case Map.pop(acc, value) do
         {nil, acc} -> {keys, acc}
         {v, acc} -> {MapSet.union(keys, v), acc}
       end

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

@@ -5,7 +5,7 @@ defmodule Electric.Shapes.Consumer.MoveHandling do
   alias Electric.Shapes.PartialModes
   alias Electric.Shapes.Shape
   alias Electric.Shapes.Shape.SubqueryMoves
-  alias Electric.Shapes.Consumer.MoveHandlingState
+  alias Electric.Shapes.Consumer.MoveIns
 
   @spec process_move_ins(State.t(), Shape.handle(), list(term())) :: State.t()
   def process_move_ins(state, _, []), do: state
@@ -42,7 +42,7 @@ defmodule Electric.Shapes.Consumer.MoveHandling do
       )
 
     move_handling_state =
-      MoveHandlingState.add_waiting_move_in(state.move_handling_state, name, pg_snapshot)
+      MoveIns.add_waiting(state.move_handling_state, name, pg_snapshot)
 
     %{state | move_handling_state: move_handling_state}
   end
@@ -53,14 +53,21 @@ defmodule Electric.Shapes.Consumer.MoveHandling do
 
   def process_move_outs(state, dep_handle, removed_values) do
     message =
-      SubqueryMoves.make_move_out_control_message(state.shape, [{dep_handle, removed_values}])
+      SubqueryMoves.make_move_out_control_message(
+        state.shape,
+        state.stack_id,
+        state.shape_handle,
+        [
+          {dep_handle, removed_values}
+        ]
+      )
 
     {{_, upper_bound}, writer} = Storage.append_control_message!(message, state.writer)
 
     {%{state | writer: writer}, {[message], upper_bound}}
   end
 
-  @spec query_complete(State.t(), MoveHandlingState.move_in_name(), list(String.t())) ::
+  @spec query_complete(State.t(), MoveIns.move_in_name(), list(String.t())) ::
           {State.t(), notification :: term()}
   def query_complete(%State{} = state, name, key_set) do
     # 1. Splice the stored data into the main log
@@ -69,7 +76,7 @@ defmodule Electric.Shapes.Consumer.MoveHandling do
 
     # 2. Remove this entry from "waiting" and add to "filtering"
     move_handling_state =
-      MoveHandlingState.change_move_in_to_filtering(state.move_handling_state, name, key_set)
+      MoveIns.change_to_filtering(state.move_handling_state, name, key_set)
 
     state = %{state | move_handling_state: move_handling_state, writer: writer}
 

packages/sync-service/lib/electric/shapes/consumer/move_handling_state.ex renamed to packages/sync-service/lib/electric/shapes/consumer/move_ins.ex

@@ -1,4 +1,4 @@
-defmodule Electric.Shapes.Consumer.MoveHandlingState do
+defmodule Electric.Shapes.Consumer.MoveIns do
   alias Electric.Replication.Changes
   alias Electric.Replication.Changes.Transaction
   alias Electric.Postgres.Xid
@@ -25,8 +25,8 @@ defmodule Electric.Shapes.Consumer.MoveHandlingState do
   Add information about a new move-in to the state for which we're waiting
   and update the buffering boundary.
   """
-  @spec add_waiting_move_in(t(), move_in_name(), pg_snapshot()) :: t()
-  def add_waiting_move_in(%__MODULE__{waiting_move_ins: waiting_move_ins} = state, name, snapshot) do
+  @spec add_waiting(t(), move_in_name(), pg_snapshot()) :: t()
+  def add_waiting(%__MODULE__{waiting_move_ins: waiting_move_ins} = state, name, snapshot) do
     new_waiting_move_ins = Map.put(waiting_move_ins, name, snapshot)
     new_buffering_snapshot = make_move_in_buffering_snapshot(new_waiting_move_ins)
 
@@ -54,8 +54,8 @@ defmodule Electric.Shapes.Consumer.MoveHandlingState do
   @doc """
   Change a move-in from "waiting" to "filtering" and update the buffering boundary.
   """
-  @spec change_move_in_to_filtering(t(), move_in_name(), list(String.t())) :: t()
-  def change_move_in_to_filtering(%__MODULE__{} = state, name, key_set) do
+  @spec change_to_filtering(t(), move_in_name(), list(String.t())) :: t()
+  def change_to_filtering(%__MODULE__{} = state, name, key_set) do
     {snapshot, waiting_move_ins} = Map.pop!(state.waiting_move_ins, name)
     filtering_move_ins = [{snapshot, key_set} | state.filtering_move_ins]
     buffering_snapshot = make_move_in_buffering_snapshot(waiting_move_ins)
@@ -77,8 +77,8 @@ defmodule Electric.Shapes.Consumer.MoveHandlingState do
   Filtering generally is applied only to transactions that are already visible
   in the snapshot, and those can only be with `xid < xmax`.
   """
-  @spec remove_completed_move_ins(t(), Transaction.t()) :: t()
-  def remove_completed_move_ins(%__MODULE__{} = state, %Transaction{xid: xid}) do
+  @spec remove_completed(t(), Transaction.t()) :: t()
+  def remove_completed(%__MODULE__{} = state, %Transaction{xid: xid}) do
     state.filtering_move_ins
     |> Enum.reject(fn {snapshot, _} -> Xid.after_snapshot?(xid, snapshot) end)
     |> then(&%{state | filtering_move_ins: &1})
@@ -92,7 +92,7 @@ defmodule Electric.Shapes.Consumer.MoveHandlingState do
   @spec check_txn(t(), Transaction.t()) :: {:continue, t()} | {:start_buffering, t()}
   def check_txn(%__MODULE__{move_in_buffering_snapshot: snapshot} = state, %Transaction{} = txn) do
     if is_nil(snapshot) or Transaction.visible_in_snapshot?(txn, snapshot) do
-      state = remove_completed_move_ins(state, txn)
+      state = remove_completed(state, txn)
       {:continue, state}
     else
       {:start_buffering, state}

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

@@ -1,7 +1,7 @@
 defmodule Electric.Shapes.Consumer.State do
   @moduledoc false
-  alias Electric.Shapes.Consumer.MoveHandlingState
-  alias Electric.Shapes.Consumer.InitialSnapshotState
+  alias Electric.Shapes.Consumer.MoveIns
+  alias Electric.Shapes.Consumer.InitialSnapshot
   alias Electric.Shapes.Shape
   alias Electric.Replication.Changes.Transaction
   alias Electric.Postgres.SnapshotQuery
@@ -18,8 +18,8 @@ defmodule Electric.Shapes.Consumer.State do
     :latest_offset,
     :storage,
     :writer,
-    initial_snapshot_state: InitialSnapshotState.new(nil),
-    move_handling_state: MoveHandlingState.new(),
+    initial_snapshot_state: InitialSnapshot.new(nil),
+    move_handling_state: MoveIns.new(),
     buffer: [],
     txn_offset_mapping: [],
     materializer_subscribed?: false,
@@ -116,11 +116,11 @@ defmodule Electric.Shapes.Consumer.State do
   #       }
 
   defguard is_snapshot_started(state)
-           when is_struct(state.initial_snapshot_state, InitialSnapshotState) and
+           when is_struct(state.initial_snapshot_state, InitialSnapshot) and
                   state.initial_snapshot_state.snapshot_started?
 
   defguard needs_initial_filtering(state)
-           when is_struct(state.initial_snapshot_state, InitialSnapshotState) and
+           when is_struct(state.initial_snapshot_state, InitialSnapshot) and
                   state.initial_snapshot_state.filtering?
 
   @spec new(Electric.stack_id(), Shape.handle(), Shape.t()) :: uninitialized_t()
@@ -156,15 +156,15 @@ defmodule Electric.Shapes.Consumer.State do
         do: LogOffset.last_before_real_offsets(),
         else: latest_offset
 
-    initial_snapshot_state = InitialSnapshotState.new(pg_snapshot)
+    initial_snapshot_state = InitialSnapshot.new(pg_snapshot)
 
     %__MODULE__{
       state
       | latest_offset: normalized_latest_offset,
         storage: storage,
         writer: writer,
         initial_snapshot_state: initial_snapshot_state,
-        buffering?: InitialSnapshotState.needs_buffering?(initial_snapshot_state)
+        buffering?: InitialSnapshot.needs_buffering?(initial_snapshot_state)
     }
   end
 
@@ -195,7 +195,7 @@ defmodule Electric.Shapes.Consumer.State do
   def add_waiter(%__MODULE__{initial_snapshot_state: initial_snapshot_state} = state, from) do
     %{
       state
-      | initial_snapshot_state: InitialSnapshotState.add_waiter(initial_snapshot_state, from)
+      | initial_snapshot_state: InitialSnapshot.add_waiter(initial_snapshot_state, from)
     }
   end
 
@@ -204,25 +204,25 @@ defmodule Electric.Shapes.Consumer.State do
         snapshot
       ) do
     initial_snapshot_state =
-      InitialSnapshotState.set_initial_snapshot(initial_snapshot_state, state.storage, snapshot)
+      InitialSnapshot.set_initial_snapshot(initial_snapshot_state, state.storage, snapshot)
 
     %{
       state
       | initial_snapshot_state: initial_snapshot_state,
-        buffering?: InitialSnapshotState.needs_buffering?(initial_snapshot_state)
+        buffering?: InitialSnapshot.needs_buffering?(initial_snapshot_state)
     }
   end
 
   def mark_snapshot_started(%__MODULE__{initial_snapshot_state: initial_snapshot_state} = state) do
     initial_snapshot_state =
-      InitialSnapshotState.mark_snapshot_started(initial_snapshot_state, state.storage)
+      InitialSnapshot.mark_snapshot_started(initial_snapshot_state, state.storage)
 
     %{state | initial_snapshot_state: initial_snapshot_state}
   end
 
   def reply_to_snapshot_waiters(state, reason) do
     initial_snapshot_state =
-      InitialSnapshotState.reply_to_waiters(state.initial_snapshot_state, reason)
+      InitialSnapshot.reply_to_waiters(state.initial_snapshot_state, reason)
 
     %{state | initial_snapshot_state: initial_snapshot_state}
   end

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

@@ -137,9 +137,17 @@ defmodule Electric.Shapes.Querying do
          shape_handle
        ) do
     additional_headers =
-      Shape.SubqueryMoves.move_in_tag_structure(shape)
-      |> Electric.Utils.deep_map(fn column_name ->
-        "$${md5('#{stack_id}#{shape_handle}' || #{column_name}::text)}"
+      shape.tag_structure
+      |> Enum.map(fn pattern ->
+        Enum.map(pattern, fn
+          column_name when is_binary(column_name) ->
+            "$${md5('#{stack_id}#{shape_handle}' || #{column_name}::text)}"
+
+          {:hash_together, columns} ->
+            column_parts = Enum.map(columns, &~s['#{&1}:' || #{&1}::text])
+            "$${md5('#{stack_id}#{shape_handle}' || #{Enum.join(column_parts, " || ")})}"
+        end)
+        |> Enum.join("/")
       end)
       |> case do
         [] -> additional_headers

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

@@ -633,17 +633,20 @@ defmodule Electric.Shapes.Shape do
     %{change | move_tags: make_tags_from_pattern(tag_structure, record, stack_id, shape_handle)}
   end
 
-  defp make_tags_from_pattern(pattern, record, stack_id, shape_handle) do
-    Utils.deep_map(pattern, fn column_name ->
-      make_value_hash(stack_id, shape_handle, Map.get(record, column_name))
+  defp make_tags_from_pattern(patterns, record, stack_id, shape_handle) do
+    Enum.map(patterns, fn pattern ->
+      Enum.map(pattern, fn
+        column_name when is_binary(column_name) ->
+          SubqueryMoves.make_value_hash(stack_id, shape_handle, Map.get(record, column_name))
+
+        {:hash_together, columns} ->
+          column_parts = Enum.map(columns, &(&1 <> ":" <> Map.get(record, &1)))
+          SubqueryMoves.make_value_hash(stack_id, shape_handle, Enum.join(column_parts, ":"))
+      end)
+      |> Enum.join("/")
     end)
   end
 
-  defp make_value_hash(stack_id, shape_handle, value) do
-    :crypto.hash(:md5, "#{stack_id}#{shape_handle}#{value}")
-    |> Base.encode16(case: :lower)
-  end
-
   defp filtered_columns_changed?(%Changes.UpdatedRecord{old_record: record, record: record}),
     do: false