Periodically refresh publication independently of schema reconciler

So far SchemaReconciler has been responsible both for updating the
publication when it detected schema changes as well as when there are no
schema changes. This latter periodic update is useful to keep the
publication up-to-date with the filters but it's not really the
responsibility of SchemaReconciler.

Author: alco

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

@@ -15,7 +15,7 @@
   "-v $(realpath ../scripts/low_privilege_db.sh):/docker-entrypoint-initdb.d/initdb-low_privilege_db.sh"]
 
 ## Start the sync service
-[invoke setup_electric_with_env "ELECTRIC_TWEAKS_SCHEMA_RECONCILER_PERIOD=2s"]
+[invoke setup_electric_with_env "ELECTRIC_TWEAKS_PUBLICATION_REFRESH_PERIOD=2s"]
 
 ## Verify that the replication pipeline has initialized successfully
 [shell electric]
@@ -62,15 +62,17 @@
   ??{"message":"Unable to create initial snapshot: must be owner of table items"}
 
 [shell electric]
-  ??[info] Configuring publication electric_publication_integration to drop [] tables, and add [] tables
 	??[info] Altering identity of public.items to FULL
   ?\[error\] Failed to configure publication: %Postgrex.Error\{message: nil, postgres: %\{code: :insufficient_privilege,.*message: "must be owner of table items"
 
-## Wait for the schema reconciler to run and verify that the table is removed from the publication
+## Wait for the publication to be refreshed and verify that the table is removed from it
 [shell electric]
   ??[info] Configuring publication electric_publication_integration to drop ["public.items"] tables, and add [] tables
-  ## Wait for one more run of the schema reconciler to make sure the previous publication update has completed
-  ??[info] Configuring publication electric_publication_integration to drop [] 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
 
 [shell psql]
   !SELECT * FROM pg_publication_tables;

packages/sync-service/config/runtime.exs

@@ -242,6 +242,12 @@ config :electric,
     env!("ELECTRIC_TCP_SEND_TIMEOUT", &Electric.Config.parse_human_readable_time!/1, nil),
   feature_flags: env!("ELECTRIC_FEATURE_FLAGS", &Electric.Config.parse_feature_flags/1, nil),
   manual_table_publishing?: env!("ELECTRIC_MANUAL_TABLE_PUBLISHING", :boolean, nil),
+  publication_refresh_period:
+    env!(
+      "ELECTRIC_TWEAKS_PUBLICATION_REFRESH_PERIOD",
+      &Electric.Config.parse_human_readable_time!/1,
+      nil
+    ),
   schema_reconciler_period:
     env!(
       "ELECTRIC_TWEAKS_SCHEMA_RECONCILER_PERIOD",

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

@@ -133,6 +133,7 @@ defmodule Electric.Application do
       tweaks: [
         publication_alter_debounce_ms: get_env(opts, :publication_alter_debounce_ms),
         registry_partitions: get_env(opts, :process_registry_partitions),
+        publication_refresh_period: get_env(opts, :publication_refresh_period),
         schema_reconciler_period: get_env(opts, :schema_reconciler_period)
       ],
       manual_table_publishing?: get_env(opts, :manual_table_publishing?)

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

@@ -91,6 +91,7 @@ defmodule Electric.Config do
     ## Misc
     process_registry_partitions: &Electric.Config.Defaults.process_registry_partitions/0,
     feature_flags: if(Mix.env() == :test, do: @known_feature_flags, else: []),
+    publication_refresh_period: 60_000,
     schema_reconciler_period: 60_000
   ]
 

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

@@ -62,7 +62,8 @@ defmodule Electric.Connection.Manager.Supervisor do
        can_alter_publication?: Keyword.fetch!(opts, :can_alter_publication?),
        manual_table_publishing?: Keyword.fetch!(opts, :manual_table_publishing?),
        db_pool: Keyword.fetch!(db_pool_opts, :name),
-       update_debounce_timeout: Keyword.get(tweaks, :publication_alter_debounce_ms, 0)}
+       update_debounce_timeout: Keyword.get(tweaks, :publication_alter_debounce_ms, 0),
+       refresh_period: Keyword.get(tweaks, :publication_refresh_period, 60_000)}
 
     shape_log_collector_spec =
       {Electric.Replication.ShapeLogCollector,

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

@@ -207,8 +207,6 @@ defmodule Electric.Postgres.Configuration do
   end
 
   defp alter_publication!(conn, publication_name, relation_filters) do
-    publication = Utils.quote_name(publication_name)
-
     prev_published_tables =
       get_publication_tables(conn, publication_name)
       |> Enum.map(fn {_oid, relation, _replident} -> Utils.relation_to_sql(relation) end)
@@ -220,13 +218,18 @@ defmodule Electric.Postgres.Configuration do
       |> Enum.map(&Utils.relation_to_sql/1)
       |> MapSet.new()
 
-    to_drop = MapSet.difference(prev_published_tables, new_published_tables)
-    to_add = MapSet.difference(new_published_tables, prev_published_tables)
+    to_drop = MapSet.difference(prev_published_tables, new_published_tables) |> Enum.to_list()
+    to_add = MapSet.difference(new_published_tables, prev_published_tables) |> Enum.to_list()
+    execute_alter_publication_statements!(conn, publication_name, to_drop, to_add)
+  end
+
+  defp execute_alter_publication_statements!(_, _, [], []), do: :ok
 
+  defp execute_alter_publication_statements!(conn, publication_name, to_drop, to_add) do
     Logger.info(
-      "Configuring publication #{publication_name} to drop #{inspect(Enum.to_list(to_drop))} tables, and add #{inspect(Enum.to_list(to_add))} tables",
-      publication_alter_drop_tables: Enum.to_list(to_drop),
-      publication_alter_add_tables: Enum.to_list(to_add)
+      "Configuring publication #{publication_name} to drop #{inspect(to_drop)} tables, and add #{inspect(to_add)} tables",
+      publication_alter_drop_tables: to_drop,
+      publication_alter_add_tables: to_add
     )
 
     alter_ops =
@@ -236,7 +239,11 @@ defmodule Electric.Postgres.Configuration do
       Postgrex.query!(conn, "SAVEPOINT before_publication", [])
 
       # PG 14 and below do not support filters on tables of publications
-      case Postgrex.query(conn, "ALTER PUBLICATION #{publication} #{op} TABLE #{table}", []) do
+      case Postgrex.query(
+             conn,
+             "ALTER PUBLICATION #{Utils.quote_name(publication_name)} #{op} TABLE #{table}",
+             []
+           ) do
         {:ok, _} ->
           Postgrex.query!(conn, "RELEASE SAVEPOINT before_publication", [])
           :ok

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

@@ -34,6 +34,7 @@ defmodule Electric.Replication.PublicationManager do
     :can_alter_publication?,
     :manual_table_publishing?,
     :configure_tables_for_replication_fn,
+    :publication_refresh_period,
     :shape_cache,
     next_update_forced?: false
   ]
