feat: increase pooling rate on no changes

To reduce overall load of the system we will start to reduce pooling speed when no changes are not registered

Author: filipecabaco

lib/extensions/postgres_cdc_rls/replication_poller.ex

@@ -1,6 +1,6 @@
 defmodule Extensions.PostgresCdcRls.ReplicationPoller do
   @moduledoc """
-  Polls the write ahead log, applies row level sucurity policies for each subscriber
+  Polls the write ahead log, applies row level security policies for each subscriber
   and broadcast records to the `MessageDispatcher`.
   """
 
@@ -46,6 +46,7 @@ defmodule Extensions.PostgresCdcRls.ReplicationPoller do
       publication: extension["publication"],
       retry_ref: nil,
       retry_count: 0,
+      empty_count: 0,
       slot_name: extension["slot_name"] <> slot_name_suffix(),
       tenant_id: tenant_id,
       rate_counter_args: rate_counter_args,
@@ -64,6 +65,7 @@ defmodule Extensions.PostgresCdcRls.ReplicationPoller do
     {:noreply, state, {:continue, :prepare}}
   end
 
+  @impl true
   def handle_continue(:prepare, state) do
     {:noreply, prepare_replication(state)}
   end
@@ -72,46 +74,40 @@ defmodule Extensions.PostgresCdcRls.ReplicationPoller do
   def handle_info(
         :poll,
         %{
-          backoff: backoff,
-          poll_interval_ms: poll_interval_ms,
-          poll_ref: poll_ref,
-          publication: publication,
-          retry_ref: retry_ref,
-          retry_count: retry_count,
+          conn: conn,
           slot_name: slot_name,
-          max_record_bytes: max_record_bytes,
+          publication: publication,
           max_changes: max_changes,
-          conn: conn,
+          max_record_bytes: max_record_bytes,
           tenant_id: tenant_id,
           subscribers_nodes_table: subscribers_nodes_table,
           rate_counter_args: rate_counter_args
         } = state
       ) do
-    cancel_timer(poll_ref)
-    cancel_timer(retry_ref)
+    cancel_timer(state.poll_ref)
+    cancel_timer(state.retry_ref)
 
-    args = [conn, slot_name, publication, max_changes, max_record_bytes]
-    {time, list_changes} = :timer.tc(Replications, :list_changes, args)
+    {time, list_changes} = :timer.tc(Replications, :list_changes, [conn, slot_name, publication, max_changes, max_record_bytes])
     record_list_changes_telemetry(time, tenant_id)
 
-    case handle_list_changes_result(list_changes, subscribers_nodes_table, tenant_id, rate_counter_args) do
-      {:ok, :peek_empty} ->
-        Backoff.reset(backoff)
-        poll_ref = Process.send_after(self(), :poll, poll_interval_ms)
-        {:noreply, %{state | backoff: backoff, poll_ref: poll_ref}}
+    case dispatch_changes(list_changes, subscribers_nodes_table, tenant_id, rate_counter_args) do
+      :peek_empty ->
+        empty_count = state.empty_count + 1
+        interval = adaptive_interval(state.poll_interval_ms, empty_count)
+        poll_ref = Process.send_after(self(), :poll, interval)
 
-      {:ok, row_count} ->
-        Backoff.reset(backoff)
+        {:noreply, %{state | backoff: Backoff.reset(state.backoff), poll_ref: poll_ref, empty_count: empty_count}}
 
-        pool_ref =
+      {:ok, row_count} ->
+        poll_ref =
           if row_count > 0 do
             send(self(), :poll)
             nil
           else
-            Process.send_after(self(), :poll, poll_interval_ms)
+            Process.send_after(self(), :poll, state.poll_interval_ms)
           end
 
-        {:noreply, %{state | backoff: backoff, poll_ref: pool_ref}}
+        {:noreply, %{state | backoff: Backoff.reset(state.backoff), poll_ref: poll_ref, empty_count: 0}}
 
       {:error, %Postgrex.Error{postgres: %{code: :object_in_use, message: msg}}} ->
         log_error("ReplicationSlotBeingUsed", msg)
@@ -122,26 +118,19 @@ defmodule Extensions.PostgresCdcRls.ReplicationPoller do
 
         Logger.warning("Database PID #{db_pid} found in pg_stat_activity with state_change diff of #{diff}")
 
-        if retry_count > 3 do
+        if state.retry_count > 3 do
           case Replications.terminate_backend(conn, slot_name) do
             {:ok, :terminated} -> Logger.warning("Replication slot in use - terminating")
             {:error, :slot_not_found} -> Logger.warning("Replication slot not found")
             {:error, error} -> Logger.warning("Error terminating backend: #{inspect(error)}")
           end
         end
 
