Removing file writes from the main WAL processing loop (#2470)

Speed up replication processing by removing file writes from the main
processing loop:
- persisting the tracked relations now only happens on a relation
change, not every transaction
- `last_processed_lsn` is no longer persisted, but it is worked out from
the last lsn seen in the shape logs. This also avoids a race condition
where the lsn is written in the shape logs but not the
PersistedReplicationState.

As part of this PR, this also fixes a bug we found where
`last_processed_lsn` was always being set to `Lsn.from_integer(0)` on
start-up rather than reading from disk.

Please note that as part of this PR, `last_processed_lsn` and
`global_last_seen_lsn` no longer include LSNs from transactions that do
not affect any shapes.

Here we can see a 20% increase in WAL processing speed:
<img width="811" alt="Screenshot 2025-03-24 at 15 11 29"
src="https://github.com/user-attachments/assets/ee299705-504d-41a7-a244-57d44bad5e39"
/>

---------

Co-authored-by: msfstef <msfstef@gmail.com>

Author: robacourt

.changeset/honest-crabs-sing.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Speed up replication processing by removing file writes from the main processing loop

packages/sync-service/lib/electric/connection/manager.ex

@@ -332,6 +332,8 @@ defmodule Electric.Connection.Manager do
             stack_id: state.stack_id,
             persistent_kv: state.persistent_kv
           )
+
+          Electric.LsnTracker.reset(state.stack_id)
         end
 
         {:ok, shapes_sup_pid} =

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

@@ -0,0 +1,44 @@
+defmodule Electric.LsnTracker do
+  alias Electric.Postgres.Lsn
+
+  def init(last_processed_lsn, stack_id) do
+    create_table(stack_id)
+
+    set_last_processed_lsn(last_processed_lsn, stack_id)
+  end
+
+  @spec set_last_processed_lsn(Lsn.t() | non_neg_integer(), String.t()) :: :ok
+  def set_last_processed_lsn(lsn, stack_id) when is_struct(lsn, Lsn) do
+    stack_id
+    |> table()
+    |> :ets.insert({:last_processed_lsn, lsn})
+  end
+
+  def set_last_processed_lsn(lsn, stack_id) when is_integer(lsn) do
+    set_last_processed_lsn(Lsn.from_integer(lsn), stack_id)
+  end
+
+  @spec get_last_processed_lsn(String.t()) :: Lsn.t()
+  def get_last_processed_lsn(stack_id) do
+    [last_processed_lsn: lsn] =
+      stack_id
+      |> table()
+      |> :ets.lookup(:last_processed_lsn)
+
+    lsn
+  end
+
+  def reset(stack_id) do
+    set_last_processed_lsn(Lsn.from_integer(0), stack_id)
+  end
+
+  defp create_table(stack_id) do
+    stack_id
+    |> table()
+    |> :ets.new([:protected, :named_table])
+  end
+
+  defp table(stack_id) do
+    :"#{stack_id}:lsn_tracker"
+  end
+end

packages/sync-service/lib/electric/postgres/lsn.ex

@@ -229,6 +229,26 @@ defmodule Electric.Postgres.Lsn do
     |> from_integer()
   end
 
+  @doc """
+  Returns the highest Lsn from the given list of Lsns.
+
+  When the list is empty, it returns #Lsn<0/0>.
+
+  ## Examples
+      iex> max([%#{Lsn}{segment: 0, offset: 1}, %#{Lsn}{segment: 0, offset: 2}])
+      %#{Lsn}{segment: 0, offset: 2}
+
+      iex> max([%#{Lsn}{segment: 1, offset: 1}, %#{Lsn}{segment: 0, offset: 2}])
+      %#{Lsn}{segment: 1, offset: 1}
+
+      iex> max([])
+      %#{Lsn}{segment: 0, offset: 0}
+
+  """
+  @spec max(Enumerable.t(t())) :: t()
+  def max([]), do: from_integer(0)
+  def max(lsns) when is_list(lsns), do: Enum.max_by(lsns, &to_integer/1)
+
   defimpl Inspect do
     def inspect(lsn, _opts) do
       "#Lsn<#{Electric.Postgres.Lsn.to_iolist(lsn)}>"

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

@@ -62,6 +62,27 @@ defmodule Electric.Replication.LogOffset do
 
   def new(%__MODULE__{} = offset), do: offset
 