@@ -50,6 +51,7 @@ defmodule Electric.Replication.PublicationManager do
            publication_name: String.t(),
            db_pool: term(),
            configure_tables_for_replication_fn: fun(),
+           publication_refresh_period: non_neg_integer(),
            shape_cache: {module(), term()},
            next_update_forced?: boolean()
          }
@@ -80,7 +82,8 @@ defmodule Electric.Replication.PublicationManager do
               required: false,
               default: &Configuration.configure_publication!/4
             ],
-            server: [type: :any, required: false]
+            server: [type: :any, required: false],
+            refresh_period: [type: :pos_integer, required: false, default: 60_000]
           )
 
   @behaviour __MODULE__
@@ -170,10 +173,11 @@ defmodule Electric.Replication.PublicationManager do
       can_alter_publication?: opts.can_alter_publication?,
       manual_table_publishing?: opts.manual_table_publishing?,
       shape_cache: Map.get(opts, :shape_cache, {Electric.ShapeCache, [stack_id: opts.stack_id]}),
-      configure_tables_for_replication_fn: opts.configure_tables_for_replication_fn
+      configure_tables_for_replication_fn: opts.configure_tables_for_replication_fn,
+      publication_refresh_period: opts.refresh_period
     }
 
-    {:ok, state}
+    {:ok, state, state.publication_refresh_period}
   end
 
   @impl true
@@ -190,7 +194,7 @@ defmodule Electric.Replication.PublicationManager do
         state = schedule_update_publication(state.update_debounce_timeout, state)
         {:noreply, state}
       else
-        {:reply, :ok, state}
+        {:reply, :ok, state, state.publication_refresh_period}
       end
     end
   end
@@ -208,7 +212,7 @@ defmodule Electric.Replication.PublicationManager do
         state = schedule_update_publication(state.update_debounce_timeout, state)
         {:noreply, state}
       else
-        {:reply, :ok, state}
+        {:reply, :ok, state, state.publication_refresh_period}
       end
     end
   end
