fix: Do not await for responses when recovering pub filters

.changeset/healthy-seas-bathe.md

@@ -0,0 +1,7 @@
+---
+"@core/sync-service": patch
+---
+
+- Do not await for responses while recovering publication filters.
+- Remove publication update debounce time - simply wait until end of current process message queue.
+

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

@@ -52,7 +52,7 @@ defmodule Electric.Replication.PublicationManager do
 
   @retry_timeout 300
   @max_retries 3
-  @default_debounce_timeout 50
+  @default_debounce_timeout 0
 
   @relation_counter :relation_counter
   @relation_where :relation_where
@@ -97,7 +97,8 @@ 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.call(server, {:recover_shape, shape})
+    GenServer.cast(server, {:recover_shape, shape})
+    :ok
   end
 
   @spec remove_shape(Shape.t(), Keyword.t()) :: :ok
@@ -113,11 +114,8 @@ defmodule Electric.Replication.PublicationManager do
   @spec refresh_publication(Keyword.t()) :: :ok
   def refresh_publication(opts \\ []) do
     server = Access.get(opts, :server, name(opts))
-
-    case GenServer.call(server, :refresh_publication) do
-      :ok -> :ok
-      {:error, err} -> raise err
-    end
+    GenServer.cast(server, :refresh_publication)
+    :ok
   end
 
   def start_link(opts) do
@@ -175,20 +173,20 @@ defmodule Electric.Replication.PublicationManager do
     {:noreply, state}
   end
 
-  def handle_call({:recover_shape, shape}, _from, state) do
-    state = update_relation_filters_for_shape(shape, :add, state)
-    {:reply, :ok, state}
-  end
-
   def handle_call({:remove_shape, shape}, from, state) do
     state = update_relation_filters_for_shape(shape, :remove, state)
     state = add_waiter(from, state)
     state = schedule_update_publication(state.update_debounce_timeout, state)
     {:noreply, state}
   end
 
-  def handle_call(:refresh_publication, from, state) do
-    state = add_waiter(from, state)
+  @impl true
+  def handle_cast({:recover_shape, shape}, state) do
+    state = update_relation_filters_for_shape(shape, :add, state)
+    {:noreply, state}
+  end
+
+  def handle_cast(:refresh_publication, state) do
     state = schedule_update_publication(state.update_debounce_timeout, state)
     {:noreply, state}
   end
@@ -265,7 +263,12 @@ defmodule Electric.Replication.PublicationManager do
         %{state | pg_version: pg_version}
 
       {:error, err} ->
-        Logger.error("Failed to get PG version, retrying after timeout: #{inspect(err)}")
+        err_msg = "Failed to get PG version, retrying after timeout: #{inspect(err)}"
+
+        if %DBConnection.ConnectionError{reason: :queue_timeout} == err,
+          do: Logger.warning(err_msg),
+          else: Logger.error(err_msg)
+
         Process.sleep(@retry_timeout)
         get_pg_version(state)
     end

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

@@ -316,7 +316,7 @@ defmodule Electric.ShapeCache do
         {:ok, _pid, _snapshot_xmin, _latest_offset} = start_shape(shape_handle, shape, state)
 
         # recover publication filter state
-        :ok = publication_manager.recover_shape(shape, publication_manager_opts)
+        publication_manager.recover_shape(shape, publication_manager_opts)
       rescue
         e ->
           Logger.error("Failed to recover shape #{shape_handle}: #{inspect(e)}")