fix(sync-service): drop subquery shapes on restart instead of restoring them

.changeset/drop-subquery-shapes-on-restart.md

@@ -0,0 +1,5 @@
+---
+'@core/sync-service': patch
+---
+
+Drop shapes that involve subqueries on server restart to prevent consistency issues.

.github/workflows/sync_service_tests.yml

@@ -169,6 +169,55 @@ jobs:
 
       - *upload_test_results_to_codecov
 
+  oracle_property_test_with_restarts:
+    name: 'Oracle property test with restarts (${{ matrix.restart_type }})'
+    runs-on: blacksmith-4vcpu-ubuntu-2404
+    defaults:
+      run:
+        working-directory: packages/sync-service
+    strategy:
+      fail-fast: false
+      matrix:
+        restart_type: [graceful, brutal]
+    env:
+      MIX_ENV: test
+      MIX_TARGET: application
+      POSTGRES_VERSION: '170000'
+      CODECOV_FLAGS: elixir,oracle-tests,sync-service,postgres-170000
+      CODECOV_TEST_RESULTS_FILES: ./junit/regular-test-junit-report.xml
+      CHECK_TIMEOUT: 60000
+      SHAPE_COUNT: 200
+      MUTATIONS_PER_TXN: 10
+      TXNS_PER_BATCH: 10
+      BATCH_COUNT: 50
+      SKIP_REPATCH_PREWARM: 'true'
+      RESTART_SERVER_EVERY: 3
+      RESTART_TYPE: ${{ matrix.restart_type }}
+      TEST_POOL_SIZE: 20
+    services:
+      postgres:
+        image: 'ghcr.io/${{ github.repository }}/postgres:17-alpine-logical'
+        env:
+          POSTGRES_PASSWORD: password
+        options: *postgres_health_check
+        ports:
+          - 54321:5432
+
+      pgbouncer: *pgbouncer_service
+    steps:
+      - *checkout_source
+      - *seed_database
+      - *setup_beam
+      - *cache_dependencies
+      - *cache_compiled_code
+      - *install_dependencies
+      - *compile_package
+
+      - name: Run oracle property test with ${{ matrix.restart_type }} server restarts
+        run: mix test --only oracle test/integration/oracle_property_test.exs
+
+      - *upload_test_results_to_codecov
+
   performance_test:
     name: 'Performance test, pg17'
     runs-on: blacksmith-4vcpu-ubuntu-2404

packages/sync-service/lib/electric/replication/shape_log_collector.ex

@@ -296,6 +296,14 @@ defmodule Electric.Replication.ShapeLogCollector do
       fn ->
         start = System.monotonic_time()
 
+        # Restoring subquery shapes consistently across a restart is not yet
+        # implemented, so for now they are dropped rather than restored and
+        # clients re-request them. This is the first restore path in the shape
+        # subsystem to read from ShapeStatus, so pruning them here — before we
+        # build routing and before the shape consumers start — is the single
+        # point that keeps every restore path from reinstating a subquery shape.
+        :ok = Electric.ShapeCache.ShapeStatus.prune_subquery_shapes(state.stack_id)
+
         {partitions, event_router, layers, count} =
           state.stack_id
           |> Electric.ShapeCache.ShapeStatus.list_shapes()

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

@@ -286,13 +286,10 @@ defmodule Electric.ShapeCache do
 
     Electric.Replication.PublicationManager.wait_for_restore(state.stack_id)
 
-    # Subquery shapes' consumers must be fully initialized before
-    # ShapeLogCollector starts dispatching events. If events flow first,
-    # the materializer can advance past the outer shape's on-disk storage;
-    # the outer consumer's later init would then seed `state.views` from
-    # the advanced materializer view and a subsequent move-in event for
-    # a value already in that seeded view would be dropped as redundant.
-    eagerly_start_subquery_shape_consumers(state)
+    # Shapes involved in a subquery are dropped rather than restored on restart
+    # (see `ShapeStatus.prune_subquery_shapes/1`, called from ShapeLogCollector's
+    # restore before any routing or consumer state is rebuilt), so there is
+    # nothing to eagerly start here.
 
     # Let ShapeLogCollector that it can start processing after finishing this function so that
     # we're subscribed to the producer before it starts forwarding its demand.
@@ -313,61 +310,6 @@ defmodule Electric.ShapeCache do
     {:noreply, state}
   end
 