-        {timeout, backoff} = Backoff.backoff(backoff)
-        retry_ref = Process.send_after(self(), :retry, timeout)
-
-        {:noreply, %{state | backoff: backoff, retry_ref: retry_ref, retry_count: retry_count + 1}}
+        {:noreply, schedule_retry(state)}
 
       {:error, reason} ->
         log_error("PoolingReplicationError", reason)
-
-        {timeout, backoff} = Backoff.backoff(backoff)
-        retry_ref = Process.send_after(self(), :retry, timeout)
-
-        {:noreply, %{state | backoff: backoff, retry_ref: retry_ref, retry_count: retry_count + 1}}
+        {:noreply, schedule_retry(state)}
     end
   end
 
@@ -158,22 +147,34 @@ defmodule Extensions.PostgresCdcRls.ReplicationPoller do
     end
   end
 
+  @max_adaptive_increase 500
+  @empty_polls_per_step 10
+  @step_increase 100
+
+  def adaptive_interval(base_interval, empty_count) do
+    increase = div(empty_count, @empty_polls_per_step) * @step_increase
+    min(base_interval + increase, base_interval + @max_adaptive_increase)
+  end
+
   defp convert_errors([_ | _] = errors), do: errors
 
   defp convert_errors(_), do: nil
 
-  defp prepare_replication(%{backoff: backoff, conn: conn, slot_name: slot_name, retry_count: retry_count} = state) do
-    case Replications.prepare_replication(conn, slot_name) do
+  defp schedule_retry(state) do
+    {timeout, backoff} = Backoff.backoff(state.backoff)
+    retry_ref = Process.send_after(self(), :retry, timeout)
+    %{state | backoff: backoff, retry_ref: retry_ref, retry_count: state.retry_count + 1}
+  end
+
+  defp prepare_replication(state) do
+    case Replications.prepare_replication(state.conn, state.slot_name) do
       {:ok, _} ->
         send(self(), :poll)
         state
 
       {:error, error} ->
         log_error("PoolingReplicationPreparationError", error)
-
-        {timeout, backoff} = Backoff.backoff(backoff)
-        retry_ref = Process.send_after(self(), :retry, timeout)
-        %{state | backoff: backoff, retry_ref: retry_ref, retry_count: retry_count + 1}
+        schedule_retry(state)
     end
   end
 
@@ -185,15 +186,15 @@ defmodule Extensions.PostgresCdcRls.ReplicationPoller do
     )
   end
 
