fix(sync-service): start missing dependency consumers before their parent

.changeset/start-dependency-consumer-on-demand.md

@@ -0,0 +1,5 @@
+---
+'@core/sync-service': patch
+---
+
+Start the consumer of an existing subquery dependency shape if it isn't running when a new parent shape resolves to it. Previously such a parent request failed with a 500 on every retry: the dependency's materializer found no consumer, the parent was invalidated, and the still-registered dependency poisoned the next attempt.

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

@@ -342,22 +342,14 @@ defmodule Electric.ShapeCache do
     # from the broadcast.
     if not is_nil(otel_ctx), do: OpenTelemetry.set_current_context(otel_ctx)
 
-    case ShapeStatus.fetch_shape_by_handle(state.stack_id, shape_handle) do
-      {:ok, shape} ->
-        {
-          :reply,
-          restore_shape_and_dependencies(shape_handle, shape, %{
-            stack_id: state.stack_id,
-            action: :restore,
-            otel_ctx: otel_ctx,
-            feature_flags: state.feature_flags
-          }),
-          state
-        }
+    opts = %{
+      stack_id: state.stack_id,
+      otel_ctx: otel_ctx,
+      feature_flags: state.feature_flags
+    }
 
-      :error ->
-        {:reply, {:error, :no_shape}, state}
-    end
+    result = try_restoring_shape_and_dependencies(state.stack_id, shape_handle, opts)
+    {:reply, result, state}
   end
 
   defp safe_maybe_create_shape(shape, opts) do
@@ -378,9 +370,13 @@ defmodule Electric.ShapeCache do
     # fetch_handle_by_shape_critical is a slower but guaranteed consistent
     # shape lookup
     with {:ok, shape_handle} <- ShapeStatus.fetch_handle_by_shape_critical(stack_id, shape),
-         {:ok, offset} <- fetch_latest_offset(stack_id, shape_handle) do
+         {:ok, offset} <- fetch_latest_offset(stack_id, shape_handle),
+         :ok <- ensure_inner_consumer_running(shape_handle, opts) do
       {:ok, {shape_handle, offset}}
     else
+      {:error, _reason} = error ->
+        error
+
       :error ->
         with {:ok, shape_handles} <- safe_maybe_create_inner_shapes(shape, opts) do
           shape = %{shape | shape_dependencies_handles: shape_handles}
@@ -403,6 +399,26 @@ defmodule Electric.ShapeCache do
     end
   end
 
+  # An existing inner shape may have no running consumer: a subquery dependency whose parent
+  # was removed survives `ShapeStatus.prune_subquery_shapes/1` on restart as a plain shape,
+  # and consumers for restored shapes only start lazily on transaction routing.
+  defp ensure_inner_consumer_running(
+         shape_handle,
+         %{is_subquery_shape?: true, stack_id: stack_id} = opts
+       ) do
+    consumer_pid = Shapes.ConsumerRegistry.whereis(stack_id, shape_handle)
+
+    if consumer_pid do
+      :ok
+    else
+      with {:ok, _pid} <- try_restoring_shape_and_dependencies(stack_id, shape_handle, opts) do
+        :ok
+      end
+    end
+  end
+
+  defp ensure_inner_consumer_running(_shape_handle, _opts), do: :ok
+
   defp safe_maybe_create_inner_shapes(%Shape{shape_dependencies: []}, _opts) do
     {:ok, []}
   end
@@ -456,6 +472,16 @@ defmodule Electric.ShapeCache do
     end
   end
 
+  defp try_restoring_shape_and_dependencies(stack_id, shape_handle, opts) do
+    case ShapeStatus.fetch_shape_by_handle(stack_id, shape_handle) do
+      {:ok, shape} ->
+        restore_shape_and_dependencies(shape_handle, shape, Map.put(opts, :action, :restore))
+
+      :error ->
+        {:error, :no_shape}
+    end
+  end
+
   # start_shape assumes that any dependent shapes already have running consumers
   # so we need to start those. this may be something we can do lazily: i.e.
   # only starting dependent shapes when they receive a write

