fix: Remove refresh_publiation API form PubMan (#3168)

Fixes #3167

We started seeing this error with an update made on 22nd of September,
and until then one of the relevant changes included is this:
#3108

Since @alco has introduced periodic forced publication updates anyway,
and upon reconciling the schema we also trigger dropping relevant shapes
(which would remove them from the publication on cleanup), we no longer
need to refresh the publication manually IMO - not much to be gained
other than unnecessary dependencies between our processes.

It's still worth investigating why we're seeing these timeouts (but was
harder to debug because we were missing sentry metadata, see
#3166) - but ultimately we
should simplify things.

1. Schema gets reconciled
2. Shapes using diverged/stale relations get dropped
3. Shape cleanup will involve removing them from the publication which
should cause a publication update (as all relevant shapes for a given
relation will be removed)
4. Even if (3) doesn't occur, which it will, we now periodically refresh
the publication anyway

In fact, I would go even further to say that since we're periodically
refreshing the publication and thus detecting missing relations there,
we should simply add an inspector cleanup and get rid of the schema
reconciler in its entirety - what do you think @alco @icehaunter ?

Author: msfstef

.changeset/little-seals-try.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Remove unncecessary `refresh_publication` API from `PublicationManager`.

integration-tests/tests/not-owner-of-the-table.lux

@@ -69,10 +69,8 @@
 [shell electric]
   ??[info] Configuring publication electric_publication_integration to drop ["public.items"] tables, and add [] tables
 
-  # Call refresh_publication() once more to ensure the previous update has completed. This is
-  # going to be a noop call since there are no changes pending for the publication.
-  !Electric.Replication.PublicationManager.refresh_publication(stack_id: "single_stack", forced?: true)
-  ??:ok
+  # sleep a bit to ensure the publication update completed
+  [sleep 3]
 
 [shell psql]
   !SELECT * FROM pg_publication_tables;

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

@@ -14,7 +14,6 @@ defmodule Electric.Replication.PublicationManager do
   @callback name(binary() | Keyword.t()) :: term()
   @callback add_shape(shape_handle(), Shape.t(), Keyword.t()) :: :ok
   @callback remove_shape(shape_handle(), Keyword.t()) :: :ok
-  @callback refresh_publication(Keyword.t()) :: :ok
 
   defstruct [
     :stack_id,
@@ -117,21 +116,6 @@ defmodule Electric.Replication.PublicationManager do
     end
   end
 
-  @impl __MODULE__
-  def refresh_publication(opts \\ []) do
-    server = Access.get(opts, :server, name(opts))
-    timeout = Access.get(opts, :timeout, 10_000)
-
-    case GenServer.call(
-           server,
-           {:refresh_publication, Access.get(opts, :forced?, false)},
-           timeout
-         ) do
-      :ok -> :ok
-      {:error, err} -> raise err
-    end
-  end
-
   def start_link(opts) do
     with {:ok, opts} <- NimbleOptions.validate(opts, @schema) do
       stack_id = Keyword.fetch!(opts, :stack_id)
@@ -201,16 +185,6 @@ defmodule Electric.Replication.PublicationManager do
     end
   end
 
-  def handle_call({:refresh_publication, forced?}, from, state) do
-    if forced? or update_needed?(state) do
-      state = add_waiter(from, state)
-      state = schedule_update_publication(state.update_debounce_timeout, forced?, state)
-      {:noreply, state}
-    else
-      {:reply, :ok, state, state.publication_refresh_period}
-    end
-  end
-
   defguardp is_fatal(err)
             when is_exception(err, Postgrex.Error) and
                    err.postgres.code in ~w|undefined_function undefined_table insufficient_privilege|a

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

@@ -13,18 +13,15 @@ defmodule Electric.Replication.SchemaReconciler do
   require Logger
 
   alias Electric.Postgres.Inspector
-  alias Electric.Replication.PublicationManager
   alias Electric.ShapeCache.ShapeCleaner
 
   @name_schema_tuple {:tuple, [:atom, :atom, :any]}
   @genserver_name_schema {:or, [:atom, @name_schema_tuple]}
-  @addressable_process {:or, [:atom, :pid, @name_schema_tuple]}
   @schema NimbleOptions.new!(
             name: [type: @genserver_name_schema, required: false],
             period: [type: :pos_integer, required: false, default: 60_000],
             inspector: [type: :any, required: true],
-            stack_id: [type: :string, required: true],
-            publication_manager: [type: @addressable_process, required: false]
+            stack_id: [type: :string, required: true]
           )
 
   def start_link(opts) do
@@ -92,10 +89,7 @@ defmodule Electric.Replication.SchemaReconciler do
     with :ok <-
            ShapeCleaner.remove_shapes_for_relations(diverged_relations, stack_id: state.stack_id),
          :ok <-
-           Enum.each(diverged_relations, fn {oid, _} ->
-             Inspector.clean(oid, state.inspector)
-           end),
-         :ok <- PublicationManager.refresh_publication(stack_id: state.stack_id, forced?: true) do
+           Enum.each(diverged_relations, fn {oid, _} -> Inspector.clean(oid, state.inspector) end) do
       :ok
     else
       {:error, reason} ->

packages/sync-service/test/electric/replication/publication_manager_db_test.exs

@@ -83,30 +83,5 @@ defmodule Electric.Replication.PublicationManagerDbTest do
     end
   end
 
-  describe "refresh_publication()" do
-    setup ctx do
-      shape = generate_shape(ctx.relation_with_oid)
-      :ok = PublicationManager.add_shape(@shape_handle_1, shape, ctx.pub_mgr_opts)
-    end
-
-    test "updates the publication if a published table is dropped", ctx do
-      Postgrex.query!(ctx.pool, "DROP TABLE items")
-
-      assert :ok == PublicationManager.refresh_publication(ctx.pub_mgr_opts ++ [forced?: true])
-      assert [] == fetch_pub_tables(ctx)
-
-      assert_receive {:remove_shapes_for_relations, [{_oid, {"public", "items"}}]}
-    end
-
-    test "updates the publication if a published table is renamed", ctx do
-      Postgrex.query!(ctx.pool, "ALTER TABLE items RENAME TO items_no_more")
-
-      assert :ok == PublicationManager.refresh_publication(ctx.pub_mgr_opts ++ [forced?: true])
-      assert [] == fetch_pub_tables(ctx)
-
-      assert_receive {:remove_shapes_for_relations, [{_oid, {"public", "items"}}]}
-    end
-  end
-
   defp fetch_pub_tables(ctx), do: fetch_publication_tables(ctx.pool, ctx.publication_name)
 end

packages/sync-service/test/electric/replication/publication_manager_manual_test.exs

@@ -84,32 +84,4 @@ defmodule Electric.Replication.PublicationManagerManualTest do
       assert_receive {:remove_shapes_for_relations, [{_oid, {"public", "items"}}]}
     end
   end
-
-  describe "refresh_publication" do
-    setup ctx do
-      Postgrex.query!(ctx.pool, "ALTER TABLE items REPLICA IDENTITY FULL")
-      Postgrex.query!(ctx.pool, "ALTER PUBLICATION #{ctx.publication_name} ADD TABLE items")
-
-      shape = generate_shape(ctx.relation_with_oid)
-      :ok = PublicationManager.add_shape(@shape_handle, shape, ctx.pub_mgr_opts)
-    end
-
-    test "verifies publication state when a table is dropped and cleans up the corresponding shape",
-         ctx do
-      Postgrex.query!(ctx.pool, "DROP TABLE items")
-
-      assert :ok == PublicationManager.refresh_publication(ctx.pub_mgr_opts ++ [forced?: true])
-
-      assert_receive {:remove_shapes_for_relations, [{_oid, {"public", "items"}}]}
-    end
-
-    test "verifies publication state when a table is renamed and cleans up the corresponding shape",
-         ctx do
-      Postgrex.query!(ctx.pool, "ALTER TABLE items RENAME TO items_no_more")
-
-      assert :ok == PublicationManager.refresh_publication(ctx.pub_mgr_opts ++ [forced?: true])
-
-      assert_receive {:remove_shapes_for_relations, [{_oid, {"public", "items"}}]}
-    end
-  end
 end

packages/sync-service/test/electric/replication/publication_manager_test.exs

@@ -65,32 +65,15 @@ defmodule Electric.Replication.PublicationManagerTest do
       refute_receive {:filters, _}, 200
     end
 
-    @tag update_debounce_timeout: 10
     test "accepts multiple relations", %{opts: opts} do
       shape1 = generate_shape({"public", "items"})
       shape2 = generate_shape({"public", "other"})
+
       assert :ok == PublicationManager.add_shape(@shape_handle_1, shape1, opts)
-      assert :ok == PublicationManager.add_shape(@shape_handle_2, shape2, opts)
+      assert_receive {:filters, [{_, {"public", "items"}}]}, 500
 
-      collected =
-        receive do
-          {:filters, filters} -> MapSet.new(filters)
-        after
-          500 -> flunk("Did not receive initial filters")
-        end
-
-      # Force refresh to ensure any pending prepared filters are committed
-      :ok = PublicationManager.refresh_publication(Keyword.merge(opts, forced?: true))
-
-      collected =
-        receive do
-          {:filters, filters} ->
-            MapSet.union(collected, MapSet.new(filters))
-        after
-          200 -> collected
-        end
-
-      assert [{_, {"public", "items"}}, {_, {"public", "other"}}] = MapSet.to_list(collected)
+      assert :ok == PublicationManager.add_shape(@shape_handle_2, shape2, opts)
+      assert_receive {:filters, [{_, {"public", "items"}}, {_, {"public", "other"}}]}, 500
     end
 
     @tag update_debounce_timeout: 100

packages/sync-service/test/electric/shape_cache/expiry_manager_test.exs

@@ -38,8 +38,6 @@ defmodule Electric.ExpiryManagerTest do
     def add_shape(_handle, _, opts) do
       send(opts[:test_pid], {:called, :prepare_tables_fn})
     end
-
-    def refresh_publication(_), do: :ok
   end
 
   setup :verify_on_exit!

packages/sync-service/test/support/component_setup.ex

@@ -15,7 +15,6 @@ defmodule Support.ComponentSetup do
     def name(_), do: :pub_man
     def add_shape(_handle, _shape, _opts), do: :ok
     def remove_shape(_handle, _opts), do: :ok
-    def refresh_publication(_opts), do: :ok
   end
 
   defmodule TestPublicationManager do
@@ -36,11 +35,6 @@ defmodule Support.ComponentSetup do
       send(parent, {TestPublicationManager, :remove_shape, handle})
       :ok
     end
-
-    def refresh_publication(%{parent: parent}) do
-      send(parent, {TestPublicationManager, :refresh_publication})
-      :ok
-    end
   end
 
   def with_stack_id_from_test(ctx) do