-  defp handle_list_changes_result(
+  defp dispatch_changes(
          {:ok, %Postgrex.Result{num_rows: 1, rows: [[nil, nil, nil, _, _, _, nil, [], ["peek_empty"]]]}},
          _subscribers_nodes_table,
          _tenant_id,
          _rate_counter_args
        ),
-       do: {:ok, :peek_empty}
+       do: :peek_empty
 
-  defp handle_list_changes_result(
+  defp dispatch_changes(
          {:ok,
           %Postgrex.Result{
             columns: columns,
@@ -249,8 +250,8 @@ defmodule Extensions.PostgresCdcRls.ReplicationPoller do
     {:ok, rows_count}
   end
 
-  defp handle_list_changes_result({:ok, _}, _, _, _), do: {:ok, 0}
-  defp handle_list_changes_result({:error, reason}, _, _, _), do: {:error, reason}
+  defp dispatch_changes({:ok, _}, _, _, _), do: {:ok, 0}
+  defp dispatch_changes({:error, reason}, _, _, _), do: {:error, reason}
 
   defp collect_subscription_nodes(subscribers_nodes_table, subscription_ids) do
     Enum.reduce_while(subscription_ids, {:ok, %{}}, fn subscription_id, {:ok, acc} ->
@@ -266,7 +267,7 @@ defmodule Extensions.PostgresCdcRls.ReplicationPoller do
       end
     end)
   rescue
-    _ -> {:error, :node_not_found}
+    ArgumentError -> {:error, :node_not_found}
   end
 
   def generate_record([

test/realtime/extensions/cdc_rls/replication_poller_test.exs

@@ -575,6 +575,142 @@ defmodule Realtime.Extensions.PostgresCdcRls.ReplicationPollerTest do
     end
   end
 
+  describe "poll adaptive interval" do
+    setup do
+      tenant = Containers.checkout_tenant(run_migrations: true)
+
+      subscribers_pids_table = :ets.new(:"#{__MODULE__}.adaptive_pids", [:public, :bag])
+      subscribers_nodes_table = :ets.new(:"#{__MODULE__}.adaptive_nodes", [:public, :set])
+
+      args =
+        hd(tenant.extensions).settings
+        |> Map.put("id", tenant.external_id)
+        |> Map.put("subscribers_pids_table", subscribers_pids_table)
+        |> Map.put("subscribers_nodes_table", subscribers_nodes_table)
+
+      %{args: args}
+    end
+
+    test "peek_empty polls slow down over time", %{args: args} do
+      test_pid = self()
+      peek_empty_result = {:ok, %Postgrex.Result{num_rows: 1, rows: [[nil, nil, nil, nil, nil, nil, nil, [], ["peek_empty"]]]}}
+
+      stub(Replications, :list_changes, fn _, _, _, _, _ ->
+        send(test_pid, {:polled_at, System.monotonic_time(:millisecond)})
+        peek_empty_result
+      end)
+
+      reject(&TenantBroadcaster.pubsub_direct_broadcast/6)
+      reject(&TenantBroadcaster.pubsub_broadcast/5)
+
+      start_link_supervised!({Poller, args})
+
+      # Collect enough polls to cross the 10-poll threshold where interval increases
+      timestamps =
+        for _ <- 1..15 do
+          assert_receive {:polled_at, ts}, 2_000
+          ts
+        end
+
+      early_gaps =
+        timestamps
+        |> Enum.take(5)
+        |> Enum.chunk_every(2, 1, :discard)
+        |> Enum.map(fn [a, b] -> b - a end)
+
+      late_gaps =
+        timestamps
+        |> Enum.drop(10)
+        |> Enum.chunk_every(2, 1, :discard)
+        |> Enum.map(fn [a, b] -> b - a end)
+
+      avg_early = Enum.sum(early_gaps) / length(early_gaps)
+      avg_late = Enum.sum(late_gaps) / length(late_gaps)
+
+      # After 10+ peek_empty polls, interval should have increased by at least 100ms
+      assert avg_late > avg_early,
+             "Expected later polls to be slower (#{avg_late}ms) than early polls (#{avg_early}ms)"
+    end
+
+    test "empty results (RLS filtered) poll at base interval without slowing down", %{args: args} do
+      test_pid = self()
+      empty_result = {:ok, %Postgrex.Result{rows: [], num_rows: 0}}
+
+      stub(Replications, :list_changes, fn _, _, _, _, _ ->
+        send(test_pid, {:polled_at, System.monotonic_time(:millisecond)})
+        empty_result
+      end)
+
+      reject(&TenantBroadcaster.pubsub_direct_broadcast/6)
+      reject(&TenantBroadcaster.pubsub_broadcast/5)
+
+      start_link_supervised!({Poller, args})
+
+      timestamps =
+        for _ <- 1..4 do
+          assert_receive {:polled_at, ts}, 1_000
+          ts
+        end
+
+      gaps = timestamps |> Enum.chunk_every(2, 1, :discard) |> Enum.map(fn [a, b] -> b - a end)
+
+      # All gaps should be roughly the base interval (100ms), none growing
+      for gap <- gaps do
+        assert gap < 200, "Expected base interval polling (~100ms), got #{gap}ms"
+      end
+    end
+
+    test "rows with changes trigger immediate re-poll without delay", %{args: args} do
+      test_pid = self()
+
+      results_with_changes =
+        build_result([
+          <<71, 36, 83, 212, 168, 9, 17, 240, 165, 186, 118, 202, 193, 157, 232, 187>>
+        ])
+
+      stub(Replications, :list_changes, fn _, _, _, _, _ ->
+        send(test_pid, {:polled_at, System.monotonic_time(:millisecond)})
+        results_with_changes
+      end)
+
+      stub(TenantBroadcaster, :pubsub_broadcast, fn _, _, _, _, _ -> :ok end)
+
+      start_link_supervised!({Poller, args})
+
+      timestamps =
+        for _ <- 1..3 do
+          assert_receive {:polled_at, ts}, 1_000
+          ts
+        end
+
+      gaps = timestamps |> Enum.chunk_every(2, 1, :discard) |> Enum.map(fn [a, b] -> b - a end)
+
+      # With changes, send(self(), :poll) means near-immediate re-poll
+      for gap <- gaps do
+        assert gap < 50, "Expected immediate re-poll, got #{gap}ms delay"
+      end
+    end
+  end
+
+  describe "adaptive_interval/2" do
+    test "returns base interval when empty_count is below threshold" do
+      assert Poller.adaptive_interval(100, 0) == 100
+      assert Poller.adaptive_interval(100, 9) == 100
+    end
+
+    test "increases by 100ms for every 10 empty polls" do
+      assert Poller.adaptive_interval(100, 10) == 200
+      assert Poller.adaptive_interval(100, 20) == 300
+      assert Poller.adaptive_interval(100, 30) == 400
+    end
+
+    test "caps at base + 500ms" do
+      assert Poller.adaptive_interval(100, 50) == 600
+      assert Poller.adaptive_interval(100, 100) == 600
+      assert Poller.adaptive_interval(100, 1000) == 600
+    end
+  end
+
   describe "slot_name_suffix/0" do
     setup do
       slot_name_suffix = Application.get_env(:realtime, :slot_name_suffix)