Fix index changes

Author: robacourt

packages/sync-service/lib/electric/shapes/consumer/event_handler/subqueries/index_changes.ex

@@ -20,8 +20,10 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.IndexChanges do
 
   For a negated (`NOT IN`) subquery:
   - Adding values to the index *narrows* the filter (fewer rows match).
-  - So a move-in does **not** update the index at enqueue time (keeping the
-    filter broad); the add is deferred until **complete**.
+  - So a dependency move-in does **not** update the index at enqueue time
+    (keeping the filter broad); the add is deferred until **complete**.
+  - A dependency move-out broadens the filter by removing the value from the
+    index immediately, and that removal remains correct after the splice.
 
   ## Effect tables
 
@@ -31,8 +33,8 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.IndexChanges do
   |------------|----------|---------------------------|
   | move_in    | positive | AddToSubqueryIndex        |
   | move_in    | negated  | *(none)*                  |
-  | move_out   | positive | RemoveFromSubqueryIndex   |
-  | move_out   | negated  | *(none)*                  |
+  | move_out   | positive | *(none)*                  |
+  | move_out   | negated  | RemoveFromSubqueryIndex   |
 
   ### When complete (splice finished, or immediate for non-buffering cases)
 
@@ -58,9 +60,11 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.IndexChanges do
   @type move :: {:move_in | :move_out, non_neg_integer(), list()}
 
   @doc """
-  Returns index effects to apply when a dependency move event is first enqueued.
+  Returns index effects to apply when a dependency move event starts buffering.
 
-  Used by buffering cases to broaden the filter before the move-in query runs.
+  Used only by buffering cases to broaden the filter before the move-in query
+  runs. Calling this for an immediate (non-buffering) move is a bug in the
+  caller.
   """
   @spec effects_for_enqueue(DnfPlan.t(), move(), [String.t()]) ::
           [Effect.AddToSubqueryIndex.t() | Effect.RemoveFromSubqueryIndex.t()]
@@ -77,10 +81,7 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.IndexChanges do
           }
         ]
 
-      {:negated, :move_in} ->
-        []
-
-      {:positive, :move_out} ->
+      {:negated, :move_out} ->
         [
           %Effect.RemoveFromSubqueryIndex{
             dep_index: dep_index,
@@ -89,8 +90,9 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.IndexChanges do
           }
         ]
 
-      {:negated, :move_out} ->
-        []
+      other ->
+        raise ArgumentError,
+              "effects_for_enqueue/3 only supports buffering cases, got #{inspect(other)}"
     end
   end
 

packages/sync-service/test/electric/shapes/consumer/event_handler_test.exs

@@ -72,14 +72,21 @@ defmodule Electric.Shapes.Consumer.EventHandlerTest do
       handler = new_handler(shape: negated_shape(), subquery_view: MapSet.new([1]))
       dep_handle = dep_handle(handler)
 
-      # Case B: negated move-out → no index effect at enqueue (filter stays broad)
+      # Case B: negated move-out → remove the value at enqueue time so the
+      # negated index reflects the post-move exclusion set while buffering.
       assert {:ok, %Buffering{} = handler, plan} =
                EventHandler.handle_event(
                  handler,
                  {:materializer_changes, dep_handle, %{move_in: [], move_out: [{1, "1"}]}}
                )
 
-      assert %Plan{effects: [%Effect.SubscribeGlobalLsn{}, %Effect.StartMoveInQuery{}]} = plan
+      assert %Plan{
+               effects: [
+                 %Effect.RemoveFromSubqueryIndex{dep_index: 0, values: [{1, "1"}]},
+                 %Effect.SubscribeGlobalLsn{},
+                 %Effect.StartMoveInQuery{}
+               ]
+             } = plan
 
       assert %Buffering{
                views_before_move: %{["$sublink", "0"] => before_view},
@@ -98,7 +105,8 @@ defmodule Electric.Shapes.Consumer.EventHandlerTest do
                  {:query_move_in_complete, [child_insert("99", "1")], lsn(10)}
                )
 
-      # Case B: negated move-out → no index effect at complete either
+      # Case B: negated move-out → no further index effect at complete because
+      # the enqueue-time removal already matches the post-splice dependency view.
       assert {:ok, %Steady{views: %{["$sublink", "0"] => view}}, plan} =
                EventHandler.handle_event(handler, global_last_seen_lsn(10))