Fix legacy move-out handling without active conditions

Author: robacourt

packages/elixir-client/lib/electric/client/tag_tracker.ex

@@ -186,25 +186,41 @@ defmodule Electric.Client.TagTracker do
             {deletes, kd_acc, ttk_acc}
 
           %{tags: current_entries, msg: msg} = data ->
-            deactivated_positions =
-              MapSet.new(removed_entries, fn {pos, _} -> pos end)
-
-            updated_ac =
-              data.active_conditions
-              |> Enum.with_index()
-              |> Enum.map(fn {val, idx} ->
-                if MapSet.member?(deactivated_positions, idx), do: false, else: val
-              end)
-
-            visible = row_visible?(updated_ac, disjunct_positions)
-
-            if not visible do
-              ttk_acc = remove_key_from_tags(ttk_acc, current_entries, key)
-
-              {[{key, msg} | deletes], Map.delete(kd_acc, key), ttk_acc}
-            else
-              updated_data = %{data | active_conditions: updated_ac}
-              {deletes, Map.put(kd_acc, key, updated_data), ttk_acc}
+            case data.active_conditions do
+              nil ->
+                # Legacy/simple shapes without active_conditions still use
+                # empty-tag-set deletion semantics rather than DNF visibility.
+                remaining_entries = MapSet.difference(current_entries, removed_entries)
+                ttk_acc = remove_key_from_tags(ttk_acc, removed_entries, key)
+
+                if MapSet.size(remaining_entries) == 0 do
+                  {[{key, msg} | deletes], Map.delete(kd_acc, key), ttk_acc}
+                else
+                  updated_data = %{data | tags: remaining_entries}
+                  {deletes, Map.put(kd_acc, key, updated_data), ttk_acc}
+                end
+
+              active_conditions ->
+                deactivated_positions =
+                  MapSet.new(removed_entries, fn {pos, _} -> pos end)
+
+                updated_ac =
+                  active_conditions
+                  |> Enum.with_index()
+                  |> Enum.map(fn {val, idx} ->
+                    if MapSet.member?(deactivated_positions, idx), do: false, else: val
+                  end)
+
+                visible = row_visible?(updated_ac, disjunct_positions)
+
+                if not visible do
+                  ttk_acc = remove_key_from_tags(ttk_acc, current_entries, key)
+
+                  {[{key, msg} | deletes], Map.delete(kd_acc, key), ttk_acc}
+                else
+                  updated_data = %{data | active_conditions: updated_ac}
+                  {deletes, Map.put(kd_acc, key, updated_data), ttk_acc}
+                end
             end
         end
       end)

packages/elixir-client/test/electric/client/tag_tracker_test.exs

@@ -221,6 +221,46 @@ defmodule Electric.Client.TagTrackerTest do
       assert new_tag_to_keys == %{}
       assert new_key_data == %{}
     end
+
+    test "falls back to empty-tag-set deletion when active_conditions are missing" do
+      msg =
+        make_change_msg("key1", :insert,
+          tags: ["tag_a", "tag_b"],
+          value: %{"id" => "1"}
+        )
+
+      {tag_to_keys, key_data, dp} = TagTracker.update_tag_index(%{}, %{}, nil, msg)
+
+      assert key_data["key1"].active_conditions == nil
+
+      {deletes, tag_to_keys, key_data} =
+        TagTracker.generate_synthetic_deletes(
+          tag_to_keys,
+          key_data,
+          dp,
+          [%{pos: 0, value: "tag_a"}],
+          DateTime.utc_now()
+        )
+
+      assert deletes == []
+      assert key_data["key1"].active_conditions == nil
+      assert key_data["key1"].tags == MapSet.new([{0, "tag_b"}])
+      assert tag_to_keys == %{{0, "tag_b"} => MapSet.new(["key1"])}
+
+      {deletes, tag_to_keys, key_data} =
+        TagTracker.generate_synthetic_deletes(
+          tag_to_keys,
+          key_data,
+          dp,
+          [%{pos: 0, value: "tag_b"}],
+          DateTime.utc_now()
+        )
+
+      assert length(deletes) == 1
+      assert hd(deletes).key == "key1"
+      assert tag_to_keys == %{}
+      assert key_data == %{}
+    end
   end
 
   describe "normalize_tags/1" do

packages/elixir-client/test/electric/client_test.exs

@@ -2401,6 +2401,87 @@ defmodule Electric.ClientTest do
       [delete] = delete_msgs
       assert delete.value["id"] == "1111"
     end
+
+    test "resume preserves legacy move-out state when active_conditions are missing", ctx do
+      body1 =
+        Jason.encode!([
+          %{
+            "key" => "row-1",
+            "headers" => %{
+              "operation" => "insert",
+              "tags" => ["legacy-tag"]
+            },
+            "offset" => "1_0",
+            "value" => %{"id" => "1111"}
+          },
+          %{"headers" => %{"control" => "up-to-date", "global_last_seen_lsn" => 9998}}
+        ])
+
+      schema = Jason.encode!(%{"id" => %{type: "text"}})
+
+      {:ok, responses} =
+        start_supervised(
+          {Agent,
+           fn ->
+             %{
+               {"-1", nil} => [
+                 &bypass_resp(&1, body1,
+                   shape_handle: "my-shape",
+                   last_offset: "1_0",
+                   schema: schema
+                 )
+               ]
+             }
+           end}
+        )
+
+      bypass_response(ctx, responses)
+
+      msgs = stream(ctx, live: false) |> Enum.to_list()
+
+      resume_msg = Enum.find(msgs, &match?(%ResumeMessage{}, &1))
+      assert resume_msg != nil
+
+      body2 =
+        Jason.encode!([
+          %{
+            "headers" => %{
+              "event" => "move-out",
+              "patterns" => [%{"pos" => 0, "value" => "legacy-tag"}]
+            }
+          },
+          %{"headers" => %{"control" => "up-to-date", "global_last_seen_lsn" => 9999}}
+        ])
+
+      {:ok, responses2} =
+        start_supervised(
+          {Agent,
+           fn ->
+             %{
+               {"1_0", "my-shape"} => [
+                 &bypass_resp(&1, body2,
+                   shape_handle: "my-shape",
+                   last_offset: "2_0"
+                 )
+               ]
+             }
+           end},
+          id: :legacy_responses2
+        )
+
+      bypass_response(ctx, responses2)
+
+      resumed_msgs = stream(ctx, resume: resume_msg, live: false) |> Enum.to_list()
+
+      delete_msgs =
+        Enum.filter(resumed_msgs, &match?(%ChangeMessage{headers: %{operation: :delete}}, &1))
+
+      assert length(delete_msgs) == 1,
+             "Legacy move-out after resume should generate synthetic delete, got: #{inspect(resumed_msgs)}"
+
+      [delete] = delete_msgs
+      assert delete.value["id"] == "1111"
+    end
   end
 
   defp bypass_response_endpoint(ctx, responses, opts) do