+  @doc """
+  Returns the LSN part of the LogOffset.
+
+  ## Examples
+
+      iex> extract_lsn(%LogOffset{tx_offset: 10, op_offset: 0})
+      #Lsn<0/A>
+
+      iex> extract_lsn(%LogOffset{tx_offset: 10, op_offset: 5})
+      #Lsn<0/A>
+
+      iex> extract_lsn(%LogOffset{tx_offset: 11, op_offset: 5})
+      #Lsn<0/B>
+
+      iex> extract_lsn(LogOffset.before_all())
+      #Lsn<0/0>
+  """
+  @spec extract_lsn(t()) :: Lsn.t()
+  def extract_lsn(%LogOffset{tx_offset: offset}) when offset < 0, do: Lsn.from_integer(0)
+  def extract_lsn(%LogOffset{tx_offset: offset}), do: Lsn.from_integer(offset)
+
   @doc """
   Compare two log offsets
 

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

@@ -1,6 +1,5 @@
 defmodule Electric.Replication.PersistentReplicationState do
   alias Electric.PersistentKV
-  alias Electric.Postgres.Lsn
   alias Electric.Replication.Changes
   require Logger
 
@@ -9,27 +8,6 @@ defmodule Electric.Replication.PersistentReplicationState do
           persistent_kv: Electric.PersistentKV.t()
         ]
 
-  @last_processed_lsn_key "last_processed_lsn"
-
-  @spec set_last_processed_lsn(Lsn.t() | non_neg_integer(), opts()) :: :ok
-  def set_last_processed_lsn(lsn, opts) when is_struct(lsn, Lsn) do
-    lsn |> Lsn.to_integer() |> set_last_processed_lsn(opts)
-  end
-
-  def set_last_processed_lsn(lsn, opts) when is_integer(lsn) do
-    Logger.debug("Updating last processed lsn to #{lsn}")
-    set(@last_processed_lsn_key, lsn, opts)
-  end
-
-  @spec get_last_processed_lsn(opts()) :: Lsn.t()
-  def get_last_processed_lsn(opts) do
-    case get(@last_processed_lsn_key, opts) do
-      {:ok, last_processed_lsn} -> last_processed_lsn
-      {:error, :not_found} -> 0
-    end
-    |> Lsn.from_integer()
-  end
-
   @base_tracked_relations %{
     table_to_id: %{},
     id_to_table_info: %{}
@@ -55,7 +33,6 @@ defmodule Electric.Replication.PersistentReplicationState do
 
   @spec reset(opts()) :: :ok
   def reset(opts) do
-    set(@last_processed_lsn_key, 0, opts)
     set_tracked_relations(@base_tracked_relations, opts)
   end
 

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

@@ -6,6 +6,7 @@ defmodule Electric.Replication.ShapeLogCollector do
   use GenStage
 
   require Electric.Postgres.Lsn
+  alias Electric.LsnTracker
   alias Electric.Replication.ShapeLogCollector.AffectedColumns
   alias Electric.Postgres.Lsn
   alias Electric.Replication.PersistentReplicationState
@@ -19,9 +20,7 @@ defmodule Electric.Replication.ShapeLogCollector do
   @schema NimbleOptions.new!(
             stack_id: [type: :string, required: true],
             inspector: [type: :mod_arg, required: true],
-            persistent_kv: [type: :any, required: true],
-            # see https://hexdocs.pm/gen_stage/GenStage.html#c:init/1-options
-            demand: [type: {:in, [:forward, :accumulate]}, default: :accumulate]
+            persistent_kv: [type: :any, required: true]
           )
 
   def start_link(opts) do
@@ -34,6 +33,10 @@ defmodule Electric.Replication.ShapeLogCollector do
     Electric.ProcessRegistry.name(stack_id, __MODULE__)
   end
 
+  def start_processing(server, last_processed_lsn) do
+    GenStage.call(server, {:start_processing, last_processed_lsn})
+  end
+
   # use `GenStage.call/2` here to make the event processing synchronous.
   #
   # Because `Electric.Shapes.Dispatcher` only sends demand to this producer
@@ -78,15 +81,13 @@ defmodule Electric.Replication.ShapeLogCollector do
         producer: nil,
         subscriptions: {0, MapSet.new()},
         persistent_replication_data_opts: persistent_replication_data_opts,
-        last_seen_lsn:
-          PersistentReplicationState.get_last_processed_lsn(persistent_replication_data_opts),
         tracked_relations: tracker_state
       })
 
     # start in demand: :accumulate mode so that the ShapeCache is able to start
     # all active consumers before we start sending transactions
     {:producer, state,
-     dispatcher: {Electric.Shapes.Dispatcher, inspector: state.inspector}, demand: opts.demand}
+     dispatcher: {Electric.Shapes.Dispatcher, inspector: state.inspector}, demand: :accumulate}
   end
 
   def handle_subscribe(:consumer, _opts, from, state) do
@@ -109,19 +110,12 @@ defmodule Electric.Replication.ShapeLogCollector do
   # last transaction and we can reply to the call and unblock the replication
   # client.
   def handle_demand(_demand, %{producer: producer} = state) do
-    GenServer.reply(producer, :ok)
-
-    :ok =
-      PersistentReplicationState.set_last_processed_lsn(
-        state.last_seen_lsn,
-        state.persistent_replication_data_opts
-      )
+    LsnTracker.set_last_processed_lsn(
+      state.last_processed_lsn,
+      state.stack_id
+    )
 
-    :ok =
-      PersistentReplicationState.set_tracked_relations(
-        state.tracked_relations,
-        state.persistent_replication_data_opts
-      )
+    GenServer.reply(producer, :ok)
 
     {:noreply, [], %{state | producer: nil}}
   end
@@ -134,6 +128,12 @@ defmodule Electric.Replication.ShapeLogCollector do
     {:noreply, [], remove_subscription(from, state)}
   end
 
+  def handle_call({:start_processing, lsn}, _from, state) do
+    LsnTracker.init(lsn, state.stack_id)
+    GenStage.demand(self(), :forward)
+    {:reply, :ok, [], Map.put(state, :last_processed_lsn, lsn)}
+  end
+
   def handle_call({:new_txn, %Transaction{xid: xid, lsn: lsn} = txn, trace_context}, from, state) do
     OpenTelemetry.set_current_context(trace_context)
 
@@ -161,17 +161,17 @@ defmodule Electric.Replication.ShapeLogCollector do
   # will prompt the `GenServer.reply/2` call.
   defp handle_transaction(txn, _from, %{subscriptions: {0, _}} = state) do
     Logger.debug(fn -> "Dropping transaction #{txn.xid}: no active consumers" end)
-    drop_transaction(txn, state)
+    drop_transaction(state)
   end
 
   # If we've already processed a transaction, then drop it without processing
-  defp handle_transaction(txn, _from, %{last_seen_lsn: last_seen_lsn} = state)
-       when not Lsn.is_larger(txn.lsn, last_seen_lsn) do
+  defp handle_transaction(txn, _from, %{last_processed_lsn: last_processed_lsn} = state)
+       when not Lsn.is_larger(txn.lsn, last_processed_lsn) do
     Logger.debug(fn ->
-      "Dropping transaction #{txn.xid}: transaction LSN #{txn.lsn} smaller than last processed #{last_seen_lsn}"
+      "Dropping transaction #{txn.xid}: transaction LSN #{txn.lsn} smaller than last processed #{last_processed_lsn}"
     end)
 
-    drop_transaction(txn, state)
+    drop_transaction(state)
   end
 
   defp handle_transaction(txn, from, state) do
@@ -191,7 +191,7 @@ defmodule Electric.Replication.ShapeLogCollector do
 
     # we don't reply to this call. we only reply when we receive demand from
     # the consumers, signifying that every one has processed this txn
-    {:noreply, [txn], %{state | producer: from} |> put_last_seen_lsn(txn.lsn)}
+    {:noreply, [txn], %{state | producer: from} |> put_last_processed_lsn(txn.lsn)}
   end
 
   defp handle_relation(rel, from, state) do
@@ -200,41 +200,27 @@ defmodule Electric.Replication.ShapeLogCollector do
     {updated_rel, tracker_state} =
       AffectedColumns.transform_relation(rel, state.tracked_relations)
 
+    :ok =
+      PersistentReplicationState.set_tracked_relations(
+        tracker_state,
+        state.persistent_replication_data_opts
+      )
+
     case state do
       %{subscriptions: {0, _}} ->
         Logger.debug(fn ->
           "Dropping relation message for #{inspect(rel.schema)}.#{inspect(rel.table)}: no active consumers"
         end)
 
-        :ok =
-          PersistentReplicationState.set_tracked_relations(
-            state.tracked_relations,
-            state.persistent_replication_data_opts
-          )
-
         {:reply, :ok, [], %{state | tracked_relations: tracker_state}}
 
       _ ->
         {:noreply, [updated_rel], %{state | producer: from, tracked_relations: tracker_state}}
     end
   end
 
-  defp drop_transaction(txn, state) do
+  defp drop_transaction(state) do
     OpenTelemetry.add_span_attributes("txn.is_dropped": true)
-
-    state =
-      if Lsn.is_larger(txn.lsn, state.last_seen_lsn) do
-        :ok =
-          PersistentReplicationState.set_last_processed_lsn(
-            txn.lsn,
-            state.persistent_replication_data_opts
-          )
-
-        put_last_seen_lsn(state, txn.lsn)
-      else
-        state
-      end
-
     {:reply, :ok, [], state}
   end
 
@@ -261,9 +247,9 @@ defmodule Electric.Replication.ShapeLogCollector do
     state
   end
 
-  defp put_last_seen_lsn(%{last_seen_lsn: last_seen_lsn} = state, lsn)
-       when Lsn.is_larger(lsn, last_seen_lsn),
-       do: %{state | last_seen_lsn: lsn}
+  defp put_last_processed_lsn(%{last_processed_lsn: last_processed_lsn} = state, lsn)
+       when Lsn.is_larger(lsn, last_processed_lsn),
+       do: %{state | last_processed_lsn: lsn}
 
-  defp put_last_seen_lsn(state, _lsn), do: state
+  defp put_last_processed_lsn(state, _lsn), do: state
 end