Subscribe and unsubscribe to LSN updates

Author: robacourt

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

@@ -17,7 +17,7 @@ defmodule Electric.Shapes.Consumer do
   alias Electric.Shapes.Consumer.Materializer
   alias Electric.Shapes.ConsumerRegistry
   alias Electric.LogItems
-  alias Electric.LsnTracker
+
   alias Electric.Postgres.Inspector
   alias Electric.Replication.Changes
   alias Electric.Replication.Changes.Transaction
@@ -126,7 +126,6 @@ defmodule Electric.Shapes.Consumer do
     metadata = [shape_handle: shape_handle, stack_id: stack_id]
     Logger.metadata(metadata)
     Electric.Telemetry.Sentry.set_tags_context(metadata)
-    {:ok, _} = LsnTracker.subscribe_to_global_lsn_updates(stack_id)
 
     # Shape initialization will be complete when we receive a message {:initialize_shape,
     # <shape>, <shape_opts>} which the ShapeCache is expected to send as soon as this process

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

@@ -5,4 +5,14 @@ defmodule Electric.Shapes.Consumer.Effect do
     @moduledoc false
     defstruct []
   end
+
+  defmodule SubscribeGlobalLsn do
+    @moduledoc false
+    defstruct []
+  end
+
+  defmodule UnsubscribeGlobalLsn do
+    @moduledoc false
+    defstruct []
+  end
 end

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

@@ -22,8 +22,7 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Buffering do
     :trigger_dep_index,
     :move_in_values,
     :views_before_move,
-    :views_after_move,
-    :latest_seen_lsn
+    :views_after_move
   ]
   defstruct [
     :shape,
@@ -98,7 +97,6 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Buffering do
       views_before_move: state.views,
       views_after_move: views_after,
       dependency_handle_to_ref: state.dependency_handle_to_ref,
-      latest_seen_lsn: state.latest_seen_lsn,
       queue: queue,
       buffer_max_transactions: state.buffer_max_transactions
     }
@@ -239,7 +237,11 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Buffering do
           txns -> txns |> List.last() |> Map.fetch!(:last_log_offset)
         end
 
-      plan = %Plan{log_ops: log_ops, ack_source_offset: ack_offset}
+      plan = %Plan{
+        log_ops: log_ops,
+        effects: [%Electric.Shapes.Consumer.Effect.UnsubscribeGlobalLsn{}],
+        ack_source_offset: ack_offset
+      }
 
       # Transition back to steady state, then drain any queued moves
       state
@@ -280,7 +282,6 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Buffering do
       dnf_plan: state.dnf_plan,
       views: state.views_after_move,
       dependency_handle_to_ref: state.dependency_handle_to_ref,
-      latest_seen_lsn: state.latest_seen_lsn,
       queue: state.queue,
       buffer_max_transactions: state.buffer_max_transactions
     }

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

@@ -19,7 +19,6 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Steady do
     :dnf_plan,
     views: %{},
     dependency_handle_to_ref: %{},
-    latest_seen_lsn: nil,
     queue: MoveQueue.new(),
     buffer_max_transactions: 1000
   ]
@@ -31,7 +30,6 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Steady do
           dnf_plan: DnfPlan.t(),
           views: %{[String.t()] => MapSet.t()},
           dependency_handle_to_ref: %{String.t() => {non_neg_integer(), [String.t()]}},
-          latest_seen_lsn: Electric.Postgres.Lsn.t() | nil,
           queue: MoveQueue.t(),
           buffer_max_transactions: pos_integer()
         }
@@ -55,8 +53,9 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Steady do
     end
   end
 
-  def handle_event(state, {:global_last_seen_lsn, lsn}) do
-    {:ok, %{state | latest_seen_lsn: Subqueries.normalize_global_lsn(lsn)}, %Plan{}}
+  def handle_event(state, {:global_last_seen_lsn, _lsn}) do
+    # Straggler message after unsubscribe; ignore.
+    {:ok, state, %Plan{}}
   end
 
   def handle_event(state, {:materializer_changes, dep_handle, payload}) do
@@ -126,7 +125,12 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Steady do
 
             plan = %{
               plan
-              | effects: plan.effects ++ [%Electric.Shapes.Consumer.Effect.StartMoveInQuery{}]
+              | effects:
+                  plan.effects ++
+                    [
+                      %Electric.Shapes.Consumer.Effect.SubscribeGlobalLsn{},
+                      %Electric.Shapes.Consumer.Effect.StartMoveInQuery{}
+                    ]
             }
 
             {:ok, buffering, plan}
