Lock ingest processing during session transfer

Author: zoldar

lib/plausible/session/transfer.ex

@@ -18,6 +18,10 @@ defmodule Plausible.Session.Transfer do
   @cmd_dump_cache :get
   @cmd_takeover_done :done
 
+  # NOTE: Timeout must be higher than respective WriteBuffer's
+  # flush interval + some margin for flush alone.
+  @lock_acquire_timeout :timer.seconds(8000)
+
   def telemetry_event, do: [:plausible, :sessions, :takeover]
 
   @doc """
@@ -109,7 +113,15 @@ defmodule Plausible.Session.Transfer do
     case request do
       {@cmd_list_cache_names, session_version} ->
         if session_version == session_version() and attempted?(parent) do
-          # NOTE: block ingest here
+          # Blocking ingest before transfer
+          Task.await_many(
+            [
+              Task.async(fn -> Plausible.Event.WriteBuffer.lock(@lock_acquire_timeout) end),
+              Task.async(fn -> Plausible.Session.WriteBuffer.lock(@lock_acquire_timeout) end)
+            ],
+            @lock_acquire_timeout + :timer.seconds(1)
+          )
+
           Cache.Adapter.get_names(:sessions)
         else
           []
@@ -119,9 +131,11 @@ defmodule Plausible.Session.Transfer do
         Cache.Adapter.cache2list(cache)
 
       @cmd_takeover_done ->
-        # NOTE: unblock ingest here
         # Wipe the cache after transfer is complete to avoid making the transferred cache stale
         Cache.Adapter.wipe(:sessions)
+        # Unblock ingest after trasnsfer is done
+        Plausible.Event.WriteBuffer.unlock()
+        Plausible.Session.WriteBuffer.unlock()
         :counters.add(given_counter, 1, 1)
     end
   end

test/plausible/session/transfer_test.exs

@@ -12,10 +12,20 @@ defmodule Plausible.Session.TransferTest do
 
     Enum.each(1..250, fn _ -> process_event(old, build(:event, name: "pageview")) end)
 
+    :ok = :peer.call(old, Plausible.Session.WriteBuffer, :flush, [])
+    :ok = :peer.call(old, Plausible.Event.WriteBuffer, :flush, [])
+
+    old_sessions_sorted = all_sessions_sorted(old)
+
     new = start_another_plausible(tmp_dir)
+
+    :ok = :peer.call(old, Plausible.Session.WriteBuffer, :flush, [])
+    :ok = :peer.call(old, Plausible.Event.WriteBuffer, :flush, [])
+
     await_transfer(new)
 
-    assert all_sessions_sorted(new) == all_sessions_sorted(old)
+    assert all_sessions_sorted(old) == []
+    assert all_sessions_sorted(new) == old_sessions_sorted
   end
 
   defp start_another_plausible(tmp_dir) do