feat: allow publication alter debounce to be configurable

icehaunter · icehaunter · commit 60160f3e8256 · 2025-06-16T19:11:27.000+03:00
diff --git a/.changeset/tidy-days-matter.md b/.changeset/tidy-days-matter.md
@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+feat: allow publication alter debounce to be configurable
diff --git a/packages/sync-service/config/runtime.exs b/packages/sync-service/config/runtime.exs
@@ -202,7 +202,13 @@ config :electric,
   profile_where_clauses?: env!("ELECTRIC_PROFILE_WHERE_CLAUSES", :boolean, false),
   persistent_kv: persistent_kv_spec,
   listen_on_ipv6?: env!("ELECTRIC_LISTEN_ON_IPV6", :boolean, nil),
-  secret: secret
+  secret: secret,
+  publication_alter_debounce_ms:
+    env!(
+      "ELECTRIC_PUBLICATION_ALTER_DEBOUNCE_TIME",
+      &Electric.Config.parse_human_readable_time!/1,
+      nil
+    )
 
 if Electric.telemetry_enabled?() do
   config :sentry,
diff --git a/packages/sync-service/lib/electric/application.ex b/packages/sync-service/lib/electric/application.ex
@@ -128,7 +128,8 @@ defmodule Electric.Application do
       chunk_bytes_threshold: get_env(opts, :chunk_bytes_threshold),
       telemetry_opts:
         telemetry_opts([instance_id: instance_id, installation_id: installation_id] ++ opts),
-      max_shapes: get_env(opts, :max_shapes)
+      max_shapes: get_env(opts, :max_shapes),
+      tweaks: [publication_alter_debounce_ms: get_env(opts, :publication_alter_debounce_ms)]
     )
   end
 
diff --git a/packages/sync-service/lib/electric/config.ex b/packages/sync-service/lib/electric/config.ex
@@ -67,7 +67,9 @@ defmodule Electric.Config do
     otel_per_process_metrics?: false,
     telemetry_top_process_count: 5,
     ## Memory
-    shape_hibernate_after: :timer.seconds(30)
+    shape_hibernate_after: :timer.seconds(30),
+    ## Performance tweaks
+    publication_alter_debounce_ms: 0
   ]
 
   @installation_id_key "electric_installation_id"
diff --git a/packages/sync-service/lib/electric/connection/supervisor.ex b/packages/sync-service/lib/electric/connection/supervisor.ex
@@ -75,14 +75,16 @@ defmodule Electric.Connection.Supervisor do
     replication_opts = Keyword.fetch!(opts, :replication_opts)
     inspector = Keyword.fetch!(shape_cache_opts, :inspector)
     persistent_kv = Keyword.fetch!(opts, :persistent_kv)
+    tweaks = Keyword.fetch!(opts, :tweaks)
 
     shape_cache_spec = {Electric.ShapeCache, shape_cache_opts}
 
     publication_manager_spec =
       {Electric.Replication.PublicationManager,
        stack_id: stack_id,
        publication_name: Keyword.fetch!(replication_opts, :publication_name),
-       db_pool: Keyword.fetch!(db_pool_opts, :name)}
+       db_pool: Keyword.fetch!(db_pool_opts, :name),
+       update_debounce_timeout: Keyword.get(tweaks, :publication_alter_debounce_ms, 0)}
 
     shape_log_collector_spec =
       {Electric.Replication.ShapeLogCollector,
diff --git a/packages/sync-service/lib/electric/replication/publication_manager.ex b/packages/sync-service/lib/electric/replication/publication_manager.ex
@@ -204,9 +204,14 @@ defmodule Electric.Replication.PublicationManager do
     else
       state = track_shape_handle(shape_handle, state)
       state = update_relation_filters_for_shape(shape, :add, state)
-      state = add_waiter(from, state)
-      state = schedule_update_publication(state.update_debounce_timeout, state)
-      {:noreply, state}
+
+      if update_needed?(state) do
+        state = add_waiter(from, state)
+        state = schedule_update_publication(state.update_debounce_timeout, state)
+        {:noreply, state}
+      else
+        {:reply, :ok, state}
+      end
     end
   end
 
@@ -217,16 +222,25 @@ defmodule Electric.Replication.PublicationManager do
     else
       state = untrack_shape_handle(shape_handle, state)
       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}
+
+      if update_needed?(state) do
+        state = add_waiter(from, state)
+        state = schedule_update_publication(state.update_debounce_timeout, state)
+        {:noreply, state}
+      else
+        {:reply, :ok, state}
+      end
     end
   end
 
   def handle_call({:refresh_publication, forced?}, from, state) do
-    state = add_waiter(from, state)
-    state = schedule_update_publication(state.update_debounce_timeout, forced?, state)
-    {:noreply, state}
+    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}
+    end
   end
 
   def handle_call({:recover_shape, shape_handle, shape}, _from, state) do
@@ -307,6 +321,18 @@ defmodule Electric.Replication.PublicationManager do
        ),
        do: %{state | next_update_forced?: forced? or state.next_update_forced?}
 
+  defp update_needed?(%__MODULE__{
+         prepared_relation_filters: commited,
+         committed_relation_filters: prepared,
+         row_filtering_enabled: row_filtering_enabled
+       }) do
+    cond do
+      prepared == commited -> false
+      not row_filtering_enabled and Map.keys(prepared) == Map.keys(commited) -> false
+      true -> true
+    end
+  end
+
   # Updates are forced when we're doing periodic checks: we expect no changes to the filters,
   # but we'll write them anyway because that'll verify that no tables have been dropped/renamed
   # since the last update. Useful when we're not altering the publication often to catch changes
diff --git a/packages/sync-service/lib/electric/replication/supervisor.ex b/packages/sync-service/lib/electric/replication/supervisor.ex
@@ -39,8 +39,8 @@ defmodule Electric.Replication.Supervisor do
       consumer_supervisor,
       publication_manager,
       log_collector,
-      schema_reconciler,
-      shape_cache
+      shape_cache,
+      schema_reconciler
     ]
 
     Supervisor.init(children, strategy: :one_for_all)
diff --git a/packages/sync-service/lib/electric/stack_supervisor.ex b/packages/sync-service/lib/electric/stack_supervisor.ex
@@ -96,6 +96,7 @@ defmodule Electric.StackSupervisor do
                      "tweaks to the behaviour of parts of the supervision tree, used mostly for tests",
                    default: [],
                    keys: [
+                     publication_alter_debounce_ms: [type: :non_neg_integer, default: 0],
                      registry_partitions: [type: :non_neg_integer, required: false],
                      monitor_opts: [
                        type: :keyword_list,

-Original file line number
+Diff line change
@@ @@ -0,0 +1,5 @@ @@
 +---
 +"@core/sync-service": patch
 +---
++
 +feat: allow publication alter debounce to be configurable
Original file line number	Diff line number	Diff line change
`@@ -128,7 +128,8 @@ defmodule Electric.Application do`
`128`	`128`	`chunk_bytes_threshold: get_env(opts, :chunk_bytes_threshold),`
`129`	`129`	`telemetry_opts:`
`130`	`130`	`telemetry_opts([instance_id: instance_id, installation_id: installation_id] ++ opts),`
`131`		`- max_shapes: get_env(opts, :max_shapes)`
	`131`	`+ max_shapes: get_env(opts, :max_shapes),`
	`132`	`+ tweaks: [publication_alter_debounce_ms: get_env(opts, :publication_alter_debounce_ms)]`
`132`	`133`	`)`
`133`	`134`	`end`
`134`	`135`