@@ -150,7 +154,12 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Steady do
 
             plan = %{
               plan
-              | effects: plan.effects ++ [%Electric.Shapes.Consumer.Effect.StartMoveInQuery{}]
+              | effects:
+                  plan.effects ++
+                    [
+                      %Electric.Shapes.Consumer.Effect.SubscribeGlobalLsn{},
+                      %Electric.Shapes.Consumer.Effect.StartMoveInQuery{}
+                    ]
             }
 
             {:ok, buffering, plan}

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

@@ -16,4 +16,6 @@ defmodule Electric.Shapes.Consumer.Plan do
 
   @type effect() ::
           %Electric.Shapes.Consumer.Effect.StartMoveInQuery{}
+          | %Electric.Shapes.Consumer.Effect.SubscribeGlobalLsn{}
+          | %Electric.Shapes.Consumer.Effect.UnsubscribeGlobalLsn{}
 end

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

@@ -123,6 +123,16 @@ defmodule Electric.Shapes.Consumer.PlanExecutor do
     acc
   end
 
+  defp execute_effect(%Effect.SubscribeGlobalLsn{}, acc) do
+    {:ok, _} = Electric.LsnTracker.subscribe_to_global_lsn_updates(acc.state.stack_id)
+    acc
+  end
+
+  defp execute_effect(%Effect.UnsubscribeGlobalLsn{}, acc) do
+    :ok = Electric.LsnTracker.unsubscribe_from_global_lsn_updates(acc.state.stack_id)
+    acc
+  end
+
   # -- Ack --
 
   defp apply_ack(acc, nil), do: acc

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

@@ -73,7 +73,7 @@ defmodule Electric.Shapes.Consumer.EventHandlerTest do
                  {:materializer_changes, dep_handle, %{move_in: [], move_out: [{1, "1"}]}}
                )
 
-      assert %Plan{effects: [%Effect.StartMoveInQuery{}]} = plan
+      assert %Plan{effects: [%Effect.SubscribeGlobalLsn{}, %Effect.StartMoveInQuery{}]} = plan
 
       assert %Buffering{
                views_before_move: %{["$sublink", "0"] => before_view},
@@ -109,7 +109,8 @@ defmodule Electric.Shapes.Consumer.EventHandlerTest do
       handler = new_handler()
       dep_handle = dep_handle(handler)
 
-      assert {:ok, %Buffering{} = handler, %Plan{effects: [%Effect.StartMoveInQuery{}]}} =
+      assert {:ok, %Buffering{} = handler,
+              %Plan{effects: [%Effect.SubscribeGlobalLsn{}, %Effect.StartMoveInQuery{}]}} =
                EventHandler.handle_event(
                  handler,
                  {:materializer_changes, dep_handle, %{move_in: [{1, "1"}], move_out: []}}
@@ -326,38 +327,6 @@ defmodule Electric.Shapes.Consumer.EventHandlerTest do
              } = plan
     end
 
-    test "uses an lsn update that was already seen before the move-in started" do
-      handler = new_handler()
-      dep_handle = dep_handle(handler)
-
-      assert {:ok, handler, %Plan{}} =
-               EventHandler.handle_event(handler, global_last_seen_lsn(20))
-
-      assert {:ok, %Buffering{} = handler, _plan} =
-               EventHandler.handle_event(
-                 handler,
-                 {:materializer_changes, dep_handle, %{move_in: [{1, "1"}], move_out: []}}
-               )
-
-      assert {:ok, %Buffering{} = handler, _plan} =
-               EventHandler.handle_event(handler, {:pg_snapshot_known, {100, 300, []}})
-
-      assert {:ok, %Steady{views: views}, plan} =
-               EventHandler.handle_event(
-                 handler,
-                 {:query_move_in_complete, [child_insert("99", "1")], lsn(20)}
-               )
-
-      assert views[["$sublink", "0"]] == MapSet.new([1])
-
-      assert %Plan{
-               log_ops: [
-                 %LogOp.AppendControl{message: %{headers: %{event: "move-in"}}},
-                 %LogOp.AppendMoveInSnapshot{rows: [%Changes.NewRecord{record: %{"id" => "99"}}]}
-               ]
-             } = plan
-    end
-
     test "defers queued move outs until after splice and starts the next move in" do
       handler = new_handler()
       dep_handle = dep_handle(handler)
@@ -403,7 +372,11 @@ defmodule Electric.Shapes.Consumer.EventHandlerTest do
                    message: %{headers: %{event: "move-out", patterns: [%{pos: 0}]}}
                  }
                ],
-               effects: [%Effect.StartMoveInQuery{}]
+               effects: [
+                 %Effect.UnsubscribeGlobalLsn{},
+                 %Effect.SubscribeGlobalLsn{},
+                 %Effect.StartMoveInQuery{}
+               ]
              } = plan
     end