fix: Synchronously refresh publication and clear cache if failed (#2490)

Followup to #2487

In the above PR I fixed an issue that would cause the publication
manager to exit, which made Electric go into a restart loop.

There were two issues that compounded that problem:
1. I had previously converted the recovery of the shapes on the
publication manager asynchronous, using `cast` instead of `call`, which
means that Electric went ahead with initialisation and marked itself
healthy despite not managing to actually recover - if it had waited for
the recovery to finish and fail the container would never have been
marked as healthy and never deployed.
2. If a recovery fails, for whatever reason, it might enter into an
infinite recovery loop as it is likely some of the persisted state from
which we are trying to recover is causing the issue. We should be able
to recover from that since we consider the storage/cache disposable.

In order to solve the above issues, I implement the following:
1. Convert the calls back to synchronous calls, increasing the timeout
for `ShapeCache` initialisation slightly, as that was the original
purpose of making them asynchronous (it might take a while to recover
things).
a. I'm not entirely sure if this is a bit of an anti-pattern - we could
place the initialisation in a `{:continue, :restore_shapes}` after the
init - my concern is that if the rest of Electric continues on and
initialises everything else before the restore is done we might have a
problem.
2. Use `try..catch..after` blocks to ensure cleanup occurs if any
recovery operation fails, such that Electric will still manage to start
after the supervisor restarts the process and the old, problematic state
is cleaned up.
a. It's a bit of a brutal approach, I've tested it by tweaking the
publication manager to fail every now and then and it works really
nicely, where it will erase the stored shapes and restart and continues
on like normal.
b. We could theoretically place this at a higher level, such as the
`start_link` of the `ShapeCache`, so that we always clear the cache if
initialisation fails, but I wanted to keep the scope tight and only to
things that we have observed during real operation.


I can spend time to add a sort of integration test with a failing
`PublicationManager` to better simulate this situation in a test suite
rather than ad hoc.

Tagging @balegas as well as we discussed this change

Author: msfstef

.changeset/hungry-beers-jog.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Synchronously recover shapes and publication to ensure Electric boot is successfuly, and clear cache if it fails.

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

@@ -107,8 +107,7 @@ defmodule Electric.Replication.PublicationManager do
   @spec recover_shape(Shape.t(), Keyword.t()) :: :ok
   def recover_shape(shape, opts \\ []) do
     server = Access.get(opts, :server, name(opts))
-    GenServer.cast(server, {:recover_shape, shape})
-    :ok
+    GenServer.call(server, {:recover_shape, shape})
   end
 
   @spec remove_shape(Shape.t(), Keyword.t()) :: :ok
@@ -124,8 +123,12 @@ defmodule Electric.Replication.PublicationManager do
   @spec refresh_publication(Keyword.t()) :: :ok
   def refresh_publication(opts \\ []) do
     server = Access.get(opts, :server, name(opts))
-    GenServer.cast(server, :refresh_publication)
-    :ok
+    timeout = Access.get(opts, :timeout, 10_000)
+
+    case GenServer.call(server, :refresh_publication, timeout) do
+      :ok -> :ok
+      {:error, err} -> raise err
+    end
   end
 
   def start_link(opts) do
@@ -194,15 +197,15 @@ defmodule Electric.Replication.PublicationManager do
     {:noreply, state}
   end
 
-  @impl true
-  def handle_cast({:recover_shape, shape}, state) do
-    state = update_relation_filters_for_shape(shape, :add, state)
+  def handle_call(:refresh_publication, from, state) do
+    state = add_waiter(from, state)
+    state = schedule_update_publication(state.update_debounce_timeout, state)
     {:noreply, state}
   end
 
-  def handle_cast(:refresh_publication, state) do
-    state = schedule_update_publication(state.update_debounce_timeout, state)
-    {:noreply, state}
+  def handle_call({:recover_shape, shape}, _from, state) do
+    state = update_relation_filters_for_shape(shape, :add, state)
+    {:reply, :ok, state}
   end
 
   @impl true

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

@@ -88,7 +88,10 @@ defmodule Electric.ShapeCache do
       GenServer.start_link(
         __MODULE__,
         Map.new(opts) |> Map.put(:db_pool, db_pool) |> Map.put(:name, name),
-        name: name
+        name: name,
+        # allow extra time for shape cache to initialise as it needs to
+        # restore existing shapes
+        timeout: Keyword.get(opts, :timeout, 20_000)
       )
     end
   end
@@ -209,9 +212,21 @@ defmodule Electric.ShapeCache do
         recover_shapes(state)
       end
 
-    # ensure publication filters are in line with existing shapes
+    # ensure publication filters are in line with existing shapes,
+    # and clean up cache if publication fails to update
     {publication_manager, publication_manager_opts} = opts.publication_manager
-    publication_manager.refresh_publication(publication_manager_opts)
+
+    try do
+      :ok = publication_manager.refresh_publication(publication_manager_opts)
+    rescue
+      error ->
+        clean_up_all_shapes(state)
+        reraise error, __STACKTRACE__
+    catch
+      :exit, reason ->
+        clean_up_all_shapes(state)
+        exit(reason)
+    end
 
     # do this after finishing this function so that we're subscribed to the
     # producer before it starts forwarding its demand
@@ -270,15 +285,21 @@ defmodule Electric.ShapeCache do
   end
 
   defp clean_up_shape(state, shape_handle) do
-    Electric.Shapes.DynamicConsumerSupervisor.stop_shape_consumer(
-      state.consumer_supervisor,
-      state.stack_id,
-      shape_handle
-    )
+    try do
+      Electric.Shapes.DynamicConsumerSupervisor.stop_shape_consumer(
+        state.consumer_supervisor,
+        state.stack_id,
+        shape_handle
+      )
+    after
+      # another failsafe for ensuring data is cleaned if consumer fails to
+      # clean itself - this is not guaranteed to run in case of an actual exit
+      # but provides an additional safeguard
 
-    shape_handle
-    |> Electric.ShapeCache.Storage.for_shape(state.storage)
-    |> Electric.ShapeCache.Storage.unsafe_cleanup!()
+      shape_handle
+      |> Electric.ShapeCache.Storage.for_shape(state.storage)
+      |> Electric.ShapeCache.Storage.unsafe_cleanup!()
+    end
 
     :ok
   end
@@ -304,9 +325,11 @@ defmodule Electric.ShapeCache do
         # recover publication filter state
         publication_manager.recover_shape(shape, publication_manager_opts)
         [LogOffset.extract_lsn(latest_offset)]
-      rescue
-        e ->
-          Logger.error("Failed to recover shape #{shape_handle}: #{inspect(e)}")
+      catch
+        kind, reason when kind in [:exit, :error] ->
+          Logger.error(
+            "Failed to recover shape #{shape_handle}: #{Exception.format(kind, reason, __STACKTRACE__)}"
+          )
 
           # clean up corrupted data to avoid persisting bad state
           Electric.ShapeCache.Storage.for_shape(shape_handle, state.storage)

packages/sync-service/test/electric/shape_cache_test.exs

@@ -893,6 +893,36 @@ defmodule Electric.ShapeCacheTest do
       assert {^shape_handle, ^offset} = ShapeCache.get_or_create_shape_handle(@shape, opts)
     end
 
+    test "invalidates shapes that we fail to restore", %{shape_cache_opts: opts} = context do
+      {shape_handle1, _} = ShapeCache.get_or_create_shape_handle(@shape, opts)
+      :started = ShapeCache.await_snapshot_start(shape_handle1, opts)
+
+      Mock.PublicationManager
+      |> stub(:remove_shape, fn _, _ -> :ok end)
+      |> expect(:recover_shape, 1, fn _, _ -> :ok end)
+      |> expect(:refresh_publication, 1, fn _ -> raise "failed recovery" end)
+      |> allow(self(), fn -> Shapes.Consumer.whereis(context[:stack_id], shape_handle1) end)
+      |> allow(self(), fn -> Process.whereis(opts[:server]) end)
+
+      # Should fail to start shape cache and clean up shapes
+      Process.flag(:trap_exit, true)
+
+      assert_raise MatchError, ~r/%RuntimeError{message: \"failed recovery\"/, fn ->
+        restart_shape_cache(%{
+          context
+          | publication_manager: {Mock.PublicationManager, []}
+        })
+      end
+
+      Process.flag(:trap_exit, false)
+
+      # Next restart should not recover shape
+      restart_shape_cache(context)
+      {shape_handle2, _} = ShapeCache.get_or_create_shape_handle(@shape, opts)
+      :started = ShapeCache.await_snapshot_start(shape_handle2, opts)
+      assert shape_handle1 != shape_handle2
+    end
+
     defp restart_shape_cache(context) do
       stop_shape_cache(context)
 
@@ -916,7 +946,9 @@ defmodule Electric.ShapeCacheTest do
           {pid, Process.monitor(pid)}
         end
 
-      Shapes.DynamicConsumerSupervisor.stop_all_consumers(ctx.consumer_supervisor)
+      if Enum.count(consumers) > 0 do
+        Shapes.DynamicConsumerSupervisor.stop_all_consumers(ctx.consumer_supervisor)
+      end
 
       for {pid, ref} <- consumers do
         assert_receive {:DOWN, ^ref, :process, ^pid, _}

packages/typescript-client/test/integration.test.ts

@@ -635,8 +635,8 @@ describe(`HTTP Sync`, () => {
     const shapeOffset = res.headers.get(`electric-offset`)!
     const fakeMidOffset = shapeOffset
       .split(`_`)
-      .map(Number)
-      .map((x, i) => (i === 0 ? x - 1 : x))
+      .map(BigInt)
+      .map((x, i) => (i === 0 ? x - BigInt(1) : x))
       .join(`_`)
     const etag = res.headers.get(`etag`)
     expect(etag, `Response should have etag header`).not.toBe(null)
@@ -927,7 +927,7 @@ describe(`HTTP Sync`, () => {
           // the next up to date message should have had
           // a 409 interleaved before it that instructed the
           // client to go and fetch data from scratch
-          expect(statusCodesReceived).toHaveLength(5)
+          expect(statusCodesReceived.length).greaterThanOrEqual(5)
           expect(statusCodesReceived[2]).toBe(409)
           expect(statusCodesReceived[3]).toBe(200)
           return res()