packages/sync-service/test/electric/shape_cache_test.exs

@@ -1078,6 +1078,82 @@ defmodule Electric.ShapeCacheTest do
       assert_receive {ShapeCache.ShapeCleaner, :cleanup, ^subshape_handle}
     end
 
+    test "starts the consumer of a registered dependency shape that has none running", ctx do
+      # After restart, an orphaned dependency may remain registered with a completed
+      # snapshot but no consumer. Reusing it must restart the consumer before the
+      # new parent's materializer starts, or the parent is repeatedly invalidated.
+      Support.TestUtils.patch_snapshotter(fn parent,
+                                             shape_handle,
+                                             _shape,
+                                             %{snapshot_fun: snapshot_fun} ->
+        GenServer.cast(parent, {:pg_snapshot_known, shape_handle, @pg_snapshot_xmin_100})
+        snapshot_fun.([])
+        GenServer.cast(parent, {:snapshot_started, shape_handle})
+      end)
+
+      {parent_handle, _} =
+        ShapeCache.get_or_create_shape_handle(@shape_with_subquery, ctx.stack_id)
+
+      assert ShapeCache.await_snapshot_start(parent_handle, ctx.stack_id) == :started
+
+      {:ok, shape} = ShapeCache.fetch_shape_by_handle(parent_handle, ctx.stack_id)
+      [dependency_handle] = shape.shape_dependencies_handles
+
+      # Remove the parent shape, orphaning the dependency.
+      ShapeCache.clean_shape(parent_handle, ctx.stack_id)
+      assert_shape_cleanup(parent_handle)
+
+      # Suspend the dependency without deleting its completed shape, reproducing
+      # the post-restart orphan state.
+      :ok =
+        Electric.Shapes.Consumer.stop(
+          ctx.stack_id,
+          dependency_handle,
+          Electric.ShapeCache.ShapeCleaner.consumer_suspend_reason()
+        )
+
+      assert wait_until(
+               fn ->
+                 is_nil(Electric.Shapes.Consumer.whereis(ctx.stack_id, dependency_handle)) and
+                   is_nil(
+                     GenServer.whereis(
+                       Electric.Shapes.Consumer.Materializer.name(
+                         ctx.stack_id,
+                         dependency_handle
+                       )
+                     )
+                   )
+               end,
+               1_000
+             )
+
+      assert ShapeStatus.has_shape_handle?(ctx.stack_id, dependency_handle)
+      assert ShapeStatus.snapshot_started?(ctx.stack_id, dependency_handle)
+
+      # A new request for the parent shape resolves the subquery to the same
+      # dependency handle. The missing consumer must be started, letting the
+      # parent initialize and the request succeed.
+      {second_parent_handle, _} =
+        ShapeCache.get_or_create_shape_handle(@shape_with_subquery, ctx.stack_id)
+
+      refute second_parent_handle == parent_handle
+
+      assert ShapeCache.await_snapshot_start(second_parent_handle, ctx.stack_id) == :started
+
+      {:ok, second_shape} =
+        ShapeCache.fetch_shape_by_handle(second_parent_handle, ctx.stack_id)
+
+      assert second_shape.shape_dependencies_handles == [dependency_handle]
+
+      dependency_pid = Electric.Shapes.Consumer.whereis(ctx.stack_id, dependency_handle)
+      assert is_pid(dependency_pid) and Process.alive?(dependency_pid)
+
+      # The restarted consumer must be initialized as a subquery (inner) shape, i.e.
+      # buffer whole transactions rather than stream fragments, like any other
+      # dependency consumer.
+      assert %{write_unit: :txn} = :sys.get_state(dependency_pid)
+    end
+
     test "should wait for consumer to come up", ctx do
       Support.TestUtils.patch_snapshotter(fn parent, shape_handle, _, _ ->
         GenServer.cast(parent, {:pg_snapshot_known, shape_handle, @pg_snapshot_xmin_100})