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