@@ -219,19 +223,21 @@ defmodule Electric.Replication.PublicationManager do
       state = schedule_update_publication(state.update_debounce_timeout, forced?, state)
       {:noreply, state}
     else
-      {:reply, :ok, state}
+      {:reply, :ok, state, state.publication_refresh_period}
     end
   end
 
   def handle_call({:recover_shape, shape_handle, shape}, _from, state) do
-    if is_tracking_shape_handle?(shape_handle, state) do
-      Logger.debug("Shape already tracked: #{inspect(shape_handle)}")
-      {:reply, :ok, state}
-    else
-      state = track_shape_handle(shape_handle, state)
-      state = update_relation_filters_for_shape(shape, :add, state)
-      {:reply, :ok, state}
-    end
+    state =
+      if is_tracking_shape_handle?(shape_handle, state) do
+        Logger.debug("Shape already tracked: #{inspect(shape_handle)}")
+        state
+      else
+        state = track_shape_handle(shape_handle, state)
+        update_relation_filters_for_shape(shape, :add, state)
+      end
+
+    {:reply, :ok, state, state.publication_refresh_period}
   end
 
   defguardp is_fatal(err)
@@ -241,16 +247,27 @@ defmodule Electric.Replication.PublicationManager do
   @impl true
   def handle_info(:update_publication, state) do
     # Clear out the timer ref
+    if state.scheduled_updated_ref, do: Process.cancel_timer(state.scheduled_updated_ref)
     state = %{state | scheduled_updated_ref: nil}
 
     # Invoke the actual handler for the publication update
-    if not state.can_alter_publication? or state.manual_table_publishing? do
-      check_publication_relations(state)
-    else
-      update_publication_state(state)
+    updated_state =
+      if not state.can_alter_publication? or state.manual_table_publishing? do
+        check_publication_relations(state)
+      else
+        update_publication_state(state)
+      end
+
+    with {:ok, state} <- updated_state do
+      # Schedule a forced refresh to happen periodically unless there's an explicit call to
+      # update the publication that happens sooner.
+      {:noreply, state, state.publication_refresh_period}
     end
   end
 
+  def handle_info(:timeout, state),
+    do: handle_info(:update_publication, %{state | next_update_forced?: true})
+
   defp check_publication_relations(
          %__MODULE__{
            committed_relation_filters: committed_filters,
@@ -260,7 +277,7 @@ defmodule Electric.Replication.PublicationManager do
        ) do
     if not forced? and filters_are_equal?(current_filters, committed_filters) do
       Logger.debug("No changes to publication, skipping checkup")
-      {:noreply, reply_to_waiters(:ok, state)}
+      {:ok, reply_to_waiters(:ok, state)}
     else
       # We cannot modify the publication, so we only check whether it is in the right state for
       # the set of currently active relation filters.
@@ -294,7 +311,7 @@ defmodule Electric.Replication.PublicationManager do
           error = %Electric.DbConfigurationError{type: reason, message: message}
 
           state = reply_to_waiters({:error, error}, state)
-          {:noreply, %{state | next_update_forced?: false}}
+          {:ok, %{state | next_update_forced?: false}}
       end
     end
   end
@@ -326,12 +343,12 @@ defmodule Electric.Replication.PublicationManager do
       {:error, err} when retries < @max_retries and not is_fatal(err) ->
         Logger.warning("Failed to configure publication, retrying: #{inspect(err)}")
         state = schedule_update_publication(@retry_timeout, %{state | retries: retries + 1})
-        {:noreply, state}
+        {:ok, state}
 
       {:error, err} ->
         Logger.error("Failed to configure publication: #{inspect(err)}")
         state = reply_to_waiters({:error, err}, state)
-        {:noreply, %{state | next_update_forced?: false}}
+        {:ok, %{state | next_update_forced?: false}}
     end
   end
 
@@ -352,7 +369,7 @@ defmodule Electric.Replication.PublicationManager do
     state = reply_to_waiters(:ok, state)
     committed_filters = MapSet.difference(state.prepared_relation_filters, invalidated_relations)
 
-    {:noreply,
+    {:ok,
      %{
        state
        | committed_relation_filters: committed_filters,

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

@@ -108,6 +108,7 @@ defmodule Electric.StackSupervisor do
                          on_cleanup: [type: {:fun, 1}]
                        ]
                      ],
+                     publication_refresh_period: [type: :non_neg_integer, default: 60_000],
                      schema_reconciler_period: [type: :non_neg_integer, default: 60_000]
                    ]
                  ],