fix(sync-service): Pass shape directly to consumer process (#3964)

There may have been a time where we didn't have a loaded shape in the
ShapeCache when starting a consumer and to avoid blocking the ShapeCache
process we moved the shape load into the consumer. But that time has
passed and the shape cache always has the Shape instance available when
launching a consumer, even for resuming a previously suspended consumer.

So instead of making the Consumer process load the shape, just include
it in the launch opts and save a query to the ShapeDb.

Most fetches from the ShapeDb are quick, especially for a new shape
where the info is probably still in the in-memory buffer and so the
fetch is just a copy from ets, but there is a very long tail, especially
under load. If the consumer process is not scheduled immediately the
ShapeDb could have time to move the shape data out of the cache and
flush it to disk, which means the shape load has to go via the disk
which can be hit and miss (pun intended).

This should be an easy win as we're removing this unstable path for new
shapes and reducing the shape fetch from twice to once for resumed
consumers.

I haven't gone through the tests to remove instances of starting a
consumer directly, thus not including the shape, hence the
`Map.get_lazy/3` call which falls back to the old shapedb fetch route.

Author: magnetised

.changeset/lovely-icons-sit.md

@@ -0,0 +1,5 @@
+---
+'@core/sync-service': patch
+---
+
+Remove redundant ShapeDb fetch from Consumer initialization

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

@@ -358,6 +358,7 @@ defmodule Electric.ShapeCache do
     start_opts =
       opts
       |> Map.put(:shape_handle, shape_handle)
+      |> Map.put(:shape, shape)
       |> Map.put(:subqueries_enabled_for_stack?, "allow_subqueries" in feature_flags)
 
     case Shapes.DynamicConsumerSupervisor.start_shape_consumer(stack_id, start_opts) do

packages/sync-service/lib/electric/shape_cache/shape_status.ex

@@ -186,6 +186,15 @@ defmodule Electric.ShapeCache.ShapeStatus do
     end)
   end
 
+  @spec fetch_shape_by_handle!(stack_id(), shape_handle()) :: Shape.t() | no_return()
+  def fetch_shape_by_handle!(stack_id, shape_handle)
+      when is_stack_id(stack_id) and is_shape_handle(shape_handle) do
+    case fetch_shape_by_handle(stack_id, shape_handle) do
+      {:ok, shape} -> shape
+      :error -> raise ArgumentError, message: "No shape found for handle #{inspect(shape_handle)}"
+    end
+  end
+
   def has_shape_handle?(stack_id, shape_handle) do
     :ets.member(shape_meta_table(stack_id), shape_handle)
   end

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

@@ -121,7 +121,10 @@ defmodule Electric.Shapes.Consumer do
       shape_handle: shape_handle
     } = state
 
-    {:ok, shape} = ShapeCache.ShapeStatus.fetch_shape_by_handle(stack_id, shape_handle)
+    shape =
+      Map.get_lazy(config, :shape, fn ->
+        ShapeCache.ShapeStatus.fetch_shape_by_handle!(stack_id, shape_handle)
+      end)
 
     state = State.initialize_shape(state, shape, config)