fix: ensure absence of a materializer doesn't crash SLC (#3640)

Author: icehaunter

.changeset/late-items-appear.md

@@ -0,0 +1,5 @@
+---
+'@core/sync-service': patch
+---
+
+fix: ensure abscense of a materializer doesn't crash a part of electric

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

@@ -55,6 +55,9 @@ defmodule Electric.Shapes.Consumer.Materializer do
 
   def get_link_values(opts) do
     GenServer.call(name(opts), :get_link_values)
+  catch
+    :exit, _reason ->
+      raise ~s|Materializer for stack "#{opts.stack_id}" and handle "#{opts.shape_handle}" is not available|
   end
 
   def get_all_as_refs(shape, stack_id) when are_deps_filled(shape) do

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

@@ -121,14 +121,14 @@ defmodule Electric.Shapes.Filter do
     OpenTelemetry.timed_fun("filter.affected_shapes.duration_µs", fn ->
       try do
         shapes_affected_by_change(filter, change)
-      rescue
-        error ->
+      catch
+        kind, error ->
           Logger.error("""
           Unexpected error in Filter.affected_shapes:
-          #{Exception.format(:error, error, __STACKTRACE__)}
+          #{Exception.format(kind, error, __STACKTRACE__)}
           """)
 
-          OpenTelemetry.record_exception(:error, error, __STACKTRACE__)
+          OpenTelemetry.record_exception(kind, error, __STACKTRACE__)
 
           # We can't tell which shapes are affected, the safest thing to do is return all shapes
           all_shape_ids(filter)

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

@@ -35,7 +35,8 @@ defmodule Electric.Shapes.Partitions do
   table) then this will expand the mapping function to add a change to the
   partition root for every change to a partition of that root.
   """
-  @spec add_shape(t(), shape_id(), Electric.Shapes.Shape.t()) :: t()
+  @spec add_shape(t(), shape_id(), Electric.Shapes.Shape.t()) ::
+          {:ok, t()} | {:error, :connection_not_available}
   def add_shape(%__MODULE__{} = state, shape_id, shape) do
     case Inspector.load_relation_info(shape.root_table_id, state.inspector) do
       {:ok, relation} ->

packages/sync-service/test/electric/replication/shape_log_collector_test.exs

@@ -929,6 +929,108 @@ defmodule Electric.Replication.ShapeLogCollectorTest do
     end
   end
 
+  describe "handle_event/2 with shapes with dependencies" do
+    @shape Shape.new!("test_table", inspector: @inspector)
+    @shape2 Shape.new!("test_table",
+              where: "id IN (SELECT id FROM test_table WHERE id > 10)",
+              inspector: @inspector
+            )
+    setup :setup_log_collector
+
+    setup ctx do
+      parent = self()
+
+      stub_inspector(
+        load_relation_oid: fn {"public", "test_table"}, _ ->
+          {:ok, {1234, {"public", "test_table"}}}
+        end,
+        load_relation_info: fn 1234, _ ->
+          {:ok, %{id: 1234, schema: "public", name: "test_table", parent: nil, children: nil}}
+        end,
+        load_column_info: fn 1234, _ ->
+          {:ok, [%{pk_position: 0, name: "id", is_generated: false}]}
+        end
+      )
+
+      consumers = [
+        {:normal,
+         start_link_supervised!(
+           {Support.TransactionConsumer,
+            id: :normal,
+            stack_id: ctx.stack_id,
+            parent: parent,
+            shape: @shape,
+            shape_handle: "normal-shape-handle"},
+           id: {:consumer, :normal}
+         )},
+        {:inner,
+         start_link_supervised!(
+           {Support.TransactionConsumer,
+            id: :inner,
+            stack_id: ctx.stack_id,
+            parent: parent,
+            shape: @shape2.shape_dependencies |> List.first(),
+            shape_handle: "inner-shape-handle"},
+           id: {:consumer, :inner}
+         )},
+        {:outer,
+         start_link_supervised!(
+           {Support.TransactionConsumer,
+            id: :outer,
+            stack_id: ctx.stack_id,
+            parent: parent,
+            shape: %{@shape2 | shape_dependencies_handles: ["inner-shape-handle"]},
+            shape_handle: "outer-shape-handle"},
+           id: {:consumer, :outer}
+         )}
+      ]
+
+      %{consumers: consumers}
+    end
+
+    test "should handle a transaction that requires a materializer", ctx do
+      start_link_supervised!(
+        {Support.StubMaterializer,
+         stack_id: ctx.stack_id,
+         shape_handle: "inner-shape-handle",
+         initial_values: MapSet.new([])}
+      )
+
+      txn =
+        transaction(100, Lsn.from_string("0/10"), [
+          %Changes.NewRecord{
+            relation: {"public", "test_table"},
+            record: %{"id" => "11"},
+            log_offset: LogOffset.new(Lsn.from_string("0/10"), 0)
+          }
+        ])
+
+      assert :ok = ShapeLogCollector.handle_event(txn, ctx.stack_id)
+
+      # Outer consumer should not receive this, because empty initial values won't satisfy the where clause
+      Support.TransactionConsumer.assert_consume(ctx.consumers |> Keyword.drop([:outer]), [txn])
+    end
+
+    test "should not crash if the materializer is not there, instead skipping the depenencies",
+         ctx do
+      txn =
+        transaction(100, Lsn.from_string("0/10"), [
+          %Changes.NewRecord{
+            relation: {"public", "test_table"},
+            record: %{"id" => "11"},
+            log_offset: LogOffset.new(Lsn.from_string("0/10"), 0)
+          }
+        ])
+
+      assert :ok = ShapeLogCollector.handle_event(txn, ctx.stack_id)
+
+      # This assertion holds because our where clause processing in SLC is best-effort:
+      # any crash, like abscence of a materializer, causes the transaction to just be sent everywhere
+      # and hope the consumers will filter it out.
+      Support.TransactionConsumer.assert_consume(ctx.consumers, [txn])
+    end
+  end
+
   defp transaction(xid, lsn, changes) do
     last_log_offset =
       case Enum.reverse(changes) do

packages/sync-service/test/support/stub_materializer.ex

@@ -0,0 +1,24 @@
+defmodule Support.StubMaterializer do
+  use GenServer, restart: :temporary
+
+  def start_link(opts) do
+    opts = Map.new(opts)
+    GenServer.start_link(__MODULE__, opts, name: Electric.Shapes.Consumer.Materializer.name(opts))
+  end
+
+  def init(opts) do
+    {:ok, %{current_values: Map.get(opts, :initial_values, MapSet.new())}}
+  end
+
+  def set_link_values(name, values) do
+    GenServer.call(Electric.Shapes.Consumer.Materializer.name(name), {:set_link_values, values})
+  end
+
+  def handle_call(:get_link_values, _from, state) do
+    {:reply, state.current_values, state}
+  end
+
+  def handle_call({:set_link_values, values}, _from, state) do
+    {:reply, :ok, %{state | current_values: MapSet.new(values)}}
+  end
+end