-  # Shapes whose where clause contains a subquery (`shape_dependencies != []`)
-  # rely on their materializer subscription to be notified of dependency-side
-  # changes. The router only delivers events for a shape when its own
-  # `root_table` changes, so a subquery dependent stays dormant after a
-  # restart until something writes to its own table — movements driven by
-  # the dependency (e.g. parent rows becoming active) never reach its
-  # on-disk view. Restoring it here re-establishes the materializer
-  # subscription so dependency updates flow in.
-  #
-  # `await_snapshot_start/2` is queued *after* the consumer's
-  # `:initialize_shape` info message, so by the time it returns
-  # `EventHandlerBuilder.build` has run and `state.views` is seeded.
-  defp eagerly_start_subquery_shape_consumers(state) do
-    opts = %{
-      stack_id: state.stack_id,
-      action: :restore,
-      otel_ctx: nil,
-      feature_flags: state.feature_flags
-    }
-
-    for {handle, %Shape{shape_dependencies: [_ | _]} = shape} <-
-          ShapeStatus.list_shapes(state.stack_id),
-        is_nil(Electric.Shapes.ConsumerRegistry.whereis(state.stack_id, handle)) do
-      case restore_shape_and_dependencies(handle, shape, opts) do
-        {:ok, _pid} ->
-          # await_snapshot_start/2 is a GenServer.call into the just-started
-          # consumer. If that consumer dies before/during the call it exits;
-          # left unguarded that would propagate out of handle_continue and
-          # crash ShapeCache before mark_as_ready — turning a single shape
-          # that reliably fails its snapshot into a stack-wide restart loop.
-          # A call timeout (the consumer is alive but wedged) exits the same
-          # way. In either case we can't confirm the shape's consumer came up
-          # subscribed-and-correct, and the eager start exists precisely to
-          # guarantee that consistency. Leaving the shape alive-but-unconfirmed
-          # would silently reintroduce the divergence this restore path fixes,
-          # so we purge it (mirroring restore_shape_and_dependencies' own
-          # clean_shape-on-failure) and let the client refetch from scratch.
-          try do
-            _ = Electric.Shapes.Consumer.await_snapshot_start(state.stack_id, handle)
-          catch
-            :exit, reason ->
-              Logger.warning(
-                "Eager subquery consumer await failed for #{handle}: #{inspect(reason)}; " <>
-                  "purging shape to force a clean refetch"
-              )
-
-              clean_shape(handle, state.stack_id)
-          end
-
-        _ ->
-          :ok
-      end
-    end
-  end
-
   @impl GenServer
   def handle_call({:create_or_wait_shape_handle, shape, otel_ctx}, _from, state) do
     if not is_nil(otel_ctx), do: OpenTelemetry.set_current_context(otel_ctx)

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

@@ -147,6 +147,64 @@ defmodule Electric.ShapeCache.ShapeStatus do
     end)
   end
 
+  @doc """
+  Given a list of `{handle, shape}` pairs (e.g. from `list_shapes/1`), return the
+  set of handles for every shape involved in a subquery: each shape with a
+  non-empty `shape_dependencies` plus all of its dependency handles.
+
+  This is the transitive closure of the subquery hierarchy — a nested dependency
+  that itself has a subquery matches the same filter and contributes its own
+  dependencies — so it covers outer shapes, intermediate dependencies and leaf
+  materializers. Used by `prune_subquery_shapes/1`.
+  """
+  @spec subquery_shape_handles([{shape_handle(), Shape.t()}]) :: MapSet.t(shape_handle())
+  def subquery_shape_handles(handles_and_shapes) do
+    for {handle, %Shape{shape_dependencies: [_ | _]} = shape} <- handles_and_shapes,
+        h <- [handle | shape.shape_dependencies_handles],
+        into: MapSet.new(),
+        do: h
+  end
+
+  @doc """
+  Remove every shape involved in a subquery (the outer shape plus its dependency
+  materializers) from shape metadata and on-disk storage.
+
+  Correctly restoring a subquery shape's on-disk view together with its
+  dependency materializer across a restart is not yet implemented, so for now we
+  drop every subquery shape on restart and let clients re-request them from
+  scratch. This is called once at the start of the shape subsystem's restore —
+  from `ShapeLogCollector`'s `restore_shapes`, before it (or the shape consumers)
+  rebuild any state from `list_shapes/1` — so no restore path ever reinstates a
+  subquery shape. At that point no consumer or routing entry exists for these
+  shapes yet, so a direct metadata + storage delete is sufficient; there is no
+  consumer to stop or routing entry to clear.
+  """
+  @spec prune_subquery_shapes(stack_id()) :: :ok
+  def prune_subquery_shapes(stack_id) when is_stack_id(stack_id) do
+    handles =
+      stack_id
+      |> list_shapes()
+      |> subquery_shape_handles()
+
+    unless Enum.empty?(handles) do
+      Logger.notice(
+        "Dropping #{MapSet.size(handles)} shape(s) involved in subqueries on restart; " <>
+          "clients will re-request them from scratch"
+      )
+
+      stack_storage = Electric.ShapeCache.Storage.for_stack(stack_id)
+
+      for handle <- handles do
+        case remove_shape(stack_id, handle) do
+          :ok -> Electric.ShapeCache.Storage.cleanup!(stack_storage, handle)
+          {:error, _reason} -> :ok
+        end
+      end
+    end
+
+    :ok
+  end
+
   @spec topological_sort([{shape_handle(), Shape.t()}]) :: [{shape_handle(), Shape.t()}]
   defp topological_sort(handles_and_shapes, acc \\ [], visited \\ MapSet.new())
   defp topological_sort([], acc, _visited), do: Enum.reverse(acc) |> List.flatten()

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

@@ -46,16 +46,6 @@ defmodule Electric.Replication.ShapeLogCollectorTest do
   @shape Shape.new!("test_table", inspector: @inspector)
   @shape_handle "the-shape-handle"
 
-  @subquery_inspector Support.StubInspector.new(
-                        tables: [{1234, {"public", "test_table"}}, {5678, {"public", "parent"}}],
-                        columns: [%{name: "id", type: "int8", type_id: {20, 1}, pk_position: 0}]
-                      )
-  @subquery_shape Shape.new!("test_table",
-                    inspector: @subquery_inspector,
-                    where: "id IN (SELECT id FROM public.parent)"
-                  )
-  @subquery_shape_handle "subquery-shape-handle"
-
   def setup_log_collector(ctx) do
     %{stack_id: stack_id} = ctx
     # Start a test Registry
@@ -235,60 +225,11 @@ defmodule Electric.Replication.ShapeLogCollectorTest do
       assert xids == [xmin]
     end
 
-    @tag restore_shapes: [{@subquery_shape_handle, @subquery_shape}],
-         inspector: @subquery_inspector
-    test "restored subquery shape routes via fallback before consumer seeds index", ctx do
-      alias Electric.Shapes.Filter.Indexes.SubqueryIndex
-
-      # After restore, the subquery shape should be in fallback because
-      # no consumer has seeded the SubqueryIndex yet.
-      index = SubqueryIndex.for_stack(ctx.stack_id)
-      assert index != nil
-      assert SubqueryIndex.fallback?(index, @subquery_shape_handle)
-
-      parent = self()
-
-      consumer =
-        start_link_supervised!(
-          {Support.TransactionConsumer,
-           [
-             id: 1,
-             stack_id: ctx.stack_id,
-             parent: parent,
-             shape: @subquery_shape,
-             shape_handle: @subquery_shape_handle,
-             stack_id: ctx.stack_id,
-             action: :restore
-           ]}
-        )
-
-      :ok =
-        Electric.Shapes.ConsumerRegistry.register_consumer(
-          consumer,
-          @subquery_shape_handle,
-          ctx.stack_id
-        )
-
-      xmin = 100
-      lsn = Lsn.from_string("0/10")
-      last_log_offset = LogOffset.new(lsn, 0)
-
-      # Any root-table change should route to the shape via fallback,
-      # even if the record wouldn't match the subquery membership.
-      txn =
-        complete_txn_fragment(xmin, lsn, [
-          %Changes.NewRecord{
-            relation: {"public", "test_table"},
-            record: %{"id" => "999"},
-            log_offset: last_log_offset
-          }
-        ])
-
-      assert :ok = ShapeLogCollector.handle_event(txn, ctx.stack_id)
-
-      xids = Support.TransactionConsumer.assert_consume([{1, consumer}], [txn])
-      assert xids == [xmin]
-    end
+    # Subquery shapes are pruned (not restored) at the start of the collector's
+    # restore via `ShapeStatus.prune_subquery_shapes/1`. Its effect — the subquery
+    # hierarchy absent from both `ShapeCache.list_shapes/1` and `active_shapes/1`
+    # after a real restart, while plain shapes are retained — is covered by the
+    # "after restart" tests in `shape_cache_test.exs`.
 
     @tag restore_shapes: [{@shape_handle, @shape}, {@shape_handle <> "-2", @shape}],
          inspector: @inspector